Kathara.manager.kubernetes.KubernetesManager

`module` `Kathara.manager.kubernetes.KubernetesManager`

`class` `KubernetesManager`

Class responsible for interacting with Kubernetes API.

`method` `KubernetesManager.init`

__init__() → None

`method` `KubernetesManager.check_image`

check_image(image_name: str) → None

Useless. The Check of the image is delegated to Kubernetes.

Args:

image_name (str): The name of the image

Returns: None

`method` `KubernetesManager.connect_machine_to_link`

connect_machine_to_link(
    machine: Kathara.model.Machine.Machine,
    link: Kathara.model.Link.Link,
    mac_address: Optional[str] = None
) → None

Connect a Kathara device to a collision domain.

Args:

machine (Kathara.model.Machine): A Kathara machine object.
link (Kathara.model.Link): A Kathara collision domain object.
mac_address (Optional[str]): The MAC address to assign to the interface.

Returns: None

Raises:

NotSupportedError: Unable to update a running device on Kubernetes.

`method` `KubernetesManager.connect_tty`

connect_tty(
    machine_name: str,
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    shell: str = None,
    logs: bool = False,
    wait: Union[bool, Tuple[int, float]] = True
) → None

Connect to a device in a running network scenario, using the specified shell.

Args:

machine_name (str): The name of the device to connect.
lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab. If None, lab_name or lab should be set.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab. If None, lab_hash or lab should be set.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name. If None, lab_hash or lab_name should be set.
shell (str): The name of the shell to use for connecting.
logs (bool): If True, print startup logs on stdout.
wait (Union[bool, Tuple[int, float]]): If True, wait indefinitely until the end of the startup commands execution before connecting. If a tuple is provided, the first value indicates the number of retries before stopping waiting and the second value indicates the time interval to wait for each retry. Default is True.

Returns: None

Raises:

InvocationError: If a running network scenario hash or name is not specified.

`method` `KubernetesManager.connect_tty_obj`

connect_tty_obj(
    machine: Kathara.model.Machine.Machine,
    shell: str = None,
    logs: bool = False,
    wait: Union[bool, Tuple[int, float]] = True
) → None

Connect to a device in a running network scenario, using the specified shell.

Args:

machine (Machine): The device to connect.
shell (str): The name of the shell to use for connecting.
logs (bool): If True, print startup logs on stdout.
wait (Union[bool, Tuple[int, float]]): If True, wait indefinitely until the end of the startup commands execution before connecting. If a tuple is provided, the first value indicates the number of retries before stopping waiting and the second value indicates the time interval to wait for each retry. Default is True.

Returns: None

Raises:

LabNotFoundError: If the specified device is not associated to any network scenario.

`method` `KubernetesManager.copy_files`

copy_files(
    machine: Kathara.model.Machine.Machine,
    guest_to_host: Dict[str, Union[str, io.IOBase]]
) → None

Copy files on a running device in the specified paths.

Args:

machine (Kathara.model.Machine): A running device object. It must have the api_object field populated.
guest_to_host (Dict[str, Union[str, io.IOBase]]): A dict containing the device path as key and a fileobj to copy in path as value or a path to a file.

Returns: None

`method` `KubernetesManager.deploy_lab`

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

Deploy a Kathara network scenario.

Args:

lab (Kathara.model.Lab): A Kathara network scenario.
selected_machines (Optional[Set[str]]): If not None, deploy only the specified devices.
excluded_machines (Optional[Set[str]]): If not None, exclude devices from being deployed.

Returns: None

Raises:

NonSequentialMachineInterfaceError: If there is a missing interface number in any device of the lab.
MachineNotFoundError: If the specified devices are not in the network scenario specified.
LabAlreadyExistsError: If a network scenario is deployed while it is terminating its execution.
ApiError: If the Kubernetes APIs throw an exception.
InvocationError: If both selected_machines and excluded_machines are specified.

`method` `KubernetesManager.deploy_link`

