› ›

解析和组织日志

编辑

解析和组织日志

编辑

如果您的日志数据是非结构化或半结构化的，您可以解析它并将其分解为有意义的字段。您可以使用这些字段来探索和分析您的数据。例如，您可以查找特定时间戳范围内的日志，或者按日志级别过滤日志以专注于潜在问题。

解析后，您可以使用结构化字段，通过配置重新路由处理器将特定日志发送到不同的目标数据流，从而进一步组织日志。

有关解析和组织日志数据的更多信息，请参阅以下章节

提取结构化字段：提取时间戳、日志级别或 IP 地址等结构化字段，以方便查询和筛选数据。
将日志数据重新路由到特定的数据流：将通用数据流中的数据路由到目标数据流，以更精细地控制数据保留、权限和处理。

提取结构化字段

编辑

通过从非结构化日志数据中提取结构化字段，使您的日志更有用。提取结构化字段可以更轻松地搜索、分析和筛选您的日志数据。

按照以下步骤查看默认情况下如何索引以下非结构化日志数据

2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%.

首先将文档存储在 logs-example-default 数据流中

要打开 控制台，请在全局搜索字段中找到 开发工具。

在 控制台 选项卡中，使用以下命令将示例日志添加到 Elasticsearch

POST logs-example-default/_doc
{
  "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%."
}

然后，您可以使用以下搜索来检索文档
```
GET /logs-example-default/_search
```

结果应如下所示

{
  ...
  "hits": {
    ...
    "hits": [
      {
        "_index": ".ds-logs-example-default-2023.08.09-000001",
        ...
        "_source": {
          "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%.",
          "@timestamp": "2023-08-09T17:19:27.73312243Z"
        }
      }
    ]
  }
}

Elasticsearch 默认索引 message 字段并添加 @timestamp 字段。由于没有设置时间戳，因此将其设置为 now。此时，您可以在 message 字段中搜索短语，例如 WARN 或 磁盘使用量超过。例如，使用以下命令在日志的 message 字段中搜索短语 WARN

GET logs-example-default/_search
{
  "query": {
    "match": {
      "message": {
        "query": "WARN"
      }
    }
  }
}

虽然您可以在 message 字段中搜索短语，但您不能使用此字段来过滤日志数据。但是，您的消息包含以下所有可以提取并用于过滤和聚合日志数据的潜在字段

@timestamp (2023-08-08T13:45:12.123Z)：提取此字段可让您按日期和时间对日志进行排序。当您想按发生顺序查看日志或确定问题发生时间时，这很有帮助。
log.level (WARN)：提取此字段可让您按严重性过滤日志。如果您想专注于高严重性 WARN 或 ERROR 级别的日志，并通过过滤掉低严重性 INFO 级别的日志来减少噪音，这将很有帮助。
host.ip (192.168.1.101)：提取此字段可让您按主机 IP 地址过滤日志。如果您想专注于您正在遇到的特定主机，或者如果您想查找主机之间的差异，这将很有帮助。
message (磁盘使用量超过 90%。)：您可以在消息字段中搜索短语或单词。

这些字段是Elastic Common Schema (ECS) 的一部分。ECS 定义了一组通用字段，您可以在存储数据（包括日志和指标数据）时在 Elasticsearch 中使用这些字段。

提取 `@timestamp` 字段

编辑

当您在上一节中将日志添加到 Elasticsearch 时，@timestamp 字段显示了添加日志的时间。显示日志实际发生时间的时间戳在非结构化的 message 字段中

        ...
        "_source": {
          "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%.",
          "@timestamp": "2023-08-09T17:19:27.73312243Z"
        }
        ...

	`message` 字段中的时间戳显示日志发生的时间。
	`@timestamp` 字段中的时间戳显示日志添加到 Elasticsearch 的时间。

在调查问题时，您希望按问题发生的时间而不是日志添加到项目的时间来筛选日志。为此，请完成以下步骤，将时间戳从非结构化的 message 字段提取到结构化的 @timestamp 字段

使用摄取管道提取 `@timestamp` 字段

编辑

摄取管道由一系列处理器组成，这些处理器在索引传入文档之前对其执行常见转换。要从示例日志中提取 @timestamp 字段，请使用具有 dissect 处理器的摄取管道。dissect 处理器基于您设置的模式从非结构化日志消息中提取结构化字段。

