kube_api_client.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
from datetime import time
import requests as req
import json
import os
import logging

from base64 import b64decode, b64encode

from openserverless.common.utils import join_host_port
from openserverless.config.app_config import AppConfig
from openserverless.error.config_exception import ConfigException

SERVICE_HOST_ENV_NAME = "KUBERNETES_SERVICE_HOST"
SERVICE_PORT_ENV_NAME = "KUBERNETES_SERVICE_PORT"
SERVICE_TOKEN_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/token"
SERVICE_CERT_FILENAME = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
class KubeApiClient:

    @staticmethod
    def build_dockerconfigjson(username: str, password: str, registry: str = "https://index.docker.io/v1/") -> str:
        """
        Crea una stringa .dockerconfigjson per un secret docker-registry.
        :param username: Username del registry
        :param password: Password del registry
        :param registry: URL del registry (default Docker Hub)
        :return: Stringa JSON da usare come valore per .dockerconfigjson
        """
        import base64
        import json as _json
        auth = base64.b64encode(f"{username}:{password}".encode()).decode()
        dockerconfig = {
            "auths": {
                registry: {
                    "auth": auth,
                    "username": username,
                    "password": password
                }
            }
        }
        return _json.dumps(dockerconfig)

    def __init__(self, environ=os.environ):
        self._environ = environ
        self.SERVICE_TOKEN_FILENAME = self._environ.get("KUBERNETES_TOKEN_FILENAME") or SERVICE_TOKEN_FILENAME
        self.SERVICE_CERT_FILENAME = self._environ.get("KUBERNETES_CERT_FILENAME") or SERVICE_CERT_FILENAME
        self._load_incluster_config()

    def _parse_b64(self, encoded_str):
        try:
            return b64decode(encoded_str).decode()
        except:
            raise ConfigException("Could not decode base64 encoded value")

    def _load_incluster_config(self):
        """
        Use the service account kubernetes gives to pods to connect to kubernetes
        cluster. It's intended for clients that expect to be running inside a pod
        running on kubernetes. It will raise an exception if called from a process
        not running in a kubernetes environment.
        """
        if (
            SERVICE_HOST_ENV_NAME not in self._environ
            or SERVICE_PORT_ENV_NAME not in self._environ
        ):
            raise ConfigException("Service host/port is not set.")

        self.host = "https://" + join_host_port(
            self._environ.get(SERVICE_HOST_ENV_NAME),
            self._environ.get(SERVICE_PORT_ENV_NAME),
        )

        self._read_token_file()

        with open(self.SERVICE_CERT_FILENAME) as f:
            if not f.read():
                raise ConfigException("Cert file exists but empty.")

        self.ssl_ca_cert = self.SERVICE_CERT_FILENAME

    def _read_token_file(self):
        with open(self.SERVICE_TOKEN_FILENAME) as f:
            content = f.read()
            if not content:
                raise ConfigException("Token file exists but empty.")
            self.token = "Bearer " + content

    def create_whisk_user(self, whisk_user_dict, namespace="nuvolaris"):
        """ "
        Creates a whisk user using a POST operation
        param: whisk_user_dict a dictionary representing the whisksusers resource to create
        param: namespace default to nuvolaris
        return: True if the operation is successfully, False otherwise
        """
        url = f"{self.host}/apis/nuvolaris.org/v1/namespaces/{namespace}/whisksusers"
        headers = {"Authorization": self.token}

        try:
            logging.info("POST request to %s", url)
            response = None
            response = req.post(
                url,
                headers=headers,
                data=json.dumps(whisk_user_dict),
                verify=self.ssl_ca_cert,
            )

            if response.status_code in [200, 201, 202]:
                logging.debug(
                    "POST to %s succeeded with %s. Body %s",
                    url,
                    response.status_code,
                    response.text,
                )
                return True

            logging.error(
                "POST to %s failed with %s. Body %s",
                url,
                response.status_code,
                response.text,
            )
            return False
        except Exception as ex:
            logging.error("create_whisk_user %s", ex)
            return False

    def delete_whisk_user(self, username: str, namespace="nuvolaris"):
        """ "
        Delete a whisk user using a DELETE operation
        param: username of the whisksusers resource to delete
        param: namespace default to nuvolaris
        return: True if the operation is successfully, False otherwise
        """
        url = f"{self.host}/apis/nuvolaris.org/v1/namespaces/{namespace}/whisksusers/{username}"
        headers = {"Authorization": self.token}

        try:
            logging.info(f"DELETE request to {url}")
            response = None
            response = req.delete(url, headers=headers, verify=self.ssl_ca_cert)

            if response.status_code in [200, 202]:
                logging.debug(
                    f"DELETE to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return True

            logging.error(
                f"DELETE to {url} failed with {response.status_code}. Body {response.text}"
            )
            return False
        except Exception as ex:
            logging.error(f"delete_whisk_user {ex}")
            return False

    def get_whisk_user(self, username: str, namespace="nuvolaris"):
        """ "
        Get a whisk user using a GET operation
        param: username of the whisksusers resource to delete
        param: namespace default to nuvolaris
        return: a dictionary representing the existing user, None otherwise
        """
        url = f"{self.host}/apis/nuvolaris.org/v1/namespaces/{namespace}/whisksusers/{username}"
        headers = {"Authorization": self.token}

        try:
            logging.info(f"GET request to {url}")
            response = None
            response = req.get(url, headers=headers, verify=self.ssl_ca_cert)

            if response.status_code in [200, 202]:
                logging.debug(
                    f"GET to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return json.loads(response.text)

            logging.error(
                f"GET to {url} failed with {response.status_code}. Body {response.text}"
            )
            return None
        except Exception as ex:
            logging.error(f"get_whisk_user {ex}")
            return None

    def update_whisk_user(self, whisk_user_dict, namespace="nuvolaris"):
        """ "
        Updates a whisk user using a PUT operation
        param: whisk_user_dict a dictionary representing the whisksusers resource to update
        param: namespace default to nuvolaris
        return: True if the operation is successfully, False otherwise
        """
        url = f"{self.host}/apis/nuvolaris.org/v1/namespaces/{namespace}/whisksusers/{whisk_user_dict['metadata']['name']}"
        headers = {"Authorization": self.token}

        try:
            logging.error(f"PUT request to {url}")
            response = None
            response = req.put(
                url,
                headers=headers,
                data=json.dumps(whisk_user_dict),
                verify=self.ssl_ca_cert,
            )

            if response.status_code in [200, 201, 202]:
                logging.debug(
                    f"PUT to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return True

            logging.error(
                f"PUT to {url} failed with {response.status_code}. Body {response.text}"
            )
            return False
        except Exception as ex:
            logging.error(f"update_whisk_user {ex}")
            return False
        
    def get_config_map(self, cm_name: str, namespace="nuvolaris"):
        """
        Get a ConfigMap by name.
        :param cm_name: Name of the ConfigMap.
        :param namespace: Namespace where the ConfigMap is located.
        :return: The ConfigMap data or None if not found.
        """
        url = f"{self.host}/api/v1/namespaces/{namespace}/configmaps/{cm_name}"
        headers = {"Authorization": self.token}

        try:
            logging.info(f"GET request to {url}")
            response = req.get(url, headers=headers, verify=self.ssl_ca_cert)

            if response.status_code == 200:
                logging.debug(
                    f"GET to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return json.loads(response.text)

            logging.error(
                f"GET to {url} failed with {response.status_code}. Body {response.text}"
            )
            return None
        except Exception as ex:
            logging.error(f"get_config_map {ex}")
            return None
    
    def post_config_map(self, cm_name: str, file_or_dir: str, namespace="nuvolaris"):
        """        
        Create a ConfigMap from a file or directory.
        :param cm_name: Name of the ConfigMap.
        :param file_or_dir: Path to the file or directory containing the data.
        :param namespace: Namespace where the ConfigMap will be created.
        :return: The created ConfigMap or None if failed.
        """
        if not os.path.exists(file_or_dir):
            raise ConfigException(f"File or directory {file_or_dir} does not exist.")
        
        configmap_data = {}
        if os.path.isfile(file_or_dir):            
            with open(file_or_dir, "r") as f:
                configmap_data[os.path.basename(file_or_dir)] = f.read()
        elif os.path.isdir(file_or_dir):
            for filename in os.listdir(file_or_dir):
                filepath = os.path.join(file_or_dir, filename)
                if os.path.isfile(filepath):
                    with open(filepath, "r") as f:
                        configmap_data[filename] = f.read()

        configmap_manifest = {
            "apiVersion": "v1",
            "kind": "ConfigMap",
            "metadata": {
                "name": cm_name
            },
            "data": configmap_data
        }
        
        url = f"{self.host}/api/v1/namespaces/{namespace}/configmaps"
        headers = {"Authorization": self.token, "Content-Type": "application/json"}

        try:
            logging.info(f"POST request to {url}")
            response = None
            response = req.post(url, data=json.dumps(configmap_manifest), headers=headers, verify=self.ssl_ca_cert)

            if response.status_code in [200, 201, 202]:
                logging.debug(
                    f"POST to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return json.loads(response.text)

            logging.error(
                f"POST to {url} failed with {response.status_code}. Body {response.text}"
            )
            return None
        except Exception as ex:
            logging.error(f"post_config_map {ex}")
            return None
    
    def delete_config_map(self, cm_name: str, namespace="nuvolaris"):
        """
        Delete a ConfigMap by name.
        :param cm_name: Name of the ConfigMap to delete.
        :param namespace: Namespace where the ConfigMap is located.
        :return: True if deletion was successful, False otherwise.
        """
        url = f"{self.host}/api/v1/namespaces/{namespace}/configmaps/{cm_name}"
        headers = {"Authorization": self.token}

        try:
            logging.info(f"DELETE request to {url}")
            response = None
            response = req.delete(url, headers=headers, verify=self.ssl_ca_cert)

            if response.status_code in [200, 202]:
                logging.debug(
                    f"DELETE to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return True

            logging.error(
                f"DELETE to {url} failed with {response.status_code}. Body {response.text}"
            )
            return False
        except Exception as ex:
            logging.error(f"delete_config_map {ex}")
            return False
     
    def get_secret(self, secret_name: str, namespace="nuvolaris"):
        """
        Get a Kubernetes secret by name.
        :param secret_name: Name of the secret.
        :param namespace: Namespace where the secret is located.
        :return: The secret data or None if not found.
        """
        url = f"{self.host}/api/v1/namespaces/{namespace}/secrets/{secret_name}"
        headers = {"Authorization": self.token}

        try:
            logging.info(f"GET request to {url}")
            response = req.get(url, headers=headers, verify=self.ssl_ca_cert)

            if response.status_code == 200:
                logging.debug(
                    f"GET to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return json.loads(response.text)

            logging.error(
                f"GET to {url} failed with {response.status_code}. Body {response.text}"
            )
            return None
        except Exception as ex:
            logging.error(f"get_secret {ex}")
            return None
    
    def post_secret(self, secret_name: str, secret_data: dict, type: str="Opaque", namespace="nuvolaris"):
        """
        Create a Kubernetes secret.
        :param secret_name: Name of the secret.
        :param secret_data: Dictionary containing the secret data.
        :param type: Type of the secret (Opaque, kubernetes.io/dockerconfigjson, kubernetes.io/tls)
        :param namespace: Namespace where the secret will be created.
        :return: The created secret or None if failed.
        """
        url = f"{self.host}/api/v1/namespaces/{namespace}/secrets"
        headers = {"Authorization": self.token, "Content-Type": "application/json"}

        if type == "kubernetes.io/dockerconfigjson":
            # secret_data should contain a key '.dockerconfigjson' with the JSON string value
            manifest_data = {
                ".dockerconfigjson": b64encode(secret_data[".dockerconfigjson"].encode()).decode()
            }
        elif type == "kubernetes.io/tls":
            # secret_data should contain 'tls.crt' and 'tls.key'
            manifest_data = {
                "tls.crt": b64encode(secret_data["tls.crt"].encode()).decode(),
                "tls.key": b64encode(secret_data["tls.key"].encode()).decode()
            }
        else:
            # Default: Opaque or other types
            manifest_data = {k: b64encode(v.encode()).decode() for k, v in secret_data.items()}

        secret_manifest = {
            "apiVersion": "v1",
            "kind": "Secret",
            "metadata": {"name": secret_name},
            "data": manifest_data,
            "type": type
        }

        try:
            logging.info(f"POST request to {url}")
            response = req.post(url, headers=headers, json=secret_manifest, verify=self.ssl_ca_cert)

            if response.status_code in [200, 201]:
                logging.debug(
                    f"POST to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return json.loads(response.text)

            logging.error(
                f"POST to {url} failed with {response.status_code}. Body {response.text}"
            )
            return None
        except Exception as ex:
            logging.error(f"post_secret {ex}")
            return None
    
    def delete_secret(self, secret_name: str, namespace="nuvolaris"):
        """
        Delete a Kubernetes secret.
        :param secret_name: Name of the secret to delete.
        :param namespace: Namespace where the secret is located.
        :return: True if deletion was successful, False otherwise.
        """
        url = f"{self.host}/api/v1/namespaces/{namespace}/secrets/{secret_name}"
        headers = {"Authorization": self.token}

        try:
            logging.info(f"DELETE request to {url}")
            response = req.delete(url, headers=headers, verify=self.ssl_ca_cert)

            if response.status_code in [200, 202]:
                logging.debug(
                    f"DELETE to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return True

            logging.error(
                f"DELETE to {url} failed with {response.status_code}. Body {response.text}"
            )
            return False
        except Exception as ex:
            logging.error(f"delete_secret {ex}")
            return False
        
    def get_jobs(self, name_filter: str = None, namespace="nuvolaris"):
        """
        Get all Kubernetes jobs in a specific namespace.
        :param namespace: Namespace to list jobs from.
        :return: List of jobs or None if failed.
        """
        url = f"{self.host}/apis/batch/v1/namespaces/{namespace}/jobs"
        headers = {"Authorization": self.token}
        try:
            logging.info(f"GET request to {url}")
            response = req.get(url, headers=headers, verify=self.ssl_ca_cert)

            if response.status_code in [200, 202]:
                logging.debug(
                    f"GET to {url} succeeded with {response.status_code}. Body {response.text}"
                )

                if name_filter:
                    jobs = json.loads(response.text)["items"]
                    filtered_jobs = [job for job in jobs if name_filter in job["metadata"]["name"]]
                    return filtered_jobs

                return json.loads(response.text)["items"]

            logging.error(
                f"GET to {url} failed with {response.status_code}. Body {response.text}"
            )
            return None
        except Exception as ex:
            logging.error(f"get_jobs {ex}")
            return None
        
    def delete_job(self, job_name: str, namespace="nuvolaris"):
        """
        Delete a Kubernetes job by name.
        :param job_name: Name of the job to delete.
        :param namespace: Namespace where the job is located.
        :return: True if deletion was successful, False otherwise.
        """
        url = f"{self.host}/apis/batch/v1/namespaces/{namespace}/jobs/{job_name}"
        headers = {"Authorization": self.token}

        try:
            logging.info(f"DELETE request to {url}")
            response = req.delete(url, headers=headers, verify=self.ssl_ca_cert)

            if response.status_code in [200, 202]:
                logging.debug(
                    f"DELETE to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return True

            logging.error(
                f"DELETE to {url} failed with {response.status_code}. Body {response.text}"
            )
            return False
        except Exception as ex:
            logging.error(f"delete_job {ex}")
            return False

    def post_job(self, job_manifest: json, namespace="nuvolaris"):
        """
        Create a Kubernetes job.
        :param job_manifest: Dictionary containing the job manifest. 
        :param namespace: Namespace where the job will be created.
        :return: The created job or None if failed.
        """
        url = f"{self.host}/apis/batch/v1/namespaces/{namespace}/jobs"
        headers = {"Authorization": self.token}
        try:
            logging.info(f"POST request to {url}")
            response = None
            response = req.post(url, headers=headers, json=job_manifest, verify=self.ssl_ca_cert)
            if response.status_code in [200, 201, 202]:
                logging.debug(
                    f"POST to {url} succeeded with {response.status_code}. Body {response.text}"
                )
                return json.loads(response.text)
            logging.error(
                f"POST to {url} failed with {response.status_code}. Body {response.text}"
            )
            return None
        except Exception as ex:
            logging.error(f"post_job {ex}")
            return None

    def get_pod_by_job_name(self, job_name: str, namespace="nuvolaris"):
        """
        Get the pod name associated with a job by its name.
        :param job_name: Name of the job.
        :param namespace: Namespace where the job is located.
        :return: The pod name if found, None otherwise.
        """
        url = f"{self.host}/api/v1/namespaces/{namespace}/pods"
        headers = {"Authorization": self.token}
        try:
            while True:
                resp = req.get(url, headers=headers, verify=self.ssl_ca_cert)
                if not resp.status_code in [200, 202]:
                    logging.error(
                        f"POST to {url} failed with {resp.status_code}. Body {resp.text}"
                    )
                    return None
                logging.debug(
                    f"POST to {url} succeeded with {resp.status_code}. Body {resp.text}"
                )
                pods = resp.json()["items"]
                for pod in pods:
                    labels = pod["metadata"].get("labels", {})
                    if labels.get("job-name") == job_name:
                        return pod["metadata"]["name"]
                time.sleep(1)
                
        except Exception as ex:
            logging.error(f"get_pod_by_job_name {ex}")
            return None

    def stream_pod_logs(self, pod_name: str, namespace="nuvolaris"):
        """
        Stream logs from a specific pod.
        :param pod_name: Name of the pod to stream logs from.
        :param namespace: Namespace where the pod is located.
        """
        url = f"{self.host}/api/v1/namespaces/{namespace}/pods/{pod_name}/log?follow=true"
        headers = {"Authorization": self.token}
        with req.get(url, headers=headers, verify=self.ssl_ca_cert, stream=True) as r:
            for line in r.iter_lines():
                if line:
                    print(line.decode())

    def check_job_status(self, job_name: str, namespace="nuvolaris"):
        """
        Check the status of a job by its name.
        :param job_name: Name of the job to check.
        :param namespace: Namespace where the job is located.
        :return: True if the job has succeeded, False otherwise.
        """
        url = f"{self.host}/apis/batch/v1/namespaces/{namespace}/jobs/{job_name}"
        headers = {"Authorization": self.token}
        try:
            resp = req.get(url, headers=headers, verify=self.ssl_ca_cert)
            resp.raise_for_status()
            status = resp.json()["status"]
            if status.get("succeeded", 0) > 0:
                return True
            else:
                return False
        except Exception as ex:
            logging.error(f"check_job_status {ex}")
            return False