From 03bf87dcb06f7021bfb2df2fa8691593c6148aff Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Wed, 30 Nov 2022 19:47:00 +0100 Subject: Adding upstream version 1.37.0. Signed-off-by: Daniel Baumann --- ml/ADCharts.cc | 233 ++++++++++++++++++++++++++++ ml/ADCharts.h | 23 +++ ml/BitBufferCounter.cc | 29 ---- ml/BitBufferCounter.h | 54 ------- ml/BitRateWindow.cc | 75 --------- ml/BitRateWindow.h | 170 --------------------- ml/Config.cc | 52 ++----- ml/Config.h | 11 +- ml/Database.cc | 127 --------------- ml/Database.h | 131 ---------------- ml/Dimension.cc | 76 ++++++--- ml/Dimension.h | 165 ++++++-------------- ml/Host.cc | 374 +++++---------------------------------------- ml/Host.h | 51 +------ ml/KMeans.cc | 43 ++++++ ml/KMeans.h | 41 +++++ ml/Makefile.am | 8 - ml/Query.h | 27 ++-- ml/README.md | 2 - ml/SamplesBuffer.cc | 150 ++++++++++++++++++ ml/SamplesBuffer.h | 146 ++++++++++++++++++ ml/SamplesBufferTests.cc | 146 ++++++++++++++++++ ml/Tests.cc | 301 ------------------------------------ ml/kmeans/KMeans.cc | 55 ------- ml/kmeans/KMeans.h | 34 ----- ml/kmeans/Makefile.am | 4 - ml/kmeans/SamplesBuffer.cc | 150 ------------------ ml/kmeans/SamplesBuffer.h | 146 ------------------ ml/kmeans/Tests.cc | 143 ----------------- ml/ml-dummy.c | 35 ++--- ml/ml-private.h | 2 +- ml/ml.cc | 99 ++---------- ml/ml.h | 13 +- 33 files changed, 993 insertions(+), 2123 deletions(-) create mode 100644 ml/ADCharts.cc create mode 100644 ml/ADCharts.h delete mode 100644 ml/BitBufferCounter.cc delete mode 100644 ml/BitBufferCounter.h delete mode 100644 ml/BitRateWindow.cc delete mode 100644 ml/BitRateWindow.h delete mode 100644 ml/Database.cc delete mode 100644 ml/Database.h create mode 100644 ml/KMeans.cc create mode 100644 ml/KMeans.h delete mode 100644 ml/Makefile.am create mode 100644 ml/SamplesBuffer.cc create mode 100644 ml/SamplesBuffer.h create mode 100644 ml/SamplesBufferTests.cc delete mode 100644 ml/Tests.cc delete mode 100644 ml/kmeans/KMeans.cc delete mode 100644 ml/kmeans/KMeans.h delete mode 100644 ml/kmeans/Makefile.am delete mode 100644 ml/kmeans/SamplesBuffer.cc delete mode 100644 ml/kmeans/SamplesBuffer.h delete mode 100644 ml/kmeans/Tests.cc (limited to 'ml') diff --git a/ml/ADCharts.cc b/ml/ADCharts.cc new file mode 100644 index 000000000..00c593c0c --- /dev/null +++ b/ml/ADCharts.cc @@ -0,0 +1,233 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#include "ADCharts.h" +#include "Config.h" + +void ml::updateDimensionsChart(RRDHOST *RH, + collected_number NumTrainedDimensions, + collected_number NumNormalDimensions, + collected_number NumAnomalousDimensions) { + static thread_local RRDSET *RS = nullptr; + static thread_local RRDDIM *NumTotalDimensionsRD = nullptr; + static thread_local RRDDIM *NumTrainedDimensionsRD = nullptr; + static thread_local RRDDIM *NumNormalDimensionsRD = nullptr; + static thread_local RRDDIM *NumAnomalousDimensionsRD = nullptr; + + if (!RS) { + std::stringstream IdSS, NameSS; + + IdSS << "dimensions_on_" << localhost->machine_guid; + NameSS << "dimensions_on_" << localhost->hostname; + + RS = rrdset_create( + RH, + "anomaly_detection", // type + IdSS.str().c_str(), // id + NameSS.str().c_str(), // name + "dimensions", // family + "anomaly_detection.dimensions", // ctx + "Anomaly detection dimensions", // title + "dimensions", // units + "netdata", // plugin + "ml", // module + 39183, // priority + RH->rrd_update_every, // update_every + RRDSET_TYPE_LINE // chart_type + ); + rrdset_flag_set(RS, RRDSET_FLAG_ANOMALY_DETECTION); + + NumTotalDimensionsRD = rrddim_add(RS, "total", NULL, + 1, 1, RRD_ALGORITHM_ABSOLUTE); + NumTrainedDimensionsRD = rrddim_add(RS, "trained", NULL, + 1, 1, RRD_ALGORITHM_ABSOLUTE); + NumNormalDimensionsRD = rrddim_add(RS, "normal", NULL, + 1, 1, RRD_ALGORITHM_ABSOLUTE); + NumAnomalousDimensionsRD = rrddim_add(RS, "anomalous", NULL, + 1, 1, RRD_ALGORITHM_ABSOLUTE); + } + + rrddim_set_by_pointer(RS, NumTotalDimensionsRD, NumNormalDimensions + NumAnomalousDimensions); + rrddim_set_by_pointer(RS, NumTrainedDimensionsRD, NumTrainedDimensions); + rrddim_set_by_pointer(RS, NumNormalDimensionsRD, NumNormalDimensions); + rrddim_set_by_pointer(RS, NumAnomalousDimensionsRD, NumAnomalousDimensions); + + rrdset_done(RS); +} + +void ml::updateHostAndDetectionRateCharts(RRDHOST *RH, collected_number AnomalyRate) { + static thread_local RRDSET *HostRateRS = nullptr; + static thread_local RRDDIM *AnomalyRateRD = nullptr; + + if (!HostRateRS) { + std::stringstream IdSS, NameSS; + + IdSS << "anomaly_rate_on_" << localhost->machine_guid; + NameSS << "anomaly_rate_on_" << localhost->hostname; + + HostRateRS = rrdset_create( + RH, + "anomaly_detection", // type + IdSS.str().c_str(), // id + NameSS.str().c_str(), // name + "anomaly_rate", // family + "anomaly_detection.anomaly_rate", // ctx + "Percentage of anomalous dimensions", // title + "percentage", // units + "netdata", // plugin + "ml", // module + 39184, // priority + RH->rrd_update_every, // update_every + RRDSET_TYPE_LINE // chart_type + ); + rrdset_flag_set(HostRateRS, RRDSET_FLAG_ANOMALY_DETECTION); + + AnomalyRateRD = rrddim_add(HostRateRS, "anomaly_rate", NULL, + 1, 100, RRD_ALGORITHM_ABSOLUTE); + } + + rrddim_set_by_pointer(HostRateRS, AnomalyRateRD, AnomalyRate); + rrdset_done(HostRateRS); + + static thread_local RRDSET *AnomalyDetectionRS = nullptr; + static thread_local RRDDIM *AboveThresholdRD = nullptr; + static thread_local RRDDIM *NewAnomalyEventRD = nullptr; + + if (!AnomalyDetectionRS) { + std::stringstream IdSS, NameSS; + + IdSS << "anomaly_detection_on_" << localhost->machine_guid; + NameSS << "anomaly_detection_on_" << localhost->hostname; + + AnomalyDetectionRS = rrdset_create( + RH, + "anomaly_detection", // type + IdSS.str().c_str(), // id + NameSS.str().c_str(), // name + "anomaly_detection", // family + "anomaly_detection.detector_events", // ctx + "Anomaly detection events", // title + "percentage", // units + "netdata", // plugin + "ml", // module + 39185, // priority + RH->rrd_update_every, // update_every + RRDSET_TYPE_LINE // chart_type + ); + rrdset_flag_set(AnomalyDetectionRS, RRDSET_FLAG_ANOMALY_DETECTION); + + AboveThresholdRD = rrddim_add(AnomalyDetectionRS, "above_threshold", NULL, + 1, 1, RRD_ALGORITHM_ABSOLUTE); + NewAnomalyEventRD = rrddim_add(AnomalyDetectionRS, "new_anomaly_event", NULL, + 1, 1, RRD_ALGORITHM_ABSOLUTE); + } + + /* + * Compute the values of the dimensions based on the host rate chart + */ + ONEWAYALLOC *OWA = onewayalloc_create(0); + time_t Now = now_realtime_sec(); + time_t Before = Now - RH->rrd_update_every; + time_t After = Before - Cfg.AnomalyDetectionQueryDuration; + RRDR_OPTIONS Options = static_cast(0x00000000); + + RRDR *R = rrd2rrdr_legacy( + OWA, HostRateRS, + 1 /* points wanted */, + After, + Before, + Cfg.AnomalyDetectionGroupingMethod, + 0 /* resampling time */, + Options, "anomaly_rate", + NULL /* group options */, + 0, /* timeout */ + 0, /* tier */ + QUERY_SOURCE_ML + ); + if(R) { + assert(R->d == 1 && R->n == 1 && R->rows == 1); + + static thread_local bool PrevAboveThreshold = false; + bool AboveThreshold = R->v[0] >= Cfg.HostAnomalyRateThreshold; + bool NewAnomalyEvent = AboveThreshold && !PrevAboveThreshold; + PrevAboveThreshold = AboveThreshold; + + rrddim_set_by_pointer(AnomalyDetectionRS, AboveThresholdRD, AboveThreshold); + rrddim_set_by_pointer(AnomalyDetectionRS, NewAnomalyEventRD, NewAnomalyEvent); + rrdset_done(AnomalyDetectionRS); + + rrdr_free(OWA, R); + } + onewayalloc_destroy(OWA); +} + +void ml::updateDetectionChart(RRDHOST *RH) { + static thread_local RRDSET *RS = nullptr; + static thread_local RRDDIM *UserRD, *SystemRD = nullptr; + + if (!RS) { + std::stringstream IdSS, NameSS; + + IdSS << "prediction_stats_" << RH->machine_guid; + NameSS << "prediction_stats_for_" << RH->hostname; + + RS = rrdset_create_localhost( + "netdata", // type + IdSS.str().c_str(), // id + NameSS.str().c_str(), // name + "ml", // family + "netdata.prediction_stats", // ctx + "Prediction thread CPU usage", // title + "milliseconds/s", // units + "netdata", // plugin + "ml", // module + 136000, // priority + RH->rrd_update_every, // update_every + RRDSET_TYPE_STACKED // chart_type + ); + + UserRD = rrddim_add(RS, "user", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL); + SystemRD = rrddim_add(RS, "system", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL); + } + + struct rusage TRU; + getrusage(RUSAGE_THREAD, &TRU); + + rrddim_set_by_pointer(RS, UserRD, TRU.ru_utime.tv_sec * 1000000ULL + TRU.ru_utime.tv_usec); + rrddim_set_by_pointer(RS, SystemRD, TRU.ru_stime.tv_sec * 1000000ULL + TRU.ru_stime.tv_usec); + rrdset_done(RS); +} + +void ml::updateTrainingChart(RRDHOST *RH, struct rusage *TRU) { + static thread_local RRDSET *RS = nullptr; + static thread_local RRDDIM *UserRD = nullptr; + static thread_local RRDDIM *SystemRD = nullptr; + + if (!RS) { + std::stringstream IdSS, NameSS; + + IdSS << "training_stats_" << RH->machine_guid; + NameSS << "training_stats_for_" << RH->hostname; + + RS = rrdset_create_localhost( + "netdata", // type + IdSS.str().c_str(), // id + NameSS.str().c_str(), // name + "ml", // family + "netdata.training_stats", // ctx + "Training thread CPU usage", // title + "milliseconds/s", // units + "netdata", // plugin + "ml", // module + 136001, // priority + RH->rrd_update_every, // update_every + RRDSET_TYPE_STACKED // chart_type + ); + + UserRD = rrddim_add(RS, "user", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL); + SystemRD = rrddim_add(RS, "system", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL); + } + + rrddim_set_by_pointer(RS, UserRD, TRU->ru_utime.tv_sec * 1000000ULL + TRU->ru_utime.tv_usec); + rrddim_set_by_pointer(RS, SystemRD, TRU->ru_stime.tv_sec * 1000000ULL + TRU->ru_stime.tv_usec); + rrdset_done(RS); +} diff --git a/ml/ADCharts.h b/ml/ADCharts.h new file mode 100644 index 000000000..0be324f7d --- /dev/null +++ b/ml/ADCharts.h @@ -0,0 +1,23 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#ifndef ML_ADCHARTS_H +#define ML_ADCHARTS_H + +#include "ml-private.h" + +namespace ml { + +void updateDimensionsChart(RRDHOST *RH, + collected_number NumTrainedDimensions, + collected_number NumNormalDimensions, + collected_number NumAnomalousDimensions); + +void updateHostAndDetectionRateCharts(RRDHOST *RH, collected_number AnomalyRate); + +void updateDetectionChart(RRDHOST *RH); + +void updateTrainingChart(RRDHOST *RH, struct rusage *TRU); + +} // namespace ml + +#endif /* ML_ADCHARTS_H */ diff --git a/ml/BitBufferCounter.cc b/ml/BitBufferCounter.cc deleted file mode 100644 index 5e1ab5aca..000000000 --- a/ml/BitBufferCounter.cc +++ /dev/null @@ -1,29 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "BitBufferCounter.h" - -using namespace ml; - -std::vector BitBufferCounter::getBuffer() const { - std::vector Buffer; - - for (size_t Idx = start(); Idx != (start() + size()); Idx++) - Buffer.push_back(V[Idx % V.size()]); - - return Buffer; -} - -void BitBufferCounter::insert(bool Bit) { - if (N >= V.size()) - NumSetBits -= (V[start()] == true); - - NumSetBits += (Bit == true); - V[N++ % V.size()] = Bit; -} - -void BitBufferCounter::print(std::ostream &OS) const { - std::vector Buffer = getBuffer(); - - for (bool B : Buffer) - OS << B; -} diff --git a/ml/BitBufferCounter.h b/ml/BitBufferCounter.h deleted file mode 100644 index db924d776..000000000 --- a/ml/BitBufferCounter.h +++ /dev/null @@ -1,54 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#ifndef BIT_BUFFER_COUNTER_H -#define BIT_BUFFER_COUNTER_H - -#include "ml-private.h" - -namespace ml { - -class BitBufferCounter { -public: - BitBufferCounter(size_t Capacity) : V(Capacity, 0), NumSetBits(0), N(0) {} - - std::vector getBuffer() const; - - void insert(bool Bit); - - void print(std::ostream &OS) const; - - bool isFilled() const { - return N >= V.size(); - } - - size_t numSetBits() const { - return NumSetBits; - } - -private: - inline size_t size() const { - return N < V.size() ? N : V.size(); - } - - inline size_t start() const { - if (N <= V.size()) - return 0; - - return N % V.size(); - } - -private: - std::vector V; - size_t NumSetBits; - - size_t N; -}; - -} // namespace ml - -inline std::ostream& operator<<(std::ostream &OS, const ml::BitBufferCounter &BBC) { - BBC.print(OS); - return OS; -} - -#endif /* BIT_BUFFER_COUNTER_H */ diff --git a/ml/BitRateWindow.cc b/ml/BitRateWindow.cc deleted file mode 100644 index c4c994c42..000000000 --- a/ml/BitRateWindow.cc +++ /dev/null @@ -1,75 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "BitRateWindow.h" - -using namespace ml; - -std::pair BitRateWindow::insert(bool Bit) { - Edge E; - - BBC.insert(Bit); - switch (CurrState) { - case State::NotFilled: { - if (BBC.isFilled()) { - if (BBC.numSetBits() < SetBitsThreshold) { - CurrState = State::BelowThreshold; - } else { - CurrState = State::AboveThreshold; - } - } else { - CurrState = State::NotFilled; - } - - E = {State::NotFilled, CurrState}; - break; - } case State::BelowThreshold: { - if (BBC.numSetBits() >= SetBitsThreshold) { - CurrState = State::AboveThreshold; - } - - E = {State::BelowThreshold, CurrState}; - break; - } case State::AboveThreshold: { - if ((BBC.numSetBits() < SetBitsThreshold) || - (CurrLength == MaxLength)) { - CurrState = State::Idle; - } - - E = {State::AboveThreshold, CurrState}; - break; - } case State::Idle: { - if (CurrLength == IdleLength) { - CurrState = State::NotFilled; - } - - E = {State::Idle, CurrState}; - break; - } - } - - Action A = EdgeActions[E]; - size_t L = (this->*A)(E.first, Bit); - return {E, L}; -} - -void BitRateWindow::print(std::ostream &OS) const { - switch (CurrState) { - case State::NotFilled: - OS << "NotFilled"; - break; - case State::BelowThreshold: - OS << "BelowThreshold"; - break; - case State::AboveThreshold: - OS << "AboveThreshold"; - break; - case State::Idle: - OS << "Idle"; - break; - default: - OS << "UnknownState"; - break; - } - - OS << ": " << BBC << " (Current Length: " << CurrLength << ")"; -} diff --git a/ml/BitRateWindow.h b/ml/BitRateWindow.h deleted file mode 100644 index 0d99008b8..000000000 --- a/ml/BitRateWindow.h +++ /dev/null @@ -1,170 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#ifndef BIT_RATE_WINDOW_H -#define BIT_RATE_WINDOW_H - -#include "BitBufferCounter.h" -#include "ml-private.h" - -namespace ml { - -class BitRateWindow { -public: - enum class State { - NotFilled, - BelowThreshold, - AboveThreshold, - Idle - }; - - using Edge = std::pair; - using Action = size_t (BitRateWindow::*)(State PrevState, bool NewBit); - -private: - std::map EdgeActions = { - // From == To - { - Edge(State::NotFilled, State::NotFilled), - &BitRateWindow::onRoundtripNotFilled, - }, - { - Edge(State::BelowThreshold, State::BelowThreshold), - &BitRateWindow::onRoundtripBelowThreshold, - }, - { - Edge(State::AboveThreshold, State::AboveThreshold), - &BitRateWindow::onRoundtripAboveThreshold, - }, - { - Edge(State::Idle, State::Idle), - &BitRateWindow::onRoundtripIdle, - }, - - - // NotFilled => {BelowThreshold, AboveThreshold} - { - Edge(State::NotFilled, State::BelowThreshold), - &BitRateWindow::onNotFilledToBelowThreshold - }, - { - Edge(State::NotFilled, State::AboveThreshold), - &BitRateWindow::onNotFilledToAboveThreshold - }, - - // BelowThreshold => AboveThreshold - { - Edge(State::BelowThreshold, State::AboveThreshold), - &BitRateWindow::onBelowToAboveThreshold - }, - - // AboveThreshold => Idle - { - Edge(State::AboveThreshold, State::Idle), - &BitRateWindow::onAboveThresholdToIdle - }, - - // Idle => NotFilled - { - Edge(State::Idle, State::NotFilled), - &BitRateWindow::onIdleToNotFilled - }, - }; - -public: - BitRateWindow(size_t MinLength, size_t MaxLength, size_t IdleLength, - size_t SetBitsThreshold) : - MinLength(MinLength), MaxLength(MaxLength), IdleLength(IdleLength), - SetBitsThreshold(SetBitsThreshold), - CurrState(State::NotFilled), CurrLength(0), BBC(MinLength) {} - - std::pair insert(bool Bit); - - void print(std::ostream &OS) const; - -private: - size_t onRoundtripNotFilled(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - CurrLength += 1; - return CurrLength; - } - - size_t onRoundtripBelowThreshold(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - CurrLength = MinLength; - return CurrLength; - } - - size_t onRoundtripAboveThreshold(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - CurrLength += 1; - return CurrLength; - } - - size_t onRoundtripIdle(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - CurrLength += 1; - return CurrLength; - } - - size_t onNotFilledToBelowThreshold(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - CurrLength = MinLength; - return CurrLength; - } - - size_t onNotFilledToAboveThreshold(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - CurrLength += 1; - return CurrLength; - } - - size_t onBelowToAboveThreshold(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - CurrLength = MinLength; - return CurrLength; - } - - size_t onAboveThresholdToIdle(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - size_t PrevLength = CurrLength; - CurrLength = 1; - return PrevLength; - } - - size_t onIdleToNotFilled(State PrevState, bool NewBit) { - (void) PrevState, (void) NewBit; - - BBC = BitBufferCounter(MinLength); - BBC.insert(NewBit); - - CurrLength = 1; - return CurrLength; - } - -private: - size_t MinLength; - size_t MaxLength; - size_t IdleLength; - size_t SetBitsThreshold; - - State CurrState; - size_t CurrLength; - BitBufferCounter BBC; -}; - -} // namespace ml - -inline std::ostream& operator<<(std::ostream &OS, const ml::BitRateWindow BRW) { - BRW.print(OS); - return OS; -} - -#endif /* BIT_RATE_WINDOW_H */ diff --git a/ml/Config.cc b/ml/Config.cc index 65b05a34d..eedd8c29f 100644 --- a/ml/Config.cc +++ b/ml/Config.cc @@ -31,8 +31,7 @@ void Config::readMLConfig(void) { unsigned MaxTrainSamples = config_get_number(ConfigSectionML, "maximum num samples to train", 4 * 3600); unsigned MinTrainSamples = config_get_number(ConfigSectionML, "minimum num samples to train", 1 * 900); unsigned TrainEvery = config_get_number(ConfigSectionML, "train every", 1 * 3600); - - unsigned DBEngineAnomalyRateEvery = config_get_number(ConfigSectionML, "dbengine anomaly rate every", 30); + unsigned NumModelsToUse = config_get_number(ConfigSectionML, "number of models per dimension", 1 * 24); unsigned DiffN = config_get_number(ConfigSectionML, "num samples to diff", 1); unsigned SmoothN = config_get_number(ConfigSectionML, "num samples to smooth", 3); @@ -42,27 +41,19 @@ void Config::readMLConfig(void) { unsigned MaxKMeansIters = config_get_number(ConfigSectionML, "maximum number of k-means iterations", 1000); double DimensionAnomalyScoreThreshold = config_get_float(ConfigSectionML, "dimension anomaly score threshold", 0.99); - double HostAnomalyRateThreshold = config_get_float(ConfigSectionML, "host anomaly rate threshold", 0.01); - - double ADMinWindowSize = config_get_float(ConfigSectionML, "minimum window size", 30); - double ADMaxWindowSize = config_get_float(ConfigSectionML, "maximum window size", 600); - double ADIdleWindowSize = config_get_float(ConfigSectionML, "idle window size", 30); - double ADWindowRateThreshold = config_get_float(ConfigSectionML, "window minimum anomaly rate", 0.25); - double ADDimensionRateThreshold = config_get_float(ConfigSectionML, "anomaly event min dimension rate threshold", 0.05); - std::stringstream SS; - SS << netdata_configured_cache_dir << "/anomaly-detection.db"; - Cfg.AnomalyDBPath = SS.str(); + double HostAnomalyRateThreshold = config_get_float(ConfigSectionML, "host anomaly rate threshold", 1.0); + std::string AnomalyDetectionGroupingMethod = config_get(ConfigSectionML, "anomaly detection grouping method", "average"); + time_t AnomalyDetectionQueryDuration = config_get_number(ConfigSectionML, "anomaly detection grouping duration", 5 * 60); /* * Clamp */ - MaxTrainSamples = clamp(MaxTrainSamples, 1 * 3600u, 24 * 3600u); - MinTrainSamples = clamp(MinTrainSamples, 1 * 900u, 6 * 3600u); - TrainEvery = clamp(TrainEvery, 1 * 3600u, 6 * 3600u); - - DBEngineAnomalyRateEvery = clamp(DBEngineAnomalyRateEvery, 1 * 30u, 15 * 60u); + MaxTrainSamples = clamp(MaxTrainSamples, 1 * 3600, 24 * 3600); + MinTrainSamples = clamp(MinTrainSamples, 1 * 900, 6 * 3600); + TrainEvery = clamp(TrainEvery, 1 * 3600, 6 * 3600); + NumModelsToUse = clamp(TrainEvery, 1, 7 * 24); DiffN = clamp(DiffN, 0u, 1u); SmoothN = clamp(SmoothN, 0u, 5u); @@ -72,13 +63,9 @@ void Config::readMLConfig(void) { MaxKMeansIters = clamp(MaxKMeansIters, 500u, 1000u); DimensionAnomalyScoreThreshold = clamp(DimensionAnomalyScoreThreshold, 0.01, 5.00); - HostAnomalyRateThreshold = clamp(HostAnomalyRateThreshold, 0.01, 1.0); - ADMinWindowSize = clamp(ADMinWindowSize, 30.0, 300.0); - ADMaxWindowSize = clamp(ADMaxWindowSize, 60.0, 900.0); - ADIdleWindowSize = clamp(ADIdleWindowSize, 30.0, 900.0); - ADWindowRateThreshold = clamp(ADWindowRateThreshold, 0.01, 0.99); - ADDimensionRateThreshold = clamp(ADDimensionRateThreshold, 0.01, 0.99); + HostAnomalyRateThreshold = clamp(HostAnomalyRateThreshold, 0.1, 10.0); + AnomalyDetectionQueryDuration = clamp(AnomalyDetectionQueryDuration, 60, 15 * 60); /* * Validate @@ -91,13 +78,6 @@ void Config::readMLConfig(void) { MaxTrainSamples = 4 * 3600; } - if (ADMinWindowSize >= ADMaxWindowSize) { - error("invalid min/max anomaly window size found (%lf >= %lf)", ADMinWindowSize, ADMaxWindowSize); - - ADMinWindowSize = 30.0; - ADMaxWindowSize = 600.0; - } - /* * Assign to config instance */ @@ -107,8 +87,7 @@ void Config::readMLConfig(void) { Cfg.MaxTrainSamples = MaxTrainSamples; Cfg.MinTrainSamples = MinTrainSamples; Cfg.TrainEvery = TrainEvery; - - Cfg.DBEngineAnomalyRateEvery = DBEngineAnomalyRateEvery; + Cfg.NumModelsToUse = NumModelsToUse; Cfg.DiffN = DiffN; Cfg.SmoothN = SmoothN; @@ -118,13 +97,10 @@ void Config::readMLConfig(void) { Cfg.MaxKMeansIters = MaxKMeansIters; Cfg.DimensionAnomalyScoreThreshold = DimensionAnomalyScoreThreshold; - Cfg.HostAnomalyRateThreshold = HostAnomalyRateThreshold; - Cfg.ADMinWindowSize = ADMinWindowSize; - Cfg.ADMaxWindowSize = ADMaxWindowSize; - Cfg.ADIdleWindowSize = ADIdleWindowSize; - Cfg.ADWindowRateThreshold = ADWindowRateThreshold; - Cfg.ADDimensionRateThreshold = ADDimensionRateThreshold; + Cfg.HostAnomalyRateThreshold = HostAnomalyRateThreshold; + Cfg.AnomalyDetectionGroupingMethod = web_client_api_request_v1_data_group(AnomalyDetectionGroupingMethod.c_str(), RRDR_GROUPING_AVERAGE); + Cfg.AnomalyDetectionQueryDuration = AnomalyDetectionQueryDuration; Cfg.HostsToSkip = config_get(ConfigSectionML, "hosts to skip from training", "!*"); Cfg.SP_HostsToSkip = simple_pattern_create(Cfg.HostsToSkip.c_str(), NULL, SIMPLE_PATTERN_EXACT); diff --git a/ml/Config.h b/ml/Config.h index 595fd072b..d876d4aa4 100644 --- a/ml/Config.h +++ b/ml/Config.h @@ -14,6 +14,7 @@ public: unsigned MaxTrainSamples; unsigned MinTrainSamples; unsigned TrainEvery; + unsigned NumModelsToUse; unsigned DBEngineAnomalyRateEvery; @@ -25,13 +26,10 @@ public: unsigned MaxKMeansIters; double DimensionAnomalyScoreThreshold; - double HostAnomalyRateThreshold; - double ADMinWindowSize; - double ADMaxWindowSize; - double ADIdleWindowSize; - double ADWindowRateThreshold; - double ADDimensionRateThreshold; + double HostAnomalyRateThreshold; + RRDR_GROUPING AnomalyDetectionGroupingMethod; + time_t AnomalyDetectionQueryDuration; bool StreamADCharts; @@ -41,7 +39,6 @@ public: std::string ChartsToSkip; SIMPLE_PATTERN *SP_ChartsToSkip; - std::string AnomalyDBPath; std::vector RandomNums; void readMLConfig(); diff --git a/ml/Database.cc b/ml/Database.cc deleted file mode 100644 index 06d0cdecb..000000000 --- a/ml/Database.cc +++ /dev/null @@ -1,127 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "Database.h" - -const char *ml::Database::SQL_CREATE_ANOMALIES_TABLE = - "CREATE TABLE IF NOT EXISTS anomaly_events( " - " anomaly_detector_name text NOT NULL, " - " anomaly_detector_version int NOT NULL, " - " host_id text NOT NULL, " - " after int NOT NULL, " - " before int NOT NULL, " - " anomaly_event_info text, " - " PRIMARY KEY( " - " anomaly_detector_name, anomaly_detector_version, " - " host_id, after, before " - " ) " - ");"; - -const char *ml::Database::SQL_INSERT_ANOMALY = - "INSERT INTO anomaly_events( " - " anomaly_detector_name, anomaly_detector_version, " - " host_id, after, before, anomaly_event_info) " - "VALUES (?1, ?2, ?3, ?4, ?5, ?6);"; - -const char *ml::Database::SQL_SELECT_ANOMALY = - "SELECT anomaly_event_info FROM anomaly_events WHERE" - " anomaly_detector_name == ?1 AND" - " anomaly_detector_version == ?2 AND" - " host_id == ?3 AND" - " after == ?4 AND" - " before == ?5;"; - -const char *ml::Database::SQL_SELECT_ANOMALY_EVENTS = - "SELECT after, before FROM anomaly_events WHERE" - " anomaly_detector_name == ?1 AND" - " anomaly_detector_version == ?2 AND" - " host_id == ?3 AND" - " after >= ?4 AND" - " before <= ?5;"; - -using namespace ml; - -bool Statement::prepare(sqlite3 *Conn) { - if (!Conn) - return false; - - if (ParsedStmt) - return true; - - int RC = sqlite3_prepare_v2(Conn, RawStmt, -1, &ParsedStmt, nullptr); - if (RC == SQLITE_OK) - return true; - - std::string Msg = "Statement \"%s\" preparation failed due to \"%s\""; - error(Msg.c_str(), RawStmt, sqlite3_errstr(RC)); - - return false; -} - -bool Statement::bindValue(size_t Pos, const std::string &Value) { - int RC = sqlite3_bind_text(ParsedStmt, Pos, Value.c_str(), -1, SQLITE_TRANSIENT); - if (RC == SQLITE_OK) - return true; - - error("Failed to bind text '%s' (pos = %zu) in statement '%s'.", Value.c_str(), Pos, RawStmt); - return false; -} - -bool Statement::bindValue(size_t Pos, const int Value) { - int RC = sqlite3_bind_int(ParsedStmt, Pos, Value); - if (RC == SQLITE_OK) - return true; - - error("Failed to bind integer %d (pos = %zu) in statement '%s'.", Value, Pos, RawStmt); - return false; -} - -bool Statement::resetAndClear(bool Ret) { - int RC = sqlite3_reset(ParsedStmt); - if (RC != SQLITE_OK) { - error("Could not reset statement: '%s'", RawStmt); - return false; - } - - RC = sqlite3_clear_bindings(ParsedStmt); - if (RC != SQLITE_OK) { - error("Could not clear bindings in statement: '%s'", RawStmt); - return false; - } - - return Ret; -} - -Database::Database(const std::string &Path) { - // Get sqlite3 connection handle. - int RC = sqlite3_open(Path.c_str(), &Conn); - if (RC != SQLITE_OK) { - std::string Msg = "Failed to initialize ML DB at %s, due to \"%s\""; - error(Msg.c_str(), Path.c_str(), sqlite3_errstr(RC)); - - sqlite3_close(Conn); - Conn = nullptr; - return; - } - - // Create anomaly events table if it does not exist. - char *ErrMsg; - RC = sqlite3_exec(Conn, SQL_CREATE_ANOMALIES_TABLE, nullptr, nullptr, &ErrMsg); - if (RC == SQLITE_OK) - return; - - error("SQLite error during database initialization, rc = %d (%s)", RC, ErrMsg); - error("SQLite failed statement: %s", SQL_CREATE_ANOMALIES_TABLE); - - sqlite3_free(ErrMsg); - sqlite3_close(Conn); - Conn = nullptr; -} - -Database::~Database() { - if (!Conn) - return; - - int RC = sqlite3_close(Conn); - if (RC != SQLITE_OK) - error("Could not close connection properly (rc=%d)", RC); -} diff --git a/ml/Database.h b/ml/Database.h deleted file mode 100644 index cc7b75872..000000000 --- a/ml/Database.h +++ /dev/null @@ -1,131 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#ifndef ML_DATABASE_H -#define ML_DATABASE_H - -#include "Dimension.h" -#include "ml-private.h" - -#include "json/single_include/nlohmann/json.hpp" - -namespace ml { - -class Statement { -public: - using RowCallback = std::function; - -public: - Statement(const char *RawStmt) : RawStmt(RawStmt), ParsedStmt(nullptr) {} - - template - bool exec(sqlite3 *Conn, RowCallback RowCb, ArgTypes ...Args) { - if (!prepare(Conn)) - return false; - - switch (bind(1, Args...)) { - case 0: - return false; - case sizeof...(Args): - break; - default: - return resetAndClear(false); - } - - while (true) { - switch (int RC = sqlite3_step(ParsedStmt)) { - case SQLITE_BUSY: case SQLITE_LOCKED: - usleep(SQLITE_INSERT_DELAY * USEC_PER_MS); - continue; - case SQLITE_ROW: - RowCb(ParsedStmt); - continue; - case SQLITE_DONE: - return resetAndClear(true); - default: - error("Stepping through '%s' returned rc=%d", RawStmt, RC); - return resetAndClear(false); - } - } - } - - ~Statement() { - if (!ParsedStmt) - return; - - int RC = sqlite3_finalize(ParsedStmt); - if (RC != SQLITE_OK) - error("Could not properly finalize statement (rc=%d)", RC); - } - -private: - bool prepare(sqlite3 *Conn); - - bool bindValue(size_t Pos, const int Value); - bool bindValue(size_t Pos, const std::string &Value); - - template - size_t bind(size_t Pos, ArgType T) { - return bindValue(Pos, T); - } - - template - size_t bind(size_t Pos, ArgType T, ArgTypes ...Args) { - return bindValue(Pos, T) + bind(Pos + 1, Args...); - } - - bool resetAndClear(bool Ret); - -private: - const char *RawStmt; - sqlite3_stmt *ParsedStmt; -}; - -class Database { -private: - static const char *SQL_CREATE_ANOMALIES_TABLE; - static const char *SQL_INSERT_ANOMALY; - static const char *SQL_SELECT_ANOMALY; - static const char *SQL_SELECT_ANOMALY_EVENTS; - -public: - Database(const std::string &Path); - - ~Database(); - - template - bool insertAnomaly(ArgTypes... Args) { - Statement::RowCallback RowCb = [](sqlite3_stmt *Stmt) { (void) Stmt; }; - return InsertAnomalyStmt.exec(Conn, RowCb, Args...); - } - - template - bool getAnomalyInfo(nlohmann::json &Json, ArgTypes&&... Args) { - Statement::RowCallback RowCb = [&](sqlite3_stmt *Stmt) { - const char *Text = static_cast(sqlite3_column_blob(Stmt, 0)); - Json = nlohmann::json::parse(Text); - }; - return GetAnomalyInfoStmt.exec(Conn, RowCb, Args...); - } - - template - bool getAnomaliesInRange(std::vector> &V, ArgTypes&&... Args) { - Statement::RowCallback RowCb = [&](sqlite3_stmt *Stmt) { - V.push_back({ - sqlite3_column_int64(Stmt, 0), - sqlite3_column_int64(Stmt, 1) - }); - }; - return GetAnomaliesInRangeStmt.exec(Conn, RowCb, Args...); - } - -private: - sqlite3 *Conn; - - Statement InsertAnomalyStmt{SQL_INSERT_ANOMALY}; - Statement GetAnomalyInfoStmt{SQL_SELECT_ANOMALY}; - Statement GetAnomaliesInRangeStmt{SQL_SELECT_ANOMALY_EVENTS}; -}; - -} - -#endif /* ML_DATABASE_H */ diff --git a/ml/Dimension.cc b/ml/Dimension.cc index 0fe07530c..bf34abb72 100644 --- a/ml/Dimension.cc +++ b/ml/Dimension.cc @@ -6,8 +6,13 @@ using namespace ml; -std::pair -TrainableDimension::getCalculatedNumbers() { +bool Dimension::isActive() const { + bool SetObsolete = rrdset_flag_check(RD->rrdset, RRDSET_FLAG_OBSOLETE); + bool DimObsolete = rrddim_flag_check(RD, RRDDIM_FLAG_OBSOLETE); + return !SetObsolete && !DimObsolete; +} + +std::pair Dimension::getCalculatedNumbers() { size_t MinN = Cfg.MinTrainSamples; size_t MaxN = Cfg.MaxTrainSamples; @@ -68,7 +73,7 @@ TrainableDimension::getCalculatedNumbers() { return { CNs, TotalValues }; } -MLResult TrainableDimension::trainModel() { +MLResult Dimension::trainModel() { auto P = getCalculatedNumbers(); CalculatedNumber *CNs = P.first; unsigned N = P.second; @@ -81,7 +86,15 @@ MLResult TrainableDimension::trainModel() { SamplesBuffer SB = SamplesBuffer(CNs, N, 1, Cfg.DiffN, Cfg.SmoothN, Cfg.LagN, SamplingRatio, Cfg.RandomNums); - KM.train(SB, Cfg.MaxKMeansIters); + std::vector Samples = SB.preprocess(); + + KMeans KM; + KM.train(Samples, Cfg.MaxKMeansIters); + + { + std::lock_guard Lock(Mutex); + Models[0] = KM; + } Trained = true; ConstantModel = true; @@ -90,16 +103,25 @@ MLResult TrainableDimension::trainModel() { return MLResult::Success; } -void PredictableDimension::addValue(CalculatedNumber Value, bool Exists) { +bool Dimension::shouldTrain(const TimePoint &TP) const { + if (ConstantModel) + return false; + + return (LastTrainedAt + Seconds(Cfg.TrainEvery * updateEvery())) < TP; +} + +bool Dimension::predict(CalculatedNumber Value, bool Exists) { if (!Exists) { CNs.clear(); - return; + AnomalyBit = false; + return false; } unsigned N = Cfg.DiffN + Cfg.SmoothN + Cfg.LagN; if (CNs.size() < N) { CNs.push_back(Value); - return; + AnomalyBit = false; + return false; } std::rotate(std::begin(CNs), std::begin(CNs) + 1, std::end(CNs)); @@ -108,28 +130,44 @@ void PredictableDimension::addValue(CalculatedNumber Value, bool Exists) { ConstantModel = false; CNs[N - 1] = Value; -} -std::pair PredictableDimension::predict() { - unsigned N = Cfg.DiffN + Cfg.SmoothN + Cfg.LagN; - if (CNs.size() != N) { + if (!isTrained() || ConstantModel) { AnomalyBit = false; - return { MLResult::MissingData, AnomalyBit }; + return false; } CalculatedNumber *TmpCNs = new CalculatedNumber[N * (Cfg.LagN + 1)](); std::memcpy(TmpCNs, CNs.data(), N * sizeof(CalculatedNumber)); - - SamplesBuffer SB = SamplesBuffer(TmpCNs, N, 1, Cfg.DiffN, Cfg.SmoothN, Cfg.LagN, + SamplesBuffer SB = SamplesBuffer(TmpCNs, N, 1, + Cfg.DiffN, Cfg.SmoothN, Cfg.LagN, 1.0, Cfg.RandomNums); - AnomalyScore = computeAnomalyScore(SB); + const DSample Sample = SB.preprocess().back(); delete[] TmpCNs; - if (AnomalyScore == std::numeric_limits::quiet_NaN()) { + std::unique_lock Lock(Mutex, std::defer_lock); + if (!Lock.try_lock()) { AnomalyBit = false; - return { MLResult::NaN, AnomalyBit }; + return false; } - AnomalyBit = AnomalyScore >= (100 * Cfg.DimensionAnomalyScoreThreshold); - return { MLResult::Success, AnomalyBit }; + for (const auto &KM : Models) { + double AnomalyScore = KM.anomalyScore(Sample); + if (AnomalyScore == std::numeric_limits::quiet_NaN()) { + AnomalyBit = false; + continue; + } + + if (AnomalyScore < (100 * Cfg.DimensionAnomalyScoreThreshold)) { + AnomalyBit = false; + return false; + } + } + + AnomalyBit = true; + return true; +} + +std::array Dimension::getModels() { + std::unique_lock Lock(Mutex); + return Models; } diff --git a/ml/Dimension.h b/ml/Dimension.h index 4fbc09b98..3ec56e098 100644 --- a/ml/Dimension.h +++ b/ml/Dimension.h @@ -3,157 +3,92 @@ #ifndef ML_DIMENSION_H #define ML_DIMENSION_H -#include "BitBufferCounter.h" +#include "Query.h" #include "Config.h" #include "ml-private.h" namespace ml { -class RrdDimension { -public: - RrdDimension(RRDDIM *RD) : RD(RD), Ops(&RD->tiers[0]->query_ops) { } - - RRDDIM *getRD() const { return RD; } - - time_t latestTime() { return Ops->latest_time(RD->tiers[0]->db_metric_handle); } - - time_t oldestTime() { return Ops->oldest_time(RD->tiers[0]->db_metric_handle); } - - unsigned updateEvery() const { return RD->update_every; } - - const std::string getID() const { - RRDSET *RS = RD->rrdset; - - std::stringstream SS; - SS << RS->context << "|" << RS->id << "|" << RD->name; - return SS.str(); - } - - bool isActive() const { - if (rrdset_flag_check(RD->rrdset, RRDSET_FLAG_OBSOLETE)) - return false; - - if (rrddim_flag_check(RD, RRDDIM_FLAG_OBSOLETE)) - return false; - - return true; - } - - void setAnomalyRateRD(RRDDIM *ARRD) { AnomalyRateRD = ARRD; } - RRDDIM *getAnomalyRateRD() const { return AnomalyRateRD; } - - void setAnomalyRateRDName(const char *Name) const { - rrddim_set_name(AnomalyRateRD->rrdset, AnomalyRateRD, Name); - } - - virtual ~RrdDimension() { - rrddim_free(AnomalyRateRD->rrdset, AnomalyRateRD); - } - -private: - RRDDIM *RD; - RRDDIM *AnomalyRateRD; - - struct rrddim_query_ops *Ops; - - std::string ID; -}; - enum class MLResult { Success = 0, MissingData, NaN, }; -class TrainableDimension : public RrdDimension { -public: - TrainableDimension(RRDDIM *RD) : - RrdDimension(RD), TrainEvery(Cfg.TrainEvery * updateEvery()) {} +static inline std::string getMLDimensionID(RRDDIM *RD) { + RRDSET *RS = RD->rrdset; - MLResult trainModel(); + std::stringstream SS; + SS << rrdset_context(RS) << "|" << rrdset_id(RS) << "|" << rrddim_name(RD); + return SS.str(); +} - CalculatedNumber computeAnomalyScore(SamplesBuffer &SB) { - return Trained ? KM.anomalyScore(SB) : 0.0; +class Dimension { +public: + Dimension(RRDDIM *RD) : + RD(RD), + LastTrainedAt(Seconds(0)), + Trained(false), + ConstantModel(false), + AnomalyScore(0.0), + AnomalyBit(0) + { } + + RRDDIM *getRD() const { + return RD; } - bool shouldTrain(const TimePoint &TP) const { - if (ConstantModel) - return false; - - return (LastTrainedAt + TrainEvery) < TP; + unsigned updateEvery() const { + return RD->update_every; } - bool isTrained() const { return Trained; } - -private: - std::pair getCalculatedNumbers(); - -public: - TimePoint LastTrainedAt{Seconds{0}}; + time_t latestTime() const { + return Query(RD).latestTime(); + } -protected: - std::atomic ConstantModel{false}; + time_t oldestTime() const { + return Query(RD).oldestTime(); + } -private: - Seconds TrainEvery; - KMeans KM; + bool isTrained() const { + return Trained; + } - std::atomic Trained{false}; -}; + bool isAnomalous() const { + return AnomalyBit; + } -class PredictableDimension : public TrainableDimension { -public: - PredictableDimension(RRDDIM *RD) : TrainableDimension(RD) {} + bool shouldTrain(const TimePoint &TP) const; - std::pair predict(); + bool isActive() const; - void addValue(CalculatedNumber Value, bool Exists); + MLResult trainModel(); - bool isAnomalous() { return AnomalyBit; } + bool predict(CalculatedNumber Value, bool Exists); - void updateAnomalyBitCounter(RRDSET *RS, unsigned Elapsed, bool IsAnomalous) { - AnomalyBitCounter += IsAnomalous; + std::pair detect(size_t WindowLength, bool Reset); - if (Elapsed == Cfg.DBEngineAnomalyRateEvery) { - double AR = static_cast(AnomalyBitCounter) / Cfg.DBEngineAnomalyRateEvery; - rrddim_set_by_pointer(RS, getAnomalyRateRD(), AR * 1000); - AnomalyBitCounter = 0; - } - } + std::array getModels(); private: - CalculatedNumber AnomalyScore{0.0}; - std::atomic AnomalyBit{false}; - unsigned AnomalyBitCounter{0}; - - std::vector CNs; -}; + std::pair getCalculatedNumbers(); -class DetectableDimension : public PredictableDimension { public: - DetectableDimension(RRDDIM *RD) : PredictableDimension(RD) {} - - std::pair detect(size_t WindowLength, bool Reset) { - bool AnomalyBit = isAnomalous(); - - if (Reset) - NumSetBits = BBC.numSetBits(); + RRDDIM *RD; - NumSetBits += AnomalyBit; - BBC.insert(AnomalyBit); + TimePoint LastTrainedAt; + std::atomic Trained; + std::atomic ConstantModel; - double AnomalyRate = static_cast(NumSetBits) / WindowLength; - return { AnomalyBit, AnomalyRate }; - } + CalculatedNumber AnomalyScore; + std::atomic AnomalyBit; -private: - BitBufferCounter BBC{static_cast(Cfg.ADMinWindowSize)}; - size_t NumSetBits{0}; + std::vector CNs; + std::array Models; + std::mutex Mutex; }; -using Dimension = DetectableDimension; - } // namespace ml #endif /* ML_DIMENSION_H */ diff --git a/ml/Host.cc b/ml/Host.cc index f8cba9d64..4a57178c7 100644 --- a/ml/Host.cc +++ b/ml/Host.cc @@ -1,278 +1,20 @@ // SPDX-License-Identifier: GPL-3.0-or-later -#include - #include "Config.h" #include "Host.h" +#include "ADCharts.h" #include "json/single_include/nlohmann/json.hpp" using namespace ml; -static void updateDimensionsChart(RRDHOST *RH, - collected_number NumTrainedDimensions, - collected_number NumNormalDimensions, - collected_number NumAnomalousDimensions) { - static thread_local RRDSET *RS = nullptr; - static thread_local RRDDIM *NumTotalDimensionsRD = nullptr; - static thread_local RRDDIM *NumTrainedDimensionsRD = nullptr; - static thread_local RRDDIM *NumNormalDimensionsRD = nullptr; - static thread_local RRDDIM *NumAnomalousDimensionsRD = nullptr; - - if (!RS) { - std::stringstream IdSS, NameSS; - - IdSS << "dimensions_on_" << localhost->machine_guid; - NameSS << "dimensions_on_" << localhost->hostname; - - RS = rrdset_create( - RH, - "anomaly_detection", // type - IdSS.str().c_str(), // id - NameSS.str().c_str(), // name - "dimensions", // family - "anomaly_detection.dimensions", // ctx - "Anomaly detection dimensions", // title - "dimensions", // units - "netdata", // plugin - "ml", // module - 39183, // priority - RH->rrd_update_every, // update_every - RRDSET_TYPE_LINE // chart_type - ); - rrdset_flag_set(RS, RRDSET_FLAG_ANOMALY_DETECTION); - - NumTotalDimensionsRD = rrddim_add(RS, "total", NULL, - 1, 1, RRD_ALGORITHM_ABSOLUTE); - NumTrainedDimensionsRD = rrddim_add(RS, "trained", NULL, - 1, 1, RRD_ALGORITHM_ABSOLUTE); - NumNormalDimensionsRD = rrddim_add(RS, "normal", NULL, - 1, 1, RRD_ALGORITHM_ABSOLUTE); - NumAnomalousDimensionsRD = rrddim_add(RS, "anomalous", NULL, - 1, 1, RRD_ALGORITHM_ABSOLUTE); - } else - rrdset_next(RS); - - rrddim_set_by_pointer(RS, NumTotalDimensionsRD, NumNormalDimensions + NumAnomalousDimensions); - rrddim_set_by_pointer(RS, NumTrainedDimensionsRD, NumTrainedDimensions); - rrddim_set_by_pointer(RS, NumNormalDimensionsRD, NumNormalDimensions); - rrddim_set_by_pointer(RS, NumAnomalousDimensionsRD, NumAnomalousDimensions); - - rrdset_done(RS); -} - -static void updateRateChart(RRDHOST *RH, collected_number AnomalyRate) { - static thread_local RRDSET *RS = nullptr; - static thread_local RRDDIM *AnomalyRateRD = nullptr; - - if (!RS) { - std::stringstream IdSS, NameSS; - - IdSS << "anomaly_rate_on_" << localhost->machine_guid; - NameSS << "anomaly_rate_on_" << localhost->hostname; - - RS = rrdset_create( - RH, - "anomaly_detection", // type - IdSS.str().c_str(), // id - NameSS.str().c_str(), // name - "anomaly_rate", // family - "anomaly_detection.anomaly_rate", // ctx - "Percentage of anomalous dimensions", // title - "percentage", // units - "netdata", // plugin - "ml", // module - 39184, // priority - RH->rrd_update_every, // update_every - RRDSET_TYPE_LINE // chart_type - ); - rrdset_flag_set(RS, RRDSET_FLAG_ANOMALY_DETECTION); - - AnomalyRateRD = rrddim_add(RS, "anomaly_rate", NULL, - 1, 100, RRD_ALGORITHM_ABSOLUTE); - } else - rrdset_next(RS); - - rrddim_set_by_pointer(RS, AnomalyRateRD, AnomalyRate); - - rrdset_done(RS); -} - -static void updateWindowLengthChart(RRDHOST *RH, collected_number WindowLength) { - static thread_local RRDSET *RS = nullptr; - static thread_local RRDDIM *WindowLengthRD = nullptr; - - if (!RS) { - std::stringstream IdSS, NameSS; - - IdSS << "detector_window_on_" << localhost->machine_guid; - NameSS << "detector_window_on_" << localhost->hostname; - - RS = rrdset_create( - RH, - "anomaly_detection", // type - IdSS.str().c_str(), // id - NameSS.str().c_str(), // name - "detector_window", // family - "anomaly_detection.detector_window", // ctx - "Anomaly detector window length", // title - "seconds", // units - "netdata", // plugin - "ml", // module - 39185, // priority - RH->rrd_update_every, // update_every - RRDSET_TYPE_LINE // chart_type - ); - rrdset_flag_set(RS, RRDSET_FLAG_ANOMALY_DETECTION); - - WindowLengthRD = rrddim_add(RS, "duration", NULL, - 1, 1, RRD_ALGORITHM_ABSOLUTE); - } else - rrdset_next(RS); - - rrddim_set_by_pointer(RS, WindowLengthRD, WindowLength * RH->rrd_update_every); - rrdset_done(RS); -} - -static void updateEventsChart(RRDHOST *RH, - std::pair P, - bool ResetBitCounter, - bool NewAnomalyEvent) { - static thread_local RRDSET *RS = nullptr; - static thread_local RRDDIM *AboveThresholdRD = nullptr; - static thread_local RRDDIM *ResetBitCounterRD = nullptr; - static thread_local RRDDIM *NewAnomalyEventRD = nullptr; - - if (!RS) { - std::stringstream IdSS, NameSS; - - IdSS << "detector_events_on_" << localhost->machine_guid; - NameSS << "detector_events_on_" << localhost->hostname; - - RS = rrdset_create( - RH, - "anomaly_detection", // type - IdSS.str().c_str(), // id - NameSS.str().c_str(), // name - "detector_events", // family - "anomaly_detection.detector_events", // ctx - "Anomaly events triggered", // title - "boolean", // units - "netdata", // plugin - "ml", // module - 39186, // priority - RH->rrd_update_every, // update_every - RRDSET_TYPE_LINE // chart_type - ); - rrdset_flag_set(RS, RRDSET_FLAG_ANOMALY_DETECTION); - - AboveThresholdRD = rrddim_add(RS, "above_threshold", NULL, - 1, 1, RRD_ALGORITHM_ABSOLUTE); - ResetBitCounterRD = rrddim_add(RS, "reset_bit_counter", NULL, - 1, 1, RRD_ALGORITHM_ABSOLUTE); - NewAnomalyEventRD = rrddim_add(RS, "new_anomaly_event", NULL, - 1, 1, RRD_ALGORITHM_ABSOLUTE); - } else - rrdset_next(RS); - - BitRateWindow::Edge E = P.first; - bool AboveThreshold = E.second == BitRateWindow::State::AboveThreshold; - - rrddim_set_by_pointer(RS, AboveThresholdRD, AboveThreshold); - rrddim_set_by_pointer(RS, ResetBitCounterRD, ResetBitCounter); - rrddim_set_by_pointer(RS, NewAnomalyEventRD, NewAnomalyEvent); - - rrdset_done(RS); -} - -static void updateDetectionChart(RRDHOST *RH) { - static thread_local RRDSET *RS = nullptr; - static thread_local RRDDIM *UserRD, *SystemRD = nullptr; - - if (!RS) { - std::stringstream IdSS, NameSS; - - IdSS << "prediction_stats_" << RH->machine_guid; - NameSS << "prediction_stats_for_" << RH->hostname; - - RS = rrdset_create_localhost( - "netdata", // type - IdSS.str().c_str(), // id - NameSS.str().c_str(), // name - "ml", // family - "netdata.prediction_stats", // ctx - "Prediction thread CPU usage", // title - "milliseconds/s", // units - "netdata", // plugin - "ml", // module - 136000, // priority - RH->rrd_update_every, // update_every - RRDSET_TYPE_STACKED // chart_type - ); - - UserRD = rrddim_add(RS, "user", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL); - SystemRD = rrddim_add(RS, "system", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL); - } else - rrdset_next(RS); - - struct rusage TRU; - getrusage(RUSAGE_THREAD, &TRU); - - rrddim_set_by_pointer(RS, UserRD, TRU.ru_utime.tv_sec * 1000000ULL + TRU.ru_utime.tv_usec); - rrddim_set_by_pointer(RS, SystemRD, TRU.ru_stime.tv_sec * 1000000ULL + TRU.ru_stime.tv_usec); - rrdset_done(RS); -} - -static void updateTrainingChart(RRDHOST *RH, struct rusage *TRU) -{ - static thread_local RRDSET *RS = nullptr; - static thread_local RRDDIM *UserRD = nullptr; - static thread_local RRDDIM *SystemRD = nullptr; - - if (!RS) { - std::stringstream IdSS, NameSS; - - IdSS << "training_stats_" << RH->machine_guid; - NameSS << "training_stats_for_" << RH->hostname; - - RS = rrdset_create_localhost( - "netdata", // type - IdSS.str().c_str(), // id - NameSS.str().c_str(), // name - "ml", // family - "netdata.training_stats", // ctx - "Training thread CPU usage", // title - "milliseconds/s", // units - "netdata", // plugin - "ml", // module - 136001, // priority - RH->rrd_update_every, // update_every - RRDSET_TYPE_STACKED // chart_type - ); - - UserRD = rrddim_add(RS, "user", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL); - SystemRD = rrddim_add(RS, "system", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL); - } else - rrdset_next(RS); - - rrddim_set_by_pointer(RS, UserRD, TRU->ru_utime.tv_sec * 1000000ULL + TRU->ru_utime.tv_usec); - rrddim_set_by_pointer(RS, SystemRD, TRU->ru_stime.tv_sec * 1000000ULL + TRU->ru_stime.tv_usec); - rrdset_done(RS); -} - void RrdHost::addDimension(Dimension *D) { - RRDDIM *AnomalyRateRD = rrddim_add(AnomalyRateRS, D->getID().c_str(), NULL, - 1, 1000, RRD_ALGORITHM_ABSOLUTE); - D->setAnomalyRateRD(AnomalyRateRD); - - { - std::lock_guard Lock(Mutex); + std::lock_guard Lock(Mutex); - DimensionsMap[D->getRD()] = D; + DimensionsMap[D->getRD()] = D; - // Default construct mutex for dimension - LocksMap[D]; - } + // Default construct mutex for dimension + LocksMap[D]; } void RrdHost::removeDimension(Dimension *D) { @@ -312,18 +54,33 @@ void RrdHost::getConfigAsJson(nlohmann::json &Json) const { Json["max-kmeans-iters"] = Cfg.MaxKMeansIters; Json["dimension-anomaly-score-threshold"] = Cfg.DimensionAnomalyScoreThreshold; - Json["host-anomaly-rate-threshold"] = Cfg.HostAnomalyRateThreshold; - Json["min-window-size"] = Cfg.ADMinWindowSize; - Json["max-window-size"] = Cfg.ADMaxWindowSize; - Json["idle-window-size"] = Cfg.ADIdleWindowSize; - Json["window-rate-threshold"] = Cfg.ADWindowRateThreshold; - Json["dimension-rate-threshold"] = Cfg.ADDimensionRateThreshold; + Json["host-anomaly-rate-threshold"] = Cfg.HostAnomalyRateThreshold; + Json["anomaly-detection-grouping-method"] = group_method2string(Cfg.AnomalyDetectionGroupingMethod); + Json["anomaly-detection-query-duration"] = Cfg.AnomalyDetectionQueryDuration; Json["hosts-to-skip"] = Cfg.HostsToSkip; Json["charts-to-skip"] = Cfg.ChartsToSkip; } +void TrainableHost::getModelsAsJson(nlohmann::json &Json) { + std::lock_guard Lock(Mutex); + + for (auto &DP : DimensionsMap) { + Dimension *D = DP.second; + + nlohmann::json JsonArray = nlohmann::json::array(); + for (const KMeans &KM : D->getModels()) { + nlohmann::json J; + KM.toJson(J); + JsonArray.push_back(J); + } + Json[getMLDimensionID(D->getRD())] = JsonArray; + } + + return; +} + std::pair> TrainableHost::findDimensionToTrain(const TimePoint &NowTP) { std::lock_guard Lock(Mutex); @@ -384,7 +141,12 @@ void TrainableHost::train() { worker_is_idle(); SleepFor = std::min(AllottedDuration - RealDuration, MaxSleepFor); - std::this_thread::sleep_for(SleepFor); + TimePoint Now = SteadyClock::now(); + auto Until = Now + SleepFor; + while (Now < Until && !netdata_exit) { + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + Now = SteadyClock::now(); + } worker_is_busy(0); } } @@ -393,78 +155,44 @@ void TrainableHost::train() { #define WORKER_JOB_UPDATE_DETECTION_CHART 1 #define WORKER_JOB_UPDATE_ANOMALY_RATES 2 #define WORKER_JOB_UPDATE_CHARTS 3 -#define WORKER_JOB_SAVE_ANOMALY_EVENT 4 #if WORKER_UTILIZATION_MAX_JOB_TYPES < 5 #error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 5 #endif void DetectableHost::detectOnce() { - auto P = BRW.insert(WindowAnomalyRate >= Cfg.HostAnomalyRateThreshold); - BitRateWindow::Edge Edge = P.first; - size_t WindowLength = P.second; - - bool ResetBitCounter = (Edge.first != BitRateWindow::State::AboveThreshold); - bool NewAnomalyEvent = (Edge.first == BitRateWindow::State::AboveThreshold) && - (Edge.second == BitRateWindow::State::Idle); - - std::vector> DimsOverThreshold; - size_t NumAnomalousDimensions = 0; size_t NumNormalDimensions = 0; size_t NumTrainedDimensions = 0; size_t NumActiveDimensions = 0; - bool CollectAnomalyRates = (++AnomalyRateTimer == Cfg.DBEngineAnomalyRateEvery); - if (CollectAnomalyRates) - rrdset_next(AnomalyRateRS); - { std::lock_guard Lock(Mutex); - DimsOverThreshold.reserve(DimensionsMap.size()); - for (auto &DP : DimensionsMap) { worker_is_busy(WORKER_JOB_DETECT_DIMENSION); Dimension *D = DP.second; - if (!D->isActive()) { - D->updateAnomalyBitCounter(AnomalyRateRS, AnomalyRateTimer, false); + if (!D->isActive()) continue; - } NumActiveDimensions++; - - auto P = D->detect(WindowLength, ResetBitCounter); - bool IsAnomalous = P.first; - double AnomalyScore = P.second; - NumTrainedDimensions += D->isTrained(); + bool IsAnomalous = D->isAnomalous(); if (IsAnomalous) NumAnomalousDimensions += 1; - - if (NewAnomalyEvent && (AnomalyScore >= Cfg.ADDimensionRateThreshold)) - DimsOverThreshold.push_back({ AnomalyScore, D->getID() }); - - D->updateAnomalyBitCounter(AnomalyRateRS, AnomalyRateTimer, IsAnomalous); } if (NumAnomalousDimensions) - WindowAnomalyRate = static_cast(NumAnomalousDimensions) / NumActiveDimensions; + HostAnomalyRate = static_cast(NumAnomalousDimensions) / NumActiveDimensions; else - WindowAnomalyRate = 0.0; + HostAnomalyRate = 0.0; NumNormalDimensions = NumActiveDimensions - NumAnomalousDimensions; } - if (CollectAnomalyRates) { - worker_is_busy(WORKER_JOB_UPDATE_ANOMALY_RATES); - AnomalyRateTimer = 0; - rrdset_done(AnomalyRateRS); - } - this->NumAnomalousDimensions = NumAnomalousDimensions; this->NumNormalDimensions = NumNormalDimensions; this->NumTrainedDimensions = NumTrainedDimensions; @@ -472,38 +200,11 @@ void DetectableHost::detectOnce() { worker_is_busy(WORKER_JOB_UPDATE_CHARTS); updateDimensionsChart(getRH(), NumTrainedDimensions, NumNormalDimensions, NumAnomalousDimensions); - updateRateChart(getRH(), WindowAnomalyRate * 10000.0); - updateWindowLengthChart(getRH(), WindowLength); - updateEventsChart(getRH(), P, ResetBitCounter, NewAnomalyEvent); + updateHostAndDetectionRateCharts(getRH(), HostAnomalyRate * 10000.0); struct rusage TRU; getResourceUsage(&TRU); updateTrainingChart(getRH(), &TRU); - - if (!NewAnomalyEvent || (DimsOverThreshold.size() == 0)) - return; - - worker_is_busy(WORKER_JOB_SAVE_ANOMALY_EVENT); - - std::sort(DimsOverThreshold.begin(), DimsOverThreshold.end()); - std::reverse(DimsOverThreshold.begin(), DimsOverThreshold.end()); - - // Make sure the JSON response won't grow beyond a specific number - // of dimensions. Log an error message if this happens, because it - // most likely means that the user specified a very-low anomaly rate - // threshold. - size_t NumMaxDimsOverThreshold = 2000; - if (DimsOverThreshold.size() > NumMaxDimsOverThreshold) { - error("Found %zu dimensions over threshold. Reducing JSON result to %zu dimensions.", - DimsOverThreshold.size(), NumMaxDimsOverThreshold); - DimsOverThreshold.resize(NumMaxDimsOverThreshold); - } - - nlohmann::json JsonResult = DimsOverThreshold; - - time_t Before = now_realtime_sec(); - time_t After = Before - (WindowLength * updateEvery()); - DB.insertAnomaly("AD1", 1, getUUID(), After, Before, JsonResult.dump(4)); } void DetectableHost::detect() { @@ -512,7 +213,6 @@ void DetectableHost::detect() { worker_register_job_name(WORKER_JOB_UPDATE_DETECTION_CHART, "detection chart"); worker_register_job_name(WORKER_JOB_UPDATE_ANOMALY_RATES, "anomaly rates"); worker_register_job_name(WORKER_JOB_UPDATE_CHARTS, "charts"); - worker_register_job_name(WORKER_JOB_SAVE_ANOMALY_EVENT, "anomaly event"); std::this_thread::sleep_for(Seconds{10}); diff --git a/ml/Host.h b/ml/Host.h index 2715008f0..52a0cd095 100644 --- a/ml/Host.h +++ b/ml/Host.h @@ -3,38 +3,17 @@ #ifndef ML_HOST_H #define ML_HOST_H -#include "BitRateWindow.h" #include "Config.h" -#include "Database.h" #include "Dimension.h" #include "ml-private.h" +#include "json/single_include/nlohmann/json.hpp" namespace ml { class RrdHost { public: - RrdHost(RRDHOST *RH) : RH(RH) { - AnomalyRateRS = rrdset_create( - RH, - "anomaly_detection", - "anomaly_rates", - NULL, // name - "anomaly_rates", - NULL, // ctx - "Average anomaly rate", - "anomaly rate", - "netdata", - "ml", - 39189, - Cfg.DBEngineAnomalyRateEvery, - RRDSET_TYPE_LINE - ); - - AnomalyRateRS->flags = static_cast( - static_cast(AnomalyRateRS->flags) | RRDSET_FLAG_HIDDEN - ); - } + RrdHost(RRDHOST *RH) : RH(RH) {}; RRDHOST *getRH() { return RH; } @@ -55,7 +34,6 @@ public: protected: RRDHOST *RH; - RRDSET *AnomalyRateRS; // Protect dimension and lock maps std::mutex Mutex; @@ -80,6 +58,8 @@ public: memcpy(RU, &ResourceUsage, sizeof(struct rusage)); } + void getModelsAsJson(nlohmann::json &Json); + private: std::pair> findDimensionToTrain(const TimePoint &NowTP); void trainDimension(Dimension *D, const TimePoint &NowTP); @@ -95,16 +75,6 @@ public: void startAnomalyDetectionThreads(); void stopAnomalyDetectionThreads(); - template - bool getAnomalyInfo(ArgTypes&&... Args) { - return DB.getAnomalyInfo(Args...); - } - - template - bool getAnomaliesInRange(ArgTypes&&... Args) { - return DB.getAnomaliesInRange(Args...); - } - void getDetectionInfoAsJson(nlohmann::json &Json) const; private: @@ -115,23 +85,12 @@ private: std::thread TrainingThread; std::thread DetectionThread; - BitRateWindow BRW{ - static_cast(Cfg.ADMinWindowSize), - static_cast(Cfg.ADMaxWindowSize), - static_cast(Cfg.ADIdleWindowSize), - static_cast(Cfg.ADMinWindowSize * Cfg.ADWindowRateThreshold) - }; - - CalculatedNumber WindowAnomalyRate{0.0}; + CalculatedNumber HostAnomalyRate{0.0}; size_t NumAnomalousDimensions{0}; size_t NumNormalDimensions{0}; size_t NumTrainedDimensions{0}; size_t NumActiveDimensions{0}; - - unsigned AnomalyRateTimer{0}; - - Database DB{Cfg.AnomalyDBPath}; }; using Host = DetectableHost; diff --git a/ml/KMeans.cc b/ml/KMeans.cc new file mode 100644 index 000000000..edc2ef49e --- /dev/null +++ b/ml/KMeans.cc @@ -0,0 +1,43 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#include "KMeans.h" +#include + +void KMeans::train(const std::vector &Samples, size_t MaxIterations) { + MinDist = std::numeric_limits::max(); + MaxDist = std::numeric_limits::min(); + + ClusterCenters.clear(); + + dlib::pick_initial_centers(NumClusters, ClusterCenters, Samples); + dlib::find_clusters_using_kmeans(Samples, ClusterCenters, MaxIterations); + + for (const auto &S : Samples) { + CalculatedNumber MeanDist = 0.0; + + for (const auto &KMCenter : ClusterCenters) + MeanDist += dlib::length(KMCenter - S); + + MeanDist /= NumClusters; + + if (MeanDist < MinDist) + MinDist = MeanDist; + + if (MeanDist > MaxDist) + MaxDist = MeanDist; + } +} + +CalculatedNumber KMeans::anomalyScore(const DSample &Sample) const { + CalculatedNumber MeanDist = 0.0; + for (const auto &CC: ClusterCenters) + MeanDist += dlib::length(CC - Sample); + + MeanDist /= NumClusters; + + if (MaxDist == MinDist) + return 0.0; + + CalculatedNumber AnomalyScore = 100.0 * std::abs((MeanDist - MinDist) / (MaxDist - MinDist)); + return (AnomalyScore > 100.0) ? 100.0 : AnomalyScore; +} diff --git a/ml/KMeans.h b/ml/KMeans.h new file mode 100644 index 000000000..0398eeb86 --- /dev/null +++ b/ml/KMeans.h @@ -0,0 +1,41 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#ifndef KMEANS_H +#define KMEANS_H + +#include +#include +#include +#include + +#include "SamplesBuffer.h" +#include "json/single_include/nlohmann/json.hpp" + +class KMeans { +public: + KMeans(size_t NumClusters = 2) : NumClusters(NumClusters) { + MinDist = std::numeric_limits::max(); + MaxDist = std::numeric_limits::min(); + }; + + void train(const std::vector &Samples, size_t MaxIterations); + CalculatedNumber anomalyScore(const DSample &Sample) const; + + void toJson(nlohmann::json &J) const { + J = nlohmann::json{ + {"CCs", ClusterCenters}, + {"MinDist", MinDist}, + {"MaxDist", MaxDist} + }; + } + +private: + size_t NumClusters; + + std::vector ClusterCenters; + + CalculatedNumber MinDist; + CalculatedNumber MaxDist; +}; + +#endif /* KMEANS_H */ diff --git a/ml/Makefile.am b/ml/Makefile.am deleted file mode 100644 index 27449d659..000000000 --- a/ml/Makefile.am +++ /dev/null @@ -1,8 +0,0 @@ -# SPDX-License-Identifier: GPL-3.0-or-later - -AUTOMAKE_OPTIONS = subdir-objects -MAINTAINERCLEANFILES = $(srcdir)/Makefile.in - -SUBDIRS = \ - kmeans \ - $(NULL) diff --git a/ml/Query.h b/ml/Query.h index 24c5fa384..78d117003 100644 --- a/ml/Query.h +++ b/ml/Query.h @@ -7,8 +7,8 @@ namespace ml { class Query { public: - Query(RRDDIM *RD) : RD(RD) { - Ops = &RD->tiers[0]->query_ops; + Query(RRDDIM *RD) : RD(RD), Initialized(false) { + Ops = RD->tiers[0]->query_ops; } time_t latestTime() { @@ -20,27 +20,36 @@ public: } void init(time_t AfterT, time_t BeforeT) { - Ops->init(RD->tiers[0]->db_metric_handle, &Handle, AfterT, BeforeT, TIER_QUERY_FETCH_SUM); + Ops->init(RD->tiers[0]->db_metric_handle, &Handle, AfterT, BeforeT); + Initialized = true; + points_read = 0; } bool isFinished() { return Ops->is_finished(&Handle); } + ~Query() { + if (Initialized) { + Ops->finalize(&Handle); + global_statistics_ml_query_completed(points_read); + points_read = 0; + } + } + std::pair nextMetric() { + points_read++; STORAGE_POINT sp = Ops->next_metric(&Handle); return { sp.start_time, sp.sum / sp.count }; } - ~Query() { - Ops->finalize(&Handle); - } - private: RRDDIM *RD; + bool Initialized; + size_t points_read; - struct rrddim_query_ops *Ops; - struct rrddim_query_handle Handle; + struct storage_engine_query_ops *Ops; + struct storage_engine_query_handle Handle; }; } // namespace ml diff --git a/ml/README.md b/ml/README.md index 2578993e2..f6fd923ab 100644 --- a/ml/README.md +++ b/ml/README.md @@ -248,8 +248,6 @@ In terms of anomaly detection, the most interesting charts would be the `anomaly - `anomaly_detection.dimensions`: Percentage of anomalous dimensions. - `anomaly_detection.detector_window`: The length of the active window used by the detector. - `anomaly_detection.detector_events`: Flags (0 or 1) to show when an anomaly event has been triggered by the detector. -- `anomaly_detection.prediction_stats`: Diagnostic metrics relating to prediction time of anomaly detection. -- `anomaly_detection.training_stats`: Diagnostic metrics relating to training time of anomaly detection. Below is an example of how these charts may look in the presence of an anomaly event. diff --git a/ml/SamplesBuffer.cc b/ml/SamplesBuffer.cc new file mode 100644 index 000000000..d276c6e09 --- /dev/null +++ b/ml/SamplesBuffer.cc @@ -0,0 +1,150 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// +#include "SamplesBuffer.h" + +#include +#include +#include + +void Sample::print(std::ostream &OS) const { + for (size_t Idx = 0; Idx != NumDims - 1; Idx++) + OS << CNs[Idx] << ", "; + + OS << CNs[NumDims - 1]; +} + +void SamplesBuffer::print(std::ostream &OS) const { + for (size_t Idx = Preprocessed ? (DiffN + (SmoothN - 1) + (LagN)) : 0; + Idx != NumSamples; Idx++) { + Sample S = Preprocessed ? getPreprocessedSample(Idx) : getSample(Idx); + OS << S << std::endl; + } +} + +std::vector SamplesBuffer::getPreprocessedSamples() const { + std::vector V; + + for (size_t Idx = Preprocessed ? (DiffN + (SmoothN - 1) + (LagN)) : 0; + Idx != NumSamples; Idx++) { + Sample S = Preprocessed ? getPreprocessedSample(Idx) : getSample(Idx); + V.push_back(S); + } + + return V; +} + +void SamplesBuffer::diffSamples() { + // Panda's DataFrame default behaviour is to subtract each element from + // itself. For us `DiffN = 0` means "disable diff-ing" when preprocessing + // the samples buffer. This deviation will make it easier for us to test + // the KMeans implementation. + if (DiffN == 0) + return; + + for (size_t Idx = 0; Idx != (NumSamples - DiffN); Idx++) { + size_t High = (NumSamples - 1) - Idx; + size_t Low = High - DiffN; + + Sample LHS = getSample(High); + Sample RHS = getSample(Low); + + LHS.diff(RHS); + } +} + +void SamplesBuffer::smoothSamples() { + // Holds the mean value of each window + CalculatedNumber *AccCNs = new CalculatedNumber[NumDimsPerSample](); + Sample Acc(AccCNs, NumDimsPerSample); + + // Used to avoid clobbering the accumulator when moving the window + CalculatedNumber *TmpCNs = new CalculatedNumber[NumDimsPerSample](); + Sample Tmp(TmpCNs, NumDimsPerSample); + + CalculatedNumber Factor = (CalculatedNumber) 1 / SmoothN; + + // Calculate the value of the 1st window + for (size_t Idx = 0; Idx != std::min(SmoothN, NumSamples); Idx++) { + Tmp.add(getSample(NumSamples - (Idx + 1))); + } + + Acc.add(Tmp); + Acc.scale(Factor); + + // Move the window and update the samples + for (size_t Idx = NumSamples; Idx != (DiffN + SmoothN - 1); Idx--) { + Sample S = getSample(Idx - 1); + + // Tmp <- Next window (if any) + if (Idx >= (SmoothN + 1)) { + Tmp.diff(S); + Tmp.add(getSample(Idx - (SmoothN + 1))); + } + + // S <- Acc + S.copy(Acc); + + // Acc <- Tmp + Acc.copy(Tmp); + Acc.scale(Factor); + } + + delete[] AccCNs; + delete[] TmpCNs; +} + +void SamplesBuffer::lagSamples() { + if (LagN == 0) + return; + + for (size_t Idx = NumSamples; Idx != LagN; Idx--) { + Sample PS = getPreprocessedSample(Idx - 1); + PS.lag(getSample(Idx - 1), LagN); + } +} + +std::vector SamplesBuffer::preprocess() { + assert(Preprocessed == false); + + std::vector DSamples; + size_t OutN = NumSamples; + + // Diff + if (DiffN >= OutN) + return DSamples; + OutN -= DiffN; + diffSamples(); + + // Smooth + if (SmoothN == 0 || SmoothN > OutN) + return DSamples; + OutN -= (SmoothN - 1); + smoothSamples(); + + // Lag + if (LagN >= OutN) + return DSamples; + OutN -= LagN; + lagSamples(); + + DSamples.reserve(OutN); + Preprocessed = true; + + uint32_t MaxMT = std::numeric_limits::max(); + uint32_t CutOff = static_cast(MaxMT) * SamplingRatio; + + for (size_t Idx = NumSamples - OutN; Idx != NumSamples; Idx++) { + if (RandNums[Idx] > CutOff) + continue; + + DSample DS; + DS.set_size(NumDimsPerSample * (LagN + 1)); + + const Sample PS = getPreprocessedSample(Idx); + PS.initDSample(DS); + + DSamples.push_back(DS); + } + + return DSamples; +} diff --git a/ml/SamplesBuffer.h b/ml/SamplesBuffer.h new file mode 100644 index 000000000..1c7215cca --- /dev/null +++ b/ml/SamplesBuffer.h @@ -0,0 +1,146 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#ifndef SAMPLES_BUFFER_H +#define SAMPLES_BUFFER_H + +#include +#include + +#include +#include +#include + +#include + +typedef double CalculatedNumber; +typedef dlib::matrix DSample; + +class Sample { +public: + Sample(CalculatedNumber *Buf, size_t N) : CNs(Buf), NumDims(N) {} + + void initDSample(DSample &DS) const { + for (size_t Idx = 0; Idx != NumDims; Idx++) { + DS(Idx) = std::abs(CNs[Idx]); + } + } + + void add(const Sample &RHS) const { + assert(NumDims == RHS.NumDims); + + for (size_t Idx = 0; Idx != NumDims; Idx++) + CNs[Idx] += RHS.CNs[Idx]; + }; + + void diff(const Sample &RHS) const { + assert(NumDims == RHS.NumDims); + + for (size_t Idx = 0; Idx != NumDims; Idx++) + CNs[Idx] -= RHS.CNs[Idx]; + }; + + void copy(const Sample &RHS) const { + assert(NumDims == RHS.NumDims); + + std::memcpy(CNs, RHS.CNs, NumDims * sizeof(CalculatedNumber)); + } + + void scale(CalculatedNumber Factor) { + for (size_t Idx = 0; Idx != NumDims; Idx++) + CNs[Idx] *= Factor; + } + + void lag(const Sample &S, size_t LagN) { + size_t N = S.NumDims; + + for (size_t Idx = 0; Idx != (LagN + 1); Idx++) { + Sample Src(S.CNs - (Idx * N), N); + Sample Dst(CNs + (Idx * N), N); + Dst.copy(Src); + } + } + + const CalculatedNumber *getCalculatedNumbers() const { + return CNs; + }; + + void print(std::ostream &OS) const; + +private: + CalculatedNumber *CNs; + size_t NumDims; +}; + +inline std::ostream& operator<<(std::ostream &OS, const Sample &S) { + S.print(OS); + return OS; +} + +class SamplesBuffer { +public: + SamplesBuffer(CalculatedNumber *CNs, + size_t NumSamples, size_t NumDimsPerSample, + size_t DiffN, size_t SmoothN, size_t LagN, + double SamplingRatio, std::vector &RandNums) : + CNs(CNs), NumSamples(NumSamples), NumDimsPerSample(NumDimsPerSample), + DiffN(DiffN), SmoothN(SmoothN), LagN(LagN), + SamplingRatio(SamplingRatio), RandNums(RandNums), + BytesPerSample(NumDimsPerSample * sizeof(CalculatedNumber)), + Preprocessed(false) {}; + + std::vector preprocess(); + std::vector getPreprocessedSamples() const; + + size_t capacity() const { return NumSamples; } + void print(std::ostream &OS) const; + +private: + size_t getSampleOffset(size_t Index) const { + assert(Index < NumSamples); + return Index * NumDimsPerSample; + } + + size_t getPreprocessedSampleOffset(size_t Index) const { + assert(Index < NumSamples); + return getSampleOffset(Index) * (LagN + 1); + } + + void setSample(size_t Index, const Sample &S) const { + size_t Offset = getSampleOffset(Index); + std::memcpy(&CNs[Offset], S.getCalculatedNumbers(), BytesPerSample); + } + + const Sample getSample(size_t Index) const { + size_t Offset = getSampleOffset(Index); + return Sample(&CNs[Offset], NumDimsPerSample); + }; + + const Sample getPreprocessedSample(size_t Index) const { + size_t Offset = getPreprocessedSampleOffset(Index); + return Sample(&CNs[Offset], NumDimsPerSample * (LagN + 1)); + }; + + void diffSamples(); + void smoothSamples(); + void lagSamples(); + +private: + CalculatedNumber *CNs; + size_t NumSamples; + size_t NumDimsPerSample; + size_t DiffN; + size_t SmoothN; + size_t LagN; + double SamplingRatio; + std::vector &RandNums; + + size_t BytesPerSample; + bool Preprocessed; +}; + +inline std::ostream& operator<<(std::ostream& OS, const SamplesBuffer &SB) { + SB.print(OS); + return OS; +} + +#endif /* SAMPLES_BUFFER_H */ diff --git a/ml/SamplesBufferTests.cc b/ml/SamplesBufferTests.cc new file mode 100644 index 000000000..5997a2a15 --- /dev/null +++ b/ml/SamplesBufferTests.cc @@ -0,0 +1,146 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#include "ml/ml-private.h" +#include + +/* + * The SamplesBuffer class implements the functionality of the following python + * code: + * >> df = pd.DataFrame(data=samples) + * >> df = df.diff(diff_n).dropna() + * >> df = df.rolling(smooth_n).mean().dropna() + * >> df = pd.concat([df.shift(n) for n in range(lag_n + 1)], axis=1).dropna() + * + * Its correctness has been verified by automatically generating random + * data frames in Python and comparing them with the correspondent preprocessed + * SampleBuffers. + * + * The following tests are meant to catch unintended changes in the SamplesBuffer + * implementation. For development purposes, one should compare changes against + * the aforementioned python code. +*/ + +TEST(SamplesBufferTest, NS_8_NDPS_1_DN_1_SN_3_LN_1) { + size_t NumSamples = 8, NumDimsPerSample = 1; + size_t DiffN = 1, SmoothN = 3, LagN = 3; + + size_t N = NumSamples * NumDimsPerSample * (LagN + 1); + CalculatedNumber *CNs = new CalculatedNumber[N](); + + CNs[0] = 0.7568336679490107; + CNs[1] = 0.4814406581763254; + CNs[2] = 0.40073555156221874; + CNs[3] = 0.5973257298194408; + CNs[4] = 0.5334727814345868; + CNs[5] = 0.2632477193454843; + CNs[6] = 0.2684839023122384; + CNs[7] = 0.851332948637479; + + std::vector RandNums(NumSamples, std::numeric_limits::max()); + SamplesBuffer SB(CNs, NumSamples, NumDimsPerSample, DiffN, SmoothN, LagN, 1.0, RandNums); + SB.preprocess(); + + std::vector Samples = SB.getPreprocessedSamples(); + EXPECT_EQ(Samples.size(), 2); + + Sample S0 = Samples[0]; + const CalculatedNumber *S0_CNs = S0.getCalculatedNumbers(); + Sample S1 = Samples[1]; + const CalculatedNumber *S1_CNs = S1.getCalculatedNumbers(); + + EXPECT_NEAR(S0_CNs[0], -0.109614, 0.001); + EXPECT_NEAR(S0_CNs[1], -0.0458293, 0.001); + EXPECT_NEAR(S0_CNs[2], 0.017344, 0.001); + EXPECT_NEAR(S0_CNs[3], -0.0531693, 0.001); + + EXPECT_NEAR(S1_CNs[0], 0.105953, 0.001); + EXPECT_NEAR(S1_CNs[1], -0.109614, 0.001); + EXPECT_NEAR(S1_CNs[2], -0.0458293, 0.001); + EXPECT_NEAR(S1_CNs[3], 0.017344, 0.001); + + delete[] CNs; +} + +TEST(SamplesBufferTest, NS_8_NDPS_1_DN_2_SN_3_LN_2) { + size_t NumSamples = 8, NumDimsPerSample = 1; + size_t DiffN = 2, SmoothN = 3, LagN = 2; + + size_t N = NumSamples * NumDimsPerSample * (LagN + 1); + CalculatedNumber *CNs = new CalculatedNumber[N](); + + CNs[0] = 0.20511885291342846; + CNs[1] = 0.13151717360306558; + CNs[2] = 0.6017085062423134; + CNs[3] = 0.46256882933941545; + CNs[4] = 0.7887758447877941; + CNs[5] = 0.9237989080034406; + CNs[6] = 0.15552559051428083; + CNs[7] = 0.6309750314597955; + + std::vector RandNums(NumSamples, std::numeric_limits::max()); + SamplesBuffer SB(CNs, NumSamples, NumDimsPerSample, DiffN, SmoothN, LagN, 1.0, RandNums); + SB.preprocess(); + + std::vector Samples = SB.getPreprocessedSamples(); + EXPECT_EQ(Samples.size(), 2); + + Sample S0 = Samples[0]; + const CalculatedNumber *S0_CNs = S0.getCalculatedNumbers(); + Sample S1 = Samples[1]; + const CalculatedNumber *S1_CNs = S1.getCalculatedNumbers(); + + EXPECT_NEAR(S0_CNs[0], 0.005016, 0.001); + EXPECT_NEAR(S0_CNs[1], 0.326450, 0.001); + EXPECT_NEAR(S0_CNs[2], 0.304903, 0.001); + + EXPECT_NEAR(S1_CNs[0], -0.154948, 0.001); + EXPECT_NEAR(S1_CNs[1], 0.005016, 0.001); + EXPECT_NEAR(S1_CNs[2], 0.326450, 0.001); + + delete[] CNs; +} + +TEST(SamplesBufferTest, NS_8_NDPS_3_DN_2_SN_4_LN_1) { + size_t NumSamples = 8, NumDimsPerSample = 3; + size_t DiffN = 2, SmoothN = 4, LagN = 1; + + size_t N = NumSamples * NumDimsPerSample * (LagN + 1); + CalculatedNumber *CNs = new CalculatedNumber[N](); + + CNs[0] = 0.34310900399667765; CNs[1] = 0.14694315994488194; CNs[2] = 0.8246677800938796; + CNs[3] = 0.48249504592307835; CNs[4] = 0.23241087965531182; CNs[5] = 0.9595348555892567; + CNs[6] = 0.44281094035598334; CNs[7] = 0.5143142171362715; CNs[8] = 0.06391303014242555; + CNs[9] = 0.7460491027783901; CNs[10] = 0.43887217459032923; CNs[11] = 0.2814395025355999; + CNs[12] = 0.9231114281214198; CNs[13] = 0.326882401786898; CNs[14] = 0.26747939220376216; + CNs[15] = 0.7787571209969636; CNs[16] =0.5851700001235088; CNs[17] = 0.34410728945321567; + CNs[18] = 0.9394494507088997; CNs[19] =0.17567223681734334; CNs[20] = 0.42732886195446984; + CNs[21] = 0.9460522396152958; CNs[22] =0.23462747016780894; CNs[23] = 0.35983249900892145; + + std::vector RandNums(NumSamples, std::numeric_limits::max()); + SamplesBuffer SB(CNs, NumSamples, NumDimsPerSample, DiffN, SmoothN, LagN, 1.0, RandNums); + SB.preprocess(); + + std::vector Samples = SB.getPreprocessedSamples(); + EXPECT_EQ(Samples.size(), 2); + + Sample S0 = Samples[0]; + const CalculatedNumber *S0_CNs = S0.getCalculatedNumbers(); + Sample S1 = Samples[1]; + const CalculatedNumber *S1_CNs = S1.getCalculatedNumbers(); + + EXPECT_NEAR(S0_CNs[0], 0.198225, 0.001); + EXPECT_NEAR(S0_CNs[1], 0.003529, 0.001); + EXPECT_NEAR(S0_CNs[2], -0.063003, 0.001); + EXPECT_NEAR(S0_CNs[3], 0.219066, 0.001); + EXPECT_NEAR(S0_CNs[4], 0.133175, 0.001); + EXPECT_NEAR(S0_CNs[5], -0.293154, 0.001); + + EXPECT_NEAR(S1_CNs[0], 0.174160, 0.001); + EXPECT_NEAR(S1_CNs[1], -0.135722, 0.001); + EXPECT_NEAR(S1_CNs[2], 0.110452, 0.001); + EXPECT_NEAR(S1_CNs[3], 0.198225, 0.001); + EXPECT_NEAR(S1_CNs[4], 0.003529, 0.001); + EXPECT_NEAR(S1_CNs[5], -0.063003, 0.001); + + delete[] CNs; +} diff --git a/ml/Tests.cc b/ml/Tests.cc deleted file mode 100644 index 7d369d48d..000000000 --- a/ml/Tests.cc +++ /dev/null @@ -1,301 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "BitBufferCounter.h" -#include "BitRateWindow.h" - -#include "gtest/gtest.h" - -using namespace ml; - -TEST(BitBufferCounterTest, Cap_4) { - size_t Capacity = 4; - BitBufferCounter BBC(Capacity); - - // No bits set - EXPECT_EQ(BBC.numSetBits(), 0); - - // All ones - for (size_t Idx = 0; Idx != (2 * Capacity); Idx++) { - BBC.insert(true); - - EXPECT_EQ(BBC.numSetBits(), std::min(Idx + 1, Capacity)); - } - - // All zeroes - for (size_t Idx = 0; Idx != Capacity; Idx++) { - BBC.insert(false); - - if (Idx < Capacity) - EXPECT_EQ(BBC.numSetBits(), Capacity - (Idx + 1)); - else - EXPECT_EQ(BBC.numSetBits(), 0); - } - - // Even ones/zeroes - for (size_t Idx = 0; Idx != (2 * Capacity); Idx++) - BBC.insert(Idx % 2 == 0); - EXPECT_EQ(BBC.numSetBits(), Capacity / 2); -} - -using State = BitRateWindow::State; -using Edge = BitRateWindow::Edge; -using Result = std::pair; - -TEST(BitRateWindowTest, Cycles) { - /* Test the FSM by going through its two cycles: - * 1) NotFilled -> AboveThreshold -> Idle -> NotFilled - * 2) NotFilled -> BelowThreshold -> AboveThreshold -> Idle -> NotFilled - * - * Check the window's length on every new state transition. - */ - - size_t MinLength = 4, MaxLength = 6, IdleLength = 5; - size_t SetBitsThreshold = 3; - - Result R; - BitRateWindow BRW(MinLength, MaxLength, IdleLength, SetBitsThreshold); - - /* - * 1st cycle - */ - - // NotFilled -> AboveThreshold - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::AboveThreshold)); - EXPECT_EQ(R.second, MinLength); - - // AboveThreshold -> Idle - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::AboveThreshold)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::AboveThreshold)); - - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::Idle)); - EXPECT_EQ(R.second, MaxLength); - - - // Idle -> NotFilled - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::NotFilled)); - EXPECT_EQ(R.second, 1); - - // NotFilled -> AboveThreshold - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::AboveThreshold)); - EXPECT_EQ(R.second, MinLength); - - /* - * 2nd cycle - */ - - BRW = BitRateWindow(MinLength, MaxLength, IdleLength, SetBitsThreshold); - - // NotFilled -> BelowThreshold - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::BelowThreshold)); - EXPECT_EQ(R.second, MinLength); - - // BelowThreshold -> BelowThreshold: - // Check the state's self loop by adding set bits that will keep the - // bit buffer below the specified threshold. - // - for (size_t Idx = 0; Idx != 2 * MaxLength; Idx++) { - R = BRW.insert(Idx % 2 == 0); - EXPECT_EQ(R.first, std::make_pair(State::BelowThreshold, State::BelowThreshold)); - EXPECT_EQ(R.second, MinLength); - } - - // Verify that at the end of the loop the internal bit buffer contains - // "1010". Do so by adding one set bit and checking that we remain below - // the specified threshold. - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::BelowThreshold, State::BelowThreshold)); - EXPECT_EQ(R.second, MinLength); - - // BelowThreshold -> AboveThreshold - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::BelowThreshold, State::AboveThreshold)); - EXPECT_EQ(R.second, MinLength); - - // AboveThreshold -> Idle: - // Do the transition without filling the max window size this time. - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::Idle)); - EXPECT_EQ(R.second, MinLength); - - // Idle -> NotFilled - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::NotFilled)); - EXPECT_EQ(R.second, 1); - - // NotFilled -> AboveThreshold - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::NotFilled)); - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::AboveThreshold)); - EXPECT_EQ(R.second, MinLength); -} - -TEST(BitRateWindowTest, ConsecutiveOnes) { - size_t MinLength = 120, MaxLength = 240, IdleLength = 30; - size_t SetBitsThreshold = 30; - - Result R; - BitRateWindow BRW(MinLength, MaxLength, IdleLength, SetBitsThreshold); - - for (size_t Idx = 0; Idx != MaxLength; Idx++) - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::BelowThreshold, State::BelowThreshold)); - EXPECT_EQ(R.second, MinLength); - - for (size_t Idx = 0; Idx != SetBitsThreshold; Idx++) { - EXPECT_EQ(R.first, std::make_pair(State::BelowThreshold, State::BelowThreshold)); - R = BRW.insert(true); - } - EXPECT_EQ(R.first, std::make_pair(State::BelowThreshold, State::AboveThreshold)); - EXPECT_EQ(R.second, MinLength); - - // At this point the window's buffer contains: - // (MinLength - SetBitsThreshold = 90) 0s, followed by - // (SetBitsThreshold = 30) 1s. - // - // To go below the threshold, we need to add (90 + 1) more 0s in the window's - // buffer. At that point, the the window's buffer will contain: - // (SetBitsThreshold = 29) 1s, followed by - // (MinLength - SetBitsThreshold = 91) 0s. - // - // Right before adding the last 0, we expect the window's length to be equal to 210, - // because the bit buffer has gone through these bits: - // (MinLength - SetBitsThreshold = 90) 0s, followed by - // (SetBitsThreshold = 30) 1s, followed by - // (MinLength - SetBitsThreshold = 90) 0s. - - for (size_t Idx = 0; Idx != (MinLength - SetBitsThreshold); Idx++) { - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::AboveThreshold)); - } - EXPECT_EQ(R.second, 2 * MinLength - SetBitsThreshold); - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::Idle)); - - // Continue with the Idle -> NotFilled edge. - for (size_t Idx = 0; Idx != IdleLength - 1; Idx++) { - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::Idle)); - } - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::Idle, State::NotFilled)); - EXPECT_EQ(R.second, 1); -} - -TEST(BitRateWindowTest, WithHoles) { - size_t MinLength = 120, MaxLength = 240, IdleLength = 30; - size_t SetBitsThreshold = 30; - - Result R; - BitRateWindow BRW(MinLength, MaxLength, IdleLength, SetBitsThreshold); - - for (size_t Idx = 0; Idx != MaxLength; Idx++) - R = BRW.insert(false); - - for (size_t Idx = 0; Idx != SetBitsThreshold / 3; Idx++) - R = BRW.insert(true); - for (size_t Idx = 0; Idx != SetBitsThreshold / 3; Idx++) - R = BRW.insert(false); - for (size_t Idx = 0; Idx != SetBitsThreshold / 3; Idx++) - R = BRW.insert(true); - for (size_t Idx = 0; Idx != SetBitsThreshold / 3; Idx++) - R = BRW.insert(false); - for (size_t Idx = 0; Idx != SetBitsThreshold / 3; Idx++) - R = BRW.insert(true); - - EXPECT_EQ(R.first, std::make_pair(State::BelowThreshold, State::AboveThreshold)); - EXPECT_EQ(R.second, MinLength); - - // The window's bit buffer contains: - // 70 0s, 10 1s, 10 0s, 10 1s, 10 0s, 10 1s. - // Where: 70 = MinLength - (5 / 3) * SetBitsThresholds, ie. we need - // to add (70 + 1) more zeros to make the bit buffer go below the - // threshold and then the window's length should be: - // 70 + 50 + 70 = 190. - - BitRateWindow::Edge E; - do { - R = BRW.insert(false); - E = R.first; - } while (E.first != State::AboveThreshold || E.second != State::Idle); - EXPECT_EQ(R.second, 2 * MinLength - (5 * SetBitsThreshold) / 3); -} - -TEST(BitRateWindowTest, MinWindow) { - size_t MinLength = 120, MaxLength = 240, IdleLength = 30; - size_t SetBitsThreshold = 30; - - Result R; - BitRateWindow BRW(MinLength, MaxLength, IdleLength, SetBitsThreshold); - - BRW.insert(true); - BRW.insert(false); - for (size_t Idx = 2; Idx != SetBitsThreshold; Idx++) - BRW.insert(true); - for (size_t Idx = SetBitsThreshold; Idx != MinLength - 1; Idx++) - BRW.insert(false); - - R = BRW.insert(true); - EXPECT_EQ(R.first, std::make_pair(State::NotFilled, State::AboveThreshold)); - EXPECT_EQ(R.second, MinLength); - - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::Idle)); -} - -TEST(BitRateWindowTest, MaxWindow) { - size_t MinLength = 100, MaxLength = 200, IdleLength = 30; - size_t SetBitsThreshold = 50; - - Result R; - BitRateWindow BRW(MinLength, MaxLength, IdleLength, SetBitsThreshold); - - for (size_t Idx = 0; Idx != MaxLength; Idx++) - R = BRW.insert(Idx % 2 == 0); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::AboveThreshold)); - EXPECT_EQ(R.second, MaxLength); - - R = BRW.insert(false); - EXPECT_EQ(R.first, std::make_pair(State::AboveThreshold, State::Idle)); -} diff --git a/ml/kmeans/KMeans.cc b/ml/kmeans/KMeans.cc deleted file mode 100644 index e66c66c16..000000000 --- a/ml/kmeans/KMeans.cc +++ /dev/null @@ -1,55 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "KMeans.h" -#include - -void KMeans::train(SamplesBuffer &SB, size_t MaxIterations) { - std::vector Samples = SB.preprocess(); - - MinDist = std::numeric_limits::max(); - MaxDist = std::numeric_limits::min(); - - { - std::lock_guard Lock(Mutex); - - ClusterCenters.clear(); - - dlib::pick_initial_centers(NumClusters, ClusterCenters, Samples); - dlib::find_clusters_using_kmeans(Samples, ClusterCenters, MaxIterations); - - for (const auto &S : Samples) { - CalculatedNumber MeanDist = 0.0; - - for (const auto &KMCenter : ClusterCenters) - MeanDist += dlib::length(KMCenter - S); - - MeanDist /= NumClusters; - - if (MeanDist < MinDist) - MinDist = MeanDist; - - if (MeanDist > MaxDist) - MaxDist = MeanDist; - } - } -} - -CalculatedNumber KMeans::anomalyScore(SamplesBuffer &SB) { - std::vector DSamples = SB.preprocess(); - - std::unique_lock Lock(Mutex, std::defer_lock); - if (!Lock.try_lock()) - return std::numeric_limits::quiet_NaN(); - - CalculatedNumber MeanDist = 0.0; - for (const auto &CC: ClusterCenters) - MeanDist += dlib::length(CC - DSamples.back()); - - MeanDist /= NumClusters; - - if (MaxDist == MinDist) - return 0.0; - - CalculatedNumber AnomalyScore = 100.0 * std::abs((MeanDist - MinDist) / (MaxDist - MinDist)); - return (AnomalyScore > 100.0) ? 100.0 : AnomalyScore; -} diff --git a/ml/kmeans/KMeans.h b/ml/kmeans/KMeans.h deleted file mode 100644 index 4ea3b6a89..000000000 --- a/ml/kmeans/KMeans.h +++ /dev/null @@ -1,34 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#ifndef KMEANS_H -#define KMEANS_H - -#include -#include -#include -#include - -#include "SamplesBuffer.h" - -class KMeans { -public: - KMeans(size_t NumClusters = 2) : NumClusters(NumClusters) { - MinDist = std::numeric_limits::max(); - MaxDist = std::numeric_limits::min(); - }; - - void train(SamplesBuffer &SB, size_t MaxIterations); - CalculatedNumber anomalyScore(SamplesBuffer &SB); - -private: - size_t NumClusters; - - std::vector ClusterCenters; - - CalculatedNumber MinDist; - CalculatedNumber MaxDist; - - std::mutex Mutex; -}; - -#endif /* KMEANS_H */ diff --git a/ml/kmeans/Makefile.am b/ml/kmeans/Makefile.am deleted file mode 100644 index babdcf0df..000000000 --- a/ml/kmeans/Makefile.am +++ /dev/null @@ -1,4 +0,0 @@ -# SPDX-License-Identifier: GPL-3.0-or-later - -AUTOMAKE_OPTIONS = subdir-objects -MAINTAINERCLEANFILES = $(srcdir)/Makefile.in diff --git a/ml/kmeans/SamplesBuffer.cc b/ml/kmeans/SamplesBuffer.cc deleted file mode 100644 index d276c6e09..000000000 --- a/ml/kmeans/SamplesBuffer.cc +++ /dev/null @@ -1,150 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later -// -#include "SamplesBuffer.h" - -#include -#include -#include - -void Sample::print(std::ostream &OS) const { - for (size_t Idx = 0; Idx != NumDims - 1; Idx++) - OS << CNs[Idx] << ", "; - - OS << CNs[NumDims - 1]; -} - -void SamplesBuffer::print(std::ostream &OS) const { - for (size_t Idx = Preprocessed ? (DiffN + (SmoothN - 1) + (LagN)) : 0; - Idx != NumSamples; Idx++) { - Sample S = Preprocessed ? getPreprocessedSample(Idx) : getSample(Idx); - OS << S << std::endl; - } -} - -std::vector SamplesBuffer::getPreprocessedSamples() const { - std::vector V; - - for (size_t Idx = Preprocessed ? (DiffN + (SmoothN - 1) + (LagN)) : 0; - Idx != NumSamples; Idx++) { - Sample S = Preprocessed ? getPreprocessedSample(Idx) : getSample(Idx); - V.push_back(S); - } - - return V; -} - -void SamplesBuffer::diffSamples() { - // Panda's DataFrame default behaviour is to subtract each element from - // itself. For us `DiffN = 0` means "disable diff-ing" when preprocessing - // the samples buffer. This deviation will make it easier for us to test - // the KMeans implementation. - if (DiffN == 0) - return; - - for (size_t Idx = 0; Idx != (NumSamples - DiffN); Idx++) { - size_t High = (NumSamples - 1) - Idx; - size_t Low = High - DiffN; - - Sample LHS = getSample(High); - Sample RHS = getSample(Low); - - LHS.diff(RHS); - } -} - -void SamplesBuffer::smoothSamples() { - // Holds the mean value of each window - CalculatedNumber *AccCNs = new CalculatedNumber[NumDimsPerSample](); - Sample Acc(AccCNs, NumDimsPerSample); - - // Used to avoid clobbering the accumulator when moving the window - CalculatedNumber *TmpCNs = new CalculatedNumber[NumDimsPerSample](); - Sample Tmp(TmpCNs, NumDimsPerSample); - - CalculatedNumber Factor = (CalculatedNumber) 1 / SmoothN; - - // Calculate the value of the 1st window - for (size_t Idx = 0; Idx != std::min(SmoothN, NumSamples); Idx++) { - Tmp.add(getSample(NumSamples - (Idx + 1))); - } - - Acc.add(Tmp); - Acc.scale(Factor); - - // Move the window and update the samples - for (size_t Idx = NumSamples; Idx != (DiffN + SmoothN - 1); Idx--) { - Sample S = getSample(Idx - 1); - - // Tmp <- Next window (if any) - if (Idx >= (SmoothN + 1)) { - Tmp.diff(S); - Tmp.add(getSample(Idx - (SmoothN + 1))); - } - - // S <- Acc - S.copy(Acc); - - // Acc <- Tmp - Acc.copy(Tmp); - Acc.scale(Factor); - } - - delete[] AccCNs; - delete[] TmpCNs; -} - -void SamplesBuffer::lagSamples() { - if (LagN == 0) - return; - - for (size_t Idx = NumSamples; Idx != LagN; Idx--) { - Sample PS = getPreprocessedSample(Idx - 1); - PS.lag(getSample(Idx - 1), LagN); - } -} - -std::vector SamplesBuffer::preprocess() { - assert(Preprocessed == false); - - std::vector DSamples; - size_t OutN = NumSamples; - - // Diff - if (DiffN >= OutN) - return DSamples; - OutN -= DiffN; - diffSamples(); - - // Smooth - if (SmoothN == 0 || SmoothN > OutN) - return DSamples; - OutN -= (SmoothN - 1); - smoothSamples(); - - // Lag - if (LagN >= OutN) - return DSamples; - OutN -= LagN; - lagSamples(); - - DSamples.reserve(OutN); - Preprocessed = true; - - uint32_t MaxMT = std::numeric_limits::max(); - uint32_t CutOff = static_cast(MaxMT) * SamplingRatio; - - for (size_t Idx = NumSamples - OutN; Idx != NumSamples; Idx++) { - if (RandNums[Idx] > CutOff) - continue; - - DSample DS; - DS.set_size(NumDimsPerSample * (LagN + 1)); - - const Sample PS = getPreprocessedSample(Idx); - PS.initDSample(DS); - - DSamples.push_back(DS); - } - - return DSamples; -} diff --git a/ml/kmeans/SamplesBuffer.h b/ml/kmeans/SamplesBuffer.h deleted file mode 100644 index 1c7215cca..000000000 --- a/ml/kmeans/SamplesBuffer.h +++ /dev/null @@ -1,146 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#ifndef SAMPLES_BUFFER_H -#define SAMPLES_BUFFER_H - -#include -#include - -#include -#include -#include - -#include - -typedef double CalculatedNumber; -typedef dlib::matrix DSample; - -class Sample { -public: - Sample(CalculatedNumber *Buf, size_t N) : CNs(Buf), NumDims(N) {} - - void initDSample(DSample &DS) const { - for (size_t Idx = 0; Idx != NumDims; Idx++) { - DS(Idx) = std::abs(CNs[Idx]); - } - } - - void add(const Sample &RHS) const { - assert(NumDims == RHS.NumDims); - - for (size_t Idx = 0; Idx != NumDims; Idx++) - CNs[Idx] += RHS.CNs[Idx]; - }; - - void diff(const Sample &RHS) const { - assert(NumDims == RHS.NumDims); - - for (size_t Idx = 0; Idx != NumDims; Idx++) - CNs[Idx] -= RHS.CNs[Idx]; - }; - - void copy(const Sample &RHS) const { - assert(NumDims == RHS.NumDims); - - std::memcpy(CNs, RHS.CNs, NumDims * sizeof(CalculatedNumber)); - } - - void scale(CalculatedNumber Factor) { - for (size_t Idx = 0; Idx != NumDims; Idx++) - CNs[Idx] *= Factor; - } - - void lag(const Sample &S, size_t LagN) { - size_t N = S.NumDims; - - for (size_t Idx = 0; Idx != (LagN + 1); Idx++) { - Sample Src(S.CNs - (Idx * N), N); - Sample Dst(CNs + (Idx * N), N); - Dst.copy(Src); - } - } - - const CalculatedNumber *getCalculatedNumbers() const { - return CNs; - }; - - void print(std::ostream &OS) const; - -private: - CalculatedNumber *CNs; - size_t NumDims; -}; - -inline std::ostream& operator<<(std::ostream &OS, const Sample &S) { - S.print(OS); - return OS; -} - -class SamplesBuffer { -public: - SamplesBuffer(CalculatedNumber *CNs, - size_t NumSamples, size_t NumDimsPerSample, - size_t DiffN, size_t SmoothN, size_t LagN, - double SamplingRatio, std::vector &RandNums) : - CNs(CNs), NumSamples(NumSamples), NumDimsPerSample(NumDimsPerSample), - DiffN(DiffN), SmoothN(SmoothN), LagN(LagN), - SamplingRatio(SamplingRatio), RandNums(RandNums), - BytesPerSample(NumDimsPerSample * sizeof(CalculatedNumber)), - Preprocessed(false) {}; - - std::vector preprocess(); - std::vector getPreprocessedSamples() const; - - size_t capacity() const { return NumSamples; } - void print(std::ostream &OS) const; - -private: - size_t getSampleOffset(size_t Index) const { - assert(Index < NumSamples); - return Index * NumDimsPerSample; - } - - size_t getPreprocessedSampleOffset(size_t Index) const { - assert(Index < NumSamples); - return getSampleOffset(Index) * (LagN + 1); - } - - void setSample(size_t Index, const Sample &S) const { - size_t Offset = getSampleOffset(Index); - std::memcpy(&CNs[Offset], S.getCalculatedNumbers(), BytesPerSample); - } - - const Sample getSample(size_t Index) const { - size_t Offset = getSampleOffset(Index); - return Sample(&CNs[Offset], NumDimsPerSample); - }; - - const Sample getPreprocessedSample(size_t Index) const { - size_t Offset = getPreprocessedSampleOffset(Index); - return Sample(&CNs[Offset], NumDimsPerSample * (LagN + 1)); - }; - - void diffSamples(); - void smoothSamples(); - void lagSamples(); - -private: - CalculatedNumber *CNs; - size_t NumSamples; - size_t NumDimsPerSample; - size_t DiffN; - size_t SmoothN; - size_t LagN; - double SamplingRatio; - std::vector &RandNums; - - size_t BytesPerSample; - bool Preprocessed; -}; - -inline std::ostream& operator<<(std::ostream& OS, const SamplesBuffer &SB) { - SB.print(OS); - return OS; -} - -#endif /* SAMPLES_BUFFER_H */ diff --git a/ml/kmeans/Tests.cc b/ml/kmeans/Tests.cc deleted file mode 100644 index 0cb595945..000000000 --- a/ml/kmeans/Tests.cc +++ /dev/null @@ -1,143 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "ml/ml-private.h" -#include - -/* - * The SamplesBuffer class implements the functionality of the following python - * code: - * >> df = pd.DataFrame(data=samples) - * >> df = df.diff(diff_n).dropna() - * >> df = df.rolling(smooth_n).mean().dropna() - * >> df = pd.concat([df.shift(n) for n in range(lag_n + 1)], axis=1).dropna() - * - * Its correctness has been verified by automatically generating random - * data frames in Python and comparing them with the correspondent preprocessed - * SampleBuffers. - * - * The following tests are meant to catch unintended changes in the SamplesBuffer - * implementation. For development purposes, one should compare changes against - * the aforementioned python code. -*/ - -TEST(SamplesBufferTest, NS_8_NDPS_1_DN_1_SN_3_LN_1) { - size_t NumSamples = 8, NumDimsPerSample = 1; - size_t DiffN = 1, SmoothN = 3, LagN = 3; - - size_t N = NumSamples * NumDimsPerSample * (LagN + 1); - CalculatedNumber *CNs = new CalculatedNumber[N](); - - CNs[0] = 0.7568336679490107; - CNs[1] = 0.4814406581763254; - CNs[2] = 0.40073555156221874; - CNs[3] = 0.5973257298194408; - CNs[4] = 0.5334727814345868; - CNs[5] = 0.2632477193454843; - CNs[6] = 0.2684839023122384; - CNs[7] = 0.851332948637479; - - SamplesBuffer SB(CNs, NumSamples, NumDimsPerSample, DiffN, SmoothN, LagN); - SB.preprocess(); - - std::vector Samples = SB.getPreprocessedSamples(); - EXPECT_EQ(Samples.size(), 2); - - Sample S0 = Samples[0]; - const CalculatedNumber *S0_CNs = S0.getCalculatedNumbers(); - Sample S1 = Samples[1]; - const CalculatedNumber *S1_CNs = S1.getCalculatedNumbers(); - - EXPECT_NEAR(S0_CNs[0], -0.109614, 0.001); - EXPECT_NEAR(S0_CNs[1], -0.0458293, 0.001); - EXPECT_NEAR(S0_CNs[2], 0.017344, 0.001); - EXPECT_NEAR(S0_CNs[3], -0.0531693, 0.001); - - EXPECT_NEAR(S1_CNs[0], 0.105953, 0.001); - EXPECT_NEAR(S1_CNs[1], -0.109614, 0.001); - EXPECT_NEAR(S1_CNs[2], -0.0458293, 0.001); - EXPECT_NEAR(S1_CNs[3], 0.017344, 0.001); - - delete[] CNs; -} - -TEST(SamplesBufferTest, NS_8_NDPS_1_DN_2_SN_3_LN_2) { - size_t NumSamples = 8, NumDimsPerSample = 1; - size_t DiffN = 2, SmoothN = 3, LagN = 2; - - size_t N = NumSamples * NumDimsPerSample * (LagN + 1); - CalculatedNumber *CNs = new CalculatedNumber[N](); - - CNs[0] = 0.20511885291342846; - CNs[1] = 0.13151717360306558; - CNs[2] = 0.6017085062423134; - CNs[3] = 0.46256882933941545; - CNs[4] = 0.7887758447877941; - CNs[5] = 0.9237989080034406; - CNs[6] = 0.15552559051428083; - CNs[7] = 0.6309750314597955; - - SamplesBuffer SB(CNs, NumSamples, NumDimsPerSample, DiffN, SmoothN, LagN); - SB.preprocess(); - - std::vector Samples = SB.getPreprocessedSamples(); - EXPECT_EQ(Samples.size(), 2); - - Sample S0 = Samples[0]; - const CalculatedNumber *S0_CNs = S0.getCalculatedNumbers(); - Sample S1 = Samples[1]; - const CalculatedNumber *S1_CNs = S1.getCalculatedNumbers(); - - EXPECT_NEAR(S0_CNs[0], 0.005016, 0.001); - EXPECT_NEAR(S0_CNs[1], 0.326450, 0.001); - EXPECT_NEAR(S0_CNs[2], 0.304903, 0.001); - - EXPECT_NEAR(S1_CNs[0], -0.154948, 0.001); - EXPECT_NEAR(S1_CNs[1], 0.005016, 0.001); - EXPECT_NEAR(S1_CNs[2], 0.326450, 0.001); - - delete[] CNs; -} - -TEST(SamplesBufferTest, NS_8_NDPS_3_DN_2_SN_4_LN_1) { - size_t NumSamples = 8, NumDimsPerSample = 3; - size_t DiffN = 2, SmoothN = 4, LagN = 1; - - size_t N = NumSamples * NumDimsPerSample * (LagN + 1); - CalculatedNumber *CNs = new CalculatedNumber[N](); - - CNs[0] = 0.34310900399667765; CNs[1] = 0.14694315994488194; CNs[2] = 0.8246677800938796; - CNs[3] = 0.48249504592307835; CNs[4] = 0.23241087965531182; CNs[5] = 0.9595348555892567; - CNs[6] = 0.44281094035598334; CNs[7] = 0.5143142171362715; CNs[8] = 0.06391303014242555; - CNs[9] = 0.7460491027783901; CNs[10] = 0.43887217459032923; CNs[11] = 0.2814395025355999; - CNs[12] = 0.9231114281214198; CNs[13] = 0.326882401786898; CNs[14] = 0.26747939220376216; - CNs[15] = 0.7787571209969636; CNs[16] =0.5851700001235088; CNs[17] = 0.34410728945321567; - CNs[18] = 0.9394494507088997; CNs[19] =0.17567223681734334; CNs[20] = 0.42732886195446984; - CNs[21] = 0.9460522396152958; CNs[22] =0.23462747016780894; CNs[23] = 0.35983249900892145; - - SamplesBuffer SB(CNs, NumSamples, NumDimsPerSample, DiffN, SmoothN, LagN); - SB.preprocess(); - - std::vector Samples = SB.getPreprocessedSamples(); - EXPECT_EQ(Samples.size(), 2); - - Sample S0 = Samples[0]; - const CalculatedNumber *S0_CNs = S0.getCalculatedNumbers(); - Sample S1 = Samples[1]; - const CalculatedNumber *S1_CNs = S1.getCalculatedNumbers(); - - EXPECT_NEAR(S0_CNs[0], 0.198225, 0.001); - EXPECT_NEAR(S0_CNs[1], 0.003529, 0.001); - EXPECT_NEAR(S0_CNs[2], -0.063003, 0.001); - EXPECT_NEAR(S0_CNs[3], 0.219066, 0.001); - EXPECT_NEAR(S0_CNs[4], 0.133175, 0.001); - EXPECT_NEAR(S0_CNs[5], -0.293154, 0.001); - - EXPECT_NEAR(S1_CNs[0], 0.174160, 0.001); - EXPECT_NEAR(S1_CNs[1], -0.135722, 0.001); - EXPECT_NEAR(S1_CNs[2], 0.110452, 0.001); - EXPECT_NEAR(S1_CNs[3], 0.198225, 0.001); - EXPECT_NEAR(S1_CNs[4], 0.003529, 0.001); - EXPECT_NEAR(S1_CNs[5], -0.063003, 0.001); - - delete[] CNs; -} diff --git a/ml/ml-dummy.c b/ml/ml-dummy.c index fad480f4b..492dfe2fc 100644 --- a/ml/ml-dummy.c +++ b/ml/ml-dummy.c @@ -24,38 +24,23 @@ char *ml_get_host_info(RRDHOST *RH) { return NULL; } -void ml_new_dimension(RRDDIM *RD) { (void) RD; } - -void ml_delete_dimension(RRDDIM *RD) { (void) RD; } - -bool ml_is_anomalous(RRDDIM *RD, double Value, bool Exists) { - (void) RD; (void) Value; (void) Exists; - return false; -} - -char *ml_get_anomaly_events(RRDHOST *RH, const char *AnomalyDetectorName, - int AnomalyDetectorVersion, time_t After, time_t Before) { - (void) RH; (void) AnomalyDetectorName; - (void) AnomalyDetectorVersion; (void) After; (void) Before; +char *ml_get_host_runtime_info(RRDHOST *RH) { + (void) RH; return NULL; } -char *ml_get_anomaly_event_info(RRDHOST *RH, const char *AnomalyDetectorName, - int AnomalyDetectorVersion, time_t After, time_t Before) { - (void) RH; (void) AnomalyDetectorName; - (void) AnomalyDetectorVersion; (void) After; (void) Before; +char *ml_get_host_models(RRDHOST *RH) { + (void) RH; return NULL; } -void ml_process_rrdr(RRDR *R, int MaxAnomalyRates) { - (void) R; - (void) MaxAnomalyRates; -} +void ml_new_dimension(RRDDIM *RD) { (void) RD; } -void ml_dimension_update_name(RRDSET *RS, RRDDIM *RD, const char *name) { - (void) RS; - (void) RD; - (void) name; +void ml_delete_dimension(RRDDIM *RD) { (void) RD; } + +bool ml_is_anomalous(RRDDIM *RD, double Value, bool Exists) { + (void) RD; (void) Value; (void) Exists; + return false; } bool ml_streaming_enabled() { diff --git a/ml/ml-private.h b/ml/ml-private.h index 7b3e00684..2bd72ac5a 100644 --- a/ml/ml-private.h +++ b/ml/ml-private.h @@ -3,7 +3,7 @@ #ifndef ML_PRIVATE_H #define ML_PRIVATE_H -#include "kmeans/KMeans.h" +#include "KMeans.h" #include "ml/ml.h" #include diff --git a/ml/ml.cc b/ml/ml.cc index 7275d88b8..1a7d6ae25 100644 --- a/ml/ml.cc +++ b/ml/ml.cc @@ -16,7 +16,7 @@ bool ml_enabled(RRDHOST *RH) { if (!Cfg.EnableAnomalyDetection) return false; - if (simple_pattern_matches(Cfg.SP_HostsToSkip, RH->hostname)) + if (simple_pattern_matches(Cfg.SP_HostsToSkip, rrdhost_hostname(RH))) return false; return true; @@ -76,7 +76,7 @@ void ml_new_dimension(RRDDIM *RD) { if (static_cast(RD->update_every) != H->updateEvery()) return; - if (simple_pattern_matches(Cfg.SP_ChartsToSkip, RS->name)) + if (simple_pattern_matches(Cfg.SP_ChartsToSkip, rrdset_name(RS))) return; Dimension *D = new Dimension(RD); @@ -108,7 +108,7 @@ char *ml_get_host_info(RRDHOST *RH) { ConfigJson["enabled"] = false; } - return strdup(ConfigJson.dump(2, '\t').c_str()); + return strdupz(ConfigJson.dump(2, '\t').c_str()); } char *ml_get_host_runtime_info(RRDHOST *RH) { @@ -124,97 +124,24 @@ char *ml_get_host_runtime_info(RRDHOST *RH) { return strdup(ConfigJson.dump(1, '\t').c_str()); } -bool ml_is_anomalous(RRDDIM *RD, double Value, bool Exists) { - Dimension *D = static_cast(RD->ml_dimension); - if (!D) - return false; - - D->addValue(Value, Exists); - bool Result = D->predict().second; - return Result; -} - -char *ml_get_anomaly_events(RRDHOST *RH, const char *AnomalyDetectorName, - int AnomalyDetectorVersion, time_t After, time_t Before) { - if (!RH || !RH->ml_host) { - error("No host"); - return nullptr; - } - - Host *H = static_cast(RH->ml_host); - std::vector> TimeRanges; - - bool Res = H->getAnomaliesInRange(TimeRanges, AnomalyDetectorName, - AnomalyDetectorVersion, - H->getUUID(), - After, Before); - if (!Res) { - error("DB result is empty"); - return nullptr; - } - - nlohmann::json Json = TimeRanges; - return strdup(Json.dump(4).c_str()); -} - -char *ml_get_anomaly_event_info(RRDHOST *RH, const char *AnomalyDetectorName, - int AnomalyDetectorVersion, time_t After, time_t Before) { - if (!RH || !RH->ml_host) { - error("No host"); - return nullptr; - } - - Host *H = static_cast(RH->ml_host); +char *ml_get_host_models(RRDHOST *RH) { + nlohmann::json ModelsJson; - nlohmann::json Json; - bool Res = H->getAnomalyInfo(Json, AnomalyDetectorName, - AnomalyDetectorVersion, - H->getUUID(), - After, Before); - if (!Res) { - error("DB result is empty"); - return nullptr; + if (RH && RH->ml_host) { + Host *H = static_cast(RH->ml_host); + H->getModelsAsJson(ModelsJson); + return strdup(ModelsJson.dump(2, '\t').c_str()); } - return strdup(Json.dump(4, '\t').c_str()); -} - -void ml_process_rrdr(RRDR *R, int MaxAnomalyRates) { - if (R->rows != 1) - return; - - if (MaxAnomalyRates < 1 || MaxAnomalyRates >= R->d) - return; - - NETDATA_DOUBLE *CNs = R->v; - RRDR_DIMENSION_FLAGS *DimFlags = R->od; - - std::vector> V; - - V.reserve(R->d); - for (int Idx = 0; Idx != R->d; Idx++) - V.emplace_back(CNs[Idx], Idx); - - std::sort(V.rbegin(), V.rend()); - - for (int Idx = MaxAnomalyRates; Idx != R->d; Idx++) { - int UnsortedIdx = V[Idx].second; - - int OldFlags = static_cast(DimFlags[UnsortedIdx]); - int NewFlags = OldFlags | RRDR_DIMENSION_HIDDEN; - - DimFlags[UnsortedIdx] = static_cast(NewFlags); - } + return nullptr; } -void ml_dimension_update_name(RRDSET *RS, RRDDIM *RD, const char *Name) { - (void) RS; - +bool ml_is_anomalous(RRDDIM *RD, double Value, bool Exists) { Dimension *D = static_cast(RD->ml_dimension); if (!D) - return; + return false; - D->setAnomalyRateRDName(Name); + return D->predict(Value, Exists); } bool ml_streaming_enabled() { diff --git a/ml/ml.h b/ml/ml.h index e07eb094e..8e62c4988 100644 --- a/ml/ml.h +++ b/ml/ml.h @@ -12,7 +12,7 @@ extern "C" { // This is a DBEngine function redeclared here so that we can free // the anomaly rate dimension, whenever its backing dimension is freed. -extern void rrddim_free(RRDSET *st, RRDDIM *rd); +void rrddim_free(RRDSET *st, RRDDIM *rd); typedef void* ml_host_t; typedef void* ml_dimension_t; @@ -28,22 +28,13 @@ void ml_delete_host(RRDHOST *RH); char *ml_get_host_info(RRDHOST *RH); char *ml_get_host_runtime_info(RRDHOST *RH); +char *ml_get_host_models(RRDHOST *RH); void ml_new_dimension(RRDDIM *RD); void ml_delete_dimension(RRDDIM *RD); bool ml_is_anomalous(RRDDIM *RD, double value, bool exists); -char *ml_get_anomaly_events(RRDHOST *RH, const char *AnomalyDetectorName, - int AnomalyDetectorVersion, time_t After, time_t Before); - -char *ml_get_anomaly_event_info(RRDHOST *RH, const char *AnomalyDetectorName, - int AnomalyDetectorVersion, time_t After, time_t Before); - -void ml_process_rrdr(RRDR *R, int MaxAnomalyRates); - -void ml_dimension_update_name(RRDSET *RS, RRDDIM *RD, const char *name); - bool ml_streaming_enabled(); #define ML_ANOMALY_RATES_CHART_ID "anomaly_detection.anomaly_rates" -- cgit v1.2.3