Elasticsearch 可以将 yyyy-MM-dd'T'HH:mm:ss.SSSZ 和 yyyy-MM-dd 格式的字符串时间戳解析为日期字段。由于日志示例的时间戳是这些格式之一，因此您不需要其他处理器。更复杂或非标准的时间戳需要日期处理器将时间戳解析为日期字段。

使用以下命令将时间戳从 message 字段提取到 @timestamp 字段

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

	管道的名称 `logs-example-default` 需要与您的数据流的名称匹配。您将在下一节中设置数据流。有关更多信息，请参阅数据流命名方案。
	您正在从中提取数据的字段，在本例中为 `message`。
	您的日志数据中元素的模式。`%{@timestamp} %{message}` 模式将时间戳 `2023-08-08T13:45:12.123Z` 提取到 `@timestamp` 字段，而消息的其余部分 `WARN 192.168.1.101 Disk usage exceeds 90%.` 保留在 `message` 字段中。dissect 处理器将查找空格作为模式定义的分隔符。

使用模拟管道 API 测试管道

编辑

模拟管道 API 运行摄取管道而不存储任何文档。这使您可以使用多个文档验证您的管道是否正常工作。运行以下命令以使用模拟管道 API 测试您的摄取管道。

POST _ingest/pipeline/logs-example-default/_simulate
{
  "docs": [
    {
      "_source": {
        "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%."
      }
    }
  ]
}

结果应显示从 message 字段中提取的 @timestamp 字段

{
  "docs": [
    {
      "doc": {
        "_index": "_index",
        "_id": "_id",
        "_version": "-3",
        "_source": {
          "message": "WARN 192.168.1.101 Disk usage exceeds 90%.",
          "@timestamp": "2023-08-08T13:45:12.123Z"
        },
        ...
      }
    }
  ]
}

在使用模拟管道 API 之前，请确保您已使用上一节中的 PUT 命令创建了摄取管道。

使用索引模板配置数据流

编辑

创建摄取管道后，运行以下命令以创建索引模板来配置数据流的支持索引

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@mappings"
  ],
  "ignore_missing_component_templates": ["logs@custom"]
}

	`index_pattern`：需要与您的日志数据流匹配。数据流的命名约定为 `<type>-<dataset>-<namespace>`。在此示例中，您的日志数据流命名为 `logs-example-*`。与此模式匹配的数据将通过您的管道。
	`data_stream`：启用数据流。
	`priority`：设置您的索引模板的优先级。优先级较高的索引模板优先于优先级较低的索引模板。如果数据流与多个索引模板匹配，则 Elasticsearch 将使用优先级较高的模板。内置模板的优先级为 `200`，因此自定义模板的优先级应高于 `200`。
	`index.default_pipeline`：您的摄取管道的名称。在本例中为 `logs-example-default`。
	`composed_of`：您可以在此处设置组件模板。组件模板是用于构建索引模板的构建块，这些模板指定索引映射、设置和别名。Elastic 有几个内置模板可帮助您摄取日志数据。

上面的示例索引模板设置以下组件模板

logs@mappings：日志数据流的通用映射，包括禁用 string 字段的自动日期检测，以及指定 data_stream ECS 字段的映射。
logs@settings：日志数据流的常规设置，包括以下内容
- 当主分片达到 50 GB 或 30 天后滚动更新的默认生命周期策略。
- 如果没有指定 @timestamp，则默认管道使用摄取时间戳，并为 logs@custom 管道放置一个钩子。如果安装了 logs@custom 管道，则该管道将应用于摄取到此数据流的日志。
- 将 ignore_malformed 标志设置为 true。当摄取大量日志数据时，单个格式错误的字段（如 IP 地址）可能会导致整个批次失败。当设置为 true 时，仍然会处理具有支持此标志的映射类型的格式错误字段。
logs@custom：一个预定义的组件模板，默认情况下未安装。使用此名称安装自定义组件模板，以覆盖或扩展任何默认映射或设置。
ecs@mappings：动态模板，自动确保您的数据流映射符合Elastic Common Schema (ECS)。

创建数据流

编辑

使用数据流命名方案创建数据流。将您的数据流命名为与您的摄取管道的名称匹配，在本例中为 logs-example-default。使用以下命令将示例日志发布到您的数据流

