Clone Kubernetes objects programmatically using the Python API

Question

The Python API is available to read objects from a cluster. By cloning we can say:

1. Get a copy of an existing Kubernetes object using kubectl get
2. Change the properties of the object
3. Apply the new object

Until recently, the option to --export api was deprecated in 1.14. How can we use the Python Kubernetes API to do the steps from 1-3 described above?

There are multiple questions about how to extract the code from Python API to YAML, but it's unclear how to transform the Kubernetes API object.

Confetti

Answer 1

Just use to_dict() which is now offered by Kubernetes Client objects. Note that it creates a partly deep copy. So to be safe:

copied_obj = copy.deepcopy(obj.to_dict())

Dicts can be passed to create* and patch* methods.

For convenience, you can also wrap the dict in Prodict.

copied_obj = Prodict.from_dict(copy.deepcopy(obj.to_dict()))

The final issue is getting rid of superfluous fields. (Unfortunately, Kubernetes sprinkles them throughout the object.) I use kopf's internal facility for getting the "essence" of an object. (It takes care of the deep copy.)

copied_obj = kopf.AnnotationsDiffBaseStorage().build(body=kopf.Body(obj.to_dict()))
copied_obj = Prodic.from_dict(copied_obj)

Hampton

Thank you Aleksandr ... I have now made your answer the preferred version! – Confetti

Answer 2

After looking at the requirement, I spent a couple of hours researching the Kubernetes Python API. Issue 340 and others ask about how to transform the Kubernetes API object into a dict, but the only workaround I found was to retrieve the raw data and then convert to JSON.

• The following code uses the Kubernetes API to get a deployment and its related hpa from the namespaced objects, but retrieving their raw values as JSON.
• Then, after transforming the data into a dict, you can alternatively clean up the data by removing null references.
• Once you are done, you can transform the dict as YAML payload to then save the YAML to the file system
• Finally, you can apply either using kubectl or the Kubernetes Python API.

Note:

• Make sure to set KUBECONFIG=config so that you can point to a cluster
• Make sure to adjust the values of origin_obj_name = "istio-ingressgateway" and origin_obj_namespace = "istio-system" with the name of the corresponding objects to be cloned in the given namespace.

import os
import logging
import yaml
import json
logging.basicConfig(level = logging.INFO)

import crayons
from kubernetes import client, config
from kubernetes.client.rest import ApiException

LOGGER = logging.getLogger(" IngressGatewayCreator ")

class IngressGatewayCreator:

    @staticmethod
    def clone_default_ingress(clone_context):
        # Clone the deployment
        IngressGatewayCreator.clone_deployment_object(clone_context)

        # Clone the deployment's HPA
        IngressGatewayCreator.clone_hpa_object(clone_context)

    @staticmethod
    def clone_deployment_object(clone_context):
        kubeconfig = os.getenv('KUBECONFIG')
        config.load_kube_config(kubeconfig)
        v1apps = client.AppsV1beta1Api()

        deployment_name = clone_context.origin_obj_name
        namespace = clone_context.origin_obj_namespace

        try:
            # gets an instance of the api without deserialization to model
            # https://github.com/kubernetes-client/python/issues/574#issuecomment-405400414
            deployment = v1apps.read_namespaced_deployment(deployment_name, namespace, _preload_content=False)

        except ApiException as error:
            if error.status == 404:
                LOGGER.info("Deployment %s not found in namespace %s", deployment_name, namespace)
                return
            raise

        # Clone the object deployment as a dic
        cloned_dict = IngressGatewayCreator.clone_k8s_object(deployment, clone_context)

        # Change additional objects
        cloned_dict["spec"]["selector"]["matchLabels"]["istio"] = clone_context.name
        cloned_dict["spec"]["template"]["metadata"]["labels"]["istio"] = clone_context.name

        # Save the deployment template in the output dir
        context.save_clone_as_yaml(cloned_dict, "deployment")

    @staticmethod
    def clone_hpa_object(clone_context):
        kubeconfig = os.getenv('KUBECONFIG')
        config.load_kube_config(kubeconfig)
        hpas = client.AutoscalingV1Api()

        hpa_name = clone_context.origin_obj_name
        namespace = clone_context.origin_obj_namespace

        try:
            # gets an instance of the api without deserialization to model
            # https://github.com/kubernetes-client/python/issues/574#issuecomment-405400414
            hpa = hpas.read_namespaced_horizontal_pod_autoscaler(hpa_name, namespace, _preload_content=False)

        except ApiException as error:
            if error.status == 404:
                LOGGER.info("HPA %s not found in namespace %s", hpa_name, namespace)
                return
            raise

        # Clone the object deployment as a dic
        cloned_dict = IngressGatewayCreator.clone_k8s_object(hpa, clone_context)

        # Change additional objects
        cloned_dict["spec"]["scaleTargetRef"]["name"] = clone_context.name

        # Save the deployment template in the output dir
        context.save_clone_as_yaml(cloned_dict, "hpa")

    @staticmethod
    def clone_k8s_object(k8s_object, clone_context):
        # Manipilate in the dict level, not k8s api, but from the fetched raw object
        # https://github.com/kubernetes-client/python/issues/574#issuecomment-405400414
        cloned_obj = json.loads(k8s_object.data)

        labels = cloned_obj['metadata']['labels']
        labels['istio'] = clone_context.name

        cloned_obj['status'] = None

        # Scrub by removing the "null" and "None" values
        cloned_obj = IngressGatewayCreator.scrub_dict(cloned_obj)

        # Patch the metadata with the name and labels adjusted
        cloned_obj['metadata'] = {
            "name": clone_context.name,
            "namespace": clone_context.origin_obj_namespace,
            "labels": labels
        }

        return cloned_obj

    # https://mcmap.net/q/41196/-efficient-way-to-remove-keys-with-empty-strings-from-a-dict/59959570#59959570
    @staticmethod
    def scrub_dict(d):
        new_dict = {}
        for k, v in d.items():
            if isinstance(v, dict):
                v = IngressGatewayCreator.scrub_dict(v)
            if isinstance(v, list):
                v = IngressGatewayCreator.scrub_list(v)
            if not v in (u'', None, {}):
                new_dict[k] = v
        return new_dict

    # https://mcmap.net/q/41196/-efficient-way-to-remove-keys-with-empty-strings-from-a-dict/59959570#59959570
    @staticmethod
    def scrub_list(d):
        scrubbed_list = []
        for i in d:
            if isinstance(i, dict):
                i = IngressGatewayCreator.scrub_dict(i)
            scrubbed_list.append(i)
        return scrubbed_list


