glances.outputs.glances_stdout_apidoc.print_processes() - Code Metrics - Inspection of "version 4.3.1" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 372380...7cfc0c )

by Nicolas

created 2025-03-22 16:51 UTC

print_processes() A

↳ Parent: glances.outputs.glances_stdout_apidoc

Complexity

Conditions

Size

Total Lines	19
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	19
nop	0
dl	0
loc	19
rs	9.45
c	0
b	0
f	0

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

"""Fields description interface class."""

import json
import time
from pprint import pformat

from glances import __apiversion__
from glances.globals import iteritems
from glances.logger import logger

API_URL = f"http://localhost:61208/api/{__apiversion__}"

APIDOC_HEADER = f"""\
.. _api:

API (Restfull/JSON) documentation
=================================

This documentation describes the Glances API version {__apiversion__} (Restfull/JSON) interface.

For Glances version 3, please have a look on:
``https://github.com/nicolargo/glances/blob/support/glancesv3/docs/api.rst``

Run the Glances API server
--------------------------

The Glances Restfull/API server could be ran using the following command line:

.. code-block:: bash

    # glances -w --disable-webui

It is also ran automatically when Glances is started in Web server mode (-w).

If you want to enable the Glances Central Browser, use:

.. code-block:: bash

    # glances -w --browser --disable-webui

API URL
-------

The default root API URL is ``http://localhost:61208/api/{__apiversion__}``.

The bind address and port could be changed using the ``--bind`` and ``--port`` command line options.

It is also possible to define an URL prefix using the ``url_prefix`` option from the [outputs] section
of the Glances configuration file.

Note: The url_prefix should always end with a slash (``/``).

For example:

.. code-block:: ini
    [outputs]
    url_prefix = /glances/

will change the root API URL to ``http://localhost:61208/glances/api/{__apiversion__}`` and the Web UI URL to
``http://localhost:61208/glances/``

API documentation URL
---------------------

The API documentation is embeded in the server and available at the following URL:
``http://localhost:61208/docs#/``.

WebUI refresh
-------------

It is possible to change the Web UI refresh rate (default is 2 seconds) using the following option in the URL:
``http://localhost:61208/?refresh=5``

"""


def indent_stat(stat, indent='    '):
    # Indent stats to pretty print it
    if isinstance(stat, list) and len(stat) > 1 and isinstance(stat[0], dict):
        # Only display two first items
        return indent + pformat(stat[0:2]).replace('\n', '\n' + indent).replace("'", '"')
    return indent + pformat(stat).replace('\n', '\n' + indent).replace("'", '"')


def print_api_status():
    sub_title = 'GET API status'
    print(sub_title)
    print('-' * len(sub_title))
    print('')
    print('This entry point should be used to check the API status.')
    print('It will the Glances version and a 200 return code if everything is OK.')
    print('')
    print('Get the Rest API status::')
    print('')
    print(f'    # curl -I {API_URL}/status')
    print(indent_stat('HTTP/1.0 200 OK'))
    print('')


def print_plugins_list(stat):
    sub_title = 'GET plugins list'
    print(sub_title)
    print('-' * len(sub_title))
    print('')
    print('Get the plugins list::')
    print('')
    print(f'    # curl {API_URL}/pluginslist')
    print(indent_stat(stat))
    print('')


def print_plugin_stats(plugin, stat):
    sub_title = f'GET {plugin}'
    print(sub_title)
    print('-' * len(sub_title))
    print('')

    print('Get plugin stats::')
    print('')
    print(f'    # curl {API_URL}/{plugin}')
    print(indent_stat(json.loads(stat.get_stats())))
    print('')


