Kibana Message Parse

[Kumbum]

I have a custom log file in a source machine and it comes as a single-line log through "message" attribute to the Dashboard

message = <INFO/ERROR/FATAL, etc>, , , , , ,

I want this message gets split as below.

message.log type or Log type= Info
message.TimeStamp or TimeStamp = Date&Time
and so on

How to split this in the source machine before it gets streamed to the dashboard?

[nickpeihl]

Hi @Kumbum. You did not say how you are ingesting your logs. Are you using Logstash or Filebeat or something else?

I usually recommend defining a custom Dissect pattern in Logstash. Or use the Dissect processor on an ingest pipeline with Filebeat.

[Kumbum]

@nickpeihl

I use filebeat

[nickpeihl]

Hi @Kumbum. Thanks. You can add your custom dissect pattern directly to your Filebeat config.

Or you can create an Ingest pipeline that uses the Dissect processor. After you create the ingest pipeline you can configure Filebeat to use the pipeline.

[Kumbum]

Hi @nickpeihl

In filebeat config file, I have added custom dissect as below and restarted the filebeat service but still message show in a single line in the kibana, doesn't split.

processors:

• fingerprint:
  fields: ["message"]
  target_field: "@metadata._id"

• add_host_metadata:
  when.not.contains.tags: forwarded

• add_cloud_metadata: ~

• add_docker_metadata: ~

• add_kubernetes_metadata: ~

• drop_event:
  when:
  regexp:
  message: "^ErrorType,"

• dissect:
  tokenizer: '"%{LOG_LEVELS},%{TIMESTAMP},%{MACHINE_NAME},%{USER},%{TASK_NAME},%{PROCESS_NAME},%{DESCRIPTION},%{OtherInformation}"'
  field: "message"
  target_prefix: ""

Kibana Results

==================
A Sample message contains:

INFO,2023/07/07 11:23:59,,,<Task_Name>,Naga Raju Kumbum,"Start","Process"
INFO,2023/07/07 11:23:59,,,<Task_Name>,Naga Raju Kumbum,"End","Process"

[nickpeihl]

Hi @Kumbum. The message field is still going to be added to the index unless you specifically add a drop_fields processor.

But we should make sure that the new fields are being created. Perhaps it will help to use the Simulate Ingest Pipeline API to check if the pattern is successful.

Try running the following in Kibana - Stack Management - DevTools.

POST _ingest/pipeline/_simulate
{
  "pipeline": {
    "description": "foo",
    "processors": [
      {
        "dissect": {
          "field": "message",
          "pattern": """%{LOG_LEVELS},%{TIMESTAMP},%{MACHINE_NAME},%{USER},%{TASK_NAME},%{PROCESS_NAME},%{DESCRIPTION},%{OtherInformation}"""
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "index",
      "_id": "id",
      "_source": {
        "message": "INFO,2023/07/07 11:23:59,,,<Task_Name>,Naga Raju Kumbum,Start,Process"
      }
    },
    {
      "_index": "index",
      "_id": "id",
      "_source": {
        "message": "INFO,2023/07/07 11:23:59,,,<Task_Name>,Naga Raju Kumbum,End,Process"
      }
    }
  ]
}

Does the simulation return with the expected fields?

[Matthew_Sturtevant]

Hi @nickpeihl We are writing everything below in the filebeat.yml and are getting an issue where filebeat will not initialize on the client machine with the dissect changes. We ran your code above in the dev tools and it appears the logs are getting parsed out as expected. Do you have any idea why the code pasted below is failing?

processors:
  -dissect: 
      field: ["message"]
      pattern: """%{LOG_LEVELS},%{TIMESTAMP},%{MACHINE_NAME},%{USER},%{TASK_NAME},%{PROCESS_NAME},%{DESCRIPTION},%{OtherInformation}"""

  - fingerprint:
      fields: ["message"]
      target_field: "@metadata._id"
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - add_docker_metadata: ~
  - add_kubernetes_metadata: ~
  - drop_event:
      when:
        regexp:
          message: "^ErrorType,"

[nickpeihl]

Hi @Matthew_Sturtevant. One thing that immediately jumps out to me is the second line. There should be a space between the "-" and "dissect", e.g. - dissect:.

processors:
  - dissect: 
      field: ["message"]
      pattern: """%{LOG_LEVELS},%{TIMESTAMP},%{MACHINE_NAME},%{USER},%{TASK_NAME},%{PROCESS_NAME},%{DESCRIPTION},%{OtherInformation}"""
...

Aside from that, can you tell me if there are any errors from Filebeat?

[Kumbum]

@nickpeihl
We have updated dissect properly(Gave a space) but now unable to restart the filebeat service as it throws the following error. Not sure what misses here.

processors:
  - dissect: 
      field: ["message"]
      pattern: """%{LOG_LEVELS},%{TIMESTAMP},%{MACHINE_NAME},%{USER},%{TASK_NAME},%{PROCESS_NAME},%{DESCRIPTION},%{OtherInformation}"""
  - fingerprint:
      fields: ["message"]
      target_field: "@metadata._id"
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - add_docker_metadata: ~
  - add_kubernetes_metadata: ~
  - drop_event:
      when:
        regexp:
          message: "^ErrorType,"

Error:

[nickpeihl]

Hi @Kumbum. I'm not an expert on Filebeat and this error is beyond my knowledge. Do you mind opening a new topic with your details in the Beats category.

[Kumbum]

Hi @nickpeihl

Below one I got from another discussion forum and working as I want.

https://discuss.elastic.co/t/add-custom-field-and-its-value-needs-to-be-derived-from-one-of-the-source-field/272119/5

  - dissect:
      tokenizer: '%{LOG_LEVELS},%{TIMESTAMP},%{MACHINE_NAME},%{USER},%{TASK_NAME},%{PROCESS_NAME},%{DESCRIPTION},%{OtherInformation}'
      field: message
      target_prefix: ""
      overwrite_keys: true