Kathara.manager.kubernetes.KubernetesMachine

module Kathara.manager.kubernetes.KubernetesMachine

Global Variables

• RP_FILTER_NAMESPACE
• MAX_RESTART_COUNT
• MAX_TIME_ERROR
• STARTUP_COMMANDS
• SHUTDOWN_COMMANDS

---

class KubernetesMachine

Class responsible for managing Kathara devices representation in Kubernetes.

method KubernetesMachine.__init__

__init__(
    kubernetes_namespace: Kathara.manager.kubernetes.KubernetesNamespace.KubernetesNamespace
) → None

---

method KubernetesMachine.connect

connect(
    lab_hash: str,
    machine_name: str,
    shell: Union[str, List[str]] = None,
    logs: bool = False
) → None

Open a stream to the Kubernetes Pod specified by machine_name using the specified shell.

Args:

• lab_hash (str): The hash of the network scenario containing the device.
• machine_name (str): The name of the device to connect.
• shell (Union[str, List[str]]): The path to the desired shell.
• logs (bool): If True, print the logs of the startup command.

Returns: None

Raises:

• MachineNotRunningError: If the specified device is not running.
• MachineNotReadyError: If the device is not ready.

---

method KubernetesMachine.copy_files

copy_files(
    machine_api_object: kubernetes.client.models.v1_deployment.V1Deployment,
    path: str,
    tar_data: bytes
) → None

Copy the files contained in tar_data in the Kubernetes deployment path specified by the machine_api_object.

Args:

• machine_api_object (client.V1Deployment): A Kubernetes deployment.
• path (str): The path of where copy the tar_data.
• tar_data (bytes): The data to copy in the deployment.

Returns: None

---

method KubernetesMachine.create

create(machine: Kathara.model.Machine.Machine) → None

Create a Kubernetes deployment and a Pod representing the device and assign it to machine.api_object.

Args:

• machine (Kathara.model.Machine.Machine): a Kathara device.

Returns: None

Raises:

• MachineAlreadyExistsError: If a device with the name specified already exists.

---

method KubernetesMachine.deploy_machines

deploy_machines(
    lab: Kathara.model.Lab.Lab,
    selected_machines: Set[str] = None,
    excluded_machines: Set[str] = None
) → None

Deploy all the devices contained in lab.machines.

Args:

• lab (Kathara.model.Lab.Lab): A Kathara network scenario.
• selected_machines (Set[str]): A set containing the name of the devices to deploy.
• excluded_machines (Set[str]): A set containing the name of the devices to exclude.

Returns: None

Raises:

• InvocationError: If both selected_machines and excluded_machines are specified.

---

method KubernetesMachine.exec

exec(
    lab_hash: str,
    machine_name: str,
    command: Union[str, List],
    tty: bool = False,
    stdin: bool = False,
    stdin_buffer: List[Union[str, bytes]] = None,
    stderr: bool = False,
    is_stream: bool = True
) → Union[Generator[Tuple[bytes, bytes], NoneType, NoneType], Tuple[bytes, bytes, int]]

Execute the command on the Kubernetes Pod specified by the lab_hash and the machine_name.

Args:

• lab_hash (str): The hash of the network scenario containing the device.
• machine_name (str): The name of the device.
• command (Union[str, List]): The command to execute.
• tty (bool): If True, open a new tty.
• stdin (bool): If True, open the stdin channel.
• stdin_buffer (List[Union[str, bytes]]): List of command to pass to the stdin.
• stderr (bool): If True, return the stderr.
• is_stream (bool): If True, return a generator object containing the stdout and the stderr of the command. If False, returns a tuple containing the complete stdout, the stderr, and the return code of the command.

Returns:

• Union[Generator[Tuple[bytes, bytes]], Tuple[bytes, bytes, int]]: A generator of tuples containing the stdout and stderr in bytes or a tuple containing the stdout, the stderr and the return code of the command.

Raises:

• MachineNotRunningError: If the specified device is not running.
• MachineBinaryError: If the command specified is not found on the device.

---

method KubernetesMachine.get_deployment_name

get_deployment_name(name: str) → str

Return the name of the Kubernetes deployment corresponding to 'name'.

Args:

• name (str): The name of a Kathara device.

Returns:

• str: The name for the Kubernetes deployment.

---

method KubernetesMachine.get_env_var_value_from_pod

get_env_var_value_from_pod(
    pod: kubernetes.client.models.v1_pod.V1Pod,
    var_name: str
) → Optional[str]

Return the value of an environment variable of the Kubernetes Pod.

Args:

• pod (client.V1Pod): A Kubernetes Pod.
• var_name (str): The name of the environment variable.

Returns:

• Optional[str]: The value of the environment variable.

---

method KubernetesMachine.get_machines_api_objects_by_filters

get_machines_api_objects_by_filters(
    lab_hash: str = None,
    machine_name: str = None
) → List[kubernetes.client.models.v1_pod.V1Pod]

Return the List of Kubernetes Pods.

Args:

• lab_hash (str): The hash of a network scenario. If specified, return all the Kubernetes Pod in the scenario.
• machine_name (str): The name of a device. If specified, return the specified Kubernetes Pod of the scenario.

Returns:

• List[client.V1Pod]: A list of Kubernetes Pods objects.

---

method KubernetesMachine.get_machines_stats

get_machines_stats(
    lab_hash: str = None,
    machine_name: str = None
) → Generator[Dict[str, Kathara.manager.kubernetes.stats.KubernetesMachineStats.KubernetesMachineStats], NoneType, NoneType]

Return a generator containing the Kubernetes devices' stats.

Args:

• lab_hash (str): The hash of a network scenario. If specified, return all the stats of the devices in the scenario.
• machine_name (str): The name of a device. If specified, return the specified device stats.

Returns:

• Generator[Dict[str, KubernetesMachineStats], None, None]: A generator containing device name as keys and KubernetesMachineStats as values.

---

method KubernetesMachine.undeploy

undeploy(
    lab_hash: str,
    selected_machines: Set[str] = None,
    excluded_machines: Set[str] = None
) → None

Undeploy all the running Kubernetes deployments and Pods contained in the scenario defined by the lab_hash.

If selected_machines is not None, undeploy only the specified devices.

Args:

• lab_hash (str): The hash of the network scenario to undeploy.
• selected_machines (Set[str]): A set containing the name of the devices to undeploy.
• excluded_machines (Set[str]): A set containing the name of the devices to exclude.

Returns: None

Raises:

• InvocationError: If both selected_machines and excluded_machines are specified.

---

method KubernetesMachine.wipe

wipe() → None

Undeploy all the running Kubernetes deployments and Pods.

Returns: None

---

This file was automatically generated via lazydocs.