@cryptotaxi247 / netdata-1 / commits / 18d3b0da7

Improve child to parent provided ML trained data (#22156)

* Add support for downstream-supplied ML models and enhance dimension handling logic - Introduced `ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED` result type for managing dimensions with downstream-provided models. - Added logic to skip retraining for dimensions flagged as supplied by downstream models. - Refactored ML dimension and host operations, adding generation tracking (`reset_generation`) and context management (`ml_host_clear_context_anomaly_rate`). - Enhanced safety and performance by improving model update checks, avoiding duplicate or stale models, and simplifying initialization/enqueue workflows. * Refactor ML dimension training logic and add unit tests for model update and requeue checks * ml: tighten downstream-model install and avoid double-enqueue on host start - Move has_received_downstream_model = true into ml_dimension_update_models() via a new from_downstream parameter, so the flag and the km_contexts install commit or cancel together under the same dim->slock. Previously the flag was set in ml_worker_add_existing_model() before unlocking, leaving a transient window where ml_dimension_update_models() could bail and leave the flag set without an installed model. - Gate ml_dimension_new()'s enqueue on host->ml_running so newly created dims aren't double-enqueued by both the create path and ml_host_start()'s sweep. Dims created before start are picked up by the sweep. - Take host->mutex in ml_host_start() before flipping ml_running so concurrent starts serialize and the flag flip is bounded by the sweep's critical section. * ml: prevent double enqueue of CREATE_NEW_MODEL requests and enhance downstream dimension handling - Add create_new_model_queued flag to track queued dimensions and avoid redundant enqueuing. - Simplify ml_dimension_enqueue_create_model by consolidating enqueue logic. - Improve short-circuiting and result handling for downstream-supplied dimensions in model training. * Add `ML_WORKER_RESULT_TRAINING_IN_PROGRESS` handling and improve requeue logic for model training * `ml: ensure deterministic stop by reasserting ml_running state under mutex`

Stelios Fragkakis committed May 14, 2026 at 09:29 UTC 18d3b0da7dc45b0e81ff69e969d138ed37e7a97b

7 files changed +326 -59

src/ml/ml-unittest.cc

+81

@@ -3,6 +3,7 @@

3 #include "ml_config.h"

4 #include "ml_features.h"

5 #include "ml_kmeans.h"

6 +#include "ml_private.h"

8 #include <algorithm>

9 #include <cmath>

@@ -862,6 +863,84 @@ static void test_kmeans_timestamp_rejection()

863 }

864 }

865

866 +static void test_downstream_model_short_circuit_and_requeue()

867 +{

868 + fprintf(stderr, " test_downstream_model_short_circuit_and_requeue...\n");

869 +

870 + enum ml_worker_result worker_res = ML_WORKER_RESULT_OK;

871 + bool should_short_circuit = ml_dimension_train_model_precheck(METRIC_TYPE_VARIABLE,

872 + true,

873 + false,

874 + &worker_res);

875 + ML_TEST_ASSERT(should_short_circuit,

876 + "downstream-supplied dimensions should short-circuit local training");

877 + ML_TEST_ASSERT(worker_res == ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED,

878 + "downstream-supplied dimensions should short-circuit local training");

879 + ML_TEST_ASSERT(!ml_should_requeue_create_new_model(worker_res),

880 + "downstream-supplied result should stop CREATE_NEW_MODEL requeueing");

881 +

882 + worker_res = ML_WORKER_RESULT_OK;

883 + should_short_circuit = ml_dimension_train_model_precheck(METRIC_TYPE_CONSTANT,

884 + true,

885 + false,

886 + &worker_res);

887 + ML_TEST_ASSERT(should_short_circuit,

888 + "constant downstream-supplied dimensions should short-circuit local training");

889 + ML_TEST_ASSERT(worker_res == ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED,

890 + "constant downstream-supplied dimensions should drain CREATE_NEW_MODEL items");

891 + ML_TEST_ASSERT(!ml_should_requeue_create_new_model(worker_res),

892 + "constant downstream-supplied result should stop CREATE_NEW_MODEL requeueing");

893 +

894 + worker_res = ML_WORKER_RESULT_OK;

895 + should_short_circuit = ml_dimension_train_model_precheck(METRIC_TYPE_VARIABLE,

896 + false,

897 + false,

898 + &worker_res);

899 + ML_TEST_ASSERT(!should_short_circuit,

900 + "dimensions without downstream models should continue to training");

901 + ML_TEST_ASSERT(ml_should_requeue_create_new_model(ML_WORKER_RESULT_OK),

902 + "ordinary training results should keep CREATE_NEW_MODEL items requeueing");

903 +

904 + worker_res = ML_WORKER_RESULT_OK;

905 + should_short_circuit = ml_dimension_train_model_precheck(METRIC_TYPE_VARIABLE,

906 + false,

907 + true,

908 + &worker_res);

909 + ML_TEST_ASSERT(should_short_circuit,

910 + "training-in-progress dimensions should short-circuit local training");

911 + ML_TEST_ASSERT(worker_res == ML_WORKER_RESULT_TRAINING_IN_PROGRESS,

912 + "training-in-progress dimensions should report the distinct result");

913 + ML_TEST_ASSERT(ml_should_requeue_create_new_model(worker_res),

914 + "training-in-progress result should keep CREATE_NEW_MODEL items requeueing "

915 + "so the dim stays in the periodic retrain cycle");

916 +}