POST logs-example-default/_doc
{
  "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%."
}

使用以下命令查看您的文档

GET /logs-example-default/_search

您应该看到管道已提取 @timestamp 字段

{
...
{
  ...
  "hits": {
    ...
    "hits": [
      {
        "_index": ".ds-logs-example-default-2023.08.09-000001",
        "_id": "RsWy3IkB8yCtA5VGOKLf",
        "_score": 1,
        "_source": {
          "message": "WARN 192.168.1.101 Disk usage exceeds 90%.",
          "@timestamp": "2023-08-08T13:45:12.123Z"
        }
      }
    ]
  }
}

提取的 @timestamp 字段。

您现在可以使用 @timestamp 字段按日志发生的日期和时间对其进行排序。

问题排查 `@timestamp` 字段

编辑

检查以下关于时间戳的常见问题和解决方案

时间戳失败：如果您的数据具有不一致的日期格式，请将日期处理器的 ignore_failure 设置为 true。这将处理格式正确的日期日志，并忽略那些存在问题的日志。
时区不正确：使用日期处理器上的 timezone 选项设置您的时区。
时间戳格式不正确：您的时间戳可以是 Java 时间模式，也可以是以下格式之一：ISO8601、UNIX、UNIX_MS 或 TAI64N。有关时间戳格式的更多信息，请参阅映射日期格式。

提取 `log.level` 字段

编辑

提取 log.level 字段可让您按严重程度进行过滤，并专注于关键问题。本节将向您展示如何从以下示例日志中提取 log.level 字段

2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%.

要提取和使用 log.level 字段

将 `log.level` 添加到您的摄取管道

编辑

使用以下命令，将 %{log.level} 选项添加到您在提取 @timestamp 字段部分中创建的摄取管道中的 dissect 处理器模式

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

现在，您的管道将提取以下字段

@timestamp 字段：2023-08-08T13:45:12.123Z
log.level 字段：WARN
message 字段：192.168.1.101 Disk usage exceeds 90%.

除了设置摄取管道之外，您还需要设置索引模板。您可以使用在提取 @timestamp 字段部分中创建的索引模板。

使用模拟 API 测试管道

编辑

使用模拟管道 API测试您的摄取管道是否按预期工作

POST _ingest/pipeline/logs-example-default/_simulate
{
  "docs": [
    {
      "_source": {
        "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%."
      }
    }
  ]
}

结果应显示从 message 字段中提取的 @timestamp 和 log.level 字段

{
  "docs": [
    {
      "doc": {
        "_index": "_index",
        "_id": "_id",
        "_version": "-3",
        "_source": {
          "message": "192.168.1.101 Disk usage exceeds 90%.",
          "log": {
            "level": "WARN"
          },
          "@timestamp": "2023-8-08T13:45:12.123Z",
        },
        ...
      }
    }
  ]
}

根据 `log.level` 查询日志

编辑

提取 log.level 字段后，您可以查询高严重性日志，例如 WARN 和 ERROR，这些日志可能需要立即关注，并过滤掉不太重要的 INFO 和 DEBUG 日志。

假设您有以下具有不同严重性的日志

2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%.
2023-08-08T13:45:14.003Z ERROR 192.168.1.103 Database connection failed.
2023-08-08T13:45:15.004Z DEBUG 192.168.1.104 Debugging connection issue.
2023-08-08T13:45:16.005Z INFO 192.168.1.102 User changed profile picture.

使用以下命令将它们添加到您的数据流

POST logs-example-default/_bulk
{ "create": {} }
{ "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%." }
{ "create": {} }
{ "message": "2023-08-08T13:45:14.003Z ERROR 192.168.1.103 Database connection failed." }
{ "create": {} }
{ "message": "2023-08-08T13:45:15.004Z DEBUG 192.168.1.104 Debugging connection issue." }
{ "create": {} }
{ "message": "2023-08-08T13:45:16.005Z INFO 192.168.1.102 User changed profile picture." }

然后，使用以下命令查询日志级别为 WARN 或 ERROR 的文档

GET logs-example-default/_search
{
  "query": {
    "terms": {
      "log.level": ["WARN", "ERROR"]
    }
  }
}

结果应仅显示高严重性日志

{
...
  },
  "hits": {
  ...
    "hits": [
      {
        "_index": ".ds-logs-example-default-2023.08.14-000001",
        "_id": "3TcZ-4kB3FafvEVY4yKx",
        "_score": 1,
        "_source": {
          "message": "192.168.1.101 Disk usage exceeds 90%.",
          "log": {
            "level": "WARN"
          },
          "@timestamp": "2023-08-08T13:45:12.123Z"
        }
      },
      {
        "_index": ".ds-logs-example-default-2023.08.14-000001",
        "_id": "3jcZ-4kB3FafvEVY4yKx",
        "_score": 1,
        "_source": {
          "message": "192.168.1.103 Database connection failed.",
          "log": {
            "level": "ERROR"
          },
          "@timestamp": "2023-08-08T13:45:14.003Z"
        }
      }
    ]
  }
}

