LAB07: Building Own Operator

LAB70 Building Own Operator (CRDs)

This lab implements custom controller which watch ingress object with specific annotation and creates the HAProxy config

Below is a high-level architecture diagram

High-Level Architecture Diagram (Text + Visual Logic)

                    ┌───────────────────────────────────────────┐
                    │            Your VPC / Network              │
                    └───────────────────────────────────────────┘
                                     │
                                     ▼
                         ┌─────────────────────┐
                         │  HAProxy Server     │
                         │  (External VM)      │
                         │  /etc/haproxy/...   │
                         └─────────────────────┘
                                     ▲
                                     │ updates via Ansible
                                     │
                       ┌─────────────────────────────────┐
                       │  HAProxy Config Operator (Pod)   │
                       │  - Watches Ingress objects       │
                       │  - Extracts rules                │
                       │  - Resolves NodePorts + NodeIPs  │
                       │  - Renders HAProxy blocks        │
                       │  - Calls ansible-playbook        │
                       └─────────────────────────────────┘
                                     │
                                     │ watches
                                     ▼
                       ┌─────────────────────────────────┐
                       │        Kubernetes API Server     │
                       │----------------------------------│
                       │ • Ingress (w/ annotation)        │
                       │ • Services (NodePort)            │
                       │ • Endpoints                      │
                       │ • Nodes (InternalIP)             │
                       └─────────────────────────────────┘
                                     │
                        routes via   │ NodePort
                                     ▼
          ┌──────────────────────────────────────────────────────────────────┐
          │                            Worker Nodes                          │
          │──────────────────────────────────────────────────────────────────│
          │    ┌────────────────────┐      ┌────────────────────┐           │
          │    │  hello-app Pod     │      │ nginx-demo Pod     │           │
          │    └────────────────────┘      └────────────────────┘           │
          │         ^     ^                      ^     ^                    │
          │         │     │                      │     │                    │
          │  NodePort SVC │              NodePort SVC │                    │
          │  hello-app-svc│              nginx-demo-svc│                   │
          └──────────────────────────────────────────────────────────────────┘

---

The Flow Explained (Short + Clear)

1. Developer creates an Ingress

With annotation:

haproxy.deepak.dev/enabled: "true"

Example host:

• hello-app.demo.local

---

2. Operator notices the annotated Ingress

It extracts:

• host

• path

• service name

• service port

---

3. Operator fetches service details

If service is NodePort:

• Collects all Node Internal IPs

• Gets assigned NodePort (e.g., 31625)

---

4. Operator generates HAProxy config

Example:

frontend fe_default_hello-app-ing
  bind *:80
  acl host_1 hdr(host) -i hello-app.demo.local
  use_backend be_default_hello-app-ing_1 if host_1

backend be_default_hello-app-ing_1
  server node1 <node-ip1>:31625
  server node2 <node-ip2>:31625
  server node3 <node-ip3>:31625

---

5. Operator pushes config using Ansible

ansible-playbook -i <haproxy_ip>,

• Inserts block into /etc/haproxy/haproxy.cfg

• Validates

• Reloads HAProxy

---

6. External user request

User hits:

curl -H "Host: hello-app.demo.local" http://<HAProxy-IP>

HAProxy forwards the traffic to a random node:

→ <node-ip>:31625 → hello-app Pod

---

Controller Implementation

::: spoiler Python controller using Kopf requirements.txt

kopf==1.37.4
kubernetes==30.1.0
pyyaml==6.0.2

controller.py

import json
import os
import subprocess
from typing import Any, Dict, List, Optional

import kopf
import kubernetes
from kubernetes import client
from kubernetes.client.exceptions import ApiException


ANNOT_PREFIX = "haproxy.deepak.dev"


def get_annotation(meta: Dict[str, Any], key: str, default=None):
    ann = meta.get("annotations") or {}
    return ann.get(f"{ANNOT_PREFIX}/{key}", default)


