# Kafka input kibana message one field

**URL:** https://discuss.elastic.co/t/kafka-input-kibana-message-one-field/296134
**Category:** Logstash
**Created:** [February 3, 2022, 12:15am UTC](https://discuss.elastic.co/t/kafka-input-kibana-message-one-field/296134 "2022-02-03T00:15:08Z")
**Posts on this page:** 4
**Page:** 1

<div class="post-metadata">

### Author: ![splitmessage88](https://avatars.discourse-cdn.com/v4/letter/s/a87d85/32.png) [@splitmessage88](https://discuss.elastic.co/u/splitmessage88)
#### Post date: [February 3, 2022, 12:15am UTC](https://discuss.elastic.co/t/kafka-input-kibana-message-one-field/296134/1 "2022-02-03T00:15:08Z")

</div>

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

```auto
    "@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"

```

```auto
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.

---

<div class="post-metadata">

### Author: ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [February 3, 2022, 12:32am UTC](https://discuss.elastic.co/t/kafka-input-kibana-message-one-field/296134/2 "2022-02-03T00:32:44Z")

</div>

Use a [json](https://www.elastic.co/guide/en/logstash/current/plugins-filters-json.html) filter

```
json { source => "message" }
```

---

<div class="post-metadata">

### Author: ![splitmessage88](https://avatars.discourse-cdn.com/v4/letter/s/a87d85/32.png) [@splitmessage88](https://discuss.elastic.co/u/splitmessage88)
#### Post date: [February 3, 2022, 7:00am UTC](https://discuss.elastic.co/t/kafka-input-kibana-message-one-field/296134/3 "2022-02-03T07:00:31Z")

</div>

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.

 ![kibana 1](https://us1.discourse-cdn.com/elastic/original/3X/a/2/a21086f71d078aa8b7f5e396155b6e8c7595435a.jpeg)

 ![kibana 2](https://us1.discourse-cdn.com/elastic/original/3X/b/d/bdccaaab651fc4ff3ccfc6dd67a7757760f6cc9c.jpeg)

 ![kibana 3](https://us1.discourse-cdn.com/elastic/original/3X/7/6/76685c20ddffb81c6e1c376e4e16d9165127390c.jpeg)

Any clue?

Thanks.

---

<div class="post-metadata">

### Author: ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)
#### Post date: [March 3, 2022, 7:00am UTC](https://discuss.elastic.co/t/kafka-input-kibana-message-one-field/296134/4 "2022-03-03T07:00:34Z")

</div>

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