917 +

918 +static void test_reset_generation_cancels_model_publish()

919 +{

920 + fprintf(stderr, " test_reset_generation_cancels_model_publish...\n");

921 +

922 + bool training_in_progress = true;

923 + bool should_publish = ml_should_publish_model_update(true, 8, 7, &training_in_progress);

924 + ML_TEST_ASSERT(!should_publish,

925 + "generation mismatch should cancel model publication");

926 + ML_TEST_ASSERT(!training_in_progress,

927 + "generation mismatch should clear training_in_progress");

928 +

929 + training_in_progress = true;

930 + should_publish = ml_should_publish_model_update(false, 7, 7, &training_in_progress);

931 + ML_TEST_ASSERT(!should_publish,

932 + "stopped hosts should cancel model publication");

933 + ML_TEST_ASSERT(!training_in_progress,

934 + "stopped-host cancellation should clear training_in_progress");

935 +

936 + training_in_progress = true;

937 + should_publish = ml_should_publish_model_update(true, 7, 7, &training_in_progress);

938 + ML_TEST_ASSERT(should_publish,

939 + "matching generation on a running host should allow model publication");

940 + ML_TEST_ASSERT(training_in_progress,

941 + "successful publication path should leave training_in_progress unchanged");

942 +}

943 +

944 extern "C" int ml_unittest()

