How to Monitor Strimzi with Vigilmon

Strimzi is the CNCF-backed Kubernetes operator that makes deploying and managing Apache Kafka clusters on Kubernetes a declarative, GitOps-native operation. When you adopt Strimzi, the entire Kafka lifecycle — brokers, topics, users, connectors — becomes a Kubernetes custom resource. But a powerful abstraction introduces new failure modes: reconciliation loops that stall silently, broker pods that restart but don't rejoin the cluster, and topic replication factors that quietly degrade after a node eviction.

This tutorial shows you how to expose the right metrics and health signals from Strimzi, then configure Vigilmon HTTP and heartbeat monitors so you know the moment your Kafka-on-Kubernetes platform is drifting from its desired state.

Why Strimzi Needs External Monitoring

Strimzi adds an operator layer on top of Kafka, which means you have two systems to monitor: the Kafka cluster itself and the Strimzi operator managing it.

Common failure modes Vigilmon catches:

• Reconciliation stalls — the operator pod is running but the Ready condition on a Kafka custom resource is stuck at False
• Broker pod churn — a broker pod is in CrashLoopBackOff but the Kafka cluster still accepts writes (using reduced replication)
• Under-replicated partitions — a topic's min.insync.replicas constraint is violated because a broker left the ISR
• KRaft/ZooKeeper degradation — in KRaft mode, the active controller fails to elect; in ZooKeeper mode, a quorum follower falls behind
• Consumer group lag — a consumer group stops committing offsets while Kafka keeps producing

Step 1: Expose a Health Endpoint from Your Cluster

Strimzi ships with a JMX Prometheus exporter sidecar, but for Vigilmon you need a simple HTTP health endpoint that aggregates key signals into a status response. Deploy a lightweight health sidecar as a Kubernetes Deployment:

