src/collectors/python.d.plugin/zscores/zscores.chart.py
# -*- coding: utf-8 -*-
# Description: zscores netdata python.d module
# Author: andrewm4894
# SPDX-License-Identifier: GPL-3.0-or-later
from datetime import datetime
import re
import requests
import numpy as np
import pandas as pd
from bases.FrameworkServices.SimpleService import SimpleService
from netdata_pandas.data import get_data, get_allmetrics
priority = 60000
update_every = 5
disabled_by_default = True
ORDER = [
'z',
'3stddev'
]
CHARTS = {
'z': {
'options': ['z', 'Z Score', 'z', 'Z Score', 'zscores.z', 'line'],
'lines': []
},
'3stddev': {
'options': ['3stddev', 'Z Score >3', 'count', '3 Stddev', 'zscores.3stddev', 'stacked'],
'lines': []
},
}
class Service(SimpleService):
def __init__(self, configuration=None, name=None):
SimpleService.__init__(self, configuration=configuration, name=name)
self.host = self.configuration.get('host', '127.0.0.1:19999')
self.charts_regex = re.compile(self.configuration.get('charts_regex', 'system.*'))
self.charts_to_exclude = self.configuration.get('charts_to_exclude', '').split(',')
self.charts_in_scope = [
c for c in
list(filter(self.charts_regex.match,
requests.get(f'http://{self.host}/api/v1/charts').json()['charts'].keys()))
if c not in self.charts_to_exclude
]
self.train_secs = self.configuration.get('train_secs', 14400)
self.offset_secs = self.configuration.get('offset_secs', 300)
self.train_every_n = self.configuration.get('train_every_n', 900)
self.z_smooth_n = self.configuration.get('z_smooth_n', 15)
self.z_clip = self.configuration.get('z_clip', 10)
self.z_abs = bool(self.configuration.get('z_abs', True))
self.burn_in = self.configuration.get('burn_in', 2)
self.mode = self.configuration.get('mode', 'per_chart')
self.per_chart_agg = self.configuration.get('per_chart_agg', 'mean')
self.order = ORDER
self.definitions = CHARTS
self.collected_dims = {'z': set(), '3stddev': set()}
self.df_mean = pd.DataFrame()
self.df_std = pd.DataFrame()
self.df_z_history = pd.DataFrame()
def check(self):
_ = get_allmetrics(self.host, self.charts_in_scope, wide=True, col_sep='.')
return True
def validate_charts(self, chart, data, algorithm='absolute', multiplier=1, divisor=1):
"""If dimension not in chart then add it.
"""
for dim in data:
if dim not in self.collected_dims[chart]:
self.collected_dims[chart].add(dim)
self.charts[chart].add_dimension([dim, dim, algorithm, multiplier, divisor])
for dim in list(self.collected_dims[chart]):
if dim not in data:
self.collected_dims[chart].remove(dim)
self.charts[chart].del_dimension(dim, hide=False)
def train_model(self):
"""Calculate the mean and stddev for all relevant metrics and store them for use in calulcating zscore at each timestep.
"""
before = int(datetime.now().timestamp()) - self.offset_secs
after = before - self.train_secs
self.df_mean = get_data(
self.host, self.charts_in_scope, after, before, points=10, group='average', col_sep='.'
).mean().to_frame().rename(columns={0: "mean"})
self.df_std = get_data(
self.host, self.charts_in_scope, after, before, points=10, group='stddev', col_sep='.'
).mean().to_frame().rename(columns={0: "std"})
def create_data(self, df_allmetrics):
"""Use x, mean, stddev to generate z scores and 3stddev flags via some pandas manipulation.
Returning two dictionaries of dimensions and measures, one for each chart.
:param df_allmetrics <pd.DataFrame>: pandas dataframe with latest data from api/v1/allmetrics.
:return: (<dict>,<dict>) tuple of dictionaries, one for zscores and the other for a flag if abs(z)>3.
"""
# calculate clipped z score for each available metric
df_z = pd.concat([self.df_mean, self.df_std, df_allmetrics], axis=1, join='inner')
df_z['z'] = ((df_z['value'] - df_z['mean']) / df_z['std']).clip(-self.z_clip, self.z_clip).fillna(0) * 100
if self.z_abs:
df_z['z'] = df_z['z'].abs()
# append last z_smooth_n rows of zscores to history table in wide format
self.df_z_history = self.df_z_history.append(
df_z[['z']].reset_index().pivot_table(values='z', columns='index'), sort=True
).tail(self.z_smooth_n)
# get average zscore for last z_smooth_n for each metric
df_z_smooth = self.df_z_history.melt(value_name='z').groupby('index')['z'].mean().to_frame()
df_z_smooth['3stddev'] = np.where(abs(df_z_smooth['z']) > 300, 1, 0)
data_z = df_z_smooth['z'].add_suffix('_z').to_dict()
# aggregate to chart level if specified
if self.mode == 'per_chart':
df_z_smooth['chart'] = ['.'.join(x[0:2]) + '_z' for x in df_z_smooth.index.str.split('.').to_list()]
if self.per_chart_agg == 'absmax':
data_z = \
list(df_z_smooth.groupby('chart').agg({'z': lambda x: max(x, key=abs)})['z'].to_dict().values())[0]
else:
data_z = list(df_z_smooth.groupby('chart').agg({'z': [self.per_chart_agg]})['z'].to_dict().values())[0]
data_3stddev = {}
for k in data_z:
data_3stddev[k.replace('_z', '')] = 1 if abs(data_z[k]) > 300 else 0
return data_z, data_3stddev
def get_data(self):
if self.runs_counter <= self.burn_in or self.runs_counter % self.train_every_n == 0:
self.train_model()
data_z, data_3stddev = self.create_data(
get_allmetrics(self.host, self.charts_in_scope, wide=True, col_sep='.').transpose())
data = {**data_z, **data_3stddev}
self.validate_charts('z', data_z, divisor=100)
self.validate_charts('3stddev', data_3stddev)
return data