Kubernetes

The K8s class reads and changes Kubernetes objects such as pods, stateful sets, config maps and ingresses. The Customizer uses it when it runs inside a Kubernetes cluster.

k8s

Kubernetes Module to implement functions to read / write Kubernetes objects.

This includes Pods, Stateful Sets, Config Maps, ...

https://github.com/kubernetes-client/python https://github.com/kubernetes-client/python/blob/master/kubernetes/README.md https://github.com/kubernetes-client/python/tree/master/examples

Classes:

Name	Description
K8s	Provides an interface to interact with the Kubernetes API.

K8s

K8s(
    kubeconfig_file: str | None = None,
    namespace: str = "default",
    logger: Logger = default_logger,
)

Provides an interface to interact with the Kubernetes API.

This class can run both in-cluster and locally using kubeconfig. It offers methods to interact with Kubernetes namespaces, pods, and various API objects like CoreV1, AppsV1, and NetworkingV1.

Attributes:

Name	Type	Description
logger	Logger	Logger for the class.
_core_v1_api	CoreV1Api	API client for Kubernetes Core V1.
_apps_v1_api	AppsV1Api	API client for Kubernetes Apps V1.
_networking_v1_api	NetworkingV1Api	API client for Kubernetes Networking V1.
_namespace	str	The namespace in which operations are performed.

Initialize the Kubernetes object.

Parameters:

Name	Type	Description	Default
kubeconfig_file	str | None	Path to a kubeconfig file. Defaults to None.	None
namespace	str	The Kubernetes name space. Defaults to "default".	'default'
logger	Logger	The logger object. Defaults to default_logger.	default_logger

Methods:

Name	Description
get_core_v1_api	Return Kubernetes Core V1 API object.
get_apps_v1_api	Return Kubernetes Apps V1 API object.
get_networking_v1_api	Return Kubernetes Networking V1 API object.
get_namespace	Return Kubernetes Namespace.
get_pod	Get a pod in the configured namespace (the namespace is defined in the class constructor).
list_pods	List all Kubernetes pods in a given namespace.
wait_pod_condition	Wait for the pod to reach a defined condition (e.g. "Ready").
exec_pod_command	Execute a command inside a Kubernetes Pod (similar to kubectl exec on command line).
exec_pod_command_interactive	Execute a command inside a Kubernetes pod (similar to kubectl exec on command line).
delete_pod	Delete a pod in the configured namespace (the namespace is defined in the class constructor).
get_config_map	Get a config map in the configured namespace (the namespace is defined in the class constructor).
list_config_maps	List all Kubernetes Config Maps in the current namespace.
find_config_map	Find a Kubernetes Config Map based on its name.
replace_config_map	Replace a Config Map with a new specification.
get_stateful_set	Get a Kubernetes Stateful Set based on its name.
get_stateful_set_scale	Get the number of replicas for a Kubernetes Stateful Set.
patch_stateful_set	Patch a Stateful set with new values.
scale_stateful_set	Scale a stateful set to a specific number of replicas.
get_service	Get a Kubernetes Service with a defined name in the current namespace.
list_services	List all Kubernetes Service in the current namespace.
patch_service	Patch a Kubernetes Service with an updated spec.
get_ingress	Get a Kubernetes Ingress with a defined name in the current namespace.
patch_ingress	Patch a Kubernetes Ingress with an updated spec.
update_ingress_backend_services	Update a backend service and port of a Kubernetes Ingress.
verify_pod_status	Verify if a pod is in a 'Ready' state by checking the status of its containers.
verify_pod_deleted	Verify if a pod is deleted within the specified timeout.
restart_deployment	Restart a Kubernetes deployment using rolling restart.
restart_stateful_set	Enterprise-grade zero-downtime rolling restart of a stateful set.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def __init__(
    self,
    kubeconfig_file: str | None = None,
    namespace: str = "default",
    logger: logging.Logger = default_logger,
) -> None:
    """Initialize the Kubernetes object.

    Args:
        kubeconfig_file (str | None, optional):
            Path to a kubeconfig file. Defaults to None.
        namespace (str, optional):
            The Kubernetes name space. Defaults to "default".
        logger (logging.Logger, optional):
            The logger object. Defaults to default_logger.

    """

    if logger != default_logger:
        self.logger = logger.getChild("k8s")
        for logfilter in logger.filters:
            self.logger.addFilter(logfilter)

    # Configure Kubernetes API authentication to use pod serviceAccount

    try:
        config.load_incluster_config()
        configured = True
    except ConfigException:
        configured = False
        self.logger.info("Failed to load in-cluster config")

    if kubeconfig_file is None:
        kubeconfig_file = os.getenv(
            "KUBECONFIG",
            os.path.expanduser("~/.kube/config"),
        )

    if not configured:
        try:
            config.load_kube_config(config_file=kubeconfig_file)
        except ConfigException:
            self.logger.info(
                "Failed to load kubernetes config with file -> '%s'",
                kubeconfig_file,
            )

    self._core_v1_api = CoreV1Api()
    self._apps_v1_api = AppsV1Api()
    self._networking_v1_api = NetworkingV1Api()

    if namespace == "default":
        # Read current namespace
        try:
            with open(
                "/var/run/secrets/kubernetes.io/serviceaccount/namespace",
                encoding="utf-8",
            ) as namespace_file:
                self._namespace = namespace_file.read()
        except FileNotFoundError:
            self._namespace = namespace
    else:
        self._namespace = namespace

get_core_v1_api

get_core_v1_api() -> CoreV1Api

Return Kubernetes Core V1 API object.

Returns:

Name	Type	Description
object	CoreV1Api	Kubernetes API object

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_core_v1_api(self) -> CoreV1Api:
    """Return Kubernetes Core V1 API object.

    Returns:
        object: Kubernetes API object

    """

    return self._core_v1_api

get_apps_v1_api

get_apps_v1_api() -> AppsV1Api

Return Kubernetes Apps V1 API object.

Returns:

Name	Type	Description
object	AppsV1Api	Kubernetes API object

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_apps_v1_api(self) -> AppsV1Api:
    """Return Kubernetes Apps V1 API object.

    Returns:
        object: Kubernetes API object

    """

    return self._apps_v1_api

get_networking_v1_api

get_networking_v1_api() -> NetworkingV1Api

Return Kubernetes Networking V1 API object.

Returns:

Name	Type	Description
object	NetworkingV1Api	Kubernetes API object

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_networking_v1_api(self) -> NetworkingV1Api:
    """Return Kubernetes Networking V1 API object.

    Returns:
        object: Kubernetes API object

    """

    return self._networking_v1_api

get_namespace

get_namespace() -> str

Return Kubernetes Namespace.

Returns:

Name	Type	Description
str	str	Kubernetes namespace

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_namespace(self) -> str:
    """Return Kubernetes Namespace.

    Returns:
        str: Kubernetes namespace

    """

    return self._namespace

get_pod

get_pod(pod_name: str) -> V1Pod

Get a pod in the configured namespace (the namespace is defined in the class constructor).

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#read_namespaced_pod

Parameters:

Name	Type	Description	Default
pod_name	str	The name of the Kubernetes pod in the current namespace.	required

Returns:

Type	Description
V1Pod	V1Pod (object) or None if the call fails.
V1Pod	api_version='v1',
V1Pod	kind='Pod',
V1Pod	metadata=V1ObjectMeta(...),
V1Pod	spec=V1PodSpec(...),
V1Pod	status=V1PodStatus(...)

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_pod(self, pod_name: str) -> V1Pod:
    """Get a pod in the configured namespace (the namespace is defined in the class constructor).

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#read_namespaced_pod

    Args:
        pod_name (str):
            The name of the Kubernetes pod in the current namespace.

    Returns:
        V1Pod (object) or None if the call fails.
        - api_version='v1',
        - kind='Pod',
        - metadata=V1ObjectMeta(...),
        - spec=V1PodSpec(...),
        - status=V1PodStatus(...)

    """

    try:
        response = self.get_core_v1_api().read_namespaced_pod(
            name=pod_name,
            namespace=self.get_namespace(),
        )
    except ApiException as e:
        if e.status == 404:
            self.logger.debug("Pod -> '%s' not found (may be deleted).", pod_name)
            return None
        else:
            self.logger.error("Failed to get pod -> '%s'!", pod_name)
            return None  # Unexpected error, return None
    return response

list_pods

list_pods(
    field_selector: str = "", label_selector: str = ""
) -> V1PodList

List all Kubernetes pods in a given namespace.

The list can be further restricted by specifying a field or label selector.

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_pod

Parameters:

Name	Type	Description	Default
field_selector	str	Filter result based on fields. Defaults to "".	''
label_selector	str	Filter result based on labels. Defaults to "".	''

Returns:

Name	Type	Description
	V1PodList	V1PodList (object) or None if the call fails
	V1PodList	Properties can be accessed with the "." notation (this is an object not a dict!):
	V1PodList	api_version: The Kubernetes API version.
	V1PodList	items: A list of V1Pod objects, each representing a pod. You can access the fields of a V1Pod object using dot notation, for example, pod.metadata.name to access the name of the pod
	V1PodList	kind: The Kubernetes object kind, which is always "PodList".
	V1PodList	metadata: Additional metadata about the pod list, such as the resource version.
See	V1PodList	https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1PodList.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def list_pods(
    self,
    field_selector: str = "",
    label_selector: str = "",
) -> V1PodList:
    """List all Kubernetes pods in a given namespace.

    The list can be further restricted by specifying a field or label selector.

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_pod

    Args:
        field_selector (str, optional):
            Filter result based on fields. Defaults to "".
        label_selector (str, optional):
            Filter result based on labels. Defaults to "".

    Returns:
        V1PodList (object) or None if the call fails
        Properties can be accessed with the "." notation (this is an object not a dict!):
        - api_version: The Kubernetes API version.
        - items: A list of V1Pod objects, each representing a pod. You can access the fields of a
                V1Pod object using dot notation, for example, pod.metadata.name to access the name of the pod
        - kind: The Kubernetes object kind, which is always "PodList".
        - metadata: Additional metadata about the pod list, such as the resource version.
        See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1PodList.md

    """

    try:
        response = self.get_core_v1_api().list_namespaced_pod(
            field_selector=field_selector,
            label_selector=label_selector,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to list pods with field selector -> '%s' and label selector -> '%s'",
            field_selector,
            label_selector,
        )
        return None

    return response

wait_pod_condition

wait_pod_condition(
    pod_name: str, condition_name: str, sleep_time: int = 30
) -> None

Wait for the pod to reach a defined condition (e.g. "Ready").

Parameters:

Name	Type	Description	Default
pod_name	str	The name of the Kubernetes pod in the current namespace.	required
condition_name	str	The name of the condition, e.g. "Ready".	required
sleep_time	int	The number of seconds to wait between repetitive status checks. Defaults to 30.	30

Returns:

Type	Description
None	None

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def wait_pod_condition(
    self,
    pod_name: str,
    condition_name: str,
    sleep_time: int = 30,
) -> None:
    """Wait for the pod to reach a defined condition (e.g. "Ready").

    Args:
        pod_name (str):
            The name of the Kubernetes pod in the current namespace.
        condition_name (str):
            The name of the condition, e.g. "Ready".
        sleep_time (int, optional):
            The number of seconds to wait between repetitive status checks. Defaults to 30.

    Returns:
        None

    """

    ready = False
    while not ready:
        try:
            pod_status = self.get_core_v1_api().read_namespaced_pod_status(
                pod_name,
                self.get_namespace(),
            )

            # Check if the pod has reached the defined condition:
            for cond in pod_status.status.conditions:
                if cond.type == condition_name and cond.status == "True":
                    self.logger.info(
                        "Pod -> '%s' is in state -> '%s'!",
                        pod_name,
                        condition_name,
                    )
                    ready = True
                    break
            else:
                self.logger.info(
                    "Pod -> '%s' is not yet in state -> '%s'. Waiting...",
                    pod_name,
                    condition_name,
                )
                time.sleep(sleep_time)
                continue

        except ApiException:
            self.logger.error(
                "Failed to wait for pod -> '%s' to reach state -> '%s'.",
                pod_name,
                condition_name,
            )

exec_pod_command

exec_pod_command(
    pod_name: str,
    command: list,
    max_retry: int = 3,
    time_retry: int = 10,
    container: str | None = None,
    timeout: int = 60,
) -> str | None

Execute a command inside a Kubernetes Pod (similar to kubectl exec on command line).

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#connect_get_namespaced_pod_exec

Parameters:

Name	Type	Description	Default
pod_name	str	The name of the Kubernetes pod in the current namespace.	required
command	list	A list of command and its parameters, e.g. ["/bin/bash", "-c", "pwd"] The "-c" is required to make the shell executing the command.	required
max_retry	int	The maximum number of attempts to execute the command. Defaults to 3.	3
time_retry	int	Wait time in seconds between retries. Defaults to 10.	10
container	str | None	The container name if the pod runs multiple containers inside. Defaults to None.	None
timeout	int	Timeout duration that is waited for any response in seconds. Defaults to 60.	60

Returns:

Type	Description
str | None	str | None: Response of the command or None if the call fails.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def exec_pod_command(
    self,
    pod_name: str,
    command: list,
    max_retry: int = 3,
    time_retry: int = 10,
    container: str | None = None,
    timeout: int = 60,
) -> str | None:
    """Execute a command inside a Kubernetes Pod (similar to kubectl exec on command line).

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#connect_get_namespaced_pod_exec

    Args:
        pod_name (str):
            The name of the Kubernetes pod in the current namespace.
        command (list):
            A list of command and its parameters, e.g. ["/bin/bash", "-c", "pwd"]
            The "-c" is required to make the shell executing the command.
        max_retry (int, optional):
            The maximum number of attempts to execute the command. Defaults to 3.
        time_retry (int, optional):
            Wait time in seconds between retries. Defaults to 10.
        container (str | None, optional):
            The container name if the pod runs multiple containers inside. Defaults to None.
        timeout (int, optional):
            Timeout duration that is waited for any response in seconds. Defaults to 60.

    Returns:
        str | None:
            Response of the command or None if the call fails.

    """

    pod = self.get_pod(pod_name=pod_name)
    if not pod:
        self.logger.error("Pod -> '%s' does not exist. Cannot execute command -> %s", pod_name, command)
        return None

    self.logger.debug("Execute command -> %s in pod -> '%s'", command, pod_name)

    retry_counter = 1

    while retry_counter <= max_retry:
        try:
            response = stream(
                self.get_core_v1_api().connect_get_namespaced_pod_exec,
                pod_name,
                self.get_namespace(),
                command=command,
                container=container,
                stderr=True,
                stdin=False,
                stdout=True,
                tty=False,
                _request_timeout=timeout,
            )
        except ApiException as exc:
            self.logger.warning(
                "Failed to execute command, retry (%s/%s) -> %s in pod -> '%s'; error -> %s",
                retry_counter,
                max_retry,
                command,
                pod_name,
                str(exc),
            )
            retry_counter = retry_counter + 1
            exception = exc
            self.logger.debug(
                "Wait %s seconds before next retry...",
                str(time_retry),
            )
            time.sleep(time_retry)
            continue
        else:
            self.logger.debug("Command execution response -> %s", response or "<empty>")
            return response

    self.logger.error(
        "Failed to execute command -> %s in pod -> '%s' after %d retries; error -> %s",
        command,
        pod_name,
        max_retry,
        str(exception),
    )

    return None

