@cryptotaxi247 / netdata-1 / commits / 0a1ef218f

Load/Store ML models (#14981)

* Pass DB connection in db_execute() * Add support for loading/saving models. * Fix ML stats when no training takes place. * Make model flushing batch size configurable. * Delete unused function * Update ML config. * Restore threshold for logs/period. * Rm whitespace. * Add missing dummy function. * Update function call arguments * Guard transactions with a lock when flushing ML models. * Mark dimensions with loaded models as trained.

vkalintiris committed May 2, 2023 at 19:09 UTC 0a1ef218f00742758585eda5321e6630680eb1bb

12 files changed +134 -83

database/sqlite/sqlite_aclk.c

+4 -4

@@ -255,7 +255,7 @@ static void sql_delete_aclk_table_list(char *host_guid)

255 if (unlikely(rc != SQLITE_OK))

256 error_report("Failed to finalize statement to clean up aclk tables, rc = %d", rc);

257

258 - rc = db_execute(buffer_tostring(sql));

258 + rc = db_execute(db_meta, buffer_tostring(sql));

259 if (unlikely(rc))

260 error("Failed to drop unused ACLK tables");

261

@@ -294,7 +294,7 @@ static int sql_maint_aclk_sync_database(void *data __maybe_unused, int argc __ma

294 {

295 char sql[512];

296 snprintfz(sql,511, SQL_ALERT_CLEANUP, (char *) argv[0], ACLK_DELETE_ACK_ALERTS_INTERNAL);

297 - if (unlikely(db_execute(sql)))

297 + if (unlikely(db_execute(db_meta, sql)))

298 error_report("Failed to clean stale ACLK alert entries");

299 return 0;

300 }

@@ -491,12 +491,12 @@ void sql_create_aclk_table(RRDHOST *host __maybe_unused, uuid_t *host_uuid __may

491 char sql[ACLK_SYNC_QUERY_SIZE];

492

493 snprintfz(sql, ACLK_SYNC_QUERY_SIZE-1, TABLE_ACLK_ALERT, uuid_str);

494 - rc = db_execute(sql);

494 + rc = db_execute(db_meta, sql);

495 if (unlikely(rc))

496 error_report("Failed to create ACLK alert table for host %s", host ? rrdhost_hostname(host) : host_guid);

497 else {

498 snprintfz(sql, ACLK_SYNC_QUERY_SIZE -1, INDEX_ACLK_ALERT, uuid_str, uuid_str);

499 - rc = db_execute(sql);

499 + rc = db_execute(db_meta, sql);

500 if (unlikely(rc))

501 error_report("Failed to create ACLK alert table index for host %s", host ? string2str(host->hostname) : host_guid);

502 }

database/sqlite/sqlite_aclk_alert.c

+5 -5

@@ -299,13 +299,13 @@ void aclk_push_alert_event(struct aclk_sync_host_config *wc)

299

300 BUFFER *sql_fix = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);

301 buffer_sprintf(sql_fix, TABLE_ACLK_ALERT, wc->uuid_str);

302 - rc = db_execute(buffer_tostring(sql_fix));

302 + rc = db_execute(db_meta, buffer_tostring(sql_fix));

303 if (unlikely(rc))

304 error_report("Failed to create ACLK alert table for host %s", rrdhost_hostname(wc->host));

305 else {

306 buffer_flush(sql_fix);

307 buffer_sprintf(sql_fix, INDEX_ACLK_ALERT, wc->uuid_str, wc->uuid_str);

308 - if (unlikely(db_execute(buffer_tostring(sql_fix))))

308 + if (unlikely(db_execute(db_meta, buffer_tostring(sql_fix))))

309 error_report("Failed to create ACLK alert table for host %s", rrdhost_hostname(wc->host));

310 }

311 buffer_free(sql_fix);

@@ -416,7 +416,7 @@ void aclk_push_alert_event(struct aclk_sync_host_config *wc)

416 "WHERE date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64 ";",

417 wc->uuid_str, first_sequence_id, last_sequence_id);

418

419 - if (unlikely(db_execute(buffer_tostring(sql))))

419 + if (unlikely(db_execute(db_meta, buffer_tostring(sql))))

420 error_report("Failed to mark ACLK alert entries as submitted for host %s", rrdhost_hostname(wc->host));

421

422 // Mark to do one more check

@@ -475,7 +475,7 @@ void sql_queue_existing_alerts_to_aclk(RRDHOST *host)

475

476 netdata_rwlock_rdlock(&host->health_log.alarm_log_rwlock);

477

478 - if (unlikely(db_execute(buffer_tostring(sql))))

478 + if (unlikely(db_execute(db_meta, buffer_tostring(sql))))

479 error_report("Failed to queue existing ACLK alert events for host %s", rrdhost_hostname(host));

480

481 netdata_rwlock_unlock(&host->health_log.alarm_log_rwlock);

@@ -685,7 +685,7 @@ void sql_process_queue_removed_alerts_to_aclk(char *node_id)

685

686 snprintfz(sql,ACLK_SYNC_QUERY_SIZE * 2 - 1, SQL_QUEUE_REMOVE_ALERTS, wc->uuid_str, wc->uuid_str, wc->uuid_str);

687

688 - if (unlikely(db_execute(sql))) {

688 + if (unlikely(db_execute(db_meta, sql))) {

689 log_access("ACLK STA [%s (%s)]: QUEUED REMOVED ALERTS FAILED", wc->node_id, rrdhost_hostname(wc->host));

690 error_report("Failed to queue ACLK alert removed entries for host %s", rrdhost_hostname(wc->host));

691 }

database/sqlite/sqlite_functions.c

+2 -2

@@ -510,13 +510,13 @@ skip:

510 }

511 // Return 0 OK

512 // Return 1 Failed

513 -int db_execute(const char *cmd)

513 +int db_execute(sqlite3 *db, const char *cmd)

514 {

515 int rc;

516 int cnt = 0;

517 while (cnt < SQL_MAX_RETRY) {

518 char *err_msg;

519 - rc = sqlite3_exec_monitored(db_meta, cmd, 0, 0, &err_msg);

519 + rc = sqlite3_exec_monitored(db, cmd, 0, 0, &err_msg);

520 if (rc != SQLITE_OK) {

521 error_report("Failed to execute '%s', rc = %d (%s) -- attempt %d", cmd, rc, err_msg, cnt);

522 sqlite3_free(err_msg);

database/sqlite/sqlite_functions.h

+1 -1

@@ -55,7 +55,7 @@ int bind_text_null(sqlite3_stmt *res, int position, const char *text, bool can_b

55 int prepare_statement(sqlite3 *database, const char *query, sqlite3_stmt **statement);

56 int execute_insert(sqlite3_stmt *res);

57 int exec_statement_with_uuid(const char *sql, uuid_t *uuid);

58 -int db_execute(const char *cmd);

58 +int db_execute(sqlite3 *database, const char *cmd);

59 void initialize_thread_key_pool(void);

61 // Look up functions

database/sqlite/sqlite_health.c

+2 -2

@@ -25,12 +25,12 @@ int sql_create_health_log_table(RRDHOST *host) {

26 snprintfz(command, MAX_HEALTH_SQL_SIZE, SQL_CREATE_HEALTH_LOG_TABLE(uuid_str));

28 - rc = db_execute(command);

28 + rc = db_execute(db_meta, command);

29 if (unlikely(rc))

30 error_report("HEALTH [%s]: SQLite error during creation of health log table", rrdhost_hostname(host));

31 else {

32 snprintfz(command, MAX_HEALTH_SQL_SIZE, "CREATE INDEX IF NOT EXISTS health_log_index_%s ON health_log_%s (unique_id); ", uuid_str, uuid_str);

33 - rc = db_execute(command);

33 + rc = db_execute(db_meta, command);

34 if (unlikely(unlikely(rc)))

35 error_report("HEALTH [%s]: SQLite error during creation of health log table index", rrdhost_hostname(host));

36 }

database/sqlite/sqlite_metadata.c

+22 -8

@@ -68,6 +68,7 @@ enum metadata_opcode {

68 METADATA_MAINTENANCE,

69 METADATA_SYNC_SHUTDOWN,

70 METADATA_UNITTEST,

71 + METADATA_ML_LOAD_MODELS,

72 // leave this last

73 // we need it to check for worker utilization

74 METADATA_MAX_ENUMERATIONS_DEFINED

@@ -184,7 +185,7 @@ static int check_and_update_chart_labels(RRDSET *st, BUFFER *work_buffer, size_t

185 struct query_build tmp = {.sql = work_buffer, .count = 0};

186 uuid_unparse_lower(st->chart_uuid, tmp.uuid_str);

187 rrdlabels_walkthrough_read(st->rrdlabels, chart_label_store_to_sql_callback, &tmp);

187 - int rc = db_execute(buffer_tostring(work_buffer));

188 + int rc = db_execute(db_meta, buffer_tostring(work_buffer));

189 if (likely(!rc)) {

190 st->rrdlabels_last_saved_version = new_version;

191 (*query_counter)++;

@@ -203,7 +204,7 @@ void migrate_localhost(uuid_t *host_uuid)

204 if (!rc)

205 rc = exec_statement_with_uuid(DELETE_NON_EXISTING_LOCALHOST, host_uuid);

206 if (!rc) {

206 - if (unlikely(db_execute(DELETE_MISSING_NODE_INSTANCES)))

207 + if (unlikely(db_execute(db_meta, DELETE_MISSING_NODE_INSTANCES)))

208 error_report("Failed to remove deleted hosts from node instances");

209 }

210 }

@@ -1012,7 +1013,7 @@ static bool metadata_scan_host(RRDHOST *host, uint32_t max_count, bool use_trans

1013 uint32_t scan_count = 1;

1014

1015 if (use_transaction)

1015 - (void)db_execute("BEGIN TRANSACTION;");

1016 + (void)db_execute(db_meta, "BEGIN TRANSACTION;");

1017

1018 rrdset_foreach_reentrant(st, host) {

1019 if (scan_count == max_count) {

@@ -1061,7 +1062,7 @@ static bool metadata_scan_host(RRDHOST *host, uint32_t max_count, bool use_trans

1062 rrdset_foreach_done(st);

1063

1064 if (use_transaction)

1064 - (void)db_execute("COMMIT TRANSACTION;");

1065 + (void)db_execute(db_meta, "COMMIT TRANSACTION;");

1066

1067 return more_to_do;

1068 }

@@ -1074,7 +1075,7 @@ static void store_host_and_system_info(RRDHOST *host, BUFFER *work_buffer, size_

1075 work_buffer = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);

1076

1077 if (build_host_system_info_statements(host, work_buffer)) {

1077 - int rc = db_execute(buffer_tostring(work_buffer));

1078 + int rc = db_execute(db_meta, buffer_tostring(work_buffer));

1079 if (unlikely(rc)) {

1080 error_report("METADATA: 'host:%s': Failed to store host updated information in the database", rrdhost_hostname(host));

1081 rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_INFO | RRDHOST_FLAG_METADATA_UPDATE);

@@ -1118,7 +1119,7 @@ static void start_metadata_hosts(uv_work_t *req __maybe_unused)

1119 worker_is_busy(UV_EVENT_METADATA_STORE);

1120

1121 if (!data->max_count)

1121 - transaction_started = !db_execute("BEGIN TRANSACTION;");

1122 + transaction_started = !db_execute(db_meta, "BEGIN TRANSACTION;");

1123

1124 dfe_start_reentrant(rrdhost_root_index, host) {

1125 if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED) || !rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_UPDATE))

@@ -1139,7 +1140,7 @@ static void start_metadata_hosts(uv_work_t *req __maybe_unused)

1140 struct query_build tmp = {.sql = work_buffer, .count = 0};

1141 uuid_unparse_lower(host->host_uuid, tmp.uuid_str);

1142 rrdlabels_walkthrough_read(host->rrdlabels, host_label_store_to_sql_callback, &tmp);

1142 - rc = db_execute(buffer_tostring(work_buffer));

1143 + rc = db_execute(db_meta, buffer_tostring(work_buffer));

1144

1145 if (unlikely(rc)) {

1146 error_report("METADATA: 'host:%s': failed to update metadata host labels", rrdhost_hostname(host));

@@ -1187,7 +1188,7 @@ static void start_metadata_hosts(uv_work_t *req __maybe_unused)

1188 dfe_done(host);

1189

1190 if (!data->max_count && transaction_started)

1190 - transaction_started = db_execute("COMMIT TRANSACTION;");

1191 + transaction_started = db_execute(db_meta, "COMMIT TRANSACTION;");

1192

1193 usec_t all_ended_ut = now_monotonic_usec(); (void)all_ended_ut;

1194 internal_error(true, "METADATA: checking all hosts completed in %0.2f ms",

@@ -1209,6 +1210,7 @@ static void metadata_event_loop(void *arg)

1210 worker_register_job_name(METADATA_STORE_CLAIM_ID, "add claim id");

1211 worker_register_job_name(METADATA_ADD_HOST_INFO, "add host info");

1212 worker_register_job_name(METADATA_MAINTENANCE, "maintenance");

1213 + worker_register_job_name(METADATA_ML_LOAD_MODELS, "ml load models");

1214

1215 int ret;

1216 uv_loop_t *loop;

@@ -1289,6 +1291,11 @@ static void metadata_event_loop(void *arg)

1291 case METADATA_DATABASE_TIMER:

1292 break;

1293

1294 + case METADATA_ML_LOAD_MODELS: {

1295 + RRDDIM *rd = (RRDDIM *) cmd.param[0];

1296 + ml_dimension_load_models(rd);

1297 + break;

1298 + }

1299 case METADATA_DEL_DIMENSION:

1300 uuid = (uuid_t *) cmd.param[0];

1301 if (likely(dimension_can_be_deleted(uuid)))

@@ -1514,6 +1521,13 @@ void metaqueue_host_update_info(RRDHOST *host)

1521 queue_metadata_cmd(METADATA_ADD_HOST_INFO, host, NULL);

1522 }

1523

1524 +void metaqueue_ml_load_models(RRDDIM *rd)

1525 +{

1526 + if (unlikely(!metasync_worker.loop))

1527 + return;

1528 + queue_metadata_cmd(METADATA_ML_LOAD_MODELS, rd, NULL);

1529 +}

1530 +

1531 void metadata_queue_load_host_context(RRDHOST *host)

1532 {

1533 if (unlikely(!metasync_worker.loop))

database/sqlite/sqlite_metadata.h

@@ -14,6 +14,7 @@ void metadata_sync_shutdown_prepare(void);

14 void metaqueue_delete_dimension_uuid(uuid_t *uuid);

15 void metaqueue_store_claim_id(uuid_t *host_uuid, uuid_t *claim_uuid);

16 void metaqueue_host_update_info(RRDHOST *host);

17 +void metaqueue_ml_load_models(RRDDIM *rd);

18 void migrate_localhost(uuid_t *host_uuid);

19 void metadata_queue_load_host_context(RRDHOST *host);

ml/Config.cc

+4 -1

@@ -44,8 +44,9 @@ void ml_config_load(ml_config_t *cfg) {

44 time_t anomaly_detection_query_duration = config_get_number(config_section_ml, "anomaly detection grouping duration", 5 * 60);

46 size_t num_training_threads = config_get_number(config_section_ml, "num training threads", 4);

47 + size_t flush_models_batch_size = config_get_number(config_section_ml, "flush models batch size", 128);

48 - bool enable_statistics_charts = config_get_boolean(config_section_ml, "enable statistics charts", false);

49 + bool enable_statistics_charts = config_get_boolean(config_section_ml, "enable statistics charts", true);

51 /*

52 * Clamp

@@ -69,6 +70,7 @@ void ml_config_load(ml_config_t *cfg) {

70 anomaly_detection_query_duration = clamp<time_t>(anomaly_detection_query_duration, 60, 15 * 60);

72 num_training_threads = clamp<size_t>(num_training_threads, 1, 128);

73 + flush_models_batch_size = clamp<size_t>(flush_models_batch_size, 8, 512);

75 /*

76 * Validate

@@ -117,6 +119,7 @@ void ml_config_load(ml_config_t *cfg) {

119 cfg->stream_anomaly_detection_charts = config_get_boolean(config_section_ml, "stream anomaly detection charts", true);

120

121 cfg->num_training_threads = num_training_threads;

122 + cfg->flush_models_batch_size = flush_models_batch_size;

123

124 cfg->enable_statistics_charts = enable_statistics_charts;

125 }

ml/ml-dummy.c

@@ -92,6 +92,11 @@ bool ml_dimension_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool

92 return false;

93 }

95 +int ml_dimension_load_models(RRDDIM *rd) {

96 + UNUSED(rd);

97 + return 0;

98 +}

99 +

100 void ml_update_global_statistics_charts(uint64_t models_consulted) {

101 UNUSED(models_consulted);

102 }

ml/ml-private.h

@@ -246,6 +246,11 @@ typedef struct {

246 RRDDIM *detector_events_new_anomaly_event_rd;

247 } ml_host_t;

248

249 +typedef struct {

250 + uuid_t metric_uuid;

251 + ml_kmeans_t kmeans;

252 +} ml_model_info_t;

253 +

254 typedef struct {

255 size_t id;

256 netdata_thread_t nd_thread;

@@ -258,6 +263,8 @@ typedef struct {

263 calculated_number_t *scratch_training_cns;

264 std::vector<DSample> training_samples;

265

266 + std::vector<ml_model_info_t> pending_model_info;

267 +

268 RRDSET *queue_stats_rs;

269 RRDDIM *queue_stats_queue_size_rd;

270 RRDDIM *queue_stats_popped_items_rd;

@@ -313,6 +320,7 @@ typedef struct {

320 std::atomic<bool> detection_stop;

321

322 size_t num_training_threads;

323 + size_t flush_models_batch_size;

324

325 std::vector<ml_training_thread_t> training_threads;

326 std::atomic<bool> training_stop;

ml/ml.cc

+78 -60

@@ -16,9 +16,10 @@

16 #define WORKER_TRAIN_UPDATE_MODELS 4

17 #define WORKER_TRAIN_RELEASE_DIMENSION 5

18 #define WORKER_TRAIN_UPDATE_HOST 6

19 -#define WORKER_TRAIN_LOAD_MODELS 7

19 +#define WORKER_TRAIN_FLUSH_MODELS 7

21 static sqlite3 *db = NULL;

22 +static netdata_mutex_t db_mutex = NETDATA_MUTEX_INITIALIZER;

24 /*

25 * Functions to convert enums to strings

@@ -406,7 +407,7 @@ ml_dimension_calculated_numbers(ml_training_thread_t *training_thread, ml_dimens

407

408 const char *db_models_create_table =

409 "CREATE TABLE IF NOT EXISTS models("

409 - " dim_id BLOB, dim_str TEXT, after INT, before INT,"

410 + " dim_id BLOB, after INT, before INT,"

411 " min_dist REAL, max_dist REAL,"

412 " c00 REAL, c01 REAL, c02 REAL, c03 REAL, c04 REAL, c05 REAL,"

413 " c10 REAL, c11 REAL, c12 REAL, c13 REAL, c14 REAL, c15 REAL,"

@@ -415,26 +416,26 @@ const char *db_models_create_table =

416

417 const char *db_models_add_model =

418 "INSERT OR REPLACE INTO models("

418 - " dim_id, dim_str, after, before,"

419 + " dim_id, after, before,"

420 " min_dist, max_dist,"

421 " c00, c01, c02, c03, c04, c05,"

422 " c10, c11, c12, c13, c14, c15)"

423 "VALUES("

423 - " @dim_id, @dim_str, @after, @before,"

424 + " @dim_id, @after, @before,"

425 " @min_dist, @max_dist,"

426 " @c00, @c01, @c02, @c03, @c04, @c05,"

427 " @c10, @c11, @c12, @c13, @c14, @c15);";

428

429 const char *db_models_load =

430 "SELECT * FROM models "

430 - "WHERE dim_id == @dim_id AND after >= @after ORDER BY before ASC;";

431 + "WHERE dim_id = @dim_id AND after >= @after ORDER BY before ASC;";

432

433 const char *db_models_delete =

434 "DELETE FROM models "

435 "WHERE dim_id = @dim_id AND before < @before;";

436

437 static int

437 -ml_dimension_add_model(ml_dimension_t *dim)

438 +ml_dimension_add_model(const uuid_t *metric_uuid, const ml_kmeans_t *km)

439 {

440 static __thread sqlite3_stmt *res = NULL;

441 int param = 0;

@@ -453,36 +454,30 @@ ml_dimension_add_model(ml_dimension_t *dim)

454 }

455 }

456

456 - rc = sqlite3_bind_blob(res, ++param, &dim->rd->metric_uuid, sizeof(dim->rd->metric_uuid), SQLITE_STATIC);

457 - if (unlikely(rc != SQLITE_OK))

458 - goto bind_fail;

459 -

460 - char id[1024];

461 - snprintfz(id, 1024 - 1, "%s.%s", rrdset_id(dim->rd->rrdset), rrddim_id(dim->rd));

462 - rc = sqlite3_bind_text(res, ++param, id, -1, SQLITE_STATIC);

457 + rc = sqlite3_bind_blob(res, ++param, metric_uuid, sizeof(*metric_uuid), SQLITE_STATIC);

458 if (unlikely(rc != SQLITE_OK))

459 goto bind_fail;

460

466 - rc = sqlite3_bind_int(res, ++param, (int) dim->kmeans.after);

461 + rc = sqlite3_bind_int(res, ++param, (int) km->after);

462 if (unlikely(rc != SQLITE_OK))

463 goto bind_fail;

464

470 - rc = sqlite3_bind_int(res, ++param, (int) dim->kmeans.before);

465 + rc = sqlite3_bind_int(res, ++param, (int) km->before);

466 if (unlikely(rc != SQLITE_OK))

467 goto bind_fail;

468

474 - rc = sqlite3_bind_double(res, ++param, dim->kmeans.min_dist);

469 + rc = sqlite3_bind_double(res, ++param, km->min_dist);

470 if (unlikely(rc != SQLITE_OK))

471 goto bind_fail;

472

478 - rc = sqlite3_bind_double(res, ++param, dim->kmeans.max_dist);

473 + rc = sqlite3_bind_double(res, ++param, km->max_dist);

474 if (unlikely(rc != SQLITE_OK))

475 goto bind_fail;

476

482 - if (dim->kmeans.cluster_centers.size() != 2)

483 - fatal("Expected 2 cluster centers, got %zu", dim->kmeans.cluster_centers.size());

477 + if (km->cluster_centers.size() != 2)

478 + fatal("Expected 2 cluster centers, got %zu", km->cluster_centers.size());

479

485 - for (const DSample &ds : dim->kmeans.cluster_centers) {

480 + for (const DSample &ds : km->cluster_centers) {

481 if (ds.size() != 6)

482 fatal("Expected dsample with 6 dimensions, got %ld", ds.size());

483

@@ -513,7 +508,7 @@ bind_fail:

508 }

509

510 static int

516 -ml_dimension_delete_models(ml_dimension_t *dim)

511 +ml_dimension_delete_models(const uuid_t *metric_uuid, time_t before)

512 {

513 static __thread sqlite3_stmt *res = NULL;

514 int rc = 0;

@@ -532,11 +527,11 @@ ml_dimension_delete_models(ml_dimension_t *dim)

527 }

528 }

529

535 - rc = sqlite3_bind_blob(res, ++param, &dim->rd->metric_uuid, sizeof(dim->rd->metric_uuid), SQLITE_STATIC);

530 + rc = sqlite3_bind_blob(res, ++param, metric_uuid, sizeof(*metric_uuid), SQLITE_STATIC);

531 if (unlikely(rc != SQLITE_OK))

532 goto bind_fail;

533

539 - rc = sqlite3_bind_int(res, ++param, (int) dim->kmeans.before - (Cfg.num_models_to_use * Cfg.train_every));

534 + rc = sqlite3_bind_int(res, ++param, (int) before);

535 if (unlikely(rc != SQLITE_OK))

536 goto bind_fail;

537

@@ -558,8 +553,18 @@ bind_fail:

553 return 1;

554 }

555

561 -static int

562 -ml_dimension_load_models(ml_dimension_t *dim) {

556 +int ml_dimension_load_models(RRDDIM *rd) {

557 + ml_dimension_t *dim = (ml_dimension_t *) rd->ml_dimension;

558 + if (!dim)

559 + return 0;

560 +

561 + netdata_mutex_lock(&dim->mutex);

562 + bool is_empty = dim->km_contexts.empty();

563 + netdata_mutex_unlock(&dim->mutex);

564 +

565 + if (!is_empty)

566 + return 0;

567 +

568 std::vector<ml_kmeans_t> V;

569

570 static __thread sqlite3_stmt *res = NULL;

@@ -587,6 +592,8 @@ ml_dimension_load_models(ml_dimension_t *dim) {

592 if (unlikely(rc != SQLITE_OK))

593 goto bind_fail;

594

595 + netdata_mutex_lock(&dim->mutex);

596 +

597 dim->km_contexts.reserve(Cfg.num_models_to_use);

598 while ((rc = sqlite3_step_monitored(res)) == SQLITE_ROW) {

599 ml_kmeans_t km;

@@ -618,6 +625,12 @@ ml_dimension_load_models(ml_dimension_t *dim) {

625 dim->km_contexts.push_back(km);

626 }

627

628 + if (!dim->km_contexts.empty()) {

629 + dim->ts = TRAINING_STATUS_TRAINED;

630 + }

631 +

632 + netdata_mutex_unlock(&dim->mutex);

633 +

634 if (unlikely(rc != SQLITE_DONE))

635 error_report("Failed to load models, rc = %d", rc);

636

@@ -635,24 +648,6 @@ bind_fail:

648 return 1;

649 }

650

638 -static int

639 -ml_dimension_update_models(ml_dimension_t *dim)

640 -{

641 - int rc;

642 -

643 - if (dim->km_contexts.empty()) {

644 - rc = ml_dimension_load_models(dim);

645 - if (rc)

646 - return rc;

647 - }

648 -

649 - rc = ml_dimension_add_model(dim);

650 - if (rc)

651 - return rc;

652 -

653 - return ml_dimension_delete_models(dim);

654 -}

655 -

651 static enum ml_training_result

652 ml_dimension_train_model(ml_training_thread_t *training_thread, ml_dimension_t *dim, const ml_training_request_t &training_request)

653 {

@@ -704,22 +699,10 @@ ml_dimension_train_model(ml_training_thread_t *training_thread, ml_dimension_t *

699 }

700

701 // update models

702 + worker_is_busy(WORKER_TRAIN_UPDATE_MODELS);

703 {

704 netdata_mutex_lock(&dim->mutex);

705

710 - // temporarily disable sqlite operations because they interfere with

711 - // training scheduling on busy parents

712 - #if 0

713 - worker_is_busy(WORKER_TRAIN_LOAD_MODELS);

714 -

715 - int rc = ml_dimension_update_models(dim);

716 - if (rc) {

717 - error("Failed to update models for %s [%u, %u]", rrddim_id(dim->rd), dim->kmeans.after, dim->kmeans.before);

718 - }

719 - #endif

720 -

721 - worker_is_busy(WORKER_TRAIN_UPDATE_MODELS);

722 -

706 if (dim->km_contexts.size() < Cfg.num_models_to_use) {

707 dim->km_contexts.push_back(std::move(dim->kmeans));

708 } else {

@@ -747,6 +730,12 @@ ml_dimension_train_model(ml_training_thread_t *training_thread, ml_dimension_t *

730 dim->tr = training_response;

731 dim->last_training_time = rrddim_last_entry_s(dim->rd);

732

733 + // Add the newly generated model to the list of pending models to flush

734 + ml_model_info_t model_info;

735 + uuid_copy(model_info.metric_uuid, dim->rd->metric_uuid);

736 + model_info.kmeans = dim->km_contexts.back();

737 + training_thread->pending_model_info.push_back(model_info);

738 +

739 netdata_mutex_unlock(&dim->mutex);

740 }

741

@@ -1133,10 +1122,9 @@ ml_detect_main(void *arg)

1122 training_stats.consumed_ut /= training_stats.num_popped_items;

1123 training_stats.remaining_ut /= training_stats.num_popped_items;

1124 } else {

1136 - training_stats.queue_size = 0;

1137 - training_stats.allotted_ut = 0;

1125 + training_stats.queue_size = ml_queue_size(training_thread->training_queue);

1126 training_stats.consumed_ut = 0;

1139 - training_stats.remaining_ut = 0;

1127 + training_stats.remaining_ut = training_stats.allotted_ut;

1128

1129 training_stats.training_result_ok = 0;

1130 training_stats.training_result_invalid_query_time_range = 0;

@@ -1352,6 +1340,8 @@ void ml_dimension_new(RRDDIM *rd)

1340 dim->km_contexts.reserve(Cfg.num_models_to_use);

1341

1342 rd->ml_dimension = (rrd_ml_dimension_t *) dim;

1343 +

1344 + metaqueue_ml_load_models(rd);

1345 }

1346

1347 void ml_dimension_delete(RRDDIM *rd)

@@ -1380,6 +1370,25 @@ bool ml_dimension_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool

1370 return is_anomalous;

1371 }

1372

1373 +static int ml_flush_pending_models(ml_training_thread_t *training_thread) {

1374 + (void) db_execute(db, "BEGIN TRANSACTION;");

1375 +

1376 + for (const auto &pending_model: training_thread->pending_model_info) {

1377 + int rc = ml_dimension_add_model(&pending_model.metric_uuid, &pending_model.kmeans);

1378 + if (rc)

1379 + return rc;

1380 +

1381 + rc = ml_dimension_delete_models(&pending_model.metric_uuid, pending_model.kmeans.before - (Cfg.num_models_to_use * Cfg.train_every));

1382 + if (rc)

1383 + return rc;

1384 + }

1385 +

1386 + (void) db_execute(db, "COMMIT TRANSACTION;");

1387 +

1388 + training_thread->pending_model_info.clear();

1389 + return 0;

1390 +}

1391 +

1392 static void *ml_train_main(void *arg) {

1393 ml_training_thread_t *training_thread = (ml_training_thread_t *) arg;

1394

@@ -1392,9 +1401,9 @@ static void *ml_train_main(void *arg) {

1401 worker_register_job_name(WORKER_TRAIN_QUERY, "query");

1402 worker_register_job_name(WORKER_TRAIN_KMEANS, "kmeans");

1403 worker_register_job_name(WORKER_TRAIN_UPDATE_MODELS, "update models");

1395 - worker_register_job_name(WORKER_TRAIN_LOAD_MODELS, "load models");

1404 worker_register_job_name(WORKER_TRAIN_RELEASE_DIMENSION, "release");

1405 worker_register_job_name(WORKER_TRAIN_UPDATE_HOST, "update host");

1406 + worker_register_job_name(WORKER_TRAIN_FLUSH_MODELS, "flush models");

1407

1408 while (!Cfg.training_stop) {

1409 worker_is_busy(WORKER_TRAIN_QUEUE_POP);

@@ -1470,6 +1479,14 @@ static void *ml_train_main(void *arg) {

1479 netdata_mutex_unlock(&training_thread->nd_mutex);

1480 }

1481

1482 + if (training_thread->pending_model_info.size() >= Cfg.flush_models_batch_size) {

1483 + worker_is_busy(WORKER_TRAIN_FLUSH_MODELS);

1484 + netdata_mutex_lock(&db_mutex);

1485 + ml_flush_pending_models(training_thread);

1486 + netdata_mutex_unlock(&db_mutex);

1487 + continue;

1488 + }

1489 +

1490 worker_is_idle();

1491 std::this_thread::sleep_for(std::chrono::microseconds{remaining_ut});

1492 }

@@ -1505,6 +1522,7 @@ void ml_init()

1522

1523 training_thread->id = idx;

1524 training_thread->training_queue = ml_queue_init();

1525 + training_thread->pending_model_info.reserve(Cfg.flush_models_batch_size);

1526 netdata_mutex_init(&training_thread->nd_mutex);

1527 }

1528

ml/ml.h

@@ -36,6 +36,8 @@ void ml_dimension_new(RRDDIM *rd);

36 void ml_dimension_delete(RRDDIM *rd);

37 bool ml_dimension_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool exists);

39 +int ml_dimension_load_models(RRDDIM *rd);

40 +

41 void ml_update_global_statistics_charts(uint64_t models_consulted);

43 #ifdef __cplusplus