集成 Datadog 监控

AutoMQ 支持将指标导出到 Datadog 进行监控和告警。本文按照 Collector 的运行环境介绍两种 Gateway 模式集成方式：

Kubernetes：使用 Datadog Agent Helm chart 的 OTEL Agent Gateway，通过 Kubernetes 服务发现抓取 AutoMQ Pod 的 Prometheus 指标。
Linux：使用运行在 Linux 上的 Gateway OTEL Collector 或 Datadog DDOT Collector 抓取 AutoMQ 的 Prometheus metrics endpoint，并转发到 Datadog。

前提条件

已部署的 AutoMQ Cloud BYOC 集群。BYOC 集群默认会暴露 AutoMQ Prometheus metrics endpoint。
有效的 Datadog API Key
Datadog Site URL（例如 datadoghq.com、us3.datadoghq.com、datadoghq.eu）

方式一：Kubernetes 环境使用 Datadog Agent Gateway

该方式下，Datadog Agent Helm chart 会运行一个 OTEL Agent Gateway Deployment。Gateway 通过 Kubernetes 服务发现抓取 AutoMQ Pod 的 Prometheus 指标，在写入 Datadog 前降低指标基数，然后导出到 Datadog。

AutoMQ Pods（在 9090 端口暴露 Prometheus 指标）
    ↓  Prometheus 抓取（Kubernetes 服务发现）
Datadog OTEL Agent Gateway
    ↓  导出到 Datadog
Datadog

步骤 1：确认 AutoMQ 指标导出

通过 AutoMQ Cloud BYOC Console 创建的 Kubernetes 集群默认会在 AutoMQ broker 和 controller Pod 上暴露 Prometheus 指标。标准 BYOC 流程不需要手动修改 AutoMQ Helm values。如果你管理的是 Kubernetes AutoMQ Software 部署，并且需要手动启用指标导出，请参考附录：为 Kubernetes Software 手动启用 AutoMQ 指标导出。

步骤 2：创建 Datadog Agent Gateway values 文件

创建 datadog-values.yaml 文件，配置 OTEL Agent Gateway。Gateway 使用 Kubernetes 服务发现来查找 AutoMQ broker 和 controller Pod。请将 <your-automq-namespace> 替换为 AutoMQ 集群所在的 Kubernetes namespace。如果需要让同一个 Datadog Agent 采集多个 AutoMQ namespace，请将每个 namespace 添加到 kubernetes_sd_configs.namespaces.names。同一套采集 profile 可同时用于 AutoMQ Software 和 AutoMQ Cloud BYOC/CMP 集群。标准大盘显示 cluster_id 作为必需的集群过滤条件；该 dashboard filter 读取的是由 AutoMQ Prometheus job label 派生出的 Datadog service tag。大盘也会在 AutoMQ 指标暴露 env_id 和 instance_id 时提供对应的可选过滤条件。该 profile 不会从 Kubernetes metadata 合成 env_id 或 instance_id。如果 AutoMQ 指标自身已经带有这些标签，honor_labels: true 会保留 AutoMQ 指标中的值。

agents:
  enabled: false
clusterAgent:
  enabled: false
  admissionController:
    enabled: false
datadog:
  kubeStateMetricsCore:
    enabled: false
  orchestratorExplorer:
    enabled: false
  operator:
    enabled: false
  instrumentationCrd:
    enabled: false
  autoscaling:
    workload:
      enabled: false
datadog-crds:
  crds:
    datadogMetrics: false
    datadogPodAutoscalers: false
    datadogPodAutoscalerClusterProfiles: false
    datadogInstrumentations: false
