在大型组织中，Confluent Cloud 通常会被许多不同用户以及业务关键型应用程序同时访问，可能跨越不同的业务线。如此多的独立组件协同工作，单个故障、错误或事件影响其他服务的风险随之增加。一次事件可能由用户点击了错误的按钮、应用程序配置错误，或仅仅是一个 bug 引起——不一而足。理想情况下，你希望系统和应用程序始终平稳运行，但如果发生事件，关键是通过将解决时间降至最低来减少损害。

在大多数情况下，你会观察到灾难（例如，UI 不可用），但不会直接知道根本原因。然后你可能会（也许在夜间）呼叫工程师或支持人员来检测和排查错误。然而，更好的方式是，一旦可能的事件发生，就主动通知负责的平台或应用程序团队，并为他们提供立即修复所需的额外信息。这正是审计日志发挥作用的地方。

审计日志提供了一种自动跟踪 Confluent Cloud 安装中发生的所有不同交互的方式。它们会为许多事件发出，例如成功授权更改 Apache Kafka ® 主题配置、创建连接器或角色绑定关联。本质上，它们可用于分析谁在你的 Confluent Cloud 安装中做了什么。所有审计日志的完整概览可在 Confluent Cloud 文档 中找到。

在故障排查对话中，我们经常发现自己在问：“审计日志说了什么？”——尤其是在事件发生之后，例如应用程序无法再向 Confluent Cloud 进行身份验证，或某个资源在无人察觉的情况下被删除。因此，我们决定撰写这篇博客，提供一些主动使用 Confluent Cloud 审计日志的最佳实践，例如通过发送通知来检测根本原因，以便尽可能快速地解决事件。

本文提供了一份完整的概念指南，用于开发你自己的审计日志告警服务。从技术上讲，它将引导你创建一个全托管管道，该管道可以：

• 使用 Cluster Linking 将审计日志从外部 Confluent Cloud 集群传输到你自己的 Confluent Cloud 集群

• 通过 Kafka Connect 将数据发送到 Splunk

• 在 Splunk 中针对某些审计日志事件定义告警

• 将告警发送到你指定的电子邮件地址，以便在可能存在问题时通知你

我们重点关注将 Splunk 与 Confluent Cloud 集成所需的各项配置，以及你应该使用哪些 Splunk 搜索和告警参数来主动发送 Confluent Cloud 审计日志事件。（安全方面、性能调优和成本仅作为附注提及。）

你可以在下面的视频中看到最终工作流程：

请注意，本指南是 Johnny Mirza 所著《How to Visualise Confluent Cloud Audit Log Data》博客的扩展。我们建议您也阅读该博客，尤其是其中关于额外 Splunk 仪表板的内容。

设置 Splunk Cloud Platform

在本博客的演示中，我们将使用 Splunk Cloud Platform。根据 Splunk 文档，您可以获得 14 天免费试用，每天允许插入 5 GB 数据，这对我们的设置来说已经足够。

注册后，您将收到一封电子邮件，其中包含您的 Splunk 实例 URL（https://

接下来，在 Splunk 中进入“Settings” → “Data inputs”，添加一个名为“Confluent Audit Logs”的新“HTTP Event Collector”，使用默认配置。稍后部署全托管的 Splunk Sink Connector 时，您将需要该收集器的 token。

我们建议通过 Terraform 创建资源。截至目前，Splunk Cloud Platform 的 Terraform Provider 不支持 HTTP Event Collector、仪表板或告警。因此，我们将通过 UI 创建所有资源。不过，如果您使用的是 Splunk Enterprise，则所有这些资源都应可通过 Splunk Enterprise 的 Terraform Provider 使用。

将审计日志从外部审计日志集群传输到 Splunk

根据 Confluent 文档，“Confluent Cloud 审计日志中的审计日志记录会在独立的 Confluent Cloud 集群上保留七天。”因此，作为最佳实践，我们将首先使用 Cluster Linking 将审计日志传输到我们的集群。将审计日志存储在我们的集群中，使我们能够配置自定义保留时间、处理日志（例如，按特定事件进行过滤），并通过全托管连接器将日志发送到 Splunk 等外部系统。

文档《Use Cluster Linking to Manage Audit Logs on Confluent Cloud》提供了设置连接的出色分步指南。简而言之，它说明了如何从外部集群检索所需信息，以及如何设置集群链接以镜像 confluent-audit-log-events topic。请注意，使用 Cluster Linking 通常需要拥有 Dedicated Confluent Cloud 集群。（不过，在本博客后文中，我们将分享针对此场景及其他挑战性场景的可能变通方案。）

