# Parsing filebeat -\> kafka -\> logstash

**URL:** <https://discuss.elastic.co/t/parsing-filebeat-kafka-logstash/85729>\
**Category:** Logstash\
**Created:** [May 15, 2017, 9:31am UTC](https://discuss.elastic.co/t/parsing-filebeat-kafka-logstash/85729 "2017-05-15T09:31:40Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![Dominic\_Prittie](https://avatars.discourse-cdn.com/v4/letter/d/94ad74/32.png) [@Dominic\_Prittie](https://discuss.elastic.co/u/Dominic_Prittie)\
**Post date:** [May 15, 2017, 9:31am UTC](https://discuss.elastic.co/t/parsing-filebeat-kafka-logstash/85729/1 "2017-05-15T09:31:40Z")

</div>

Hi,

I am using the following versions of software:  
filebeat: 5.4.0  
kafka: 0.10.0.1  
logstash: 5.4.0

When I directly connect filebeat to logstash my log's fields such as host and source get parsed sensibly, eg:

```auto
{
  "_index": "logstash-2017.05.15",
  "_type": "log",
  "_id": "AVwLa6bjfzTWzNZQ1MEK",
  "_version": 1,
  "_score": null,
  "_source": {
    "@timestamp": "2017-05-15T09:23:40.739Z",
    "offset": 13692098,
    "@version": "1",
    "beat": {
      "hostname": "host.domain.com",
      "name": "host.domain.com",
      "version": "5.4.0"
    },
    "input_type": "log",
    "host": "host.domain.com",
    "source": "/var/log/messages",
    "message": "May 15 10:23:39 host root: test",
    "type": "log",
    "tags": [
      "beats_input_codec_plain_applied"
    ]
  },
  "fields": {
    "@timestamp": [
      1494840220739
    ]
  },
  "sort": [
    1494840220739
  ]
}

```

When I use kafka as transport in between filebeat and logstash pretty much all the information remains in the message field, eg

```auto
{
  "_index": "logstash-2017.05.15",
  "_type": "logs",
  "_id": "AVwLRbwQL-zbtw-GlIKo",
  "_version": 1,
  "_score": null,
  "_source": {
    "@timestamp": "2017-05-15T08:42:18.502Z",
    "@version": "1",
    "message": "{\"@timestamp\":\"2017-05-15T08:42:16.927Z\",\"beat\":{\"hostname\":\"host.domain.com\",\"name\":\"host.domain.com\",\"version\":\"5.4.0\"},\"input_type\":\"log\",\"message\":\"May 15 09:42:16 host root: test\",\"offset\":10735391,\"source\":\"/var/log/messages\",\"type\":\"log\"}"
  },
  "fields": {
    "@timestamp": [
      1494837738502
    ]
  },
  "sort": [
    1494837738502
  ]
}

```

My configuration files:  
**logstash.conf**

```auto
input {
        beats {
                port => 5001
        }

        kafka {
                bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
                topics => ["logging"]
        }
}

filter {
}

output {
        elasticsearch {
                hosts => "elasticsearch:9200"
        }
}

```

**filebeat.yml**

```auto
filebeat.prospectors:
- input_type: log
  paths:
    - /var/log/*.log
    - /var/log/messages

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]
  topic: logging
  version: 0.10.0

output.logstash:
  hosts: ["10.64.3.108:5001"]

```

Does anyone know how I can configure logstash/filebeats so that I get the same behaviour when using kafka as transport as I do without kafka?

---

<div class="post-metadata">

**Author:** ![Dominic\_Prittie](https://avatars.discourse-cdn.com/v4/letter/d/94ad74/32.png) [@Dominic\_Prittie](https://discuss.elastic.co/u/Dominic_Prittie)\
**Post date:** [May 16, 2017, 3:14pm UTC](https://discuss.elastic.co/t/parsing-filebeat-kafka-logstash/85729/2 "2017-05-16T15:14:33Z")

</div>

solution was to set the codec to json for the kafka input:

```auto
        kafka {
                bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
                codec => "json"
                topics => ["logging"]
        }

```

---

<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:** [June 30, 2017, 5:40pm UTC](https://discuss.elastic.co/t/parsing-filebeat-kafka-logstash/85729/4 "2017-06-30T17:40:10Z")

</div>

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