Fix unmatched quote and non-whitespaces in csv fields (Filebeat and Ingest Pipeline)

I am shipping csv data using Filebeat to the Elasticsearch output, I have an ingest pipeline in place to process the events but I am encountering two events failure:

  1. (status=400): {"type":"illegal_argument_exception","reason":"Unmatched quote"}
  2. (status=400): {"type":"illegal_argument_exception","reason":"Illegal character inside unquoted field at 71"}

Clearly failure one doesn't like the csv field value "........,\"mscorlib" and I think failure two is related to the non-whitespaces here "System.Activities.Statements.WorkflowTerminatedException: Action Failed\n at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()\n at \"

The csv data is genereted by UiPath RPA bots and I can't control the exported messages from source. How can I overcome the 2 issues by manipulating/processing the data either in Filebeat or the Pipeline? :smiley:

Below are the complete messages being sent in each of the events, and the ingest pipeline used:

Event Messages

Message 1

"message":"Error,Robot.tenantId:1.UiPath.Orchestrator.Application.Services.Logs.LogsWriter,29044,4,2021-04-18T09:30:13.0243363Z,Robot,3,2,\"mscorlib"

Message 2

"message": "System.Activities.Statements.WorkflowTerminatedException: Action Failed\n   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()\n   at UiPath.Executor.BaseRunner.EndExecute(IAsyncResult result)\n   at UiPath.Executor.InProcessRunner.EndExecute(IAsyncResult result)\n   at UiPath.Core.Activities.ExecutorInvokeActivity.EndExecute(AsyncCodeActivityContext context, IAsyncResult result)\n   at System.Activities.AsyncCodeActivity.System.Activities.IAsyncCodeActivity.FinishExecution(AsyncCodeActivityContext context, IAsyncResult result)\n   at System.Activities.AsyncCodeActivity.CompleteAsyncCodeActivityData.CompleteAsyncCodeActivityWorkItem.Execute(ActivityExecutor executor, BookmarkManager bookmarkManager)\",User,123,123\\123,123,123,1.0.64,123,123,13577,Main,123,,,\"{\"\"message\"\":\"\"mscorlib\\r\\nSystem.Activities.Statements.WorkflowTerminatedException: Action Failed\\r\\n   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw()\\r\\n   at UiPath.Executor.BaseRunner.EndExecute(IAsyncResult result)\\r\\n   at UiPath.Executor.InProcessRunner.EndExecute(IAsyncResult result)\\r\\n   at UiPath.Core.Activities.ExecutorInvokeActivity.EndExecute(AsyncCodeActivityContext context, IAsyncResult result)\\r\\n   at System.Activities.AsyncCodeActivity.System.Activities.IAsyncCodeActivity.FinishExecution(AsyncCodeActivityContext context, IAsyncResult result)\\r\\n   at System.Activities.AsyncCodeActivity.CompleteAsyncCodeActivityData.CompleteAsyncCodeActivityWorkItem.Execute(ActivityExecutor executor, BookmarkManager bookmarkManager)\"\",\"\"level\"\":\"\"Error\"\",\"\"logType\"\":\"\"User\"\",\"\"timeStamp\"\":\"\"2021-04-18T12:30:13.0243363+03:00\"\",\"\"fingerprint\"\":\"\"444\"\",\"\"windowsIdentity\"\":\"\"123\\\\123\"\",\"\"machineName\"\":\"\"21\"\",\"\"processName\"\":\"\"123\"\",\"\"processVersion\"\":\"\"1.0.64\"\",\"\"jobId\"\":\"\"12\"\",\"\"robotName\"\":\"\"123\"\",\"\"machineId\"\":2,\"\"organizationUnitId\"\":2,\"\"fileName\"\":\"\"Main\"\",\"\"logF_BusinessProcessName\"\":\"\"2\"\"}\",2,2,,,,,,,,,,1"

