| 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 | static BUFFER *preferred_sender_buffer(RRDHOST *host) { |
| 8 | if(host->stream.snd.commit.receiver_tid == gettid_cached()) |
| 9 | return sender_host_buffer(host); |
| 10 | else |
| 11 | return sender_thread_buffer(host->sender, HOST_THREAD_BUFFER_INITIAL_SIZE); |
| 12 | } |
| 13 | |
| 14 | ALWAYS_INLINE RRDSET_STREAM_BUFFER stream_send_metrics_init(RRDSET *st, time_t wall_clock_time) { |
| 15 | RRDHOST *host = st->rrdhost; |
| 16 | |
| 17 | // fetch the flags we need to check with one atomic operation |
| 18 | RRDHOST_FLAGS host_flags = __atomic_load_n(&host->flags, __ATOMIC_SEQ_CST); |
| 19 | |
| 20 | // check if we are not connected |
| 21 | if(unlikely(!(host_flags & RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS))) { |
| 22 | |
| 23 | if(unlikely((host_flags & RRDHOST_FLAG_COLLECTOR_ONLINE) && |
| 24 | !(host_flags & RRDHOST_FLAG_STREAM_SENDER_ADDED))) |
| 25 | stream_sender_start_host(host); |
| 26 | |
| 27 | if(unlikely(!(host_flags & RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS))) { |
| 28 | rrdhost_flag_set(host, RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS); |
| 29 | |
| 30 | // this message is logged in 2 cases: |
| 31 | // - the parent is connected, but not yet available for streaming data |
| 32 | // - the parent just disconnected, so local data are not streamed to parent |
| 33 | |
| 34 | nd_log(NDLS_DAEMON, NDLP_INFO, |
| 35 | "STREAM SND '%s': streaming is not ready, not sending data to a parent...", |
| 36 | rrdhost_hostname(host)); |
| 37 | } |
| 38 | |
| 39 | return (RRDSET_STREAM_BUFFER) { .wb = NULL, }; |
| 40 | } |
| 41 | else if(unlikely(host_flags & RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS)) { |
| 42 | nd_log(NDLS_DAEMON, NDLP_INFO, |
| 43 | "STREAM SND '%s': streaming is ready, sending metrics to parent...", |
| 44 | rrdhost_hostname(host)); |
| 45 | rrdhost_flag_clear(host, RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS); |
| 46 | } |
| 47 | |
| 48 | if(unlikely(host_flags & RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED)) { |
| 49 | BUFFER *wb = preferred_sender_buffer(host); |
| 50 | stream_sender_send_global_rrdhost_functions(host, wb, stream_has_capability(host->sender, STREAM_CAP_DYNCFG)); |
| 51 | sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); |
| 52 | } |
| 53 | |
| 54 | bool exposed_upstream = rrdset_check_upstream_exposed(st); |
| 55 | RRDSET_FLAGS rrdset_flags = rrdset_flag_get(st); |
| 56 | bool replication_in_progress = !(rrdset_flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED); |
| 57 | |
| 58 | if(unlikely((exposed_upstream && replication_in_progress) || |
| 59 | !should_send_rrdset_matching(st, rrdset_flags))) |
| 60 | return (RRDSET_STREAM_BUFFER) { .wb = NULL, }; |
| 61 | |
| 62 | if(unlikely(!exposed_upstream)) { |
| 63 | BUFFER *wb = preferred_sender_buffer(host); |
| 64 | replication_in_progress = stream_sender_send_rrdset_definition(wb, st); |
| 65 | sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); |
| 66 | } |
| 67 | |
| 68 | if(unlikely(replication_in_progress)) |
| 69 | return (RRDSET_STREAM_BUFFER) { .wb = NULL, }; |
| 70 | |
| 71 | return (RRDSET_STREAM_BUFFER) { |
| 72 | .capabilities = host->sender->capabilities, |
| 73 | .v2 = stream_has_capability(host->sender, STREAM_CAP_INTERPOLATED), |
| 74 | .rrdset_flags = rrdset_flags, |
| 75 | .wb = preferred_sender_buffer(host), |
| 76 | .wall_clock_time = wall_clock_time, |
| 77 | }; |
| 78 | } |