otelAgentGateway:
  enabled: true
  replicas: 1
  rbac:
    create: true
    rules:
      - apiGroups: [""]
        resources: ["pods"]
        verbs: ["get", "list", "watch"]
  config: |
    receivers:
      prometheus:
        config:
          scrape_configs:
            - job_name: "automq"
              scrape_interval: 15s
              honor_labels: true
              kubernetes_sd_configs:
                - role: pod
                  namespaces:
                    names:
                      - <your-automq-namespace>
                  selectors:
                    - role: pod
                      label: 'app.kubernetes.io/name=automq-enterprise,app.kubernetes.io/component in (broker,controller)'
                      field: 'status.phase=Running'
              relabel_configs:
                - source_labels: [__meta_kubernetes_pod_container_name]
                  action: keep
                  regex: 'automq-(broker|controller)'
                - source_labels: [__meta_kubernetes_pod_container_port_number]
                  action: keep
                  regex: "9090"
              metric_relabel_configs:
                - source_labels: [__name__]
                  action: keep
                  regex: "^(consumer_(offset_lag_count|estimated_time_lag_milliseconds)|jvm_cpu_(count|time_seconds_total)|kafka_(node(_metadata|_info)?|broker_(active_count|fenced_count|network_io_bytes_total)|group_(count|stable_count|empty_count|dead_count|preparing_rebalance_count|completing_rebalance_count|commit_offset)|log_(size|end_offset)|message_count_total|network_io_bytes_total|partition_total_count|request_error_count_total|stream_topic_partition_count|tabletopic_(delay_milliseconds|fps_fields_per_second)|topic_(count|request_count_total)))$"
                - action: labeldrop
                  regex: "^(active_controller|customer|exported_env_id|exported_instance_id|exported_job|instance|link_id|operation_name|operation_type|otel_scope_name|pool_type|request|request_type|size|source_node_id)$"
    exporters:
      datadog:
        api:
          key: ${env:DD_API_KEY}
          site: ${env:DD_SITE}
    processors:
      cumulativetodelta/automq_counters:
        initial_value: auto
        include:
          match_type: strict
          metrics:
            - kafka_network_io_bytes_total
            - kafka_message_count_total
            - kafka_topic_request_count_total
            - kafka_broker_network_io_bytes_total
            - kafka_request_error_count_total
      transform/automq_rollups:
        error_mode: ignore
        metric_statements:
          - context: metric
            statements:
              - copy_metric(name="automq_topic_log_size") where metric.name == "kafka_log_size"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic"]) where metric.name == "automq_topic_log_size"
              - copy_metric(name="automq_topic_log_end_offset") where metric.name == "kafka_log_end_offset"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic"]) where metric.name == "automq_topic_log_end_offset"
              - copy_metric(name="automq_topic_network_io_bytes_total") where metric.name == "kafka_network_io_bytes_total"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic", "direction"]) where metric.name == "automq_topic_network_io_bytes_total"
              - copy_metric(name="automq_topic_message_count_total") where metric.name == "kafka_message_count_total"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic", "direction"]) where metric.name == "automq_topic_message_count_total"
              - copy_metric(name="automq_topic_request_count_total") where metric.name == "kafka_topic_request_count_total"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic", "type"]) where metric.name == "automq_topic_request_count_total"
              - copy_metric(name="automq_cluster_broker_network_io_bytes_total") where metric.name == "kafka_broker_network_io_bytes_total"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "direction"]) where metric.name == "automq_cluster_broker_network_io_bytes_total"
              - copy_metric(name="automq_request_error_count_total") where metric.name == "kafka_request_error_count_total"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "error"]) where metric.name == "automq_request_error_count_total"
              - copy_metric(name="automq_topic_partition_count") where metric.name == "kafka_stream_topic_partition_count"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic"]) where metric.name == "automq_topic_partition_count"
              - copy_metric(name="automq_topic_rack_partition_count") where metric.name == "kafka_stream_topic_partition_count"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic", "rack"]) where metric.name == "automq_topic_rack_partition_count"
              - copy_metric(name="automq_group_count") where metric.name == "kafka_group_count"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_count"
              - copy_metric(name="automq_group_stable_count") where metric.name == "kafka_group_stable_count"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_stable_count"
              - copy_metric(name="automq_group_empty_count") where metric.name == "kafka_group_empty_count"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_empty_count"
              - copy_metric(name="automq_group_dead_count") where metric.name == "kafka_group_dead_count"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_dead_count"
              - copy_metric(name="automq_group_preparing_rebalance_count") where metric.name == "kafka_group_preparing_rebalance_count"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_preparing_rebalance_count"
              - copy_metric(name="automq_group_completing_rebalance_count") where metric.name == "kafka_group_completing_rebalance_count"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_completing_rebalance_count"
              - copy_metric(name="automq_consumer_group_topic_commit_offset") where metric.name == "kafka_group_commit_offset"
              - copy_metric(name="automq_group_topic_offset_lag") where metric.name == "consumer_offset_lag_count"
              - copy_metric(name="automq_group_topic_estimated_time_lag_milliseconds") where metric.name == "consumer_estimated_time_lag_milliseconds"
              - aggregate_on_attributes("max", ["env_id", "instance_id", "service", "group", "topic"]) where metric.name == "automq_group_topic_estimated_time_lag_milliseconds"
          - context: datapoint
            statements:
              - set(datapoint.attributes["group"], datapoint.attributes["consumer_group"]) where metric.name == "automq_consumer_group_topic_commit_offset" and datapoint.attributes["consumer_group"] != nil
              - set(datapoint.attributes["consumer_group"], datapoint.attributes["group"]) where metric.name == "automq_group_topic_offset_lag" and datapoint.attributes["group"] != nil
          - context: metric
            statements:
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "consumer_group", "group", "topic"]) where metric.name == "automq_consumer_group_topic_commit_offset"
              - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "consumer_group", "group", "topic"]) where metric.name == "automq_group_topic_offset_lag"
      filter/drop_raw_replaced_metrics:
        metrics:
          metric:
            - 'name == "kafka_log_size"'
            - 'name == "kafka_log_end_offset"'
            - 'name == "kafka_network_io_bytes_total"'
            - 'name == "kafka_message_count_total"'
            - 'name == "kafka_broker_network_io_bytes_total"'
            - 'name == "kafka_topic_request_count_total"'
            - 'name == "kafka_request_error_count_total"'
            - 'name == "kafka_stream_topic_partition_count"'
            - 'name == "kafka_group_count"'
            - 'name == "kafka_group_stable_count"'
            - 'name == "kafka_group_empty_count"'
            - 'name == "kafka_group_dead_count"'
            - 'name == "kafka_group_preparing_rebalance_count"'
            - 'name == "kafka_group_completing_rebalance_count"'
            - 'name == "kafka_group_commit_offset"'
            - 'name == "consumer_offset_lag_count"'
            - 'name == "consumer_estimated_time_lag_milliseconds"'
      infraattributes:
        cardinality: 2
    service:
      pipelines:
        metrics:
          receivers: [prometheus]
          processors: [cumulativetodelta/automq_counters, transform/automq_rollups, filter/drop_raw_replaced_metrics, infraattributes]
          exporters: [datadog]

