作者：Carly Richmond

记录日志还是不记录日志？一直是软件工程师仍在努力解决的一个难题，这对站点可靠性工程（SRE）同事来说是不利的。开发人员并不总能正确了解他们在应用程序中捕获的警告和错误的级别或上下文，并且经常记录可能并不总是对 SRE 有帮助的消息。我可以承认自己是这些开发人员之一！这种混乱通常会导致大量事件被摄取到日志平台中，从而使 SRE 的应用程序监控和问题调查有点像这样：

来源: GIPHY

当寻求减少日志量时，可以删除两个维度的信息：事件内的字段或整个事件本身。删除感兴趣的维度可确保我们能够专注于已知的感兴趣事件和可能感兴趣的未知事件。

在本博客中，我们将讨论通过各种收集器从日志中删除已知的不相关事件和字段的各种方法。具体来说，我们将重点关注 Beats、Logstash、Elastic Agent、Ingest Pipelines 以及使用 OpenTelemetry Collectors 进行过滤。

Beats

Beats 是一个轻量级数据采集程序系列，允许转发来自特定源的事件。它们通常用于将事件从源提取到 Elasticsearch 中，还包括其他输出，例如 Logstash、Kafka 或 Redis，如 Filebeat 文档中所示。有六种可用的 Beats 类型，此处总结如下。

我们的示例将专门关注 Filebeat，但这里讨论的两个 drop 处理器都适用于所有 Beats。按照 Filebeat 文档中的快速入门指南进行操作后，你将拥有一个正在运行的进程，该进程使用配置文件 filebeat.yml 指示你正在使用 Filebeat 从任何受支持的输入类型监控哪些日志文件。希望你的配置指定一系列类似于以下格式的输入：

