diff --git a/config/platform-infra/kafka.yaml b/config/platform-infra/kafka.yaml index 7233f7f5..e8695fc5 100644 --- a/config/platform-infra/kafka.yaml +++ b/config/platform-infra/kafka.yaml @@ -251,7 +251,7 @@ management: deleteBatchSize: 50 automatic: enabled: true - schedule: "*/15 * * * *" + schedule: "*/10 * * * *" image: quay.io/strimzi/kafka:1.1.0-kafka-4.3.0 consumerGroupsCommand: /opt/kafka/bin/kafka-consumer-groups.sh configMapName: platform-infra-kafka-group-retention diff --git a/scripts/native/platform-infra/kafka-consumer-group-retention.sh b/scripts/native/platform-infra/kafka-consumer-group-retention.sh index 9ce632e0..63be4c00 100644 --- a/scripts/native/platform-infra/kafka-consumer-group-retention.sh +++ b/scripts/native/platform-infra/kafka-consumer-group-retention.sh @@ -43,6 +43,17 @@ state_allowed() { return 1 } +declare -A group_states=() +while read -r group state; do + [[ -n "$group" && -n "$state" ]] || continue + group_states["$group"]="$state" +done < <( + timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --describe --all-groups --state 2>/dev/null | + awk 'NF >= 2 && $1 != "GROUP" { print $1, $(NF-1) }' +) + +selected_groups=() + while IFS= read -r group; do [[ -n "$group" ]] || continue scanned=$((scanned + 1)) @@ -56,18 +67,34 @@ while IFS= read -r group; do age_matched=$((age_matched + 1)) (( selected < KAFKA_CANDIDATE_LIMIT )) || continue - state="$(timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --describe --group "$group" --state 2>/dev/null | awk 'NR > 1 && NF >= 2 { value=$(NF-1) } END { print value }')" + state="${group_states[$group]:-}" state_allowed "$state" || continue inactive_matched=$((inactive_matched + 1)) selected=$((selected + 1)) - (( selected <= KAFKA_DELETE_BATCH_SIZE )) || continue - if timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --delete --group "$group" >/dev/null 2>&1; then - deleted=$((deleted + 1)) - else - failed=$((failed + 1)) - fi + selected_groups+=("$group") done < <(timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --list 2>/dev/null) +delete_args=() +delete_count="${#selected_groups[@]}" +(( delete_count <= KAFKA_DELETE_BATCH_SIZE )) || delete_count="$KAFKA_DELETE_BATCH_SIZE" +for ((index = 0; index < delete_count; index += 1)); do + delete_args+=(--group "${selected_groups[$index]}") +done +if (( delete_count > 0 )); then + timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --delete "${delete_args[@]}" >/dev/null 2>&1 || true + declare -A remaining_groups=() + while IFS= read -r group; do + [[ -n "$group" ]] && remaining_groups["$group"]=1 + done < <(timeout "$KAFKA_COMMAND_TIMEOUT_SECONDS" "${groups_cli[@]}" --list 2>/dev/null) + for ((index = 0; index < delete_count; index += 1)); do + if [[ -n "${remaining_groups[${selected_groups[$index]}]:-}" ]]; then + failed=$((failed + 1)) + else + deleted=$((deleted + 1)) + fi + done +fi + printf '{"ok":%s,"policy":"%s","scanned":%d,"prefixMatched":%d,"patternMatched":%d,"ageMatched":%d,"inactiveMatched":%d,"selected":%d,"deleted":%d,"failed":%d,"valuesPrinted":false}\n' \ "$([[ "$failed" -eq 0 ]] && printf true || printf false)" \ "$KAFKA_POLICY_ID" "$scanned" "$prefix_matched" "$pattern_matched" "$age_matched" \ diff --git a/scripts/src/platform-infra-kafka.ts b/scripts/src/platform-infra-kafka.ts index 6e9d2851..d2368bc8 100644 --- a/scripts/src/platform-infra-kafka.ts +++ b/scripts/src/platform-infra-kafka.ts @@ -2094,7 +2094,7 @@ function renderStatus(result: Record): RenderedCliResult { const topics = arrayRecords(summary.topics).map((topic) => [stringValue(topic.topicName), boolText(topic.ready), stringValue(topic.resource)]); const users = arrayRecords(summary.users).map((user) => [stringValue(user.name), boolText(user.ready), stringValue(user.secret)]); const pods = arrayRecords(summary.pods).map((pod) => [stringValue(pod.name), stringValue(pod.phase), boolText(pod.ready)]); - const retention = arrayRecords(summary.retention).map((job) => [stringValue(job.name), stringValue(job.schedule), boolText(!job.suspended), stringValue(job.lastSuccessfulTime)]); + const retention = arrayRecords(summary.retention).map((job) => [stringValue(job.name), stringValue(job.schedule), boolText(!job.suspended), stringValue(job.active, "0"), stringValue(job.lastSuccessfulTime)]); return rendered(result, "platform-infra kafka status", [ "PLATFORM-INFRA KAFKA STATUS", ...table(["TARGET", "ROUTE", "NAMESPACE", "READY"], [[stringValue(summary.target), stringValue(summary.route), stringValue(summary.namespace), boolText(summary.ready)]]), @@ -2119,7 +2119,7 @@ function renderStatus(result: Record): RenderedCliResult { ...(pods.length === 0 ? ["-"] : table(["POD", "PHASE", "READY"], pods)), "", "GROUP RETENTION", - ...(retention.length === 0 ? ["-"] : table(["CRONJOB", "SCHEDULE", "ENABLED", "LAST_SUCCESS"], retention)), + ...(retention.length === 0 ? ["-"] : table(["CRONJOB", "SCHEDULE", "ENABLED", "ACTIVE", "LAST_SUCCESS"], retention)), ...remoteErrorLines(result), "", `NEXT bun scripts/cli.ts platform-infra kafka validate --target ${stringValue(summary.target)}`,