exec_pod_command_interactive

exec_pod_command_interactive(
    pod_name: str,
    commands: list,
    timeout: int = 30,
    write_stderr_to_error_log: bool = True,
) -> str | None

Execute a command inside a Kubernetes pod (similar to kubectl exec on command line).

Other than exec_pod_command() method above this is an interactive execution using stdin and reading the output from stdout and stderr. This is required for longer running commands. It is currently used for restarting the spawner of Archive Center. The output of the command is pushed into the logging.

Parameters:

Name	Type	Description	Default
pod_name	str	The name of the Kubernetes pod in the current namespace	required
commands	list	A list of command and its parameters, e.g. ["/bin/bash", "/etc/init.d/spawner restart"] Here we should NOT have a "-c" parameter!	required
timeout	int	Timeout duration that is waited for any response. Defaults to 30. Each time a response is found in stdout or stderr we wait another timeout duration to make sure we get the full output of the command.	30
write_stderr_to_error_log	bool	Flag to control if output in stderr should be written to the debug or the error log stream. Defaults to True (write to error log).	True

Returns:

Type	Description
str | None	str | None: Response of the command or None if the call fails.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def exec_pod_command_interactive(
    self,
    pod_name: str,
    commands: list,
    timeout: int = 30,
    write_stderr_to_error_log: bool = True,
) -> str | None:
    """Execute a command inside a Kubernetes pod (similar to kubectl exec on command line).

    Other than exec_pod_command() method above this is an interactive execution using
    stdin and reading the output from stdout and stderr. This is required for longer
    running commands. It is currently used for restarting the spawner of Archive Center.
    The output of the command is pushed into the logging.

    Args:
        pod_name (str):
            The name of the Kubernetes pod in the current namespace
        commands (list):
            A list of command and its parameters, e.g. ["/bin/bash", "/etc/init.d/spawner restart"]
            Here we should NOT have a "-c" parameter!
        timeout (int, optional):
            Timeout duration that is waited for any response. Defaults to 30.
            Each time a response is found in stdout or stderr we wait another timeout duration
            to make sure we get the full output of the command.
        write_stderr_to_error_log (bool, optional):
            Flag to control if output in stderr should be written to the debug or the error log stream.
            Defaults to True (write to error log).

    Returns:
        str | None:
            Response of the command or None if the call fails.

    """

    if not commands:
        self.logger.error("No commands to execute on pod ->'%s'!", pod_name)
        return None

    pod = self.get_pod(pod_name=pod_name)
    if not pod:
        self.logger.error("Pod -> '%s' does not exist. Cannot execute commands -> %s", pod_name, commands)
        return None

    # Get first command - this should be the shell:
    command = commands.pop(0)

    try:
        response = stream(
            self.get_core_v1_api().connect_get_namespaced_pod_exec,
            pod_name,
            self.get_namespace(),
            command=command,
            stderr=True,
            stdin=True,  # This is important!
            stdout=True,
            tty=False,
            _preload_content=False,  # This is important!
            _request_timeout=timeout,
        )
    except ApiException:
        self.logger.error(
            "Failed to execute command -> %s in pod -> '%s'",
            command,
            pod_name,
        )
        return None

    while response.is_open():
        got_response = False
        response.update(timeout=timeout)
        if response.peek_stdout():
            self.logger.debug(response.read_stdout().replace("\n", " "))
            got_response = True
        if response.peek_stderr():
            if write_stderr_to_error_log:
                self.logger.error(response.read_stderr().replace("\n", " "))
            else:
                self.logger.debug(response.read_stderr().replace("\n", " "))
            got_response = True
        if commands:
            command = commands.pop(0)
            self.logger.debug(
                "Execute command -> %s in pod -> '%s'",
                command,
                pod_name,
            )
            response.write_stdin(command + "\n")
        # We continue as long as we get some response during timeout period
        elif not got_response:
            break

    response.close()

    return response

delete_pod

delete_pod(pod_name: str) -> V1Pod | None

Delete a pod in the configured namespace (the namespace is defined in the class constructor).

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#delete_namespaced_pod

Parameters:

Name	Type	Description	Default
pod_name	str	The name of the Kubernetes pod in the current namespace.	required

Returns:

Type	Description
V1Pod | None	V1Pod (object) or None if the call fails.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def delete_pod(self, pod_name: str) -> V1Pod | None:
    """Delete a pod in the configured namespace (the namespace is defined in the class constructor).

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#delete_namespaced_pod

    Args:
        pod_name (str):
            The name of the Kubernetes pod in the current namespace.

    Returns:
        V1Pod (object) or None if the call fails.

    """

    pod = self.get_pod(pod_name=pod_name)
    if not pod:
        self.logger.error("Pod -> '%s' does not exist! Cannot delete it.", pod_name)
        return None

    try:
        response = self.get_core_v1_api().delete_namespaced_pod(
            pod_name,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to delete pod -> '%s'",
            pod_name,
        )
        return None

    return response

get_config_map

get_config_map(config_map_name: str) -> V1ConfigMap

Get a config map in the configured namespace (the namespace is defined in the class constructor).

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#read_namespaced_config_map

Parameters:

Name	Type	Description	Default
config_map_name	str	The name of the Kubernetes config map in the current namespace.	required

Returns:

Name	Type	Description
V1ConfigMap	object	Kubernetes Config Map object that includes these fields: - api_version: The Kubernetes API version. - metadata: A V1ObjectMeta object representing metadata about the V1ConfigMap object, such as its name, labels, and annotations. - data: A dictionary containing the non-binary data stored in the ConfigMap, where the keys represent the keys of the data items and the values represent the values of the data items. - binary_data: A dictionary containing the binary data stored in the ConfigMap, where the keys represent the keys of the binary data items and the values represent the values of the binary data items. Binary data is encoded as base64 strings in the dictionary values.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_config_map(self, config_map_name: str) -> V1ConfigMap:
    """Get a config map in the configured namespace (the namespace is defined in the class constructor).

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#read_namespaced_config_map

    Args:
        config_map_name (str):
            The name of the Kubernetes config map in the current namespace.

    Returns:
        V1ConfigMap (object):
            Kubernetes Config Map object that includes these fields:
            - api_version:
                The Kubernetes API version.
            - metadata:
                A V1ObjectMeta object representing metadata about the V1ConfigMap object,
                such as its name, labels, and annotations.
            - data:
                A dictionary containing the non-binary data stored in the ConfigMap,
                where the keys represent the keys of the data items and the values represent
                the values of the data items.
            - binary_data:
                A dictionary containing the binary data stored in the ConfigMap,
                where the keys represent the keys of the binary data items and the values
                represent the values of the binary data items. Binary data is encoded as base64
                strings in the dictionary values.

    """

    try:
        response = self.get_core_v1_api().read_namespaced_config_map(
            name=config_map_name,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to get config map -> '%s'",
            config_map_name,
        )
        return None

    return response

list_config_maps

list_config_maps(
    field_selector: str | None = None,
    label_selector: str | None = None,
) -> V1ConfigMapList

List all Kubernetes Config Maps in the current namespace.

The list can be filtered by providing field selectors and label selectors. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_config_map

Parameters:

Name	Type	Description	Default
field_selector	str | None	To filter the result based on fields. Defaults to None.	None
label_selector	str | None	To filter result based on labels. Defaults to None.	None

