Connecting Filebeat to CKafka

Introduction to Filebeats

Beats Platform hosts various single-purpose data shippers which can be used as lightweight agents after installation to send collected data from hundreds or thousands of machines to the target.

Beats offers a wide variety of shippers. You can download the most appropriate one based on your needs. This article takes Filebeats (lightweight log collector) as an example to introduce how to connect Filebeats to CKafka and how to solve it by FAQs.

Steps

Configure file

Follow these steps to configure the monitoring file:

#======= Filebeat prospectors ==========
filebeat.prospectors:

- input_type: log 

# Here is the path to the listener file
  paths:
    - /var/log/messages

#=======  Outputs =========

#------------------ kafka -------------------------------------
output.kafka:
  version:0.10.2 // Configuration based on the open source version of different CKafka instances
  # Set as the instance address
  hosts: ["127.1.2.3:9092"]
  # Set the target topic
  topic: 'test'
  partition.round_robin:
    reachable_only: false

  required_acks: 1
  compression: none
  max_message_bytes: 1000000

  # The following information needs to be configured for SASL. If not required, they can be ignored.
  username: "instance-will#user"
  password: "password"

Run the file

1. Run the following command to start the client.
   sudo ./filebeat -e -c filebeat.yml
2. Add data to the listener file (the example here is a testlog file written to the listener).

echo ckafka1 >> testlog
echo ckafka2 >> testlog
echo ckafka3 >> testlog

3. Start the consumer to consume the corresponding topic and get the following data.

{"@timestamp":"2017-09-29T10:01:27.936Z","beat":{"hostname":"10.193.9.26","name":"10.193.9.26","version":"5.6.2"},"input_type":"log","message":"ckafka1","offset":500,"source":"/data/ryanyyang/hcmq/beats/filebeat-5.6.2-linux-x86_64/testlog","type":"log"}
{"@timestamp":"2017-09-29T10:01:30.936Z","beat":{"hostname":"10.193.9.26","name":"10.193.9.26","version":"5.6.2"},"input_type":"log","message":"ckafka2","offset":508,"source":"/data/ryanyyang/hcmq/beats/filebeat-5.6.2-linux-x86_64/testlog","type":"log"}
{"@timestamp":"2017-09-29T10:01:33.937Z","beat":{"hostname":"10.193.9.26","name":"10.193.9.26","version":"5.6.2"},"input_type":"log","message":"ckafka3","offset":516,"source":"/data/ryanyyang/hcmq/beats/filebeat-5.6.2-linux-x86_64/testlog","type":"log"}

SASL/PLAINTEXT Mode

If you need to configure SALS/PLAINTEXT, you need to set the username and password under Kafka configuration.

# The following information needs to be configured for SASL. If not required, they can be ignored.
  username: "instance-will#user"
  password: "password"

FAQ

In the Filebeat log (default path /var/log/filebeat/filebeat ), it is found that there are a large number of INFO logs, such as:

2019-03-20T08:55:02.198+0800    INFO    kafka/log.go:53 producer/broker/544 starting up
2019-03-20T08:55:02.198+0800    INFO    kafka/log.go:53 producer/broker/544 state change to [open] on wp-news-filebeat/4
2019-03-20T08:55:02.198+0800    INFO    kafka/log.go:53 producer/leader/wp-news-filebeat/4 selected broker 544
2019-03-20T08:55:02.198+0800    INFO    kafka/log.go:53 producer/broker/478 state change to [closing] because EOF
2019-03-20T08:55:02.199+0800    INFO    kafka/log.go:53 Closed connection to broker bitar1d12:9092
2019-03-20T08:55:02.199+0800    INFO    kafka/log.go:53 producer/leader/wp-news-filebeat/5 state change to [retrying-3]
2019-03-20T08:55:02.199+0800    INFO    kafka/log.go:53 producer/leader/wp-news-filebeat/4 state change to [flushing-3]
2019-03-20T08:55:02.199+0800    INFO    kafka/log.go:53 producer/leader/wp-news-filebeat/5 abandoning broker 478
2019-03-20T08:55:02.199+0800    INFO    kafka/log.go:53 producer/leader/wp-news-filebeat/2 state change to [retrying-2]
2019-03-20T08:55:02.199+0800    INFO    kafka/log.go:53 producer/leader/wp-news-filebeat/2 abandoning broker 541
2019-03-20T08:55:02.199+0800    INFO    kafka/log.go:53 producer/leader/wp-news-filebeat/3 state change to [retrying-2]
2019-03-20T08:55:02.199+0800    INFO    kafka/log.go:53 producer/broker/478 shut down

The emergence of a large number of INFO may be a problem with the Filebeat version, because the Elastic family of products are released frequently and there are many incompatibility between different major versions.
For example, the default version of 6.5.x Kafka is 1.0.0, while 5.6.x defaults to 0.8.2.0.

You need to check the version configuration in the configuration file:

output.kafka:
  version:0.10.2 // Configuration based on the open source version of different CKafka instances