Kubernetes Operator Pythonic Framework (Kopf)

Kopf is written in Python, which means it consumes more memory and is slower than Go-based operators. If your primary requirement is extremely low memory footprint (e.g., reducing usage from ~300 MB to ~30 MB), you should consider using Go instead.

Key performance characteristics:

• Memory usage: Most memory is consumed by imported Python modules (up to ~75% when using the official kubernetes client) and TCP/SSL contexts, rather than the Kopf runtime itself.
• Runtime: Switching from CPython to PyPy does not provide significant benefits for Kopf.
• Use case: Kopf is optimized for expressive power, ease of use, quick prototyping, and building minimally viable products (MVPs) or ad-hoc operators rapidly, rather than matching the raw resource efficiency of Go.

When developing operators, you should only depend on the public interface. All other modules are implementation details and are prefixed with underscores to discourage use. The only modules with public guarantees on names and signatures are:

• kopf: The main entry point.
• kopf.on: Used for defining handlers and decorators.
• kopf.testing: Used for testing your operators.

Instead of using the kopf run CLI command, you can embed a Kopf operator directly into your own Python application. This allows you to run the operator in a side thread while the main thread performs other application activities (such as a UI loop or orchestration logic).

Note that when running in a side thread, OS signals (like SIGTERM) are ignored by the operator thread, as they are typically handled by the main thread.

python example.py

When running against a real cluster, use the --dev flag. This sets the operator's priority to 666 and automatically pauses all other running operators (which default to priority 0) to prevent collisions and infinite loops.

Alternatively, you can manually pause or resume all other operators using the kopf freeze and kopf resume commands.

The operator's pod must have sufficient permissions via RBAC (Role-Based Access Control) to access and manipulate both built-in Kubernetes objects (like Pod, Job, or PersistentVolumeClaim) and your domain-specific custom resources.

To apply these permissions, create a ServiceAccount, Role/ClusterRole, and RoleBinding/ClusterRoleBinding, then attach the ServiceAccount to the operator's pod in the Deployment specification.

Kopf resource specifications do not support masks, globs, or multiple values in a single selector. To handle multiple independent resources, apply multiple @kopf.on.event decorators to the same handler function. Kopf automatically deduplicates these so the function is only triggered once per resource even if specifications overlap.

import kopf
from typing import Any

@kopf.on.event('kopfexamples')
@kopf.on.event('v1', 'pods')
def fn(**_: Any) -> None:
    pass

You can prevent specific resources from being added to an index by having your indexing function return None (or by not returning a value at all). When None is returned, the index is not updated for that resource, and existing values for that resource are preserved.

Note: If you return a dictionary containing None as a value (e.g., {'key': None}), the None value will be indexed as a placeholder. This is useful when you want to record the existence of a key without providing a specific value.

import kopf
from typing import Any

# This resource will be skipped and existing index values preserved
@kopf.index('pods')
async def empty_index(**_: Any) -> Any:
    pass

# This will index the key with a None value
@kopf.index('pods')
async def index_of_nones(**_: Any) -> Any:
    return {'key': None}

The recommended way to deploy Kopf is using a Kubernetes Deployment object.

Critical Requirements:

• Replicas: Set replicas: 1. Running multiple replicas for the same objects causes unpredictable collisions.
• Update Strategy: Use .spec.strategy.type: Recreate to ensure that during pod restarts or upgrades, only one pod is running at a time.
• Networking: No Services or Ingresses are required because the operator only makes outgoing calls to the Kubernetes API.

kubectl apply -f deployment.yaml

Once a liveness endpoint is configured via the --liveness flag, you can use it in your Kubernetes Deployment specification. If the operator fails to respond to the probe, Kubernetes will restart the pod.

Warning: Ensure that exactly one pod of your operator is running at a time, especially during restarts, to avoid conflicts.

apiVersion: apps/v1
kind: Deployment
spec:
  template:
    spec:
      containers:
      - name: the-only-one
        image: ...
        livenessProbe:
          httpGet:
            path: /healthz
            port: 8080