本示例使用以下组件：

组件	用途
`otelAgentGateway`	为该采集 profile 运行一个 OTEL Agent Gateway Deployment。单个 Gateway 负责 scrape、rollup，并将指标导出到 Datadog。
`prometheus` receiver	通过 Kubernetes 服务发现找到指定 namespace 中正在运行的 AutoMQ broker 和 controller Pod，并抓取 9090 端口。`honor_labels: true` 会保留 AutoMQ 指标自带的 Prometheus `job`、`env_id`、`instance_id` 等标签。container-name relabel 规则只保留 AutoMQ broker/controller 容器，并排除 Schema Registry 等 sidecar。
`metric_relabel_configs`	只保留标准最小化大盘或 `automq_*` rollup processor 必需的原始指标。该 profile 不会把 `job` 复制为 `cluster_id`；Datadog 会将 Prometheus `job` label 映射为 `service` tag，dashboard 仍保留 `cluster_id` 过滤器名称，但其取值来自 `service`。
`cumulativetodelta/automq_counters` processor	在 rollup 前将选定的 Prometheus cumulative counter 转换为 delta，避免先把多个 AutoMQ 节点的 cumulative counter 聚合后再计算 rate。
`transform/automq_rollups` processor	在 Datadog ingest 之前创建低基数 `automq_*` 聚合指标。该 profile 使用 Collector 侧 OTTL 聚合。
`filter/drop_raw_replaced_metrics` processor	在聚合指标创建完成后丢弃已被 rollup 替代的原始源指标，从而降低 Datadog custom metrics 数量，同时保持面向客户的大盘面板可用。
`datadog` exporter	将指标发送到 Datadog。API key 和 site 从 Agent 的环境变量中读取。
`infraattributes` processor	可选。在相关资源属性可用时，为遥测数据添加 Datadog 基础设施标签。

建议将 OTEL Agent Gateway 调度到通用节点或独立的观测节点，不需要与 AutoMQ broker/controller 运行在同一节点。若你的 Kubernetes 集群需要自定义 Gateway 的调度策略、资源规格、Pod annotation 或 label，请在 Datadog Helm values 中配置 otelAgentGateway.nodeSelector、otelAgentGateway.affinity、otelAgentGateway.tolerations、otelAgentGateway.podAnnotations、otelAgentGateway.additionalLabels、otelAgentGateway.containers.otelAgent.resources 等参数。可参考 Datadog DDOT Collector Gateway Helm 安装文档和 Datadog Helm chart values.yaml。

标准大盘采集 profile

推荐配置面向 AutoMQ 客户标准 Datadog 大盘，只保留这些大盘所需的最小维度：

指标范围	保留维度
集群健康和容量	`service`
CPU 和 broker metadata	`service`、`broker_id`
请求、错误、延迟和 broker 流量	`service`、`topic`
Topic 吞吐、消息数、容量和 offset	`service`、`topic`
Topic partition 在 rack 上的分布	`service`、`topic`、`rack`
Consumer group lag、offset 和吞吐	`service`、`consumer_group` 或 `group`、`topic`