filebeat.inputs:
- type: filestream
  id: my-logging-app
  paths:
    - /var/log/*.log

Filebeat 有许多可供配置的选项，文档中的 filebeat.reference.yml 中给出了完整的列表。然而，drop_event 和 drop_fields 处理器尤其可以帮助我们排除无用的消息并分别隔离给定事件中的相关字段。使用 drop_event 处理器时，你需要确保至少存在一个条件才能接收你想要的消息；否则，如果没有指定条件，处理器将丢弃所有消息。如果 drop_event 处理器中未指定条件，则所有事件都将被删除。请确保至少满足一项条件，以确保你收到所需的消息。例如，如果我们对针对 /profile 端点的 HTTP 请求不感兴趣，我们可以修改配置以使用以下条件：

filebeat.inputs:
- type: filestream
  id: my-logging-app
  paths:
    - /var/tmp/other.log
    - /var/log/*.log
processors:
  - drop_event:
      when:
          and:
            - equals:
              url.scheme: http
            - equals:
              url.path: /profile

同时，如果满足指定条件，drop_fields 处理器将删除指定字段（@timestamp 和 type 字段除外）。与 drop_event 处理器类似，如果条件缺失，则字段将始终被删除。如果我们想排除成功的 HTTP 请求的错误消息字段，我们可以配置类似于以下的处理器：

filebeat.inputs:
- type: filestream
  id: my-logging-app
  paths:
    - /var/tmp/other.log
    - /var/log/*.log
processors:
  - drop_fields:
      when:
          and:
            - equals:
              url.scheme: http
            - equals:
              http.response.status_code: 200
          fields: ["event.message"]
          ignore_missing: false

删除字段时，该字段始终有可能不存在于给定的日志消息中。如果 Filebeat 中正在处理的所有事件中不存在该字段，则如果将 ignore_missing 指定为 true 而不是默认值 false ，否则会引发错误。

Logstash过滤

Logstash 是一个免费、开放的数据处理管道工具，允许你在无数来源之间提取、转换和输出数据。它位于提取、转换和加载（或 ETL）域内。在前面对 Beats 的讨论中，需要注意的是，如果你想集中转换逻辑，建议使用 Logstash 而不是 Beats。同时，Beats 或 Elastic Agent 允许提前丢弃事件，这可以提前减少网络流量需求。 Logstash 提供了各种开箱即用的转换插件，可用于格式化和转换来自 Logstash 连接到的任何源的事件。 Logstash 配置文件 logstash.yml 中的典型管道包含三个主要部分：

input 表示管道的数据源。
filter 包含相关的数据转换逻辑。
转换数据的目标在 output 属性中配置。为了防止事件到达输出，drop 过滤器插件将删除满足规定条件的所有事件。

从输入文件读取、删除 INFO 级别事件并输出到 Elasticsearch 的典型示例如下：

input {
  file {
    id => "my-logging-app"
    path => [ "/var/tmp/other.log", "/var/log/*.log" ]
  }
}

filter {
  if [url.scheme] == "http" && [url.path] == "/profile" {
    drop {
      percentage => 80
    }
  }
}

output {
  elasticsearch {
        hosts => "https://my-elasticsearch:9200"
        data_stream => "true"
    }
}

此过滤器鲜为人知的选项之一是能够使用百分比（percentage）选项配置丢弃率。过滤日志事件的可怕之处之一是担心你会无意中删除未知但相关的条目，这些条目在中断情况下可能有用。或者，你的软件可能会发送大量消息，这些消息可能会淹没你的实例、占用重要的热存储并增加你的成本。百分比属性通过允许将事件的子集引入 Elasticsearch 来涵盖这种情况，这可以解决这些挑战。在上面的示例中，我们将 20% 与 Elasticsearch 条件匹配的消息摄取。与 Beats 中的 drop_fields 处理器类似，Logstash 有一个 remove_field 过滤器，用于删除单个字段。尽管这些可以在许多 Logstash 插件中使用，但它们通常在 mutate 插件中用于转换事件，类似于以下内容：

# Input configuration omitted

filter {
  if [url.scheme] == "http" && [http.response.status_code] == 200 {
    drop {
      percentage => 80
    }
    mutate {
      remove_field: [ "event.message" ]
    }
  }
}

# Output configuration omitted

就像我们的 Beats 示例一样，这将从 drop 过滤器保留的事件中删除 event.message。

Elastic Agent

Elastic Agent 是日志、指标和安全数据的单一代理，可以在你的主机上执行并将事件从多个服务和基础设施发送到 Elasticsearch。与 Beats 类似，你可以在任何支持处理器的集成中使用 drop_event 和 drop_fields 处理器。对于独立安装，你应该在 elastic-agent.yml 配置中指定处理器。使用 Fleet 时，通常在高级选项弹出部分下配置集成时指定处理转换，如下所示：

将上面的示例与我们的 Beats 示例进行比较，你会注意到它们对两个处理器使用相同的基于 YAML 的格式。使用 Elastic Agent 处理器时需要注意一些限制，这些限制在 Fleet 文档中进行了介绍。如果你不确定通过 Elastic Agent 处理器处理数据是否适合你的用例，请查看这个方便的矩阵。

Ingest pipeline

上一节中讨论的 Elastic Agent 处理器将处理原始事件数据，这意味着它们在摄取管道（ingest pipeline）之前执行。因此，在使用这两种方法时，请谨慎行事，因为删除或更改摄取管道所需的字段可能会导致管道中断。正如创建和管理管道文档中所述，可以在 Stack Management > Ingest Pipelines 屏幕中或通过我们将使用的 _ingest API 创建新管道。就像本文中介绍的其他工具一样，drop 处理器将允许删除任何满足所需条件的事件。同样的情况是，如果没有指定条件，所有通过的事件都将被丢弃。不同的是，条件逻辑是使用 Painless（一种类似 Java 的脚本语言）编写的，而不是我们之前使用的 YAML 语法：

PUT _ingest/pipeline/my-logging-app-pipeline
{
  "description": "Event and field dropping for my-logging-app",
  "processors": [
    {
      "drop": {
        "description" : "Drop event",
        "if": "ctx?.url?.scheme == 'http' && ctx?.url?.path == '/profile'",
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "description" : "Drop field",
        "field" : "event.message",
        "if": "ctx?.url?.scheme == 'http' && ctx?.http?.response?.status_code == 200",
        "ignore_failure": false
      }
    }
  ]
}

ctx 变量是通过管道传输的文档中字段的映射表示形式，这意味着我们的示例将比较 url.scheme 和 http.response.status_code 字段的值。 JavaScript 开发人员会认识到 ? 表示空安全运算符，它对字段访问执行非空检查。如上例中的第二个处理器所示，Painless 条件逻辑的使用也与 remove 处理器相关。当指定字段与指定条件匹配时，该处理器将从事件中删除指定字段。摄取处理器比其他方法更具优势的功能之一是能够指定管道上或指定处理器上的故障处理器。尽管 Beat 确实有一个 ignore_missing 选项（如前所述），但摄取处理器允许我们添加异常处理，例如添加错误消息以提供处理器异常的详细信息：

PUT _ingest/pipeline/my-logging-app-pipeline
{
  "description": "Event and field dropping for my-logging-app with failures",
  "processors": [
    {
      "drop": {
        "description" : "Drop event",
        "if": "ctx?.url?.scheme == 'http' && ctx?.url?.path == '/profile'",
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "description" : "Drop field",
        "field" : "event.message",
        "if": "ctx?.url?.scheme == 'http' && ctx?.http?.response?.status_code == 200",
        "ignore_failure": false
      }
    }
  ],
  "on_failure": [
    {
      "set": {
        "description": "Set 'ingest.failure.message'",
        "field": "ingest.failure.message",
        "value": "Ingestion issue"
        }
      }
  ]
}

然后，该管道可以用于单个索引请求，设置为索引的默认管道，甚至可以与 Beats 和 Elastic Agent 一起使用。

OpenTelemetry 收集器

OpenTelemetry 或 OTel 是一种开放标准，提供 API、工具和集成，以支持从应用程序捕获遥测数据，例如日志、指标和跟踪。应用程序开发人员通常使用 OpenTelemetry 代理作为他们选择的编程语言，将跟踪数据和指标直接发送到 Elastic Stack，因为 Elastic 支持 OpenTelemetry 协议 (OTLP)。在某些情况下，让每个应用程序直接向可观察平台发送行为信息可能是不明智的。大型企业生态系统可能具有集中的可观察能力，或者可能运行大型微服务生态系统，而在这些生态系统中采用标准跟踪实践可能很困难。然而，随着应用程序和服务数量的增长，事件和跟踪的清理也面临挑战。在这些情况下，使用一个或多个收集器作为 Elastic Stack 的数据路由器是有意义的。如 OpenTelemetry 文档和 Elastic APM 文档中的示例收集器所示，OTel 收集器的基本配置有四个主要部分：

定义数据源的 receivers，可以是基于推送或拉取的。
processors 可以在导出之前过滤或转换接收到的数据，这是我们感兴趣的事情。
exporters 定义如何将数据发送到最终目的地，在本例中是 Elastic！
service 部分，用于定义其他元素所需的收集器中启用的组件。可以分别使用收集器配置中的 filter 和 attribute 处理器来删除事件和字段。

两种过滤器的选择示例如下所示：

receivers: 
  filelog:
    include: [ /var/tmp/other.log, /var/log/*.log ]
processors: 
  filter/denylist:
    error_mode: ignore
    logs:
      log_record:
        - 'url.scheme == "info"'
        - 'url.path == "/profile"'
        - 'http.response.status_code == 200'
  attributes/errors:
    actions:
      - key: error.message
        action: delete
  memory_limiter:
    check_interval: 1s
    limit_mib: 2000
  batch:
exporters:
  # Exporters configuration omitted 
service:
  pipelines:
    # Pipelines configuration omitted

如果事件符合任何指定条件，则应用于 telemetry 类型（日志、指标或跟踪）的 filter 处理器将删除该事件。同时，应用于错误字段的 attribute 处理器将从所有事件中删除 error.message 属性。还可以使用 pattern 属性代替 key 选项来删除与指定正则表达式匹配的字段。正如我们在 Beats、Logstash 和摄取管道示例中所做的那样，基于条件的字段删除不是规范的一部分。然而，另一种方法是使用 transform 处理器指定复杂的转换来设置字段的值，然后删除。

结论

DevOps 运动的目标是协调软件工程和 SRE 的流程和实践。这包括共同努力确保相关日志、指标和跟踪从应用程序发送到我们的可观察性平台。正如我们在 Beats、Logstash、Elastic Agent、Ingest Pipelines 和 OTel 收集器中亲眼所见，删除事件和单个字段的方法根据所使用的工具而有所不同。你可能想知道哪个选项适合你？

如果担心通过网络发送大型消息的开销，那么使用 Beats 或 Logstash 进行更接近源的转换是更好的选择。如果你希望最大限度地减少收集和转换中使用的系统资源，Beats 可能比 Logstash 更受青睐，因为它们占用空间较小。
为了集中转换逻辑以应用于许多应用程序日志，使用 OTel 收集器中的处理器可能是正确的方法。
如果你想通过流行的服务和系统（例如 Kafka、Nginx 或 AWS）使用集中管理的摄取和转换策略，建议使用由 Fleet 管理的 Elastic Agent 来采集数据。
如果你不太关心网络开销并且希望将逻辑集中在 Elasticsearch 中，那么摄取管道非常适合在摄取时转换事件。尽管此处未介绍，但也可以使用其他技术（例如运行时字段、索引级压缩和 _synthetic_source 使用）来减少磁盘存储要求和 CPU 开销。如果此处未列出你最喜欢的删除事件或字段的方式，请告诉我们！