glances.plugins.containers.glances_podman.PodmanPodStatsFetcher._get_network_stats() - Code Metrics - Inspection of "Merge branch 'master' into develop" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 05aaee...10b5c2 )

by Nicolas

created 2023-05-16 07:38 UTC

PodmanPodStatsFetcher._get_network_stats() A

↳ Parent: glances.plugins.containers.glances_podman

Complexity

Conditions

Size

Total Lines	25
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
eloc	13
nop	2
dl	0
loc	25
rs	9.75
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only

"""Podman Extension unit for Glances' Containers plugin."""
from datetime import datetime

from glances.compat import iterkeys, itervalues, nativestr, pretty_date, string_value_to_float
from glances.logger import logger
from glances.plugins.containers.stats_streamer import StatsStreamer

# Podman library (optional and Linux-only)
# https://pypi.org/project/podman/
try:
    from podman import PodmanClient
except Exception as e:
    import_podman_error_tag = True
    # Display debug message if import KeyError
    logger.debug("Error loading Podman deps Lib. Podman feature in the Containers plugin is disabled ({})".format(e))
else:
    import_podman_error_tag = False


class PodmanContainerStatsFetcher:
    MANDATORY_FIELDS = ["CPU", "MemUsage", "MemLimit", "NetInput", "NetOutput", "BlockInput", "BlockOutput"]

    def __init__(self, container):
        self._container = container

        # Threaded Streamer
        stats_iterable = container.stats(decode=True)
        self._streamer = StatsStreamer(stats_iterable, initial_stream_value={})

    def _log_debug(self, msg, exception=None):
        logger.debug("containers (Podman) ID: {} - {} ({})".format(self._container.id, msg, exception))
        logger.debug(self._streamer.stats)

    def stop(self):
        self._streamer.stop()

    @property
    def stats(self):
        stats = self._streamer.stats
        if stats["Error"]:
            self._log_debug("Stats fetching failed", stats["Error"])

        return stats["Stats"][0]

    @property
    def activity_stats(self):
        result_stats = {"cpu": {}, "memory": {}, "io": {}, "network": {}}
        api_stats = self.stats

        if any(field not in api_stats for field in self.MANDATORY_FIELDS):
            self._log_debug("Missing mandatory fields")
            return result_stats

        try:
            cpu_usage = float(api_stats.get("CPU", 0))

            mem_usage = float(api_stats["MemUsage"])
            mem_limit = float(api_stats["MemLimit"])

            rx = float(api_stats["NetInput"])
            tx = float(api_stats["NetOutput"])

            ior = float(api_stats["BlockInput"])
            iow = float(api_stats["BlockOutput"])

            # Hardcode `time_since_update` to 1 as podman already sends the calculated rate
            result_stats = {
                "cpu": {"total": cpu_usage},
                "memory": {"usage": mem_usage, "limit": mem_limit},
                "io": {"ior": ior, "iow": iow, "time_since_update": 1},
                "network": {"rx": rx, "tx": tx, "time_since_update": 1},
            }
        except ValueError as e:
            self._log_debug("Non float stats values found", e)

        return result_stats


