错误处理
Elasticsearch 中的摄取管道（Ingest pipelines）是在索引数据之前转换和丰富数据的强大工具。然而，处理过程中可能会发生错误。本指南概述了有效处理此类错误的策略。
摄取管道在文档被 Elasticsearch 索引之前执行。您可以处理文档处理期间（即转换 json 对象时）发生的错误，但不能处理索引期间触发的错误，例如映射冲突。针对这种情况，可以使用 Elasticsearch 失败存储（Failure Store）。
摄取管道中的错误通常分为以下几类
- 解析错误：当处理器无法解析字段（例如日期或数字）时发生。
- 字段缺失：当文档中缺少必需字段时发生。
创建一个 error-handling-pipeline，将 event.kind 设置为 pipeline_error，并将错误消息以及来自失败处理器的标签存储在 error.message 字段中。在使用多个 grok、dissect 或 script 处理器时，包含标签特别有用，因为它有助于识别是哪一个处理器导致了失败。
on_failure 参数既可以针对单个处理器定义，也可以在管道级别定义，以捕获文档处理过程中可能发生的异常。ignore_failure 选项允许特定处理器静默跳过错误，而不会影响管道的其余部分。
以下示例演示了如何在管道级别而不是在单个处理器内部使用 on_failure 处理程序。虽然这种方法确保了管道在失败时能优雅地退出，但也意味着处理会在出错点停止。
在此示例中，配置 dissect 处理器时出现了一个拼写错误，该处理器旨在从消息中提取 user.name。使用了逗号 (,) 而不是正确的冒号 (:)。
POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"@timestamp": "2025-04-03T10:00:00.000Z",
"message": "user: philipp has logged in"
}
}
],
"pipeline": {
"processors": [
{
"dissect": {
"field": "message",
"pattern": "%{}, %{user.name} %{}",
"tag": "dissect for user.name"
}
},
{
"append": {
"field": "event.category",
"value": "authentication"
}
}
],
"on_failure": [
{
"set": {
"field": "event.kind",
"value": "pipeline_error"
}
},
{
"append": {
"field": "error.message",
"value": "Processor {{ _ingest.on_failure_processor_type }} with tag {{ _ingest.on_failure_processor_tag }} in pipeline {{ _ingest.on_failure_pipeline }} failed with message: {{ _ingest.on_failure_message }}"
}
}
]
}
}
第二个将 event.category 设置为 authentication 的处理器不再执行，因为第一个 dissect 处理器失败并触发了全局 on_failure 处理程序。生成的文档显示了是哪个处理器导致了错误、它尝试应用的模式以及它接收到的输入。
"@timestamp": "2025-04-03T10:00:00.000Z",
"message": "user: philipp has logged in",
"event": {
"kind": "pipeline_error"
},
"error": {
"message": "Processor dissect with tag dissect for user.name in pipeline _simulate_pipeline failed with message: Unable to find match for dissect pattern: %{}, %{user.name} %{} against source: user: philipp has logged in"
}
我们可以通过将 on_failure 处理直接移入处理器本身来重构管道。这允许管道继续执行。在这种情况下，event.category 处理器仍然会运行。您还可以保留全局 on_failure 来处理来自其他处理器的错误，同时在需要的地方添加处理器特定的错误处理。
虽然在 dissect 错误处理程序中执行两个 set 处理器可能并不总是理想的，但它起到了演示作用。
对于 dissect 处理器，请考虑设置一个临时字段，例如 _tmp.error: dissect_failure。然后，您可以在后续处理器中使用 if 条件，仅在解析失败时执行它们，从而实现更可控和灵活的错误处理。
POST _ingest/pipeline/_simulate
{
"docs": [
{
"_source": {
"@timestamp": "2025-04-03T10:00:00.000Z",
"message": "user: philipp has logged in"
}
}
],
"pipeline": {
"processors": [
{
"dissect": {
"field": "message",
"pattern": "%{}, %{user.name} %{}",
"on_failure": [
{
"set": {
"field": "event.kind",
"value": "pipeline_error"
}
},
{
"append": {
"field": "error.message",
"value": "Processor {{ _ingest.on_failure_processor_type }} with tag {{ _ingest.on_failure_processor_tag }} in pipeline {{ _ingest.on_failure_pipeline }} failed with message: {{ _ingest.on_failure_message }}"
}
}
],
"tag": "dissect for user.name"
}
},
{
"append": {
"field": "event.category",
"value": "authentication"
}
}
],
"on_failure": [
{
"set": {
"field": "event.kind",
"value": "pipeline_error"
}
},
{
"set": {
"field": "error.message",
"value": "Processor {{ _ingest.on_failure_processor_type }} with tag {{ _ingest.on_failure_processor_tag }} in pipeline {{ _ingest.on_failure_pipeline }} failed with message: {{ _ingest.on_failure_message }}"
}
}
]
}
}