deploy_link(link: Kathara.model.Link.Link) → None

Deploy a Kathara collision domain.

Args:

link (Kathara.model.Link): A Kathara collision domain object.

Returns: None

Raises:

LabNotFoundError: If the collision domain specified is not associated to any network scenario.

`method` `KubernetesManager.deploy_machine`

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

Deploy a Kathara device.

Args:

machine (Kathara.model.Machine): A Kathara machine object.

Returns: None

Raises:

LabNotFoundError: If the specified device is not associated to any network scenario.
NonSequentialMachineInterfaceError: If there is a missing interface number in any device of the lab.

`method` `KubernetesManager.disconnect_machine_from_link`

disconnect_machine_from_link(
    machine: Kathara.model.Machine.Machine,
    link: Kathara.model.Link.Link
) → None

Disconnect a Kathara device from a collision domain.

Args:

machine (Kathara.model.Machine): A Kathara machine object.
link (Kathara.model.Link): The Kathara collision domain from which disconnect the device.

Returns: None

Raises:

NotSupportedError: Unable to update a running device on Kubernetes.

`method` `KubernetesManager.exec`

exec(
    machine_name: str,
    command: Union[List[str], str],
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    wait: Union[bool, Tuple[int, float]] = False,
    stream: bool = True
) → Union[Generator[Tuple[bytes, bytes], NoneType, NoneType], Tuple[bytes, bytes, int]]

Exec a command on a device in a running network scenario.

Args:

machine_name (str): The name of the device to connect.
command (Union[List[str], str]): The command to exec on the device.
lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab. If None, lab_name or lab should be set.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab. If None, lab_hash or lab should be set.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name. If None, lab_hash or lab_name should be set.
wait (Union[bool, Tuple[int, float]]): If True, wait indefinitely until the end of the startup commands execution before executing the command. If a tuple is provided, the first value indicates the number of retries before stopping waiting and the second value indicates the time interval to wait for each retry. Default is False.
stream (bool): If True, return a generator object containing the command output. 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:

InvocationError: If a running network scenario hash or name is not specified.
MachineNotRunningError: If the specified device is not running.
MachineBinaryError: If the binary of the command is not found.

`method` `KubernetesManager.exec_obj`

exec_obj(
    machine: Kathara.model.Machine.Machine,
    command: Union[List[str], str],
    wait: Union[bool, Tuple[int, float]] = False,
    stream: bool = True
) → Union[Generator[Tuple[bytes, bytes], NoneType, NoneType], Tuple[bytes, bytes, int]]

Exec a command on a device in a running network scenario.

Args:

machine (Machine): The device to connect.
command (Union[List[str], str]): The command to exec on the device.
wait (Union[bool, Tuple[int, float]]): If True, wait indefinitely until the end of the startup commands execution before executing the command. If a tuple is provided, the first value indicates the number of retries before stopping waiting and the second value indicates the time interval to wait for each retry. Default is False.
stream (bool): If True, return a generator object containing the command output. 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:

LabNotFoundError: If the specified device is not associated to any network scenario.
MachineNotRunningError: If the specified device is not running.
MachineBinaryError: If the binary of the command is not found.

`method` `KubernetesManager.get_formatted_manager_name`

get_formatted_manager_name() → str

Return a formatted string containing the current manager name.

Returns:

str: A formatted string containing the current manager name.

`method` `KubernetesManager.get_lab_from_api`

get_lab_from_api(lab_hash: str = None, lab_name: str = None) → Lab

Return the network scenario (specified by the hash or name), building it from API objects.

Args:

lab_hash (str): The hash of the network scenario. Can be used as an alternative to lab_name.
lab_name (str): The name of the network scenario. Can be used as an alternative to lab_name.

Returns:

Lab: The built network scenario.

Raises:

InvocationError: If a running network scenario hash or name is not specified.

`method` `KubernetesManager.get_link_api_object`

get_link_api_object(
    link_name: str,
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    all_users: bool = False
) → Any