提取 `host.ip` 字段

编辑

提取 host.ip 字段可让您按主机 IP 地址过滤日志，从而使您可以专注于您遇到问题的特定主机或查找主机之间的差异。

host.ip 字段是Elastic Common Schema (ECS)的一部分。通过 ECS，host.ip 字段被映射为ip 字段类型。ip 字段类型允许范围查询，因此您可以查找具有特定范围内的 IP 地址的日志。您还可以使用无类别域间路由 (CIDR) 表示法查询 ip 字段类型，以查找来自特定网络或子网的日志。

本节将向您展示如何从以下示例日志中提取 host.ip 字段，并基于提取的字段进行查询

2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%.
2023-08-08T13:45:14.003Z ERROR 192.168.1.103 Database connection failed.
2023-08-08T13:45:15.004Z DEBUG 192.168.1.104 Debugging connection issue.
2023-08-08T13:45:16.005Z INFO 192.168.1.102 User changed profile picture.

要提取和使用 host.ip 字段

将 `host.ip` 添加到您的摄取管道

编辑

将 %{host.ip} 选项添加到您在提取 @timestamp 字段部分中创建的摄取管道中的 dissect 处理器模式

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}"
      }
    }
  ]
}

您的管道将提取以下字段

@timestamp 字段：2023-08-08T13:45:12.123Z
log.level 字段：WARN
host.ip 字段：192.168.1.101
message 字段：Disk usage exceeds 90%.

除了设置摄取管道之外，您还需要设置索引模板。您可以使用在提取 @timestamp 字段部分中创建的索引模板。

使用模拟 API 测试管道

编辑

使用模拟管道 API测试您的摄取管道是否按预期工作

POST _ingest/pipeline/logs-example-default/_simulate
{
  "docs": [
    {
      "_source": {
        "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%."
      }
    }
  ]
}

结果应显示从 message 字段中提取的 host.ip、@timestamp 和 log.level 字段

{
  "docs": [
    {
      "doc": {
        ...
        "_source": {
          "host": {
            "ip": "192.168.1.101"
          },
          "@timestamp": "2023-08-08T13:45:12.123Z",
          "message": "Disk usage exceeds 90%.",
          "log": {
            "level": "WARN"
          }
        },
        ...
      }
    }
  ]
}

根据 `host.ip` 查询日志

编辑

您可以使用不同的方式（包括使用 CIDR 表示法和范围查询）根据 host.ip 字段查询日志。

在查询日志之前，请使用以下命令将它们添加到您的数据流

POST logs-example-default/_bulk
{ "create": {} }
{ "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%." }
{ "create": {} }
{ "message": "2023-08-08T13:45:14.003Z ERROR 192.168.1.103 Database connection failed." }
{ "create": {} }
{ "message": "2023-08-08T13:45:15.004Z DEBUG 192.168.1.104 Debugging connection issue." }
{ "create": {} }
{ "message": "2023-08-08T13:45:16.005Z INFO 192.168.1.102 User changed profile picture." }

CIDR 表示法

编辑

您可以使用CIDR 表示法来查询您的日志数据，该数据使用位于特定网段内的 IP 地址块。CIDR 表示法使用 [IP 地址]/[前缀长度] 的格式。以下命令查询 192.168.1.0/24 子网中的 IP 地址，这意味着 IP 地址从 192.168.1.0 到 192.168.1.255。

GET logs-example-default/_search
{
  "query": {
    "term": {
      "host.ip": "192.168.1.0/24"
    }
  }
}