945 {

946 fprintf(stderr, "\nML unit tests:\n");

@@ -888,6 +967,8 @@ extern "C" int ml_unittest()

967 test_parameter_combinations();

968 test_kmeans_timestamp_roundtrip();

969 test_kmeans_timestamp_rejection();

970 + test_downstream_model_short_circuit_and_requeue();

971 + test_reset_generation_cancels_model_publish();

972

973 fprintf(stderr, "\nML tests: %d run, %d failed\n", tests_run, tests_failed);

974

src/ml/ml.cc

+129 -40

@@ -671,19 +671,73 @@ static void ml_dimension_stream_kmeans(ml_worker_t *worker, const ml_dimension_t

671 pulse_ml_models_sent();

672 }

673

674 -static void ml_dimension_update_models(ml_worker_t *worker, ml_dimension_t *dim)

674 +bool ml_dimension_train_model_precheck(enum ml_metric_type mt,

675 + bool has_received_downstream_model,

676 + bool training_in_progress,

677 + enum ml_worker_result *worker_res)

678 +{

679 + if (has_received_downstream_model) {

680 + *worker_res = ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED;

681 + return true;

682 + }

683 +

684 + if (mt == METRIC_TYPE_CONSTANT) {

685 + *worker_res = ML_WORKER_RESULT_OK;

686 + return true;

687 + }

688 +

689 + if (training_in_progress) {

690 + *worker_res = ML_WORKER_RESULT_TRAINING_IN_PROGRESS;

691 + return true;

692 + }

693 +

694 + return false;

695 +}

696 +

697 +bool ml_should_requeue_create_new_model(enum ml_worker_result worker_res)

698 +{

699 + // TRAINING_IN_PROGRESS keeps requeueing so the dim stays in the periodic

700 + // retrain cycle; the worker loop is paced by Cfg.train_every, so this is

701 + // not a tight CPU spin.

702 + return worker_res != ML_WORKER_RESULT_NULL_ACQUIRED_DIMENSION &&

703 + worker_res != ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED;

704 +}

705 +

706 +bool ml_should_publish_model_update(bool host_running,

707 + uint32_t current_generation,

708 + uint32_t expected_generation,

709 + bool *training_in_progress)

710 +{

711 + if (!host_running || current_generation != expected_generation) {

712 + if (training_in_progress)

713 + *training_in_progress = false;

714 + return false;

715 + }

716 +

717 + return true;

718 +}

719 +

720 +static bool ml_dimension_update_models(ml_worker_t *worker, ml_dimension_t *dim, uint32_t expected_generation, bool from_downstream)

721 {

722 worker_is_busy(WORKER_TRAIN_UPDATE_MODELS);

723

724 spinlock_lock(&dim->slock);

725

726 ml_host_t *host = (ml_host_t *) dim->rd->rrdset->rrdhost->ml_host;

681 - if (!host || !host->ml_running) {

682 - dim->training_in_progress = false;

727 + if (!ml_should_publish_model_update(host && host->ml_running,

728 + dim->reset_generation,

729 + expected_generation,

730 + &dim->training_in_progress)) {

731 spinlock_unlock(&dim->slock);

684 - return;

732 + return false;

733 }

734

735 + // Mark the dim as downstream-supplied only after the publish-check passes

736 + // and under the same slock as the install. Setting it earlier would risk

737 + // suppressing local training if the install was cancelled.

738 + if (from_downstream)

739 + dim->has_received_downstream_model = true;

740 +

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

742 dim->km_contexts.emplace_back(dim->kmeans);

743 } else {

@@ -712,7 +766,7 @@ static void ml_dimension_update_models(ml_worker_t *worker, ml_dimension_t *dim)

766 dim->suppression_anomaly_counter = 0;

767 dim->suppression_window_counter = 0;

768

715 - // Add the newly generated model to the list of pending models to flush

769 + // Add the latest model to the list of pending models to flush.

770 ml_model_info_t model_info;

771 nd_uuid_t *rd_uuid = uuidmap_uuid_ptr(dim->rd->uuid);

772 uuid_copy(model_info.metric_uuid, *rd_uuid);

@@ -725,6 +779,7 @@ static void ml_dimension_update_models(ml_worker_t *worker, ml_dimension_t *dim)

779 dim->training_in_progress = false;

780

781 spinlock_unlock(&dim->slock);

782 + return true;

783 }

784

785 static enum ml_worker_result

@@ -733,20 +788,22 @@ ml_dimension_train_model(ml_worker_t *worker, ml_dimension_t *dim)

788 worker_is_busy(WORKER_TRAIN_QUERY);

789

790 spinlock_lock(&dim->slock);

736 - if (dim->mt == METRIC_TYPE_CONSTANT) {

791 + ml_worker_result precheck;

792 + if (ml_dimension_train_model_precheck(dim->mt,

793 + dim->has_received_downstream_model,

794 + dim->training_in_progress,

795 + &precheck)) {

796 + if (precheck == ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED)

797 + dim->create_new_model_queued = false;

798 spinlock_unlock(&dim->slock);

738 - return ML_WORKER_RESULT_OK;

799 + return precheck;

800 }

801

741 - // Check if training is already in progress for this dimension

742 - // If so, skip this training request to prevent concurrent access to dim->kmeans

743 - if (dim->training_in_progress) {

744 - spinlock_unlock(&dim->slock);

745 - return ML_WORKER_RESULT_OK;

746 - }

747 -

748 - // Mark training as in progress

802 + // Mark training as in progress and snapshot the generation so that

803 + // ml_dimension_update_models() can detect a stop/reset that happened

804 + // while training was running.

805 dim->training_in_progress = true;

806 + uint32_t generation = dim->reset_generation;

807 spinlock_unlock(&dim->slock);

808

809 auto P = ml_dimension_calculated_numbers(worker, dim);

@@ -779,14 +836,14 @@ ml_dimension_train_model(ml_worker_t *worker, ml_dimension_t *dim)

836 worker->scratch_training_cns, training_response.total_values,

837 worker->training_cns, training_response.total_values

838 };

782 -

839 +

840 // Calculate dynamic sampling ratio based on expected output size

841 // After diff and smooth, we'll have approximately this many vectors

842 size_t expected_vectors = training_response.total_values;

843 if (Cfg.diff_n > 0) expected_vectors--;

844 if (smoothing_window > 1) expected_vectors = expected_vectors - smoothing_window + 1;

845 expected_vectors = expected_vectors - Cfg.lag_n;

789 -

846 +

847 double sampling_ratio = 1.0;

848 if (expected_vectors > Cfg.max_training_vectors) {

849 sampling_ratio = (double)Cfg.max_training_vectors / expected_vectors;

@@ -800,7 +857,7 @@ ml_dimension_train_model(ml_worker_t *worker, ml_dimension_t *dim)

857 }

858

859 // update models

803 - ml_dimension_update_models(worker, dim);

860 + (void) ml_dimension_update_models(worker, dim, generation, /*from_downstream=*/false);

861

862 return worker_result;

863 }

@@ -1053,12 +1110,15 @@ ml_host_detect_once(ml_host_t *host, ONEWAYALLOC *owa)

1110 auto &um = host->context_anomaly_rate;

1111 auto it = um.find(key);

1112 if (it == um.end()) {

1056 - um[key] = ml_context_anomaly_rate_t {

1113 + STRING *owned_key = string_dup(key);

1114 + auto insert_result = um.emplace(owned_key, ml_context_anomaly_rate_t {

1115 .rd = NULL,

1116 .normal_dimensions = 0,

1117 .anomalous_dimensions = 0

1060 - };

1061 - it = um.find(key);

1118 + });

1119 + if (!insert_result.second)

1120 + string_freez(owned_key);

1121 + it = insert_result.first;

1122 }

1123

1124 it->second.anomalous_dimensions += chart_mls.num_anomalous_dimensions;

@@ -1084,15 +1144,6 @@ ml_host_detect_once(ml_host_t *host, ONEWAYALLOC *owa)

1144 ml_update_host_and_detection_rate_charts(host, host->host_anomaly_rate * 10000.0, owa);

1145 } else {

1146 host->host_anomaly_rate = 0.0;

1087 -

1088 - auto &um = host->context_anomaly_rate;

1089 - for (auto &entry: um) {

1090 - entry.second = ml_context_anomaly_rate_t {

1091 - .rd = NULL,

1092 - .normal_dimensions = 0,

1093 - .anomalous_dimensions = 0

1094 - };

1095 - }

1147 }

