加载中

计算摄取延迟元数据

摄取延迟是指从读取文档到 Elasticsearch 接收到文档所花费的时间。将此值以分钟、秒或毫秒为单位存储，并使用它来创建可视化和警报。

基本计算公式为

event.ingested - @timestamp

您可以通过两种方式获取 event.ingested 时间戳

  • _ingest.timestamp
    在除 script 之外的所有处理器中，可通过 mustache 标记 {{_ingest.timestamp}} 使用。

  • metadata().now 仅在 script 处理器中可用。在编写脚本时，请使用此项代替 _ingest.timestamp。

注意

event.ingested 选项通常在 Fleet 最终管道中设置，它是摄取过程中的最后一步。对于大多数用例，以秒为单位计算延迟已足够。

以下脚本是该解决方案的核心。它创建了一个新字段 event.ingestion.latency，您可以使用它来监控管道中的摄取性能。

{
  "script": {
    "description": "Calculates entire ingestion flow latency",
    "if": "ctx['@timestamp'] != null",
    "source": """
      ZonedDateTime start = ZonedDateTime.parse(ctx['@timestamp']);
      ctx.putIfAbsent("event", [:]);
      ctx.event.putIfAbsent("ingestion", [:]);
      ctx.event.ingestion.latency= ChronoUnit.SECONDS.between(start, metadata().now);
    """
  }
}

@timestamp 的值可能因数据源而异。它可能代表 Elastic Agent 读取文档的时间，也可能是解析后从文档本身提取的实际时间戳。

这种区别会影响摄取延迟的计算方式。例如，当 Elastic Agent 读取 Windows 事件日志时，它会根据日志的原始时间戳设置 @timestamp。然而，并非所有来源都适用此行为，例如 syslog 消息或 Linux 日志文件，在这些情况下，@timestamp 通常在管道解析后设置。

如果不加以考虑，这种不一致可能导致延迟测量不准确。

POST _ingest/pipeline/_simulate
{
    "docs": [{
        "_source": {
            "@timestamp": "2025-04-03T10:00:00.000Z",
            "message": "2025-03-01T09:00:00.000Z user: philipp has logged in"
        }
    }],
    "pipeline": {
        "processors": [
          {"script": {
            "if": "ctx['@timestamp'] != null",
            "source": """
              ZonedDateTime start = ZonedDateTime.parse(ctx['@timestamp']);
              ctx.latency= ChronoUnit.SECONDS.between(start, metadata().now);
            """
          }}
        ]
    }
}

在前面的示例中，读取时间戳是 4 月 3 日 10:00，而存储上的实际日志消息来自 3 月 3 日。如果您在第一步（任何解析之前）计算差值，结果将是准确的。但是，如果您在管道的最后一步进行计算（通常使用 @custom 管道的 Elastic 集成就是这种情况），则 2025-03-01 的时间戳将被用作 @timestamp，从而导致延迟计算错误。

对于许多用例，仅使用 @timestamp 就足够了，因为我们期望 Elastic Agent 尽可能快地获取日志。在初始接入新数据源期间，由于摄取了历史数据或旧数据，可能会出现较高的延迟。

如上所述，@timestamp 被设置为收集的日志内的时间戳。当我们想要计算真实值时，结果可能会有偏差，因为 Elastic Agent 可能需要几秒、几分钟、几小时、几天甚至更长时间才能采集到数据。想象一下，如果您第一次接入一台旧服务器，您的延迟会瞬间飙升，因为您可能会采集到一年前的数据。event.created 不会自动添加到 Elastic Agent 中的任何日志中。为此，您需要在集成的高级设置中添加以下处理器。

- script:
    lang: javascript
    source: >
      function process(event) {
          event.Put("event.created", Date.now());
      }

无论选择哪种架构，都请在管道末尾添加一个 remove 处理器，以删除 _tmp 字段。不需要来自各个处理步骤的原始时间戳，因为以秒为单位的延迟已经足够。有关其他管道架构的信息，请参考 摄取架构 (Ingest architectures)。

使用 @timestamp 和 event.ingested 来计算差值。这将为您提供以下文档。event.ingestion.latency 的单位是秒。

{
  "event": {
    "ingestion": {
      "latency": 443394
    }
  }
}

此脚本包含摄取管道计算摄取延迟所需的处理器。

POST _ingest/pipeline/_simulate
{
    "docs": [{
        "_source": {
            "@timestamp": "2025-04-03T10:00:00.000Z",
            "message": "user: philipp has logged in",
            "_tmp": {
              "logstash": "2025-04-03T10:00:02.456Z"
            }

        }
    }],
    "pipeline": {
        "processors": [
          {
            "script": {
              "description": "Calculates entire ingestion flow latency",
              "if": "ctx['@timestamp'] != null",
              "source": """
                ZonedDateTime start = ZonedDateTime.parse(ctx['@timestamp']);
                ctx.putIfAbsent("event", [:]);
                ctx.event.putIfAbsent("ingestion", [:]);
                ctx.event.ingestion.latency= ChronoUnit.SECONDS.between(start, metadata().now);
              """
            }
          }
        ]
    }
}

当架构中添加了 Logstash 时，我们必须添加一个时间戳，这只能通过使用 Ruby 来完成，最简单的形式如下

ruby {
  code => "event.set('[_tmp][logstash_seen]', Time.now());"
}

Elastic Agent 会填充 @timestamp 字段，但 Logstash 默认不会添加任何时间戳。添加一个临时时间戳，例如通过设置 _tmp.logstash_seen。有了它，您就可以计算以下延迟值

  • 总延迟：(@timestamp - event.ingested)
  • Elastic Agent 到 Logstash：(@timestamp - _tmp.logstash_seen)
  • Logstash 到 Elasticsearch：(_tmp.logstash_seen - event.ingested)

