Perhaps everyone has used operators in Kubernetes, for example, PostgreSQL operator, VictoriaMetrics Operator.
But what’s going on under the hood? How and to what are CustomResourceDefinition (CRD) applied, and what is an “operator”?
And finally, what is the difference between a Kubernetes Operator and a Kubernetes Controller?
In the previous part – Kubernetes: Kubernetes APIs, API Groups, CRDs, etcd – we dug a little deeper into how the Kubernetes API works and what a CRD is, and now we can try to write our own micro-operator, a simple MVP, and use it as an example to understand the details.
Contents
Kubernetes Controller vs Kubernetes Operator
So, what is the main difference between Controllers and Operators?
What is: Kubernetes Controller
Simply put, a Controller is just some service that monitors resources in a cluster and brings their state in line with how this state is described in the database – etcd.
In Kubernetes, we have a set of default controllers – Core Controllers within the Kube Controller Manager, such as the ReplicaSet Controller, which checks the number of pods in the Deployment against the replicas value, or the Deployment Controller, which controls the creation and update of ReplicaSets, or the PersistentVolume Controller and PersistentVolumeClaim Binder for working with disks, etc.
In addition to these default controllers, you can create your own controller or use an existing one, such as ExternalDNS Controller. These are examples of custom controllers.
Controllers work in a control loop – a cyclic process in which they constantly check the resources assigned to them – either to change existing resources in the system or to respond to the addition of new ones.
During each check(reconciliation loop), the Controller compares the current state of the resource and compares it with the desired state – that is, the parameters specified in its manifest when the resource was created or updated.
If the desired state does not correspond to the current state, the controller performs the necessary actions to bring these states into alignment.
What is: Kubernetes Operator
Kubernetes Operator, in turn, is a kind of “controller on steroids”: in fact, Operator is a Custom Controller in the sense that it has its own service in the form of a Pod that communicates with the Kubernetes API to receive and update information about resources.
But if ordinary controllers work with “default” resource types (Pod, Endpoint Slice, Node, PVC), then for Operator we describe our own custom resources using a manifest with Custom Resource.
And how these resources will look like and what parameters they will have – we set through CustomResourceDefinition which are written to the Kubernetes database and added to the Kubernetes API, and thus the Kubernetes API allows our custom Controller to operate with these resources.
That is:
- Controller is a component, a service, and Operator is a combination of one or more custom Controllers and corresponding CRDs
- Controller – responds to changes in resources, and Operator – adds new types of resources + controller that controls these resources
Kubernetes Operator frameworks
There are several solutions that simplify the creation of operators.
The main ones are Kubebuilder, a framework for creating controllers in Go, and Kopf, a framework in Python.
There is also the Operator SDK, which allows you to work with controllers even with Helm, without code.
At first, I was thinking of doing it in bare Go, without any frameworks, to better understand how everything works under the hood – but this post started to turn into 95% Golang.
And since the main idea of the post was to show conceptually what a Kubernetes Operator is, what role CustomResourceDefinitions play, and how they interact with each other and allow you to manage resources, I decided to use Kopf because it’s very simple and quite suitable for these purposes.
Creating a CustomResourceDefinition
Let’s start with writing the CRD.
Actually, CustomResourceDefinition is just a description of what fields our custom resource will have so that the controller can use them through the Kubernetes API to create real resources – whether they are some resources in Kubernetes itself, or external ones like AWS Load Balancer or AWS Route 53.
What we will do: we will write a CRD that will describe the MyApp resource, and this resource will have fields for the Docker image and a custom field with some text that will then be written to the Kubernetes Pod logs.
Kubernetes documentation on CRD – Extend the Kubernetes API with CustomResourceDefinitions.
Create the file myapp-crd.yaml:
apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: name: myapps.demo.rtfm.co.ua spec: group: demo.rtfm.co.ua versions: - name: v1 served: true storage: true schema: openAPIV3Schema: type: object properties: spec: type: object properties: image: type: string banner: type: string description: "Optional banner text for the application" scope: Namespaced names: plural: myapps singular: myapp kind: MyApp shortNames: - ma
Here:
spec.group: demo.rtfm.co.ua: create a new API Group, all resources of this type will be available at/apis/demo.rtfm.co.ua/...versions: list of versions of the new resourcename.v1: we will have an only one versionserved: true: add a new resource to the Kube API – you can dokubectl get myapp(GET /apis/demo.rtfm.co.ua/v1/myapps)storage: true: this version will be used for storage inetcd(if several versions are described, only one should be withstorage: true)schema:openAPIV3Schema: describe the API scheme according to the OpenAPI v3type: object: describe an object with nested fields (key: value)
-
-
properties: what fields the object will havespec: what we can use in YAML manifests when creatingtype: object– describe the following fields:properties:image.type: string: a Docker image
-
-
-
-
-
-
banner.type: string: our custom field through which we will add some entry to the resource logs
-
-
-
-
scope:Namespaced: all resources of this type will exist in a specific Kubernetes Namespacenames:plural: myapps: the resources will be available through/apis/demo.rtfm.co.ua/v1/namespaces/<ns>/myapps/, and how we can “access” the resource (kubectl get myapp), used in RBAC where you need to specifyresources:["myapps"]
singular: myap: alias for convenienceshortNames:[ma]short alias for convenience
Let’s start Minikube:
$ minikube start
Add the CRD:
$ kk apply -f myapp-crd.yaml customresourcedefinition.apiextensions.k8s.io/myapps.demo.rtfm.co.ua created
Let’s look at the Groups API:
$kubectl api-versions ... demo.rtfm.co.ua/v1 ...
And a new resource in this API Group:
$ kubectl api-resources --api-group=demo.rtfm.co.ua NAME SHORTNAMES APIVERSION NAMESPACED KIND myapps ma demo.rtfm.co.ua/v1 true MyApp
OK – we have created a CRD, and now we can even create a CustomResource (CR).
Create the file myapp-example-resource.yaml:
apiVersion: demo.rtfm.co.ua/v1 # matches the CRD's group and version kind: MyApp # kind from the CRD's 'spec.names.kind' metadata: name: example-app # name of this custom resource namespace: default # namespace (CRD has scope: Namespaced) spec: image: nginx:latest # container image to use (from our schema) banner: "This pod was created by MyApp operator 🚀"
Deploy:
$ kk apply -f myapp-example-resource.yaml myapp.demo.rtfm.co.ua/example-app created
And checking:
$ kk get myapp NAME AGE example-app 15s
But there are no resources of type Pod – because we do not have a controller that will work with this type of resources.
Creating a Kubernetes Operator with Kopf
So, we will use Kopf to create a Kubernetes Pod, but using our own CRD.
Create a Python virtual environment:
$ python -m venv venv $ . ./venv/bin/activate (venv)
Add dependencies – requirements.txt file :
kopf kubernetes PyYAML
Install them – with pip or uv:
$ pip install -r requirements.txt
Let’s write the operator code:
import os
import kopf
import kubernetes
import yaml
# use kopf to register a handler for the creation of MyApp custom resources
@kopf.on.create('demo.rtfm.co.ua', 'v1', 'myapps')
# this function will be called when a new MyApp resource is created
def create_myapp(spec, name, namespace, logger, **kwargs):
# get image value from the spec of the CustomResource manifest
image = spec.get('image')
if not image:
raise kopf.PermanentError("Field 'spec.image' must be provided.")
# get optional banner value from the CR manifest spec
banner = spec.get('banner')
# load pod template YAML from file
path = os.path.join(os.path.dirname(__file__), 'pod.yaml')
with open(path, 'rt') as f:
pod_template = f.read()
# render pod YAML with provided values
pod_yaml = pod_template.format(
name=f"{name}-pod",
image=image,
app_name=name,
)
# create Pod difinition from the rendered YAML
# it uses PyYAML to parse the YAML string into a Python dictionary
# which can be used by Kubernetes API client
# it is used to create a Pod object in Kubernetes
pod_spec = yaml.safe_load(pod_yaml)
# inject banner as environment variable if provided
if banner:
# it is used to add a new environment variable into the container spec
container = pod_spec['spec']['containers'][0]
env = container.setdefault('env', [])
env.append({
'name': 'BANNER',
'value': banner
})
# create Kubernetes CoreV1 API client
# used to interact with the Kubernetes API
api = kubernetes.client.CoreV1Api()
try:
# it sends a request to the Kubernetes API to create a new Pod
# uses 'create_namespaced_pod' method to create the Pod in the specified namespace
# 'namespace' is the namespace where the Pod will be created
# 'body' is the Pod specification that was created from the YAML template
api.create_namespaced_pod(namespace=namespace, body=pod_spec)
logger.info(f"Pod {name}-pod created.")
except kubernetes.client.exceptions.ApiException as e:
logger.error(f"Failed to create pod {name}-pod: {e}")
Create a template that will be used by our Operator to create resources:
apiVersion: v1
kind: Pod
metadata:
name: {name}
labels:
app: {app_name}
spec:
containers:
- name: {app_name}
image: {image}
ports:
- containerPort: 80
env:
- name: BANNER
value: "" # will be overridden in code if provided
command: ["/bin/sh", "-c"]
args:
- |
if [ -n "$BANNER" ]; then
echo "$BANNER";
fi
exec sleep infinity
Run the operator with kopf run myoperator.py.
We already have a CustomResource created, and the Operator should see it and create a Kubernetes Pod:
$ kopf run myoperator.py --verbose
...
[2025-07-18 13:59:58,201] kopf._cogs.clients.w [DEBUG ] Starting the watch-stream for customresourcedefinitions.v1.apiextensions.k8s.io cluster-wide.
[2025-07-18 13:59:58,201] kopf._cogs.clients.w [DEBUG ] Starting the watch-stream for myapps.v1.demo.rtfm.co.ua cluster-wide.
[2025-07-18 13:59:58,305] kopf.objects [DEBUG ] [default/example-app] Creation is in progress: {'apiVersion': 'demo.rtfm.co.ua/v1', 'kind': 'MyApp', 'metadata': {'annotations': {'kubectl.kubernetes.io/last-applied-configuration': '{"apiVersion":"demo.rtfm.co.ua/v1","kind":"MyApp","metadata":{"annotations":{},"name":"example-app","namespace":"default"},"spec":{"banner":"This pod was created by MyApp operator 🚀","image":"nginx:latest","replicas":3}}\n'}, 'creationTimestamp': '2025-07-18T09:55:42Z', 'generation': 2, 'managedFields': [{'apiVersion': 'demo.rtfm.co.ua/v1', 'fieldsType': 'FieldsV1', 'fieldsV1': {'f:metadata': {'f:annotations': {'.': {}, 'f:kubectl.kubernetes.io/last-applied-configuration': {}}}, 'f:spec': {'.': {}, 'f:banner': {}, 'f:image': {}, 'f:replicas': {}}}, 'manager': 'kubectl-client-side-apply', 'operation': 'Update', 'time': '2025-07-18T10:48:27Z'}], 'name': 'example-app', 'namespace': 'default', 'resourceVersion': '2955', 'uid': '8b674a99-05ab-4d4b-8205-725de450890a'}, 'spec': {'banner': 'This pod was created by MyApp operator 🚀', 'image': 'nginx:latest', 'replicas': 3}}
...
[2025-07-18 13:59:58,325] kopf.objects [INFO ] [default/example-app] Pod example-app-pod created.
[2025-07-18 13:59:58,326] kopf.objects [INFO ] [default/example-app] Handler 'create_myapp' succeeded.
...
Check the Pod:
$ kk get pod NAME READY STATUS RESTARTS AGE example-app-pod 1/1 Running 0 68s
And its logs:
$ kk logs -f example-app-pod This pod was created by MyApp operator 🚀
So, the Operator launched the Pod using our CustomResource in which he took the spec.banner field with the string “This pod was created by MyApp operator 🚀“, and executed the command /bin/sh -c " $BANNER” command in the pod.
Resource templates: Kopf and Kubebuilder
Instead of having a separate pod-template.yaml file, we could describe everything directly in the operator code.
That is, you can describe something like this:
...
# get optional banner value
banner = spec.get('banner', '')
# define Pod spec as a Python dict
pod_spec = {
"apiVersion": "v1",
"kind": "Pod",
"metadata": {
"name": f"{name}-pod",
"labels": {
"app": name,
},
},
"spec": {
"containers": [
{
"name": name,
"image": image,
"env": [
{
"name": "BANNER",
"value": banner
}
],
"command": ["/bin/sh", "-c"],
"args": [f'echo "$BANNER"; exec sleep infinity'],
"ports": [
{
"containerPort": 80
}
]
}
]
}
}
# create Kubernetes API client
api = kubernetes.client.CoreV1Api()
...
And in the case of Kubebuilder, a function is usually created that uses the CustomResource manifest (cr *myappv1.MyApp) and forms an object of type *corev1.Pod using the Go structures corev1.PodSpec and corev1.Container:
...
// newPod is a helper function that builds a Kubernetes Pod object
// based on the custom MyApp resource. It returns a pointer to corev1.Pod,
// which is later passed to controller-runtime's client.Create(...) to create the Pod in the cluster.
func newPod(cr *myappv1.MyApp) *corev1.Pod {
// `cr` is a pointer to your CustomResource of kind MyApp
// type MyApp is generated by Kubebuilder and lives in your `api/v1/myapp_types.go`
// it contains fields like cr.Spec.Image, cr.Spec.Banner, cr.Name, cr.Namespace, etc.
return &corev1.Pod{
// corev1.Pod is a Go struct representing the built-in Kubernetes Pod type
// it's defined in "k8s.io/api/core/v1" package (aliased here as corev1)
// we return a pointer to it (`*corev1.Pod`) because client-go methods like
// `client.Create()` expect pointer types
ObjectMeta: metav1.ObjectMeta{
// metav1.ObjectMeta comes from "k8s.io/apimachinery/pkg/apis/meta/v1"
// it defines metadata like name, namespace, labels, annotations, ownerRefs, etc.
Name: cr.Name + "-pod", // generate Pod name based on the CR's name
Namespace: cr.Namespace, // place the Pod in the same namespace as the CR
Labels: map[string]string{ // set a label for identification or selection
"app": cr.Name, // e.g., `app=example-app`
},
},
Spec: corev1.PodSpec{
// corev1.PodSpec defines everything about how the Pod runs
// including containers, volumes, restart policy, etc.
Containers: []corev1.Container{
// define a single container inside the Pod
{
Name: cr.Name, // use CR name as container name (must be DNS compliant)
Image: cr.Spec.Image, // container image (e.g., "nginx:1.25")
Env: []corev1.EnvVar{
// corev1.EnvVar is a struct that defines environment variables
{
Name: "BANNER", // name of the variable
Value: cr.Spec.Banner, // value from the CR spec
},
},
Command: []string{"/bin/sh", "-c"},
// override container ENTRYPOINT to run a shell command
Args: []string{
// run a command that prints the banner and sleeps forever
// fmt.Sprintf(...) injects the value at runtime into the string
fmt.Sprintf(`echo "%s"; exec sleep infinity`, cr.Spec.Banner),
},
// optional: could also add ports, readiness/liveness probes, etc.
},
},
},
}
}
...
And what about in real operators?
But we did this for “internal” Kubernetes resources.
What about external resources?
Here’s just an example – I haven’t tested it, but the general idea is this: just take an SDK (in the Python example, it’s boto3), and using the fields from the CustomResource (for example, subnets or schema), make the appropriate API requests to AWS through the SDK.
An example of such a CustomResource:
apiVersion: demo.rtfm.co.ua/v1 kind: MyIngress metadata: name: myapp spec: subnets: - subnet-abc - subnet-def scheme: internet-facing
And the code that could create an AWS ALB from it:
import kopf
import boto3
import botocore
import logging
# create a global boto3 client for AWS ELBv2 service
# this client will be reused for all requests from the operator
# NOTE: region must match where your subnets and VPC exist
elbv2 = boto3.client("elbv2", region_name="us-east-1")
# define a handler that is triggered when a new MyIngress resource is created
@kopf.on.create('demo.rtfm.co.ua', 'v1', 'myingresses')
def create_ingress(spec, name, namespace, status, patch, logger, **kwargs):
# extract the list of subnet IDs from the CustomResource 'spec.subnets' field
# these subnets must belong to the same VPC and be public if scheme=internet-facing
subnets = spec.get('subnets')
# extract optional scheme (default to 'internet-facing' if not provided)
scheme = spec.get('scheme', 'internet-facing')
# validate input: at least 2 subnets are required to create an ALB
if not subnets:
raise kopf.PermanentError("spec.subnets is required.")
# attempt to create an ALB in AWS using the provided spec
# using the boto3 ELBv2 client
try:
response = elbv2.create_load_balancer(
Name=f"{name}-alb", # ALB name will be derived from CR name
Subnets=subnets, # list of subnet IDs provided by user
Scheme=scheme, # 'internet-facing' or 'internal'
Type='application', # we are creating an ALB (not NLB)
IpAddressType='ipv4', # only IPv4 supported here (could be 'dualstack')
Tags=[ # add tags for ownership tracking
{'Key': 'ManagedBy', 'Value': 'kopf'},
]
)
except botocore.exceptions.ClientError as e:
# if AWS API fails (e.g. invalid subnet, quota exceeded), retry later
raise kopf.TemporaryError(f"Failed to create ALB: {e}", delay=30)
# parse ALB metadata from AWS response
lb = response['LoadBalancers'][0] # ALB list should contain exactly one entry
dns_name = lb['DNSName'] # external DNS of the ALB (e.g. abc.elb.amazonaws.com)
arn = lb['LoadBalancerArn'] # unique ARN of the ALB (used for deletion or listeners)
# log the creation for operator diagnostics
logger.info(f"Created ALB: {dns_name}")
# save ALB info into the CustomResource status field
# this updates .status.alb.dns and .status.alb.arn in the CR object
patch.status['alb'] = {
'dns': dns_name,
'arn': arn,
}
# return a dict, will be stored in the finalizer state
# used later during deletion to clean up the ALB
return {'alb-arn': arn}
In the case of Go and Kubebuilder, we would use the aws-sdk-go library:
import (
"context"
"fmt"
elbv2 "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
"github.com/aws/aws-sdk-go-v2/aws"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
networkingv1 "k8s.io/api/networking/v1"
)
func newALB(ctx context.Context, client *elbv2.Client, cr *networkingv1.Ingress) (string, error) {
// build input for the ALB
input := &elbv2.CreateLoadBalancerInput{
Name: aws.String(fmt.Sprintf("%s-alb", cr.Name)),
Subnets: []string{"subnet-abc123", "subnet-def456"}, // replace with real subnets
Scheme: elbv2.LoadBalancerSchemeEnumInternetFacing,
Type: elbv2.LoadBalancerTypeEnumApplication,
IpAddressType: elbv2.IpAddressTypeIpv4,
Tags: []types.Tag{
{
Key: aws.String("ManagedBy"),
Value: aws.String("MyIngressOperator"),
},
},
}
// create ALB
output, err := client.CreateLoadBalancer(ctx, input)
if err != nil {
return "", fmt.Errorf("failed to create ALB: %w", err)
}
if len(output.LoadBalancers) == 0 {
return "", fmt.Errorf("ALB was not returned by AWS")
}
// return the DNS name of the ALB
return aws.ToString(output.LoadBalancers[0].DNSName), nil
}
In the real AWS ALB Ingress Controller, the creation of an ALB is called in the elbv2.go file :
...
func (c *elbv2Client) CreateLoadBalancerWithContext(ctx context.Context, input *elasticloadbalancingv2.CreateLoadBalancerInput) (*elasticloadbalancingv2.CreateLoadBalancerOutput, error) {
client, err := c.getClient(ctx, "CreateLoadBalancer")
if err != nil {
return nil, err
}
return client.CreateLoadBalancer(ctx, input)
}
...
Actually, that’s all there is to it.