def print_plugin_description(plugin, stat):
    if stat.fields_description:
        # For each plugins with a description
        print('Fields descriptions:')
        print('')
        time_since_update = False
        for field, description in iteritems(stat.fields_description):
            print(
                '* **{}**: {} (unit is *{}*)'.format(
                    field,
                    (
                        description['description'][:-1]
                        if description['description'].endswith('.')
                        else description['description']
                    ),
                    description['unit'] if 'unit' in description else 'None',
                )
            )
            if 'rate' in description and description['rate']:
                time_since_update = True
                print(
                    '* **{}**: {} (unit is *{}* per second)'.format(
                        field + '_rate_per_sec',
                        (
                            description['description'][:-1]
                            if description['description'].endswith('.')
                            else description['description']
                        )
                        + ' per second',
                        description['unit'] if 'unit' in description else 'None',
                    )
                )
                print(
                    '* **{}**: {} (unit is *{}*)'.format(
                        field + '_gauge',
                        (
                            description['description'][:-1]
                            if description['description'].endswith('.')
                            else description['description']
                        )
                        + ' (cumulative)',
                        description['unit'] if 'unit' in description else 'None',
                    )
                )

        if time_since_update:
            print(
                '* **{}**: {} (unit is *{}*)'.format(
                    'time_since_update', 'Number of seconds since last update', 'seconds'
                )
            )

        print('')
    else:
        logger.error(f'No fields_description variable defined for plugin {plugin}')


def print_plugin_item_value(plugin, stat, stat_export):
    item = None
    value = None
    if isinstance(stat_export, dict):
        item = list(stat_export.keys())[0]
        value = None
    elif isinstance(stat_export, list) and len(stat_export) > 0 and isinstance(stat_export[0], dict):
        if 'key' in stat_export[0]:
            item = stat_export[0]['key']
        else:
            item = list(stat_export[0].keys())[0]
    if item and stat.get_stats_item(item):
        stat_item = json.loads(stat.get_stats_item(item))
        if isinstance(stat_item[item], list):
            value = stat_item[item][0]
        else:
            value = stat_item[item]
        print('Get a specific field::')
        print('')
        print(f'    # curl {API_URL}/{plugin}/{item}')
        print(indent_stat(stat_item))
        print('')
    if item and value and stat.get_stats_value(item, value):
        print('Get a specific item when field matches the given value::')
        print('')
        print(f'    # curl {API_URL}/{plugin}/{item}/value/{value}')
        print(indent_stat(json.loads(stat.get_stats_value(item, value))))
        print('')


def print_all():
    sub_title = 'GET all stats'
    print(sub_title)
    print('-' * len(sub_title))
    print('')
    print('Get all Glances stats::')
    print('')
    print(f'    # curl {API_URL}/all')
    print('    Return a very big dictionary with all stats')
    print('')
    print('Note: Update is done automatically every time /all or /<plugin> is called.')
    print('')


def print_processes():
    sub_title = 'GET stats of a specific process'
    print(sub_title)
    print('-' * len(sub_title))
    print('')
    print('Get stats for process with PID == 777::')
    print('')
    print(f'    # curl {API_URL}/processes/777')
    print('    Return stats for process (dict)')
    print('')
    print('Enable extended stats for process with PID == 777 (only one process at a time can be enabled)::')
    print('')
    print(f'    # curl -X POST {API_URL}/processes/extended/777')
    print(f'    # curl {API_URL}/all')
    print(f'    # curl {API_URL}/processes/777')
    print('    Return stats for process (dict)')
    print('')
    print('Note: Update *is not* done automatically when you call /processes/<pid>.')
    print('')


def print_top(stats):
    time.sleep(1)
    stats.update()
    sub_title = 'GET top n items of a specific plugin'
    print(sub_title)
    print('-' * len(sub_title))
    print('')
    print('Get top 2 processes of the processlist plugin::')
    print('')
    print(f'    # curl {API_URL}/processlist/top/2')
    print(indent_stat(stats.get_plugin('processlist').get_export()[:2]))
    print('')
    print('Note: Only work for plugin with a list of items')
    print('')


def print_fields_info(stats):
    sub_title = 'GET item description'
    print(sub_title)
    print('-' * len(sub_title))
    print('Get item description (human readable) for a specific plugin/item::')
    print('')
    print(f'    # curl {API_URL}/diskio/read_bytes/description')
    print(indent_stat(stats.get_plugin('diskio').get_item_info('read_bytes', 'description')))
    print('')
    print('Note: the description is defined in the fields_description variable of the plugin.')
    print('')
    sub_title = 'GET item unit'
    print(sub_title)
    print('-' * len(sub_title))
    print('Get item unit for a specific plugin/item::')
    print('')
    print(f'    # curl {API_URL}/diskio/read_bytes/unit')
    print(indent_stat(stats.get_plugin('diskio').get_item_info('read_bytes', 'unit')))
    print('')
    print('Note: the description is defined in the fields_description variable of the plugin.')
    print('')