标准最小化 profile 不保留原始 partition 级序列，不保留 topic-by-broker traffic / offset / size 序列，不保留 broker/node 级 JVM、connection、request、latency 或 traffic drilldown 序列，不保留 consumer_group * topic * broker 序列，也不保留未被标准大盘使用的 kafka_stream_* 内部指标、Kafka Linking 指标或 Detailed Metrics 独有的内部诊断维度。

该示例是面向 AutoMQ 客户标准最小化大盘的推荐采集配置。它只采集标准大盘或 rollup processor 必需的原始指标，将 request、traffic、topic、partition、consumer group 源指标转换为低基数 automq_* 聚合指标，并在聚合后丢弃原始源指标，以降低 Datadog custom metrics 数量和标签基数。标准最小化 profile 保留 cluster、topic、consumer group、rack 维度，以及 CPU 和 broker metadata，但不保留 partition 级序列、topic-by-broker 序列、broker/node 级 drilldown 序列，也不保留 consumer-group/topic/broker 三维序列。

该示例为 AutoMQ 配置了 Prometheus 抓取任务和指标 pipeline，并使用可选的 infraattributes processor 在相关资源属性可用时添加 Datadog 基础设施标签。OTLP receiver 和其他 pipeline（traces、logs）已省略。如需通过同一 Collector 采集其他应用的 OTLP 数据，请参考 Datadog OTEL Agent Gateway 文档添加相应的 receiver 和 pipeline。

步骤 3：部署 Datadog Agent

添加 Datadog Helm 仓库并创建 Kubernetes Secret 存储 API key：

helm repo add datadog https://helm.datadoghq.com
helm repo update

kubectl create namespace datadog
kubectl -n datadog create secret generic datadog-secret \
  --from-literal api-key=<your-datadog-api-key>

使用步骤 2 中的 values 文件安装 Datadog Agent：

helm install datadog-agent datadog/datadog \
  -f datadog-values.yaml \
  -n datadog \
  --set datadog.apiKeyExistingSecret=datadog-secret \
  --set datadog.site=<your-datadog-site>

将 <your-datadog-site> 替换为你的 Datadog Site（例如 datadoghq.com、us3.datadoghq.com、datadoghq.eu）。

步骤 4：验证集成结果

确认 Datadog Agent service account 具备发现 Pod 的权限：

kubectl auth can-i list pods \
  --as=system:serviceaccount:datadog:datadog-agent \
  --all-namespaces

确认 OTEL Agent Gateway Deployment 已正常运行：

kubectl -n datadog rollout status deployment/datadog-agent-datadog-otel-agent-gateway

查看 OTEL Collector 日志，并确认 AutoMQ scrape job 已被添加：

kubectl -n datadog logs deployment/datadog-agent-datadog-otel-agent-gateway --since=10m | grep -i automq

然后在 Datadog 中验证指标：

打开 Datadog，进入 Metrics > Explorer。
在搜索栏中输入指标名称前缀，例如 kafka_broker_active_count。
使用 service:<your-cluster-id> 过滤，确认指标属于预期的 AutoMQ 集群。在导入的大盘中，过滤器名称仍为 cluster_id，但其取值来自 Datadog service tag。如果 AutoMQ 环境原生暴露了 env_id 或 instance_id 标签，也可以将它们作为额外过滤条件。
新指标可能需要 1–3 分钟才能显示。

步骤 5：导入 AutoMQ Datadog 仪表盘

AutoMQ 指标在 Datadog 中可见后，下载 Datadog 仪表盘 JSON 模板，并导入到你的 Datadog 组织中。

仪表盘	JSON 模板
Cluster Overview	下载 JSON
Topic Metrics	下载 JSON
Group Metrics	下载 JSON

导入仪表盘时，打开 Datadog，进入 Dashboards，创建或打开一个仪表盘，选择 dashboard JSON 导入选项，然后粘贴或上传下载的 JSON 内容。

这些仪表盘模板与步骤 2 中的指标名称、标签和采集降维配置配套使用。Cluster、Topic 和 Group 视图会使用 OTEL transform processor 创建的 automq_* 聚合指标。标准大盘显示 cluster_id 作为必需过滤条件，并映射到 Datadog service tag；当指标存在 env_id 和 instance_id 时，也会提供对应的可选过滤条件。AutoMQ Version 应从节点元数据 gauge（kafka_node_metadata、kafka_node_info 或 kafka_node）发现，并由 dashboard 生成/导入流程渲染。不要从 Helm Chart/App version label 派生，因为该 label 可能与运行时镜像版本不一致。

