| # 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. | |
| # | |
| 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="openserverless"): | |
| """ " | |
| Creates a whisk user using a POST operation | |
| param: whisk_user_dict a dictionary representing the whisksusers resource to create | |
| param: namespace default to openserverless | |
| return: True if the operation is successfully, False otherwise | |
| """ | |
| url = f"{self.host}/apis/openserverless.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="openserverless"): | |
| """ " | |
| Delete a whisk user using a DELETE operation | |
| param: username of the whisksusers resource to delete | |
| param: namespace default to openserverless | |
| return: True if the operation is successfully, False otherwise | |
| """ | |
| url = f"{self.host}/apis/openserverless.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="openserverless"): | |
| """ " | |
| Get a whisk user using a GET operation | |
| param: username of the whisksusers resource to delete | |
| param: namespace default to openserverless | |
| return: a dictionary representing the existing user, None otherwise | |
| """ | |
| url = f"{self.host}/apis/openserverless.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="openserverless"): | |
| """ " | |
| Updates a whisk user using a PUT operation | |
| param: whisk_user_dict a dictionary representing the whisksusers resource to update | |
| param: namespace default to openserverless | |
| return: True if the operation is successfully, False otherwise | |
| """ | |
| url = f"{self.host}/apis/openserverless.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="openserverless"): | |
| """ | |
| 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="openserverless"): | |
| """ | |
| 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="openserverless"): | |
| """ | |
| 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="openserverless"): | |
| """ | |
| 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="openserverless"): | |
| """ | |
| 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="openserverless"): | |
| """ | |
| 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 = None, namespace="openserverless"): | |
| """ | |
| Get all Kubernetes jobs in a specific namespace. | |
| :param name_filter: Optional filter to match job names. | |
| :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="openserverless"): | |
| """ | |
| 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: dict, namespace="openserverless"): | |
| """ | |
| 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="openserverless"): | |
| """ | |
| 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="openserverless"): | |
| """ | |
| 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="openserverless"): | |
| """ | |
| 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 | |
| def get_pod(self, pod_name: str, namespace="openserverless"): | |
| """ | |
| Get pod details by name. | |
| :param pod_name: Name of the pod. | |
| :param namespace: Namespace where the pod is located. | |
| :return: Pod object or None if not found. | |
| """ | |
| url = f"{self.host}/api/v1/namespaces/{namespace}/pods/{pod_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_pod {ex}") | |
| return None | |
| def wait_for_init_container_completion(self, job_name: str, init_container_name: str, namespace="openserverless", timeout_seconds=300): | |
| """ | |
| Wait for a specific init container in a job's pod to complete (successfully or with error). | |
| :param job_name: Name of the job. | |
| :param init_container_name: Name of the init container to wait for. | |
| :param namespace: Namespace where the job is located. | |
| :param timeout_seconds: Maximum time to wait in seconds (default: 300 = 5 minutes). | |
| :return: True if init container completed (success or error), False if timeout or other failure. | |
| """ | |
| import time | |
| start_time = time.time() | |
| pod_name = None | |
| logging.info(f"Waiting for init container '{init_container_name}' in job '{job_name}' to complete") | |
| # First, wait for the pod to be created | |
| while time.time() - start_time < timeout_seconds: | |
| pod_name = self.get_pod_by_job_name(job_name, namespace) | |
| if pod_name: | |
| logging.info(f"Found pod '{pod_name}' for job '{job_name}'") | |
| break | |
| logging.debug(f"Pod for job '{job_name}' not yet created, waiting...") | |
| time.sleep(2) | |
| if not pod_name: | |
| logging.error(f"Timeout waiting for pod to be created for job '{job_name}'") | |
| return False | |
| # Now wait for the init container to complete | |
| while time.time() - start_time < timeout_seconds: | |
| pod = self.get_pod(pod_name, namespace) | |
| if not pod: | |
| logging.error(f"Failed to get pod '{pod_name}'") | |
| return False | |
| # Check init container status | |
| init_container_statuses = pod.get("status", {}).get("initContainerStatuses", []) | |
| for status in init_container_statuses: | |
| if status.get("name") == init_container_name: | |
| state = status.get("state", {}) | |
| # Check if terminated (completed or failed) | |
| if "terminated" in state: | |
| terminated = state["terminated"] | |
| exit_code = terminated.get("exitCode", -1) | |
| reason = terminated.get("reason", "Unknown") | |
| if exit_code == 0: | |
| logging.info(f"Init container '{init_container_name}' completed successfully") | |
| else: | |
| logging.warning(f"Init container '{init_container_name}' terminated with exit code {exit_code}, reason: {reason}") | |
| return True | |
| # Check if still running | |
| if "running" in state: | |
| logging.debug(f"Init container '{init_container_name}' is still running") | |
| # Check if waiting | |
| if "waiting" in state: | |
| waiting = state["waiting"] | |
| reason = waiting.get("reason", "Unknown") | |
| logging.debug(f"Init container '{init_container_name}' is waiting, reason: {reason}") | |
| time.sleep(2) | |
| logging.error(f"Timeout waiting for init container '{init_container_name}' to complete") | |
| return False |