Returns:

Name	Type	Description
	V1ConfigMapList	V1ConfigMapList (object) or None if the call fails
	V1ConfigMapList	Properties can be accessed with the "." notation (this is an object not a dict!):
	V1ConfigMapList	api_version: The Kubernetes API version.
	V1ConfigMapList	items: A list of V1ConfigMap objects, each representing a config map. You can access the fields of a V1Pod object using dot notation, for example, cm.metadata.name to access the name of the config map
	V1ConfigMapList	kind: The Kubernetes object kind, which is always "ConfigMapList".
	V1ConfigMapList	metadata: Additional metadata about the config map list, such as the resource version.
See	V1ConfigMapList	https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1ConfigMapList.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def list_config_maps(
    self,
    field_selector: str | None = None,
    label_selector: str | None = None,
) -> V1ConfigMapList:
    """List all Kubernetes Config Maps in the current namespace.

    The list can be filtered by providing field selectors and label selectors.
    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_config_map

    Args:
        field_selector (str | None, optional):
            To filter the result based on fields. Defaults to None.
        label_selector (str | None, optional):
            To filter result based on labels. Defaults to None.

    Returns:
        V1ConfigMapList (object) or None if the call fails
        Properties can be accessed with the "." notation (this is an object not a dict!):
        - api_version: The Kubernetes API version.
        - items: A list of V1ConfigMap objects, each representing a config map. You can access the fields of a
                 V1Pod object using dot notation, for example, cm.metadata.name to access the name of the config map
        - kind: The Kubernetes object kind, which is always "ConfigMapList".
        - metadata: Additional metadata about the config map list, such as the resource version.
        See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1ConfigMapList.md

    """

    try:
        response = self.get_core_v1_api().list_namespaced_config_map(
            field_selector=field_selector,
            label_selector=label_selector,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to list config maps with field selector -> '%s' and label selector -> '%s'",
            field_selector,
            label_selector,
        )
        return None

    return response

find_config_map

find_config_map(config_map_name: str) -> V1ConfigMapList

Find a Kubernetes Config Map based on its name.

This is just a wrapper method for list_config_maps() that uses the name as a field selector.

Parameters:

Name	Type	Description	Default
config_map_name	str	The name of the Kubernetes Config Map to search for.	required

Returns:

Name	Type	Description
object	V1ConfigMapList	V1ConfigMapList (object) or None if the call fails.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def find_config_map(self, config_map_name: str) -> V1ConfigMapList:
    """Find a Kubernetes Config Map based on its name.

    This is just a wrapper method for list_config_maps()
    that uses the name as a field selector.

    Args:
        config_map_name (str):
            The name of the Kubernetes Config Map to search for.

    Returns:
        object:
            V1ConfigMapList (object) or None if the call fails.

    """

    try:
        response = self.list_config_maps(
            field_selector="metadata.name={}".format(config_map_name),
        )
    except ApiException:
        self.logger.error(
            "Failed to find config map -> '%s'",
            config_map_name,
        )
        return None

    return response

replace_config_map

replace_config_map(
    config_map_name: str, config_map_data: dict
) -> V1ConfigMap

Replace a Config Map with a new specification.

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#replace_namespaced_config_map

Parameters:

Name	Type	Description	Default
config_map_name	str	The name of the Kubernetes Config Map to replace.	required
config_map_data	dict	The updated specification of the Config Map.	required

Returns:

Name	Type	Description
V1ConfigMap	object	Updated Kubernetes Config Map object or None if the call fails. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1ConfigMap.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def replace_config_map(
    self,
    config_map_name: str,
    config_map_data: dict,
) -> V1ConfigMap:
    """Replace a Config Map with a new specification.

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#replace_namespaced_config_map

    Args:
        config_map_name (str):
            The name of the Kubernetes Config Map to replace.
        config_map_data (dict):
            The updated specification of the Config Map.

    Returns:
        V1ConfigMap (object):
            Updated Kubernetes Config Map object or None if the call fails.
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1ConfigMap.md

    """

    try:
        response = self.get_core_v1_api().replace_namespaced_config_map(
            name=config_map_name,
            namespace=self.get_namespace(),
            body=client.V1ConfigMap(
                metadata=client.V1ObjectMeta(
                    name=config_map_name,
                ),
                data=config_map_data,
            ),
        )
    except ApiException:
        self.logger.error(
            "Failed to replace config map -> '%s'",
            config_map_name,
        )
        return None

    return response

get_stateful_set

get_stateful_set(sts_name: str) -> V1StatefulSet

Get a Kubernetes Stateful Set based on its name.

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1Api.md#read_namespaced_stateful_set

Parameters:

Name	Type	Description	Default
sts_name	str	The name of the Kubernetes stateful set	required

Returns:

Name	Type	Description
V1StatefulSet	object	Kubernetes Stateful Set object or None if the call fails. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1StatefulSet.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_stateful_set(self, sts_name: str) -> V1StatefulSet:
    """Get a Kubernetes Stateful Set based on its name.

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1Api.md#read_namespaced_stateful_set

    Args:
        sts_name (str):
            The name of the Kubernetes stateful set

    Returns:
        V1StatefulSet (object):
            Kubernetes Stateful Set object or None if the call fails.
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1StatefulSet.md

    """

    try:
        response = self.get_apps_v1_api().read_namespaced_stateful_set(
            name=sts_name,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to get stateful set -> '%s'",
            sts_name,
        )
        return None

    return response

get_stateful_set_scale

get_stateful_set_scale(sts_name: str) -> V1Scale

Get the number of replicas for a Kubernetes Stateful Set.

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1Api.md#read_namespaced_stateful_set_scale

Parameters:

Name	Type	Description	Default
sts_name	str	The name of the Kubernetes Stateful Set.	required

Returns:

Name	Type	Description
V1Scale	object	Kubernetes Scale object or None if the call fails. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Scale.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_stateful_set_scale(self, sts_name: str) -> V1Scale:
    """Get the number of replicas for a Kubernetes Stateful Set.

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1Api.md#read_namespaced_stateful_set_scale

    Args:
        sts_name (str):
            The name of the Kubernetes Stateful Set.

    Returns:
        V1Scale (object):
            Kubernetes Scale object or None if the call fails.
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Scale.md

    """

    try:
        response = self.get_apps_v1_api().read_namespaced_stateful_set_scale(
            name=sts_name,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to get scaling (replicas) of stateful set -> '%s'",
            sts_name,
        )
        return None

    return response

patch_stateful_set

patch_stateful_set(
    sts_name: str, sts_body: dict
) -> V1StatefulSet

Patch a Stateful set with new values.

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1Api.md#patch_namespaced_stateful_set

Parameters:

Name	Type	Description	Default
sts_name	str	The name of the Kubernetes stateful set in the current namespace.	required
sts_body	dict	The patch body.	required

Returns:

Name	Type	Description
V1StatefulSet	object	The patched Kubernetes Stateful Set object or None if the call fails. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1StatefulSet.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def patch_stateful_set(self, sts_name: str, sts_body: dict) -> V1StatefulSet:
    """Patch a Stateful set with new values.

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1Api.md#patch_namespaced_stateful_set

    Args:
        sts_name (str):
            The name of the Kubernetes stateful set in the current namespace.
        sts_body (dict):
            The patch body.

    Returns:
        V1StatefulSet (object):
            The patched Kubernetes Stateful Set object or None if the call fails.
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1StatefulSet.md

    """

    try:
        response = self.get_apps_v1_api().patch_namespaced_stateful_set(
            name=sts_name,
            namespace=self.get_namespace(),
            body=sts_body,
        )
    except ApiException:
        self.logger.error(
            "Failed to patch stateful set -> '%s' with -> %s",
            sts_name,
            str(sts_body),
        )
        return None

    return response