AutoMQ 指标使用 Prometheus 命名规范（下划线分隔，例如 kafka_broker_active_count），而 Datadog 内置的 Apache Kafka 仪表盘依赖 Datadog Kafka 集成，使用点号分隔的命名（例如 kafka.request.time）。因此，AutoMQ 指标不会出现在内置仪表盘中。请使用 Datadog Metrics Explorer 直接查询 AutoMQ 指标，或使用下划线分隔的名称构建自定义仪表盘。可用指标请参考 Prometheus Metrics。

以下截图展示了命名差异 — Datadog 内置仪表盘使用点号分隔格式，而 AutoMQ 指标使用下划线分隔格式：

方式二：Linux 环境使用 Gateway OTEL Collector

该方式下，单个 OTEL Collector 或 Datadog DDOT Collector 以 Gateway 方式运行在 Linux 上，并通过 static targets 抓取所有 AutoMQ Prometheus metrics endpoints。Linux 方式复用 Kubernetes 方式中的标准最小化指标降维 profile，区别只在服务发现：Linux 使用静态抓取目标列表，不使用 Kubernetes service discovery。

AutoMQ Prometheus metrics endpoints
    ↓  Prometheus scrape /metrics
Linux 上的 Gateway OTEL Collector / DDOT Collector
    ↓  转发到 Datadog
Datadog

步骤 1：确认 AutoMQ 指标导出

AutoMQ Cloud BYOC 集群默认会暴露 Prometheus metrics endpoint。标准 BYOC 流程不需要修改 server.properties。

步骤 2：安装 Collector

请选择一种 Collector 发行版，并保持配置路径和启动方式与该发行版一致。

Datadog DDOT Collector
Upstream otelcol-contrib

安装内嵌 DDOT Collector 的 Datadog Agent。Datadog 官方 Linux 安装流程通过 datadog-agent 管理 Collector，OpenTelemetry Collector 配置文件路径为 /etc/datadog-agent/otel-config.yaml。

DD_API_KEY="<your-api-key>" \
DD_SITE="<your-datadog-site>" \
DD_OTELCOLLECTOR_ENABLED=true \
DD_AGENT_MAJOR_VERSION=7 \
bash -c "$(curl -L https://install.datadoghq.com/scripts/install_script_agent7.sh)"

sudo datadog-agent status

如果主机上已经安装 Datadog Agent，请参考 Datadog DDOT Collector 文档选择与当前 Agent 版本匹配的安装方式。

安装包含 Prometheus receiver、Datadog exporter、transform processor、filter processor、cumulative-to-delta processor、memory limiter processor、batch processor 和 health check extension 的 OpenTelemetry Collector 发行版。下面的示例使用 /etc/otelcol/config.yaml 和自定义 otelcol systemd 服务。确认 Collector binary 可用：

<otel-collector-binary> --version

步骤 3：配置 Collector

创建 Collector 配置文件。下面是标准最小化 profile 的 Linux 版本：使用 static Prometheus scrape targets，只保留标准大盘或 rollup processor 需要的原始指标，创建低基数 automq_* 聚合指标，并在聚合后丢弃原始源指标。请为每个 AutoMQ 集群运行一份该 Gateway Collector 配置，并确保 Collector 所在主机可以访问所有 AutoMQ metrics endpoints。不要在每台 AutoMQ 节点上使用只抓取 127.0.0.1:9090 的相同 profile，因为本地 Collector 会先各自对 counter 做局部 rollup，再导出到 Datadog。根据选择的 Collector 使用对应配置路径：

Datadog DDOT Collector：/etc/datadog-agent/otel-config.yaml
Upstream otelcol-contrib：/etc/otelcol/config.yaml

COLLECTOR_CONFIG_PATH="/etc/datadog-agent/otel-config.yaml" # Datadog DDOT Collector
# COLLECTOR_CONFIG_PATH="/etc/otelcol/config.yaml"          # Upstream otelcol-contrib

sudo mkdir -p "$(dirname "$COLLECTOR_CONFIG_PATH")"

COLLECTOR_CONFIG_PATH="/etc/datadog-agent/otel-config.yaml" # Datadog DDOT Collector
# COLLECTOR_CONFIG_PATH="/etc/otelcol/config.yaml"          # Upstream otelcol-contrib

