Kafka input kibana message one field

[splitmessage88]

Hello!

I'm new here and have been stuck with an issue for 2 days.

I have a working flow with winlogbeat - kafka - logstash - Elasticsearch.

I receive the winlogbeat logs in kibana but the problem is that the full log is contained in the "message" field. I want to be able to split the message into separate fields so I can search my logs easier. Below is my config.

rubydebug

    "@timestamp" => 2022-02-03T00:00:03.066Z,
       "message" => "{\"@timestamp\":\"2022-02-02T22:59:38.580Z\",\"@metadata\":{\"beat\":\"winlogbeat\",\"type\":\"_doc\",\"version\":\"7.16.3\"},\"log\":{\"level\":\"information\"},\"message\":\"Cryptographic operation.\\n\\nSubject:\\n\\tSecurity ID:\\t\\tS-1-5-18\\n\\tAccount Name:\\t\\tXXX$\\n\\tAccount Domain:\\t\\tXXX\\n\\tLogon ID:\\t\\t0x3E7\\n\\nCryptographic Parameters:\\n\\tProvider Name:\\tMicrosoft Software Key Storage Provider\\n\\tAlgorithm Name:\\tRSA\\n\\tKey Name:\\tbfaacff2-115f-2df5-4db7-434a81582a16\\n\\tKey Type:\\tUser key.\\n\\nCryptographic Operation:\\n\\tOperation:\\tOpen Key.\\n\\tReturn Code:\\t0x0\",\"host\":{\"os\":{\"platform\":\"windows\",\"version\":\"10.0\",\"family\":\"windows\",\"name\":\"Windows Server 2019 Standard\",\"kernel\":\"10.0.17763.1577 (WinBuild.160101.0800)\",\"build\":\"17763.1577\",\"type\":\"windows\"},\"id\":\"491222d5-4e07-4a32-9080-05cf9de8600c\",\"ip\":[\"fe80::c551:cc3d:2138:4452\",\"10.99.55.87\"],\"name\":\"XXX\",\"mac\":[\"00:0c:29:9f:89:08\"],\"hostname\":\"XXX\",\"architecture\":\"x86_64\"},\"agent\":{\"name\":\"XXX"type\":\"winlogbeat\",\"version\":\"7.16.3\",\"hostname\":\"XXX\",\"ephemeral_id\":\"3b43a8e6-f1dd-4b3f-ac45-6c8839118345\",\"id\":\"aec98a48-98b6-4039-874c-b536d3e22e2c\"},\"ecs\":{\"version\":\"1.12.0\"},\"winlog\":{\"process\":{\"pid\":636,\"thread\":{\"id\":2884}},\"event_id\":\"5061\",\"task\":\"System Integrity\",\"api\":\"wineventlog\",\"provider_name\":\"Microsoft-Windows-Security-Auditing\",\"computer_name\":\"XXX\",\"provider_guid\":\"{54849625-5478-4994-a5ba-3e3b0328c30d}\",\"event_data\":{\"KeyType\":\"%%2500\",\"SubjectUserName\":\"XXX$\",\"KeyName\":\"bfaacff2-115f-2df5-4db7-434a81582a16\",\"SubjectDomainName\":\"XXX\",\"ProviderName\":\"Microsoft Software Key Storage Provider\",\"Operation\":\"%%2480\",\"SubjectLogonId\":\"0x3e7\",\"AlgorithmName\":\"RSA\",\"ReturnCode\":\"0x0\",\"SubjectUserSid\":\"S-1-5-18\"},\"record_id\":1425351,\"keywords\":[\"Audit Success\"],\"opcode\":\"Info\",\"channel\":\"Security\"},\"event\":{\"kind\":\"event\",\"provider\":\"Microsoft-Windows-Security-Auditing\",\"outcome\":\"success\",\"action\":\"System Integrity\",\"created\":\"2022-02-02T22:59:40.600Z\",\"code\":\"5061\"}}",
      "@version" => "1"

input {
   kafka {
    bootstrap_servers => "10.99.55.57:9092"
    topics => ["test-topic"]
    decorate_events => true
    }

filter {
        split {
	}
}

output { stdout {codec => rubydebug } }

So my question is, what do I need to do to split the message into separate fields?

Thank you.

[Badger]

Use a json filter

json { source => "message" }

[splitmessage88]

Hello Badger,

That seems to fix the output for the rubydebug code. The message field is now seperated into own fields.

However, I can't see any events under kibana and the index seems to be out of order.

Any clue?

Thanks.