Example: Set up Filebeat modules to work with Kafka and Logstash

If you haven’t already set up the Filebeat index template and sample Kibana dashboards, run the Filebeat setup command to do that now:

filebeat -e setup

The -e flag is optional and sends output to standard error instead of syslog.

A connection to Elasticsearch and Kibana is required for this one-time setup step because Filebeat needs to create the index template in Elasticsearch and load the sample dashboards into Kibana. For more information about configuring the connection to Elasticsearch, see the Filebeat quick start.

After the template and dashboards are loaded, you’ll see the message INFO {kib} dashboards successfully loaded. Loaded dashboards.

Run the modules enable command to enable the modules that you want to run. For example:

filebeat modules enable system

You can further configure the module by editing the config file under the Filebeat modules.d directory. For example, if the log files are not in the location expected by the module, you can set the var.paths option.

Run the setup command with the --pipelines and --modules options specified to load ingest pipelines for the modules you’ve enabled. This step also requires a connection to Elasticsearch. If you want use a Logstash pipeline instead of ingest node to parse the data, skip this step.

filebeat setup --pipelines --modules system

Configure Filebeat to send log lines to Kafka. To do this, in the filebeat.yml config file, disable the Elasticsearch output by commenting it out, and enable the Kafka output. For example:

#output.elasticsearch:
  #hosts: ["localhost:9200"]
output.kafka:
  hosts: ["kafka:9092"]
  topic: "filebeat"
  codec.json:
    pretty: false

Start Filebeat. For example:

filebeat -e

Filebeat will attempt to send messages to Logstash and continue until Logstash is available to receive them.

On the system where Logstash is installed, create a Logstash pipeline configuration that reads from a Kafka input and sends events to an Elasticsearch output:

input {
  kafka {
    bootstrap_servers => "myhost:9092"
    topics => ["filebeat"]
    codec => json
  }
}

output {
  if [@metadata][pipeline] {
    elasticsearch {
      hosts => "https://myEShost:9200"
      manage_template => false
      index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
      pipeline => "%{[@metadata][pipeline]}" 
      user => "elastic"
      password => "secret"
    }
  } else {
    elasticsearch {
      hosts => "https://myEShost:9200"
      manage_template => false
      index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
      user => "elastic"
      password => "secret"
    }
  }
}

If you want use a Logstash pipeline instead of ingest node to parse the data, see the filter and output settings in the examples under Use Logstash pipelines for parsing.

Start Logstash, passing in the pipeline configuration file you just defined. For example:

bin/logstash -f mypipeline.conf

Logstash should start a pipeline and begin receiving events from the Kafka input.