sudo tee "$COLLECTOR_CONFIG_PATH" > /dev/null << 'EOF'
receivers:
  prometheus:
    config:
      scrape_configs:
        - job_name: "<your-cluster-id>"
          scrape_interval: 15s
          honor_labels: true
          static_configs:
            - targets:
                - "<automq-node-1-host>:9090"
                - "<automq-node-2-host>:9090"
                - "<automq-node-3-host>:9090"
              labels:
                env_id: "<your-env-id>"
                instance_id: "<your-instance-id>"
          metric_relabel_configs:
            - source_labels: [__name__]
              action: keep
              regex: "^(consumer_(offset_lag_count|estimated_time_lag_milliseconds)|jvm_cpu_(count|time_seconds_total)|kafka_(node(_metadata|_info)?|broker_(active_count|fenced_count|network_io_bytes_total)|group_(count|stable_count|empty_count|dead_count|preparing_rebalance_count|completing_rebalance_count|commit_offset)|log_(size|end_offset)|message_count_total|network_io_bytes_total|partition_total_count|request_error_count_total|stream_topic_partition_count|tabletopic_(delay_milliseconds|fps_fields_per_second)|topic_(count|request_count_total)))$"
            - action: labeldrop
              regex: "^(active_controller|customer|exported_env_id|exported_instance_id|exported_job|instance|link_id|operation_name|operation_type|otel_scope_name|pool_type|request|request_type|size|source_node_id)$"
exporters:
  datadog:
    api:
      # Upstream otelcol-contrib：保留这些环境变量引用，并在启动服务前设置 DD_API_KEY/DD_SITE。
      # Datadog DDOT Collector：可以为 datadog-agent 设置这些环境变量，或删除整个 api 配置块以使用 core Agent 配置。
      key: ${env:DD_API_KEY}
      site: ${env:DD_SITE}
extensions:
  health_check:
    endpoint: 0.0.0.0:13133
processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 1024
    spike_limit_mib: 256
  batch:
    timeout: 10s
    send_batch_size: 10000
  cumulativetodelta/automq_counters:
    initial_value: auto
    include:
      match_type: strict
      metrics:
        - kafka_network_io_bytes_total
        - kafka_message_count_total
        - kafka_topic_request_count_total
        - kafka_broker_network_io_bytes_total
        - kafka_request_error_count_total
  transform/automq_rollups:
    error_mode: ignore
    metric_statements:
      - context: metric
        statements:
          - copy_metric(name="automq_topic_log_size") where metric.name == "kafka_log_size"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic"]) where metric.name == "automq_topic_log_size"
          - copy_metric(name="automq_topic_log_end_offset") where metric.name == "kafka_log_end_offset"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic"]) where metric.name == "automq_topic_log_end_offset"
          - copy_metric(name="automq_topic_network_io_bytes_total") where metric.name == "kafka_network_io_bytes_total"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic", "direction"]) where metric.name == "automq_topic_network_io_bytes_total"
          - copy_metric(name="automq_topic_message_count_total") where metric.name == "kafka_message_count_total"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic", "direction"]) where metric.name == "automq_topic_message_count_total"
          - copy_metric(name="automq_topic_request_count_total") where metric.name == "kafka_topic_request_count_total"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic", "type"]) where metric.name == "automq_topic_request_count_total"
          - copy_metric(name="automq_cluster_broker_network_io_bytes_total") where metric.name == "kafka_broker_network_io_bytes_total"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "direction"]) where metric.name == "automq_cluster_broker_network_io_bytes_total"
          - copy_metric(name="automq_request_error_count_total") where metric.name == "kafka_request_error_count_total"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "error"]) where metric.name == "automq_request_error_count_total"
          - copy_metric(name="automq_topic_partition_count") where metric.name == "kafka_stream_topic_partition_count"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic"]) where metric.name == "automq_topic_partition_count"
          - copy_metric(name="automq_topic_rack_partition_count") where metric.name == "kafka_stream_topic_partition_count"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "topic", "rack"]) where metric.name == "automq_topic_rack_partition_count"
          - copy_metric(name="automq_group_count") where metric.name == "kafka_group_count"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_count"
          - copy_metric(name="automq_group_stable_count") where metric.name == "kafka_group_stable_count"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_stable_count"
          - copy_metric(name="automq_group_empty_count") where metric.name == "kafka_group_empty_count"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_empty_count"
          - copy_metric(name="automq_group_dead_count") where metric.name == "kafka_group_dead_count"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_dead_count"
          - copy_metric(name="automq_group_preparing_rebalance_count") where metric.name == "kafka_group_preparing_rebalance_count"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_preparing_rebalance_count"
          - copy_metric(name="automq_group_completing_rebalance_count") where metric.name == "kafka_group_completing_rebalance_count"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service"]) where metric.name == "automq_group_completing_rebalance_count"
          - copy_metric(name="automq_consumer_group_topic_commit_offset") where metric.name == "kafka_group_commit_offset"
          - copy_metric(name="automq_group_topic_offset_lag") where metric.name == "consumer_offset_lag_count"
          - copy_metric(name="automq_group_topic_estimated_time_lag_milliseconds") where metric.name == "consumer_estimated_time_lag_milliseconds"
          - aggregate_on_attributes("max", ["env_id", "instance_id", "service", "group", "topic"]) where metric.name == "automq_group_topic_estimated_time_lag_milliseconds"
      - context: datapoint
        statements:
          - set(datapoint.attributes["group"], datapoint.attributes["consumer_group"]) where metric.name == "automq_consumer_group_topic_commit_offset" and datapoint.attributes["consumer_group"] != nil
          - set(datapoint.attributes["consumer_group"], datapoint.attributes["group"]) where metric.name == "automq_group_topic_offset_lag" and datapoint.attributes["group"] != nil
      - context: metric
        statements:
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "consumer_group", "group", "topic"]) where metric.name == "automq_consumer_group_topic_commit_offset"
          - aggregate_on_attributes("sum", ["env_id", "instance_id", "service", "consumer_group", "group", "topic"]) where metric.name == "automq_group_topic_offset_lag"
  filter/drop_raw_replaced_metrics:
    metrics:
      metric:
        - 'name == "kafka_log_size"'
        - 'name == "kafka_log_end_offset"'
        - 'name == "kafka_network_io_bytes_total"'
        - 'name == "kafka_message_count_total"'
        - 'name == "kafka_broker_network_io_bytes_total"'
        - 'name == "kafka_topic_request_count_total"'
        - 'name == "kafka_request_error_count_total"'
        - 'name == "kafka_stream_topic_partition_count"'
        - 'name == "kafka_group_count"'
        - 'name == "kafka_group_stable_count"'
        - 'name == "kafka_group_empty_count"'
        - 'name == "kafka_group_dead_count"'
        - 'name == "kafka_group_preparing_rebalance_count"'
        - 'name == "kafka_group_completing_rebalance_count"'
        - 'name == "kafka_group_commit_offset"'
        - 'name == "consumer_offset_lag_count"'
        - 'name == "consumer_estimated_time_lag_milliseconds"'
