From dd7f028a5bc65fde24d3b3fb22d64bf2f2bc5d5e Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 10 Jul 2026 01:53:53 +0200 Subject: [PATCH] feat: add bounded kafka group cleanup --- .../unidesk-cicd/references/platform-ops.md | 4 +- config/platform-infra/kafka.yaml | 16 ++ docs/reference/platform-infra.md | 4 +- .../platform-infra/kafka-consumer-groups.py | 212 ++++++++++++++++++ scripts/src/platform-infra-kafka.ts | 194 ++++++++++++++++ scripts/src/platform-infra/entry.ts | 1 + 6 files changed, 429 insertions(+), 2 deletions(-) create mode 100644 scripts/native/platform-infra/kafka-consumer-groups.py diff --git a/.agents/skills/unidesk-cicd/references/platform-ops.md b/.agents/skills/unidesk-cicd/references/platform-ops.md index 136eddc2..9b52d92f 100644 --- a/.agents/skills/unidesk-cicd/references/platform-ops.md +++ b/.agents/skills/unidesk-cicd/references/platform-ops.md @@ -42,11 +42,13 @@ bun scripts/cli.ts platform-infra sub2api plan|apply|status|validate bun scripts/cli.ts platform-infra sub2api codex-pool plan|sync|validate|expose|configure-local bun scripts/cli.ts platform-infra gitea plan|apply|status|validate|mirror --target JD01 bun scripts/cli.ts platform-infra pipelines-as-code plan|apply|status|closeout|history|webhook-test --target JD01 [--consumer ] +bun scripts/cli.ts platform-infra kafka groups --node [--limit N|--full|--raw] +bun scripts/cli.ts platform-infra kafka groups cleanup --node --policy [--confirm] bun scripts/cli.ts platform-infra wechat-archive plan|apply|status|validate|pull bun scripts/cli.ts platform-infra wechat-archive wcf-host-status|collector-plan|collector-apply|collector-status ``` -`platform-infra` 是 UniDesk 运维的平台基础设施控制面；新增平台服务优先进入该命名空间或对应 YAML 声明目标，旧 `devops-infra` 只作为渐进迁移来源。Sub2API 日常部署、Codex pool、FRP 暴露、master `~/.codex` 配置、验收和排障统一使用 `$unidesk-sub2api`。Gitea mirror 和 Pipelines-as-Code 是 JD01 migrated CI source/trigger 平台服务，source-of-truth 分别是 `config/platform-infra/gitea.yaml` 和 `config/platform-infra/pipelines-as-code.yaml`；正式架构和三 consumer 矩阵见 [gitea-pac.md](gitea-pac.md)。PaC closeout 是 migrated lane 正式交付入口，PaC status 是当前状态入口，PaC history 是触发/耗时/env reuse 审计入口，默认输出必须包含 `READ_ERRORS` 并在目标 node 聚合，不能把读取失败或 namespace 过大误报为空表成功。不用 Gitea Actions、act_runner、branch-follower、自维护脚本或 consumer 专用手动 publish 兜底。`agentrun-jd01-v02` 是默认 consumer；哨兵使用 `--consumer sentinel-jd01-v03`，HWLAB 使用 `--consumer hwlab-jd01-v03` 查看 PaC/Tekton/Argo/env reuse 证据。WeChat archive 是 platform-infra 的 YAML-first 工作流入口；只读 collector 的副本、镜像、WCF host、端口和版本 pin 都以 YAML 为准。 +`platform-infra` 是 UniDesk 运维的平台基础设施控制面；新增平台服务优先进入该命名空间或对应 YAML 声明目标，旧 `devops-infra` 只作为渐进迁移来源。Sub2API 日常部署、Codex pool、FRP 暴露、master `~/.codex` 配置、验收和排障统一使用 `$unidesk-sub2api`。Kafka groups cleanup 默认只生成 plan，policy 与单次上限来自 `config/platform-infra/kafka.yaml`；长期安全边界以 [Platform Infrastructure](../../../../docs/reference/platform-infra.md#kafka-event-bus-boundary) 为唯一权威。Gitea mirror 和 Pipelines-as-Code 是 JD01 migrated CI source/trigger 平台服务，source-of-truth 分别是 `config/platform-infra/gitea.yaml` 和 `config/platform-infra/pipelines-as-code.yaml`；正式架构和三 consumer 矩阵见 [gitea-pac.md](gitea-pac.md)。PaC closeout 是 migrated lane 正式交付入口，PaC status 是当前状态入口，PaC history 是触发/耗时/env reuse 审计入口，默认输出必须包含 `READ_ERRORS` 并在目标 node 聚合，不能把读取失败或 namespace 过大误报为空表成功。不用 Gitea Actions、act_runner、branch-follower、自维护脚本或 consumer 专用手动 publish 兜底。`agentrun-jd01-v02` 是默认 consumer；哨兵使用 `--consumer sentinel-jd01-v03`，HWLAB 使用 `--consumer hwlab-jd01-v03` 查看 PaC/Tekton/Argo/env reuse 证据。WeChat archive 是 platform-infra 的 YAML-first 工作流入口；只读 collector 的副本、镜像、WCF host、端口和版本 pin 都以 YAML 为准。 ## CI Tools Image diff --git a/config/platform-infra/kafka.yaml b/config/platform-infra/kafka.yaml index 64430a46..c76e8a31 100644 --- a/config/platform-infra/kafka.yaml +++ b/config/platform-infra/kafka.yaml @@ -162,5 +162,21 @@ management: maxTailLimit: 50 defaultGroupListLimit: 25 maxGroupListLimit: 200 + consumerGroupCleanup: + defaultCandidateLimit: 25 + maxCandidateLimit: 200 + commandTimeoutSeconds: 30 + deleteBatchSize: 50 + policies: + - id: hwlab-cloud-api-sse + prefix: hwlab-v03-cloud-api-sse- + inactiveStates: + - Empty + minimumGroupAgeHours: 1 + timestamp: + source: group-name-regex + pattern: ^hwlab-v03-cloud-api-sse-([0-9]{13})-[0-9a-f]{8}$ + captureGroup: 1 + unit: unix-ms defaultShadowTopic: hwlab.event.v1 shadowProducerId: unidesk-platform-infra-kafka-cli diff --git a/docs/reference/platform-infra.md b/docs/reference/platform-infra.md index ece6f4f7..e13fa278 100644 --- a/docs/reference/platform-infra.md +++ b/docs/reference/platform-infra.md @@ -60,7 +60,9 @@ - Kafka for the HWLAB v0.3 / AgentRun v0.2 event-bus POC is a target-scoped UniDesk-operated platform service in namespace `platform-infra`; NC01 is the current validation target for pikasTech/HWLAB#2449. It is not owned by `hwlab-v03`, `agentrun-v02`, a per-lane Kafka namespace, or a service repository deployment file. - The canonical source of truth is `config/platform-infra/kafka.yaml`; target, namespace, Strimzi release URL, cluster name, storage class/size, topic list, client declarations, DLQ names, runtime switch and validation smoke topic must stay in that YAML. Current version numbers and retention values belong only in YAML, not in this reference. -- The canonical entrypoint is `bun scripts/cli.ts platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|produce --target `; `--node ` is an equivalent selector for node-targeted operations. Formal mutation must use that path; raw `kubectl` is bounded diagnosis only. +- The canonical entrypoint is `bun scripts/cli.ts platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|produce --target `; `--node ` is an equivalent selector for node-targeted operations. Consumer-group retirement uses `platform-infra kafka groups cleanup --node --policy [--confirm]`. Formal mutation must use that path; raw `kubectl` is bounded diagnosis only. +- `platform-infra kafka groups` renders a YAML-bounded real group table plus total/shown/omitted counts by default; `--limit` may raise the view only to the YAML maximum, while `--full` and `--raw` are explicit drill-down paths. +- Consumer-group cleanup is plan-only unless `--confirm` is present and accepts only a policy declared under `config/platform-infra/kafka.yaml#management.consumerGroupCleanup.policies`. A group must match that policy's exact prefix and full name pattern, decode a group-name timestamp old enough for the YAML minimum, and be in a YAML-allowed inactive Kafka state. The candidate limit is also the per-command mutation ceiling; `--full` and `--raw` never expand it. Confirmed execution re-reads broker state after deletion and reports deleted, state-changed and failed groups separately. - HWLAB v0.3 and AgentRun v0.2 are client namespaces. Runtime readiness alone does not prove Workbench projection, SSE or AgentRun command ingestion has migrated. App producer/consumer switchover is owned by the corresponding service repo and must be verified from that service's original entrypoint, not by Kafka topic readiness alone. - The event-bus authority split is three topics: `codex-stdio.raw.v1` stores AgentRun runner's raw Codex stdio frames; `agentrun.event.v1` stores AgentRun manager durable event mirrors before HWLAB conversion; `hwlab.event.v1` stores HWLAB event stream after HWLAB-side conversion and is the stream that SSE should proxy to Workbench in the event-bus design. Do not mix these schemas into a single topic or let a downstream consumer rewrite the upstream source topic. - Shadow produce may write Kafka events for observation only when YAML keeps consumer cutover disabled; it must not replace the current business read path until the corresponding service has an explicit consumer switchover decision and replay/refresh semantics. diff --git a/scripts/native/platform-infra/kafka-consumer-groups.py b/scripts/native/platform-infra/kafka-consumer-groups.py new file mode 100644 index 00000000..9ab17026 --- /dev/null +++ b/scripts/native/platform-infra/kafka-consumer-groups.py @@ -0,0 +1,212 @@ +import base64 +import datetime +import json +import os +import re +import subprocess +import sys +import time + + +def required(name): + value = os.environ.get(name, "").strip() + if not value: + raise RuntimeError(f"missing required environment variable {name}") + return value + + +def run(command, timeout_seconds): + return subprocess.run( + command, + check=False, + capture_output=True, + text=True, + timeout=timeout_seconds, + ) + + +def stderr_tail(result, limit=1200): + return (result.stderr or "")[-limit:] + + +def list_group_states(kafka_command, timeout_seconds): + result = run(kafka_command + ["--list", "--state"], timeout_seconds) + if result.returncode != 0: + raise RuntimeError(f"Kafka group state listing failed: {stderr_tail(result)}") + states = {} + header_seen = False + for raw_line in result.stdout.splitlines(): + fields = raw_line.split() + if len(fields) >= 2 and fields[0] == "GROUP" and fields[1] == "STATE": + header_seen = True + continue + if header_seen and len(fields) >= 2: + states[fields[0]] = fields[1] + if not header_seen: + raise RuntimeError("Kafka group state listing did not include the GROUP STATE header") + return states, stderr_tail(result) + + +def iso_time(timestamp_ms): + return datetime.datetime.fromtimestamp(timestamp_ms / 1000, datetime.timezone.utc).isoformat().replace("+00:00", "Z") + + +def classify(states, policy, now_ms): + pattern = re.compile(policy["timestamp"]["pattern"]) + capture_group = int(policy["timestamp"]["captureGroup"]) + minimum_age_ms = int(policy["minimumGroupAgeHours"]) * 60 * 60 * 1000 + inactive_states = set(policy["inactiveStates"]) + prefix_matched = [] + pattern_matched = [] + inactive_matched = [] + age_eligible = [] + candidates = [] + invalid_timestamp_count = 0 + for name, state in states.items(): + if not name.startswith(policy["prefix"]): + continue + prefix_matched.append(name) + match = pattern.fullmatch(name) + if match is None: + continue + pattern_matched.append(name) + try: + created_ms = int(match.group(capture_group)) + except (IndexError, TypeError, ValueError): + invalid_timestamp_count += 1 + continue + age_ms = now_ms - created_ms + if age_ms < 0: + invalid_timestamp_count += 1 + continue + if state in inactive_states: + inactive_matched.append(name) + if age_ms >= minimum_age_ms: + age_eligible.append(name) + if state in inactive_states and age_ms >= minimum_age_ms: + candidates.append({ + "name": name, + "state": state, + "createdAt": iso_time(created_ms), + "ageHours": round(age_ms / 3600000, 3), + }) + candidates.sort(key=lambda item: (-item["ageHours"], item["name"])) + return { + "prefixMatchedCount": len(prefix_matched), + "patternMatchedCount": len(pattern_matched), + "inactiveMatchedCount": len(inactive_matched), + "ageEligibleCount": len(age_eligible), + "invalidTimestampCount": invalid_timestamp_count, + "candidates": candidates, + } + + +def delete_candidates(kafka_command, candidates, batch_size, timeout_seconds): + batch_evidence = [] + for start in range(0, len(candidates), batch_size): + batch = candidates[start:start + batch_size] + command = kafka_command + ["--delete"] + for candidate in batch: + command.extend(["--group", candidate["name"]]) + result = run(command, timeout_seconds) + batch_evidence.append({ + "size": len(batch), + "exitCode": result.returncode, + "stderrTail": stderr_tail(result), + }) + return batch_evidence + + +def main(): + target = required("KAFKA_TARGET") + namespace = required("KAFKA_NAMESPACE") + cluster = required("KAFKA_CLUSTER") + bootstrap = required("KAFKA_BOOTSTRAP") + confirm = required("KAFKA_GROUP_CONFIRM") == "true" + full = required("KAFKA_FULL") == "true" + candidate_limit = int(required("KAFKA_CANDIDATE_LIMIT")) + timeout_seconds = int(required("KAFKA_COMMAND_TIMEOUT_SECONDS")) + batch_size = int(required("KAFKA_DELETE_BATCH_SIZE")) + policy = json.loads(base64.b64decode(required("KAFKA_POLICY_B64")).decode("utf-8")) + pod_result = run([ + "kubectl", "-n", namespace, "get", "pod", + "-l", f"strimzi.io/cluster={cluster},strimzi.io/kind=Kafka", + "-o", "jsonpath={.items[0].metadata.name}", + ], timeout_seconds) + pod = pod_result.stdout.strip() + if pod_result.returncode != 0 or not pod: + raise RuntimeError(f"Kafka broker pod not found: {stderr_tail(pod_result)}") + kafka_command = [ + "kubectl", "-n", namespace, "exec", pod, "--", + "bin/kafka-consumer-groups.sh", "--bootstrap-server", bootstrap, + ] + observed_at_ms = int(time.time() * 1000) + before_states, list_stderr = list_group_states(kafka_command, timeout_seconds) + selection = classify(before_states, policy, observed_at_ms) + eligible_candidates = selection.pop("candidates") + selected_candidates = eligible_candidates[:candidate_limit] + batch_evidence = [] + deletions = [] + after_states = before_states + if confirm and selected_candidates: + batch_evidence = delete_candidates(kafka_command, selected_candidates, batch_size, timeout_seconds) + after_states, _ = list_group_states(kafka_command, timeout_seconds) + inactive_states = set(policy["inactiveStates"]) + for candidate in selected_candidates: + name = candidate["name"] + state_after = after_states.get(name) + if state_after is None: + status = "deleted" + elif state_after not in inactive_states: + status = "state-changed" + else: + status = "failed" + deletions.append({"name": name, "status": status, "stateAfter": state_after}) + display_deletions = deletions if full else deletions[:candidate_limit] + deleted_count = sum(1 for item in deletions if item["status"] == "deleted") + state_changed_count = sum(1 for item in deletions if item["status"] == "state-changed") + failed_count = sum(1 for item in deletions if item["status"] == "failed") + payload = { + "ok": failed_count == 0, + "target": target, + "namespace": namespace, + "cluster": cluster, + "pod": pod, + "mode": "confirmed" if confirm else "plan", + "mutation": confirm, + "observedAt": iso_time(observed_at_ms), + "policy": policy, + "summary": { + "totalGroupCount": len(before_states), + **selection, + "eligibleCount": len(eligible_candidates), + "selectedCount": len(selected_candidates), + "omittedCount": max(0, len(eligible_candidates) - len(selected_candidates)), + "deleteAttemptedCount": len(deletions), + "deletedCount": deleted_count, + "stateChangedCount": state_changed_count, + "failedCount": failed_count, + }, + "candidates": selected_candidates, + "eligibleCandidates": eligible_candidates if full else [], + "deletions": display_deletions, + "batchEvidence": batch_evidence if full else [], + "stderrTail": list_stderr, + "valuesPrinted": False, + } + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 0 if payload["ok"] else 1 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except Exception as error: + confirmed = os.environ.get("KAFKA_GROUP_CONFIRM", "").strip() == "true" + print(json.dumps({ + "ok": False, + "error": str(error), + "mutation": confirmed, + "mutationOutcome": "unknown" if confirmed else "not-attempted", + }, ensure_ascii=False)) + sys.exit(1) diff --git a/scripts/src/platform-infra-kafka.ts b/scripts/src/platform-infra-kafka.ts index 991a4d8b..7369dcf2 100644 --- a/scripts/src/platform-infra-kafka.ts +++ b/scripts/src/platform-infra-kafka.ts @@ -1,3 +1,4 @@ +import { readFileSync } from "node:fs"; import type { UniDeskConfig } from "./config"; import { rootPath } from "./config"; import type { RenderedCliResult } from "./output"; @@ -15,6 +16,7 @@ const configFile = rootPath("config", "platform-infra", "kafka.yaml"); const configLabel = "config/platform-infra/kafka.yaml"; const fieldManager = "unidesk-platform-infra-kafka"; const y = createYamlFieldReader(configLabel); +const kafkaConsumerGroupsNativeScript = readFileSync(rootPath("scripts", "native", "platform-infra", "kafka-consumer-groups.py"), "utf8"); interface KafkaTarget { id: string; @@ -46,6 +48,19 @@ interface KafkaClientSpec { dlqTopics: string[]; } +interface KafkaGroupCleanupPolicy { + id: string; + prefix: string; + inactiveStates: Array<"Empty" | "Dead">; + minimumGroupAgeHours: number; + timestamp: { + source: "group-name-regex"; + pattern: string; + captureGroup: number; + unit: "unix-ms"; + }; +} + interface PlatformKafkaConfig { version: number; kind: "platform-infra-kafka"; @@ -105,6 +120,13 @@ interface PlatformKafkaConfig { maxTailLimit: number; defaultGroupListLimit: number; maxGroupListLimit: number; + consumerGroupCleanup: { + defaultCandidateLimit: number; + maxCandidateLimit: number; + commandTimeoutSeconds: number; + deleteBatchSize: number; + policies: KafkaGroupCleanupPolicy[]; + }; defaultShadowTopic: string; shadowProducerId: string; }; @@ -136,6 +158,12 @@ interface KafkaProduceOptions extends CommonOptions { payload: string | null; } +interface KafkaGroupCleanupOptions extends CommonOptions { + policyId: string; + confirm: boolean; + limit: number | null; +} + export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args: string[]): Promise | RenderedCliResult> { const [action = "plan"] = args; if (action === "plan") { @@ -156,6 +184,11 @@ export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args: return options.full || options.raw ? result : renderTopics(result); } if (action === "groups") { + if (args[1] === "cleanup") { + const options = parseGroupCleanupOptions(args.slice(2)); + const result = await cleanupGroups(config, options); + return options.full || options.raw ? result : renderGroupCleanup(result); + } const options = parseInspectOptions(args.slice(1)); const result = await groups(config, options); return options.full || options.raw ? result : renderGroups(result); @@ -192,6 +225,7 @@ export function kafkaHelp(): Record { "bun scripts/cli.ts platform-infra kafka validate --target NC01 [--full|--raw]", "bun scripts/cli.ts platform-infra kafka topics --node NC01", "bun scripts/cli.ts platform-infra kafka groups --node NC01 [--limit N] [--full|--raw]", + "bun scripts/cli.ts platform-infra kafka groups cleanup --node NC01 --policy hwlab-cloud-api-sse [--confirm]", "bun scripts/cli.ts platform-infra kafka offsets --node NC01 [--topic hwlab.event.v1]", "bun scripts/cli.ts platform-infra kafka tail --node NC01 --topic hwlab.event.v1 --limit 5", "bun scripts/cli.ts platform-infra kafka produce --node NC01 --topic hwlab.event.v1 --key ", @@ -316,16 +350,54 @@ function parseClient(record: Record, index: number): KafkaClien function parseManagement(root: Record): PlatformKafkaConfig["management"] { const management = y.objectField(root, "management", ""); + const cleanup = y.objectField(management, "consumerGroupCleanup", "management"); return { defaultTailLimit: positiveInteger(management, "defaultTailLimit", "management"), maxTailLimit: positiveInteger(management, "maxTailLimit", "management"), defaultGroupListLimit: positiveInteger(management, "defaultGroupListLimit", "management"), maxGroupListLimit: positiveInteger(management, "maxGroupListLimit", "management"), + consumerGroupCleanup: { + defaultCandidateLimit: positiveInteger(cleanup, "defaultCandidateLimit", "management.consumerGroupCleanup"), + maxCandidateLimit: positiveInteger(cleanup, "maxCandidateLimit", "management.consumerGroupCleanup"), + commandTimeoutSeconds: positiveInteger(cleanup, "commandTimeoutSeconds", "management.consumerGroupCleanup"), + deleteBatchSize: positiveInteger(cleanup, "deleteBatchSize", "management.consumerGroupCleanup"), + policies: y.arrayOfRecords(cleanup.policies, "management.consumerGroupCleanup.policies").map(parseGroupCleanupPolicy), + }, defaultShadowTopic: topicNameField(management, "defaultShadowTopic", "management"), shadowProducerId: y.stringField(management, "shadowProducerId", "management"), }; } +function parseGroupCleanupPolicy(record: Record, index: number): KafkaGroupCleanupPolicy { + const path = `management.consumerGroupCleanup.policies[${index}]`; + const timestamp = y.objectField(record, "timestamp", path); + const prefix = y.stringField(record, "prefix", path); + if (!/^[A-Za-z0-9._-]+$/u.test(prefix) || !prefix.endsWith("-")) throw new Error(`${configLabel}.${path}.prefix must be a Kafka-safe, delimiter-terminated prefix`); + const inactiveStates = y.stringArrayField(record, "inactiveStates", path).map((value) => { + if (value !== "Empty" && value !== "Dead") throw new Error(`${configLabel}.${path}.inactiveStates only accepts Empty or Dead`); + return value; + }); + if (inactiveStates.length === 0) throw new Error(`${configLabel}.${path}.inactiveStates must not be empty`); + const pattern = y.stringField(timestamp, "pattern", `${path}.timestamp`); + try { + new RegExp(pattern, "u"); + } catch (error) { + throw new Error(`${configLabel}.${path}.timestamp.pattern is invalid: ${error instanceof Error ? error.message : String(error)}`); + } + return { + id: y.stringField(record, "id", path), + prefix, + inactiveStates, + minimumGroupAgeHours: positiveInteger(record, "minimumGroupAgeHours", path), + timestamp: { + source: y.enumField(timestamp, "source", `${path}.timestamp`, ["group-name-regex"] as const), + pattern, + captureGroup: positiveInteger(timestamp, "captureGroup", `${path}.timestamp`), + unit: y.enumField(timestamp, "unit", `${path}.timestamp`, ["unix-ms"] as const), + }, + }; +} + function validateConfig(kafka: PlatformKafkaConfig): void { resolveTarget(kafka, kafka.defaults.targetId); if (!/^https:\/\/github\.com\/strimzi\/strimzi-kafka-operator\/releases\/download\//u.test(kafka.operator.manifestUrl)) { @@ -337,6 +409,15 @@ function validateConfig(kafka: PlatformKafkaConfig): void { if (!kafka.topics.some((topic) => topic.name === kafka.management.defaultShadowTopic)) throw new Error(`${configLabel}.management.defaultShadowTopic must reference one of topics[].name`); if (kafka.management.defaultTailLimit > kafka.management.maxTailLimit) throw new Error(`${configLabel}.management.defaultTailLimit must be <= management.maxTailLimit`); if (kafka.management.defaultGroupListLimit > kafka.management.maxGroupListLimit) throw new Error(`${configLabel}.management.defaultGroupListLimit must be <= management.maxGroupListLimit`); + if (kafka.management.consumerGroupCleanup.defaultCandidateLimit > kafka.management.consumerGroupCleanup.maxCandidateLimit) throw new Error(`${configLabel}.management.consumerGroupCleanup.defaultCandidateLimit must be <= maxCandidateLimit`); + const cleanupPolicyIds = new Set(); + const cleanupPrefixes = new Set(); + for (const policy of kafka.management.consumerGroupCleanup.policies) { + if (cleanupPolicyIds.has(policy.id)) throw new Error(`${configLabel}.management.consumerGroupCleanup.policies contains duplicate id ${policy.id}`); + if (cleanupPrefixes.has(policy.prefix)) throw new Error(`${configLabel}.management.consumerGroupCleanup.policies contains duplicate prefix ${policy.prefix}`); + cleanupPolicyIds.add(policy.id); + cleanupPrefixes.add(policy.prefix); + } if (kafka.defaults.switch.shadowConsumeEnabled) throw new Error(`${configLabel}.defaults.switch.shadowConsumeEnabled must stay false for shadow-produce-only stage`); const topicNames = new Set(); for (const topic of kafka.topics) { @@ -500,6 +581,49 @@ async function groups(config: UniDeskConfig, options: KafkaInspectOptions): Prom }; } +async function cleanupGroups(config: UniDeskConfig, options: KafkaGroupCleanupOptions): Promise> { + const kafka = readKafkaConfig(); + const target = resolveTarget(kafka, options.targetId); + const policy = kafka.management.consumerGroupCleanup.policies.find((item) => item.id === options.policyId); + if (policy === undefined) throw new Error(`unknown Kafka consumer group cleanup policy ${options.policyId}; configured policies: ${kafka.management.consumerGroupCleanup.policies.map((item) => item.id).join(", ")}`); + const candidateLimit = boundedGroupCleanupLimit(kafka, options.limit); + const runtimeEnvironment = { + KAFKA_GROUP_CONFIRM: options.confirm ? "true" : "false", + KAFKA_TARGET: target.id, + KAFKA_NAMESPACE: target.namespace, + KAFKA_CLUSTER: kafka.cluster.name, + KAFKA_BOOTSTRAP: brokerBootstrap(kafka), + KAFKA_POLICY_B64: Buffer.from(JSON.stringify(policy), "utf8").toString("base64"), + KAFKA_CANDIDATE_LIMIT: String(candidateLimit), + KAFKA_FULL: options.full || options.raw ? "true" : "false", + KAFKA_COMMAND_TIMEOUT_SECONDS: String(kafka.management.consumerGroupCleanup.commandTimeoutSeconds), + KAFKA_DELETE_BATCH_SIZE: String(kafka.management.consumerGroupCleanup.deleteBatchSize), + }; + const remoteScript = [ + ...Object.entries(runtimeEnvironment).map(([key, value]) => `export ${key}=${shQuote(value)}`), + "python3 - <<'PY'", + kafkaConsumerGroupsNativeScript.trimEnd(), + "PY", + ].join("\n"); + const result = await capture(config, target.route, ["sh"], remoteScript); + const parsed = parseJsonOutput(result.stdout); + return { + ok: result.exitCode === 0 && parsed?.ok === true, + action: "platform-infra-kafka-groups-cleanup", + mode: options.confirm ? "confirmed" : "plan", + mutation: options.confirm, + target: targetSummary(target), + config: { configPath: configLabel, policyId: policy.id, valuesPrinted: false }, + cleanup: parsed ?? null, + remote: options.raw && parsed !== null ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }), + next: { + plan: `bun scripts/cli.ts platform-infra kafka groups cleanup --node ${target.id} --policy ${policy.id}`, + confirm: `bun scripts/cli.ts platform-infra kafka groups cleanup --node ${target.id} --policy ${policy.id} --confirm`, + groups: `bun scripts/cli.ts platform-infra kafka groups --node ${target.id}`, + }, + }; +} + async function offsets(config: UniDeskConfig, options: KafkaInspectOptions): Promise> { const kafka = readKafkaConfig(); const target = resolveTarget(kafka, options.targetId); @@ -1424,6 +1548,7 @@ function compactConfigSummary(kafka: PlatformKafkaConfig, target: KafkaTarget): maxTailLimit: kafka.management.maxTailLimit, defaultGroupListLimit: kafka.management.defaultGroupListLimit, maxGroupListLimit: kafka.management.maxGroupListLimit, + consumerGroupCleanup: kafka.management.consumerGroupCleanup, shadowProducerId: kafka.management.shadowProducerId, }, valuesPrinted: false, @@ -1643,6 +1768,37 @@ function renderGroups(result: Record): RenderedCliResult { ]); } +function renderGroupCleanup(result: Record): RenderedCliResult { + const cleanup = record(result.cleanup); + const policy = record(cleanup.policy); + const summary = record(cleanup.summary); + const candidates = arrayRecords(cleanup.candidates).map((item) => [stringValue(item.name), stringValue(item.state), stringValue(item.ageHours), stringValue(item.createdAt)]); + const deletions = arrayRecords(cleanup.deletions).map((item) => [stringValue(item.name), stringValue(item.status), stringValue(item.stateAfter)]); + const next = record(result.next); + return rendered(result, "platform-infra kafka groups cleanup", [ + "PLATFORM-INFRA KAFKA GROUP CLEANUP", + ...table(["TARGET", "NAMESPACE", "CLUSTER", "POD", "MODE"], [[stringValue(cleanup.target), stringValue(cleanup.namespace), stringValue(cleanup.cluster), stringValue(cleanup.pod), stringValue(result.mode)]]), + "", + "POLICY", + ...table(["ID", "PREFIX", "INACTIVE", "MIN_AGE_HOURS"], [[stringValue(policy.id), stringValue(policy.prefix), Array.isArray(policy.inactiveStates) ? policy.inactiveStates.join(",") : "-", stringValue(policy.minimumGroupAgeHours)]]), + "", + "SELECTION", + ...table(["TOTAL", "PREFIX", "PATTERN", "INACTIVE", "AGE_OK", "ELIGIBLE", "SELECTED", "OMITTED"], [[stringValue(summary.totalGroupCount, "0"), stringValue(summary.prefixMatchedCount, "0"), stringValue(summary.patternMatchedCount, "0"), stringValue(summary.inactiveMatchedCount, "0"), stringValue(summary.ageEligibleCount, "0"), stringValue(summary.eligibleCount, "0"), stringValue(summary.selectedCount, "0"), stringValue(summary.omittedCount, "0")]]), + "", + ...(candidates.length === 0 ? ["CANDIDATES -"] : table(["GROUP", "STATE", "AGE_HOURS", "CREATED_AT"], candidates)), + ...(result.mode === "confirmed" ? [ + "", + "RESULT", + ...table(["ATTEMPTED", "DELETED", "STATE_CHANGED", "FAILED"], [[stringValue(summary.deleteAttemptedCount, "0"), stringValue(summary.deletedCount, "0"), stringValue(summary.stateChangedCount, "0"), stringValue(summary.failedCount, "0")]]), + ...(deletions.length === 0 ? [] : ["", ...table(["GROUP", "STATUS", "STATE_AFTER"], deletions)]), + ] : []), + ...remoteErrorLines(result), + "", + `NEXT ${result.mode === "confirmed" ? stringValue(next.groups) : stringValue(next.confirm)}`, + "Boundary: only the YAML-selected batch can mutate; prefix, full name pattern, inactive state and minimum group age must all match. --full/--raw never expands the mutation batch.", + ]); +} + function renderOffsets(result: Record): RenderedCliResult { const query = record(result.query); const offsets = arrayRecords(query.offsets).map((item) => [stringValue(item.topic), stringValue(item.partition), stringValue(item.endOffset)]); @@ -1769,6 +1925,36 @@ function parseInspectOptions(args: string[]): KafkaInspectOptions { return { ...parseCommonOptions(commonArgs), topic, group, limit }; } +function parseGroupCleanupOptions(args: string[]): KafkaGroupCleanupOptions { + const commonArgs: string[] = []; + let policyId: string | null = null; + let confirm = false; + let limit: number | null = null; + for (let index = 0; index < args.length; index += 1) { + const arg = args[index]; + if (arg === "--policy") { + policyId = parseSimpleOption(args[index + 1], arg); + index += 1; + } else if (arg === "--confirm") { + confirm = true; + } else if (arg === "--limit") { + const value = args[index + 1]; + if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value"); + limit = Number(value); + if (!Number.isInteger(limit) || limit < 1) throw new Error("--limit must be a positive integer"); + index += 1; + } else { + commonArgs.push(arg); + if (arg === "--target" || arg === "--node") { + commonArgs.push(args[index + 1] ?? ""); + index += 1; + } + } + } + if (policyId === null) throw new Error("Kafka consumer group cleanup requires --policy "); + return { ...parseCommonOptions(commonArgs), policyId, confirm, limit }; +} + async function parseProduceOptions(args: string[]): Promise { const commonArgs: string[] = []; let topic: string | null = null; @@ -1832,6 +2018,14 @@ function boundedGroupListLimit(kafka: PlatformKafkaConfig, limit: number | null) return value; } +function boundedGroupCleanupLimit(kafka: PlatformKafkaConfig, limit: number | null): number { + const management = kafka.management.consumerGroupCleanup; + const value = limit ?? management.defaultCandidateLimit; + if (!Number.isInteger(value) || value < 1) throw new Error("group cleanup candidate limit must be a positive integer"); + if (value > management.maxCandidateLimit) throw new Error(`limit ${value} exceeds ${configLabel}.management.consumerGroupCleanup.maxCandidateLimit=${management.maxCandidateLimit}`); + return value; +} + function buildShadowPayload(kafka: PlatformKafkaConfig, target: KafkaTarget, options: KafkaProduceOptions, topic: string): string { let userPayload: unknown = null; if (options.payload !== null) { diff --git a/scripts/src/platform-infra/entry.ts b/scripts/src/platform-infra/entry.ts index e7c87a3b..bd711d17 100644 --- a/scripts/src/platform-infra/entry.ts +++ b/scripts/src/platform-infra/entry.ts @@ -454,6 +454,7 @@ export function platformInfraHelp(): unknown { "bun scripts/cli.ts platform-infra kafka validate --target NC01", "bun scripts/cli.ts platform-infra kafka topics --node NC01", "bun scripts/cli.ts platform-infra kafka groups --node NC01 [--limit N] [--full|--raw]", + "bun scripts/cli.ts platform-infra kafka groups cleanup --node NC01 --policy hwlab-cloud-api-sse [--confirm]", "bun scripts/cli.ts platform-infra kafka offsets --node NC01 --topic hwlab.event.v1", "bun scripts/cli.ts platform-infra kafka tail --node NC01 --topic hwlab.event.v1 --limit 5", "bun scripts/cli.ts platform-infra kafka produce --node NC01 --topic hwlab.event.v1 --key ",