scale_stateful_set

scale_stateful_set(
    sts_name: str, scale: int
) -> V1StatefulSet

Scale a stateful set to a specific number of replicas.

It uses the class method patch_stateful_set() above.

Parameters:

Name	Type	Description	Default
sts_name	str	The name of the Kubernetes stateful set in the current namespace.	required
scale	int	The number of replicas (pods) the stateful set shall be scaled to.	required

Returns:

Name	Type	Description
V1StatefulSet	object	Kubernetes Stateful Set object or None if the call fails. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1StatefulSet.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def scale_stateful_set(self, sts_name: str, scale: int) -> V1StatefulSet:
    """Scale a stateful set to a specific number of replicas.

    It uses the class method patch_stateful_set() above.

    Args:
        sts_name (str):
            The name of the Kubernetes stateful set in the current namespace.
        scale (int):
            The number of replicas (pods) the stateful set shall be scaled to.

    Returns:
        V1StatefulSet (object):
            Kubernetes Stateful Set object or None if the call fails.
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1StatefulSet.md

    """

    try:
        response = self.patch_stateful_set(
            sts_name=sts_name,
            sts_body={"spec": {"replicas": scale}},
        )
    except ApiException:
        self.logger.error(
            "Failed to scale stateful set -> '%s' to -> %s replicas",
            sts_name,
            scale,
        )
        return None

    return response

get_service

get_service(service_name: str) -> V1Service

Get a Kubernetes Service with a defined name in the current namespace.

Parameters:

Name	Type	Description	Default
service_name	str	The name of the Kubernetes Service in the current namespace.	required

Returns:

Name	Type	Description
V1Service	object	Kubernetes Service object or None if the call fails This is NOT a dict but an object - then you have to use the "." syntax to access to returned elements. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Service.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_service(self, service_name: str) -> V1Service:
    """Get a Kubernetes Service with a defined name in the current namespace.

    Args:
        service_name (str):
            The name of the Kubernetes Service in the current namespace.

    Returns:
        V1Service (object):
            Kubernetes Service object or None if the call fails
            This is NOT a dict but an object - then you have to use the "." syntax to access to returned elements.
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Service.md

    """

    try:
        response = self.get_core_v1_api().read_namespaced_service(
            name=service_name,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to get service -> '%s'",
            service_name,
        )
        return None

    return response

list_services

list_services(
    field_selector: str = "", label_selector: str = ""
) -> None

List all Kubernetes Service in the current namespace.

The list can be filtered by providing field selectors and label selectors. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_service

Parameters:

Name	Type	Description	Default
field_selector	str	To filter result based on fields. Defaults to "".	''
label_selector	str	To filter result based on labels. Defaults to "".	''

Returns:

Name	Type	Description
V1ServiceList	object	A list of Kubernetes Services or None if the call fails. Properties can be accessed with the "." notation (this is an object not a dict!): - api_version: The Kubernetes API version. - items: A list of V1Service objects, each representing a service. You can access the fields of a V1Service object using dot notation, for example, service.metadata.name to access the name of the service - kind: The Kubernetes object kind, which is always "ServiceList". - metadata: Additional metadata about the pod list, such as the resource version. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1ServiceList.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def list_services(self, field_selector: str = "", label_selector: str = "") -> None:
    """List all Kubernetes Service in the current namespace.

    The list can be filtered by providing field selectors and label selectors.
    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_service

    Args:
        field_selector (str, optional):
            To filter result based on fields. Defaults to "".
        label_selector (str, optional):
            To filter result based on labels. Defaults to "".

    Returns:
        V1ServiceList (object):
            A list of Kubernetes Services or None if the call fails.
            Properties can be accessed with the "." notation (this is an object not a dict!):
            - api_version: The Kubernetes API version.
            - items: A list of V1Service objects, each representing a service.
                    You can access the fields of a V1Service object using dot notation,
                    for example, service.metadata.name to access the name of the service
            - kind: The Kubernetes object kind, which is always "ServiceList".
            - metadata: Additional metadata about the pod list, such as the resource version.
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1ServiceList.md

    """

    try:
        response = self.get_core_v1_api().list_namespaced_service(
            field_selector=field_selector,
            label_selector=label_selector,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to list services with field selector -> '%s' and label selector -> '%s'",
            field_selector,
            label_selector,
        )
        return None

    return response

patch_service

patch_service(
    service_name: str, service_body: dict
) -> V1Service

Patch a Kubernetes Service with an updated spec.

Parameters:

Name	Type	Description	Default
service_name	str	The name of the Kubernetes Service in the current namespace.	required
service_body	dict	The new / updated Service body spec. (will be merged with existing values)	required

Returns:

Name	Type	Description
V1Service	object	The patched Kubernetes Service or None if the call fails. This is NOT a dict but an object - you have to use the "." syntax to access to returned elements See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Service.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def patch_service(self, service_name: str, service_body: dict) -> V1Service:
    """Patch a Kubernetes Service with an updated spec.

    Args:
        service_name (str):
            The name of the Kubernetes Service in the current namespace.
        service_body (dict):
            The new / updated Service body spec.
            (will be merged with existing values)

    Returns:
        V1Service (object):
            The patched Kubernetes Service or None if the call fails.
            This is NOT a dict but an object - you have to use the "." syntax
            to access to returned elements
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Service.md

    """

    try:
        response = self.get_core_v1_api().patch_namespaced_service(
            name=service_name,
            namespace=self.get_namespace(),
            body=service_body,
        )
    except ApiException:
        self.logger.error(
            "Failed to patch service -> '%s' with -> %s",
            service_name,
            service_body,
        )
        return None

    return response

get_ingress

get_ingress(ingress_name: str) -> V1Ingress

Get a Kubernetes Ingress with a defined name in the current namespace.

Parameters:

Name	Type	Description	Default
ingress_name	str	The name of the Kubernetes Ingress in the current namespace.	required

Returns:

Name	Type	Description
V1Ingress	object	Kubernetes Ingress or None if the call fails This is NOT a dict but an object - then you have to use the "." syntax to access to returned elements. See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Ingress.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def get_ingress(self, ingress_name: str) -> V1Ingress:
    """Get a Kubernetes Ingress with a defined name in the current namespace.

    Args:
        ingress_name (str):
            The name of the Kubernetes Ingress in the current namespace.

    Returns:
        V1Ingress (object):
            Kubernetes Ingress or None if the call fails
            This is NOT a dict but an object - then you have to use the "." syntax to access to returned elements.
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Ingress.md

    """

    try:
        response = self.get_networking_v1_api().read_namespaced_ingress(
            name=ingress_name,
            namespace=self.get_namespace(),
        )
    except ApiException:
        self.logger.error(
            "Failed to get ingress -> '%s'!",
            ingress_name,
        )
        return None

    return response

patch_ingress

patch_ingress(
    ingress_name: str, ingress_body: dict
) -> V1Ingress

Patch a Kubernetes Ingress with an updated spec.

See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/NetworkingV1Api.md#patch_namespaced_ingress

Parameters:

Name	Type	Description	Default
ingress_name	str	The name of the Kubernetes Ingress in the current namespace.	required
ingress_body	dict	The new / updated ingress body spec. (will be merged with existing values)	required

Returns:

Name	Type	Description
V1Ingress	object	The patched Kubernetes Ingress object or None if the call fails This is NOT a dict but an object - you have to use the "." syntax to access to returned elements See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Ingress.md

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def patch_ingress(self, ingress_name: str, ingress_body: dict) -> V1Ingress:
    """Patch a Kubernetes Ingress with an updated spec.

    See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/NetworkingV1Api.md#patch_namespaced_ingress

    Args:
        ingress_name (str):
            The name of the Kubernetes Ingress in the current namespace.
        ingress_body (dict):
            The new / updated ingress body spec.
            (will be merged with existing values)

    Returns:
        V1Ingress (object):
            The patched Kubernetes Ingress object or None if the call fails
            This is NOT a dict but an object - you have to use the
            "." syntax to access to returned elements
            See: https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1Ingress.md

    """

    try:
        response = self.get_networking_v1_api().patch_namespaced_ingress(
            name=ingress_name,
            namespace=self.get_namespace(),
            body=ingress_body,
        )
    except ApiException:
        self.logger.error(
            "Failed to patch ingress -> '%s' with -> %s",
            ingress_name,
            ingress_body,
        )
        return None

    return response

update_ingress_backend_services

update_ingress_backend_services(
    ingress_name: str,
    hostname: str,
    service_name: str,
    service_port: int,
    path: str = "/",
) -> V1Ingress

Update a backend service and port of a Kubernetes Ingress.

This method updates the backend service and port for a specific hostname and path in a Kubernetes Ingress. It supports multiple paths configured for a host.

Parameters:

Name	Type	Description	Default
ingress_name	str	The name of the Kubernetes Ingress in the current namespace.	required
hostname	str	The hostname that should get an updated backend service / port.	required
service_name	str	The new backend service name.	required
service_port	int	The new backend service port.	required
path	str	The path to match for the backend service update. Defaults to "/".	'/'

Returns:

Name	Type	Description
V1Ingress	object	The updated Kubernetes Ingress object or None if the call fails.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def update_ingress_backend_services(
    self,
    ingress_name: str,
    hostname: str,
    service_name: str,
    service_port: int,
    path: str = "/",
) -> V1Ingress:
    """Update a backend service and port of a Kubernetes Ingress.

    This method updates the backend service and port for a specific hostname and path
    in a Kubernetes Ingress. It supports multiple paths configured for a host.

    Args:
        ingress_name (str):
            The name of the Kubernetes Ingress in the current namespace.
        hostname (str):
            The hostname that should get an updated backend service / port.
        service_name (str):
            The new backend service name.
        service_port (int):
            The new backend service port.
        path (str, optional):
            The path to match for the backend service update. Defaults to "/".

    Returns:
        V1Ingress (object):
            The updated Kubernetes Ingress object or None if the call fails.

    """

    ingress = self.get_ingress(ingress_name=ingress_name)
    if not ingress:
        return None

    host = ""
    rules = ingress.spec.rules
    rule_index = 0
    path_index = -1
    for rule in rules:
        if hostname in rule.host:
            host = rule.host
            for i, rule_path in enumerate(rule.http.paths):
                if rule_path.path == path:
                    path_index = i
                    backend = rule_path.backend
                    service = backend.service

                    self.logger.debug(
                        "Replace backend service -> '%s' (%s) with new backend service -> '%s' (%s)",
                        service.name,
                        service.port.number,
                        service_name,
                        service_port,
                    )

                    service.name = service_name
                    service.port.number = service_port
                    break
            if path_index != -1:
                break
        rule_index += 1

    if not host or path_index == -1:
        self.logger.error(
            "Cannot find host and path to upgrade the Kubernetes Ingress -> '%s'",
            ingress_name,
        )
        return None

    body = [
        {
            "op": "replace",
            "path": "/spec/rules/{}/http/paths/{}/backend/service/name".format(rule_index, path_index),
            "value": service_name,
        },
        {
            "op": "replace",
            "path": "/spec/rules/{}/http/paths/{}/backend/service/port/number".format(rule_index, path_index),
            "value": service_port,
        },
    ]

    return self.patch_ingress(ingress_name, body)

verify_pod_status

verify_pod_status(
    pod_name: str,
    timeout: int = 1800,
    total_containers: int = 1,
    ready_containers: int = 1,
    retry_interval: int = 30,
) -> bool

Verify if a pod is in a 'Ready' state by checking the status of its containers.

This function waits for a Kubernetes pod to reach the 'Ready' state, where a specified number of containers are ready. It checks the pod status at regular intervals and reports the status using logs. If the pod does not reach the 'Ready' state within the specified timeout, it returns False.

Parameters:

Name	Type	Description	Default
pod_name	str	The name of the pod to check the status for.	required
timeout	int	The maximum time (in seconds) to wait for the pod to become ready. Defaults to 1800.	1800
total_containers	int	The total number of containers expected to be running in the pod. Defaults to 1.	1
ready_containers	int	The minimum number of containers that need to be in a ready state. Defaults to 1.	1
retry_interval	int	Time interval (in seconds) between each retry to check pod readiness. Defaults to 30.	30

Returns:

Name	Type	Description
bool	bool	Returns True if the pod reaches the 'Ready' state with the specified number of containers ready within the timeout. Otherwise, returns False.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def verify_pod_status(
    self,
    pod_name: str,
    timeout: int = 1800,
    total_containers: int = 1,
    ready_containers: int = 1,
    retry_interval: int = 30,
) -> bool:
    """Verify if a pod is in a 'Ready' state by checking the status of its containers.

    This function waits for a Kubernetes pod to reach the 'Ready' state, where a specified number
    of containers are ready. It checks the pod status at regular intervals and reports the status
    using logs. If the pod does not reach the 'Ready' state within the specified timeout,
    it returns `False`.

    Args:
        pod_name (str):
            The name of the pod to check the status for.
        timeout (int, optional):
            The maximum time (in seconds) to wait for the pod to become ready. Defaults to 1800.
        total_containers (int, optional):
            The total number of containers expected to be running in the pod. Defaults to 1.
        ready_containers (int, optional):
            The minimum number of containers that need to be in a ready state. Defaults to 1.
        retry_interval (int, optional):
            Time interval (in seconds) between each retry to check pod readiness. Defaults to 30.

    Returns:
        bool:
            Returns `True` if the pod reaches the 'Ready' state with the specified number of containers ready
            within the timeout. Otherwise, returns `False`.

    """

    def wait_for_pod_ready(pod_name: str, timeout: int) -> bool:
        """Wait until the pod is in the 'Ready' state with the specified number of containers ready.

        This sub method repeatedly checks the readiness of the pod, logging the
        status of the containers. If the pod does not exist, it retries after waiting
        and logs detailed information at each step.

        Args:
            pod_name (str):
                The name of the pod to check the status for.
            timeout (int):
                The maximum time (in seconds) to wait for the pod to become ready.

        Returns:
            bool:
                Returns `True` if the pod is ready with the specified number of containers in a 'Ready' state.
                Otherwise, returns `False`.

        """

        elapsed_time = 0  # Initialize elapsed time

        while elapsed_time < timeout:
            pod = self.get_pod(pod_name=pod_name)

            if not pod:
                self.logger.warning(
                    "Pod -> '%s' does not exist, waiting 300 seconds to retry.",
                    pod_name,
                )
                time.sleep(300)
                pod = self.get_pod(pod_name=pod_name)

            if not pod:
                self.logger.error(
                    "Pod -> '%s' still does not exist after retry!",
                    pod_name,
                )
                return False

            # Get the ready status of containers
            container_statuses = pod.status.container_statuses
            if container_statuses and all(container.ready for container in container_statuses):
                current_ready_containers = sum(1 for c in container_statuses if c.ready)
                total_containers_in_pod = len(container_statuses)

                if current_ready_containers >= ready_containers and total_containers_in_pod == total_containers:
                    self.logger.info(
                        "Pod -> '%s' is ready with %d/%d containers.",
                        pod_name,
                        current_ready_containers,
                        total_containers_in_pod,
                    )
                    return True
                else:
                    self.logger.debug(
                        "Pod -> '%s' is not yet ready (%d/%d).",
                        pod_name,
                        current_ready_containers,
                        total_containers_in_pod,
                    )
            else:
                self.logger.debug("Pod -> '%s' is not yet ready.", pod_name)

            self.logger.info(
                "Waiting %s seconds before next pod status check.",
                retry_interval,
            )
            time.sleep(
                retry_interval,
            )  # Sleep for the retry interval before checking again
            elapsed_time += retry_interval

        self.logger.error(
            "Pod -> '%s' is not ready after %d seconds.",
            pod_name,
            timeout,
        )
        return False

    # end method definition

    # Wait until the pod is ready
    return wait_for_pod_ready(pod_name=pod_name, timeout=timeout)

