@cryptotaxi247 / netdata-1 / commits / f5f237e27

Optimize ML prediction hot path: circular buffer and reduce allocations (#22042)

* Refactor anomaly detection: replace rotating vector logic with circular buffer, optimize feature preprocessing, and improve memory management. * Fix compilation issues * Improve ML preprocessing and k-means error handling - Add null checks for `preprocessed_features` in preprocessing and k-means training. - Avoid crashes in `ml_kmeans_train` when features are uninitialized. - Fix logical error in value comparison during circular buffer update. * Fix circular buffer indexing in ML value comparison - Resolve logical error in determining the newest index for value comparison during circular buffer updates. * address review : Add null check and error log for `preprocessed_features` in ML preprocessing * Fix value comparison logic in circular buffer for ML dimensions - Compare incoming value against the newest sample instead of the oldest. - Prevent misclassification of constant vs variable dimensions during training and statistics reporting. - Ensure reliable detection of transitions in changing series. * Refactor ML preprocessing and k-means logic to simplify interfaces and improve memory safety - Remove redundant pointer-based management of preprocessed features, replacing it with `std::vector` references for cleaner and safer memory handling. - Update `ml_features_preprocess` and `ml_kmeans_train` to use direct vector references. - Adjust all related test cases and logic to align with the refactored interfaces. - Add unit test to validate correct comparison of the newest sample in `ml_dimension_predict`. * Add unit test and logic to ensure smooth_n=0 behaves like smooth_n=1 in ML preprocessing - Introduced `ml_effective_smooth_n` to handle `smooth_n=0` as equivalent to `smooth_n=1`. - Updated smoothing logic for consistency across preprocessing and prediction. - Added comprehensive unit test `test_features_zero_smooth_matches_one` to validate feature consistency for `smooth_n=0` and `smooth_n=1`. * Fix documentation comments in ML dimension value comparison logic - Clarify reasoning behind comparing against the newest sample in the circular buffer. - Simplify and standardize comments describing training skip logic and statistics reporting. * Normalize `smooth_n=0` handling for ML preprocessing and prediction - Treat `smooth_n=0` as equivalent to a smoothing window of 1 across all paths. - Update signature of `ml_features_preprocess_predict` to use `DSample` references instead of pointers for improved safety. - Adjust tests, comments, and logic to reflect updated `smooth_n=0` behavior and interface changes. * Fix circular buffer rebuild logic in ML smoothing window updates - Ensure buffer clears and resets when smoothing window changes to maintain state consistency. - Replace `assert` with `fatal_assert` for static buffer size validation during prediction. * Refine circular buffer rebuild logic in ML smoothing - Add `cns_size` optimization to reduce redundant size checks. - Clarify logic for detecting effective window changes and resetting buffer state. - Improve comments for consistency in ring buffer handling. * Clarify circular buffer reset conditions for ML smoothing window changes * Simplify ML preprocessing prediction logic by unifying feature extraction paths - Replace manual feature extraction logic in unit tests with `ml_features_preprocess_predict`. - Add input validation via `ml_validate_features_input` to both training and prediction paths. - Ensure consistent buffer validation and minimum sample checks across preprocessing functions. * Address review comments * Address build failure with -std=gnu++14

Stelios Fragkakis committed Mar 30, 2026 at 10:42 UTC f5f237e27e137e17e6e906ceea88a9ea52ff59ba

8 files changed +259 -132

src/ml/ml-unittest.cc

+120 -81

@@ -47,8 +47,7 @@ static void test_features_diff()

47 std::vector<DSample> pf;

48 ml_features_t features = {

49 1, 1, 1, // diff_n=1, smooth_n=1, lag_n=1

50 - dst, n, src, n,

51 - pf

50 + dst, n, src, n

51 };

53 // ml_features_preprocess calls diff, smooth, lag in sequence.

@@ -62,7 +61,7 @@ static void test_features_diff()

61 // feature vectors: [src[i], src[i+1]] for i in 0..6

62 // n_vectors = 9 - 1 - 1 + 1 - 1 = 7

65 - ml_features_preprocess(&features, 1.0);

64 + ml_features_preprocess(&features, pf, 1.0);

66 ML_TEST_ASSERT(pf.size() == 7, "lag should produce 7 feature vectors");

67 if (pf.size() >= 1) {

@@ -89,11 +88,10 @@ static void test_features_no_diff()

88 std::vector<DSample> pf;

89 ml_features_t features = {

90 0, 1, 1, // diff_n=0, smooth_n=1, lag_n=1

92 - dst, n, src, n,

93 - pf

91 + dst, n, src, n

92 };

96 - ml_features_preprocess(&features, 1.0);

94 + ml_features_preprocess(&features, pf, 1.0);

96 // With diff_n=0, smooth_n=1 (no-op), lag_n=1:

97 // n_vectors = 6 - 0 - 1 + 1 - 1 = 5

@@ -121,11 +119,10 @@ static void test_features_smooth()

119 std::vector<DSample> pf;

120 ml_features_t features = {

121 0, 3, 1, // diff_n=0, smooth_n=3, lag_n=1

124 - dst, n, src, n,

125 - pf

122 + dst, n, src, n

123 };

124

128 - ml_features_preprocess(&features, 1.0);

125 + ml_features_preprocess(&features, pf, 1.0);

126

127 // With diff_n=0: no diff

128 // Smooth with smooth_n=3, operating on src_n - diff_n = 9 elements:

@@ -151,6 +148,61 @@ static void test_features_smooth()

148 }

149 }

