weights.c · cryptotaxi247/netdata · siGit Code & Deploy

Code Commits Issues Pulls Sessions CI/CD

master

c 2,701 lines 104 KB

Raw

1	// SPDX-License-Identifier: GPL-3.0-or-later
2
3	#include "database/rrd.h"
4	#include "KolmogorovSmirnovDist.h"
5
6	#define MAX_POINTS 10000
7	int metric_correlations_version = 1;
8
9	typedef struct weights_stats {
10	NETDATA_DOUBLE max_base_high_ratio;
11	size_t db_points;
12	size_t result_points;
13	size_t db_queries;
14	size_t db_points_per_tier[RRD_STORAGE_TIERS];
15	size_t binary_searches;
16	} WEIGHTS_STATS;
17
18	// ----------------------------------------------------------------------------
19	// parse and render metric correlations methods
20
21	static struct {
22	const char *name;
23	WEIGHTS_METHOD value;
24	} weights_methods[] = {
25	{ "ks2" , WEIGHTS_METHOD_MC_KS2}
26	, { "volume" , WEIGHTS_METHOD_MC_VOLUME}
27	, { "anomaly-rate" , WEIGHTS_METHOD_ANOMALY_RATE}
28	, { "value" , WEIGHTS_METHOD_VALUE}
29	, { NULL , 0 }
30	};
31
32	WEIGHTS_METHOD weights_string_to_method(const char *method) {
33	for(int i = 0; weights_methods[i].name ;i++)
34	if(strcmp(method, weights_methods[i].name) == 0)
35	return weights_methods[i].value;
36
37	return WEIGHTS_METHOD_MC_KS2;
38	}
39
40	const char *weights_method_to_string(WEIGHTS_METHOD method) {
41	for(int i = 0; weights_methods[i].name ;i++)
42	if(weights_methods[i].value == method)
43	return weights_methods[i].name;
44
45	return "ks2";
46	}
47
48	// ----------------------------------------------------------------------------
49	// The results per dimension are aggregated into a dictionary
50
51	typedef enum {
52	RESULT_IS_BASE_HIGH_RATIO = (1 << 0),
53	RESULT_IS_PERCENTAGE_OF_TIME = (1 << 1),
54	} RESULT_FLAGS;
55
56	struct register_result {
57	RESULT_FLAGS flags;
58	RRDHOST *host;
59	RRDCONTEXT_ACQUIRED *rca;
60	RRDINSTANCE_ACQUIRED *ria;
61	RRDMETRIC_ACQUIRED *rma;
62	NETDATA_DOUBLE value;
63	STORAGE_POINT highlighted;
64	STORAGE_POINT baseline;
65	usec_t duration_ut;
66	};
67
68	static DICTIONARY *register_result_init() {
69	DICTIONARY *results = dictionary_create_advanced(DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct register_result));
70	return results;
71	}
72
73	static DICTIONARY *register_result_init_single_threaded() {
74	DICTIONARY *results = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct register_result));
75	return results;
76	}
77
78	static void register_result_destroy(DICTIONARY *results) {
79	dictionary_destroy(results);
80	}
81
82	// Merge results from local dictionary into main dictionary
83	static void merge_results_dictionaries(DICTIONARY main_results, DICTIONARY local_results) {
84	if (!local_results \|\| !main_results)
85	return;
86
87	struct register_result *local_result;
88	dfe_start_read(local_results, local_result) {
89	// Try to get existing result in main dictionary
90	struct register_result *main_result = dictionary_get(main_results, local_result_dfe.name);
91	if (main_result) {
92	// Merge the results - keep the higher weight
93	if (local_result->value > main_result->value) {
94	// Create a copy with the new values and replace the entire entry
95	struct register_result merged_result = *local_result;
96	dictionary_set(main_results, local_result_dfe.name, &merged_result, sizeof(struct register_result));
97	}
98	// If local value is not higher, keep the existing main result (do nothing)
99	} else {
100	// Insert new result - copy the entire structure
101	dictionary_set(main_results, local_result_dfe.name, local_result, sizeof(struct register_result));
102	}
103	}
104	dfe_done(local_result);
105	}
106
107	// Forward declarations
108	static ssize_t weights_do_node_callback(void data, RRDHOST host, bool queryable);
109	static ssize_t weights_do_context_callback(void data, RRDCONTEXT_ACQUIRED rca, bool queryable_context);
110
111	static void register_result(DICTIONARY results, RRDHOST host, RRDCONTEXT_ACQUIRED rca, RRDINSTANCE_ACQUIRED ria,
112	RRDMETRIC_ACQUIRED *rma, NETDATA_DOUBLE value, RESULT_FLAGS flags,
113	STORAGE_POINT highlighted, STORAGE_POINT baseline, WEIGHTS_STATS *stats,
114	bool register_zero, usec_t duration_ut) {
115
116	if(!netdata_double_isnumber(value)) return;
117
118	// make it positive
119	NETDATA_DOUBLE v = fabsndd(value);
120
121	// no need to store zero scored values
122	if(unlikely(fpclassify(v) == FP_ZERO && !register_zero))
123	return;
124
125	// keep track of the max of the baseline / highlight ratio
126	if((flags & RESULT_IS_BASE_HIGH_RATIO) && v > stats->max_base_high_ratio)
127	stats->max_base_high_ratio = v;
128
129	struct register_result t = {
130	.flags = flags,
131	.host = host,
132	.rca = rca,
133	.ria = ria,
134	.rma = rma,
135	.value = v,
136	.duration_ut = duration_ut,
137	};
138
139	if(highlighted)
140	t.highlighted = *highlighted;
141
142	if(baseline)
143	t.baseline = *baseline;
144
145	// Use the original pointer address approach - revert the stable key change
146	char buf[20 + 1];
147	ssize_t len = snprintfz(buf, sizeof(buf) - 1, "%p", rma);
148	dictionary_set_advanced(results, buf, len, &t, sizeof(struct register_result), NULL);
149	}
150
151	// ----------------------------------------------------------------------------
152	// Generation of JSON output for the results
153
154	static void results_header_to_json(DICTIONARY results __maybe_unused, BUFFER wb,
155	time_t after, time_t before,
156	time_t baseline_after, time_t baseline_before,
157	size_t points, WEIGHTS_METHOD method,
158	RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
159	size_t examined_dimensions __maybe_unused, usec_t duration,
160	WEIGHTS_STATS *stats) {
161
162	buffer_json_member_add_time_t_formatted(wb, "after", after, options & RRDR_OPTION_RFC3339);
163	buffer_json_member_add_time_t_formatted(wb, "before", before, options & RRDR_OPTION_RFC3339);
164	buffer_json_member_add_time_t(wb, "duration", before - after);
165	buffer_json_member_add_uint64(wb, "points", points);
166
167	if(method == WEIGHTS_METHOD_MC_KS2 \|\| method == WEIGHTS_METHOD_MC_VOLUME) {
168	buffer_json_member_add_time_t_formatted(wb, "baseline_after", baseline_after, options & RRDR_OPTION_RFC3339);
169	buffer_json_member_add_time_t_formatted(wb, "baseline_before", baseline_before, options & RRDR_OPTION_RFC3339);
170	buffer_json_member_add_time_t(wb, "baseline_duration", baseline_before - baseline_after);
171	buffer_json_member_add_uint64(wb, "baseline_points", points << shifts);
172	}
173
174	buffer_json_member_add_object(wb, "statistics");
175	{
176	buffer_json_member_add_double(wb, "query_time_ms", (double) duration / (double) USEC_PER_MS);
177	buffer_json_member_add_uint64(wb, "db_queries", stats->db_queries);
178	buffer_json_member_add_uint64(wb, "query_result_points", stats->result_points);
179	buffer_json_member_add_uint64(wb, "binary_searches", stats->binary_searches);
180	buffer_json_member_add_uint64(wb, "db_points_read", stats->db_points);
181
182	buffer_json_member_add_array(wb, "db_points_per_tier");
183	{
184	for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++)
185	buffer_json_add_array_item_uint64(wb, stats->db_points_per_tier[tier]);
186	}
187	buffer_json_array_close(wb);
188	}
189	buffer_json_object_close(wb);
190
191	buffer_json_member_add_string(wb, "group", time_grouping_tostring(group));
192	buffer_json_member_add_string(wb, "method", weights_method_to_string(method));
193	rrdr_options_to_buffer_json_array(wb, "options", options);
194	}
195
196	static size_t registered_results_to_json_charts(DICTIONARY results, BUFFER wb,
197	time_t after, time_t before,
198	time_t baseline_after, time_t baseline_before,
199	size_t points, WEIGHTS_METHOD method,
200	RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
201	size_t examined_dimensions, usec_t duration,
202	WEIGHTS_STATS *stats) {
203
204	buffer_json_initialize(wb, "\"", "\"", 0, true, (options & RRDR_OPTION_MINIFY) ? BUFFER_JSON_OPTIONS_MINIFY : BUFFER_JSON_OPTIONS_DEFAULT);
205
206	results_header_to_json(results, wb, after, before, baseline_after, baseline_before,
207	points, method, group, options, shifts, examined_dimensions, duration, stats);
208
209	buffer_json_member_add_object(wb, "correlated_charts");
210
211	size_t charts = 0, total_dimensions = 0;
212	struct register_result *t;
213	RRDINSTANCE_ACQUIRED *last_ria = NULL; // never access this - we use it only for comparison
214	dfe_start_read(results, t) {
215	if(t->ria != last_ria) {
216	last_ria = t->ria;
217
218	if(charts) {
219	buffer_json_object_close(wb); // dimensions
220	buffer_json_object_close(wb); // chart:id
221	}
222
223	buffer_json_member_add_object(wb, rrdinstance_acquired_id(t->ria));
224	buffer_json_member_add_string(wb, "context", rrdcontext_acquired_id(t->rca));
225	buffer_json_member_add_object(wb, "dimensions");
226	charts++;
227	}
228	buffer_json_member_add_double(wb, rrdmetric_acquired_name(t->rma), t->value);
229	total_dimensions++;
230	}
231	dfe_done(t);
232
233	// close dimensions and chart
234	if (total_dimensions) {
235	buffer_json_object_close(wb); // dimensions
236	buffer_json_object_close(wb); // chart:id
237	}
238
239	buffer_json_object_close(wb);
240
241	buffer_json_member_add_uint64(wb, "correlated_dimensions", total_dimensions);
242	buffer_json_member_add_uint64(wb, "total_dimensions_count", examined_dimensions);
243	buffer_json_finalize(wb);
244
245	return total_dimensions;
246	}
247
248	static size_t registered_results_to_json_contexts(DICTIONARY results, BUFFER wb,
249	time_t after, time_t before,
250	time_t baseline_after, time_t baseline_before,
251	size_t points, WEIGHTS_METHOD method,
252	RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
253	size_t examined_dimensions, usec_t duration,
254	WEIGHTS_STATS *stats) {
255
256	buffer_json_initialize(wb, "\"", "\"", 0, true, (options & RRDR_OPTION_MINIFY) ? BUFFER_JSON_OPTIONS_MINIFY : BUFFER_JSON_OPTIONS_DEFAULT);
257
258	results_header_to_json(results, wb, after, before, baseline_after, baseline_before,
259	points, method, group, options, shifts, examined_dimensions, duration, stats);
260
261	buffer_json_member_add_object(wb, "contexts");
262
263	size_t contexts = 0, charts = 0, total_dimensions = 0, context_dims = 0, chart_dims = 0;
264	NETDATA_DOUBLE contexts_total_weight = 0.0, charts_total_weight = 0.0;
265	struct register_result *t;
266	RRDCONTEXT_ACQUIRED *last_rca = NULL;
267	RRDINSTANCE_ACQUIRED *last_ria = NULL;
268	dfe_start_read(results, t) {
269
270	if(t->rca != last_rca) {
271	last_rca = t->rca;
272
273	if(contexts) {
274	buffer_json_object_close(wb); // dimensions
275	buffer_json_member_add_double(wb, "weight", charts_total_weight / (double) chart_dims);
276	buffer_json_object_close(wb); // chart:id
277	buffer_json_object_close(wb); // charts
278	buffer_json_member_add_double(wb, "weight", contexts_total_weight / (double) context_dims);
279	buffer_json_object_close(wb); // context
280	}
281
282	buffer_json_member_add_object(wb, rrdcontext_acquired_id(t->rca));
283	buffer_json_member_add_object(wb, "charts");
284
285	contexts++;
286	charts = 0;
287	context_dims = 0;
288	contexts_total_weight = 0.0;
289
290	last_ria = NULL;
291	}
292
293	if(t->ria != last_ria) {
294	last_ria = t->ria;
295
296	if(charts) {
297	buffer_json_object_close(wb); // dimensions
298	buffer_json_member_add_double(wb, "weight", charts_total_weight / (double) chart_dims);
299	buffer_json_object_close(wb); // chart:id
300	}
301
302	buffer_json_member_add_object(wb, rrdinstance_acquired_id(t->ria));
303	buffer_json_member_add_object(wb, "dimensions");
304
305	charts++;
306	chart_dims = 0;
307	charts_total_weight = 0.0;
308	}
309
310	buffer_json_member_add_double(wb, rrdmetric_acquired_name(t->rma), t->value);
311	charts_total_weight += t->value;
312	contexts_total_weight += t->value;
313	chart_dims++;
314	context_dims++;
315	total_dimensions++;
316	}
317	dfe_done(t);
318
319	// close dimensions and chart
320	if (total_dimensions) {
321	buffer_json_object_close(wb); // dimensions
322	buffer_json_member_add_double(wb, "weight", charts_total_weight / (double) chart_dims);
323	buffer_json_object_close(wb); // chart:id
324	buffer_json_object_close(wb); // charts
325	buffer_json_member_add_double(wb, "weight", contexts_total_weight / (double) context_dims);
326	buffer_json_object_close(wb); // context
327	}
328
329	buffer_json_object_close(wb);
330
331	buffer_json_member_add_uint64(wb, "correlated_dimensions", total_dimensions);
332	buffer_json_member_add_uint64(wb, "total_dimensions_count", examined_dimensions);
333	buffer_json_finalize(wb);
334
335	return total_dimensions;
336	}
337
338	// Workload statistics for progress tracking and thread optimization
339	struct workload_stats {
340	size_t nodes;
341	size_t contexts;
342	size_t metrics;
343	};
344
345	struct query_weights_data {
346	QUERY_WEIGHTS_REQUEST *qwr;
347
348	SIMPLE_PATTERN *scope_nodes_sp;
349	SIMPLE_PATTERN *scope_contexts_sp;
350	SIMPLE_PATTERN *scope_instances_sp;
351	SIMPLE_PATTERN *scope_labels_sp;
352	SIMPLE_PATTERN *scope_dimensions_sp;
353	SIMPLE_PATTERN *nodes_sp;
354	SIMPLE_PATTERN *contexts_sp;
355	SIMPLE_PATTERN *instances_sp;
356	SIMPLE_PATTERN *dimensions_sp;
357	SIMPLE_PATTERN *labels_sp;
358	SIMPLE_PATTERN *alerts_sp;
359
360	struct pattern_array *scope_labels_pa;
361	struct pattern_array *labels_pa;
362
363	usec_t timeout_us;
364	bool timed_out;
365	bool interrupted;
366
367	struct query_timings timings;
368
369	size_t examined_dimensions;
370	bool register_zero;
371
372	DICTIONARY *results;
373	WEIGHTS_STATS stats;
374	RRDHOST **hosts_array;
375	size_t total_hosts;
376	size_t hosts_array_capacity;
377
378	uint32_t shifts;
379
380	struct query_versions versions;
381	struct workload_stats total_workload; // Overall workload statistics for progress tracking
382	};
383
384	// Thread-local data for parallel processing
385	struct query_weights_thread_data {
386	struct query_weights_data *main_qwd;
387	DICTIONARY *local_results;
388	WEIGHTS_STATS local_stats;
389	size_t local_examined_dimensions;
390	struct query_versions local_versions;
391	RRDHOST **hosts;
392	struct completion completion;
393	size_t host_count;
394	size_t thread_id;
395	};
396
397	// Worker thread function for parallel host processing
398	void query_weights_worker_thread(void *arg)
399	{
400	struct query_weights_thread_data thread_data = (struct query_weights_thread_data )arg;
401	struct query_weights_data *main_qwd = thread_data->main_qwd;
402
403	// Initialize local statistics
404	memset(&thread_data->local_stats, 0, sizeof(WEIGHTS_STATS));
405	thread_data->local_examined_dimensions = 0;
406	memset(&thread_data->local_versions, 0, sizeof(struct query_versions));
407
408	// Process assigned hosts
409	for (size_t i = 0; i < thread_data->host_count; i++) {
410	RRDHOST *host = thread_data->hosts[i];
411	if (!host) continue;
412
413	// Check for timeout/interruption
414	if (__atomic_load_n(&main_qwd->timed_out, __ATOMIC_RELAXED) \|\|
415	__atomic_load_n(&main_qwd->interrupted, __ATOMIC_RELAXED)) {
416	break;
417	}
418
419	// Check timeout
420	if (now_monotonic_usec() > (main_qwd->timings.received_ut + main_qwd->timeout_us)) {
421	__atomic_store_n(&main_qwd->timed_out, true, __ATOMIC_RELAXED);
422	break;
423	}
424
425	// Check interruption callback
426	if (main_qwd->qwr->interrupt_callback &&
427	main_qwd->qwr->interrupt_callback(main_qwd->qwr->interrupt_callback_data)) {
428	__atomic_store_n(&main_qwd->interrupted, true, __ATOMIC_RELAXED);
429	break;
430	}
431
432	// Create a local query_weights_data for this thread
433	struct query_weights_data local_qwd = *main_qwd;
434	local_qwd.results = thread_data->local_results;
435	local_qwd.stats = thread_data->local_stats;
436	local_qwd.examined_dimensions = thread_data->local_examined_dimensions;
437	local_qwd.versions = thread_data->local_versions;
438
439	char uuid[UUID_STR_LEN];
440	if(!UUIDiszero(host->node_id))
441	uuid_unparse_lower(host->node_id.uuid, uuid);
442	else
443	uuid[0] = '\0';
444
445	SIMPLE_PATTERN_RESULT match = SP_MATCHED_POSITIVE;
446	if(main_qwd->scope_nodes_sp) {
447	match = simple_pattern_matches_string_extract(main_qwd->scope_nodes_sp, host->hostname, NULL, 0);
448	if(match == SP_NOT_MATCHED) {
449	match = simple_pattern_matches_extract(main_qwd->scope_nodes_sp, host->machine_guid, NULL, 0);
450	if(match == SP_NOT_MATCHED && *uuid)
451	match = simple_pattern_matches_extract(main_qwd->scope_nodes_sp, uuid, NULL, 0);
452	}
453	}
454
455	if(match != SP_MATCHED_POSITIVE)
456	continue;
457
458	if(main_qwd->nodes_sp) {
459	match = simple_pattern_matches_string_extract(main_qwd->nodes_sp, host->hostname, NULL, 0);
460	if(match == SP_NOT_MATCHED) {
461	match = simple_pattern_matches_extract(main_qwd->nodes_sp, host->machine_guid, NULL, 0);
462	if(match == SP_NOT_MATCHED && *uuid)
463	match = simple_pattern_matches_extract(main_qwd->nodes_sp, uuid, NULL, 0);
464	}
465	}
466
467	bool queryable_host = (match == SP_MATCHED_POSITIVE);
468
469	// Update local version hashes
470	thread_data->local_versions.contexts_hard_hash += dictionary_version(host->rrdctx.contexts);
471	thread_data->local_versions.contexts_soft_hash += rrdcontext_queue_version(&host->rrdctx.hub_queue);
472	thread_data->local_versions.alerts_hard_hash += dictionary_version(host->rrdcalc_root_index);
473	thread_data->local_versions.alerts_soft_hash += __atomic_load_n(&host->health_transitions, __ATOMIC_RELAXED);
474
475	// Process the host using the callback
476	ssize_t ret = weights_do_node_callback(&local_qwd, host, queryable_host);
477	if (ret < 0)
478	break;
479
480	// Update thread-local counters
481	thread_data->local_examined_dimensions = local_qwd.examined_dimensions;
482	thread_data->local_stats = local_qwd.stats;
483	}
484	}
485
486	// Thread-safe statistics merging - use simple addition since we're in single-threaded merge
487	static void merge_weights_stats(WEIGHTS_STATS dest, const WEIGHTS_STATS src) {
488	dest->db_queries += src->db_queries;
489	dest->db_points += src->db_points;
490	dest->result_points += src->result_points;
491	dest->binary_searches += src->binary_searches;
492
493	// Update max ratio if needed
494	if (src->max_base_high_ratio > dest->max_base_high_ratio) {
495	dest->max_base_high_ratio = src->max_base_high_ratio;
496	}
497
498	for(size_t tier = 0; tier < RRD_STORAGE_TIERS; tier++) {
499	dest->db_points_per_tier[tier] += src->db_points_per_tier[tier];
500	}
501	}
502
503	#define AGGREGATED_WEIGHT_EMPTY (struct aggregated_weight) { \
504	.min = NAN, \
505	.max = NAN, \
506	.sum = NAN, \
507	.count = 0, \
508	.hsp = STORAGE_POINT_UNSET, \
509	.bsp = STORAGE_POINT_UNSET, \
510	}
511
512	#define merge_into_aw(aw, t) do { \
513	if(!(aw).count) { \
514	(aw).count = 1; \
515	(aw).min = (aw).max = (aw).sum = (t)->value; \
516	(aw).hsp = (t)->highlighted; \
517	if(baseline) \
518	(aw).bsp = (t)->baseline; \
519	} \
520	else { \
521	(aw).count++; \
522	(aw).sum += (t)->value; \
523	if((t)->value < (aw).min) \
524	(aw).min = (t)->value; \
525	if((t)->value > (aw).max) \
526	(aw).max = (t)->value; \
527	storage_point_merge_to((aw).hsp, (t)->highlighted); \
528	if(baseline) \
529	storage_point_merge_to((aw).bsp, (t)->baseline); \
530	} \
531	} while(0)
532
533	static void results_header_to_json_v2(DICTIONARY results __maybe_unused, BUFFER wb, struct query_weights_data *qwd,
534	time_t after, time_t before,
535	time_t baseline_after, time_t baseline_before,
536	size_t points, WEIGHTS_METHOD method,
537	RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
538	size_t examined_dimensions __maybe_unused, usec_t duration __maybe_unused,
539	WEIGHTS_STATS *stats, bool group_by) {
540
541	buffer_json_member_add_object(wb, "request");
542	buffer_json_member_add_string(wb, "method", weights_method_to_string(method));
543	rrdr_options_to_buffer_json_array(wb, "options", options);
544
545	buffer_json_member_add_object(wb, "scope");
546	buffer_json_member_add_string(wb, "scope_nodes", qwd->qwr->scope_nodes ? qwd->qwr->scope_nodes : "*");
547	buffer_json_member_add_string(wb, "scope_contexts", qwd->qwr->scope_contexts ? qwd->qwr->scope_contexts : "*");
548	buffer_json_member_add_string(wb, "scope_instances", qwd->qwr->scope_instances ? qwd->qwr->scope_instances : "*");
549	buffer_json_member_add_string(wb, "scope_labels", qwd->qwr->scope_labels ? qwd->qwr->scope_labels : "*");
550	buffer_json_object_close(wb);
551
552	buffer_json_member_add_object(wb, "selectors");
553	buffer_json_member_add_string(wb, "nodes", qwd->qwr->nodes ? qwd->qwr->nodes : "*");
554	buffer_json_member_add_string(wb, "contexts", qwd->qwr->contexts ? qwd->qwr->contexts : "*");
555	buffer_json_member_add_string(wb, "instances", qwd->qwr->instances ? qwd->qwr->instances : "*");
556	buffer_json_member_add_string(wb, "dimensions", qwd->qwr->dimensions ? qwd->qwr->dimensions : "*");
557	buffer_json_member_add_string(wb, "labels", qwd->qwr->labels ? qwd->qwr->labels : "*");
558	buffer_json_member_add_string(wb, "alerts", qwd->qwr->alerts ? qwd->qwr->alerts : "*");
559	buffer_json_object_close(wb);
560
561	buffer_json_member_add_object(wb, "window");
562	buffer_json_member_add_time_t_formatted(wb, "after", qwd->qwr->after, options & RRDR_OPTION_RFC3339);
563	buffer_json_member_add_time_t_formatted(wb, "before", qwd->qwr->before, options & RRDR_OPTION_RFC3339);
564	buffer_json_member_add_uint64(wb, "points", qwd->qwr->points);
565	if(qwd->qwr->options & RRDR_OPTION_SELECTED_TIER)
566	buffer_json_member_add_uint64(wb, "tier", qwd->qwr->tier);
567	else
568	buffer_json_member_add_string(wb, "tier", NULL);
569	buffer_json_object_close(wb);
570
571	if(method == WEIGHTS_METHOD_MC_KS2 \|\| method == WEIGHTS_METHOD_MC_VOLUME) {
572	buffer_json_member_add_object(wb, "baseline");
573	buffer_json_member_add_time_t_formatted(wb, "baseline_after", qwd->qwr->baseline_after, options & RRDR_OPTION_RFC3339);
574	buffer_json_member_add_time_t_formatted(wb, "baseline_before", qwd->qwr->baseline_before, options & RRDR_OPTION_RFC3339);
575	buffer_json_object_close(wb);
576	}
577
578	buffer_json_member_add_object(wb, "aggregations");
579	buffer_json_member_add_object(wb, "time");
580	buffer_json_member_add_string(wb, "time_group", time_grouping_tostring(qwd->qwr->time_group_method));
581	buffer_json_member_add_string(wb, "time_group_options", qwd->qwr->time_group_options);
582	buffer_json_object_close(wb); // time
583
584	buffer_json_member_add_array(wb, "metrics");
585	buffer_json_add_array_item_object(wb);
586	{
587	buffer_json_member_add_array(wb, "group_by");
588	buffer_json_group_by_to_array(wb, qwd->qwr->group_by.group_by);
589	buffer_json_array_close(wb);
590
591	// buffer_json_member_add_array(wb, "group_by_label");
592	// buffer_json_array_close(wb);
593
594	buffer_json_member_add_string(wb, "aggregation", group_by_aggregate_function_to_string(qwd->qwr->group_by.aggregation));
595	}
596	buffer_json_object_close(wb); // 1st group by
597	buffer_json_array_close(wb); // array
598	buffer_json_object_close(wb); // aggregations
599
600	buffer_json_member_add_uint64(wb, "timeout", qwd->qwr->timeout_ms);
601	buffer_json_object_close(wb); // request
602
603	buffer_json_member_add_object(wb, "view");
604	buffer_json_member_add_string(wb, "format", (group_by)?"grouped":"full");
605	buffer_json_member_add_string(wb, "time_group", time_grouping_tostring(group));
606
607	buffer_json_member_add_object(wb, "window");
608	buffer_json_member_add_time_t_formatted(wb, "after", after, options & RRDR_OPTION_RFC3339);
609	buffer_json_member_add_time_t_formatted(wb, "before", before, options & RRDR_OPTION_RFC3339);
610	buffer_json_member_add_time_t(wb, "duration", before - after);
611	buffer_json_member_add_uint64(wb, "points", points);
612	buffer_json_object_close(wb);
613
614	if(method == WEIGHTS_METHOD_MC_KS2 \|\| method == WEIGHTS_METHOD_MC_VOLUME) {
615	buffer_json_member_add_object(wb, "baseline");
616	buffer_json_member_add_time_t_formatted(wb, "after", baseline_after, options & RRDR_OPTION_RFC3339);
617	buffer_json_member_add_time_t_formatted(wb, "before", baseline_before, options & RRDR_OPTION_RFC3339);
618	buffer_json_member_add_time_t(wb, "duration", baseline_before - baseline_after);
619	buffer_json_member_add_uint64(wb, "points", points << shifts);
620	buffer_json_object_close(wb);
621	}
622
623	buffer_json_object_close(wb); // view
624
625	buffer_json_member_add_object(wb, "db");
626	{
627	buffer_json_member_add_uint64(wb, "db_queries", stats->db_queries);
628	buffer_json_member_add_uint64(wb, "query_result_points", stats->result_points);
629	buffer_json_member_add_uint64(wb, "binary_searches", stats->binary_searches);
630	buffer_json_member_add_uint64(wb, "db_points_read", stats->db_points);
631
632	buffer_json_member_add_array(wb, "db_points_per_tier");
633	{
634	for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++)
635	buffer_json_add_array_item_uint64(wb, stats->db_points_per_tier[tier]);
636	}
637	buffer_json_array_close(wb);
638	}
639	buffer_json_object_close(wb); // db
640	}
641
642	typedef enum {
643	WPT_DIMENSION = 0,
644	WPT_INSTANCE = 1,
645	WPT_CONTEXT = 2,
646	WPT_NODE = 3,
647	WPT_GROUP = 4,
648	} WEIGHTS_POINT_TYPE;
649
650	struct aggregated_weight {
651	const char *name;
652	NETDATA_DOUBLE min;
653	NETDATA_DOUBLE max;
654	NETDATA_DOUBLE sum;
655	size_t count;
656	STORAGE_POINT hsp;
657	STORAGE_POINT bsp;
658	};
659
660	static inline void storage_point_to_json(BUFFER wb, WEIGHTS_POINT_TYPE type, ssize_t di, ssize_t ii, ssize_t ci, ssize_t ni, struct aggregated_weight aw, RRDR_OPTIONS options __maybe_unused, bool baseline) {
661	if(type != WPT_GROUP) {
662	buffer_json_add_array_item_array(wb);
663	buffer_json_add_array_item_uint64(wb, type); // "type"
664	buffer_json_add_array_item_int64(wb, ni);
665	if (type != WPT_NODE) {
666	buffer_json_add_array_item_int64(wb, ci);
667	if (type != WPT_CONTEXT) {
668	buffer_json_add_array_item_int64(wb, ii);
669	if (type != WPT_INSTANCE)
670	buffer_json_add_array_item_int64(wb, di);
671	else
672	buffer_json_add_array_item_string(wb, NULL);
673	}
674	else {
675	buffer_json_add_array_item_string(wb, NULL);
676	buffer_json_add_array_item_string(wb, NULL);
677	}
678	}
679	else {
680	buffer_json_add_array_item_string(wb, NULL);
681	buffer_json_add_array_item_string(wb, NULL);
682	buffer_json_add_array_item_string(wb, NULL);
683	}
684	buffer_json_add_array_item_double(wb, (aw->count) ? aw->sum / (NETDATA_DOUBLE)aw->count : 0.0); // "weight"
685	}
686	else {
687	buffer_json_member_add_array(wb, "v");
688	buffer_json_add_array_item_array(wb);
689	buffer_json_add_array_item_double(wb, aw->min); // "min"
690	buffer_json_add_array_item_double(wb, (aw->count) ? aw->sum / (NETDATA_DOUBLE)aw->count : 0.0); // "avg"
691	buffer_json_add_array_item_double(wb, aw->max); // "max"
692	buffer_json_add_array_item_double(wb, aw->sum); // "sum"
693	buffer_json_add_array_item_uint64(wb, aw->count); // "count"
694	buffer_json_array_close(wb);
695	}
696
697	buffer_json_add_array_item_array(wb);
698	buffer_json_add_array_item_double(wb, aw->hsp.min); // "min"
699	buffer_json_add_array_item_double(wb, (aw->hsp.count) ? aw->hsp.sum / (NETDATA_DOUBLE) aw->hsp.count : 0.0); // "avg"
700	buffer_json_add_array_item_double(wb, aw->hsp.max); // "max"
701	buffer_json_add_array_item_double(wb, aw->hsp.sum); // "sum"
702	buffer_json_add_array_item_uint64(wb, aw->hsp.count); // "count"
703	buffer_json_add_array_item_uint64(wb, aw->hsp.anomaly_count); // "anomaly_count"
704	buffer_json_array_close(wb);
705
706	if(baseline) {
707	buffer_json_add_array_item_array(wb);
708	buffer_json_add_array_item_double(wb, aw->bsp.min); // "min"
709	buffer_json_add_array_item_double(wb, (aw->bsp.count) ? aw->bsp.sum / (NETDATA_DOUBLE) aw->bsp.count : 0.0); // "avg"
710	buffer_json_add_array_item_double(wb, aw->bsp.max); // "max"
711	buffer_json_add_array_item_double(wb, aw->bsp.sum); // "sum"
712	buffer_json_add_array_item_uint64(wb, aw->bsp.count); // "count"
713	buffer_json_add_array_item_uint64(wb, aw->bsp.anomaly_count); // "anomaly_count"
714	buffer_json_array_close(wb);
715	}
716
717	buffer_json_array_close(wb);
718	}
719
720	static void multinode_data_schema(BUFFER wb, RRDR_OPTIONS options __maybe_unused, const char key, bool baseline, bool group_by) {
721	buffer_json_member_add_object(wb, key); // schema
722
723	buffer_json_member_add_string(wb, "type", "array");
724	buffer_json_member_add_array(wb, "items");
725
726	if(group_by) {
727	buffer_json_add_array_item_object(wb);
728	{
729	buffer_json_member_add_string(wb, "name", "weight");
730	buffer_json_member_add_string(wb, "type", "array");
731	buffer_json_member_add_array(wb, "labels");
732	{
733	buffer_json_add_array_item_string(wb, "min");
734	buffer_json_add_array_item_string(wb, "avg");
735	buffer_json_add_array_item_string(wb, "max");
736	buffer_json_add_array_item_string(wb, "sum");
737	buffer_json_add_array_item_string(wb, "count");
738	}
739	buffer_json_array_close(wb);
740	}
741	buffer_json_object_close(wb);
742	}
743	else {
744	buffer_json_add_array_item_object(wb);
745	buffer_json_member_add_string(wb, "name", "row_type");
746	buffer_json_member_add_string(wb, "type", "integer");
747	buffer_json_member_add_array(wb, "value");
748	buffer_json_add_array_item_string(wb, "dimension");
749	buffer_json_add_array_item_string(wb, "instance");
750	buffer_json_add_array_item_string(wb, "context");
751	buffer_json_add_array_item_string(wb, "node");
752	buffer_json_array_close(wb);
753	buffer_json_object_close(wb);
754
755	buffer_json_add_array_item_object(wb);
756	{
757	buffer_json_member_add_string(wb, "name", "ni");
758	buffer_json_member_add_string(wb, "type", "integer");
759	buffer_json_member_add_string(wb, "dictionary", "nodes");
760	}
761	buffer_json_object_close(wb);
762
763	buffer_json_add_array_item_object(wb);
764	{
765	buffer_json_member_add_string(wb, "name", "ci");
766	buffer_json_member_add_string(wb, "type", "integer");
767	buffer_json_member_add_string(wb, "dictionary", "contexts");
768	}
769	buffer_json_object_close(wb);
770
771	buffer_json_add_array_item_object(wb);
772	{
773	buffer_json_member_add_string(wb, "name", "ii");
774	buffer_json_member_add_string(wb, "type", "integer");
775	buffer_json_member_add_string(wb, "dictionary", "instances");
776	}
777	buffer_json_object_close(wb);
778
779	buffer_json_add_array_item_object(wb);
780	{
781	buffer_json_member_add_string(wb, "name", "di");
782	buffer_json_member_add_string(wb, "type", "integer");
783	buffer_json_member_add_string(wb, "dictionary", "dimensions");
784	}
785	buffer_json_object_close(wb);
786
787	buffer_json_add_array_item_object(wb);
788	{
789	buffer_json_member_add_string(wb, "name", "weight");
790	buffer_json_member_add_string(wb, "type", "number");
791	}
792	buffer_json_object_close(wb);
793	}
794
795	buffer_json_add_array_item_object(wb);
796	{
797	buffer_json_member_add_string(wb, "name", "timeframe");
798	buffer_json_member_add_string(wb, "type", "array");
799	buffer_json_member_add_array(wb, "labels");
800	{
801	buffer_json_add_array_item_string(wb, "min");
802	buffer_json_add_array_item_string(wb, "avg");
803	buffer_json_add_array_item_string(wb, "max");
804	buffer_json_add_array_item_string(wb, "sum");
805	buffer_json_add_array_item_string(wb, "count");
806	buffer_json_add_array_item_string(wb, "anomaly_count");
807	}
808	buffer_json_array_close(wb);
809	buffer_json_member_add_object(wb, "calculations");
810	buffer_json_member_add_string(wb, "anomaly rate", "anomaly_count * 100 / count");
811	buffer_json_object_close(wb);
812	}
813	buffer_json_object_close(wb);
814
815	if(baseline) {
816	buffer_json_add_array_item_object(wb);
817	{
818	buffer_json_member_add_string(wb, "name", "baseline timeframe");
819	buffer_json_member_add_string(wb, "type", "array");
820	buffer_json_member_add_array(wb, "labels");
821	{
822	buffer_json_add_array_item_string(wb, "min");
823	buffer_json_add_array_item_string(wb, "avg");
824	buffer_json_add_array_item_string(wb, "max");
825	buffer_json_add_array_item_string(wb, "sum");
826	buffer_json_add_array_item_string(wb, "count");
827	buffer_json_add_array_item_string(wb, "anomaly_count");
828	}
829	buffer_json_array_close(wb);
830	buffer_json_member_add_object(wb, "calculations");
831	buffer_json_member_add_string(wb, "anomaly rate", "anomaly_count * 100 / count");
832	buffer_json_object_close(wb);
833	}
834	buffer_json_object_close(wb);
835	}
836
837	buffer_json_array_close(wb); // items
838	buffer_json_object_close(wb); // schema
839	}
840
841	struct dict_unique_node {
842	bool existing;
843	bool exposed;
844	uint32_t i;
845	RRDHOST *host;
846	usec_t duration_ut;
847	};
848
849	struct dict_unique_name_units {
850	bool existing;
851	bool exposed;
852	uint32_t i;
853	const char *units;
854	};
855
856	struct dict_unique_id_name {
857	bool existing;
858	bool exposed;
859	uint32_t i;
860	const char *id;
861	const char *name;
862	};
863
864	static inline struct dict_unique_node dict_unique_node_add(DICTIONARY dict, RRDHOST host, ssize_t max_id) {
865	struct dict_unique_node *dun = dictionary_set(dict, host->machine_guid, NULL, sizeof(struct dict_unique_node));
866	if(!dun->existing) {
867	dun->existing = true;
868	dun->host = host;
869	dun->i = *max_id;
870	(*max_id)++;
871	}
872
873	return dun;
874	}
875
876	static inline struct dict_unique_name_units dict_unique_name_units_add(DICTIONARY dict, const char name, const char units, ssize_t *max_id) {
877	struct dict_unique_name_units *dun = dictionary_set(dict, name, NULL, sizeof(struct dict_unique_name_units));
878	if(!dun->existing) {
879	dun->units = units;
880	dun->existing = true;
881	dun->i = *max_id;
882	(*max_id)++;
883	}
884
885	return dun;
886	}
887
888	static inline struct dict_unique_id_name dict_unique_id_name_add(DICTIONARY dict, const char id, const char name, ssize_t *max_id) {
889	char key[1024 + 1];
890	snprintfz(key, sizeof(key) - 1, "%s:%s", id, name);
891	struct dict_unique_id_name *dun = dictionary_set(dict, key, NULL, sizeof(struct dict_unique_id_name));
892	if(!dun->existing) {
893	dun->existing = true;
894	dun->i = *max_id;
895	(*max_id)++;
896	dun->id = id;
897	dun->name = name;
898	}
899
900	return dun;
901	}
902
903	static size_t registered_results_to_json_multinode_no_group_by(
904	DICTIONARY results, BUFFER wb,
905	time_t after, time_t before,
906	time_t baseline_after, time_t baseline_before,
907	size_t points, WEIGHTS_METHOD method,
908	RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
909	size_t examined_dimensions, struct query_weights_data *qwd,
910	WEIGHTS_STATS *stats,
911	struct query_versions *versions) {
912	buffer_json_initialize(wb, "\"", "\"", 0, true, (options & RRDR_OPTION_MINIFY) ? BUFFER_JSON_OPTIONS_MINIFY : BUFFER_JSON_OPTIONS_DEFAULT);
913	buffer_json_member_add_uint64(wb, "api", 2);
914
915	results_header_to_json_v2(results, wb, qwd, after, before, baseline_after, baseline_before,
916	points, method, group, options, shifts, examined_dimensions,
917	qwd->timings.executed_ut - qwd->timings.received_ut, stats, false);
918
919	version_hashes_api_v2(wb, versions);
920
921	bool baseline = method == WEIGHTS_METHOD_MC_KS2 \|\| method == WEIGHTS_METHOD_MC_VOLUME;
922	multinode_data_schema(wb, options, "schema", baseline, false);
923
924	DICTIONARY *dict_nodes = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct dict_unique_node));
925	DICTIONARY *dict_contexts = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct dict_unique_name_units));
926	DICTIONARY *dict_instances = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct dict_unique_id_name));
927	DICTIONARY *dict_dimensions = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct dict_unique_id_name));
928
929	buffer_json_member_add_array(wb, "result");
930
931	struct aggregated_weight node_aw = AGGREGATED_WEIGHT_EMPTY, context_aw = AGGREGATED_WEIGHT_EMPTY, instance_aw = AGGREGATED_WEIGHT_EMPTY;
932	struct register_result *t;
933	RRDHOST *last_host = NULL;
934	RRDCONTEXT_ACQUIRED *last_rca = NULL;
935	RRDINSTANCE_ACQUIRED *last_ria = NULL;
936	struct dict_unique_name_units *context_dun = NULL;
937	struct dict_unique_node *node_dun = NULL;
938	struct dict_unique_id_name *instance_dun = NULL;
939	struct dict_unique_id_name *dimension_dun = NULL;
940	ssize_t di = -1, ii = -1, ci = -1, ni = -1;
941	ssize_t di_max = 0, ii_max = 0, ci_max = 0, ni_max = 0;
942	size_t total_dimensions = 0;
943	dfe_start_read(results, t) {
944
945	// close instance
946	if(t->ria != last_ria && last_ria) {
947	storage_point_to_json(wb, WPT_INSTANCE, di, ii, ci, ni, &instance_aw, options, baseline);
948	instance_dun->exposed = true;
949	last_ria = NULL;
950	instance_aw = AGGREGATED_WEIGHT_EMPTY;
951	}
952
953	// close context
954	if(t->rca != last_rca && last_rca) {
955	storage_point_to_json(wb, WPT_CONTEXT, di, ii, ci, ni, &context_aw, options, baseline);
956	context_dun->exposed = true;
957	last_rca = NULL;
958	context_aw = AGGREGATED_WEIGHT_EMPTY;
959	}
960
961	// close node
962	if(t->host != last_host && last_host) {
963	storage_point_to_json(wb, WPT_NODE, di, ii, ci, ni, &node_aw, options, baseline);
964	node_dun->exposed = true;
965	last_host = NULL;
966	node_aw = AGGREGATED_WEIGHT_EMPTY;
967	}
968
969	// open node
970	if(t->host != last_host) {
971	last_host = t->host;
972	node_dun = dict_unique_node_add(dict_nodes, t->host, &ni_max);
973	ni = node_dun->i;
974	}
975
976	// open context
977	if(t->rca != last_rca) {
978	last_rca = t->rca;
979	context_dun = dict_unique_name_units_add(dict_contexts, rrdcontext_acquired_id(t->rca),
980	rrdcontext_acquired_units(t->rca), &ci_max);
981	ci = context_dun->i;
982	}
983
984	// open instance
985	if(t->ria != last_ria) {
986	last_ria = t->ria;
987	instance_dun = dict_unique_id_name_add(dict_instances, rrdinstance_acquired_id(t->ria), rrdinstance_acquired_name(t->ria), &ii_max);
988	ii = instance_dun->i;
989	}
990
991	dimension_dun = dict_unique_id_name_add(dict_dimensions, rrdmetric_acquired_id(t->rma), rrdmetric_acquired_name(t->rma), &di_max);
992	di = dimension_dun->i;
993
994	struct aggregated_weight aw = {
995	.min = t->value,
996	.max = t->value,
997	.sum = t->value,
998	.count = 1,
999	.hsp = t->highlighted,
1000	.bsp = t->baseline,
1001	};
1002
1003	storage_point_to_json(wb, WPT_DIMENSION, di, ii, ci, ni, &aw, options, baseline);
1004	node_dun->exposed = true;
1005	context_dun->exposed = true;
1006	instance_dun->exposed = true;
1007	dimension_dun->exposed = true;
1008
1009	merge_into_aw(instance_aw, t);
1010	merge_into_aw(context_aw, t);
1011	merge_into_aw(node_aw, t);
1012
1013	node_dun->duration_ut += t->duration_ut;
1014	total_dimensions++;
1015	}
1016	dfe_done(t);
1017
1018	// close instance
1019	if(last_ria) {
1020	storage_point_to_json(wb, WPT_INSTANCE, di, ii, ci, ni, &instance_aw, options, baseline);
1021	instance_dun->exposed = true;
1022	}
1023
1024	// close context
1025	if(last_rca) {
1026	storage_point_to_json(wb, WPT_CONTEXT, di, ii, ci, ni, &context_aw, options, baseline);
1027	context_dun->exposed = true;
1028	}
1029
1030	// close node
1031	if(last_host) {
1032	storage_point_to_json(wb, WPT_NODE, di, ii, ci, ni, &node_aw, options, baseline);
1033	node_dun->exposed = true;
1034	}
1035
1036	buffer_json_array_close(wb); // points
1037
1038	buffer_json_member_add_object(wb, "dictionaries");
1039	buffer_json_member_add_array(wb, "nodes");
1040	{
1041	struct dict_unique_node *dun;
1042	dfe_start_read(dict_nodes, dun) {
1043	if(!dun->exposed)
1044	continue;
1045
1046	buffer_json_add_array_item_object(wb);
1047	buffer_json_node_add_v2(wb, dun->host, dun->i, dun->duration_ut, true);
1048	buffer_json_object_close(wb);
1049	}
1050	dfe_done(dun);
1051	}
1052	buffer_json_array_close(wb);
1053
1054	buffer_json_member_add_array(wb, "contexts");
1055	{
1056	struct dict_unique_name_units *dun;
1057	dfe_start_read(dict_contexts, dun) {
1058	if(!dun->exposed)
1059	continue;
1060
1061	buffer_json_add_array_item_object(wb);
1062	buffer_json_member_add_string(wb, "id", dun_dfe.name);
1063	buffer_json_member_add_string(wb, "units", dun->units);
1064	buffer_json_member_add_int64(wb, "ci", dun->i);
1065	buffer_json_object_close(wb);
1066	}
1067	dfe_done(dun);
1068	}
1069	buffer_json_array_close(wb);
1070
1071	buffer_json_member_add_array(wb, "instances");
1072	{
1073	struct dict_unique_id_name *dun;
1074	dfe_start_read(dict_instances, dun) {
1075	if(!dun->exposed)
1076	continue;
1077
1078	buffer_json_add_array_item_object(wb);
1079	buffer_json_member_add_string(wb, "id", dun->id);
1080	if(dun->id != dun->name)
1081	buffer_json_member_add_string(wb, "nm", dun->name);
1082	buffer_json_member_add_int64(wb, "ii", dun->i);
1083	buffer_json_object_close(wb);
1084	}
1085	dfe_done(dun);
1086	}
1087	buffer_json_array_close(wb);
1088
1089	buffer_json_member_add_array(wb, "dimensions");
1090	{
1091	struct dict_unique_id_name *dun;
1092	dfe_start_read(dict_dimensions, dun) {
1093	if(!dun->exposed)
1094	continue;
1095
1096	buffer_json_add_array_item_object(wb);
1097	buffer_json_member_add_string(wb, "id", dun->id);
1098	if(dun->id != dun->name)
1099	buffer_json_member_add_string(wb, "nm", dun->name);
1100	buffer_json_member_add_int64(wb, "di", dun->i);
1101	buffer_json_object_close(wb);
1102	}
1103	dfe_done(dun);
1104	}
1105	buffer_json_array_close(wb);
1106
1107	buffer_json_object_close(wb); //dictionaries
1108
1109	buffer_json_agents_v2(wb, &qwd->timings, 0, false, true, rrdr_options_to_contexts_options(options));
1110	buffer_json_member_add_uint64(wb, "correlated_dimensions", total_dimensions);
1111	buffer_json_member_add_uint64(wb, "total_dimensions_count", examined_dimensions);
1112	buffer_json_finalize(wb);
1113
1114	dictionary_destroy(dict_nodes);
1115	dictionary_destroy(dict_contexts);
1116	dictionary_destroy(dict_instances);
1117	dictionary_destroy(dict_dimensions);
1118
1119	return total_dimensions;
1120	}
1121
1122	static size_t registered_results_to_json_multinode_group_by(
1123	DICTIONARY results, BUFFER wb,
1124	time_t after, time_t before,
1125	time_t baseline_after, time_t baseline_before,
1126	size_t points, WEIGHTS_METHOD method,
1127	RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
1128	size_t examined_dimensions, struct query_weights_data *qwd,
1129	WEIGHTS_STATS *stats,
1130	struct query_versions *versions) {
1131	buffer_json_initialize(wb, "\"", "\"", 0, true, (options & RRDR_OPTION_MINIFY) ? BUFFER_JSON_OPTIONS_MINIFY : BUFFER_JSON_OPTIONS_DEFAULT);
1132	buffer_json_member_add_uint64(wb, "api", 2);
1133
1134	results_header_to_json_v2(results, wb, qwd, after, before, baseline_after, baseline_before,
1135	points, method, group, options, shifts, examined_dimensions,
1136	qwd->timings.executed_ut - qwd->timings.received_ut, stats, true);
1137
1138	version_hashes_api_v2(wb, versions);
1139
1140	bool baseline = method == WEIGHTS_METHOD_MC_KS2 \|\| method == WEIGHTS_METHOD_MC_VOLUME;
1141	multinode_data_schema(wb, options, "v_schema", baseline, true);
1142
1143	DICTIONARY *group_by = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED \| DICT_OPTION_DONT_OVERWRITE_VALUE \| DICT_OPTION_FIXED_SIZE,
1144	NULL, sizeof(struct aggregated_weight));
1145
1146	struct register_result *t;
1147	size_t total_dimensions = 0;
1148	BUFFER *key = buffer_create(0, NULL);
1149	BUFFER *name = buffer_create(0, NULL);
1150	dfe_start_read(results, t) {
1151	char node_uuid[UUID_STR_LEN];
1152
1153	if(UUIDiszero(t->host->node_id))
1154	uuid_unparse_lower(t->host->host_id.uuid, node_uuid);
1155	else
1156	uuid_unparse_lower(t->host->node_id.uuid, node_uuid);
1157
1158	buffer_flush(key);
1159	buffer_flush(name);
1160
1161	if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_DIMENSION) {
1162	buffer_strcat(key, rrdmetric_acquired_name(t->rma));
1163	buffer_strcat(name, rrdmetric_acquired_name(t->rma));
1164	}
1165	if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_INSTANCE) {
1166	if(buffer_strlen(key)) {
1167	buffer_fast_strcat(key, ",", 1);
1168	buffer_fast_strcat(name, ",", 1);
1169	}
1170
1171	buffer_strcat(key, rrdinstance_acquired_id(t->ria));
1172	buffer_strcat(name, rrdinstance_acquired_name(t->ria));
1173
1174	if(!(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_NODE)) {
1175	buffer_fast_strcat(key, "@", 1);
1176	buffer_fast_strcat(name, "@", 1);
1177	buffer_strcat(key, node_uuid);
1178	buffer_strcat(name, rrdhost_hostname(t->host));
1179	}
1180	}
1181	if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_NODE) {
1182	if(buffer_strlen(key)) {
1183	buffer_fast_strcat(key, ",", 1);
1184	buffer_fast_strcat(name, ",", 1);
1185	}
1186
1187	buffer_strcat(key, node_uuid);
1188	buffer_strcat(name, rrdhost_hostname(t->host));
1189	}
1190	if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_CONTEXT) {
1191	if(buffer_strlen(key)) {
1192	buffer_fast_strcat(key, ",", 1);
1193	buffer_fast_strcat(name, ",", 1);
1194	}
1195
1196	buffer_strcat(key, rrdcontext_acquired_id(t->rca));
1197	buffer_strcat(name, rrdcontext_acquired_id(t->rca));
1198	}
1199	if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_UNITS) {
1200	if(buffer_strlen(key)) {
1201	buffer_fast_strcat(key, ",", 1);
1202	buffer_fast_strcat(name, ",", 1);
1203	}
1204
1205	buffer_strcat(key, rrdcontext_acquired_units(t->rca));
1206	buffer_strcat(name, rrdcontext_acquired_units(t->rca));
1207	}
1208
1209	struct aggregated_weight *aw = dictionary_set(group_by, buffer_tostring(key), NULL, sizeof(struct aggregated_weight));
1210	if(!aw->name) {
1211	aw->name = strdupz(buffer_tostring(name));
1212	aw->min = aw->max = aw->sum = t->value;
1213	aw->count = 1;
1214	aw->hsp = t->highlighted;
1215	aw->bsp = t->baseline;
1216	}
1217	else
1218	merge_into_aw(*aw, t);
1219
1220	total_dimensions++;
1221	}
1222	dfe_done(t);
1223	buffer_free(key); key = NULL;
1224	buffer_free(name); name = NULL;
1225
1226	struct aggregated_weight *aw;
1227	buffer_json_member_add_array(wb, "result");
1228	dfe_start_read(group_by, aw) {
1229	const char *k = aw_dfe.name;
1230	const char *n = aw->name;
1231
1232	buffer_json_add_array_item_object(wb);
1233	buffer_json_member_add_string(wb, "id", k);
1234
1235	if(strcmp(k, n) != 0)
1236	buffer_json_member_add_string(wb, "nm", n);
1237
1238	storage_point_to_json(wb, WPT_GROUP, 0, 0, 0, 0, aw, options, baseline);
1239	buffer_json_object_close(wb);
1240
1241	freez((void *)aw->name);
1242	}
1243	dfe_done(aw);
1244	buffer_json_array_close(wb); // result
1245
1246	buffer_json_agents_v2(wb, &qwd->timings, 0, false, true, rrdr_options_to_contexts_options(options));
1247	buffer_json_member_add_uint64(wb, "correlated_dimensions", total_dimensions);
1248	buffer_json_member_add_uint64(wb, "total_dimensions_count", examined_dimensions);
1249	buffer_json_finalize(wb);
1250
1251	dictionary_destroy(group_by);
1252
1253	return total_dimensions;
1254	}
1255
1256	// ----------------------------------------------------------------------------
1257	// KS2 algorithm functions
1258
1259	typedef long int DIFFS_NUMBERS;
1260	#define DOUBLE_TO_INT_MULTIPLIER 100000
1261
1262	static inline int binary_search_bigger_than(const DIFFS_NUMBERS arr[], int left, int size, DIFFS_NUMBERS K) {
1263	// binary search to find the index the smallest index
1264	// of the first value in the array that is greater than K
1265
1266	int right = size;
1267	while(left < right) {
1268	int middle = (int)(((unsigned int)(left + right)) >> 1);
1269
1270	if(arr[middle] > K)
1271	right = middle;
1272
1273	else
1274	left = middle + 1;
1275	}
1276
1277	return left;
1278	}
1279
1280	int compare_diffs(const void left, const void right) {
1281	DIFFS_NUMBERS lt = (DIFFS_NUMBERS )left;
1282	DIFFS_NUMBERS rt = (DIFFS_NUMBERS )right;
1283
1284	// https://stackoverflow.com/a/3886497/1114110
1285	return (lt > rt) - (lt < rt);
1286	}
1287
1288	static size_t calculate_pairs_diff(DIFFS_NUMBERS diffs, NETDATA_DOUBLE arr, size_t size) {
1289	NETDATA_DOUBLE *last = &arr[size - 1];
1290	size_t added = 0;
1291
1292	while(last > arr) {
1293	NETDATA_DOUBLE second = *last--;
1294	NETDATA_DOUBLE first = *last;
1295	diffs++ = (DIFFS_NUMBERS)((first - second) (NETDATA_DOUBLE)DOUBLE_TO_INT_MULTIPLIER);
1296	added++;
1297	}
1298
1299	return added;
1300	}
1301
1302	static double ks_2samp(
1303	DIFFS_NUMBERS baseline_diffs[], int base_size,
1304	DIFFS_NUMBERS highlight_diffs[], int high_size,
1305	uint32_t base_shifts) {
1306
1307	qsort(baseline_diffs, base_size, sizeof(DIFFS_NUMBERS), compare_diffs);
1308	qsort(highlight_diffs, high_size, sizeof(DIFFS_NUMBERS), compare_diffs);
1309
1310	// Now we should be calculating this:
1311	//
1312	// For each number in the diffs arrays, we should find the index of the
1313	// number bigger than them in both arrays and calculate the % of this index
1314	// vs the total array size. Once we have the 2 percentages, we should find
1315	// the min and max across the delta of all of them.
1316	//
1317	// It should look like this:
1318	//
1319	// base_pcent = binary_search_bigger_than(...) / base_size;
1320	// high_pcent = binary_search_bigger_than(...) / high_size;
1321	// delta = base_pcent - high_pcent;
1322	// if(delta < min) min = delta;
1323	// if(delta > max) max = delta;
1324	//
1325	// This would require a lot of multiplications and divisions.
1326	//
1327	// To speed it up, we do the binary search to find the index of each number
1328	// but, then we divide the base index by the power of two number (shifts) it
1329	// is bigger than high index. So the 2 indexes are now comparable.
1330	// We also keep track of the original indexes with min and max, to properly
1331	// calculate their percentages once the loops finish.
1332
1333
1334	// initialize min and max using the first number of baseline_diffs
1335	DIFFS_NUMBERS K = baseline_diffs[0];
1336	int base_idx = binary_search_bigger_than(baseline_diffs, 1, base_size, K);
1337	int high_idx = binary_search_bigger_than(highlight_diffs, 0, high_size, K);
1338	int delta = base_idx - (high_idx << base_shifts);
1339	int min = delta, max = delta;
1340	int base_min_idx = base_idx;
1341	int base_max_idx = base_idx;
1342	int high_min_idx = high_idx;
1343	int high_max_idx = high_idx;
1344
1345	// do the baseline_diffs starting from 1 (we did position 0 above)
1346	for(int i = 1; i < base_size; i++) {
1347	K = baseline_diffs[i];
1348	base_idx = binary_search_bigger_than(baseline_diffs, i + 1, base_size, K); // starting from i, since data1 is sorted
1349	high_idx = binary_search_bigger_than(highlight_diffs, 0, high_size, K);
1350
1351	delta = base_idx - (high_idx << base_shifts);
1352	if(delta < min) {
1353	min = delta;
1354	base_min_idx = base_idx;
1355	high_min_idx = high_idx;
1356	}
1357	else if(delta > max) {
1358	max = delta;
1359	base_max_idx = base_idx;
1360	high_max_idx = high_idx;
1361	}
1362	}
1363
1364	// do the highlight_diffs starting from 0
1365	for(int i = 0; i < high_size; i++) {
1366	K = highlight_diffs[i];
1367	base_idx = binary_search_bigger_than(baseline_diffs, 0, base_size, K);
1368	high_idx = binary_search_bigger_than(highlight_diffs, i + 1, high_size, K); // starting from i, since data2 is sorted
1369
1370	delta = base_idx - (high_idx << base_shifts);
1371	if(delta < min) {
1372	min = delta;
1373	base_min_idx = base_idx;
1374	high_min_idx = high_idx;
1375	}
1376	else if(delta > max) {
1377	max = delta;
1378	base_max_idx = base_idx;
1379	high_max_idx = high_idx;
1380	}
1381	}
1382
1383	// now we have the min, max and their indexes
1384	// properly calculate min and max as dmin and dmax
1385	double dbase_size = (double)base_size;
1386	double dhigh_size = (double)high_size;
1387	double dmin = ((double)base_min_idx / dbase_size) - ((double)high_min_idx / dhigh_size);
1388	double dmax = ((double)base_max_idx / dbase_size) - ((double)high_max_idx / dhigh_size);
1389
1390	dmin = -dmin;
1391	if(islessequal(dmin, 0.0)) dmin = 0.0;
1392	else if(isgreaterequal(dmin, 1.0)) dmin = 1.0;
1393
1394	double d;
1395	if(isgreaterequal(dmin, dmax)) d = dmin;
1396	else d = dmax;
1397
1398	double en = round(dbase_size * dhigh_size / (dbase_size + dhigh_size));
1399
1400	// under these conditions, KSfbar() crashes
1401	if(unlikely(isnan(en) \|\| isinf(en) \|\| en == 0.0 \|\| isnan(d) \|\| isinf(d)))
1402	return NAN;
1403
1404	return KSfbar((int)en, d);
1405	}
1406
1407	static double kstwo(
1408	ONEWAYALLOC *owa,
1409	NETDATA_DOUBLE baseline[], int baseline_points,
1410	NETDATA_DOUBLE highlight[], int highlight_points,
1411	uint32_t base_shifts) {
1412
1413	if(unlikely(baseline_points <= 1 \|\| highlight_points <= 1))
1414	return NAN;
1415
1416	// -1 in size, since the calculate_pairs_diffs() returns one less point
1417	DIFFS_NUMBERS baseline_diffs = onewayalloc_mallocz(owa, (size_t)(baseline_points - 1) sizeof(*baseline_diffs));
1418	DIFFS_NUMBERS highlight_diffs = onewayalloc_mallocz(owa, (size_t)(highlight_points - 1) sizeof(*highlight_diffs));
1419
1420	int base_size = (int)calculate_pairs_diff(baseline_diffs, baseline, baseline_points);
1421	int high_size = (int)calculate_pairs_diff(highlight_diffs, highlight, highlight_points);
1422
1423	if(unlikely(!base_size \|\| !high_size))
1424	return NAN;
1425
1426	if(unlikely(base_size != baseline_points - 1 \|\| high_size != highlight_points - 1)) {
1427	netdata_log_error("Metric correlations: internal error - calculate_pairs_diff() returns the wrong number of entries");
1428	return NAN;
1429	}
1430
1431	return ks_2samp(baseline_diffs, base_size, highlight_diffs, high_size, base_shifts);
1432	}
1433
1434	NETDATA_DOUBLE *rrd2rrdr_ks2(
1435	ONEWAYALLOC owa, RRDHOST host,
1436	RRDCONTEXT_ACQUIRED rca, RRDINSTANCE_ACQUIRED ria, RRDMETRIC_ACQUIRED *rma,
1437	time_t after, time_t before, size_t points, RRDR_OPTIONS options,
1438	RRDR_TIME_GROUPING time_group_method, const char *time_group_options, size_t tier,
1439	WEIGHTS_STATS *stats,
1440	size_t *entries,
1441	STORAGE_POINT *sp
1442	) {
1443
1444	NETDATA_DOUBLE *ret = NULL;
1445
1446	QUERY_TARGET_REQUEST qtr = {
1447	.version = 1,
1448	.host = host,
1449	.rca = rca,
1450	.ria = ria,
1451	.rma = rma,
1452	.after = after,
1453	.before = before,
1454	.points = points,
1455	.options = options,
1456	.time_group_method = time_group_method,
1457	.time_group_options = time_group_options,
1458	.tier = tier,
1459	.query_source = QUERY_SOURCE_API_WEIGHTS,
1460	.priority = STORAGE_PRIORITY_SYNCHRONOUS_FIRST,
1461	};
1462
1463	QUERY_TARGET *qt = query_target_create(&qtr);
1464	stream_control_user_weights_query_started();
1465	RRDR *r = rrd2rrdr(owa, qt);
1466	stream_control_user_weights_query_finished();
1467
1468	if(!r)
1469	goto cleanup;
1470
1471	stats->db_queries++;
1472	stats->result_points += r->stats.result_points_generated;
1473	stats->db_points += r->stats.db_points_read;
1474	for(size_t tr = 0; tr < nd_profile.storage_tiers; tr++)
1475	stats->db_points_per_tier[tr] += r->internal.qt->db.tiers[tr].points;
1476
1477	if(!r->d \|\| !r->internal.qt->query.used) {
1478	// the result is empty - no data to query for this metric
1479	goto cleanup;
1480	}
1481
1482	if(r->d != 1 \|\| r->internal.qt->query.used != 1) {
1483	netdata_log_error("WEIGHTS: on query '%s' expected 1 dimension in RRDR but got %zu r->d and %zu qt->query.used",
1484	r->internal.qt->id, r->d, (size_t)r->internal.qt->query.used);
1485	goto cleanup;
1486	}
1487
1488	if(unlikely(r->od[0] & RRDR_DIMENSION_HIDDEN))
1489	goto cleanup;
1490
1491	if(unlikely(!(r->od[0] & RRDR_DIMENSION_QUERIED)))
1492	goto cleanup;
1493
1494	if(unlikely(!(r->od[0] & RRDR_DIMENSION_NONZERO)))
1495	goto cleanup;
1496
1497	if(rrdr_rows(r) < 2)
1498	goto cleanup;
1499
1500	*entries = rrdr_rows(r);
1501	ret = onewayalloc_mallocz(owa, sizeof(NETDATA_DOUBLE) * rrdr_rows(r));
1502
1503	if(sp)
1504	*sp = r->internal.qt->query.array[0].query_points;
1505
1506	// copy the points of the dimension to a contiguous array
1507	// there is no need to check for empty values, since empty values are already zero
1508	// https://github.com/netdata/netdata/blob/6e3144683a73a2024d51425b20ecfd569034c858/web/api/queries/average/average.c#L41-L43
1509	memcpy(ret, r->v, rrdr_rows(r) * sizeof(NETDATA_DOUBLE));
1510
1511	cleanup:
1512	rrdr_free(owa, r);
1513	query_target_release(qt);
1514	return ret;
1515	}
1516
1517	static void rrdset_metric_correlations_ks2(
1518	RRDHOST *host,
1519	RRDCONTEXT_ACQUIRED rca, RRDINSTANCE_ACQUIRED ria, RRDMETRIC_ACQUIRED *rma,
1520	DICTIONARY *results,
1521	time_t baseline_after, time_t baseline_before,
1522	time_t after, time_t before,
1523	size_t points, RRDR_OPTIONS options,
1524	RRDR_TIME_GROUPING time_group_method, const char *time_group_options, size_t tier,
1525	uint32_t shifts,
1526	WEIGHTS_STATS *stats, bool register_zero
1527	) {
1528
1529	options \|= RRDR_OPTION_NATURAL_POINTS;
1530
1531	usec_t started_ut = now_monotonic_usec();
1532	ONEWAYALLOC owa = onewayalloc_create(16 1024);
1533
1534	size_t high_points = 0;
1535	STORAGE_POINT highlighted_sp;
1536	NETDATA_DOUBLE highlight = NULL, baseline = NULL;
1537
1538	highlight = rrd2rrdr_ks2(
1539	owa, host, rca, ria, rma, after, before, points,
1540	options, time_group_method, time_group_options, tier, stats, &high_points, &highlighted_sp);
1541
1542	if(!highlight)
1543	goto cleanup;
1544
1545	size_t base_points = 0;
1546	STORAGE_POINT baseline_sp;
1547	baseline = rrd2rrdr_ks2(
1548	owa, host, rca, ria, rma, baseline_after, baseline_before, high_points << shifts,
1549	options, time_group_method, time_group_options, tier, stats, &base_points, &baseline_sp);
1550
1551	if(!baseline)
1552	goto cleanup;
1553
1554	stats->binary_searches += 2 * (base_points - 1) + 2 * (high_points - 1);
1555
1556	double prob = kstwo(owa, baseline, (int)base_points, highlight, (int)high_points, shifts);
1557	if(!isnan(prob) && !isinf(prob)) {
1558
1559	// these conditions should never happen, but still let's check
1560	if(unlikely(prob < 0.0)) {
1561	netdata_log_error("Metric correlations: kstwo() returned a negative number: %f", prob);
1562	prob = -prob;
1563	}
1564	if(unlikely(prob > 1.0)) {
1565	netdata_log_error("Metric correlations: kstwo() returned a number above 1.0: %f", prob);
1566	prob = 1.0;
1567	}
1568
1569	usec_t ended_ut = now_monotonic_usec();
1570
1571	// to spread the results evenly, 0.0 needs to be the less correlated and 1.0 the most correlated
1572	// so, we flip the result of kstwo()
1573	register_result(results, host, rca, ria, rma, 1.0 - prob, RESULT_IS_BASE_HIGH_RATIO, &highlighted_sp,
1574	&baseline_sp, stats, register_zero, ended_ut - started_ut);
1575	}
1576
1577	cleanup:
1578	onewayalloc_freez(owa, highlight);
1579	onewayalloc_freez(owa, baseline);
1580	onewayalloc_destroy(owa);
1581	}
1582
1583	// ----------------------------------------------------------------------------
1584	// VOLUME algorithm functions
1585
1586	static void merge_query_value_to_stats(QUERY_VALUE qv, WEIGHTS_STATS stats, size_t queries) {
1587	stats->db_queries += queries;
1588	stats->result_points += qv->result_points;
1589	stats->db_points += qv->points_read;
1590	for(size_t tier = 0; tier < nd_profile.storage_tiers; tier++)
1591	stats->db_points_per_tier[tier] += qv->storage_points_per_tier[tier];
1592	}
1593
1594	static void rrdset_metric_correlations_volume(
1595	RRDHOST *host,
1596	RRDCONTEXT_ACQUIRED rca, RRDINSTANCE_ACQUIRED ria, RRDMETRIC_ACQUIRED *rma,
1597	DICTIONARY *results,
1598	time_t baseline_after, time_t baseline_before,
1599	time_t after, time_t before,
1600	RRDR_OPTIONS options, RRDR_TIME_GROUPING time_group_method, const char *time_group_options,
1601	size_t tier,
1602	WEIGHTS_STATS *stats, bool register_zero) {
1603
1604	options \|= RRDR_OPTION_MATCH_IDS \| RRDR_OPTION_ABSOLUTE \| RRDR_OPTION_NATURAL_POINTS;
1605
1606	QUERY_VALUE baseline_average = rrdmetric2value(host, rca, ria, rma, baseline_after, baseline_before,
1607	options, time_group_method, time_group_options, tier, 0,
1608	QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);
1609	merge_query_value_to_stats(&baseline_average, stats, 1);
1610
1611	if(!netdata_double_isnumber(baseline_average.value)) {
1612	// this means no data for the baseline window, but we may have data for the highlighted one - assume zero
1613	baseline_average.value = 0.0;
1614	}
1615
1616	QUERY_VALUE highlight_average = rrdmetric2value(host, rca, ria, rma, after, before,
1617	options, time_group_method, time_group_options, tier, 0,
1618	QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);
1619	merge_query_value_to_stats(&highlight_average, stats, 1);
1620
1621	if(!netdata_double_isnumber(highlight_average.value))
1622	return;
1623
1624	if(baseline_average.value == highlight_average.value) {
1625	// they are the same - let's move on
1626	return;
1627	}
1628
1629	if((options & RRDR_OPTION_ANOMALY_BIT) && highlight_average.value < baseline_average.value) {
1630	// when working on anomaly bits, we are looking for an increase in the anomaly rate
1631	return;
1632	}
1633
1634	char highlight_countif_options[50 + 1];
1635	snprintfz(highlight_countif_options, 50, "%s" NETDATA_DOUBLE_FORMAT, highlight_average.value < baseline_average.value ? "<" : ">", baseline_average.value);
1636	QUERY_VALUE highlight_countif = rrdmetric2value(host, rca, ria, rma, after, before,
1637	options, RRDR_GROUPING_COUNTIF, highlight_countif_options, tier, 0,
1638	QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);
1639	merge_query_value_to_stats(&highlight_countif, stats, 1);
1640
1641	if(!netdata_double_isnumber(highlight_countif.value)) {
1642	netdata_log_info("WEIGHTS: highlighted countif query failed, but highlighted average worked - strange...");
1643	return;
1644	}
1645
1646	// this represents the percentage of time
1647	// the highlighted window was above/below the baseline window
1648	// (above or below depending on their averages)
1649	highlight_countif.value = highlight_countif.value / 100.0; // countif returns 0 - 100.0
1650
1651	RESULT_FLAGS flags;
1652	NETDATA_DOUBLE pcent = NAN;
1653	if(isgreater(baseline_average.value, 0.0) \|\| isless(baseline_average.value, 0.0)) {
1654	flags = RESULT_IS_BASE_HIGH_RATIO;
1655	pcent = (highlight_average.value - baseline_average.value) / baseline_average.value * highlight_countif.value;
1656	}
1657	else {
1658	flags = RESULT_IS_PERCENTAGE_OF_TIME;
1659	pcent = highlight_countif.value;
1660	}
1661
1662	register_result(results, host, rca, ria, rma, pcent, flags, &highlight_average.sp, &baseline_average.sp, stats,
1663	register_zero, baseline_average.duration_ut + highlight_average.duration_ut + highlight_countif.duration_ut);
1664	}
1665
1666	// ----------------------------------------------------------------------------
1667	// VALUE / ANOMALY RATE algorithm functions
1668
1669	static void rrdset_weights_value(
1670	RRDHOST *host,
1671	RRDCONTEXT_ACQUIRED rca, RRDINSTANCE_ACQUIRED ria, RRDMETRIC_ACQUIRED *rma,
1672	DICTIONARY *results,
1673	time_t after, time_t before,
1674	RRDR_OPTIONS options, RRDR_TIME_GROUPING time_group_method, const char *time_group_options,
1675	size_t tier,
1676	WEIGHTS_STATS *stats, bool register_zero) {
1677
1678	options \|= RRDR_OPTION_MATCH_IDS \| RRDR_OPTION_NATURAL_POINTS;
1679
1680	QUERY_VALUE qv = rrdmetric2value(host, rca, ria, rma, after, before,
1681	options, time_group_method, time_group_options, tier, 0,
1682	QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS_FIRST);
1683
1684	merge_query_value_to_stats(&qv, stats, 1);
1685
1686	if(netdata_double_isnumber(qv.value))
1687	register_result(results, host, rca, ria, rma, qv.value, 0, &qv.sp, NULL, stats, register_zero, qv.duration_ut);
1688	}
1689
1690	static void rrdset_weights_multi_dimensional_value(struct query_weights_data *qwd) {
1691	QUERY_TARGET_REQUEST qtr = {
1692	.version = 1,
1693	.scope_nodes = qwd->qwr->scope_nodes,
1694	.scope_contexts = qwd->qwr->scope_contexts,
1695	.scope_instances = qwd->qwr->scope_instances,
1696	.scope_labels = qwd->qwr->scope_labels,
1697	.scope_dimensions = qwd->qwr->scope_dimensions,
1698	.nodes = qwd->qwr->nodes,
1699	.contexts = qwd->qwr->contexts,
1700	.instances = qwd->qwr->instances,
1701	.dimensions = qwd->qwr->dimensions,
1702	.labels = qwd->qwr->labels,
1703	.alerts = qwd->qwr->alerts,
1704	.after = qwd->qwr->after,
1705	.before = qwd->qwr->before,
1706	.points = 1,
1707	.options = qwd->qwr->options \| RRDR_OPTION_NATURAL_POINTS,
1708	.time_group_method = qwd->qwr->time_group_method,
1709	.time_group_options = qwd->qwr->time_group_options,
1710	.tier = qwd->qwr->tier,
1711	.timeout_ms = qwd->qwr->timeout_ms,
1712	.query_source = QUERY_SOURCE_API_WEIGHTS,
1713	.priority = STORAGE_PRIORITY_SYNCHRONOUS_FIRST,
1714	};
1715
1716	ONEWAYALLOC owa = onewayalloc_create(16 1024);
1717	QUERY_TARGET *qt = query_target_create(&qtr);
1718	stream_control_user_weights_query_started();
1719	RRDR *r = rrd2rrdr(owa, qt);
1720	stream_control_user_weights_query_finished();
1721
1722	if(!r \|\| rrdr_rows(r) != 1 \|\| !r->d \|\| r->d != r->internal.qt->query.used)
1723	goto cleanup;
1724
1725	QUERY_VALUE qv = {
1726	.after = r->view.after,
1727	.before = r->view.before,
1728	.points_read = r->stats.db_points_read,
1729	.result_points = r->stats.result_points_generated,
1730	};
1731
1732	size_t queries = 0;
1733	for(size_t d = 0; d < r->d ;d++) {
1734	qwd->examined_dimensions++;
1735
1736	if(!rrdr_dimension_should_be_exposed(r->od[d], qwd->qwr->options))
1737	continue;
1738
1739	long i = 0; // only one row
1740	NETDATA_DOUBLE cn = &r->v[ i r->d ];
1741	NETDATA_DOUBLE ar = &r->ar[ i r->d ];
1742
1743	qv.value = cn[d];
1744	qv.anomaly_rate = ar[d];
1745	storage_point_merge_to(qv.sp, r->internal.qt->query.array[d].query_points);
1746
1747	if(netdata_double_isnumber(qv.value)) {
1748	QUERY_METRIC *qm = query_metric(r->internal.qt, d);
1749	QUERY_DIMENSION *qd = query_dimension(r->internal.qt, qm->link.query_dimension_id);
1750	QUERY_INSTANCE *qi = query_instance(r->internal.qt, qm->link.query_instance_id);
1751	QUERY_CONTEXT *qc = query_context(r->internal.qt, qm->link.query_context_id);
1752	QUERY_NODE *qn = query_node(r->internal.qt, qm->link.query_node_id);
1753
1754	register_result(qwd->results, qn->rrdhost, qc->rca, qi->ria, qd->rma, qv.value, 0,
1755	&r->internal.qt->query.array[d].query_points, NULL,
1756	&qwd->stats, qwd->register_zero, qm->duration_ut);
1757	}
1758
1759	queries++;
1760	}
1761
1762	merge_query_value_to_stats(&qv, &qwd->stats, queries);
1763
1764	cleanup:
1765	rrdr_free(owa, r);
1766	query_target_release(qt);
1767	onewayalloc_destroy(owa);
1768	}
1769
1770	// ----------------------------------------------------------------------------
1771
1772	int compare_netdata_doubles(const void left, const void right) {
1773	NETDATA_DOUBLE lt = (NETDATA_DOUBLE )left;
1774	NETDATA_DOUBLE rt = (NETDATA_DOUBLE )right;
1775
1776	// https://stackoverflow.com/a/3886497/1114110
1777	return (lt > rt) - (lt < rt);
1778	}
1779
1780	static inline int binary_search_bigger_than_netdata_double(const NETDATA_DOUBLE arr[], int left, int size, NETDATA_DOUBLE K) {
1781	// binary search to find the index the smallest index
1782	// of the first value in the array that is greater than K
1783
1784	int right = size;
1785	while(left < right) {
1786	int middle = (int)(((unsigned int)(left + right)) >> 1);
1787
1788	if(arr[middle] > K)
1789	right = middle;
1790
1791	else
1792	left = middle + 1;
1793	}
1794
1795	return left;
1796	}
1797
1798	// ----------------------------------------------------------------------------
1799	// spread the results evenly according to their value
1800
1801	static size_t spread_results_evenly(DICTIONARY results, WEIGHTS_STATS stats) {
1802	struct register_result *t;
1803
1804	// count the dimensions
1805	size_t dimensions = dictionary_entries(results);
1806	if(!dimensions) return 0;
1807
1808	if(stats->max_base_high_ratio == 0.0)
1809	stats->max_base_high_ratio = 1.0;
1810
1811	// create an array of the right size and copy all the values in it
1812	NETDATA_DOUBLE slots = mallocz(dimensions sizeof(*slots));
1813	dimensions = 0;
1814	dfe_start_read(results, t) {
1815	if(t->flags & RESULT_IS_PERCENTAGE_OF_TIME)
1816	t->value = t->value * stats->max_base_high_ratio;
1817
1818	slots[dimensions++] = t->value;
1819	}
1820	dfe_done(t);
1821
1822	if(!dimensions) {
1823	freez(slots);
1824	return 0; // Coverity fix
1825	}
1826
1827	// sort the array with the values of all dimensions
1828	qsort(slots, dimensions, sizeof(NETDATA_DOUBLE), compare_netdata_doubles);
1829
1830	// skip the duplicates in the sorted array
1831	NETDATA_DOUBLE last_value = NAN;
1832	size_t unique_values = 0;
1833	for(size_t i = 0; i < dimensions ;i++) {
1834	if(likely(slots[i] != last_value))
1835	slots[unique_values++] = last_value = slots[i];
1836	}
1837
1838	// this cannot happen, but coverity thinks otherwise...
1839	if(!unique_values)
1840	unique_values = dimensions;
1841
1842	// calculate the weight of each slot, using the number of unique values
1843	NETDATA_DOUBLE slot_weight = 1.0 / (NETDATA_DOUBLE)unique_values;
1844
1845	dfe_start_read(results, t) {
1846	int slot = binary_search_bigger_than_netdata_double(slots, 0, (int)unique_values, t->value);
1847	NETDATA_DOUBLE v = slot * slot_weight;
1848	if(unlikely(v > 1.0)) v = 1.0;
1849	v = 1.0 - v;
1850	t->value = v;
1851	}
1852	dfe_done(t);
1853
1854	freez(slots);
1855	return dimensions;
1856	}
1857
1858	// ----------------------------------------------------------------------------
1859	// MCP format output
1860
1861	// Comparator for sorting results by value (descending order - highest scores first)
1862	static int registered_results_value_compare(const DICTIONARY_ITEM item1, const DICTIONARY_ITEM item2) {
1863	struct register_result r1 = dictionary_acquired_item_value(item1);
1864	struct register_result r2 = dictionary_acquired_item_value(item2);
1865
1866	// Sort by value in descending order (highest first)
1867	if (r1->value < r2->value) return 1;
1868	if (r1->value > r2->value) return -1;
1869	return 0;
1870	}
1871
1872	// Callback for sorted dictionary walkthrough
1873	struct mcp_output_state {
1874	BUFFER *wb;
1875	WEIGHTS_METHOD method;
1876	size_t count;
1877	size_t limit;
1878	};
1879
1880	static int registered_results_to_json_mcp_callback(const DICTIONARY_ITEM item __maybe_unused, void value, void *data) {
1881	struct mcp_output_state state = (struct mcp_output_state )data;
1882	struct register_result t = (struct register_result )value;
1883
1884	// Check if we've reached the cardinality limit
1885	if (state->count >= state->limit)
1886	return -1; // Stop iteration
1887
1888	BUFFER *wb = state->wb;
1889
1890	buffer_json_add_array_item_array(wb); // Start row array
1891
1892	// Add score/value based on method
1893	switch(state->method) {
1894	case WEIGHTS_METHOD_MC_KS2:
1895	case WEIGHTS_METHOD_MC_VOLUME:
1896	buffer_json_add_array_item_double(wb, t->value);
1897	break;
1898
1899	case WEIGHTS_METHOD_ANOMALY_RATE:
1900	// For anomaly rate, the value is already a percentage
1901	buffer_json_add_array_item_double(wb, t->value);
1902	break;
1903
1904	case WEIGHTS_METHOD_VALUE:
1905	// For CV or other aggregations
1906	buffer_json_add_array_item_double(wb, t->value);
1907	break;
1908	}
1909
1910	// Add the 5 statistical values
1911	// 1. Min
1912	if(storage_point_is_unset(t->highlighted) \|\| storage_point_is_gap(t->highlighted))
1913	buffer_json_add_array_item_double(wb, NAN);
1914	else
1915	buffer_json_add_array_item_double(wb, t->highlighted.min);
1916
1917	// 2. Max
1918	if(storage_point_is_unset(t->highlighted) \|\| storage_point_is_gap(t->highlighted))
1919	buffer_json_add_array_item_double(wb, NAN);
1920	else
1921	buffer_json_add_array_item_double(wb, t->highlighted.max);
1922
1923	// 3. Average
1924	if(storage_point_is_unset(t->highlighted) \|\| storage_point_is_gap(t->highlighted) \|\| t->highlighted.count == 0)
1925	buffer_json_add_array_item_double(wb, NAN);
1926	else
1927	buffer_json_add_array_item_double(wb, t->highlighted.sum / (NETDATA_DOUBLE)t->highlighted.count);
1928
1929	// 4. Number of samples in window
1930	buffer_json_add_array_item_uint64(wb, t->highlighted.count);
1931
1932	// 5. Number of anomalous samples in window
1933	buffer_json_add_array_item_double(wb, t->highlighted.anomaly_count);
1934
1935	// Add metadata
1936	// Add node name
1937	buffer_json_add_array_item_string(wb, rrdhost_hostname(t->host));
1938
1939	// Add context
1940	buffer_json_add_array_item_string(wb, rrdcontext_acquired_id(t->rca));
1941
1942	// Add instance
1943	buffer_json_add_array_item_string(wb, rrdinstance_acquired_id(t->ria));
1944
1945	// Add dimension
1946	buffer_json_add_array_item_string(wb, rrdmetric_acquired_name(t->rma));
1947
1948	// Add labels (as object or null)
1949	RRDLABELS *labels = rrdinstance_acquired_labels(t->ria);
1950	if(labels && rrdlabels_entries(labels) > 0) {
1951	buffer_json_add_array_item_object(wb);
1952	rrdlabels_to_buffer_json_members(labels, wb);
1953	buffer_json_object_close(wb);
1954	}
1955	else {
1956	buffer_json_add_array_item_string(wb, NULL);
1957	}
1958
1959	buffer_json_array_close(wb); // End row array
1960
1961	state->count++;
1962	return 0; // Continue iteration
1963	}
1964
1965	static size_t registered_results_to_json_mcp(
1966	DICTIONARY results, BUFFER wb,
1967	time_t after __maybe_unused, time_t before __maybe_unused,
1968	time_t baseline_after __maybe_unused, time_t baseline_before __maybe_unused,
1969	size_t points __maybe_unused, WEIGHTS_METHOD method,
1970	RRDR_TIME_GROUPING group __maybe_unused, RRDR_OPTIONS options, uint32_t shifts __maybe_unused,
1971	size_t examined_dimensions __maybe_unused, struct query_weights_data *qwd,
1972	WEIGHTS_STATS *stats __maybe_unused,
1973	struct query_versions *versions __maybe_unused) {
1974
1975	buffer_json_initialize(wb, "\"", "\"", 0, true, (options & RRDR_OPTION_MINIFY) ? BUFFER_JSON_OPTIONS_MINIFY : BUFFER_JSON_OPTIONS_DEFAULT);
1976
1977	// Add columns array based on method
1978	buffer_json_member_add_array(wb, "columns");
1979
1980	switch(method) {
1981	case WEIGHTS_METHOD_MC_KS2:
1982	buffer_json_add_array_item_string(wb, "KS2 Score");
1983	break;
1984
1985	case WEIGHTS_METHOD_MC_VOLUME:
1986	buffer_json_add_array_item_string(wb, "Volume Score");
1987	break;
1988
1989	case WEIGHTS_METHOD_ANOMALY_RATE:
1990	buffer_json_add_array_item_string(wb, "Anomaly Rate");
1991	break;
1992
1993	case WEIGHTS_METHOD_VALUE:
1994	buffer_json_add_array_item_string(wb, "Coefficient of Variation");
1995	break;
1996	}
1997
1998	// Common statistical columns for all methods
1999	buffer_json_add_array_item_string(wb, "Minimum Sample Value");
2000	buffer_json_add_array_item_string(wb, "Maximum Sample Value");
2001	buffer_json_add_array_item_string(wb, "Average Sample Value");
2002	buffer_json_add_array_item_string(wb, "# of Samples in Window");
2003	buffer_json_add_array_item_string(wb, "# of Anomalous Samples in Window");
2004
2005	// Metadata columns
2006	buffer_json_add_array_item_string(wb, "Hostname");
2007	buffer_json_add_array_item_string(wb, "Context / Metric Name");
2008	buffer_json_add_array_item_string(wb, "Metrics Instance");
2009	buffer_json_add_array_item_string(wb, "Dimension");
2010	buffer_json_add_array_item_string(wb, "Instance Labels");
2011
2012	buffer_json_array_close(wb); // columns
2013
2014	// Add results array
2015	buffer_json_member_add_array(wb, "results");
2016
2017	// Get cardinality limit from query weights data
2018	size_t cardinality_limit = qwd && qwd->qwr ? qwd->qwr->cardinality_limit : 50;
2019	if (cardinality_limit < 30) cardinality_limit = 30;
2020
2021	// Set up state for callback
2022	struct mcp_output_state state = {
2023	.wb = wb,
2024	.method = method,
2025	.count = 0,
2026	.limit = cardinality_limit
2027	};
2028
2029	// Walk through dictionary in sorted order (by value descending)
2030	dictionary_sorted_walkthrough_rw(results, 'r', registered_results_to_json_mcp_callback, &state, registered_results_value_compare);
2031
2032	buffer_json_array_close(wb); // results
2033
2034	// Add metadata
2035	buffer_json_member_add_object(wb, "metadata");
2036	buffer_json_member_add_uint64(wb, "total_time_series_analyzed", examined_dimensions);
2037	buffer_json_member_add_uint64(wb, "total_time_series_returned", state.count);
2038	buffer_json_member_add_string(wb, "method", weights_method_to_string(method));
2039	if (state.count >= cardinality_limit) {
2040	buffer_json_member_add_uint64(wb, "cardinality_limit", cardinality_limit);
2041	buffer_json_member_add_boolean(wb, "truncated", true);
2042	}
2043	buffer_json_object_close(wb); // metadata
2044
2045	buffer_json_finalize(wb);
2046
2047	return state.count;
2048	}
2049
2050	static ssize_t weights_count_for_rrdmetric(
2051	void *data,
2052	RRDHOST *host __maybe_unused,
2053	RRDCONTEXT_ACQUIRED *rca __maybe_unused,
2054	RRDINSTANCE_ACQUIRED *ria __maybe_unused,
2055	RRDMETRIC_ACQUIRED *rma __maybe_unused)
2056	{
2057	struct query_weights_data *qwd = data;
2058
2059	__atomic_fetch_add(&qwd->total_workload.metrics, 1, __ATOMIC_RELAXED);
2060	return 1;
2061	}
2062
2063	// ----------------------------------------------------------------------------
2064	// The main function
2065
2066	static ssize_t weights_for_rrdmetric(void data, RRDHOST host, RRDCONTEXT_ACQUIRED rca, RRDINSTANCE_ACQUIRED ria, RRDMETRIC_ACQUIRED *rma) {
2067	struct query_weights_data *qwd = data;
2068	QUERY_WEIGHTS_REQUEST *qwr = qwd->qwr;
2069
2070	if(qwd->qwr->interrupt_callback && qwd->qwr->interrupt_callback(qwd->qwr->interrupt_callback_data)) {
2071	__atomic_store_n(&qwd->interrupted, true, __ATOMIC_RELAXED);
2072	return -1;
2073	}
2074
2075	__atomic_fetch_add(&qwd->examined_dimensions, 1, __ATOMIC_RELAXED);
2076
2077	switch(qwr->method) {
2078	case WEIGHTS_METHOD_VALUE:
2079	rrdset_weights_value(
2080	host, rca, ria, rma,
2081	qwd->results,
2082	qwr->after, qwr->before,
2083	qwr->options, qwr->time_group_method, qwr->time_group_options, qwr->tier,
2084	&qwd->stats, qwd->register_zero
2085	);
2086	break;
2087
2088	case WEIGHTS_METHOD_ANOMALY_RATE:
2089	qwr->options \|= RRDR_OPTION_ANOMALY_BIT;
2090	rrdset_weights_value(
2091	host, rca, ria, rma,
2092	qwd->results,
2093	qwr->after, qwr->before,
2094	qwr->options, qwr->time_group_method, qwr->time_group_options, qwr->tier,
2095	&qwd->stats, qwd->register_zero
2096	);
2097	break;
2098
2099	case WEIGHTS_METHOD_MC_VOLUME:
2100	rrdset_metric_correlations_volume(
2101	host, rca, ria, rma,
2102	qwd->results,
2103	qwr->baseline_after, qwr->baseline_before,
2104	qwr->after, qwr->before,
2105	qwr->options, qwr->time_group_method, qwr->time_group_options, qwr->tier,
2106	&qwd->stats, qwd->register_zero
2107	);
2108	break;
2109
2110	default:
2111	case WEIGHTS_METHOD_MC_KS2:
2112	rrdset_metric_correlations_ks2(
2113	host, rca, ria, rma,
2114	qwd->results,
2115	qwr->baseline_after, qwr->baseline_before,
2116	qwr->after, qwr->before, qwr->points,
2117	qwr->options, qwr->time_group_method, qwr->time_group_options, qwr->tier, qwd->shifts,
2118	&qwd->stats, qwd->register_zero
2119	);
2120	break;
2121	}
2122
2123	qwd->timings.executed_ut = now_monotonic_usec();
2124	if(qwd->timings.executed_ut - qwd->timings.received_ut > qwd->timeout_us) {
2125	qwd->timed_out = true;
2126	return -1;
2127	}
2128
2129	query_progress_done_step(qwr->transaction, 1);
2130
2131	return 1;
2132	}
2133
2134	static ssize_t weights_count_context_callback(void data, RRDCONTEXT_ACQUIRED rca, bool queryable_context) {
2135	if(!queryable_context)
2136	return false;
2137
2138	struct query_weights_data *qwd = data;
2139
2140	bool has_retention = false;
2141	switch(qwd->qwr->method) {
2142	case WEIGHTS_METHOD_VALUE:
2143	case WEIGHTS_METHOD_ANOMALY_RATE:
2144	has_retention = rrdcontext_retention_match(rca, qwd->qwr->after, qwd->qwr->before);
2145	break;
2146
2147	case WEIGHTS_METHOD_MC_KS2:
2148	case WEIGHTS_METHOD_MC_VOLUME:
2149	has_retention = rrdcontext_retention_match(rca, qwd->qwr->after, qwd->qwr->before);
2150	if(has_retention)
2151	has_retention = rrdcontext_retention_match(rca, qwd->qwr->baseline_after, qwd->qwr->baseline_before);
2152	break;
2153	}
2154
2155	if(!has_retention)
2156	return 0;
2157
2158	__atomic_fetch_add(&qwd->total_workload.contexts, 1, __ATOMIC_RELAXED);
2159	ssize_t ret = weights_foreach_rrdmetric_in_context(rca,
2160	qwd->scope_instances_sp,
2161	qwd->scope_labels_pa,
2162	qwd->scope_dimensions_sp,
2163	qwd->instances_sp,
2164	NULL,
2165	qwd->labels_pa,
2166	qwd->alerts_sp,
2167	qwd->dimensions_sp,
2168	true, true, qwd->qwr->version,
2169	weights_count_for_rrdmetric, qwd);
2170	if (ret >= 1)
2171	return 1;
2172	else
2173	return 0;
2174	}
2175
2176	static ssize_t weights_count_node_callback(void data, RRDHOST host, bool queryable) {
2177	if(!queryable)
2178	return 0;
2179
2180	struct query_weights_data *qwd = data;
2181	if (qwd->total_hosts >= qwd->hosts_array_capacity) {
2182	qwd->hosts_array_capacity *= 2;
2183	qwd->hosts_array = reallocz(qwd->hosts_array, sizeof(RRDHOST ) qwd->hosts_array_capacity);
2184	}
2185	qwd->hosts_array[qwd->total_hosts++] = host;
2186
2187	__atomic_fetch_add(&qwd->total_workload.nodes, 1, __ATOMIC_RELAXED);
2188	ssize_t ret = query_scope_foreach_context(host, qwd->qwr->scope_contexts,
2189	qwd->scope_contexts_sp, qwd->contexts_sp,
2190	weights_count_context_callback, queryable, qwd);
2191
2192	return ret;
2193	}
2194
2195	static ssize_t weights_do_context_callback(void data, RRDCONTEXT_ACQUIRED rca, bool queryable_context) {
2196	if(!queryable_context)
2197	return false;
2198
2199	struct query_weights_data *qwd = data;
2200
2201	bool has_retention = false;
2202	switch(qwd->qwr->method) {
2203	case WEIGHTS_METHOD_VALUE:
2204	case WEIGHTS_METHOD_ANOMALY_RATE:
2205	has_retention = rrdcontext_retention_match(rca, qwd->qwr->after, qwd->qwr->before);
2206	break;
2207
2208	case WEIGHTS_METHOD_MC_KS2:
2209	case WEIGHTS_METHOD_MC_VOLUME:
2210	has_retention = rrdcontext_retention_match(rca, qwd->qwr->after, qwd->qwr->before);
2211	if(has_retention)
2212	has_retention = rrdcontext_retention_match(rca, qwd->qwr->baseline_after, qwd->qwr->baseline_before);
2213	break;
2214	}
2215
2216	if(!has_retention)
2217	return 0;
2218
2219	ssize_t ret = weights_foreach_rrdmetric_in_context(rca,
2220	qwd->scope_instances_sp,
2221	qwd->scope_labels_pa,
2222	qwd->scope_dimensions_sp,
2223	qwd->instances_sp,
2224	NULL,
2225	qwd->labels_pa,
2226	qwd->alerts_sp,
2227	qwd->dimensions_sp,
2228	true, true, qwd->qwr->version,
2229	weights_for_rrdmetric, qwd);
2230	return ret;
2231	}
2232
2233	// Parallel version of query_scope_foreach_host
2234	static ssize_t query_scope_foreach_host_parallel(SIMPLE_PATTERN scope_hosts_sp, SIMPLE_PATTERN hosts_sp,
2235	struct query_weights_data *qwd)
2236	{
2237	#ifndef ENABLE_DBENGINE
2238	return query_scope_foreach_host(scope_hosts_sp, hosts_sp,
2239	weights_do_node_callback, qwd,
2240	&qwd->versions, NULL);
2241
2242	#else
2243	size_t host_count = dictionary_entries(rrdhost_root_index);
2244	qwd->hosts_array = mallocz(sizeof(RRDHOST ) host_count);
2245	qwd->hosts_array_capacity = host_count;
2246	qwd->total_hosts = 0;
2247
2248	(void) query_scope_foreach_host(scope_hosts_sp, hosts_sp, weights_count_node_callback, qwd, &qwd->versions, NULL);
2249
2250	size_t active_hosts = qwd->total_hosts;
2251
2252	size_t num_threads = netdata_conf_cpus();
2253	if (num_threads < 1) num_threads = 1;
2254
2255	// If we have fewer hosts than threads, reduce thread count
2256	if (active_hosts < num_threads) {
2257	num_threads = active_hosts;
2258	}
2259
2260	if (num_threads <= 1 \|\| active_hosts <= 1) {
2261	// Fall back to single-threaded processing
2262	freez(qwd->hosts_array);
2263	return query_scope_foreach_host(scope_hosts_sp, hosts_sp,
2264	weights_do_node_callback, qwd,
2265	&qwd->versions, NULL);
2266	}
2267
2268	// Calculate hosts per thread
2269	size_t hosts_per_thread = active_hosts / num_threads;
2270	size_t remaining_hosts = active_hosts % num_threads;
2271
2272	// Prepare thread data
2273	struct query_weights_thread_data thread_data = mallocz(sizeof(struct query_weights_thread_data) num_threads);
2274	ND_THREAD *threads = mallocz(sizeof(ND_THREAD ) * num_threads);
2275
2276	size_t current_host_idx = 0;
2277	for (size_t i = 0; i < num_threads; i++) {
2278	thread_data[i].main_qwd = qwd;
2279	thread_data[i].local_results = register_result_init_single_threaded();
2280	thread_data[i].thread_id = i;
2281	thread_data[i].hosts = &qwd->hosts_array[current_host_idx];
2282
2283	// Distribute hosts evenly, giving extra hosts to first threads
2284	thread_data[i].host_count = hosts_per_thread + (i < remaining_hosts ? 1 : 0);
2285	current_host_idx += thread_data[i].host_count;
2286
2287	completion_init(&thread_data[i].completion);
2288	rrdeng_enq_cmd(NULL, RRDENG_OPCODE_PARALLEL_WEIGHT, &thread_data[i], &thread_data[i].completion, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
2289	}
2290
2291	// Wait for all threads to complete
2292	ssize_t total_added = 0;
2293	for (size_t i = 0; i < num_threads; i++) {
2294	completion_wait_for(&thread_data[i].completion);
2295	completion_destroy(&thread_data[i].completion);
2296
2297	// Merge results from this thread
2298	merge_results_dictionaries(qwd->results, thread_data[i].local_results);
2299	merge_weights_stats(&qwd->stats, &thread_data[i].local_stats);
2300
2301	// Accumulate examined dimensions
2302	__atomic_fetch_add(&qwd->examined_dimensions, thread_data[i].local_examined_dimensions, __ATOMIC_RELAXED);
2303
2304	// Merge version hashes
2305	qwd->versions.contexts_hard_hash += thread_data[i].local_versions.contexts_hard_hash;
2306	qwd->versions.contexts_soft_hash += thread_data[i].local_versions.contexts_soft_hash;
2307	qwd->versions.alerts_hard_hash += thread_data[i].local_versions.alerts_hard_hash;
2308	qwd->versions.alerts_soft_hash += thread_data[i].local_versions.alerts_soft_hash;
2309
2310	// Clean up thread data
2311	register_result_destroy(thread_data[i].local_results);
2312	}
2313
2314	total_added = (ssize_t) dictionary_entries(qwd->results);
2315
2316	// Cleanup
2317	freez(thread_data);
2318	freez(threads);
2319	freez(qwd->hosts_array);
2320
2321	return total_added;
2322	#endif
2323	}
2324
2325	static ssize_t weights_do_node_callback(void data, RRDHOST host, bool queryable) {
2326	if(!queryable)
2327	return 0;
2328
2329	struct query_weights_data *qwd = data;
2330
2331	ssize_t ret = query_scope_foreach_context(host, qwd->qwr->scope_contexts,
2332	qwd->scope_contexts_sp, qwd->contexts_sp,
2333	weights_do_context_callback, queryable, qwd);
2334
2335	return ret;
2336	}
2337
2338	int web_api_v12_weights(BUFFER wb, QUERY_WEIGHTS_REQUEST qwr) {
2339
2340	char *error = NULL;
2341	int resp = HTTP_RESP_OK;
2342
2343	// if the user didn't give a timeout
2344	// assume 60 seconds
2345	if(!qwr->timeout_ms)
2346	qwr->timeout_ms = 5 * 60 * MSEC_PER_SEC;
2347
2348	// if the timeout is less than 1 second
2349	// make it at least 1 second
2350	if(qwr->timeout_ms < (long)(1 * MSEC_PER_SEC))
2351	qwr->timeout_ms = 1 * MSEC_PER_SEC;
2352
2353	struct query_weights_data qwd = {
2354	.qwr = qwr,
2355
2356	.scope_nodes_sp = string_to_simple_pattern(qwr->scope_nodes),
2357	.scope_contexts_sp = string_to_simple_pattern(qwr->scope_contexts),
2358	.scope_instances_sp = string_to_simple_pattern(qwr->scope_instances),
2359	.scope_labels_sp = string_to_simple_pattern(qwr->scope_labels),
2360	.scope_dimensions_sp = string_to_simple_pattern(qwr->scope_dimensions),
2361	.nodes_sp = string_to_simple_pattern(qwr->nodes),
2362	.contexts_sp = string_to_simple_pattern(qwr->contexts),
2363	.instances_sp = string_to_simple_pattern(qwr->instances),
2364	.dimensions_sp = string_to_simple_pattern(qwr->dimensions),
2365	.labels_sp = string_to_simple_pattern(qwr->labels),
2366	.alerts_sp = string_to_simple_pattern(qwr->alerts),
2367	.scope_labels_pa = NULL,
2368	.labels_pa = NULL,
2369	.timeout_us = qwr->timeout_ms * USEC_PER_MS,
2370	.timed_out = false,
2371	.examined_dimensions = 0,
2372	.register_zero = true,
2373	.results = register_result_init(),
2374	.stats = {},
2375	.shifts = 0,
2376	.total_workload = {0}, // Initialize workload statistics
2377	.timings = {
2378	.received_ut = now_monotonic_usec(),
2379	}
2380	};
2381
2382	// Pre-compile pattern arrays for labels
2383	if(qwd.scope_labels_sp)
2384	qwd.scope_labels_pa = pattern_array_add_simple_pattern(NULL, qwd.scope_labels_sp, ':');
2385	if(qwd.labels_sp)
2386	qwd.labels_pa = pattern_array_add_simple_pattern(NULL, qwd.labels_sp, ':');
2387
2388	if(!rrdr_relative_window_to_absolute_query(&qwr->after, &qwr->before, NULL, false))
2389	buffer_no_cacheable(wb);
2390	else
2391	buffer_cacheable(wb);
2392
2393	if (qwr->before <= qwr->after) {
2394	resp = HTTP_RESP_BAD_REQUEST;
2395	error = "Invalid selected time-range.";
2396	goto cleanup;
2397	}
2398
2399	if(qwr->method == WEIGHTS_METHOD_MC_KS2 \|\| qwr->method == WEIGHTS_METHOD_MC_VOLUME) {
2400	if(!qwr->points) qwr->points = 500;
2401
2402	if(qwr->baseline_before <= API_RELATIVE_TIME_MAX)
2403	qwr->baseline_before += qwr->after;
2404
2405	rrdr_relative_window_to_absolute_query(&qwr->baseline_after, &qwr->baseline_before, NULL, false);
2406
2407	if (qwr->baseline_before <= qwr->baseline_after) {
2408	resp = HTTP_RESP_BAD_REQUEST;
2409	error = "Invalid baseline time-range.";
2410	goto cleanup;
2411	}
2412
2413	// baseline should be a power of two multiple of highlight
2414	long long base_delta = qwr->baseline_before - qwr->baseline_after;
2415	long long high_delta = qwr->before - qwr->after;
2416	uint32_t multiplier = (uint32_t)round((double)base_delta / (double)high_delta);
2417
2418	// check if the multiplier is a power of two
2419	// https://stackoverflow.com/a/600306/1114110
2420	if((multiplier & (multiplier - 1)) != 0) {
2421	// it is not power of two
2422	// let's find the closest power of two
2423	// https://stackoverflow.com/a/466242/1114110
2424	multiplier--;
2425	multiplier \|= multiplier >> 1;
2426	multiplier \|= multiplier >> 2;
2427	multiplier \|= multiplier >> 4;
2428	multiplier \|= multiplier >> 8;
2429	multiplier \|= multiplier >> 16;
2430	multiplier++;
2431	}
2432
2433	// convert the multiplier to the number of shifts
2434	// we need to do, to divide baseline numbers to match
2435	// the highlight ones
2436	while(multiplier > 1) {
2437	qwd.shifts++;
2438	multiplier = multiplier >> 1;
2439	}
2440
2441	// if the baseline size will not comply to MAX_POINTS
2442	// lower the window of the baseline
2443	while(qwd.shifts && (qwr->points << qwd.shifts) > MAX_POINTS)
2444	qwd.shifts--;
2445
2446	// if the baseline size still does not comply to MAX_POINTS
2447	// lower the resolution of the highlight and the baseline
2448	while((qwr->points << qwd.shifts) > MAX_POINTS)
2449	qwr->points = qwr->points >> 1;
2450
2451	if(qwr->points < 15) {
2452	resp = HTTP_RESP_BAD_REQUEST;
2453	error = "Too few points available, at least 15 are needed.";
2454	goto cleanup;
2455	}
2456
2457	// adjust the baseline to be multiplier times bigger than the highlight
2458	qwr->baseline_after = qwr->baseline_before - (high_delta << qwd.shifts);
2459	}
2460
2461	if(qwr->options & RRDR_OPTION_NONZERO) {
2462	qwd.register_zero = false;
2463
2464	// remove it to run the queries without it
2465	qwr->options &= ~RRDR_OPTION_NONZERO;
2466	}
2467
2468	if(qwr->host && qwr->version == 1)
2469	weights_do_node_callback(&qwd, qwr->host, true);
2470	else {
2471	if((qwd.qwr->method == WEIGHTS_METHOD_VALUE \|\| qwd.qwr->method == WEIGHTS_METHOD_ANOMALY_RATE) && (qwd.contexts_sp \|\| qwd.scope_contexts_sp)) {
2472
2473	if(qwd.qwr->format == WEIGHTS_FORMAT_MCP && qwd.qwr->method == WEIGHTS_METHOD_ANOMALY_RATE)
2474	qwd.qwr->options \|= RRDR_OPTION_ANOMALY_BIT;
2475
2476	rrdset_weights_multi_dimensional_value(&qwd);
2477	}
2478	else {
2479	query_scope_foreach_host_parallel(qwd.scope_nodes_sp, qwd.nodes_sp, &qwd);
2480	}
2481	}
2482
2483	if(!qwd.register_zero) {
2484	// put it back, to show it in the response
2485	qwr->options \|= RRDR_OPTION_NONZERO;
2486	}
2487
2488	if(__atomic_load_n(&qwd.timed_out, __ATOMIC_RELAXED)) {
2489	error = "timed out";
2490	resp = HTTP_RESP_GATEWAY_TIMEOUT;
2491	goto cleanup;
2492	}
2493
2494	if(__atomic_load_n(&qwd.interrupted, __ATOMIC_RELAXED)) {
2495	error = "interrupted";
2496	resp = HTTP_RESP_CLIENT_CLOSED_REQUEST;
2497	goto cleanup;
2498	}
2499
2500	if(!qwd.register_zero)
2501	qwr->options \|= RRDR_OPTION_NONZERO;
2502
2503	if(!(qwr->options & RRDR_OPTION_RETURN_RAW) &&
2504	qwr->method != WEIGHTS_METHOD_VALUE &&
2505	qwr->format != WEIGHTS_FORMAT_MCP)
2506	spread_results_evenly(qwd.results, &qwd.stats);
2507
2508	usec_t ended_usec = qwd.timings.executed_ut = now_monotonic_usec();
2509
2510	// generate the json output we need
2511	buffer_flush(wb);
2512
2513	size_t added_dimensions = 0;
2514	switch(qwr->format) {
2515	case WEIGHTS_FORMAT_CHARTS:
2516	added_dimensions =
2517	registered_results_to_json_charts(
2518	qwd.results, wb,
2519	qwr->after, qwr->before,
2520	qwr->baseline_after, qwr->baseline_before,
2521	qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
2522	qwd.examined_dimensions,
2523	ended_usec - qwd.timings.received_ut, &qwd.stats);
2524	break;
2525
2526	case WEIGHTS_FORMAT_CONTEXTS:
2527	added_dimensions =
2528	registered_results_to_json_contexts(
2529	qwd.results, wb,
2530	qwr->after, qwr->before,
2531	qwr->baseline_after, qwr->baseline_before,
2532	qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
2533	qwd.examined_dimensions,
2534	ended_usec - qwd.timings.received_ut, &qwd.stats);
2535	break;
2536
2537	case WEIGHTS_FORMAT_MCP:
2538	added_dimensions =
2539	registered_results_to_json_mcp(
2540	qwd.results, wb,
2541	qwr->after, qwr->before,
2542	qwr->baseline_after, qwr->baseline_before,
2543	qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
2544	qwd.examined_dimensions,
2545	&qwd, &qwd.stats, &qwd.versions);
2546	break;
2547
2548	default:
2549	case WEIGHTS_FORMAT_MULTINODE:
2550	// we don't support these groupings in weights
2551	qwr->group_by.group_by &= ~(RRDR_GROUP_BY_LABEL\|RRDR_GROUP_BY_SELECTED\|RRDR_GROUP_BY_PERCENTAGE_OF_INSTANCE);
2552	if(qwr->group_by.group_by == RRDR_GROUP_BY_NONE) {
2553	added_dimensions =
2554	registered_results_to_json_multinode_no_group_by(
2555	qwd.results, wb,
2556	qwr->after, qwr->before,
2557	qwr->baseline_after, qwr->baseline_before,
2558	qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
2559	qwd.examined_dimensions,
2560	&qwd, &qwd.stats, &qwd.versions);
2561	}
2562	else {
2563	added_dimensions =
2564	registered_results_to_json_multinode_group_by(
2565	qwd.results, wb,
2566	qwr->after, qwr->before,
2567	qwr->baseline_after, qwr->baseline_before,
2568	qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
2569	qwd.examined_dimensions,
2570	&qwd, &qwd.stats, &qwd.versions);
2571	}
2572	break;
2573	}
2574
2575	if(!added_dimensions && qwr->version < 2) {
2576	error = "no results produced.";
2577	resp = HTTP_RESP_NOT_FOUND;
2578	}
2579
2580	cleanup:
2581	simple_pattern_free(qwd.scope_nodes_sp);
2582	simple_pattern_free(qwd.scope_contexts_sp);
2583	simple_pattern_free(qwd.scope_instances_sp);
2584	simple_pattern_free(qwd.scope_labels_sp);
2585	simple_pattern_free(qwd.scope_dimensions_sp);
2586	simple_pattern_free(qwd.nodes_sp);
2587	simple_pattern_free(qwd.contexts_sp);
2588	simple_pattern_free(qwd.instances_sp);
2589	simple_pattern_free(qwd.dimensions_sp);
2590	simple_pattern_free(qwd.labels_sp);
2591	simple_pattern_free(qwd.alerts_sp);
2592
2593	pattern_array_free(qwd.scope_labels_pa);
2594	pattern_array_free(qwd.labels_pa);
2595
2596	register_result_destroy(qwd.results);
2597
2598	if(error) {
2599	buffer_flush(wb);
2600	buffer_sprintf(wb, "{\"error\": \"%s\" }", error);
2601	}
2602
2603	return resp;
2604	}
2605
2606	// ----------------------------------------------------------------------------
2607	// unittest
2608
2609	/*
2610
2611	Unit tests against the output of this:
2612
2613	https://github.com/scipy/scipy/blob/4cf21e753cf937d1c6c2d2a0e372fbc1dbbeea81/scipy/stats/_stats_py.py#L7275-L7449
2614
2615	import matplotlib.pyplot as plt
2616	import pandas as pd
2617	import numpy as np
2618	import scipy as sp
2619	from scipy import stats
2620
2621	data1 = np.array([ 1111, -2222, 33, 100, 100, 15555, -1, 19999, 888, 755, -1, -730 ])
2622	data2 = np.array([365, -123, 0])
2623	data1 = np.sort(data1)
2624	data2 = np.sort(data2)
2625	n1 = data1.shape[0]
2626	n2 = data2.shape[0]
2627	data_all = np.concatenate([data1, data2])
2628	cdf1 = np.searchsorted(data1, data_all, side='right') / n1
2629	cdf2 = np.searchsorted(data2, data_all, side='right') / n2
2630	print(data_all)
2631	print("\ndata1", data1, cdf1)
2632	print("\ndata2", data2, cdf2)
2633	cddiffs = cdf1 - cdf2
2634	print("\ncddiffs", cddiffs)
2635	minS = np.clip(-np.min(cddiffs), 0, 1)
2636	maxS = np.max(cddiffs)
2637	print("\nmin", minS)
2638	print("max", maxS)
2639	m, n = sorted([float(n1), float(n2)], reverse=True)
2640	en = m * n / (m + n)
2641	d = max(minS, maxS)
2642	prob = stats.distributions.kstwo.sf(d, np.round(en))
2643	print("\nprob", prob)
2644
2645	*/
2646
2647	static int double_expect(double v, const char str, const char descr) {
2648	char buf[100 + 1];
2649	snprintfz(buf, sizeof(buf) - 1, "%0.6f", v);
2650	int ret = strcmp(buf, str) ? 1 : 0;
2651
2652	fprintf(stderr, "%s %s, expected %s, got %s\n", ret?"FAILED":"OK", descr, str, buf);
2653	return ret;
2654	}
2655
2656	static int mc_unittest1(void) {
2657	int bs = 3, hs = 3;
2658	DIFFS_NUMBERS base[3] = { 1, 2, 3 };
2659	DIFFS_NUMBERS high[3] = { 3, 4, 6 };
2660
2661	double prob = ks_2samp(base, bs, high, hs, 0);
2662	return double_expect(prob, "0.222222", "3x3");
2663	}
2664
2665	static int mc_unittest2(void) {
2666	int bs = 6, hs = 3;
2667	DIFFS_NUMBERS base[6] = { 1, 2, 3, 10, 10, 15 };
2668	DIFFS_NUMBERS high[3] = { 3, 4, 6 };
2669
2670	double prob = ks_2samp(base, bs, high, hs, 1);
2671	return double_expect(prob, "0.500000", "6x3");
2672	}
2673
2674	static int mc_unittest3(void) {
2675	int bs = 12, hs = 3;
2676	DIFFS_NUMBERS base[12] = { 1, 2, 3, 10, 10, 15, 111, 19999, 8, 55, -1, -73 };
2677	DIFFS_NUMBERS high[3] = { 3, 4, 6 };
2678
2679	double prob = ks_2samp(base, bs, high, hs, 2);
2680	return double_expect(prob, "0.347222", "12x3");
2681	}
2682
2683	static int mc_unittest4(void) {
2684	int bs = 12, hs = 3;
2685	DIFFS_NUMBERS base[12] = { 1111, -2222, 33, 100, 100, 15555, -1, 19999, 888, 755, -1, -730 };
2686	DIFFS_NUMBERS high[3] = { 365, -123, 0 };
2687
2688	double prob = ks_2samp(base, bs, high, hs, 2);
2689	return double_expect(prob, "0.777778", "12x3");
2690	}
2691
2692	int mc_unittest(void) {
2693	int errors = 0;
2694
2695	errors += mc_unittest1();
2696	errors += mc_unittest2();
2697	errors += mc_unittest3();
2698	errors += mc_unittest4();
2699
2700	return errors;
2701	}