一旦审计事件开始生产到您的 Confluent Cloud 集群，您就需要部署全托管的 Splunk Sink Connector。对于身份验证配置，您应从 Splunk 注册电子邮件中获取 HEC URI，并从 HTTP Event Collector 获取 HEC Token。您可以使用 HEC SSL Validate Certificates 来启用或禁用 HTTPS 证书验证，这对于连接器与 Splunk 实例之间的安全连接非常重要。如果您希望启用 SSL，则需要上传 Splunk HEC SSL Trust Store 文件（包含验证 SSL 连接所需的证书）以及 Splunk HEC SSL Trust Store Password。如需更多信息，我们建议查看相应的 Splunk 文档。在演示中，为简单起见，我们将把 HEC SSL Validate Certificates 设置为“false”。

完整配置如下所示（尽管大多数配置只是默认值）：

{
  "topics": "confluent-audit-log-events",
  "schema.context.name": "default",
  "input.data.format": "JSON",
  "connector.class": "SplunkSink",
  "name": "",
  "kafka.auth.mode": "KAFKA_API_KEY",
  "kafka.api.key": "",
  "kafka.api.secret": "***",
  "splunk.hec.uri": "https://abc.splunkcloud.com:8088",
  "splunk.hec.token": "***",
  "splunk.hec.ssl.validate.certs": "false",
  "splunk.indexes": "default",
  "splunk.hec.raw": "false",
  "splunk.hec.track.data": "false",
  "splunk.hec.http.keepalive": "true",
  "splunk.hec.max.http.connection.per.channel": "2",
  "splunk.hec.total.channels": "2",
  "splunk.hec.socket.timeout": "10",
  "splunk.hec.use.record.timestamp": "true",
  "splunk.hec.threads": "1",
  "splunk.hec.max.outstanding.events": "10000",
  "splunk.hec.max.retries": "-1",
  "splunk.hec.backoff.threshhold.seconds": "60",
  "splunk.hec.json.event.formatted": "false",
  "splunk.hec.max.batch.size": "500",
  "splunk.hec.lb.poll.interval": "120",
  "splunk.flush.window": "30",
  "splunk.hec.ack.enabled": "false",
  "splunk.header.support": "false",
  "max.poll.interval.ms": "300000",
  "max.poll.records": "500",
  "tasks.max": "1"
}

变通方案

在某些情况下，无法遵循最佳实践并使用 Cluster Linking 和全托管的 Splunk Sink Connector 将数据从外部集群传输到 Splunk。在下表中，我们分享了针对这些不同场景的一些变通方法：

| 管道部分 | 场景 | 变通方法 |
| --- | --- | --- |
| 将审计日志从外部集群传输到您的集群 | • 您没有 Dedicated 集群 • 由于网络限制，您无法使用 Cluster Linking（另请参阅 Manage Private Networking for Cluster Linking on Confluent Cloud） | Replicator |
| 将审计日志从您的集群传输到 Splunk | • 您无法使用全托管连接器（例如，由于网络限制） | 自管理 Splunk Sink Connector |
| 将审计日志直接传输到 Splunk | • 您需要将审计日志从外部集群直接传输到 Splunk（出于上述任何原因） | 带有自定义配置的自管理 Splunk Sink Connector（请参阅 How to Visualise Confluent Cloud Audit Log Data） |

Splunk 中的告警

您可能想先观看 Splunk 提供的关于设置告警的通用视频教程：Creating Alerts in Splunk Enterprise。

定义 Confluent Cloud 集群删除告警

接下来，我们提供一个分步示例，用于设置当 Confluent Cloud 集群被删除时触发的告警。从业务角度来看，此告警并不一定意味着正在发生可疑事件，因为集群可能因测试目的而被创建和删除，但是，如果被删除的集群是生产集群，则可能至关重要。

在 Splunk 中，转到“Search & Reporting”，您可以在其中查询数据。您将在“Confluent Audit Logs”源（我们为 HTTP Event Collector 设置的名称）下找到您的数据，方法名称为“DeleteKafkaCluster”。

接下来，在“Save As”下，定义告警（所有示例设置请参见下图）。遗憾的是，默认情况下，Splunk Cloud 中禁用了实时告警，因此我们需要使用计划告警。基本上，如果删除了一个或多个 Confluent Cloud 集群，则每小时将触发一次告警。我们选择发送电子邮件，但您也可以配置 webhook 将消息发送到您首选的系统，例如 Slack 或 Microsoft Teams。

关于搜索和告警，请注意 Splunk 文档中的以下内容：“在 Victoria Experience 上的 Splunk Enterprise 和 Splunk Cloud Platform 中，实时搜索默认开启。在 Classic Experience 上的 Splunk Cloud Platform 中，您必须提交支持工单才能开启实时搜索。”