class IngressGatewayContext:

    def __init__(self, manifest_dir, name, hostname, nats, type):
        self.manifest_dir = manifest_dir
        self.name = name
        self.hostname = hostname
        self.nats = nats
        self.ingress_type = type

        self.origin_obj_name = "istio-ingressgateway"
        self.origin_obj_namespace = "istio-system"

    def save_clone_as_yaml(self, k8s_object, kind):
        try:
            # Just try to create if it doesn't exist
            os.makedirs(self.manifest_dir)

        except FileExistsError:
            LOGGER.debug("Dir already exists %s", self.manifest_dir)

        full_file_path = os.path.join(self.manifest_dir, self.name + '-' + kind + '.yaml')

        # Store in the file-system with the name provided
        # https://mcmap.net/q/134599/-how-can-i-write-data-in-yaml-format-in-a-file/18210750#18210750
        with open(full_file_path, 'w') as yaml_file:
            yaml.dump(k8s_object, yaml_file, default_flow_style=False)

        LOGGER.info(crayons.yellow("Saved %s '%s' at %s: \n%s"), kind, self.name, full_file_path, k8s_object)

try:
    k8s_clone_name = "http2-ingressgateway"
    hostname = "my-nlb-awesome.a.company.com"
    nats = ["123.345.678.11", "333.444.222.111", "33.221.444.23"]
    manifest_dir = "out/clones"

    context = IngressGatewayContext(manifest_dir, k8s_clone_name, hostname, nats, "nlb")

    IngressGatewayCreator.clone_default_ingress(context)

except Exception as err:
  print("ERROR: {}".format(err))

Confetti

Objects have a to_dict method now. – Hampton

Answer 3

Not python, but I've used jq in the past to quickly clone something with the small customisations required for each use case (usually cloning secrets into a new namespace).

kc get pod whatever-85pmk -o json \
 | jq 'del(.status, .metadata ) | .metadata.name="newname"' \
 | kc apply -f - -o yaml --dry-run 

Subtreasury

that's awesome, but I had seen the jq integration before... It works, but not flexible enough when we are manipulating the clone... The question is about pyhon – Confetti

Answer 4

This is really easy to do with Hikaru.

Here is an example from my own open source project:

def duplicate_without_fields(obj: HikaruBase, omitted_fields: List[str]):
    """
    Duplicate a hikaru object, omitting the specified fields
    This is useful when you want to compare two versions of an object and first "cleanup" fields that shouldn't be
    compared.
    :param HikaruBase obj: A kubernetes object
    :param List[str] omitted_fields: List of fields to be omitted. Field name format should be '.' separated
                                     For example: ["status", "metadata.generation"]
    """
    if obj is None:
        return None

    duplication = obj.dup()

    for field_name in omitted_fields:
        field_parts = field_name.split(".")
        try:
            if len(field_parts) > 1:
                parent_obj = duplication.object_at_path(field_parts[:-1])
            else:
                parent_obj = duplication

            setattr(parent_obj, field_parts[-1], None)
        except Exception:
            pass  # in case the field doesn't exist on this object

    return duplication

Dumping the object to yaml afterwards or re-applying it to the cluster is trivial with Hikaru

We're using this to clean up objects so that can show users a github-style diff when objects change, without spammy fields that change often like generation

Whitlow