Docker部署Logstash如何为http_poller输入添加标识定位报错来源

你之前使用的%X{}语法是Log4j2用于读取MDC（线程上下文容器）内属性的语法，Logstash事件本身的字段默认不会写入MDC，因此无法直接通过该语法读取，可通过以下两种方式实现需求：

• 方案1：开启Elasticsearch输出插件DEBUG级日志
  在你的自定义log4j2.properties文件中添加如下配置，即可让ES输出报错时打印完整的事件字段内容：

logger.elasticsearchoutput.name = logstash.outputs.elasticsearch
logger.elasticsearchoutput.level = debug

同时修改日志输出的pattern，确保完整输出异常信息：

appender.rolling.layout.pattern = [%d{ISO8601}][%-5p][%-25c]%notEmpty{[%X{pipeline.id}]}%notEmpty{[%X{plugin.id}]} %m%n%ex

• 方案2：通过死信队列捕获失败事件自定义输出日志
  1. 先修改logstash.yml开启死信队列功能：

dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 1024mb

2. 新增独立Pipeline消费死信队列，输出你需要的关键字段：

input {
  dead_letter_queue {
    path => "/usr/share/logstash/data/dead_letter_queue"
    pipeline_id => "main" # 替换为你业务Pipeline的实际ID
  }
}
output {
  stdout {
    codec => line {
      format => "ES写入失败 来源请求标识: %{request_source} 错误原因: %{[@metadata][dead_letter_queue][reason]} 事件内容: %{message}"
    }
  }
}

补充说明：你提供的报错是因为事件中不存在objectType字段，导致索引名模板prefix-%{objectType}没有被替换，直接将带占位符的字符串作为索引名写入ES，而ES的自动创建索引规则不允许创建该名称的索引。

有两种常用实现方案，你可以根据配置习惯选择：

• 方案1：拆分http_poller输入块，通过公共参数add_field添加标识
  每个URL单独配置一个http_poller输入块，分别添加自定义字段：

input {
    # 第一个URL的拉取配置
    http_poller {
        id => "medium-pull-url1"
        urls => {
            url1 => {               
                method => post
                url => "${ROOT}/endpoint1"
                body => '{"ids": [],  "limit": 10000,  "page": 1}'
            }
        }
        add_field => { "request_source" => "endpoint1" }
    }
    # 第二个URL的拉取配置
    http_poller {
        id => "medium-pull-url2"
        urls => {
            url2 => {               
                method => post
                url => "${ROOT}/endpoint2"
                body => '{"ids": [],  "limit": 10000,  "page": 1}'
            }
        }
        add_field => { "request_source" => "endpoint2" }
    }
}

• 方案2：单http_poller块+Filter阶段匹配添加标识
  http_poller默认会将请求的目标地址写入事件的url根字段，你可以在Filter阶段根据该字段匹配添加自定义标识，无需拆分输入配置：

filter {
    # 先默认填充完整请求URL作为标识
    mutate {
        add_field => { "request_source" => "%{url}" }
    }
    # 匹配URL替换为自定义别名
    if [url] == "${ROOT}/endpoint1" {
        mutate {
            update => { "request_source" => "业务接口1" }
        }
    }
    if [url] == "${ROOT}/endpoint2" {
        mutate {
            update => { "request_source" => "业务接口2" }
        }
    }
}

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