2

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.

Improve this question

4 Answers 4

Reset to default
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)
Improve this answer
Sign up to request clarification or add additional context in comments.

1 Comment

Thank you Aleksandr ... I have now made your answer the preferred version!
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 ")

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://stackoverflow.com/questions/12118695/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://stackoverflow.com/questions/12118695/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://stackoverflow.com/questions/12470665/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))
Improve this answer

1 Comment

Objects have a to_dict method now.
0

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
Improve this answer

1 Comment

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
0

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

Improve this answer

Comments

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.