streaming nodes accounting (#19466)
* fix labels for pulse streaming * add max values to streaming function so that the range selectors work * remove the filtering from some fields * fix for zero first_time_s * add waiting replication state * fix ws_client logs to include ACLK, make the disconnection by the remote ws server to be reflected in aclk status
Costa Tsaousis committed
Jan 23, 2025 at 17:42 UTC
2dc54fa939200c754cf0e39b1361de0f59842e9a
8 files changed
+193
-97
src/aclk/mqtt_websockets/mqtt_wss_client.c
+3
@@ -898,6 +898,9 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
898
client->poll_fds[POLLFD_SOCKET].events |= POLLIN;
899
break;
900
901
+ case WS_CLIENT_CONNECTION_REMOTE_CLOSED:
902
+ return MQTT_WSS_ERR_REMOTE_CLOSED;
903
+
904
case WS_CLIENT_CONNECTION_CLOSED:
905
return MQTT_WSS_ERR_CONN_DROP;
906
src/aclk/mqtt_websockets/ws_client.c
+55
-38
@@ -75,13 +75,14 @@ void ws_client_reset(ws_client *client)
75
client->state = WS_RAW;
76
client->hs.hdr_state = WS_HDR_HTTP;
77
client->rx.parse_state = WS_FIRST_2BYTES;
78
+ client->rx.remote_closed = false;
79
}
80
81
#define MAX_HTTP_HDR_COUNT 128
82
int ws_client_add_http_header(ws_client *client, struct http_header *hdr)
83
{
84
if (client->hs.hdr_count > MAX_HTTP_HDR_COUNT) {
84
- nd_log(NDLS_DAEMON, NDLP_ERR, "Too many HTTP response header fields");
85
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Too many HTTP response header fields");
86
return -1;
87
}
88
@@ -114,8 +115,10 @@ int ws_client_start_handshake(ws_client *client)
115
const EVP_MD *md;
116
int rc = 1;
117
118
+ client->rx.remote_closed = false;
119
+
120
if(!client->host || !*client->host) {
118
- nd_log(NDLS_DAEMON, NDLP_ERR, "Hostname has not been set. We should not be able to come here!");
121
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Hostname has not been set. We should not be able to come here!");
122
return 1;
123
}
124
@@ -129,13 +132,13 @@ int ws_client_start_handshake(ws_client *client)
132
md_ctx = EVP_MD_CTX_new();
133
#endif
134
if (md_ctx == NULL) {
132
- nd_log(NDLS_DAEMON, NDLP_ERR, "Can't create EVP_MD context");
135
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Can't create EVP_MD context");
136
return 1;
137
}
138
139
md = EVP_sha1(); // Use SHA-1 for WebSocket handshake
140
if (!md) {
138
- nd_log(NDLS_DAEMON, NDLP_ERR, "Unknown message digest SHA-1");
141
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Unknown message digest SHA-1");
142
goto exit_with_error;
143
}
144
@@ -144,7 +147,7 @@ int ws_client_start_handshake(ws_client *client)
147
// Format and push the upgrade header to the write buffer
148
size_t bytes = snprintf(second, TEMP_BUF_SIZE, websocket_upgrage_hdr, *client->host, nonce_b64);
149
if(rbuf_bytes_free(client->buf_write) < bytes) {
147
- nd_log(NDLS_DAEMON, NDLP_ERR, "Write buffer capacity too low.");
150
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Write buffer capacity too low.");
151
goto exit_with_error;
152
}
153
rbuf_push(client->buf_write, second, bytes);
@@ -155,17 +158,17 @@ int ws_client_start_handshake(ws_client *client)
158
bytes = snprintf(second, TEMP_BUF_SIZE, "%s%s", nonce_b64, mqtt_protoid);
159
160
if (!EVP_DigestInit_ex(md_ctx, md, NULL)) {
158
- nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to initialize digest context");
161
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Failed to initialize digest context");
162
goto exit_with_error;
163
}
164
165
if (!EVP_DigestUpdate(md_ctx, second, bytes)) {
163
- nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to update digest");
166
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Failed to update digest");
167
goto exit_with_error;
168
}
169
170
if (!EVP_DigestFinal_ex(md_ctx, digest, &md_len)) {
168
- nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to finalize digest");
171
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Failed to finalize digest");
172
goto exit_with_error;
173
}
174
@@ -187,7 +190,7 @@ exit_with_error:
190
191
#define BUF_READ_MEMCMP_CONST(const, err) \
192
if (rbuf_memcmp_n(client->buf_read, const, strlen(const))) { \
190
- nd_log(NDLS_DAEMON, NDLP_ERR, err); \
193
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: %s", err); \
194
rbuf_flush(client->buf_read); \
195
return WS_CLIENT_PROTOCOL_ERROR; \
196
}
@@ -213,7 +216,7 @@ exit_with_error:
216
217
#define HTTP_HDR_LINE_CHECK_LIMIT(x) \
218
if ((x) >= MAX_HTTP_LINE_LENGTH) { \
216
- nd_log(NDLS_DAEMON, NDLP_ERR, "HTTP line received is too long. Maximum is %d", MAX_HTTP_LINE_LENGTH); \
219
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: HTTP line received is too long. Maximum is %d", MAX_HTTP_LINE_LENGTH); \
220
return WS_CLIENT_PROTOCOL_ERROR; \
221
}
222
@@ -236,13 +239,13 @@ int ws_client_parse_handshake_resp(ws_client *client)
239
BUF_READ_CHECK_AT_LEAST(HTTP_SC_LENGTH); // "XXX " http return code
240
rbuf_pop(client->buf_read, buf, HTTP_SC_LENGTH);
241
if (buf[HTTP_SC_LENGTH - 1] != 0x20) {
239
- nd_log(NDLS_DAEMON, NDLP_ERR, "HTTP status code received is not terminated by space (0x20)");
242
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: HTTP status code received is not terminated by space (0x20)");
243
return WS_CLIENT_PROTOCOL_ERROR;
244
}
245
buf[HTTP_SC_LENGTH - 1] = 0;
246
client->hs.http_code = atoi(buf);
247
if (client->hs.http_code < 100 || client->hs.http_code >= 600) {
245
- nd_log(NDLS_DAEMON, NDLP_ERR, "HTTP status code received not in valid range 100-600");
248
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: HTTP status code received not in valid range 100-600");
249
return WS_CLIENT_PROTOCOL_ERROR;
250
}
251
client->hs.hdr_state = WS_HDR_ENDLINE;
@@ -281,16 +284,16 @@ int ws_client_parse_handshake_resp(ws_client *client)
284
285
ptr = rbuf_find_bytes(client->buf_read, HTTP_HDR_SEPARATOR, strlen(HTTP_HDR_SEPARATOR), &idx_sep);
286
if (!ptr || idx_sep > idx_crlf) {
284
- nd_log(NDLS_DAEMON, NDLP_ERR, "Expected HTTP hdr field key/value separator \": \" before endline in non empty HTTP header line");
287
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Expected HTTP hdr field key/value separator \": \" before endline in non empty HTTP header line");
288
return WS_CLIENT_PROTOCOL_ERROR;
289
}
290
if (idx_crlf == idx_sep + (int)strlen(HTTP_HDR_SEPARATOR)) {
288
- nd_log(NDLS_DAEMON, NDLP_ERR, "HTTP Header value cannot be empty");
291
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: HTTP Header value cannot be empty");
292
return WS_CLIENT_PROTOCOL_ERROR;
293
}
294
295
if (idx_sep > HTTP_HEADER_NAME_MAX_LEN) {
293
- nd_log(NDLS_DAEMON, NDLP_ERR, "HTTP header too long (%d)", idx_sep);
296
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: HTTP header too long (%d)", idx_sep);
297
return WS_CLIENT_PROTOCOL_ERROR;
298
}
299
@@ -312,7 +315,7 @@ int ws_client_parse_handshake_resp(ws_client *client)
315
316
if (!strcmp(hdr->key, WS_CONN_ACCEPT)) {
317
if (strcmp(client->hs.nonce_reply, hdr->value)) {
315
- nd_log(NDLS_DAEMON, NDLP_ERR, "Received NONCE \"%s\" does not match expected nonce of \"%s\"", hdr->value, client->hs.nonce_reply);
318
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Received NONCE \"%s\" does not match expected nonce of \"%s\"", hdr->value, client->hs.nonce_reply);
319
return WS_CLIENT_PROTOCOL_ERROR;
320
}
321
client->hs.nonce_matched = 1;
@@ -322,21 +325,21 @@ int ws_client_parse_handshake_resp(ws_client *client)
325
326
case WS_HDR_PARSE_DONE:
327
if (!client->hs.nonce_matched) {
325
- nd_log(NDLS_DAEMON, NDLP_ERR, "Missing " WS_CONN_ACCEPT " header");
328
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Missing " WS_CONN_ACCEPT " header");
329
return WS_CLIENT_PROTOCOL_ERROR;
330
}
331
if (client->hs.http_code != 101) {
329
- nd_log(NDLS_DAEMON, NDLP_ERR, "HTTP return code not 101. Received %d with msg \"%s\".", client->hs.http_code, client->hs.http_reply_msg);
332
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: HTTP return code not 101. Received %d with msg \"%s\".", client->hs.http_code, client->hs.http_reply_msg);
333
return WS_CLIENT_PROTOCOL_ERROR;
334
}
335
336
client->state = WS_ESTABLISHED;
337
client->hs.hdr_state = WS_HDR_ALL_DONE;
335
- nd_log(NDLS_DAEMON, NDLP_INFO, "Websocket Connection Accepted By Server");
338
+ nd_log(NDLS_DAEMON, NDLP_INFO, "ACLK: Websocket Connection Accepted By Server");
339
return WS_CLIENT_PARSING_DONE;
340
341
case WS_HDR_ALL_DONE:
339
- nd_log(NDLS_DAEMON, NDLP_CRIT, "This is error we should never come here!");
342
+ nd_log(NDLS_DAEMON, NDLP_CRIT, "ACLK: This is error we should never come here!");
343
return WS_CLIENT_PROTOCOL_ERROR;
344
}
345
return 0;
@@ -437,13 +440,13 @@ static int check_opcode(enum websocket_opcode oc)
440
case WS_OP_PING:
441
return 0;
442
case WS_OP_CONTINUATION_FRAME:
440
- nd_log(NDLS_DAEMON, NDLP_ERR, "WS_OP_CONTINUATION_FRAME NOT IMPLEMENTED YET!!!!");
443
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: WS_OP_CONTINUATION_FRAME NOT IMPLEMENTED YET!!!!");
444
return 0;
445
case WS_OP_TEXT_FRAME:
443
- nd_log(NDLS_DAEMON, NDLP_ERR, "WS_OP_TEXT_FRAME NOT IMPLEMENTED YET!!!!");
446
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: WS_OP_TEXT_FRAME NOT IMPLEMENTED YET!!!!");
447
return 0;
448
case WS_OP_PONG:
446
- nd_log(NDLS_DAEMON, NDLP_ERR, "WS_OP_PONG NOT IMPLEMENTED YET!!!!");
449
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: WS_OP_PONG NOT IMPLEMENTED YET!!!!");
450
return 0;
451
default:
452
return WS_CLIENT_PROTOCOL_ERROR;
@@ -480,7 +483,7 @@ int ws_client_process_rx_ws(ws_client *client)
483
client->rx.opcode = buf[0] & (char)~BYTE_MSB;
484
485
if (!(buf[0] & (char)~WS_FINAL_FRAG)) {
483
- nd_log(NDLS_DAEMON, NDLP_ERR, "Not supporting fragmented messages yet!");
486
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Not supporting fragmented messages yet!");
487
return WS_CLIENT_PROTOCOL_ERROR;
488
}
489
@@ -488,7 +491,7 @@ int ws_client_process_rx_ws(ws_client *client)
491
return WS_CLIENT_PROTOCOL_ERROR;
492
493
if (buf[1] & (char)WS_PAYLOAD_MASKED) {
491
- nd_log(NDLS_DAEMON, NDLP_ERR, "Mask is not allowed in Server->Client Websocket direction.");
494
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Mask is not allowed in Server->Client Websocket direction.");
495
return WS_CLIENT_PROTOCOL_ERROR;
496
}
497
@@ -538,11 +541,12 @@ int ws_client_process_rx_ws(ws_client *client)
541
// b) 2byte reason code
542
// c) 2byte reason code followed by message
543
if (client->rx.payload_length == 1) {
541
- nd_log(NDLS_DAEMON, NDLP_ERR, "WebScoket CONNECTION_CLOSE can't have payload of size 1");
544
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: WebScoket CONNECTION_CLOSE can't have payload of size 1");
545
return WS_CLIENT_PROTOCOL_ERROR;
546
}
547
+ client->rx.remote_closed = true;
548
if (!client->rx.payload_length) {
545
- nd_log(NDLS_DAEMON, NDLP_INFO, "WebSocket server closed the connection without giving reason.");
549
+ nd_log(NDLS_DAEMON, NDLP_INFO, "ACLK: WebSocket server closed the connection without giving reason.");
550
client->rx.parse_state = WS_PACKET_DONE;
551
break;
552
}
@@ -555,8 +559,9 @@ int ws_client_process_rx_ws(ws_client *client)
559
client->rx.specific_data.op_close.ec = be16toh(*((uint16_t *)buf));
560
client->rx.payload_processed += sizeof(uint16_t);
561
562
+ client->rx.remote_closed = true;
563
if(client->rx.payload_processed == client->rx.payload_length) {
559
- nd_log(NDLS_DAEMON, NDLP_INFO, "WebSocket server closed the connection with EC=%d. Without message.",
564
+ nd_log(NDLS_DAEMON, NDLP_INFO, "ACLK: WebSocket server closed the connection with EC=%d. Without message.",
565
client->rx.specific_data.op_close.ec);
566
client->rx.parse_state = WS_PACKET_DONE;
567
break;
@@ -575,23 +580,24 @@ int ws_client_process_rx_ws(ws_client *client)
580
client->rx.payload_length - client->rx.payload_processed);
581
}
582
client->rx.specific_data.op_close.reason[client->rx.payload_length] = 0;
578
- nd_log(NDLS_DAEMON, NDLP_INFO, "WebSocket server closed the connection with EC=%d and reason \"%s\"",
583
+ nd_log(NDLS_DAEMON, NDLP_INFO, "ACLK: WebSocket server closed the connection with EC=%d and reason \"%s\"",
584
client->rx.specific_data.op_close.ec,
585
client->rx.specific_data.op_close.reason);
586
freez(client->rx.specific_data.op_close.reason);
587
+ client->rx.remote_closed = true;
588
client->rx.specific_data.op_close.reason = NULL;
589
client->rx.parse_state = WS_PACKET_DONE;
590
break;
591
case WS_PAYLOAD_SKIP_UNKNOWN_PAYLOAD:
592
BUF_READ_CHECK_AT_LEAST(client->rx.payload_length);
587
- nd_log(NDLS_DAEMON, NDLP_WARNING, "Skipping Websocket Packet of unsupported/unknown type");
593
+ nd_log(NDLS_DAEMON, NDLP_WARNING, "ACLK: Skipping Websocket Packet of unsupported/unknown type");
594
if (client->rx.payload_length)
595
rbuf_bump_tail(client->buf_read, client->rx.payload_length);
596
client->rx.parse_state = WS_PACKET_DONE;
597
return WS_CLIENT_PARSING_DONE;
598
case WS_PAYLOAD_PING_REQ_PAYLOAD:
599
if (client->rx.payload_length > rbuf_get_capacity(client->buf_read) / 2) {
594
- nd_log(NDLS_DAEMON, NDLP_ERR, "Ping arrived with payload which is too big!");
600
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Ping arrived with payload which is too big!");
601
return WS_CLIENT_INTERNAL_ERROR;
602
}
603
BUF_READ_CHECK_AT_LEAST(client->rx.payload_length);
@@ -601,7 +607,7 @@ int ws_client_process_rx_ws(ws_client *client)
607
// then attempt to send as soon as buffer space clears up
608
size = ws_client_send(client, WS_OP_PONG, client->rx.specific_data.ping_msg, client->rx.payload_length);
609
if (size != client->rx.payload_length) {
604
- nd_log(NDLS_DAEMON, NDLP_ERR, "Unable to send the PONG as one packet back. Closing connection.");
610
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Unable to send the PONG as one packet back. Closing connection.");
611
return WS_CLIENT_PROTOCOL_ERROR;
612
}
613
client->rx.parse_state = WS_PACKET_DONE;
@@ -609,11 +615,15 @@ int ws_client_process_rx_ws(ws_client *client)
615
case WS_PACKET_DONE:
616
client->rx.parse_state = WS_FIRST_2BYTES;
617
client->rx.payload_processed = 0;
612
- if (client->rx.opcode == WS_OP_CONNECTION_CLOSE)
613
- return WS_CLIENT_CONNECTION_CLOSED;
618
+ if (client->rx.opcode == WS_OP_CONNECTION_CLOSE) {
619
+ if(client->rx.remote_closed)
620
+ return WS_CLIENT_CONNECTION_REMOTE_CLOSED;
621
+ else
622
+ return WS_CLIENT_CONNECTION_CLOSED;
623
+ }
624
return WS_CLIENT_PARSING_DONE;
625
default:
616
- nd_log(NDLS_DAEMON, NDLP_ERR, "Unknown parse state");
626
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Unknown parse state");
627
return WS_CLIENT_INTERNAL_ERROR;
628
}
629
return 0;
@@ -646,6 +656,9 @@ int ws_client_process(ws_client *client)
656
case WS_CLIENT_PROTOCOL_ERROR:
657
client->state = WS_ERROR;
658
break;
659
+ case WS_CLIENT_CONNECTION_REMOTE_CLOSED:
660
+ client->state = WS_CONN_CLOSED_GRACEFUL_BY_REMOTE;
661
+ break;
662
case WS_CLIENT_CONNECTION_CLOSED:
663
client->state = WS_CONN_CLOSED_GRACEFUL;
664
break;
@@ -660,15 +673,19 @@ int ws_client_process(ws_client *client)
673
break;
674
case WS_ERROR:
675
worker_is_busy(WORKER_ACLK_PROCESS_ERROR);
663
- nd_log(NDLS_DAEMON, NDLP_ERR, "ws_client is in error state. Restart the connection!");
676
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: ws_client is in error state. Restart the connection!");
677
return WS_CLIENT_PROTOCOL_ERROR;
678
case WS_CONN_CLOSED_GRACEFUL:
679
worker_is_busy(WORKER_ACLK_PROCESS_CLOSED_GRACEFULLY);
667
- nd_log(NDLS_DAEMON, NDLP_ERR, "Connection has been gracefully closed. Calling this is useless (and probably bug) until you reconnect again.");
680
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Connection has been gracefully closed.");
681
return WS_CLIENT_CONNECTION_CLOSED;
682
+ case WS_CONN_CLOSED_GRACEFUL_BY_REMOTE:
683
+ worker_is_busy(WORKER_ACLK_PROCESS_CLOSED_GRACEFULLY);
684
+ nd_log(NDLS_DAEMON, NDLP_ERR, "ACLK: Connection has been gracefully closed by remote end.");
685
+ return WS_CLIENT_CONNECTION_REMOTE_CLOSED;
686
default:
687
worker_is_busy(WORKER_ACLK_PROCESS_UNKNOWN);
671
- nd_log(NDLS_DAEMON, NDLP_CRIT, "Unknown connection state! Probably memory corruption.");
688
+ nd_log(NDLS_DAEMON, NDLP_CRIT, "ACLK: Unknown connection state! Probably memory corruption.");
689
return WS_CLIENT_INTERNAL_ERROR;
690
}
691
return ret;
src/aclk/mqtt_websockets/ws_client.h
+10
-7
@@ -3,19 +3,21 @@
3
#ifndef WS_CLIENT_H
4
#define WS_CLIENT_H
5
6
-#define WS_CLIENT_NEED_MORE_BYTES 0x10
7
-#define WS_CLIENT_PARSING_DONE 0x11
8
-#define WS_CLIENT_CONNECTION_CLOSED 0x12
9
-#define WS_CLIENT_PROTOCOL_ERROR -0x10
10
-#define WS_CLIENT_BUFFER_FULL -0x11
11
-#define WS_CLIENT_INTERNAL_ERROR -0x12
6
+#define WS_CLIENT_NEED_MORE_BYTES 0x10
7
+#define WS_CLIENT_PARSING_DONE 0x11
8
+#define WS_CLIENT_CONNECTION_CLOSED 0x12
9
+#define WS_CLIENT_CONNECTION_REMOTE_CLOSED 0x13
10
+#define WS_CLIENT_PROTOCOL_ERROR -0x10
11
+#define WS_CLIENT_BUFFER_FULL -0x11
12
+#define WS_CLIENT_INTERNAL_ERROR -0x12
13
14
enum websocket_client_conn_state {
15
WS_RAW = 0,
16
WS_HANDSHAKE,
17
WS_ESTABLISHED,
18
WS_ERROR, // connection has to be restarted if this is reached
18
- WS_CONN_CLOSED_GRACEFUL
19
+ WS_CONN_CLOSED_GRACEFUL,
20
+ WS_CONN_CLOSED_GRACEFUL_BY_REMOTE,
21
};
22
23
enum websocket_client_hdr_parse_state {
@@ -77,6 +79,7 @@ typedef struct websocket_client {
79
struct ws_rx {
80
enum websocket_client_rx_ws_parse_state parse_state;
81
enum websocket_opcode opcode;
82
+ bool remote_closed;
83
uint64_t payload_length;
84
uint64_t payload_processed;
85
union {
src/daemon/pulse/pulse-parents.c
+12
-4
@@ -35,6 +35,7 @@ struct {
35
ssize_t nodes_offline;
36
ssize_t nodes_waiting;
37
ssize_t nodes_replicating;
38
+ ssize_t nodes_replication_waiting;
39
ssize_t nodes_running;
40
} parent;
41
@@ -137,6 +138,10 @@ static void pulse_host_add_sub_status(PULSE_HOST_STATUS status, ssize_t val, STR
138
reason = 0;
139
break;
140
141
+ case PULSE_HOST_STATUS_RCV_REPLICATION_WAIT:
142
+ __atomic_add_fetch(&p.parent.nodes_replication_waiting, val, __ATOMIC_RELAXED);
143
+ break;
144
+
145
case PULSE_HOST_STATUS_RCV_REPLICATING:
146
__atomic_add_fetch(&p.parent.nodes_replicating, val, __ATOMIC_RELAXED);
147
break;
@@ -191,7 +196,7 @@ void pulse_host_status(RRDHOST *host, PULSE_HOST_STATUS status, STREAM_HANDSHAKE
196
status = pulse_host_detect_receiver_status(host);
197
198
PULSE_HOST_STATUS basic = PULSE_HOST_STATUS_LOCAL|PULSE_HOST_STATUS_VIRTUAL| PULSE_HOST_STATUS_LOADING |PULSE_HOST_STATUS_ARCHIVED|PULSE_HOST_STATUS_DELETED;
194
- PULSE_HOST_STATUS rcv = PULSE_HOST_STATUS_RCV_OFFLINE|PULSE_HOST_STATUS_RCV_WAITING|PULSE_HOST_STATUS_RCV_REPLICATING|PULSE_HOST_STATUS_RCV_RUNNING;
199
+ PULSE_HOST_STATUS rcv = PULSE_HOST_STATUS_RCV_OFFLINE|PULSE_HOST_STATUS_RCV_WAITING|PULSE_HOST_STATUS_RCV_REPLICATING|PULSE_HOST_STATUS_RCV_REPLICATION_WAIT|PULSE_HOST_STATUS_RCV_RUNNING;
200
PULSE_HOST_STATUS snd = PULSE_HOST_STATUS_SND_OFFLINE|PULSE_HOST_STATUS_SND_PENDING|PULSE_HOST_STATUS_SND_CONNECTING|PULSE_HOST_STATUS_SND_WAITING|PULSE_HOST_STATUS_SND_REPLICATING|PULSE_HOST_STATUS_SND_RUNNING|PULSE_HOST_STATUS_SND_NO_DST;
201
202
if(status & basic)
@@ -254,7 +259,7 @@ static void chart_by_reason(struct by_reason *b, const char *id, const char *con
259
, "Streaming"
260
, context
261
, title
257
- , "events"
262
+ , "events/s"
263
, "netdata"
264
, "pulse"
265
, priority
@@ -294,6 +299,7 @@ void pulse_parents_do(bool extended) {
299
static RRDDIM *rd_archived = NULL;
300
static RRDDIM *rd_offline = NULL;
301
static RRDDIM *rd_waiting = NULL;
302
+ static RRDDIM *rd_replication_waiting = NULL;
303
static RRDDIM *rd_replicating = NULL;
304
static RRDDIM *rd_running = NULL;
305
@@ -316,9 +322,10 @@ void pulse_parents_do(bool extended) {
322
rd_local = rrddim_add(st_nodes, "local", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
323
rd_virtual = rrddim_add(st_nodes, "virtual", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
324
rd_loading = rrddim_add(st_nodes, "loading", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
319
- rd_archived = rrddim_add(st_nodes, "stale", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
320
- rd_offline = rrddim_add(st_nodes, "offline", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
325
+ rd_archived = rrddim_add(st_nodes, "stale archived", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
326
+ rd_offline = rrddim_add(st_nodes, "stale disconnected", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
327
rd_waiting = rrddim_add(st_nodes, "waiting", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
328
+ rd_replication_waiting = rrddim_add(st_nodes, "waiting replication", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
329
rd_replicating = rrddim_add(st_nodes, "replicating", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
330
rd_running = rrddim_add(st_nodes, "running", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
331
}
@@ -329,6 +336,7 @@ void pulse_parents_do(bool extended) {
336
rrddim_set_by_pointer(st_nodes, rd_archived, (collected_number)__atomic_load_n(&p.parent.nodes_archived, __ATOMIC_RELAXED));
337
rrddim_set_by_pointer(st_nodes, rd_offline, (collected_number)__atomic_load_n(&p.parent.nodes_offline, __ATOMIC_RELAXED));
338
rrddim_set_by_pointer(st_nodes, rd_waiting, (collected_number)__atomic_load_n(&p.parent.nodes_waiting, __ATOMIC_RELAXED));
339
+ rrddim_set_by_pointer(st_nodes, rd_replication_waiting, (collected_number)__atomic_load_n(&p.parent.nodes_replication_waiting, __ATOMIC_RELAXED));
340
rrddim_set_by_pointer(st_nodes, rd_replicating, (collected_number)__atomic_load_n(&p.parent.nodes_replicating, __ATOMIC_RELAXED));
341
rrddim_set_by_pointer(st_nodes, rd_running, (collected_number)__atomic_load_n(&p.parent.nodes_running, __ATOMIC_RELAXED));
342
src/daemon/pulse/pulse-parents.h
+18
-17
@@ -7,23 +7,24 @@
7
#include "streaming/stream-handshake.h"
8
9
typedef enum {
10
- PULSE_HOST_STATUS_NONE = 0,
11
- PULSE_HOST_STATUS_LOCAL = (1 << 0),
12
- PULSE_HOST_STATUS_VIRTUAL = (1 << 1),
13
- PULSE_HOST_STATUS_LOADING = (1 << 2),
14
- PULSE_HOST_STATUS_ARCHIVED = (1 << 3),
15
- PULSE_HOST_STATUS_RCV_OFFLINE = (1 << 4),
16
- PULSE_HOST_STATUS_RCV_WAITING = (1 << 5),
17
- PULSE_HOST_STATUS_RCV_REPLICATING = (1 << 6),
18
- PULSE_HOST_STATUS_RCV_RUNNING = (1 << 7),
19
- PULSE_HOST_STATUS_SND_OFFLINE = (1 << 8),
20
- PULSE_HOST_STATUS_SND_PENDING = (1 << 9),
21
- PULSE_HOST_STATUS_SND_CONNECTING = (1 << 10),
22
- PULSE_HOST_STATUS_SND_NO_DST = (1 << 11),
23
- PULSE_HOST_STATUS_SND_WAITING = (1 << 12),
24
- PULSE_HOST_STATUS_SND_REPLICATING = (1 << 13),
25
- PULSE_HOST_STATUS_SND_RUNNING = (1 << 14),
26
- PULSE_HOST_STATUS_DELETED = (1 << 15),
10
+ PULSE_HOST_STATUS_NONE = 0,
11
+ PULSE_HOST_STATUS_LOCAL = (1 << 0),
12
+ PULSE_HOST_STATUS_VIRTUAL = (1 << 1),
13
+ PULSE_HOST_STATUS_LOADING = (1 << 2),
14
+ PULSE_HOST_STATUS_ARCHIVED = (1 << 3),
15
+ PULSE_HOST_STATUS_RCV_OFFLINE = (1 << 4),
16
+ PULSE_HOST_STATUS_RCV_WAITING = (1 << 5),
17
+ PULSE_HOST_STATUS_RCV_REPLICATING = (1 << 6),
18
+ PULSE_HOST_STATUS_RCV_REPLICATION_WAIT = (1 << 7),
19
+ PULSE_HOST_STATUS_RCV_RUNNING = (1 << 8),
20
+ PULSE_HOST_STATUS_SND_OFFLINE = (1 << 9),
21
+ PULSE_HOST_STATUS_SND_PENDING = (1 << 10),
22
+ PULSE_HOST_STATUS_SND_CONNECTING = (1 << 11),
23
+ PULSE_HOST_STATUS_SND_NO_DST = (1 << 12),
24
+ PULSE_HOST_STATUS_SND_WAITING = (1 << 13),
25
+ PULSE_HOST_STATUS_SND_REPLICATING = (1 << 14),
26
+ PULSE_HOST_STATUS_SND_RUNNING = (1 << 15),
27
+ PULSE_HOST_STATUS_DELETED = (1 << 16),
28
} PULSE_HOST_STATUS;
29
30
void pulse_host_status(RRDHOST *host, PULSE_HOST_STATUS status, STREAM_HANDSHAKE reason);
src/database/contexts/worker.c
+1
-1
@@ -113,7 +113,7 @@ void rrdcontext_recalculate_host_retention(RRDHOST *host, RRD_FLAGS reason, bool
113
dfe_start_read(host->rrdctx.contexts, rc) {
114
rrdcontext_recalculate_context_retention(rc, reason, worker_jobs);
115
116
- if(!first_time_s || rc->first_time_s < first_time_s)
116
+ if(!first_time_s || (rc->first_time_s && rc->first_time_s < first_time_s))
117
first_time_s = rc->first_time_s;
118
119
if(!last_time_s || rc->last_time_s > last_time_s)
src/streaming/stream-receiver.c
+4
-1
@@ -491,7 +491,10 @@ void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct re
491
parser->h2o_ctx = rpt->h2o_ctx;
492
#endif
493
494
- pulse_host_status(rpt->host, PULSE_HOST_STATUS_RCV_RUNNING, 0);
494
+ if(stream_receive.replication.enabled)
495
+ pulse_host_status(rpt->host, PULSE_HOST_STATUS_RCV_REPLICATION_WAIT, 0);
496
+ else
497
+ pulse_host_status(rpt->host, PULSE_HOST_STATUS_RCV_RUNNING, 0);
498
499
// keep this last - it needs everything ready since to sends data to the child
500
stream_receiver_send_node_and_claim_id_to_child(rpt->host);
src/web/api/functions/function-streaming.c
+90
-29
@@ -22,6 +22,30 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
22
size_t max_db_metrics = 0, max_db_instances = 0, max_db_contexts = 0;
23
size_t max_collection_replication_instances = 0, max_streaming_replication_instances = 0;
24
size_t max_ml_anomalous = 0, max_ml_normal = 0, max_ml_trained = 0, max_ml_pending = 0, max_ml_silenced = 0;
25
+
26
+ time_t
27
+ max_db_duration,
28
+ max_db_from,
29
+ max_db_to,
30
+ max_in_age,
31
+ max_out_age,
32
+ max_out_attempt_age;
33
+
34
+ uint64_t
35
+ max_in_since,
36
+ max_out_since,
37
+ max_out_attempt_since;
38
+
39
+ int16_t
40
+ max_in_hops,
41
+ max_out_hops;
42
+
43
+ int
44
+ max_in_local_port,
45
+ max_in_remote_port,
46
+ max_out_local_port,
47
+ max_out_remote_port;
48
+
49
{
50
RRDHOST *host;
51
dfe_start_read(rrdhost_root_index, host) {
@@ -53,11 +77,18 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
77
78
// retention
79
buffer_json_add_array_item_string(wb, rrdhost_hostname(s.host)); // Node
80
+
81
buffer_json_add_array_item_uint64(wb, s.db.first_time_s * MSEC_PER_SEC); // dbFrom
82
+ if(s.db.first_time_s > max_db_from) max_db_from = s.db.first_time_s;
83
+
84
buffer_json_add_array_item_uint64(wb, s.db.last_time_s * MSEC_PER_SEC); // dbTo
85
+ if(s.db.last_time_s > max_db_to) max_db_to = s.db.last_time_s;
86
59
- if(s.db.first_time_s && s.db.last_time_s && s.db.last_time_s > s.db.first_time_s)
60
- buffer_json_add_array_item_uint64(wb, s.db.last_time_s - s.db.first_time_s); // dbDuration
87
+ if(s.db.first_time_s && s.db.last_time_s && s.db.last_time_s > s.db.first_time_s) {
88
+ time_t db_duration = s.db.last_time_s - s.db.first_time_s;
89
+ buffer_json_add_array_item_uint64(wb, db_duration); // dbDuration
90
+ if(db_duration > max_db_duration) max_db_duration = db_duration;
91
+ }
92
else
93
buffer_json_add_array_item_string(wb, NULL); // dbDuration
94
@@ -72,21 +103,34 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
103
104
// collection
105
if(s.ingest.since) {
75
- buffer_json_add_array_item_uint64(wb, s.ingest.since * MSEC_PER_SEC); // InSince
76
- buffer_json_add_array_item_time_t(wb, s.now - s.ingest.since); // InAge
106
+ uint64_t in_since = s.ingest.since * MSEC_PER_SEC;
107
+ buffer_json_add_array_item_uint64(wb, in_since); // InSince
108
+ if(in_since > max_in_since) max_in_since = in_since;
109
+
110
+ time_t in_age = s.now - s.ingest.since;
111
+ buffer_json_add_array_item_time_t(wb, in_age); // InAge
112
+ if(in_age > max_in_age) max_in_age = in_age;
113
}
114
else {
115
buffer_json_add_array_item_string(wb, NULL); // InSince
116
buffer_json_add_array_item_string(wb, NULL); // InAge
117
}
118
buffer_json_add_array_item_string(wb, stream_handshake_error_to_string(s.ingest.reason)); // InReason
119
+
120
buffer_json_add_array_item_int64(wb, s.ingest.hops); // InHops
121
+ if(s.ingest.hops > max_in_hops) max_in_hops = s.ingest.hops;
122
+
123
buffer_json_add_array_item_double(wb, s.ingest.replication.completion); // InReplCompletion
124
buffer_json_add_array_item_uint64(wb, s.ingest.replication.instances); // InReplInstances
125
buffer_json_add_array_item_string(wb, s.ingest.peers.local.ip); // InLocalIP
126
+
127
buffer_json_add_array_item_uint64(wb, s.ingest.peers.local.port); // InLocalPort
128
+ if(s.ingest.peers.local.port > max_in_local_port) max_in_local_port = s.ingest.peers.local.port;
129
+
130
buffer_json_add_array_item_string(wb, s.ingest.peers.peer.ip); // InRemoteIP
131
buffer_json_add_array_item_uint64(wb, s.ingest.peers.peer.port); // InRemotePort
132
+ if(s.ingest.peers.peer.port > max_in_remote_port) max_in_remote_port = s.ingest.peers.peer.port;
133
+
134
buffer_json_add_array_item_string(wb, s.ingest.ssl ? "SSL" : "PLAIN"); // InSSL
135
stream_capabilities_to_json_array(wb, s.ingest.capabilities, NULL); // InCapabilities
136
@@ -96,21 +140,33 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
140
141
// streaming
142
if(s.stream.since) {
99
- buffer_json_add_array_item_uint64(wb, s.stream.since * MSEC_PER_SEC); // OutSince
100
- buffer_json_add_array_item_time_t(wb, s.now - s.stream.since); // OutAge
143
+ uint64_t out_since = s.stream.since * MSEC_PER_SEC;
144
+ buffer_json_add_array_item_uint64(wb, out_since); // OutSince
145
+ if(out_since > max_out_since) max_out_since = out_since;
146
+
147
+ time_t out_age = s.now - s.stream.since;
148
+ buffer_json_add_array_item_time_t(wb, out_age); // OutAge
149
+ if(out_age > max_out_age) max_out_age = out_age;
150
}
151
else {
152
buffer_json_add_array_item_string(wb, NULL); // OutSince
153
buffer_json_add_array_item_string(wb, NULL); // OutAge
154
}
155
buffer_json_add_array_item_string(wb, stream_handshake_error_to_string(s.stream.reason)); // OutReason
156
+
157
buffer_json_add_array_item_uint64(wb, s.stream.hops); // OutHops
158
+ if(s.stream.hops > max_out_hops) max_out_hops = s.stream.hops;
159
+
160
buffer_json_add_array_item_double(wb, s.stream.replication.completion); // OutReplCompletion
161
buffer_json_add_array_item_uint64(wb, s.stream.replication.instances); // OutReplInstances
162
buffer_json_add_array_item_string(wb, s.stream.peers.local.ip); // OutLocalIP
163
buffer_json_add_array_item_uint64(wb, s.stream.peers.local.port); // OutLocalPort
164
+ if(s.stream.peers.local.port > max_out_local_port) max_out_local_port = s.stream.peers.local.port;
165
+
166
buffer_json_add_array_item_string(wb, s.stream.peers.peer.ip); // OutRemoteIP
167
buffer_json_add_array_item_uint64(wb, s.stream.peers.peer.port); // OutRemotePort
168
+ if(s.stream.peers.peer.port > max_out_remote_port) max_out_remote_port = s.stream.peers.peer.port;
169
+
170
buffer_json_add_array_item_string(wb, s.stream.ssl ? "SSL" : "PLAIN"); // OutSSL
171
buffer_json_add_array_item_string(wb, s.stream.compression ? "COMPRESSED" : "UNCOMPRESSED"); // OutCompression
172
stream_capabilities_to_json_array(wb, s.stream.capabilities, NULL); // OutCapabilities
@@ -128,8 +184,13 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
184
buffer_json_add_array_item_string(wb, NULL); // OutAttemptAge
185
}
186
else {
131
- buffer_json_add_array_item_uint64(wb, last_attempt / USEC_PER_MS); // OutAttemptSince
132
- buffer_json_add_array_item_time_t(wb, s.now - (time_t)(last_attempt / USEC_PER_SEC)); // OutAttemptAge
187
+ uint64_t out_attempt_since = last_attempt / USEC_PER_MS;
188
+ buffer_json_add_array_item_uint64(wb, out_attempt_since); // OutAttemptSince
189
+ if(out_attempt_since > max_out_attempt_since) max_out_attempt_since = out_attempt_since;
190
+
191
+ time_t out_attempt_age = s.now - (time_t)(last_attempt / USEC_PER_SEC);
192
+ buffer_json_add_array_item_time_t(wb, out_attempt_age); // OutAttemptAge
193
+ if(out_attempt_age > max_out_attempt_age) max_out_attempt_age = out_attempt_age;
194
}
195
196
// ML
@@ -184,20 +245,20 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
245
246
buffer_rrdf_table_add_field(wb, field_id++, "dbFrom", "DB Data Retention From",
247
RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
187
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
188
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
248
+ 0, NULL, (double)max_db_from * MSEC_PER_SEC, RRDF_FIELD_SORT_ASCENDING, NULL,
249
+ RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_NONE,
250
RRDF_FIELD_OPTS_NONE, NULL);
251
252
buffer_rrdf_table_add_field(wb, field_id++, "dbTo", "DB Data Retention To",
253
RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
193
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
194
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
254
+ 0, NULL, (double)max_db_to * MSEC_PER_SEC, RRDF_FIELD_SORT_ASCENDING, NULL,
255
+ RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_NONE,
256
RRDF_FIELD_OPTS_NONE, NULL);
257
258
buffer_rrdf_table_add_field(wb, field_id++, "dbDuration", "DB Data Retention Duration",
259
RRDF_FIELD_TYPE_DURATION, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DURATION_S,
199
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
200
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
260
+ 0, NULL, (double)max_db_duration, RRDF_FIELD_SORT_ASCENDING, NULL,
261
+ RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_NONE,
262
RRDF_FIELD_OPTS_VISIBLE, NULL);
263
264
buffer_rrdf_table_add_field(wb, field_id++, "dbMetrics", "Time-series Metrics in the DB",
@@ -243,13 +304,13 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
304
305
buffer_rrdf_table_add_field(wb, field_id++, "InSince", "Last Data Collection Status Change",
306
RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
246
- 0, NULL, NAN, RRDF_FIELD_SORT_DESCENDING, NULL,
247
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
307
+ 0, NULL, (double)max_in_since, RRDF_FIELD_SORT_DESCENDING, NULL,
308
+ RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_NONE,
309
RRDF_FIELD_OPTS_NONE, NULL);
310
311
buffer_rrdf_table_add_field(wb, field_id++, "InAge", "Last Data Collection Online Status Change Age",
312
RRDF_FIELD_TYPE_DURATION, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DURATION_S,
252
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
313
+ 0, NULL, (double)max_in_age, RRDF_FIELD_SORT_ASCENDING, NULL,
314
RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
315
RRDF_FIELD_OPTS_VISIBLE, NULL);
316
@@ -261,7 +322,7 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
322
323
buffer_rrdf_table_add_field(wb, field_id++, "InHops", "Data Collection Distance Hops from Origin Node",
324
RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
264
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
325
+ 0, NULL, (double)max_in_hops, RRDF_FIELD_SORT_ASCENDING, NULL,
326
RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
327
RRDF_FIELD_OPTS_VISIBLE, NULL);
328
@@ -286,7 +347,7 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
347
348
buffer_rrdf_table_add_field(wb, field_id++, "InLocalPort", "Inbound Local Port",
349
RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
289
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
350
+ 0, NULL, (double)max_in_local_port, RRDF_FIELD_SORT_ASCENDING, NULL,
351
RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_RANGE,
352
RRDF_FIELD_OPTS_NONE, NULL);
353
@@ -298,7 +359,7 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
359
360
buffer_rrdf_table_add_field(wb, field_id++, "InRemotePort", "Inbound Remote Port",
361
RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
301
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
362
+ 0, NULL, (double)max_in_remote_port, RRDF_FIELD_SORT_ASCENDING, NULL,
363
RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_RANGE,
364
RRDF_FIELD_OPTS_NONE, NULL);
365
@@ -336,13 +397,13 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
397
398
buffer_rrdf_table_add_field(wb, field_id++, "OutSince", "Last Streaming Status Change",
399
RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
339
- 0, NULL, NAN, RRDF_FIELD_SORT_DESCENDING, NULL,
340
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
400
+ 0, NULL, (double)max_out_since, RRDF_FIELD_SORT_DESCENDING, NULL,
401
+ RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_NONE,
402
RRDF_FIELD_OPTS_NONE, NULL);
403
404
buffer_rrdf_table_add_field(wb, field_id++, "OutAge", "Last Streaming Status Change Age",
405
RRDF_FIELD_TYPE_DURATION, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DURATION_S,
345
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
406
+ 0, NULL, (double)max_out_age, RRDF_FIELD_SORT_ASCENDING, NULL,
407
RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
408
RRDF_FIELD_OPTS_VISIBLE, NULL);
409
@@ -354,7 +415,7 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
415
416
buffer_rrdf_table_add_field(wb, field_id++, "OutHops", "Streaming Distance Hops from Origin Node",
417
RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
357
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
418
+ 0, NULL, (double)max_out_hops, RRDF_FIELD_SORT_ASCENDING, NULL,
419
RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
420
RRDF_FIELD_OPTS_VISIBLE, NULL);
421
@@ -380,7 +441,7 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
441
buffer_rrdf_table_add_field(wb, field_id++, "OutLocalPort", "Outbound Local Port",
442
RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
443
0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
383
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_RANGE,
444
+ RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_NONE,
445
RRDF_FIELD_OPTS_NONE, NULL);
446
447
buffer_rrdf_table_add_field(wb, field_id++, "OutRemoteIP", "Outbound Remote IP",
@@ -391,7 +452,7 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
452
453
buffer_rrdf_table_add_field(wb, field_id++, "OutRemotePort", "Outbound Remote Port",
454
RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
394
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
455
+ 0, NULL, (double)max_out_remote_port, RRDF_FIELD_SORT_ASCENDING, NULL,
456
RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_RANGE,
457
RRDF_FIELD_OPTS_NONE, NULL);
458
@@ -454,14 +515,14 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
515
buffer_rrdf_table_add_field(wb, field_id++, "OutAttemptSince",
516
"Last Outbound Connection Attempt Status Change Time",
517
RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
457
- 0, NULL, NAN, RRDF_FIELD_SORT_DESCENDING, NULL,
458
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
518
+ 0, NULL, (double)max_out_attempt_since, RRDF_FIELD_SORT_DESCENDING, NULL,
519
+ RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_NONE,
520
RRDF_FIELD_OPTS_NONE, NULL);
521
522
buffer_rrdf_table_add_field(wb, field_id++, "OutAttemptAge",
523
"Last Outbound Connection Attempt Status Change Age",
524
RRDF_FIELD_TYPE_DURATION, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DURATION_S,
464
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
525
+ 0, NULL, (double)max_out_attempt_age, RRDF_FIELD_SORT_ASCENDING, NULL,
526
RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
527
RRDF_FIELD_OPTS_VISIBLE, NULL);
528