glances.plugins.containers.PluginModel.build_header() - Code Metrics - Inspection of "version 4.3.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( ce0fc3...e09530 )

by Nicolas

created 2024-12-27 14:16 UTC

PluginModel.build_header() D

↳ Parent: glances.plugins.containers

Complexity

Conditions

Size

Total Lines	36
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	12
eloc	27
nop	3
dl	0
loc	36
rs	4.8
c	0
b	0
f	0

How to fix Complexity

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

"""Containers plugin."""

from copy import deepcopy
from functools import partial, reduce
from typing import Any, Optional

from glances.globals import iteritems, itervalues, nativestr
from glances.logger import logger
from glances.plugins.containers.engines import ContainersExtension
from glances.plugins.containers.engines.docker import DockerExtension, disable_plugin_docker
from glances.plugins.containers.engines.podman import PodmanExtension, disable_plugin_podman
from glances.plugins.plugin.model import GlancesPluginModel
from glances.processes import glances_processes
from glances.processes import sort_stats as sort_stats_processes

# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'name': {
        'description': 'Container name',
    },
    'id': {
        'description': 'Container ID',
    },
    'image': {
        'description': 'Container image',
    },
    'status': {
        'description': 'Container status',
    },
    'created': {
        'description': 'Container creation date',
    },
    'command': {
        'description': 'Container command',
    },
    'cpu_percent': {
        'description': 'Container CPU consumption',
        'unit': 'percent',
    },
    'memory_usage': {
        'description': 'Container memory usage',
        'unit': 'byte',
    },
    'io_rx': {
        'description': 'Container IO bytes read rate',
        'unit': 'bytepersecond',
    },
    'io_wx': {
        'description': 'Container IO bytes write rate',
        'unit': 'bytepersecond',
    },
    'network_rx': {
        'description': 'Container network RX bitrate',
        'unit': 'bitpersecond',
    },
    'network_tx': {
        'description': 'Container network TX bitrate',
        'unit': 'bitpersecond',
    },
    'uptime': {
        'description': 'Container uptime',
    },
    'engine': {
        'description': 'Container engine (Docker and Podman are currently supported)',
    },
    'pod_name': {
        'description': 'Pod name (only with Podman)',
    },
    'pod_id': {
        'description': 'Pod ID (only with Podman)',
    },
}

# Define the items history list (list of items to add to history)
# TODO: For the moment limited to the CPU. Had to change the graph exports
#       method to display one graph per container.
# items_history_list = [{'name': 'cpu_percent',
#                        'description': 'Container CPU consumption in %',
#                        'y_unit': '%'},
#                       {'name': 'memory_usage',
#                        'description': 'Container memory usage in bytes',
#                        'y_unit': 'B'},
#                       {'name': 'network_rx',
#                        'description': 'Container network RX bitrate in bits per second',
#                        'y_unit': 'bps'},
#                       {'name': 'network_tx',
#                        'description': 'Container network TX bitrate in bits per second',
#                        'y_unit': 'bps'},
#                       {'name': 'io_r',
#                        'description': 'Container IO bytes read per second',
#                        'y_unit': 'Bps'},
#                       {'name': 'io_w',
#                        'description': 'Container IO bytes write per second',
#                        'y_unit': 'Bps'}]
items_history_list = [{'name': 'cpu_percent', 'description': 'Container CPU consumption in %', 'y_unit': '%'}]

# List of key to remove before export
export_exclude_list = ['cpu', 'io', 'memory', 'network']

# Sort dictionary for human
sort_for_human = {
    'io_counters': 'disk IO',
    'cpu_percent': 'CPU consumption',
    'memory_usage': 'memory consumption',
    'cpu_times': 'uptime',
    'name': 'container name',
    None: 'None',
}


