How to Monitor Kopf Operators with Vigilmon

Kopf (Kubernetes Operator Pythonic Framework) is a Python framework for building Kubernetes operators. Instead of Go boilerplate, you write decorated Python functions that respond to resource events — a @kopf.on.create handler runs when a custom resource is created, @kopf.on.update when it changes, and @kopf.on.delete when it's removed. Kopf operators run as Deployments inside Kubernetes and are increasingly common for data pipelines, ML training jobs, and domain-specific automation written by teams more comfortable in Python than Go.

But Kopf operators are production infrastructure. When your Python handler crashes, raises an unhandled exception, or gets stuck in a retry loop, the custom resources it manages stop being reconciled — jobs don't get scheduled, resources don't get cleaned up, and state drifts silently. In this tutorial you'll set up uptime and response-time monitoring for your Kopf operators using Vigilmon — free tier, no credit card required.

Why Kopf operators need external monitoring

Internal Kubernetes probes on your Kopf operator Pod only detect container crashes. They miss the Python-specific failure modes:

• Handler exception loop — an unhandled exception in a @kopf.on.create handler puts Kopf into a retry loop with exponential backoff; the pod stays Running but the resource never reaches its desired state
• Asyncio event loop blocked — a synchronous blocking call inside an async handler freezes the entire event loop; no handlers execute while the pod appears healthy
• Peering conflict — Kopf uses a KopfPeering object for leader election across replicas; if two instances disagree on who the leader is, reconciliations may be duplicated or dropped
• External dependency timeout — your handler calls an external API or database; slow or failed responses cascade into timeout errors without surfacing through pod health checks
• Memory leak from CRD event storm — a large cluster with many CRD objects can trigger a rapid sequence of watch events that consumes memory until the OOM killer terminates the pod

External monitoring from Vigilmon catches operator unavailability before your users see stuck resources.

What you'll need

• A Kopf operator running as a Kubernetes Deployment
• Python 3.8+ with Kopf installed
• A free Vigilmon account

Step 1: Add an HTTP health endpoint to your Kopf operator

Kopf does not expose a built-in HTTP health server. Add one using aiohttp (which Kopf already depends on) so Vigilmon has an endpoint to probe:

# operator.py
import asyncio
import logging
import kopf
from aiohttp import web

logger = logging.getLogger(__name__)

# Track operator health state
_health_state = {"status": "starting", "last_event": None, "errors": []}


@kopf.on.startup()
async def startup(settings: kopf.OperatorSettings, **kwargs):
    settings.peering.priority = 100
    settings.posting.level = logging.WARNING
    _health_state["status"] = "healthy"
    logger.info("Operator started")


@kopf.on.create("mygroup.io", "v1", "myresources")
async def create_handler(spec, name, namespace, **kwargs):
    try:
        # Your business logic here
        logger.info(f"Creating resource {name} in {namespace}")
        _health_state["last_event"] = f"create:{namespace}/{name}"
        _health_state["status"] = "healthy"
    except Exception as e:
        _health_state["status"] = "degraded"
        _health_state["errors"].append(str(e))
        raise


@kopf.on.update("mygroup.io", "v1", "myresources")
async def update_handler(spec, name, namespace, **kwargs):
    _health_state["last_event"] = f"update:{namespace}/{name}"


# Health HTTP server running alongside the operator
async def health_handler(request):
    if _health_state["status"] == "healthy":
        return web.json_response({"status": "ok", "last_event": _health_state["last_event"]})
    return web.json_response(
        {"status": "error", "errors": _health_state["errors"][-5:]},
        status=503,
    )


async def run_health_server():
    app = web.Application()
    app.router.add_get("/health", health_handler)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", 9090)
    await site.start()
    logger.info("Health server listening on :9090")
    # Keep running until cancelled
    while True:
        await asyncio.sleep(3600)


@kopf.on.startup()
async def start_health_server(settings: kopf.OperatorSettings, **kwargs):
    asyncio.create_task(run_health_server())

Build and push the operator image, then expose the health port:

# operator-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: my-kopf-operator
  namespace: my-operator-system
spec:
  replicas: 1
  selector:
    matchLabels:
      app: my-kopf-operator
  template:
    metadata:
      labels:
        app: my-kopf-operator
    spec:
      serviceAccountName: my-kopf-operator
      containers:
        - name: operator
          image: your-registry/my-kopf-operator:latest
          command: ["kopf", "run", "--standalone", "/app/operator.py"]
          ports:
            - containerPort: 9090
              name: health
          livenessProbe:
            httpGet:
              path: /health
              port: 9090
            initialDelaySeconds: 15
            periodSeconds: 30
          readinessProbe:
            httpGet:
              path: /health
              port: 9090
            initialDelaySeconds: 5
            periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: my-kopf-operator-health
  namespace: my-operator-system
spec:
  type: LoadBalancer
  selector:
    app: my-kopf-operator
  ports:
    - port: 9090
      targetPort: 9090
      protocol: TCP

kubectl apply -f operator-deployment.yaml
kubectl get svc my-kopf-operator-health -n my-operator-system
# NAME                       TYPE           EXTERNAL-IP     PORT(S)
# my-kopf-operator-health    LoadBalancer   203.0.113.53    9090:32090/TCP

