@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
h 170 lines 7.34 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #ifndef NETDATA_STREAM_SENDER_INTERNALS_H
4 #define NETDATA_STREAM_SENDER_INTERNALS_H
5
6 #include "stream.h"
7 #include "stream-thread.h"
8 #include "h2o-common.h"
9 #include "aclk/https_client.h"
10 #include "stream-parents.h"
11 #include "stream-circular-buffer.h"
12
13 // connector thread
14 #define WORKER_SENDER_CONNECTOR_JOB_CONNECTING 0
15 #define WORKER_SENDER_CONNECTOR_JOB_CONNECTED 1
16 #define WORKER_SENDER_CONNECTOR_JOB_REMOVED 2
17 #define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_BAD_HANDSHAKE 3
18 #define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_TIMEOUT 4
19 #define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_CANT_UPGRADE_CONNECTION 5
20 #define WORKER_SENDER_CONNECTOR_JOB_QUEUED_NODES 6
21 #define WORKER_SENDER_CONNECTOR_JOB_CONNECTED_NODES 7
22 #define WORKER_SENDER_CONNECTOR_JOB_FAILED_NODES 8
23 #define WORKER_SENDER_CONNECTOR_JOB_CANCELLED_NODES 9
24
25 #define CONNECTED_TO_SIZE 100
26
27 #include "stream-compression/compression.h"
28 #include "stream-conf.h"
29
30 typedef void (*stream_defer_action_t)(struct sender_state *s, void *data);
31 typedef void (*stream_defer_cleanup_t)(struct sender_state *s, void *data);
32
33 struct sender_state {
34 SPINLOCK spinlock;
35 STREAM_CAPABILITIES capabilities;
36 STREAM_CAPABILITIES disabled_capabilities;
37 int16_t hops;
38 bool parent_using_h2o;
39 WAITQ waitq;
40 ND_SOCK sock;
41 RRDHOST *host;
42
43 time_t last_state_since_t; // the timestamp of the last state (online/offline) change
44 STREAM_CIRCULAR_BUFFER *scb; // sender buffer
45
46 struct {
47 struct stream_opcode msg; // the template for sending a message to the dispatcher - protected by sender_lock()
48
49 // this is a property of stream_sender_send_msg_to_dispatcher()
50 // protected by dispatcher->messages.spinlock
51 // DO NOT READ OR WRITE ANYWHERE
52 uint32_t msg_slot; // ensures a opcode queue that can never get full
53
54 struct compressor_state compressor;
55
56 struct {
57 size_t size;
58 char *b;
59 ssize_t read_len;
60 struct line_splitter line;
61 } rbuf;
62
63 struct {
64 const char *end_keyword;
65 BUFFER *payload;
66 stream_defer_action_t action;
67 stream_defer_cleanup_t cleanup;
68 void *action_data;
69 } defer;
70
71 nd_poll_event_t wanted;
72 usec_t last_traffic_ut;
73 struct pollfd_meta meta;
74 } thread;
75
76 struct {
77 int8_t id; // the connector id - protected by sender_lock()
78 } connector;
79
80 struct {
81 bool shutdown; // when set, the sender should stop sending this host
82 STREAM_HANDSHAKE reason; // the reason we decided to stop this sender
83 } exit;
84
85 struct {
86 uint32_t last_counter_sum; // copy from the host, to detect progress
87 usec_t last_progress_ut; // last time we found some progress (monotonic)
88 usec_t last_checked_ut; // last time we checked for stalled progress (monotonic)
89
90 DICTIONARY *requests; // de-duplication of replication requests, per chart
91 time_t oldest_request_after_t; // the timestamp of the oldest replication request
92 time_t latest_completed_before_t; // the timestamp of the latest replication request
93
94 struct {
95 size_t pending_requests; // the currently outstanding replication requests
96 size_t charts_replicating; // the number of unique charts having pending replication requests (on every request one is added and is removed when we finish it - it does not track completion of the replication for this chart)
97 bool reached_max; // true when the sender buffer should not get more replication responses
98 } atomic;
99 } replication;
100
101 #ifdef NETDATA_LOG_STREAM_SENDER
102 struct {
103 SPINLOCK spinlock;
104 struct timespec first_call;
105 BUFFER *received;
106 FILE *fp;
107 } log;
108 #endif
109
110 char remote_ip[CONNECTED_TO_SIZE + 1]; // We don't know which proxy we connect to, passed back from socket.c
111 };
112
113 #define stream_sender_lock(sender) spinlock_lock(&(sender)->spinlock)
114 #define stream_sender_unlock(sender) spinlock_unlock(&(sender)->spinlock)
115 #define stream_sender_trylock(sender) spinlock_trylock(&(sender)->spinlock)
116
117 #define stream_sender_replication_buffer_full_set(sender, value) __atomic_store_n(&((sender)->replication.atomic.reached_max), value, __ATOMIC_SEQ_CST)
118 #define stream_sender_replication_buffer_full_get(sender) __atomic_load_n(&((sender)->replication.atomic.reached_max), __ATOMIC_SEQ_CST)
119
120 #define stream_sender_replicating_charts(sender) __atomic_load_n(&((sender)->replication.atomic.charts_replicating), __ATOMIC_RELAXED)
121 #define stream_sender_replicating_charts_plus_one(sender) __atomic_add_fetch(&((sender)->replication.atomic.charts_replicating), 1, __ATOMIC_RELAXED)
122 #define stream_sender_replicating_charts_minus_one(sender) __atomic_sub_fetch(&((sender)->replication.atomic.charts_replicating), 1, __ATOMIC_RELAXED)
123 #define stream_sender_replicating_charts_zero(sender) __atomic_store_n(&((sender)->replication.atomic.charts_replicating), 0, __ATOMIC_RELAXED)
124
125 #define stream_sender_pending_replication_requests(sender) __atomic_load_n(&((sender)->replication.atomic.pending_requests), __ATOMIC_RELAXED)
126 #define stream_sender_pending_replication_requests_plus_one(sender) __atomic_add_fetch(&((sender)->replication.atomic.pending_requests), 1, __ATOMIC_RELAXED)
127 #define stream_sender_pending_replication_requests_minus_one(sender) __atomic_sub_fetch(&((sender)->replication.atomic.pending_requests), 1, __ATOMIC_RELAXED)
128 #define stream_sender_pending_replication_requests_zero(sender) __atomic_store_n(&((sender)->replication.atomic.pending_requests), 0, __ATOMIC_RELAXED)
129
130 void stream_sender_add_to_connector_queue(RRDHOST *host);
131
132 void stream_sender_execute_commands_cleanup(struct sender_state *s);
133 void stream_sender_execute_commands(struct sender_state *s);
134
135 bool stream_connect(struct sender_state *s, uint16_t default_port, time_t timeout);
136
137 bool stream_sender_is_host_stopped(struct sender_state *s);
138
139 void stream_sender_send_opcode(struct sender_state *s, struct stream_opcode msg);
140
141 void stream_sender_add_to_queue(struct sender_state *s);
142
143 // stream connector
144 typedef enum __attribute__((packed)) {
145 STRCNT_CMD_NONE = 0,
146 STRCNT_CMD_CONNECT,
147 STRCNT_CMD_REMOVE,
148
149 // terminator
150 STRCNT_CMD_MAX,
151 } STRCNT_CMD;
152
153 bool stream_connector_init(struct sender_state *s);
154 void stream_connector_cancel_threads(void);
155 void stream_connector_add(struct sender_state *s);
156 void stream_connector_requeue(struct sender_state *s, STRCNT_CMD cmd);
157 bool stream_connector_is_signaled_to_stop(struct sender_state *s);
158
159 void stream_sender_on_connect(struct sender_state *s);
160 void stream_sender_on_disconnect(struct sender_state *s);
161
162 void stream_sender_remove(struct sender_state *s, STREAM_HANDSHAKE reason);
163
164 #ifdef NETDATA_LOG_STREAM_SENDER
165 void stream_sender_log_payload(struct sender_state *s, BUFFER *payload, STREAM_TRAFFIC_TYPE type, bool inbound);
166 #else
167 #define stream_sender_log_payload(s, payload, type, inbound) debug_dummy()
168 #endif
169
170 #endif //NETDATA_STREAM_SENDER_INTERNALS_H