Filebeat无法向Logstash发送数据问题求助及报错排查

Let's break down your problems one by one and fix them step by step:

1. Why All Data Went to mylog-%{+YYYY.MM.dd} Initially? #

Your first pipeline's conditional logic was incorrect. In Logstash, the tags field is an array, not a single string. When you used if [tag] == "mylog", you were checking for a non-existent singular tag field, which broke the routing logic entirely.

The correct way to check for tags is using the in operator to verify if a tag exists in the array:

output {
  if "mylog" in [tags] {
    elasticsearch {
      hosts => [ "10.206.81.246:9200", "10.206.81.236:9200", "10.206.81.243:9200" ]
      user => "Test"
      password => "123456"
      index => "mylog-%{+YYYY.MM.dd}"
    }
  }
  if "syslog" in [tags] {
    elasticsearch {
      hosts => [ "10.206.81.246:9200", "10.206.81.236:9200", "10.206.81.243:9200" ]
      user => "Test"
      password => "123456"
      index => "syslog-%{+YYYY.MM.dd}"
    }
  }
}

Or use else if (note the space—Logstash doesn't recognize elseif as valid syntax):

output {
  if "mylog" in [tags] {
    elasticsearch { ... }
  } else if "syslog" in [tags] {
    elasticsearch { ... }
  }
}

2. What Do Those Filebeat Errors Mean? #

Let's decode the errors you saw:

• Failed to publish events caused by: EOF: This means the connection between Filebeat and Logstash was abruptly closed. Common triggers include Logstash restarting, reloading its config (which restarts input plugins), or a temporary network blip.
• client is not connected: Filebeat lost its active connection to Logstash and couldn't send events immediately.

3. Why Did Data Still Arrive in Kibana Despite the Errors? #

Filebeat has a built-in retry mechanism. When an initial send fails, it caches the unsent events and keeps trying to reconnect to Logstash until it succeeds. The errors you saw are logs of those failed initial attempts—but the retries worked, so your data eventually made it through.

4. What Caused the Connection Issues? #

Based on your workflow, the most likely culprit is:

• When you modified your Logstash pipeline config and reloaded the service (or Logstash auto-reloaded), the Beats input on port 5044 was restarted. This dropped the existing connection with Filebeat, triggering the errors.
• Another possibility: If you had a syntax error in your pipeline config when switching back, Logstash might have failed to reload properly, causing the input plugin to go offline temporarily. Always test configs before applying them with:

logstash -f /etc/logstash/conf.d/test.conf --config.test_and_exit

5. Recommended Fixes & Best Practices #

a. Avoid Duplicate Inputs #

You were using both Filebeat and Logstash's file input to read the same log files—this will lead to duplicate data. Stick to Filebeat for log collection (it's designed for this purpose, with better tailing and reliability), then route data in Logstash using Filebeat's source field (which contains the full path of the log file):

input {
  beats { port => "5044" }
}
filter {
  mutate {
    # Map log paths to clean tags
    add_tag => "%{if [source] == '/home/centos/logs/mylogs.log' { 'mylog' } else if [source] == '/home/centos/logs/syslog.log' { 'syslog' } }"
  }
}
output {
  # Use the cleaned tags for routing
  if "mylog" in [tags] {
    elasticsearch { ... }
  } else if "syslog" in [tags] {
    elasticsearch { ... }
  }
}

b. Tune Filebeat's Reconnection Settings #

Add these to your Filebeat config to make it more resilient to connection drops:

output.logstash:
  hosts: ["10.206.81.234:5044"]
  reconnect_interval: 5s  # Increase retry interval to avoid flooding Logstash
  max_retries: 30         # Allow more retries before discarding events (default is 3)
  bulk_max_size: 1024     # Adjust batch size based on your log volume

c. Check Logstash Logs #

Whenever you have connection issues, check Logstash's logs at /var/log/logstash/logstash-plain.log—it will tell you if there were config errors, input plugin failures, or other issues that caused the connection drop.

内容的提问来源于stack exchange，提问作者dorinand