glances.plugins.containers.PluginModel.build_container_data() - Code Metrics - Inspection of "Reduced complexity of `glances.plugins.containers...." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — develop (#2919)

unknown

created 2024-08-12 21:41 UTC

PluginModel.build_container_data() A

↳ Parent: glances.plugins.containers

Complexity

Conditions

Size

Total Lines	17
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	14
nop	3
dl	0
loc	17
rs	9.7
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 reduce
from typing import Any, Dict, List, Optional, Tuple

from glances.globals import iteritems, itervalues
from glances.logger import logger
from glances.plugins.containers.engines import ContainersExtension
from glances.plugins.containers.engines.docker import DockerExtension, import_docker_error_tag
from glances.plugins.containers.engines.podman import PodmanExtension, import_podman_error_tag
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 import_docker_error_tag:
            self.watchers['docker'] = DockerExtension()

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

        # Sort key
        self.sort_key = None

        # 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 len(conf_podman_sock) == 0:
            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 len(all_tag) == 0:
            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):
            version, containers = watcher.update(all_tag=self._all_tag())
            for container in containers:
                container["engine"] = engine
            stats.extend(containers)

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

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

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

        if not self.stats:
            return False

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

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

        return True

    def build_tile(self, ret):
        msg = '{}'.format('CONTAINERS')
        ret.append(self.curse_add_line(msg, "TITLE"))
        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)

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

        msgs = ['{:>10}'.format('Status'), '{:>10}'.format('Uptime')]
        ret = reduce(self.add_msg_to_line, msgs, ret)

        msg = '{:>6}'.format('CPU%')
        ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'cpu_percent' else 'DEFAULT'))
        msg = '{:>7}'.format('MEM')
        ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'memory_usage' else 'DEFAULT'))

        msgs = [
            '/{:<7}'.format('MAX'),
            '{:>7}'.format('IOR/s'),
            ' {:<7}'.format('IOW/s'),
            '{:>7}'.format('Rx/s'),
            ' {:<7}'.format('Tx/s'),
            ' {: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('_')

            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
        ret = []

        # 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 ret

        # 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()

        # Title
        ret = self.build_tile(ret)
        # Header
        ret = self.build_header(ret, name_max_width)

        # Data
        return self.build_data_line(name_max_width, args)(ret)

    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,
                self.build_container_name(name_max_width),
                self.build_status_name,
                self.build_uptime_line,
                self.build_cpu_line,
                self.build_memory_line,
                self.build_io_line,
                self.build_net_line(args),
                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 reduce
13		from typing import Any, Dict, List, Optional, Tuple
14
15		from glances.globals import iteritems, itervalues
16		from glances.logger import logger
17		from glances.plugins.containers.engines import ContainersExtension
18		from glances.plugins.containers.engines.docker import DockerExtension, import_docker_error_tag
19		from glances.plugins.containers.engines.podman import PodmanExtension, import_podman_error_tag
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 import_docker_error_tag:
149		self.watchers['docker'] = DockerExtension()
150
151		# Init the Podman API
152		if not import_podman_error_tag:
153		self.watchers['podman'] = PodmanExtension(podman_sock=self._podman_sock())
154
155		# Sort key
156		self.sort_key = None
157
158		# Force a first update because we need two update to have the first stat
159		self.update()
160		self.refresh_timer.set(0)
161
162		def _podman_sock(self) -> str:
163		"""Return the podman sock.
164		Could be desfined in the [docker] section thanks to the podman_sock option.
165		Default value: unix:///run/user/1000/podman/podman.sock
166		"""
167		conf_podman_sock = self.get_conf_value('podman_sock')
168		if len(conf_podman_sock) == 0:
169		return "unix:///run/user/1000/podman/podman.sock"
170		return conf_podman_sock[0]
171
172		def exit(self) -> None:
173		"""Overwrite the exit method to close threads."""
174		for watcher in itervalues(self.watchers):
175		watcher.stop()
176
177		# Call the father class
178		super().exit()
179
180		def get_key(self) -> str:
181		"""Return the key of the list."""
182		return 'name'
183
184		def get_export(self) -> List[Dict]:
185		"""Overwrite the default export method.
186
187		- Only exports containers
188		- The key is the first container name
189		"""
190		try:
191		ret = deepcopy(self.stats)
192		except KeyError as e:
193		logger.debug(f"docker plugin - Docker export error {e}")
194		ret = []
195
196		# Remove fields uses to compute rate
197		for container in ret:
198		for i in export_exclude_list:
199		container.pop(i)
200
201		return ret
202
203		def _all_tag(self) -> bool:
204		"""Return the all tag of the Glances/Docker configuration file.
205
206		# By default, Glances only display running containers
207		# Set the following key to True to display all containers
208		all=True
209		"""
210		all_tag = self.get_conf_value('all')
211		if len(all_tag) == 0:
212		return False
213		return all_tag[0].lower() == 'true'
214
215	View Code Duplication	@GlancesPluginModel._check_decorator
		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...
216		@GlancesPluginModel._log_result_decorator
217		def update(self) -> List[Dict]:
218		"""Update Docker and podman stats using the input method."""
219		# Connection should be ok
220		if not self.watchers:
221		return self.get_init_value()
222
223		if self.input_method != 'local':
224		return self.get_init_value()
225
226		# Update stats
227		stats = []
228		for engine, watcher in iteritems(self.watchers):
229		version, containers = watcher.update(all_tag=self._all_tag())
230		for container in containers:
231		container["engine"] = engine
232		stats.extend(containers)
233
234		# Sort and update the stats
235		# @TODO: Have a look because sort did not work for the moment (need memory stats ?)
236		self.sort_key, self.stats = sort_docker_stats(stats)
237		return self.stats
238
239		@staticmethod
240		def memory_usage_no_cache(mem: Dict[str, float]) -> float:
241		"""Return the 'real' memory usage by removing inactive_file to usage"""
242		# Ref: https://github.com/docker/docker-py/issues/3210
243		return mem['usage'] - (mem['inactive_file'] if 'inactive_file' in mem else 0)
244
245		def update_views(self) -> bool:
246		"""Update stats views."""
247		# Call the father's method
248		super().update_views()
249
250		if not self.stats:
251		return False
252
253		# Add specifics information
254		# Alert
255		for i in self.stats:
256		# Init the views for the current container (key = container name)
257		self.views[i[self.get_key()]] = {'cpu': {}, 'mem': {}}
258		# CPU alert
259		if 'cpu' in i and 'total' in i['cpu']:
260		# Looking for specific CPU container threshold in the conf file
261		alert = self.get_alert(i['cpu']['total'], header=i['name'] + '_cpu', action_key=i['name'])
262		if alert == 'DEFAULT':
263		# Not found ? Get back to default CPU threshold value
264		alert = self.get_alert(i['cpu']['total'], header='cpu')
265		self.views[i[self.get_key()]]['cpu']['decoration'] = alert
266		# MEM alert
267		if 'memory' in i and 'usage' in i['memory']:
268		# Looking for specific MEM container threshold in the conf file
269		alert = self.get_alert(
270		self.memory_usage_no_cache(i['memory']),
271		maximum=i['memory']['limit'],
272		header=i['name'] + '_mem',
273		action_key=i['name'],
274		)
275		if alert == 'DEFAULT':
276		# Not found ? Get back to default MEM threshold value
277		alert = self.get_alert(
278		self.memory_usage_no_cache(i['memory']), maximum=i['memory']['limit'], header='mem'
279		)
280		self.views[i[self.get_key()]]['mem']['decoration'] = alert
281
282		# Display Engine and Pod name ?
283		show_pod_name = False
284		if any(ct.get("pod_name") for ct in self.stats):
285		show_pod_name = True
286		self.views['show_pod_name'] = show_pod_name
287		show_engine_name = False
288		if len({ct["engine"] for ct in self.stats}) > 1:
289		show_engine_name = True
290		self.views['show_engine_name'] = show_engine_name
291
292		return True
293
294		def build_tile(self, ret):
295		msg = '{}'.format('CONTAINERS')
296		ret.append(self.curse_add_line(msg, "TITLE"))
297		msg = f' {len(self.stats)}'
298		ret.append(self.curse_add_line(msg))
299		msg = f' sorted by {sort_for_human[self.sort_key]}'
300		ret.append(self.curse_add_line(msg))
301		if not self.views['show_engine_name']:
302		msg = f' (served by {self.stats[0].get("engine", "")})'
303		ret.append(self.curse_add_line(msg))
304		ret.append(self.curse_new_line())
305		return ret
306
307		def maybe_add_engine_name_or_pod_line(self, ret):
308		if self.views['show_engine_name']:
309		ret = self.add_msg_to_line(ret, ' {:{width}}'.format('Engine', width=6))
310		if self.views['show_pod_name']:
311		ret = self.add_msg_to_line(ret, ' {:{width}}'.format('Pod', width=12))
312
313		return ret
314
315		def maybe_add_engine_name_or_pod_name(self, ret, container):
316		ret.append(self.curse_new_line())
317		if self.views['show_engine_name']:
318		ret.append(self.curse_add_line(' {:{width}}'.format(container["engine"], width=6)))
319		if self.views['show_pod_name']:
320		ret.append(self.curse_add_line(' {:{width}}'.format(container.get("pod_id", "-"), width=12)))
321
322		return ret
323
324		def build_container_name(self, name_max_width):
325		def build_for_this_max_length(ret, container):
326		ret.append(
327		self.curse_add_line(' {:{width}}'.format(container['name'][:name_max_width], width=name_max_width))
328		)
329
330		return ret
331
332		return build_for_this_max_length
333
334		def build_header(self, ret, name_max_width):
335		ret.append(self.curse_new_line())
336
337		ret = self.maybe_add_engine_name_or_pod_line(ret)
338
339		msg = ' {:{width}}'.format('Name', width=name_max_width)
340		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'name' else 'DEFAULT'))
341
342		msgs = ['{:>10}'.format('Status'), '{:>10}'.format('Uptime')]
343		ret = reduce(self.add_msg_to_line, msgs, ret)
344
345		msg = '{:>6}'.format('CPU%')
346		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'cpu_percent' else 'DEFAULT'))
347		msg = '{:>7}'.format('MEM')
348		ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'memory_usage' else 'DEFAULT'))
349
350		msgs = [
351		'/{:<7}'.format('MAX'),
352		'{:>7}'.format('IOR/s'),
353		' {:<7}'.format('IOW/s'),
354		'{:>7}'.format('Rx/s'),
355		' {:<7}'.format('Tx/s'),
356		' {:8}'.format('Command'),
357		]
358
359		return reduce(self.add_msg_to_line, msgs, ret)
360
361		def add_msg_to_line(self, ret, msg):
362		ret.append(self.curse_add_line(msg))
363
364		return ret
365
366		def get_max_of_container_names(self):
367		return min(
368		self.config.get_int_value('containers', 'max_name_size', default=20) if self.config is not None else 20,
369		len(max(self.stats, key=lambda x: len(x['name']))['name']),
370		)
371
372		def build_status_name(self, ret, container):
373		status = self.container_alert(container['status'])
374		msg = '{:>10}'.format(container['status'][0:10])
375		ret.append(self.curse_add_line(msg, status))
376
377		return ret
378
379		def build_uptime_line(self, ret, container):
380		if container['uptime']:
381		msg = '{:>10}'.format(container['uptime'])
382		else:
383		msg = '{:>10}'.format('_')
384
385		return self.add_msg_to_line(ret, msg)
386
387		def build_cpu_line(self, ret, container):
388		try:
389		msg = '{:>6.1f}'.format(container['cpu']['total'])
390		except (KeyError, TypeError):
391		msg = '{:>6}'.format('_')
392		ret.append(self.curse_add_line(msg, self.get_views(item=container['name'], key='cpu', option='decoration')))
393
394		return ret
395
396		def build_memory_line(self, ret, container):
397		try:
398		msg = '{:>7}'.format(self.auto_unit(self.memory_usage_no_cache(container['memory'])))
399		except KeyError:
400		msg = '{:>7}'.format('_')
401		ret.append(self.curse_add_line(msg, self.get_views(item=container['name'], key='mem', option='decoration')))
402		try:
403		msg = '/{:<7}'.format(self.auto_unit(container['memory']['limit']))
404		except (KeyError, TypeError):
405		msg = '/{:<7}'.format('_')
406		ret.append(self.curse_add_line(msg))
407
408		return ret
409
410		def build_io_line(self, ret, container):
411		unit = 'B'
412		try:
413		value = self.auto_unit(int(container['io_rx'])) + unit
414		msg = f'{value:>7}'
415		except (KeyError, TypeError):
416		msg = '{:>7}'.format('_')
417		ret.append(self.curse_add_line(msg))
418		try:
419		value = self.auto_unit(int(container['io_wx'])) + unit
420		msg = f' {value:<7}'
421		except (KeyError, TypeError):
422		msg = ' {:<7}'.format('_')
423		ret.append(self.curse_add_line(msg))
424
425		return ret
426
427		def build_net_line(self, args):
428		def build_with_this_args(ret, container):
429		if args.byte:
430		# Bytes per second (for dummy)
431		to_bit = 1
432		unit = ''
433		else:
434		# Bits per second (for real network administrator \| Default)
435		to_bit = 8
436		unit = 'b'
437		try:
438		value = self.auto_unit(int(container['network_rx'] * to_bit)) + unit
439		msg = f'{value:>7}'
440		except (KeyError, TypeError):
441		msg = '{:>7}'.format('_')
442		ret.append(self.curse_add_line(msg))
443		try:
444		value = self.auto_unit(int(container['network_tx'] * to_bit)) + unit
445		msg = f' {value:<7}'
446		except (KeyError, TypeError):
447		msg = ' {:<7}'.format('_')
448
449		return ret
450
451		return build_with_this_args
452
453		def build_cmd_line(self, ret, container):
454		if container['command'] is not None:
455		msg = ' {}'.format(container['command'])
456		else:
457		msg = ' {}'.format('_')
458		ret.append(self.curse_add_line(msg, splittable=True))
459
460		return ret
461
462		def msg_curse(self, args=None, max_width: Optional[int] = None) -> List[str]:
463		"""Return the dict to display in the curse interface."""
464		# Init the return message
465		ret = []
466
467		# Only process if stats exist (and non null) and display plugin enable...
468		conditions = [not self.stats, len(self.stats) == 0, self.is_disabled()]
469		if any(conditions):
470		return ret
471
472		# Build the string message
473		# Get the maximum containers name
474		# Max size is configurable. See feature request #1723.
475		name_max_width = self.get_max_of_container_names()
476
477		# Title
478		ret = self.build_tile(ret)
479		# Header
480		ret = self.build_header(ret, name_max_width)
481
482		# Data
483		return self.build_data_line(name_max_width, args)(ret)
484
485		def build_data_line(self, name_max_width, args):
486		def build_for_this_params(ret):
487		build_data_with_params = self.build_container_data(name_max_width, args)
488		return reduce(build_data_with_params, self.stats, ret)
489
490		return build_for_this_params
491
492		def build_container_data(self, name_max_width, args):
493		def build_with_this_params(ret, container):
494		steps = [
495		self.maybe_add_engine_name_or_pod_name,
496		self.build_container_name(name_max_width),
497		self.build_status_name,
498		self.build_uptime_line,
499		self.build_cpu_line,
500		self.build_memory_line,
501		self.build_io_line,
502		self.build_net_line(args),
503		self.build_cmd_line,
504		]
505
506		return reduce(lambda ret, step: step(ret, container), steps, ret)
507
508		return build_with_this_params
509
510		@staticmethod
511		def container_alert(status: str) -> str:
512		"""Analyse the container status.
513		One of created, restarting, running, removing, paused, exited, or dead
514		"""
515		if status == 'running':
516		return 'OK'
517		if status == 'dead':
518		return 'ERROR'
519		if status in ['created', 'restarting', 'exited']:
520		return 'WARNING'
521		return 'INFO'
522
523
524	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...
525		# Make VM sort related to process sort
526		if glances_processes.sort_key == 'memory_percent':
527		sort_by = 'memory_usage'
528		sort_by_secondary = 'cpu_percent'
529		elif glances_processes.sort_key == 'name':
530		sort_by = 'name'
531		sort_by_secondary = 'cpu_percent'
532		else:
533		sort_by = 'cpu_percent'
534		sort_by_secondary = 'memory_usage'
535
536		# Sort docker stats
537		sort_stats_processes(
538		stats,
539		sorted_by=sort_by,
540		sorted_by_secondary=sort_by_secondary,
541		# Reverse for all but name
542		reverse=glances_processes.sort_key != 'name',
543		)
544
545		# Return the main sort key and the sorted stats
546		return sort_by, stats
547

nicolargo / glances

Pull Request — develop (#2919)

PluginModel.build_container_data() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like