class PluginModel(GlancesPluginModel):
    """Glances Docker plugin.

    stats is a dict: {'version': {...}, 'containers': [{}, {}]}
    """

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super().__init__(
            args=args, config=config, items_history_list=items_history_list, fields_description=fields_description
        )

        # The plugin can be disabled using: args.disable_docker
        self.args = args

        # Default config keys
        self.config = config

        # We want to display the stat in the curse interface
        self.display_curse = True

        self.watchers: dict[str, ContainersExtension] = {}

        # Init the Docker API
        if not disable_plugin_docker:
            self.watchers['docker'] = DockerExtension()

        # Init the Podman API
        if not disable_plugin_podman:
            self.watchers['podman'] = PodmanExtension(podman_sock=self._podman_sock())

        # Sort key
        self.sort_key = None

        # Set the key's list be disabled in order to only display specific attribute in the container list
        self.disable_stats = self.get_conf_value('disable_stats')

        # Force a first update because we need two update to have the first stat
        self.update()
        self.refresh_timer.set(0)

    def _podman_sock(self) -> str:
        """Return the podman sock.
        Could be desfined in the [docker] section thanks to the podman_sock option.
        Default value: unix:///run/user/1000/podman/podman.sock
        """
        conf_podman_sock = self.get_conf_value('podman_sock')
        if not conf_podman_sock:
            return "unix:///run/user/1000/podman/podman.sock"
        return conf_podman_sock[0]

    def exit(self) -> None:
        """Overwrite the exit method to close threads."""
        for watcher in itervalues(self.watchers):
            watcher.stop()

        # Call the father class
        super().exit()

    def get_key(self) -> str:
        """Return the key of the list."""
        return 'name'

    def get_export(self) -> list[dict]:
        """Overwrite the default export method.

        - Only exports containers
        - The key is the first container name
        """
        try:
            ret = deepcopy(self.stats)
        except KeyError as e:
            logger.debug(f"docker plugin - Docker export error {e}")
            ret = []

        # Remove fields uses to compute rate
        for container in ret:
            for i in export_exclude_list:
                container.pop(i)

        return ret

    def _all_tag(self) -> bool:
        """Return the all tag of the Glances/Docker configuration file.

        # By default, Glances only display running containers
        # Set the following key to True to display all containers
        all=True
        """
        all_tag = self.get_conf_value('all')
        if not all_tag:
            return False
        return all_tag[0].lower() == 'true'

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self) -> list[dict]:
        """Update Docker and podman stats using the input method."""
        # Connection should be ok
        if not self.watchers:
            return self.get_init_value()

        if self.input_method != 'local':
            return self.get_init_value()

        # Update stats
        stats = []
        for engine, watcher in iteritems(self.watchers):
            _, containers = watcher.update(all_tag=self._all_tag())
            containers_filtered = []
            for container in containers:
                container["engine"] = engine
                if 'key' in container and container['key'] in container:
                    if not self.is_hide(nativestr(container[container['key']])):
                        containers_filtered.append(container)
                else:
                    containers_filtered.append(container)
            stats.extend(containers_filtered)

        # Sort and update the stats
        # @TODO: Have a look because sort did not work for the moment (need memory stats ?)
        self.sort_key, self.stats = sort_docker_stats(stats)
        return self.stats

    @staticmethod
    def memory_usage_no_cache(mem: dict[str, float]) -> float:
        """Return the 'real' memory usage by removing inactive_file to usage"""
        # Ref: https://github.com/docker/docker-py/issues/3210
        return mem['usage'] - (mem['inactive_file'] if 'inactive_file' in mem else 0)

    def update_views(self) -> bool:
        """Update stats views."""
        # Call the father's method
        super().update_views()

        if not self.stats:
            return False

        # Add specifics information
        # Alert
        for i in self.stats:
            # Init the views for the current container (key = container name)
            self.views[i[self.get_key()]] = {'cpu': {}, 'mem': {}}
            # CPU alert
            if 'cpu' in i and 'total' in i['cpu']:
                # Looking for specific CPU container threshold in the conf file
                alert = self.get_alert(i['cpu']['total'], header=i['name'] + '_cpu', action_key=i['name'])
                if alert == 'DEFAULT':
                    # Not found ? Get back to default CPU threshold value
                    alert = self.get_alert(i['cpu']['total'], header='cpu')
                self.views[i[self.get_key()]]['cpu']['decoration'] = alert
            # MEM alert
            if 'memory' in i and 'usage' in i['memory']:
                # Looking for specific MEM container threshold in the conf file
                alert = self.get_alert(
                    self.memory_usage_no_cache(i['memory']),
                    maximum=i['memory']['limit'],
                    header=i['name'] + '_mem',
                    action_key=i['name'],
                )
                if alert == 'DEFAULT':
                    # Not found ? Get back to default MEM threshold value
                    alert = self.get_alert(
                        self.memory_usage_no_cache(i['memory']), maximum=i['memory']['limit'], header='mem'
                    )
                self.views[i[self.get_key()]]['mem']['decoration'] = alert

        # Display Engine and Pod name ?
        show_pod_name = False
        if any(ct.get("pod_name") for ct in self.stats):
            show_pod_name = True
        self.views['show_pod_name'] = show_pod_name
        show_engine_name = False
        if len({ct["engine"] for ct in self.stats}) > 1:
            show_engine_name = True
        self.views['show_engine_name'] = show_engine_name

        return True

    def build_title(self, ret):
        msg = '{}'.format('CONTAINERS')
        ret.append(self.curse_add_line(msg, "TITLE"))
        if len(self.stats) > 1:
            msg = f' {len(self.stats)}'
            ret.append(self.curse_add_line(msg))
            msg = f' sorted by {sort_for_human[self.sort_key]}'
            ret.append(self.curse_add_line(msg))
        if not self.views['show_engine_name']:
            msg = f' (served by {self.stats[0].get("engine", "")})'
        ret.append(self.curse_add_line(msg))
        ret.append(self.curse_new_line())
        return ret

    def maybe_add_engine_name_or_pod_line(self, ret):
        if self.views['show_engine_name']:
            ret = self.add_msg_to_line(ret, ' {:{width}}'.format('Engine', width=6))
        if self.views['show_pod_name']:
            ret = self.add_msg_to_line(ret, ' {:{width}}'.format('Pod', width=12))

        return ret

    def maybe_add_engine_name_or_pod_name(self, ret, container):
        ret.append(self.curse_new_line())
        if self.views['show_engine_name']:
            ret.append(self.curse_add_line(' {:{width}}'.format(container["engine"], width=6)))
        if self.views['show_pod_name']:
            ret.append(self.curse_add_line(' {:{width}}'.format(container.get("pod_id", "-"), width=12)))

        return ret

    def build_container_name(self, name_max_width):
        def build_for_this_max_length(ret, container):
            ret.append(
                self.curse_add_line(' {:{width}}'.format(container['name'][:name_max_width], width=name_max_width))
            )

            return ret

        return build_for_this_max_length

    def build_header(self, ret, name_max_width):
        ret.append(self.curse_new_line())

        ret = self.maybe_add_engine_name_or_pod_line(ret)

        if 'name' not in self.disable_stats:
            msg = ' {:{width}}'.format('Name', width=name_max_width)
            ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'name' else 'DEFAULT'))

        msgs = []
        if 'status' not in self.disable_stats:
            msgs.append('{:>10}'.format('Status'))
        if 'uptime' not in self.disable_stats:
            msgs.append('{:>10}'.format('Uptime'))
        ret = reduce(self.add_msg_to_line, msgs, ret)

        if 'cpu' not in self.disable_stats:
            msg = '{:>6}'.format('CPU%')
            ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'cpu_percent' else 'DEFAULT'))

        msgs = []
        if 'mem' not in self.disable_stats:
            msg = '{:>7}'.format('MEM')
            ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'memory_usage' else 'DEFAULT'))
            msgs.append('/{:<7}'.format('MAX'))

        if 'diskio' not in self.disable_stats:
            msgs.extend(['{:>7}'.format('IOR/s'), ' {:<7}'.format('IOW/s')])

        if 'networkio' not in self.disable_stats:
            msgs.extend(['{:>7}'.format('Rx/s'), ' {:<7}'.format('Tx/s')])

        if 'command' not in self.disable_stats:
            msgs.append(' {:8}'.format('Command'))

        return reduce(self.add_msg_to_line, msgs, ret)

    def add_msg_to_line(self, ret, msg):
        ret.append(self.curse_add_line(msg))

        return ret

    def get_max_of_container_names(self):
        return min(
            self.config.get_int_value('containers', 'max_name_size', default=20) if self.config is not None else 20,
            len(max(self.stats, key=lambda x: len(x['name']))['name']),
        )

    def build_status_name(self, ret, container):
        status = self.container_alert(container['status'])
        msg = '{:>10}'.format(container['status'][0:10])
        ret.append(self.curse_add_line(msg, status))

        return ret

    def build_uptime_line(self, ret, container):
        if container['uptime']:
            msg = '{:>10}'.format(container['uptime'])
        else:
            msg = '{:>10}'.format('_')

        return self.add_msg_to_line(ret, msg)

    def build_cpu_line(self, ret, container):
        try:
            msg = '{:>6.1f}'.format(container['cpu']['total'])
        except (KeyError, TypeError):
            msg = '{:>6}'.format('_')
        ret.append(self.curse_add_line(msg, self.get_views(item=container['name'], key='cpu', option='decoration')))

        return ret

    def build_memory_line(self, ret, container):
        try:
            msg = '{:>7}'.format(self.auto_unit(self.memory_usage_no_cache(container['memory'])))
        except KeyError:
            msg = '{:>7}'.format('_')
        ret.append(self.curse_add_line(msg, self.get_views(item=container['name'], key='mem', option='decoration')))
        try:
            msg = '/{:<7}'.format(self.auto_unit(container['memory']['limit']))
        except (KeyError, TypeError):
            msg = '/{:<7}'.format('_')
        ret.append(self.curse_add_line(msg))

        return ret

    def build_io_line(self, ret, container):
        unit = 'B'
        try:
            value = self.auto_unit(int(container['io_rx'])) + unit
            msg = f'{value:>7}'
        except (KeyError, TypeError):
            msg = '{:>7}'.format('_')
        ret.append(self.curse_add_line(msg))
        try:
            value = self.auto_unit(int(container['io_wx'])) + unit
            msg = f' {value:<7}'
        except (KeyError, TypeError):
            msg = ' {:<7}'.format('_')
        ret.append(self.curse_add_line(msg))

        return ret

    def build_net_line(self, args):
        def build_with_this_args(ret, container):
            if args.byte:
                # Bytes per second (for dummy)
                to_bit = 1
                unit = ''
            else:
                # Bits per second (for real network administrator | Default)
                to_bit = 8
                unit = 'b'
            try:
                value = self.auto_unit(int(container['network_rx'] * to_bit)) + unit
                msg = f'{value:>7}'
            except (KeyError, TypeError):
                msg = '{:>7}'.format('_')
            ret.append(self.curse_add_line(msg))
            try:
                value = self.auto_unit(int(container['network_tx'] * to_bit)) + unit
                msg = f' {value:<7}'
            except (KeyError, TypeError):
                msg = ' {:<7}'.format('_')
            ret.append(self.curse_add_line(msg))

            return ret

        return build_with_this_args

    def build_cmd_line(self, ret, container):
        if container['command'] is not None:
            msg = ' {}'.format(container['command'])
        else:
            msg = ' {}'.format('_')
        ret.append(self.curse_add_line(msg, splittable=True))

        return ret

    def msg_curse(self, args=None, max_width: Optional[int] = None) -> list[str]:
        """Return the dict to display in the curse interface."""
        # Init the return message
        init = []

        # Only process if stats exist (and non null) and display plugin enable...
        conditions = [not self.stats, len(self.stats) == 0, self.is_disabled()]
        if any(conditions):
            return init

        # Build the string message
        # Get the maximum containers name
        # Max size is configurable. See feature request #1723.
        name_max_width = self.get_max_of_container_names()

        steps = [
            self.build_title,
            partial(self.build_header, name_max_width=name_max_width),
            self.build_data_line(name_max_width, args),
        ]

        return reduce(lambda ret, step: step(ret), steps, init)

    def build_data_line(self, name_max_width, args):
        def build_for_this_params(ret):
            build_data_with_params = self.build_container_data(name_max_width, args)
            return reduce(build_data_with_params, self.stats, ret)

        return build_for_this_params

    def build_container_data(self, name_max_width, args):
        def build_with_this_params(ret, container):
            steps = [self.maybe_add_engine_name_or_pod_name]
            if 'name' not in self.disable_stats:
                steps.append(self.build_container_name(name_max_width))
            if 'status' not in self.disable_stats:
                steps.append(self.build_status_name)
            if 'uptime' not in self.disable_stats:
                steps.append(self.build_uptime_line)
            if 'cpu' not in self.disable_stats:
                steps.append(self.build_cpu_line)
            if 'mem' not in self.disable_stats:
                steps.append(self.build_memory_line)
            if 'diskio' not in self.disable_stats:
                steps.append(self.build_io_line)
            if 'networkio' not in self.disable_stats:
                steps.append(self.build_net_line(args))
            if 'command' not in self.disable_stats:
                steps.append(self.build_cmd_line)

            return reduce(lambda ret, step: step(ret, container), steps, ret)

        return build_with_this_params

    @staticmethod
    def container_alert(status: str) -> str:
        """Analyse the container status.
        One of created, restarting, running, removing, paused, exited, or dead
        """
        if status == 'running':
            return 'OK'
        if status == 'dead':
            return 'ERROR'
        if status in ['created', 'restarting', 'exited']:
            return 'WARNING'
        return 'INFO'