150

151 +// Test: smooth_n=0 is normalized to the same effective smoothing window as

152 +// smooth_n=1, for both training and prediction.

153 +static void test_features_zero_smooth_matches_one()

154 +{

155 + fprintf(stderr, " test_features_zero_smooth_matches_one...\n");

156 +

157 + const size_t n = 6;

158 + calculated_number_t input[16] = {10, 20, 30, 40, 50, 60};

159 +

160 + calculated_number_t src0[16], dst0[16];

161 + memcpy(src0, input, n * sizeof(calculated_number_t));

162 + memcpy(dst0, input, n * sizeof(calculated_number_t));

163 + std::vector<DSample> pf0;

164 + ml_features_t features0 = {

165 + 0, 0, 1,

166 + dst0, n, src0, n

167 + };

168 + ml_features_preprocess(&features0, pf0, 1.0);

169 +

170 + calculated_number_t src1[16], dst1[16];

171 + memcpy(src1, input, n * sizeof(calculated_number_t));

172 + memcpy(dst1, input, n * sizeof(calculated_number_t));

173 + std::vector<DSample> pf1;

174 + ml_features_t features1 = {

175 + 0, 1, 1,

176 + dst1, n, src1, n

177 + };

178 + ml_features_preprocess(&features1, pf1, 1.0);

179 +

180 + ML_TEST_ASSERT(pf0.size() == pf1.size(), "smooth_n=0 and smooth_n=1 should produce the same number of vectors");

181 + for (size_t i = 0; i < pf0.size() && i < pf1.size(); i++) {

182 + for (long j = 0; j < pf0[i].size(); j++) {

183 + char msg[128];

184 + snprintf(msg, sizeof(msg), "smooth_n=0 should match smooth_n=1 at feature[%zu](%ld)", i, j);

185 + ML_TEST_ASSERT_DOUBLE_EQ(pf0[i](j), pf1[i](j), 1e-12, msg);

186 + }

187 + }

188 +

189 + DSample sample0, sample1;

190 + memcpy(src0, input, n * sizeof(calculated_number_t));

191 + memcpy(dst0, input, n * sizeof(calculated_number_t));

192 + ml_features_preprocess_predict(&features0, sample0);

193 +

194 + memcpy(src1, input, n * sizeof(calculated_number_t));

195 + memcpy(dst1, input, n * sizeof(calculated_number_t));

196 + ml_features_preprocess_predict(&features1, sample1);

197 +

198 + ML_TEST_ASSERT(sample0.size() == sample1.size(), "prediction sample size should match for smooth_n=0 and smooth_n=1");

199 + for (size_t i = 0; i < features0.lag_n + 1; i++) {

200 + char msg[128];

201 + snprintf(msg, sizeof(msg), "prediction smooth_n=0 should match smooth_n=1 at sample(%zu)", i);

202 + ML_TEST_ASSERT_DOUBLE_EQ(sample0(i), sample1(i), 1e-12, msg);

203 + }

204 +}

205 +

206 // Test: full pipeline with default-like params (diff_n=1, smooth_n=3, lag_n=5)

207 // Validates the feature vector shape and that a round-trip through

208 // train + score produces sensible anomaly scores.

@@ -185,10 +237,9 @@ static void test_full_pipeline()

237 std::vector<DSample> pf;

238 ml_features_t features = {

239 diff_n, smooth_n, lag_n,

188 - dst, n, src, n,

189 - pf

240 + dst, n, src, n

241 };

191 - ml_features_preprocess(&features, 1.0);

242 + ml_features_preprocess(&features, pf, 1.0);

243

244 // With these params:

245 // n_vectors = n - diff_n - smooth_n + 1 - lag_n = 9 - 1 - 3 + 1 - 5 = 1

@@ -208,15 +259,10 @@ static void test_full_pipeline()

259

260 // Train a kmeans model on the normal data

261 std::vector<DSample> training_features = std::move(all_features);

211 - ml_features_t train_ft = {

212 - diff_n, smooth_n, lag_n,

213 - nullptr, 0, nullptr, 0,

214 - training_features

215 - };

262

263 ml_kmeans_t kmeans;

264 ml_kmeans_init(&kmeans);

219 - ml_kmeans_train(&kmeans, &train_ft, 1000, 0, 100);

