# How to grok then send to kafka as JSON text?

**URL:** https://discuss.elastic.co/t/how-to-grok-then-send-to-kafka-as-json-text/201456
**Category:** Logstash
**Created:** [September 27, 2019, 6:48pm UTC](https://discuss.elastic.co/t/how-to-grok-then-send-to-kafka-as-json-text/201456 "2019-09-27T18:48:23Z")
**Posts on this page:** 2
**Page:** 1

<div class="post-metadata">

### Author: ![DeliciousPickle](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/deliciouspickle/32/55045_2.png) [@DeliciousPickle](https://discuss.elastic.co/u/DeliciousPickle)
#### Post date: [September 27, 2019, 6:48pm UTC](https://discuss.elastic.co/t/how-to-grok-then-send-to-kafka-as-json-text/201456/1 "2019-09-27T18:48:23Z")

</div>

In my Logstash config I want to read in a syslog, parse the message text to get some field values, then assemble those fields/values into a JSON string which I then output to a kafka topic. I must be missing something, because no matter what I do with my filter and output, all it ever sends to kafka is my original message string. logstash.conf source is shown below. Can anyone tell me what I need to do to send the JSON string to kafka instead of the original message?

```
input {
    syslog {
        port => 9601
    }
}

filter {
    metrics {
        meter => "events"
        add_tag => "metric"
    }
    if ! ( "metric" in [tags] ) {
        grok {
            match => {
                "message" => [
                    "(?<eventTime>\|FIELD1\|.*\|)",
                    "(?<localIpAddress>\|FIELD2\|.*\|)"
                ]
            }
        }
        json_encode {
            source => "eventTime"
            add_field => {
                # "eventTime" => "%{eventTime}"
                "localIpAddress" => "%{localIpAddress}"
            }
        }
    }
}

output {
    # stdout { codec => line }

    # only emit events with the 'metric' tag
    if "metric" in [tags] {
        stdout {
          codec => line {
              format => "events/second: %{[events][rate_1m]}"
          }
        }
    } else {
        kafka {
            codec => json_lines

            bootstrap_servers => (redacted)
            sasl_jaas_config => (redacted)
            topic_id => (redacted)
            security_protocol => (redacted)
            sasl_mechanism => (redacted)
        }
    }
}
```

---

<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: [October 25, 2019, 6:48pm UTC](https://discuss.elastic.co/t/how-to-grok-then-send-to-kafka-as-json-text/201456/2 "2019-10-25T18:48:24Z")

</div>

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