# strimzi-health-sidecar.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: strimzi-health-probe
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: strimzi-health-probe
  template:
    metadata:
      labels:
        app: strimzi-health-probe
    spec:
      serviceAccountName: strimzi-health-sa
      containers:
        - name: probe
          image: python:3.12-slim
          command: ["python", "-u", "/app/probe.py"]
          env:
            - name: KAFKA_NAMESPACE
              value: "kafka"
            - name: KAFKA_CLUSTER_NAME
              value: "my-cluster"
            - name: BOOTSTRAP_SERVERS
              value: "my-cluster-kafka-bootstrap.kafka.svc:9092"
          ports:
            - containerPort: 8080
          volumeMounts:
            - name: probe-script
              mountPath: /app
      volumes:
        - name: probe-script
          configMap:
            name: strimzi-health-probe-script
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: strimzi-health-sa
  namespace: kafka
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: strimzi-health-reader
rules:
  - apiGroups: ["kafka.strimzi.io"]
    resources: ["kafkas", "kafkatopics"]
    verbs: ["get", "list"]
  - apiGroups: [""]
    resources: ["pods"]
    verbs: ["get", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: strimzi-health-reader-binding
subjects:
  - kind: ServiceAccount
    name: strimzi-health-sa
    namespace: kafka
roleRef:
  kind: ClusterRole
  name: strimzi-health-reader
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: v1
kind: Service
metadata:
  name: strimzi-health-probe
  namespace: kafka
spec:
  selector:
    app: strimzi-health-probe
  ports:
    - port: 8080
      targetPort: 8080

And the probe script:

# probe.py
import os, json, subprocess
from http.server import HTTPServer, BaseHTTPRequestHandler
from kafka import KafkaAdminClient, KafkaConsumer
from kafka.admin import NewTopic

KAFKA_NAMESPACE = os.environ.get('KAFKA_NAMESPACE', 'kafka')
KAFKA_CLUSTER_NAME = os.environ.get('KAFKA_CLUSTER_NAME', 'my-cluster')
BOOTSTRAP = os.environ.get('BOOTSTRAP_SERVERS', 'localhost:9092')

def get_kafka_cr_status():
    """Read Kafka CR readiness via kubectl."""
    result = subprocess.run(
        ['kubectl', 'get', 'kafka', KAFKA_CLUSTER_NAME, '-n', KAFKA_NAMESPACE,
         '-o', 'jsonpath={.status.conditions}'],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        return None, 'kubectl_failed'
    try:
        conditions = json.loads(result.stdout)
        ready = next((c for c in conditions if c.get('type') == 'Ready'), None)
        return ready, None
    except Exception:
        return None, 'parse_failed'

def get_broker_pod_status():
    result = subprocess.run(
        ['kubectl', 'get', 'pods', '-n', KAFKA_NAMESPACE,
         '-l', f'strimzi.io/cluster={KAFKA_CLUSTER_NAME},strimzi.io/kind=Kafka',
         '-o', 'jsonpath={range .items[*]}{.metadata.name}:{.status.phase}\\n{end}'],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        return [], 'kubectl_failed'
    lines = [l for l in result.stdout.strip().split('\n') if l]
    pods = [{'name': p.split(':')[0], 'phase': p.split(':')[1]} for p in lines if ':' in p]
    return pods, None

def get_under_replicated():
    try:
        admin = KafkaAdminClient(bootstrap_servers=BOOTSTRAP, request_timeout_ms=5000)
        topics = admin.list_topics()
        metadata = admin.describe_topics(topics)
        under_rep = []
        for topic in metadata:
            for partition in topic.get('partitions', []):
                isr = partition.get('isr', [])
                replicas = partition.get('replicas', [])
                if len(isr) < len(replicas):
                    under_rep.append(f"{topic['name']}:{partition['partition']}")
        admin.close()
        return under_rep, None
    except Exception as e:
        return [], str(e)

class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/health/strimzi':
            self._handle_strimzi()
        elif self.path == '/health/strimzi/brokers':
            self._handle_brokers()
        elif self.path == '/health/strimzi/replication':
            self._handle_replication()
        else:
            self.send_response(404)
            self.end_headers()

    def _respond(self, code, body):
        self.send_response(code)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(body).encode())

    def _handle_strimzi(self):
        issues = []
        ready_cond, err = get_kafka_cr_status()
        if err:
            return self._respond(503, {'status': 'down', 'reason': err})
        if not ready_cond:
            issues.append('kafka_cr_no_ready_condition')
        elif ready_cond.get('status') != 'True':
            issues.append(f"kafka_cr_not_ready: {ready_cond.get('message', 'unknown')}")

        pods, err = get_broker_pod_status()
        if err:
            issues.append(f'pod_check_failed: {err}')
        else:
            not_running = [p['name'] for p in pods if p['phase'] != 'Running']
            if not_running:
                issues.append(f"broker_pods_not_running: {','.join(not_running)}")

        if issues:
            return self._respond(503, {'status': 'degraded', 'issues': issues})
        self._respond(200, {'status': 'ok', 'broker_pods': len(pods)})

    def _handle_brokers(self):
        pods, err = get_broker_pod_status()
        if err:
            return self._respond(503, {'status': 'down', 'reason': err})
        not_running = [p for p in pods if p['phase'] != 'Running']
        if not_running:
            return self._respond(503, {'status': 'degraded', 'not_running': not_running})
        self._respond(200, {'status': 'ok', 'pods': pods})

    def _handle_replication(self):
        under_rep, err = get_under_replicated()
        if err:
            return self._respond(503, {'status': 'down', 'reason': err})
        if under_rep:
            return self._respond(503, {'status': 'degraded', 'under_replicated_partitions': under_rep})
        self._respond(200, {'status': 'ok', 'under_replicated_partitions': []})

    def log_message(self, *args):
        pass

if __name__ == '__main__':
    print('Strimzi health probe listening on :8080')
    HTTPServer(('0.0.0.0', 8080), HealthHandler).serve_forever()

Apply the resources:

kubectl apply -f strimzi-health-sidecar.yaml

Step 2: Monitor Operator Reconciliation Status

The Kafka custom resource exposes reconciliation state through its .status.conditions array. A healthy cluster has a Ready condition with status: "True". When reconciliation stalls — for example, after a rolling update that hits a pod-disruption budget limit — the condition reads:

conditions:
  - type: Ready
    status: "False"
    reason: "UpdatePodError"
    message: "Pod my-cluster-kafka-1 is not ready"

Your health endpoint at /health/strimzi already parses this condition. Expose it via a NodePort or Ingress for Vigilmon to reach:

# ingress-health-probe.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: strimzi-health-probe
  namespace: kafka
  annotations:
    nginx.ingress.kubernetes.io/rewrite-target: /
spec:
  rules:
    - host: kafka-health.internal.example.com
      http:
        paths:
          - path: /health
            pathType: Prefix
            backend:
              service:
                name: strimzi-health-probe
                port:
                  number: 8080

Step 3: Monitor Consumer Group Lag

Consumer group lag is one of the most actionable signals in a Kafka deployment. Add a dedicated lag endpoint:

# Add to probe.py
from kafka import KafkaConsumer, TopicPartition

def get_consumer_lag(group_id, topic):
    consumer = KafkaConsumer(
        group_id=f'__vigilmon_probe_{group_id}',
        bootstrap_servers=BOOTSTRAP,
        enable_auto_commit=False,
        request_timeout_ms=5000,
    )
    partitions = consumer.partitions_for_topic(topic)
    if not partitions:
        consumer.close()
        return None, f'topic_{topic}_not_found'

    tps = [TopicPartition(topic, p) for p in partitions]
    consumer.assign(tps)
    consumer.seek_to_end(*tps)
    end_offsets = {tp: consumer.position(tp) for tp in tps}

    consumer2 = KafkaConsumer(
        group_id=group_id,
        bootstrap_servers=BOOTSTRAP,
        enable_auto_commit=False,
    )
    committed = {tp: (consumer2.committed(tp) or 0) for tp in tps}
    consumer.close()
    consumer2.close()

    lag = {str(tp): end_offsets[tp] - committed[tp] for tp in tps}
    total_lag = sum(lag.values())
    return {'total_lag': total_lag, 'per_partition': lag}, None

Add the route in HealthHandler.do_GET:

elif self.path.startswith('/health/strimzi/lag'):
    # Usage: /health/strimzi/lag?group=my-group&topic=my-topic
    from urllib.parse import urlparse, parse_qs
    params = parse_qs(urlparse(self.path).query)
    group = (params.get('group') or [''])[0]
    topic = (params.get('topic') or [''])[0]
    if not group or not topic:
        return self._respond(400, {'error': 'group and topic required'})
    lag_data, err = get_consumer_lag(group, topic)
    if err:
        return self._respond(503, {'status': 'down', 'reason': err})
    threshold = int(os.environ.get('LAG_ALERT_THRESHOLD', '10000'))
    if lag_data['total_lag'] > threshold:
        return self._respond(503, {'status': 'degraded',
                                   'total_lag': lag_data['total_lag'],
                                   'threshold': threshold})
    self._respond(200, {'status': 'ok', **lag_data})

Step 4: Configure Vigilmon Monitors

1. Log in to vigilmon.online and go to Monitors → New Monitor
2. Choose HTTP / HTTPS
3. Set the URL to your Strimzi health endpoint: https://kafka-health.internal.example.com/health/strimzi
4. Set the check interval to 1 minute
5. Under Expected response, configure:
   • Status code: 200
   • Response body contains: "status":"ok"
   • Response time threshold: 5000ms
6. Under Alert channels, assign your Slack or PagerDuty integration

Set up all four monitors:

| Monitor URL | Purpose | Interval | Alert | |---|---|---|---| | /health/strimzi | Operator readiness + broker pod health | 1 min | P1 – PagerDuty | | /health/strimzi/brokers | Broker pod phase | 1 min | P1 – PagerDuty | | /health/strimzi/replication | Under-replicated partitions | 2 min | P1 – PagerDuty | | /health/strimzi/lag?group=my-group&topic=my-topic | Consumer group lag | 1 min | P2 – Slack |

Step 5: Heartbeat Monitor for the Operator

The Strimzi operator itself runs as a Deployment in the strimzi-system namespace. Add a liveness probe that pings Vigilmon from a CronJob running every two minutes:

# strimzi-heartbeat-cronjob.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
  name: strimzi-operator-heartbeat
  namespace: strimzi-system
spec:
  schedule: "*/2 * * * *"
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: heartbeat
              image: curlimages/curl:8.5.0
              command:
                - sh
                - -c
                - |
                  # Only ping if the operator deployment is healthy
                  READY=$(kubectl get deployment strimzi-cluster-operator \
                    -n strimzi-system \
                    -o jsonpath='{.status.readyReplicas}' 2>/dev/null)
                  if [ "$READY" = "1" ]; then
                    curl -sf "$VIGILMON_HEARTBEAT_URL" || true
                  fi
              env:
                - name: VIGILMON_HEARTBEAT_URL
                  valueFrom:
                    secretKeyRef:
                      name: vigilmon-secrets
                      key: operator-heartbeat-url

Create the Vigilmon heartbeat monitor first:

1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
2. Name it strimzi-operator
3. Expected interval: 5 minutes, grace period: 10 minutes
4. Copy the heartbeat URL into your vigilmon-secrets Kubernetes secret

Step 6: Alert Routing

| Monitor | Priority | Alert Channel | Why | |---|---|---|---| | Operator readiness (/health/strimzi) | P1 | PagerDuty | Stalled reconciliation blocks all Kafka config changes | | Under-replicated partitions | P1 | PagerDuty | Data durability is compromised | | Broker pod phase | P1 | PagerDuty | Reduced broker count increases partition load | | Consumer lag | P2 | Slack | Processing delay, may cascade to P1 | | Operator heartbeat | P2 | Slack + email | Operator absent but cluster still running |

Summary

Strimzi lifts Kafka management onto Kubernetes, but that abstraction introduces new monitoring surfaces: the operator reconciliation loop, the Kafka CR ready condition, broker pod phases, topic ISR health, and consumer group lag. A single Vigilmon HTTP probe per signal and one heartbeat monitor for the operator gives you end-to-end visibility with no Prometheus or Grafana required.

| Monitor | What It Catches | |---|---| | /health/strimzi | Reconciliation stalls, CR not-ready, pod churn | | /health/strimzi/brokers | Broker pods in CrashLoop or Pending | | /health/strimzi/replication | Under-replicated partitions, ISR shrinkage | | /health/strimzi/lag | Consumer group falling behind | | Heartbeat: strimzi-operator | Operator pod absent or crashlooping |

Get started free at vigilmon.online — your first Strimzi monitor is live in under five minutes.