Ingest Pipeline

        "description" : "Parsing the local logs",
         "processors" : [
          {
            "csv" : {
              "field" : "message",
              "target_fields" : [
                "level",
                "logger",
                "user",
                "levelOrdinal",
                "timeStamp",
                "Source",
                "tenantId",
                "organizationUnitId",
                "message",
                "logType",
                "fingerprint",
                "windowsIdentity",
                "machineName",
                "processName",
                "processVersion",
                "jobId",
                "robotName",
                "machineId",
                "fileName",
                "indexName",
                "transactionExecutionTime",
                "totalExecutionTimeInSeconds",
                "rawMessage",
                "tenantKey",
                "userKey",
                "transactionStatus",
                "transactionState",
                "transactionId",
                "queueName",
                "processingExceptionType",
                "processingExceptionReason",
                "queueItemPriority",
                "queueItemReviewStatus",
                "activityInfo",
                "logF_BusinessProcessName"
              ]
            }
          },
          {
            "json" : {
              "field" : "rawMessage",
              "target_field" : "customFields"
            }
          },
          {
            "rename" : {
              "field" : "customFields",
              "target_field" : "{{indexName}}"
            }
          },
          {
          "date" : {
          "field" : "timeStamp",
          "formats" : ["ISO8601"]
            }
          },
          {
           "date" : {
           "field" : "timeStamp",
           "target_field" : "timeStamp",
             "formats" : ["ISO8601"]
          }
         },
           {
            "date_index_name" : {
              "field" : "@timestamp",
              "index_name_prefix" : "{{indexName}}-",
              "date_rounding" : "M",
              "index_name_format" : "yyyy.MM"
            }
          },
          {
            "remove" : {
              "field" : [
                "{{indexName}}.fileName",
                "{{indexName}}.fingerprint",
                "{{indexName}}.jobId",
                "{{indexName}}.level",
                "{{indexName}}.logType",
                "{{indexName}}.machineId",
                "{{indexName}}.machineName",
                "{{indexName}}.message",
                "{{indexName}}.organizationUnitId",
                "{{indexName}}.processingExceptionType",
                "{{indexName}}.processingExceptionReason",
                "{{indexName}}.processName",
                "{{indexName}}.processVersion",
                "{{indexName}}.queueItemReviewStatus",
                "{{indexName}}.queueItemPriority",
                "{{indexName}}.queueName",
                "{{indexName}}.robotName",
                "{{indexName}}.totalExecutionTime",
                "{{indexName}}.totalExecutionTimeInSeconds",
                "{{indexName}}.transactionExecutionTime",
                "{{indexName}}.timeStamp",
                "{{indexName}}.transactionId",
                "{{indexName}}.transactionState",
                "{{indexName}}.transactionStatus",
                "{{indexName}}.windowsIdentity"
              ],
              "ignore_missing" : true
            }
          },
          {
            "set" : {
              "field" : "_id",
              "value" : "{{fingerprint}}"
            }
          },
          {
            "remove" : {
              "field" : [
                "cloud",
                "log",
                "agent",
                "input",
                "ecs",
                "host",
                "_type"
              ],
              "ignore_missing" : true
            }
          },		
          {
            "convert" : {
              "field" : "machineId",
              "type" : "integer",
              "ignore_missing" : true
            }
          },
          {
            "convert" : {
              "field" : "levelOrdinal",
              "type" : "integer",
              "ignore_missing" : true
            }
          },
          {
            "convert" : {
              "field" : "organizationUnitId",
              "type" : "integer",
              "ignore_missing" : true
            }
          },
          {
            "convert" : {
              "field" : "tenantId",
              "type" : "integer",
              "ignore_missing" : true
            }
          },
          {
            "convert" : {
              "field" : "user",
              "type" : "integer",
              "ignore_missing" : true
            }
          },
          {
            "convert" : {
              "field" : "totalExecutionTimeInSeconds",
              "type" : "integer",
              "ignore_missing" : true
            }
          },
          {
            "convert" : {
              "field" : "transactionExecutionTime",
              "type" : "float",
              "ignore_missing" : true
            }
          }
        ]
    }

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.