就是这样！您可以在 Splunk 的“Alerts”下检查和编辑已定义的告警。因此，现在如果 Confluent Cloud 集群被删除，您就有了相应的告警定义，因此您将自动收到通知。

尝试通过在 Confluent Cloud 中删除一个集群来触发告警（但不要删除用于此管道的集群！）。在下一个整点（由于所需的计划任务限制），您将收到一封标题为“CCloud Cluster deletion”的告警电子邮件。

该电子邮件还将包含原始审计日志事件，以便您可以进一步分析哪个集群在何时被何人删除。

如果你只想在某些集群被删除时收到通知（例如生产环境），可以修改你的 Splunk 搜索，添加：(data.cloudResources{}.resource.resourceId="

本质上，上述步骤可以针对你想要定义的每一个其他告警进行修改和重复。确保你通过 Splunk 搜索获得正确的结果（一个好的起点是扫描你想要设置告警的对应审计日志事件的字段），并且你定义了正确的触发条件。

常用告警

以下常用告警表基于与同事和客户的讨论。由于客户有不同的需求，且 Confluent Cloud 频繁发布新功能，此表并非详尽无遗。如果你希望添加其他告警，我们鼓励你在相应的 GitHub 仓库中协作，或通过 LinkedIn 直接联系我。

| 告警 | 审计日志 | Splunk 搜索 |
| --- | --- | --- |
| CC 集群删除 | DeleteKafkaCluster | source="http:Confluent Audit Logs" data.methodName="DeleteKafkaCluster" |
| 环境删除 | DeleteEnvironment | source="http:Confluent Audit Logs" data.methodName="DeleteEnvironment" |
| CC 集群认证失败 | kafka.Authentication | source="http:Confluent Audit Logs" data.methodName="kafka.Authentication" data.result.status="UNAUTHENTICATED" |
| Schema Registry 认证失败 | schema-registry.Authentication | source="http:Confluent Audit Logs" data.methodName="schema-registry.Authentication" data.authenticationInfo.result="UNAUTHENTICATED" |
| CC 集群资源授权失败 | io.confluent.kafka.server/authorization | source="http:Confluent Audit Logs" type="io.confluent.kafka.server/authorization" data.authorizationInfo.granted="false" |
| 基于 IP 过滤器的授权失败 | ip-filter.Authorize | source="http:Confluent Audit Logs" data.methodName="ip-filter.Authorize" data.authorizationInfo.result="DENY" |
| OrganizationAdmin 角色绑定关联 | BindRoleForPrincipal | source="http:Confluent Audit Logs" data.methodName="BindRoleForPrincipal" data.request.data.role_name="OrganizationAdmin" |

检查管道的健康状况

由于你的管道涉及多个组件，我们建议监控其健康状况：

• 对于 Cluster Linking，你可以使用 Confluent Cloud Metrics API 检查集群链接数量和镜像主题偏移滞后。

• 全托管的 Confluent Cloud 连接器在任务或连接器失败时会发出 Connector 事件。我们建议将这些通知（连接器处于“FAILED”状态）通过电子邮件发送给自己，或发送到你偏好的系统，如 Slack 或 Microsoft Teams。更多信息请参阅 Notifications for Confluent Cloud 。

• 你也可以考虑在 Splunk 中添加一个健康指标告警，检查在给定时间间隔内是否至少有一个新的审计日志事件产生到 Splunk。相应的 Splunk 搜索就是 source="http:Confluent Audit Logs" ，告警的时间间隔取决于你的个人需求，例如“在 x 小时/天内小于 0”。

最终思考

我们成功创建了一个管道，当 Confluent Cloud 安装中发生某些事件时，它会基于审计日志主动通知我们。

我们希望它能帮助你用“我主动知道我的 Confluent Cloud 安装中何时发生可疑情况，并且我立即拥有正确的数据来进一步分析它”取代“让我分析我的审计日志看看发生了什么”。

你可以在这个 GitHub 仓库 中找到演示的代码。

• Patrick 是 Confluent 的客户成功技术架构师，致力于通过创新性地使用 Apache Kafka 和流数据处理技术为客户推动价值和成功。凭借在数据科学和软件开发方面的基础，他为每个项目带来了对数据架构和分析的全面理解。值得一提的是，Patrick 曾在 Kafka Summit 上作为演讲者分享了他的专业知识，并撰写了专注于流数据架构的博客文章

  
  

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

原文来源：Confluent Blog

原创地址：https://www.confluent.io/blog/alerting-pipeline-confluent-cloud-audit-logs/