去年，我们推出了 Connect with Confluent 合作伙伴计划，使我们的技术合作伙伴能够开发与 Confluent Cloud 的原生集成。这使我们的客户能够从他们喜爱的应用程序中访问 Confluent 数据流，并让他们从数据中获取最大价值。目前该计划已有 40 多个合作伙伴，其中包括两项 Amazon Web Services (AWS) 服务——Amazon Athena 和 AWS Lambda——我们正在迅速扩展 Confluent 本已庞大的数据流生态系统。今天，我们很高兴地宣布 Amazon EventBridge 已加入 Connect with Confluent 计划，成为第三项加入这一合作努力的 AWS 服务。

EventBridge 是一项完全托管的服务，可促进 AWS 与第三方服务之间的无缝事件路由。超过 150 万客户使用 EventBridge 每月发送超过 2.6 万亿条消息。

目前，有两种类型的 Confluent 与 EventBridge 集成可用。首先，API（应用程序编程接口）目标使客户能够将 EventBridge 数据发送到 Confluent。其次，还有 EventBridge Pipes（一种完全托管的点对点集成）和 EventBridge Kafka Connector，两者都将数据从 Confluent 下沉到事件。在这篇博文中，我们将主要关注 API 目标集成。此外，我们将讨论如何使用 Confluent Cloud 作为源来设置与 EventBridge Pipes 和连接器的集成。

EventBridge 到 Confluent 集成的用例

Confluent 和 EventBridge 相似，但服务于不同的用例。Confluent Cloud 是一个完整的数据流平台，专为大规模管理和处理实时数据流而设计。Confluent 提供流处理、模式管理和连接器等高级功能，用于与各种数据源和数据接收器集成，使其成为在多云或混合环境中构建事件驱动架构的多功能选择。对于具有复杂实时事件处理需求的高吞吐量和低延迟用例，它是理想的选择。例如，Confluent 为 GBps+ 或每秒数千万事件提供低于 100 毫秒的 p99 端到端延迟。此外，组织可以利用 Confluent 提供的 120 多个预构建连接器，将数据无缝流式传输到 AWS 上几乎任何数据存储或从中流出。

EventBridge 是一个事件路由器，连接各种 AWS 和第三方服务。它还提供 EventBridge Pipes 用于点对点集成，以及 EventBridge Scheduler，这是一种无服务器任务调度器，可简化跨 AWS 服务创建、执行和管理数百万个计划的过程，而无需预置或管理底层基础设施。超过 160 项 AWS 服务产生通过 EventBridge 路由的操作和管理事件。此外，EventBridge 支持来自 30 多个第三方服务的事件。将 Confluent 数据流平台与 EventBridge 集成，使组织能够实时从所有这些数据源中释放价值。

通过将两种服务集成在一起，我们可以解锁众多用例，其中一个突出的例子是实时分析 AWS CloudTrail 事件。这使组织能够实时捕获和处理 CloudTrail 事件，用于各种用例，如安全监控、合规审计和运营分析。这样做可以快速检测恶意行为，并在任何潜在损害发生之前加以阻止。

EventBridge API 目标集成

API 目标是可以作为 EventBridge 目标调用的 HTTP 端点。在与 Confluent 集成的情况下，这涉及利用 Confluent 的 Produce REST API 将数据从 EventBridge 无缝传输到 Confluent。这使客户能够利用 Confluent 丰富的工具集实时流式传输、处理和分析数据。

解决方案概述

在此场景中，AWS CloudTrail 事件被路由到 EventBridge，EventBridge 再将其转发到托管在 AWS 上的公共 Confluent Cloud 集群。随后，利用 Confluent Cloud for Apache Flink ® 实时检测任何异常活动。检测到的异常将被发送到 Confluent Cloud 中的“flagged”主题。在实际应用中，专用消费者将实时对这些事件做出及时响应。

先决条件

已创建“audit_trail_topic”的 Confluent Cloud 集群
具有集群和主题访问权限的 Confluent Cloud API 密钥
AWS 命令行界面（AWS CLI）
已启用日志记录的 CloudTrail

在 EventBridge 中捕获 CloudTrail 只读事件

要捕获 CloudTrail 管理事件，我们需要在启用 CloudTrail 的同一区域中在 EventBridge 中创建规则。创建时还应使用定义我们想要捕获的确切事件的模式。以下使用 AWS CLI 创建该规则：

'''
aws events put-rule \
--name "ruleForManagementEvents" \
--region\
--event-bus-name "default" \
--state "ENABLED_WITH_ALL_CLOUDTRAIL_MANAGEMENT_EVENTS" \
--event-pattern '{
"detail-type": ["AWS API Call via CloudTrail"],
"detail": {
"userIdentity": {
"type": [{"anything-but":"AWSService"}]
},
"eventSource": [{
"anything-but": ["sts.amazonaws.com","ssm.amazonaws.com"]
}]
}
}' \
--role-arn'''

上述命令创建了一个 EventBridge 规则，用于捕获通过 CloudTrail 由 AWS API 调用生成的管理事件，同时过滤掉用户身份为 AWS 服务且事件源为“sts.amazonaws.com”或“ssm.amazonaws.com”的事件。您可以根据自己的用例更改事件模式。

