Stéphane Tailland
5770f78c99
Topics are declared rather than auto-created: the application account has no Create ACL, so a producer would have been denied, and an auto-created topic takes broker defaults nobody chose. The credentials store lets an application namespace read its own KafkaUser password and the cluster CA from the kafka namespace, so the password stays generated and rotatable by Strimzi instead of being copied by hand. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
216 lines
9.7 KiB
Makefile
216 lines
9.7 KiB
Makefile
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Kafka cluster management — Strimzi / k3s
|
|
# Requires: kubectl + helm configured for the target cluster
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
NAMESPACE := kafka
|
|
BROKER := kafka-dual-role-0
|
|
BOOTSTRAP := kafka-kafka-bootstrap.$(NAMESPACE).svc.cluster.local:9093
|
|
SCRIPT_DIR := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))
|
|
KAFKA_NAME := kafka
|
|
|
|
# Overridable per invocation
|
|
TOPIC ?= smoke-test
|
|
MESSAGE ?= hello
|
|
GROUP ?= my-group
|
|
MAX ?= 10
|
|
PARTITIONS ?= 1
|
|
REPLICAS ?= 1
|
|
KAFKA_VER ?= 4.3.1
|
|
METADATA_VER ?= 4.3-IV0
|
|
|
|
KUBECTL := kubectl -n $(NAMESPACE)
|
|
KEXEC := $(KUBECTL) exec $(BROKER) --
|
|
KAFKA_BIN := /opt/kafka/bin
|
|
|
|
.DEFAULT_GOAL := help
|
|
.PHONY: help \
|
|
deploy smoke-test upgrade teardown \
|
|
_config \
|
|
topic-list topic-create topic-delete \
|
|
produce consume list-messages \
|
|
group-list group-offsets
|
|
|
|
# ── Help ──────────────────────────────────────────────────────────────────────
|
|
|
|
help:
|
|
@printf "Usage: make <target> [VARIABLE=value ...]\n\n"
|
|
@printf "Cluster lifecycle:\n"
|
|
@printf " deploy Install Strimzi operator + Kafka cluster + users\n"
|
|
@printf " smoke-test Run produce/consume end-to-end test\n"
|
|
@printf " upgrade Upgrade Kafka (KAFKA_VER=) or Strimzi operator\n"
|
|
@printf " teardown Delete the Kafka cluster and namespace\n"
|
|
@printf "\nTopics:\n"
|
|
@printf " topic-list List all topics\n"
|
|
@printf " topic-create Create TOPIC (idempotent)\n"
|
|
@printf " topic-delete Delete TOPIC\n"
|
|
@printf "\nMessages:\n"
|
|
@printf " produce Produce MESSAGE into TOPIC\n"
|
|
@printf " consume Consume MAX messages from TOPIC (from latest)\n"
|
|
@printf " list-messages Consume MAX messages from TOPIC (from beginning)\n"
|
|
@printf "\nConsumer groups:\n"
|
|
@printf " group-list List all consumer groups\n"
|
|
@printf " group-offsets Show offsets and lag for GROUP\n"
|
|
@printf "\nVariables (current values):\n"
|
|
@printf " TOPIC=%-12s KAFKA_VER=%-10s PARTITIONS=%s\n" "$(TOPIC)" "$(KAFKA_VER)" "$(PARTITIONS)"
|
|
@printf " MESSAGE=%-12s GROUP=%-10s REPLICAS=%s\n" "$(MESSAGE)" "$(GROUP)" "$(REPLICAS)"
|
|
@printf " MAX=%-12s METADATA_VER=%s\n\n" "$(MAX)" "$(METADATA_VER)"
|
|
|
|
# ── Cluster lifecycle ─────────────────────────────────────────────────────────
|
|
|
|
deploy:
|
|
@echo "==> [1/6] Creating namespace '$(NAMESPACE)'"
|
|
@kubectl create namespace $(NAMESPACE) --dry-run=client -o yaml | kubectl apply -f -
|
|
@echo "==> [2/6] Installing Strimzi operator"
|
|
@helm repo add strimzi https://strimzi.io/charts/ 2>/dev/null || true
|
|
@helm repo update strimzi 2>/dev/null
|
|
@helm upgrade --install strimzi-kafka-operator strimzi/strimzi-kafka-operator \
|
|
--namespace $(NAMESPACE) \
|
|
--set watchAnyNamespace=false \
|
|
--wait --timeout 5m
|
|
@echo "==> [3/6] Waiting for Strimzi CRDs"
|
|
@for crd in kafkas.kafka.strimzi.io kafkanodepools.kafka.strimzi.io kafkausers.kafka.strimzi.io; do \
|
|
until kubectl get crd "$$crd" -o jsonpath='{.status.conditions[?(@.type=="Established")].status}' 2>/dev/null | grep -q True; do sleep 2; done; \
|
|
echo " $$crd ready"; \
|
|
done
|
|
@echo "==> [4/6] Deploying Kafka cluster (KRaft, SASL_SSL, SCRAM-SHA-512)"
|
|
@kubectl apply -f $(SCRIPT_DIR)kafka.yaml
|
|
@kubectl wait kafka/$(KAFKA_NAME) --for=condition=Ready --timeout=10m -n $(NAMESPACE)
|
|
@echo "==> [5/6] Deploying KafkaUsers"
|
|
@kubectl apply -f $(SCRIPT_DIR)kafka-users.yaml
|
|
@for user in kafka-admin kafka-client; do \
|
|
kubectl wait kafkauser/$$user --for=condition=Ready --timeout=120s -n $(NAMESPACE); \
|
|
done
|
|
@echo "==> [6/6] Deploying topics and the credentials store"
|
|
@kubectl apply -f $(SCRIPT_DIR)kafka-topics.yaml
|
|
@kubectl apply -f $(SCRIPT_DIR)credentials-access.yaml
|
|
@echo ""
|
|
@echo "Cluster ready. Bootstrap: $(BOOTSTRAP)"
|
|
|
|
smoke-test:
|
|
@echo "==> Running smoke test"
|
|
@$(KUBECTL) delete job kafka-smoke-test --ignore-not-found > /dev/null
|
|
@kubectl apply -f $(SCRIPT_DIR)kafka-smoke-test.yaml
|
|
@until $(KUBECTL) get pod -l job-name=kafka-smoke-test \
|
|
-o jsonpath='{.items[0].status.phase}' 2>/dev/null | grep -qE 'Succeeded|Failed'; do sleep 3; done
|
|
@$(KUBECTL) logs -l job-name=kafka-smoke-test --tail=20
|
|
@$(KUBECTL) get pod -l job-name=kafka-smoke-test \
|
|
-o jsonpath='{.items[0].status.phase}' | grep -q Succeeded \
|
|
&& echo "==> Smoke test PASSED" \
|
|
|| { echo "==> Smoke test FAILED"; exit 1; }
|
|
|
|
upgrade:
|
|
@echo "==> Upgrading Strimzi operator"
|
|
@helm repo update strimzi 2>/dev/null
|
|
@helm upgrade strimzi-kafka-operator strimzi/strimzi-kafka-operator \
|
|
--namespace $(NAMESPACE) \
|
|
--set watchAnyNamespace=false \
|
|
--wait --timeout 5m
|
|
@echo "==> Upgrading Kafka to $(KAFKA_VER) (metadataVersion=$(METADATA_VER))"
|
|
@$(KUBECTL) patch kafka $(KAFKA_NAME) --type=merge \
|
|
-p '{"spec":{"kafka":{"version":"$(KAFKA_VER)","metadataVersion":"$(METADATA_VER)"}}}'
|
|
@kubectl wait kafka/$(KAFKA_NAME) --for=condition=Ready --timeout=10m -n $(NAMESPACE)
|
|
@echo "==> Upgrade complete"
|
|
@$(KUBECTL) get kafka $(KAFKA_NAME)
|
|
|
|
teardown:
|
|
@echo "==> Deleting KafkaUsers and Kafka cluster"
|
|
@kubectl delete -f $(SCRIPT_DIR)credentials-access.yaml --ignore-not-found
|
|
@kubectl delete -f $(SCRIPT_DIR)kafka-topics.yaml --ignore-not-found
|
|
@kubectl delete -f $(SCRIPT_DIR)kafka-users.yaml --ignore-not-found
|
|
@kubectl delete -f $(SCRIPT_DIR)kafka.yaml --ignore-not-found
|
|
@echo "==> Waiting for pods to terminate"
|
|
@$(KUBECTL) wait pod -l strimzi.io/cluster=$(KAFKA_NAME) \
|
|
--for=delete --timeout=120s 2>/dev/null || true
|
|
@echo "==> Uninstalling Strimzi operator"
|
|
@helm uninstall strimzi-kafka-operator -n $(NAMESPACE) 2>/dev/null || true
|
|
@echo "==> Deleting namespace '$(NAMESPACE)'"
|
|
@kubectl delete namespace $(NAMESPACE) --ignore-not-found
|
|
@echo "==> Teardown complete"
|
|
|
|
# ── Internal: write client config into the broker pod ─────────────────────────
|
|
|
|
_config:
|
|
@$(KUBECTL) get secret kafka-cluster-ca-cert \
|
|
-o jsonpath='{.data.ca\.crt}' | base64 -d \
|
|
| $(KUBECTL) exec -i $(BROKER) -- tee /tmp/mk-ca.crt > /dev/null
|
|
@$(KUBECTL) get secret kafka-client \
|
|
-o jsonpath='{.data.sasl\.jaas\.config}' | base64 -d \
|
|
| $(KUBECTL) exec -i $(BROKER) -- tee /tmp/mk-jaas.conf > /dev/null
|
|
@$(KEXEC) /bin/bash -c \
|
|
'printf "security.protocol=SASL_SSL\nssl.truststore.type=PEM\nssl.truststore.location=/tmp/mk-ca.crt\nsasl.mechanism=SCRAM-SHA-512\nsasl.jaas.config=$$(cat /tmp/mk-jaas.conf)\n" \
|
|
> /tmp/mk-client.properties'
|
|
|
|
# ── Topics ────────────────────────────────────────────────────────────────────
|
|
|
|
topic-list: _config
|
|
@$(KEXEC) $(KAFKA_BIN)/kafka-topics.sh \
|
|
--bootstrap-server $(BOOTSTRAP) \
|
|
--command-config /tmp/mk-client.properties \
|
|
--list
|
|
|
|
topic-create: _config
|
|
@$(KEXEC) $(KAFKA_BIN)/kafka-topics.sh \
|
|
--bootstrap-server $(BOOTSTRAP) \
|
|
--command-config /tmp/mk-client.properties \
|
|
--create \
|
|
--topic $(TOPIC) \
|
|
--partitions $(PARTITIONS) \
|
|
--replication-factor $(REPLICAS) \
|
|
--if-not-exists
|
|
@echo "Topic '$(TOPIC)' ready (partitions=$(PARTITIONS) replicas=$(REPLICAS))"
|
|
|
|
topic-delete: _config
|
|
@$(KEXEC) $(KAFKA_BIN)/kafka-topics.sh \
|
|
--bootstrap-server $(BOOTSTRAP) \
|
|
--command-config /tmp/mk-client.properties \
|
|
--delete \
|
|
--topic $(TOPIC)
|
|
@echo "Topic '$(TOPIC)' deleted"
|
|
|
|
# ── Messages ──────────────────────────────────────────────────────────────────
|
|
|
|
produce: _config
|
|
@echo "$(MESSAGE)" \
|
|
| $(KUBECTL) exec -i $(BROKER) -- \
|
|
$(KAFKA_BIN)/kafka-console-producer.sh \
|
|
--bootstrap-server $(BOOTSTRAP) \
|
|
--producer.config /tmp/mk-client.properties \
|
|
--topic $(TOPIC)
|
|
@echo "Produced to '$(TOPIC)': $(MESSAGE)"
|
|
|
|
consume: _config
|
|
@$(KEXEC) $(KAFKA_BIN)/kafka-console-consumer.sh \
|
|
--bootstrap-server $(BOOTSTRAP) \
|
|
--consumer.config /tmp/mk-client.properties \
|
|
--topic $(TOPIC) \
|
|
--max-messages $(MAX) \
|
|
--timeout-ms 10000 2>/dev/null
|
|
|
|
list-messages: _config
|
|
@$(KEXEC) $(KAFKA_BIN)/kafka-console-consumer.sh \
|
|
--bootstrap-server $(BOOTSTRAP) \
|
|
--consumer.config /tmp/mk-client.properties \
|
|
--topic $(TOPIC) \
|
|
--from-beginning \
|
|
--max-messages $(MAX) \
|
|
--timeout-ms 15000 2>/dev/null
|
|
|
|
# ── Consumer groups ───────────────────────────────────────────────────────────
|
|
|
|
group-list: _config
|
|
@$(KEXEC) $(KAFKA_BIN)/kafka-consumer-groups.sh \
|
|
--bootstrap-server $(BOOTSTRAP) \
|
|
--command-config /tmp/mk-client.properties \
|
|
--list 2>/dev/null
|
|
|
|
group-offsets: _config
|
|
@$(KEXEC) /bin/bash -c \
|
|
'out=$$( $(KAFKA_BIN)/kafka-consumer-groups.sh \
|
|
--bootstrap-server $(BOOTSTRAP) \
|
|
--command-config /tmp/mk-client.properties \
|
|
--group $(GROUP) --describe 2>/dev/null); \
|
|
echo "$$out" | grep -q GroupIdNotFoundException \
|
|
&& echo "Group $(GROUP) not found (no consumer has connected yet)" \
|
|
|| echo "$$out"'
|