加载中

过滤和聚合日志

筛选和聚合您的日志数据以查找特定信息、获取洞察并更高效地监控您的系统。您可以根据从日志数据中提取的结构化字段（如时间戳、日志级别和 IP 地址）进行筛选和聚合。

本指南向您展示如何：

  • 筛选日志：通过应用特定条件缩小您的日志数据范围。
  • 聚合日志：分析和总结数据以发现模式并获取洞察。
可用的 Agent 技能

有一个技能可用于帮助 AI 智能体处理此主题。

详细了解适用于 Elastic 的 Agent 技能

获取此技能
注意

对于 Observability serverless 项目，需要 Admin 角色或更高权限才能创建引入管道（ingest pipelines）并设置索引模板。要了解更多信息，请参考 分配用户角色和权限。

本页面上的示例使用以下引入管道和索引模板。在执行以下步骤创建数据流之前，需要先设置管道和模板。在 Developer Tools 中设置它们，您可以通过在 全局搜索字段 中搜索 Developer Tools 来找到它。

如果您尚未通过引入管道和索引模板来解析日志数据并提取结构化字段，请从 解析和组织日志 文档开始。

使用以下命令设置引入管道

				PUT _ingest/pipeline/logs-example-default
					{
  "description": "Extracts the timestamp log level and host ip",
  "processors": [
    {
      "dissect": {
        "field": "message",
        "pattern": "%{@timestamp} %{log.level} %{host.ip} %{message}"
      }
    }
  ]
}

使用以下命令设置索引模板

				PUT _index_template/logs-example-default-template
					{
  "index_patterns": [ "logs-example-*" ],
  "data_stream": { },
  "priority": 500,
  "template": {
    "settings": {
      "index.default_pipeline":"logs-example-default"
    }
  },
  "composed_of": [
    "logs-mappings",
    "logs-settings",
    "logs@custom",
    "ecs@dynamic_templates"
  ],
  "ignore_missing_component_templates": ["logs@custom"]
}

使用您提取的字段筛选数据，以便专注于具有特定日志级别、时间戳范围或主机 IP 的日志数据。您可以通过以下几种方式筛选日志数据：

Discover 是一个根据数据视图和索引模式提供日志数据视图的工具。要打开 Discover，请在 全局搜索字段 中查找 Discover。

在 Discover 中，从 Data views 菜单中打开 logs-* 或 All logs 数据视图。在此处，您可以在搜索栏中使用 Kibana 查询语言 (KQL) 来缩小显示的日志数据范围。例如，您可能想要调查在特定时间范围内发生的事件。

向您的数据流中添加一些具有不同时间戳和日志级别的日志

  1. 在您的 Observability 项目中，转到 Developer Tools。
  2. 在 Console 选项卡中，运行以下命令
				POST logs-example-default/_bulk
					{ "create": {} }
{ "message": "2025-04-15T08:15:20.234Z WARN 192.168.1.101 Disk usage exceeds 90%." }
{ "create": {} }
{ "message": "2025-04-14T10:30:45.789Z ERROR 192.168.1.102 Critical system failure detected." }
{ "create": {} }
{ "message": "2025-04-10T14:20:45.789Z ERROR 192.168.1.105 Database connection lost." }
{ "create": {} }
{ "message": "2025-04-20T09:40:32.345Z INFO 192.168.1.106 User logout initiated." }

对于此示例，让我们查找在 4 月 14 日或 15 日发生的、日志级别为 WARN 或 ERROR 的日志。在 Discover 中：

  1. 确保在 Data views 菜单中选择了 All logs。

  2. 在搜索栏中添加以下 KQL 查询，以筛选日志级别为 WARN 或 ERROR 的日志

    log.level: ("ERROR" or "WARN")

  3. 将时间过滤器设置为一个绝对范围，起点为 Apr 14, 2025 @ 00:00:00.000，终点为 Apr 15, 2025 @ 23:59:59.999。

在 Documents 选项卡下，您将看到与您的查询匹配的已筛选日志数据。

logs kql filter

有关使用 Discover 的更多信息，请参考 Discover 文档。

Query DSL 是一种基于 JSON 的语言，用于发送请求并从索引和数据流中检索数据。您可以在 Developer Tools 中使用 Query DSL 筛选日志数据。

例如，您可能想要对在特定日期或特定时间发生的问题进行故障排除。为此，请使用带有 范围查询 的布尔查询来筛选特定的时间戳范围，并使用 词条查询 来筛选 WARN 和 ERROR 日志级别。

首先，在 Developer Tools 中，使用以下命令向您的数据流中添加一些具有不同时间戳和日志级别的日志

				POST logs-example-default/_bulk
					{ "create": {} }
{ "message": "2025-04-15T08:15:20.234Z WARN 192.168.1.101 Disk usage exceeds 90%." }
{ "create": {} }
{ "message": "2025-04-14T10:30:45.789Z ERROR 192.168.1.102 Critical system failure detected." }
{ "create": {} }
{ "message": "2025-04-10T14:20:45.789Z ERROR 192.168.1.105 Database connection lost." }
{ "create": {} }
{ "message": "2025-04-20T09:40:32.345Z INFO 192.168.1.106 User logout initiated." }

