Codex
213 lines
7.8 KiB
Python
213 lines
7.8 KiB
Python
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)
|