Having problem Parsing Message via ingest pipeline

[Alaxwora (julz)]

Hello,
I was trying to parse CSV files of Message Tracking Logs with filebeat and elasticsearch ingest pipeline, so I used Grok Debugger to create a pattern that matches the message and I had no error then I added the log path of Message tracking logs in filebeat inputs section and added pipeline Id in output.elasticsearch section here is the the error message after running filebeat

Below is the pattern I used in creating pipeline

 
  PUT _ingest/pipeline/GrokTest
{
  "description" : "Convert Exchange csv data to indexed data",
  "processors" : [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{TIMESTAMP_ISO8601:date-time},%{DATA:client-ip},%{DATA:client-hostname},%{DATA:Server-ip},%{DATA:server-hostname},%{DATA:source-context},%{DATA:connector-id},%{DATA:source},%{DATA:event-id},%{DATA:internal-message-id},%{DATA:message-id},%{DATA:network-message-id},%{DATA:recipient-address},%{DATA:recipient-status},%{DATA:total-bytes},%{DATA:recipient-count},%{DATA:related-recipient-address},%{DATA:reference},%{DATA:message-subject},%{DATA:sender-address},%{DATA:return-path},%{DATA:message-info},%{DATA:directionality},%{DATA:tenant-id},%{DATA:original-client-ip},%{DATA:original-server-ip},%{DATA:custom-data},%{DATA:transport-traffic-type}"]

      }
    },
      {
        "remove" : {
          "field" : "message"
        }
      }],
  "on_failure": [
    {
      "set" : {
        "field" : "error",
        "value" : " - Error processing message - "
      }
    }
  ]
}

Log Message to parse

a#Software: Microsoft Exchange Server
#Version: 15.01.0225.037
#Log-type: Message Tracking Log
#Date: 2020-07-14T17:13:37.768Z
#Fields: date-time,client-ip,client-hostname,server-ip,server-hostname,source-context,connector-id,source,event-id,internal-message-id,message-id,network-message-id,recipient-address,recipient-status,total-bytes,recipient-count,related-recipient-address,reference,message-subject,sender-address,return-path,message-info,directionality,tenant-id,original-client-ip,original-server-ip,custom-data,transport-traffic-type
2020-07-14T17:13:38.009Z,10.0.0.2,QQQ0c.Random.CU,10.0.0.3,QQQ0c,08D9463CB6EE1755;2020-07-14T17:13:37.574Z;0,QQQ0c\Default QQQ0c,SMTP,RECEIVE,4449933333,<c579aad48e654c03ea75132a124446d4@Random.aa.ee>,1c62d03e-54c8-43e7-e5bc-05d946eab7f5,,,10429,1,,,0000003d-0000-0000-0000-0000b70d6900-MapiSubmitLAMProbe,HealthMailbox035ff7daed444848be3484e733389269@AAA.Random.cu,HealthMailbox035ff7daedec3848be3487354c583569@Random.cu.qw,0cI: ,Originating,,::1,10.0.0.3,S:FirstForestHop=QQQc.RandomCU;S:FromEntity=Hosted;S:ProxiedClientIPAddress=10.0.03;S:ProxiedClientHostname=QQQc.Random.CU;S:ProxyHop1=QQQc.Random.CU(10.0.0.3);S:DeliveryPriority=Normal;S:AccountForest=Random.cu;S:IsProbe=true;S:PersistProbeTrace=False,Email

filebeat config

[Felix_Roessel]

Wouldn't it be easier using the CSV processor directly?

CSV processor | Elasticsearch Guide [master] | Elastic

[Alaxwora (julz)]

in this case, Yes I used dissect, CSV and Grok and CSV was the best choice but my main problem the documents are not parsed even after adding Pipeline Id in the config file

[Felix_Roessel]

Your Screenshot has truncated the error message. I can just see that it fails. Usually it also says why it fails.

Can you add the full error message so what we can better read it? Also your example log starts with a before the # sign.. If this is not a typo then you need to exclude this line also.

[Alaxwora (julz)]

That was the full error the rest lines in PowerShell showing the same error messages for the new lines in log file

[Felix_Roessel]

This Error says that you want to ingest a document that has a conflicting data structure / mapping for field source.
The existing field has a concrete value while your new document wants to ingest an object (multiple fields) .
If you dont need the existing data the easiest way is to delete the index.
If you need the existing data you can switch the index or change the target field for your parsing.

[Alaxwora (julz)]

I changed the index "source" to another name and it worked
thanks <3