265 + ml_kmeans_train(&kmeans, training_features, 1000, 0, 100);

266

267 ML_TEST_ASSERT(kmeans.cluster_centers.size() == 2, "kmeans should have 2 cluster centers");

268 ML_TEST_ASSERT(kmeans.min_dist < kmeans.max_dist, "min_dist < max_dist after training");

@@ -246,10 +292,9 @@ static void test_full_pipeline()

292 std::vector<DSample> pf;

293 ml_features_t features = {

294 diff_n, smooth_n, lag_n,

249 - dst, n, src, n,

250 - pf

295 + dst, n, src, n

296 };

252 - ml_features_preprocess(&features, 1.0);

297 + ml_features_preprocess(&features, pf, 1.0);

298

299 if (pf.size() >= 1) {

300 calculated_number_t anomaly_score = ml_kmeans_anomaly_score(&inlined_km, pf[0]);

@@ -357,10 +402,9 @@ static void test_circular_buffer_equivalence()

402 std::vector<DSample> pf;

403 ml_features_t features = {

404 diff_n, smooth_n, lag_n,

360 - dst, n, src, n,

361 - pf

405 + dst, n, src, n

406 };

363 - ml_features_preprocess(&features, 1.0);

407 + ml_features_preprocess(&features, pf, 1.0);

408

409 if (pf.size() >= 1)

410 rotate_results.push_back(pf[0]);

@@ -391,10 +435,9 @@ static void test_circular_buffer_equivalence()

435 std::vector<DSample> pf;

436 ml_features_t features = {

437 diff_n, smooth_n, lag_n,

394 - dst, n, src, n,

395 - pf

438 + dst, n, src, n

439 };

397 - ml_features_preprocess(&features, 1.0);

440 + ml_features_preprocess(&features, pf, 1.0);

441

442 if (pf.size() >= 1)

443 circ_results.push_back(pf[0]);

@@ -412,6 +455,33 @@ static void test_circular_buffer_equivalence()

455 }

456 }

457

458 +// Test: same_value must compare against the previous newest sample, not the

459 +// oldest slot being overwritten. This locks in the intentional semantic change

460 +// in ml_dimension_predict().

461 +static void test_same_value_uses_newest_sample()

462 +{

463 + fprintf(stderr, " test_same_value_uses_newest_sample...\n");

464 +

465 + const size_t n = 5;

466 + std::vector<calculated_number_t> cns = {7.0, 2.0, 3.0, 4.0, 5.0};

467 + size_t cns_head = 0;

468 + calculated_number_t incoming = 7.0;

469 +

470 + // Circular buffer state:

471 + // oldest slot being overwritten = cns[cns_head] = 7.0

472 + // previous newest sample = cns[(cns_head + n - 1) % n] = 5.0

473 + // If we compared against the oldest slot, same_value would be true and we'd

474 + // miss the transition from 5.0 -> 7.0. Comparing against newest is correct.

475 + bool old_rotate_equivalent = (cns[cns_head] == incoming);

476 + size_t newest_idx = (cns_head + n - 1) % n;

477 + bool new_ring_semantics = (cns[newest_idx] == incoming);

478 +

479 + ML_TEST_ASSERT(old_rotate_equivalent,

480 + "oldest-slot comparison should report same_value for this edge case");

481 + ML_TEST_ASSERT(!new_ring_semantics,

482 + "newest-sample comparison should detect the changed incoming value");

483 +}

484 +

485 // Test: ml_features_preprocess with a prediction-sized window produces the same

486 // feature vector as a manual reimplementation of diff + smooth + extract.

487 // This validates the preprocessing math and serves as a baseline for verifying

@@ -468,10 +538,9 @@ static void test_preprocess_predict_equivalence()

538 std::vector<DSample> pf;

539 ml_features_t features1 = {

540 diff_n, smooth_n, lag_n,

471 - dst1, n, src1, n,

472 - pf

541 + dst1, n, src1, n

542 };

474 - ml_features_preprocess(&features1, 1.0);

543 + ml_features_preprocess(&features1, pf, 1.0);

544

545 // With prediction-sized window: n_vectors = n - diff_n - smooth_n + 1 - lag_n = 1

546 char msg[256];

@@ -480,53 +549,25 @@ static void test_preprocess_predict_equivalence()

549 ML_TEST_ASSERT(pf.size() == 1, msg);

550 if (pf.size() != 1) continue;

551

483 - // Path 2: manual extraction matching what ml_features_preprocess_predict does:

484 - // diff + smooth, then read first lag_n+1 values from src.

485 - // This validates the logic without depending on the branch function existing.

552 + // Path 2: ml_features_preprocess_predict should produce the same

553 + // prediction-sized feature vector as the training preprocess path.

554 calculated_number_t src2[128], dst2[128];

555 memset(src2, 0, sizeof(src2));

556 memcpy(src2, input, n * sizeof(calculated_number_t));