1148 }

1149

@@ -1219,9 +1270,6 @@ static enum ml_worker_result ml_worker_create_new_model(ml_worker_t *worker, ml_

1270 }

1271

1272 static enum ml_worker_result ml_worker_add_existing_model(ml_worker_t *worker, ml_request_add_existing_model_t req) {

1222 - UNUSED(worker);

1223 - UNUSED(req);

1224 -

1273 AcquiredDimension AcqDim(req.DLI);

1274

1275 if (!AcqDim.acquired()) {

@@ -1234,20 +1282,55 @@ static enum ml_worker_result ml_worker_add_existing_model(ml_worker_t *worker, m

1282 return ML_WORKER_RESULT_OK;

1283 }

1284

1237 - // Check if training is in progress and skip if so to avoid race condition

1285 + ml_host_t *host = (ml_host_t *) Dim->rd->rrdset->rrdhost->ml_host;

1286 + if (!host || !host->ml_running) {

1287 + pulse_ml_models_ignored();

1288 + return ML_WORKER_RESULT_OK;

1289 + }

1290 +

1291 spinlock_lock(&Dim->slock);

1292 +

1293 + // Loop detection: skip if we already have this exact model.

1294 + // The (after, before) pair uniquely identifies a model per dimension and is

1295 + // preserved across hops, so a model that loops back is detected as a duplicate.

1296 + for (const auto &km : Dim->km_contexts) {

1297 + if (km.after == req.inlined_km.after && km.before == req.inlined_km.before) {

1298 + spinlock_unlock(&Dim->slock);

1299 + pulse_ml_models_ignored();

1300 + return ML_WORKER_RESULT_OK;

1301 + }

1302 + }

1303 +

1304 + // Reject models that are not newer than the newest accepted model. This

1305 + // prevents an older model from being re-accepted after it has been evicted

1306 + // from km_contexts and later loops back from downstream.

1307 + if (!Dim->km_contexts.empty()) {

1308 + const auto &latest_km = Dim->km_contexts.back();

1309 + if (req.inlined_km.before <= latest_km.before) {

1310 + spinlock_unlock(&Dim->slock);

1311 + pulse_ml_models_ignored();

1312 + return ML_WORKER_RESULT_OK;

1313 + }

1314 + }

1315 +

1316 + // Skip if training is in progress to avoid race condition.

1317 if (Dim->training_in_progress) {

1318 spinlock_unlock(&Dim->slock);

1319 pulse_ml_models_ignored();

1320 return ML_WORKER_RESULT_OK;

1321 }

1244 - spinlock_unlock(&Dim->slock);

1322

1246 - // Safe without Dim->slock: per-host work is serialized through a single worker queue,

1247 - // and stop/reset no longer writes Dim->kmeans from non-worker threads.

1323 + // Stage the incoming kmeans into the dim's working buffer; the actual

1324 + // install into km_contexts and the has_received_downstream_model flag-set

1325 + // happen inside ml_dimension_update_models() under the same slock as the

1326 + // publish-check, so a concurrent ml_host_stop() either commits both or

1327 + // cancels both.

1328 Dim->kmeans = req.inlined_km;

1249 - ml_dimension_update_models(worker, Dim);

1250 - pulse_ml_models_received();

1329 + uint32_t generation = Dim->reset_generation;

1330 + spinlock_unlock(&Dim->slock);

1331 + if (ml_dimension_update_models(worker, Dim, generation, /*from_downstream=*/true))

1332 + pulse_ml_models_received();

1333 +

1334 return ML_WORKER_RESULT_OK;

1335 }

1336

@@ -1296,7 +1379,7 @@ void ml_train_main(void *arg) {

1379 switch (item.type) {

1380 case ML_QUEUE_ITEM_TYPE_CREATE_NEW_MODEL: {

1381 worker_res = ml_worker_create_new_model(worker, item.create_new_model);

1299 - if (worker_res != ML_WORKER_RESULT_NULL_ACQUIRED_DIMENSION) {

1382 + if (ml_should_requeue_create_new_model(worker_res)) {

1383 ml_queue_push(worker->queue, item);

1384 }

1385 break;

@@ -1346,6 +1429,12 @@ void ml_train_main(void *arg) {

1429 case ML_WORKER_RESULT_CHART_UNDER_REPLICATION:

1430 loop_stats.item_result_chart_under_replication = 1;

1431 break;

1432 + case ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED:

1433 + loop_stats.item_result_ok = 1;

1434 + break;

1435 + case ML_WORKER_RESULT_TRAINING_IN_PROGRESS:

1436 + loop_stats.item_result_ok = 1;

1437 + break;

1438 }

1439

1440 netdata_mutex_lock(&worker->nd_mutex);

src/ml/ml_dimension.h

@@ -17,7 +17,10 @@ struct ml_dimension_t {

17 SPINLOCK slock;

18 uint32_t suppression_window_counter;

19 uint32_t suppression_anomaly_counter;

20 + uint32_t reset_generation;

21 bool training_in_progress;

22 + bool has_received_downstream_model;

23 + bool create_new_model_queued;

24 size_t cns_head;

26 std::vector<calculated_number_t> cns;

src/ml/ml_enums.cc

@@ -57,6 +57,10 @@ ml_worker_result_to_string(enum ml_worker_result tr)

57 return "null-acquired-dim";

58 case ML_WORKER_RESULT_CHART_UNDER_REPLICATION:

59 return "chart-under-replication";

60 + case ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED:

61 + return "downstream-model-supplied";

62 + case ML_WORKER_RESULT_TRAINING_IN_PROGRESS:

63 + return "training-in-progress";

64 default:

65 return "unknown";

66 }

src/ml/ml_enums.h

@@ -51,6 +51,13 @@ enum ml_worker_result {

52 // Chart is under replication

53 ML_WORKER_RESULT_CHART_UNDER_REPLICATION,

54 +

55 + // This dimension is now supplied by downstream ML models; stop local requeueing

56 + ML_WORKER_RESULT_DOWNSTREAM_MODEL_SUPPLIED,

57 +

58 + // Another worker is already training this dimension; the item is requeued

59 + // so the dim stays in the periodic retrain cycle.

60 + ML_WORKER_RESULT_TRAINING_IN_PROGRESS,

61 };

63 const char *ml_worker_result_to_string(enum ml_worker_result tr);

src/ml/ml_private.h

+10

@@ -11,6 +11,16 @@

11 void ml_train_main(void *arg);

12 void ml_detect_main(void *arg);

14 +bool ml_dimension_train_model_precheck(enum ml_metric_type mt,

15 + bool has_received_downstream_model,

16 + bool training_in_progress,

17 + enum ml_worker_result *worker_res);

18 +bool ml_should_requeue_create_new_model(enum ml_worker_result worker_res);

19 +bool ml_should_publish_model_update(bool host_running,

20 + uint32_t current_generation,

21 + uint32_t expected_generation,

22 + bool *training_in_progress);

23 +

24 extern sqlite3 *ml_db;

25 extern const char *db_models_create_table;

src/ml/ml_public.cc

+92 -19

@@ -8,6 +8,50 @@

9 #define ML_METADATA_VERSION 2

11 +static void ml_host_clear_context_anomaly_rate(ml_host_t *host)

12 +{

13 + spinlock_lock(&host->context_anomaly_rate_spinlock);

14 +

15 + for (auto &entry : host->context_anomaly_rate)

16 + string_freez(entry.first);

17 +

18 + host->context_anomaly_rate.clear();

19 +

20 + spinlock_unlock(&host->context_anomaly_rate_spinlock);

21 +}

22 +

23 +static void ml_dimension_enqueue_create_model(RRDHOST *rh, RRDDIM *rd)

24 +{

25 + ml_host_t *host = (ml_host_t *) rh->ml_host;

26 + if (!host)

27 + return;

28 +

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

30 + if (!dim)

31 + return;

32 +

33 + spinlock_lock(&dim->slock);

34 + bool should_enqueue = !dim->create_new_model_queued &&

35 + dim->ts == TRAINING_STATUS_UNTRAINED &&

36 + (!dim->has_received_downstream_model || dim->km_contexts.empty());

37 + if (should_enqueue)

38 + dim->create_new_model_queued = true;

39 + spinlock_unlock(&dim->slock);

40 +

41 + if (!should_enqueue)

42 + return;

43 +

44 + ml_queue_item_t item;

45 + item.type = ML_QUEUE_ITEM_TYPE_CREATE_NEW_MODEL;

46 + item.create_new_model.DLI = DimensionLookupInfo(

47 + &rh->machine_guid[0],

48 + rd->rrdset->id,

49 + rd->id

50 + );

51 +

52 + ml_queue_push(host->queue, item);

53 +}

54 +

55 bool ml_capable()

56 {

57 return true;

@@ -60,6 +104,7 @@ void ml_host_delete(RRDHOST *rh)

104 if (!host)

105 return;

106

107 + ml_host_clear_context_anomaly_rate(host);

108 netdata_mutex_destroy(&host->mutex);

109

110 delete host;

@@ -71,7 +116,33 @@ void ml_host_start(RRDHOST *rh) {

116 if (!host)

117 return;

118

119 + // Set ml_running and run the sweep under host->mutex so concurrent

120 + // ml_host_start() calls are serialized and the visibility window of the

121 + // flag flip is bounded by the same critical section that performs the

122 + // sweep.

123 + netdata_mutex_lock(&host->mutex);

124 +

125 + if (host->ml_running) {

126 + netdata_mutex_unlock(&host->mutex);

127 + return;

128 + }

129 +

130 host->ml_running = true;

131 +

132 + void *rsp = NULL;

133 + rrdset_foreach_read(rsp, host->rh) {

134 + RRDSET *rs = static_cast<RRDSET *>(rsp);

135 +

136 + void *rdp = NULL;

137 + rrddim_foreach_read(rdp, rs) {

138 + RRDDIM *rd = static_cast<RRDDIM *>(rdp);

139 + ml_dimension_enqueue_create_model(rh, rd);

140 + }

141 + rrddim_foreach_done(rdp);

142 + }

143 + rrdset_foreach_done(rsp);

144 +

145 + netdata_mutex_unlock(&host->mutex);

146 }

147

148 void ml_host_stop(RRDHOST *rh) {

@@ -84,8 +155,14 @@ void ml_host_stop(RRDHOST *rh) {

155

156 netdata_mutex_lock(&host->mutex);

157

158 + // Re-assert under the mutex so stop deterministically wins over a racing

159 + // ml_host_start() that may have flipped the flag back to true after our

160 + // early write but before we acquired the mutex.

161 + host->ml_running = false;

162 +

163 // reset host stats

164 host->mls = ml_machine_learning_stats_t();

165 + ml_host_clear_context_anomaly_rate(host);

166

167 // reset charts/dims

168 void *rsp = NULL;

@@ -117,6 +194,10 @@ void ml_host_stop(RRDHOST *rh) {

194 dim->cns.clear();

195 dim->cns_head = 0;

196 dim->km_contexts.clear();

197 + dim->has_received_downstream_model = false;

198 + // create_new_model_queued not reset here: stop does not drain the

199 + // worker queue, so pending CREATE_NEW_MODEL items remain valid.

200 + dim->reset_generation++;

201

202 spinlock_unlock(&dim->slock);

203 }

@@ -204,7 +285,7 @@ bool ml_host_running(RRDHOST *rh) {

285 if(!host)

286 return false;

287

207 - return true;

288 + return host->ml_running;

289 }

290

291 void ml_host_get_models(RRDHOST *rh, BUFFER *wb)

@@ -274,6 +355,9 @@ void ml_dimension_new(RRDDIM *rd)

355 dim->suppression_anomaly_counter = 0;

356 dim->suppression_window_counter = 0;

357 dim->training_in_progress = false;

358 + dim->has_received_downstream_model = false;

359 + dim->create_new_model_queued = false;

360 + dim->reset_generation = 0;

361 dim->cns_head = 0;

362

363 ml_kmeans_init(&dim->kmeans);

@@ -291,24 +375,13 @@ void ml_dimension_new(RRDDIM *rd)

375

376 metaqueue_ml_load_models(rd);

377

294 - // add to worker queue

295 - {

296 - RRDHOST *rh = rd->rrdset->rrdhost;

297 - ml_host_t *host = (ml_host_t *) rh->ml_host;

298 -

299 - ml_queue_item_t item;

300 - item.type = ML_QUEUE_ITEM_TYPE_CREATE_NEW_MODEL;

301 -

302 - ml_request_create_new_model_t req;

303 - req.DLI = DimensionLookupInfo(

304 - &rh->machine_guid[0],

305 - rd->rrdset->id,

306 - rd->id

307 - );

308 - item.create_new_model = req;

309 -

310 - ml_queue_push(host->queue, item);

311 - }

378 + // Only enqueue once ml is running for this host. Otherwise, ml_host_start()

379 + // will sweep all untrained dimensions and enqueue them when it runs.

380 + // This avoids double-enqueueing the same dim from both paths.

381 + RRDHOST *rh = rd->rrdset->rrdhost;

382 + ml_host_t *host = (ml_host_t *) rh->ml_host;

383 + if (host && host->ml_running)

384 + ml_dimension_enqueue_create_model(rh, rd);

385 }

386

387 void ml_dimension_delete(RRDDIM *rd)