verify_pod_deleted

verify_pod_deleted(
    pod_name: str,
    timeout: int = 300,
    retry_interval: int = 30,
) -> bool

Verify if a pod is deleted within the specified timeout.

Parameters:

Name	Type	Description	Default
pod_name	str	The name of the pod to check.	required
timeout	int	Maximum time to wait for the pod to be deleted (in seconds). Defaults to 300.	300
retry_interval	int	Time interval between retries (in seconds). Defaults to 30.	30

Returns:

Name	Type	Description
bool	bool	True if the pod is deleted, False otherwise.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def verify_pod_deleted(
    self,
    pod_name: str,
    timeout: int = 300,
    retry_interval: int = 30,
) -> bool:
    """Verify if a pod is deleted within the specified timeout.

    Args:
        pod_name (str):
            The name of the pod to check.
        timeout (int, optional):
            Maximum time to wait for the pod to be deleted (in seconds). Defaults to 300.
        retry_interval (int, optional):
            Time interval between retries (in seconds). Defaults to 30.

    Returns:
        bool:
            True if the pod is deleted, False otherwise.

    """

    elapsed_time = 0  # Initialize elapsed time

    while elapsed_time < timeout:
        pod = self.get_pod(pod_name=pod_name)

        if not pod:
            self.logger.info("Pod -> '%s' has been deleted successfully.", pod_name)
            return True

        pod_status = self.get_core_v1_api().read_namespaced_pod_status(
            pod_name,
            self.get_namespace(),
        )

        self.logger.info(
            "Pod -> '%s' still exists with conditions -> %s. Waiting %s seconds before next check.",
            pod_name,
            str(
                [
                    pod_condition.type
                    for pod_condition in pod_status.status.conditions
                    if pod_condition.status == "True"
                ]
            ),
            retry_interval,
        )
        time.sleep(retry_interval)
        elapsed_time += retry_interval

    self.logger.error("Pod -> '%s' was not deleted within %d seconds.", pod_name, timeout)

    return False

restart_deployment

restart_deployment(
    deployment_name: str,
    force: bool = False,
    wait: bool = False,
    wait_timeout: int = 1800,
) -> bool

Restart a Kubernetes deployment using rolling restart.

Parameters:

Name	Type	Description	Default
deployment_name	str	Name of the Kubernetes deployment.	required
force	bool	If True, all pod instances will be forcefully deleted. Defaults to False.	False
wait	bool	If True, wait for the deployment to be ready again. Defaults to False.	False
wait_timeout	int	Maximum time to wait for the deployment to be ready again (in seconds). Defaults to 1800.	1800

Returns:

Name	Type	Description
bool	bool	True if successful, False otherwise.

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def restart_deployment(
    self, deployment_name: str, force: bool = False, wait: bool = False, wait_timeout: int = 1800
) -> bool:
    """Restart a Kubernetes deployment using rolling restart.

    Args:
        deployment_name (str):
            Name of the Kubernetes deployment.
        force (bool, optional):
            If True, all pod instances will be forcefully deleted. Defaults to False.
        wait (bool, optional):
            If True, wait for the deployment to be ready again. Defaults to False.
        wait_timeout (int, optional):
            Maximum time to wait for the deployment to be ready again (in seconds). Defaults to 1800.

    Returns:
        bool:
            True if successful, False otherwise.

    """

    success = True

    if not force:
        now = datetime.now(UTC).isoformat(timespec="seconds") + "Z"

        body = {
            "spec": {
                "template": {
                    "metadata": {
                        "annotations": {
                            "kubectl.kubernetes.io/restartedAt": now,
                        },
                    },
                },
            },
        }
        try:
            self.get_apps_v1_api().patch_namespaced_deployment(
                deployment_name,
                self.get_namespace(),
                body,
                pretty="true",
            )
            self.logger.debug("Triggered restart of deployment -> '%s'.", deployment_name)

        except ApiException as api_exception:
            self.logger.error(
                "Failed to restart deployment -> '%s'; error -> %s!", deployment_name, str(api_exception)
            )
            return False

    # If force is set, all pod instances will be forcefully deleted.
    elif force:
        self.logger.info("Force deleting all pods of deployment -> '%s'.", deployment_name)

        try:
            # Get the Deployment to retrieve its pod labels
            deployment = self.get_apps_v1_api().read_namespaced_deployment(
                name=deployment_name, namespace=self.get_namespace()
            )

            # Get the label selector for the Deployment
            label_selector = deployment.spec.selector.match_labels

            # List pods matching the label selector
            pods = (
                self.get_core_v1_api()
                .list_namespaced_pod(
                    namespace=self.get_namespace(),
                    label_selector=",".join([f"{k}={v}" for k, v in label_selector.items()]),
                )
                .items
            )

            # Loop through the pods and delete each one
            for pod in pods:
                pod_name = pod.metadata.name
                try:
                    # Define the delete options with force and grace period set to 0
                    body = client.V1DeleteOptions(grace_period_seconds=0, propagation_policy="Foreground")

                    # Call the delete_namespaced_pod method
                    self.get_core_v1_api().delete_namespaced_pod(
                        name=pod_name, namespace=self.get_namespace(), body=body
                    )
                    self.logger.info(
                        "Pod '%s' in namespace '%s' has been deleted forcefully.", pod_name, self.get_namespace()
                    )
                except Exception as e:
                    self.logger.error("Error occurred while deleting pod '%s': %s", pod_name, e)
                    success = False

        except Exception as e:
            self.logger.error("Error occurred while getting Deployment '%s': %s", deployment_name, e)
            success = False

    start_time = time.time()
    while wait:
        self.logger.info("Waiting for restart of deployment -> '%s' to complete.", deployment_name)
        # Get the deployment
        deployment = self.get_apps_v1_api().read_namespaced_deployment_status(deployment_name, self.get_namespace())

        # Check the availability status
        ready_replicas = deployment.status.ready_replicas or 0
        updated_replicas = deployment.status.updated_replicas or 0
        unavailable_replicas = deployment.status.unavailable_replicas or 0
        total_replicas = deployment.status.replicas or 0
        desired_replicas = deployment.spec.replicas or 0

        self.logger.debug(
            "Deployment status -> updated pods: %s/%s -> ready replicas: %s/%s",
            updated_replicas,
            desired_replicas,
            ready_replicas,
            total_replicas,
        )

        if (
            updated_replicas == desired_replicas
            and unavailable_replicas == 0
            and total_replicas == desired_replicas
        ):
            self.logger.info("Restart of deployment -> '%s' completed successfully", deployment_name)
            break

        if (time.time() - start_time) > wait_timeout:
            self.logger.error("Timed out waiting for restart of deployment -> '%s' to complete.", deployment_name)
            success = False
            break

        # Sleep for a while before checking again
        time.sleep(20)

    return success

