Elastic Serverless Forwarder 配置选项

了解有关 Elastic Serverless Forwarder 配置选项的更多信息，包括权限和策略的详细信息、AWS 服务日志的自动路由以及如何使用 AWS Secrets Manager 进行身份验证。

您可以转换或丰富来自 AWS 的数据，因为转发器会对其进行解析。这包括一些示例，例如使用标签和过滤器来组织数据并排除特定消息，自动发现和收集 JSON 内容以及有效地管理多行消息。

Lambda 函数的执行角色是一个 AWS Identity and Access Management (IAM) 角色，它授予函数访问 AWS 服务和资源的权限。此角色在部署函数时自动创建，并且在调用函数时，Lambda 会承担此角色。

当您提供转发器将与其交互的 AWS 资源的 ARN 时，Cloudformation 模板将创建具有相应 IAM 策略的正确 IAM 角色。

您可以从 配置 > 权限 部分查看与您的 Lambda 函数关联的执行角色。默认情况下，此角色名称以 serverlessrepo- 开头。创建角色时，会添加自定义策略以授予 Lambda 最小权限，以便能够使用已配置的 SQS 队列、S3 存储桶、Kinesis 数据流、CloudWatch Logs 日志组、Secrets Manager（如果使用）和 SQS 重播队列。

转发器被授予以下 ManagedPolicyArns 权限，这些权限默认情况下会自动添加到 Events 配置（如果相关）。

arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
arn:aws:iam::aws:policy/service-role/AWSLambdaKinesisExecutionRole
arn:aws:iam::aws:policy/service-role/AWSLambdaSQSQueueExecutionRole

除了这些基本权限外，函数的 Cloudformation 模板创建函数时还会添加以下权限：

对于要用作转发器触发器的 CloudWatch Logs 订阅筛选器日志组资源，以下内容在单独的策略声明中被允许作为基于资源的策略：

  * Principal: logs.%AWS_REGION%.amazonaws.com
  * Action: lambda:InvokeFunction
  * Source ARN: arn:aws:logs:%AWS_REGION%:%AWS_ACCOUNT_ID%:log-group:%LOG_GROUP_NAME%:*

AWS Secrets Manager 使您能够使用对 Secrets Manager 的 API 调用以编程方式检索密钥来替换代码中硬编码的凭据（包括密码）。有关更多信息，请参阅 AWS Secrets Manager 文档。

可以使用两种类型的密钥：

以下代码显示了对 AWS Secrets Manager 的 API 调用：

inputs:
  - type: "s3-sqs"
    id: "arn:aws:sqs:%REGION%:%ACCOUNT%:%QUEUENAME%"
    outputs:
      - type: "elasticsearch"
        args:
          elasticsearch_url: "arn:aws:secretsmanager:eu-central-1:123456789:secret:es_url"
          username: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:username"
          password: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:password"
          es_datastream_name: "logs-generic-default"

要使用 纯文本 或 二进制 密钥，请注意 ARN 的以下格式：

arn:aws:secretsmanager:AWS_REGION:AWS_ACCOUNT_ID:secret:SECRET_NAME

为了使用 键/值 对密钥，您需要在 arn 的末尾提供键，如下所示：

arn:aws:secretsmanager:AWS_REGION:AWS_ACCOUNT_ID:secret:SECRET_NAME:SECRET_KEY

对于 S3 SQS 事件通知 输入，Elastic Serverless Forwarder 支持将多个 AWS 服务日志自动路由到相应的 集成数据流，以便在 Elasticsearch 集群中进一步处理和存储。

Elastic Serverless Forwarder 支持将以下日志自动路由到相应默认的集成数据流：

对于这些用例，在配置文件中设置 es_datastream_name 字段是可选的。

对于大多数其他用例，您需要在配置文件中设置 es_datastream_name 字段才能将数据路由到特定数据流或索引。在以下用例中应设置此值：

