能否基于Kafka消息内容动态切换Logstash的Elasticsearch输出地址？

Absolutely, this is totally achievable with Logstash's flexible filtering and output configuration capabilities. Here's how you can implement this dynamic routing:

1. Parse the Target ES URL from Kafka Messages #

First, you need to extract the specific target ES URL information from your Kafka messages and store it as a Logstash event field. For example, if your Kafka messages are JSON-formatted with a field like target_es_host:

filter {
  # Parse JSON-formatted Kafka messages
  json {
    source => "message"
  }

  # Store the target ES URL from the message into a dedicated field
  mutate {
    add_field => { "target_es_url" => "%{target_es_host}" }
    # Optional: Clean up accidental spaces in the URL
    gsub => [ "target_es_url", " ", "" ]
  }
}

2. Use the Dynamic Field in Elasticsearch Output #

Logstash's Elasticsearch output plugin supports using event fields to dynamically configure the hosts parameter. This lets you route each message directly to the ES server specified in the Kafka message:

output {
  # Only send to ES if the target URL is valid (prevents runtime errors)
  if [target_es_url] and [target_es_url] =~ /^http(s)?:\/\/.+/ {
    elasticsearch {
      hosts => "%{target_es_url}"
      index => "dynamic-routed-%{+YYYY.MM.dd}"
      # Add other required configs (authentication, timeout, etc.)
    }
  } else {
    # Fallback to your default ES server from environment variable
    elasticsearch {
      hosts => "${DEFAULT_ES_URL}"
      index => "fallback-%{+YYYY.MM.dd}"
    }
  }
}

3. Handle Multi-Cluster Routing (If Needed) #

If you need to route messages to pre-defined clusters based on a message field (instead of full URLs), use conditional logic in the output section:

output {
  if [es_cluster] == "analytics" {
    elasticsearch {
      hosts => "http://analytics-es:9200"
      index => "analytics-%{+YYYY.MM.dd}"
    }
  } elsif [es_cluster] == "operations" {
    elasticsearch {
      hosts => "http://ops-es:9200"
      index => "operations-%{+YYYY.MM.dd}"
    }
  } else {
    # Fallback to default ES server
    elasticsearch {
      hosts => "${DEFAULT_ES_URL}"
      index => "unclassified-%{+YYYY.MM.dd}"
    }
  }
}

Key Notes #

• Validate Fields: Always add checks for valid target URLs/cluster names to avoid Logstash runtime errors.
• Test First: Use logstash -f your-config.conf --config.test_and_exit to validate your config, then test with sample Kafka messages to ensure fields are parsed correctly.
• Authentication: If your ES clusters require auth, use conditional configs for auth parameters tied to each cluster (avoid hardcoding credentials in dynamic URLs).

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