Skip to content

Write a Minimal Mutating Webhook

In this article, let's write a minial mutating webhook that add a new label to the targeted pods.

Folder Structure

├── deploy
   ├── test.yml
   └── webhook.yml
├── requirements.txt
└── webhook


Install Requirements

=== requirements.txt


Webhook Codes

We use fastapi to develop the webhook.

Webhook Codes

from typing import Optional

from pydantic import BaseModel

class WebhookRequestDetail(BaseModel):
    uid: str
    kind: dict
    resource: dict
    name: Optional[str]
    namespace: str
    operation: str
    object: dict

class WebhookRequest(BaseModel):
    apiVersion: str
    kind: str
    request: WebhookRequestDetail

class WebhookResponseDetail(BaseModel):
    allowed: bool
    uid: str
    patch: Optional[str]
    patchType: Optional[str]

class WebhookResponse(BaseModel):
    apiVersion: str
    kind: str
    response: WebhookResponseDetail
from kubernetes.client.api_client import ApiClient

class Converter(ApiClient):
    def from_dict(self, data, kind):
        return self._ApiClient__deserialize(data, kind)
import base64
import logging
from typing import Optional

import jsonpatch
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from loguru import logger

from .k8s import Converter
from .models import WebhookRequest, WebhookResponse, WebhookResponseDetail


app = FastAPI()
    "/mutate-pods", response_model=WebhookResponse, response_model_exclude_unset=True
async def mutate_pods(webhook_request: WebhookRequest, timeout: Optional[str] = "10s"):
    resp = WebhookResponse(
        response=WebhookResponseDetail(allowed=True, uid=webhook_request.request.uid),
    pod = Converter().from_dict(webhook_request.request.object, "V1Pod")
    original = pod.to_dict()
        pod.metadata.labels["is_patched"] = "true"
    except KeyError as exc:
        logger.error("Patch failed: key not found")
        return resp
    # patch = jsonpatch.JsonPatch.from_diff(webhook_request.request.object, patched_pod)
    patch = jsonpatch.JsonPatch.from_diff(original, pod.to_dict())
    resp.response.patchType = "JSONPatch"
    resp.response.patch = base64.b64encode(str(patch).encode()).decode()"Response.apiVersion: {}", resp.apiVersion)"Response.kind: {}", resp.kind)"Response.response.allowed: {}", resp.response.allowed)"Response.response.uid: {}", resp.response.uid)"Response.response.patch: {}", resp.response.patch)"Response.response.patchType: {}", resp.response.patchType)
    return resp

Run the Server

$ uvicorn --reload --host= --port=8000 --debug
INFO:     Will watch for changes in these directories: ['/home/test/webhook']
INFO:     Uvicorn running on (Press CTRL+C to quit)
INFO:     Started reloader process [31408] using watchgod
INFO:     Started server process [31417]
INFO:     Waiting for application startup.
INFO:     Application startup complete.