Skip to content

panther_docker ¤

append_to_hosts_file ¤

append_to_hosts_file(entry)

summary Append a new entry to the /etc/hosts file.

Parameters:

  • entry (_type_) –

    description

Source code in panther/panther_docker.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
def append_to_hosts_file(entry):
    """_summary_
    Append a new entry to the /etc/hosts file.

    Args:
        entry (_type_): _description_
    """
    try:
        command = f"echo '{entry.strip()}' | sudo tee -a /etc/hosts"
        execute_command(command)
        logging.info(f"Added entry to /etc/hosts: {entry.strip()}")
    except subprocess.CalledProcessError as e:
        logging.error(f"Error adding entry to /etc/hosts: {e}")

container_exists ¤

container_exists(client, container_name)

summary Check if the Docker container exists.

Parameters:

  • client (_type_) –

    description

  • container_name (_type_) –

    description

Returns:

  • _type_ –

    description

Source code in panther/panther_docker.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def container_exists(client, container_name):
    """_summary_
    Check if the Docker container exists.

    Args:
        client (_type_): _description_
        container_name (_type_): _description_

    Returns:
        _type_: _description_
    """
    try:
        client.containers.get(container_name)
        return True
    except docker.errors.NotFound:
        return False
    except Exception as e:
        logging.error(f"Error checking container existence: {e}")
        return False

create_docker_network ¤

create_docker_network()

summary Create the docker network

Source code in panther/panther_docker.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def create_docker_network():
    """_summary_
    Create the docker network
    """
    network_name = "net"
    gateway = "172.27.1.1"
    subnet = "172.27.1.0/24"

    client = docker.from_env()

    if network_exists(client, network_name):
        print(f"Network '{network_name}' already exists.")
    else:
        create_network(client, network_name, gateway, subnet)

create_network ¤

create_network(client, network_name, gateway, subnet)

summary Create a Docker network with the specified gateway and subnet.

Parameters:

  • client (_type_) –

    description

  • network_name (_type_) –

    description

  • gateway (_type_) –

    description

  • subnet (_type_) –

    description

Source code in panther/panther_docker.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def create_network(client, network_name, gateway, subnet):
    """_summary_
    Create a Docker network with the specified gateway and subnet.

    Args:
        client (_type_): _description_
        network_name (_type_): _description_
        gateway (_type_): _description_
        subnet (_type_): _description_
    """
    try:
        client.networks.create(
            name=network_name,
            driver="bridge",
            ipam={"Config": [{"Subnet": subnet, "Gateway": gateway}]},
        )
        print(f"Network '{network_name}' created successfully.")
    except Exception as e:
        print(f"Error creating network: {e}")

get_container_ip ¤

get_container_ip(client, container_name)

summary Get the IP address of the Docker container.

Parameters:

  • client (_type_) –

    description

  • container_name (_type_) –

    description

Returns:

  • _type_ –

    description

Source code in panther/panther_docker.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_container_ip(client, container_name):
    """_summary_
    Get the IP address of the Docker container.

    Args:
        client (_type_): _description_
        container_name (_type_): _description_

    Returns:
        _type_: _description_
    """
    try:
        container = client.containers.get(container_name)
        ip_address = container.attrs["NetworkSettings"]["Networks"].values()
        return list(ip_address)[0]["IPAddress"]
    except Exception as e:
        logging.error(f"Error getting IP address for container '{container_name}': {e}")
        return None

get_panther_container ¤

get_panther_container()

summary

Returns:

  • _type_ –

    description

Source code in panther/panther_docker.py 
74
75
76
77
78
79
80
81
82
83
84
85
def get_panther_container():
    """_summary_

    Returns:
        _type_: _description_
    """
    client = docker.from_env()
    panther_containers = []
    for container in client.containers.list():
        if "panther" in container.name:
            panther_containers.append(container.name)
    return panther_containers

log_docker_output ¤

log_docker_output(generator, task_name: str = 'docker command execution') -> None

summary

Parameters:

  • generator (_type_) –

    description

  • task_name (str, default: 'docker command execution' ) –

    description. Defaults to "docker command execution".

Raises:

  • ValueError –

    description

Source code in panther/panther_docker.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def log_docker_output(generator, task_name: str = "docker command execution") -> None:
    """_summary_

    Args:
        generator (_type_): _description_
        task_name (str, optional): _description_. Defaults to "docker command execution".

    Raises:
        ValueError: _description_
    """

    while True:
        try:
            output = generator.__next__()
            if "stream" in output:
                output_str = output["stream"].strip("\r\n").strip("\n")
                logging.info(f"{task_name}: {output_str}")
            elif "error" in output:
                raise ValueError(f'Error from {task_name}: {output["error"]}')
        except StopIteration:
            logging.info(f"{task_name} complete.")
            break
        except ValueError:
            logging.error(f"Error parsing output from {task_name}: {output}")

monitor_docker_usage ¤

monitor_docker_usage(docker_to_monitor, interval=1.0, duration=10.0)