由于所有示例日志都在此范围内，您将获得以下结果

{
  ...
  },
  "hits": {
    ...
      {
        "_index": ".ds-logs-example-default-2023.08.16-000001",
        "_id": "ak4oAIoBl7fe5ItIixuB",
        "_score": 1,
        "_source": {
          "host": {
            "ip": "192.168.1.101"
          },
          "@timestamp": "2023-08-08T13:45:12.123Z",
          "message": "Disk usage exceeds 90%.",
          "log": {
            "level": "WARN"
          }
        }
      },
      {
        "_index": ".ds-logs-example-default-2023.08.16-000001",
        "_id": "a04oAIoBl7fe5ItIixuC",
        "_score": 1,
        "_source": {
          "host": {
            "ip": "192.168.1.103"
          },
          "@timestamp": "2023-08-08T13:45:14.003Z",
          "message": "Database connection failed.",
          "log": {
            "level": "ERROR"
          }
        }
      },
      {
        "_index": ".ds-logs-example-default-2023.08.16-000001",
        "_id": "bE4oAIoBl7fe5ItIixuC",
        "_score": 1,
        "_source": {
          "host": {
            "ip": "192.168.1.104"
          },
          "@timestamp": "2023-08-08T13:45:15.004Z",
          "message": "Debugging connection issue.",
          "log": {
            "level": "DEBUG"
          }
        }
      },
      {
        "_index": ".ds-logs-example-default-2023.08.16-000001",
        "_id": "bU4oAIoBl7fe5ItIixuC",
        "_score": 1,
        "_source": {
          "host": {
            "ip": "192.168.1.102"
          },
          "@timestamp": "2023-08-08T13:45:16.005Z",
          "message": "User changed profile picture.",
          "log": {
            "level": "INFO"
          }
        }
      }
    ]
  }
}

范围查询

编辑

使用范围查询来查询特定范围内的日志。

以下命令搜索大于或等于 192.168.1.100 且小于或等于 192.168.1.102 的 IP 地址。

GET logs-example-default/_search
{
  "query": {
    "range": {
      "host.ip": {
        "gte": "192.168.1.100",
        "lte": "192.168.1.102"
      }
    }
  }
}

	大于或等于 `192.168.1.100`。
	小于或等于 `192.168.1.102`。

您将获得以下结果，仅显示您设置的范围内的日志

{
  ...
  },
  "hits": {
    ...
      {
        "_index": ".ds-logs-example-default-2023.08.16-000001",
        "_id": "ak4oAIoBl7fe5ItIixuB",
        "_score": 1,
        "_source": {
          "host": {
            "ip": "192.168.1.101"
          },
          "@timestamp": "2023-08-08T13:45:12.123Z",
          "message": "Disk usage exceeds 90%.",
          "log": {
            "level": "WARN"
          }
        }
      },
      {
        "_index": ".ds-logs-example-default-2023.08.16-000001",
        "_id": "bU4oAIoBl7fe5ItIixuC",
        "_score": 1,
        "_source": {
          "host": {
            "ip": "192.168.1.102"
          },
          "@timestamp": "2023-08-08T13:45:16.005Z",
          "message": "User changed profile picture.",
          "log": {
            "level": "INFO"
          }
        }
      }
    ]
  }
}

将日志数据重新路由到特定的数据流

编辑

默认情况下，摄取管道会将您的日志数据发送到单个数据流。为了简化日志数据管理，请使用重新路由处理器将数据从通用数据流路由到目标数据流。例如，您可能希望将高严重性日志发送到特定的数据流，以帮助进行分类。

本节向您展示如何使用重新路由处理器将以下示例日志中的高严重性日志（WARN 或 ERROR）发送到特定的数据流，并将常规日志（DEBUG 和 INFO）保留在默认数据流中

2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%.
2023-08-08T13:45:14.003Z ERROR 192.168.1.103 Database connection failed.
2023-08-08T13:45:15.004Z DEBUG 192.168.1.104 Debugging connection issue.
2023-08-08T13:45:16.005Z INFO 192.168.1.102 User changed profile picture.

将数据路由到不同的数据流时，我们建议选择具有有限数量的不同值的字段，以防止数据流数量过度增加。有关更多详细信息，请参阅调整分片大小文档。

要使用重新路由处理器

