| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "rrddim-collection.h" |
| 4 | |
| 5 | ALWAYS_INLINE void store_metric_collection_completed() { |
| 6 | pulse_queries_rrdset_collection_completed(rrdset_done_statistics_points_stored_per_tier); |
| 7 | } |
| 8 | |
| 9 | static inline time_t tier_next_point_time_s(RRDDIM *rd, struct rrddim_tier *t, time_t now_s) { |
| 10 | time_t loop = (time_t)rd->rrdset->update_every * (time_t)t->tier_grouping; |
| 11 | return now_s + loop - ((now_s + loop) % loop); |
| 12 | } |
| 13 | |
| 14 | #define LAST_COMPLETED_POINT_EXISTS(t) (t->last_completed_point.end_time_s != 0) |
| 15 | |
| 16 | ALWAYS_INLINE_HOT |
| 17 | void store_metric_at_tier_flush_last_completed(RRDDIM *rd __maybe_unused, size_t tier, struct rrddim_tier *t) { |
| 18 | // when there is no end_time_s we do not have a saved last_completed_point |
| 19 | if(!LAST_COMPLETED_POINT_EXISTS(t)) return; |
| 20 | |
| 21 | STORAGE_POINT *sp = &t->last_completed_point; |
| 22 | if(likely(!storage_point_is_unset(t->last_completed_point))) { |
| 23 | storage_engine_store_metric( |
| 24 | t->sch, |
| 25 | sp->end_time_s * USEC_PER_SEC, |
| 26 | sp->sum, |
| 27 | sp->min, |
| 28 | sp->max, |
| 29 | sp->count, |
| 30 | sp->anomaly_count, |
| 31 | sp->flags); |
| 32 | } |
| 33 | else { |
| 34 | storage_engine_store_metric( |
| 35 | t->sch, |
| 36 | sp->end_time_s * USEC_PER_SEC, |
| 37 | NAN, |
| 38 | NAN, |
| 39 | NAN, |
| 40 | 0, |
| 41 | 0, SN_FLAG_NONE); |
| 42 | } |
| 43 | |
| 44 | rrdset_done_statistics_points_stored_per_tier[tier]++; |
| 45 | |
| 46 | // make the point unset |
| 47 | t->last_completed_point.count = 0; // make it unset |
| 48 | t->last_completed_point.end_time_s = 0; // make it not saved |
| 49 | } |
| 50 | |
| 51 | ALWAYS_INLINE_HOT |
| 52 | static void store_metric_at_tier_save_last_completed(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAGE_POINT sp) { |
| 53 | // make sure the last_completed_point is empty |
| 54 | store_metric_at_tier_flush_last_completed(rd, tier, t); |
| 55 | |
| 56 | // copy the point |
| 57 | t->last_completed_point = sp; |
| 58 | |
| 59 | // set the end_time_s, so that we will know we have saved a last_completed_point |
| 60 | t->last_completed_point.end_time_s = t->next_point_end_time_s; |
| 61 | } |
| 62 | |
| 63 | ALWAYS_INLINE_HOT |
| 64 | void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut __maybe_unused) { |
| 65 | if(LAST_COMPLETED_POINT_EXISTS(t) && sp.start_time_s % t->last_completed_point_flush_modulo == 0) |
| 66 | store_metric_at_tier_flush_last_completed(rd, tier, t); |
| 67 | |
| 68 | if (unlikely(!t->next_point_end_time_s)) |
| 69 | t->next_point_end_time_s = tier_next_point_time_s(rd, t, sp.end_time_s); |
| 70 | |
| 71 | if(unlikely(sp.start_time_s >= t->next_point_end_time_s)) { |
| 72 | // flush the virtual point, it is done |
| 73 | |
| 74 | if (likely(!storage_point_is_unset(t->virtual_point))) |
| 75 | store_metric_at_tier_save_last_completed(rd, tier, t, t->virtual_point); |
| 76 | else |
| 77 | store_metric_at_tier_save_last_completed(rd, tier, t, STORAGE_POINT_UNSET); |
| 78 | |
| 79 | t->virtual_point.count = 0; // make the point unset |
| 80 | t->next_point_end_time_s = tier_next_point_time_s(rd, t, sp.end_time_s); |
| 81 | } |
| 82 | |
| 83 | // merge the dates into our virtual point |
| 84 | if (unlikely(sp.start_time_s < t->virtual_point.start_time_s)) |
| 85 | t->virtual_point.start_time_s = sp.start_time_s; |
| 86 | |
| 87 | if (likely(sp.end_time_s > t->virtual_point.end_time_s)) |
| 88 | t->virtual_point.end_time_s = sp.end_time_s; |
| 89 | |
| 90 | // merge the values into our virtual point |
| 91 | if (likely(!storage_point_is_gap(sp))) { |
| 92 | // we aggregate only non NULLs into higher tiers |
| 93 | |
| 94 | if (likely(!storage_point_is_unset(t->virtual_point))) { |
| 95 | // merge the collected point to our virtual one |
| 96 | t->virtual_point.sum += sp.sum; |
| 97 | t->virtual_point.min = MIN(t->virtual_point.min, sp.min); |
| 98 | t->virtual_point.max = MAX(t->virtual_point.max, sp.max); |
| 99 | t->virtual_point.count += sp.count; |
| 100 | t->virtual_point.anomaly_count += sp.anomaly_count; |
| 101 | t->virtual_point.flags |= sp.flags; |
| 102 | } |
| 103 | else { |
| 104 | // reset our virtual point to this one |
| 105 | t->virtual_point = sp; |
| 106 | } |
| 107 | } |
| 108 | } |
| 109 | |
| 110 | NOT_INLINE_HOT |
| 111 | #ifdef NETDATA_LOG_COLLECTION_ERRORS |
| 112 | void rrddim_store_metric_with_trace(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags, const char *function) { |
| 113 | #else // !NETDATA_LOG_COLLECTION_ERRORS |
| 114 | void rrddim_store_metric(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags) { |
| 115 | #endif // !NETDATA_LOG_COLLECTION_ERRORS |
| 116 | |
| 117 | static __thread struct log_stack_entry lgs[] = { |
| 118 | [0] = ND_LOG_FIELD_STR(NDF_NIDL_DIMENSION, NULL), |
| 119 | [1] = ND_LOG_FIELD_END(), |
| 120 | }; |
| 121 | lgs[0].str = rd->id; |
| 122 | log_stack_push(lgs); |
| 123 | |
| 124 | #ifdef NETDATA_LOG_COLLECTION_ERRORS |
| 125 | rd->rrddim_store_metric_count++; |
| 126 | |
| 127 | if(likely(rd->rrddim_store_metric_count > 1)) { |
| 128 | usec_t expected = rd->rrddim_store_metric_last_ut + rd->update_every * USEC_PER_SEC; |
| 129 | |
| 130 | if(point_end_time_ut != rd->rrddim_store_metric_last_ut) { |
| 131 | internal_error(true, |
| 132 | "%s COLLECTION: 'host:%s/chart:%s/dim:%s' granularity %d, collection %zu, expected to store at tier 0 a value at %llu, but it gave %llu [%s%llu usec] (called from %s(), previously by %s())", |
| 133 | (point_end_time_ut < rd->rrddim_store_metric_last_ut) ? "**PAST**" : "GAP", |
| 134 | rrdhost_hostname(rd->rrdset->rrdhost), rrdset_id(rd->rrdset), rrddim_id(rd), |
| 135 | rd->update_every, |
| 136 | rd->rrddim_store_metric_count, |
| 137 | expected, point_end_time_ut, |
| 138 | (point_end_time_ut < rd->rrddim_store_metric_last_ut)?"by -" : "gap ", |
| 139 | expected - point_end_time_ut, |
| 140 | function, |
| 141 | rd->rrddim_store_metric_last_caller?rd->rrddim_store_metric_last_caller:"none"); |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | rd->rrddim_store_metric_last_ut = point_end_time_ut; |
| 146 | rd->rrddim_store_metric_last_caller = function; |
| 147 | #endif // NETDATA_LOG_COLLECTION_ERRORS |
| 148 | |
| 149 | // store the metric on tier 0 |
| 150 | storage_engine_store_metric(rd->tiers[0].sch, point_end_time_ut, |
| 151 | n, 0, 0, |
| 152 | 1, 0, flags); |
| 153 | |
| 154 | rrdset_done_statistics_points_stored_per_tier[0]++; |
| 155 | |
| 156 | time_t now_s = (time_t)(point_end_time_ut / USEC_PER_SEC); |
| 157 | |
| 158 | STORAGE_POINT sp = { |
| 159 | .start_time_s = now_s - rd->rrdset->update_every, |
| 160 | .end_time_s = now_s, |
| 161 | .min = n, |
| 162 | .max = n, |
| 163 | .sum = n, |
| 164 | .count = 1, |
| 165 | .anomaly_count = (flags & SN_FLAG_NOT_ANOMALOUS) ? 0 : 1, |
| 166 | .flags = flags |
| 167 | }; |
| 168 | |
| 169 | for(size_t tier = 1; tier < nd_profile.storage_tiers;tier++) { |
| 170 | if(unlikely(!rd->tiers[tier].smh)) continue; |
| 171 | |
| 172 | struct rrddim_tier *t = &rd->tiers[tier]; |
| 173 | |
| 174 | if(!rrddim_option_check(rd, RRDDIM_OPTION_BACKFILLED_HIGH_TIERS)) { |
| 175 | // we have not collected this tier before |
| 176 | // let's fill any gap that may exist |
| 177 | backfill_tier_from_smaller_tiers(rd, tier, now_s); |
| 178 | } |
| 179 | |
| 180 | store_metric_at_tier(rd, tier, t, sp, point_end_time_ut); |
| 181 | } |
| 182 | rrddim_option_set(rd, RRDDIM_OPTION_BACKFILLED_HIGH_TIERS); |
| 183 | |
| 184 | rrdcontext_collected_rrddim(rd); |
| 185 | log_stack_pop(&lgs); |
| 186 | } |