weights.c 79 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101
  1. // SPDX-License-Identifier: GPL-3.0-or-later
  2. #include "daemon/common.h"
  3. #include "database/KolmogorovSmirnovDist.h"
  4. #define MAX_POINTS 10000
  5. int enable_metric_correlations = CONFIG_BOOLEAN_YES;
  6. int metric_correlations_version = 1;
  7. WEIGHTS_METHOD default_metric_correlations_method = WEIGHTS_METHOD_MC_KS2;
  8. typedef struct weights_stats {
  9. NETDATA_DOUBLE max_base_high_ratio;
  10. size_t db_points;
  11. size_t result_points;
  12. size_t db_queries;
  13. size_t db_points_per_tier[RRD_STORAGE_TIERS];
  14. size_t binary_searches;
  15. } WEIGHTS_STATS;
  16. // ----------------------------------------------------------------------------
  17. // parse and render metric correlations methods
  18. static struct {
  19. const char *name;
  20. WEIGHTS_METHOD value;
  21. } weights_methods[] = {
  22. { "ks2" , WEIGHTS_METHOD_MC_KS2}
  23. , { "volume" , WEIGHTS_METHOD_MC_VOLUME}
  24. , { "anomaly-rate" , WEIGHTS_METHOD_ANOMALY_RATE}
  25. , { "value" , WEIGHTS_METHOD_VALUE}
  26. , { NULL , 0 }
  27. };
  28. WEIGHTS_METHOD weights_string_to_method(const char *method) {
  29. for(int i = 0; weights_methods[i].name ;i++)
  30. if(strcmp(method, weights_methods[i].name) == 0)
  31. return weights_methods[i].value;
  32. return default_metric_correlations_method;
  33. }
  34. const char *weights_method_to_string(WEIGHTS_METHOD method) {
  35. for(int i = 0; weights_methods[i].name ;i++)
  36. if(weights_methods[i].value == method)
  37. return weights_methods[i].name;
  38. return "unknown";
  39. }
  40. // ----------------------------------------------------------------------------
  41. // The results per dimension are aggregated into a dictionary
  42. typedef enum {
  43. RESULT_IS_BASE_HIGH_RATIO = (1 << 0),
  44. RESULT_IS_PERCENTAGE_OF_TIME = (1 << 1),
  45. } RESULT_FLAGS;
  46. struct register_result {
  47. RESULT_FLAGS flags;
  48. RRDHOST *host;
  49. RRDCONTEXT_ACQUIRED *rca;
  50. RRDINSTANCE_ACQUIRED *ria;
  51. RRDMETRIC_ACQUIRED *rma;
  52. NETDATA_DOUBLE value;
  53. STORAGE_POINT highlighted;
  54. STORAGE_POINT baseline;
  55. usec_t duration_ut;
  56. };
  57. static DICTIONARY *register_result_init() {
  58. DICTIONARY *results = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED | DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct register_result));
  59. return results;
  60. }
  61. static void register_result_destroy(DICTIONARY *results) {
  62. dictionary_destroy(results);
  63. }
  64. static void register_result(DICTIONARY *results, RRDHOST *host, RRDCONTEXT_ACQUIRED *rca, RRDINSTANCE_ACQUIRED *ria,
  65. RRDMETRIC_ACQUIRED *rma, NETDATA_DOUBLE value, RESULT_FLAGS flags,
  66. STORAGE_POINT *highlighted, STORAGE_POINT *baseline, WEIGHTS_STATS *stats,
  67. bool register_zero, usec_t duration_ut) {
  68. if(!netdata_double_isnumber(value)) return;
  69. // make it positive
  70. NETDATA_DOUBLE v = fabsndd(value);
  71. // no need to store zero scored values
  72. if(unlikely(fpclassify(v) == FP_ZERO && !register_zero))
  73. return;
  74. // keep track of the max of the baseline / highlight ratio
  75. if((flags & RESULT_IS_BASE_HIGH_RATIO) && v > stats->max_base_high_ratio)
  76. stats->max_base_high_ratio = v;
  77. struct register_result t = {
  78. .flags = flags,
  79. .host = host,
  80. .rca = rca,
  81. .ria = ria,
  82. .rma = rma,
  83. .value = v,
  84. .duration_ut = duration_ut,
  85. };
  86. if(highlighted)
  87. t.highlighted = *highlighted;
  88. if(baseline)
  89. t.baseline = *baseline;
  90. // we can use the pointer address or RMA as a unique key for each metric
  91. char buf[20 + 1];
  92. ssize_t len = snprintfz(buf, 20, "%p", rma);
  93. dictionary_set_advanced(results, buf, len + 1, &t, sizeof(struct register_result), NULL);
  94. }
  95. // ----------------------------------------------------------------------------
  96. // Generation of JSON output for the results
  97. static void results_header_to_json(DICTIONARY *results __maybe_unused, BUFFER *wb,
  98. time_t after, time_t before,
  99. time_t baseline_after, time_t baseline_before,
  100. size_t points, WEIGHTS_METHOD method,
  101. RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
  102. size_t examined_dimensions __maybe_unused, usec_t duration,
  103. WEIGHTS_STATS *stats) {
  104. buffer_json_member_add_time_t(wb, "after", after);
  105. buffer_json_member_add_time_t(wb, "before", before);
  106. buffer_json_member_add_time_t(wb, "duration", before - after);
  107. buffer_json_member_add_uint64(wb, "points", points);
  108. if(method == WEIGHTS_METHOD_MC_KS2 || method == WEIGHTS_METHOD_MC_VOLUME) {
  109. buffer_json_member_add_time_t(wb, "baseline_after", baseline_after);
  110. buffer_json_member_add_time_t(wb, "baseline_before", baseline_before);
  111. buffer_json_member_add_time_t(wb, "baseline_duration", baseline_before - baseline_after);
  112. buffer_json_member_add_uint64(wb, "baseline_points", points << shifts);
  113. }
  114. buffer_json_member_add_object(wb, "statistics");
  115. {
  116. buffer_json_member_add_double(wb, "query_time_ms", (double) duration / (double) USEC_PER_MS);
  117. buffer_json_member_add_uint64(wb, "db_queries", stats->db_queries);
  118. buffer_json_member_add_uint64(wb, "query_result_points", stats->result_points);
  119. buffer_json_member_add_uint64(wb, "binary_searches", stats->binary_searches);
  120. buffer_json_member_add_uint64(wb, "db_points_read", stats->db_points);
  121. buffer_json_member_add_array(wb, "db_points_per_tier");
  122. {
  123. for (size_t tier = 0; tier < storage_tiers; tier++)
  124. buffer_json_add_array_item_uint64(wb, stats->db_points_per_tier[tier]);
  125. }
  126. buffer_json_array_close(wb);
  127. }
  128. buffer_json_object_close(wb);
  129. buffer_json_member_add_string(wb, "group", time_grouping_tostring(group));
  130. buffer_json_member_add_string(wb, "method", weights_method_to_string(method));
  131. web_client_api_request_v1_data_options_to_buffer_json_array(wb, "options", options);
  132. }
  133. static size_t registered_results_to_json_charts(DICTIONARY *results, BUFFER *wb,
  134. time_t after, time_t before,
  135. time_t baseline_after, time_t baseline_before,
  136. size_t points, WEIGHTS_METHOD method,
  137. RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
  138. size_t examined_dimensions, usec_t duration,
  139. WEIGHTS_STATS *stats) {
  140. buffer_json_initialize(wb, "\"", "\"", 0, true, options & RRDR_OPTION_MINIFY);
  141. results_header_to_json(results, wb, after, before, baseline_after, baseline_before,
  142. points, method, group, options, shifts, examined_dimensions, duration, stats);
  143. buffer_json_member_add_object(wb, "correlated_charts");
  144. size_t charts = 0, total_dimensions = 0;
  145. struct register_result *t;
  146. RRDINSTANCE_ACQUIRED *last_ria = NULL; // never access this - we use it only for comparison
  147. dfe_start_read(results, t) {
  148. if(t->ria != last_ria) {
  149. last_ria = t->ria;
  150. if(charts) {
  151. buffer_json_object_close(wb); // dimensions
  152. buffer_json_object_close(wb); // chart:id
  153. }
  154. buffer_json_member_add_object(wb, rrdinstance_acquired_id(t->ria));
  155. buffer_json_member_add_string(wb, "context", rrdcontext_acquired_id(t->rca));
  156. buffer_json_member_add_object(wb, "dimensions");
  157. charts++;
  158. }
  159. buffer_json_member_add_double(wb, rrdmetric_acquired_name(t->rma), t->value);
  160. total_dimensions++;
  161. }
  162. dfe_done(t);
  163. // close dimensions and chart
  164. if (total_dimensions) {
  165. buffer_json_object_close(wb); // dimensions
  166. buffer_json_object_close(wb); // chart:id
  167. }
  168. buffer_json_object_close(wb);
  169. buffer_json_member_add_uint64(wb, "correlated_dimensions", total_dimensions);
  170. buffer_json_member_add_uint64(wb, "total_dimensions_count", examined_dimensions);
  171. buffer_json_finalize(wb);
  172. return total_dimensions;
  173. }
  174. static size_t registered_results_to_json_contexts(DICTIONARY *results, BUFFER *wb,
  175. time_t after, time_t before,
  176. time_t baseline_after, time_t baseline_before,
  177. size_t points, WEIGHTS_METHOD method,
  178. RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
  179. size_t examined_dimensions, usec_t duration,
  180. WEIGHTS_STATS *stats) {
  181. buffer_json_initialize(wb, "\"", "\"", 0, true, options & RRDR_OPTION_MINIFY);
  182. results_header_to_json(results, wb, after, before, baseline_after, baseline_before,
  183. points, method, group, options, shifts, examined_dimensions, duration, stats);
  184. buffer_json_member_add_object(wb, "contexts");
  185. size_t contexts = 0, charts = 0, total_dimensions = 0, context_dims = 0, chart_dims = 0;
  186. NETDATA_DOUBLE contexts_total_weight = 0.0, charts_total_weight = 0.0;
  187. struct register_result *t;
  188. RRDCONTEXT_ACQUIRED *last_rca = NULL;
  189. RRDINSTANCE_ACQUIRED *last_ria = NULL;
  190. dfe_start_read(results, t) {
  191. if(t->rca != last_rca) {
  192. last_rca = t->rca;
  193. if(contexts) {
  194. buffer_json_object_close(wb); // dimensions
  195. buffer_json_member_add_double(wb, "weight", charts_total_weight / (double) chart_dims);
  196. buffer_json_object_close(wb); // chart:id
  197. buffer_json_object_close(wb); // charts
  198. buffer_json_member_add_double(wb, "weight", contexts_total_weight / (double) context_dims);
  199. buffer_json_object_close(wb); // context
  200. }
  201. buffer_json_member_add_object(wb, rrdcontext_acquired_id(t->rca));
  202. buffer_json_member_add_object(wb, "charts");
  203. contexts++;
  204. charts = 0;
  205. context_dims = 0;
  206. contexts_total_weight = 0.0;
  207. last_ria = NULL;
  208. }
  209. if(t->ria != last_ria) {
  210. last_ria = t->ria;
  211. if(charts) {
  212. buffer_json_object_close(wb); // dimensions
  213. buffer_json_member_add_double(wb, "weight", charts_total_weight / (double) chart_dims);
  214. buffer_json_object_close(wb); // chart:id
  215. }
  216. buffer_json_member_add_object(wb, rrdinstance_acquired_id(t->ria));
  217. buffer_json_member_add_object(wb, "dimensions");
  218. charts++;
  219. chart_dims = 0;
  220. charts_total_weight = 0.0;
  221. }
  222. buffer_json_member_add_double(wb, rrdmetric_acquired_name(t->rma), t->value);
  223. charts_total_weight += t->value;
  224. contexts_total_weight += t->value;
  225. chart_dims++;
  226. context_dims++;
  227. total_dimensions++;
  228. }
  229. dfe_done(t);
  230. // close dimensions and chart
  231. if (total_dimensions) {
  232. buffer_json_object_close(wb); // dimensions
  233. buffer_json_member_add_double(wb, "weight", charts_total_weight / (double) chart_dims);
  234. buffer_json_object_close(wb); // chart:id
  235. buffer_json_object_close(wb); // charts
  236. buffer_json_member_add_double(wb, "weight", contexts_total_weight / (double) context_dims);
  237. buffer_json_object_close(wb); // context
  238. }
  239. buffer_json_object_close(wb);
  240. buffer_json_member_add_uint64(wb, "correlated_dimensions", total_dimensions);
  241. buffer_json_member_add_uint64(wb, "total_dimensions_count", examined_dimensions);
  242. buffer_json_finalize(wb);
  243. return total_dimensions;
  244. }
  245. struct query_weights_data {
  246. QUERY_WEIGHTS_REQUEST *qwr;
  247. SIMPLE_PATTERN *scope_nodes_sp;
  248. SIMPLE_PATTERN *scope_contexts_sp;
  249. SIMPLE_PATTERN *nodes_sp;
  250. SIMPLE_PATTERN *contexts_sp;
  251. SIMPLE_PATTERN *instances_sp;
  252. SIMPLE_PATTERN *dimensions_sp;
  253. SIMPLE_PATTERN *labels_sp;
  254. SIMPLE_PATTERN *alerts_sp;
  255. usec_t timeout_us;
  256. bool timed_out;
  257. bool interrupted;
  258. struct query_timings timings;
  259. size_t examined_dimensions;
  260. bool register_zero;
  261. DICTIONARY *results;
  262. WEIGHTS_STATS stats;
  263. uint32_t shifts;
  264. struct query_versions versions;
  265. };
  266. #define AGGREGATED_WEIGHT_EMPTY (struct aggregated_weight) { \
  267. .min = NAN, \
  268. .max = NAN, \
  269. .sum = NAN, \
  270. .count = 0, \
  271. .hsp = STORAGE_POINT_UNSET, \
  272. .bsp = STORAGE_POINT_UNSET, \
  273. }
  274. #define merge_into_aw(aw, t) do { \
  275. if(!(aw).count) { \
  276. (aw).count = 1; \
  277. (aw).min = (aw).max = (aw).sum = (t)->value; \
  278. (aw).hsp = (t)->highlighted; \
  279. if(baseline) \
  280. (aw).bsp = (t)->baseline; \
  281. } \
  282. else { \
  283. (aw).count++; \
  284. (aw).sum += (t)->value; \
  285. if((t)->value < (aw).min) \
  286. (aw).min = (t)->value; \
  287. if((t)->value > (aw).max) \
  288. (aw).max = (t)->value; \
  289. storage_point_merge_to((aw).hsp, (t)->highlighted); \
  290. if(baseline) \
  291. storage_point_merge_to((aw).bsp, (t)->baseline); \
  292. } \
  293. } while(0)
  294. static void results_header_to_json_v2(DICTIONARY *results __maybe_unused, BUFFER *wb, struct query_weights_data *qwd,
  295. time_t after, time_t before,
  296. time_t baseline_after, time_t baseline_before,
  297. size_t points, WEIGHTS_METHOD method,
  298. RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
  299. size_t examined_dimensions __maybe_unused, usec_t duration __maybe_unused,
  300. WEIGHTS_STATS *stats, bool group_by) {
  301. buffer_json_member_add_object(wb, "request");
  302. buffer_json_member_add_string(wb, "method", weights_method_to_string(method));
  303. web_client_api_request_v1_data_options_to_buffer_json_array(wb, "options", options);
  304. buffer_json_member_add_object(wb, "scope");
  305. buffer_json_member_add_string(wb, "scope_nodes", qwd->qwr->scope_nodes ? qwd->qwr->scope_nodes : "*");
  306. buffer_json_member_add_string(wb, "scope_contexts", qwd->qwr->scope_contexts ? qwd->qwr->scope_contexts : "*");
  307. buffer_json_object_close(wb);
  308. buffer_json_member_add_object(wb, "selectors");
  309. buffer_json_member_add_string(wb, "nodes", qwd->qwr->nodes ? qwd->qwr->nodes : "*");
  310. buffer_json_member_add_string(wb, "contexts", qwd->qwr->contexts ? qwd->qwr->contexts : "*");
  311. buffer_json_member_add_string(wb, "instances", qwd->qwr->instances ? qwd->qwr->instances : "*");
  312. buffer_json_member_add_string(wb, "dimensions", qwd->qwr->dimensions ? qwd->qwr->dimensions : "*");
  313. buffer_json_member_add_string(wb, "labels", qwd->qwr->labels ? qwd->qwr->labels : "*");
  314. buffer_json_member_add_string(wb, "alerts", qwd->qwr->alerts ? qwd->qwr->alerts : "*");
  315. buffer_json_object_close(wb);
  316. buffer_json_member_add_object(wb, "window");
  317. buffer_json_member_add_time_t(wb, "after", qwd->qwr->after);
  318. buffer_json_member_add_time_t(wb, "before", qwd->qwr->before);
  319. buffer_json_member_add_uint64(wb, "points", qwd->qwr->points);
  320. if(qwd->qwr->options & RRDR_OPTION_SELECTED_TIER)
  321. buffer_json_member_add_uint64(wb, "tier", qwd->qwr->tier);
  322. else
  323. buffer_json_member_add_string(wb, "tier", NULL);
  324. buffer_json_object_close(wb);
  325. if(method == WEIGHTS_METHOD_MC_KS2 || method == WEIGHTS_METHOD_MC_VOLUME) {
  326. buffer_json_member_add_object(wb, "baseline");
  327. buffer_json_member_add_time_t(wb, "baseline_after", qwd->qwr->baseline_after);
  328. buffer_json_member_add_time_t(wb, "baseline_before", qwd->qwr->baseline_before);
  329. buffer_json_object_close(wb);
  330. }
  331. buffer_json_member_add_object(wb, "aggregations");
  332. buffer_json_member_add_object(wb, "time");
  333. buffer_json_member_add_string(wb, "time_group", time_grouping_tostring(qwd->qwr->time_group_method));
  334. buffer_json_member_add_string(wb, "time_group_options", qwd->qwr->time_group_options);
  335. buffer_json_object_close(wb); // time
  336. buffer_json_member_add_array(wb, "metrics");
  337. buffer_json_add_array_item_object(wb);
  338. {
  339. buffer_json_member_add_array(wb, "group_by");
  340. buffer_json_group_by_to_array(wb, qwd->qwr->group_by.group_by);
  341. buffer_json_array_close(wb);
  342. // buffer_json_member_add_array(wb, "group_by_label");
  343. // buffer_json_array_close(wb);
  344. buffer_json_member_add_string(wb, "aggregation", group_by_aggregate_function_to_string(qwd->qwr->group_by.aggregation));
  345. }
  346. buffer_json_object_close(wb); // 1st group by
  347. buffer_json_array_close(wb); // array
  348. buffer_json_object_close(wb); // aggregations
  349. buffer_json_member_add_uint64(wb, "timeout", qwd->qwr->timeout_ms);
  350. buffer_json_object_close(wb); // request
  351. buffer_json_member_add_object(wb, "view");
  352. buffer_json_member_add_string(wb, "format", (group_by)?"grouped":"full");
  353. buffer_json_member_add_string(wb, "time_group", time_grouping_tostring(group));
  354. buffer_json_member_add_object(wb, "window");
  355. buffer_json_member_add_time_t(wb, "after", after);
  356. buffer_json_member_add_time_t(wb, "before", before);
  357. buffer_json_member_add_time_t(wb, "duration", before - after);
  358. buffer_json_member_add_uint64(wb, "points", points);
  359. buffer_json_object_close(wb);
  360. if(method == WEIGHTS_METHOD_MC_KS2 || method == WEIGHTS_METHOD_MC_VOLUME) {
  361. buffer_json_member_add_object(wb, "baseline");
  362. buffer_json_member_add_time_t(wb, "after", baseline_after);
  363. buffer_json_member_add_time_t(wb, "before", baseline_before);
  364. buffer_json_member_add_time_t(wb, "duration", baseline_before - baseline_after);
  365. buffer_json_member_add_uint64(wb, "points", points << shifts);
  366. buffer_json_object_close(wb);
  367. }
  368. buffer_json_object_close(wb); // view
  369. buffer_json_member_add_object(wb, "db");
  370. {
  371. buffer_json_member_add_uint64(wb, "db_queries", stats->db_queries);
  372. buffer_json_member_add_uint64(wb, "query_result_points", stats->result_points);
  373. buffer_json_member_add_uint64(wb, "binary_searches", stats->binary_searches);
  374. buffer_json_member_add_uint64(wb, "db_points_read", stats->db_points);
  375. buffer_json_member_add_array(wb, "db_points_per_tier");
  376. {
  377. for (size_t tier = 0; tier < storage_tiers; tier++)
  378. buffer_json_add_array_item_uint64(wb, stats->db_points_per_tier[tier]);
  379. }
  380. buffer_json_array_close(wb);
  381. }
  382. buffer_json_object_close(wb); // db
  383. }
  384. typedef enum {
  385. WPT_DIMENSION = 0,
  386. WPT_INSTANCE = 1,
  387. WPT_CONTEXT = 2,
  388. WPT_NODE = 3,
  389. WPT_GROUP = 4,
  390. } WEIGHTS_POINT_TYPE;
  391. struct aggregated_weight {
  392. const char *name;
  393. NETDATA_DOUBLE min;
  394. NETDATA_DOUBLE max;
  395. NETDATA_DOUBLE sum;
  396. size_t count;
  397. STORAGE_POINT hsp;
  398. STORAGE_POINT bsp;
  399. };
  400. 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) {
  401. if(type != WPT_GROUP) {
  402. buffer_json_add_array_item_array(wb);
  403. buffer_json_add_array_item_uint64(wb, type); // "type"
  404. buffer_json_add_array_item_int64(wb, ni);
  405. if (type != WPT_NODE) {
  406. buffer_json_add_array_item_int64(wb, ci);
  407. if (type != WPT_CONTEXT) {
  408. buffer_json_add_array_item_int64(wb, ii);
  409. if (type != WPT_INSTANCE)
  410. buffer_json_add_array_item_int64(wb, di);
  411. else
  412. buffer_json_add_array_item_string(wb, NULL);
  413. }
  414. else {
  415. buffer_json_add_array_item_string(wb, NULL);
  416. buffer_json_add_array_item_string(wb, NULL);
  417. }
  418. }
  419. else {
  420. buffer_json_add_array_item_string(wb, NULL);
  421. buffer_json_add_array_item_string(wb, NULL);
  422. buffer_json_add_array_item_string(wb, NULL);
  423. }
  424. buffer_json_add_array_item_double(wb, (aw->count) ? aw->sum / (NETDATA_DOUBLE)aw->count : 0.0); // "weight"
  425. }
  426. else {
  427. buffer_json_member_add_array(wb, "v");
  428. buffer_json_add_array_item_array(wb);
  429. buffer_json_add_array_item_double(wb, aw->min); // "min"
  430. buffer_json_add_array_item_double(wb, (aw->count) ? aw->sum / (NETDATA_DOUBLE)aw->count : 0.0); // "avg"
  431. buffer_json_add_array_item_double(wb, aw->max); // "max"
  432. buffer_json_add_array_item_double(wb, aw->sum); // "sum"
  433. buffer_json_add_array_item_uint64(wb, aw->count); // "count"
  434. buffer_json_array_close(wb);
  435. }
  436. buffer_json_add_array_item_array(wb);
  437. buffer_json_add_array_item_double(wb, aw->hsp.min); // "min"
  438. buffer_json_add_array_item_double(wb, (aw->hsp.count) ? aw->hsp.sum / (NETDATA_DOUBLE) aw->hsp.count : 0.0); // "avg"
  439. buffer_json_add_array_item_double(wb, aw->hsp.max); // "max"
  440. buffer_json_add_array_item_double(wb, aw->hsp.sum); // "sum"
  441. buffer_json_add_array_item_uint64(wb, aw->hsp.count); // "count"
  442. buffer_json_add_array_item_uint64(wb, aw->hsp.anomaly_count); // "anomaly_count"
  443. buffer_json_array_close(wb);
  444. if(baseline) {
  445. buffer_json_add_array_item_array(wb);
  446. buffer_json_add_array_item_double(wb, aw->bsp.min); // "min"
  447. buffer_json_add_array_item_double(wb, (aw->bsp.count) ? aw->bsp.sum / (NETDATA_DOUBLE) aw->bsp.count : 0.0); // "avg"
  448. buffer_json_add_array_item_double(wb, aw->bsp.max); // "max"
  449. buffer_json_add_array_item_double(wb, aw->bsp.sum); // "sum"
  450. buffer_json_add_array_item_uint64(wb, aw->bsp.count); // "count"
  451. buffer_json_add_array_item_uint64(wb, aw->bsp.anomaly_count); // "anomaly_count"
  452. buffer_json_array_close(wb);
  453. }
  454. buffer_json_array_close(wb);
  455. }
  456. static void multinode_data_schema(BUFFER *wb, RRDR_OPTIONS options __maybe_unused, const char *key, bool baseline, bool group_by) {
  457. buffer_json_member_add_object(wb, key); // schema
  458. buffer_json_member_add_string(wb, "type", "array");
  459. buffer_json_member_add_array(wb, "items");
  460. if(group_by) {
  461. buffer_json_add_array_item_object(wb);
  462. {
  463. buffer_json_member_add_string(wb, "name", "weight");
  464. buffer_json_member_add_string(wb, "type", "array");
  465. buffer_json_member_add_array(wb, "labels");
  466. {
  467. buffer_json_add_array_item_string(wb, "min");
  468. buffer_json_add_array_item_string(wb, "avg");
  469. buffer_json_add_array_item_string(wb, "max");
  470. buffer_json_add_array_item_string(wb, "sum");
  471. buffer_json_add_array_item_string(wb, "count");
  472. }
  473. buffer_json_array_close(wb);
  474. }
  475. buffer_json_object_close(wb);
  476. }
  477. else {
  478. buffer_json_add_array_item_object(wb);
  479. buffer_json_member_add_string(wb, "name", "row_type");
  480. buffer_json_member_add_string(wb, "type", "integer");
  481. buffer_json_member_add_array(wb, "value");
  482. buffer_json_add_array_item_string(wb, "dimension");
  483. buffer_json_add_array_item_string(wb, "instance");
  484. buffer_json_add_array_item_string(wb, "context");
  485. buffer_json_add_array_item_string(wb, "node");
  486. buffer_json_array_close(wb);
  487. buffer_json_object_close(wb);
  488. buffer_json_add_array_item_object(wb);
  489. {
  490. buffer_json_member_add_string(wb, "name", "ni");
  491. buffer_json_member_add_string(wb, "type", "integer");
  492. buffer_json_member_add_string(wb, "dictionary", "nodes");
  493. }
  494. buffer_json_object_close(wb);
  495. buffer_json_add_array_item_object(wb);
  496. {
  497. buffer_json_member_add_string(wb, "name", "ci");
  498. buffer_json_member_add_string(wb, "type", "integer");
  499. buffer_json_member_add_string(wb, "dictionary", "contexts");
  500. }
  501. buffer_json_object_close(wb);
  502. buffer_json_add_array_item_object(wb);
  503. {
  504. buffer_json_member_add_string(wb, "name", "ii");
  505. buffer_json_member_add_string(wb, "type", "integer");
  506. buffer_json_member_add_string(wb, "dictionary", "instances");
  507. }
  508. buffer_json_object_close(wb);
  509. buffer_json_add_array_item_object(wb);
  510. {
  511. buffer_json_member_add_string(wb, "name", "di");
  512. buffer_json_member_add_string(wb, "type", "integer");
  513. buffer_json_member_add_string(wb, "dictionary", "dimensions");
  514. }
  515. buffer_json_object_close(wb);
  516. buffer_json_add_array_item_object(wb);
  517. {
  518. buffer_json_member_add_string(wb, "name", "weight");
  519. buffer_json_member_add_string(wb, "type", "number");
  520. }
  521. buffer_json_object_close(wb);
  522. }
  523. buffer_json_add_array_item_object(wb);
  524. {
  525. buffer_json_member_add_string(wb, "name", "timeframe");
  526. buffer_json_member_add_string(wb, "type", "array");
  527. buffer_json_member_add_array(wb, "labels");
  528. {
  529. buffer_json_add_array_item_string(wb, "min");
  530. buffer_json_add_array_item_string(wb, "avg");
  531. buffer_json_add_array_item_string(wb, "max");
  532. buffer_json_add_array_item_string(wb, "sum");
  533. buffer_json_add_array_item_string(wb, "count");
  534. buffer_json_add_array_item_string(wb, "anomaly_count");
  535. }
  536. buffer_json_array_close(wb);
  537. buffer_json_member_add_object(wb, "calculations");
  538. buffer_json_member_add_string(wb, "anomaly rate", "anomaly_count * 100 / count");
  539. buffer_json_object_close(wb);
  540. }
  541. buffer_json_object_close(wb);
  542. if(baseline) {
  543. buffer_json_add_array_item_object(wb);
  544. {
  545. buffer_json_member_add_string(wb, "name", "baseline timeframe");
  546. buffer_json_member_add_string(wb, "type", "array");
  547. buffer_json_member_add_array(wb, "labels");
  548. {
  549. buffer_json_add_array_item_string(wb, "min");
  550. buffer_json_add_array_item_string(wb, "avg");
  551. buffer_json_add_array_item_string(wb, "max");
  552. buffer_json_add_array_item_string(wb, "sum");
  553. buffer_json_add_array_item_string(wb, "count");
  554. buffer_json_add_array_item_string(wb, "anomaly_count");
  555. }
  556. buffer_json_array_close(wb);
  557. buffer_json_member_add_object(wb, "calculations");
  558. buffer_json_member_add_string(wb, "anomaly rate", "anomaly_count * 100 / count");
  559. buffer_json_object_close(wb);
  560. }
  561. buffer_json_object_close(wb);
  562. }
  563. buffer_json_array_close(wb); // items
  564. buffer_json_object_close(wb); // schema
  565. }
  566. struct dict_unique_node {
  567. bool existing;
  568. bool exposed;
  569. uint32_t i;
  570. RRDHOST *host;
  571. usec_t duration_ut;
  572. };
  573. struct dict_unique_name_units {
  574. bool existing;
  575. bool exposed;
  576. uint32_t i;
  577. const char *units;
  578. };
  579. struct dict_unique_id_name {
  580. bool existing;
  581. bool exposed;
  582. uint32_t i;
  583. const char *id;
  584. const char *name;
  585. };
  586. static inline struct dict_unique_node *dict_unique_node_add(DICTIONARY *dict, RRDHOST *host, ssize_t *max_id) {
  587. struct dict_unique_node *dun = dictionary_set(dict, host->machine_guid, NULL, sizeof(struct dict_unique_node));
  588. if(!dun->existing) {
  589. dun->existing = true;
  590. dun->host = host;
  591. dun->i = *max_id;
  592. (*max_id)++;
  593. }
  594. return dun;
  595. }
  596. static inline struct dict_unique_name_units *dict_unique_name_units_add(DICTIONARY *dict, const char *name, const char *units, ssize_t *max_id) {
  597. struct dict_unique_name_units *dun = dictionary_set(dict, name, NULL, sizeof(struct dict_unique_name_units));
  598. if(!dun->existing) {
  599. dun->units = units;
  600. dun->existing = true;
  601. dun->i = *max_id;
  602. (*max_id)++;
  603. }
  604. return dun;
  605. }
  606. static inline struct dict_unique_id_name *dict_unique_id_name_add(DICTIONARY *dict, const char *id, const char *name, ssize_t *max_id) {
  607. char key[1024 + 1];
  608. snprintfz(key, 1024, "%s:%s", id, name);
  609. struct dict_unique_id_name *dun = dictionary_set(dict, key, NULL, sizeof(struct dict_unique_id_name));
  610. if(!dun->existing) {
  611. dun->existing = true;
  612. dun->i = *max_id;
  613. (*max_id)++;
  614. dun->id = id;
  615. dun->name = name;
  616. }
  617. return dun;
  618. }
  619. static size_t registered_results_to_json_multinode_no_group_by(
  620. DICTIONARY *results, BUFFER *wb,
  621. time_t after, time_t before,
  622. time_t baseline_after, time_t baseline_before,
  623. size_t points, WEIGHTS_METHOD method,
  624. RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
  625. size_t examined_dimensions, struct query_weights_data *qwd,
  626. WEIGHTS_STATS *stats,
  627. struct query_versions *versions) {
  628. buffer_json_initialize(wb, "\"", "\"", 0, true, options & RRDR_OPTION_MINIFY);
  629. buffer_json_member_add_uint64(wb, "api", 2);
  630. results_header_to_json_v2(results, wb, qwd, after, before, baseline_after, baseline_before,
  631. points, method, group, options, shifts, examined_dimensions,
  632. qwd->timings.executed_ut - qwd->timings.received_ut, stats, false);
  633. version_hashes_api_v2(wb, versions);
  634. bool baseline = method == WEIGHTS_METHOD_MC_KS2 || method == WEIGHTS_METHOD_MC_VOLUME;
  635. multinode_data_schema(wb, options, "schema", baseline, false);
  636. 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));
  637. 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));
  638. 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));
  639. 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));
  640. buffer_json_member_add_array(wb, "result");
  641. struct aggregated_weight node_aw = AGGREGATED_WEIGHT_EMPTY, context_aw = AGGREGATED_WEIGHT_EMPTY, instance_aw = AGGREGATED_WEIGHT_EMPTY;
  642. struct register_result *t;
  643. RRDHOST *last_host = NULL;
  644. RRDCONTEXT_ACQUIRED *last_rca = NULL;
  645. RRDINSTANCE_ACQUIRED *last_ria = NULL;
  646. struct dict_unique_name_units *context_dun = NULL;
  647. struct dict_unique_node *node_dun = NULL;
  648. struct dict_unique_id_name *instance_dun = NULL;
  649. struct dict_unique_id_name *dimension_dun = NULL;
  650. ssize_t di = -1, ii = -1, ci = -1, ni = -1;
  651. ssize_t di_max = 0, ii_max = 0, ci_max = 0, ni_max = 0;
  652. size_t total_dimensions = 0;
  653. dfe_start_read(results, t) {
  654. // close instance
  655. if(t->ria != last_ria && last_ria) {
  656. storage_point_to_json(wb, WPT_INSTANCE, di, ii, ci, ni, &instance_aw, options, baseline);
  657. instance_dun->exposed = true;
  658. last_ria = NULL;
  659. instance_aw = AGGREGATED_WEIGHT_EMPTY;
  660. }
  661. // close context
  662. if(t->rca != last_rca && last_rca) {
  663. storage_point_to_json(wb, WPT_CONTEXT, di, ii, ci, ni, &context_aw, options, baseline);
  664. context_dun->exposed = true;
  665. last_rca = NULL;
  666. context_aw = AGGREGATED_WEIGHT_EMPTY;
  667. }
  668. // close node
  669. if(t->host != last_host && last_host) {
  670. storage_point_to_json(wb, WPT_NODE, di, ii, ci, ni, &node_aw, options, baseline);
  671. node_dun->exposed = true;
  672. last_host = NULL;
  673. node_aw = AGGREGATED_WEIGHT_EMPTY;
  674. }
  675. // open node
  676. if(t->host != last_host) {
  677. last_host = t->host;
  678. node_dun = dict_unique_node_add(dict_nodes, t->host, &ni_max);
  679. ni = node_dun->i;
  680. }
  681. // open context
  682. if(t->rca != last_rca) {
  683. last_rca = t->rca;
  684. context_dun = dict_unique_name_units_add(dict_contexts, rrdcontext_acquired_id(t->rca),
  685. rrdcontext_acquired_units(t->rca), &ci_max);
  686. ci = context_dun->i;
  687. }
  688. // open instance
  689. if(t->ria != last_ria) {
  690. last_ria = t->ria;
  691. instance_dun = dict_unique_id_name_add(dict_instances, rrdinstance_acquired_id(t->ria), rrdinstance_acquired_name(t->ria), &ii_max);
  692. ii = instance_dun->i;
  693. }
  694. dimension_dun = dict_unique_id_name_add(dict_dimensions, rrdmetric_acquired_id(t->rma), rrdmetric_acquired_name(t->rma), &di_max);
  695. di = dimension_dun->i;
  696. struct aggregated_weight aw = {
  697. .min = t->value,
  698. .max = t->value,
  699. .sum = t->value,
  700. .count = 1,
  701. .hsp = t->highlighted,
  702. .bsp = t->baseline,
  703. };
  704. storage_point_to_json(wb, WPT_DIMENSION, di, ii, ci, ni, &aw, options, baseline);
  705. node_dun->exposed = true;
  706. context_dun->exposed = true;
  707. instance_dun->exposed = true;
  708. dimension_dun->exposed = true;
  709. merge_into_aw(instance_aw, t);
  710. merge_into_aw(context_aw, t);
  711. merge_into_aw(node_aw, t);
  712. node_dun->duration_ut += t->duration_ut;
  713. total_dimensions++;
  714. }
  715. dfe_done(t);
  716. // close instance
  717. if(last_ria) {
  718. storage_point_to_json(wb, WPT_INSTANCE, di, ii, ci, ni, &instance_aw, options, baseline);
  719. instance_dun->exposed = true;
  720. }
  721. // close context
  722. if(last_rca) {
  723. storage_point_to_json(wb, WPT_CONTEXT, di, ii, ci, ni, &context_aw, options, baseline);
  724. context_dun->exposed = true;
  725. }
  726. // close node
  727. if(last_host) {
  728. storage_point_to_json(wb, WPT_NODE, di, ii, ci, ni, &node_aw, options, baseline);
  729. node_dun->exposed = true;
  730. }
  731. buffer_json_array_close(wb); // points
  732. buffer_json_member_add_object(wb, "dictionaries");
  733. buffer_json_member_add_array(wb, "nodes");
  734. {
  735. struct dict_unique_node *dun;
  736. dfe_start_read(dict_nodes, dun) {
  737. if(!dun->exposed)
  738. continue;
  739. buffer_json_add_array_item_object(wb);
  740. buffer_json_node_add_v2(wb, dun->host, dun->i, dun->duration_ut);
  741. buffer_json_object_close(wb);
  742. }
  743. dfe_done(dun);
  744. }
  745. buffer_json_array_close(wb);
  746. buffer_json_member_add_array(wb, "contexts");
  747. {
  748. struct dict_unique_name_units *dun;
  749. dfe_start_read(dict_contexts, dun) {
  750. if(!dun->exposed)
  751. continue;
  752. buffer_json_add_array_item_object(wb);
  753. buffer_json_member_add_string(wb, "id", dun_dfe.name);
  754. buffer_json_member_add_string(wb, "units", dun->units);
  755. buffer_json_member_add_int64(wb, "ci", dun->i);
  756. buffer_json_object_close(wb);
  757. }
  758. dfe_done(dun);
  759. }
  760. buffer_json_array_close(wb);
  761. buffer_json_member_add_array(wb, "instances");
  762. {
  763. struct dict_unique_id_name *dun;
  764. dfe_start_read(dict_instances, dun) {
  765. if(!dun->exposed)
  766. continue;
  767. buffer_json_add_array_item_object(wb);
  768. buffer_json_member_add_string(wb, "id", dun->id);
  769. if(dun->id != dun->name)
  770. buffer_json_member_add_string(wb, "nm", dun->name);
  771. buffer_json_member_add_int64(wb, "ii", dun->i);
  772. buffer_json_object_close(wb);
  773. }
  774. dfe_done(dun);
  775. }
  776. buffer_json_array_close(wb);
  777. buffer_json_member_add_array(wb, "dimensions");
  778. {
  779. struct dict_unique_id_name *dun;
  780. dfe_start_read(dict_dimensions, dun) {
  781. if(!dun->exposed)
  782. continue;
  783. buffer_json_add_array_item_object(wb);
  784. buffer_json_member_add_string(wb, "id", dun->id);
  785. if(dun->id != dun->name)
  786. buffer_json_member_add_string(wb, "nm", dun->name);
  787. buffer_json_member_add_int64(wb, "di", dun->i);
  788. buffer_json_object_close(wb);
  789. }
  790. dfe_done(dun);
  791. }
  792. buffer_json_array_close(wb);
  793. buffer_json_object_close(wb); //dictionaries
  794. buffer_json_agents_array_v2(wb, &qwd->timings, 0);
  795. buffer_json_member_add_uint64(wb, "correlated_dimensions", total_dimensions);
  796. buffer_json_member_add_uint64(wb, "total_dimensions_count", examined_dimensions);
  797. buffer_json_finalize(wb);
  798. dictionary_destroy(dict_nodes);
  799. dictionary_destroy(dict_contexts);
  800. dictionary_destroy(dict_instances);
  801. dictionary_destroy(dict_dimensions);
  802. return total_dimensions;
  803. }
  804. static size_t registered_results_to_json_multinode_group_by(
  805. DICTIONARY *results, BUFFER *wb,
  806. time_t after, time_t before,
  807. time_t baseline_after, time_t baseline_before,
  808. size_t points, WEIGHTS_METHOD method,
  809. RRDR_TIME_GROUPING group, RRDR_OPTIONS options, uint32_t shifts,
  810. size_t examined_dimensions, struct query_weights_data *qwd,
  811. WEIGHTS_STATS *stats,
  812. struct query_versions *versions) {
  813. buffer_json_initialize(wb, "\"", "\"", 0, true, options & RRDR_OPTION_MINIFY);
  814. buffer_json_member_add_uint64(wb, "api", 2);
  815. results_header_to_json_v2(results, wb, qwd, after, before, baseline_after, baseline_before,
  816. points, method, group, options, shifts, examined_dimensions,
  817. qwd->timings.executed_ut - qwd->timings.received_ut, stats, true);
  818. version_hashes_api_v2(wb, versions);
  819. bool baseline = method == WEIGHTS_METHOD_MC_KS2 || method == WEIGHTS_METHOD_MC_VOLUME;
  820. multinode_data_schema(wb, options, "v_schema", baseline, true);
  821. DICTIONARY *group_by = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED | DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE,
  822. NULL, sizeof(struct aggregated_weight));
  823. struct register_result *t;
  824. size_t total_dimensions = 0;
  825. BUFFER *key = buffer_create(0, NULL);
  826. BUFFER *name = buffer_create(0, NULL);
  827. dfe_start_read(results, t) {
  828. buffer_flush(key);
  829. buffer_flush(name);
  830. if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_DIMENSION) {
  831. buffer_strcat(key, rrdmetric_acquired_name(t->rma));
  832. buffer_strcat(name, rrdmetric_acquired_name(t->rma));
  833. }
  834. if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_INSTANCE) {
  835. if(buffer_strlen(key)) {
  836. buffer_fast_strcat(key, ",", 1);
  837. buffer_fast_strcat(name, ",", 1);
  838. }
  839. buffer_strcat(key, rrdinstance_acquired_id(t->ria));
  840. buffer_strcat(name, rrdinstance_acquired_name(t->ria));
  841. if(!(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_NODE)) {
  842. buffer_fast_strcat(key, "@", 1);
  843. buffer_fast_strcat(name, "@", 1);
  844. buffer_strcat(key, t->host->machine_guid);
  845. buffer_strcat(name, rrdhost_hostname(t->host));
  846. }
  847. }
  848. if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_NODE) {
  849. if(buffer_strlen(key)) {
  850. buffer_fast_strcat(key, ",", 1);
  851. buffer_fast_strcat(name, ",", 1);
  852. }
  853. buffer_strcat(key, t->host->machine_guid);
  854. buffer_strcat(name, rrdhost_hostname(t->host));
  855. }
  856. if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_CONTEXT) {
  857. if(buffer_strlen(key)) {
  858. buffer_fast_strcat(key, ",", 1);
  859. buffer_fast_strcat(name, ",", 1);
  860. }
  861. buffer_strcat(key, rrdcontext_acquired_id(t->rca));
  862. buffer_strcat(name, rrdcontext_acquired_id(t->rca));
  863. }
  864. if(qwd->qwr->group_by.group_by & RRDR_GROUP_BY_UNITS) {
  865. if(buffer_strlen(key)) {
  866. buffer_fast_strcat(key, ",", 1);
  867. buffer_fast_strcat(name, ",", 1);
  868. }
  869. buffer_strcat(key, rrdcontext_acquired_units(t->rca));
  870. buffer_strcat(name, rrdcontext_acquired_units(t->rca));
  871. }
  872. struct aggregated_weight *aw = dictionary_set(group_by, buffer_tostring(key), NULL, sizeof(struct aggregated_weight));
  873. if(!aw->name) {
  874. aw->name = strdupz(buffer_tostring(name));
  875. aw->min = aw->max = aw->sum = t->value;
  876. aw->count = 1;
  877. aw->hsp = t->highlighted;
  878. aw->bsp = t->baseline;
  879. }
  880. else
  881. merge_into_aw(*aw, t);
  882. total_dimensions++;
  883. }
  884. dfe_done(t);
  885. buffer_free(key); key = NULL;
  886. buffer_free(name); name = NULL;
  887. struct aggregated_weight *aw;
  888. buffer_json_member_add_array(wb, "result");
  889. dfe_start_read(group_by, aw) {
  890. const char *k = aw_dfe.name;
  891. const char *n = aw->name;
  892. buffer_json_add_array_item_object(wb);
  893. buffer_json_member_add_string(wb, "id", k);
  894. if(strcmp(k, n) != 0)
  895. buffer_json_member_add_string(wb, "nm", n);
  896. storage_point_to_json(wb, WPT_GROUP, 0, 0, 0, 0, aw, options, baseline);
  897. buffer_json_object_close(wb);
  898. freez((void *)aw->name);
  899. }
  900. dfe_done(aw);
  901. buffer_json_array_close(wb); // result
  902. buffer_json_agents_array_v2(wb, &qwd->timings, 0);
  903. buffer_json_member_add_uint64(wb, "correlated_dimensions", total_dimensions);
  904. buffer_json_member_add_uint64(wb, "total_dimensions_count", examined_dimensions);
  905. buffer_json_finalize(wb);
  906. dictionary_destroy(group_by);
  907. return total_dimensions;
  908. }
  909. // ----------------------------------------------------------------------------
  910. // KS2 algorithm functions
  911. typedef long int DIFFS_NUMBERS;
  912. #define DOUBLE_TO_INT_MULTIPLIER 100000
  913. static inline int binary_search_bigger_than(const DIFFS_NUMBERS arr[], int left, int size, DIFFS_NUMBERS K) {
  914. // binary search to find the index the smallest index
  915. // of the first value in the array that is greater than K
  916. int right = size;
  917. while(left < right) {
  918. int middle = (int)(((unsigned int)(left + right)) >> 1);
  919. if(arr[middle] > K)
  920. right = middle;
  921. else
  922. left = middle + 1;
  923. }
  924. return left;
  925. }
  926. int compare_diffs(const void *left, const void *right) {
  927. DIFFS_NUMBERS lt = *(DIFFS_NUMBERS *)left;
  928. DIFFS_NUMBERS rt = *(DIFFS_NUMBERS *)right;
  929. // https://stackoverflow.com/a/3886497/1114110
  930. return (lt > rt) - (lt < rt);
  931. }
  932. static size_t calculate_pairs_diff(DIFFS_NUMBERS *diffs, NETDATA_DOUBLE *arr, size_t size) {
  933. NETDATA_DOUBLE *last = &arr[size - 1];
  934. size_t added = 0;
  935. while(last > arr) {
  936. NETDATA_DOUBLE second = *last--;
  937. NETDATA_DOUBLE first = *last;
  938. *diffs++ = (DIFFS_NUMBERS)((first - second) * (NETDATA_DOUBLE)DOUBLE_TO_INT_MULTIPLIER);
  939. added++;
  940. }
  941. return added;
  942. }
  943. static double ks_2samp(
  944. DIFFS_NUMBERS baseline_diffs[], int base_size,
  945. DIFFS_NUMBERS highlight_diffs[], int high_size,
  946. uint32_t base_shifts) {
  947. qsort(baseline_diffs, base_size, sizeof(DIFFS_NUMBERS), compare_diffs);
  948. qsort(highlight_diffs, high_size, sizeof(DIFFS_NUMBERS), compare_diffs);
  949. // Now we should be calculating this:
  950. //
  951. // For each number in the diffs arrays, we should find the index of the
  952. // number bigger than them in both arrays and calculate the % of this index
  953. // vs the total array size. Once we have the 2 percentages, we should find
  954. // the min and max across the delta of all of them.
  955. //
  956. // It should look like this:
  957. //
  958. // base_pcent = binary_search_bigger_than(...) / base_size;
  959. // high_pcent = binary_search_bigger_than(...) / high_size;
  960. // delta = base_pcent - high_pcent;
  961. // if(delta < min) min = delta;
  962. // if(delta > max) max = delta;
  963. //
  964. // This would require a lot of multiplications and divisions.
  965. //
  966. // To speed it up, we do the binary search to find the index of each number
  967. // but, then we divide the base index by the power of two number (shifts) it
  968. // is bigger than high index. So the 2 indexes are now comparable.
  969. // We also keep track of the original indexes with min and max, to properly
  970. // calculate their percentages once the loops finish.
  971. // initialize min and max using the first number of baseline_diffs
  972. DIFFS_NUMBERS K = baseline_diffs[0];
  973. int base_idx = binary_search_bigger_than(baseline_diffs, 1, base_size, K);
  974. int high_idx = binary_search_bigger_than(highlight_diffs, 0, high_size, K);
  975. int delta = base_idx - (high_idx << base_shifts);
  976. int min = delta, max = delta;
  977. int base_min_idx = base_idx;
  978. int base_max_idx = base_idx;
  979. int high_min_idx = high_idx;
  980. int high_max_idx = high_idx;
  981. // do the baseline_diffs starting from 1 (we did position 0 above)
  982. for(int i = 1; i < base_size; i++) {
  983. K = baseline_diffs[i];
  984. base_idx = binary_search_bigger_than(baseline_diffs, i + 1, base_size, K); // starting from i, since data1 is sorted
  985. high_idx = binary_search_bigger_than(highlight_diffs, 0, high_size, K);
  986. delta = base_idx - (high_idx << base_shifts);
  987. if(delta < min) {
  988. min = delta;
  989. base_min_idx = base_idx;
  990. high_min_idx = high_idx;
  991. }
  992. else if(delta > max) {
  993. max = delta;
  994. base_max_idx = base_idx;
  995. high_max_idx = high_idx;
  996. }
  997. }
  998. // do the highlight_diffs starting from 0
  999. for(int i = 0; i < high_size; i++) {
  1000. K = highlight_diffs[i];
  1001. base_idx = binary_search_bigger_than(baseline_diffs, 0, base_size, K);
  1002. high_idx = binary_search_bigger_than(highlight_diffs, i + 1, high_size, K); // starting from i, since data2 is sorted
  1003. delta = base_idx - (high_idx << base_shifts);
  1004. if(delta < min) {
  1005. min = delta;
  1006. base_min_idx = base_idx;
  1007. high_min_idx = high_idx;
  1008. }
  1009. else if(delta > max) {
  1010. max = delta;
  1011. base_max_idx = base_idx;
  1012. high_max_idx = high_idx;
  1013. }
  1014. }
  1015. // now we have the min, max and their indexes
  1016. // properly calculate min and max as dmin and dmax
  1017. double dbase_size = (double)base_size;
  1018. double dhigh_size = (double)high_size;
  1019. double dmin = ((double)base_min_idx / dbase_size) - ((double)high_min_idx / dhigh_size);
  1020. double dmax = ((double)base_max_idx / dbase_size) - ((double)high_max_idx / dhigh_size);
  1021. dmin = -dmin;
  1022. if(islessequal(dmin, 0.0)) dmin = 0.0;
  1023. else if(isgreaterequal(dmin, 1.0)) dmin = 1.0;
  1024. double d;
  1025. if(isgreaterequal(dmin, dmax)) d = dmin;
  1026. else d = dmax;
  1027. double en = round(dbase_size * dhigh_size / (dbase_size + dhigh_size));
  1028. // under these conditions, KSfbar() crashes
  1029. if(unlikely(isnan(en) || isinf(en) || en == 0.0 || isnan(d) || isinf(d)))
  1030. return NAN;
  1031. return KSfbar((int)en, d);
  1032. }
  1033. static double kstwo(
  1034. NETDATA_DOUBLE baseline[], int baseline_points,
  1035. NETDATA_DOUBLE highlight[], int highlight_points,
  1036. uint32_t base_shifts) {
  1037. // -1 in size, since the calculate_pairs_diffs() returns one less point
  1038. DIFFS_NUMBERS baseline_diffs[baseline_points - 1];
  1039. DIFFS_NUMBERS highlight_diffs[highlight_points - 1];
  1040. int base_size = (int)calculate_pairs_diff(baseline_diffs, baseline, baseline_points);
  1041. int high_size = (int)calculate_pairs_diff(highlight_diffs, highlight, highlight_points);
  1042. if(unlikely(!base_size || !high_size))
  1043. return NAN;
  1044. if(unlikely(base_size != baseline_points - 1 || high_size != highlight_points - 1)) {
  1045. error("Metric correlations: internal error - calculate_pairs_diff() returns the wrong number of entries");
  1046. return NAN;
  1047. }
  1048. return ks_2samp(baseline_diffs, base_size, highlight_diffs, high_size, base_shifts);
  1049. }
  1050. NETDATA_DOUBLE *rrd2rrdr_ks2(
  1051. ONEWAYALLOC *owa, RRDHOST *host,
  1052. RRDCONTEXT_ACQUIRED *rca, RRDINSTANCE_ACQUIRED *ria, RRDMETRIC_ACQUIRED *rma,
  1053. time_t after, time_t before, size_t points, RRDR_OPTIONS options,
  1054. RRDR_TIME_GROUPING time_group_method, const char *time_group_options, size_t tier,
  1055. WEIGHTS_STATS *stats,
  1056. size_t *entries,
  1057. STORAGE_POINT *sp
  1058. ) {
  1059. NETDATA_DOUBLE *ret = NULL;
  1060. QUERY_TARGET_REQUEST qtr = {
  1061. .version = 1,
  1062. .host = host,
  1063. .rca = rca,
  1064. .ria = ria,
  1065. .rma = rma,
  1066. .after = after,
  1067. .before = before,
  1068. .points = points,
  1069. .options = options,
  1070. .time_group_method = time_group_method,
  1071. .time_group_options = time_group_options,
  1072. .tier = tier,
  1073. .query_source = QUERY_SOURCE_API_WEIGHTS,
  1074. .priority = STORAGE_PRIORITY_SYNCHRONOUS,
  1075. };
  1076. QUERY_TARGET *qt = query_target_create(&qtr);
  1077. RRDR *r = rrd2rrdr(owa, qt);
  1078. if(!r)
  1079. goto cleanup;
  1080. stats->db_queries++;
  1081. stats->result_points += r->stats.result_points_generated;
  1082. stats->db_points += r->stats.db_points_read;
  1083. for(size_t tr = 0; tr < storage_tiers ; tr++)
  1084. stats->db_points_per_tier[tr] += r->internal.qt->db.tiers[tr].points;
  1085. if(r->d != 1 || r->internal.qt->query.used != 1) {
  1086. error("WEIGHTS: on query '%s' expected 1 dimension in RRDR but got %zu r->d and %zu qt->query.used",
  1087. r->internal.qt->id, r->d, (size_t)r->internal.qt->query.used);
  1088. goto cleanup;
  1089. }
  1090. if(unlikely(r->od[0] & RRDR_DIMENSION_HIDDEN))
  1091. goto cleanup;
  1092. if(unlikely(!(r->od[0] & RRDR_DIMENSION_QUERIED)))
  1093. goto cleanup;
  1094. if(unlikely(!(r->od[0] & RRDR_DIMENSION_NONZERO)))
  1095. goto cleanup;
  1096. if(rrdr_rows(r) < 2)
  1097. goto cleanup;
  1098. *entries = rrdr_rows(r);
  1099. ret = onewayalloc_mallocz(owa, sizeof(NETDATA_DOUBLE) * rrdr_rows(r));
  1100. if(sp)
  1101. *sp = r->internal.qt->query.array[0].query_points;
  1102. // copy the points of the dimension to a contiguous array
  1103. // there is no need to check for empty values, since empty values are already zero
  1104. // https://github.com/netdata/netdata/blob/6e3144683a73a2024d51425b20ecfd569034c858/web/api/queries/average/average.c#L41-L43
  1105. memcpy(ret, r->v, rrdr_rows(r) * sizeof(NETDATA_DOUBLE));
  1106. cleanup:
  1107. rrdr_free(owa, r);
  1108. query_target_release(qt);
  1109. return ret;
  1110. }
  1111. static void rrdset_metric_correlations_ks2(
  1112. RRDHOST *host,
  1113. RRDCONTEXT_ACQUIRED *rca, RRDINSTANCE_ACQUIRED *ria, RRDMETRIC_ACQUIRED *rma,
  1114. DICTIONARY *results,
  1115. time_t baseline_after, time_t baseline_before,
  1116. time_t after, time_t before,
  1117. size_t points, RRDR_OPTIONS options,
  1118. RRDR_TIME_GROUPING time_group_method, const char *time_group_options, size_t tier,
  1119. uint32_t shifts,
  1120. WEIGHTS_STATS *stats, bool register_zero
  1121. ) {
  1122. options |= RRDR_OPTION_NATURAL_POINTS;
  1123. usec_t started_ut = now_monotonic_usec();
  1124. ONEWAYALLOC *owa = onewayalloc_create(16 * 1024);
  1125. size_t high_points = 0;
  1126. STORAGE_POINT highlighted_sp;
  1127. NETDATA_DOUBLE *highlight = rrd2rrdr_ks2(
  1128. owa, host, rca, ria, rma, after, before, points,
  1129. options, time_group_method, time_group_options, tier, stats, &high_points, &highlighted_sp);
  1130. if(!highlight)
  1131. goto cleanup;
  1132. size_t base_points = 0;
  1133. STORAGE_POINT baseline_sp;
  1134. NETDATA_DOUBLE *baseline = rrd2rrdr_ks2(
  1135. owa, host, rca, ria, rma, baseline_after, baseline_before, high_points << shifts,
  1136. options, time_group_method, time_group_options, tier, stats, &base_points, &baseline_sp);
  1137. if(!baseline)
  1138. goto cleanup;
  1139. stats->binary_searches += 2 * (base_points - 1) + 2 * (high_points - 1);
  1140. double prob = kstwo(baseline, (int)base_points, highlight, (int)high_points, shifts);
  1141. if(!isnan(prob) && !isinf(prob)) {
  1142. // these conditions should never happen, but still let's check
  1143. if(unlikely(prob < 0.0)) {
  1144. error("Metric correlations: kstwo() returned a negative number: %f", prob);
  1145. prob = -prob;
  1146. }
  1147. if(unlikely(prob > 1.0)) {
  1148. error("Metric correlations: kstwo() returned a number above 1.0: %f", prob);
  1149. prob = 1.0;
  1150. }
  1151. usec_t ended_ut = now_monotonic_usec();
  1152. // to spread the results evenly, 0.0 needs to be the less correlated and 1.0 the most correlated
  1153. // so, we flip the result of kstwo()
  1154. register_result(results, host, rca, ria, rma, 1.0 - prob, RESULT_IS_BASE_HIGH_RATIO, &highlighted_sp,
  1155. &baseline_sp, stats, register_zero, ended_ut - started_ut);
  1156. }
  1157. cleanup:
  1158. onewayalloc_destroy(owa);
  1159. }
  1160. // ----------------------------------------------------------------------------
  1161. // VOLUME algorithm functions
  1162. static void merge_query_value_to_stats(QUERY_VALUE *qv, WEIGHTS_STATS *stats, size_t queries) {
  1163. stats->db_queries += queries;
  1164. stats->result_points += qv->result_points;
  1165. stats->db_points += qv->points_read;
  1166. for(size_t tier = 0; tier < storage_tiers ; tier++)
  1167. stats->db_points_per_tier[tier] += qv->storage_points_per_tier[tier];
  1168. }
  1169. static void rrdset_metric_correlations_volume(
  1170. RRDHOST *host,
  1171. RRDCONTEXT_ACQUIRED *rca, RRDINSTANCE_ACQUIRED *ria, RRDMETRIC_ACQUIRED *rma,
  1172. DICTIONARY *results,
  1173. time_t baseline_after, time_t baseline_before,
  1174. time_t after, time_t before,
  1175. RRDR_OPTIONS options, RRDR_TIME_GROUPING time_group_method, const char *time_group_options,
  1176. size_t tier,
  1177. WEIGHTS_STATS *stats, bool register_zero) {
  1178. options |= RRDR_OPTION_MATCH_IDS | RRDR_OPTION_ABSOLUTE | RRDR_OPTION_NATURAL_POINTS;
  1179. QUERY_VALUE baseline_average = rrdmetric2value(host, rca, ria, rma, baseline_after, baseline_before,
  1180. options, time_group_method, time_group_options, tier, 0,
  1181. QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS);
  1182. merge_query_value_to_stats(&baseline_average, stats, 1);
  1183. if(!netdata_double_isnumber(baseline_average.value)) {
  1184. // this means no data for the baseline window, but we may have data for the highlighted one - assume zero
  1185. baseline_average.value = 0.0;
  1186. }
  1187. QUERY_VALUE highlight_average = rrdmetric2value(host, rca, ria, rma, after, before,
  1188. options, time_group_method, time_group_options, tier, 0,
  1189. QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS);
  1190. merge_query_value_to_stats(&highlight_average, stats, 1);
  1191. if(!netdata_double_isnumber(highlight_average.value))
  1192. return;
  1193. if(baseline_average.value == highlight_average.value) {
  1194. // they are the same - let's move on
  1195. return;
  1196. }
  1197. if((options & RRDR_OPTION_ANOMALY_BIT) && highlight_average.value < baseline_average.value) {
  1198. // when working on anomaly bits, we are looking for an increase in the anomaly rate
  1199. return;
  1200. }
  1201. char highlight_countif_options[50 + 1];
  1202. snprintfz(highlight_countif_options, 50, "%s" NETDATA_DOUBLE_FORMAT, highlight_average.value < baseline_average.value ? "<" : ">", baseline_average.value);
  1203. QUERY_VALUE highlight_countif = rrdmetric2value(host, rca, ria, rma, after, before,
  1204. options, RRDR_GROUPING_COUNTIF, highlight_countif_options, tier, 0,
  1205. QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS);
  1206. merge_query_value_to_stats(&highlight_countif, stats, 1);
  1207. if(!netdata_double_isnumber(highlight_countif.value)) {
  1208. info("WEIGHTS: highlighted countif query failed, but highlighted average worked - strange...");
  1209. return;
  1210. }
  1211. // this represents the percentage of time
  1212. // the highlighted window was above/below the baseline window
  1213. // (above or below depending on their averages)
  1214. highlight_countif.value = highlight_countif.value / 100.0; // countif returns 0 - 100.0
  1215. RESULT_FLAGS flags;
  1216. NETDATA_DOUBLE pcent = NAN;
  1217. if(isgreater(baseline_average.value, 0.0) || isless(baseline_average.value, 0.0)) {
  1218. flags = RESULT_IS_BASE_HIGH_RATIO;
  1219. pcent = (highlight_average.value - baseline_average.value) / baseline_average.value * highlight_countif.value;
  1220. }
  1221. else {
  1222. flags = RESULT_IS_PERCENTAGE_OF_TIME;
  1223. pcent = highlight_countif.value;
  1224. }
  1225. register_result(results, host, rca, ria, rma, pcent, flags, &highlight_average.sp, &baseline_average.sp, stats,
  1226. register_zero, baseline_average.duration_ut + highlight_average.duration_ut + highlight_countif.duration_ut);
  1227. }
  1228. // ----------------------------------------------------------------------------
  1229. // VALUE / ANOMALY RATE algorithm functions
  1230. static void rrdset_weights_value(
  1231. RRDHOST *host,
  1232. RRDCONTEXT_ACQUIRED *rca, RRDINSTANCE_ACQUIRED *ria, RRDMETRIC_ACQUIRED *rma,
  1233. DICTIONARY *results,
  1234. time_t after, time_t before,
  1235. RRDR_OPTIONS options, RRDR_TIME_GROUPING time_group_method, const char *time_group_options,
  1236. size_t tier,
  1237. WEIGHTS_STATS *stats, bool register_zero) {
  1238. options |= RRDR_OPTION_MATCH_IDS | RRDR_OPTION_NATURAL_POINTS;
  1239. QUERY_VALUE qv = rrdmetric2value(host, rca, ria, rma, after, before,
  1240. options, time_group_method, time_group_options, tier, 0,
  1241. QUERY_SOURCE_API_WEIGHTS, STORAGE_PRIORITY_SYNCHRONOUS);
  1242. merge_query_value_to_stats(&qv, stats, 1);
  1243. if(netdata_double_isnumber(qv.value))
  1244. register_result(results, host, rca, ria, rma, qv.value, 0, &qv.sp, NULL, stats, register_zero, qv.duration_ut);
  1245. }
  1246. static void rrdset_weights_multi_dimensional_value(struct query_weights_data *qwd) {
  1247. QUERY_TARGET_REQUEST qtr = {
  1248. .version = 1,
  1249. .scope_nodes = qwd->qwr->scope_nodes,
  1250. .scope_contexts = qwd->qwr->scope_contexts,
  1251. .nodes = qwd->qwr->nodes,
  1252. .contexts = qwd->qwr->contexts,
  1253. .instances = qwd->qwr->instances,
  1254. .dimensions = qwd->qwr->dimensions,
  1255. .labels = qwd->qwr->labels,
  1256. .alerts = qwd->qwr->alerts,
  1257. .after = qwd->qwr->after,
  1258. .before = qwd->qwr->before,
  1259. .points = 1,
  1260. .options = qwd->qwr->options | RRDR_OPTION_NATURAL_POINTS,
  1261. .time_group_method = qwd->qwr->time_group_method,
  1262. .time_group_options = qwd->qwr->time_group_options,
  1263. .tier = qwd->qwr->tier,
  1264. .timeout_ms = qwd->qwr->timeout_ms,
  1265. .query_source = QUERY_SOURCE_API_WEIGHTS,
  1266. .priority = STORAGE_PRIORITY_NORMAL,
  1267. };
  1268. ONEWAYALLOC *owa = onewayalloc_create(16 * 1024);
  1269. QUERY_TARGET *qt = query_target_create(&qtr);
  1270. RRDR *r = rrd2rrdr(owa, qt);
  1271. if(!r || rrdr_rows(r) != 1 || !r->d || r->d != r->internal.qt->query.used)
  1272. goto cleanup;
  1273. QUERY_VALUE qv = {
  1274. .after = r->view.after,
  1275. .before = r->view.before,
  1276. .points_read = r->stats.db_points_read,
  1277. .result_points = r->stats.result_points_generated,
  1278. };
  1279. size_t queries = 0;
  1280. for(size_t d = 0; d < r->d ;d++) {
  1281. if(!rrdr_dimension_should_be_exposed(r->od[d], qwd->qwr->options))
  1282. continue;
  1283. long i = 0; // only one row
  1284. NETDATA_DOUBLE *cn = &r->v[ i * r->d ];
  1285. NETDATA_DOUBLE *ar = &r->ar[ i * r->d ];
  1286. qv.value = cn[d];
  1287. qv.anomaly_rate = ar[d];
  1288. storage_point_merge_to(qv.sp, r->internal.qt->query.array[d].query_points);
  1289. if(netdata_double_isnumber(qv.value)) {
  1290. QUERY_METRIC *qm = query_metric(r->internal.qt, d);
  1291. QUERY_DIMENSION *qd = query_dimension(r->internal.qt, qm->link.query_dimension_id);
  1292. QUERY_INSTANCE *qi = query_instance(r->internal.qt, qm->link.query_instance_id);
  1293. QUERY_CONTEXT *qc = query_context(r->internal.qt, qm->link.query_context_id);
  1294. QUERY_NODE *qn = query_node(r->internal.qt, qm->link.query_node_id);
  1295. register_result(qwd->results, qn->rrdhost, qc->rca, qi->ria, qd->rma, qv.value, 0, &qv.sp,
  1296. NULL, &qwd->stats, qwd->register_zero, qm->duration_ut);
  1297. }
  1298. queries++;
  1299. }
  1300. merge_query_value_to_stats(&qv, &qwd->stats, queries);
  1301. cleanup:
  1302. rrdr_free(owa, r);
  1303. query_target_release(qt);
  1304. onewayalloc_destroy(owa);
  1305. }
  1306. // ----------------------------------------------------------------------------
  1307. int compare_netdata_doubles(const void *left, const void *right) {
  1308. NETDATA_DOUBLE lt = *(NETDATA_DOUBLE *)left;
  1309. NETDATA_DOUBLE rt = *(NETDATA_DOUBLE *)right;
  1310. // https://stackoverflow.com/a/3886497/1114110
  1311. return (lt > rt) - (lt < rt);
  1312. }
  1313. static inline int binary_search_bigger_than_netdata_double(const NETDATA_DOUBLE arr[], int left, int size, NETDATA_DOUBLE K) {
  1314. // binary search to find the index the smallest index
  1315. // of the first value in the array that is greater than K
  1316. int right = size;
  1317. while(left < right) {
  1318. int middle = (int)(((unsigned int)(left + right)) >> 1);
  1319. if(arr[middle] > K)
  1320. right = middle;
  1321. else
  1322. left = middle + 1;
  1323. }
  1324. return left;
  1325. }
  1326. // ----------------------------------------------------------------------------
  1327. // spread the results evenly according to their value
  1328. static size_t spread_results_evenly(DICTIONARY *results, WEIGHTS_STATS *stats) {
  1329. struct register_result *t;
  1330. // count the dimensions
  1331. size_t dimensions = dictionary_entries(results);
  1332. if(!dimensions) return 0;
  1333. if(stats->max_base_high_ratio == 0.0)
  1334. stats->max_base_high_ratio = 1.0;
  1335. // create an array of the right size and copy all the values in it
  1336. NETDATA_DOUBLE slots[dimensions];
  1337. dimensions = 0;
  1338. dfe_start_read(results, t) {
  1339. if(t->flags & RESULT_IS_PERCENTAGE_OF_TIME)
  1340. t->value = t->value * stats->max_base_high_ratio;
  1341. slots[dimensions++] = t->value;
  1342. }
  1343. dfe_done(t);
  1344. if(!dimensions) return 0; // Coverity fix
  1345. // sort the array with the values of all dimensions
  1346. qsort(slots, dimensions, sizeof(NETDATA_DOUBLE), compare_netdata_doubles);
  1347. // skip the duplicates in the sorted array
  1348. NETDATA_DOUBLE last_value = NAN;
  1349. size_t unique_values = 0;
  1350. for(size_t i = 0; i < dimensions ;i++) {
  1351. if(likely(slots[i] != last_value))
  1352. slots[unique_values++] = last_value = slots[i];
  1353. }
  1354. // this cannot happen, but coverity thinks otherwise...
  1355. if(!unique_values)
  1356. unique_values = dimensions;
  1357. // calculate the weight of each slot, using the number of unique values
  1358. NETDATA_DOUBLE slot_weight = 1.0 / (NETDATA_DOUBLE)unique_values;
  1359. dfe_start_read(results, t) {
  1360. int slot = binary_search_bigger_than_netdata_double(slots, 0, (int)unique_values, t->value);
  1361. NETDATA_DOUBLE v = slot * slot_weight;
  1362. if(unlikely(v > 1.0)) v = 1.0;
  1363. v = 1.0 - v;
  1364. t->value = v;
  1365. }
  1366. dfe_done(t);
  1367. return dimensions;
  1368. }
  1369. // ----------------------------------------------------------------------------
  1370. // The main function
  1371. static ssize_t weights_for_rrdmetric(void *data, RRDHOST *host, RRDCONTEXT_ACQUIRED *rca, RRDINSTANCE_ACQUIRED *ria, RRDMETRIC_ACQUIRED *rma) {
  1372. struct query_weights_data *qwd = data;
  1373. QUERY_WEIGHTS_REQUEST *qwr = qwd->qwr;
  1374. if(qwd->qwr->interrupt_callback && qwd->qwr->interrupt_callback(qwd->qwr->interrupt_callback_data)) {
  1375. qwd->interrupted = true;
  1376. return -1;
  1377. }
  1378. qwd->examined_dimensions++;
  1379. switch(qwr->method) {
  1380. case WEIGHTS_METHOD_VALUE:
  1381. rrdset_weights_value(
  1382. host, rca, ria, rma,
  1383. qwd->results,
  1384. qwr->after, qwr->before,
  1385. qwr->options, qwr->time_group_method, qwr->time_group_options, qwr->tier,
  1386. &qwd->stats, qwd->register_zero
  1387. );
  1388. break;
  1389. case WEIGHTS_METHOD_ANOMALY_RATE:
  1390. qwr->options |= RRDR_OPTION_ANOMALY_BIT;
  1391. rrdset_weights_value(
  1392. host, rca, ria, rma,
  1393. qwd->results,
  1394. qwr->after, qwr->before,
  1395. qwr->options, qwr->time_group_method, qwr->time_group_options, qwr->tier,
  1396. &qwd->stats, qwd->register_zero
  1397. );
  1398. break;
  1399. case WEIGHTS_METHOD_MC_VOLUME:
  1400. rrdset_metric_correlations_volume(
  1401. host, rca, ria, rma,
  1402. qwd->results,
  1403. qwr->baseline_after, qwr->baseline_before,
  1404. qwr->after, qwr->before,
  1405. qwr->options, qwr->time_group_method, qwr->time_group_options, qwr->tier,
  1406. &qwd->stats, qwd->register_zero
  1407. );
  1408. break;
  1409. default:
  1410. case WEIGHTS_METHOD_MC_KS2:
  1411. rrdset_metric_correlations_ks2(
  1412. host, rca, ria, rma,
  1413. qwd->results,
  1414. qwr->baseline_after, qwr->baseline_before,
  1415. qwr->after, qwr->before, qwr->points,
  1416. qwr->options, qwr->time_group_method, qwr->time_group_options, qwr->tier, qwd->shifts,
  1417. &qwd->stats, qwd->register_zero
  1418. );
  1419. break;
  1420. }
  1421. qwd->timings.executed_ut = now_monotonic_usec();
  1422. if(qwd->timings.executed_ut - qwd->timings.received_ut > qwd->timeout_us) {
  1423. qwd->timed_out = true;
  1424. return -1;
  1425. }
  1426. return 1;
  1427. }
  1428. static ssize_t weights_do_context_callback(void *data, RRDCONTEXT_ACQUIRED *rca, bool queryable_context) {
  1429. if(!queryable_context)
  1430. return false;
  1431. struct query_weights_data *qwd = data;
  1432. bool has_retention = false;
  1433. switch(qwd->qwr->method) {
  1434. case WEIGHTS_METHOD_VALUE:
  1435. case WEIGHTS_METHOD_ANOMALY_RATE:
  1436. has_retention = rrdcontext_retention_match(rca, qwd->qwr->after, qwd->qwr->before);
  1437. break;
  1438. case WEIGHTS_METHOD_MC_KS2:
  1439. case WEIGHTS_METHOD_MC_VOLUME:
  1440. has_retention = rrdcontext_retention_match(rca, qwd->qwr->after, qwd->qwr->before);
  1441. if(has_retention)
  1442. has_retention = rrdcontext_retention_match(rca, qwd->qwr->baseline_after, qwd->qwr->baseline_before);
  1443. break;
  1444. }
  1445. if(!has_retention)
  1446. return 0;
  1447. ssize_t ret = weights_foreach_rrdmetric_in_context(rca,
  1448. qwd->instances_sp,
  1449. NULL,
  1450. qwd->labels_sp,
  1451. qwd->alerts_sp,
  1452. qwd->dimensions_sp,
  1453. true, true, qwd->qwr->version,
  1454. weights_for_rrdmetric, qwd);
  1455. return ret;
  1456. }
  1457. ssize_t weights_do_node_callback(void *data, RRDHOST *host, bool queryable) {
  1458. if(!queryable)
  1459. return 0;
  1460. struct query_weights_data *qwd = data;
  1461. ssize_t ret = query_scope_foreach_context(host, qwd->qwr->scope_contexts,
  1462. qwd->scope_contexts_sp, qwd->contexts_sp,
  1463. weights_do_context_callback, queryable, qwd);
  1464. return ret;
  1465. }
  1466. int web_api_v12_weights(BUFFER *wb, QUERY_WEIGHTS_REQUEST *qwr) {
  1467. char *error = NULL;
  1468. int resp = HTTP_RESP_OK;
  1469. // if the user didn't give a timeout
  1470. // assume 60 seconds
  1471. if(!qwr->timeout_ms)
  1472. qwr->timeout_ms = 5 * 60 * MSEC_PER_SEC;
  1473. // if the timeout is less than 1 second
  1474. // make it at least 1 second
  1475. if(qwr->timeout_ms < (long)(1 * MSEC_PER_SEC))
  1476. qwr->timeout_ms = 1 * MSEC_PER_SEC;
  1477. struct query_weights_data qwd = {
  1478. .qwr = qwr,
  1479. .scope_nodes_sp = string_to_simple_pattern(qwr->scope_nodes),
  1480. .scope_contexts_sp = string_to_simple_pattern(qwr->scope_contexts),
  1481. .nodes_sp = string_to_simple_pattern(qwr->nodes),
  1482. .contexts_sp = string_to_simple_pattern(qwr->contexts),
  1483. .instances_sp = string_to_simple_pattern(qwr->instances),
  1484. .dimensions_sp = string_to_simple_pattern(qwr->dimensions),
  1485. .labels_sp = string_to_simple_pattern(qwr->labels),
  1486. .alerts_sp = string_to_simple_pattern(qwr->alerts),
  1487. .timeout_us = qwr->timeout_ms * USEC_PER_MS,
  1488. .timed_out = false,
  1489. .examined_dimensions = 0,
  1490. .register_zero = true,
  1491. .results = register_result_init(),
  1492. .stats = {},
  1493. .shifts = 0,
  1494. .timings = {
  1495. .received_ut = now_monotonic_usec(),
  1496. }
  1497. };
  1498. if(!rrdr_relative_window_to_absolute(&qwr->after, &qwr->before, NULL))
  1499. buffer_no_cacheable(wb);
  1500. if (qwr->before <= qwr->after) {
  1501. resp = HTTP_RESP_BAD_REQUEST;
  1502. error = "Invalid selected time-range.";
  1503. goto cleanup;
  1504. }
  1505. if(qwr->method == WEIGHTS_METHOD_MC_KS2 || qwr->method == WEIGHTS_METHOD_MC_VOLUME) {
  1506. if(!qwr->points) qwr->points = 500;
  1507. if(qwr->baseline_before <= API_RELATIVE_TIME_MAX)
  1508. qwr->baseline_before += qwr->after;
  1509. rrdr_relative_window_to_absolute(&qwr->baseline_after, &qwr->baseline_before, NULL);
  1510. if (qwr->baseline_before <= qwr->baseline_after) {
  1511. resp = HTTP_RESP_BAD_REQUEST;
  1512. error = "Invalid baseline time-range.";
  1513. goto cleanup;
  1514. }
  1515. // baseline should be a power of two multiple of highlight
  1516. long long base_delta = qwr->baseline_before - qwr->baseline_after;
  1517. long long high_delta = qwr->before - qwr->after;
  1518. uint32_t multiplier = (uint32_t)round((double)base_delta / (double)high_delta);
  1519. // check if the multiplier is a power of two
  1520. // https://stackoverflow.com/a/600306/1114110
  1521. if((multiplier & (multiplier - 1)) != 0) {
  1522. // it is not power of two
  1523. // let's find the closest power of two
  1524. // https://stackoverflow.com/a/466242/1114110
  1525. multiplier--;
  1526. multiplier |= multiplier >> 1;
  1527. multiplier |= multiplier >> 2;
  1528. multiplier |= multiplier >> 4;
  1529. multiplier |= multiplier >> 8;
  1530. multiplier |= multiplier >> 16;
  1531. multiplier++;
  1532. }
  1533. // convert the multiplier to the number of shifts
  1534. // we need to do, to divide baseline numbers to match
  1535. // the highlight ones
  1536. while(multiplier > 1) {
  1537. qwd.shifts++;
  1538. multiplier = multiplier >> 1;
  1539. }
  1540. // if the baseline size will not comply to MAX_POINTS
  1541. // lower the window of the baseline
  1542. while(qwd.shifts && (qwr->points << qwd.shifts) > MAX_POINTS)
  1543. qwd.shifts--;
  1544. // if the baseline size still does not comply to MAX_POINTS
  1545. // lower the resolution of the highlight and the baseline
  1546. while((qwr->points << qwd.shifts) > MAX_POINTS)
  1547. qwr->points = qwr->points >> 1;
  1548. if(qwr->points < 15) {
  1549. resp = HTTP_RESP_BAD_REQUEST;
  1550. error = "Too few points available, at least 15 are needed.";
  1551. goto cleanup;
  1552. }
  1553. // adjust the baseline to be multiplier times bigger than the highlight
  1554. qwr->baseline_after = qwr->baseline_before - (high_delta << qwd.shifts);
  1555. }
  1556. if(qwr->options & RRDR_OPTION_NONZERO) {
  1557. qwd.register_zero = false;
  1558. // remove it to run the queries without it
  1559. qwr->options &= ~RRDR_OPTION_NONZERO;
  1560. }
  1561. if(qwr->host && qwr->version == 1)
  1562. weights_do_node_callback(&qwd, qwr->host, true);
  1563. else {
  1564. if((qwd.qwr->method == WEIGHTS_METHOD_VALUE || qwd.qwr->method == WEIGHTS_METHOD_ANOMALY_RATE) && (qwd.contexts_sp || qwd.scope_contexts_sp)) {
  1565. rrdset_weights_multi_dimensional_value(&qwd);
  1566. }
  1567. else {
  1568. query_scope_foreach_host(qwd.scope_nodes_sp, qwd.nodes_sp,
  1569. weights_do_node_callback, &qwd,
  1570. &qwd.versions,
  1571. NULL);
  1572. }
  1573. }
  1574. if(!qwd.register_zero) {
  1575. // put it back, to show it in the response
  1576. qwr->options |= RRDR_OPTION_NONZERO;
  1577. }
  1578. if(qwd.timed_out) {
  1579. error = "timed out";
  1580. resp = HTTP_RESP_GATEWAY_TIMEOUT;
  1581. goto cleanup;
  1582. }
  1583. if(qwd.interrupted) {
  1584. error = "interrupted";
  1585. resp = HTTP_RESP_BACKEND_FETCH_FAILED;
  1586. goto cleanup;
  1587. }
  1588. if(!qwd.register_zero)
  1589. qwr->options |= RRDR_OPTION_NONZERO;
  1590. if(!(qwr->options & RRDR_OPTION_RETURN_RAW) && qwr->method != WEIGHTS_METHOD_VALUE)
  1591. spread_results_evenly(qwd.results, &qwd.stats);
  1592. usec_t ended_usec = qwd.timings.executed_ut = now_monotonic_usec();
  1593. // generate the json output we need
  1594. buffer_flush(wb);
  1595. size_t added_dimensions = 0;
  1596. switch(qwr->format) {
  1597. case WEIGHTS_FORMAT_CHARTS:
  1598. added_dimensions =
  1599. registered_results_to_json_charts(
  1600. qwd.results, wb,
  1601. qwr->after, qwr->before,
  1602. qwr->baseline_after, qwr->baseline_before,
  1603. qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
  1604. qwd.examined_dimensions,
  1605. ended_usec - qwd.timings.received_ut, &qwd.stats);
  1606. break;
  1607. case WEIGHTS_FORMAT_CONTEXTS:
  1608. added_dimensions =
  1609. registered_results_to_json_contexts(
  1610. qwd.results, wb,
  1611. qwr->after, qwr->before,
  1612. qwr->baseline_after, qwr->baseline_before,
  1613. qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
  1614. qwd.examined_dimensions,
  1615. ended_usec - qwd.timings.received_ut, &qwd.stats);
  1616. break;
  1617. default:
  1618. case WEIGHTS_FORMAT_MULTINODE:
  1619. // we don't support these groupings in weights
  1620. qwr->group_by.group_by &= ~(RRDR_GROUP_BY_LABEL|RRDR_GROUP_BY_SELECTED|RRDR_GROUP_BY_PERCENTAGE_OF_INSTANCE);
  1621. if(qwr->group_by.group_by == RRDR_GROUP_BY_NONE) {
  1622. added_dimensions =
  1623. registered_results_to_json_multinode_no_group_by(
  1624. qwd.results, wb,
  1625. qwr->after, qwr->before,
  1626. qwr->baseline_after, qwr->baseline_before,
  1627. qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
  1628. qwd.examined_dimensions,
  1629. &qwd, &qwd.stats, &qwd.versions);
  1630. }
  1631. else {
  1632. added_dimensions =
  1633. registered_results_to_json_multinode_group_by(
  1634. qwd.results, wb,
  1635. qwr->after, qwr->before,
  1636. qwr->baseline_after, qwr->baseline_before,
  1637. qwr->points, qwr->method, qwr->time_group_method, qwr->options, qwd.shifts,
  1638. qwd.examined_dimensions,
  1639. &qwd, &qwd.stats, &qwd.versions);
  1640. }
  1641. break;
  1642. }
  1643. if(!added_dimensions && qwr->version < 2) {
  1644. error = "no results produced.";
  1645. resp = HTTP_RESP_NOT_FOUND;
  1646. }
  1647. cleanup:
  1648. simple_pattern_free(qwd.scope_nodes_sp);
  1649. simple_pattern_free(qwd.scope_contexts_sp);
  1650. simple_pattern_free(qwd.nodes_sp);
  1651. simple_pattern_free(qwd.contexts_sp);
  1652. simple_pattern_free(qwd.instances_sp);
  1653. simple_pattern_free(qwd.dimensions_sp);
  1654. simple_pattern_free(qwd.labels_sp);
  1655. simple_pattern_free(qwd.alerts_sp);
  1656. register_result_destroy(qwd.results);
  1657. if(error) {
  1658. buffer_flush(wb);
  1659. buffer_sprintf(wb, "{\"error\": \"%s\" }", error);
  1660. }
  1661. return resp;
  1662. }
  1663. // ----------------------------------------------------------------------------
  1664. // unittest
  1665. /*
  1666. Unit tests against the output of this:
  1667. https://github.com/scipy/scipy/blob/4cf21e753cf937d1c6c2d2a0e372fbc1dbbeea81/scipy/stats/_stats_py.py#L7275-L7449
  1668. import matplotlib.pyplot as plt
  1669. import pandas as pd
  1670. import numpy as np
  1671. import scipy as sp
  1672. from scipy import stats
  1673. data1 = np.array([ 1111, -2222, 33, 100, 100, 15555, -1, 19999, 888, 755, -1, -730 ])
  1674. data2 = np.array([365, -123, 0])
  1675. data1 = np.sort(data1)
  1676. data2 = np.sort(data2)
  1677. n1 = data1.shape[0]
  1678. n2 = data2.shape[0]
  1679. data_all = np.concatenate([data1, data2])
  1680. cdf1 = np.searchsorted(data1, data_all, side='right') / n1
  1681. cdf2 = np.searchsorted(data2, data_all, side='right') / n2
  1682. print(data_all)
  1683. print("\ndata1", data1, cdf1)
  1684. print("\ndata2", data2, cdf2)
  1685. cddiffs = cdf1 - cdf2
  1686. print("\ncddiffs", cddiffs)
  1687. minS = np.clip(-np.min(cddiffs), 0, 1)
  1688. maxS = np.max(cddiffs)
  1689. print("\nmin", minS)
  1690. print("max", maxS)
  1691. m, n = sorted([float(n1), float(n2)], reverse=True)
  1692. en = m * n / (m + n)
  1693. d = max(minS, maxS)
  1694. prob = stats.distributions.kstwo.sf(d, np.round(en))
  1695. print("\nprob", prob)
  1696. */
  1697. static int double_expect(double v, const char *str, const char *descr) {
  1698. char buf[100 + 1];
  1699. snprintfz(buf, 100, "%0.6f", v);
  1700. int ret = strcmp(buf, str) ? 1 : 0;
  1701. fprintf(stderr, "%s %s, expected %s, got %s\n", ret?"FAILED":"OK", descr, str, buf);
  1702. return ret;
  1703. }
  1704. static int mc_unittest1(void) {
  1705. int bs = 3, hs = 3;
  1706. DIFFS_NUMBERS base[3] = { 1, 2, 3 };
  1707. DIFFS_NUMBERS high[3] = { 3, 4, 6 };
  1708. double prob = ks_2samp(base, bs, high, hs, 0);
  1709. return double_expect(prob, "0.222222", "3x3");
  1710. }
  1711. static int mc_unittest2(void) {
  1712. int bs = 6, hs = 3;
  1713. DIFFS_NUMBERS base[6] = { 1, 2, 3, 10, 10, 15 };
  1714. DIFFS_NUMBERS high[3] = { 3, 4, 6 };
  1715. double prob = ks_2samp(base, bs, high, hs, 1);
  1716. return double_expect(prob, "0.500000", "6x3");
  1717. }
  1718. static int mc_unittest3(void) {
  1719. int bs = 12, hs = 3;
  1720. DIFFS_NUMBERS base[12] = { 1, 2, 3, 10, 10, 15, 111, 19999, 8, 55, -1, -73 };
  1721. DIFFS_NUMBERS high[3] = { 3, 4, 6 };
  1722. double prob = ks_2samp(base, bs, high, hs, 2);
  1723. return double_expect(prob, "0.347222", "12x3");
  1724. }
  1725. static int mc_unittest4(void) {
  1726. int bs = 12, hs = 3;
  1727. DIFFS_NUMBERS base[12] = { 1111, -2222, 33, 100, 100, 15555, -1, 19999, 888, 755, -1, -730 };
  1728. DIFFS_NUMBERS high[3] = { 365, -123, 0 };
  1729. double prob = ks_2samp(base, bs, high, hs, 2);
  1730. return double_expect(prob, "0.777778", "12x3");
  1731. }
  1732. int mc_unittest(void) {
  1733. int errors = 0;
  1734. errors += mc_unittest1();
  1735. errors += mc_unittest2();
  1736. errors += mc_unittest3();
  1737. errors += mc_unittest4();
  1738. return errors;
  1739. }