restart_stateful_set

restart_stateful_set(
    sts_name: str,
    wait: bool = True,
    wait_timeout: int = 1800,
    endpoint_service_name: str | None = None,
) -> bool

Enterprise-grade zero-downtime rolling restart of a stateful set.

Guarantees: - Zero-downtime traffic handling (via readiness probes) - Kubernetes-native rollout - Deterministic state transitions - Traffic draining via Service endpoints - Readiness-gated pod rotation - Failure detection - Rollout completion validation

Parameters:

Name	Type	Description	Default
sts_name	str	Name of the stateful set.	required
wait	bool	Whether to wait for rollout completion. Defaults to True.	True
wait_timeout	int	Max seconds to wait. Defaults to 1800.	1800
endpoint_service_name	str | None	Service name for endpoint gating. Defaults to None (no endpoint gating).	None

Returns:

Name	Type	Description
bool	bool	True if successful, False otherwise

Source code in packages/pyxecm/src/pyxecm_customizer/k8s.py

def restart_stateful_set(
    self,
    sts_name: str,
    wait: bool = True,
    wait_timeout: int = 1800,
    endpoint_service_name: str | None = None,
) -> bool:
    """Enterprise-grade zero-downtime rolling restart of a stateful set.

    Guarantees:
    - Zero-downtime traffic handling (via readiness probes)
    - Kubernetes-native rollout
    - Deterministic state transitions
    - Traffic draining via Service endpoints
    - Readiness-gated pod rotation
    - Failure detection
    - Rollout completion validation

    Args:
        sts_name (str):
            Name of the stateful set.
        wait (bool, optional):
            Whether to wait for rollout completion. Defaults to True.
        wait_timeout (int, optional):
            Max seconds to wait. Defaults to 1800.
        endpoint_service_name (str | None, optional):
            Service name for endpoint gating. Defaults to None (no endpoint gating).

    Returns:
        bool:
            True if successful, False otherwise

    """

    self.logger.info("Starting rolling restart for stateful set -> '%s'", sts_name)

    # ------------------------------------------------------------------
    # Load stateful set
    # ------------------------------------------------------------------

    sts = self.get_stateful_set(sts_name)
    if not sts:
        self.logger.error("Stateful set -> '%s' not found!", sts_name)
        return False

    replicas = sts.spec.replicas or 0

    # ------------------------------------------------------------------
    # Safety checks
    # ------------------------------------------------------------------

    if replicas < 2:
        self.logger.info(
            "Stateful set -> '%s' has only %s replica(s). Proceeding with restart in downtime-allowed mode.",
            sts_name,
            replicas,
        )

    # ------------------------------------------------------------------
    # Readiness probe validation
    # ------------------------------------------------------------------

    containers = sts.spec.template.spec.containers
    for container in containers:
        if not container.readiness_probe:
            self.logger.error(
                "Container -> '%s' in stateful set -> '%s' has no readiness probe. Zero-downtime restart not possible!",
                container.name,
                sts_name,
            )
            return False
    # end for container in containers

    # ------------------------------------------------------------------
    # Trigger rolling restart via annotation
    # ------------------------------------------------------------------

    now = datetime.now(UTC).isoformat(timespec="seconds") + "Z"

    patch_body = {"spec": {"template": {"metadata": {"annotations": {"kubectl.kubernetes.io/restartedAt": now}}}}}

    sts = self.patch_stateful_set(sts_name, patch_body)
    if not sts:
        self.logger.error("Failed to trigger rolling restart for stateful set -> '%s'", sts_name)
        return False

    self.logger.info("Rolling restart triggered for stateful set -> '%s'", sts_name)

    if not wait:
        return True

    target_generation = sts.metadata.generation
    self.logger.debug("Rollout triggered. Target generation -> %s", target_generation)

    # ------------------------------------------------------------------
    # Rollout monitoring
    # ------------------------------------------------------------------

    start_time = time.time()
    last_logged_revision = None
    wait_time = 3

    while True:
        # Avoid tight loop
        time.sleep(wait_time)

        elapsed = time.time() - start_time
        if elapsed > wait_timeout:
            self.logger.error(
                "Timeout waiting for rolling restart of stateful set -> '%s'",
                sts_name,
            )
            return False

        sts = self.get_stateful_set(sts_name)
        if not sts:
            self.logger.warning("Stateful set -> '%s' not found during rollout monitoring. Retrying...", sts_name)
            continue

        # The ONLY reliable way to know if the loop is seeing the NEW rollout
        if (sts.status.observed_generation or 0) < target_generation:
            self.logger.debug(
                "Kubernetes controller hasn't processed the new generation -> %s yet...", target_generation
            )
            continue

        status = sts.status

        desired = sts.spec.replicas or 0
        ready = status.ready_replicas or 0
        updated = status.updated_replicas or 0
        available = status.available_replicas or 0
        current_revision = status.current_revision
        update_revision = status.update_revision

        # Log revision changes
        if update_revision != last_logged_revision:
            self.logger.debug(
                "Stateful set -> '%s' rollout revision: current=%s update=%s",
                sts_name,
                current_revision,
                update_revision,
            )
            last_logged_revision = update_revision

        # ------------------------------------------------------------------
        # Rollout completion condition
        # ------------------------------------------------------------------

        self.logger.info(
            "Rolling restart in progress for stateful set -> '%s'... (ready=%s, updated=%s, available=%s, desired=%s)",
            sts_name,
            ready,
            updated,
            available,
            desired,
        )

        if ready == desired and updated == desired and available == desired and current_revision == update_revision:
            self.logger.info(
                "Stateful set -> '%s' rollout completed successfully (%s/%s pods ready)",
                sts_name,
                ready,
                desired,
            )
            break  # the while loop will exit here if the rollout is successful

        wait_time = 30

    # end while rollout monitoring

    # ------------------------------------------------------------------
    # Endpoint gating (optional)
    # ------------------------------------------------------------------

    if endpoint_service_name:
        self.logger.info(
            "Validating registration of stateful set -> '%s' at service -> '%s'",
            sts_name,
            endpoint_service_name,
        )

        # Extract the match_labels from the STS spec
        match_labels = sts.spec.selector.match_labels

        # Convert the dictionary into a comma-separated string (e.g., "app=my-app,tier=frontend")
        selector = ",".join([f"{k}={v}" for k, v in match_labels.items()])

        # Use the dynamic selector to list the pods
        pod_list = self.list_pods(label_selector=selector)

        pods = pod_list.items

        pod_ips = [p.status.pod_ip for p in pods if p.status.pod_ip]

        start_time = time.time()
        while True:
            if time.time() - start_time > 300:
                self.logger.error(
                    "Timeout waiting for endpoint registration for stateful set -> '%s'.",
                    sts_name,
                )
                return False

            endpoints = self.get_core_v1_api().read_namespaced_endpoints(
                name=endpoint_service_name,
                namespace=self.get_namespace(),
            )

            registered_ips = [addr.ip for subset in (endpoints.subsets or []) for addr in (subset.addresses or [])]

            if all(ip in registered_ips for ip in pod_ips):
                self.logger.info(
                    "All pods for stateful set -> '%s' registered in service -> '%s' endpoint.",
                    sts_name,
                    endpoint_service_name,
                )
                break

            self.logger.info(
                "Waiting for service endpoint registration for stateful set -> '%s'...",
                sts_name,
            )
            time.sleep(5)
        # end while endpoint registration
    # end if endpoint_service_name

    self.logger.info("Rolling restart completed for stateful set ->'%s'", sts_name)

    return True