| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "stream-replication-receiver.h" |
| 4 | #include "stream-receiver-internals.h" |
| 5 | |
| 6 | struct replication_request_details { |
| 7 | struct { |
| 8 | send_command callback; |
| 9 | struct parser *parser; |
| 10 | } caller; |
| 11 | |
| 12 | RRDHOST *host; |
| 13 | RRDSET *st; |
| 14 | |
| 15 | struct { |
| 16 | time_t first_entry_t; // the first entry time the child has |
| 17 | time_t last_entry_t; // the last entry time the child has |
| 18 | time_t wall_clock_time; // the current time of the child |
| 19 | bool fixed_last_entry; // when set we set the last entry to wall clock time |
| 20 | } child_db; |
| 21 | |
| 22 | struct { |
| 23 | time_t first_entry_t; // the first entry time we have |
| 24 | time_t last_entry_t; // the last entry time we have |
| 25 | time_t wall_clock_time; // the current local world clock time |
| 26 | } local_db; |
| 27 | |
| 28 | struct { |
| 29 | time_t from; // the starting time of the entire gap we have |
| 30 | time_t to; // the ending time of the entire gap we have |
| 31 | } gap; |
| 32 | |
| 33 | struct { |
| 34 | time_t after; // the start time we requested previously from this child |
| 35 | time_t before; // the end time we requested previously from this child |
| 36 | } last_request; |
| 37 | |
| 38 | struct { |
| 39 | time_t after; // the start time of this replication request - the child will add 1 second |
| 40 | time_t before; // the end time of this replication request |
| 41 | bool start_streaming; // true when we want the child to send anything remaining and start streaming - the child will overwrite 'before' |
| 42 | } wanted; |
| 43 | }; |
| 44 | |
| 45 | static void replicate_log_request(struct replication_request_details *r, const char *msg) { |
| 46 | #ifdef NETDATA_INTERNAL_CHECKS |
| 47 | internal_error(true, |
| 48 | #else |
| 49 | nd_log_limit_static_global_var(erl, 1, 0); |
| 50 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_NOTICE, |
| 51 | #endif |
| 52 | "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' child sent: " |
| 53 | "db from %ld to %ld%s, wall clock time %ld, " |
| 54 | "last request from %ld to %ld, " |
| 55 | "issue: %s - " |
| 56 | "sending replication request from %ld to %ld, start streaming %s", |
| 57 | rrdhost_hostname(r->st->rrdhost), rrdset_id(r->st), |
| 58 | r->child_db.first_entry_t, |
| 59 | r->child_db.last_entry_t, r->child_db.fixed_last_entry ? " (fixed)" : "", |
| 60 | r->child_db.wall_clock_time, |
| 61 | r->last_request.after, |
| 62 | r->last_request.before, |
| 63 | msg, |
| 64 | r->wanted.after, |
| 65 | r->wanted.before, |
| 66 | r->wanted.start_streaming ? "true" : "false"); |
| 67 | } |
| 68 | |
| 69 | static bool send_replay_chart_cmd(struct replication_request_details *r, const char *msg, bool log) { |
| 70 | RRDSET *st = r->st; |
| 71 | |
| 72 | if(log) |
| 73 | replicate_log_request(r, msg); |
| 74 | |
| 75 | if(st->rrdhost->receiver && (!st->rrdhost->receiver->replication.first_time_s || r->wanted.after < st->rrdhost->receiver->replication.first_time_s)) |
| 76 | st->rrdhost->receiver->replication.first_time_s = r->wanted.after; |
| 77 | |
| 78 | #ifdef NETDATA_LOG_REPLICATION_REQUESTS |
| 79 | st->replay.log_next_data_collection = true; |
| 80 | |
| 81 | char wanted_after_buf[LOG_DATE_LENGTH + 1] = "", wanted_before_buf[LOG_DATE_LENGTH + 1] = ""; |
| 82 | |
| 83 | if(r->wanted.after) |
| 84 | log_date(wanted_after_buf, LOG_DATE_LENGTH, r->wanted.after); |
| 85 | |
| 86 | if(r->wanted.before) |
| 87 | log_date(wanted_before_buf, LOG_DATE_LENGTH, r->wanted.before); |
| 88 | |
| 89 | internal_error(true, |
| 90 | "STREAM SND REPLAY: 'host:%s/chart:%s' sending replication request %ld [%s] to %ld [%s], start streaming '%s': %s: " |
| 91 | "last[%ld - %ld] child[%ld - %ld, now %ld %s] local[%ld - %ld, now %ld] gap[%ld - %ld %s] %s" |
| 92 | , rrdhost_hostname(r->host), rrdset_id(r->st) |
| 93 | , r->wanted.after, wanted_after_buf |
| 94 | , r->wanted.before, wanted_before_buf |
| 95 | , r->wanted.start_streaming ? "YES" : "NO" |
| 96 | , msg |
| 97 | , r->last_request.after, r->last_request.before |
| 98 | , r->child_db.first_entry_t, r->child_db.last_entry_t |
| 99 | , r->child_db.wall_clock_time, (r->child_db.wall_clock_time == r->local_db.wall_clock_time) ? "SAME" : (r->child_db.wall_clock_time < r->local_db.wall_clock_time) ? "BEHIND" : "AHEAD" |
| 100 | , r->local_db.first_entry_t, r->local_db.last_entry_t |
| 101 | , r->local_db.wall_clock_time |
| 102 | , r->gap.from, r->gap.to |
| 103 | , (r->gap.from == r->wanted.after) ? "FULL" : "PARTIAL" |
| 104 | , (st->replay.after != 0 || st->replay.before != 0) ? "OVERLAPPING" : "" |
| 105 | ); |
| 106 | |
| 107 | st->replay.start_streaming = r->wanted.start_streaming; |
| 108 | st->replay.after = r->wanted.after; |
| 109 | st->replay.before = r->wanted.before; |
| 110 | #endif // NETDATA_LOG_REPLICATION_REQUESTS |
| 111 | |
| 112 | char buffer[2048 + 1]; |
| 113 | snprintfz(buffer, sizeof(buffer) - 1, PLUGINSD_KEYWORD_REPLAY_CHART " \"%s\" \"%s\" %llu %llu\n", |
| 114 | rrdset_id(st), r->wanted.start_streaming ? "true" : "false", |
| 115 | (unsigned long long)r->wanted.after, (unsigned long long)r->wanted.before); |
| 116 | |
| 117 | ssize_t ret = r->caller.callback(buffer, r->caller.parser, STREAM_TRAFFIC_TYPE_REPLICATION); |
| 118 | if (ret < 0) { |
| 119 | netdata_log_error("STREAM SND REPLAY ERROR: 'host:%s/chart:%s' failed to send replication request to child (error %zd)", |
| 120 | rrdhost_hostname(r->host), rrdset_id(r->st), ret); |
| 121 | return false; |
| 122 | } |
| 123 | |
| 124 | __atomic_add_fetch(&st->rrdhost->stream.rcv.status.replication.counter_out, 1, __ATOMIC_RELAXED); |
| 125 | |
| 126 | #ifdef REPLICATION_TRACKING |
| 127 | st->stream.rcv.who = REPLAY_WHO_THEM; |
| 128 | #endif |
| 129 | |
| 130 | return true; |
| 131 | } |
| 132 | |
| 133 | bool replicate_chart_request(send_command callback, struct parser *parser, RRDHOST *host, RRDSET *st, |
| 134 | time_t child_first_entry, time_t child_last_entry, time_t child_wall_clock_time, |
| 135 | time_t prev_first_entry_wanted, time_t prev_last_entry_wanted) |
| 136 | { |
| 137 | struct replication_request_details r = { |
| 138 | .caller = { |
| 139 | .callback = callback, |
| 140 | .parser = parser, |
| 141 | }, |
| 142 | |
| 143 | .host = host, |
| 144 | .st = st, |
| 145 | |
| 146 | .child_db = { |
| 147 | .first_entry_t = child_first_entry, |
| 148 | .last_entry_t = child_last_entry, |
| 149 | .wall_clock_time = child_wall_clock_time, |
| 150 | .fixed_last_entry = false, |
| 151 | }, |
| 152 | |
| 153 | .local_db = { |
| 154 | .first_entry_t = 0, |
| 155 | .last_entry_t = 0, |
| 156 | .wall_clock_time = now_realtime_sec(), |
| 157 | }, |
| 158 | |
| 159 | .last_request = { |
| 160 | .after = prev_first_entry_wanted, |
| 161 | .before = prev_last_entry_wanted, |
| 162 | }, |
| 163 | |
| 164 | .wanted = { |
| 165 | .after = 0, |
| 166 | .before = 0, |
| 167 | .start_streaming = true, |
| 168 | }, |
| 169 | }; |
| 170 | |
| 171 | if(r.child_db.last_entry_t > r.child_db.wall_clock_time) { |
| 172 | replicate_log_request(&r, "child's db last entry > child's wall clock time"); |
| 173 | r.child_db.last_entry_t = r.child_db.wall_clock_time; |
| 174 | r.child_db.fixed_last_entry = true; |
| 175 | } |
| 176 | |
| 177 | rrdset_get_retention_of_tier_for_collected_chart(r.st, &r.local_db.first_entry_t, &r.local_db.last_entry_t, r.local_db.wall_clock_time, 0); |
| 178 | |
| 179 | // let's find the GAP we have |
| 180 | if(!r.last_request.after || !r.last_request.before) { |
| 181 | // there is no previous request |
| 182 | |
| 183 | if(r.local_db.last_entry_t) |
| 184 | // we have some data, let's continue from the last point we have |
| 185 | r.gap.from = r.local_db.last_entry_t; |
| 186 | else |
| 187 | // we don't have any data, the gap is the max timeframe we are allowed to replicate |
| 188 | r.gap.from = r.local_db.wall_clock_time - r.host->stream.replication.period; |
| 189 | |
| 190 | } |
| 191 | else { |
| 192 | // we had sent a request - let's continue at the point we left it |
| 193 | // for this we don't take into account the actual data in our db |
| 194 | // because the child may also have gaps, and we need to get over it |
| 195 | r.gap.from = r.last_request.before; |
| 196 | } |
| 197 | |
| 198 | // we want all the data up to now |
| 199 | r.gap.to = r.local_db.wall_clock_time; |
| 200 | |
| 201 | // The gap is now r.gap.from -> r.gap.to |
| 202 | |
| 203 | if (unlikely(!rrdhost_option_check(host, RRDHOST_OPTION_REPLICATION))) |
| 204 | return send_replay_chart_cmd(&r, "sending empty replication request, replication is disabled", false); |
| 205 | |
| 206 | if (unlikely(!rrdset_number_of_dimensions(st))) |
| 207 | return send_replay_chart_cmd(&r, "sending empty replication request, chart has no dimensions", false); |
| 208 | |
| 209 | if (unlikely(!r.child_db.first_entry_t || !r.child_db.last_entry_t)) |
| 210 | return send_replay_chart_cmd(&r, "sending empty replication request, child has no stored data", false); |
| 211 | |
| 212 | if (unlikely(r.child_db.first_entry_t < 0 || r.child_db.last_entry_t < 0)) |
| 213 | return send_replay_chart_cmd(&r, "sending empty replication request, child db timestamps are invalid", true); |
| 214 | |
| 215 | if (unlikely(r.child_db.first_entry_t > r.child_db.wall_clock_time)) |
| 216 | return send_replay_chart_cmd(&r, "sending empty replication request, child db first entry is after its wall clock time", true); |
| 217 | |
| 218 | if (unlikely(r.child_db.first_entry_t > r.child_db.last_entry_t)) |
| 219 | return send_replay_chart_cmd(&r, "sending empty replication request, child timings are invalid (first entry > last entry)", true); |
| 220 | |
| 221 | // Check if parent is already caught up with or ahead of child |
| 222 | // This check uses >= (not just >) to handle the case where parent and child are exactly equal |
| 223 | // When equal, there's no gap to replicate, so we should finish replication |
| 224 | if (unlikely(r.local_db.last_entry_t >= r.child_db.last_entry_t)) { |
| 225 | // Parent is at or ahead of child - no replication needed |
| 226 | // Send empty request (after=0, before=0) with start_streaming=true to finish replication |
| 227 | // The child will receive this, recognize it as empty, and respond with start_streaming=true |
| 228 | // which will properly terminate the replication process |
| 229 | return send_replay_chart_cmd(&r, "sending empty replication request, local last entry is at or later than the child one", false); |
| 230 | } |
| 231 | |
| 232 | // let's find what the child can provide to fill that gap |
| 233 | |
| 234 | if(r.child_db.first_entry_t > r.gap.from) |
| 235 | // the child does not have all the data - let's get what it has |
| 236 | r.wanted.after = r.child_db.first_entry_t; |
| 237 | else |
| 238 | // ok, the child can fill the entire gap we have |
| 239 | r.wanted.after = r.gap.from; |
| 240 | |
| 241 | if(r.gap.to - r.wanted.after > host->stream.replication.step) |
| 242 | // the duration is too big for one request - let's take the first step |
| 243 | r.wanted.before = r.wanted.after + host->stream.replication.step; |
| 244 | else |
| 245 | // wow, we can do it in one request |
| 246 | r.wanted.before = r.gap.to; |
| 247 | |
| 248 | // don't ask from the child more than it has |
| 249 | if(r.wanted.before > r.child_db.last_entry_t) |
| 250 | r.wanted.before = r.child_db.last_entry_t; |
| 251 | |
| 252 | if(r.wanted.after > r.wanted.before) { |
| 253 | r.wanted.after = 0; |
| 254 | r.wanted.before = 0; |
| 255 | r.wanted.start_streaming = true; |
| 256 | return send_replay_chart_cmd(&r, "sending empty replication request, because wanted 'after' computed bigger than wanted 'before'", true); |
| 257 | } |
| 258 | |
| 259 | // the child should start streaming immediately if the wanted duration is small, or we reached the last entry of the child |
| 260 | r.wanted.start_streaming = (r.local_db.wall_clock_time - r.wanted.after <= host->stream.replication.step || |
| 261 | r.wanted.before >= r.child_db.last_entry_t || |
| 262 | r.wanted.before >= r.child_db.wall_clock_time || |
| 263 | r.wanted.before >= r.local_db.wall_clock_time); |
| 264 | |
| 265 | // the wanted timeframe is now r.wanted.after -> r.wanted.before |
| 266 | // send it |
| 267 | return send_replay_chart_cmd(&r, "OK", false); |
| 268 | } |
| 269 | |
| 270 | ALWAYS_INLINE bool stream_parse_enable_streaming(const char *start_streaming_txt) { |
| 271 | bool start_streaming; |
| 272 | |
| 273 | if(unlikely(!start_streaming_txt || !*start_streaming_txt)) { |
| 274 | start_streaming = false; |
| 275 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 276 | "REPLAY: malformed start_streaming boolean value empty"); |
| 277 | } |
| 278 | else if(likely(strcmp(start_streaming_txt, "false") == 0)) |
| 279 | start_streaming = false; |
| 280 | else if(likely(strcmp(start_streaming_txt, "true") == 0)) |
| 281 | start_streaming = true; |
| 282 | else { |
| 283 | start_streaming = false; |
| 284 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 285 | "REPLAY: malformed start_streaming boolean value '%s'", start_streaming_txt); |
| 286 | } |
| 287 | |
| 288 | return start_streaming; |
| 289 | } |