如果未指定 es_datastream_name，并且日志无法与上述任何 AWS 服务匹配，则数据集将设置为 generic，命名空间设置为 default，指向数据流名称 logs-generic-default。

您可以使用标签和过滤器根据正则表达式标记和筛选消息。

您可以添加自定义标签来筛选和分类事件中的项目。

inputs:
  - type: "s3-sqs"
    id: "arn:aws:sqs:%REGION%:%ACCOUNT%:%QUEUENAME%"
    tags:
      - "tag1"
      - "tag2"
      - "tag3"
    outputs:
      - type: "elasticsearch"
        args:
          elasticsearch_url: "arn:aws:secretsmanager:eu-central-1:123456789:secret:es_url"
          username: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:username"
          password: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:password"
          es_datastream_name: "logs-generic-default"

使用上述示例配置，标签将按以下方式设置：

["forwarded", "generic", "tag1", "tag2", "tag3"]

forwarded 标签始终附加，此示例中的 generic 标签来自数据集。

您可以为输入定义多个过滤器，以包含或排除数据摄取中的事件。

inputs:
  - type: "s3-sqs"
    id: "arn:aws:sqs:%REGION%:%ACCOUNT%:%QUEUENAME%"
    include:
      - "[a-zA-Z]"
    exclude:
      - "skip this"
      - "skip also this"
    outputs:
      - type: "elasticsearch"
        args:
          elasticsearch_url: "arn:aws:secretsmanager:eu-central-1:123456789:secret:es_url"
          username: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:username"
          password: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:password"
          es_datastream_name: "logs-generic-default"

您可以在 inputs.[].include 中定义正则表达式列表。如果填充此列表，则只有匹配任何已定义正则表达式的消息才会转发到输出。

您可以在 inputs.[].exclude 中定义正则表达式列表。如果填充此列表，则只有不匹配任何已定义正则表达式的消息才会转发到输出，即除非消息匹配任何已定义的正则表达式，否则每条消息都会转发到输出。

Elastic Serverless Forwarder 能够自动发现输入有效负载中的 JSON 内容，并收集有效负载中包含的 JSON 对象。

JSON 对象可以位于单行上，也可以跨多行。在第二种情况下，转发器期望跨多行的不同 JSON 对象以换行符分隔。

当 JSON 对象跨越多行时，会应用 1000 行的限制。任何跨越超过 1000 行的 JSON 对象都将不会被收集。组成整个 JSON 对象的每一行将单独转发。

如果您已知有效负载内容包含跨越超过 1000 行的单个 JSON 对象，或者您发现依赖于 JSON 内容的自动发现会对性能产生重大影响，则可以在输入中配置 JSON 内容类型来解决此问题。这将改变解析逻辑并提高性能，同时克服 1000 行的限制。

如果已知内容为纯文本，则可以通过完全禁用自动 JSON 内容发现来提高整体性能。

要更改此配置选项，请将inputs.[].json_content_type设置为以下值之一

您可以从 JSON 文件中的特定字段提取要摄取的事件列表。

inputs:
  - type: "s3-sqs"
    id: "arn:aws:sqs:%REGION%:%ACCOUNT%:%QUEUENAME%"
    expand_event_list_from_field: "Records"
    # root_fields_to_add_to_expanded_event: "all"
    # root_fields_to_add_to_expanded_event: ["owner", "logGroup", "logStream"]
    outputs:
      - type: "elasticsearch"
        args:
          elasticsearch_url: "arn:aws:secretsmanager:eu-central-1:123456789:secret:es_url"
          username: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:username"
          password: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:password"
          es_datastream_name: "logs-generic-default"

您可以将inputs.[].expand_event_list_from_field定义为一个字符串，其值为 JSON 中包含必须作为事件发送的元素列表（而不是包含 JSON）的键。

要将 JSON 对象中的根字段注入展开的事件中，请定义inputs.[].root_fields_to_add_to_expanded_event。此配置采用以下值之一

使用以下输入

