glances.plugins.containers.PluginModel.msg_curse() - Code Metrics - Inspection of "Refactor plugin folder in the same way than export..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 30cc9f...48251c )

by Nicolas

created 2023-10-15 16:05 UTC

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

↳ Parent: glances.plugins.containers

Complexity

Conditions

Size

Total Lines	155
Code Lines	119

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	27
eloc	119
nop	3
dl	0
loc	155
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Containers plugin."""

import os
from copy import deepcopy

from glances.logger import logger
from glances.plugins.containers.engines.docker import DockerContainersExtension, import_docker_error_tag
from glances.plugins.containers.engines.podman import PodmanContainersExtension, 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

# 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(PluginModel, self).__init__(args=args, config=config, items_history_list=items_history_list)

        # 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

        # Init the Docker API
        self.docker_extension = DockerContainersExtension() if not import_docker_error_tag else None

        # Init the Podman API
        if import_podman_error_tag:
            self.podman_client = None
        else:
            self.podman_client = PodmanContainersExtension(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):
        """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"
        else:
            return conf_podman_sock[0]

    def exit(self):
        """Overwrite the exit method to close threads."""
        if self.docker_extension:
            self.docker_extension.stop()
        if self.podman_client:
            self.podman_client.stop()
        # Call the father class
        super(PluginModel, self).exit()

    def get_key(self):
        """Return the key of the list."""
        return 'name'

    def get_export(self):
        """Overwrite the default export method.

        - Only exports containers
        - The key is the first container name
        """
        try:
            ret = deepcopy(self.stats['containers'])
        except KeyError as e:
            logger.debug("docker plugin - Docker export error {}".format(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):
        """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
        else:
            return all_tag[0].lower() == 'true'

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update Docker and podman stats using the input method."""
        # Connection should be ok
        if self.docker_extension is None and self.podman_client is None:
            return self.get_init_value()

        if self.input_method == 'local':
            # Update stats
            stats_docker = self.update_docker() if self.docker_extension else {}
            stats_podman = self.update_podman() if self.podman_client else {}
            stats = {
                'version': stats_docker.get('version', {}),
                'version_podman': stats_podman.get('version', {}),
                'containers': stats_docker.get('containers', []) + stats_podman.get('containers', []),
            }
        elif self.input_method == 'snmp':
            # Update stats using SNMP
            # Not available
            pass

        # 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

    def update_docker(self):
        """Update Docker stats using the input method."""
        version, containers = self.docker_extension.update(all_tag=self._all_tag())
        for container in containers:
            container["engine"] = 'docker'
        return {"version": version, "containers": containers}

    def update_podman(self):
        """Update Podman stats."""
        version, containers = self.podman_client.update(all_tag=self._all_tag())
        for container in containers:
            container["engine"] = 'podman'
        return {"version": version, "containers": containers}

    def get_user_ticks(self):
        """Return the user ticks by reading the environment variable."""
        return os.sysconf(os.sysconf_names['SC_CLK_TCK'])

    def get_stats_action(self):
        """Return stats for the action.

        Docker will return self.stats['containers']
        """
        return self.stats['containers']

    def update_views(self):
        """Update stats views."""
        # Call the father's method
        super(PluginModel, self).update_views()

        if 'containers' not in self.stats:
            return False

        # Add specifics information
        # Alert
        for i in self.stats['containers']:
            # 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(
                    i['memory']['usage'], 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(i['memory']['usage'], maximum=i['memory']['limit'], header='mem')
                self.views[i[self.get_key()]]['mem']['decoration'] = alert

        return True

    def msg_curse(self, args=None, max_width=None):
        """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 'containers' not in self.stats or len(self.stats['containers']) == 0 or self.is_disabled():
            return ret

        show_pod_name = False
        if any(ct.get("pod_name") for ct in self.stats["containers"]):
            show_pod_name = True

        show_engine_name = False
        if len(set(ct["engine"] for ct in self.stats["containers"])) > 1:
            show_engine_name = True

        # Build the string message
        # Title
        msg = '{}'.format('CONTAINERS')
        ret.append(self.curse_add_line(msg, "TITLE"))
        msg = ' {}'.format(len(self.stats['containers']))
        ret.append(self.curse_add_line(msg))
        msg = ' sorted by {}'.format(sort_for_human[self.sort_key])
        ret.append(self.curse_add_line(msg))
        # msg = ' (served by Docker {})'.format(self.stats['version']["Version"])
        # 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['containers'], key=lambda x: len(x['name']))['name']),
        )

        if show_engine_name:
            msg = ' {:{width}}'.format('Engine', width=6)
            ret.append(self.curse_add_line(msg))
        if 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['containers']:
            ret.append(self.curse_new_line())
            if show_engine_name:
                ret.append(self.curse_add_line(' {:{width}}'.format(container["engine"], width=6)))
            if show_pod_name:
                ret.append(self.curse_add_line(' {:{width}}'.format(container.get("pod_id", "-"), width=12)))
            # Name
            ret.append(self.curse_add_line(self._msg_name(container=container, max_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:
                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(container['memory']['usage']))
            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:
                msg = '/{:<7}'.format('_')
            ret.append(self.curse_add_line(msg))
            # IO R/W
            unit = 'B'
            try:
                value = self.auto_unit(int(container['io']['ior'] // container['io']['time_since_update'])) + unit
                msg = '{:>7}'.format(value)
            except KeyError:
                msg = '{:>7}'.format('_')
            ret.append(self.curse_add_line(msg))
            try:
                value = self.auto_unit(int(container['io']['iow'] // container['io']['time_since_update'])) + unit
                msg = ' {:<7}'.format(value)
            except KeyError:
                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'] // container['network']['time_since_update'] * to_bit)
                    )
                    + unit
                )
                msg = '{:>7}'.format(value)
            except KeyError:
                msg = '{:>7}'.format('_')
            ret.append(self.curse_add_line(msg))
            try:
                value = (
                    self.auto_unit(
                        int(container['network']['tx'] // container['network']['time_since_update'] * to_bit)
                    )
                    + unit
                )
                msg = ' {:<7}'.format(value)
            except KeyError:
                msg = ' {:<7}'.format('_')
            ret.append(self.curse_add_line(msg))
            # Command
            if container['Command'] is not None:
                msg = ' {}'.format(' '.join(container['Command']))
            else:
                msg = ' {}'.format('_')
            ret.append(self.curse_add_line(msg, splittable=True))

        return ret

    def _msg_name(self, container, max_width):
        """Build the container name."""
        name = container['name'][:max_width]
        return ' {:{width}}'.format(name, width=max_width)

    def container_alert(self, status):
        """Analyse the container status."""
        if status == 'running':
            return 'OK'
        elif status == 'exited':
            return 'WARNING'
        elif status == 'dead':
            return 'CRITICAL'
        else:
            return 'CAREFUL'


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

nicolargo / glances

Push — develop ( 30cc9f...48251c )

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