summary Monitor the CPU and memory usage of a Docker container.

Parameters:

  • docker_to_monitor (_type_) –

    description

  • interval (float, default: 1.0 ) –

    description. Defaults to 1.0.

  • duration (float, default: 10.0 ) –

    description. Defaults to 10.0.

Source code in panther/panther_docker.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def monitor_docker_usage(docker_to_monitor, interval=1.0, duration=10.0):
    """_summary_
    Monitor the CPU and memory usage of a Docker container.

    Args:
        docker_to_monitor (_type_): _description_
        interval (float, optional): _description_. Defaults to 1.0.
        duration (float, optional): _description_. Defaults to 10.0.
    """
    client = docker.from_env()

    for container_name in docker_to_monitor:
        try:
            container = client.containers.get(container_name)
        except docker.errors.NotFound:
            logging.info(f"No container found with name or ID {container_name}")
            return

    start_time = time.time()
    duration_condition = lambda: (
        (time.time() - start_time) < duration if duration > 0 else True
    )
    while duration_condition():
        execute_command("clear")
        for container_name in docker_to_monitor:
            try:
                container = client.containers.get(container_name)

                stats = container.stats(stream=False)

                # Check for missing keys and default to 0 if missing
                cpu_delta = stats["cpu_stats"]["cpu_usage"].get(
                    "total_usage", 0
                ) - stats["precpu_stats"]["cpu_usage"].get("total_usage", 0)
                system_cpu_delta = stats["cpu_stats"].get(
                    "system_cpu_usage", 0
                ) - stats["precpu_stats"].get("system_cpu_usage", 0)
                number_cpus = len(
                    stats["cpu_stats"]["cpu_usage"].get("percpu_usage", [])
                )
                cpu_usage = (
                    (cpu_delta / system_cpu_delta) * number_cpus * 100.0
                    if system_cpu_delta > 0
                    else 0.0
                )

                memory_usage = stats["memory_stats"].get("usage", 0) / (
                    1024 * 1024
                )  # Convert to MB
                memory_limit = stats["memory_stats"].get("limit", 1) / (
                    1024 * 1024
                )  # Convert to MB
                memory_percentage = (
                    (memory_usage / memory_limit) * 100.0 if memory_limit > 0 else 0.0
                )

                logging.info(
                    f"Name {container_name}\n\t - Time: {time.time() - start_time:.2f}s\n\t - CPU Usage: {cpu_usage:.2f}%\n\t - Memory Usage: {memory_usage:.2f}MB ({memory_percentage:.2f}%)"
                )
            except docker.errors.APIError as e:
                logging.error(f"An error occurred: {e}")
                break
            except KeyError as e:
                logging.warning(f"Missing key in stats: {e}")
        time.sleep(interval)

network_exists ¤

network_exists(client, network_name)

summary Check if the Docker network exists.

Parameters:

  • client (_type_) –

    description

  • network_name (_type_) –

    description

Returns:

  • _type_ –

    description

Source code in panther/panther_docker.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def network_exists(client, network_name):
    """_summary_
    Check if the Docker network exists.

    Args:
        client (_type_): _description_
        network_name (_type_): _description_

    Returns:
        _type_: _description_
    """
    try:
        client.networks.get(network_name)
        return True
    except NotFound:
        return False
    except Exception as e:
        print(f"Error checking network existence: {e}")
        return False

push_image_to_registry ¤

push_image_to_registry(image_name, registry_url='elniak', tag='latest')

summary Push a Docker image to a registry.

Parameters:

  • image_name (_type_) –

    description

  • registry_url (str, default: 'elniak' ) –

    description. Defaults to "elniak".

  • tag (str, default: 'latest' ) –

    description. Defaults to "latest".

Source code in panther/panther_docker.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def push_image_to_registry(image_name, registry_url="elniak", tag="latest"):
    """_summary_
    Push a Docker image to a registry.

    Args:
        image_name (_type_): _description_
        registry_url (str, optional): _description_. Defaults to "elniak".
        tag (str, optional): _description_. Defaults to "latest".
    """
    try:
        command = f"docker tag {image_name} {registry_url}/{image_name}:{tag}"
        execute_command(command)
        command = f"docker push {registry_url}/{image_name}:{tag}"
        execute_command(command)
        logging.info(f"Pushed image '{image_name}' to registry '{registry_url}'.")
    except subprocess.CalledProcessError as e:
        logging.error(f"Error pushing image to registry: {e}")

restore_hosts_file ¤

restore_hosts_file()

summary Restore the original /etc/hosts file from the backup.

Source code in panther/panther_docker.py 
105
106
107
108
109
110
111
112
113
def restore_hosts_file():
    """_summary_
    Restore the original /etc/hosts file from the backup.
    """
    try:
        execute_command("sudo cp /etc/hosts.bak /etc/hosts")
        logging.info("Restored the original /etc/hosts file.")
    except subprocess.CalledProcessError as e:
        logging.error(f"Error restoring /etc/hosts: {e}")