这些值对于调试特别有帮助，因为它们允许您快速确定延迟是在哪里引入的，以及延迟是由从 Elastic Agent 到 Logstash 的传输引起的，还是由从 Logstash 到 Elasticsearch 的传输引起的。

此脚本计算这些差值，为上述提到的每个阶段提供延迟值。

{
  "event": {
    "ingestion": {
      "latency_logstash_to_elasticsearch": 443091,
      "latency": 443093,
      "latency_elastic_agent_to_logstash": 1
    }
  }
}

此脚本包含摄取管道计算摄取延迟所需的处理器。如果您想删除第一次计算，请确保对象 event.ingestion 可用。

POST _ingest/pipeline/_simulate
{
    "docs": [{
        "_source": {
            "@timestamp": "2025-04-03T10:00:00.000Z",
            "message": "user: philipp has logged in",
            "_tmp": {
              "logstash": "2025-04-03T10:00:02.456Z"
            }

        }
    }],
    "pipeline": {
        "processors": [
          {
            "script": {
              "description": "Calculates entire ingestion flow latency",
              "if": "ctx['@timestamp'] != null",
              "source": """
                ZonedDateTime start = ZonedDateTime.parse(ctx['@timestamp']);
                ctx.putIfAbsent("event", [:]);
                ctx.event.putIfAbsent("ingestion", [:]);
                ctx.event.ingestion.latency= ChronoUnit.SECONDS.between(start, metadata().now);
              """
            }
          },
          {
            "script": {
              "description": "Calculates logstash to Elasticsearch latency",
              "if": "ctx._tmp?.logstash_seen != null",
              "source": """
                ZonedDateTime start = ZonedDateTime.parse(ctx._tmp.logstash_seen);
                ctx.event.ingestion.latency_logstash_to_elasticsearch=ChronoUnit.SECONDS.between(start, metadata().now);
              """
            }
          },
          {
            "script": {
              "description": "Calculates Elastic Agent to Logstash latency",
              "if": "ctx._tmp?.logstash_seen != null",
              "source": """
                ZonedDateTime start = ZonedDateTime.parse(ctx['@timestamp']);
                ZonedDateTime end = ZonedDateTime.parse(ctx._tmp.logstash_seen);
                ctx.event.ingestion.latency_elastic_agent_to_logstash=ChronoUnit.SECONDS.between(start, end);
              """
            }
          }
        ]
    }
}

与前一种情况一样，增加一个额外的跳转会引入另一个可能发生延迟的点。建议添加另一个临时时间戳字段。有关更多详细信息，请参考上一节。

这是一个计算管道中每一步延迟的脚本。将生成以下值

{
  "event": {
    "ingestion": {
      "latency_logstash_to_elasticsearch": 443091,
      "latency_logstash_to_logstash": 1,
      "latency": 443093,
      "latency_elastic_agent_to_logstash": 1
    }
  }
}

此脚本包含摄取管道计算摄取延迟所需的处理器。要删除第一次计算，请确保对象 event.ingestion 可用。您也可以将所有步骤合并为一个更大的脚本。

POST _ingest/pipeline/_simulate
{
    "docs": [{
        "_source": {
            "@timestamp": "2025-04-03T10:00:00.000Z",
            "message": "user: philipp has logged in",
            "_tmp": {
              "logstash_pre_kafka": "2025-04-03T10:00:01.233Z",
              "logstash_post_kafka": "2025-04-03T10:00:02.456Z"
            }

        }
    }],
    "pipeline": {
        "processors": [
          {
            "script": {
              "description": "Calculates entire ingestion flow latency",
              "if": "ctx['@timestamp'] != null",
              "source": """
                ZonedDateTime start = ZonedDateTime.parse(ctx['@timestamp']);
                ctx.putIfAbsent("event", [:]);
                ctx.event.putIfAbsent("ingestion", [:]);
                ctx.event.ingestion.latency= ChronoUnit.SECONDS.between(start, metadata().now);
              """
            }
          },
          {
            "script": {
              "description": "Calculates logstash to logstash latency",
              "if": "ctx._tmp?.logstash_pre_kafka != null && ctx._tmp?.logstash_post_kafka != null",
              "source": """
                ZonedDateTime start = ZonedDateTime.parse(ctx._tmp.logstash_pre_kafka);
                ZonedDateTime end = ZonedDateTime.parse(ctx._tmp.logstash_post_kafka);
                ctx.event.ingestion.latency_logstash_to_logstash=ChronoUnit.SECONDS.between(start, end);
              """
            }
          },
          {
            "script": {
              "description": "Calculates logstash post Kafka to Elasticsearch latency",
              "if": "ctx._tmp?.logstash_post_kafka != null",
              "source": """
                ZonedDateTime start = ZonedDateTime.parse(ctx._tmp.logstash_post_kafka);
                ctx.event.ingestion.latency_logstash_to_elasticsearch=ChronoUnit.SECONDS.between(start, metadata().now);
              """
            }
          },
          {
            "script": {
              "description": "Calculates Elastic Agent to pre kafka Logstash latency",
              "if": "ctx._tmp?.logstash_pre_kafka != null",
              "source": """
                ZonedDateTime start = ZonedDateTime.parse(ctx['@timestamp']);
                ZonedDateTime end = ZonedDateTime.parse(ctx._tmp.logstash_pre_kafka);
                ctx.event.ingestion.latency_elastic_agent_to_logstash=ChronoUnit.SECONDS.between(start, end);
              """
            }
          }
        ]
    }
}
© . This website operates independently and is not affiliated with or endorsed by Elasticsearch B.V. All brand names, logos, and trademarks are the property of their respective owners.