Return the corresponding API object of a collision domain in a network scenario.

Args:

link_name (str): The name of the collision domain.
lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab. If None, lab_name or lab should be set.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab. If None, lab_hash or lab should be set.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name. If None, lab_hash or lab_name should be set.
all_users (bool): If True, return information about collision domains of all users.

Returns:

Any: Kubernetes API object of the network.

Raises:

InvocationError: If a running network scenario hash or name is not specified.
LinkNotFoundError: If the collision domain is not found.

`method` `KubernetesManager.get_link_stats`

get_link_stats(
    link_name: str,
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    all_users: bool = False
) → Generator[Optional[Kathara.manager.kubernetes.stats.KubernetesLinkStats.KubernetesLinkStats], NoneType, NoneType]

Return information of the specified deployed network in a specified network scenario.

Args:

link_name (str): The name of the collision domain for which statistics are requested.
lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab. If None, lab_name or lab should be set.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab. If None, lab_hash or lab should be set.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name. If None, lab_hash or lab_name should be set.
all_users (bool): If True, return information about the networks of all users.

Returns:

Generator[Optional[KubernetesLinkStats], None, None]: A generator containing the KubernetesLinkStats object with the network info. Returns None if the network is not found.

Raises:

InvocationError: If a running network scenario hash or name is not specified.

`method` `KubernetesManager.get_link_stats_obj`

get_link_stats_obj(
    link: Kathara.model.Link.Link,
    all_users: bool = False
) → Generator[Optional[Kathara.manager.kubernetes.stats.KubernetesLinkStats.KubernetesLinkStats], NoneType, NoneType]

Return information of the specified deployed network in a specified network scenario.

Args:

link (Link): The collision domain for which statistics are requested.
all_users (bool): If True, return information about the networks of all users.

Returns:

Generator[Optional[ILinkStats], None, None]: A generator containing the ILinkStats object with the network info. Returns None if the network is not found.

Raises:

LabNotFoundError: If the specified device is not associated to any network scenario.

`method` `KubernetesManager.get_links_api_objects`

get_links_api_objects(
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    all_users: bool = False
) → List[Any]

Return API objects of collision domains in a network scenario.

Args:

lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name.
all_users (bool): If True, return information about collision domains of all users.

Returns:

List[Any]: Kubernetes API objects of networks.

`method` `KubernetesManager.get_links_stats`

get_links_stats(
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    link_name: str = None,
    all_users: bool = False
) → Generator[Dict[str, Kathara.manager.kubernetes.stats.KubernetesLinkStats.KubernetesLinkStats], NoneType, NoneType]

Return information about deployed networks.

Args:

lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name.
link_name (str): If specified return all the networks with link_name.
all_users (bool): If True, return information about the networks of all users.

Returns:

Generator[Dict[str, KubernetesLinkStats], None, None]: A generator containing dicts that has API Object identifier as keys and KubernetesLinkStats objects as values.

`method` `KubernetesManager.get_machine_api_object`

get_machine_api_object(
    machine_name: str,
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    all_users: bool = False
) → V1Pod

Return the corresponding API object of a running device in a network scenario.

Args:

machine_name (str): The name of the device.
lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab. If None, lab_name or lab should be set.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab. If None, lab_hash or lab should be set.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name. If None, lab_hash or lab_name should be set.
all_users (bool): If True, return information about devices of all users.

Returns:

client.V1Pod: A Kubernetes Pod.

Raises:

InvocationError: If a running network scenario hash or name is not specified.
MachineNotFoundError: If the device is not found.

`method` `KubernetesManager.get_machine_stats`

get_machine_stats(
    machine_name: str,
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    all_users: bool = False
) → Generator[Optional[Kathara.manager.kubernetes.stats.KubernetesMachineStats.KubernetesMachineStats], NoneType, NoneType]

Return information of the specified device in a specified network scenario.

Args:

machine_name (str): The name of the device for which statistics are requested.
lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab. If None, lab_name or lab should be set.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab. If None, lab_hash or lab should be set.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name. If None, lab_hash or lab_name should be set.
all_users (bool): If True, search the device among all the users devices.

Returns:

Generator[Optional[KubernetesMachineStats], None, None]: A generator containing the KubernetesMachineStats object with the device info. Returns None if the device is not found.

Raises:

InvocationError: If a running network scenario hash or name is not specified.

`method` `KubernetesManager.get_machine_stats_obj`

get_machine_stats_obj(
    machine: Kathara.model.Machine.Machine,
    all_users: bool = False
) → Generator[Optional[Kathara.manager.kubernetes.stats.KubernetesMachineStats.KubernetesMachineStats], NoneType, NoneType]

Return information of the specified device in a specified network scenario.

Args:

machine (Machine): The device for which statistics are requested.
all_users (bool): If True, search the device among all the users devices.

Returns:

Generator[Optional[IMachineStats], None, None]: A generator containing the IMachineStats object with the device info. Returns None if the device is not found.

Raises:

LabNotFoundError: If the specified device is not associated to any network scenario.
MachineNotRunningError: If the specified device is not running.

`method` `KubernetesManager.get_machines_api_objects`

get_machines_api_objects(
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    all_users: bool = False
) → List[kubernetes.client.models.v1_pod.V1Pod]

Return API objects of running devices.

Args:

lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name.
all_users (bool): If True, return information about devices of all users.

Returns:

List[client.V1Pod]: Kubernetes Pod objects of devices.

`method` `KubernetesManager.get_machines_stats`

get_machines_stats(
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    machine_name: str = None,
    all_users: bool = False
) → Generator[Dict[str, Kathara.manager.kubernetes.stats.KubernetesMachineStats.KubernetesMachineStats], NoneType, NoneType]

Return information about the running devices.

Args:

lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name.
machine_name (str): If specified return all the devices with machine_name.
all_users (bool): If True, return information about the device of all users.

Returns:

Generator[Dict[str, KubernetesMachineStats], None, None]: A generator containing dicts that has API Object identifier as keys and KubernetesMachineStats objects as values.

`method` `KubernetesManager.get_release_version`

get_release_version() → str

Return the current manager version.

Returns:

str: The current manager version.

`method` `KubernetesManager.undeploy_lab`

undeploy_lab(
    lab_hash: Optional[str] = None,
    lab_name: Optional[str] = None,
    lab: Optional[Kathara.model.Lab.Lab] = None,
    selected_machines: Optional[Set[str]] = None,
    excluded_machines: Optional[Set[str]] = None
) → None

Undeploy a Kathara network scenario.

Args:

lab_hash (Optional[str]): The hash of the network scenario. Can be used as an alternative to lab_name and lab. If None, lab_name or lab should be set.
lab_name (Optional[str]): The name of the network scenario. Can be used as an alternative to lab_hash and lab. If None, lab_hash or lab should be set.
lab (Optional[Kathara.model.Lab]): The network scenario object. Can be used as an alternative to lab_hash and lab_name. If None, lab_hash or lab_name should be set.
selected_machines (Optional[Set[str]]): If not None, undeploy only the specified devices.
excluded_machines (Optional[Set[str]]): If not None, exclude devices from being undeployed.

Returns: None

Raises:

InvocationError: If a running network scenario hash or name is not specified, or if both selected_machines and excluded_machines are specified.

`method` `KubernetesManager.undeploy_link`

undeploy_link(link: Kathara.model.Link.Link) → None

Undeploy a Kathara collision domain.

Args:

link (Kathara.model.Link): A Kathara collision domain object.

Returns: None

Raises:

LabNotFoundError: If the collision domain specified is not associated to any network scenario.

`method` `KubernetesManager.undeploy_machine`

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

Undeploy a Kathara device.

Args:

machine (Kathara.model.Machine): A Kathara machine object.

Returns: None

Raises:

LabNotFoundError: If the specified machine is not associated to a network scenario.