假设您想要调查在 4 月 14 日至 15 日之间发生的事件。以下布尔查询用于筛选在这些天内发生且日志级别为 ERROR 或 WARN 的日志。

				POST /logs-example-default/_search
					{
  "query": {
    "bool": {
      "filter": [
        {
          "range": {
            "@timestamp": {
              "gte": "2025-04-14T00:00:00",
              "lte": "2025-04-15T23:59:59"
            }
          }
        },
        {
          "terms": {
            "log.level": ["WARN", "ERROR"]
          }
        }
      ]
    }
  }
}

筛选结果应显示在该时间戳范围内发生的 WARN 和 ERROR 日志

{
  ...
  "hits": {
    ...
    "hits": [
      {
        "_index": ".ds-logs-example-default-2025.04.25-000001",
        "_id": "JkwPzooBTddK4OtTQToP",
        "_score": 0,
        "_source": {
          "message": "192.168.1.101 Disk usage exceeds 90%.",
          "log": {
            "level": "WARN"
          },
          "@timestamp": "2025-04-15T08:15:20.234Z"
        }
      },
      {
        "_index": ".ds-logs-example-default-2025.04.25-000001",
        "_id": "A5YSzooBMYFrNGNwH75O",
        "_score": 0,
        "_source": {
          "message": "192.168.1.102 Critical system failure detected.",
          "log": {
            "level": "ERROR"
          },
          "@timestamp": "2025-04-14T10:30:45.789Z"
        }
      }
    ]
  }
}

使用聚合来分析和总结您的日志数据，以发现模式并获取洞察。分桶聚合 (Bucket aggregations) 将日志数据组织成有意义的组，从而更容易识别日志中的模式、趋势和异常。

例如，您可能希望通过分析每个日志级别的日志数量来了解错误分布情况。

首先，在 Developer Tools 中，使用以下命令向您的数据流中添加一些具有不同日志级别的日志

				POST logs-example-default/_bulk
					{ "create": {} }
{ "message": "2025-04-15T08:15:20.234Z WARN 192.168.1.101 Disk usage exceeds 90%." }
{ "create": {} }
{ "message": "2025-04-14T10:30:45.789Z ERROR 192.168.1.102 Critical system failure detected." }
{ "create": {} }
{ "message": "2025-04-15T12:45:55.123Z INFO 192.168.1.103 Application successfully started." }
{ "create": {} }
{ "message": "2025-04-14T15:20:10.789Z WARN 192.168.1.104 Network latency exceeding threshold." }
{ "create": {} }
{ "message": "2025-04-10T14:20:45.789Z ERROR 192.168.1.105 Database connection lost." }
{ "create": {} }
{ "message": "2025-04-20T09:40:32.345Z INFO 192.168.1.106 User logout initiated." }
{ "create": {} }
{ "message": "2025-04-21T15:20:55.678Z DEBUG 192.168.1.102 Database connection established." }

接下来，运行此命令以使用 log.level 字段聚合您的日志数据

				POST logs-example-default/_search?size=0&filter_path=aggregations
					{
"size": 0,
"aggs": {
    "log_level_distribution": {
      "terms": {
        "field": "log.level"
      }
    }
  }
}
  1. 带有聚合的搜索会同时返回查询结果和聚合结果，因此您将同时看到匹配该数据的日志和聚合。将 size 设置为 0 可将结果限制为仅显示聚合。

结果应显示每个日志级别的日志数量

{
  "aggregations": {
    "error_distribution": {
      "doc_count_error_upper_bound": 0,
      "sum_other_doc_count": 0,
      "buckets": [
        {
          "key": "ERROR",
          "doc_count": 2
        },
        {
          "key": "INFO",
          "doc_count": 2
        },
        {
          "key": "WARN",
          "doc_count": 2
        },
        {
          "key": "DEBUG",
          "doc_count": 1
        }
      ]
    }
  }
}

您还可以将聚合与查询组合使用。例如，您可能希望通过添加范围查询来限制先前聚合的范围

				GET /logs-example-default/_search
					{
  "size": 0,
  "query": {
    "range": {
      "@timestamp": {
        "gte": "2025-04-14T00:00:00",
        "lte": "2025-04-15T23:59:59"
      }
    }
  },
  "aggs": {
    "my-agg-name": {
      "terms": {
        "field": "log.level"
      }
    }
  }
}

结果应显示在您的时间戳范围内发生的日志的聚合

{
  ...
  "hits": {
    ...
    "hits": []
  },
  "aggregations": {
    "my-agg-name": {
      "doc_count_error_upper_bound": 0,
      "sum_other_doc_count": 0,
      "buckets": [
        {
          "key": "WARN",
          "doc_count": 2
        },
        {
          "key": "ERROR",
          "doc_count": 1
        },
        {
          "key": "INFO",
          "doc_count": 1
        }
      ]
    }
  }
}

有关聚合类型和可用聚合的更多信息，请参考 Aggregations 文档。

© . 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.