class PodmanPodStatsFetcher:
    def __init__(self, pod_manager):
        self._pod_manager = pod_manager

        # Threaded Streamer
        # Temporary patch to get podman extension working
        stats_iterable = (pod_manager.stats(decode=True) for _ in iter(int, 1))
        self._streamer = StatsStreamer(stats_iterable, initial_stream_value={}, sleep_duration=2)

    def _log_debug(self, msg, exception=None):
        logger.debug("containers (Podman): Pod Manager - {} ({})".format(msg, exception))
        logger.debug(self._streamer.stats)

    def stop(self):
        self._streamer.stop()

    @property
    def activity_stats(self):
        result_stats = {}
        container_stats = self._streamer.stats
        for stat in container_stats:
            io_stats = self._get_io_stats(stat)
            cpu_stats = self._get_cpu_stats(stat)
            memory_stats = self._get_memory_stats(stat)
            network_stats = self._get_network_stats(stat)

            computed_stats = {
                "name": stat["Name"],
                "cid": stat["CID"],
                "pod_id": stat["Pod"],
                "io": io_stats or {},
                "memory": memory_stats or {},
                "network": network_stats or {},
                "cpu": cpu_stats or {"total": 0.0},
            }
            result_stats[stat["CID"]] = computed_stats

        return result_stats

    def _get_cpu_stats(self, stats):
        """Return the container CPU usage.

        Output: a dict {'total': 1.49}
        """
        if "CPU" not in stats:
            self._log_debug("Missing CPU usage fields")
            return None

        cpu_usage = string_value_to_float(stats["CPU"].rstrip("%"))
        return {"total": cpu_usage}

    def _get_memory_stats(self, stats):
        """Return the container MEMORY.

        Output: a dict {'rss': 1015808, 'cache': 356352,  'usage': ..., 'max_usage': ...}
        """
        if "MemUsage" not in stats or "/" not in stats["MemUsage"]:
            self._log_debug("Missing MEM usage fields")
            return None

        memory_usage_str = stats["MemUsage"]
        usage_str, limit_str = memory_usage_str.split("/")

        try:
            usage = string_value_to_float(usage_str)
            limit = string_value_to_float(limit_str)
        except ValueError as e:
            self._log_debug("Compute MEM usage failed", e)
            return None

        return {"usage": usage, "limit": limit}

    def _get_network_stats(self, stats):
        """Return the container network usage using the Docker API (v1.0 or higher).

        Output: a dict {'time_since_update': 3000, 'rx': 10, 'tx': 65}.
        with:
            time_since_update: number of seconds elapsed between the latest grab
            rx: Number of bytes received
            tx: Number of bytes transmitted
        """
        if "NetIO" not in stats or "/" not in stats["NetIO"]:
            self._log_debug("Compute MEM usage failed")
            return None

        net_io_str = stats["NetIO"]
        rx_str, tx_str = net_io_str.split("/")

        try:
            rx = string_value_to_float(rx_str)
            tx = string_value_to_float(tx_str)
        except ValueError as e:
            self._log_debug("Compute MEM usage failed", e)
            return None

        # Hardcode `time_since_update` to 1 as podman docs don't specify the rate calculated procedure
        return {"rx": rx, "tx": tx, "time_since_update": 1}

    def _get_io_stats(self, stats):
        """Return the container IO usage using the Docker API (v1.0 or higher).

        Output: a dict {'time_since_update': 3000, 'ior': 10, 'iow': 65}.
        with:
            time_since_update: number of seconds elapsed between the latest grab
            ior: Number of bytes read
            iow: Number of bytes written
        """
        if "BlockIO" not in stats or "/" not in stats["BlockIO"]:
            self._log_debug("Missing BlockIO usage fields")
            return None

        block_io_str = stats["BlockIO"]
        ior_str, iow_str = block_io_str.split("/")

        try:
            ior = string_value_to_float(ior_str)
            iow = string_value_to_float(iow_str)
        except ValueError as e:
            self._log_debug("Compute BlockIO usage failed", e)
            return None

        # Hardcode `time_since_update` to 1 as podman docs don't specify the rate calculated procedure
        return {"ior": ior, "iow": iow, "time_since_update": 1}


