Clone Kubernetes Objects Programmatically Using The Python Api

The Python API is available to read objects from a cluster. By cloning we can say: Get a copy of an existing Kubernetes object using kubectl get Change the properties of the objec

Solution 1:

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 ")

classIngressGatewayCreator:

    @staticmethoddefclone_default_ingress(clone_context):
        # Clone the deployment
        IngressGatewayCreator.clone_deployment_object(clone_context)

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

    @staticmethoddefclone_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)
                returnraise# 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")

    @staticmethoddefclone_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)
                returnraise# 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")

    @staticmethoddefclone_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://stackoverflow.com/questions/12118695/efficient-way-to-remove-keys-with-empty-strings-from-a-dict/59959570#59959570    @staticmethoddefscrub_dict(d):
        new_dict = {}
        for k, v in d.items():
            ifisinstance(v, dict):
                v = IngressGatewayCreator.scrub_dict(v)
            ifisinstance(v, list):
                v = IngressGatewayCreator.scrub_list(v)
            ifnot v in (u'', None, {}):
                new_dict[k] = v
        return new_dict

    # https://stackoverflow.com/questions/12118695/efficient-way-to-remove-keys-with-empty-strings-from-a-dict/59959570#59959570    @staticmethoddefscrub_list(d):
        scrubbed_list = []
        for i in d:
            ifisinstance(i, dict):
                i = IngressGatewayCreator.scrub_dict(i)
            scrubbed_list.append(i)
        return scrubbed_list


classIngressGatewayContext:

    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"defsave_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://stackoverflow.com/questions/12470665/how-can-i-write-data-in-yaml-format-in-a-file/18210750#18210750withopen(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))

Solution 2:

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 

Solution 3:

This is really easy to do with Hikaru.

Here is an example from my own open source project:

defduplicate_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"fieldsthatshouldn'tbecompared.:paramHikaruBase obj:Akubernetesobject:paramList[str]omitted_fields:Listoffieldstobeomitted.Fieldnameformatshouldbe'.'separatedFor 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:iflen(field_parts)>1:parent_obj=duplication.object_at_path(field_parts[:-1])else:parent_obj=duplicationsetattr(parent_obj,field_parts[-1],None)except Exception:pass# in case the field doesn't exist on this objectreturnduplication

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