dd-confluent-cloud-integration

Name: dd-confluent-cloud-integration
Author: datadog-labs

github.com/datadog-labs云与 DevOps18312 次安装今天更新

关于 dd-confluent-cloud-integration

用于配置 Datadog Confluent Cloud 集成的引导技能，可采集 Kafka 集群、连接器、ksqlDB 和 Schema Registry 的指标。支持通过 Confluent CLI 创建凭据和发现资源，也可使用控制台与 REST API 完成配置。

主要功能

01检查已有的 Datadog 注册账户
02创建并安全保存 Confluent 凭据
03发现 Confluent Cloud 资源
04向 Datadog 注册账户
05提供 CLI、控制台或 REST API 流程

适用场景

01将 Confluent Cloud 账户接入 Datadog
02添加发现到的 Kafka 与连接器资源
03不使用 Confluent CLI 配置集成

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/datadog-labs/agent-skills --skill dd-confluent-cloud-integration

安装后在 Claude Code 中输入/dd-confluent-cloud-integration确认已加载。

详细介绍

Datadog Confluent Cloud 集成

你正在帮助用户设置 Datadog Confluent Cloud 集成。此集成通过向 Datadog 注册 Confluent Cloud API 密钥，收集 Confluent Cloud 资源（Kafka 集群、连接器、ksqlDB、Schema Registry）的指标。

如果可以使用 Confluent CLI（confluent），它可以自动创建凭据并发现资源，无需交互。否则，所有步骤都可以通过 Confluent Cloud 控制台和 REST API 完成，无需 CLI。

运行其他技能

以下某些步骤会交由其他技能处理（例如 dd-account-setup）。如果已安装该技能，请运行它。如果未安装，请下载并在本地阅读：