def print_history(stats):
    time.sleep(1)
    stats.update()
    time.sleep(1)
    stats.update()
    sub_title = 'GET stats history'
    print(sub_title)
    print('-' * len(sub_title))
    print('')
    print('History of a plugin::')
    print('')
    print(f'    # curl {API_URL}/cpu/history')
    print(indent_stat(json.loads(stats.get_plugin('cpu').get_stats_history(nb=3))))
    print('')
    print('Limit history to last 2 values::')
    print('')
    print(f'    # curl {API_URL}/cpu/history/2')
    print(indent_stat(json.loads(stats.get_plugin('cpu').get_stats_history(nb=2))))
    print('')
    print('History for a specific field::')
    print('')
    print(f'    # curl {API_URL}/cpu/system/history')
    print(indent_stat(json.loads(stats.get_plugin('cpu').get_stats_history('system'))))
    print('')
    print('Limit history for a specific field to last 2 values::')
    print('')
    print(f'    # curl {API_URL}/cpu/system/history')
    print(indent_stat(json.loads(stats.get_plugin('cpu').get_stats_history('system', nb=2))))
    print('')


def print_limits(stats):
    sub_title = 'GET limits (used for thresholds)'
    print(sub_title)
    print('-' * len(sub_title))
    print('')
    print('All limits/thresholds::')
    print('')
    print(f'    # curl {API_URL}/all/limits')
    print(indent_stat(stats.getAllLimitsAsDict()))
    print('')
    print('Limits/thresholds for the cpu plugin::')
    print('')
    print(f'    # curl {API_URL}/cpu/limits')
    print(indent_stat(stats.get_plugin('cpu').limits))
    print('')


def print_plugin_post_events():
    sub_title = 'POST clear events'
    print(sub_title)
    print('-' * len(sub_title))
    print('')
    print('Clear all alarms from the list::')
    print('')
    print(f'    # curl -H "Content-Type: application/json" -X POST {API_URL}/events/clear/all')
    print('')
    print('Clear warning alarms from the list::')
    print('')
    print(f'    # curl -H "Content-Type: application/json" -X POST {API_URL}/events/clear/warning')
    print('')


class GlancesStdoutApiDoc:
    """This class manages the fields description display."""

    def __init__(self, config=None, args=None):
        # Init
        self.config = config
        self.args = args

    def end(self):
        pass

    def update(self, stats, duration=1):
        """Display issue"""

        # Display header
        print(APIDOC_HEADER)

        # Display API status
        print_api_status()

        # Display plugins list
        print_plugins_list(sorted(stats._plugins))

        # Loop over plugins
        for plugin in sorted(stats._plugins):
            stat = stats.get_plugin(plugin)
            print_plugin_stats(plugin, stat)
            print_plugin_description(plugin, stat)
            if plugin == 'alert':
                print_plugin_post_events()

            stat_export = stat.get_export()
            if stat_export is None or stat_export == [] or stat_export == {}:
                continue
            print_plugin_item_value(plugin, stat, stat_export)

        # Get all stats
        print_all()

        # Get process stats
        print_processes()

        # Get top stats (only for plugins with a list of items)
        # Example for processlist plugin: get top 2 processes
        print_top(stats)

        # Fields description
        print_fields_info(stats)

        # History
        print_history(stats)

        # Limits
        print_limits(stats)

        # Return True to exit directly (no refresh)
        return True