def sort_docker_stats(stats: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:

    # Make VM sort related to process sort
    if glances_processes.sort_key == 'memory_percent':
        sort_by = 'memory_usage'
        sort_by_secondary = 'cpu_percent'
    elif glances_processes.sort_key == 'name':
        sort_by = 'name'
        sort_by_secondary = 'cpu_percent'
    else:
        sort_by = 'cpu_percent'
        sort_by_secondary = 'memory_usage'

    # Sort docker stats
    sort_stats_processes(
        stats,
        sorted_by=sort_by,
        sorted_by_secondary=sort_by_secondary,
        # Reverse for all but name
        reverse=glances_processes.sort_key != 'name',
    )

    # Return the main sort key and the sorted stats
    return sort_by, stats


1		#
2		# This file is part of Glances.
3		#
4		# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
5		#
6		# SPDX-License-Identifier: LGPL-3.0-only
7		#
8
9		"""Containers plugin."""
10
11		from copy import deepcopy
12		from functools import partial, reduce
13		from typing import Any, Optional
14
15		from glances.globals import iteritems, itervalues, nativestr
16		from glances.logger import logger
17		from glances.plugins.containers.engines import ContainersExtension
18		from glances.plugins.containers.engines.docker import DockerExtension, disable_plugin_docker
19		from glances.plugins.containers.engines.podman import PodmanExtension, disable_plugin_podman
20		from glances.plugins.plugin.model import GlancesPluginModel
21		from glances.processes import glances_processes
22		from glances.processes import sort_stats as sort_stats_processes
23
24		# Fields description
25		# description: human readable description
26		# short_name: shortname to use un UI
27		# unit: unit type
28		# rate: is it a rate ? If yes, // by time_since_update when displayed,
29		# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
30		fields_description = {
31		'name': {
32		'description': 'Container name',
33		},
34		'id': {
35		'description': 'Container ID',
36		},
37		'image': {
38		'description': 'Container image',
39		},
40		'status': {
41		'description': 'Container status',
42		},
43		'created': {
44		'description': 'Container creation date',
45		},
46		'command': {
47		'description': 'Container command',
48		},
49		'cpu_percent': {
50		'description': 'Container CPU consumption',
51		'unit': 'percent',
52		},
53		'memory_usage': {
54		'description': 'Container memory usage',
55		'unit': 'byte',
56		},
57		'io_rx': {
58		'description': 'Container IO bytes read rate',
59		'unit': 'bytepersecond',
60		},
61		'io_wx': {
62		'description': 'Container IO bytes write rate',
63		'unit': 'bytepersecond',
64		},
65		'network_rx': {
66		'description': 'Container network RX bitrate',
67		'unit': 'bitpersecond',
68		},
69		'network_tx': {
70		'description': 'Container network TX bitrate',
71		'unit': 'bitpersecond',
72		},
73		'uptime': {
74		'description': 'Container uptime',
75		},
76		'engine': {
77		'description': 'Container engine (Docker and Podman are currently supported)',
78		},
79		'pod_name': {
80		'description': 'Pod name (only with Podman)',
81		},
82		'pod_id': {
83		'description': 'Pod ID (only with Podman)',
84		},
85		}
86
87		# Define the items history list (list of items to add to history)
88		# TODO: For the moment limited to the CPU. Had to change the graph exports
89		# method to display one graph per container.
90		# items_history_list = [{'name': 'cpu_percent',
91		# 'description': 'Container CPU consumption in %',
92		# 'y_unit': '%'},
93		# {'name': 'memory_usage',
94		# 'description': 'Container memory usage in bytes',
95		# 'y_unit': 'B'},
96		# {'name': 'network_rx',
97		# 'description': 'Container network RX bitrate in bits per second',
98		# 'y_unit': 'bps'},
99		# {'name': 'network_tx',
100		# 'description': 'Container network TX bitrate in bits per second',
101		# 'y_unit': 'bps'},
102		# {'name': 'io_r',
103		# 'description': 'Container IO bytes read per second',
104		# 'y_unit': 'Bps'},
105		# {'name': 'io_w',
106		# 'description': 'Container IO bytes write per second',
107		# 'y_unit': 'Bps'}]
108		items_history_list = [{'name': 'cpu_percent', 'description': 'Container CPU consumption in %', 'y_unit': '%'}]
109
110		# List of key to remove before export
111		export_exclude_list = ['cpu', 'io', 'memory', 'network']
112
113		# Sort dictionary for human
114		sort_for_human = {
115		'io_counters': 'disk IO',
116		'cpu_percent': 'CPU consumption',
117		'memory_usage': 'memory consumption',
118		'cpu_times': 'uptime',
119		'name': 'container name',
120		None: 'None',
121		}
122
123
124		class PluginModel(GlancesPluginModel):
125		"""Glances Docker plugin.
126
127		stats is a dict: {'version': {...}, 'containers': [{}, {}]}
128		"""
129
130		def __init__(self, args=None, config=None):
131		"""Init the plugin."""
132		super().__init__(
133		args=args, config=config, items_history_list=items_history_list, fields_description=fields_description
134		)
135
136		# The plugin can be disabled using: args.disable_docker
137		self.args = args
138
139		# Default config keys
140		self.config = config
141
142		# We want to display the stat in the curse interface
143		self.display_curse = True
144
145		self.watchers: dict[str, ContainersExtension] = {}
146
147		# Init the Docker API
148		if not disable_plugin_docker:
149		self.watchers['docker'] = DockerExtension()
150
151		# Init the Podman API
152		if not disable_plugin_podman:
153		self.watchers['podman'] = PodmanExtension(podman_sock=self._podman_sock())
154
155		# Sort key
156		self.sort_key = None
157
158		# Set the key's list be disabled in order to only display specific attribute in the container list
159		self.disable_stats = self.get_conf_value('disable_stats')
160
161		# Force a first update because we need two update to have the first stat
162		self.update()
163		self.refresh_timer.set(0)
164
165		def _podman_sock(self) -> str:
166		"""Return the podman sock.
167		Could be desfined in the [docker] section thanks to the podman_sock option.
168		Default value: unix:///run/user/1000/podman/podman.sock
169		"""
170		conf_podman_sock = self.get_conf_value('podman_sock')
171		if not conf_podman_sock:
172		return "unix:///run/user/1000/podman/podman.sock"
173		return conf_podman_sock[0]
174
175		def exit(self) -> None:
176		"""Overwrite the exit method to close threads."""
177		for watcher in itervalues(self.watchers):
178		watcher.stop()
179
180		# Call the father class
181		super().exit()
182
183		def get_key(self) -> str:
184		"""Return the key of the list."""
185		return 'name'
186
187		def get_export(self) -> list[dict]:
188		"""Overwrite the default export method.
189
190		- Only exports containers
191		- The key is the first container name
192		"""
193		try:
194		ret = deepcopy(self.stats)
195		except KeyError as e:
196		logger.debug(f"docker plugin - Docker export error {e}")
197		ret = []
198
199		# Remove fields uses to compute rate
200		for container in ret:
201		for i in export_exclude_list:
202		container.pop(i)
203
204		return ret
205
206		def _all_tag(self) -> bool:
207		"""Return the all tag of the Glances/Docker configuration file.
208
209		# By default, Glances only display running containers
210		# Set the following key to True to display all containers
211		all=True
212		"""
213		all_tag = self.get_conf_value('all')
214		if not all_tag:
215		return False
216		return all_tag[0].lower() == 'true'
217
218		@GlancesPluginModel._check_decorator
219		@GlancesPluginModel._log_result_decorator
220		def update(self) -> list[dict]:
221		"""Update Docker and podman stats using the input method."""
222		# Connection should be ok
223		if not self.watchers:
224		return self.get_init_value()
225
226		if self.input_method != 'local':
227		return self.get_init_value()
228
229		# Update stats
230		stats = []
231		for engine, watcher in iteritems(self.watchers):
232		_, containers = watcher.update(all_tag=self._all_tag())
233		containers_filtered = []
234		for container in containers:
235		container["engine"] = engine
236		if 'key' in container and container['key'] in container:
237		if not self.is_hide(nativestr(container[container['key']])):
238		containers_filtered.append(container)
239		else:
240		containers_filtered.append(container)
241		stats.extend(containers_filtered)
242
243		# Sort and update the stats
244		# @TODO: Have a look because sort did not work for the moment (need memory stats ?)
245		self.sort_key, self.stats = sort_docker_stats(stats)
246		return self.stats
247
248		@staticmethod
249		def memory_usage_no_cache(mem: dict[str, float]) -> float:
250		"""Return the 'real' memory usage by removing inactive_file to usage"""
251		# Ref: https://github.com/docker/docker-py/issues/3210
252		return mem['usage'] - (mem['inactive_file'] if 'inactive_file' in mem else 0)
253
254		def update_views(self) -> bool:
255		"""Update stats views."""
256		# Call the father's method
257		super().update_views()
258
259		if not self.stats:
260		return False
261
262		# Add specifics information
263		# Alert
264		for i in self.stats:
265		# Init the views for the current container (key = container name)
266		self.views[i[self.get_key()]] = {'cpu': {}, 'mem': {}}
267		# CPU alert
268		if 'cpu' in i and 'total' in i['cpu']:
269		# Looking for specific CPU container threshold in the conf file
270		alert = self.get_alert(i['cpu']['total'], header=i['name'] + '_cpu', action_key=i['name'])
271		if alert == 'DEFAULT':
272		# Not found ? Get back to default CPU threshold value
273		alert = self.get_alert(i['cpu']['total'], header='cpu')
274		self.views[i[self.get_key()]]['cpu']['decoration'] = alert
275		# MEM alert
276		if 'memory' in i and 'usage' in i['memory']:
277		# Looking for specific MEM container threshold in the conf file
278		alert = self.get_alert(
279		self.memory_usage_no_cache(i['memory']),
280		maximum=i['memory']['limit'],
281		header=i['name'] + '_mem',
282		action_key=i['name'],
283		)
284		if alert == 'DEFAULT':
285		# Not found ? Get back to default MEM threshold value
286		alert = self.get_alert(
287		self.memory_usage_no_cache(i['memory']), maximum=i['memory']['limit'], header='mem'
288		)
289		self.views[i[self.get_key()]]['mem']['decoration'] = alert
290
291		# Display Engine and Pod name ?
292		show_pod_name = False
293		if any(ct.get("pod_name") for ct in self.stats):
294		show_pod_name = True
295		self.views['show_pod_name'] = show_pod_name
296		show_engine_name = False
297		if len({ct["engine"] for ct in self.stats}) > 1:
298		show_engine_name = True
299		self.views['show_engine_name'] = show_engine_name
300
301		return True
302
303		def build_title(self, ret):
304		msg = '{}'.format('CONTAINERS')
305		ret.append(self.curse_add_line(msg, "TITLE"))
306		if len(self.stats) > 1:
307		msg = f' {len(self.stats)}'
308		ret.append(self.curse_add_line(msg))
309		msg = f' sorted by {sort_for_human[self.sort_key]}'
310		ret.append(self.curse_add_line(msg))
311		if not self.views['show_engine_name']:
312		msg = f' (served by {self.stats[0].get("engine", "")})'
313		ret.append(self.curse_add_line(msg))
314		ret.append(self.curse_new_line())
315		return ret
316
317		def maybe_add_engine_name_or_pod_line(self, ret):
318		if self.views['show_engine_name']:
319		ret = self.add_msg_to_line(ret, ' {:{width}}'.format('Engine', width=6))
320		if self.views['show_pod_name']:
321		ret = self.add_msg_to_line(ret, ' {:{width}}'.format('Pod', width=12))
322
323		return ret
324
325		def maybe_add_engine_name_or_pod_name(self, ret, container):
326		ret.append(self.curse_new_line())
327		if self.views['show_engine_name']:
328		ret.append(self.curse_add_line(' {:{width}}'.format(container["engine"], width=6)))
329		if self.views['show_pod_name']:
330		ret.append(self.curse_add_line(' {:{width}}'.format(container.get("pod_id", "-"), width=12)))
331
332		return ret
333
334		def build_container_name(self, name_max_width):
335		def build_for_this_max_length(ret, container):
336		ret.append(
337		self.curse_add_line(' {:{width}}'.format(container['name'][:name_max_width], width=name_max_width))
338		)
339
340		return ret
341
342		return build_for_this_max_length
343
344		def build_header(self, ret, name_max_width):
345		ret.append(self.curse_new_line())
346
347		ret = self.maybe_add_engine_name_or_pod_line(ret)
348
349		if 'name' not in self.disable_stats:
350		msg = ' {:{width}}'.format('Name', width=name_max_width)
351		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'name' else 'DEFAULT'))
352
353		msgs = []
354		if 'status' not in self.disable_stats:
355		msgs.append('{:>10}'.format('Status'))
356		if 'uptime' not in self.disable_stats:
357		msgs.append('{:>10}'.format('Uptime'))
358		ret = reduce(self.add_msg_to_line, msgs, ret)
359
360		if 'cpu' not in self.disable_stats:
361		msg = '{:>6}'.format('CPU%')
362		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'cpu_percent' else 'DEFAULT'))
363
364		msgs = []
365		if 'mem' not in self.disable_stats:
366		msg = '{:>7}'.format('MEM')
367		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'memory_usage' else 'DEFAULT'))
368		msgs.append('/{:<7}'.format('MAX'))
369
370		if 'diskio' not in self.disable_stats:
371		msgs.extend(['{:>7}'.format('IOR/s'), ' {:<7}'.format('IOW/s')])
372
373		if 'networkio' not in self.disable_stats:
374		msgs.extend(['{:>7}'.format('Rx/s'), ' {:<7}'.format('Tx/s')])
375
376		if 'command' not in self.disable_stats:
377		msgs.append(' {:8}'.format('Command'))
378
379		return reduce(self.add_msg_to_line, msgs, ret)
380
381		def add_msg_to_line(self, ret, msg):
382		ret.append(self.curse_add_line(msg))
383
384		return ret
385
386		def get_max_of_container_names(self):
387		return min(
388		self.config.get_int_value('containers', 'max_name_size', default=20) if self.config is not None else 20,
389		len(max(self.stats, key=lambda x: len(x['name']))['name']),
390		)
391
392		def build_status_name(self, ret, container):
393		status = self.container_alert(container['status'])
394		msg = '{:>10}'.format(container['status'][0:10])
395		ret.append(self.curse_add_line(msg, status))
396
397		return ret
398
399		def build_uptime_line(self, ret, container):
400		if container['uptime']:
401		msg = '{:>10}'.format(container['uptime'])
402		else:
403		msg = '{:>10}'.format('_')
404
405		return self.add_msg_to_line(ret, msg)
406
407		def build_cpu_line(self, ret, container):
408		try:
409		msg = '{:>6.1f}'.format(container['cpu']['total'])
410		except (KeyError, TypeError):
411		msg = '{:>6}'.format('_')
412		ret.append(self.curse_add_line(msg, self.get_views(item=container['name'], key='cpu', option='decoration')))
413
414		return ret
415
416		def build_memory_line(self, ret, container):
417		try:
418		msg = '{:>7}'.format(self.auto_unit(self.memory_usage_no_cache(container['memory'])))
419		except KeyError:
420		msg = '{:>7}'.format('_')
421		ret.append(self.curse_add_line(msg, self.get_views(item=container['name'], key='mem', option='decoration')))
422		try:
423		msg = '/{:<7}'.format(self.auto_unit(container['memory']['limit']))
424		except (KeyError, TypeError):
425		msg = '/{:<7}'.format('_')
426		ret.append(self.curse_add_line(msg))
427
428		return ret
429
430		def build_io_line(self, ret, container):
431		unit = 'B'
432		try:
433		value = self.auto_unit(int(container['io_rx'])) + unit
434		msg = f'{value:>7}'
435		except (KeyError, TypeError):
436		msg = '{:>7}'.format('_')
437		ret.append(self.curse_add_line(msg))
438		try:
439		value = self.auto_unit(int(container['io_wx'])) + unit
440		msg = f' {value:<7}'
441		except (KeyError, TypeError):
442		msg = ' {:<7}'.format('_')
443		ret.append(self.curse_add_line(msg))
444
445		return ret
446
447		def build_net_line(self, args):
448		def build_with_this_args(ret, container):
449		if args.byte:
450		# Bytes per second (for dummy)
451		to_bit = 1
452		unit = ''
453		else:
454		# Bits per second (for real network administrator \| Default)
455		to_bit = 8
456		unit = 'b'
457		try:
458		value = self.auto_unit(int(container['network_rx'] * to_bit)) + unit
459		msg = f'{value:>7}'
460		except (KeyError, TypeError):
461		msg = '{:>7}'.format('_')
462		ret.append(self.curse_add_line(msg))
463		try:
464		value = self.auto_unit(int(container['network_tx'] * to_bit)) + unit
465		msg = f' {value:<7}'
466		except (KeyError, TypeError):
467		msg = ' {:<7}'.format('_')
468		ret.append(self.curse_add_line(msg))
469
470		return ret
471
472		return build_with_this_args
473
474		def build_cmd_line(self, ret, container):
475		if container['command'] is not None:
476		msg = ' {}'.format(container['command'])
477		else:
478		msg = ' {}'.format('_')
479		ret.append(self.curse_add_line(msg, splittable=True))
480
481		return ret
482
483		def msg_curse(self, args=None, max_width: Optional[int] = None) -> list[str]:
484		"""Return the dict to display in the curse interface."""
485		# Init the return message
486		init = []
487
488		# Only process if stats exist (and non null) and display plugin enable...
489		conditions = [not self.stats, len(self.stats) == 0, self.is_disabled()]
490		if any(conditions):
491		return init
492
493		# Build the string message
494		# Get the maximum containers name
495		# Max size is configurable. See feature request #1723.
496		name_max_width = self.get_max_of_container_names()
497
498		steps = [
499		self.build_title,
500		partial(self.build_header, name_max_width=name_max_width),
501		self.build_data_line(name_max_width, args),
502		]
503
504		return reduce(lambda ret, step: step(ret), steps, init)
505
506		def build_data_line(self, name_max_width, args):
507		def build_for_this_params(ret):
508		build_data_with_params = self.build_container_data(name_max_width, args)
509		return reduce(build_data_with_params, self.stats, ret)
510
511		return build_for_this_params
512
513		def build_container_data(self, name_max_width, args):
514		def build_with_this_params(ret, container):
515		steps = [self.maybe_add_engine_name_or_pod_name]
516		if 'name' not in self.disable_stats:
517		steps.append(self.build_container_name(name_max_width))
518		if 'status' not in self.disable_stats:
519		steps.append(self.build_status_name)
520		if 'uptime' not in self.disable_stats:
521		steps.append(self.build_uptime_line)
522		if 'cpu' not in self.disable_stats:
523		steps.append(self.build_cpu_line)
524		if 'mem' not in self.disable_stats:
525		steps.append(self.build_memory_line)
526		if 'diskio' not in self.disable_stats:
527		steps.append(self.build_io_line)
528		if 'networkio' not in self.disable_stats:
529		steps.append(self.build_net_line(args))
530		if 'command' not in self.disable_stats:
531		steps.append(self.build_cmd_line)
532
533		return reduce(lambda ret, step: step(ret, container), steps, ret)
534
535		return build_with_this_params
536
537		@staticmethod
538		def container_alert(status: str) -> str:
539		"""Analyse the container status.
540		One of created, restarting, running, removing, paused, exited, or dead
541		"""
542		if status == 'running':
543		return 'OK'
544		if status == 'dead':
545		return 'ERROR'
546		if status in ['created', 'restarting', 'exited']:
547		return 'WARNING'
548		return 'INFO'
549
550
551	View Code Duplication	def sort_docker_stats(stats: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
		0 ignored issues – show Duplication introduced 2024-07-24 13:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
552		# Make VM sort related to process sort
553		if glances_processes.sort_key == 'memory_percent':
554		sort_by = 'memory_usage'
555		sort_by_secondary = 'cpu_percent'
556		elif glances_processes.sort_key == 'name':
557		sort_by = 'name'
558		sort_by_secondary = 'cpu_percent'
559		else:
560		sort_by = 'cpu_percent'
561		sort_by_secondary = 'memory_usage'
562
563		# Sort docker stats
564		sort_stats_processes(
565		stats,
566		sorted_by=sort_by,
567		sorted_by_secondary=sort_by_secondary,
568		# Reverse for all but name
569		reverse=glances_processes.sort_key != 'name',
570		)
571
572		# Return the main sort key and the sorted stats
573		return sort_by, stats
574

nicolargo / glances

Push — master ( ce0fc3...e09530 )

PluginModel.build_header() D

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like