{"Records":[{"key": "value #1"},{"key": "value #2"}]}
{"Records":[{"key": "value #3"},{"key": "value #4"}]}

不设置expand_event_list_from_field，将转发两个事件

{"@timestamp": "2022-06-16T04:06:03.064Z", "message": "{\"Records\":[{\"key\": \"value #1\"},{\"key\": \"value #2\"}]}"}
{"@timestamp": "2022-06-16T04:06:13.888Z", "message": "{\"Records\":[{\"key\": \"value #3\"},{\"key\": \"value #4\"}]}"}

如果expand_event_list_from_field设置为Records，将转发四个事件

{"@timestamp": "2022-06-16T04:06:21.105Z", "message": "{\"key\": \"value #1\"}"}
{"@timestamp": "2022-06-16T04:06:27.204Z", "message": "{\"key\": \"value #2\"}"}
{"@timestamp": "2022-06-16T04:06:31.154Z", "message": "{\"key\": \"value #3\"}"}
{"@timestamp": "2022-06-16T04:06:36.189Z", "message": "{\"key\": \"value #4\"}"}

使用以下输入

{"Records":[{"key": "value #1"},{"key": "value #2"}], "field1": "value 1a", "field2": "value 2a"}
{"Records":[{"key": "value #3"},{"key": "value #4"}], "field1": "value 1b", "field2": "value 2b"}

如果expand_event_list_from_field设置为Records，并且root_fields_to_add_to_expanded_event设置为all，将转发四个事件

{"@timestamp": "2022-06-16T04:06:21.105Z", "message": "{\"key\": \"value #1\", \"field1\": \"value 1a\", \"field2\": \"value 2a\""}
{"@timestamp": "2022-06-16T04:06:27.204Z", "message": "{\"key\": \"value #2\", \"field1\": \"value 1a\", \"field2\": \"value 2a\""}
{"@timestamp": "2022-06-16T04:06:31.154Z", "message": "{\"key\": \"value #3\", \"field1\": \"value 1b\", \"field2\": \"value 2b\""}
{"@timestamp": "2022-06-16T04:06:36.189Z", "message": "{\"key\": \"value #4\", \"field1\": \"value 1b\", \"field2\": \"value 2b\""}

如果expand_event_list_from_field设置为Records，并且root_fields_to_add_to_expanded_event设置为["field1"]，将转发四个事件

{"@timestamp": "2022-06-16T04:06:21.105Z", "message": "{\"key\": \"value #1\", \"field1\": \"value 1a\""}
{"@timestamp": "2022-06-16T04:06:27.204Z", "message": "{\"key\": \"value #2\", \"field1\": \"value 1a\""}
{"@timestamp": "2022-06-16T04:06:31.154Z", "message": "{\"key\": \"value #3\", \"field1\": \"value 1b\""}
{"@timestamp": "2022-06-16T04:06:36.189Z", "message": "{\"key\": \"value #4\", \"field1\": \"value 1b\""}

转发的內容可能包含跨越多行的消息。例如，包含 Java 堆栈跟踪的文件中常见的都是多行消息。为了正确处理这些多行事件，您需要为特定输入配置multiline设置，以指定哪些行属于单个事件。

您可以在config.yaml文件中为特定输入指定以下选项，以控制 Elastic Serverless Forwarder 如何处理跨越多行的消息。

inputs:
  - type: "s3-sqs"
    id: "arn:aws:sqs:%REGION%:%ACCOUNT%:%QUEUENAME%"
    multiline:
      type: pattern
      pattern: '^\\['
      negate: true
      match: after
    outputs:
      - type: "elasticsearch"
        args:
          elasticsearch_url: "arn:aws:secretsmanager:eu-central-1:123456789:secret:es_url"
          username: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:username"
          password: "arn:aws:secretsmanager:eu-west-1:123456789:secret:es_secrets:password"
          es_datastream_name: "logs-generic-default"

