playground/tekton_api.py

import functools
import hashlib
import json
import textwrap
import typing
import uuid
from pathlib import Path

import kubernetes.config
import yaml


def make_client() -> kubernetes.client.CoreV1Api:
    kubernetes.config.load_kube_config('/Users/abdus/Desktop/kubeconfig-kube.yml')
    return kubernetes.client.CoreV1Api()


class Kubernetes:
    def __init__(self, client: kubernetes.client.CoreV1Api, namespace='default'):
        self.client = client
        self.custom_objects = kubernetes.client.CustomObjectsApi(self.client.api_client)
        self.namespace = namespace

    @classmethod
    def from_kubeconfig(cls, kubeconfig_path: str, namespace: str) -> 'Kubernetes':
        kubernetes.config.load_kube_config(kubeconfig_path)
        return cls(kubernetes.client.CoreV1Api(), namespace=namespace)

    def read_logs(self, pod_name: str, container_name: typing.Optional[str] = None, timestamps: bool = False) -> str:
        return self.client.read_namespaced_pod_log(
            pod_name,
            self.namespace,
            container=container_name,
            timestamps=timestamps,
            pretty=True,
        )

    def get_tekton_object(self, plural: str, name: str = None):
        getter = functools.partial(
            self.custom_objects.get_namespaced_custom_object,
            group='tekton.dev',
            version='v1beta1',
            namespace=self.namespace,
        )
        return getter(plural=plural, name=name)

    def list_tekton_object(self, plural: str):
        getter = functools.partial(
            self.custom_objects.list_namespaced_custom_object,
            group='tekton.dev',
            version='v1beta1',
            namespace=self.namespace,
        )
        return getter(plural=plural)['items']

    def create_from_yaml(self, yaml_text: str) -> list[dict]:
        specs = list(yaml.safe_load_all(yaml_text))
        return kubernetes.utils.create_from_yaml(
            kubernetes.client.CustomObjectsApi(self.client.api_client),
            yaml_objects=specs,
            namespace=self.namespace,
        )


def read_logs():
    task_hello_world = k8s.get_tekton_object(plural='tasks', name='echo-hello-world')
    taskruns = k8s.list_tekton_object(plural='taskruns')

    pod_name = taskruns['items'][0]['status']['podName']
    container_name = taskruns['items'][0]['status']['containerName']
    logs = k8s.read_logs(pod_name, container_name)
    print(logs)


k8s = Kubernetes.from_kubeconfig('/Users/abdus/Desktop/kubeconfig.yaml', namespace='api-server')


def get_pipelineru_status():
    result = k8s.get_tekton_object('pipelineruns', name='pr-2rr6t')
    status = result['status']['conditions'][0]['status']
    reason = result['status']['conditions'][0]['reason']
    failed_taskruns = [
        name for name, it in result['status']['taskRuns'].items() if it['status']['conditions'][0]['status'] == 'False'
    ]

    logs = []
    for it in result['status']['taskRuns'].values():
        task_name = it['pipelineTaskName']
        task_status = it['status']['conditions'][0]['reason']
        pod_name = it['status']['podName']

        if task_status != 'Failed':
            continue

        # fetch all logs for all steps to get a better picture of the error
        # (regardless of whether it is successful or not)
        for i, step in enumerate(it['status']['steps'], start=1):
            container_name = step['container']
            header = f'# {task_name}, step {i}: {step["name"]}'
            logs.append('#' * len(header))
            logs.append(header)
            logs.append('#' * len(header))
            step_logs = k8s.read_logs(pod_name, container_name, timestamps=True)
            logs.append(step_logs)

    log_text = '\n'.join(logs)
    return {
        'pipelinerun_name': result['metadata']['name'],
        'namespace': result['metadata']['namespace'],
        'status': status,
        'reason': reason,
        'failed_taskruns': failed_taskruns,
        'logs': log_text,
    }


def create_taskrun():
    k = kubernetes.client.CustomObjectsApi()
    res = k.create_namespaced_custom_object(
        group='tekton.dev',
        version='v1beta1',
        namespace='default',
        plural='taskruns',
        body={
            'apiVersion': 'tekton.dev/v1beta1',
            'kind': 'TaskRun',
            'metadata': {'name': 'testing1234' + uuid.uuid4().hex},
            'spec': {
                'taskRef': {'name': 'task-with-json'},
                'params': [
                    {
                        'name': 'json_value',
                        'value': '{"foo": "bar", "nested": {"a": "1"}}',
                    }
                ],
            },
        },
    )
    print(res)


def migrate_pipelines():
    k = kubernetes.client.CustomObjectsApi()

    pipeline_yaml = Path('sample_pipeline.yaml').read_text()
    parsed_pipeline = yaml.safe_load(pipeline_yaml)

    params = dict(group='tekton.dev', version='v1beta1', namespace='default', plural='pipelines')
    k.delete_namespaced_custom_object(
        **params,
        name=parsed_pipeline['metadata']['name'],
    )
    res = k.create_namespaced_custom_object(
        **params,
        body=parsed_pipeline,
    )
    return parsed_pipeline


def trigger_pipeline():
    k = kubernetes.client.CustomObjectsApi()
    pipeline = migrate_pipelines()
    pipeline_name = pipeline['metadata']['name']

    payload = {
        'image': 'nginx',
        'json_value': {'nested': {'nested2': {'a': 1}}},
    }

    res = k.create_namespaced_custom_object(
        group='tekton.dev',
        version='v1beta1',
        namespace='default',
        plural='pipelineruns',
        body={
            'apiVersion': 'tekton.dev/v1beta1',
            'kind': 'PipelineRun',
            'metadata': {'name': 'deploy-app-' + uuid.uuid4().hex},
            'spec': {
                'pipelineRef': {'name': pipeline_name},
                'params': [
                    {'name': k, 'value': json.dumps(v) if not isinstance(v, str) else v} for k, v in payload.items()
                ],
            },
        },
    )


if __name__ == '__main__':
    main()