glances.plugins.containers.ContainersPlugin.msg_curse() - Code Metrics - Inspection of "Merge pull request #3180 from ariel-anieli/issue-2..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( e118d7...ef83c4 )

by Nicolas

created 2025-05-25 15:31 UTC

glances.plugins.containers.ContainersPlugin.msg_curse() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	22
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	11
nop	3
dl	0
loc	22
rs	9.85
c	0
b	0
f	0

#
# 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 itertools import chain
from typing import Any, Optional

from glances.globals import 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 ContainersPlugin(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 self.watchers.values():
            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()

        def is_key_in_container_and_not_hidden(container):
            return (key := container.get('key')) in container and not self.is_hide(nativestr(container.get(key)))

        def is_key_absent_in_container(container):
            return 'key' not in container or container.get('key') not in container

        def add_engine_into_container(engine, container):
            return container | {"engine": engine}

        def get_containers_from_updated_watcher(watcher):
            _, containers = watcher.update(all_tag=self._all_tag())
            return containers

        # Update stats
        stats = list(
            chain.from_iterable(
                (
                    add_engine_into_container(engine, container)
                    for container in get_containers_from_updated_watcher(watcher)
                    if is_key_in_container_and_not_hidden(container) or is_key_absent_in_container(container)
                )
                for engine, watcher in self.watchers.items()
            )
        )

        # 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')
                if 'cpu' in self.views[i[self.get_key()]]:
                    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'
                    )
                if 'mem' in self.views[i[self.get_key()]]:
                    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...
        if any([not self.stats, len(self.stats) == 0, self.is_disabled()]):
            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]
            options = {
                'name': self.build_container_name(name_max_width),
                'status': self.build_status_name,
                'uptime': self.build_uptime_line,
                'cpu': self.build_cpu_line,
                'mem': self.build_memory_line,
                'diskio': self.build_io_line,
                'networkio': self.build_net_line(args),
                'command': self.build_cmd_line,
            }

            steps.extend(v for k, v in options.items() if k not in self.disable_stats)
            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
    sort_by, sort_by_secondary = {
        'memory_percent': ('memory_usage', 'cpu_percent'),
        'name': ('name', 'cpu_percent'),
    }.get(glances_processes.sort_key, ('cpu_percent', 'memory_usage'))

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

nicolargo / glances

Push — develop ( e118d7...ef83c4 )

glances.plugins.containers.ContainersPlugin.msg_curse() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like