curl http://203.0.113.53:9090/health
# {"status": "ok", "last_event": null}

Step 2: Add structured exception tracking

Improve the health endpoint to reflect handler error rates, which is more useful than a binary up/down signal:

import time
from collections import deque

# Ring buffer of recent errors (last 20)
_recent_errors = deque(maxlen=20)
_error_window_start = time.monotonic()


def record_error(handler_name: str, error: str):
    _recent_errors.append({
        "handler": handler_name,
        "error": error,
        "time": time.time(),
    })


async def health_handler(request):
    # Count errors in the last 5 minutes
    cutoff = time.time() - 300
    recent = [e for e in _recent_errors if e["time"] > cutoff]
    status = "ok" if len(recent) < 5 else "degraded"
    http_status = 200 if status == "ok" else 503

    return web.json_response(
        {
            "status": status,
            "errors_last_5m": len(recent),
            "last_event": _health_state.get("last_event"),
        },
        status=http_status,
    )

Step 3: Set up HTTP monitoring in Vigilmon

With your health endpoint running and exposed, add it to Vigilmon:

1. Log in to vigilmon.online and go to Monitors → New Monitor
2. Choose HTTP / HTTPS
3. Add monitors for your operator:

| Monitor name | URL | Expected status | |---|---|---| | Kopf operator health | http://203.0.113.53:9090/health | 200 | | Operator readiness check | http://203.0.113.53:9090/health | 200 with "status":"ok" |

4. Set the check interval to 1 minute
5. Under Expected response, set status code 200 and optionally match "status":"ok" in the response body
6. Save each monitor

Multi-region monitoring from Vigilmon ensures that a regional network issue with the LoadBalancer or a cloud provider hiccup doesn't produce a false positive — multiple independent probes must agree before an alert fires.

Step 4: Monitor the TCP layer

Add a TCP monitor for the health port to detect port-level failures independently of the application:

1. Go to Monitors → New Monitor
2. Choose TCP Port
3. Enter your operator host and port: 203.0.113.53 / 9090
4. Save the monitor

A TCP failure means the container's network stack or the LoadBalancer routing is broken — the operator may have crashed or been OOM-killed, causing the port to stop accepting connections before the pod shows NotReady.

Step 5: Configure alert channels

Kopf handler failures are silent until someone notices a resource stuck in a pending state. Get alerted the moment the operator stops responding.

Email alerts

1. Go to Alert Channels → Add Channel → Email
2. Enter your platform or data-engineering team's on-call address
3. Assign the channel to your Kopf operator monitors

Webhook alerts for incident routing

1. Go to Alert Channels → Add Channel → Webhook
2. Enter your incident management system's webhook URL
3. Vigilmon sends this payload on failure:

{
  "monitor_name": "Kopf operator health",
  "status": "down",
  "url": "http://203.0.113.53:9090/health",
  "started_at": "2024-01-15T10:23:00Z",
  "duration_seconds": 120
}

Build a runbook triggered by this webhook that checks the operator pod logs and lists custom resources in a non-terminal state.

Step 6: Correlate Vigilmon alerts with Kopf diagnostics

When you get a Kopf operator alert, run these checks:

# 1. Check the operator pod
kubectl get pods -n my-operator-system -l app=my-kopf-operator

# 2. Read operator logs — Kopf logs handler activity and errors
kubectl logs -n my-operator-system \
  -l app=my-kopf-operator --tail=100

# 3. Check for handler retry backoff — look for "retrying" in logs
kubectl logs -n my-operator-system \
  -l app=my-kopf-operator --tail=200 \
  | grep -i "retrying\|backoff\|exception\|error"

# 4. Check KopfPeering — operator may have lost leader election
kubectl get kopfpeering -A

# 5. List custom resources stuck in non-terminal states
kubectl get myresource -A \
  -o jsonpath='{range .items[*]}{.metadata.namespace}/{.metadata.name}{"\t"}{.status.kopf.progress}{"\n"}{end}' \
  | grep -v "^$"

# 6. Check pod resource usage — Python memory leaks are common
kubectl top pod -n my-operator-system -l app=my-kopf-operator

If the pod is Running but /health returns 503, your error rate threshold was crossed — look at the recent error log in the health response body and trace it back to a specific handler.

Step 7: Create a status page for operator-managed workflows

If multiple teams trigger custom resources managed by your Kopf operator (ML jobs, data pipelines, batch tasks), a status page reduces "why is my job stuck?" questions:

1. Go to Status Pages → New Status Page
2. Name it: "Workflow Operator Status"
3. Add your monitors: Kopf operator health
4. Share the URL with teams that submit custom resources

Summary

| What you set up | What it catches | |---|---| | HTTP health endpoint in Kopf operator | Handler exceptions, retry loops, asyncio stalls | | HTTP monitor on /health | Operator crashes, degraded handler error rate | | TCP monitor on health port | OOM kills, port-level outages | | Email + webhook alert channels | Immediate notification before resources pile up in stuck state | | Status page | Team self-service for stuck job/workflow diagnosis |

Kopf operators bring Python's expressiveness to the Kubernetes control plane. External monitoring keeps that control plane observable and your team informed when Python exceptions or resource exhaustion take the operator offline.

Get started at vigilmon.online — free tier, no credit card required.