class PodmanContainersExtension:
    """Glances' Containers Plugin's Docker Extension unit"""

    CONTAINER_ACTIVE_STATUS = ['running', 'paused']

    def __init__(self, podman_sock):
        if import_podman_error_tag:
            raise Exception("Missing libs required to run Podman Extension (Containers)")

        self.client = None
        self.ext_name = "containers (Podman)"
        self.podman_sock = podman_sock
        self.pods_stats_fetcher = None
        self.container_stats_fetchers = {}

        self.connect()

    def connect(self):
        """Connect to Podman."""
        try:
            self.client = PodmanClient(base_url=self.podman_sock)
            # PodmanClient works lazily, so make a ping to determine if socket is open
            self.client.ping()
        except Exception as e:
            logger.error("{} plugin - Can't connect to Podman ({})".format(self.ext_name, e))
            self.client = None

    def update_version(self):
        # Long and not useful anymore because the information is no more displayed in UIs
        # return self.client.version()
        return {}

    def stop(self):
        # Stop all streaming threads
        for t in itervalues(self.container_stats_fetchers):
            t.stop()

        if self.pods_stats_fetcher:
            self.pods_stats_fetcher.stop()

    def update(self, all_tag):
        """Update Podman stats using the input method."""

        if not self.client:
            return {}, []

        version_stats = self.update_version()

        # Update current containers list
        try:
            # Issue #1152: Podman module doesn't export details about stopped containers
            # The Containers/all key of the configuration file should be set to True
            containers = self.client.containers.list(all=all_tag)
            if not self.pods_stats_fetcher:
                self.pods_stats_fetcher = PodmanPodStatsFetcher(self.client.pods)
        except Exception as e:
            logger.error("{} plugin - Can't get containers list ({})".format(self.ext_name, e))
            return version_stats, []

        # Start new thread for new container
        for container in containers:
            if container.id not in self.container_stats_fetchers:
                # StatsFetcher did not exist in the internal dict
                # Create it, add it to the internal dict
                logger.debug("{} plugin - Create thread for container {}".format(self.ext_name, container.id[:12]))
                self.container_stats_fetchers[container.id] = PodmanContainerStatsFetcher(container)

        # Stop threads for non-existing containers
        absent_containers = set(iterkeys(self.container_stats_fetchers)) - set(c.id for c in containers)
        for container_id in absent_containers:
            # Stop the StatsFetcher
            logger.debug("{} plugin - Stop thread for old container {}".format(self.ext_name, container_id[:12]))
            self.container_stats_fetchers[container_id].stop()
            # Delete the StatsFetcher from the dict
            del self.container_stats_fetchers[container_id]

        # Get stats for all containers
        container_stats = [self.generate_stats(container) for container in containers]

        pod_stats = self.pods_stats_fetcher.activity_stats
        for stats in container_stats:
            if stats["Id"][:12] in pod_stats:
                stats["pod_name"] = pod_stats[stats["Id"][:12]]["name"]
                stats["pod_id"] = pod_stats[stats["Id"][:12]]["pod_id"]

        return version_stats, container_stats

    @property
    def key(self):
        """Return the key of the list."""
        return 'name'

    def generate_stats(self, container):
        # Init the stats for the current container
        stats = {
            'key': self.key,
            # Export name
            'name': nativestr(container.name),
            # Container Id
            'Id': container.id,
            # Container Image
            'Image': str(container.image.tags),
            # Container Status (from attrs)
            'Status': container.attrs['State'],
            'Created': container.attrs['Created'],
            'Command': container.attrs.get('Command') or [],
        }

        if stats['Status'] in self.CONTAINER_ACTIVE_STATUS:
            started_at = datetime.fromtimestamp(container.attrs['StartedAt'])
            stats_fetcher = self.container_stats_fetchers[container.id]
            activity_stats = stats_fetcher.activity_stats
            stats.update(activity_stats)

            # Additional fields
            stats['cpu_percent'] = stats["cpu"]['total']
            stats['memory_usage'] = stats["memory"].get('usage')
            if stats['memory'].get('cache') is not None:
                stats['memory_usage'] -= stats['memory']['cache']
            stats['io_r'] = stats['io'].get('ior')
            stats['io_w'] = stats['io'].get('iow')
            stats['network_rx'] = stats['network'].get('rx')
            stats['network_tx'] = stats['network'].get('tx')
            stats['Uptime'] = pretty_date(started_at)
        else:
            stats['io'] = {}
            stats['cpu'] = {}
            stats['memory'] = {}
            stats['network'] = {}
            stats['io_r'] = None
            stats['io_w'] = None
            stats['cpu_percent'] = None
            stats['memory_percent'] = None
            stats['network_rx'] = None
            stats['network_tx'] = None
            stats['Uptime'] = None

        return stats


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
6			#
7			# SPDX-License-Identifier: LGPL-3.0-only
8
9			"""Podman Extension unit for Glances' Containers plugin."""
10			from datetime import datetime
11
12			from glances.compat import iterkeys, itervalues, nativestr, pretty_date, string_value_to_float
13			from glances.logger import logger
14			from glances.plugins.containers.stats_streamer import StatsStreamer
15
16			# Podman library (optional and Linux-only)
17			# https://pypi.org/project/podman/
18			try:
19			from podman import PodmanClient
20			except Exception as e:
21			import_podman_error_tag = True
22			# Display debug message if import KeyError
23			logger.debug("Error loading Podman deps Lib. Podman feature in the Containers plugin is disabled ({})".format(e))
24			else:
25			import_podman_error_tag = False
26
27
28			class PodmanContainerStatsFetcher:
29			MANDATORY_FIELDS = ["CPU", "MemUsage", "MemLimit", "NetInput", "NetOutput", "BlockInput", "BlockOutput"]
30
31			def __init__(self, container):
32			self._container = container
33
34			# Threaded Streamer
35			stats_iterable = container.stats(decode=True)
36			self._streamer = StatsStreamer(stats_iterable, initial_stream_value={})
37
38			def _log_debug(self, msg, exception=None):
39			logger.debug("containers (Podman) ID: {} - {} ({})".format(self._container.id, msg, exception))
40			logger.debug(self._streamer.stats)
41
42			def stop(self):
43			self._streamer.stop()
44
45			@property
46			def stats(self):
47			stats = self._streamer.stats
48			if stats["Error"]:
49			self._log_debug("Stats fetching failed", stats["Error"])
50
51			return stats["Stats"][0]
52
53			@property
54			def activity_stats(self):
55			result_stats = {"cpu": {}, "memory": {}, "io": {}, "network": {}}
56			api_stats = self.stats
57
58			if any(field not in api_stats for field in self.MANDATORY_FIELDS):
59			self._log_debug("Missing mandatory fields")
60			return result_stats
61
62			try:
63			cpu_usage = float(api_stats.get("CPU", 0))
64
65			mem_usage = float(api_stats["MemUsage"])
66			mem_limit = float(api_stats["MemLimit"])
67
68			rx = float(api_stats["NetInput"])
69			tx = float(api_stats["NetOutput"])
70
71			ior = float(api_stats["BlockInput"])
72			iow = float(api_stats["BlockOutput"])
73
74			# Hardcode `time_since_update` to 1 as podman already sends the calculated rate
75			result_stats = {
76			"cpu": {"total": cpu_usage},
77			"memory": {"usage": mem_usage, "limit": mem_limit},
78			"io": {"ior": ior, "iow": iow, "time_since_update": 1},
79			"network": {"rx": rx, "tx": tx, "time_since_update": 1},
80			}
81			except ValueError as e:
82			self._log_debug("Non float stats values found", e)
83
84			return result_stats
85
86
87			class PodmanPodStatsFetcher:
88			def __init__(self, pod_manager):
89			self._pod_manager = pod_manager
90
91			# Threaded Streamer
92			# Temporary patch to get podman extension working
93			stats_iterable = (pod_manager.stats(decode=True) for _ in iter(int, 1))
94			self._streamer = StatsStreamer(stats_iterable, initial_stream_value={}, sleep_duration=2)
95
96			def _log_debug(self, msg, exception=None):
97			logger.debug("containers (Podman): Pod Manager - {} ({})".format(msg, exception))
98			logger.debug(self._streamer.stats)
99
100			def stop(self):
101			self._streamer.stop()
102
103			@property
104			def activity_stats(self):
105			result_stats = {}
106			container_stats = self._streamer.stats
107			for stat in container_stats:
108			io_stats = self._get_io_stats(stat)
109			cpu_stats = self._get_cpu_stats(stat)
110			memory_stats = self._get_memory_stats(stat)
111			network_stats = self._get_network_stats(stat)
112
113			computed_stats = {
114			"name": stat["Name"],
115			"cid": stat["CID"],
116			"pod_id": stat["Pod"],
117			"io": io_stats or {},
118			"memory": memory_stats or {},
119			"network": network_stats or {},
120			"cpu": cpu_stats or {"total": 0.0},
121			}
122			result_stats[stat["CID"]] = computed_stats
123
124			return result_stats
125
126			def _get_cpu_stats(self, stats):
127			"""Return the container CPU usage.
128
129			Output: a dict {'total': 1.49}
130			"""
131			if "CPU" not in stats:
132			self._log_debug("Missing CPU usage fields")
133			return None
134
135			cpu_usage = string_value_to_float(stats["CPU"].rstrip("%"))
136			return {"total": cpu_usage}
137
138			def _get_memory_stats(self, stats):
139			"""Return the container MEMORY.
140
141			Output: a dict {'rss': 1015808, 'cache': 356352, 'usage': ..., 'max_usage': ...}
142			"""
143			if "MemUsage" not in stats or "/" not in stats["MemUsage"]:
144			self._log_debug("Missing MEM usage fields")
145			return None
146
147			memory_usage_str = stats["MemUsage"]
148			usage_str, limit_str = memory_usage_str.split("/")
149
150			try:
151			usage = string_value_to_float(usage_str)
152			limit = string_value_to_float(limit_str)
153			except ValueError as e:
154			self._log_debug("Compute MEM usage failed", e)
155			return None
156
157			return {"usage": usage, "limit": limit}
158
159			def _get_network_stats(self, stats):
160			"""Return the container network usage using the Docker API (v1.0 or higher).
161
162			Output: a dict {'time_since_update': 3000, 'rx': 10, 'tx': 65}.
163			with:
164			time_since_update: number of seconds elapsed between the latest grab
165			rx: Number of bytes received
166			tx: Number of bytes transmitted
167			"""
168			if "NetIO" not in stats or "/" not in stats["NetIO"]:
169			self._log_debug("Compute MEM usage failed")
170			return None
171
172			net_io_str = stats["NetIO"]
173			rx_str, tx_str = net_io_str.split("/")
174
175			try:
176			rx = string_value_to_float(rx_str)
177			tx = string_value_to_float(tx_str)
178			except ValueError as e:
179			self._log_debug("Compute MEM usage failed", e)
180			return None
181
182			# Hardcode `time_since_update` to 1 as podman docs don't specify the rate calculated procedure
183			return {"rx": rx, "tx": tx, "time_since_update": 1}
184
185			def _get_io_stats(self, stats):
186			"""Return the container IO usage using the Docker API (v1.0 or higher).
187
188			Output: a dict {'time_since_update': 3000, 'ior': 10, 'iow': 65}.
189			with:
190			time_since_update: number of seconds elapsed between the latest grab
191			ior: Number of bytes read
192			iow: Number of bytes written
193			"""
194			if "BlockIO" not in stats or "/" not in stats["BlockIO"]:
195			self._log_debug("Missing BlockIO usage fields")
196			return None
197
198			block_io_str = stats["BlockIO"]
199			ior_str, iow_str = block_io_str.split("/")
200
201			try:
202			ior = string_value_to_float(ior_str)
203			iow = string_value_to_float(iow_str)
204			except ValueError as e:
205			self._log_debug("Compute BlockIO usage failed", e)
206			return None
207
208			# Hardcode `time_since_update` to 1 as podman docs don't specify the rate calculated procedure
209			return {"ior": ior, "iow": iow, "time_since_update": 1}
210
211
212			class PodmanContainersExtension:
213			"""Glances' Containers Plugin's Docker Extension unit"""
214
215			CONTAINER_ACTIVE_STATUS = ['running', 'paused']
216
217			def __init__(self, podman_sock):
218			if import_podman_error_tag:
219			raise Exception("Missing libs required to run Podman Extension (Containers)")
220
221			self.client = None
222			self.ext_name = "containers (Podman)"
223			self.podman_sock = podman_sock
224			self.pods_stats_fetcher = None
225			self.container_stats_fetchers = {}
226
227			self.connect()
228
229			def connect(self):
230			"""Connect to Podman."""
231			try:
232			self.client = PodmanClient(base_url=self.podman_sock)
233			# PodmanClient works lazily, so make a ping to determine if socket is open
234			self.client.ping()
235			except Exception as e:
236			logger.error("{} plugin - Can't connect to Podman ({})".format(self.ext_name, e))
237			self.client = None
238
239			def update_version(self):
240			# Long and not useful anymore because the information is no more displayed in UIs
241			# return self.client.version()
242			return {}
243
244			def stop(self):
245			# Stop all streaming threads
246			for t in itervalues(self.container_stats_fetchers):
247			t.stop()
248
249			if self.pods_stats_fetcher:
250			self.pods_stats_fetcher.stop()
251
252			def update(self, all_tag):
253			"""Update Podman stats using the input method."""
254
255			if not self.client:
256			return {}, []
257
258			version_stats = self.update_version()
259
260			# Update current containers list
261			try:
262			# Issue #1152: Podman module doesn't export details about stopped containers
263			# The Containers/all key of the configuration file should be set to True
264			containers = self.client.containers.list(all=all_tag)
265			if not self.pods_stats_fetcher:
266			self.pods_stats_fetcher = PodmanPodStatsFetcher(self.client.pods)
267			except Exception as e:
268			logger.error("{} plugin - Can't get containers list ({})".format(self.ext_name, e))
269			return version_stats, []
270
271			# Start new thread for new container
272			for container in containers:
273			if container.id not in self.container_stats_fetchers:
274			# StatsFetcher did not exist in the internal dict
275			# Create it, add it to the internal dict
276			logger.debug("{} plugin - Create thread for container {}".format(self.ext_name, container.id[:12]))
277			self.container_stats_fetchers[container.id] = PodmanContainerStatsFetcher(container)
278
279			# Stop threads for non-existing containers
280			absent_containers = set(iterkeys(self.container_stats_fetchers)) - set(c.id for c in containers)
281			for container_id in absent_containers:
282			# Stop the StatsFetcher
283			logger.debug("{} plugin - Stop thread for old container {}".format(self.ext_name, container_id[:12]))
284			self.container_stats_fetchers[container_id].stop()
285			# Delete the StatsFetcher from the dict
286			del self.container_stats_fetchers[container_id]
287
288			# Get stats for all containers
289			container_stats = [self.generate_stats(container) for container in containers]
290
291			pod_stats = self.pods_stats_fetcher.activity_stats
292			for stats in container_stats:
293			if stats["Id"][:12] in pod_stats:
294			stats["pod_name"] = pod_stats[stats["Id"][:12]]["name"]
295			stats["pod_id"] = pod_stats[stats["Id"][:12]]["pod_id"]
296
297			return version_stats, container_stats
298
299			@property
300			def key(self):
301			"""Return the key of the list."""
302			return 'name'
303
304			def generate_stats(self, container):
305			# Init the stats for the current container
306			stats = {
307			'key': self.key,
308			# Export name
309			'name': nativestr(container.name),
310			# Container Id
311			'Id': container.id,
312			# Container Image
313			'Image': str(container.image.tags),
314			# Container Status (from attrs)
315			'Status': container.attrs['State'],
316			'Created': container.attrs['Created'],
317			'Command': container.attrs.get('Command') or [],
318			}
319
320			if stats['Status'] in self.CONTAINER_ACTIVE_STATUS:
321			started_at = datetime.fromtimestamp(container.attrs['StartedAt'])
322			stats_fetcher = self.container_stats_fetchers[container.id]
323			activity_stats = stats_fetcher.activity_stats
324			stats.update(activity_stats)
325
326			# Additional fields
327			stats['cpu_percent'] = stats["cpu"]['total']
328			stats['memory_usage'] = stats["memory"].get('usage')
329			if stats['memory'].get('cache') is not None:
330			stats['memory_usage'] -= stats['memory']['cache']
331			stats['io_r'] = stats['io'].get('ior')
332			stats['io_w'] = stats['io'].get('iow')
333			stats['network_rx'] = stats['network'].get('rx')
334			stats['network_tx'] = stats['network'].get('tx')
335			stats['Uptime'] = pretty_date(started_at)
336			else:
337			stats['io'] = {}
338			stats['cpu'] = {}
339			stats['memory'] = {}
340			stats['network'] = {}
341			stats['io_r'] = None
342			stats['io_w'] = None
343			stats['cpu_percent'] = None
344			stats['memory_percent'] = None
345			stats['network_rx'] = None
346			stats['network_tx'] = None
347			stats['Uptime'] = None
348
349			return stats
350

nicolargo / glances

Push — master ( 05aaee...10b5c2 )

PodmanPodStatsFetcher._get_network_stats() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like