转发器获取所有不以[开头的行，并将它们与之前以[开头的行组合在一起。例如，您可以使用此配置将多行消息的以下行合并到单个事件中

[beat-logstash-some-name-832-2015.11.28] IndexNotFoundException[no such index]
    at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver$WildcardExpressionResolver.resolve(IndexNameExpressionResolver.java:566)
    at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.concreteIndices(IndexNameExpressionResolver.java:133)
    at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.concreteIndices(IndexNameExpressionResolver.java:77)
    at org.elasticsearch.action.admin.indices.delete.TransportDeleteIndexAction.checkBlock(TransportDeleteIndexAction.java:75)

inputs.[].multiline.type定义要使用的聚合方法。默认值为pattern。其他选项包括count（允许您聚合固定数量的行）和while_pattern（允许根据模式聚合行，但不进行匹配选项）。

inputs.[].multiline.pattern与 Logstash 支持的模式不同。有关支持的正则表达式模式列表，请参阅Python 3.9 正则表达式语法。根据您如何配置其他多行选项，与指定正则表达式匹配的行被视为先前行的延续或新的多行事件的开始。

inputs.[].multiline.negate定义是否否定模式。默认值为false。此设置仅适用于pattern和while_pattern类型。

inputs.[].multiline.match根据以下模式更改多行的分组（仅适用于pattern类型）

inputs.[].multiline.flush_pattern指定一个正则表达式，在该表达式中，当前多行将从内存中刷新，结束多行消息。仅适用于pattern类型。

inputs.[].multiline.max_lines定义可以组合到一个事件中的最大行数。如果多行消息包含超过max_lines行，则任何附加行都将从事件中截断。默认值为500。

inputs.[].multiline.max_bytes定义可以组合到一个事件中的最大字节数。如果多行消息包含超过max_bytes字节，则任何附加内容都将从事件中截断。默认值为10485760。

inputs.[].multiline.count_lines定义要聚合到单个事件中的行数。仅适用于count类型。

inputs.[].multiline.skip_newline定义是否必须连接多行事件，同时去除行分隔符。如果设置为true，则将去除行分隔符。默认值为false。

本节中的示例涵盖以下用例

Java 堆栈跟踪由多行组成，初始行后的每一行都以空格开头，例如：

Exception in thread "main" java.lang.NullPointerException
        at com.example.myproject.Book.getTitle(Book.java:16)
        at com.example.myproject.Author.getBookTitles(Author.java:25)
        at com.example.myproject.Bootstrap.main(Bootstrap.java:14)

此配置合并任何以空格开头的行到前一行

multiline:
  type: pattern
  pattern: '^\s'
  negate: false
  match: after

这是一个稍微复杂的 Java 堆栈跟踪示例

Exception in thread "main" java.lang.IllegalStateException: A book has a null property
       at com.example.myproject.Author.getBookIds(Author.java:38)
       at com.example.myproject.Bootstrap.main(Bootstrap.java:14)
Caused by: java.lang.NullPointerException
       at com.example.myproject.Book.getId(Book.java:22)
       at com.example.myproject.Author.getBookIds(Author.java:35)
       ... 1 more

要将这些行合并到单个事件中，请使用以下多行配置

multiline:
  type: pattern
  pattern: '^\s+(at|.{3})\s+\\b|^Caused by:'
  negate: false
  match: after

在此示例中，模式匹配并合并以下行：- 以空格开头，后跟at或...的行 - 以Caused by:开头的行

几种编程语言使用行尾的反斜杠（\）字符来表示行延续，例如：

printf ("%10.10ld  \t %10.10ld \t %s\
  %f", w, x, y, z );

要将这些行合并到单个事件中，请使用以下多行配置

multiline:
  type: pattern
  pattern: '\\\\$'
  negate: false
  match: after

此配置将任何以\字符结尾的行与下一行合并。

Elasticsearch 等服务的活动日志通常以时间戳开头，后跟有关特定活动的信息，例如：

[2015-08-24 11:49:14,389][INFO ][env                      ] [Letha] using [1] data paths, mounts [[/
(/dev/disk1)]], net usable_space [34.5gb], net total_space [118.9gb], types [hfs]

要将这些行合并到单个事件中，请使用以下多行配置

multiline:
  type: pattern
  pattern: '^\\[[0-9]{4}-[0-9]{2}-[0-9]{2}'
  negate: true
  match: after

此配置使用negate: true和match: after设置来指定任何不匹配指定模式的行都属于前一行。

有时您的应用程序日志包含以自定义标记开头和结尾的事件，例如以下示例

[2015-08-24 11:49:14,389] Start new event
[2015-08-24 11:49:14,395] Content of processing something
[2015-08-24 11:49:14,399] End event

要将这些行合并到单个事件中，请使用以下多行配置

multiline:
  type: pattern
  pattern: 'Start new event'
  negate: true
  match: after
  flush_pattern: 'End event'

flush_pattern选项指定一个正则表达式，在该表达式处将刷新当前多行。如果您认为pattern选项指定事件的开头，则flush_pattern选项将指定事件的结尾或最后一行。

[2015-08-24 11:49:14,389] Start new event
[2015-08-24 11:49:14,395] Content of processing something
[2015-08-24 11:49:14,399] End event
[2015-08-24 11:50:14,389] Some other log
[2015-08-24 11:50:14,395] Some other log
[2015-08-24 11:50:14,399] Some other log
[2015-08-24 11:51:14,389] Start new event
[2015-08-24 11:51:14,395] Content of processing something
[2015-08-24 11:51:14,399] End event

为了方便您测试多行配置中的正则表达式模式，您可以使用此多行正则表达式测试工具

从 v1.5.0 版本开始，ESF 引入了 SSL 指纹选项，用于使用自签名证书访问 Elasticsearch 集群。

要设置ssl_assert_fingerprint选项，必须编辑存储在 S3 存储桶中的配置文件。

假设您在存储桶中有一个config.yml文件，其内容如下所示

inputs:
  - type: "s3-sqs"
    id: "arn:aws:sqs:eu-west-1:123456789:dev-access-logs"
    outputs:
      - type: "elasticsearch"
        args:
          api_key: "<REDACTED>"
          es_datastream_name: "logs-aws.s3access-default"
          batch_max_actions: 500
          batch_max_bytes: 10485760
          ssl_assert_fingerprint: ""

如果配置省略了ssl_assert_fingerprint，或者像此示例一样为空（默认选项），则 HTTP 客户端将验证 Elasticsearch 集群的证书。

下一步是获取 Elasticsearch 集群当前使用的 HTTPS 证书的指纹。

您可以使用 OpenSSL 获取证书的指纹。以下是一个使用托管在 Elastic Cloud 上的 Elasticsearch 集群的示例

$ openssl s_client \
    -connect my-deployment.es.eastus2.azure.elastic-cloud.com:443 \
    -showcerts </dev/null 2>/dev/null | openssl x509 -noout -fingerprint

SHA1 Fingerprint=1C:46:32:75:AA:D6:F1:E2:8E:10:A3:64:44:B1:36:C9:7D:44:35:B4

您可以使用您的 DNS 名称、IP 地址和端口号替换上述示例中的my-deployment.es.eastus2.azure.elastic-cloud.com:443。

复制您的指纹值以进行下一步操作。

最后一步是编辑您的config.yml文件以使用 SSL 指纹。

inputs:
  - type: "s3-sqs"
    id: "arn:aws:sqs:eu-west-1:123456789:dev-access-logs"
    outputs:
      - type: "elasticsearch"
        args:
          api_key: "<REDACTED>"
          es_datastream_name: "logs-aws.s3access-default"
          batch_max_actions: 500
          batch_max_bytes: 10485760
          ssl_assert_fingerprint: "1C:46:32:75:AA:D6:F1:E2:8E:10:A3:64:44:B1:36:C9:7D:44:35:B4"