`method` `KubernetesManager.update_lab_from_api`

update_lab_from_api(lab: Kathara.model.Lab.Lab) → None

Update the passed network scenario from API objects.

Args:

lab (Lab): The network scenario to update.

Raises:

NotSupportedError: Unable to update a running network scenario on Kubernetes.

`method` `KubernetesManager.wipe`

wipe(all_users: bool = False) → None

Undeploy all the running network scenarios.

Args:

all_users (bool): If false, undeploy only the current user network scenarios. If true, undeploy the running network scenarios of all users.

Returns: None

This file was automatically generated via lazydocs.

Website: kathara.org

Contact us: contact@kathara.org

Wiki Home

Manual

Official Labs

Old Netkit Labs

Docker Images

Kathará Python API

Installation Guide

Linux
Windows
Mac
Megalos (Kathará over Kubernetes)

Kathara.manager.kubernetes.KubernetesManager

module Kathara.manager.kubernetes.KubernetesManager

class KubernetesManager

method KubernetesManager.__init__

method KubernetesManager.check_image

method KubernetesManager.connect_machine_to_link

method KubernetesManager.connect_tty

method KubernetesManager.connect_tty_obj

method KubernetesManager.copy_files

method KubernetesManager.deploy_lab

method KubernetesManager.deploy_link

method KubernetesManager.deploy_machine

method KubernetesManager.disconnect_machine_from_link

method KubernetesManager.exec

method KubernetesManager.exec_obj

method KubernetesManager.get_formatted_manager_name

method KubernetesManager.get_lab_from_api

method KubernetesManager.get_link_api_object

method KubernetesManager.get_link_stats

method KubernetesManager.get_link_stats_obj

method KubernetesManager.get_links_api_objects

method KubernetesManager.get_links_stats

method KubernetesManager.get_machine_api_object

method KubernetesManager.get_machine_stats

method KubernetesManager.get_machine_stats_obj

method KubernetesManager.get_machines_api_objects

method KubernetesManager.get_machines_stats

method KubernetesManager.get_release_version

method KubernetesManager.undeploy_lab

method KubernetesManager.undeploy_link

method KubernetesManager.undeploy_machine

method KubernetesManager.update_lab_from_api

method KubernetesManager.wipe

Clone this wiki locally

`module` `Kathara.manager.kubernetes.KubernetesManager`

`class` `KubernetesManager`

`method` `KubernetesManager.init`

`method` `KubernetesManager.check_image`

`method` `KubernetesManager.connect_machine_to_link`

`method` `KubernetesManager.connect_tty`

`method` `KubernetesManager.connect_tty_obj`

`method` `KubernetesManager.copy_files`

`method` `KubernetesManager.deploy_lab`

`method` `KubernetesManager.deploy_link`

`method` `KubernetesManager.deploy_machine`

`method` `KubernetesManager.disconnect_machine_from_link`

`method` `KubernetesManager.exec`

`method` `KubernetesManager.exec_obj`

`method` `KubernetesManager.get_formatted_manager_name`

`method` `KubernetesManager.get_lab_from_api`

`method` `KubernetesManager.get_link_api_object`

`method` `KubernetesManager.get_link_stats`

`method` `KubernetesManager.get_link_stats_obj`

`method` `KubernetesManager.get_links_api_objects`

`method` `KubernetesManager.get_links_stats`

`method` `KubernetesManager.get_machine_api_object`

`method` `KubernetesManager.get_machine_stats`

`method` `KubernetesManager.get_machine_stats_obj`

`method` `KubernetesManager.get_machines_api_objects`

`method` `KubernetesManager.get_machines_stats`

`method` `KubernetesManager.get_release_version`

`method` `KubernetesManager.undeploy_lab`

`method` `KubernetesManager.undeploy_link`

`method` `KubernetesManager.undeploy_machine`

`method` `KubernetesManager.update_lab_from_api`

`method` `KubernetesManager.wipe`