| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "commands.h" |
| 4 | #include "../stream-sender-internals.h" |
| 5 | #include "plugins.d/pluginsd_internals.h" |
| 6 | |
| 7 | // chart labels |
| 8 | static int stream_send_clabels_callback(const char *name, const char *value, RRDLABEL_SRC ls, void *data) { |
| 9 | BUFFER *wb = (BUFFER *)data; |
| 10 | buffer_sprintf(wb, PLUGINSD_KEYWORD_CLABEL " \"%s\" \"%s\" %d\n", name, value, ls & ~(RRDLABEL_FLAG_INTERNAL)); |
| 11 | return 1; |
| 12 | } |
| 13 | |
| 14 | static void stream_send_clabels(BUFFER *wb, RRDSET *st) { |
| 15 | if (st->rrdlabels) { |
| 16 | if(rrdlabels_walkthrough_read(st->rrdlabels, stream_send_clabels_callback, wb) > 0) |
| 17 | buffer_sprintf(wb, PLUGINSD_KEYWORD_CLABEL_COMMIT "\n"); |
| 18 | } |
| 19 | } |
| 20 | |
| 21 | // Send the current chart definition. |
| 22 | // Assumes that collector thread has already called sender_start for mutex / buffer state. |
| 23 | bool stream_sender_send_rrdset_definition(BUFFER *wb, RRDSET *st) { |
| 24 | uint32_t version = rrdset_metadata_version(st); |
| 25 | |
| 26 | RRDHOST *host = st->rrdhost; |
| 27 | NUMBER_ENCODING integer_encoding = stream_has_capability(host->sender, STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_HEX; |
| 28 | bool with_slots = stream_has_capability(host->sender, STREAM_CAP_SLOTS) ? true : false; |
| 29 | |
| 30 | bool replication_progress = false; |
| 31 | |
| 32 | // properly set the name for the remote end to parse it |
| 33 | char *name = ""; |
| 34 | if(likely(st->name)) { |
| 35 | if(unlikely(st->id != st->name)) { |
| 36 | // they differ |
| 37 | name = strchr(rrdset_name(st), '.'); |
| 38 | if(name) |
| 39 | name++; |
| 40 | else |
| 41 | name = ""; |
| 42 | } |
| 43 | } |
| 44 | |
| 45 | buffer_fast_strcat(wb, PLUGINSD_KEYWORD_CHART, sizeof(PLUGINSD_KEYWORD_CHART) - 1); |
| 46 | |
| 47 | if(with_slots) { |
| 48 | buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2); |
| 49 | buffer_print_uint64_encoded(wb, integer_encoding, st->stream.snd.chart_slot); |
| 50 | } |
| 51 | |
| 52 | // send the chart |
| 53 | buffer_sprintf( |
| 54 | wb |
| 55 | , " \"%s\" \"%s\" \"%s\" \"%s\" \"%s\" \"%s\" \"%s\" %d %d \"%s %s %s\" \"%s\" \"%s\"\n" |
| 56 | , rrdset_id(st) |
| 57 | , name |
| 58 | , rrdset_title(st) |
| 59 | , rrdset_units(st) |
| 60 | , rrdset_family(st) |
| 61 | , rrdset_context(st) |
| 62 | , rrdset_type_name(st->chart_type) |
| 63 | , st->priority |
| 64 | , st->update_every |
| 65 | , rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)?"obsolete":"" |
| 66 | , rrdset_flag_check(st, RRDSET_FLAG_STORE_FIRST)?"store_first":"" |
| 67 | , rrdset_flag_check(st, RRDSET_FLAG_HIDDEN)?"hidden":"" |
| 68 | , rrdset_plugin_name(st) |
| 69 | , rrdset_module_name(st) |
| 70 | ); |
| 71 | |
| 72 | // send the chart labels |
| 73 | if (stream_has_capability(host->sender, STREAM_CAP_CLABELS)) |
| 74 | stream_send_clabels(wb, st); |
| 75 | |
| 76 | // send the dimensions |
| 77 | RRDDIM *rd; |
| 78 | rrddim_foreach_read(rd, st) { |
| 79 | buffer_fast_strcat(wb, PLUGINSD_KEYWORD_DIMENSION, sizeof(PLUGINSD_KEYWORD_DIMENSION) - 1); |
| 80 | |
| 81 | if(with_slots) { |
| 82 | buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2); |
| 83 | buffer_print_uint64_encoded(wb, integer_encoding, rd->stream.snd.dim_slot); |
| 84 | } |
| 85 | |
| 86 | buffer_sprintf( |
| 87 | wb |
| 88 | , " \"%s\" \"%s\" \"%s\" %d %d \"%s %s %s %s\"\n" |
| 89 | , rrddim_id(rd) |
| 90 | , rrddim_name(rd) |
| 91 | , rrd_algorithm_name(rd->algorithm) |
| 92 | , rd->multiplier |
| 93 | , rd->divisor |
| 94 | , rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE)?"obsolete":"" |
| 95 | , rrddim_option_check(rd, RRDDIM_OPTION_HIDDEN)?"hidden":"" |
| 96 | , rrddim_option_check(rd, RRDDIM_OPTION_DONT_DETECT_RESETS_OR_OVERFLOWS)?"noreset":"" |
| 97 | , rrddim_is_float(rd)?"type=float":"type=int" |
| 98 | ); |
| 99 | } |
| 100 | rrddim_foreach_done(rd); |
| 101 | |
| 102 | // send the chart functions |
| 103 | if(stream_has_capability(host->sender, STREAM_CAP_FUNCTIONS)) |
| 104 | stream_sender_send_rrdset_functions(st, wb); |
| 105 | |
| 106 | // send the chart local custom variables |
| 107 | rrdvar_print_to_streaming_custom_chart_variables(st, wb); |
| 108 | |
| 109 | if (stream_has_capability(host->sender, STREAM_CAP_REPLICATION)) { |
| 110 | time_t db_first_time_t, db_last_time_t; |
| 111 | |
| 112 | time_t now = now_realtime_sec(); |
| 113 | rrdset_get_retention_of_tier_for_collected_chart(st, &db_first_time_t, &db_last_time_t, now, 0); |
| 114 | |
| 115 | buffer_sprintf(wb, PLUGINSD_KEYWORD_CHART_DEFINITION_END " %llu %llu %llu\n", |
| 116 | (unsigned long long)db_first_time_t, |
| 117 | (unsigned long long)db_last_time_t, |
| 118 | (unsigned long long)now); |
| 119 | |
| 120 | // The receiver skips replication for obsolete charts (stream-receiver.c), |
| 121 | // so do not enter the replication bookkeeping here either: it would pin |
| 122 | // rrdhost_sender_replicating_charts above zero and keep the host's |
| 123 | // pulse SND_REPLICATING / SND_RUNNING observability stuck on the wrong |
| 124 | // state. |
| 125 | if(!rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)) { |
| 126 | // Claim before publish: increment the host counter BEFORE setting |
| 127 | // RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS, so any concurrent |
| 128 | // observer (obsoleter at rrdset.c, finalizer at |
| 129 | // stream-replication-sender.c, reset at stream-sender.c) that |
| 130 | // sees the flag set in its CAS old-value also sees a counter |
| 131 | // already incremented to match. Without this ordering, an |
| 132 | // observer can clear the flag and call rrdhost_sender_replicating |
| 133 | // _charts_minus_one() before the sender's increment, causing a |
| 134 | // transient underflow that other concurrent inc/dec can latch. |
| 135 | bool first_claim = (rrdhost_sender_replicating_charts_plus_one(st->rrdhost) == 1); |
| 136 | |
| 137 | RRDSET_FLAGS old = rrdset_flag_set_and_clear(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS, RRDSET_FLAG_SENDER_REPLICATION_FINISHED); |
| 138 | bool we_caused_transition = !(old & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS); |
| 139 | |
| 140 | if(we_caused_transition) { |
| 141 | if(first_claim) |
| 142 | pulse_host_status(st->rrdhost, PULSE_HOST_STATUS_SND_REPLICATING, 0); |
| 143 | } |
| 144 | else { |
| 145 | // Lost race: another sender already had IN_PROGRESS set, so |
| 146 | // our +1 is one too many. Roll it back; mirror the natural- |
| 147 | // finalize pulse-status flip on the 0 boundary. |
| 148 | if(rrdhost_sender_replicating_charts_minus_one(st->rrdhost) == 0) |
| 149 | pulse_host_status(st->rrdhost, PULSE_HOST_STATUS_SND_RUNNING, 0); |
| 150 | } |
| 151 | |
| 152 | // Recheck after our CAS: a concurrent obsoleter may have set |
| 153 | // RRDSET_FLAG_OBSOLETE, OR a concurrent disconnect may have |
| 154 | // cleared the host's metadata-readiness flag. In either case the |
| 155 | // parent will not drive replication for this chart to completion |
| 156 | // and the natural decrement never fires; undo our state to keep |
| 157 | // the host counter and pulse status balanced. The atomic CAS |
| 158 | // ensures only the thread that observes IN_PROGRESS=1 actually |
| 159 | // decrements (handles the case where stream_sender_charts_and_ |
| 160 | // replication_reset() already cleared the flag during a disconnect |
| 161 | // racing with this push). |
| 162 | if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE) || |
| 163 | !rrdhost_can_stream_metadata_to_parent(st->rrdhost))) { |
| 164 | if(we_caused_transition) { |
| 165 | RRDSET_FLAGS undo = rrdset_flag_set_and_clear( |
| 166 | st, |
| 167 | RRDSET_FLAG_SENDER_REPLICATION_FINISHED, |
| 168 | RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS); |
| 169 | if(undo & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS) { |
| 170 | if(rrdhost_sender_replicating_charts_minus_one(st->rrdhost) == 0) |
| 171 | pulse_host_status(st->rrdhost, PULSE_HOST_STATUS_SND_RUNNING, 0); |
| 172 | } |
| 173 | } |
| 174 | } |
| 175 | else { |
| 176 | replication_progress = true; |
| 177 | } |
| 178 | } |
| 179 | |
| 180 | #ifdef NETDATA_LOG_REPLICATION_REQUESTS |
| 181 | internal_error(true, "REPLAY: 'host:%s/chart:%s' replication starts", |
| 182 | rrdhost_hostname(st->rrdhost), rrdset_id(st)); |
| 183 | #endif |
| 184 | } |
| 185 | |
| 186 | // we can set the exposed flag, after we commit the buffer |
| 187 | // because replication may pick it up prematurely |
| 188 | rrddim_foreach_read(rd, st) { |
| 189 | rrddim_metadata_exposed_upstream(rd, version); |
| 190 | } |
| 191 | rrddim_foreach_done(rd); |
| 192 | rrdset_metadata_exposed_upstream(st, version); |
| 193 | |
| 194 | st->stream.snd.resync_time_s = st->last_collected_time.tv_sec + (stream_send.initial_clock_resync_iterations * st->update_every); |
| 195 | return replication_progress; |
| 196 | } |
| 197 | |
| 198 | bool should_send_rrdset_matching(RRDSET *st, RRDSET_FLAGS flags) { |
| 199 | if(!(flags & RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED)) |
| 200 | return false; |
| 201 | |
| 202 | if(unlikely(!(flags & (RRDSET_FLAG_UPSTREAM_SEND | RRDSET_FLAG_UPSTREAM_IGNORE)))) { |
| 203 | RRDHOST *host = st->rrdhost; |
| 204 | |
| 205 | if (flags & RRDSET_FLAG_ANOMALY_DETECTION) { |
| 206 | if(ml_streaming_enabled()) |
| 207 | rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_SEND); |
| 208 | else |
| 209 | rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_IGNORE); |
| 210 | } |
| 211 | else { |
| 212 | int negative = 0, positive = 0; |
| 213 | SIMPLE_PATTERN_RESULT r; |
| 214 | |
| 215 | r = simple_pattern_matches_string_extract(host->stream.snd.charts_matching, st->context, NULL, 0); |
| 216 | if(r == SP_MATCHED_POSITIVE) positive++; |
| 217 | else if(r == SP_MATCHED_NEGATIVE) negative++; |
| 218 | |
| 219 | if(!negative) { |
| 220 | r = simple_pattern_matches_string_extract(host->stream.snd.charts_matching, st->name, NULL, 0); |
| 221 | if (r == SP_MATCHED_POSITIVE) positive++; |
| 222 | else if (r == SP_MATCHED_NEGATIVE) negative++; |
| 223 | } |
| 224 | |
| 225 | if(!negative) { |
| 226 | r = simple_pattern_matches_string_extract(host->stream.snd.charts_matching, st->id, NULL, 0); |
| 227 | if (r == SP_MATCHED_POSITIVE) positive++; |
| 228 | else if (r == SP_MATCHED_NEGATIVE) negative++; |
| 229 | } |
| 230 | |
| 231 | if(!negative && positive) |
| 232 | rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_SEND); |
| 233 | else |
| 234 | rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_IGNORE); |
| 235 | } |
| 236 | |
| 237 | // get the flags again, to know how to respond |
| 238 | flags = rrdset_flag_check(st, RRDSET_FLAG_UPSTREAM_SEND|RRDSET_FLAG_UPSTREAM_IGNORE); |
| 239 | } |
| 240 | |
| 241 | return flags & RRDSET_FLAG_UPSTREAM_SEND; |
| 242 | } |
| 243 | |
| 244 | // Called from the internal collectors to mark a chart obsolete. |
| 245 | bool stream_sender_send_rrdset_definition_now(RRDSET *st) { |
| 246 | RRDHOST *host = st->rrdhost; |
| 247 | |
| 248 | if(unlikely(!rrdhost_can_stream_metadata_to_parent(host) || !should_send_rrdset_matching(st, rrdset_flag_get(st)))) |
| 249 | return false; |
| 250 | |
| 251 | CLEAN_BUFFER *wb = buffer_create(0, NULL); |
| 252 | stream_sender_send_rrdset_definition(wb, st); |
| 253 | sender_commit_clean_buffer(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); |
| 254 | |
| 255 | return true; |
| 256 | } |