将重新路由处理器添加到摄取管道

编辑

使用以下命令将重新路由处理器添加到您的摄取管道

PUT _ingest/pipeline/logs-example-default
{
  "description": "Extracts fields and reroutes WARN",
  "processors": [
    {
      "dissect": {
        "field": "message",
        "pattern": "%{@timestamp} %{log.level} %{host.ip} %{message}"
      },
      "reroute": {
        "tag": "high_severity_logs",
        "if" : "ctx.log?.level == 'WARN' || ctx.log?.level == 'ERROR'",
        "dataset": "critical"
      }
    }
  ]
}

	`tag`：可用于调试和指标的处理器的标识符。在示例中，标签设置为 `high_severity_logs`。
	`if`：有条件地运行处理器。在示例中，`"ctx.log?.level == 'WARN' \|\| ctx.log?.level == 'ERROR'"` 表示当 `log.level` 字段为 `WARN` 或 `ERROR` 时，处理器运行。
	`dataset`：如果前面的条件为 `true`，则将您的文档路由到的数据流数据集。在示例中，`log.level` 为 `WARN` 或 `ERROR` 的日志会被路由到 `logs-critical-default` 数据流。

除了设置摄取管道之外，您还需要设置索引模板。您可以使用在提取 @timestamp 字段部分中创建的索引模板。

将日志添加到数据流

编辑

使用以下命令将示例日志添加到您的数据流

POST logs-example-default/_bulk
{ "create": {} }
{ "message": "2023-08-08T13:45:12.123Z WARN 192.168.1.101 Disk usage exceeds 90%." }
{ "create": {} }
{ "message": "2023-08-08T13:45:14.003Z ERROR 192.168.1.103 Database connection failed." }
{ "create": {} }
{ "message": "2023-08-08T13:45:15.004Z DEBUG 192.168.1.104 Debugging connection issue." }
{ "create": {} }
{ "message": "2023-08-08T13:45:16.005Z INFO 192.168.1.102 User changed profile picture." }

验证重新路由处理器是否正常工作

编辑

重新路由处理器应将 log.level 为 WARN 或 ERROR 的任何日志路由到 logs-critical-default 数据流。使用以下命令查询数据流以验证日志数据是否按预期路由

GET logs-critical-default/_search

您应该看到与以下结果类似的结果，表明高严重性日志现在位于 critical 数据集中

{
  ...
  "hits": {
    ...
    "hits": [
        ...
        "_source": {
          "host": {
            "ip": "192.168.1.101"
          },
          "@timestamp": "2023-08-08T13:45:12.123Z",
          "message": "Disk usage exceeds 90%.",
          "log": {
            "level": "WARN"
          },
          "data_stream": {
            "namespace": "default",
            "type": "logs",
            "dataset": "critical"
          },
          {
        ...
        "_source": {
          "host": {
            "ip": "192.168.1.103"
           },
          "@timestamp": "2023-08-08T13:45:14.003Z",
          "message": "Database connection failed.",
          "log": {
            "level": "ERROR"
          },
          "data_stream": {
            "namespace": "default",
            "type": "logs",
            "dataset": "critical"
          }
        }
      }
    ]
  }
}

« 流式传输任何日志文件过滤和聚合日志 »

解析和组织日志

解析和组织日志

提取结构化字段

提取 @timestamp 字段

使用摄取管道提取 @timestamp 字段

使用模拟管道 API 测试管道

使用索引模板配置数据流

创建数据流

问题排查 @timestamp 字段

提取 log.level 字段

将 log.level 添加到您的摄取管道

使用模拟 API 测试管道

根据 log.level 查询日志

提取 host.ip 字段

将 host.ip 添加到您的摄取管道

使用模拟 API 测试管道

根据 host.ip 查询日志

CIDR 表示法

范围查询

将日志数据重新路由到特定的数据流

将重新路由处理器添加到摄取管道

将日志添加到数据流

验证重新路由处理器是否正常工作

提取 `@timestamp` 字段

使用摄取管道提取 `@timestamp` 字段

问题排查 `@timestamp` 字段

提取 `log.level` 字段

将 `log.level` 添加到您的摄取管道

根据 `log.level` 查询日志

提取 `host.ip` 字段

将 `host.ip` 添加到您的摄取管道

根据 `host.ip` 查询日志