glances.plugins.containers.PluginModel.msg_curse() - Code Metrics - Inspection of "Glances 4.1.2" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 1826f0...f8aa98 )

by Nicolas

created 2024-07-03 17:44 UTC

glances.plugins.containers.PluginModel.msg_curse() F

↳ Parent: glances.plugins.containers

Complexity

Conditions

Size

Total Lines	137
Code Lines	108

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	24
eloc	108
nop	3
dl	0
loc	137
rs	0
c	0
b	0
f	0

How to fix Long Method 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 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"] = 'docker'
            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 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...
        if not self.stats or len(self.stats) == 0 or self.is_disabled():
            return ret

        # Build the string message
        # Title
        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))
        ret.append(self.curse_new_line())
        # Header
        ret.append(self.curse_new_line())
        # Get the maximum containers name
        # Max size is configurable. See feature request #1723.
        name_max_width = 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']),
        )

        if self.views['show_engine_name']:
            msg = ' {:{width}}'.format('Engine', width=6)
            ret.append(self.curse_add_line(msg))
        if self.views['show_pod_name']:
            msg = ' {:{width}}'.format('Pod', width=12)
            ret.append(self.curse_add_line(msg))
        msg = ' {:{width}}'.format('Name', width=name_max_width)
        ret.append(self.curse_add_line(msg, 'SORT' if self.sort_key == 'name' else 'DEFAULT'))
        msg = '{:>10}'.format('Status')
        ret.append(self.curse_add_line(msg))
        msg = '{:>10}'.format('Uptime')
        ret.append(self.curse_add_line(msg))
        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'))
        msg = '/{:<7}'.format('MAX')
        ret.append(self.curse_add_line(msg))
        msg = '{:>7}'.format('IOR/s')
        ret.append(self.curse_add_line(msg))
        msg = ' {:<7}'.format('IOW/s')
        ret.append(self.curse_add_line(msg))
        msg = '{:>7}'.format('Rx/s')
        ret.append(self.curse_add_line(msg))
        msg = ' {:<7}'.format('Tx/s')
        ret.append(self.curse_add_line(msg))
        msg = ' {:8}'.format('Command')
        ret.append(self.curse_add_line(msg))

        # Data
        for container in self.stats:
            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)))
            # Name
            ret.append(
                self.curse_add_line(' {:{width}}'.format(container['name'][:name_max_width], width=name_max_width))
            )
            # Status
            status = self.container_alert(container['status'])
            msg = '{:>10}'.format(container['status'][0:10])
            ret.append(self.curse_add_line(msg, status))
            # Uptime
            if container['uptime']:
                msg = '{:>10}'.format(container['uptime'])
            else:
                msg = '{:>10}'.format('_')
            ret.append(self.curse_add_line(msg))
            # CPU
            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')))
            # MEM
            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))
            # IO R/W
            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))
            # NET RX/TX
            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))
            # Command
            if container['command'] is not None:
                msg = ' {}'.format(container['command'])
            else:
                msg = ' {}'.format('_')
            ret.append(self.curse_add_line(msg, splittable=True))

        return ret

    @staticmethod
    def container_alert(status: str) -> str:
        """Analyse the container status."""
        if status == 'running':
            return 'OK'
        if status == 'exited':
            return 'WARNING'
        if status == 'dead':
            return 'CRITICAL'
        return 'CAREFUL'


def sort_docker_stats(stats: List[Dict[str, Any]]) -> Tuple[str, List[Dict[str, Any]]]:
    # Sort Docker stats using the same function than processes
    sort_by = glances_processes.sort_key
    sort_by_secondary = 'memory_usage'
    if sort_by == 'memory_percent':
        sort_by = 'memory_usage'
        sort_by_secondary = 'cpu_percent'
    elif sort_by in ['username', 'io_counters', 'cpu_times']:
        sort_by = 'cpu_percent'

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

nicolargo / glances

Push — master ( 1826f0...f8aa98 )

glances.plugins.containers.PluginModel.msg_curse() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like