def extract_backend_from_ingress(spec: Dict[str, Any]) -> List[Dict[str, Any]]:
    results = []
    rules = spec.get("rules") or []
    for rule in rules:
        host = rule.get("host")
        http = rule.get("http") or {}
        paths = http.get("paths") or []
        for p in paths:
            backend = p.get("backend") or {}
            service = backend.get("service") or {}
            svc_name = service.get("name")
            port = service.get("port") or {}
            svc_port = port.get("number") or port.get("name")
            results.append({
                "host": host,
                "path": p.get("path", "/"),
                "service_name": svc_name,
                "service_port": svc_port,
            })
    return results


def get_node_ips():
    api = client.CoreV1Api()
    nodes = api.list_node().items
    ips = []
    for node in nodes:
        for addr in node.status.addresses:
            if addr.type == "InternalIP":
                ips.append(addr.address)
    return ips


def resolve_nodeport(namespace, service_name, logger):
    """Return NodePort if available, else None (and do NOT fail)."""
    api = client.CoreV1Api()

    try:
        svc = api.read_namespaced_service(service_name, namespace)
    except ApiException as e:
        logger.warning(f"Service {service_name} not found: {e.reason}")
        return None

    if not svc.spec.ports:
        logger.warning(f"Service {service_name} has no ports defined.")
        return None

    for port in svc.spec.ports:
        if port.node_port:
            return port.node_port

    logger.warning(f"Service {service_name} has no NodePort.")
    return None


def run_ansible(haproxy_host, playbook, extra_vars, logger):
    extra_vars_json = json.dumps(extra_vars)

    cmd = [
        "ansible-playbook",
        "-i", f"{haproxy_host},",
        "--user", "ubuntu",
        playbook,
        "--extra-vars", extra_vars_json,
    ]

    logger.info(f"Running Ansible: {' '.join(cmd)}")

    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        env=os.environ,
    )

    logger.info(f"STDOUT:\n{result.stdout}")
    if result.stderr:
        logger.warning(f"STDERR:\n{result.stderr}")

    if result.returncode != 0:
        raise kopf.TemporaryError(f"ansible-playbook failed with exit code {result.returncode}", delay=60)


def patch_ingress_status(name, namespace, external_ip, logger):
    api = client.NetworkingV1Api()
    body = {
        "status": {
            "loadBalancer": {
                "ingress": [{"ip": external_ip}]
            }
        }
    }
    api.patch_namespaced_ingress_status(name, namespace, body)
    logger.info(f"Updated ingress.status for {namespace}/{name} → {external_ip}")


def reconcile_ingress(body, spec, meta, status, logger, **kwargs):

    enabled = get_annotation(meta, "enabled", "false")
    if enabled.lower() != "true":
        logger.info("Skipping: not enabled for HAProxy")
        return

    name = meta["name"]
    namespace = meta["namespace"]

    haproxy_host = os.environ.get("HAPROXY_HOST")
    external_ip = haproxy_host
    playbook = "/ansible/haproxy-update.yml"

    if not haproxy_host:
        raise kopf.PermanentError("Missing env: HAPROXY_HOST")

    logger.info(f"Reconciling {namespace}/{name} for HAProxy {haproxy_host}")

    backends = extract_backend_from_ingress(spec)
    if not backends:
        raise kopf.PermanentError("No backends found in ingress")

    node_ips = get_node_ips()

    # --- Resolve NodePorts gracefully ---
    for b in backends:
        node_port = resolve_nodeport(namespace, b["service_name"], logger)
        if not node_port:
            logger.warning(f"Skipping backend: {b['service_name']} has no NodePort.")
            continue
        b["node_port"] = node_port
        b["node_ips"] = node_ips

    # --- FILTER VALID BACKENDS ---
    valid_backends = [b for b in backends if b.get("node_port")]

    if not valid_backends:
        logger.warning(f"No usable backends for ingress {name}. Skipping HAProxy update.")
        return  # Do NOT run Ansible, Do NOT retry

    extra_vars = {
        "ingress_name": name,
        "ingress_namespace": namespace,
        "haproxy_external_ip": external_ip,
        "backends": valid_backends,
    }

    run_ansible(haproxy_host, playbook, extra_vars, logger)

    patch_ingress_status(name, namespace, external_ip, logger)

    logger.info(f"Reconciliation complete for {namespace}/{name}")


@kopf.on.startup()
def configure(settings: kopf.OperatorSettings, **_):
    kubernetes.config.load_incluster_config()
    settings.persistence.finalizer = f"{ANNOT_PREFIX}/finalizer"