557 memcpy(dst2, src2, n * sizeof(calculated_number_t));

558

491 - // Replicate diff

492 - if (diff_n > 0) {

493 - for (size_t idx = 0; idx != (n - diff_n); idx++) {

494 - size_t high = (n - 1) - idx;

495 - size_t low = high - diff_n;

496 - dst2[low] = src2[high] - src2[low];

497 - }

498 - memcpy(src2, dst2, (n - diff_n) * sizeof(calculated_number_t));

499 - for (size_t idx = n - diff_n; idx != n; idx++)

500 - src2[idx] = 0.0;

501 - }

502 -

503 - // Replicate smooth

504 - {

505 - calculated_number_t sum = 0.0;

506 - size_t idx = 0;

507 - for (; idx != smooth_n - 1; idx++)

508 - sum += src2[idx];

509 - for (; idx != (n - diff_n); idx++) {

510 - sum += src2[idx];

511 - calculated_number_t prev = src2[idx - (smooth_n - 1)];

512 - src2[idx - (smooth_n - 1)] = sum / smooth_n;

513 - sum -= prev;

514 - }

515 - for (idx = 0; idx != smooth_n; idx++)

516 - src2[(n - 1) - idx] = 0.0;

517 - }

518 -

519 - // Extract feature: first lag_n+1 values (what preprocess_predict does)

520 - DSample direct_feature;

521 - direct_feature.set_size(lag_n + 1);

522 - for (size_t fi = 0; fi != lag_n + 1; fi++)

523 - direct_feature(fi) = src2[fi];

559 + ml_features_t features2 = {

560 + diff_n, smooth_n, lag_n,

561 + dst2, n, src2, n

562 + };

563 + DSample predicted_feature;

564 + ml_features_preprocess_predict(&features2, predicted_feature);

565

525 - // Compare: pf[0] from full pipeline must match direct extraction

566 + // Compare: pf[0] from full pipeline must match the direct prediction path.

567 for (size_t fi = 0; fi < lag_n + 1; fi++) {

527 - snprintf(msg, sizeof(msg), "params(%zu,%zu,%zu) %s: feature[%zu] preprocess vs direct",

568 + snprintf(msg, sizeof(msg), "params(%zu,%zu,%zu) %s: feature[%zu] preprocess vs predict",

569 diff_n, smooth_n, lag_n, filler_names[f], fi);

529 - ML_TEST_ASSERT_DOUBLE_EQ(pf[0](fi), direct_feature(fi), 1e-12, msg);

570 + ML_TEST_ASSERT_DOUBLE_EQ(pf[0](fi), predicted_feature(fi), 1e-12, msg);

571 }

572 }

573 }

@@ -551,10 +592,9 @@ static void test_constant_input()

592 std::vector<DSample> pf;

593 ml_features_t features = {

594 diff_n, smooth_n, lag_n,

554 - dst, n, src, n,

555 - pf

595 + dst, n, src, n

596 };

557 - ml_features_preprocess(&features, 1.0);

597 + ml_features_preprocess(&features, pf, 1.0);

598

599 ML_TEST_ASSERT(pf.size() == 1, "constant input should produce 1 feature vector");

600

@@ -577,10 +617,9 @@ static void test_constant_input()

617 std::vector<DSample> pf2;

618 ml_features_t features2 = {

619 0, smooth_n, lag_n,

580 - dst2, n, src2, n,

581 - pf2

620 + dst2, n, src2, n

621 };

583 - ml_features_preprocess(&features2, 1.0);

622 + ml_features_preprocess(&features2, pf2, 1.0);

623

624 // With diff_n=0, smooth on constant values gives the same constant.

625 // Feature vector should be all 42.0.

@@ -670,10 +709,9 @@ static void test_parameter_combinations()

709 std::vector<DSample> pf;

710 ml_features_t features = {

711 diff_n, smooth_n, lag_n,

673 - dst, n, src, n,

674 - pf

712 + dst, n, src, n

713 };

676 - ml_features_preprocess(&features, 1.0);

714 + ml_features_preprocess(&features, pf, 1.0);

715

716 char msg[256];

