improvements to anomalies collector following dogfooding (#11003)
* improvements to anomalies collector following dogfooding - add check for python 3. - add ability to reinitialize models and data regulary if needed (off by default). - change check() to be less error prone when running on a parent node (direct learning from dogfooding on netdata parent). - add some model initialization functions to enable reinitialization if model gets into a "bad state". - add some logic to check for models that have not been trained and reinitialize and train them as needed. - add logic to validate_charts() to maintain dimensions by adding and removing as needed. * add newline at end of file * rename `verify` param to `tls_verify` for clarity
Andrew Maguire committed
Apr 21, 2021 at 23:47 UTC
9bcac122c4ef0b1f0b41eebae4024f58ca1bcd4a
3 files changed
+120
-29
collectors/python.d.plugin/anomalies/README.md
+16
-5
@@ -35,18 +35,26 @@ Then, as the issue passes, the anomaly probabilities should settle back down int
35
## Requirements
36
37
- This collector will only work with Python 3 and requires the packages below be installed.
38
+- Typically you will not need to do this, but, if needed, to ensure Python 3 is used you can add the below line to the `[plugin:python.d]` section of `netdata.conf`
39
+
40
+```conf
41
+[plugin:python.d]
42
+ # update every = 1
43
+ command options = -ppython3
44
+```
45
+
46
+Install the required python libraries.
47
48
```bash
49
# become netdata user
50
sudo su -s /bin/bash netdata
51
# install required packages for the netdata user
43
-pip3 install --user netdata-pandas==0.0.32 numba==0.50.1 scikit-learn==0.23.2 pyod==0.8.3
52
+pip3 install --user netdata-pandas==0.0.38 numba==0.50.1 scikit-learn==0.23.2 pyod==0.8.3
53
```
54
55
## Configuration
56
48
-Install the Python requirements above, enable the collector and [restart
49
-Netdata](/docs/configure/start-stop-restart.md).
57
+Install the Python requirements above, enable the collector and restart Netdata.
58
59
```bash
60
cd /etc/netdata/
@@ -69,7 +77,7 @@ sudo ./edit-config python.d/anomalies.conf
77
78
The default configuration should look something like this. Here you can see each parameter (with sane defaults) and some information about each one and what it does.
79
72
-```yaml
80
+```conf
81
# ----------------------------------------------------------------------
82
# JOBS (data collection sources)
83
@@ -87,6 +95,9 @@ local:
95
# Use http or https to pull data
96
protocol: 'http'
97
98
+ # SSL verify parameter for requests.get() calls
99
+ tls_verify: true
100
+
101
# What charts to pull data for - A regex like 'system\..*|' or 'system\..*|apps.cpu|apps.mem' etc.
102
charts_regex: 'system\..*'
103
@@ -229,4 +240,4 @@ If you would like to go deeper on what exactly the anomalies collector is doing
240
- Good [blog post](https://www.anodot.com/blog/what-is-anomaly-detection/) from Anodot on time series anomaly detection. Anodot also have some great whitepapers in this space too that some may find useful.
241
- Novelty and outlier detection in the [scikit-learn documentation](https://scikit-learn.org/stable/modules/outlier_detection.html).
242
232
-[]()
243
+[]()
\ No newline at end of file
collectors/python.d.plugin/anomalies/anomalies.chart.py
+101
-24
@@ -3,6 +3,7 @@
3
# Author: andrewm4894
4
# SPDX-License-Identifier: GPL-3.0-or-later
5
6
+import sys
7
import time
8
from datetime import datetime
9
import re
@@ -51,14 +52,17 @@ class Service(SimpleService):
52
self.basic_init()
53
self.charts_init()
54
self.custom_models_init()
55
+ self.data_init()
56
self.model_params_init()
57
self.models_init()
58
+ self.collected_dims = {'probability': set(), 'anomaly': set()}
59
60
def check(self):
58
- _ = get_allmetrics_async(
59
- host_charts_dict=self.host_charts_dict, host_prefix=True, host_sep='::', wide=True, sort_cols=True,
60
- protocol=self.protocol, numeric_only=True, float_size='float32', user=self.username, pwd=self.password
61
- )
61
+ python_version = float('{}.{}'.format(sys.version_info[0], sys.version_info[1]))
62
+ if python_version < 3.6:
63
+ self.error("anomalies collector only works with Python>=3.6")
64
+ if len(self.host_charts_dict[self.host]) > 0:
65
+ _ = get_allmetrics_async(host_charts_dict=self.host_charts_dict, protocol=self.protocol, user=self.username, pwd=self.password)
66
return True
67
68
def basic_init(self):
@@ -70,17 +74,18 @@ class Service(SimpleService):
74
self.host = self.configuration.get('host', '127.0.0.1:19999')
75
self.username = self.configuration.get('username', None)
76
self.password = self.configuration.get('password', None)
77
+ self.tls_verify = self.configuration.get('tls_verify', True)
78
self.fitted_at = {}
79
self.df_allmetrics = pd.DataFrame()
75
- self.data_latest = {}
80
self.last_train_at = 0
81
self.include_average_prob = bool(self.configuration.get('include_average_prob', True))
82
+ self.reinitialize_at_every_step = bool(self.configuration.get('reinitialize_at_every_step', False))
83
84
def charts_init(self):
85
"""Do some initialisation of charts in scope related variables.
86
"""
87
self.charts_regex = re.compile(self.configuration.get('charts_regex','None'))
83
- self.charts_available = [c for c in list(requests.get(f'{self.protocol}://{self.host}/api/v1/charts').json().get('charts', {}).keys())]
88
+ self.charts_available = [c for c in list(requests.get(f'{self.protocol}://{self.host}/api/v1/charts', verify=self.tls_verify).json().get('charts', {}).keys())]
89
self.charts_in_scope = list(filter(self.charts_regex.match, self.charts_available))
90
self.charts_to_exclude = self.configuration.get('charts_to_exclude', '').split(',')
91
if len(self.charts_to_exclude) > 0:
@@ -115,6 +120,14 @@ class Service(SimpleService):
120
self.models_in_scope = [f'{self.host}::{c}' for c in self.charts_in_scope]
121
self.host_charts_dict = {self.host: self.charts_in_scope}
122
self.model_display_names = {model: model.split('::')[1] if '::' in model else model for model in self.models_in_scope}
123
+ #self.info(f'self.host_charts_dict (len={len(self.host_charts_dict[self.host])}): {self.host_charts_dict}')
124
+
125
+ def data_init(self):
126
+ """Initialize some empty data objects.
127
+ """
128
+ self.data_probability_latest = {f'{m}_prob': 0 for m in self.charts_in_scope}
129
+ self.data_anomaly_latest = {f'{m}_anomaly': 0 for m in self.charts_in_scope}
130
+ self.data_latest = {**self.data_probability_latest, **self.data_anomaly_latest}
131
132
def model_params_init(self):
133
"""Model parameters initialisation.
@@ -153,12 +166,55 @@ class Service(SimpleService):
166
self.models = {model: HBOS(contamination=self.contamination) for model in self.models_in_scope}
167
self.custom_model_scalers = {model: MinMaxScaler() for model in self.models_in_scope}
168
156
- def validate_charts(self, name, data, algorithm='absolute', multiplier=1, divisor=1):
169
+ def model_init(self, model):
170
+ """Model initialisation of a single model.
171
+ """
172
+ if self.model == 'pca':
173
+ self.models[model] = PCA(contamination=self.contamination)
174
+ elif self.model == 'loda':
175
+ self.models[model] = LODA(contamination=self.contamination)
176
+ elif self.model == 'iforest':
177
+ self.models[model] = IForest(n_estimators=50, bootstrap=True, behaviour='new', contamination=self.contamination)
178
+ elif self.model == 'cblof':
179
+ self.models[model] = CBLOF(n_clusters=3, contamination=self.contamination)
180
+ elif self.model == 'feature_bagging':
181
+ self.models[model] = FeatureBagging(base_estimator=PCA(contamination=self.contamination), contamination=self.contamination)
182
+ elif self.model == 'copod':
183
+ self.models[model] = COPOD(contamination=self.contamination)
184
+ elif self.model == 'hbos':
185
+ self.models[model] = HBOS(contamination=self.contamination)
186
+ else:
187
+ self.models[model] = HBOS(contamination=self.contamination)
188
+ self.custom_model_scalers[model] = MinMaxScaler()
189
+
190
+ def reinitialize(self):
191
+ """Reinitialize charts, models and data to a begining state.
192
+ """
193
+ self.charts_init()
194
+ self.custom_models_init()
195
+ self.data_init()
196
+ self.model_params_init()
197
+ self.models_init()
198
+
199
+ def save_data_latest(self, data, data_probability, data_anomaly):
200
+ """Save the most recent data objects to be used if needed in the future.
201
+ """
202
+ self.data_latest = data
203
+ self.data_probability_latest = data_probability
204
+ self.data_anomaly_latest = data_anomaly
205
+
206
+ def validate_charts(self, chart, data, algorithm='absolute', multiplier=1, divisor=1):
207
"""If dimension not in chart then add it.
208
"""
209
for dim in data:
160
- if dim not in self.charts[name]:
161
- self.charts[name].add_dimension([dim, dim, algorithm, multiplier, divisor])
210
+ if dim not in self.collected_dims[chart]:
211
+ self.collected_dims[chart].add(dim)
212
+ self.charts[chart].add_dimension([dim, dim, algorithm, multiplier, divisor])
213
+
214
+ for dim in list(self.collected_dims[chart]):
215
+ if dim not in data:
216
+ self.collected_dims[chart].remove(dim)
217
+ self.charts[chart].del_dimension(dim, hide=False)
218
219
def add_custom_models_dims(self, df):
220
"""Given a df, select columns used by custom models, add custom model name as prefix, and append to df.
@@ -242,8 +298,9 @@ class Service(SimpleService):
298
# get training data
299
df_train = get_data(
300
host_charts_dict=self.host_charts_dict, host_prefix=True, host_sep='::', after=after, before=before,
245
- sort_cols=True, numeric_only=True, protocol=self.protocol, float_size='float32', user=self.username, pwd=self.password
246
- ).ffill()
301
+ sort_cols=True, numeric_only=True, protocol=self.protocol, float_size='float32', user=self.username, pwd=self.password,
302
+ verify=self.tls_verify
303
+ ).ffill()
304
if self.custom_models:
305
df_train = self.add_custom_models_dims(df_train)
306
@@ -262,6 +319,8 @@ class Service(SimpleService):
319
models_to_train = list(self.models.keys())
320
self.n_fit_fail, self.n_fit_success = 0, 0
321
for model in models_to_train:
322
+ if model not in self.models:
323
+ self.model_init(model)
324
X_train = self.make_features(
325
df_train[df_train.columns[df_train.columns.str.startswith(f'{model}|')]].values,
326
train=True, model=model)
@@ -303,13 +362,16 @@ class Service(SimpleService):
362
data_probability, data_anomaly = {}, {}
363
for model in self.fitted_at.keys():
364
model_display_name = self.model_display_names[model]
306
- X_model = np.nan_to_num(self.make_features(
307
- self.df_allmetrics[self.df_allmetrics.columns[self.df_allmetrics.columns.str.startswith(f'{model}|')]].values,
308
- model=model)[-1,:].reshape(1, -1))
365
try:
366
+ X_model = np.nan_to_num(
367
+ self.make_features(
368
+ self.df_allmetrics[self.df_allmetrics.columns[self.df_allmetrics.columns.str.startswith(f'{model}|')]].values,
369
+ model=model
370
+ )[-1,:].reshape(1, -1)
371
+ )
372
data_probability[model_display_name + '_prob'] = np.nan_to_num(self.models[model].predict_proba(X_model)[-1][1]) * 10000
373
data_anomaly[model_display_name + '_anomaly'] = self.models[model].predict(X_model)[-1]
312
- except Exception:
374
+ except Exception as _:
375
#self.info(e)
376
if model_display_name + '_prob' in self.data_latest:
377
#self.info(f'prediction failed for {model} at run_counter {self.runs_counter}, using last prediction instead.')
@@ -323,27 +385,42 @@ class Service(SimpleService):
385
386
def get_data(self):
387
388
+ # initialize to whats available right now
389
+ if self.reinitialize_at_every_step or len(self.host_charts_dict[self.host]) == 0:
390
+ self.charts_init()
391
+ self.custom_models_init()
392
+ self.model_params_init()
393
+
394
# if not all models have been trained then train those we need to
327
- if len(self.fitted_at) < len(self.models):
395
+ if len(self.fitted_at) < len(self.models_in_scope):
396
self.train(
329
- models_to_train=[m for m in self.models if m not in self.fitted_at],
397
+ models_to_train=[m for m in self.models_in_scope if m not in self.fitted_at],
398
train_data_after=self.initial_train_data_after,
331
- train_data_before=self.initial_train_data_before)
399
+ train_data_before=self.initial_train_data_before
400
+ )
401
# retrain all models as per schedule from config
402
elif self.train_every_n > 0 and self.runs_counter % self.train_every_n == 0:
403
+ self.reinitialize()
404
self.train()
405
406
# roll forward previous predictions around a training step to avoid the possibility of having the training itself trigger an anomaly
407
if (self.runs_counter - self.last_train_at) <= self.train_no_prediction_n:
338
- data = self.data_latest
408
+ data_probability = self.data_probability_latest
409
+ data_anomaly = self.data_anomaly_latest
410
else:
411
data_probability, data_anomaly = self.predict()
412
if self.include_average_prob:
342
- data_probability['average_prob'] = np.mean(list(data_probability.values()))
343
- data = {**data_probability, **data_anomaly}
344
- self.validate_charts('probability', data_probability, divisor=100)
345
- self.validate_charts('anomaly', data_anomaly)
413
+ average_prob = np.mean(list(data_probability.values()))
414
+ data_probability['average_prob'] = 0 if np.isnan(average_prob) else average_prob
415
+
416
+ data = {**data_probability, **data_anomaly}
417
347
- self.data_latest = data
418
+ self.validate_charts('probability', data_probability, divisor=100)
419
+ self.validate_charts('anomaly', data_anomaly)
420
+
421
+ self.save_data_latest(data, data_probability, data_anomaly)
422
+
423
+ #self.info(f'len(data)={len(data)}')
424
+ #self.info(f'data')
425
426
return data
collectors/python.d.plugin/anomalies/anomalies.conf
+3
@@ -44,6 +44,9 @@ local:
44
# Use http or https to pull data
45
protocol: 'http'
46
47
+ # SSL verify parameter for requests.get() calls
48
+ tls_verify: true
49
+
50
# What charts to pull data for - A regex like 'system\..*|' or 'system\..*|apps.cpu|apps.mem' etc.
51
charts_regex: 'system\..*'
52