@cryptotaxi247 / netdata-1 / commits / 69ea17d6e

Track anomaly rates with DBEngine. (#12083)

* Track anomaly rates with DBEngine. This commit adds support for tracking anomaly rates with DBEngine. We do so by creating a single chart with id "anomaly_detection.anomaly_rates" for each trainable/predictable host, which is responsible for tracking the anomaly rate of each dimension that we train/predict for that host. The rrdset->state->is_ar_chart boolean flag is set to true only for anomaly rates charts. We use this flag to: - Disable exposing the anomaly rates charts through the functionality in backends/, exporting/ and streaming/. - Skip generation of configuration options for the name, algorithm, multiplier, divisor of each dimension in an anomaly rates chart. - Skip the creation of health variables for anomaly rates dimensions. - Skip the chart/dim queue of ACLK. - Post-process the RRDR result of an anomaly rates chart, so that we can return a sorted, trimmed number of anomalous dimensions. In a child/parent configuration where both the child and the parent run ML for the child, we want to be able to stream the rest of the ML-related charts to the parent. To be able to do this without any chart name collisions, the charts are now created on localhost and their IDs and titles have the node's machine_guid and hostname as a suffix, respectively. * Fix exporting_engine tests. * Restore default ML configuration. The reverted changes where meant for local testing only. This commit restores the default values that we want to have when someone runs anomaly detection on their node. * Set context for anomaly_detection.* charts. * Check for anomaly rates chart only with a valid pointer. * Remove duplicate code. * Use a more descriptive name for id/title pair variable

vkalintiris committed Feb 24, 2022 at 10:57 UTC 69ea17d6ec534e1ed796a92fd042bd76a3ca9215
20 files changed +307 -82
backends/backends.c
+4
@@ -193,6 +193,10 @@ inline int backends_can_send_rrdset(BACKEND_OPTIONS backend_options, RRDSET *st)
193 RRDHOST *host = st->rrdhost;
194 (void)host;
195
196 + // Do not send anomaly rates charts.
197 + if (unlikely(st->state->is_ar_chart))
198 + return 0;
199 +
200 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_BACKEND_IGNORE)))
201 return 0;
202
database/rrd.h
+1
@@ -436,6 +436,7 @@ struct rrdset_volatile {
436 uuid_t hash_id;
437 struct label *new_labels;
438 struct label_index labels;
439 + bool is_ar_chart;
440 };
441
442 // ----------------------------------------------------------------------------
database/rrddim.c
+8 -2
@@ -73,9 +73,15 @@ inline int rrddim_set_name(RRDSET *st, RRDDIM *rd, const char *name) {
73 snprintfz(varname, CONFIG_MAX_NAME, "dim %s name", rd->id);
74 rd->name = config_set_default(st->config_section, varname, name);
75 rd->hash_name = simple_hash(rd->name);
76 - rrddimvar_rename_all(rd);
76 +
77 + if (!st->state->is_ar_chart)
78 + rrddimvar_rename_all(rd);
79 +
80 rd->exposed = 0;
81 rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
82 +
83 + ml_dimension_update_name(st, rd, name);
84 +
85 return 1;
86 }
87
@@ -438,7 +444,7 @@ RRDDIM *rrddim_add_custom(RRDSET *st, const char *id, const char *name, collecte
444 td->next = rd;
445 }
446
441 - if(host->health_enabled) {
447 + if(host->health_enabled && !st->state->is_ar_chart) {
448 rrddimvar_create(rd, RRDVAR_TYPE_CALCULATED, NULL, NULL, &rd->last_stored_value, RRDVAR_OPTION_DEFAULT);
449 rrddimvar_create(rd, RRDVAR_TYPE_COLLECTED, NULL, "_raw", &rd->last_collected_value, RRDVAR_OPTION_DEFAULT);
450 rrddimvar_create(rd, RRDVAR_TYPE_TIME_T, NULL, "_last_collected_t", &rd->last_collected_time.tv_sec, RRDVAR_OPTION_DEFAULT);
database/rrdhost.c
+5 -7
@@ -385,17 +385,15 @@ RRDHOST *rrdhost_create(const char *hostname,
385 // ------------------------------------------------------------------------
386 // init new ML host and update system_info to let upstreams know
387 // about ML functionality
388 + //
389
389 - ml_new_host(host);
390 if (is_localhost && host->system_info) {
391 -#ifndef ENABLE_ML
392 - host->system_info->ml_capable = 0;
393 -#else
394 - host->system_info->ml_capable = 1;
395 -#endif
396 - host->system_info->ml_enabled = host->ml_host != NULL;
391 + host->system_info->ml_capable = ml_capable();
392 + host->system_info->ml_enabled = ml_enabled(host);
393 }
394
395 + ml_new_host(host);
396 +
397 info("Host '%s' (at registry as '%s') with guid '%s' initialized"
398 ", os '%s'"
399 ", timezone '%s'"
database/rrdset.c
+11 -4
@@ -847,9 +847,12 @@ RRDSET *rrdset_create_custom(
847 st->type = strdupz(type);
848
849 st->state = callocz(1, sizeof(*st->state));
850 +
851 st->family = family ? strdupz(family) : strdupz(st->type);
852 json_fix_string(st->family);
853
854 + st->state->is_ar_chart = strcmp(st->id, ML_ANOMALY_RATES_CHART_ID) == 0;
855 +
856 st->units = units ? strdupz(units) : strdupz("");
857 json_fix_string(st->units);
858
@@ -1395,10 +1398,12 @@ void rrdset_done(RRDSET *st) {
1398 rrdset_rdlock(st);
1399
1400 #ifdef ENABLE_ACLK
1398 - if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1399 - if (st->counter_done >= RRDSET_MINIMUM_LIVE_COUNT && st->dimensions) {
1400 - if (likely(!queue_chart_to_aclk(st)))
1401 - rrdset_flag_set(st, RRDSET_FLAG_ACLK);
1401 + if (likely(!st->state->is_ar_chart)) {
1402 + if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1403 + if (st->counter_done >= RRDSET_MINIMUM_LIVE_COUNT && st->dimensions) {
1404 + if (likely(!queue_chart_to_aclk(st)))
1405 + rrdset_flag_set(st, RRDSET_FLAG_ACLK);
1406 + }
1407 }
1408 }
1409 #endif
@@ -1825,6 +1830,7 @@ after_second_database_work:
1830 continue;
1831
1832 #if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
1833 + if (likely(!st->state->is_ar_chart)) {
1834 if (!rrddim_flag_check(rd, RRDDIM_FLAG_HIDDEN)) {
1835 int live = ((mark - rd->last_collected_time.tv_sec) < (RRDSET_MINIMUM_LIVE_COUNT * rd->update_every));
1836 if (unlikely(live != rd->state->aclk_live_status)) {
@@ -1836,6 +1842,7 @@ after_second_database_work:
1842 }
1843 }
1844 }
1845 + }
1846 #endif
1847 if(unlikely(!rd->updated))
1848 continue;
database/sqlite/sqlite_aclk_node.c
+2 -2
@@ -22,8 +22,8 @@ void sql_build_node_info(struct aclk_database_worker_config *wc, struct aclk_dat
22 node_info.claim_id = is_agent_claimed();
23 node_info.machine_guid = wc->host_guid;
24 node_info.child = (wc->host != localhost);
25 - node_info.ml_info.ml_capable = localhost->system_info->ml_capable;
26 - node_info.ml_info.ml_enabled = wc->host->ml_host != NULL;
25 + node_info.ml_info.ml_capable = ml_capable(localhost);
26 + node_info.ml_info.ml_enabled = ml_enabled(wc->host);
27 now_realtime_timeval(&node_info.updated_at);
28
29 RRDHOST *host = wc->host;
exporting/check_filters.c
+4
@@ -47,6 +47,10 @@ int rrdset_is_exportable(struct instance *instance, RRDSET *st)
47 RRDHOST *host = st->rrdhost;
48 #endif
49
50 + // Do not export anomaly rates charts.
51 + if (st->state && st->state->is_ar_chart)
52 + return 0;
53 +
54 if (st->exporting_flags == NULL)
55 st->exporting_flags = callocz(instance->engine->instance_num, sizeof(size_t));
56
exporting/prometheus/prometheus.c
+4
@@ -20,6 +20,10 @@ inline int can_send_rrdset(struct instance *instance, RRDSET *st)
20 RRDHOST *host = st->rrdhost;
21 #endif
22
23 + // Do not send anomaly rates charts.
24 + if (st->state && st->state->is_ar_chart)
25 + return 0;
26 +
27 if (unlikely(rrdset_flag_check(st, RRDSET_FLAG_EXPORTING_IGNORE)))
28 return 0;
29
ml/Config.cc
+9 -1
@@ -32,6 +32,8 @@ void Config::readMLConfig(void) {
32 unsigned MinTrainSamples = config_get_number(ConfigSectionML, "minimum num samples to train", 1 * 3600);
33 unsigned TrainEvery = config_get_number(ConfigSectionML, "train every", 1 * 3600);
34
35 + unsigned DBEngineAnomalyRateEvery = config_get_number(ConfigSectionML, "dbengine anomaly rate every", 60);
36 +
37 unsigned DiffN = config_get_number(ConfigSectionML, "num samples to diff", 1);
38 unsigned SmoothN = config_get_number(ConfigSectionML, "num samples to smooth", 3);
39 unsigned LagN = config_get_number(ConfigSectionML, "num samples to lag", 5);
@@ -59,6 +61,8 @@ void Config::readMLConfig(void) {
61 MinTrainSamples = clamp(MinTrainSamples, 1 * 3600u, 6 * 3600u);
62 TrainEvery = clamp(TrainEvery, 1 * 3600u, 6 * 3600u);
63
64 + DBEngineAnomalyRateEvery = clamp(DBEngineAnomalyRateEvery, 1 * 60u, 15 * 60u);
65 +
66 DiffN = clamp(DiffN, 0u, 1u);
67 SmoothN = clamp(SmoothN, 0u, 5u);
68 LagN = clamp(LagN, 0u, 5u);
@@ -102,6 +106,8 @@ void Config::readMLConfig(void) {
106 Cfg.MinTrainSamples = MinTrainSamples;
107 Cfg.TrainEvery = TrainEvery;
108
109 + Cfg.DBEngineAnomalyRateEvery = DBEngineAnomalyRateEvery;
110 +
111 Cfg.DiffN = DiffN;
112 Cfg.SmoothN = SmoothN;
113 Cfg.LagN = LagN;
@@ -120,7 +126,9 @@ void Config::readMLConfig(void) {
126 Cfg.HostsToSkip = config_get(ConfigSectionML, "hosts to skip from training", "!*");
127 Cfg.SP_HostsToSkip = simple_pattern_create(Cfg.HostsToSkip.c_str(), NULL, SIMPLE_PATTERN_EXACT);
128
123 - Cfg.ChartsToSkip = config_get(ConfigSectionML, "charts to skip from training",
129 + // Always exclude anomaly_detection charts from training.
130 + Cfg.ChartsToSkip = "anomaly_detection.* ";
131 + Cfg.ChartsToSkip += config_get(ConfigSectionML, "charts to skip from training",
132 "!system.* !cpu.* !mem.* !disk.* !disk_* "
133 "!ip.* !ipv4.* !ipv6.* !net.* !net_* !netfilter.* "
134 "!services.* !apps.* !groups.* !user.* !ebpf.* !netdata.* *");
ml/Config.h
+2
@@ -15,6 +15,8 @@ public:
15 unsigned MinTrainSamples;
16 unsigned TrainEvery;
17
18 + unsigned DBEngineAnomalyRateEvery;
19 +
20 unsigned DiffN;
21 unsigned SmoothN;
22 unsigned LagN;
ml/Dimension.h
+29 -7
@@ -12,11 +12,7 @@ namespace ml {
12
13 class RrdDimension {
14 public:
15 - RrdDimension(RRDDIM *RD) : RD(RD), Ops(&RD->state->query_ops) {
16 - std::stringstream SS;
17 - SS << RD->rrdset->id << "|" << RD->name;
18 - ID = SS.str();
19 - }
15 + RrdDimension(RRDDIM *RD) : RD(RD), Ops(&RD->state->query_ops) { }
16
17 RRDDIM *getRD() const { return RD; }
18
@@ -26,12 +22,27 @@ public:
22
23 unsigned updateEvery() const { return RD->update_every; }
24
29 - const std::string getID() const { return ID; }
25 + const std::string getID() const {
26 + std::stringstream SS;
27 + SS << RD->rrdset->id << "|" << RD->name;
28 + return SS.str();
29 + }
30 +
31 + void setAnomalyRateRD(RRDDIM *ARRD) { AnomalyRateRD = ARRD; }
32 + RRDDIM *getAnomalyRateRD() const { return AnomalyRateRD; }
33
31 - virtual ~RrdDimension() {}
34 + void setAnomalyRateRDName(const char *Name) const {
35 + rrddim_set_name(AnomalyRateRD->rrdset, AnomalyRateRD, Name);
36 + }
37 +
38 + virtual ~RrdDimension() {
39 + rrddim_free_custom(AnomalyRateRD->rrdset, AnomalyRateRD, 0);
40 + }
41
42 private:
43 RRDDIM *RD;
44 + RRDDIM *AnomalyRateRD;
45 +
46 struct rrddim_volatile::rrddim_query_ops *Ops;
47
48 std::string ID;
@@ -94,9 +105,20 @@ public:
105
106 bool isAnomalous() { return AnomalyBit; }
107
108 + void updateAnomalyBitCounter(RRDSET *RS, unsigned Elapsed, bool IsAnomalous) {
109 + AnomalyBitCounter += IsAnomalous;
110 +
111 + if (Elapsed == Cfg.DBEngineAnomalyRateEvery) {
112 + double AR = static_cast<double>(AnomalyBitCounter) / Cfg.DBEngineAnomalyRateEvery;
113 + rrddim_set_by_pointer(RS, getAnomalyRateRD(), AR * 1000);
114 + AnomalyBitCounter = 0;
115 + }
116 + }
117 +
118 private:
119 CalculatedNumber AnomalyScore{0.0};
120 std::atomic<bool> AnomalyBit{false};
121 + unsigned AnomalyBitCounter{0};
122
123 std::vector<CalculatedNumber> CNs;
124 };
ml/Host.cc
+87 -41
@@ -9,6 +9,17 @@
9
10 using namespace ml;
11
12 +static std::pair<std::string, std::string>
13 +getHostSpecificIdAndTitle(RRDHOST *RH, const std::string &IdPrefix,
14 + const std::string &TitlePrefix) {
15 + std::stringstream IdSS, TitleSS;
16 +
17 + IdSS << IdPrefix << "_" << RH->machine_guid;
18 + TitleSS << TitlePrefix << " " << RH->hostname;
19 +
20 + return {IdSS.str(), TitleSS.str()};
21 +}
22 +
23 static void updateDimensionsChart(RRDHOST *RH,
24 collected_number NumTrainedDimensions,
25 collected_number NumNormalDimensions,
@@ -20,14 +31,17 @@ static void updateDimensionsChart(RRDHOST *RH,
31 static thread_local RRDDIM *NumAnomalousDimensionsRD = nullptr;
32
33 if (!RS) {
23 - RS = rrdset_create(
24 - RH, // host
34 + std::string IdPrefix = "dimensions";
35 + std::string TitlePrefix = "Anomaly detection dimensions for host";
36 + auto IdTitlePair = getHostSpecificIdAndTitle(RH, IdPrefix, TitlePrefix);
37 +
38 + RS = rrdset_create_localhost(
39 "anomaly_detection", // type
26 - "dimensions", // id
40 + IdTitlePair.first.c_str(), // id
41 NULL, // name
42 "dimensions", // family
29 - NULL, // ctx
30 - "Anomaly detection dimensions", // title
43 + "anomaly_detection.dimensions", // ctx
44 + IdTitlePair.second.c_str(), // title
45 "dimensions", // units
46 "netdata", // plugin
47 "ml", // module
@@ -60,14 +74,17 @@ static void updateRateChart(RRDHOST *RH, collected_number AnomalyRate) {
74 static thread_local RRDDIM *AnomalyRateRD = nullptr;
75
76 if (!RS) {
63 - RS = rrdset_create(
64 - RH, // host
77 + std::string IdPrefix = "anomaly_rate";
78 + std::string TitlePrefix = "Percentage of anomalous dimensions for host";
79 + auto IdTitlePair = getHostSpecificIdAndTitle(RH, IdPrefix, TitlePrefix);
80 +
81 + RS = rrdset_create_localhost(
82 "anomaly_detection", // type
66 - "anomaly_rate", // id
83 + IdTitlePair.first.c_str(), // id
84 NULL, // name
85 "anomaly_rate", // family
69 - NULL, // ctx
70 - "Percentage of anomalous dimensions", // title
86 + "anomaly_detection.anomaly_rate", // ctx
87 + IdTitlePair.second.c_str(), // title
88 "percentage", // units
89 "netdata", // plugin
90 "ml", // module
@@ -91,14 +108,17 @@ static void updateWindowLengthChart(RRDHOST *RH, collected_number WindowLength)
108 static thread_local RRDDIM *WindowLengthRD = nullptr;
109
110 if (!RS) {
94 - RS = rrdset_create(
95 - RH, // host
111 + std::string IdPrefix = "detector_window";
112 + std::string TitlePrefix = "Anomaly detector window length for host";
113 + auto IdTitlePair = getHostSpecificIdAndTitle(RH, IdPrefix, TitlePrefix);
114 +
115 + RS = rrdset_create_localhost(
116 "anomaly_detection", // type
97 - "detector_window", // id
117 + IdTitlePair.first.c_str(), // id
118 NULL, // name
119 "detector_window", // family
100 - NULL, // ctx
101 - "Anomaly detector window length", // title
120 + "anomaly_detection.detector_window", // ctx
121 + IdTitlePair.second.c_str(), // title
122 "seconds", // units
123 "netdata", // plugin
124 "ml", // module
@@ -126,14 +146,17 @@ static void updateEventsChart(RRDHOST *RH,
146 static thread_local RRDDIM *NewAnomalyEventRD = nullptr;
147
148 if (!RS) {
129 - RS = rrdset_create(
130 - RH, // host
149 + std::string IdPrefix = "detector_events";
150 + std::string TitlePrefix = "Anomaly events triggered for host";
151 + auto IdTitlePair = getHostSpecificIdAndTitle(RH, IdPrefix, TitlePrefix);
152 +
153 + RS = rrdset_create_localhost(
154 "anomaly_detection", // type
132 - "detector_events", // id
155 + IdTitlePair.first.c_str(), // id
156 NULL, // name
157 "detector_events", // family
135 - NULL, // ctx
136 - "Anomaly events triggered", // title
158 + "anomaly_detection.detector_events", // ctx
159 + IdTitlePair.second.c_str(), // title
160 "boolean", // units
161 "netdata", // plugin
162 "ml", // module
@@ -166,14 +189,17 @@ static void updateDetectionChart(RRDHOST *RH, collected_number PredictionDuratio
189 static thread_local RRDDIM *PredictiobDurationRD = nullptr;
190
191 if (!RS) {
169 - RS = rrdset_create(
170 - RH, // host
192 + std::string IdPrefix = "prediction_stats";
193 + std::string TitlePrefix = "Time it took to run prediction for host";
194 + auto IdTitlePair = getHostSpecificIdAndTitle(RH, IdPrefix, TitlePrefix);
195 +
196 + RS = rrdset_create_localhost(
197 "anomaly_detection", // type
172 - "prediction_stats", // id
198 + IdTitlePair.first.c_str(), // id
199 NULL, // name
200 "prediction_stats", // family
175 - NULL, // ctx
176 - "Time it took to run prediction", // title
201 + "anomaly_detection.prediction_stats", // ctx
202 + IdTitlePair.second.c_str(), // title
203 "milliseconds", // units
204 "netdata", // plugin
205 "ml", // module
@@ -201,14 +227,17 @@ static void updateTrainingChart(RRDHOST *RH,
227 static thread_local RRDDIM *MaxTrainingDurationRD = nullptr;
228
229 if (!RS) {
204 - RS = rrdset_create(
205 - RH, // host
230 + std::string IdPrefix = "training_stats";
231 + std::string TitlePrefix = "Training step statistics for host";
232 + auto IdTitlePair = getHostSpecificIdAndTitle(RH, IdPrefix, TitlePrefix);
233 +
234 + RS = rrdset_create_localhost(
235 "anomaly_detection", // type
207 - "training_stats", // id
236 + IdTitlePair.first.c_str(), // id
237 NULL, // name
238 "training_stats", // family
210 - NULL, // ctx
211 - "Training step statistics", // title
239 + "anomaly_detection.training_stats", // ctx
240 + IdTitlePair.second.c_str(), // title
241 "milliseconds", // units
242 "netdata", // plugin
243 "ml", // module
@@ -231,12 +260,18 @@ static void updateTrainingChart(RRDHOST *RH,
260 }
261
262 void RrdHost::addDimension(Dimension *D) {
234 - std::lock_guard<std::mutex> Lock(Mutex);
263 + RRDDIM *AnomalyRateRD = rrddim_add(AnomalyRateRS, D->getID().c_str(), NULL,
264 + 1, 1000, RRD_ALGORITHM_ABSOLUTE);
265 + D->setAnomalyRateRD(AnomalyRateRD);
266 +
267 + {
268 + std::lock_guard<std::mutex> Lock(Mutex);
269
236 - DimensionsMap[D->getRD()] = D;
270 + DimensionsMap[D->getRD()] = D;
271
238 - // Default construct mutex for dimension
239 - LocksMap[D];
272 + // Default construct mutex for dimension
273 + LocksMap[D];
274 + }
275 }
276
277 void RrdHost::removeDimension(Dimension *D) {
@@ -344,7 +379,7 @@ void TrainableHost::train() {
379 }
380
381 void DetectableHost::detectOnce() {
347 - auto P = BRW.insert(AnomalyRate >= Cfg.HostAnomalyRateThreshold);
382 + auto P = BRW.insert(WindowAnomalyRate >= Cfg.HostAnomalyRateThreshold);
383 BitRateWindow::Edge Edge = P.first;
384 size_t WindowLength = P.second;
385
@@ -361,6 +396,10 @@ void DetectableHost::detectOnce() {
396 double TotalTrainingDuration = 0.0;
397 double MaxTrainingDuration = 0.0;
398
399 + bool CollectAnomalyRates = (++AnomalyRateTimer == Cfg.DBEngineAnomalyRateEvery);
400 + if (CollectAnomalyRates)
401 + rrdset_next(AnomalyRateRS);
402 +
403 {
404 std::lock_guard<std::mutex> Lock(Mutex);
405
@@ -371,7 +410,7 @@ void DetectableHost::detectOnce() {
410
411 auto P = D->detect(WindowLength, ResetBitCounter);
412 bool IsAnomalous = P.first;
374 - double AnomalyRate = P.second;
413 + double AnomalyScore = P.second;
414
415 NumTrainedDimensions += D->isTrained();
416
@@ -382,24 +421,31 @@ void DetectableHost::detectOnce() {
421 if (IsAnomalous)
422 NumAnomalousDimensions += 1;
423
385 - if (NewAnomalyEvent && (AnomalyRate >= Cfg.ADDimensionRateThreshold))
386 - DimsOverThreshold.push_back({ AnomalyRate, D->getID() });
424 + if (NewAnomalyEvent && (AnomalyScore >= Cfg.ADDimensionRateThreshold))
425 + DimsOverThreshold.push_back({ AnomalyScore, D->getID() });
426 +
427 + D->updateAnomalyBitCounter(AnomalyRateRS, AnomalyRateTimer, IsAnomalous);
428 }
429
430 if (NumAnomalousDimensions)
390 - AnomalyRate = static_cast<double>(NumAnomalousDimensions) / DimensionsMap.size();
431 + WindowAnomalyRate = static_cast<double>(NumAnomalousDimensions) / DimensionsMap.size();
432 else
392 - AnomalyRate = 0.0;
433 + WindowAnomalyRate = 0.0;
434
435 NumNormalDimensions = DimensionsMap.size() - NumAnomalousDimensions;
436 }
437
438 + if (CollectAnomalyRates) {
439 + AnomalyRateTimer = 0;
440 + rrdset_done(AnomalyRateRS);
441 + }
442 +
443 this->NumAnomalousDimensions = NumAnomalousDimensions;
444 this->NumNormalDimensions = NumNormalDimensions;
445 this->NumTrainedDimensions = NumTrainedDimensions;
446
447 updateDimensionsChart(getRH(), NumTrainedDimensions, NumNormalDimensions, NumAnomalousDimensions);
402 - updateRateChart(getRH(), AnomalyRate * 10000.0);
448 + updateRateChart(getRH(), WindowAnomalyRate * 10000.0);
449 updateWindowLengthChart(getRH(), WindowLength);
450 updateEventsChart(getRH(), P, ResetBitCounter, NewAnomalyEvent);
451 updateTrainingChart(getRH(), TotalTrainingDuration * 1000.0, MaxTrainingDuration * 1000.0);
ml/Host.h
+25 -4
@@ -14,7 +14,25 @@ namespace ml {
14
15 class RrdHost {
16 public:
17 - RrdHost(RRDHOST *RH) : RH(RH) {}
17 + RrdHost(RRDHOST *RH) : RH(RH) {
18 + AnomalyRateRS = rrdset_create(
19 + RH,
20 + "anomaly_detection",
21 + "anomaly_rates",
22 + NULL, // name
23 + "anomaly_rates",
24 + NULL, // ctx
25 + "Average anomaly rate",
26 + "anomaly rate",
27 + "netdata",
28 + "ml",
29 + 39189,
30 + Cfg.DBEngineAnomalyRateEvery,
31 + RRDSET_TYPE_LINE
32 + );
33 +
34 + rrdset_flag_set(AnomalyRateRS, RRDSET_FLAG_HIDDEN);
35 + }
36
37 RRDHOST *getRH() { return RH; }
38
@@ -35,12 +53,13 @@ public:
53
54 protected:
55 RRDHOST *RH;
56 + RRDSET *AnomalyRateRS;
57
58 // Protect dimension and lock maps
59 std::mutex Mutex;
60
42 - std::map<RRDDIM *, Dimension *> DimensionsMap;
43 - std::map<Dimension *, std::mutex> LocksMap;
61 + std::unordered_map<RRDDIM *, Dimension *> DimensionsMap;
62 + std::unordered_map<Dimension *, std::mutex> LocksMap;
63 };
64
65 class TrainableHost : public RrdHost {
@@ -88,12 +107,14 @@ private:
107 static_cast<size_t>(Cfg.ADMinWindowSize * Cfg.ADWindowRateThreshold)
108 };
109
91 - CalculatedNumber AnomalyRate{0.0};
110 + CalculatedNumber WindowAnomalyRate{0.0};
111
112 size_t NumAnomalousDimensions{0};
113 size_t NumNormalDimensions{0};
114 size_t NumTrainedDimensions{0};
115
116 + unsigned AnomalyRateTimer{0};
117 +
118 Database DB{Cfg.AnomalyDBPath};
119 };
120
ml/ml-dummy.c
+20
@@ -4,6 +4,15 @@
4
5 #if !defined(ENABLE_ML)
6
7 +bool ml_capable() {
8 + return false;
9 +}
10 +
11 +bool ml_enabled(RRDHOST *RH) {
12 + (void) RH;
13 + return false;
14 +}
15 +
16 void ml_init(void) {}
17
18 void ml_new_host(RRDHOST *RH) { (void) RH; }
@@ -38,4 +47,15 @@ char *ml_get_anomaly_event_info(RRDHOST *RH, const char *AnomalyDetectorName,
47 return NULL;
48 }
49
50 +void ml_process_rrdr(RRDR *R, int MaxAnomalyRates) {
51 + (void) R;
52 + (void) MaxAnomalyRates;
53 +}
54 +
55 +void ml_dimension_update_name(RRDSET *RS, RRDDIM *RD, const char *name) {
56 + (void) RS;
57 + (void) RD;
58 + (void) name;
59 +}
60 +
61 #endif
ml/ml.cc
+55 -4
@@ -6,6 +6,20 @@
6
7 using namespace ml;
8
9 +bool ml_capable() {
10 + return true;
11 +}
12 +
13 +bool ml_enabled(RRDHOST *RH) {
14 + if (!Cfg.EnableAnomalyDetection)
15 + return false;
16 +
17 + if (simple_pattern_matches(Cfg.SP_HostsToSkip, RH->hostname))
18 + return false;
19 +
20 + return true;
21 +}
22 +
23 /*
24 * Assumptions:
25 * 1) hosts outlive their sets, and sets outlive their dimensions,
@@ -17,10 +31,7 @@ void ml_init(void) {
31 }
32
33 void ml_new_host(RRDHOST *RH) {
20 - if (!Cfg.EnableAnomalyDetection)
21 - return;
22 -
23 - if (simple_pattern_matches(Cfg.SP_HostsToSkip, RH->hostname))
34 + if (!ml_enabled(RH))
35 return;
36
37 Host *H = new Host(RH);
@@ -150,6 +161,44 @@ char *ml_get_anomaly_event_info(RRDHOST *RH, const char *AnomalyDetectorName,
161 return strdup(Json.dump(4, '\t').c_str());
162 }
163
164 +void ml_process_rrdr(RRDR *R, int MaxAnomalyRates) {
165 + if (R->rows != 1)
166 + return;
167 +
168 + if (MaxAnomalyRates < 1 || MaxAnomalyRates >= R->d)
169 + return;
170 +
171 + calculated_number *CNs = R->v;
172 + RRDR_DIMENSION_FLAGS *DimFlags = R->od;
173 +
174 + std::vector<std::pair<calculated_number, int>> V;
175 +
176 + V.reserve(R->d);
177 + for (int Idx = 0; Idx != R->d; Idx++)
178 + V.emplace_back(CNs[Idx], Idx);
179 +
180 + std::sort(V.rbegin(), V.rend());
181 +
182 + for (int Idx = MaxAnomalyRates; Idx != R->d; Idx++) {
183 + int UnsortedIdx = V[Idx].second;
184 +
185 + int OldFlags = static_cast<int>(DimFlags[UnsortedIdx]);
186 + int NewFlags = OldFlags | RRDR_DIMENSION_HIDDEN;
187 +
188 + DimFlags[UnsortedIdx] = static_cast<rrdr_dimension_flag>(NewFlags);
189 + }
190 +}
191 +
192 +void ml_dimension_update_name(RRDSET *RS, RRDDIM *RD, const char *Name) {
193 + (void) RS;
194 +
195 + Dimension *D = static_cast<Dimension *>(RD->state->ml_dimension);
196 + if (!D)
197 + return;
198 +
199 + D->setAnomalyRateRDName(Name);
200 +}
201 +
202 #if defined(ENABLE_ML_TESTS)
203
204 #include "gtest/gtest.h"
@@ -163,3 +212,5 @@ int test_ml(int argc, char *argv[]) {
212 }
213
214 #endif // ENABLE_ML_TESTS
215 +
216 +#include "ml-private.h"
ml/ml.h
+15
@@ -8,10 +8,19 @@ extern "C" {
8 #endif
9
10 #include "daemon/common.h"
11 +#include "web/api/queries/rrdr.h"
12 +
13 +// This is an internal DBEngine function redeclared here so that we can free
14 +// the anomaly rate dimension, whenever its backing dimension is freed.
15 +extern void rrddim_free_custom(RRDSET *st, RRDDIM *rd, int db_rotated);
16
17 typedef void* ml_host_t;
18 typedef void* ml_dimension_t;
19
20 +bool ml_capable();
21 +
22 +bool ml_enabled(RRDHOST *RH);
23 +
24 void ml_init(void);
25
26 void ml_new_host(RRDHOST *RH);
@@ -31,6 +40,12 @@ char *ml_get_anomaly_events(RRDHOST *RH, const char *AnomalyDetectorName,
40 char *ml_get_anomaly_event_info(RRDHOST *RH, const char *AnomalyDetectorName,
41 int AnomalyDetectorVersion, time_t After, time_t Before);
42
43 +void ml_process_rrdr(RRDR *R, int MaxAnomalyRates);
44 +
45 +void ml_dimension_update_name(RRDSET *RS, RRDDIM *RD, const char *name);
46 +
47 +#define ML_ANOMALY_RATES_CHART_ID "anomaly_detection.anomaly_rates"
48 +
49 #if defined(ENABLE_ML_TESTS)
50 int test_ml(int argc, char *argv[]);
51 #endif
streaming/rrdpush.c
+5 -1
@@ -129,6 +129,10 @@ unsigned int remote_clock_resync_iterations = 60;
129
130
131 static inline int should_send_chart_matching(RRDSET *st) {
132 + // Do not stream anomaly rates charts.
133 + if (unlikely(st->state->is_ar_chart))
134 + return false;
135 +
136 if(unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ENABLED))) {
137 rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_SEND);
138 rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_IGNORE);
@@ -781,4 +785,4 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
785
786 buffer_flush(w->response.data);
787 return 200;
784 -}
\ No newline at end of file
788 +}
web/api/formatters/rrd2json.c
+4
@@ -217,6 +217,7 @@ int rrdset2anything_api_v1(
217 , time_t *latest_timestamp
218 , struct context_param *context_param_list
219 , char *chart_label_key
220 + , int max_anomaly_rates
221 ) {
222
223 if (context_param_list && !(context_param_list->flags & CONTEXT_FLAGS_ARCHIVE))
@@ -228,6 +229,9 @@ int rrdset2anything_api_v1(
229 return HTTP_RESP_INTERNAL_SERVER_ERROR;
230 }
231
232 + if (st && st->state->is_ar_chart)
233 + ml_process_rrdr(r, max_anomaly_rates);
234 +
235 RRDDIM *temp_rd = context_param_list ? context_param_list->rd : NULL;
236
237 if(r->result_options & RRDR_RESULT_OPTION_RELATIVE)
web/api/formatters/rrd2json.h
+1
@@ -67,6 +67,7 @@ extern int rrdset2anything_api_v1(
67 , time_t *latest_timestamp
68 , struct context_param *context_param_list
69 , char *chart_label_key
70 + , int max_anomaly_rates
71 );
72
73 extern int rrdset2value_api_v1(
web/api/web_api_v1.c
+16 -9
@@ -401,13 +401,14 @@ inline int web_client_api_request_v1_data(RRDHOST *host, struct web_client *w, c
401
402 time_t last_timestamp_in_data = 0, google_timestamp = 0;
403
404 - char *chart = NULL
405 - , *before_str = NULL
406 - , *after_str = NULL
407 - , *group_time_str = NULL
408 - , *points_str = NULL
409 - , *context = NULL
410 - , *chart_label_key = NULL;
404 + char *chart = NULL;
405 + char *before_str = NULL;
406 + char *after_str = NULL;
407 + char *group_time_str = NULL;
408 + char *points_str = NULL;
409 + char *max_anomaly_rates_str = NULL;
410 + char *context = NULL;
411 + char *chart_label_key = NULL;
412
413 int group = RRDR_GROUPING_AVERAGE;
414 uint32_t format = DATASOURCE_JSON;
@@ -484,6 +485,9 @@ inline int web_client_api_request_v1_data(RRDHOST *host, struct web_client *w, c
485 outFileName = tqx_value;
486 }
487 }
488 + else if(!strcmp(name, "max_anomaly_rates")) {
489 + max_anomaly_rates_str = value;
490 + }
491 }
492
493 // validate the google parameters given
@@ -564,6 +568,7 @@ inline int web_client_api_request_v1_data(RRDHOST *host, struct web_client *w, c
568 long long after = (after_str && *after_str) ?str2l(after_str):-600;
569 int points = (points_str && *points_str)?str2i(points_str):0;
570 long group_time = (group_time_str && *group_time_str)?str2l(group_time_str):0;
571 + int max_anomaly_rates = (max_anomaly_rates_str && *max_anomaly_rates_str) ? str2i(max_anomaly_rates_str) : 0;
572
573 debug(D_WEB_CLIENT, "%llu: API command 'data' for chart '%s', dimensions '%s', after '%lld', before '%lld', points '%d', group '%d', format '%u', options '0x%08x'"
574 , w->id
@@ -606,8 +611,10 @@ inline int web_client_api_request_v1_data(RRDHOST *host, struct web_client *w, c
611 buffer_strcat(w->response.data, "(");
612 }
613
609 - ret = rrdset2anything_api_v1(st, w->response.data, dimensions, format, points, after, before, group, group_time
610 - , options, &last_timestamp_in_data, context_param_list, chart_label_key);
614 + ret = rrdset2anything_api_v1(st, w->response.data, dimensions, format,
615 + points, after, before, group, group_time,
616 + options, &last_timestamp_in_data, context_param_list,
617 + chart_label_key, max_anomaly_rates);
618
619 free_context_param_list(&context_param_list);
620