1			#
2			# This file is part of Glances.
3			#
4			# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
5			#
6			# SPDX-License-Identifier: LGPL-3.0-only
7			#
8
9			"""Fields description interface class."""
10
11			import json
12			import time
13			from pprint import pformat
14
15			from glances import __apiversion__
16			from glances.globals import iteritems
17			from glances.logger import logger
18
19			API_URL = f"http://localhost:61208/api/{__apiversion__}"
20
21			APIDOC_HEADER = f"""\
22			.. _api:
23
24			API (Restfull/JSON) documentation
25			=================================
26
27			This documentation describes the Glances API version {__apiversion__} (Restfull/JSON) interface.
28
29			For Glances version 3, please have a look on:
30			``https://github.com/nicolargo/glances/blob/support/glancesv3/docs/api.rst``
31
32			Run the Glances API server
33			--------------------------
34
35			The Glances Restfull/API server could be ran using the following command line:
36
37			.. code-block:: bash
38
39			# glances -w --disable-webui
40
41			It is also ran automatically when Glances is started in Web server mode (-w).
42
43			If you want to enable the Glances Central Browser, use:
44
45			.. code-block:: bash
46
47			# glances -w --browser --disable-webui
48
49			API URL
50			-------
51
52			The default root API URL is ``http://localhost:61208/api/{__apiversion__}``.
53
54			The bind address and port could be changed using the ``--bind`` and ``--port`` command line options.
55
56			It is also possible to define an URL prefix using the ``url_prefix`` option from the [outputs] section
57			of the Glances configuration file.
58
59			Note: The url_prefix should always end with a slash (``/``).
60
61			For example:
62
63			.. code-block:: ini
64			[outputs]
65			url_prefix = /glances/
66
67			will change the root API URL to ``http://localhost:61208/glances/api/{__apiversion__}`` and the Web UI URL to
68			``http://localhost:61208/glances/``
69
70			API documentation URL
71			---------------------
72
73			The API documentation is embeded in the server and available at the following URL:
74			``http://localhost:61208/docs#/``.
75
76			WebUI refresh
77			-------------
78
79			It is possible to change the Web UI refresh rate (default is 2 seconds) using the following option in the URL:
80			``http://localhost:61208/?refresh=5``
81
82			"""
83
84
85			def indent_stat(stat, indent=' '):
86			# Indent stats to pretty print it
87			if isinstance(stat, list) and len(stat) > 1 and isinstance(stat[0], dict):
88			# Only display two first items
89			return indent + pformat(stat[0:2]).replace('\n', '\n' + indent).replace("'", '"')
90			return indent + pformat(stat).replace('\n', '\n' + indent).replace("'", '"')
91
92
93			def print_api_status():
94			sub_title = 'GET API status'
95			print(sub_title)
96			print('-' * len(sub_title))
97			print('')
98			print('This entry point should be used to check the API status.')
99			print('It will the Glances version and a 200 return code if everything is OK.')
100			print('')
101			print('Get the Rest API status::')
102			print('')
103			print(f' # curl -I {API_URL}/status')
104			print(indent_stat('HTTP/1.0 200 OK'))
105			print('')
106
107
108			def print_plugins_list(stat):
109			sub_title = 'GET plugins list'
110			print(sub_title)
111			print('-' * len(sub_title))
112			print('')
113			print('Get the plugins list::')
114			print('')
115			print(f' # curl {API_URL}/pluginslist')
116			print(indent_stat(stat))
117			print('')
118
119
120			def print_plugin_stats(plugin, stat):
121			sub_title = f'GET {plugin}'
122			print(sub_title)
123			print('-' * len(sub_title))
124			print('')
125
126			print('Get plugin stats::')
127			print('')
128			print(f' # curl {API_URL}/{plugin}')
129			print(indent_stat(json.loads(stat.get_stats())))
130			print('')
131
132
133			def print_plugin_description(plugin, stat):
134			if stat.fields_description:
135			# For each plugins with a description
136			print('Fields descriptions:')
137			print('')
138			time_since_update = False
139			for field, description in iteritems(stat.fields_description):
140			print(
141			'* {}: {} (unit is {})'.format(
142			field,
143			(
144			description['description'][:-1]
145			if description['description'].endswith('.')
146			else description['description']
147			),
148			description['unit'] if 'unit' in description else 'None',
149			)
150			)
151			if 'rate' in description and description['rate']:
152			time_since_update = True
153			print(
154			'* {}: {} (unit is {} per second)'.format(
155			field + '_rate_per_sec',
156			(
157			description['description'][:-1]
158			if description['description'].endswith('.')
159			else description['description']
160			)
161			+ ' per second',
162			description['unit'] if 'unit' in description else 'None',
163			)
164			)
165			print(
166			'* {}: {} (unit is {})'.format(
167			field + '_gauge',
168			(
169			description['description'][:-1]
170			if description['description'].endswith('.')
171			else description['description']
172			)
173			+ ' (cumulative)',
174			description['unit'] if 'unit' in description else 'None',
175			)
176			)
177
178			if time_since_update:
179			print(
180			'* {}: {} (unit is {})'.format(
181			'time_since_update', 'Number of seconds since last update', 'seconds'
182			)
183			)
184
185			print('')
186			else:
187			logger.error(f'No fields_description variable defined for plugin {plugin}')
188
189
190			def print_plugin_item_value(plugin, stat, stat_export):
191			item = None
192			value = None
193			if isinstance(stat_export, dict):
194			item = list(stat_export.keys())[0]
195			value = None
196			elif isinstance(stat_export, list) and len(stat_export) > 0 and isinstance(stat_export[0], dict):
197			if 'key' in stat_export[0]:
198			item = stat_export[0]['key']
199			else:
200			item = list(stat_export[0].keys())[0]
201			if item and stat.get_stats_item(item):
202			stat_item = json.loads(stat.get_stats_item(item))
203			if isinstance(stat_item[item], list):
204			value = stat_item[item][0]
205			else:
206			value = stat_item[item]
207			print('Get a specific field::')
208			print('')
209			print(f' # curl {API_URL}/{plugin}/{item}')
210			print(indent_stat(stat_item))
211			print('')
212			if item and value and stat.get_stats_value(item, value):
213			print('Get a specific item when field matches the given value::')
214			print('')
215			print(f' # curl {API_URL}/{plugin}/{item}/value/{value}')
216			print(indent_stat(json.loads(stat.get_stats_value(item, value))))
217			print('')
218
219
220			def print_all():
221			sub_title = 'GET all stats'
222			print(sub_title)
223			print('-' * len(sub_title))
224			print('')
225			print('Get all Glances stats::')
226			print('')
227			print(f' # curl {API_URL}/all')
228			print(' Return a very big dictionary with all stats')
229			print('')
230			print('Note: Update is done automatically every time /all or /<plugin> is called.')
231			print('')
232
233
234			def print_processes():
235			sub_title = 'GET stats of a specific process'
236			print(sub_title)
237			print('-' * len(sub_title))
238			print('')
239			print('Get stats for process with PID == 777::')
240			print('')
241			print(f' # curl {API_URL}/processes/777')
242			print(' Return stats for process (dict)')
243			print('')
244			print('Enable extended stats for process with PID == 777 (only one process at a time can be enabled)::')
245			print('')
246			print(f' # curl -X POST {API_URL}/processes/extended/777')
247			print(f' # curl {API_URL}/all')
248			print(f' # curl {API_URL}/processes/777')
249			print(' Return stats for process (dict)')
250			print('')
251			print('Note: Update is not done automatically when you call /processes/<pid>.')
252			print('')
253
254
255			def print_top(stats):
256			time.sleep(1)
257			stats.update()
258			sub_title = 'GET top n items of a specific plugin'
259			print(sub_title)
260			print('-' * len(sub_title))
261			print('')
262			print('Get top 2 processes of the processlist plugin::')
263			print('')
264			print(f' # curl {API_URL}/processlist/top/2')
265			print(indent_stat(stats.get_plugin('processlist').get_export()[:2]))
266			print('')
267			print('Note: Only work for plugin with a list of items')
268			print('')
269
270
271			def print_fields_info(stats):
272			sub_title = 'GET item description'
273			print(sub_title)
274			print('-' * len(sub_title))
275			print('Get item description (human readable) for a specific plugin/item::')
276			print('')
277			print(f' # curl {API_URL}/diskio/read_bytes/description')
278			print(indent_stat(stats.get_plugin('diskio').get_item_info('read_bytes', 'description')))
279			print('')
280			print('Note: the description is defined in the fields_description variable of the plugin.')
281			print('')
282			sub_title = 'GET item unit'
283			print(sub_title)
284			print('-' * len(sub_title))
285			print('Get item unit for a specific plugin/item::')
286			print('')
287			print(f' # curl {API_URL}/diskio/read_bytes/unit')
288			print(indent_stat(stats.get_plugin('diskio').get_item_info('read_bytes', 'unit')))
289			print('')
290			print('Note: the description is defined in the fields_description variable of the plugin.')
291			print('')
292
293
294			def print_history(stats):
295			time.sleep(1)
296			stats.update()
297			time.sleep(1)
298			stats.update()
299			sub_title = 'GET stats history'
300			print(sub_title)
301			print('-' * len(sub_title))
302			print('')
303			print('History of a plugin::')
304			print('')
305			print(f' # curl {API_URL}/cpu/history')
306			print(indent_stat(json.loads(stats.get_plugin('cpu').get_stats_history(nb=3))))
307			print('')
308			print('Limit history to last 2 values::')
309			print('')
310			print(f' # curl {API_URL}/cpu/history/2')
311			print(indent_stat(json.loads(stats.get_plugin('cpu').get_stats_history(nb=2))))
312			print('')
313			print('History for a specific field::')
314			print('')
315			print(f' # curl {API_URL}/cpu/system/history')
316			print(indent_stat(json.loads(stats.get_plugin('cpu').get_stats_history('system'))))
317			print('')
318			print('Limit history for a specific field to last 2 values::')
319			print('')
320			print(f' # curl {API_URL}/cpu/system/history')
321			print(indent_stat(json.loads(stats.get_plugin('cpu').get_stats_history('system', nb=2))))
322			print('')
323
324
325			def print_limits(stats):
326			sub_title = 'GET limits (used for thresholds)'
327			print(sub_title)
328			print('-' * len(sub_title))
329			print('')
330			print('All limits/thresholds::')
331			print('')
332			print(f' # curl {API_URL}/all/limits')
333			print(indent_stat(stats.getAllLimitsAsDict()))
334			print('')
335			print('Limits/thresholds for the cpu plugin::')
336			print('')
337			print(f' # curl {API_URL}/cpu/limits')
338			print(indent_stat(stats.get_plugin('cpu').limits))
339			print('')
340
341
342			def print_plugin_post_events():
343			sub_title = 'POST clear events'
344			print(sub_title)
345			print('-' * len(sub_title))
346			print('')
347			print('Clear all alarms from the list::')
348			print('')
349			print(f' # curl -H "Content-Type: application/json" -X POST {API_URL}/events/clear/all')
350			print('')
351			print('Clear warning alarms from the list::')
352			print('')
353			print(f' # curl -H "Content-Type: application/json" -X POST {API_URL}/events/clear/warning')
354			print('')
355
356
357			class GlancesStdoutApiDoc:
358			"""This class manages the fields description display."""
359
360			def __init__(self, config=None, args=None):
361			# Init
362			self.config = config
363			self.args = args
364
365			def end(self):
366			pass
367
368			def update(self, stats, duration=1):
369			"""Display issue"""
370
371			# Display header
372			print(APIDOC_HEADER)
373
374			# Display API status
375			print_api_status()
376
377			# Display plugins list
378			print_plugins_list(sorted(stats._plugins))
379
380			# Loop over plugins
381			for plugin in sorted(stats._plugins):
382			stat = stats.get_plugin(plugin)
383			print_plugin_stats(plugin, stat)
384			print_plugin_description(plugin, stat)
385			if plugin == 'alert':
386			print_plugin_post_events()
387
388			stat_export = stat.get_export()
389			if stat_export is None or stat_export == [] or stat_export == {}:
390			continue
391			print_plugin_item_value(plugin, stat, stat_export)
392
393			# Get all stats
394			print_all()
395
396			# Get process stats
397			print_processes()
398
399			# Get top stats (only for plugins with a list of items)
400			# Example for processlist plugin: get top 2 processes
401			print_top(stats)
402
403			# Fields description
404			print_fields_info(stats)
405
406			# History
407			print_history(stats)
408
409			# Limits
410			print_limits(stats)
411
412			# Return True to exit directly (no refresh)
413			return True
414

nicolargo / glances

Push — master ( 372380...7cfc0c )

print_processes() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like