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

Test Failed

Push — master ( 372380...7cfc0c )

by Nicolas

created 2025-03-22 16:51 UTC

PluginModel.update_views() D

↳ Parent: glances.plugins.containers

Complexity

Conditions

Size

Total Lines	50
Code Lines	32

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

nicolargo / glances

Push — master ( 372380...7cfc0c )

PluginModel.update_views() D

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like