« 解析和组织日志流式传输应用程序日志 »

› ›

过滤和聚合日志

过滤和聚合日志

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

本指南将向您展示如何

过滤日志 — 通过应用特定条件缩小日志数据的范围。
聚合日志 — 分析和汇总数据以查找模式并获取洞察。

开始之前

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

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

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 的日志数据。您可以通过不同的方式过滤日志数据

在日志资源管理器中过滤日志 – 使用日志资源管理器在 Kibana 中过滤和可视化日志数据。
使用查询 DSL 过滤日志 – 使用查询 DSL 从开发者工具中过滤日志数据。

在日志资源管理器中过滤日志

日志资源管理器是一个 Kibana 工具，它根据集成和数据流自动提供日志数据的视图。要打开 日志资源管理器，请在全局搜索字段中找到 日志资源管理器。

在日志资源管理器中，您可以使用搜索栏中的 Kibana 查询语言 (KQL) 缩小日志资源管理器中显示的日志数据的范围。例如，您可能希望查看在特定时间范围内发生的事件。

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

要打开 控制台，请在全局搜索字段中找到 Dev Tools。
在 控制台 选项卡中，运行以下命令

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 的日志。从日志资源管理器

在搜索栏中添加以下 KQL 查询以过滤日志级别为 WARN 或 ERROR 的日志
```
log.level: ("ERROR" or "WARN")
```
单击当前时间范围，选择绝对，并将 开始日期 设置为 2023 年 9 月 14 日 00:00:00.000。
单击当前时间范围的结束，选择绝对，并将 结束日期 设置为 2023 年 9 月 15 日 23:59:59.999。

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

Filter data by log level using KQL

有关使用日志资源管理器的更多信息，请参阅发现文档。

使用查询 DSL 过滤日志

查询 DSL 是一种基于 JSON 的语言，用于向索引和数据流发送请求并检索数据。您可以从 开发者工具 中使用查询 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
        }
      ]
    }
  }
}

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

« 解析和组织日志流式传输应用程序日志 »