Refactor ML code. (#14659)
* Refactor ML code. This commit introduces only non-functional changes. Originally, the C++ code exposed C functions to be called from the rest of the agent. When we migrated from C++ to C, we did not eliminate these wrapper functions to make the PR easier to understand and keep the total LOC low. This commit removes the wrapper functions and "reclaims" the `ml_` prefix that we used for the public API of the old implementation. Also, the nlohmann Json library has been removed and its functionality was replaced with the equivalent Json functionality that we added in libnetdata's BUFFERs. * Remove missing headers from build systems. * Fix CMake build. * rrddim_free is outside of rrd "internals" now.
vkalintiris committed
Mar 10, 2023 at 12:20 UTC
37a06960f90c046f21c125c2b4265713da04f851
18 files changed
+1306
-1424
.gitmodules
-4
@@ -9,7 +9,3 @@
9
url = https://github.com/davisking/dlib.git
10
shallow = true
11
ignore = dirty
12
-[submodule "ml/json"]
13
- path = ml/json
14
- url = https://github.com/nlohmann/json.git
15
- shallow = true
CMakeLists.txt
+15
-21
@@ -352,8 +352,7 @@ ENDIF()
352
# Detect ml dependencies
353
file(STRINGS "${CMAKE_SOURCE_DIR}/config.h" DEFINE_ENABLE_ML REGEX "^#define ENABLE_ML 1$")
354
IF(DEFINE_ENABLE_ML MATCHES ".+" AND
355
- EXISTS "${CMAKE_SOURCE_DIR}/ml/dlib/dlib/all/source.cpp" AND
356
- EXISTS "${CMAKE_SOURCE_DIR}/ml/json/single_include/nlohmann/json.hpp")
355
+ EXISTS "${CMAKE_SOURCE_DIR}/ml/dlib/dlib/all/source.cpp")
356
set(ENABLE_ML True)
357
list(APPEND NETDATA_COMMON_CFLAGS "-DDLIB_NO_GUI_SUPPORT")
358
list(APPEND NETDATA_COMMON_INCLUDE_DIRS "ml/dlib")
@@ -718,12 +717,21 @@ set(STATSD_PLUGIN_FILES
717
)
718
719
set(RRD_PLUGIN_FILES
720
+ database/contexts/api_v1.c
721
+ database/contexts/api_v2.c
722
+ database/contexts/context.c
723
+ database/contexts/instance.c
724
+ database/contexts/internal.h
725
+ database/contexts/metric.c
726
+ database/contexts/query_scope.c
727
+ database/contexts/query_target.c
728
+ database/contexts/rrdcontext.c
729
+ database/contexts/rrdcontext.h
730
+ database/contexts/worker.c
731
database/rrdcalc.c
732
database/rrdcalc.h
733
database/rrdcalctemplate.c
734
database/rrdcalctemplate.h
725
- database/rrdcontext.c
726
- database/rrdcontext.h
735
database/rrddim.c
736
database/rrddimvar.c
737
database/rrddimvar.h
@@ -747,7 +755,7 @@ set(RRD_PLUGIN_FILES
755
database/sqlite/sqlite_metadata.h
756
database/sqlite/sqlite_functions.c
757
database/sqlite/sqlite_functions.h
750
- database/sqlite/sqlite_context.c
758
+ database/sqlite/sqlite_context.c
759
database/sqlite/sqlite_context.h
760
database/sqlite/sqlite_db_migration.c
761
database/sqlite/sqlite_db_migration.h
@@ -1037,24 +1045,10 @@ set(ML_FILES
1045
IF(ENABLE_ML)
1046
message(STATUS "ML: enabled")
1047
list(APPEND ML_FILES
1040
- ml/ADCharts.h
1041
- ml/ADCharts.cc
1042
- ml/Chart.h
1043
- ml/Chart.cc
1044
- ml/Config.h
1048
+ ml/ad_charts.h
1049
+ ml/ad_charts.cc
1050
ml/Config.cc
1046
- ml/Dimension.h
1047
- ml/Dimension.cc
1048
- ml/Host.h
1049
- ml/Host.cc
1050
- ml/Mutex.h
1051
- ml/Query.h
1052
- ml/KMeans.h
1053
- ml/KMeans.cc
1054
- ml/SamplesBuffer.h
1055
- ml/SamplesBuffer.cc
1051
ml/dlib/dlib/all/source.cpp
1057
- ml/json/single_include/nlohmann/json.hpp
1052
ml/ml.cc
1053
ml/ml-private.h
1054
)
Makefile.am
+1
-3
@@ -239,9 +239,7 @@ ML_FILES += \
239
ml/ad_charts.cc \
240
ml/Config.cc \
241
ml/dlib/dlib/all/source.cpp \
242
- ml/json/single_include/nlohmann/json.hpp \
243
- ml/nml.h \
244
- ml/nml.cc \
242
+ ml/ml-private.h \
243
ml/ml.cc \
244
$(NULL)
245
collectors/plugins.d/pluginsd_parser.c
+2
-2
@@ -1693,10 +1693,10 @@ PARSER_RC pluginsd_set_v2(char **words, size_t num_words, void *user) {
1693
flags = SN_EMPTY_SLOT;
1694
1695
if(u->v2.ml_locked)
1696
- ml_is_anomalous(rd, u->v2.end_time, 0, false);
1696
+ ml_dimension_is_anomalous(rd, u->v2.end_time, 0, false);
1697
}
1698
else if(u->v2.ml_locked) {
1699
- if (ml_is_anomalous(rd, u->v2.end_time, value, true)) {
1699
+ if (ml_dimension_is_anomalous(rd, u->v2.end_time, value, true)) {
1700
// clear anomaly bit: 0 -> is anomalous, 1 -> not anomalous
1701
flags &= ~((storage_number) SN_FLAG_NOT_ANOMALOUS);
1702
}
configure.ac
+1
-1
@@ -1140,7 +1140,7 @@ fi
1140
1141
# Check if submodules have not been fetched. Fail if ML was explicitly requested.
1142
AC_MSG_CHECKING([if git submodules are present for machine learning functionality])
1143
-if test -f "ml/dlib/dlib/all/source.cpp" -a -f "ml/json/single_include/nlohmann/json.hpp"; then
1143
+if test -f "ml/dlib/dlib/all/source.cpp"; then
1144
AC_MSG_RESULT([yes])
1145
have_ml_submodules="yes"
1146
else
database/rrd.h
+8
-8
@@ -30,9 +30,9 @@ typedef struct rrdhost_acquired RRDHOST_ACQUIRED;
30
typedef struct rrdset_acquired RRDSET_ACQUIRED;
31
typedef struct rrddim_acquired RRDDIM_ACQUIRED;
32
33
-typedef struct ml_host ml_host_t;
34
-typedef struct ml_chart ml_chart_t;
35
-typedef struct ml_dimension ml_dimension_t;
33
+typedef struct ml_host rrd_ml_host_t;
34
+typedef struct ml_chart rrd_ml_chart_t;
35
+typedef struct ml_dimension rrd_ml_dimension_t;
36
37
typedef enum __attribute__ ((__packed__)) {
38
QUERY_SOURCE_UNKNOWN = 0,
@@ -363,7 +363,7 @@ struct rrddim {
363
// ------------------------------------------------------------------------
364
// operational state members
365
366
- ml_dimension_t *ml_dimension; // machine learning data about this dimension
366
+ rrd_ml_dimension_t *ml_dimension; // machine learning data about this dimension
367
368
// ------------------------------------------------------------------------
369
// linking to siblings and parents
@@ -626,7 +626,7 @@ struct rrdset {
626
DICTIONARY *rrddimvar_root_index; // dimension variables
627
// we use this dictionary to manage their allocation
628
629
- ml_chart_t *ml_chart;
629
+ rrd_ml_chart_t *ml_chart;
630
631
// ------------------------------------------------------------------------
632
// operational state members
@@ -1067,7 +1067,7 @@ struct rrdhost {
1067
1068
// ------------------------------------------------------------------------
1069
// ML handle
1070
- ml_host_t *ml_host;
1070
+ rrd_ml_host_t *ml_host;
1071
1072
// ------------------------------------------------------------------------
1073
// Support for host-level labels
@@ -1358,13 +1358,13 @@ void rrdset_delete_files(RRDSET *st);
1358
void rrdset_save(RRDSET *st);
1359
void rrdset_free(RRDSET *st);
1360
1361
+void rrddim_free(RRDSET *st, RRDDIM *rd);
1362
+
1363
#ifdef NETDATA_RRD_INTERNALS
1364
1365
char *rrdhost_cache_dir_for_rrdset_alloc(RRDHOST *host, const char *id);
1366
const char *rrdset_cache_dir(RRDSET *st);
1367
1366
-void rrddim_free(RRDSET *st, RRDDIM *rd);
1367
-
1368
void rrdset_reset(RRDSET *st);
1369
void rrdset_delete_obsolete_dimensions(RRDSET *st);
1370
database/rrdhost.c
+3
-3
@@ -525,7 +525,7 @@ int is_legacy = 1;
525
rrdhost_load_rrdcontext_data(host);
526
if (!archived) {
527
ml_host_new(host);
528
- ml_start_training_thread(host);
528
+ ml_host_start_training_thread(host);
529
} else
530
rrdhost_flag_set(host, RRDHOST_FLAG_ARCHIVED | RRDHOST_FLAG_ORPHAN);
531
@@ -642,7 +642,7 @@ static void rrdhost_update(RRDHOST *host
642
host->rrdpush_replication_step = rrdpush_replication_step;
643
644
ml_host_new(host);
645
- ml_start_training_thread(host);
645
+ ml_host_start_training_thread(host);
646
647
rrdhost_load_rrdcontext_data(host);
648
info("Host %s is not in archived mode anymore", rrdhost_hostname(host));
@@ -1145,7 +1145,7 @@ void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host, bool force) {
1145
rrdcalctemplate_index_destroy(host);
1146
1147
// cleanup ML resources
1148
- ml_stop_training_thread(host);
1148
+ ml_host_stop_training_thread(host);
1149
ml_host_delete(host);
1150
1151
freez(host->exporting_flags);
database/rrdset.c
+3
-3
@@ -1406,7 +1406,7 @@ static inline size_t rrdset_done_interpolate(
1406
time_t current_time_s = (time_t) (next_store_ut / USEC_PER_SEC);
1407
1408
if(unlikely(!store_this_entry)) {
1409
- (void) ml_is_anomalous(rd, current_time_s, 0, false);
1409
+ (void) ml_dimension_is_anomalous(rd, current_time_s, 0, false);
1410
1411
if(rsb->wb && rsb->v2)
1412
rrddim_push_metrics_v2(rsb, rd, next_store_ut, NAN, SN_FLAG_NONE);
@@ -1418,7 +1418,7 @@ static inline size_t rrdset_done_interpolate(
1418
if(likely(rd->updated && rd->collections_counter > 1 && iterations < gap_when_lost_iterations_above)) {
1419
uint32_t dim_storage_flags = storage_flags;
1420
1421
- if (ml_is_anomalous(rd, current_time_s, new_value, true)) {
1421
+ if (ml_dimension_is_anomalous(rd, current_time_s, new_value, true)) {
1422
// clear anomaly bit: 0 -> is anomalous, 1 -> not anomalous
1423
dim_storage_flags &= ~((storage_number)SN_FLAG_NOT_ANOMALOUS);
1424
}
@@ -1430,7 +1430,7 @@ static inline size_t rrdset_done_interpolate(
1430
rd->last_stored_value = new_value;
1431
}
1432
else {
1433
- (void) ml_is_anomalous(rd, current_time_s, 0, false);
1433
+ (void) ml_dimension_is_anomalous(rd, current_time_s, 0, false);
1434
1435
rrdset_debug(st, "%s: STORE[%ld] = NON EXISTING ", rrddim_name(rd), current_entry);
1436
ml/Config.cc
+3
-3
@@ -1,12 +1,12 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
-#include "nml.h"
3
+#include "ml-private.h"
4
5
/*
6
* Global configuration instance to be shared between training and
7
* prediction threads.
8
*/
9
-nml_config_t Cfg;
9
+ml_config_t Cfg;
10
11
template <typename T>
12
static T clamp(const T& Value, const T& Min, const T& Max) {
@@ -16,7 +16,7 @@ static T clamp(const T& Value, const T& Min, const T& Max) {
16
/*
17
* Initialize global configuration variable.
18
*/
19
-void nml_config_load(nml_config_t *cfg) {
19
+void ml_config_load(ml_config_t *cfg) {
20
const char *config_section_ml = CONFIG_SECTION_ML;
21
22
bool enable_anomaly_detection = config_get_boolean(config_section_ml, "enabled", true);
ml/ad_charts.cc
+3
-3
@@ -2,7 +2,7 @@
2
3
#include "ad_charts.h"
4
5
-void nml_update_dimensions_chart(nml_host_t *host, const nml_machine_learning_stats_t &mls) {
5
+void ml_update_dimensions_chart(ml_host_t *host, const ml_machine_learning_stats_t &mls) {
6
/*
7
* Machine learning status
8
*/
@@ -182,7 +182,7 @@ void nml_update_dimensions_chart(nml_host_t *host, const nml_machine_learning_st
182
183
}
184
185
-void nml_update_host_and_detection_rate_charts(nml_host_t *host, collected_number AnomalyRate) {
185
+void ml_update_host_and_detection_rate_charts(ml_host_t *host, collected_number AnomalyRate) {
186
/*
187
* Anomaly rate
188
*/
@@ -301,7 +301,7 @@ void nml_update_host_and_detection_rate_charts(nml_host_t *host, collected_numbe
301
}
302
}
303
304
-void nml_update_training_statistics_chart(nml_host_t *host, const nml_training_stats_t &ts) {
304
+void ml_update_training_statistics_chart(ml_host_t *host, const ml_training_stats_t &ts) {
305
/*
306
* queue stats
307
*/
ml/ad_charts.h
+4
-4
@@ -3,12 +3,12 @@
3
#ifndef ML_ADCHARTS_H
4
#define ML_ADCHARTS_H
5
6
-#include "nml.h"
6
+#include "ml-private.h"
7
8
-void nml_update_dimensions_chart(nml_host_t *host, const nml_machine_learning_stats_t &mls);
8
+void ml_update_dimensions_chart(ml_host_t *host, const ml_machine_learning_stats_t &mls);
9
10
-void nml_update_host_and_detection_rate_charts(nml_host_t *host, collected_number anomaly_rate);
10
+void ml_update_host_and_detection_rate_charts(ml_host_t *host, collected_number anomaly_rate);
11
12
-void nml_update_training_statistics_chart(nml_host_t *host, const nml_training_stats_t &ts);
12
+void ml_update_training_statistics_chart(ml_host_t *host, const ml_training_stats_t &ts);
13
14
#endif /* ML_ADCHARTS_H */
ml/json
deleted
-1
@@ -1 +0,0 @@
1
-Subproject commit 0b345b20c888f7dc8888485768e4bf9a6be29de0
ml/ml-dummy.c
+41
-38
@@ -8,78 +8,81 @@ bool ml_capable() {
8
return false;
9
}
10
11
-bool ml_enabled(RRDHOST *RH) {
12
- (void) RH;
11
+bool ml_enabled(RRDHOST *rh) {
12
+ UNUSED(rh);
13
+ return false;
14
+}
15
+
16
+bool ml_streaming_enabled() {
17
return false;
18
}
19
20
void ml_init(void) {}
21
18
-void ml_host_new(RRDHOST *RH) {
19
- UNUSED(RH);
22
+void ml_host_new(RRDHOST *rh) {
23
+ UNUSED(rh);
24
}
25
22
-void ml_host_delete(RRDHOST *RH) {
23
- UNUSED(RH);
26
+void ml_host_delete(RRDHOST *rh) {
27
+ UNUSED(rh);
28
}
29
26
-void ml_chart_new(RRDSET *RS) {
27
- UNUSED(RS);
30
+void ml_host_start_training_thread(RRDHOST *rh) {
31
+ UNUSED(rh);
32
}
33
30
-void ml_chart_delete(RRDSET *RS) {
31
- UNUSED(RS);
34
+void ml_host_stop_training_thread(RRDHOST *rh) {
35
+ UNUSED(rh);
36
}
37
34
-void ml_dimension_new(RRDDIM *RD) {
35
- UNUSED(RD);
38
+void ml_host_cancel_training_thread(RRDHOST *rh) {
39
+ UNUSED(rh);
40
}
41
38
-void ml_dimension_delete(RRDDIM *RD) {
39
- UNUSED(RD);
42
+void ml_host_get_info(RRDHOST *rh, BUFFER *wb) {
43
+ UNUSED(rh);
44
+ UNUSED(wb);
45
}
46
42
-void ml_start_training_thread(RRDHOST *RH) {
43
- UNUSED(RH);
47
+void ml_host_get_models(RRDHOST *rh, BUFFER *wb) {
48
+ UNUSED(rh);
49
+ UNUSED(wb);
50
}
51
46
-void ml_stop_training_thread(RRDHOST *RH) {
47
- UNUSED(RH);
52
+void ml_host_get_runtime_info(RRDHOST *rh) {
53
+ UNUSED(rh);
54
}
55
50
-void ml_get_host_info(RRDHOST *RH, BUFFER *wb) {
51
- (void) RH;
52
- (void) wb;
56
+void ml_chart_new(RRDSET *rs) {
57
+ UNUSED(rs);
58
}
59
55
-char *ml_get_host_runtime_info(RRDHOST *RH) {
56
- (void) RH;
57
- return NULL;
60
+void ml_chart_delete(RRDSET *rs) {
61
+ UNUSED(rs);
62
}
63
60
-bool ml_chart_update_begin(RRDSET *RS) {
61
- (void) RS;
64
+bool ml_chart_update_begin(RRDSET *rs) {
65
+ UNUSED(rs);
66
return false;
67
}
68
65
-void ml_chart_update_end(RRDSET *RS) {
66
- (void) RS;
69
+void ml_chart_update_end(RRDSET *rs) {
70
+ UNUSED(rs);
71
}
72
69
-char *ml_get_host_models(RRDHOST *RH) {
70
- (void) RH;
71
- return NULL;
73
+void ml_dimension_new(RRDDIM *rd) {
74
+ UNUSED(rd);
75
}
76
74
-bool ml_is_anomalous(RRDDIM *RD, time_t CurrT, double Value, bool Exists) {
75
- (void) RD;
76
- (void) CurrT;
77
- (void) Value;
78
- (void) Exists;
79
- return false;
77
+void ml_dimension_delete(RRDDIM *rd) {
78
+ UNUSED(rd);
79
}
80
82
-bool ml_streaming_enabled() {
81
+bool ml_dimension_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool exists) {
82
+ UNUSED(rd);
83
+ UNUSED(curr_time);
84
+ UNUSED(value);
85
+ UNUSED(exists);
86
return false;
87
}
88
ml/ml-private.h
renamed
+33
-68
@@ -1,7 +1,7 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
-#ifndef NETDATA_NML_H
4
-#define NETDATA_NML_H
3
+#ifndef NETDATA_ML_PRIVATE_H
4
+#define NETDATA_ML_PRIVATE_H
5
6
#include "dlib/matrix.h"
7
#include "ml/ml.h"
@@ -28,7 +28,7 @@ typedef struct {
28
size_t src_n;
29
30
std::vector<DSample> &preprocessed_features;
31
-} nml_features_t;
31
+} ml_features_t;
32
33
/*
34
* KMeans
@@ -41,9 +41,7 @@ typedef struct {
41
42
calculated_number_t min_dist;
43
calculated_number_t max_dist;
44
-} nml_kmeans_t;
45
-
46
-#include "json/single_include/nlohmann/json.hpp"
44
+} ml_kmeans_t;
45
46
typedef struct machine_learning_stats_t {
47
size_t num_machine_learning_status_enabled;
@@ -59,11 +57,9 @@ typedef struct machine_learning_stats_t {
57
58
size_t num_anomalous_dimensions;
59
size_t num_normal_dimensions;
62
-} nml_machine_learning_stats_t;
60
+} ml_machine_learning_stats_t;
61
62
typedef struct training_stats_t {
65
- struct rusage training_ru;
66
-
63
size_t queue_size;
64
size_t num_popped_items;
65
@@ -76,9 +72,9 @@ typedef struct training_stats_t {
72
size_t training_result_not_enough_collected_values;
73
size_t training_result_null_acquired_dimension;
74
size_t training_result_chart_under_replication;
79
-} nml_training_stats_t;
75
+} ml_training_stats_t;
76
81
-enum nml_metric_type {
77
+enum ml_metric_type {
78
// The dimension has constant values, no need to train
79
METRIC_TYPE_CONSTANT,
80
@@ -86,7 +82,7 @@ enum nml_metric_type {
82
METRIC_TYPE_VARIABLE,
83
};
84
89
-enum nml_machine_learning_status {
85
+enum ml_machine_learning_status {
86
// Enable training/prediction
87
MACHINE_LEARNING_STATUS_ENABLED,
88
@@ -94,7 +90,7 @@ enum nml_machine_learning_status {
90
MACHINE_LEARNING_STATUS_DISABLED_DUE_TO_EXCLUDED_CHART,
91
};
92
97
-enum nml_training_status {
93
+enum ml_training_status {
94
// We don't have a model for this dimension
95
TRAINING_STATUS_UNTRAINED,
96
@@ -108,7 +104,7 @@ enum nml_training_status {
104
TRAINING_STATUS_TRAINED,
105
};
106
111
-enum nml_training_result {
107
+enum ml_training_result {
108
// We managed to create a KMeans model
109
TRAINING_RESULT_OK,
110
@@ -137,7 +133,7 @@ typedef struct {
133
// at the point the request was made
134
time_t first_entry_on_request;
135
time_t last_entry_on_request;
140
-} nml_training_request_t;
136
+} ml_training_request_t;
137
138
typedef struct {
139
// Time when the request for this response was made
@@ -166,71 +162,52 @@ typedef struct {
162
size_t total_values;
163
164
// Result of training response
169
- enum nml_training_result result;
170
-} nml_training_response_t;
165
+ enum ml_training_result result;
166
+} ml_training_response_t;
167
168
/*
169
* Queue
170
*/
175
-
171
typedef struct {
177
- std::queue<nml_training_request_t> internal;
172
+ std::queue<ml_training_request_t> internal;
173
netdata_mutex_t mutex;
174
pthread_cond_t cond_var;
175
std::atomic<bool> exit;
181
-} nml_queue_t;
176
+} ml_queue_t;
177
183
-nml_queue_t *nml_queue_init(void);
184
-void nml_queue_destroy(nml_queue_t *q);
185
-
186
-void nml_queue_push(nml_queue_t *q, const nml_training_request_t req);
187
-nml_training_request_t nml_queue_pop(nml_queue_t *q);
188
-size_t nml_queue_size(nml_queue_t *q);
189
-
190
-void nml_queue_signal(nml_queue_t *q);
178
179
typedef struct {
180
RRDDIM *rd;
181
195
- enum nml_metric_type mt;
196
- enum nml_training_status ts;
197
- enum nml_machine_learning_status mls;
182
+ enum ml_metric_type mt;
183
+ enum ml_training_status ts;
184
+ enum ml_machine_learning_status mls;
185
199
- nml_training_response_t tr;
186
+ ml_training_response_t tr;
187
time_t last_training_time;
188
189
std::vector<calculated_number_t> cns;
190
204
- std::vector<nml_kmeans_t> km_contexts;
191
+ std::vector<ml_kmeans_t> km_contexts;
192
netdata_mutex_t mutex;
206
- nml_kmeans_t kmeans;
193
+ ml_kmeans_t kmeans;
194
std::vector<DSample> feature;
208
-} nml_dimension_t;
209
-
210
-nml_dimension_t *nml_dimension_new(RRDDIM *rd);
211
-void nml_dimension_delete(nml_dimension_t *dim);
212
-
213
-bool nml_dimension_predict(nml_dimension_t *d, time_t curr_t, calculated_number_t value, bool exists);
195
+} ml_dimension_t;
196
197
typedef struct {
198
RRDSET *rs;
217
- nml_machine_learning_stats_t mls;
199
+ ml_machine_learning_stats_t mls;
200
201
netdata_mutex_t mutex;
220
-} nml_chart_t;
221
-
222
-nml_chart_t *nml_chart_new(RRDSET *rs);
223
-void nml_chart_delete(nml_chart_t *chart);
202
+} ml_chart_t;
203
225
-void nml_chart_update_begin(nml_chart_t *chart);
226
-void nml_chart_update_end(nml_chart_t *chart);
227
-void nml_chart_update_dimension(nml_chart_t *chart, nml_dimension_t *dim, bool is_anomalous);
204
+void ml_chart_update_dimension(ml_chart_t *chart, ml_dimension_t *dim, bool is_anomalous);
205
206
typedef struct {
207
RRDHOST *rh;
208
232
- nml_machine_learning_stats_t mls;
233
- nml_training_stats_t ts;
209
+ ml_machine_learning_stats_t mls;
210
+ ml_training_stats_t ts;
211
212
calculated_number_t host_anomaly_rate;
213
@@ -238,7 +215,7 @@ typedef struct {
215
std::atomic<bool> threads_cancelled;
216
std::atomic<bool> threads_joined;
217
241
- nml_queue_t *training_queue;
218
+ ml_queue_t *training_queue;
219
220
netdata_mutex_t mutex;
221
@@ -288,17 +265,7 @@ typedef struct {
265
RRDDIM *training_results_not_enough_collected_values_rd;
266
RRDDIM *training_results_null_acquired_dimension_rd;
267
RRDDIM *training_results_chart_under_replication_rd;
291
-} nml_host_t;
292
-
293
-nml_host_t *nml_host_new(RRDHOST *rh);
294
-void nml_host_delete(nml_host_t *host);
295
-
296
-void nml_host_start_training_thread(nml_host_t *host);
297
-void nml_host_stop_training_thread(nml_host_t *host, bool join);
298
-
299
-void nml_host_get_config_as_json(nml_host_t *host, BUFFER *wb);
300
-void nml_host_get_models_as_json(nml_host_t *host, nlohmann::json &j);
301
-void nml_host_get_detection_info_as_json(nml_host_t *host, nlohmann::json &j);
268
+} ml_host_t;
269
270
typedef struct {
271
bool enable_anomaly_detection;
@@ -335,12 +302,10 @@ typedef struct {
302
std::vector<uint32_t> random_nums;
303
304
netdata_thread_t detection_thread;
338
-} nml_config_t;
339
-
340
-void nml_config_load(nml_config_t *cfg);
305
+} ml_config_t;
306
342
-void *nml_detect_main(void *arg);
307
+void ml_config_load(ml_config_t *cfg);
308
344
-extern nml_config_t Cfg;
309
+extern ml_config_t Cfg;
310
346
-#endif /* NETDATA_NML_H */
311
+#endif /* NETDATA_ML_PRIVATE_H */
ml/ml.cc
+1169
-93
@@ -1,14 +1,969 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
-#include "nml.h"
3
+#include <dlib/clustering.h>
4
+
5
+#include "ml-private.h"
6
7
#include <random>
8
7
-bool ml_capable() {
9
+#include "ad_charts.h"
10
+
11
+typedef struct {
12
+ calculated_number_t *training_cns;
13
+ calculated_number_t *scratch_training_cns;
14
+
15
+ std::vector<DSample> training_samples;
16
+} ml_tls_data_t;
17
+
18
+static thread_local ml_tls_data_t tls_data;
19
+
20
+/*
21
+ * Functions to convert enums to strings
22
+*/
23
+
24
+__attribute__((unused)) static const char *
25
+ml_machine_learning_status_to_string(enum ml_machine_learning_status mls)
26
+{
27
+ switch (mls) {
28
+ case MACHINE_LEARNING_STATUS_ENABLED:
29
+ return "enabled";
30
+ case MACHINE_LEARNING_STATUS_DISABLED_DUE_TO_EXCLUDED_CHART:
31
+ return "disabled-sp";
32
+ default:
33
+ return "unknown";
34
+ }
35
+}
36
+
37
+__attribute__((unused)) static const char *
38
+ml_metric_type_to_string(enum ml_metric_type mt)
39
+{
40
+ switch (mt) {
41
+ case METRIC_TYPE_CONSTANT:
42
+ return "constant";
43
+ case METRIC_TYPE_VARIABLE:
44
+ return "variable";
45
+ default:
46
+ return "unknown";
47
+ }
48
+}
49
+
50
+__attribute__((unused)) static const char *
51
+ml_training_status_to_string(enum ml_training_status ts)
52
+{
53
+ switch (ts) {
54
+ case TRAINING_STATUS_PENDING_WITH_MODEL:
55
+ return "pending-with-model";
56
+ case TRAINING_STATUS_PENDING_WITHOUT_MODEL:
57
+ return "pending-without-model";
58
+ case TRAINING_STATUS_TRAINED:
59
+ return "trained";
60
+ case TRAINING_STATUS_UNTRAINED:
61
+ return "untrained";
62
+ default:
63
+ return "unknown";
64
+ }
65
+}
66
+
67
+__attribute__((unused)) static const char *
68
+ml_training_result_to_string(enum ml_training_result tr)
69
+{
70
+ switch (tr) {
71
+ case TRAINING_RESULT_OK:
72
+ return "ok";
73
+ case TRAINING_RESULT_INVALID_QUERY_TIME_RANGE:
74
+ return "invalid-query";
75
+ case TRAINING_RESULT_NOT_ENOUGH_COLLECTED_VALUES:
76
+ return "missing-values";
77
+ case TRAINING_RESULT_NULL_ACQUIRED_DIMENSION:
78
+ return "null-acquired-dim";
79
+ case TRAINING_RESULT_CHART_UNDER_REPLICATION:
80
+ return "chart-under-replication";
81
+ default:
82
+ return "unknown";
83
+ }
84
+}
85
+
86
+/*
87
+ * Features
88
+*/
89
+
90
+// subtract elements that are `diff_n` positions apart
91
+static void
92
+ml_features_diff(ml_features_t *features)
93
+{
94
+ if (features->diff_n == 0)
95
+ return;
96
+
97
+ for (size_t idx = 0; idx != (features->src_n - features->diff_n); idx++) {
98
+ size_t high = (features->src_n - 1) - idx;
99
+ size_t low = high - features->diff_n;
100
+
101
+ features->dst[low] = features->src[high] - features->src[low];
102
+ }
103
+
104
+ size_t n = features->src_n - features->diff_n;
105
+ memcpy(features->src, features->dst, n * sizeof(calculated_number_t));
106
+
107
+ for (size_t idx = features->src_n - features->diff_n; idx != features->src_n; idx++)
108
+ features->src[idx] = 0.0;
109
+}
110
+
111
+// a function that computes the window average of an array inplace
112
+static void
113
+ml_features_smooth(ml_features_t *features)
114
+{
115
+ calculated_number_t sum = 0.0;
116
+
117
+ size_t idx = 0;
118
+ for (; idx != features->smooth_n - 1; idx++)
119
+ sum += features->src[idx];
120
+
121
+ for (; idx != (features->src_n - features->diff_n); idx++) {
122
+ sum += features->src[idx];
123
+ calculated_number_t prev_cn = features->src[idx - (features->smooth_n - 1)];
124
+ features->src[idx - (features->smooth_n - 1)] = sum / features->smooth_n;
125
+ sum -= prev_cn;
126
+ }
127
+
128
+ for (idx = 0; idx != features->smooth_n; idx++)
129
+ features->src[(features->src_n - 1) - idx] = 0.0;
130
+}
131
+
132
+// create lag'd vectors out of the preprocessed buffer
133
+static void
134
+ml_features_lag(ml_features_t *features)
135
+{
136
+ size_t n = features->src_n - features->diff_n - features->smooth_n + 1 - features->lag_n;
137
+ features->preprocessed_features.resize(n);
138
+
139
+ unsigned target_num_samples = Cfg.max_train_samples * Cfg.random_sampling_ratio;
140
+ double sampling_ratio = std::min(static_cast<double>(target_num_samples) / n, 1.0);
141
+
142
+ uint32_t max_mt = std::numeric_limits<uint32_t>::max();
143
+ uint32_t cutoff = static_cast<double>(max_mt) * sampling_ratio;
144
+
145
+ size_t sample_idx = 0;
146
+
147
+ for (size_t idx = 0; idx != n; idx++) {
148
+ DSample &DS = features->preprocessed_features[sample_idx++];
149
+ DS.set_size(features->lag_n);
150
+
151
+ if (Cfg.random_nums[idx] > cutoff) {
152
+ sample_idx--;
153
+ continue;
154
+ }
155
+
156
+ for (size_t feature_idx = 0; feature_idx != features->lag_n + 1; feature_idx++)
157
+ DS(feature_idx) = features->src[idx + feature_idx];
158
+ }
159
+
160
+ features->preprocessed_features.resize(sample_idx);
161
+}
162
+
163
+static void
164
+ml_features_preprocess(ml_features_t *features)
165
+{
166
+ ml_features_diff(features);
167
+ ml_features_smooth(features);
168
+ ml_features_lag(features);
169
+}
170
+
171
+/*
172
+ * KMeans
173
+*/
174
+
175
+static void
176
+ml_kmeans_init(ml_kmeans_t *kmeans, size_t num_clusters, size_t max_iterations)
177
+{
178
+ kmeans->num_clusters = num_clusters;
179
+ kmeans->max_iterations = max_iterations;
180
+
181
+ kmeans->cluster_centers.reserve(kmeans->num_clusters);
182
+ kmeans->min_dist = std::numeric_limits<calculated_number_t>::max();
183
+ kmeans->max_dist = std::numeric_limits<calculated_number_t>::min();
184
+}
185
+
186
+static void
187
+ml_kmeans_train(ml_kmeans_t *kmeans, const ml_features_t *features)
188
+{
189
+ kmeans->min_dist = std::numeric_limits<calculated_number_t>::max();
190
+ kmeans->max_dist = std::numeric_limits<calculated_number_t>::min();
191
+
192
+ kmeans->cluster_centers.clear();
193
+
194
+ dlib::pick_initial_centers(kmeans->num_clusters, kmeans->cluster_centers, features->preprocessed_features);
195
+ dlib::find_clusters_using_kmeans(features->preprocessed_features, kmeans->cluster_centers, kmeans->max_iterations);
196
+
197
+ for (const auto &preprocessed_feature : features->preprocessed_features) {
198
+ calculated_number_t mean_dist = 0.0;
199
+
200
+ for (const auto &cluster_center : kmeans->cluster_centers) {
201
+ mean_dist += dlib::length(cluster_center - preprocessed_feature);
202
+ }
203
+
204
+ mean_dist /= kmeans->num_clusters;
205
+
206
+ if (mean_dist < kmeans->min_dist)
207
+ kmeans->min_dist = mean_dist;
208
+
209
+ if (mean_dist > kmeans->max_dist)
210
+ kmeans->max_dist = mean_dist;
211
+ }
212
+}
213
+
214
+static calculated_number_t
215
+ml_kmeans_anomaly_score(const ml_kmeans_t *kmeans, const DSample &DS)
216
+{
217
+ calculated_number_t mean_dist = 0.0;
218
+ for (const auto &CC: kmeans->cluster_centers)
219
+ mean_dist += dlib::length(CC - DS);
220
+
221
+ mean_dist /= kmeans->num_clusters;
222
+
223
+ if (kmeans->max_dist == kmeans->min_dist)
224
+ return 0.0;
225
+
226
+ calculated_number_t anomaly_score = 100.0 * std::abs((mean_dist - kmeans->min_dist) / (kmeans->max_dist - kmeans->min_dist));
227
+ return (anomaly_score > 100.0) ? 100.0 : anomaly_score;
228
+}
229
+
230
+/*
231
+ * Queue
232
+*/
233
+
234
+static ml_queue_t *
235
+ml_queue_init()
236
+{
237
+ ml_queue_t *q = new ml_queue_t();
238
+
239
+ netdata_mutex_init(&q->mutex);
240
+ pthread_cond_init(&q->cond_var, NULL);
241
+ q->exit = false;
242
+ return q;
243
+}
244
+
245
+static void
246
+ml_queue_destroy(ml_queue_t *q)
247
+{
248
+ netdata_mutex_destroy(&q->mutex);
249
+ pthread_cond_destroy(&q->cond_var);
250
+ delete q;
251
+}
252
+
253
+static void
254
+ml_queue_push(ml_queue_t *q, const ml_training_request_t req)
255
+{
256
+ netdata_mutex_lock(&q->mutex);
257
+ q->internal.push(req);
258
+ pthread_cond_signal(&q->cond_var);
259
+ netdata_mutex_unlock(&q->mutex);
260
+}
261
+
262
+static ml_training_request_t
263
+ml_queue_pop(ml_queue_t *q)
264
+{
265
+ netdata_mutex_lock(&q->mutex);
266
+
267
+ ml_training_request_t req = { NULL, NULL, 0, 0, 0 };
268
+
269
+ while (q->internal.empty()) {
270
+ pthread_cond_wait(&q->cond_var, &q->mutex);
271
+
272
+ if (q->exit) {
273
+ netdata_mutex_unlock(&q->mutex);
274
+
275
+ // We return a dummy request because the queue has been signaled
276
+ return req;
277
+ }
278
+ }
279
+
280
+ req = q->internal.front();
281
+ q->internal.pop();
282
+
283
+ netdata_mutex_unlock(&q->mutex);
284
+ return req;
285
+}
286
+
287
+static size_t
288
+ml_queue_size(ml_queue_t *q)
289
+{
290
+ netdata_mutex_lock(&q->mutex);
291
+ size_t size = q->internal.size();
292
+ netdata_mutex_unlock(&q->mutex);
293
+ return size;
294
+}
295
+
296
+static void
297
+ml_queue_signal(ml_queue_t *q)
298
+{
299
+ netdata_mutex_lock(&q->mutex);
300
+ q->exit = true;
301
+ pthread_cond_signal(&q->cond_var);
302
+ netdata_mutex_unlock(&q->mutex);
303
+}
304
+
305
+/*
306
+ * Dimension
307
+*/
308
+
309
+static std::pair<calculated_number_t *, ml_training_response_t>
310
+ml_dimension_calculated_numbers(ml_dimension_t *dim, const ml_training_request_t &training_request)
311
+{
312
+ ml_training_response_t training_response = {};
313
+
314
+ training_response.request_time = training_request.request_time;
315
+ training_response.first_entry_on_request = training_request.first_entry_on_request;
316
+ training_response.last_entry_on_request = training_request.last_entry_on_request;
317
+
318
+ training_response.first_entry_on_response = rrddim_first_entry_s_of_tier(dim->rd, 0);
319
+ training_response.last_entry_on_response = rrddim_last_entry_s_of_tier(dim->rd, 0);
320
+
321
+ size_t min_n = Cfg.min_train_samples;
322
+ size_t max_n = Cfg.max_train_samples;
323
+
324
+ // Figure out what our time window should be.
325
+ training_response.query_before_t = training_response.last_entry_on_response;
326
+ training_response.query_after_t = std::max(
327
+ training_response.query_before_t - static_cast<time_t>((max_n - 1) * dim->rd->update_every),
328
+ training_response.first_entry_on_response
329
+ );
330
+
331
+ if (training_response.query_after_t >= training_response.query_before_t) {
332
+ training_response.result = TRAINING_RESULT_INVALID_QUERY_TIME_RANGE;
333
+ return { NULL, training_response };
334
+ }
335
+
336
+ if (rrdset_is_replicating(dim->rd->rrdset)) {
337
+ training_response.result = TRAINING_RESULT_CHART_UNDER_REPLICATION;
338
+ return { NULL, training_response };
339
+ }
340
+
341
+ /*
342
+ * Execute the query
343
+ */
344
+ struct storage_engine_query_ops *ops = dim->rd->tiers[0].query_ops;
345
+ struct storage_engine_query_handle handle;
346
+
347
+ ops->init(dim->rd->tiers[0].db_metric_handle,
348
+ &handle,
349
+ training_response.query_after_t,
350
+ training_response.query_before_t,
351
+ STORAGE_PRIORITY_BEST_EFFORT);
352
+
353
+ size_t idx = 0;
354
+ memset(tls_data.training_cns, 0, sizeof(calculated_number_t) * max_n * (Cfg.lag_n + 1));
355
+ calculated_number_t last_value = std::numeric_limits<calculated_number_t>::quiet_NaN();
356
+
357
+ while (!ops->is_finished(&handle)) {
358
+ if (idx == max_n)
359
+ break;
360
+
361
+ STORAGE_POINT sp = ops->next_metric(&handle);
362
+
363
+ time_t timestamp = sp.end_time_s;
364
+ calculated_number_t value = sp.sum / sp.count;
365
+
366
+ if (netdata_double_isnumber(value)) {
367
+ if (!training_response.db_after_t)
368
+ training_response.db_after_t = timestamp;
369
+ training_response.db_before_t = timestamp;
370
+
371
+ tls_data.training_cns[idx] = value;
372
+ last_value = tls_data.training_cns[idx];
373
+ training_response.collected_values++;
374
+ } else
375
+ tls_data.training_cns[idx] = last_value;
376
+
377
+ idx++;
378
+ }
379
+ ops->finalize(&handle);
380
+
381
+ global_statistics_ml_query_completed(/* points_read */ idx);
382
+
383
+ training_response.total_values = idx;
384
+ if (training_response.collected_values < min_n) {
385
+ training_response.result = TRAINING_RESULT_NOT_ENOUGH_COLLECTED_VALUES;
386
+ return { NULL, training_response };
387
+ }
388
+
389
+ // Find first non-NaN value.
390
+ for (idx = 0; std::isnan(tls_data.training_cns[idx]); idx++, training_response.total_values--) { }
391
+
392
+ // Overwrite NaN values.
393
+ if (idx != 0)
394
+ memmove(tls_data.training_cns, &tls_data.training_cns[idx], sizeof(calculated_number_t) * training_response.total_values);
395
+
396
+ training_response.result = TRAINING_RESULT_OK;
397
+ return { tls_data.training_cns, training_response };
398
+}
399
+
400
+static enum ml_training_result
401
+ml_dimension_train_model(ml_dimension_t *dim, const ml_training_request_t &training_request)
402
+{
403
+ auto P = ml_dimension_calculated_numbers(dim, training_request);
404
+ ml_training_response_t training_response = P.second;
405
+
406
+ if (training_response.result != TRAINING_RESULT_OK) {
407
+ netdata_mutex_lock(&dim->mutex);
408
+
409
+ dim->mt = METRIC_TYPE_CONSTANT;
410
+
411
+ switch (dim->ts) {
412
+ case TRAINING_STATUS_PENDING_WITH_MODEL:
413
+ dim->ts = TRAINING_STATUS_TRAINED;
414
+ break;
415
+ case TRAINING_STATUS_PENDING_WITHOUT_MODEL:
416
+ dim->ts = TRAINING_STATUS_UNTRAINED;
417
+ break;
418
+ default:
419
+ break;
420
+ }
421
+
422
+ dim->tr = training_response;
423
+
424
+ dim->last_training_time = training_response.last_entry_on_response;
425
+ enum ml_training_result result = training_response.result;
426
+ netdata_mutex_unlock(&dim->mutex);
427
+
428
+ return result;
429
+ }
430
+
431
+ // compute kmeans
432
+ {
433
+ memcpy(tls_data.scratch_training_cns, tls_data.training_cns,
434
+ training_response.total_values * sizeof(calculated_number_t));
435
+
436
+ ml_features_t features = {
437
+ Cfg.diff_n, Cfg.smooth_n, Cfg.lag_n,
438
+ tls_data.scratch_training_cns, training_response.total_values,
439
+ tls_data.training_cns, training_response.total_values,
440
+ tls_data.training_samples
441
+ };
442
+ ml_features_preprocess(&features);
443
+
444
+ ml_kmeans_init(&dim->kmeans, 2, 1000);
445
+ ml_kmeans_train(&dim->kmeans, &features);
446
+ }
447
+
448
+ // update kmeans models
449
+ {
450
+ netdata_mutex_lock(&dim->mutex);
451
+
452
+ if (dim->km_contexts.size() < Cfg.num_models_to_use) {
453
+ dim->km_contexts.push_back(std::move(dim->kmeans));
454
+ } else {
455
+ std::rotate(std::begin(dim->km_contexts), std::begin(dim->km_contexts) + 1, std::end(dim->km_contexts));
456
+ dim->km_contexts[dim->km_contexts.size() - 1] = std::move(dim->kmeans);
457
+ }
458
+
459
+ dim->mt = METRIC_TYPE_CONSTANT;
460
+ dim->ts = TRAINING_STATUS_TRAINED;
461
+ dim->tr = training_response;
462
+ dim->last_training_time = rrddim_last_entry_s(dim->rd);
463
+
464
+ netdata_mutex_unlock(&dim->mutex);
465
+ }
466
+
467
+ return training_response.result;
468
+}
469
+
470
+static void
471
+ml_dimension_schedule_for_training(ml_dimension_t *dim, time_t curr_time)
472
+{
473
+ switch (dim->mt) {
474
+ case METRIC_TYPE_CONSTANT:
475
+ return;
476
+ default:
477
+ break;
478
+ }
479
+
480
+ bool schedule_for_training = false;
481
+
482
+ switch (dim->ts) {
483
+ case TRAINING_STATUS_PENDING_WITH_MODEL:
484
+ case TRAINING_STATUS_PENDING_WITHOUT_MODEL:
485
+ schedule_for_training = false;
486
+ break;
487
+ case TRAINING_STATUS_UNTRAINED:
488
+ schedule_for_training = true;
489
+ dim->ts = TRAINING_STATUS_PENDING_WITHOUT_MODEL;
490
+ break;
491
+ case TRAINING_STATUS_TRAINED:
492
+ if ((dim->last_training_time + (Cfg.train_every * dim->rd->update_every)) < curr_time) {
493
+ schedule_for_training = true;
494
+ dim->ts = TRAINING_STATUS_PENDING_WITH_MODEL;
495
+ }
496
+ break;
497
+ }
498
+
499
+ if (schedule_for_training) {
500
+ ml_host_t *host = (ml_host_t *) dim->rd->rrdset->rrdhost->ml_host;
501
+ ml_training_request_t req = {
502
+ string_dup(dim->rd->rrdset->id), string_dup(dim->rd->id),
503
+ curr_time, rrddim_first_entry_s(dim->rd), rrddim_last_entry_s(dim->rd),
504
+ };
505
+ ml_queue_push(host->training_queue, req);
506
+ }
507
+}
508
+
509
+static bool
510
+ml_dimension_predict(ml_dimension_t *dim, time_t curr_time, calculated_number_t value, bool exists)
511
+{
512
+ // Nothing to do if ML is disabled for this dimension
513
+ if (dim->mls != MACHINE_LEARNING_STATUS_ENABLED)
514
+ return false;
515
+
516
+ // Don't treat values that don't exist as anomalous
517
+ if (!exists) {
518
+ dim->cns.clear();
519
+ return false;
520
+ }
521
+
522
+ // Save the value and return if we don't have enough values for a sample
523
+ unsigned n = Cfg.diff_n + Cfg.smooth_n + Cfg.lag_n;
524
+ if (dim->cns.size() < n) {
525
+ dim->cns.push_back(value);
526
+ return false;
527
+ }
528
+
529
+ // Push the value and check if it's different from the last one
530
+ bool same_value = true;
531
+ std::rotate(std::begin(dim->cns), std::begin(dim->cns) + 1, std::end(dim->cns));
532
+ if (dim->cns[n - 1] != value)
533
+ same_value = false;
534
+ dim->cns[n - 1] = value;
535
+
536
+ // Create the sample
537
+ assert((n * (Cfg.lag_n + 1) <= 128) &&
538
+ "Static buffers too small to perform prediction. "
539
+ "This should not be possible with the default clamping of feature extraction options");
540
+ calculated_number_t src_cns[128];
541
+ calculated_number_t dst_cns[128];
542
+
543
+ memset(src_cns, 0, n * (Cfg.lag_n + 1) * sizeof(calculated_number_t));
544
+ memcpy(src_cns, dim->cns.data(), n * sizeof(calculated_number_t));
545
+ memcpy(dst_cns, dim->cns.data(), n * sizeof(calculated_number_t));
546
+
547
+ ml_features_t features = {
548
+ Cfg.diff_n, Cfg.smooth_n, Cfg.lag_n,
549
+ dst_cns, n, src_cns, n,
550
+ dim->feature
551
+ };
552
+ ml_features_preprocess(&features);
553
+
554
+ /*
555
+ * Lock to predict and possibly schedule the dimension for training
556
+ */
557
+ if (netdata_mutex_trylock(&dim->mutex) != 0)
558
+ return false;
559
+
560
+ // Mark the metric time as variable if we received different values
561
+ if (!same_value)
562
+ dim->mt = METRIC_TYPE_VARIABLE;
563
+
564
+ // Decide if the dimension needs to be scheduled for training
565
+ ml_dimension_schedule_for_training(dim, curr_time);
566
+
567
+ // Nothing to do if we don't have a model
568
+ switch (dim->ts) {
569
+ case TRAINING_STATUS_UNTRAINED:
570
+ case TRAINING_STATUS_PENDING_WITHOUT_MODEL: {
571
+ netdata_mutex_unlock(&dim->mutex);
572
+ return false;
573
+ }
574
+ default:
575
+ break;
576
+ }
577
+
578
+ /*
579
+ * Use the KMeans models to check if the value is anomalous
580
+ */
581
+
582
+ size_t sum = 0;
583
+ size_t models_consulted = 0;
584
+
585
+ for (const auto &km_ctx : dim->km_contexts) {
586
+ models_consulted++;
587
+
588
+ calculated_number_t anomaly_score = ml_kmeans_anomaly_score(&km_ctx, features.preprocessed_features[0]);
589
+ if (anomaly_score == std::numeric_limits<calculated_number_t>::quiet_NaN())
590
+ continue;
591
+
592
+ if (anomaly_score < (100 * Cfg.dimension_anomaly_score_threshold)) {
593
+ global_statistics_ml_models_consulted(models_consulted);
594
+ netdata_mutex_unlock(&dim->mutex);
595
+ return false;
596
+ }
597
+
598
+ sum += 1;
599
+ }
600
+
601
+ netdata_mutex_unlock(&dim->mutex);
602
+
603
+ global_statistics_ml_models_consulted(models_consulted);
604
+ return sum;
605
+}
606
+
607
+/*
608
+ * Chart
609
+*/
610
+
611
+static bool
612
+ml_chart_is_available_for_ml(ml_chart_t *chart)
613
+{
614
+ return rrdset_is_available_for_exporting_and_alarms(chart->rs);
615
+}
616
+
617
+void
618
+ml_chart_update_dimension(ml_chart_t *chart, ml_dimension_t *dim, bool is_anomalous)
619
+{
620
+ switch (dim->mls) {
621
+ case MACHINE_LEARNING_STATUS_DISABLED_DUE_TO_EXCLUDED_CHART:
622
+ chart->mls.num_machine_learning_status_disabled_sp++;
623
+ return;
624
+ case MACHINE_LEARNING_STATUS_ENABLED: {
625
+ chart->mls.num_machine_learning_status_enabled++;
626
+
627
+ switch (dim->mt) {
628
+ case METRIC_TYPE_CONSTANT:
629
+ chart->mls.num_metric_type_constant++;
630
+ chart->mls.num_training_status_trained++;
631
+ chart->mls.num_normal_dimensions++;
632
+ return;
633
+ case METRIC_TYPE_VARIABLE:
634
+ chart->mls.num_metric_type_variable++;
635
+ break;
636
+ }
637
+
638
+ switch (dim->ts) {
639
+ case TRAINING_STATUS_UNTRAINED:
640
+ chart->mls.num_training_status_untrained++;
641
+ return;
642
+ case TRAINING_STATUS_PENDING_WITHOUT_MODEL:
643
+ chart->mls.num_training_status_pending_without_model++;
644
+ return;
645
+ case TRAINING_STATUS_TRAINED:
646
+ chart->mls.num_training_status_trained++;
647
+
648
+ chart->mls.num_anomalous_dimensions += is_anomalous;
649
+ chart->mls.num_normal_dimensions += !is_anomalous;
650
+ return;
651
+ case TRAINING_STATUS_PENDING_WITH_MODEL:
652
+ chart->mls.num_training_status_pending_with_model++;
653
+
654
+ chart->mls.num_anomalous_dimensions += is_anomalous;
655
+ chart->mls.num_normal_dimensions += !is_anomalous;
656
+ return;
657
+ }
658
+
659
+ return;
660
+ }
661
+ }
662
+}
663
+
664
+/*
665
+ * Host detection & training functions
666
+*/
667
+
668
+#define WORKER_JOB_DETECTION_COLLECT_STATS 0
669
+#define WORKER_JOB_DETECTION_DIM_CHART 1
670
+#define WORKER_JOB_DETECTION_HOST_CHART 2
671
+#define WORKER_JOB_DETECTION_STATS 3
672
+
673
+static void
674
+ml_host_detect_once(ml_host_t *host)
675
+{
676
+ worker_is_busy(WORKER_JOB_DETECTION_COLLECT_STATS);
677
+
678
+ host->mls = {};
679
+ ml_machine_learning_stats_t mls_copy = {};
680
+ ml_training_stats_t ts_copy = {};
681
+
682
+ {
683
+ netdata_mutex_lock(&host->mutex);
684
+
685
+ /*
686
+ * prediction/detection stats
687
+ */
688
+ void *rsp = NULL;
689
+ rrdset_foreach_read(rsp, host->rh) {
690
+ RRDSET *rs = static_cast<RRDSET *>(rsp);
691
+
692
+ ml_chart_t *chart = (ml_chart_t *) rs->ml_chart;
693
+ if (!chart)
694
+ continue;
695
+
696
+ if (!ml_chart_is_available_for_ml(chart))
697
+ continue;
698
+
699
+ ml_machine_learning_stats_t chart_mls = chart->mls;
700
+
701
+ host->mls.num_machine_learning_status_enabled += chart_mls.num_machine_learning_status_enabled;
702
+ host->mls.num_machine_learning_status_disabled_sp += chart_mls.num_machine_learning_status_disabled_sp;
703
+
704
+ host->mls.num_metric_type_constant += chart_mls.num_metric_type_constant;
705
+ host->mls.num_metric_type_variable += chart_mls.num_metric_type_variable;
706
+
707
+ host->mls.num_training_status_untrained += chart_mls.num_training_status_untrained;
708
+ host->mls.num_training_status_pending_without_model += chart_mls.num_training_status_pending_without_model;
709
+ host->mls.num_training_status_trained += chart_mls.num_training_status_trained;
710
+ host->mls.num_training_status_pending_with_model += chart_mls.num_training_status_pending_with_model;
711
+
712
+ host->mls.num_anomalous_dimensions += chart_mls.num_anomalous_dimensions;
713
+ host->mls.num_normal_dimensions += chart_mls.num_normal_dimensions;
714
+ }
715
+ rrdset_foreach_done(rsp);
716
+
717
+ host->host_anomaly_rate = 0.0;
718
+ size_t NumActiveDimensions = host->mls.num_anomalous_dimensions + host->mls.num_normal_dimensions;
719
+ if (NumActiveDimensions)
720
+ host->host_anomaly_rate = static_cast<double>(host->mls.num_anomalous_dimensions) / NumActiveDimensions;
721
+
722
+ mls_copy = host->mls;
723
+
724
+ /*
725
+ * training stats
726
+ */
727
+ ts_copy = host->ts;
728
+
729
+ host->ts.queue_size = 0;
730
+ host->ts.num_popped_items = 0;
731
+
732
+ host->ts.allotted_ut = 0;
733
+ host->ts.consumed_ut = 0;
734
+ host->ts.remaining_ut = 0;
735
+
736
+ host->ts.training_result_ok = 0;
737
+ host->ts.training_result_invalid_query_time_range = 0;
738
+ host->ts.training_result_not_enough_collected_values = 0;
739
+ host->ts.training_result_null_acquired_dimension = 0;
740
+ host->ts.training_result_chart_under_replication = 0;
741
+
742
+ netdata_mutex_unlock(&host->mutex);
743
+ }
744
+
745
+ // Calc the avg values
746
+ if (ts_copy.num_popped_items) {
747
+ ts_copy.queue_size /= ts_copy.num_popped_items;
748
+ ts_copy.allotted_ut /= ts_copy.num_popped_items;
749
+ ts_copy.consumed_ut /= ts_copy.num_popped_items;
750
+ ts_copy.remaining_ut /= ts_copy.num_popped_items;
751
+
752
+ ts_copy.training_result_ok /= ts_copy.num_popped_items;
753
+ ts_copy.training_result_invalid_query_time_range /= ts_copy.num_popped_items;
754
+ ts_copy.training_result_not_enough_collected_values /= ts_copy.num_popped_items;
755
+ ts_copy.training_result_null_acquired_dimension /= ts_copy.num_popped_items;
756
+ ts_copy.training_result_chart_under_replication /= ts_copy.num_popped_items;
757
+ } else {
758
+ ts_copy.queue_size = 0;
759
+ ts_copy.allotted_ut = 0;
760
+ ts_copy.consumed_ut = 0;
761
+ ts_copy.remaining_ut = 0;
762
+ }
763
+
764
+ worker_is_busy(WORKER_JOB_DETECTION_DIM_CHART);
765
+ ml_update_dimensions_chart(host, mls_copy);
766
+
767
+ worker_is_busy(WORKER_JOB_DETECTION_HOST_CHART);
768
+ ml_update_host_and_detection_rate_charts(host, host->host_anomaly_rate * 10000.0);
769
+
770
+ worker_is_busy(WORKER_JOB_DETECTION_STATS);
771
+ ml_update_training_statistics_chart(host, ts_copy);
772
+}
773
+
774
+typedef struct {
775
+ RRDDIM_ACQUIRED *acq_rd;
776
+ ml_dimension_t *dim;
777
+} ml_acquired_dimension_t;
778
+
779
+static ml_acquired_dimension_t
780
+ml_acquired_dimension_get(RRDHOST *rh, STRING *chart_id, STRING *dimension_id)
781
+{
782
+ RRDDIM_ACQUIRED *acq_rd = NULL;
783
+ ml_dimension_t *dim = NULL;
784
+
785
+ RRDSET *rs = rrdset_find(rh, string2str(chart_id));
786
+ if (rs) {
787
+ acq_rd = rrddim_find_and_acquire(rs, string2str(dimension_id));
788
+ if (acq_rd) {
789
+ RRDDIM *rd = rrddim_acquired_to_rrddim(acq_rd);
790
+ if (rd)
791
+ dim = (ml_dimension_t *) rd->ml_dimension;
792
+ }
793
+ }
794
+
795
+ ml_acquired_dimension_t acq_dim = {
796
+ acq_rd, dim
797
+ };
798
+
799
+ return acq_dim;
800
+}
801
+
802
+static void
803
+ml_acquired_dimension_release(ml_acquired_dimension_t acq_dim)
804
+{
805
+ if (!acq_dim.acq_rd)
806
+ return;
807
+
808
+ rrddim_acquired_release(acq_dim.acq_rd);
809
+}
810
+
811
+static enum ml_training_result
812
+ml_acquired_dimension_train(ml_acquired_dimension_t acq_dim, const ml_training_request_t &TR)
813
+{
814
+ if (!acq_dim.dim)
815
+ return TRAINING_RESULT_NULL_ACQUIRED_DIMENSION;
816
+
817
+ return ml_dimension_train_model(acq_dim.dim, TR);
818
+}
819
+
820
+#define WORKER_JOB_TRAINING_FIND 0
821
+#define WORKER_JOB_TRAINING_TRAIN 1
822
+#define WORKER_JOB_TRAINING_STATS 2
823
+
824
+static void
825
+ml_host_train(ml_host_t *host)
826
+{
827
+ worker_register("MLTRAIN");
828
+ worker_register_job_name(WORKER_JOB_TRAINING_FIND, "find");
829
+ worker_register_job_name(WORKER_JOB_TRAINING_TRAIN, "train");
830
+ worker_register_job_name(WORKER_JOB_TRAINING_STATS, "stats");
831
+
832
+ service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, (force_quit_t ) ml_host_cancel_training_thread, host->rh, true);
833
+
834
+ while (service_running(SERVICE_ML_TRAINING)) {
835
+ ml_training_request_t training_req = ml_queue_pop(host->training_queue);
836
+ size_t queue_size = ml_queue_size(host->training_queue) + 1;
837
+
838
+ if (host->threads_cancelled) {
839
+ info("Stopping training thread for host %s because it was cancelled", rrdhost_hostname(host->rh));
840
+ break;
841
+ }
842
+
843
+ usec_t allotted_ut = (Cfg.train_every * host->rh->rrd_update_every * USEC_PER_SEC) / queue_size;
844
+ if (allotted_ut > USEC_PER_SEC)
845
+ allotted_ut = USEC_PER_SEC;
846
+
847
+ usec_t start_ut = now_monotonic_usec();
848
+ enum ml_training_result training_res;
849
+ {
850
+ worker_is_busy(WORKER_JOB_TRAINING_FIND);
851
+ ml_acquired_dimension_t acq_dim = ml_acquired_dimension_get(host->rh, training_req.chart_id, training_req.dimension_id);
852
+
853
+ worker_is_busy(WORKER_JOB_TRAINING_TRAIN);
854
+ training_res = ml_acquired_dimension_train(acq_dim, training_req);
855
+
856
+ string_freez(training_req.chart_id);
857
+ string_freez(training_req.dimension_id);
858
+
859
+ ml_acquired_dimension_release(acq_dim);
860
+ }
861
+ usec_t consumed_ut = now_monotonic_usec() - start_ut;
862
+
863
+ worker_is_busy(WORKER_JOB_TRAINING_STATS);
864
+
865
+ usec_t remaining_ut = 0;
866
+ if (consumed_ut < allotted_ut)
867
+ remaining_ut = allotted_ut - consumed_ut;
868
+
869
+ {
870
+ netdata_mutex_lock(&host->mutex);
871
+
872
+ host->ts.queue_size += queue_size;
873
+ host->ts.num_popped_items += 1;
874
+
875
+ host->ts.allotted_ut += allotted_ut;
876
+ host->ts.consumed_ut += consumed_ut;
877
+ host->ts.remaining_ut += remaining_ut;
878
+
879
+ switch (training_res) {
880
+ case TRAINING_RESULT_OK:
881
+ host->ts.training_result_ok += 1;
882
+ break;
883
+ case TRAINING_RESULT_INVALID_QUERY_TIME_RANGE:
884
+ host->ts.training_result_invalid_query_time_range += 1;
885
+ break;
886
+ case TRAINING_RESULT_NOT_ENOUGH_COLLECTED_VALUES:
887
+ host->ts.training_result_not_enough_collected_values += 1;
888
+ break;
889
+ case TRAINING_RESULT_NULL_ACQUIRED_DIMENSION:
890
+ host->ts.training_result_null_acquired_dimension += 1;
891
+ break;
892
+ case TRAINING_RESULT_CHART_UNDER_REPLICATION:
893
+ host->ts.training_result_chart_under_replication += 1;
894
+ break;
895
+ }
896
+
897
+ netdata_mutex_unlock(&host->mutex);
898
+ }
899
+
900
+ worker_is_idle();
901
+ std::this_thread::sleep_for(std::chrono::microseconds{remaining_ut});
902
+ worker_is_busy(0);
903
+ }
904
+}
905
+
906
+static void *
907
+train_main(void *arg)
908
+{
909
+ size_t max_elements_needed_for_training = Cfg.max_train_samples * (Cfg.lag_n + 1);
910
+ tls_data.training_cns = new calculated_number_t[max_elements_needed_for_training]();
911
+ tls_data.scratch_training_cns = new calculated_number_t[max_elements_needed_for_training]();
912
+
913
+ ml_host_t *host = (ml_host_t *) arg;
914
+ ml_host_train(host);
915
+ return NULL;
916
+}
917
+
918
+static void *
919
+ml_detect_main(void *arg)
920
+{
921
+ UNUSED(arg);
922
+
923
+ worker_register("MLDETECT");
924
+ worker_register_job_name(WORKER_JOB_DETECTION_COLLECT_STATS, "collect stats");
925
+ worker_register_job_name(WORKER_JOB_DETECTION_DIM_CHART, "dim chart");
926
+ worker_register_job_name(WORKER_JOB_DETECTION_HOST_CHART, "host chart");
927
+ worker_register_job_name(WORKER_JOB_DETECTION_STATS, "training stats");
928
+
929
+ service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, NULL, NULL, true);
930
+
931
+ heartbeat_t hb;
932
+ heartbeat_init(&hb);
933
+
934
+ while (service_running((SERVICE_TYPE)(SERVICE_ML_PREDICTION | SERVICE_COLLECTORS))) {
935
+ worker_is_idle();
936
+ heartbeat_next(&hb, USEC_PER_SEC);
937
+
938
+ void *rhp;
939
+ dfe_start_reentrant(rrdhost_root_index, rhp) {
940
+ RRDHOST *rh = (RRDHOST *) rhp;
941
+
942
+ if (!rh->ml_host)
943
+ continue;
944
+
945
+ ml_host_detect_once((ml_host_t *) rh->ml_host);
946
+ }
947
+ dfe_done(rhp);
948
+ }
949
+
950
+ return NULL;
951
+}
952
+
953
+/*
954
+ * Public API
955
+*/
956
+
957
+bool ml_capable()
958
+{
959
return true;
960
}
961
11
-bool ml_enabled(RRDHOST *rh) {
962
+bool ml_enabled(RRDHOST *rh)
963
+{
964
+ if (!rh)
965
+ return false;
966
+
967
if (!Cfg.enable_anomaly_detection)
968
return false;
969
@@ -18,15 +973,15 @@ bool ml_enabled(RRDHOST *rh) {
973
return true;
974
}
975
21
-/*
22
- * Assumptions:
23
- * 1) hosts outlive their sets, and sets outlive their dimensions,
24
- * 2) dimensions always have a set that has a host.
25
- */
976
+bool ml_streaming_enabled()
977
+{
978
+ return Cfg.stream_anomaly_detection_charts;
979
+}
980
27
-void ml_init(void) {
981
+void ml_init()
982
+{
983
// Read config values
29
- nml_config_load(&Cfg);
984
+ ml_config_load(&Cfg);
985
986
if (!Cfg.enable_anomaly_detection)
987
return;
@@ -45,149 +1000,270 @@ void ml_init(void) {
1000
char tag[NETDATA_THREAD_TAG_MAX + 1];
1001
1002
snprintfz(tag, NETDATA_THREAD_TAG_MAX, "%s", "PREDICT");
48
- netdata_thread_create(&Cfg.detection_thread, tag, NETDATA_THREAD_OPTION_JOINABLE, nml_detect_main, NULL);
1003
+ netdata_thread_create(&Cfg.detection_thread, tag, NETDATA_THREAD_OPTION_JOINABLE, ml_detect_main, NULL);
1004
}
1005
51
-void ml_host_new(RRDHOST *rh) {
1006
+void ml_host_new(RRDHOST *rh)
1007
+{
1008
if (!ml_enabled(rh))
1009
return;
1010
55
- nml_host_t *host = nml_host_new(rh);
56
- rh->ml_host = reinterpret_cast<ml_host_t *>(host);
1011
+ ml_host_t *host = new ml_host_t();
1012
+
1013
+ host->rh = rh;
1014
+ host->mls = ml_machine_learning_stats_t();
1015
+ host->ts = ml_training_stats_t();
1016
+
1017
+ host->host_anomaly_rate = 0.0;
1018
+ host->threads_running = false;
1019
+ host->threads_cancelled = false;
1020
+ host->threads_joined = false;
1021
+
1022
+ host->training_queue = ml_queue_init();
1023
+
1024
+ netdata_mutex_init(&host->mutex);
1025
+
1026
+ rh->ml_host = (rrd_ml_host_t *) host;
1027
}
1028
59
-void ml_host_delete(RRDHOST *rh) {
60
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rh->ml_host);
1029
+void ml_host_delete(RRDHOST *rh)
1030
+{
1031
+ ml_host_t *host = (ml_host_t *) rh->ml_host;
1032
if (!host)
1033
return;
1034
64
- nml_host_delete(host);
1035
+ netdata_mutex_destroy(&host->mutex);
1036
+ ml_queue_destroy(host->training_queue);
1037
+
1038
+ delete host;
1039
rh->ml_host = NULL;
1040
}
1041
68
-void ml_chart_new(RRDSET *rs) {
69
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rs->rrdhost->ml_host);
70
- if (!host)
1042
+void ml_host_get_info(RRDHOST *rh, BUFFER *wb)
1043
+{
1044
+ ml_host_t *host = (ml_host_t *) rh->ml_host;
1045
+ if (!host) {
1046
+ buffer_json_member_add_boolean(wb, "enabled", false);
1047
return;
1048
+ }
1049
+
1050
+ buffer_json_member_add_uint64(wb, "version", 1);
1051
+
1052
+ buffer_json_member_add_boolean(wb, "enabled", Cfg.enable_anomaly_detection);
1053
+
1054
+ buffer_json_member_add_uint64(wb, "min-train-samples", Cfg.min_train_samples);
1055
+ buffer_json_member_add_uint64(wb, "max-train-samples", Cfg.max_train_samples);
1056
+ buffer_json_member_add_uint64(wb, "train-every", Cfg.train_every);
1057
+
1058
+ buffer_json_member_add_uint64(wb, "diff-n", Cfg.diff_n);
1059
+ buffer_json_member_add_uint64(wb, "smooth-n", Cfg.smooth_n);
1060
+ buffer_json_member_add_uint64(wb, "lag-n", Cfg.lag_n);
1061
+
1062
+ buffer_json_member_add_double(wb, "random-sampling-ratio", Cfg.random_sampling_ratio);
1063
+ buffer_json_member_add_uint64(wb, "max-kmeans-iters", Cfg.random_sampling_ratio);
1064
+
1065
+ buffer_json_member_add_double(wb, "dimension-anomaly-score-threshold", Cfg.dimension_anomaly_score_threshold);
1066
+
1067
+ buffer_json_member_add_string(wb, "anomaly-detection-grouping-method",
1068
+ time_grouping_method2string(Cfg.anomaly_detection_grouping_method));
1069
+
1070
+ buffer_json_member_add_int64(wb, "anomaly-detection-query-duration", Cfg.anomaly_detection_query_duration);
1071
73
- nml_chart_t *chart = nml_chart_new(rs);
74
- rs->ml_chart = reinterpret_cast<ml_chart_t *>(chart);
1072
+ buffer_json_member_add_string(wb, "hosts-to-skip", Cfg.hosts_to_skip.c_str());
1073
+ buffer_json_member_add_string(wb, "charts-to-skip", Cfg.charts_to_skip.c_str());
1074
}
1075
77
-void ml_chart_delete(RRDSET *rs) {
78
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rs->rrdhost->ml_host);
1076
+void ml_host_get_detection_info(RRDHOST *rh, BUFFER *wb)
1077
+{
1078
+ ml_host_t *host = (ml_host_t *) rh->ml_host;
1079
if (!host)
1080
return;
1081
82
- nml_chart_t *chart = reinterpret_cast<nml_chart_t *>(rs->ml_chart);
1082
+ netdata_mutex_lock(&host->mutex);
1083
84
- nml_chart_delete(chart);
85
- rs->ml_chart = NULL;
1084
+ buffer_json_member_add_uint64(wb, "version", 1);
1085
+ buffer_json_member_add_uint64(wb, "anomalous-dimensions", host->mls.num_anomalous_dimensions);
1086
+ buffer_json_member_add_uint64(wb, "normal-dimensions", host->mls.num_normal_dimensions);
1087
+ buffer_json_member_add_uint64(wb, "total-dimensions", host->mls.num_anomalous_dimensions +
1088
+ host->mls.num_normal_dimensions);
1089
+ buffer_json_member_add_uint64(wb, "trained-dimensions", host->mls.num_training_status_trained +
1090
+ host->mls.num_training_status_pending_with_model);
1091
+ netdata_mutex_unlock(&host->mutex);
1092
}
1093
88
-void ml_dimension_new(RRDDIM *rd) {
89
- nml_chart_t *chart = reinterpret_cast<nml_chart_t *>(rd->rrdset->ml_chart);
90
- if (!chart)
91
- return;
1094
+void ml_host_get_models(RRDHOST *rh, BUFFER *wb)
1095
+{
1096
+ UNUSED(rh);
1097
+ UNUSED(wb);
1098
93
- nml_dimension_t *dim = nml_dimension_new(rd);
94
- rd->ml_dimension = reinterpret_cast<ml_dimension_t *>(dim);
1099
+ // TODO: To be implemented
1100
+ error("Fetching KMeans models is not supported yet");
1101
}
1102
97
-void ml_dimension_delete(RRDDIM *rd) {
98
- nml_dimension_t *dim = reinterpret_cast<nml_dimension_t *>(rd->ml_dimension);
99
- if (!dim)
1103
+void ml_host_start_training_thread(RRDHOST *rh)
1104
+{
1105
+ if (!rh || !rh->ml_host)
1106
return;
1107
102
- nml_dimension_delete(dim);
103
- rd->ml_dimension = NULL;
104
-}
1108
+ ml_host_t *host = (ml_host_t *) rh->ml_host;
1109
106
-void ml_get_host_info(RRDHOST *rh, BUFFER *wb) {
107
- if (rh && rh->ml_host) {
108
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rh->ml_host);
109
- nml_host_get_config_as_json(host, wb);
110
- } else {
111
- buffer_json_member_add_boolean(wb, "enabled", false);
1110
+ if (host->threads_running) {
1111
+ error("Anomaly detections threads for host %s are already-up and running.", rrdhost_hostname(host->rh));
1112
+ return;
1113
}
1114
+
1115
+ host->threads_running = true;
1116
+ host->threads_cancelled = false;
1117
+ host->threads_joined = false;
1118
+
1119
+ char tag[NETDATA_THREAD_TAG_MAX + 1];
1120
+
1121
+ snprintfz(tag, NETDATA_THREAD_TAG_MAX, "MLTR[%s]", rrdhost_hostname(host->rh));
1122
+ netdata_thread_create(&host->training_thread, tag, NETDATA_THREAD_OPTION_JOINABLE, train_main, static_cast<void *>(host));
1123
}
1124
115
-char *ml_get_host_runtime_info(RRDHOST *rh) {
116
- nlohmann::json config_json;
1125
+void ml_host_cancel_training_thread(RRDHOST *rh)
1126
+{
1127
+ if (!rh || !rh->ml_host)
1128
+ return;
1129
+
1130
+ ml_host_t *host = (ml_host_t *) rh->ml_host;
1131
118
- if (rh && rh->ml_host) {
119
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rh->ml_host);
120
- nml_host_get_detection_info_as_json(host, config_json);
121
- } else {
122
- return NULL;
1132
+ if (!host->threads_running) {
1133
+ error("Anomaly detections threads for host %s have already been stopped.", rrdhost_hostname(host->rh));
1134
+ return;
1135
}
1136
125
- return strdup(config_json.dump(1, '\t').c_str());
1137
+ if (!host->threads_cancelled) {
1138
+ host->threads_cancelled = true;
1139
+
1140
+ // Signal the training queue to stop popping-items
1141
+ ml_queue_signal(host->training_queue);
1142
+ netdata_thread_cancel(host->training_thread);
1143
+ }
1144
}
1145
128
-char *ml_get_host_models(RRDHOST *rh) {
129
- nlohmann::json j;
1146
+void ml_host_stop_training_thread(RRDHOST *rh)
1147
+{
1148
+ if (!rh || !rh->ml_host)
1149
+ return;
1150
+
1151
+ ml_host_cancel_training_thread(rh);
1152
+
1153
+ ml_host_t *host = (ml_host_t *) rh->ml_host;
1154
131
- if (rh && rh->ml_host) {
132
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rh->ml_host);
133
- nml_host_get_models_as_json(host, j);
134
- return strdup(j.dump(2, '\t').c_str());
1155
+ if (!host->threads_joined) {
1156
+ host->threads_joined = true;
1157
+ host->threads_running = false;
1158
+
1159
+ delete[] tls_data.training_cns;
1160
+ delete[] tls_data.scratch_training_cns;
1161
+
1162
+ netdata_thread_join(host->training_thread, NULL);
1163
}
1164
+}
1165
137
- return NULL;
1166
+void ml_chart_new(RRDSET *rs)
1167
+{
1168
+ ml_host_t *host = (ml_host_t *) rs->rrdhost->ml_host;
1169
+ if (!host)
1170
+ return;
1171
+
1172
+ ml_chart_t *chart = new ml_chart_t();
1173
+
1174
+ chart->rs = rs;
1175
+ chart->mls = ml_machine_learning_stats_t();
1176
+
1177
+ netdata_mutex_init(&chart->mutex);
1178
+
1179
+ rs->ml_chart = (rrd_ml_chart_t *) chart;
1180
+}
1181
+
1182
+void ml_chart_delete(RRDSET *rs)
1183
+{
1184
+ ml_host_t *host = (ml_host_t *) rs->rrdhost->ml_host;
1185
+ if (!host)
1186
+ return;
1187
+
1188
+ ml_chart_t *chart = (ml_chart_t *) rs->ml_chart;
1189
+
1190
+ netdata_mutex_destroy(&chart->mutex);
1191
+
1192
+ delete chart;
1193
+ rs->ml_chart = NULL;
1194
}
1195
140
-bool ml_chart_update_begin(RRDSET *rs) {
141
- nml_chart_t *chart = reinterpret_cast<nml_chart_t *>(rs->ml_chart);
1196
+bool ml_chart_update_begin(RRDSET *rs)
1197
+{
1198
+ ml_chart_t *chart = (ml_chart_t *) rs->ml_chart;
1199
if (!chart)
1200
return false;
1201
145
- nml_chart_update_begin(chart);
1202
+ netdata_mutex_lock(&chart->mutex);
1203
+ chart->mls = {};
1204
return true;
1205
}
1206
149
-void ml_chart_update_end(RRDSET *rs) {
150
- nml_chart_t *chart = reinterpret_cast<nml_chart_t *>(rs->ml_chart);
1207
+void ml_chart_update_end(RRDSET *rs)
1208
+{
1209
+ ml_chart_t *chart = (ml_chart_t *) rs->ml_chart;
1210
if (!chart)
1211
return;
1212
154
- nml_chart_update_end(chart);
1213
+ netdata_mutex_unlock(&chart->mutex);
1214
}
1215
157
-bool ml_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool exists) {
158
- nml_dimension_t *dim = reinterpret_cast<nml_dimension_t *>(rd->ml_dimension);
159
- if (!dim)
160
- return false;
1216
+void ml_dimension_new(RRDDIM *rd)
1217
+{
1218
+ ml_chart_t *chart = (ml_chart_t *) rd->rrdset->ml_chart;
1219
+ if (!chart)
1220
+ return;
1221
162
- nml_chart_t *chart = reinterpret_cast<nml_chart_t *>(rd->rrdset->ml_chart);
1222
+ ml_dimension_t *dim = new ml_dimension_t();
1223
164
- bool is_anomalous = nml_dimension_predict(dim, curr_time, value, exists);
165
- nml_chart_update_dimension(chart, dim, is_anomalous);
1224
+ dim->rd = rd;
1225
167
- return is_anomalous;
168
-}
1226
+ dim->mt = METRIC_TYPE_CONSTANT;
1227
+ dim->ts = TRAINING_STATUS_UNTRAINED;
1228
170
-bool ml_streaming_enabled() {
171
- return Cfg.stream_anomaly_detection_charts;
172
-}
1229
+ dim->last_training_time = 0;
1230
174
-void ml_start_training_thread(RRDHOST *rh) {
175
- if (rh && rh->ml_host) {
176
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rh->ml_host);
177
- nml_host_start_training_thread(host);
178
- }
1231
+ ml_kmeans_init(&dim->kmeans, 2, 1000);
1232
+
1233
+ if (simple_pattern_matches(Cfg.sp_charts_to_skip, rrdset_name(rd->rrdset)))
1234
+ dim->mls = MACHINE_LEARNING_STATUS_DISABLED_DUE_TO_EXCLUDED_CHART;
1235
+ else
1236
+ dim->mls = MACHINE_LEARNING_STATUS_ENABLED;
1237
+
1238
+ netdata_mutex_init(&dim->mutex);
1239
+
1240
+ dim->km_contexts.reserve(Cfg.num_models_to_use);
1241
+
1242
+ rd->ml_dimension = (rrd_ml_dimension_t *) dim;
1243
}
1244
181
-void ml_stop_training_thread(RRDHOST *rh) {
182
- if (rh && rh->ml_host) {
183
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rh->ml_host);
184
- nml_host_stop_training_thread(host, /* join */ true);
185
- }
1245
+void ml_dimension_delete(RRDDIM *rd)
1246
+{
1247
+ ml_dimension_t *dim = (ml_dimension_t *) rd->ml_dimension;
1248
+ if (!dim)
1249
+ return;
1250
+
1251
+ netdata_mutex_destroy(&dim->mutex);
1252
+
1253
+ delete dim;
1254
+ rd->ml_dimension = NULL;
1255
}
1256
188
-void ml_cancel_training_thread(RRDHOST *rh) {
189
- if (rh && rh->ml_host) {
190
- nml_host_t *host = reinterpret_cast<nml_host_t *>(rh->ml_host);
191
- nml_host_stop_training_thread(host, /* join */ false);
192
- }
1257
+bool ml_dimension_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool exists)
1258
+{
1259
+ ml_dimension_t *dim = (ml_dimension_t *) rd->ml_dimension;
1260
+ if (!dim)
1261
+ return false;
1262
+
1263
+ ml_chart_t *chart = (ml_chart_t *) rd->rrdset->ml_chart;
1264
+
1265
+ bool is_anomalous = ml_dimension_predict(dim, curr_time, value, exists);
1266
+ ml_chart_update_dimension(chart, dim, is_anomalous);
1267
+
1268
+ return is_anomalous;
1269
}
ml/ml.h
+12
-21
@@ -10,39 +10,30 @@ extern "C" {
10
#include "daemon/common.h"
11
#include "web/api/queries/rrdr.h"
12
13
-// This is a DBEngine function redeclared here so that we can free
14
-// the anomaly rate dimension, whenever its backing dimension is freed.
15
-void rrddim_free(RRDSET *rs, RRDDIM *rd);
16
-
13
bool ml_capable();
18
-
14
bool ml_enabled(RRDHOST *rh);
20
-
15
+bool ml_streaming_enabled();
16
void ml_init(void);
17
18
void ml_host_new(RRDHOST *rh);
19
void ml_host_delete(RRDHOST *rh);
20
26
-void ml_chart_new(RRDSET *rs);
27
-void ml_chart_delete(RRDSET *rs);
28
-
29
-void ml_dimension_new(RRDDIM *rd);
30
-void ml_dimension_delete(RRDDIM *rd);
21
+void ml_host_get_info(RRDHOST *RH, BUFFER *wb);
22
+void ml_host_get_detection_info(RRDHOST *RH, BUFFER *wb);
23
+void ml_host_get_models(RRDHOST *RH, BUFFER *wb);
24
32
-void ml_get_host_info(RRDHOST *RH, BUFFER *wb);
33
-char *ml_get_host_runtime_info(RRDHOST *RH);
34
-char *ml_get_host_models(RRDHOST *RH);
25
+void ml_host_start_training_thread(RRDHOST *rh);
26
+void ml_host_cancel_training_thread(RRDHOST *rh);
27
+void ml_host_stop_training_thread(RRDHOST *rh);
28
29
+void ml_chart_new(RRDSET *rs);
30
+void ml_chart_delete(RRDSET *rs);
31
bool ml_chart_update_begin(RRDSET *rs);
32
void ml_chart_update_end(RRDSET *rs);
33
39
-bool ml_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool exists);
40
-
41
-bool ml_streaming_enabled();
42
-
43
-void ml_start_training_thread(RRDHOST *rh);
44
-void ml_cancel_training_thread(RRDHOST *rh);
45
-void ml_stop_training_thread(RRDHOST *rh);
34
+void ml_dimension_new(RRDDIM *rd);
35
+void ml_dimension_delete(RRDDIM *rd);
36
+bool ml_dimension_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool exists);
37
38
#ifdef __cplusplus
39
};
ml/nml.cc
deleted
-1134
@@ -1,1134 +0,0 @@
1
-// SPDX-License-Identifier: GPL-3.0-or-later
2
-
3
-#include <dlib/clustering.h>
4
-
5
-#include "nml.h"
6
-
7
-#include <random>
8
-
9
-#include "ad_charts.h"
10
-
11
-typedef struct {
12
- calculated_number_t *training_cns;
13
- calculated_number_t *scratch_training_cns;
14
-
15
- std::vector<DSample> training_samples;
16
-} nml_tls_data_t;
17
-
18
-static thread_local nml_tls_data_t tls_data;
19
-
20
-/*
21
- * Functions to convert enums to strings
22
-*/
23
-
24
-static const char *nml_machine_learning_status_to_string(enum nml_machine_learning_status mls) {
25
- switch (mls) {
26
- case MACHINE_LEARNING_STATUS_ENABLED:
27
- return "enabled";
28
- case MACHINE_LEARNING_STATUS_DISABLED_DUE_TO_EXCLUDED_CHART:
29
- return "disabled-sp";
30
- default:
31
- return "unknown";
32
- }
33
-}
34
-
35
-static const char *nml_metric_type_to_string(enum nml_metric_type mt) {
36
- switch (mt) {
37
- case METRIC_TYPE_CONSTANT:
38
- return "constant";
39
- case METRIC_TYPE_VARIABLE:
40
- return "variable";
41
- default:
42
- return "unknown";
43
- }
44
-}
45
-
46
-static const char *nml_training_status_to_string(enum nml_training_status ts) {
47
- switch (ts) {
48
- case TRAINING_STATUS_PENDING_WITH_MODEL:
49
- return "pending-with-model";
50
- case TRAINING_STATUS_PENDING_WITHOUT_MODEL:
51
- return "pending-without-model";
52
- case TRAINING_STATUS_TRAINED:
53
- return "trained";
54
- case TRAINING_STATUS_UNTRAINED:
55
- return "untrained";
56
- default:
57
- return "unknown";
58
- }
59
-}
60
-
61
-static const char *nml_training_result_to_string(enum nml_training_result tr) {
62
- switch (tr) {
63
- case TRAINING_RESULT_OK:
64
- return "ok";
65
- case TRAINING_RESULT_INVALID_QUERY_TIME_RANGE:
66
- return "invalid-query";
67
- case TRAINING_RESULT_NOT_ENOUGH_COLLECTED_VALUES:
68
- return "missing-values";
69
- case TRAINING_RESULT_NULL_ACQUIRED_DIMENSION:
70
- return "null-acquired-dim";
71
- case TRAINING_RESULT_CHART_UNDER_REPLICATION:
72
- return "chart-under-replication";
73
- default:
74
- return "unknown";
75
- }
76
-}
77
-
78
-/*
79
- * Features
80
-*/
81
-
82
-// subtract elements that are `diff_n` positions apart
83
-static void nml_features_diff(nml_features_t *features) {
84
- if (features->diff_n == 0)
85
- return;
86
-
87
- for (size_t idx = 0; idx != (features->src_n - features->diff_n); idx++) {
88
- size_t high = (features->src_n - 1) - idx;
89
- size_t low = high - features->diff_n;
90
-
91
- features->dst[low] = features->src[high] - features->src[low];
92
- }
93
-
94
- size_t n = features->src_n - features->diff_n;
95
- memcpy(features->src, features->dst, n * sizeof(calculated_number_t));
96
-
97
- for (size_t idx = features->src_n - features->diff_n; idx != features->src_n; idx++)
98
- features->src[idx] = 0.0;
99
-}
100
-
101
-// a function that computes the window average of an array inplace
102
-static void nml_features_smooth(nml_features_t *features) {
103
- calculated_number_t sum = 0.0;
104
-
105
- size_t idx = 0;
106
- for (; idx != features->smooth_n - 1; idx++)
107
- sum += features->src[idx];
108
-
109
- for (; idx != (features->src_n - features->diff_n); idx++) {
110
- sum += features->src[idx];
111
- calculated_number_t prev_cn = features->src[idx - (features->smooth_n - 1)];
112
- features->src[idx - (features->smooth_n - 1)] = sum / features->smooth_n;
113
- sum -= prev_cn;
114
- }
115
-
116
- for (idx = 0; idx != features->smooth_n; idx++)
117
- features->src[(features->src_n - 1) - idx] = 0.0;
118
-}
119
-
120
-// create lag'd vectors out of the preprocessed buffer
121
-static void nml_features_lag(nml_features_t *features) {
122
- size_t n = features->src_n - features->diff_n - features->smooth_n + 1 - features->lag_n;
123
- features->preprocessed_features.resize(n);
124
-
125
- unsigned target_num_samples = Cfg.max_train_samples * Cfg.random_sampling_ratio;
126
- double sampling_ratio = std::min(static_cast<double>(target_num_samples) / n, 1.0);
127
-
128
- uint32_t max_mt = std::numeric_limits<uint32_t>::max();
129
- uint32_t cutoff = static_cast<double>(max_mt) * sampling_ratio;
130
-
131
- size_t sample_idx = 0;
132
-
133
- for (size_t idx = 0; idx != n; idx++) {
134
- DSample &DS = features->preprocessed_features[sample_idx++];
135
- DS.set_size(features->lag_n);
136
-
137
- if (Cfg.random_nums[idx] > cutoff) {
138
- sample_idx--;
139
- continue;
140
- }
141
-
142
- for (size_t feature_idx = 0; feature_idx != features->lag_n + 1; feature_idx++)
143
- DS(feature_idx) = features->src[idx + feature_idx];
144
- }
145
-
146
- features->preprocessed_features.resize(sample_idx);
147
-}
148
-
149
-static void nml_features_preprocess(nml_features_t *features) {
150
- nml_features_diff(features);
151
- nml_features_smooth(features);
152
- nml_features_lag(features);
153
-}
154
-
155
-/*
156
- * KMeans
157
-*/
158
-
159
-static void nml_kmeans_init(nml_kmeans_t *kmeans, size_t num_clusters, size_t max_iterations) {
160
- kmeans->num_clusters = num_clusters;
161
- kmeans->max_iterations = max_iterations;
162
-
163
- kmeans->cluster_centers.reserve(kmeans->num_clusters);
164
- kmeans->min_dist = std::numeric_limits<calculated_number_t>::max();
165
- kmeans->max_dist = std::numeric_limits<calculated_number_t>::min();
166
-}
167
-
168
-static void nml_kmeans_train(nml_kmeans_t *kmeans, const nml_features_t *features) {
169
- kmeans->min_dist = std::numeric_limits<calculated_number_t>::max();
170
- kmeans->max_dist = std::numeric_limits<calculated_number_t>::min();
171
-
172
- kmeans->cluster_centers.clear();
173
-
174
- dlib::pick_initial_centers(kmeans->num_clusters, kmeans->cluster_centers, features->preprocessed_features);
175
- dlib::find_clusters_using_kmeans(features->preprocessed_features, kmeans->cluster_centers, kmeans->max_iterations);
176
-
177
- for (const auto &preprocessed_feature : features->preprocessed_features) {
178
- calculated_number_t mean_dist = 0.0;
179
-
180
- for (const auto &cluster_center : kmeans->cluster_centers) {
181
- mean_dist += dlib::length(cluster_center - preprocessed_feature);
182
- }
183
-
184
- mean_dist /= kmeans->num_clusters;
185
-
186
- if (mean_dist < kmeans->min_dist)
187
- kmeans->min_dist = mean_dist;
188
-
189
- if (mean_dist > kmeans->max_dist)
190
- kmeans->max_dist = mean_dist;
191
- }
192
-}
193
-
194
-static calculated_number_t nml_kmeans_anomaly_score(const nml_kmeans_t *kmeans, const DSample &DS) {
195
- calculated_number_t mean_dist = 0.0;
196
- for (const auto &CC: kmeans->cluster_centers)
197
- mean_dist += dlib::length(CC - DS);
198
-
199
- mean_dist /= kmeans->num_clusters;
200
-
201
- if (kmeans->max_dist == kmeans->min_dist)
202
- return 0.0;
203
-
204
- calculated_number_t anomaly_score = 100.0 * std::abs((mean_dist - kmeans->min_dist) / (kmeans->max_dist - kmeans->min_dist));
205
- return (anomaly_score > 100.0) ? 100.0 : anomaly_score;
206
-}
207
-
208
-/*
209
- * Queue
210
-*/
211
-
212
-nml_queue_t *nml_queue_init() {
213
- nml_queue_t *q = new nml_queue_t();
214
-
215
- netdata_mutex_init(&q->mutex);
216
- pthread_cond_init(&q->cond_var, NULL);
217
- q->exit = false;
218
- return q;
219
-}
220
-
221
-void nml_queue_destroy(nml_queue_t *q) {
222
- netdata_mutex_destroy(&q->mutex);
223
- pthread_cond_destroy(&q->cond_var);
224
- delete q;
225
-}
226
-
227
-void nml_queue_push(nml_queue_t *q, const nml_training_request_t req) {
228
- netdata_mutex_lock(&q->mutex);
229
- q->internal.push(req);
230
- pthread_cond_signal(&q->cond_var);
231
- netdata_mutex_unlock(&q->mutex);
232
-}
233
-
234
-nml_training_request_t nml_queue_pop(nml_queue_t *q) {
235
- netdata_mutex_lock(&q->mutex);
236
-
237
- nml_training_request_t req = { NULL, NULL, 0, 0, 0 };
238
-
239
- while (q->internal.empty()) {
240
- pthread_cond_wait(&q->cond_var, &q->mutex);
241
-
242
- if (q->exit) {
243
- netdata_mutex_unlock(&q->mutex);
244
-
245
- // We return a dummy request because the queue has been signaled
246
- return req;
247
- }
248
- }
249
-
250
- req = q->internal.front();
251
- q->internal.pop();
252
-
253
- netdata_mutex_unlock(&q->mutex);
254
- return req;
255
-}
256
-
257
-size_t nml_queue_size(nml_queue_t *q) {
258
- netdata_mutex_lock(&q->mutex);
259
- size_t size = q->internal.size();
260
- netdata_mutex_unlock(&q->mutex);
261
- return size;
262
-}
263
-
264
-void nml_queue_signal(nml_queue_t *q) {
265
- netdata_mutex_lock(&q->mutex);
266
- q->exit = true;
267
- pthread_cond_signal(&q->cond_var);
268
- netdata_mutex_unlock(&q->mutex);
269
-}
270
-
271
-/*
272
- * Dimension
273
-*/
274
-
275
-static std::pair<calculated_number_t *, nml_training_response_t>
276
-nml_dimension_calculated_numbers(nml_dimension_t *dim, const nml_training_request_t &training_request) {
277
- nml_training_response_t training_response = {};
278
-
279
- training_response.request_time = training_request.request_time;
280
- training_response.first_entry_on_request = training_request.first_entry_on_request;
281
- training_response.last_entry_on_request = training_request.last_entry_on_request;
282
-
283
- training_response.first_entry_on_response = rrddim_first_entry_s_of_tier(dim->rd, 0);
284
- training_response.last_entry_on_response = rrddim_last_entry_s_of_tier(dim->rd, 0);
285
-
286
- size_t min_n = Cfg.min_train_samples;
287
- size_t max_n = Cfg.max_train_samples;
288
-
289
- // Figure out what our time window should be.
290
- training_response.query_before_t = training_response.last_entry_on_response;
291
- training_response.query_after_t = std::max(
292
- training_response.query_before_t - static_cast<time_t>((max_n - 1) * dim->rd->update_every),
293
- training_response.first_entry_on_response
294
- );
295
-
296
- if (training_response.query_after_t >= training_response.query_before_t) {
297
- training_response.result = TRAINING_RESULT_INVALID_QUERY_TIME_RANGE;
298
- return { NULL, training_response };
299
- }
300
-
301
- if (rrdset_is_replicating(dim->rd->rrdset)) {
302
- training_response.result = TRAINING_RESULT_CHART_UNDER_REPLICATION;
303
- return { NULL, training_response };
304
- }
305
-
306
- /*
307
- * Execute the query
308
- */
309
- struct storage_engine_query_ops *ops = dim->rd->tiers[0].query_ops;
310
- struct storage_engine_query_handle handle;
311
-
312
- ops->init(dim->rd->tiers[0].db_metric_handle,
313
- &handle,
314
- training_response.query_after_t,
315
- training_response.query_before_t,
316
- STORAGE_PRIORITY_BEST_EFFORT);
317
-
318
- size_t idx = 0;
319
- memset(tls_data.training_cns, 0, sizeof(calculated_number_t) * max_n * (Cfg.lag_n + 1));
320
- calculated_number_t last_value = std::numeric_limits<calculated_number_t>::quiet_NaN();
321
-
322
- while (!ops->is_finished(&handle)) {
323
- if (idx == max_n)
324
- break;
325
-
326
- STORAGE_POINT sp = ops->next_metric(&handle);
327
-
328
- time_t timestamp = sp.end_time_s;
329
- calculated_number_t value = sp.sum / sp.count;
330
-
331
- if (netdata_double_isnumber(value)) {
332
- if (!training_response.db_after_t)
333
- training_response.db_after_t = timestamp;
334
- training_response.db_before_t = timestamp;
335
-
336
- tls_data.training_cns[idx] = value;
337
- last_value = tls_data.training_cns[idx];
338
- training_response.collected_values++;
339
- } else
340
- tls_data.training_cns[idx] = last_value;
341
-
342
- idx++;
343
- }
344
- ops->finalize(&handle);
345
-
346
- global_statistics_ml_query_completed(/* points_read */ idx);
347
-
348
- training_response.total_values = idx;
349
- if (training_response.collected_values < min_n) {
350
- training_response.result = TRAINING_RESULT_NOT_ENOUGH_COLLECTED_VALUES;
351
- return { NULL, training_response };
352
- }
353
-
354
- // Find first non-NaN value.
355
- for (idx = 0; std::isnan(tls_data.training_cns[idx]); idx++, training_response.total_values--) { }
356
-
357
- // Overwrite NaN values.
358
- if (idx != 0)
359
- memmove(tls_data.training_cns, &tls_data.training_cns[idx], sizeof(calculated_number_t) * training_response.total_values);
360
-
361
- training_response.result = TRAINING_RESULT_OK;
362
- return { tls_data.training_cns, training_response };
363
-}
364
-
365
-static enum nml_training_result
366
-nml_dimension_train_model(nml_dimension_t *dim, const nml_training_request_t &training_request) {
367
- auto P = nml_dimension_calculated_numbers(dim, training_request);
368
- nml_training_response_t training_response = P.second;
369
-
370
- if (training_response.result != TRAINING_RESULT_OK) {
371
- netdata_mutex_lock(&dim->mutex);
372
-
373
- dim->mt = METRIC_TYPE_CONSTANT;
374
-
375
- switch (dim->ts) {
376
- case TRAINING_STATUS_PENDING_WITH_MODEL:
377
- dim->ts = TRAINING_STATUS_TRAINED;
378
- break;
379
- case TRAINING_STATUS_PENDING_WITHOUT_MODEL:
380
- dim->ts = TRAINING_STATUS_UNTRAINED;
381
- break;
382
- default:
383
- break;
384
- }
385
-
386
- dim->tr = training_response;
387
-
388
- dim->last_training_time = training_response.last_entry_on_response;
389
- enum nml_training_result result = training_response.result;
390
- netdata_mutex_unlock(&dim->mutex);
391
-
392
- return result;
393
- }
394
-
395
- // compute kmeans
396
- {
397
- memcpy(tls_data.scratch_training_cns, tls_data.training_cns,
398
- training_response.total_values * sizeof(calculated_number_t));
399
-
400
- nml_features_t features = {
401
- Cfg.diff_n, Cfg.smooth_n, Cfg.lag_n,
402
- tls_data.scratch_training_cns, training_response.total_values,
403
- tls_data.training_cns, training_response.total_values,
404
- tls_data.training_samples
405
- };
406
- nml_features_preprocess(&features);
407
-
408
- nml_kmeans_init(&dim->kmeans, 2, 1000);
409
- nml_kmeans_train(&dim->kmeans, &features);
410
- }
411
-
412
- // update kmeans models
413
- {
414
- netdata_mutex_lock(&dim->mutex);
415
-
416
- if (dim->km_contexts.size() < Cfg.num_models_to_use) {
417
- dim->km_contexts.push_back(std::move(dim->kmeans));
418
- } else {
419
- std::rotate(std::begin(dim->km_contexts), std::begin(dim->km_contexts) + 1, std::end(dim->km_contexts));
420
- dim->km_contexts[dim->km_contexts.size() - 1] = std::move(dim->kmeans);
421
- }
422
-
423
- dim->mt = METRIC_TYPE_CONSTANT;
424
- dim->ts = TRAINING_STATUS_TRAINED;
425
- dim->tr = training_response;
426
- dim->last_training_time = rrddim_last_entry_s(dim->rd);
427
-
428
- netdata_mutex_unlock(&dim->mutex);
429
- }
430
-
431
- return training_response.result;
432
-}
433
-
434
-static void nml_dimension_schedule_for_training(nml_dimension_t *dim, time_t curr_time) {
435
- switch (dim->mt) {
436
- case METRIC_TYPE_CONSTANT:
437
- return;
438
- default:
439
- break;
440
- }
441
-
442
- bool schedule_for_training = false;
443
-
444
- switch (dim->ts) {
445
- case TRAINING_STATUS_PENDING_WITH_MODEL:
446
- case TRAINING_STATUS_PENDING_WITHOUT_MODEL:
447
- schedule_for_training = false;
448
- break;
449
- case TRAINING_STATUS_UNTRAINED:
450
- schedule_for_training = true;
451
- dim->ts = TRAINING_STATUS_PENDING_WITHOUT_MODEL;
452
- break;
453
- case TRAINING_STATUS_TRAINED:
454
- if ((dim->last_training_time + (Cfg.train_every * dim->rd->update_every)) < curr_time) {
455
- schedule_for_training = true;
456
- dim->ts = TRAINING_STATUS_PENDING_WITH_MODEL;
457
- }
458
- break;
459
- }
460
-
461
- if (schedule_for_training) {
462
- nml_host_t *host = reinterpret_cast<nml_host_t *>(dim->rd->rrdset->rrdhost->ml_host);
463
- nml_training_request_t req = {
464
- string_dup(dim->rd->rrdset->id), string_dup(dim->rd->id),
465
- curr_time, rrddim_first_entry_s(dim->rd), rrddim_last_entry_s(dim->rd),
466
- };
467
- nml_queue_push(host->training_queue, req);
468
- }
469
-}
470
-
471
-bool nml_dimension_predict(nml_dimension_t *dim, time_t curr_time, calculated_number_t value, bool exists) {
472
- // Nothing to do if ML is disabled for this dimension
473
- if (dim->mls != MACHINE_LEARNING_STATUS_ENABLED)
474
- return false;
475
-
476
- // Don't treat values that don't exist as anomalous
477
- if (!exists) {
478
- dim->cns.clear();
479
- return false;
480
- }
481
-
482
- // Save the value and return if we don't have enough values for a sample
483
- unsigned n = Cfg.diff_n + Cfg.smooth_n + Cfg.lag_n;
484
- if (dim->cns.size() < n) {
485
- dim->cns.push_back(value);
486
- return false;
487
- }
488
-
489
- // Push the value and check if it's different from the last one
490
- bool same_value = true;
491
- std::rotate(std::begin(dim->cns), std::begin(dim->cns) + 1, std::end(dim->cns));
492
- if (dim->cns[n - 1] != value)
493
- same_value = false;
494
- dim->cns[n - 1] = value;
495
-
496
- // Create the sample
497
- assert((n * (Cfg.lag_n + 1) <= 128) &&
498
- "Static buffers too small to perform prediction. "
499
- "This should not be possible with the default clamping of feature extraction options");
500
- calculated_number_t src_cns[128];
501
- calculated_number_t dst_cns[128];
502
-
503
- memset(src_cns, 0, n * (Cfg.lag_n + 1) * sizeof(calculated_number_t));
504
- memcpy(src_cns, dim->cns.data(), n * sizeof(calculated_number_t));
505
- memcpy(dst_cns, dim->cns.data(), n * sizeof(calculated_number_t));
506
-
507
- nml_features_t features = {
508
- Cfg.diff_n, Cfg.smooth_n, Cfg.lag_n,
509
- dst_cns, n, src_cns, n,
510
- dim->feature
511
- };
512
- nml_features_preprocess(&features);
513
-
514
- /*
515
- * Lock to predict and possibly schedule the dimension for training
516
- */
517
- if (netdata_mutex_trylock(&dim->mutex) != 0)
518
- return false;
519
-
520
- // Mark the metric time as variable if we received different values
521
- if (!same_value)
522
- dim->mt = METRIC_TYPE_VARIABLE;
523
-
524
- // Decide if the dimension needs to be scheduled for training
525
- nml_dimension_schedule_for_training(dim, curr_time);
526
-
527
- // Nothing to do if we don't have a model
528
- switch (dim->ts) {
529
- case TRAINING_STATUS_UNTRAINED:
530
- case TRAINING_STATUS_PENDING_WITHOUT_MODEL: {
531
- netdata_mutex_unlock(&dim->mutex);
532
- return false;
533
- }
534
- default:
535
- break;
536
- }
537
-
538
- /*
539
- * Use the KMeans models to check if the value is anomalous
540
- */
541
-
542
- size_t sum = 0;
543
- size_t models_consulted = 0;
544
-
545
- for (const auto &km_ctx : dim->km_contexts) {
546
- models_consulted++;
547
-
548
- calculated_number_t anomaly_score = nml_kmeans_anomaly_score(&km_ctx, features.preprocessed_features[0]);
549
- if (anomaly_score == std::numeric_limits<calculated_number_t>::quiet_NaN())
550
- continue;
551
-
552
- if (anomaly_score < (100 * Cfg.dimension_anomaly_score_threshold)) {
553
- global_statistics_ml_models_consulted(models_consulted);
554
- netdata_mutex_unlock(&dim->mutex);
555
- return false;
556
- }
557
-
558
- sum += 1;
559
- }
560
-
561
- netdata_mutex_unlock(&dim->mutex);
562
-
563
- global_statistics_ml_models_consulted(models_consulted);
564
- return sum;
565
-}
566
-
567
-void nml_dimension_dump(nml_dimension_t *dim) {
568
- const char *chart_id = rrdset_id(dim->rd->rrdset);
569
- const char *dimension_id = rrddim_id(dim->rd);
570
-
571
- const char *mls_str = nml_machine_learning_status_to_string(dim->mls);
572
- const char *mt_str = nml_metric_type_to_string(dim->mt);
573
- const char *ts_str = nml_training_status_to_string(dim->ts);
574
- const char *tr_str = nml_training_result_to_string(dim->tr.result);
575
-
576
- const char *fmt =
577
- "[ML] %s.%s: MLS=%s, MT=%s, TS=%s, Result=%s, "
578
- "ReqTime=%ld, FEOReq=%ld, LEOReq=%ld, "
579
- "FEOResp=%ld, LEOResp=%ld, QTR=<%ld, %ld>, DBTR=<%ld, %ld>, Collected=%zu, Total=%zu";
580
-
581
- error(fmt,
582
- chart_id, dimension_id, mls_str, mt_str, ts_str, tr_str,
583
- dim->tr.request_time, dim->tr.first_entry_on_request, dim->tr.last_entry_on_request,
584
- dim->tr.first_entry_on_response, dim->tr.last_entry_on_response,
585
- dim->tr.query_after_t, dim->tr.query_before_t, dim->tr.db_after_t, dim->tr.db_before_t, dim->tr.collected_values, dim->tr.total_values
586
- );
587
-}
588
-
589
-nml_dimension_t *nml_dimension_new(RRDDIM *rd) {
590
- nml_dimension_t *dim = new nml_dimension_t();
591
-
592
- dim->rd = rd;
593
-
594
- dim->mt = METRIC_TYPE_CONSTANT;
595
- dim->ts = TRAINING_STATUS_UNTRAINED;
596
-
597
- dim->last_training_time = 0;
598
-
599
- nml_kmeans_init(&dim->kmeans, 2, 1000);
600
-
601
- if (simple_pattern_matches(Cfg.sp_charts_to_skip, rrdset_name(rd->rrdset)))
602
- dim->mls = MACHINE_LEARNING_STATUS_DISABLED_DUE_TO_EXCLUDED_CHART;
603
- else
604
- dim->mls = MACHINE_LEARNING_STATUS_ENABLED;
605
-
606
- netdata_mutex_init(&dim->mutex);
607
-
608
- dim->km_contexts.reserve(Cfg.num_models_to_use);
609
-
610
- return dim;
611
-}
612
-
613
-void nml_dimension_delete(nml_dimension_t *dim) {
614
- netdata_mutex_destroy(&dim->mutex);
615
- delete dim;
616
-}
617
-
618
-nml_chart_t *nml_chart_new(RRDSET *rs) {
619
- nml_chart_t *chart = new nml_chart_t();
620
-
621
- chart->rs = rs;
622
- chart->mls = nml_machine_learning_stats_t();
623
-
624
- netdata_mutex_init(&chart->mutex);
625
-
626
- return chart;
627
-}
628
-
629
-void nml_chart_delete(nml_chart_t *chart) {
630
- netdata_mutex_destroy(&chart->mutex);
631
- delete chart;
632
-}
633
-
634
-static bool nml_chart_is_available_for_ml(nml_chart_t *chart) {
635
- return rrdset_is_available_for_exporting_and_alarms(chart->rs);
636
-}
637
-
638
-static std::string ml_dimension_get_id(RRDDIM *rd) {
639
- RRDSET *rs = rd->rrdset;
640
-
641
- std::stringstream ss;
642
- ss << rrdset_context(rs) << "|" << rrdset_id(rs) << "|" << rrddim_name(rd);
643
- return ss.str();
644
-}
645
-
646
-static void nml_chart_get_models_as_json(nml_chart_t *chart, nlohmann::json &j) {
647
- netdata_mutex_lock(&chart->mutex);
648
-
649
- void *rdp = NULL;
650
- rrddim_foreach_read(rdp, chart->rs) {
651
- RRDDIM *rd = static_cast<RRDDIM *>(rdp);
652
- nml_dimension_t *dim = reinterpret_cast<nml_dimension_t *>(rd->ml_dimension);
653
- if (!dim)
654
- continue;
655
-
656
- nlohmann::json jarray = nlohmann::json::array();
657
-#if 0
658
- for (const KMeans &KM : nml_dimension_models(D)) {
659
- nlohmann::json tmp;
660
-
661
- KM.toJson(tmp);
662
- jarray.push_back(tmp);
663
- j[ml_dimension_get_id(D->rd)] = jarray;
664
- }
665
-#else
666
- j[ml_dimension_get_id(dim->rd)] = jarray;
667
-#endif
668
- }
669
- rrdset_foreach_done(rdp);
670
-
671
- netdata_mutex_unlock(&chart->mutex);
672
-}
673
-
674
-void nml_chart_update_begin(nml_chart_t *chart) {
675
- netdata_mutex_lock(&chart->mutex);
676
- chart->mls = {};
677
-}
678
-
679
-void nml_chart_update_end(nml_chart_t *chart) {
680
- netdata_mutex_unlock(&chart->mutex);
681
-}
682
-
683
-void nml_chart_update_dimension(nml_chart_t *chart, nml_dimension_t *dim, bool is_anomalous) {
684
- switch (dim->mls) {
685
- case MACHINE_LEARNING_STATUS_DISABLED_DUE_TO_EXCLUDED_CHART:
686
- chart->mls.num_machine_learning_status_disabled_sp++;
687
- return;
688
- case MACHINE_LEARNING_STATUS_ENABLED: {
689
- chart->mls.num_machine_learning_status_enabled++;
690
-
691
- switch (dim->mt) {
692
- case METRIC_TYPE_CONSTANT:
693
- chart->mls.num_metric_type_constant++;
694
- chart->mls.num_training_status_trained++;
695
- chart->mls.num_normal_dimensions++;
696
- return;
697
- case METRIC_TYPE_VARIABLE:
698
- chart->mls.num_metric_type_variable++;
699
- break;
700
- }
701
-
702
- switch (dim->ts) {
703
- case TRAINING_STATUS_UNTRAINED:
704
- chart->mls.num_training_status_untrained++;
705
- return;
706
- case TRAINING_STATUS_PENDING_WITHOUT_MODEL:
707
- chart->mls.num_training_status_pending_without_model++;
708
- return;
709
- case TRAINING_STATUS_TRAINED:
710
- chart->mls.num_training_status_trained++;
711
-
712
- chart->mls.num_anomalous_dimensions += is_anomalous;
713
- chart->mls.num_normal_dimensions += !is_anomalous;
714
- return;
715
- case TRAINING_STATUS_PENDING_WITH_MODEL:
716
- chart->mls.num_training_status_pending_with_model++;
717
-
718
- chart->mls.num_anomalous_dimensions += is_anomalous;
719
- chart->mls.num_normal_dimensions += !is_anomalous;
720
- return;
721
- }
722
-
723
- return;
724
- }
725
- }
726
-}
727
-
728
-nml_host_t *nml_host_new(RRDHOST *rh) {
729
- nml_host_t *host = new nml_host_t();
730
-
731
- host->rh = rh;
732
- host->mls = nml_machine_learning_stats_t();
733
- host->ts = nml_training_stats_t();
734
-
735
- host->host_anomaly_rate = 0.0;
736
- host->threads_running = false;
737
- host->threads_cancelled = false;
738
- host->threads_joined = false;
739
-
740
- host->training_queue = nml_queue_init();
741
-
742
- netdata_mutex_init(&host->mutex);
743
-
744
- return host;
745
-}
746
-
747
-void nml_host_delete(nml_host_t *host) {
748
- netdata_mutex_destroy(&host->mutex);
749
- nml_queue_destroy(host->training_queue);
750
- delete host;
751
-}
752
-
753
-void nml_host_get_config_as_json(nml_host_t *host, BUFFER *wb) {
754
- // Unused for now, until we add support for per-host configs
755
- (void) host;
756
-
757
- buffer_json_member_add_uint64(wb, "version", 1);
758
-
759
- buffer_json_member_add_boolean(wb, "enabled", Cfg.enable_anomaly_detection);
760
-
761
- buffer_json_member_add_uint64(wb, "min-train-samples", Cfg.min_train_samples);
762
- buffer_json_member_add_uint64(wb, "max-train-samples", Cfg.max_train_samples);
763
- buffer_json_member_add_uint64(wb, "train-every", Cfg.train_every);
764
-
765
- buffer_json_member_add_uint64(wb, "diff-n", Cfg.diff_n);
766
- buffer_json_member_add_uint64(wb, "smooth-n", Cfg.smooth_n);
767
- buffer_json_member_add_uint64(wb, "lag-n", Cfg.lag_n);
768
-
769
- buffer_json_member_add_double(wb, "random-sampling-ratio", Cfg.random_sampling_ratio);
770
- buffer_json_member_add_uint64(wb, "max-kmeans-iters", Cfg.random_sampling_ratio);
771
-
772
- buffer_json_member_add_double(wb, "dimension-anomaly-score-threshold", Cfg.dimension_anomaly_score_threshold);
773
-
774
- buffer_json_member_add_string(wb, "anomaly-detection-grouping-method",
775
- time_grouping_method2string(Cfg.anomaly_detection_grouping_method));
776
-
777
- buffer_json_member_add_int64(wb, "anomaly-detection-query-duration", Cfg.anomaly_detection_query_duration);
778
-
779
- buffer_json_member_add_string(wb, "hosts-to-skip", Cfg.hosts_to_skip.c_str());
780
- buffer_json_member_add_string(wb, "charts-to-skip", Cfg.charts_to_skip.c_str());
781
-}
782
-
783
-void nml_host_get_models_as_json(nml_host_t *host, nlohmann::json &j) {
784
- netdata_mutex_lock(&host->mutex);
785
-
786
- void* rsp = NULL;
787
- rrdset_foreach_read(rsp, host->rh) {
788
- RRDSET *rs = static_cast<RRDSET *>(rsp);
789
- nml_chart_t *chart = reinterpret_cast<nml_chart_t *>(rs->ml_chart);
790
-
791
- if (!chart)
792
- continue;
793
-
794
- nml_chart_get_models_as_json(chart, j);
795
- }
796
- rrdset_foreach_done(rsp);
797
-
798
- netdata_mutex_unlock(&host->mutex);
799
-}
800
-
801
-#define WORKER_JOB_DETECTION_PREP 0
802
-#define WORKER_JOB_DETECTION_DIM_CHART 1
803
-#define WORKER_JOB_DETECTION_HOST_CHART 2
804
-#define WORKER_JOB_DETECTION_STATS 3
805
-#define WORKER_JOB_DETECTION_RESOURCES 4
806
-
807
-static void nml_host_detect_once(nml_host_t *host) {
808
- worker_is_busy(WORKER_JOB_DETECTION_PREP);
809
-
810
- host->mls = {};
811
- nml_machine_learning_stats_t mls_copy = {};
812
- nml_training_stats_t ts_copy = {};
813
-
814
- {
815
- netdata_mutex_lock(&host->mutex);
816
-
817
- /*
818
- * prediction/detection stats
819
- */
820
- void *rsp = NULL;
821
- rrdset_foreach_read(rsp, host->rh) {
822
- RRDSET *rs = static_cast<RRDSET *>(rsp);
823
-
824
- nml_chart_t *chart = reinterpret_cast<nml_chart_t *>(rs->ml_chart);
825
- if (!chart)
826
- continue;
827
-
828
- if (!nml_chart_is_available_for_ml(chart))
829
- continue;
830
-
831
- nml_machine_learning_stats_t chart_mls = chart->mls;
832
-
833
- host->mls.num_machine_learning_status_enabled += chart_mls.num_machine_learning_status_enabled;
834
- host->mls.num_machine_learning_status_disabled_sp += chart_mls.num_machine_learning_status_disabled_sp;
835
-
836
- host->mls.num_metric_type_constant += chart_mls.num_metric_type_constant;
837
- host->mls.num_metric_type_variable += chart_mls.num_metric_type_variable;
838
-
839
- host->mls.num_training_status_untrained += chart_mls.num_training_status_untrained;
840
- host->mls.num_training_status_pending_without_model += chart_mls.num_training_status_pending_without_model;
841
- host->mls.num_training_status_trained += chart_mls.num_training_status_trained;
842
- host->mls.num_training_status_pending_with_model += chart_mls.num_training_status_pending_with_model;
843
-
844
- host->mls.num_anomalous_dimensions += chart_mls.num_anomalous_dimensions;
845
- host->mls.num_normal_dimensions += chart_mls.num_normal_dimensions;
846
- }
847
- rrdset_foreach_done(rsp);
848
-
849
- host->host_anomaly_rate = 0.0;
850
- size_t NumActiveDimensions = host->mls.num_anomalous_dimensions + host->mls.num_normal_dimensions;
851
- if (NumActiveDimensions)
852
- host->host_anomaly_rate = static_cast<double>(host->mls.num_anomalous_dimensions) / NumActiveDimensions;
853
-
854
- mls_copy = host->mls;
855
-
856
- /*
857
- * training stats
858
- */
859
- ts_copy = host->ts;
860
-
861
- host->ts.queue_size = 0;
862
- host->ts.num_popped_items = 0;
863
-
864
- host->ts.allotted_ut = 0;
865
- host->ts.consumed_ut = 0;
866
- host->ts.remaining_ut = 0;
867
-
868
- host->ts.training_result_ok = 0;
869
- host->ts.training_result_invalid_query_time_range = 0;
870
- host->ts.training_result_not_enough_collected_values = 0;
871
- host->ts.training_result_null_acquired_dimension = 0;
872
- host->ts.training_result_chart_under_replication = 0;
873
-
874
- netdata_mutex_unlock(&host->mutex);
875
- }
876
-
877
- // Calc the avg values
878
- if (ts_copy.num_popped_items) {
879
- ts_copy.queue_size /= ts_copy.num_popped_items;
880
- ts_copy.allotted_ut /= ts_copy.num_popped_items;
881
- ts_copy.consumed_ut /= ts_copy.num_popped_items;
882
- ts_copy.remaining_ut /= ts_copy.num_popped_items;
883
-
884
- ts_copy.training_result_ok /= ts_copy.num_popped_items;
885
- ts_copy.training_result_invalid_query_time_range /= ts_copy.num_popped_items;
886
- ts_copy.training_result_not_enough_collected_values /= ts_copy.num_popped_items;
887
- ts_copy.training_result_null_acquired_dimension /= ts_copy.num_popped_items;
888
- ts_copy.training_result_chart_under_replication /= ts_copy.num_popped_items;
889
- } else {
890
- ts_copy.queue_size = 0;
891
- ts_copy.allotted_ut = 0;
892
- ts_copy.consumed_ut = 0;
893
- ts_copy.remaining_ut = 0;
894
- }
895
-
896
- worker_is_busy(WORKER_JOB_DETECTION_DIM_CHART);
897
- nml_update_dimensions_chart(host, mls_copy);
898
-
899
- worker_is_busy(WORKER_JOB_DETECTION_HOST_CHART);
900
- nml_update_host_and_detection_rate_charts(host, host->host_anomaly_rate * 10000.0);
901
-
902
-#ifdef NETDATA_ML_RESOURCE_CHARTS
903
- worker_is_busy(WORKER_JOB_DETECTION_RESOURCES);
904
- struct rusage PredictionRU;
905
- getrusage(RUSAGE_THREAD, &PredictionRU);
906
- updateResourceUsageCharts(RH, PredictionRU, TSCopy.TrainingRU);
907
-#endif
908
-
909
- worker_is_busy(WORKER_JOB_DETECTION_STATS);
910
- nml_update_training_statistics_chart(host, ts_copy);
911
-}
912
-
913
-typedef struct {
914
- RRDDIM_ACQUIRED *acq_rd;
915
- nml_dimension_t *dim;
916
-} nml_acquired_dimension_t;
917
-
918
-static nml_acquired_dimension_t nml_acquired_dimension_get(RRDHOST *rh, STRING *chart_id, STRING *dimension_id) {
919
- RRDDIM_ACQUIRED *acq_rd = NULL;
920
- nml_dimension_t *dim = NULL;
921
-
922
- RRDSET *rs = rrdset_find(rh, string2str(chart_id));
923
- if (rs) {
924
- acq_rd = rrddim_find_and_acquire(rs, string2str(dimension_id));
925
- if (acq_rd) {
926
- RRDDIM *rd = rrddim_acquired_to_rrddim(acq_rd);
927
- if (rd)
928
- dim = reinterpret_cast<nml_dimension_t *>(rd->ml_dimension);
929
- }
930
- }
931
-
932
- nml_acquired_dimension_t acq_dim = {
933
- acq_rd, dim
934
- };
935
-
936
- return acq_dim;
937
-}
938
-
939
-static void nml_acquired_dimension_release(nml_acquired_dimension_t acq_dim) {
940
- if (!acq_dim.acq_rd)
941
- return;
942
-
943
- rrddim_acquired_release(acq_dim.acq_rd);
944
-}
945
-
946
-static enum nml_training_result nml_acquired_dimension_train(nml_acquired_dimension_t acq_dim, const nml_training_request_t &TR) {
947
- if (!acq_dim.dim)
948
- return TRAINING_RESULT_NULL_ACQUIRED_DIMENSION;
949
-
950
- return nml_dimension_train_model(acq_dim.dim, TR);
951
-}
952
-
953
-#define WORKER_JOB_TRAINING_FIND 0
954
-#define WORKER_JOB_TRAINING_TRAIN 1
955
-#define WORKER_JOB_TRAINING_STATS 2
956
-
957
-void nml_host_get_detection_info_as_json(nml_host_t *host, nlohmann::json &j) {
958
- j["version"] = 1;
959
- j["anomalous-dimensions"] = host->mls.num_anomalous_dimensions;
960
- j["normal-dimensions"] = host->mls.num_normal_dimensions;
961
- j["total-dimensions"] = host->mls.num_anomalous_dimensions + host->mls.num_normal_dimensions;
962
- j["trained-dimensions"] = host->mls.num_training_status_trained + host->mls.num_training_status_pending_with_model;
963
-}
964
-
965
-void nml_host_train(nml_host_t *host) {
966
- worker_register("MLTRAIN");
967
- worker_register_job_name(WORKER_JOB_TRAINING_FIND, "find");
968
- worker_register_job_name(WORKER_JOB_TRAINING_TRAIN, "train");
969
- worker_register_job_name(WORKER_JOB_TRAINING_STATS, "stats");
970
-
971
- service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, (force_quit_t )ml_cancel_training_thread, host->rh, true);
972
-
973
- while (service_running(SERVICE_ML_TRAINING)) {
974
- nml_training_request_t training_req = nml_queue_pop(host->training_queue);
975
- size_t queue_size = nml_queue_size(host->training_queue) + 1;
976
-
977
- if (host->threads_cancelled) {
978
- info("Stopping training thread for host %s because it was cancelled", rrdhost_hostname(host->rh));
979
- break;
980
- }
981
-
982
- usec_t allotted_ut = (Cfg.train_every * host->rh->rrd_update_every * USEC_PER_SEC) / queue_size;
983
- if (allotted_ut > USEC_PER_SEC)
984
- allotted_ut = USEC_PER_SEC;
985
-
986
- usec_t start_ut = now_monotonic_usec();
987
- enum nml_training_result training_res;
988
- {
989
- worker_is_busy(WORKER_JOB_TRAINING_FIND);
990
- nml_acquired_dimension_t acq_dim = nml_acquired_dimension_get(host->rh, training_req.chart_id, training_req.dimension_id);
991
-
992
- worker_is_busy(WORKER_JOB_TRAINING_TRAIN);
993
- training_res = nml_acquired_dimension_train(acq_dim, training_req);
994
-
995
- string_freez(training_req.chart_id);
996
- string_freez(training_req.dimension_id);
997
-
998
- nml_acquired_dimension_release(acq_dim);
999
- }
1000
- usec_t consumed_ut = now_monotonic_usec() - start_ut;
1001
-
1002
- worker_is_busy(WORKER_JOB_TRAINING_STATS);
1003
-
1004
- usec_t remaining_ut = 0;
1005
- if (consumed_ut < allotted_ut)
1006
- remaining_ut = allotted_ut - consumed_ut;
1007
-
1008
- {
1009
- netdata_mutex_lock(&host->mutex);
1010
-
1011
- if (host->ts.allotted_ut == 0) {
1012
- struct rusage TRU;
1013
- getrusage(RUSAGE_THREAD, &TRU);
1014
- host->ts.training_ru = TRU;
1015
- }
1016
-
1017
- host->ts.queue_size += queue_size;
1018
- host->ts.num_popped_items += 1;
1019
-
1020
- host->ts.allotted_ut += allotted_ut;
1021
- host->ts.consumed_ut += consumed_ut;
1022
- host->ts.remaining_ut += remaining_ut;
1023
-
1024
- switch (training_res) {
1025
- case TRAINING_RESULT_OK:
1026
- host->ts.training_result_ok += 1;
1027
- break;
1028
- case TRAINING_RESULT_INVALID_QUERY_TIME_RANGE:
1029
- host->ts.training_result_invalid_query_time_range += 1;
1030
- break;
1031
- case TRAINING_RESULT_NOT_ENOUGH_COLLECTED_VALUES:
1032
- host->ts.training_result_not_enough_collected_values += 1;
1033
- break;
1034
- case TRAINING_RESULT_NULL_ACQUIRED_DIMENSION:
1035
- host->ts.training_result_null_acquired_dimension += 1;
1036
- break;
1037
- case TRAINING_RESULT_CHART_UNDER_REPLICATION:
1038
- host->ts.training_result_chart_under_replication += 1;
1039
- break;
1040
- }
1041
-
1042
- netdata_mutex_unlock(&host->mutex);
1043
- }
1044
-
1045
- worker_is_idle();
1046
- std::this_thread::sleep_for(std::chrono::microseconds{remaining_ut});
1047
- worker_is_busy(0);
1048
- }
1049
-}
1050
-
1051
-static void *train_main(void *arg) {
1052
- size_t max_elements_needed_for_training = Cfg.max_train_samples * (Cfg.lag_n + 1);
1053
- tls_data.training_cns = new calculated_number_t[max_elements_needed_for_training]();
1054
- tls_data.scratch_training_cns = new calculated_number_t[max_elements_needed_for_training]();
1055
-
1056
- nml_host_t *host = reinterpret_cast<nml_host_t *>(arg);
1057
- nml_host_train(host);
1058
- return NULL;
1059
-}
1060
-
1061
-void nml_host_start_training_thread(nml_host_t *host) {
1062
- if (host->threads_running) {
1063
- error("Anomaly detections threads for host %s are already-up and running.", rrdhost_hostname(host->rh));
1064
- return;
1065
- }
1066
-
1067
- host->threads_running = true;
1068
- host->threads_cancelled = false;
1069
- host->threads_joined = false;
1070
-
1071
- char tag[NETDATA_THREAD_TAG_MAX + 1];
1072
-
1073
- snprintfz(tag, NETDATA_THREAD_TAG_MAX, "MLTR[%s]", rrdhost_hostname(host->rh));
1074
- netdata_thread_create(&host->training_thread, tag, NETDATA_THREAD_OPTION_JOINABLE, train_main, static_cast<void *>(host));
1075
-}
1076
-
1077
-void nml_host_stop_training_thread(nml_host_t *host, bool join) {
1078
- if (!host->threads_running) {
1079
- error("Anomaly detections threads for host %s have already been stopped.", rrdhost_hostname(host->rh));
1080
- return;
1081
- }
1082
-
1083
- if (!host->threads_cancelled) {
1084
- host->threads_cancelled = true;
1085
-
1086
- // Signal the training queue to stop popping-items
1087
- nml_queue_signal(host->training_queue);
1088
- netdata_thread_cancel(host->training_thread);
1089
- }
1090
-
1091
- if (join && !host->threads_joined) {
1092
- host->threads_joined = true;
1093
- host->threads_running = false;
1094
-
1095
- delete[] tls_data.training_cns;
1096
- delete[] tls_data.scratch_training_cns;
1097
-
1098
- netdata_thread_join(host->training_thread, NULL);
1099
- }
1100
-}
1101
-
1102
-void *nml_detect_main(void *arg) {
1103
- UNUSED(arg);
1104
-
1105
- worker_register("MLDETECT");
1106
- worker_register_job_name(WORKER_JOB_DETECTION_PREP, "prep");
1107
- worker_register_job_name(WORKER_JOB_DETECTION_DIM_CHART, "dim chart");
1108
- worker_register_job_name(WORKER_JOB_DETECTION_HOST_CHART, "host chart");
1109
- worker_register_job_name(WORKER_JOB_DETECTION_STATS, "stats");
1110
- worker_register_job_name(WORKER_JOB_DETECTION_RESOURCES, "resources");
1111
-
1112
- service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, NULL, NULL, true);
1113
-
1114
- heartbeat_t hb;
1115
- heartbeat_init(&hb);
1116
-
1117
- while (service_running((SERVICE_TYPE)(SERVICE_ML_PREDICTION | SERVICE_COLLECTORS))) {
1118
- worker_is_idle();
1119
- heartbeat_next(&hb, USEC_PER_SEC);
1120
-
1121
- void *rhp;
1122
- dfe_start_reentrant(rrdhost_root_index, rhp) {
1123
- RRDHOST *rh = (RRDHOST *) rhp;
1124
-
1125
- if (!rh->ml_host)
1126
- continue;
1127
-
1128
- nml_host_detect_once(reinterpret_cast<nml_host_t *>(rh->ml_host));
1129
- }
1130
- dfe_done(rhp);
1131
- }
1132
-
1133
- return NULL;
1134
-}
web/api/web_api_v1.c
+8
-14
@@ -1191,7 +1191,7 @@ inline int web_client_api_request_v1_info_fill_buffer(RRDHOST *host, BUFFER *wb)
1191
1192
#if defined(ENABLE_ML)
1193
buffer_json_member_add_object(wb, "ml-info");
1194
- ml_get_host_info(host, wb);
1194
+ ml_host_get_info(host, wb);
1195
buffer_json_object_close(wb);
1196
#endif
1197
@@ -1206,17 +1206,16 @@ int web_client_api_request_v1_ml_info(RRDHOST *host, struct web_client *w, char
1206
if (!netdata_ready)
1207
return HTTP_RESP_BACKEND_FETCH_FAILED;
1208
1209
- char *s = ml_get_host_runtime_info(host);
1210
- if (!s)
1211
- s = strdupz("{\"error\": \"json string is empty\" }\n");
1212
-
1209
BUFFER *wb = w->response.data;
1210
buffer_flush(wb);
1211
wb->content_type = CT_APPLICATION_JSON;
1216
- buffer_strcat(wb, s);
1212
+
1213
+ buffer_json_initialize(wb, "\"", "\"", 0, true, false);
1214
+ ml_host_get_detection_info(host, wb);
1215
+ buffer_json_finalize(wb);
1216
+
1217
buffer_no_cacheable(wb);
1218
1219
- freez(s);
1219
return HTTP_RESP_OK;
1220
}
1221
@@ -1226,20 +1225,15 @@ int web_client_api_request_v1_ml_models(RRDHOST *host, struct web_client *w, cha
1225
if (!netdata_ready)
1226
return HTTP_RESP_BACKEND_FETCH_FAILED;
1227
1229
- char *s = ml_get_host_models(host);
1230
- if (!s)
1231
- s = strdupz("{\"error\": \"json string is empty\" }\n");
1232
-
1228
BUFFER *wb = w->response.data;
1229
buffer_flush(wb);
1230
wb->content_type = CT_APPLICATION_JSON;
1236
- buffer_strcat(wb, s);
1231
+ ml_host_get_models(host, wb);
1232
buffer_no_cacheable(wb);
1233
1239
- freez(s);
1234
return HTTP_RESP_OK;
1235
}
1242
-#endif
1236
+#endif // ENABLE_ML
1237
1238
inline int web_client_api_request_v1_info(RRDHOST *host, struct web_client *w, char *url) {
1239
(void)url;