@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 143 lines 4.91 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "stream-sender-internals.h"
4 #include "stream-replication-sender.h"
5
6 bool stream_sender_has_capabilities(struct rrdhost *host, STREAM_CAPABILITIES capabilities) {
7 return host && stream_has_capability(host->sender, capabilities);
8 }
9
10 bool stream_sender_is_connected_with_ssl(struct rrdhost *host) {
11 return host && rrdhost_can_stream_metadata_to_parent(host) && nd_sock_is_ssl(&host->sender->sock);
12 }
13
14 bool stream_sender_has_compression(struct rrdhost *host) {
15 return host && host->sender && host->sender->thread.compressor.initialized;
16 }
17
18 void stream_sender_structures_init(RRDHOST *host, bool stream, STRING *parents, STRING *api_key, STRING *send_charts_matching) {
19 if(rrdhost_flag_check(host, RRDHOST_FLAG_STREAM_SENDER_INITIALIZED))
20 return;
21
22 if(!stream || !parents || !api_key) {
23 rrdhost_option_clear(host, RRDHOST_OPTION_SENDER_ENABLED);
24 return;
25 }
26
27 rrdhost_flag_set(host, RRDHOST_FLAG_STREAM_SENDER_INITIALIZED);
28
29 if (host->sender) return;
30
31 host->sender = callocz(1, sizeof(*host->sender));
32 __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_senders, sizeof(*host->sender), __ATOMIC_RELAXED);
33
34 host->sender->connector.id = -1;
35 host->sender->host = host;
36 host->sender->scb = stream_circular_buffer_create();
37 waitq_init(&host->sender->waitq);
38 host->sender->capabilities = stream_our_capabilities(host, true);
39
40 nd_sock_init(&host->sender->sock, netdata_ssl_streaming_sender_ctx, netdata_ssl_validate_certificate_sender);
41 host->sender->disabled_capabilities = STREAM_CAP_NONE;
42
43 if(!stream_send.compression.enabled)
44 host->sender->disabled_capabilities |= STREAM_CAP_COMPRESSIONS_AVAILABLE;
45
46 spinlock_init(&host->sender->spinlock);
47 replication_sender_init(host->sender);
48
49 // gracefully swap destination
50 if(host->stream.snd.destination != parents) {
51 STRING *t = string_dup(parents);
52 SWAP(host->stream.snd.destination, t);
53 string_freez(t);
54 }
55 rrdhost_stream_parents_update_from_destination(host);
56
57 // gracefully swap api_key
58 if(host->stream.snd.api_key != api_key) {
59 STRING *t = string_dup(api_key);
60 SWAP(host->stream.snd.api_key, t);
61 string_freez(t);
62 }
63
64 // gracefully swap send_charts_matching
65 {
66 SIMPLE_PATTERN *t = simple_pattern_create(
67 string2str(send_charts_matching), NULL, SIMPLE_PATTERN_EXACT, true);
68 SWAP(host->stream.snd.charts_matching, t);
69 simple_pattern_free(t);
70 }
71
72 rrdhost_option_set(host, RRDHOST_OPTION_SENDER_ENABLED);
73 }
74
75 void stream_sender_structures_free(struct rrdhost *host) {
76 rrdhost_option_clear(host, RRDHOST_OPTION_SENDER_ENABLED);
77
78 if (unlikely(!host->sender)) return;
79
80 // stop a possibly running thread
81 stream_sender_signal_to_stop_and_wait(host, STREAM_HANDSHAKE_SND_DISCONNECT_HOST_CLEANUP, true);
82 stream_circular_buffer_destroy(host->sender->scb);
83 host->sender->scb = NULL;
84 waitq_destroy(&host->sender->waitq);
85 stream_compressor_destroy(&host->sender->thread.compressor);
86
87 replication_sender_cleanup(host->sender);
88
89 __atomic_sub_fetch(&netdata_buffers_statistics.rrdhost_senders, sizeof(*host->sender), __ATOMIC_RELAXED);
90
91 freez(host->sender);
92 host->sender = NULL;
93
94 sender_host_buffer_free(host);
95 rrdhost_stream_parents_free(host, false);
96
97 rrdhost_flag_clear(host, RRDHOST_FLAG_STREAM_SENDER_INITIALIZED);
98 }
99
100 void stream_sender_start_host(struct rrdhost *host) {
101 internal_fatal(!rrdhost_has_stream_sender_enabled(host),
102 "Host '%s' does not have streaming enabled, but %s() was called",
103 rrdhost_hostname(host), __FUNCTION__);
104
105 stream_sender_add_to_connector_queue(host);
106 }
107
108 void stream_sender_start_localhost(void *ptr __maybe_unused)
109 {
110 if (!localhost)
111 return;
112 stream_sender_start_host(localhost);
113 }
114
115 // Either the receiver lost the connection or the host is being destroyed.
116 // The sender mutex guards thread creation, any spurious data is wiped on reconnection.
117 void stream_sender_signal_to_stop_and_wait(struct rrdhost *host, STREAM_HANDSHAKE reason, bool wait) {
118 if (!host->sender)
119 return;
120
121 stream_sender_lock(host->sender);
122
123 if(rrdhost_flag_check(host, RRDHOST_FLAG_STREAM_SENDER_ADDED)) {
124 __atomic_store_n(&host->sender->exit.shutdown, true, __ATOMIC_RELAXED);
125 host->sender->exit.reason = reason;
126 }
127
128 struct stream_opcode msg = host->sender->thread.msg;
129 stream_sender_unlock(host->sender);
130
131 if(reason == STREAM_HANDSHAKE_SND_DISCONNECT_HOST_CLEANUP)
132 msg.opcode = STREAM_OPCODE_SENDER_STOP_HOST_CLEANUP;
133 else
134 msg.opcode = STREAM_OPCODE_SENDER_STOP_RECEIVER_LEFT;
135 msg.reason = reason;
136
137 stream_sender_send_opcode(host->sender, msg);
138
139 while(wait && rrdhost_flag_check(host, RRDHOST_FLAG_STREAM_SENDER_ADDED)) {
140 sleep_usec(10 * USEC_PER_MS);
141 stream_connector_remove_host(host);
142 }
143 }