将 CloudTrail 事件发送到 Confluent Cloud

EventBridge 使用 API 目标将数据发送到 Confluent Cloud。此方法利用 Confluent Cloud 的全托管 Produce API。因此，初始步骤涉及创建 API 目标并在 EventBridge 中为 Confluent Cloud 建立连接。连接是一个逻辑容器，用于保存目标的授权详细信息。

创建 API 目标及到 Confluent Cloud 的连接

在 EventBridge API 目标控制台中，点击“Create API destination”。
按如下方式填写 API 目标详细信息：API 目标端点：Confluent REST 端点和路径应如下所示：
在“Connection type”部分，点击 Create a new connection 并填写连接详细信息。在“Destination type”部分，选择 Partners 并选择“Confluent”。
输入用户名（Confluent Cloud 集群 API 密钥）和密码（Confluent Cloud 集群 API 密钥）。
点击 Create 以创建 API 目标及到 Confluent Cloud 的连接。

将 API 目标添加为 EventBridge 的目标

接下来，我们还需要将 API 目标添加为之前创建的规则的目标，以便在规则匹配过滤后的事件时调用 API 目标。我们使用 AWS CLI 将 Confluent Cloud API 目标添加为步骤 1 中创建的规则的目标。

'''
aws events put-targets \
    --rule "ruleForManagementEvents" \
    --event-bus-name "default" \
    --region\
    --targets '[{
        "Id": "cwcapidestination",
        "Arn":,
        "RoleArn":,
        "InputTransformer": {
            "InputPathsMap": {
                "event": "$.detail"
            },
            "InputTemplate": "{\"value\":{\"type\":\"JSON\",\"data\":}}"
        }
    }]'
'''

在此命令中，InputTransformer 指定了在将事件数据发送到目标之前应如何转换事件数据。

"InputPathsMap"：指定输入路径的映射。在此情况下，它将 "event" 键映射到整个事件详情。
"InputTemplate"：指定用于转换事件数据的模板。它将事件数据包装为一个带有键 "value" 的 JSON 对象，并将事件数据嵌入其中。然后，Confluent Produce API 将使用此值来发送数据。

成功运行此命令后，EventBridge 即配置为将事件从 CloudTrail 发送到 Confluent Cloud。

在 Confluent Cloud 中检查事件

在 Confluent Cloud 控制台中，转到主题页面。
选择 Confluent Cloud 环境和集群，然后单击 Continue。
选择 audit_trail_topic，并确保它正在接收 CloudTrail 事件。

实时审计日志分析

现在，我们可以使用 Confluent Cloud for Apache Flink 实时处理、分析和转换审计日志，使我们能够实时采取行动并获得洞察。

首先，我们需要创建一个 Flink 计算池。在 Flink 控制台中，选择环境并单击“Continue”。然后点击“Create compute pool”；我们选择与当前工作相同的区域，并为该池命名。

编写 Flink 语句

一个常见的用例是检测 API 调用中的异常。假设我们是安全团队的一员，我们的目标是在任何 1 分钟窗口内，标记同一主体调用特定 S3 API 超过 10 次的情况。Flink 语句将如下所示：

'''
SELECT
  window_start,
  window_end,
  JSON_VALUE(CAST(val AS STRING), '$.eventName') AS eventName,
  JSON_VALUE(CAST(val AS STRING), '$.eventSource') AS AWSService,
  JSON_VALUE(CAST(val AS STRING), '$.userIdentity.principalId') AS userIdentity,
  COUNT(*) AS eventCount
FROM TABLE(
  TUMBLE(TABLE eventbridge, DESCRIPTOR($rowtime), INTERVAL '1' MINUTE)
)
WHERE JSON_VALUE(CAST(val AS STRING), '$.eventSource') = 's3.amazonaws.com'
GROUP BY window_start, window_end, JSON_VALUE(CAST(val AS STRING), '$.eventName'), JSON_VALUE(CAST(val AS STRING), '$.eventSource'), JSON_VALUE(CAST(val AS STRING), '$.userIdentity.principalId')
HAVING COUNT(*) > 10;
'''

要测试这一点，请使用 AWS CLI 运行 '''aws s3 ls''' 15 次。这样做将在 Flink 语句输出中产生以下行：

部署 Flink 语句

当我们对 Flink SQL 语句满意后，可以将其包装起来，使其开始将标记的活动发送到目标主题。

