死信队列 (DLQ)

默认情况下，当 Logstash 遇到由于数据包含映射错误或其他问题而无法处理的事件时，Logstash 管道会挂起或丢弃不成功的事件。为了防止在这种情况下丢失数据，您可以 配置 Logstash 将不成功的事件写入死信队列，而不是丢弃它们。

写入死信队列的每个事件都包括原始事件、描述事件无法处理的原因的元数据、有关写入事件的插件的信息以及事件进入死信队列的时间戳。

要处理死信队列中的事件，请创建一个 Logstash 管道配置，该配置使用 dead_letter_queue 输入插件 从队列中读取。有关更多信息，请参阅 处理死信队列中的事件。

HTTP 请求失败。 如果 HTTP 请求失败（因为 Elasticsearch 无法访问或因为它返回了 HTTP 错误代码），Elasticsearch 输出会无限期地重试整个请求。在这些情况下，死信队列没有机会进行拦截。

HTTP 请求成功。 Elasticsearch 批量 API 可以使用同一个请求执行多个操作。如果批量 API 请求成功，它会返回 200 OK，即使批处理中的一些文档 失败。在这种情况下，请求的 errors 标志将为 true。

响应正文可以包含元数据，指示批量请求中的一个或多个特定操作无法执行，以及每个条目的 HTTP 风格状态代码，以指示操作无法执行的原因。如果配置了 DLQ，则会将各个索引失败路由到那里。

即使您定期处理事件，事件也会保留在死信队列中。死信队列需要 手动干预 才能清除它。

默认情况下，死信队列处于禁用状态。要启用死信队列，请在 logstash.yml 设置文件 中设置 dead_letter_queue_enable 选项

dead_letter_queue.enable: true

死信队列作为文件存储在 Logstash 实例的本地目录中。默认情况下，死信队列文件存储在 path.data/dead_letter_queue 中。每个管道都有一个单独的队列。例如，main 管道的死信队列默认存储在 LOGSTASH_HOME/data/dead_letter_queue/main 中。队列文件按顺序编号：1.log、2.log，依此类推。

您可以在 logstash.yml 文件中设置 path.dead_letter_queue 以指定文件的不同路径

path.dead_letter_queue: "path/to/data/dead_letter_queue"

死信队列条目被写入临时文件，然后重命名为死信队列段文件，然后该文件可用于摄取。当此临时文件被视为已满或自上次将符合条件的死信队列事件写入临时文件以来经过一段时间后，就会发生重命名。

可以使用 dead_letter_queue.flush_interval 设置来设置这段时间。此设置以毫秒为单位，默认为 5000 毫秒。此处的值较低意味着，如果很少写入死信队列，则可能会写入更多、更小的队列文件，而值较大则会在“写入”到死信队列的项目与可供读取的项目之间引入更多延迟由 dead_letter_queue 输入。

Note that this value cannot be set to lower than 1000ms.

dead_letter_queue.flush_interval: 5000

dead_letter_queue.retain.age: 2d

input {
  dead_letter_queue {
  	path => "/path/to/data/dead_letter_queue"
  	pipeline_id => "main"
    clean_consumed => true
  }
}

当您准备好处理死信队列中的事件时，您可以创建一个管道，该管道使用 dead_letter_queue 输入插件 从死信队列中读取。您使用的管道配置当然取决于您需要执行的操作。例如，如果死信队列包含 Elasticsearch 中映射错误导致的事件，您可以创建一个管道来读取“死”事件，删除导致映射问题的字段，并将清理后的事件重新索引到 Elasticsearch 中。

以下示例显示了一个简单的管道，该管道从死信队列中读取事件并将事件（包括元数据）写入标准输出

input {
  dead_letter_queue {
    path => "/path/to/data/dead_letter_queue" 
    commit_offsets => true 
    pipeline_id => "main" 
  }
}

output {
  stdout {
    codec => rubydebug { metadata => true }
  }
}

有关另一个示例，请参阅示例：处理具有映射错误的数据。

当管道处理完死信队列中的所有事件后，它将继续运行并处理流入队列的新事件。 这意味着您无需停止生产系统即可处理死信队列中的事件。

从死信队列读取数据时，您可能不想处理队列中的所有事件，尤其是在队列中有很多旧事件的情况下。 您可以使用 start_timestamp 选项从队列中的特定点开始处理事件。 此选项将管道配置为根据事件进入队列的时间戳开始处理事件

input {
  dead_letter_queue {
    path => "/path/to/data/dead_letter_queue"
    start_timestamp => "2017-06-06T23:40:37"
    pipeline_id => "main"
  }
}

对于此示例，管道开始读取在 2017 年 6 月 6 日 23:40:37 或之后传递到死信队列的所有事件。

在此示例中，用户尝试索引包含 geo_ip 数据的文档，但由于数据包含映射错误，因此无法处理该数据

{"geoip":{"location":"home"}}

索引失败，因为 Logstash 输出插件需要 location 字段中的 geo_point 对象，但该值是一个字符串。 失败的事件将写入死信队列，以及有关导致失败的错误的元数据

{
   "@metadata" => {
    "dead_letter_queue" => {
       "entry_time" => #<Java::OrgLogstash::Timestamp:0x5b5dacd5>,
        "plugin_id" => "fb80f1925088497215b8d037e622dec5819b503e-4",
      "plugin_type" => "elasticsearch",
           "reason" => "Could not index event to Elasticsearch. status: 400, action: [\"index\", {:_id=>nil, :_index=>\"logstash-2017.06.22\", :_type=>\"doc\", :_routing=>nil}, 2017-06-22T01:29:29.804Z My-MacBook-Pro-2.local {\"geoip\":{\"location\":\"home\"}}], response: {\"index\"=>{\"_index\"=>\"logstash-2017.06.22\", \"_type\"=>\"doc\", \"_id\"=>\"AVzNayPze1iR9yDdI2MD\", \"status\"=>400, \"error\"=>{\"type\"=>\"mapper_parsing_exception\", \"reason\"=>\"failed to parse\", \"caused_by\"=>{\"type\"=>\"illegal_argument_exception\", \"reason\"=>\"illegal latitude value [266.30859375] for geoip.location\"}}}}"
    }
  },
  "@timestamp" => 2017-06-22T01:29:29.804Z,
    "@version" => "1",
       "geoip" => {
    "location" => "home"
  },
        "host" => "My-MacBook-Pro-2.local",
     "message" => "{\"geoip\":{\"location\":\"home\"}}"
}

要处理失败的事件，请创建以下管道，该管道从死信队列读取数据并删除映射问题

input {
  dead_letter_queue {
    path => "/path/to/data/dead_letter_queue/" 
  }
}
filter {
  mutate {
    remove_field => "[geoip][location]" 
  }
}
output {
  elasticsearch{
    hosts => [ "localhost:9200" ] 
  }
}

在死信队列成为问题之前监控其大小。 通过定期检查，您可以确定每个管道有意义的最大队列大小。

每个管道的 DLQ 大小在节点统计信息 API 中可用。

pipelines.${pipeline_id}.dead_letter_queue.queue_size_in_bytes.

其中 {pipeline_id} 是启用了 DLQ 的管道的名称。

在上游管道运行时无法清除死信队列。

死信队列是页面的目录。 要清除它，请停止管道并删除 location/<文件名>。

${path.data}/dead_letter_queue/${pipeline_id}

其中 {pipeline_id} 是启用了 DLQ 的管道的名称。

管道在再次启动时会创建一个新的死信队列。