过滤和聚合日志

Elastic Stack Serverless

过滤和聚合您的日志数据，以查找特定信息、获得洞察力并更有效地监控您的系统。您可以基于结构化字段（例如时间戳、日志级别和从日志数据中提取的 IP 地址）进行过滤和聚合。

本指南向您展示如何：

过滤日志：通过应用特定条件来缩小您的日志数据范围。
聚合日志：分析和总结数据以查找模式并获得洞察力。

开始之前

注意

对于可观测性无服务器项目，需要 Admin 角色或更高的角色才能创建摄取管道并设置索引模板。要了解更多信息，请参阅分配用户角色和权限。

此页面上的示例使用以下摄取管道和索引模板，您可以在 开发者工具 中设置它们。如果您尚未使用摄取管道和索引模板来解析您的日志数据并提取结构化字段，请从解析和组织日志文档开始。

使用以下命令设置摄取管道

		 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 中过滤和可视化日志数据。
使用 Query DSL 过滤日志：使用 Query DSL 从开发者工具过滤日志数据。

在 Discover 中过滤日志

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

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

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

在您的可观测性项目中，转到 开发者工具。
在 控制台 选项卡中，运行以下命令

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

	

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

确保在 数据视图 菜单中选择了 所有日志。
在搜索栏中添加以下 KQL 查询，以过滤日志级别为 WARN 或 ERROR 的日志：
```
log.level: ("ERROR" or "WARN")
```
单击当前时间范围，选择绝对，并将 开始日期 设置为 Sep 14, 2023 @ 00:00:00.000。
单击当前时间范围的末尾，选择绝对，并将 结束日期 设置为 Sep 15, 2023 @ 23:59:59.999。

在文档选项卡下，您将看到与您的查询匹配的已过滤日志数据。

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

使用 Query DSL 过滤日志

Query DSL 是一种基于 JSON 的语言，用于发送请求并从索引和数据流中检索数据。您可以使用 开发者工具 中的 Query DSL 过滤您的日志数据。

例如，您可能想要排查在特定日期或特定时间发生的事件。为此，请使用带有范围查询的布尔查询来过滤特定时间戳范围，并使用词项查询来过滤 WARN 和 ERROR 日志级别。

首先，从 开发者工具 中，使用以下命令将一些具有不同时间戳和日志级别的日志添加到您的数据流中：

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

	

假设您想要调查 9 月 14 日和 15 日之间发生的事件。以下布尔查询会过滤在这些日期内具有时间戳的日志，并且这些日志的日志级别也为 ERROR 或 WARN。

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

	

过滤后的结果应显示在时间戳范围内发生的 WARN 和 ERROR 日志。

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

	

聚合日志

使用聚合来分析和总结您的日志数据，以查找模式并获得洞察力。桶聚合将日志数据组织成有意义的组，从而更容易识别日志中的模式、趋势和异常情况。

例如，您可能想要通过分析每个日志级别的日志计数来了解错误分布。

首先，从 开发者工具 中，使用以下命令将一些具有不同日志级别的日志添加到您的数据流中：

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

	

具有聚合的搜索会同时返回查询结果和聚合，因此您将看到与数据匹配的日志和聚合。将 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": "2023-09-14T00:00:00",
        "lte": "2023-09-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
        }
      ]
    }
  }
}

	

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