回到 Workspaces 控制台，添加一个新表：''' CREATE TABLE target_table ( window_start TIMESTAMP, window_end TIMESTAMP, eventName STRING, AWSService STRING, userIdentity STRING, eventCount BIGINT ); ''' 此语句创建一个名为 "flagged_topic" 的新主题，并在 Schema Registry 中为该主题注册一个 schema。
现在编辑正在运行的语句，将数据插入到刚刚创建的主题中：''' INSERT INTO flagged_topic (window_start, window_end, eventName, AWSService, userIdentity, eventCount)

无代码字段掩码

请注意，某些用户信息以明文显示。虽然这些信息对下游消费者采取适当行动可能很有价值，但我们将选择对 "userIdentity" 字段进行掩码，以演示我们的新 Topic Actions 功能。Topic Actions 提供预配置的流处理语句，旨在解决常见的、与领域无关的任务。

在 Data Portal 中，选择环境、"flagged_topic" 主题，然后点击 Actions 按钮。
目前有两个操作：Deduplicate topic 和 Mask fields。选择 Mask fields。
在 Settings 中，输入 userIdentity 字段，并将除最后两个字符外的所有内容用 '*' 掩码。您可以填写其余部分并点击 Confirm。
userIdentity 字段现在已被掩码，可以在 final_flagged_topic 中查看。

将输出数据连接到其他系统

现在输出已存储在 Confluent 主题中，有几种选项可以将消费者连接起来或将数据下沉到 AWS 上首选的数据存储中。一种选项是将消费者直接连接到输出主题。此外，Confluent 提供超过 120 个预构建连接器，可用于将数据下沉到 AWS 上的各种数据存储中。

（可选）将数据发送回 EventBridge

另一种可能性是使用 EventBridge Sink 连接器或 EventBridge Pipes 将数据下沉回 EventBridge 或 AWS 服务，例如 Step Functions。EventBridge Sink 连接器可以将数据发送到任何已配置的事件总线，支持 Confluent Cloud Schema Registry。另一方面，EventBridge Pipes 提供不同源与 15 个不同 AWS 目标之间的全托管点对点集成。

在撰写本文时，EventBridge Pipes 不支持 Confluent Schema Registry。因此，使用 Avro schema 序列化的事件可以被转发，但不会在 EventBridge Pipes 上进行过滤。然而，我们可以使用 AWS Lambda 函数作为 Pipe 中的增强步骤，在下游应用程序发送之前反序列化事件，如下所述。

要使用 Confluent Cloud 设置 EventBridge Sink 连接器，请遵循 Confluent Hub 和 GitHub 仓库中列出的步骤。另一方面，要配置 EventBridge Pipes 以从 Confluent 接收数据，请执行以下步骤：

使用 AWS CLI 在 Secrets Manager 中创建一个新的 Confluent 密钥。''' aws secretsmanager create-secret \ --name confluent_cloud \ --secret-string '{"username":
在 EventBridge Pipes 控制台中，从源列表中选择 Confluent。
填写引导服务器、主题名称，即我们在上一步中创建的密钥。
在 Filtering 页面中，我们可以选择在将任何事件发送到目标之前将其过滤掉。但是，由于事件使用 Avro 格式序列化，目前无法使用 Pipe 过滤。因此，过滤功能仅限于具有 JSON 编码或纯字符串事件的主题。您可以在过滤页面中保留默认值，然后点击 Next。
在 Enrichment 步骤中，我们可以触发 Lambda 函数进行事件反序列化，然后再将其路由回 Pipes。如果留空，数据在发送到目标目的地时将保持编码状态。您可以将其留空并点击 Next。
您可以选择 EventBridge Pipes 支持的其中一个目标，然后点击“Create pipe”。

将 EventBridge API 目标与 Confluent 结合使用的注意事项

API 目标使用 Confluent Produce API 将数据发送到 Confluent。因此，此集成有一些需要考虑的限制：

该集成不支持批处理，这可能会限制吞吐量
API 目标和 Confluent Produce API 不支持 Confluent Schema Registry，因此，该集成只会将无模式数据写入 Confluent。
EventBridge API 目标仅支持公共端点，因此仅支持公共 Confluent Cloud 集群。对于私有 Confluent Cloud 集群，可以使用 Lambda 函数从 EventBridge 读取数据并将其写入 Confluent，如下所示。
REST API 直接向 Confluent Cloud 集群生成记录的最大吞吐量为 10 MBps。如需更高吞吐量，应使用其他方法。同样，每个 API 目标端点有每秒 300 次调用的软限制，以保护下游目标。此限制可以在 AWS 配额控制台中进行调整。

结论

在本文中，我们探讨了 EventBridge 与 Confluent Cloud 的集成，促进了数据从各种 AWS 服务无缝流向 Confluent Cloud，以进行实时处理和分析。通过实际演示，我们展示了如何实时摄取和分析来自 CloudTrail 的审计事件，使组织能够快速检测和响应异常，从而加强其安全性和合规性措施。

尽管此集成带来了诸多好处，但需要注意的是，它目前仅支持 REST。话虽如此，我们将继续与 EventBridge 团队合作，以增强集成并引入对 Kafka 原生生产者等其他机制的支持。

后续步骤

还不是 Confluent 客户？在 AWS Marketplace 中试用 Confluent Cloud。新注册用户可获得 400 美元的免费额度，可在前 30 天内使用。通过 AWS Marketplace 订阅 Confluent 后，您的额度将立即显示在您的 Confluent 账户中。

转载声明：本文为转载整理，原文版权归原作者及原站点所有。

原文来源：Confluent Blog

原创地址：https://www.confluent.io/blog/analyze-aws-audit-logs-in-real-time/

ITPub博客

如何实时分析AWS审计日志