service:
  extensions: [health_check]
  pipelines:
    metrics:
      receivers: [prometheus]
      processors: [memory_limiter, cumulativetodelta/automq_counters, transform/automq_rollups, filter/drop_raw_replaced_metrics, batch]
      exporters: [datadog]
EOF

当 honor_labels: true 时，如果被抓取的 AutoMQ endpoint 已经暴露 Prometheus job label，则会保留 endpoint 提供的值，job_name 不会覆盖该 label。标准 AutoMQ 指标通常会将 job 暴露为集群标识，Datadog 会将该 label 映射为标准大盘 cluster_id 过滤器使用的 service tag。如果 endpoint 未暴露 job，则 job_name 会作为 fallback 值。静态 env_id 和 instance_id 标签是可选配置；仅当你希望 Linux 采集路径下这些 Datadog dashboard filters 有可选值时才需要设置。

该 Gateway profile 的 rollup 在 Collector 侧完成，用于降低 Datadog custom metrics 基数。它不等价于 Prometheus recording rule，不能表达跨 scrape target 的 max by (..., partition) 后再 sum by (...)。如果需要在分区迁移期间严格去重同一个 partition 被多个 AutoMQ 节点短暂同时暴露的 gauge，请在 Datadog 之前增加 Prometheus recording rule / federation 层，或保留 partition 维度并在 Datadog 查询侧单独处理。

步骤 4：启动 Collector

Datadog DDOT Collector
Upstream otelcol-contrib

重启 Datadog Agent，并确认内嵌 Collector 状态：

sudo systemctl restart datadog-agent
sudo datadog-agent status

前台运行 Collector 做 smoke test：

DD_API_KEY="<your-api-key>" \
DD_SITE="<your-datadog-site>" \
<otel-collector-binary> --config /etc/otelcol/config.yaml

如果需要使用 systemd 部署，创建专用服务用户和环境变量文件：

sudo useradd --system --home /var/lib/otelcol --shell /usr/sbin/nologin otelcol || true
sudo install -d -o otelcol -g otelcol /var/lib/otelcol

sudo tee /etc/default/otelcol > /dev/null << 'EOF'
DD_API_KEY=<your-api-key>
DD_SITE=<your-datadog-site>
EOF

由于该文件包含 Datadog API key，需要收紧访问权限。同时，避免把真实 key 粘贴到共享的 shell 历史或日志中。

sudo chmod 600 /etc/default/otelcol

sudo tee /etc/systemd/system/otelcol.service > /dev/null << 'EOF'
[Unit]
Description=OpenTelemetry Collector for AutoMQ Datadog metrics
After=network-online.target
Requires=network-online.target