@kopf.on.create("networking.k8s.io", "v1", "ingresses")
def on_create(body, spec, meta, status, logger, **kwargs):
    return reconcile_ingress(body, spec, meta, status, logger, **kwargs)


@kopf.on.update("networking.k8s.io", "v1", "ingresses")
def on_update(body, spec, meta, status, logger, **kwargs):
    return reconcile_ingress(body, spec, meta, status, logger, **kwargs)


@kopf.on.resume("networking.k8s.io", "v1", "ingresses")
def on_resume(body, spec, meta, status, logger, **kwargs):
    return reconcile_ingress(body, spec, meta, status, logger, **kwargs)


@kopf.on.delete("networking.k8s.io", "v1", "ingresses")
def on_delete(body, spec, meta, status, logger, **kwargs):
    logger.info("Ingress deleted: cleanup logic not implemented yet.")

Dockerfile

FROM python:3.11-slim

# Install system packages required by ansible + ssh
RUN apt-get update && apt-get install -y \
    openssh-client \
    git \
    curl \
    ansible \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/*

# Make a working directory
WORKDIR /app

# Install python libs
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy controller code
COPY controller.py /app/controller.py

# Copy optional ansible playbooks (if you want)
COPY ansible /ansible

# Entry point for Kopf
CMD ["kopf", "run", "--standalone", "/app/controller.py"]

:::

Deployment Manifests

1. 00-namespace.yaml

apiVersion: v1
kind: Namespace
metadata:
  name: haproxy-operator

1. 02-rbac.yaml

apiVersion: v1
kind: ServiceAccount
metadata:
  name: haproxy-operator
  namespace: haproxy-operator
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: haproxy-operator
rules:
  # Operator needs to read CRDs (kopf requires this)
  - apiGroups: ["apiextensions.k8s.io"]
    resources: ["customresourcedefinitions"]
    verbs: ["get", "list", "watch"]

  # Watch all ingresses across namespaces
  - apiGroups: ["networking.k8s.io"]
    resources: ["ingresses"]
    verbs: ["get", "list", "watch", "patch", "update"]

  # Patch ingress/status
  - apiGroups: ["networking.k8s.io"]
    resources: ["ingresses/status"]
    verbs: ["patch", "update"]

  # Allow watching events (kopf uses this)
  - apiGroups: [""]
    resources: ["events"]
    verbs: ["create", "patch", "update"]

  # Allow reading namespaces (optional but useful)
  - apiGroups: [""]
    resources: ["namespaces"]
    verbs: ["get", "list", "watch"]

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: haproxy-operator
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: haproxy-operator
subjects:
  - kind: ServiceAccount
    name: haproxy-operator
    namespace: haproxy-operator

1. 03-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: haproxy-config-updater
  namespace: haproxy-operator
spec:
  replicas: 1
  selector:
    matchLabels:
      app: haproxy-config-updater
  template:
    metadata:
      labels:
        app: haproxy-config-updater
    spec:
      serviceAccountName: haproxy-operator
      containers:
        - name: operator
          image: openlabfree/haproxy-config-updater:latest
          imagePullPolicy: Always
          command: ["kopf", "run", "--standalone", "/app/controller.py"]

          # ✨ Environment variables used by operator (replaces annotations)
          env:
            - name: HAPROXY_HOST
              value: "34.237.52.128"         # your HAProxy server IP
            - name: HAPROXY_EXTERNAL_IP
              value: "34.237.52.128"         # Ingress.status.loadBalancer.ingress[0].ip
            - name: ANSIBLE_INVENTORY
              value: "/etc/ansible/hosts"
            - name: ANSIBLE_PLAYBOOK
              value: "/ansible/haproxy-update.yml"

          volumeMounts:
            - name: ssh
              mountPath: /root/.ssh
              readOnly: true
            - name: ansible
              mountPath: /ansible

      volumes:
        - name: ssh
          secret:
            secretName: haproxy-operator-ssh
            defaultMode: 0600

        - name: ansible
          configMap:
            name: haproxy-operator-playbooks

1. 04-ssh-secret.yaml

apiVersion: v1
kind: Secret
metadata:
  name: haproxy-operator-ssh
  namespace: haproxy-operator
type: Opaque
data:
  id_rsa: <base64_of_ssh_key>
  known_hosts: <base64_of_knownhostfile> 

1. 05-playbook-configmap.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  name: haproxy-operator-playbooks
  namespace: haproxy-operator
data:

  haproxy-update.yml: |
    ---
    - name: Update HAProxy configuration for Kubernetes Ingress
      hosts: all
      remote_user: ubuntu
      become: yes

      vars:
        haproxy_config_dir: "/etc/haproxy/conf.d"

      tasks:

        - name: Ensure HAProxy conf.d directory exists
          file:
            path: "{{ haproxy_config_dir }}"
            state: directory
            mode: '0755'

        - name: Create backend configuration for this ingress
          template:
            src: "haproxy-backend.j2"
            dest: "{{ haproxy_config_dir }}/{{ ingress_namespace }}-{{ ingress_name }}.cfg"
            mode: "0644"
          notify: Reload HAProxy

        - name: Validate HAProxy configuration
          command: haproxy -c -f /etc/haproxy/haproxy.cfg
          register: haproxy_check
          failed_when: haproxy_check.rc != 0
          changed_when: false

      handlers:
        - name: Reload HAProxy
          service:
            name: haproxy
            state: reloaded


  haproxy-remove.yml: |
    ---
    - name: Remove HAProxy configuration for deleted Ingress
      hosts: all
      remote_user: ubuntu
      become: yes

      vars:
        haproxy_config_dir: "/etc/haproxy/conf.d"

      tasks:
        - name: Delete backend config file
          file:
            path: "{{ haproxy_config_dir }}/{{ ingress_namespace }}-{{ ingress_name }}.cfg"
            state: absent
          notify: Reload HAProxy

        - name: Validate HAProxy configuration after removal
          command: haproxy -c -f /etc/haproxy/haproxy.cfg
          register: haproxy_check
          failed_when: haproxy_check.rc != 0
          changed_when: false

      handlers:
        - name: Reload HAProxy
          service:
            name: haproxy
            state: reloaded


  haproxy-backend.j2: |
    # ---------------------------------------------
    # Auto-generated by Kubernetes HAProxy Operator
    # Ingress: {{ ingress_namespace }}/{{ ingress_name }}
    # External IP: {{ haproxy_external_ip }}
    # ---------------------------------------------

    # FRONTEND SECTION
    frontend {{ ingress_namespace }}_{{ ingress_name }}_frontend
        bind *:80
    {% for item in backends %}
        acl host_{{ loop.index }} hdr(host) -i {{ item.host }}
        acl path_{{ loop.index }} path_beg {{ item.path }}
        use_backend {{ ingress_namespace }}_{{ ingress_name }}_backend_{{ loop.index }} if host_{{ loop.index }} path_{{ loop.index }}
    {% endfor %}

    # BACKEND SECTION
    {% for item in backends %}
    backend {{ ingress_namespace }}_{{ ingress_name }}_backend_{{ loop.index }}
        mode http
        option httpchk GET /
        # Using internal K8s DNS (ClusterIP)
        server svc {{ item.service_name }}.{{ ingress_namespace }}.svc.cluster.local:{{ item.service_port }} check resolve-prefer ipv4
    {% endfor %}

Test

1. nginx-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx-demo
  namespace: default
spec:
  replicas: 1
  selector:
    matchLabels:
      app: nginx-demo
  template:
    metadata:
      labels:
        app: nginx-demo
    spec:
      containers:
        - name: nginx
          image: nginx:stable
          ports:
            - containerPort: 80

1. ingress.yaml

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: nginx-demo-ing
  namespace: default
  annotations:
    haproxy.deepak.dev/enabled: "true"
spec:
  rules:
    - host: nginx.demo.local
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: nginx-demo-svc
                port:
                  number: 80

1. service.yaml

apiVersion: v1
kind: Service
metadata:
  name: nginx-demo-svc
  namespace: default
spec:
  type: ClusterIP
  selector:
    app: nginx-demo
  ports:
    - port: 80
      targetPort: 80
      protocol: TCP