A simple Python script to backup all Kubernetes resource manifests in your cluster

backup-kubernetes-manifests.py

import json
import subprocess
from dataclasses import dataclass
from pathlib import Path

from typing import Set

MANIFESTS_PATH = "exported-manifests"


@dataclass
class Resource:
    api_version: str
    namespace: str
    name: str


def get_api_resources() -> Set[str]:
    output = subprocess.check_output(["kubectl", "api-resources", "-o", "name"])
    api_resources = output.split()

    return {ar.decode() for ar in api_resources if ar.decode()}


def get_resources(api_resources: Set[str]) -> [Resource]:
    resources: [Resource] = []

    for api_resource in api_resources:
        try:
            output = subprocess.check_output(
                ["kubectl", "get", api_resource, "--all-namespaces", "-o", "json"]
            )
        except subprocess.CalledProcessError:
            continue

        for resource in json.loads(output)["items"]:
            resources.append(
                Resource(
                    api_version=api_resource,
                    namespace=resource["metadata"].get("namespace"),
                    name=resource["metadata"]["name"],
                )
            )

    return resources


def export_resource_manifests(resources: [Resource]):
    for idx, resource in enumerate(resources):
        if idx % 25 == 0:
            print(f"Processing {idx} of {len(resources)} resources")
        
        export_resource_manifest(resource)


def export_resource_manifest(resource: Resource):
    ns = resource.namespace if resource.namespace else "global"
    path = Path(MANIFESTS_PATH) / ns / resource.api_version

    Path(path).mkdir(parents=True, exist_ok=True)
    file_path = path / f"{resource.name}.yaml"
    if file_path.exists():
        return

    cmd = ["kubectl", "get", resource.api_version, resource.name, "-o", "yaml"]
    if resource.namespace:
        cmd += ["-n", resource.namespace]

    try:
        manifest = subprocess.check_output(cmd)
    except:
        print(cmd)
        raise

    with open(file_path, "wb") as f:
        f.write(manifest)


if __name__ == "__main__":
    api_resources = get_api_resources()
    resources = get_resources(api_resources)
    export_resource_manifests(resources)