@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
c 481 lines 20.2 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #define PULSE_INTERNALS
4 #include "pulse.h"
5
6 DEFINE_JUDYL_TYPED(PHOST, PULSE_HOST_STATUS);
7
8 // --------------------------------------------------------------------------------------------------------------------
9 // parents
10
11 struct by_reason {
12 size_t counters[STREAM_HANDSHAKE_NEGATIVE_MAX + 3];
13 RRDSET *st;
14 RRDDIM *rd[STREAM_HANDSHAKE_NEGATIVE_MAX + 3];
15 };
16
17 #define STREAM_HANDSHAKE_STREAM_INFO (STREAM_HANDSHAKE_NEGATIVE_MAX)
18 #define STREAM_HANDSHAKE_CONNECT (STREAM_HANDSHAKE_NEGATIVE_MAX + 1)
19 #define STREAM_HANDSHAKE_OTHER (STREAM_HANDSHAKE_NEGATIVE_MAX + 2)
20
21 struct {
22 SPINLOCK spinlock;
23 PHOST_JudyLSet index;
24
25 struct {
26 // counters
27 struct by_reason events_by_reason;
28 struct by_reason disconnects_by_reason;
29
30 // gauges
31 struct {
32 RRDSET *st_nodes;
33 RRDDIM *rd_loading;
34 RRDDIM *rd_local;
35 RRDDIM *rd_virtual;
36 RRDDIM *rd_archived;
37 RRDDIM *rd_offline;
38 RRDDIM *rd_waiting;
39 RRDDIM *rd_replication_waiting;
40 RRDDIM *rd_replicating;
41 RRDDIM *rd_running;
42
43 ssize_t nodes_local;
44 ssize_t nodes_virtual;
45 ssize_t nodes_loading;
46 ssize_t nodes_archived;
47 ssize_t nodes_offline;
48 ssize_t nodes_waiting;
49 ssize_t nodes_replicating;
50 ssize_t nodes_replication_waiting;
51 ssize_t nodes_running;
52 } type[2];
53 } parent;
54
55 struct {
56 // counters
57 struct by_reason stream_info_failed_by_reason;
58 struct by_reason events_by_reason;
59 struct by_reason disconnects_by_reason;
60
61 // gauges
62 ssize_t nodes_offline;
63 ssize_t nodes_connecting;
64 ssize_t nodes_pending;
65 ssize_t nodes_waiting;
66 ssize_t nodes_replicating;
67 ssize_t nodes_running;
68 ssize_t nodes_no_dst;
69 ssize_t nodes_no_dst_failed;
70 } sender;
71
72 } p = { 0 };
73
74 static PULSE_HOST_STATUS pulse_host_detect_receiver_status(RRDHOST *host) {
75 RRDHOST_STATUS status = { 0 };
76 rrdhost_status(host, now_realtime_sec(), &status, RRDHOST_STATUS_BASIC);
77
78 PULSE_HOST_STATUS rc = 0;
79
80 if(status.db.status == RRDHOST_DB_STATUS_INITIALIZING || status.ingest.status == RRDHOST_INGEST_STATUS_INITIALIZING)
81 rc = PULSE_HOST_STATUS_LOADING;
82
83 else if(status.ingest.type == RRDHOST_INGEST_TYPE_LOCALHOST)
84 rc = PULSE_HOST_STATUS_LOCAL;
85
86 else if(status.ingest.type == RRDHOST_INGEST_TYPE_VIRTUAL)
87 rc = PULSE_HOST_STATUS_VIRTUAL;
88
89 else if(status.ingest.status == RRDHOST_INGEST_STATUS_ARCHIVED)
90 rc = PULSE_HOST_STATUS_ARCHIVED;
91
92 else if(status.ingest.status == RRDHOST_INGEST_STATUS_REPLICATING)
93 rc = PULSE_HOST_STATUS_RCV_REPLICATING;
94
95 else if(status.ingest.status == RRDHOST_INGEST_STATUS_OFFLINE)
96 rc = PULSE_HOST_STATUS_RCV_OFFLINE;
97
98 else if(status.ingest.status == RRDHOST_INGEST_STATUS_ONLINE)
99 rc = PULSE_HOST_STATUS_RCV_RUNNING;
100
101 return rc;
102 }
103
104 static void update_reason(struct by_reason *b, STREAM_HANDSHAKE reason) {
105 int r = reason;
106
107 if(r >= 0)
108 r = 0;
109 else if(r > -STREAM_HANDSHAKE_NEGATIVE_MAX)
110 r = -reason;
111 else
112 r = STREAM_HANDSHAKE_NEGATIVE_MAX;
113
114 __atomic_add_fetch(&b->counters[r], 1, __ATOMIC_RELAXED);
115 }
116
117 static void pulse_host_add_sub_status(PULSE_HOST_STATUS status, ssize_t val, STREAM_HANDSHAKE reason) {
118 size_t idx = status & PULSE_HOST_STATUS_EPHEMERAL ? 1 : 0;
119 status &= ~(PULSE_HOST_STATUS_EPHEMERAL | PULSE_HOST_STATUS_PERMANENT);
120
121 while(status) {
122 PULSE_HOST_STATUS s = 1 << (__builtin_ffs(status) - 1);
123 status &= ~s;
124
125 bool do_parent_reason = false, do_sender_reason = false;
126
127 switch(s) {
128 default:
129 break;
130
131 case PULSE_HOST_STATUS_LOCAL:
132 __atomic_add_fetch(&p.parent.type[idx].nodes_local, val, __ATOMIC_RELAXED);
133 break;
134
135 case PULSE_HOST_STATUS_VIRTUAL:
136 __atomic_add_fetch(&p.parent.type[idx].nodes_virtual, val, __ATOMIC_RELAXED);
137 break;
138
139 case PULSE_HOST_STATUS_LOADING:
140 __atomic_add_fetch(&p.parent.type[idx].nodes_loading, val, __ATOMIC_RELAXED);
141 break;
142
143 case PULSE_HOST_STATUS_ARCHIVED:
144 __atomic_add_fetch(&p.parent.type[idx].nodes_archived, val, __ATOMIC_RELAXED);
145 break;
146
147 case PULSE_HOST_STATUS_RCV_OFFLINE:
148 __atomic_add_fetch(&p.parent.type[idx].nodes_offline, val, __ATOMIC_RELAXED);
149 do_parent_reason = true;
150 break;
151
152 case PULSE_HOST_STATUS_RCV_WAITING:
153 __atomic_add_fetch(&p.parent.type[idx].nodes_waiting, val, __ATOMIC_RELAXED);
154 do_parent_reason = true;
155 reason = 0;
156 break;
157
158 case PULSE_HOST_STATUS_RCV_REPLICATION_WAIT:
159 __atomic_add_fetch(&p.parent.type[idx].nodes_replication_waiting, val, __ATOMIC_RELAXED);
160 break;
161
162 case PULSE_HOST_STATUS_RCV_REPLICATING:
163 __atomic_add_fetch(&p.parent.type[idx].nodes_replicating, val, __ATOMIC_RELAXED);
164 break;
165
166 case PULSE_HOST_STATUS_RCV_RUNNING:
167 __atomic_add_fetch(&p.parent.type[idx].nodes_running, val, __ATOMIC_RELAXED);
168 break;
169
170 case PULSE_HOST_STATUS_SND_OFFLINE:
171 __atomic_add_fetch(&p.sender.nodes_offline, val, __ATOMIC_RELAXED);
172 do_sender_reason = true;
173 break;
174
175 case PULSE_HOST_STATUS_SND_PENDING:
176 __atomic_add_fetch(&p.sender.nodes_pending, val, __ATOMIC_RELAXED);
177 break;
178
179 case PULSE_HOST_STATUS_SND_CONNECTING:
180 __atomic_add_fetch(&p.sender.nodes_connecting, val, __ATOMIC_RELAXED);
181 __atomic_add_fetch(&p.sender.events_by_reason.counters[STREAM_HANDSHAKE_CONNECT], 1, __ATOMIC_RELAXED);
182 break;
183
184 case PULSE_HOST_STATUS_SND_WAITING:
185 __atomic_add_fetch(&p.sender.nodes_waiting, val, __ATOMIC_RELAXED);
186 break;
187
188 case PULSE_HOST_STATUS_SND_REPLICATING:
189 __atomic_add_fetch(&p.sender.nodes_replicating, val, __ATOMIC_RELAXED);
190 break;
191
192 case PULSE_HOST_STATUS_SND_RUNNING:
193 __atomic_add_fetch(&p.sender.nodes_running, val, __ATOMIC_RELAXED);
194 break;
195
196 case PULSE_HOST_STATUS_SND_NO_DST:
197 __atomic_add_fetch(&p.sender.nodes_no_dst, val, __ATOMIC_RELAXED);
198 break;
199
200 case PULSE_HOST_STATUS_SND_NO_DST_FAILED:
201 __atomic_add_fetch(&p.sender.nodes_no_dst_failed, val, __ATOMIC_RELAXED);
202 break;
203 }
204
205 if(do_parent_reason && val > 0)
206 update_reason(&p.parent.disconnects_by_reason, reason);
207
208 if(do_sender_reason && val > 0)
209 update_reason(&p.sender.disconnects_by_reason, reason);
210 }
211 }
212
213 void pulse_host_status(RRDHOST *host, PULSE_HOST_STATUS status, STREAM_HANDSHAKE reason) {
214 PULSE_HOST_STATUS remove = 0;
215
216 if(!status)
217 status = pulse_host_detect_receiver_status(host);
218
219 static const PULSE_HOST_STATUS ephemerality =
220 PULSE_HOST_STATUS_EPHEMERAL | PULSE_HOST_STATUS_PERMANENT;
221
222 static const PULSE_HOST_STATUS basic =
223 PULSE_HOST_STATUS_LOCAL | PULSE_HOST_STATUS_VIRTUAL | PULSE_HOST_STATUS_LOADING |
224 PULSE_HOST_STATUS_ARCHIVED | PULSE_HOST_STATUS_DELETED;
225
226 static const PULSE_HOST_STATUS receiver =
227 PULSE_HOST_STATUS_RCV_OFFLINE | PULSE_HOST_STATUS_RCV_WAITING | PULSE_HOST_STATUS_RCV_REPLICATING |
228 PULSE_HOST_STATUS_RCV_REPLICATION_WAIT | PULSE_HOST_STATUS_RCV_RUNNING;
229
230 static const PULSE_HOST_STATUS sender =
231 PULSE_HOST_STATUS_SND_OFFLINE | PULSE_HOST_STATUS_SND_PENDING | PULSE_HOST_STATUS_SND_CONNECTING |
232 PULSE_HOST_STATUS_SND_WAITING | PULSE_HOST_STATUS_SND_REPLICATING | PULSE_HOST_STATUS_SND_RUNNING |
233 PULSE_HOST_STATUS_SND_NO_DST | PULSE_HOST_STATUS_SND_NO_DST_FAILED;
234
235 if((status & (basic | receiver)) && !(status & ephemerality))
236 status |= rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST) ?
237 PULSE_HOST_STATUS_EPHEMERAL : PULSE_HOST_STATUS_PERMANENT;
238
239 if(status & basic)
240 remove = basic | receiver | ephemerality | sender;
241 else if(status & receiver)
242 remove = basic | receiver | ephemerality;
243 else if(status & sender)
244 remove = sender;
245
246 spinlock_lock(&p.spinlock);
247 PULSE_HOST_STATUS old = PHOST_GET(&p.index, (uintptr_t)host);
248 if(status == PULSE_HOST_STATUS_DELETED) {
249 PHOST_DEL(&p.index, (uintptr_t)host);
250 status = 0; // do not add anything, just remove the old flags
251 }
252 else
253 PHOST_SET(&p.index, (uintptr_t)host, (old & ~remove) | status);
254 spinlock_unlock(&p.spinlock);
255
256 remove &= old;
257
258 pulse_host_add_sub_status(remove, -1, 0);
259 pulse_host_add_sub_status(status, 1, reason);
260 }
261
262 void pulse_parent_stream_info_received_request(void) {
263 __atomic_add_fetch(&p.parent.events_by_reason.counters[STREAM_HANDSHAKE_STREAM_INFO], 1, __ATOMIC_RELAXED);
264 }
265
266 void pulse_parent_receiver_request(void) {
267 __atomic_add_fetch(&p.parent.events_by_reason.counters[STREAM_HANDSHAKE_CONNECT], 1, __ATOMIC_RELAXED);
268 }
269
270 void pulse_parent_receiver_rejected(STREAM_HANDSHAKE reason) {
271 update_reason(&p.parent.events_by_reason, reason);
272 }
273
274 // --------------------------------------------------------------------------------------------------------------------
275 // children / senders
276
277 void pulse_stream_info_sent_request(void) {
278 __atomic_add_fetch(&p.sender.events_by_reason.counters[STREAM_HANDSHAKE_STREAM_INFO], 1, __ATOMIC_RELAXED);
279 }
280
281 void pulse_sender_stream_info_failed(const char *destination __maybe_unused, STREAM_HANDSHAKE reason) {
282 update_reason(&p.sender.stream_info_failed_by_reason, reason);
283 }
284
285 void pulse_sender_connection_failed(const char *destination __maybe_unused, STREAM_HANDSHAKE reason) {
286 update_reason(&p.sender.events_by_reason, reason);
287 }
288
289 // --------------------------------------------------------------------------------------------------------------------
290
291 static void chart_by_reason(struct by_reason *b, const char *id, const char *context, const char *title, const char *label, int priority) {
292 if(!b->st) {
293 b->st = rrdset_create_localhost(
294 "netdata"
295 , id
296 , NULL
297 , "Streaming"
298 , context
299 , title
300 , "events/s"
301 , "netdata"
302 , "pulse"
303 , priority
304 , localhost->rrd_update_every
305 , RRDSET_TYPE_LINE
306 );
307
308 for(int i = 0; i < STREAM_HANDSHAKE_NEGATIVE_MAX ;i++) {
309 char buf[1024];
310 if(!i)
311 strncpyz(buf, "connected", sizeof(buf) - 1);
312 else
313 strncpyz(buf, stream_handshake_error_to_string(-i), sizeof(buf) - 1);
314 for(int c = 0; buf[c] ;c++)
315 buf[c] = (char)tolower(buf[c]);
316
317 b->rd[i] = rrddim_add(b->st, buf, NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
318 }
319
320 rrdlabels_add(b->st->rrdlabels, "type", label, RRDLABEL_SRC_AUTO);
321
322 b->rd[STREAM_HANDSHAKE_STREAM_INFO] = rrddim_add(b->st, "info", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
323 b->rd[STREAM_HANDSHAKE_CONNECT] = rrddim_add(b->st, "connect", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
324 b->rd[STREAM_HANDSHAKE_OTHER] = rrddim_add(b->st, "other", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
325 }
326
327 for(size_t i = 0; i <= STREAM_HANDSHAKE_OTHER ;i++)
328 rrddim_set_by_pointer(b->st, b->rd[i], (collected_number)__atomic_load_n(&b->counters[i], __ATOMIC_RELAXED));
329
330 rrdset_done(b->st);
331 }
332
333 void pulse_parents_do(bool extended) {
334 if(netdata_conf_is_parent()) {
335 for(size_t idx = 0; idx < _countof(p.parent.type) ; idx++) {
336 if (unlikely(!p.parent.type[idx].st_nodes)) {
337 const char *type;
338 const char *id;
339 if(idx == 0) {
340 type = "permanent";
341 id = "netdata.streaming_inbound_permanent";
342 }
343 else {
344 type = "ephemeral";
345 id = "netdata.streaming_inbound_ephemeral";
346 }
347
348 p.parent.type[idx].st_nodes = rrdset_create_localhost(
349 "netdata"
350 , id
351 , NULL
352 , "Streaming"
353 , "netdata.streaming_inbound"
354 , "Inbound Nodes"
355 , "nodes"
356 , "netdata"
357 , "pulse"
358 , 130150
359 , localhost->rrd_update_every
360 , RRDSET_TYPE_LINE
361 );
362
363 rrdlabels_add(p.parent.type[idx].st_nodes->rrdlabels, "type", type, RRDLABEL_SRC_AUTO);
364
365 p.parent.type[idx].rd_local = rrddim_add(p.parent.type[idx].st_nodes, "local", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
366 p.parent.type[idx].rd_virtual = rrddim_add(p.parent.type[idx].st_nodes, "virtual", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
367 p.parent.type[idx].rd_loading = rrddim_add(p.parent.type[idx].st_nodes, "loading", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
368 p.parent.type[idx].rd_archived = rrddim_add(p.parent.type[idx].st_nodes, "stale archived", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
369 p.parent.type[idx].rd_offline = rrddim_add(p.parent.type[idx].st_nodes, "stale disconnected", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
370 p.parent.type[idx].rd_waiting = rrddim_add(p.parent.type[idx].st_nodes, "waiting", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
371 p.parent.type[idx].rd_replication_waiting = rrddim_add(p.parent.type[idx].st_nodes, "waiting replication", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
372 p.parent.type[idx].rd_replicating = rrddim_add(p.parent.type[idx].st_nodes, "replicating", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
373 p.parent.type[idx].rd_running = rrddim_add(p.parent.type[idx].st_nodes, "running", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
374 }
375
376 rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_local, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_local, __ATOMIC_RELAXED));
377 rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_virtual, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_virtual, __ATOMIC_RELAXED));
378 rrddim_set_by_pointer(p.parent.type[idx].st_nodes,p.parent.type[idx].rd_loading, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_loading, __ATOMIC_RELAXED));
379 rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_archived, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_archived, __ATOMIC_RELAXED));
380 rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_offline, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_offline, __ATOMIC_RELAXED));
381 rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_waiting, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_waiting, __ATOMIC_RELAXED));
382 rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_replication_waiting, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_replication_waiting, __ATOMIC_RELAXED));
383 rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_replicating, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_replicating, __ATOMIC_RELAXED));
384 rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_running, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_running, __ATOMIC_RELAXED));
385
386 rrdset_done(p.parent.type[idx].st_nodes);
387 }
388
389 if(extended) {
390 chart_by_reason(
391 &p.parent.events_by_reason,
392 "streaming_rejections_inbound",
393 "netdata.streaming_events_inbound",
394 "Inbound Streaming Events",
395 "rejections",
396 130151);
397 chart_by_reason(
398 &p.parent.disconnects_by_reason,
399 "streaming_disconnects_inbound",
400 "netdata.streaming_events_inbound",
401 "Inbound Streaming Events",
402 "disconnects",
403 130151);
404 }
405 }
406
407 if(stream_conf_is_child()) {
408 {
409 static RRDSET *st_nodes = NULL;
410 static RRDDIM *rd_pending = NULL;
411 static RRDDIM *rd_connecting = NULL;
412 static RRDDIM *rd_offline = NULL;
413 static RRDDIM *rd_waiting = NULL;
414 static RRDDIM *rd_replicating = NULL;
415 static RRDDIM *rd_running = NULL;
416 static RRDDIM *rd_no_dst = NULL;
417 static RRDDIM *rd_no_dst_failed = NULL;
418
419 if (unlikely(!st_nodes)) {
420 st_nodes = rrdset_create_localhost(
421 "netdata"
422 , "streaming_outbound"
423 , NULL
424 , "Streaming"
425 , "netdata.streaming_outbound"
426 , "Outbound Nodes"
427 , "nodes"
428 , "netdata"
429 , "pulse"
430 , 130153
431 , localhost->rrd_update_every
432 , RRDSET_TYPE_LINE
433 );
434
435 rd_connecting = rrddim_add(st_nodes, "connecting", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
436 rd_pending = rrddim_add(st_nodes, "pending", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
437 rd_offline = rrddim_add(st_nodes, "offline", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
438 rd_waiting = rrddim_add(st_nodes, "waiting", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
439 rd_replicating = rrddim_add(st_nodes, "replicating", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
440 rd_running = rrddim_add(st_nodes, "running", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
441 rd_no_dst = rrddim_add(st_nodes, "no dst", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
442 rd_no_dst_failed = rrddim_add(st_nodes, "failed", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
443 }
444
445 rrddim_set_by_pointer(st_nodes, rd_connecting, (collected_number)__atomic_load_n(&p.sender.nodes_connecting, __ATOMIC_RELAXED));
446 rrddim_set_by_pointer(st_nodes, rd_pending, (collected_number)__atomic_load_n(&p.sender.nodes_pending, __ATOMIC_RELAXED));
447 rrddim_set_by_pointer(st_nodes, rd_offline, (collected_number)__atomic_load_n(&p.sender.nodes_offline, __ATOMIC_RELAXED));
448 rrddim_set_by_pointer(st_nodes, rd_waiting, (collected_number)__atomic_load_n(&p.sender.nodes_waiting, __ATOMIC_RELAXED));
449 rrddim_set_by_pointer(st_nodes, rd_replicating, (collected_number)__atomic_load_n(&p.sender.nodes_replicating, __ATOMIC_RELAXED));
450 rrddim_set_by_pointer(st_nodes, rd_running, (collected_number)__atomic_load_n(&p.sender.nodes_running, __ATOMIC_RELAXED));
451 rrddim_set_by_pointer(st_nodes, rd_no_dst, (collected_number)__atomic_load_n(&p.sender.nodes_no_dst, __ATOMIC_RELAXED));
452 rrddim_set_by_pointer(st_nodes, rd_no_dst_failed, (collected_number)__atomic_load_n(&p.sender.nodes_no_dst_failed, __ATOMIC_RELAXED));
453
454 rrdset_done(st_nodes);
455 }
456
457 if(extended) {
458 chart_by_reason(
459 &p.sender.stream_info_failed_by_reason,
460 "streaming_info_failed_outbound",
461 "netdata.streaming_events_outbound",
462 "Outbound Streaming Events",
463 "stream-info",
464 130154);
465 chart_by_reason(
466 &p.sender.events_by_reason,
467 "streaming_rejections_outbound",
468 "netdata.streaming_events_outbound",
469 "Outbound Streaming Events",
470 "rejections",
471 130154);
472 chart_by_reason(
473 &p.sender.disconnects_by_reason,
474 "streaming_disconnects_outbound",
475 "netdata.streaming_events_outbound",
476 "Outbound Streaming Events",
477 "disconnects",
478 130154);
479 }
480 }
481 }