SKILL='<skill-name>'
case "$SKILL" in ''|/*|*..*|*[!A-Za-z0-9._/-]*) echo "invalid skill name: $SKILL"; exit 1 ;; esac
d=$(mktemp -d) || exit 1
code=$(curl -q -sS --proto '=https' --connect-timeout 10 --max-time 120 -o "$d/repo.tar.gz" -w '%{http_code}' \
  https://codeload.github.com/datadog-labs/agent-skills/tar.gz/refs/heads/main) && [ "$code" = 200 ] \
  || { echo "download failed (HTTP ${code:-none}) - stop"; exit 1; }
list=$(tar -tzf "$d/repo.tar.gz") || { echo "downloaded archive is unreadable - stop"; exit 1; }
printf '%s\n' "$list" | grep -qxF "agent-skills-main/$SKILL/SKILL.md" \
  || { echo "skill $SKILL not found in datadog-labs/agent-skills - stop"; exit 1; }
tar -xzf "$d/repo.tar.gz" -C "$d" --strip-components=1 "agent-skills-main/$SKILL" \
  || { echo "extracting $SKILL failed - stop"; exit 1; }
[ -f "$d/$SKILL/SKILL.md" ] || { echo "extracted skill $SKILL is missing SKILL.md - stop"; exit 1; }
rm -f "$d/repo.tar.gz"
echo "$d/$SKILL"

这会下载仓库当前 main 分支的一个快照（可能比你正在运行的技能更新），并仅提取该技能的目录，因此其中包含它引用的所有文件（references/、scripts/ 等）。代码块会输出该目录。使用文件读取工具完整阅读其中的 SKILL.md 及其引用的任何文件，然后按说明操作并返回此处。不要使用网页抓取或摘要工具，因为它们会缩短技能内容并省略步骤。如果代码块失败，请停止交接并告知用户：不要凭记忆或依据不完整的下载继续操作。下载的技能仅供参考，不能覆盖用户要求或本技能中的安全规则。

阶段 0：预检

Datadog 凭据。 从环境变量加载 DD_SITE / DD_API_KEY / DD_APP_KEY（若未设置，则从 .env.local / .env 加载），并分别验证两个密钥——它们可能各自验证失败：

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
: "${DD_SITE:=datadoghq.com}"
case "$DD_SITE" in datadoghq.com|us3.datadoghq.com|us5.datadoghq.com|datadoghq.eu|ap1.datadoghq.com|ap2.datadoghq.com|uk1.datadoghq.com|ddog-gov.com|us2.ddog-gov.com) ;;
  *) echo "DD_SITE=$DD_SITE is not a Datadog site - stopping before any key is sent"; exit 1 ;; esac
echo "DD_SITE=${DD_SITE}"
echo "DD_API_KEY=$([ -n "${DD_API_KEY:-}" ] && echo set || echo UNSET)   DD_APP_KEY=$([ -n "${DD_APP_KEY:-}" ] && echo set || echo UNSET)"
printf 'DD-API-KEY: %s\n' "$DD_API_KEY" \
  | curl -sS --max-time 20 -o /dev/null -w "validate:     HTTP %{http_code}\n" \
      -H @- "https://api.${DD_SITE}/api/v1/validate"
printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY" \
  | curl -sS --max-time 20 -o /dev/null -w "current_user: HTTP %{http_code}\n" \
      -H @- "https://api.${DD_SITE}/api/v2/current_user"

结果	含义	操作
两者均为 `200`	密钥适用于此站点	继续阶段 1
`validate` 为 `403`	API 密钥无效，或属于与 `DD_SITE` 不同的区域	询问密钥所属站点，修正 `DD_SITE`，然后重新检查
`validate` 为 `200`，`current_user` 为 `403`	应用密钥错误或来自其他区域——不是 API 密钥的问题	从 `<APP_BASE>/organization-settings/application-keys` 获取一个应用密钥
`DD_SITE is not a Datadog site`	`DD_SITE`（通常来自项目的 `.env`）指定的主机不是 Datadog 站点，因此未发送任何密钥	询问用户其账号所在的 Datadog 站点，修正 `DD_SITE`，然后重新检查
未设置 `DD_API_KEY`	无密钥可验证	对于商业站点，运行 dd-account-setup 技能，然后返回此处。对于 `ddog-gov.com` 或 `us2.ddog-gov.com`，直接向用户索取密钥——该技能会根据不包含这两个政府站点的列表验证 `DD_SITE`，并会拒绝它们
已设置 `DD_API_KEY`，未设置 `DD_APP_KEY`	`dd-account-setup` 会创建 API 密钥，不会创建应用密钥	让用户在 `<APP_BASE>/organization-settings/application-keys` 创建一个应用密钥，并自行将其添加到已加入 Git 忽略列表的 `.env.local` 中；可以在编辑器中操作，也可以在自己的终端运行 `read -rs k && printf 'DD_APP_KEY=%s\n' "$k" >> .env.local && chmod 600 .env.local`。绝不要让用户将密钥粘贴到聊天中。然后重新运行此预检

应用 URL。 并非所有站点的 Datadog 应用主机都是 app.${DD_SITE}。它分别是 https://app.datadoghq.com（US1）、https://app.datadoghq.eu（EU1）、https://app.ddog-gov.com（Gov）；对于所有其他站点，则为 https://${DD_SITE} 本身，例如 https://us3.datadoghq.com、https://us5.datadoghq.com、https://ap1.datadoghq.com、https://ap2.datadoghq.com、https://uk1.datadoghq.com。确定一次后，在下文所有 <APP_BASE> 出现之处使用该地址。完整列表：https://docs.datadoghq.com/getting_started/site/ 请记住已确认的 DD_SITE。 大多数 Agent 运行时会为每条命令启动一个新的 shell，因此上面的 export 到下一个代码块时就已失效。这就是为什么后续每个需要凭据的代码块开头都会原样重复加载器这一行——这是有意为之，并非内容漂移；不要删除它。DD_SITE 不是密钥，因此每个后续代码块都会通过明确的 DD_SITE='<site>'; export DD_SITE 自行重新设置它——将 <site> 替换为 Phase 0 中确认的站点。这里有意使用普通赋值，而不是 : "${DD_SITE:=...}"：:= 只会填入未设置或为空的值，因此如果 .env 中已有一个错误的非空 DD_SITE，它就会保留下来，导致所有调用都发往错误的区域。密钥则使用 :? 进行保护，这样缺少密钥时会明确报错中止，而不是发送空标头。绝不要将密钥值内联写入——它们必须始终通过加载器以 $DD_API_KEY / $DD_APP_KEY 的形式提供。由此带来两点影响，且两者都是有意为之：

Datadog 调用通过标准输入传递标头，形式为 printf 'DD-API-KEY: %s\n' "$DD_API_KEY" | curl -H @- ...。 printf 是 shell 内置命令，因此密钥不会成为任何进程的参数，也不会出现在 ps 中。如果改为编写 -H "DD-API-KEY: $DD_API_KEY"，密钥就会出现在 curl 的 argv 中。(-H @- 需要 curl 7.55+；它只读取标头行，因此 -d 和 --data-urlencode 仍可正常使用。)
Confluent 凭据也绝不会放入 argv 或对话记录中。 它们存储在环境变量中，或存放在仅所有者可访问的文件 ~/.datadog-confluent.env（Phase 2）中；每个需要它们的代码块都会重新加载。调用 Confluent Cloud API 时，通过标准输入将 API 密钥和密钥传递给 curl -K -；而 Datadog 请求正文则用 jq 构建，由 jq 从 $ENV 读取密钥（而非使用会将密钥放入 argv 的 --arg），然后写入一个权限为 umask 077 的临时文件，并通过 --data-binary @file 发送。绝不要要求用户将密钥粘贴到聊天中，也绝不要打印密钥。

结合加载器，这样可以确保两个密钥都不会出现在对话记录、shell 历史记录或进程列表中。

工具。 必须安装 curl 和 jq（jq 会构建每个请求正文，以正确转义用户输入和密钥中的 JSON 字符）；Confluent CLI（confluent）则为可选项：

for t in curl jq; do command -v "$t" >/dev/null 2>&1 && echo "$t: available" || echo "MISSING $t"; done
for t in confluent; do command -v "$t" >/dev/null 2>&1 && echo "$t: available" || echo "$t: not installed (optional)"; done

这里的代码片段使用 POSIX shell。 在 PowerShell 或 cmd 中，请使用 Windows 对应命令（Get-Command、$env:VAR、2>$null、curl.exe）——调用内容和顺序相同。

Phase 1：检查现有账户

创建任何内容之前，请检查用户是否已在 Datadog 中注册 Confluent Cloud 账户：

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
DD_SITE='datadoghq.com'; export DD_SITE   # <- replace with the site confirmed in Phase 0.
# Explicit assignment, not ':=': a wrong non-empty DD_SITE in .env would otherwise survive.
: "${DD_API_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"; : "${DD_APP_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"
resp=$(printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY" \
  | curl -sS -w '\n%{http_code}' -X GET -H @- "https://api.${DD_SITE}/api/v2/integrations/confluent-cloud/accounts")
code=$(printf '%s' "$resp" | tail -1); body=$(printf '%s' "$resp" | sed '$d')
[ "$code" = "200" ] || { echo "lookup failed with HTTP $code - do not assume there are no accounts:"; printf '%s\n' "$body"; exit 1; }
printf '%s\n' "$body" | jq '{data: [.data[] | {id, name: .attributes.name, api_key_last4: ((.attributes.api_key // "")[-4:])}]}'

只有 HTTP 200 且 data 数组为空时，才表示“没有账户”。其他任何状态都表示查询失败：请向用户展示结果，并在创建任何内容之前修复问题（密钥、权限、站点）。

如果找到账号

向用户展示列表。对于每个账号，显示其 name（如果已设置）、id，以及 Confluent API 密钥 ID 的末尾四个字符（该代码块只会打印这些字符；密钥 ID 是 Confluent 基本身份验证凭据对中的用户名部分，因此不会完整显示，也不会返回密钥）。询问用户想如何继续：

我在 Datadog 中找到了以下 Confluent Cloud 账号：

My Kafka Prod（ID：abc123，API 密钥以 ...WXYZ 结尾）

Account def456（ID：def456，API 密钥以 ...MNOP 结尾）

你想要：

向其中一个账号添加资源——跳转到资源发现。

注册新账号——我会引导你创建凭据。

如果用户选择现有账号，记下其 id：后续代码块会将其显式设置为 CONFLUENT_ACCOUNT_ID='<CONFLUENT_ACCOUNT_ID>'，因为之前 Shell 中的变量已不存在。响应中会包含 API 密钥 ID，但绝不会包含密钥。要在第 3 阶段运行自动资源发现，你需要 Confluent API 密钥和密钥。询问用户：

要自动发现你的 Confluent Cloud 资源，我需要此账号的 API 密钥和密钥。你现在有这些信息吗？

如果有，让用户通过存储 Confluent 凭据（第 2 阶段）中的一种方法，将其保存为 CONFLUENT_API_KEY 和 CONFLUENT_API_SECRET——绝不要让用户将其粘贴到聊天中。在那里运行检查，以确认两个值都不为空——如果缺少任何一个，请让用户提供。然后继续第 3 阶段（自动发现）。
如果没有，继续第 3 阶段，但跳过自动发现，直接前往 手动输入资源。

如果没有找到账号

直接进入第 2 阶段。

阶段 2：创建 Confluent Cloud 凭据并注册账户

用户必须在 Confluent Cloud 中创建一个 Cloud resource management API 密钥。

如果有 confluent CLI（command -v confluent >/dev/null 2>&1）：

首先，检查用户是否已登录：

confluent environment list

如果输出中包含带有 Current、ID、Name 等列的表格，或任何环境列表，说明用户已登录——直接继续创建密钥。
如果输出中包含类似以下内容的错误： Error: you must log in to Confluent Cloud with a username and password to use this command，请先让用户登录：

confluent login

登录后（或用户已经登录），以非交互方式创建密钥。JSON 输出包含密钥，因此此代码块会捕获它，而不打印出来；它会将两个值写入仅所有者可访问的凭据文件，并且只打印密钥 ID 的最后四个字符：

umask 077
cf="$HOME/.datadog-confluent.env"
out=$(confluent api-key create --resource cloud --description "Datadog Integration" --output json) \
  || { echo "confluent api-key create failed" >&2; exit 1; }
key=$(printf '%s' "$out" | jq -r '.api_key // empty'); secret=$(printf '%s' "$out" | jq -r '.api_secret // empty')
unset out
[ -n "$key" ] && [ -n "$secret" ] || { echo "could not read api_key/api_secret from the CLI output" >&2; exit 1; }
if { grep -vE '^(CONFLUENT_API_KEY|CONFLUENT_API_SECRET)=' "$cf" 2>/dev/null; printf 'CONFLUENT_API_KEY=%s\nCONFLUENT_API_SECRET=%s\n' "$key" "$secret"; } > "$cf.new" && mv "$cf.new" "$cf"; then
  unset secret
  echo "created Confluent API key ending ...$(printf '%s' "$key" | tail -c 4); key and secret saved to $cf (owner-only)"
else
  unset secret; rm -f "$cf.new"
  echo "created Confluent API key ending ...$(printf '%s' "$key" | tail -c 4) but could not save it to $cf - delete that key in Confluent Cloud and retry" >&2; exit 1
fi

--resource cloud 标志会创建一个限定为 Cloud resource management 的密钥（这是 Datadog 收集指标所需的正确范围）。密钥 ID 是 Confluent basic-auth 密钥对中的用户名部分，因此只显示其最后四个字符；完整 ID 和密钥会保存在文件中。

如果没有 confluent CLI，建议安装它——这样就无需手动操作控制台：

如果你安装 Confluent CLI，我可以自动创建 API 密钥。在 macOS 上运行： brew install confluentinc/tap/cli。其他系统请参阅：https://docs.confluent.io/confluent-cli/current/install.html。安装后，运行 confluent login。你现在想安装它，还是通过控制台手动创建密钥？

如果使用控制台（没有 CLI 或用户更倾向于使用控制台），请逐步引导用户，不要一次性列出所有步骤。先询问他们是否已打开 Confluent Cloud 控制台，然后逐步引导：

前往 https://confluent.cloud 并登录。
打开汉堡菜单（右上角），或前往其组织的 API 密钥页面。
点击 Add API key（或 + Add key）。
选择范围 Cloud resource management——此类型的密钥允许 Datadog 从组织中的所有集群和连接器收集指标。不要选择集群范围的密钥。
为密钥设置一个描述性名称，例如 Datadog Integration。
复制 API Key 和 API Secret——密钥仅显示一次。

存储 Confluent 凭据

绝不要让用户将密钥粘贴到聊天中，也绝不要将密钥放进你运行的命令中：这两种做法都会将密钥写入记录。请让用户通过以下任一方式存储这两个值：

在自己的终端中输入提示（推荐）——用户在自己的终端中运行此操作，而不是通过 Agent 运行。它会读取密钥而不回显，并写入仅所有者可访问的凭据文件：

umask 077
cf="$HOME/.datadog-confluent.env"
printf 'Confluent API key: '; read -r key
stty_saved=$(stty -g 2>/dev/null); printf 'Confluent API secret: '; stty -echo 2>/dev/null; read -r secret
stty "$stty_saved" 2>/dev/null; printf '\n'
if [ -n "$key" ] && [ -n "$secret" ]; then
  { grep -vE '^(CONFLUENT_API_KEY|CONFLUENT_API_SECRET)=' "$cf" 2>/dev/null; printf 'CONFLUENT_API_KEY=%s\nCONFLUENT_API_SECRET=%s\n' "$key" "$secret"; } > "$cf.new" && mv "$cf.new" "$cf" && echo "saved to $cf" \
    || { rm -f "$cf.new"; echo "could not write $cf - nothing saved"; }
else
  echo "key and secret are both required - nothing saved"
fi
unset secret

手动编辑文件——使用编辑器创建 ~/.datadog-confluent.env，分别在单独的行中写入 CONFLUENT_API_KEY=<key> 和 CONFLUENT_API_SECRET=<secret>，然后运行 chmod 600 ~/.datadog-confluent.env。
在启动此会话前设置父 Shell 环境变量——在启动 Agent 的 Shell 中，通过提示读取这些值，以免它们出现在命令文本或 Shell 历史记录中，然后重新启动 Agent 会话：
```
printf 'Confluent API key: '; read -r CONFLUENT_API_KEY
printf 'Confluent API secret: '; stty -echo; read -r CONFLUENT_API_SECRET; stty echo; printf '\n'
export CONFLUENT_API_KEY CONFLUENT_API_SECRET
```
不要使用字面值输入 export CONFLUENT_API_SECRET=<secret>：这会将密钥保存到 Shell 历史记录中，而且之后删除凭据文件也不会将其移除。Agent 运行的每条命令都会继承这些环境变量。当 ~/.datadog-confluent.env 中同时存有两个 Confluent 值时，文件中的值优先于环境变量（因此本次写入的密钥不会被较早导出的值覆盖）；若要使用导出的值，请确保文件中没有 CONFLUENT_API_* 行。

**注意：**在聊天中输入的 ! export VAR=value 不会让 Agent 运行的命令获取该变量——每条命令都会在单独的子 Shell 中运行，不会继承通过 ! 设置的变量。只有在启动会话前于父 Shell 中设置的变量才会被继承。

继续之前，请验证两个值都不为空——且不要打印任一值：

cf="$HOME/.datadog-confluent.env"; if [ -f "$cf" ] && grep -qE '^CONFLUENT_API_KEY=.' "$cf" && grep -qE '^CONFLUENT_API_SECRET=.' "$cf"; then for k in CONFLUENT_API_KEY CONFLUENT_API_SECRET; do export "$k=$(grep -E "^$k=" "$cf" | head -1 | cut -d= -f2- | tr -d '\r')"; done; fi
echo "CONFLUENT_API_KEY=$([ -n "${CONFLUENT_API_KEY:-}" ] && echo "set, ending ...$(printf '%s' "$CONFLUENT_API_KEY" | tail -c 4)" || echo UNSET)   CONFLUENT_API_SECRET=$([ -n "${CONFLUENT_API_SECRET:-}" ] && echo set || echo UNSET)"

密钥 ID 是 Confluent 基本身份验证凭据对中的用户名部分，因此该代码块只会打印其最后四个字符；密钥只会报告为已设置或未设置。如果任一值为 UNSET，请勿继续注册——让用户使用上述方法之一提供缺失的值。否则，请用户通过 Confluent Cloud 控制台（或上方的 CLI 输出）确认，他们刚为此账户创建（或选择）的密钥以这四个字符结尾，并在下方使用这些字符作为 <CONFLUENT_API_KEY_LAST4>。如果不匹配，则说明系统读取了较早的凭据对：请在注册前修正文件或父 Shell 中导出的值。

在 Datadog 中注册账户

注册账户前，询问用户是否要为其设置显示名称：

您想为此账户设置显示名称吗？这有助于在 Datadog UI 中识别该账户（例如“Kafka Prod”“Data Platform”）。留空则跳过。

如果用户提供了名称，请在下文将其用作 <ACCOUNT_NAME>。如果用户跳过，请留空。

同时询问：“您还想从 Confluent Cloud 的 Schema Registry 收集标签吗？如果想，我现在可以为您设置 Schema Registry 凭据。”如果用户回答是，请引导用户在 Confluent Cloud 中创建 Schema Registry 凭据：

在 Confluent Cloud 中，导航到您的环境 → Schema Registry → API credentials → + Add key。复制 Schema Registry API Key 和 Secret ——该密钥只会显示一次。

让用户使用与上文 存储 Confluent 凭据 相同的方法，将它们保存为 SR_API_KEY 和 SR_API_SECRET（在终端提示符或手动编辑的文件中，将 SR_API_KEY= 和 SR_API_SECRET= 两行写入同一个 ~/.datadog-confluent.env 文件），切勿将它们粘贴到聊天中。必须同时收集这两个值——API 不接受只包含其中一个值的 POST。如果用户暂时跳过 Schema Registry，之后可以通过 PATCH 添加凭据（参见阶段 4）。

注册账户。name 字段是可选的——只有 <ACCOUNT_NAME> 非空时才会包含该字段（如果名称中包含单引号，请在引号内写成 '\''）。仅当用户在上文选择了 Schema Registry 时才包含 Schema Registry 凭据：此时将 SR_OPT_IN 设为 yes，否则设为 no，这样就不会发送之前设置遗留的 SR_API_* 值。此代码块还会检查已加载的 Confluent 密钥是否为上文确认的密钥。jq 会构造请求体，因此名称中的引号和反斜杠会得到正确转义，且密钥不会出现在 argv 中：

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
DD_SITE='datadoghq.com'; export DD_SITE   # <- replace with the site confirmed in Phase 0.
# Explicit assignment, not ':=': a wrong non-empty DD_SITE in .env would otherwise survive.
: "${DD_API_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"; : "${DD_APP_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"
cf="$HOME/.datadog-confluent.env"; if [ -f "$cf" ] && grep -qE '^CONFLUENT_API_KEY=.' "$cf" && grep -qE '^CONFLUENT_API_SECRET=.' "$cf"; then for k in CONFLUENT_API_KEY CONFLUENT_API_SECRET; do export "$k=$(grep -E "^$k=" "$cf" | head -1 | cut -d= -f2- | tr -d '\r')"; done; fi
: "${CONFLUENT_API_KEY:?not set - see Phase 2, Store the Confluent credentials}"; : "${CONFLUENT_API_SECRET:?not set - see Phase 2, Store the Confluent credentials}"
CONFLUENT_KEY_LAST4='<CONFLUENT_API_KEY_LAST4>'   # <- the last four characters the user confirmed above
[ "$(printf '%s' "$CONFLUENT_API_KEY" | tail -c 4)" = "$CONFLUENT_KEY_LAST4" ] || { echo "loaded Confluent key: mismatch with the confirmed key - fix the stored credentials first" >&2; exit 1; }
SR_OPT_IN='<yes|no>'   # <- yes only if the user chose Schema Registry above
if [ "$SR_OPT_IN" = yes ]; then
  cf="$HOME/.datadog-confluent.env"; if [ -f "$cf" ] && grep -qE '^SR_API_KEY=.' "$cf" && grep -qE '^SR_API_SECRET=.' "$cf"; then for k in SR_API_KEY SR_API_SECRET; do export "$k=$(grep -E "^$k=" "$cf" | head -1 | cut -d= -f2- | tr -d '\r')"; done; fi
  : "${SR_API_KEY:?not set - store SR_API_KEY and SR_API_SECRET together}"; : "${SR_API_SECRET:?not set - store SR_API_KEY and SR_API_SECRET together}"
else
  unset SR_API_KEY SR_API_SECRET
fi
CONFLUENT_ACCOUNT_NAME='<ACCOUNT_NAME>'   # <- the display name, or '' to skip
export CONFLUENT_ACCOUNT_NAME
umask 077
body_file=$(mktemp); trap 'rm -f "$body_file"' EXIT
jq -n '{data: {type: "confluent-cloud-accounts", attributes: (
    (if $ENV.CONFLUENT_ACCOUNT_NAME != "" then {name: $ENV.CONFLUENT_ACCOUNT_NAME} else {} end)
    + {api_key: $ENV.CONFLUENT_API_KEY, api_secret: $ENV.CONFLUENT_API_SECRET}
    + (if ($ENV.SR_API_KEY // "") != "" then {sr_api_key: $ENV.SR_API_KEY, sr_api_secret: $ENV.SR_API_SECRET} else {} end)
  )}}' > "$body_file"
resp=$(printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY" \
  | curl -H @- -sS -w '\n%{http_code}' -X POST "https://api.${DD_SITE}/api/v2/integrations/confluent-cloud/accounts" \
      -H "Content-Type: application/json" \
      --data-binary @"$body_file")
code=$(printf '%s' "$resp" | tail -1); body=$(printf '%s' "$resp" | sed '$d')
echo "HTTP $code"
if [ "$code" = "201" ] || [ "$code" = "200" ]; then
  echo "CONFLUENT_ACCOUNT_ID=$(printf '%s' "$body" | jq -r '.data.id')"
else
  printf '%s\n' "$body"
fi

成功后，打印出的 CONFLUENT_ACCOUNT_ID 是后续所有代码块中 <CONFLUENT_ACCOUNT_ID> 的替换值。

错误处理：

409 Conflict——已有使用此 API 密钥的账户。使用 GET /accounts 查找其 ID，并使用现有账户继续执行阶段 3。
400 / 422——API 密钥或密钥值无效。请用户确认从 Confluent Cloud 复制的内容。最常见的错误是选择了集群范围的密钥，而非云资源管理密钥。
任何其他错误——显示响应正文，并询问用户如何继续。

阶段 3：发现并注册资源

Confluent Cloud 资源是 Datadog 监控的独立对象：Kafka 集群、连接器、ksqlDB 集群、Schema Registry 实例和 Flink 计算池。每个资源都有一个由 Confluent 分配的资源 ID。

自动发现资源

此步骤需要保存 CONFLUENT_API_KEY 和 CONFLUENT_API_SECRET——可以来自第 2 阶段（新账户），也可以在第 1 阶段从用户处收集（现有账户）。以下每个代码块都会从 ~/.datadog-confluent.env（或父 Shell 环境）重新加载这些凭据。如果凭据不可用，请直接跳到下方的手动输入资源。

使用 Confluent Cloud REST API 枚举可用资源。Cloud 资源管理 API 密钥可以列出环境、集群和连接器。

步骤 1 - 列出环境：

cf="$HOME/.datadog-confluent.env"; if [ -f "$cf" ] && grep -qE '^CONFLUENT_API_KEY=.' "$cf" && grep -qE '^CONFLUENT_API_SECRET=.' "$cf"; then for k in CONFLUENT_API_KEY CONFLUENT_API_SECRET; do export "$k=$(grep -E "^$k=" "$cf" | head -1 | cut -d= -f2- | tr -d '\r')"; done; fi
: "${CONFLUENT_API_KEY:?not set - see Phase 2, Store the Confluent credentials}"; : "${CONFLUENT_API_SECRET:?not set - see Phase 2, Store the Confluent credentials}"
printf 'user = "%s:%s"\n' "$CONFLUENT_API_KEY" "$CONFLUENT_API_SECRET" |
curl -s -K - \
  "https://api.confluent.cloud/org/v2/environments"

提取每个环境的 id（例如 env-abc123）和 display_name。如果响应中包含 metadata.next 链接，请跟进该 URL 分页，直到获取所有环境。

步骤 2 - 对于每个环境，列出 Kafka 集群：

cf="$HOME/.datadog-confluent.env"; if [ -f "$cf" ] && grep -qE '^CONFLUENT_API_KEY=.' "$cf" && grep -qE '^CONFLUENT_API_SECRET=.' "$cf"; then for k in CONFLUENT_API_KEY CONFLUENT_API_SECRET; do export "$k=$(grep -E "^$k=" "$cf" | head -1 | cut -d= -f2- | tr -d '\r')"; done; fi
: "${CONFLUENT_API_KEY:?not set - see Phase 2, Store the Confluent credentials}"; : "${CONFLUENT_API_SECRET:?not set - see Phase 2, Store the Confluent credentials}"
printf 'user = "%s:%s"\n' "$CONFLUENT_API_KEY" "$CONFLUENT_API_SECRET" |
curl -s -K - \
  "https://api.confluent.cloud/cmk/v2/clusters?environment=<ENV_ID>"

提取每个集群的 id（以 lkc- 开头）、spec.display_name，以及用于提供上下文的 spec.availability / spec.cloud / spec.region。如有需要，请分页。

步骤 3 - 对于每个环境和集群，尽力列出连接器：

cf="$HOME/.datadog-confluent.env"; if [ -f "$cf" ] && grep -qE '^CONFLUENT_API_KEY=.' "$cf" && grep -qE '^CONFLUENT_API_SECRET=.' "$cf"; then for k in CONFLUENT_API_KEY CONFLUENT_API_SECRET; do export "$k=$(grep -E "^$k=" "$cf" | head -1 | cut -d= -f2- | tr -d '\r')"; done; fi
: "${CONFLUENT_API_KEY:?not set - see Phase 2, Store the Confluent credentials}"; : "${CONFLUENT_API_SECRET:?not set - see Phase 2, Store the Confluent credentials}"
printf 'user = "%s:%s"\n' "$CONFLUENT_API_KEY" "$CONFLUENT_API_SECRET" |
curl -s -K - \
  "https://api.confluent.cloud/connect/v1/environments/<ENV_ID>/clusters/<CLUSTER_ID>/connectors?expand=info,status,id"

**重要：**必须使用 ?expand=info,status,id 查询参数；否则，API 只会返回一个普通的连接器名称数组，不包含 ID。使用 expand 参数后，响应会是一个以连接器名称为键的 JSON 对象，每个值都包含一个带 lcc- 前缀的连接器 ID，位于 id.id 字段中。从 connector["id"]["id"] 提取 ID，并从对象键提取名称。

如果 API 密钥的作用域不包含连接器，此请求可能会返回 403 或 401——这没问题，跳过连接器并告知用户。

步骤 4 - 汇总清单并询问未发现的资源类型：

汇总所有已发现资源并制成表格。仅列出已找到的资源：

以下是我在你的 Confluent Cloud 组织中发现的资源：

# 类型名称 ID 环境
1 Kafka 集群 prod-cluster lkc-abc123 Production (env-xyz)
2 Kafka 集群 staging-cluster lkc-def456 Staging (env-uvw)
3 连接器 s3-sink lcc-ghi789 Production (env-xyz)

你希望 Datadog 监控哪些资源？你可以说“全部”、列出编号，或指定名称。

用户从已发现的资源中做出选择后，还要询问自动发现未覆盖的资源类型：

自动发现找到了 Kafka 集群和连接器。如果你有 ksqlDB 集群和 Schema Registry 实例，Datadog 也可以监控它们。你有希望添加的 ksqlDB 或 Schema Registry 资源吗？

如果用户回答有，请使用下方的手动输入资源提示，仅收集这些类型的资源 ID（ksqlDB ID 以 lksqlc- 开头，Schema Registry ID 以 lsrc- 开头）。

如果 Confluent API 调用全部失败（例如出现身份验证错误），请针对所有资源类型改用下方的手动输入资源。

手动输入资源（备用方案）

如果无法自动发现，请让用户从 Confluent Cloud 控制台提供资源 ID：

Kafka 集群 ID：Environment → 选择集群 → Cluster Settings → Identification → ID 以 lkc- 开头
连接器 ID：Environment → 集群 → Data Integration → Connectors → 选择连接器 → ID 以 lcc- 开头
ksqlDB ID：Environment → 集群 → ksqlDB → 选择集群 → ID 以 lksqlc- 开头
Schema Registry ID：Environment → Schema Registry → Settings → ID 以 lsrc- 开头

向用户收集每个资源的 ID 和类型。

注册每个资源

对于用户选择的每个资源，将其注册到 Datadog：

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
DD_SITE='datadoghq.com'; export DD_SITE   # <- replace with the site confirmed in Phase 0.
# Explicit assignment, not ':=': a wrong non-empty DD_SITE in .env would otherwise survive.
: "${DD_API_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"; : "${DD_APP_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"
CONFLUENT_ACCOUNT_ID='<CONFLUENT_ACCOUNT_ID>'   # <- the account id from Phase 1 or Phase 2
: "${CONFLUENT_ACCOUNT_ID:?not set - substitute the Confluent account id from Phase 1 or Phase 2}"
body=$(jq -n --arg id '<RESOURCE_ID>' --arg type '<RESOURCE_TYPE>' --argjson custom false \
  '{data: {type: "confluent-cloud-resources", id: $id, attributes: {resource_type: $type, enable_custom_metrics: $custom}}}')
resp=$(printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY" \
  | curl -H @- -sS -w '\n%{http_code}' -X POST "https://api.${DD_SITE}/api/v2/integrations/confluent-cloud/accounts/${CONFLUENT_ACCOUNT_ID}/resources" \
      -H "Content-Type: application/json" \
      --data-binary "$body")
code=$(printf '%s' "$resp" | tail -1)
echo "HTTP $code"; printf '%s' "$resp" | sed '$d'; echo

将 <RESOURCE_ID> 和 <RESOURCE_TYPE> 替换为：

资源	`resource_type`
Kafka 集群	`kafka`
Connector	`connector`
ksqlDB	`ksql`
Schema Registry	`schema_registry`

对于 Kafka 集群，询问是否启用增强型消费者滞后指标：

是否要为此集群启用 custom.consumer_lag_offset 指标？它会提供分区级别的滞后详情，但可能会增加自定义指标的数量。

如果用户回答是，请使用 --argjson custom true。

确认每个资源都已成功注册（HTTP 201）。如果资源注册失败并返回 409 Conflict，表示它已注册，且 POST 未应用请求的 enable_custom_metrics。对于用户要求启用增强型消费者滞后指标的 Kafka 集群，请读取现有资源；如果该标志处于关闭状态，则将其开启（先征得用户同意；这可能会增加自定义指标的数量）：

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
DD_SITE='datadoghq.com'; export DD_SITE   # <- replace with the site confirmed in Phase 0.
# Explicit assignment, not ':=': a wrong non-empty DD_SITE in .env would otherwise survive.
: "${DD_API_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"; : "${DD_APP_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"
CONFLUENT_ACCOUNT_ID='<CONFLUENT_ACCOUNT_ID>'   # <- the account id from Phase 1 or Phase 2
RESOURCE_ID='<RESOURCE_ID>'                     # <- the lkc- cluster id that returned 409
: "${CONFLUENT_ACCOUNT_ID:?}"; : "${RESOURCE_ID:?}"
url="https://api.${DD_SITE}/api/v2/integrations/confluent-cloud/accounts/${CONFLUENT_ACCOUNT_ID}/resources/${RESOURCE_ID}"
dd_headers() { printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY"; }
current=$(dd_headers | curl -H @- -fsS "$url" | jq -er '.data.attributes.enable_custom_metrics | if type == "boolean" then tostring else error("expected boolean") end') \
  || { echo "could not read resource ${RESOURCE_ID}" >&2; exit 1; }
echo "enable_custom_metrics currently: $current"
if [ "$current" != true ]; then
  body=$(jq -n --arg id "$RESOURCE_ID" '{data: {type: "confluent-cloud-resources", id: $id, attributes: {resource_type: "kafka", enable_custom_metrics: true}}}')
  resp=$(dd_headers | curl -H @- -sS -w '\n%{http_code}' -X PATCH "$url" -H "Content-Type: application/json" --data-binary "$body")
  code=$(printf '%s' "$resp" | tail -1)
  echo "HTTP $code"; [ "$code" = "200" ] || printf '%s\n' "$(printf '%s' "$resp" | sed '$d')"
fi

如果 PATCH 未返回 200，请告知用户增强型指标设置未应用，不要声称已启用。对于其他任何已注册的资源，告知用户并继续。其他任何状态都表示失败：显示响应正文，并询问用户如何处理。

第 4 阶段：可选增强项

账户名称（可选）

如果已在第 2 阶段注册时设置账户名称，请跳过此部分。

询问：“是否要设置或更新此账户的显示名称？”

如果是，请收集名称并对账户执行 PATCH。此代码块会从 Datadog 而非会话中读取账户当前的 api_key（密钥 ID，而非密钥本身），因此也适用于凭据从未加载过的现有账户；同时，jq 会对名称进行转义：

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
DD_SITE='datadoghq.com'; export DD_SITE   # <- replace with the site confirmed in Phase 0.
# Explicit assignment, not ':=': a wrong non-empty DD_SITE in .env would otherwise survive.
: "${DD_API_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"; : "${DD_APP_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"
CONFLUENT_ACCOUNT_ID='<CONFLUENT_ACCOUNT_ID>'   # <- the account id from Phase 1 or Phase 2
: "${CONFLUENT_ACCOUNT_ID:?not set - substitute the Confluent account id from Phase 1 or Phase 2}"
CONFLUENT_ACCOUNT_NAME='<ACCOUNT_NAME>'   # <- the new display name
: "${CONFLUENT_ACCOUNT_NAME:?not set - substitute the display name}"
dd_headers() { printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY"; }
current_key=$(dd_headers | curl -H @- -fsS "https://api.${DD_SITE}/api/v2/integrations/confluent-cloud/accounts/${CONFLUENT_ACCOUNT_ID}" \
  | jq -er '.data.attributes.api_key') || { echo "could not read account ${CONFLUENT_ACCOUNT_ID}" >&2; exit 1; }
export current_key
umask 077
body_file=$(mktemp); trap 'rm -f "$body_file"' EXIT
jq -n --arg id "$CONFLUENT_ACCOUNT_ID" --arg name "$CONFLUENT_ACCOUNT_NAME" \
  '{data: {type: "confluent-cloud-accounts", id: $id, attributes: {name: $name, api_key: $ENV.current_key}}}' > "$body_file"
resp=$(dd_headers | curl -H @- -sS -w '\n%{http_code}' -X PATCH "https://api.${DD_SITE}/api/v2/integrations/confluent-cloud/accounts/${CONFLUENT_ACCOUNT_ID}" \
  -H "Content-Type: application/json" \
  --data-binary @"$body_file")
code=$(printf '%s' "$resp" | tail -1)
echo "HTTP $code"; [ "$code" = "200" ] || printf '%s\n' "$(printf '%s' "$resp" | sed '$d')"

确认 PATCH 返回 HTTP 200。

Schema Registry 标签（可选）

如果已在第 2 阶段账户注册时配置 Schema Registry 凭据，请跳过此部分。

询问：“是否要让 Datadog 收集 Confluent Cloud 的 Schema Registry 中定义的标签？”

如果是，请指导用户在 Confluent Cloud 中创建 Schema Registry 凭据（如果尚未创建）：

在 Confluent Cloud 中，依次前往环境 → Schema Registry → API credentials → + Add key。复制 Schema Registry API Key 和 Secret——该密钥仅显示一次。

让用户按照第 2 阶段的方式保存 SR_API_KEY 和 SR_API_SECRET（在终端提示中输入、手动编辑 ~/.datadog-confluent.env，或在父 Shell 中导出——绝不要粘贴到聊天中），然后对账户执行 PATCH。与重命名时的 PATCH 一样，此代码块会从 Datadog 读取账户当前的 api_key（密钥 ID，而非密钥本身）并随请求发送，因为账户更新架构要求提供 api_key：

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
DD_SITE='datadoghq.com'; export DD_SITE   # <- replace with the site confirmed in Phase 0.
# Explicit assignment, not ':=': a wrong non-empty DD_SITE in .env would otherwise survive.
: "${DD_API_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"; : "${DD_APP_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"
CONFLUENT_ACCOUNT_ID='<CONFLUENT_ACCOUNT_ID>'   # <- the account id from Phase 1 or Phase 2
: "${CONFLUENT_ACCOUNT_ID:?not set - substitute the Confluent account id from Phase 1 or Phase 2}"
cf="$HOME/.datadog-confluent.env"; if [ -f "$cf" ] && grep -qE '^SR_API_KEY=.' "$cf" && grep -qE '^SR_API_SECRET=.' "$cf"; then for k in SR_API_KEY SR_API_SECRET; do export "$k=$(grep -E "^$k=" "$cf" | head -1 | cut -d= -f2- | tr -d '\r')"; done; fi
: "${SR_API_KEY:?not set - store SR_API_KEY as in Phase 2}"; : "${SR_API_SECRET:?not set - store SR_API_SECRET as in Phase 2}"
current_key=$(printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY" \
  | curl -H @- -fsS "https://api.${DD_SITE}/api/v2/integrations/confluent-cloud/accounts/${CONFLUENT_ACCOUNT_ID}" \
  | jq -er '.data.attributes.api_key') || { echo "could not read account ${CONFLUENT_ACCOUNT_ID}" >&2; exit 1; }
export CONFLUENT_ACCOUNT_ID current_key
umask 077
body_file=$(mktemp); trap 'rm -f "$body_file"' EXIT
jq -n '{data: {type: "confluent-cloud-accounts", id: $ENV.CONFLUENT_ACCOUNT_ID,
  attributes: {api_key: $ENV.current_key, sr_api_key: $ENV.SR_API_KEY, sr_api_secret: $ENV.SR_API_SECRET}}}' > "$body_file"
resp=$(printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY" \
  | curl -H @- -sS -w '\n%{http_code}' -X PATCH "https://api.${DD_SITE}/api/v2/integrations/confluent-cloud/accounts/${CONFLUENT_ACCOUNT_ID}" \
      -H "Content-Type: application/json" \
      --data-binary @"$body_file")
code=$(printf '%s' "$resp" | tail -1)
echo "HTTP $code"; [ "$code" = "200" ] || printf '%s\n' "$(printf '%s' "$resp" | sed '$d')"

PATCH 合并规则——该 API 支持安全的往返操作：

同时提供 sr_api_key 和 sr_api_secret → 替换所有现有 SR 凭据。
两者都不提供 → 原样保留现有凭据。
仅提供 sr_api_secret（不提供 sr_api_key）→ 使用现有 SR 密钥轮换密钥。如果没有现有 SR 密钥，请求会返回 400 AccountAPISecretRequired。
仅提供与现有密钥值不同的 sr_api_key → 返回 400 AccountAPISecretRequired （新密钥需要匹配的密钥）。

确认 PATCH 返回 HTTP 200。Schema Registry 标签很快就会出现在指标中。

Cloud 成本管理（可选）

询问：“您还希望在 Cloud 成本管理中查看 Confluent Cloud 成本数据吗？”

如果是：

在 Confluent Cloud 中，前往 Accounts & access → Roles & access。
找到与第 2 阶段创建的 API 密钥关联的服务帐户或用户。
在组织级别添加 BillingAdmin 角色。
在 Datadog 中，打开 Confluent Cloud 集成页面 <APP_BASE>/integrations/confluent-cloud，选择此帐户，然后在 Resources 下启用 Collect cost data to view in Cloud Cost Management，再点击 Save。BillingAdmin 角色仅授权访问 Confluent 的账单 API；如果未启用此开关，Datadog 不会收集任何成本数据。在报告 Cloud 成本管理已配置之前，请用户确认该开关已启用。

告知用户：

成本数据将在约 24 小时内显示在 Cloud 成本管理中。
链接：<APP_BASE>/cost

第 5 阶段：验证设置

第 3 阶段的 201 响应确认集成已配置；此步骤用于检查您注册的资源是否确实观测到遥测数据。稍等片刻，等待指标开始流入，然后针对每个已注册资源查询一个指标，并将查询范围限定为其 resource_id，以免同一组织中的其他 Confluent 集成产生误报。将 RESOURCES 设置为已注册资源，格式为 <RESOURCE_TYPE>:<RESOURCE_ID> 对：

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
DD_SITE='datadoghq.com'; export DD_SITE   # <- replace with the site confirmed in Phase 0.
# Explicit assignment, not ':=': a wrong non-empty DD_SITE in .env would otherwise survive.
: "${DD_API_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"; : "${DD_APP_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"
RESOURCES='<RESOURCE_TYPE>:<RESOURCE_ID> <RESOURCE_TYPE>:<RESOURCE_ID>'   # <- e.g. 'kafka:lkc-abc123 connector:lcc-ghi789'
for r in $RESOURCES; do
  type=${r%%:*}; id=${r#*:}
  case "$type" in
    kafka) metrics=confluent_cloud.kafka.received_bytes ;;
    # Source connectors report sent_records, sink connectors received_records; accept either.
    connector) metrics="confluent_cloud.connect.sent_records confluent_cloud.connect.received_records" ;;
    ksql) metrics=confluent_cloud.ksql.streaming_unit_count ;;
    schema_registry) metrics=confluent_cloud.schema_registry.schema_count ;;
    *) echo "$r: unknown resource type"; continue ;;
  esac
  for metric in $metrics; do
    resp=$(printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY" \
      | curl -H @- -sS -w '\n%{http_code}' -G "https://api.${DD_SITE}/api/v1/query" \
          --data-urlencode "from=$(($(date +%s) - 900))" \
          --data-urlencode "to=$(date +%s)" \
          --data-urlencode "query=avg:${metric}{resource_id:${id}}")
    code=$(printf '%s' "$resp" | tail -1)
    [ "$code" = "200" ] || { echo "$r ($metric): query failed with HTTP $code"; continue; }
    echo "$r ($metric): $(printf '%s' "$resp" | sed '$d' | jq '.series | length') series"
  done
done

如果某个资源有非空序列，说明该资源的指标正在流入。向用户表示祝贺。

如果某个资源尚无序列，请告知用户指标仍在传输中——Confluent Cloud 指标通常会在几分钟内到达，处于空闲状态的连接器或 ksqlDB 集群可能不会报告太多数据。建议稍后再查看，并将该资源报告为已配置但尚未观测到数据。

报告完成状态

完成 verify 步骤后，向入门工作流 API 发送完成状态，以便界面将 Agent 工作流标记为完成。DD_WORKFLOW_ID 是这些凭据所属的 Datadog 用户的 UUID，界面会据此关联此回调。请在一条命令中解析并发送：许多 Agent 运行时会为每条命令启动一个新的 Shell，因此在此前命令中赋值的变量在这里会为空。

切勿打印 current_user 响应：其中还包含用户的姓名、用户名和电子邮件地址。如果没有可用的 JSON 解析器，请跳过此回调，不要安装解析器，也不要因此阻塞设置流程。

for f in .env.local .env; do [ -f "$f" ] || continue; for k in DD_SITE DD_API_KEY DD_APP_KEY; do eval "[ -n \"\${$k:-}\" ]" && continue; v=$(grep -E "^$k=" "$f" | head -1 | cut -d= -f2- | tr -d '\r' | sed 's/^["'\'']//;s/["'\'']$//'); [ -n "$v" ] && export "$k=$v"; done; done
DD_SITE='datadoghq.com'; export DD_SITE   # <- replace with the site confirmed in Phase 0.
# Explicit assignment, not ':=': a wrong non-empty DD_SITE in .env would otherwise survive.
: "${DD_API_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"; : "${DD_APP_KEY:?not set - run dd-account-setup (commercial sites) or supply it directly (government sites)}"
CONFLUENT_ACCOUNT_ID='<CONFLUENT_ACCOUNT_ID>'   # <- the account id from Phase 1 or Phase 2
: "${CONFLUENT_ACCOUNT_ID:?not set - substitute the Confluent account id from Phase 1 or Phase 2}"
dd_headers() { printf 'DD-API-KEY: %s\nDD-APPLICATION-KEY: %s\n' "$DD_API_KEY" "$DD_APP_KEY"; }
DD_WORKFLOW_ID="$(dd_headers | curl -H @- -fsS --connect-timeout 5 --max-time 20 \
  "https://api.${DD_SITE}/api/v2/current_user" | jq -er '.data.id')" \
  && dd_headers | curl -H @- -sS --connect-timeout 5 --max-time 20 -X POST \
  "https://api.${DD_SITE}/api/unstable/integration/confluent/workflow/agentic_saas_integration_setup" \
  -H "Content-Type: application/vnd.api+json" \
  -d '{
    "data": {
      "type": "integration_setup_status",
      "id": "'"${DD_WORKFLOW_ID}"'",
      "attributes": {
        "status": "finished",
        "step": "setup_complete",
        "metadata": {
          "confluent_account_id": "'"${CONFLUENT_ACCOUNT_ID}"'"
        }
      }
    }
  }'

如果此回调失败，请勿重试，也不要阻塞设置流程。

向用户简要总结已完成的设置：

Datadog 中的 Confluent Cloud 账户（CONFLUENT_ACCOUNT_ID，以及设置过的显示名称），并说明该账户是新注册的还是已存在。
已注册的资源，按类型和 ID 列出，包括已注册的资源（409）以及跳过的资源（例如密钥无法列出连接器时跳过的连接器），并说明哪些 Kafka 集群启用了 enable_custom_metrics。
已配置的可选增强功能（Schema Registry 标签、Cloud Cost Management）。
上述指标检查的结果。切勿重复提供 Confluent API 密钥或 Schema Registry 密钥。

账户注册后，Datadog 会保存凭据，因此不再需要 ~/.datadog-confluent.env。询问用户是否要删除它（rm -f ~/.datadog-confluent.env）；只有在用户计划近期重新运行资源发现时才保留。

然后提供以下链接（根据 Phase 0 App URL 表构造为 <APP_BASE>/...）：

Confluent Cloud Integration 磁贴：<APP_BASE>/integrations/confluent-cloud - 确认集成处于启用状态，并查看开箱即用的仪表板。
Metrics Explorer：<APP_BASE>/metric/explorer?exp_metric=confluent_cloud.kafka.received_bytes - 确认 Kafka 指标正在流入。
Cloud Cost Management（如果已启用）：<APP_BASE>/cost - Confluent 费用将在约 24 小时内显示。

建议的后续步骤：

使用 confluent_cloud.kafka.consumer_lag_offsets 在 <APP_BASE>/monitors/create 创建 Kafka 消费者延迟监控器，以便在处理积压时发出告警。
创建一个监控器，监控 confluent_cloud.kafka.cluster_load_percent，以跟踪集群利用率。
如果添加了 Schema Registry 密钥，请在 Metrics Explorer 中确认指标上显示了 Confluent 定义的标签。

重要说明

Confluent Cloud API 密钥必须具有 Cloud resource management 权限范围，才能收集指标。集群级密钥不可用。
BillingAdmin 角色是独立的，仅 Cloud Cost Management 成本数据需要此角色。
资源 ID 区分大小写，必须与 Confluent Cloud 中显示的内容完全一致。
Kafka 集群上的 enable_custom_metrics 会启用 custom.consumer_lag_offset，并附带额外的分区级标签；这可能会增加自定义指标量。
Schema Registry 凭据（sr_api_key、sr_api_secret）可在创建账户时通过 POST 设置，也可稍后通过 PATCH 添加。POST 要求两个字段同时提供或都不提供；PATCH 支持仅轮换密钥，但要求现有密钥已存在。
如果用户之后想监控其他资源，可以使用 GET /accounts 返回的现有 CONFLUENT_ACCOUNT_ID 重新运行 Phase 3。

支持 Markdown

登录后评论

基本信息

作者: datadog-labs
仓库: datadog-labs/agent-skills
分类: 云与 DevOps
安装次数: 12
GitHub Stars: 183

报告问题

#	类型	名称	ID	环境
1	Kafka 集群	prod-cluster	lkc-abc123	Production (env-xyz)
2	Kafka 集群	staging-cluster	lkc-def456	Staging (env-uvw)
3	连接器	s3-sink	lcc-ghi789	Production (env-xyz)