[Service]
User=otelcol
Group=otelcol
EnvironmentFile=/etc/default/otelcol
ExecStart=<otel-collector-binary-path> --config /etc/otelcol/config.yaml
Restart=always
RestartSec=5
NoNewPrivileges=true
ProtectSystem=full
ProtectHome=true
PrivateTmp=true
StateDirectory=otelcol

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl daemon-reload
sudo systemctl enable otelcol
sudo systemctl start otelcol

步骤 5：验证

打开 Datadog Metrics > Explorer，搜索 AutoMQ 指标，例如 kafka_broker_active_count，并使用 service:<expected-cluster-id> 过滤。在导入的大盘中，过滤器名称仍为 cluster_id，但其取值来自 Datadog service tag。如果没有出现预期的 service 值，请检查一个 AutoMQ /metrics endpoint 中实际暴露的 job label，因为 honor_labels: true 会保留 endpoint 提供的 job 值。如果 AutoMQ 环境原生暴露了 env_id、instance_id 或 broker_id 标签，也可以将它们作为额外过滤条件。新指标可能需要 1–3 分钟才会出现。

附录：为 Kubernetes Software 手动启用 AutoMQ 指标导出

AutoMQ Cloud BYOC 集群默认会暴露 Prometheus metrics endpoint。仅当你管理 Kubernetes AutoMQ Software 部署，并且需要手动启用指标导出时，才需要参考本附录。对于在 Kubernetes 上通过 Helm 部署的 AutoMQ Software，请在 Helm values 中将指标导出器设置为 Prometheus 模式：

global:
  config: |
    s3.telemetry.metrics.exporter.uri=prometheus://?host=0.0.0.0&port=9090

此配置会在每个 AutoMQ Pod 的 9090 端口暴露 Prometheus 格式的指标。本文中的 Datadog Agent Gateway profile 使用基于 selector 的 Kubernetes 服务发现，因此不依赖 Prometheus scrape annotation 来发现 AutoMQ Pod。如果你还需要被该 profile 之外的 annotation-based Prometheus 或 Datadog collector 发现，可以在 AutoMQ Helm values 中添加以下 annotation：

controller:
  annotations:
    prometheus.io/automq-scrape: "true"
    prometheus.io/automq-port: "9090"
    prometheus.io/automq-path: "/metrics"

broker:
  annotations:
    prometheus.io/automq-scrape: "true"
    prometheus.io/automq-port: "9090"
    prometheus.io/automq-path: "/metrics"

产品介绍

快速开始

AutoMQ on Kubernetes

订阅和计费

管理环境

管理账号成员

管理 AutoMQ

迁移到 AutoMQ

安全配置

Kafka Connect

监控报警

Table Topic

最佳实践

消除跨可用区数据传输

版本记录

服务支持

附录

前提条件

方式一：Kubernetes 环境使用 Datadog Agent Gateway

步骤 1：确认 AutoMQ 指标导出

步骤 2：创建 Datadog Agent Gateway values 文件

标准大盘采集 profile

步骤 3：部署 Datadog Agent

步骤 4：验证集成结果

步骤 5：导入 AutoMQ Datadog 仪表盘

方式二：Linux 环境使用 Gateway OTEL Collector

步骤 1：确认 AutoMQ 指标导出

步骤 2：安装 Collector

步骤 3：配置 Collector

步骤 4：启动 Collector

步骤 5：验证

附录：为 Kubernetes Software 手动启用 AutoMQ 指标导出

参考资料

​前提条件

​方式一：Kubernetes 环境使用 Datadog Agent Gateway

​步骤 1：确认 AutoMQ 指标导出

​步骤 2：创建 Datadog Agent Gateway values 文件

​标准大盘采集 profile

​步骤 3：部署 Datadog Agent

​步骤 4：验证集成结果

​步骤 5：导入 AutoMQ Datadog 仪表盘

​方式二：Linux 环境使用 Gateway OTEL Collector

​步骤 1：确认 AutoMQ 指标导出

​步骤 2：安装 Collector

​步骤 3：配置 Collector

​步骤 4：启动 Collector

​步骤 5：验证

​附录：为 Kubernetes Software 手动启用 AutoMQ 指标导出

​参考资料

前提条件

方式一：Kubernetes 环境使用 Datadog Agent Gateway

步骤 1：确认 AutoMQ 指标导出

步骤 2：创建 Datadog Agent Gateway values 文件

标准大盘采集 profile

步骤 3：部署 Datadog Agent

步骤 4：验证集成结果

步骤 5：导入 AutoMQ Datadog 仪表盘

方式二：Linux 环境使用 Gateway OTEL Collector

步骤 1：确认 AutoMQ 指标导出

步骤 2：安装 Collector

步骤 3：配置 Collector

步骤 4：启动 Collector

步骤 5：验证

附录：为 Kubernetes Software 手动启用 AutoMQ 指标导出

参考资料