| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "query-internal.h" |
| 4 | |
| 5 | // ---------------------------------------------------------------------------- |
| 6 | // helpers to find our way in RRDR |
| 7 | |
| 8 | ALWAYS_INLINE |
| 9 | static RRDR_VALUE_FLAGS *UNUSED_FUNCTION(rrdr_line_options)(RRDR *r, long rrdr_line) { |
| 10 | return &r->o[ rrdr_line * r->d ]; |
| 11 | } |
| 12 | |
| 13 | ALWAYS_INLINE |
| 14 | static NETDATA_DOUBLE *UNUSED_FUNCTION(rrdr_line_values)(RRDR *r, long rrdr_line) { |
| 15 | return &r->v[ rrdr_line * r->d ]; |
| 16 | } |
| 17 | |
| 18 | ALWAYS_INLINE |
| 19 | static long rrdr_line_init(RRDR *r __maybe_unused, time_t t __maybe_unused, long rrdr_line) { |
| 20 | rrdr_line++; |
| 21 | |
| 22 | internal_fatal(rrdr_line >= (long)r->n, |
| 23 | "QUERY: requested to step above RRDR size for query '%s'", |
| 24 | r->internal.qt->id); |
| 25 | |
| 26 | internal_fatal(r->t[rrdr_line] != t, |
| 27 | "QUERY: wrong timestamp at RRDR line %ld, expected %ld, got %ld, of query '%s'", |
| 28 | rrdr_line, r->t[rrdr_line], t, r->internal.qt->id); |
| 29 | |
| 30 | return rrdr_line; |
| 31 | } |
| 32 | |
| 33 | ALWAYS_INLINE |
| 34 | static NETDATA_DOUBLE query_point_grouping_value( |
| 35 | QUERY_POINT point, QUERY_ENGINE_OPS *ops, RRDR_TIME_GROUPING add_flush) { |
| 36 | if(likely(add_flush != RRDR_GROUPING_SUM || !ops->qm->values_stored_as_rates)) |
| 37 | return point.value; |
| 38 | |
| 39 | if(unlikely(storage_point_is_unset(point.sp) || storage_point_is_gap(point.sp) || !point.sp.count)) |
| 40 | return point.value; |
| 41 | |
| 42 | time_t duration = point.sp.end_time_s - point.sp.start_time_s; |
| 43 | if(unlikely(duration <= 0)) |
| 44 | return point.value; |
| 45 | |
| 46 | return point.value * (NETDATA_DOUBLE)duration / (NETDATA_DOUBLE)point.sp.count; |
| 47 | } |
| 48 | |
| 49 | // ---------------------------------------------------------------------------- |
| 50 | // dimension level query engine |
| 51 | |
| 52 | #define query_interpolate_point(this_point, last_point, now) do { \ |
| 53 | if(likely( \ |
| 54 | /* the point to interpolate is more than 1s wide */ \ |
| 55 | (this_point).sp.end_time_s - (this_point).sp.start_time_s > 1 \ |
| 56 | \ |
| 57 | /* the two points are exactly next to each other */ \ |
| 58 | && (last_point).sp.end_time_s == (this_point).sp.start_time_s \ |
| 59 | \ |
| 60 | /* both points are valid numbers */ \ |
| 61 | && netdata_double_isnumber((this_point).value) \ |
| 62 | && netdata_double_isnumber((last_point).value) \ |
| 63 | \ |
| 64 | )) { \ |
| 65 | (this_point).value = (last_point).value + ((this_point).value - (last_point).value) * (1.0 - (NETDATA_DOUBLE)((this_point).sp.end_time_s - (now)) / (NETDATA_DOUBLE)((this_point).sp.end_time_s - (this_point).sp.start_time_s)); \ |
| 66 | (this_point).sp.end_time_s = now; \ |
| 67 | } \ |
| 68 | } while(0) |
| 69 | |
| 70 | #define query_add_point_to_group(r, point, ops, add_flush) do { \ |
| 71 | if(likely(netdata_double_isnumber((point).value))) { \ |
| 72 | if(likely(fpclassify((point).value) != FP_ZERO)) \ |
| 73 | (ops)->group_points_non_zero++; \ |
| 74 | \ |
| 75 | if(unlikely((point).sp.flags & SN_FLAG_RESET)) \ |
| 76 | (ops)->group_value_flags |= RRDR_VALUE_RESET; \ |
| 77 | \ |
| 78 | NETDATA_DOUBLE grouping_value = \ |
| 79 | query_point_grouping_value(point, ops, add_flush); \ |
| 80 | time_grouping_add(r, grouping_value, add_flush); \ |
| 81 | \ |
| 82 | storage_point_merge_to((ops)->group_point, (point).sp); \ |
| 83 | if(!(point).added) \ |
| 84 | storage_point_merge_to((ops)->query_point, (point).sp); \ |
| 85 | } \ |
| 86 | \ |
| 87 | (ops)->group_points_added++; \ |
| 88 | } while(0) |
| 89 | |
| 90 | NOT_INLINE_HOT void rrd2rrdr_query_execute(RRDR *r, size_t dim_id_in_rrdr, QUERY_ENGINE_OPS *ops) { |
| 91 | QUERY_TARGET *qt = r->internal.qt; |
| 92 | QUERY_METRIC *qm = ops->qm; |
| 93 | |
| 94 | const RRDR_TIME_GROUPING add_flush = r->time_grouping.add_flush; |
| 95 | |
| 96 | ops->group_point = STORAGE_POINT_UNSET; |
| 97 | ops->query_point = STORAGE_POINT_UNSET; |
| 98 | |
| 99 | RRDR_OPTIONS options = qt->window.options; |
| 100 | size_t points_wanted = qt->window.points; |
| 101 | time_t after_wanted = qt->window.after; |
| 102 | time_t before_wanted = qt->window.before; (void)before_wanted; |
| 103 | |
| 104 | // bool debug_this = false; |
| 105 | // if(strcmp("user", string2str(rd->id)) == 0 && strcmp("system.cpu", string2str(rd->rrdset->id)) == 0) |
| 106 | // debug_this = true; |
| 107 | |
| 108 | size_t points_added = 0; |
| 109 | |
| 110 | long rrdr_line = -1; |
| 111 | bool use_anomaly_bit_as_value = (r->internal.qt->window.options & RRDR_OPTION_ANOMALY_BIT) ? true : false; |
| 112 | |
| 113 | NETDATA_DOUBLE min = r->view.min, max = r->view.max; |
| 114 | |
| 115 | QUERY_POINT last2_point = QUERY_POINT_EMPTY; |
| 116 | QUERY_POINT last1_point = QUERY_POINT_EMPTY; |
| 117 | QUERY_POINT new_point = QUERY_POINT_EMPTY; |
| 118 | |
| 119 | // ONE POINT READ-AHEAD |
| 120 | // when we switch plans, we read-ahead a point from the next plan |
| 121 | // to join them smoothly at the exact time the next plan begins |
| 122 | STORAGE_POINT next1_point = STORAGE_POINT_UNSET; |
| 123 | |
| 124 | time_t now_start_time = after_wanted - ops->query_granularity; |
| 125 | time_t now_end_time = after_wanted + ops->view_update_every - ops->query_granularity; |
| 126 | |
| 127 | size_t db_points_read_since_plan_switch = 0; (void)db_points_read_since_plan_switch; |
| 128 | size_t query_is_finished_counter = 0; |
| 129 | |
| 130 | // The main loop, based on the query granularity we need |
| 131 | for( ; points_added < points_wanted && query_is_finished_counter <= 10 ; |
| 132 | now_start_time = now_end_time, now_end_time += ops->view_update_every) { |
| 133 | |
| 134 | if(unlikely(query_plan_should_switch_plan(ops, now_end_time))) { |
| 135 | query_planer_next_plan(ops, now_end_time, new_point.sp.end_time_s); |
| 136 | db_points_read_since_plan_switch = 0; |
| 137 | } |
| 138 | |
| 139 | // read all the points of the db, prior to the time we need (now_end_time) |
| 140 | |
| 141 | size_t count_same_end_time = 0; |
| 142 | while(count_same_end_time < 100) { |
| 143 | if(likely(count_same_end_time == 0)) { |
| 144 | last2_point = last1_point; |
| 145 | last1_point = new_point; |
| 146 | } |
| 147 | |
| 148 | if(unlikely(storage_engine_query_is_finished(ops->seqh))) { |
| 149 | query_is_finished_counter++; |
| 150 | |
| 151 | if(count_same_end_time != 0) { |
| 152 | last2_point = last1_point; |
| 153 | last1_point = new_point; |
| 154 | } |
| 155 | new_point = QUERY_POINT_EMPTY; |
| 156 | new_point.sp.start_time_s = last1_point.sp.end_time_s; |
| 157 | new_point.sp.end_time_s = now_end_time; |
| 158 | // |
| 159 | // if(debug_this) netdata_log_info("QUERY: is finished() returned true"); |
| 160 | // |
| 161 | break; |
| 162 | } |
| 163 | else |
| 164 | query_is_finished_counter = 0; |
| 165 | |
| 166 | // fetch the new point |
| 167 | { |
| 168 | STORAGE_POINT sp; |
| 169 | if(likely(storage_point_is_unset(next1_point))) { |
| 170 | db_points_read_since_plan_switch++; |
| 171 | sp = storage_engine_query_next_metric(ops->seqh); |
| 172 | ops->db_points_read_per_tier[ops->tier]++; |
| 173 | ops->db_total_points_read++; |
| 174 | |
| 175 | if(unlikely(options & RRDR_OPTION_ABSOLUTE)) |
| 176 | storage_point_make_positive(sp); |
| 177 | } |
| 178 | else { |
| 179 | // ONE POINT READ-AHEAD |
| 180 | sp = next1_point; |
| 181 | storage_point_unset(next1_point); |
| 182 | db_points_read_since_plan_switch = 1; |
| 183 | } |
| 184 | |
| 185 | // ONE POINT READ-AHEAD |
| 186 | if(unlikely(query_plan_should_switch_plan(ops, sp.end_time_s) && |
| 187 | query_planer_next_plan(ops, now_end_time, new_point.sp.end_time_s))) { |
| 188 | |
| 189 | // The end time of the current point, crosses our plans (tiers) |
| 190 | // so, we switched plan (tier) |
| 191 | // |
| 192 | // There are 2 cases now: |
| 193 | // |
| 194 | // A. the entire point of the previous plan is to the future of point from the next plan |
| 195 | // B. part of the point of the previous plan overlaps with the point from the next plan |
| 196 | |
| 197 | STORAGE_POINT sp2 = storage_engine_query_next_metric(ops->seqh); |
| 198 | ops->db_points_read_per_tier[ops->tier]++; |
| 199 | ops->db_total_points_read++; |
| 200 | |
| 201 | if(unlikely(options & RRDR_OPTION_ABSOLUTE)) |
| 202 | storage_point_make_positive(sp); |
| 203 | |
| 204 | if(sp.start_time_s > sp2.start_time_s) |
| 205 | // the point from the previous plan is useless |
| 206 | sp = sp2; |
| 207 | else |
| 208 | // let the query run from the previous plan |
| 209 | // but setting this will also cut off the interpolation |
| 210 | // of the point from the previous plan |
| 211 | next1_point = sp2; |
| 212 | } |
| 213 | |
| 214 | new_point.sp = sp; |
| 215 | new_point.added = false; |
| 216 | query_point_set_id(new_point, ops->db_total_points_read); |
| 217 | |
| 218 | // if(debug_this) |
| 219 | // netdata_log_info("QUERY: got point %zu, from time %ld to %ld // now from %ld to %ld // query from %ld to %ld", |
| 220 | // new_point.id, new_point.start_time, new_point.end_time, now_start_time, now_end_time, after_wanted, before_wanted); |
| 221 | // |
| 222 | // get the right value from the point we got |
| 223 | if(likely(!storage_point_is_unset(sp) && !storage_point_is_gap(sp))) { |
| 224 | |
| 225 | if(unlikely(use_anomaly_bit_as_value)) |
| 226 | new_point.value = storage_point_anomaly_rate(new_point.sp); |
| 227 | |
| 228 | else { |
| 229 | switch (ops->tier_query_fetch) { |
| 230 | default: |
| 231 | case TIER_QUERY_FETCH_AVERAGE: |
| 232 | new_point.value = sp.sum / (NETDATA_DOUBLE)sp.count; |
| 233 | break; |
| 234 | |
| 235 | case TIER_QUERY_FETCH_MIN: |
| 236 | new_point.value = sp.min; |
| 237 | break; |
| 238 | |
| 239 | case TIER_QUERY_FETCH_MAX: |
| 240 | new_point.value = sp.max; |
| 241 | break; |
| 242 | |
| 243 | case TIER_QUERY_FETCH_SUM: |
| 244 | new_point.value = sp.sum; |
| 245 | break; |
| 246 | } |
| 247 | } |
| 248 | } |
| 249 | else |
| 250 | new_point.value = NAN; |
| 251 | } |
| 252 | |
| 253 | // check if the db is giving us zero duration points |
| 254 | if(unlikely(db_points_read_since_plan_switch > 1 && |
| 255 | new_point.sp.start_time_s == new_point.sp.end_time_s)) { |
| 256 | |
| 257 | internal_error(true, "QUERY: '%s', dimension '%s' next_metric() returned " |
| 258 | "point %zu from %ld to %ld, that are both equal", |
| 259 | qt->id, query_metric_id(qt, qm), |
| 260 | new_point.id, new_point.sp.start_time_s, new_point.sp.end_time_s); |
| 261 | |
| 262 | new_point.sp.start_time_s = new_point.sp.end_time_s - ops->tier_ptr->db_update_every_s; |
| 263 | } |
| 264 | |
| 265 | // check if the db is advancing the query |
| 266 | if(unlikely(db_points_read_since_plan_switch > 1 && |
| 267 | new_point.sp.end_time_s <= last1_point.sp.end_time_s)) { |
| 268 | |
| 269 | internal_error(true, |
| 270 | "QUERY: '%s', dimension '%s' next_metric() returned " |
| 271 | "point %zu from %ld to %ld, before the " |
| 272 | "last point %zu from %ld to %ld, " |
| 273 | "now is %ld to %ld", |
| 274 | qt->id, query_metric_id(qt, qm), |
| 275 | new_point.id, new_point.sp.start_time_s, new_point.sp.end_time_s, |
| 276 | last1_point.id, last1_point.sp.start_time_s, last1_point.sp.end_time_s, |
| 277 | now_start_time, now_end_time); |
| 278 | |
| 279 | count_same_end_time++; |
| 280 | continue; |
| 281 | } |
| 282 | count_same_end_time = 0; |
| 283 | |
| 284 | // decide how to use this point |
| 285 | if(likely(new_point.sp.end_time_s < now_end_time)) { // likely to favor tier0 |
| 286 | // this db point ends before our now_end_time |
| 287 | |
| 288 | if(likely(new_point.sp.end_time_s >= now_start_time)) { // likely to favor tier0 |
| 289 | // this db point ends after our now_start time |
| 290 | |
| 291 | query_add_point_to_group(r, new_point, ops, add_flush); |
| 292 | new_point.added = true; |
| 293 | } |
| 294 | else { |
| 295 | // we don't need this db point |
| 296 | // it is totally outside our current time-frame |
| 297 | |
| 298 | // this is desirable for the first point of the query |
| 299 | // because it allows us to interpolate the next point |
| 300 | // at exactly the time we will want |
| 301 | |
| 302 | // we only log if this is not point 1 |
| 303 | internal_error(new_point.sp.end_time_s < ops->plan_expanded_after && |
| 304 | db_points_read_since_plan_switch > 1, |
| 305 | "QUERY: '%s', dimension '%s' next_metric() " |
| 306 | "returned point %zu from %ld time %ld, " |
| 307 | "which is entirely before our current timeframe %ld to %ld " |
| 308 | "(and before the entire query, after %ld, before %ld)", |
| 309 | qt->id, query_metric_id(qt, qm), |
| 310 | new_point.id, new_point.sp.start_time_s, new_point.sp.end_time_s, |
| 311 | now_start_time, now_end_time, |
| 312 | ops->plan_expanded_after, ops->plan_expanded_before); |
| 313 | } |
| 314 | |
| 315 | } |
| 316 | else { |
| 317 | // the point ends in the future |
| 318 | // so, we will interpolate it below, at the inner loop |
| 319 | break; |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | if(unlikely(count_same_end_time)) { |
| 324 | internal_error(true, |
| 325 | "QUERY: '%s', dimension '%s', the database does not advance the query," |
| 326 | " it returned an end time less or equal to the end time of the last " |
| 327 | "point we got %ld, %zu times", |
| 328 | qt->id, query_metric_id(qt, qm), |
| 329 | last1_point.sp.end_time_s, count_same_end_time); |
| 330 | |
| 331 | if(unlikely(new_point.sp.end_time_s <= last1_point.sp.end_time_s)) |
| 332 | new_point.sp.end_time_s = now_end_time; |
| 333 | } |
| 334 | |
| 335 | time_t stop_time = new_point.sp.end_time_s; |
| 336 | if(unlikely(!storage_point_is_unset(next1_point) && next1_point.start_time_s >= now_end_time)) { |
| 337 | // ONE POINT READ-AHEAD |
| 338 | // the point crosses the start time of the |
| 339 | // read ahead storage point we have read |
| 340 | stop_time = next1_point.start_time_s; |
| 341 | } |
| 342 | |
| 343 | // the inner loop |
| 344 | // we have 3 points in memory: last2, last1, new |
| 345 | // we select the one to use based on their timestamps |
| 346 | |
| 347 | internal_fatal(now_end_time > stop_time || points_added >= points_wanted, |
| 348 | "QUERY: first part of query provides invalid point to interpolate (now_end_time %ld, stop_time %ld", |
| 349 | now_end_time, stop_time); |
| 350 | |
| 351 | do { |
| 352 | // now_start_time is wrong in this loop |
| 353 | // but, we don't need it |
| 354 | |
| 355 | QUERY_POINT current_point; |
| 356 | |
| 357 | if(likely(now_end_time > new_point.sp.start_time_s)) { |
| 358 | // it is time for our NEW point to be used |
| 359 | current_point = new_point; |
| 360 | new_point.added = true; // first copy, then set it, so that new_point will not be added again |
| 361 | query_interpolate_point(current_point, last1_point, now_end_time); |
| 362 | |
| 363 | // internal_error(current_point.id > 0 |
| 364 | // && last1_point.id == 0 |
| 365 | // && current_point.end_time > after_wanted |
| 366 | // && current_point.end_time > now_end_time, |
| 367 | // "QUERY: '%s', dimension '%s', after %ld, before %ld, view update every %ld," |
| 368 | // " query granularity %ld, interpolating point %zu (from %ld to %ld) at %ld," |
| 369 | // " but we could really favor by having last_point1 in this query.", |
| 370 | // qt->id, string2str(qm->dimension.id), |
| 371 | // after_wanted, before_wanted, |
| 372 | // ops.view_update_every, ops.query_granularity, |
| 373 | // current_point.id, current_point.start_time, current_point.end_time, |
| 374 | // now_end_time); |
| 375 | } |
| 376 | else if(likely(now_end_time <= last1_point.sp.end_time_s)) { |
| 377 | // our LAST point is still valid |
| 378 | current_point = last1_point; |
| 379 | last1_point.added = true; // first copy, then set it, so that last1_point will not be added again |
| 380 | query_interpolate_point(current_point, last2_point, now_end_time); |
| 381 | |
| 382 | // internal_error(current_point.id > 0 |
| 383 | // && last2_point.id == 0 |
| 384 | // && current_point.end_time > after_wanted |
| 385 | // && current_point.end_time > now_end_time, |
| 386 | // "QUERY: '%s', dimension '%s', after %ld, before %ld, view update every %ld," |
| 387 | // " query granularity %ld, interpolating point %zu (from %ld to %ld) at %ld," |
| 388 | // " but we could really favor by having last_point2 in this query.", |
| 389 | // qt->id, string2str(qm->dimension.id), |
| 390 | // after_wanted, before_wanted, ops.view_update_every, ops.query_granularity, |
| 391 | // current_point.id, current_point.start_time, current_point.end_time, |
| 392 | // now_end_time); |
| 393 | } |
| 394 | else { |
| 395 | // a GAP, we don't have a value this time |
| 396 | current_point = QUERY_POINT_EMPTY; |
| 397 | } |
| 398 | |
| 399 | query_add_point_to_group(r, current_point, ops, add_flush); |
| 400 | |
| 401 | rrdr_line = rrdr_line_init(r, now_end_time, rrdr_line); |
| 402 | size_t rrdr_o_v_index = rrdr_line * r->d + dim_id_in_rrdr; |
| 403 | |
| 404 | // find the place to store our values |
| 405 | RRDR_VALUE_FLAGS *rrdr_value_options_ptr = &r->o[rrdr_o_v_index]; |
| 406 | |
| 407 | // update the dimension options |
| 408 | if(likely(ops->group_points_non_zero)) |
| 409 | r->od[dim_id_in_rrdr] |= RRDR_DIMENSION_NONZERO; |
| 410 | |
| 411 | // store the specific point options |
| 412 | *rrdr_value_options_ptr = ops->group_value_flags; |
| 413 | |
| 414 | // store the group value |
| 415 | NETDATA_DOUBLE group_value = time_grouping_flush(r, rrdr_value_options_ptr, add_flush); |
| 416 | r->v[rrdr_o_v_index] = group_value; |
| 417 | |
| 418 | r->ar[rrdr_o_v_index] = storage_point_anomaly_rate(ops->group_point); |
| 419 | |
| 420 | if(likely(points_added || r->internal.queries_count)) { |
| 421 | // find the min/max across all dimensions |
| 422 | |
| 423 | if(unlikely(group_value < min)) min = group_value; |
| 424 | if(unlikely(group_value > max)) max = group_value; |
| 425 | |
| 426 | } |
| 427 | else { |
| 428 | // runs only when r->internal.queries_count == 0 && points_added == 0 |
| 429 | // so, on the first point added for the query. |
| 430 | min = max = group_value; |
| 431 | } |
| 432 | |
| 433 | points_added++; |
| 434 | ops->group_points_added = 0; |
| 435 | ops->group_value_flags = RRDR_VALUE_NOTHING; |
| 436 | ops->group_points_non_zero = 0; |
| 437 | ops->group_point = STORAGE_POINT_UNSET; |
| 438 | |
| 439 | now_end_time += ops->view_update_every; |
| 440 | } while(now_end_time <= stop_time && points_added < points_wanted); |
| 441 | |
| 442 | // the loop above increased "now" by ops->view_update_every, |
| 443 | // but the main loop will increase it too, |
| 444 | // so, let's undo the last iteration of this loop |
| 445 | now_end_time -= ops->view_update_every; |
| 446 | } |
| 447 | query_planer_finalize_remaining_plans(ops); |
| 448 | |
| 449 | qm->query_points = ops->query_point; |
| 450 | |
| 451 | // fill the rest of the points with empty values |
| 452 | while (points_added < points_wanted) { |
| 453 | rrdr_line++; |
| 454 | size_t rrdr_o_v_index = rrdr_line * r->d + dim_id_in_rrdr; |
| 455 | r->o[rrdr_o_v_index] = RRDR_VALUE_EMPTY; |
| 456 | r->v[rrdr_o_v_index] = 0.0; |
| 457 | r->ar[rrdr_o_v_index] = 0.0; |
| 458 | points_added++; |
| 459 | } |
| 460 | |
| 461 | r->internal.queries_count++; |
| 462 | r->view.min = min; |
| 463 | r->view.max = max; |
| 464 | |
| 465 | r->stats.result_points_generated += points_added; |
| 466 | r->stats.db_points_read += ops->db_total_points_read; |
| 467 | for(size_t tr = 0; tr < nd_profile.storage_tiers; tr++) |
| 468 | qt->db.tiers[tr].points += ops->db_points_read_per_tier[tr]; |
| 469 | } |