feat: add bounded kafka group cleanup

2026-07-10 01:53:53 +02:00
parent 374ff576b1
commit dd7f028a5b
6 changed files with 429 additions and 2 deletions
@@ -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)