717 snprintf(msg, sizeof(msg), "params(%zu,%zu,%zu): expected %zu vectors, got %zu",

@@ -708,10 +746,9 @@ static void test_parameter_combinations()

746 std::vector<DSample> pf_large;

747 ml_features_t features_large = {

748 diff_n, smooth_n, lag_n,

711 - dst_large, large_n, src_large, large_n,

712 - pf_large

749 + dst_large, large_n, src_large, large_n

750 };

714 - ml_features_preprocess(&features_large, 1.0);

751 + ml_features_preprocess(&features_large, pf_large, 1.0);

752

753 size_t expected_large = large_n - diff_n - smooth_n + 1 - lag_n;

754 snprintf(msg, sizeof(msg), "params(%zu,%zu,%zu) large window: expected %zu vectors",

@@ -736,9 +773,11 @@ extern "C" int ml_unittest()

773 test_features_diff();

774 test_features_no_diff();

775 test_features_smooth();

776 + test_features_zero_smooth_matches_one();

777 test_kmeans_scoring();

778 test_full_pipeline();

779 test_circular_buffer_equivalence();

780 + test_same_value_uses_newest_sample();

781 test_preprocess_predict_equivalence();

782 test_constant_input();

783 test_parameter_combinations();

src/ml/ml.cc

+73 -26

@@ -28,6 +28,17 @@ static void __attribute__((destructor)) destroy_mutex(void) {

28 netdata_mutex_destroy(&db_mutex);

29 }

31 +static inline size_t ml_dimension_smoothing_window(const ml_dimension_t *dim)

32 +{

33 + unsigned chart_update_every = dim->rd->rrdset->update_every;

34 + if (chart_update_every > nd_profile.update_every)

35 + return 1;

36 +

37 + // max_samples_to_smooth == 0 is normalized to an effective smoothing window

38 + // of 1 for feature extraction.

39 + return std::max<size_t>(Cfg.max_samples_to_smooth, 1);

40 +}

41 +

42 typedef struct {

43 // First/last entry of the dimension in DB when generating the response

44 time_t first_entry_on_response;

@@ -57,7 +68,7 @@ ml_dimension_calculated_numbers(ml_worker_t *worker, ml_dimension_t *dim)

68 training_response.last_entry_on_response = rrddim_last_entry_s_of_tier(dim->rd, 0);

70 unsigned chart_update_every = dim->rd->rrdset->update_every;

60 - size_t smoothing_window = (chart_update_every > nd_profile.update_every) ? 1 : Cfg.max_samples_to_smooth;

71 + size_t smoothing_window = ml_dimension_smoothing_window(dim);

72 size_t min_required_samples = Cfg.diff_n + smoothing_window + Cfg.lag_n;

74 auto round_up_div = [](time_t window, unsigned step) -> size_t {

@@ -724,13 +735,12 @@ ml_dimension_train_model(ml_worker_t *worker, ml_dimension_t *dim)

735 memcpy(worker->scratch_training_cns, worker->training_cns,

736 training_response.total_values * sizeof(calculated_number_t));

737

727 - size_t smoothing_window = (dim->rd->rrdset->update_every > nd_profile.update_every) ? 1 : Cfg.max_samples_to_smooth;

738 + size_t smoothing_window = ml_dimension_smoothing_window(dim);

739

740 ml_features_t features = {

741 Cfg.diff_n, smoothing_window, Cfg.lag_n,

742 worker->scratch_training_cns, training_response.total_values,

732 - worker->training_cns, training_response.total_values,

733 - worker->training_samples

743 + worker->training_cns, training_response.total_values

744 };

745

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

@@ -746,10 +756,10 @@ ml_dimension_train_model(ml_worker_t *worker, ml_dimension_t *dim)

756 }

757

758 // Apply sampling during lag feature extraction

749 - ml_features_preprocess(&features, sampling_ratio);

759 + ml_features_preprocess(&features, worker->training_samples, sampling_ratio);

760

761 ml_kmeans_init(&dim->kmeans);

752 - ml_kmeans_train(&dim->kmeans, &features, Cfg.max_kmeans_iters, training_response.query_after_t, training_response.query_before_t);

762 + ml_kmeans_train(&dim->kmeans, worker->training_samples, Cfg.max_kmeans_iters, training_response.query_after_t, training_response.query_before_t);

763 }

764

765 // update models

@@ -772,42 +782,79 @@ ml_dimension_predict(ml_dimension_t *dim, calculated_number_t value, bool exists

782 // Don't treat values that don't exist as anomalous

783 if (!exists) {

784 dim->cns.clear();

785 + dim->cns_head = 0;

786 spinlock_unlock(&dim->slock);

787 return false;

788 }

789

790 // Save the value and return if we don't have enough values for a sample

780 - unsigned n = Cfg.diff_n + Cfg.max_samples_to_smooth + Cfg.lag_n;

781 - if (dim->cns.size() < n) {

791 + size_t smoothing_window = ml_dimension_smoothing_window(dim);

792 + unsigned n = Cfg.diff_n + smoothing_window + Cfg.lag_n;

793 +

794 + size_t cns_size = dim->cns.size();

795 +

796 + // The ring buffer modulus is derived from the current effective smoothing

797 + // window. When the effective window changes, the existing history is only

798 + // reusable if it is still a linear chronological prefix, which in this

799 + // representation means cns_head == 0. Wrapped ring state from the old

800 + // modulus must be discarded before warmup/indexing/linearization continue.

801 + bool invalid_head = (cns_size > 0 && dim->cns_head >= cns_size);

802 + bool size_changed_with_wrapped_state = (cns_size != n && dim->cns_head != 0);

803 + bool shrunk_below_existing_history = (cns_size > n);

804 + if (invalid_head || size_changed_with_wrapped_state || shrunk_below_existing_history) {

805 + dim->cns.clear();

806 + dim->cns_head = 0;

807 + cns_size = 0;

808 + }

809 +

810 + if (cns_size < n) {

811 dim->cns.push_back(value);

812 spinlock_unlock(&dim->slock);

813 return false;

814 }

815

787 - // Push the value and check if it's different from the last one

788 - bool same_value = true;

789 - std::rotate(std::begin(dim->cns), std::begin(dim->cns) + 1, std::end(dim->cns));

790 - if (dim->cns[n - 1] != value)

791 - same_value = false;

792 - dim->cns[n - 1] = value;

816 + // Compare incoming value against the most recent sample (newest_idx).

817 + //

818 + // The old std::rotate code compared against the oldest element being dropped — that

819 + // was a side effect of rotate mechanics, not intentional design.

820 + //

821 + // Downstream effect: when same_value is false, we set dim->mt = METRIC_TYPE_VARIABLE.

822 + // This controls two things:

823 + // 1. ml_dimension_train_model() skips training when mt == METRIC_TYPE_CONSTANT.

824 + // 2. Statistics reporting counts constant vs variable dimensions.

825 + //

826 + // Comparing against newest is safe (and more correct) because:

827 + // - For truly constant series, all elements are equal — either comparison works.

828 + // - For changing series, comparing against newest detects the transition on the

829 + // first differing tick. The old oldest-comparison could miss transitions when

830 + // the oldest element happened to equal the new value by coincidence.

831 + // - mt is reset to METRIC_TYPE_CONSTANT after each training cycle,

832 + // so a single false negative cannot cause a permanent misclassification.

833 + size_t newest_idx = (dim->cns_head + n - 1) % n;

834 + bool same_value = (dim->cns[newest_idx] == value);

835 + dim->cns[dim->cns_head] = value;

836 + dim->cns_head = (dim->cns_head + 1) % n;

837

838 // Create the sample

795 - assert((n * (Cfg.lag_n + 1) <= 128) &&

796 - "Static buffers too small to perform prediction. "

797 - "This should not be possible with the default clamping of feature extraction options");

839 calculated_number_t src_cns[128];

840 calculated_number_t dst_cns[128];

800 -

801 - memset(src_cns, 0, n * (Cfg.lag_n + 1) * sizeof(calculated_number_t));

802 - memcpy(src_cns, dim->cns.data(), n * sizeof(calculated_number_t));

803 - memcpy(dst_cns, dim->cns.data(), n * sizeof(calculated_number_t));

841 + constexpr size_t src_cns_capacity = sizeof(src_cns) / sizeof(src_cns[0]);

842 + constexpr size_t dst_cns_capacity = sizeof(dst_cns) / sizeof(dst_cns[0]);

843 + fatal_assert((n <= src_cns_capacity && n <= dst_cns_capacity) &&

844 + "Static buffers too small to perform prediction. "

845 + "This should not be possible with the default clamping of feature extraction options");

846 +

847 + size_t first_chunk = n - dim->cns_head;

848 + memcpy(src_cns, dim->cns.data() + dim->cns_head, first_chunk * sizeof(calculated_number_t));

849 + if (dim->cns_head)

850 + memcpy(src_cns + first_chunk, dim->cns.data(), dim->cns_head * sizeof(calculated_number_t));

851 + memcpy(dst_cns, src_cns, n * sizeof(calculated_number_t));

852

853 ml_features_t features = {

806 - Cfg.diff_n, Cfg.max_samples_to_smooth, Cfg.lag_n,

807 - dst_cns, n, src_cns, n,

808 - dim->feature

854 + Cfg.diff_n, smoothing_window, Cfg.lag_n,

855 + dst_cns, n, src_cns, n

856 };

810 - ml_features_preprocess(&features, 1.0);

857 + ml_features_preprocess_predict(&features, dim->feature);

858

859 // Mark the metric time as variable if we received different values

860 if (!same_value)

@@ -831,7 +878,7 @@ ml_dimension_predict(ml_dimension_t *dim, calculated_number_t value, bool exists

878 for (const auto &km_ctx : dim->km_contexts) {

879 models_consulted++;

880

834 - calculated_number_t anomaly_score = ml_kmeans_anomaly_score(&km_ctx, features.preprocessed_features[0]);

881 + calculated_number_t anomaly_score = ml_kmeans_anomaly_score(&km_ctx, dim->feature);

882 if (std::isnan(anomaly_score))

883 continue;

884

src/ml/ml_dimension.h

+2 -1

@@ -18,12 +18,13 @@ struct ml_dimension_t {

18 uint32_t suppression_window_counter;

19 uint32_t suppression_anomaly_counter;

20 bool training_in_progress;

21 + size_t cns_head;

23 std::vector<calculated_number_t> cns;

25 std::vector<ml_kmeans_inlined_t> km_contexts;

26 ml_kmeans_t kmeans;

26 - std::vector<DSample> feature;

27 + DSample feature;

28 };

30 bool

src/ml/ml_features.cc

+49 -14

@@ -3,6 +3,29 @@

3 #include "ml_config.h"

4 #include "ml_features.h"

6 +static inline size_t ml_effective_smooth_n(const ml_features_t *features)

7 +{

8 + // smooth_n == 0 is normalized to an effective window of 1, preserving the

9 + // existing feature-extraction shape without introducing a separate no-op path.

10 + return features->smooth_n == 0 ? 1 : features->smooth_n;

11 +}

12 +

13 +static inline void ml_validate_features_input(const ml_features_t *features, bool prediction_path)

14 +{

15 + size_t smooth_n = ml_effective_smooth_n(features);

16 + size_t min_required_samples = features->diff_n + smooth_n + features->lag_n;

17 +

18 + fatal_assert(features->dst_n >= features->src_n &&

19 + "ml_features: dst buffer must be at least as large as src buffer");

20 + if (prediction_path) {

21 + fatal_assert(features->src_n >= min_required_samples &&

22 + "ml_features_preprocess_predict: src buffer is smaller than diff_n + effective_smooth_n + lag_n");

23 + } else {

24 + fatal_assert(features->src_n >= min_required_samples &&

25 + "ml_features_preprocess: src buffer is smaller than diff_n + effective_smooth_n + lag_n");

26 + }

27 +}

28 +

29 static void ml_features_diff(ml_features_t *features)

30 {

31 if (features->diff_n == 0)

@@ -24,27 +47,29 @@ static void ml_features_diff(ml_features_t *features)

48 static void ml_features_smooth(ml_features_t *features)

49 {

50 + size_t smooth_n = ml_effective_smooth_n(features);

51 calculated_number_t sum = 0.0;

53 size_t idx = 0;

30 - for (; idx != features->smooth_n - 1; idx++)

54 + for (; idx != smooth_n - 1; idx++)

55 sum += features->src[idx];

57 for (; idx != (features->src_n - features->diff_n); idx++) {

58 sum += features->src[idx];

35 - calculated_number_t prev_cn = features->src[idx - (features->smooth_n - 1)];

36 - features->src[idx - (features->smooth_n - 1)] = sum / features->smooth_n;

59 + calculated_number_t prev_cn = features->src[idx - (smooth_n - 1)];

60 + features->src[idx - (smooth_n - 1)] = sum / smooth_n;

61 sum -= prev_cn;

62 }

40 - for (idx = 0; idx != features->smooth_n; idx++)

64 + for (idx = 0; idx != smooth_n; idx++)

65 features->src[(features->src_n - 1) - idx] = 0.0;

66 }

44 -static void ml_features_lag(ml_features_t *features, double sampling_ratio)

68 +static void ml_features_lag(ml_features_t *features, std::vector<DSample> &preprocessed_features, double sampling_ratio)

69 {

46 - size_t n = features->src_n - features->diff_n - features->smooth_n + 1 - features->lag_n;

47 - features->preprocessed_features.resize(n);

70 + size_t n = features->src_n - features->diff_n - ml_effective_smooth_n(features) + 1 - features->lag_n;

71 + preprocessed_features.clear();

72 + preprocessed_features.reserve(n);

74 uint32_t max_mt = std::numeric_limits<uint32_t>::max();

75 uint32_t cutoff = static_cast<double>(max_mt) * sampling_ratio;

@@ -52,24 +77,34 @@ static void ml_features_lag(ml_features_t *features, double sampling_ratio)

77 size_t sample_idx = 0;

79 for (size_t idx = 0; idx != n; idx++) {

55 - DSample &DS = features->preprocessed_features[sample_idx++];

56 - DS.set_size(features->lag_n + 1);

57 -

80 if (Cfg.random_nums[idx % Cfg.random_nums.size()] > cutoff) {

59 - sample_idx--;

81 continue;

82 }

84 + preprocessed_features.emplace_back();

85 + DSample &DS = preprocessed_features[sample_idx++];

86 + DS.set_size(features->lag_n + 1);

87 +

88 for (size_t feature_idx = 0; feature_idx != features->lag_n + 1; feature_idx++)

89 DS(feature_idx) = features->src[idx + feature_idx];

90 }

91 +}

67 - features->preprocessed_features.resize(sample_idx);

93 +void ml_features_preprocess(ml_features_t *features, std::vector<DSample> &preprocessed_features, double sampling_ratio)

94 +{

95 + ml_validate_features_input(features, false);

96 + ml_features_diff(features);

97 + ml_features_smooth(features);

98 + ml_features_lag(features, preprocessed_features, sampling_ratio);

99 }

100

70 -void ml_features_preprocess(ml_features_t *features, double sampling_ratio)

101 +void ml_features_preprocess_predict(ml_features_t *features, DSample &sample)

102 {

103 + ml_validate_features_input(features, true);

104 ml_features_diff(features);

105 ml_features_smooth(features);

74 - ml_features_lag(features, sampling_ratio);

106 +

107 + sample.set_size(features->lag_n + 1);

108 + for (size_t feature_idx = 0; feature_idx != features->lag_n + 1; feature_idx++)

109 + sample(feature_idx) = features->src[feature_idx];

110 }

src/ml/ml_features.h

+5 -3

@@ -17,10 +17,12 @@ typedef struct {

18 calculated_number_t *src;

19 size_t src_n;

20 -

21 - std::vector<DSample> &preprocessed_features;

20 } ml_features_t;

24 -void ml_features_preprocess(ml_features_t *features, double sampling_ratio);

22 +// Training path: diff + smooth + lag into preprocessed_features.

23 +void ml_features_preprocess(ml_features_t *features, std::vector<DSample> &preprocessed_features, double sampling_ratio);

24 +

25 +// Prediction path: diff + smooth, then fill a single DSample from the first lag_n+1 values.

26 +void ml_features_preprocess_predict(ml_features_t *features, DSample &sample);

28 #endif /* ML_FEATURES_H */

src/ml/ml_kmeans.cc

+7 -6

@@ -22,7 +22,7 @@ ml_kmeans_init(ml_kmeans_t *kmeans)

22 }

24 void

25 -ml_kmeans_train(ml_kmeans_t *kmeans, const ml_features_t *features, unsigned max_iters, time_t after, time_t before)

25 +ml_kmeans_train(ml_kmeans_t *kmeans, const std::vector<DSample> &preprocessed_features, unsigned max_iters, time_t after, time_t before)

26 {

27 kmeans->after = (uint32_t) after;

28 kmeans->before = (uint32_t) before;

@@ -32,8 +32,9 @@ ml_kmeans_train(ml_kmeans_t *kmeans, const ml_features_t *features, unsigned max

33 kmeans->cluster_centers.clear();

35 - if (features->preprocessed_features.size() < 2) {

36 - netdata_log_error("ml_kmeans_train: not enough features to train kmeans (size=%zu)", features->preprocessed_features.size());

35 + if (preprocessed_features.size() < 2) {

36 + netdata_log_error("ml_kmeans_train: not enough features to train kmeans (size=%zu)",

37 + preprocessed_features.size());

38 return;

39 }

@@ -43,10 +44,10 @@ ml_kmeans_train(ml_kmeans_t *kmeans, const ml_features_t *features, unsigned max

44 // causing heap-use-after-free when multiple threads train models concurrently.

45 kmeans->cluster_centers.reserve(2);

46 - dlib::pick_initial_centers(2, kmeans->cluster_centers, features->preprocessed_features);

47 - dlib::find_clusters_using_kmeans(features->preprocessed_features, kmeans->cluster_centers, max_iters);

47 + dlib::pick_initial_centers(2, kmeans->cluster_centers, preprocessed_features);

48 + dlib::find_clusters_using_kmeans(preprocessed_features, kmeans->cluster_centers, max_iters);

49 - for (const auto &preprocessed_feature : features->preprocessed_features) {

50 + for (const auto &preprocessed_feature : preprocessed_features) {

51 calculated_number_t mean_dist = 0.0;

53 for (const auto &cluster_center : kmeans->cluster_centers) {

src/ml/ml_kmeans.h

+1 -1

@@ -92,7 +92,7 @@ inline ml_kmeans_t &ml_kmeans_t::operator=(const ml_kmeans_inlined_t &inlined_km

93 void ml_kmeans_init(ml_kmeans_t *kmeans);

95 -void ml_kmeans_train(ml_kmeans_t *kmeans, const ml_features_t *features, unsigned max_iters, time_t after, time_t before);

95 +void ml_kmeans_train(ml_kmeans_t *kmeans, const std::vector<DSample> &preprocessed_features, unsigned max_iters, time_t after, time_t before);

97 calculated_number_t ml_kmeans_anomaly_score(const ml_kmeans_inlined_t *kmeans, const DSample &DS);

src/ml/ml_public.cc

@@ -115,6 +115,7 @@ void ml_host_stop(RRDHOST *rh) {

115 dim->suppression_anomaly_counter = 0;

116 dim->suppression_window_counter = 0;

117 dim->cns.clear();

118 + dim->cns_head = 0;

119 dim->km_contexts.clear();

120

121 spinlock_unlock(&dim->slock);

@@ -273,6 +274,7 @@ void ml_dimension_new(RRDDIM *rd)

274 dim->suppression_anomaly_counter = 0;

275 dim->suppression_window_counter = 0;

276 dim->training_in_progress = false;

277 + dim->cns_head = 0;

278

279 ml_kmeans_init(&dim->kmeans);

280