Issue with Filebeats Kafka module

[ryan.dyer]

I am having an issue with the kafka module where it does not appear to process the lines properly. I have enabled debugging of filebeat and see the following:
2018-03-16T18:36:54Z DBG [publish] Publish event: { "@timestamp": "2018-03-16T18:36:54.052Z", "@metadata": { "beat": "filebeat", "type": "doc", "version": "6.1.2", "pipeline": "filebeat-6.1.2-kafka-log-pipeline" }, "source": "/var/log/kafka/controller.log", "offset": 84336, "message": "[2018-03-16 18:36:45,144] DEBUG [Controller 2]: topics not in preferred replica Map() (kafka.controller.KafkaController)", "fileset": { "name": "log", "module": "kafka" }, "prospector": { "type": "log" }, "beat": { "name": "eventing2030-ct-kafka-2", "hostname": "eventing2030-ct-kafka-2", "version": "6.1.2" } }

The line appears to match what the module shows so I'm not sure why its not being processed. The syslog module on the other hand is working fine.

I am referring to this https://www.elastic.co/guide/en/beats/filebeat/master/filebeat-module-kafka.html

Based on this document I expect filebeats to read the log lines, parse it and create appropriate fields before sending it to kafka(our logging buffer; not the same as the kafka where the filebeats is running)

Any assistance would be appreciated.

Thanks,
Ryan

[kvch (Noémi Ványi)]

Could you please share more debug logs of Filebeat and its config file? Please, format them using </>.

[ryan.dyer]

filebeat.yml

  --- 
fields_under_root: true
filebeat.config.modules: 
enabled: true
path: "${path.config}/modules.d/*.yml"
filebeat.prospectors: 
- 
    fields: 
    type: json
    fields_under_root: true
    json.add_error_key: true
    json.keys_under_root: true
    json.overwrite_keys: true
    paths: 
    - /var/log/**/*.json
    type: log
- 
    backoff: 1s
    clean_removed: true
    close_eof: false
    close_inactive: 5m
    close_removed: true
    close_renamed: false
    exclude_files: 
    - .gz$
    - /var/log/storm/access-drpc.log
    - /var/log/syslog
    - /var/log/auth.log
    - /var/log/kafka/
    fields_under_root: false
    ignore_older: 30m
    max_backoff: 10s
    paths: 
    - /var/log/**/*.log
    - /opt/hsp_service/logs/**/*.log
    - /opt/nitro/**/*.log
    scan_frequency: 10s
    type: log
logging.files: ~
logging.to_files: true
logging.to_syslog: true
output.kafka: 
enabled: true
hosts: "127.0.0.1"
topic: filebeat
processors: 
- 
    decode_json_fields: 
    fields: 
        - logLine
    target: logLine_json
    when: 
        equals: 
        type: json
recursive_glob.enabled: true

modules.d/kafka.yml

- module: kafka
log:
    enabled: true
    var.paths:
    - "/var/log/kafka/controller.log*"
    - "/var/log/kafka/server.log*"
    - "/var/log/kafka/state-change.log*"
    - "/var/log/kafka/kafka-*.log*"
    exclude_files: [".gz$"]

syslog event

2018-05-02T22:00:28Z DBG  [publish] Publish event: {
"@timestamp": "2018-05-02T22:00:23.881Z",
"@metadata": {
    "beat": "filebeat",
    "type": "doc",
    "version": "6.1.2",
    "pipeline": "filebeat-6.1.2-system-syslog-pipeline"
},
"source": "/var/log/syslog",
"offset": 5038,
"message": "May  2 21:59:55 eventing2030-ct-kafka-1 systemd[1]: Started filebeat.",
"prospector": {
    "type": "log"
},
"fileset": {
    "name": "syslog",
    "module": "system"
},
"beat": {
    "name": "eventing2030-ct-kafka-1",
    "hostname": "eventing2030-ct-kafka-1",
    "version": "6.1.2"
}
}

kafka event

2018-03-16T18:36:54Z DBG [publish] Publish event: { "@timestamp": "2018-03-16T18:36:54.052Z", "@metadata": { "beat": "filebeat", "type": "doc", "version": "6.1.2", "pipeline": "filebeat-6.1.2-kafka-log-pipeline" }, "source": "/var/log/kafka/controller.log", "offset": 84336, "message": "[2018-03-16 18:36:45,144] DEBUG [Controller 2]: topics not in preferred replica Map() (kafka.controller.KafkaController)", "fileset": { "name": "log", "module": "kafka" }, "prospector": { "type": "log" }, "beat": { "name": "eventing2030-ct-kafka-2", "hostname": "eventing2030-ct-kafka-2", "version": "6.1.2" } }

From kibana when I search for the syslog, I have additional fields specific for syslog. The kafka one, not so much.