glances.plugins.containers.ContainersPlugin.build_memory_line() - Code Metrics - Inspection of "plugins: containers; Refactor" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — develop (#3180)

unknown

created 2025-05-23 16:49 UTC

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

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	13
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	12
nop	3
dl	0
loc	13
rs	9.8
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 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 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 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()

        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_as_key_in_container(engine, container):
            container["engine"] = engine
            return container

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

        # Update stats
        stats = list(
            chain.from_iterable(
                (
                    add_engine_as_key_in_container(engine, container)
                    for container in get_containers_from_watcher(watcher)
                    if is_key_in_container_and_not_hidden(container) or is_key_absent_in_container(container)
                )
                for engine, watcher in iteritems(self.watchers)
            )
        )

        # 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 iteritems, itervalues, 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 itervalues(self.watchers):
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_as_key_in_container(engine, container):
237			container["engine"] = engine
238			return container
239
240			def get_containers_from_watcher(watcher):
241			_, containers = watcher.update(all_tag=self._all_tag())
242			return containers
243
244			# Update stats
245			stats = list(
246			chain.from_iterable(
247			(
248			add_engine_as_key_in_container(engine, container)
249			for container in get_containers_from_watcher(watcher)
250			if is_key_in_container_and_not_hidden(container) or is_key_absent_in_container(container)
251			)
252			for engine, watcher in iteritems(self.watchers)
253			)
254			)
255
256			# Sort and update the stats
257			# @TODO: Have a look because sort did not work for the moment (need memory stats ?)
258			self.sort_key, self.stats = sort_docker_stats(stats)
259			return self.stats
260
261			@staticmethod
262			def memory_usage_no_cache(mem: dict[str, float]) -> float:
263			"""Return the 'real' memory usage by removing inactive_file to usage"""
264			# Ref: https://github.com/docker/docker-py/issues/3210
265			return mem['usage'] - (mem['inactive_file'] if 'inactive_file' in mem else 0)
266
267			def update_views(self) -> bool:
268			"""Update stats views."""
269			# Call the father's method
270			super().update_views()
271
272			if not self.stats:
273			return False
274
275			# Add specifics information
276			# Alert
277			for i in self.stats:
278			# Init the views for the current container (key = container name)
279			self.views[i[self.get_key()]] = {'cpu': {}, 'mem': {}}
280			# CPU alert
281			if 'cpu' in i and 'total' in i['cpu']:
282			# Looking for specific CPU container threshold in the conf file
283			alert = self.get_alert(i['cpu']['total'], header=i['name'] + '_cpu', action_key=i['name'])
284			if alert == 'DEFAULT':
285			# Not found ? Get back to default CPU threshold value
286			alert = self.get_alert(i['cpu']['total'], header='cpu')
287			if 'cpu' in self.views[i[self.get_key()]]:
288			self.views[i[self.get_key()]]['cpu']['decoration'] = alert
289			# MEM alert
290			if 'memory' in i and 'usage' in i['memory']:
291			# Looking for specific MEM container threshold in the conf file
292			alert = self.get_alert(
293			self.memory_usage_no_cache(i['memory']),
294			maximum=i['memory']['limit'],
295			header=i['name'] + '_mem',
296			action_key=i['name'],
297			)
298			if alert == 'DEFAULT':
299			# Not found ? Get back to default MEM threshold value
300			alert = self.get_alert(
301			self.memory_usage_no_cache(i['memory']), maximum=i['memory']['limit'], header='mem'
302			)
303			if 'mem' in self.views[i[self.get_key()]]:
304			self.views[i[self.get_key()]]['mem']['decoration'] = alert
305
306			# Display Engine and Pod name ?
307			show_pod_name = False
308			if any(ct.get("pod_name") for ct in self.stats):
309			show_pod_name = True
310			self.views['show_pod_name'] = show_pod_name
311			show_engine_name = False
312			if len({ct["engine"] for ct in self.stats}) > 1:
313			show_engine_name = True
314			self.views['show_engine_name'] = show_engine_name
315
316			return True
317
318			def build_title(self, ret):
319			msg = '{}'.format('CONTAINERS')
320			ret.append(self.curse_add_line(msg, "TITLE"))
321			if len(self.stats) > 1:
322			msg = f' {len(self.stats)}'
323			ret.append(self.curse_add_line(msg))
324			msg = f' sorted by {sort_for_human[self.sort_key]}'
325			ret.append(self.curse_add_line(msg))
326			if not self.views['show_engine_name']:
327			msg = f' (served by {self.stats[0].get("engine", "")})'
328			ret.append(self.curse_add_line(msg))
329			ret.append(self.curse_new_line())
330			return ret
331
332			def maybe_add_engine_name_or_pod_line(self, ret):
333			if self.views['show_engine_name']:
334			ret = self.add_msg_to_line(ret, ' {:{width}}'.format('Engine', width=6))
335			if self.views['show_pod_name']:
336			ret = self.add_msg_to_line(ret, ' {:{width}}'.format('Pod', width=12))
337
338			return ret
339
340			def maybe_add_engine_name_or_pod_name(self, ret, container):
341			ret.append(self.curse_new_line())
342			if self.views['show_engine_name']:
343			ret.append(self.curse_add_line(' {:{width}}'.format(container["engine"], width=6)))
344			if self.views['show_pod_name']:
345			ret.append(self.curse_add_line(' {:{width}}'.format(container.get("pod_id", "-"), width=12)))
346
347			return ret
348
349			def build_container_name(self, name_max_width):
350			def build_for_this_max_length(ret, container):
351			ret.append(
352			self.curse_add_line(' {:{width}}'.format(container['name'][:name_max_width], width=name_max_width))
353			)
354
355			return ret
356
357			return build_for_this_max_length
358
359			def build_header(self, ret, name_max_width):
360			ret.append(self.curse_new_line())
361
362			ret = self.maybe_add_engine_name_or_pod_line(ret)
363
364			if 'name' not in self.disable_stats:
365			msg = ' {:{width}}'.format('Name', width=name_max_width)
366			ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'name' else 'DEFAULT'))
367
368			msgs = []
369			if 'status' not in self.disable_stats:
370			msgs.append('{:>10}'.format('Status'))
371			if 'uptime' not in self.disable_stats:
372			msgs.append('{:>10}'.format('Uptime'))
373			ret = reduce(self.add_msg_to_line, msgs, ret)
374
375			if 'cpu' not in self.disable_stats:
376			msg = '{:>6}'.format('CPU%')
377			ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'cpu_percent' else 'DEFAULT'))
378
379			msgs = []
380			if 'mem' not in self.disable_stats:
381			msg = '{:>7}'.format('MEM')
382			ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'memory_usage' else 'DEFAULT'))
383			msgs.append('/{:<7}'.format('MAX'))
384
385			if 'diskio' not in self.disable_stats:
386			msgs.extend(['{:>7}'.format('IOR/s'), ' {:<7}'.format('IOW/s')])
387
388			if 'networkio' not in self.disable_stats:
389			msgs.extend(['{:>7}'.format('Rx/s'), ' {:<7}'.format('Tx/s')])
390
391			if 'command' not in self.disable_stats:
392			msgs.append(' {:8}'.format('Command'))
393
394			return reduce(self.add_msg_to_line, msgs, ret)
395
396			def add_msg_to_line(self, ret, msg):
397			ret.append(self.curse_add_line(msg))
398
399			return ret
400
401			def get_max_of_container_names(self):
402			return min(
403			self.config.get_int_value('containers', 'max_name_size', default=20) if self.config is not None else 20,
404			len(max(self.stats, key=lambda x: len(x['name']))['name']),
405			)
406
407			def build_status_name(self, ret, container):
408			status = self.container_alert(container['status'])
409			msg = '{:>10}'.format(container['status'][0:10])
410			ret.append(self.curse_add_line(msg, status))
411
412			return ret
413
414			def build_uptime_line(self, ret, container):
415			if container['uptime']:
416			msg = '{:>10}'.format(container['uptime'])
417			else:
418			msg = '{:>10}'.format('_')
419
420			return self.add_msg_to_line(ret, msg)
421
422			def build_cpu_line(self, ret, container):
423			try:
424			msg = '{:>6.1f}'.format(container['cpu']['total'])
425			except (KeyError, TypeError):
426			msg = '{:>6}'.format('_')
427			ret.append(self.curse_add_line(msg, self.get_views(item=container['name'], key='cpu', option='decoration')))
428
429			return ret
430
431			def build_memory_line(self, ret, container):
432			try:
433			msg = '{:>7}'.format(self.auto_unit(self.memory_usage_no_cache(container['memory'])))
434			except KeyError:
435			msg = '{:>7}'.format('_')
436			ret.append(self.curse_add_line(msg, self.get_views(item=container['name'], key='mem', option='decoration')))
437			try:
438			msg = '/{:<7}'.format(self.auto_unit(container['memory']['limit']))
439			except (KeyError, TypeError):
440			msg = '/{:<7}'.format('_')
441			ret.append(self.curse_add_line(msg))
442
443			return ret
444
445			def build_io_line(self, ret, container):
446			unit = 'B'
447			try:
448			value = self.auto_unit(int(container['io_rx'])) + unit
449			msg = f'{value:>7}'
450			except (KeyError, TypeError):
451			msg = '{:>7}'.format('_')
452			ret.append(self.curse_add_line(msg))
453			try:
454			value = self.auto_unit(int(container['io_wx'])) + unit
455			msg = f' {value:<7}'
456			except (KeyError, TypeError):
457			msg = ' {:<7}'.format('_')
458			ret.append(self.curse_add_line(msg))
459
460			return ret
461
462			def build_net_line(self, args):
463			def build_with_this_args(ret, container):
464			if args.byte:
465			# Bytes per second (for dummy)
466			to_bit = 1
467			unit = ''
468			else:
469			# Bits per second (for real network administrator \| Default)
470			to_bit = 8
471			unit = 'b'
472			try:
473			value = self.auto_unit(int(container['network_rx'] * to_bit)) + unit
474			msg = f'{value:>7}'
475			except (KeyError, TypeError):
476			msg = '{:>7}'.format('_')
477			ret.append(self.curse_add_line(msg))
478			try:
479			value = self.auto_unit(int(container['network_tx'] * to_bit)) + unit
480			msg = f' {value:<7}'
481			except (KeyError, TypeError):
482			msg = ' {:<7}'.format('_')
483			ret.append(self.curse_add_line(msg))
484
485			return ret
486
487			return build_with_this_args
488
489			def build_cmd_line(self, ret, container):
490			if container['command'] is not None:
491			msg = ' {}'.format(container['command'])
492			else:
493			msg = ' {}'.format('_')
494			ret.append(self.curse_add_line(msg, splittable=True))
495
496			return ret
497
498			def msg_curse(self, args=None, max_width: Optional[int] = None) -> list[str]:
499			"""Return the dict to display in the curse interface."""
500			# Init the return message
501			init = []
502
503			# Only process if stats exist (and non null) and display plugin enable...
504			if any([not self.stats, len(self.stats) == 0, self.is_disabled()]):
505			return init
506
507			# Build the string message
508			# Get the maximum containers name
509			# Max size is configurable. See feature request #1723.
510			name_max_width = self.get_max_of_container_names()
511
512			steps = [
513			self.build_title,
514			partial(self.build_header, name_max_width=name_max_width),
515			self.build_data_line(name_max_width, args),
516			]
517
518			return reduce(lambda ret, step: step(ret), steps, init)
519
520			def build_data_line(self, name_max_width, args):
521			def build_for_this_params(ret):
522			build_data_with_params = self.build_container_data(name_max_width, args)
523			return reduce(build_data_with_params, self.stats, ret)
524
525			return build_for_this_params
526
527			def build_container_data(self, name_max_width, args):
528			def build_with_this_params(ret, container):
529			steps = [self.maybe_add_engine_name_or_pod_name]
530			options = {
531			'name': self.build_container_name(name_max_width),
532			'status': self.build_status_name,
533			'uptime': self.build_uptime_line,
534			'cpu': self.build_cpu_line,
535			'mem': self.build_memory_line,
536			'diskio': self.build_io_line,
537			'networkio': self.build_net_line(args),
538			'command': self.build_cmd_line,
539			}
540
541			steps.extend(v for k, v in options.items() if k not in self.disable_stats)
542			return reduce(lambda ret, step: step(ret, container), steps, ret)
543
544			return build_with_this_params
545
546			@staticmethod
547			def container_alert(status: str) -> str:
548			"""Analyse the container status.
549			One of created, restarting, running, removing, paused, exited, or dead
550			"""
551			if status == 'running':
552			return 'OK'
553			if status == 'dead':
554			return 'ERROR'
555			if status in ['created', 'restarting', 'exited']:
556			return 'WARNING'
557			return 'INFO'
558
559
560			def sort_docker_stats(stats: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
561			# Make VM sort related to process sort
562			sort_by, sort_by_secondary = {
563			'memory_percent': ('memory_usage', 'cpu_percent'),
564			'name': ('name', 'cpu_percent'),
565			}.get(glances_processes.sort_key, ('cpu_percent', 'memory_usage'))
566
567			# Sort docker stats
568			stats = sort_stats_processes(
569			stats,
570			sorted_by=sort_by,
571			sorted_by_secondary=sort_by_secondary,
572			# Reverse for all but name
573			reverse=glances_processes.sort_key != 'name',
574			)
575
576			# Return the main sort key and the sorted stats
577			return sort_by, stats
578

nicolargo / glances

Pull Request — develop (#3180)

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like