1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
|
# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2023 Nicolas Hennion <nicolas@nicolargo.com>
#
# SPDX-License-Identifier: LGPL-3.0-only
#
"""Alert plugin."""
from datetime import datetime
from time import tzname
import pytz
from glances.events import glances_events
# from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel
# {
# "begin": "begin",
# "end": "end",
# "state": "WARNING|CRITICAL",
# "type": "CPU|LOAD|MEM",
# "max": MAX,
# "avg": AVG,
# "min": MIN,
# "sum": SUM,
# "count": COUNT,
# "top": [top3 process list],
# "desc": "Processes description",
# "sort": "top sort key"
# "global": "global alert message"
# }
# 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 = {
'begin': {
'description': 'Begin timestamp of the event',
'unit': 'timestamp',
},
'end': {
'description': 'End timestamp of the event (or -1 if ongoing)',
'unit': 'timestamp',
},
'state': {
'description': 'State of the event (WARNING|CRITICAL)',
'unit': 'string',
},
'type': {
'description': 'Type of the event (CPU|LOAD|MEM)',
'unit': 'string',
},
'max': {
'description': 'Maximum value during the event period',
'unit': 'float',
},
'avg': {
'description': 'Average value during the event period',
'unit': 'float',
},
'min': {
'description': 'Minimum value during the event period',
'unit': 'float',
},
'sum': {
'description': 'Sum of the values during the event period',
'unit': 'float',
},
'count': {
'description': 'Number of values during the event period',
'unit': 'int',
},
'top': {
'description': 'Top 3 processes name during the event period',
'unit': 'list',
},
'desc': {
'description': 'Description of the event',
'unit': 'string',
},
'sort': {
'description': 'Sort key of the top processes',
'unit': 'string',
},
'global': {
'description': 'Global alert message',
'unit': 'string',
}
}
class PluginModel(GlancesPluginModel):
"""Glances alert plugin.
Only for display.
"""
def __init__(self, args=None, config=None):
"""Init the plugin."""
super(PluginModel, self).__init__(
args=args, config=config,
stats_init_value=[],
fields_description=fields_description
)
# We want to display the stat in the curse interface
self.display_curse = True
# Set the message position
self.align = 'bottom'
# Set the maximum number of events to display
if config is not None and (config.has_section('alert') or config.has_section('alerts')):
glances_events.set_max_events(config.get_int_value('alert', 'max_events', default=10))
glances_events.set_min_duration(config.get_int_value('alert', 'min_duration', default=6))
glances_events.set_min_interval(config.get_int_value('alert', 'min_interval', default=6))
def update(self):
"""Nothing to do here. Just return the global glances_log."""
# Set the stats to the glances_events
self.stats = glances_events.get()
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 display plugin enable...
if not self.stats or self.is_disabled():
return ret
# Build the string message
# Header with the global message
if len(self.stats) > 0 and self.stats[0]['end'] < 0 and 'global' in self.stats[0]:
ret.append(self.curse_add_line(self.stats[0]['global'], "TITLE"))
else:
ret.append(self.curse_add_line("ALERTS", "TITLE"))
# Loop over alerts
for alert in self.stats:
# New line
ret.append(self.curse_new_line())
# Start
msg = str(datetime.fromtimestamp(alert['begin'],
tz=pytz.timezone(tzname[0] if tzname[0] else 'UTC')))
ret.append(self.curse_add_line(msg))
# Duration
if alert['end'] > 0:
# If finished display duration
msg = ' ({})'.format(datetime.fromtimestamp(alert['end']) - datetime.fromtimestamp(alert['begin']))
else:
msg = ' (ongoing)'
ret.append(self.curse_add_line(msg))
ret.append(self.curse_add_line(" - "))
# Infos
if alert['end'] > 0:
# If finished do not display status
msg = '{} on {}'.format(alert['state'], alert['type'])
ret.append(self.curse_add_line(msg))
else:
msg = str(alert['type'])
ret.append(self.curse_add_line(msg, decoration=alert['state']))
# Min / Mean / Max
if self.approx_equal(alert['min'], alert['max'], tolerance=0.1):
msg = ' ({:.1f})'.format(alert['avg'])
else:
msg = ' (Min:{:.1f} Mean:{:.1f} Max:{:.1f})'.format(alert['min'],
alert['avg'],
alert['max'])
ret.append(self.curse_add_line(msg))
# Top processes
top_process = ', '.join(alert['top'])
if top_process != '':
msg = ': {}'.format(top_process)
ret.append(self.curse_add_line(msg))
return ret
def approx_equal(self, a, b, tolerance=0.0):
"""Compare a with b using the tolerance (if numerical)."""
if str(int(a)).isdigit() and str(int(b)).isdigit():
return abs(a - b) <= max(abs(a), abs(b)) * tolerance
else:
return a == b
|
