Stream with version (#7851)
* stream_forward: Fix protocol This commit brings the necessary fixes to the protocol * stream_forward: Fix old slave support This commit fixes the communication with old versions of Netdata * stream_forward: Remove declaration There was a wrong declaration inside a block, so I am removing it * stream_forward: USe version This commit brings the use of version instead flags to stream * stream_forward: Remove variable This commit removes useless variable from hand shake * stream_forward: Change message Change the message setting the protocol version on it * stream_forward: Fix version number * stream_forward: readable definition The definition and the variables were using the same data type, but with different declaration, this commit fixes this. * stream_forward: Set master version inside message This commit updates the message used that there was a successfull connection with master * stream_forward: FIx wrong version This commit fixes the multiple set for stream version * stream_forward: Reorganize code This commit reorganizes code to speed up the processing * stream_forward: Adjust code This commit removes an unecessary else * stream_forward: Brings old structure This commits returns a previous necessary to the code * stream_forward: fix error report This commit fixes the error report that was happening when the stream version does not match * stream_forward: Fixes msg and remove unecessary call
thiagoftsm committed
Feb 5, 2020 at 20:13 UTC
d9f977b864a8abc2213201817a93a041c30d17b4
4 files changed
+90
-46
database/rrd.h
+1
@@ -702,6 +702,7 @@ struct rrdhost {
702
int rrdpush_sender_pipe[2]; // collector to sender thread signaling
703
BUFFER *rrdpush_sender_buffer; // collector fills it, sender sends it
704
705
+ uint32_t stream_version; //Set the current version of the stream.
706
707
// ------------------------------------------------------------------------
708
// streaming of data from remote hosts - rrdpush
database/rrdhost.c
+6
-1
@@ -147,6 +147,8 @@ RRDHOST *rrdhost_create(const char *hostname,
147
host->rrdpush_sender_pipe[0] = -1;
148
host->rrdpush_sender_pipe[1] = -1;
149
host->rrdpush_sender_socket = -1;
150
+
151
+ host->stream_version = STREAMING_PROTOCOL_CURRENT_VERSION;
152
#ifdef ENABLE_HTTPS
153
host->ssl.conn = NULL;
154
host->ssl.flags = NETDATA_SSL_START;
@@ -405,6 +407,7 @@ RRDHOST *rrdhost_find_or_create(
407
}
408
else {
409
host->health_enabled = health_enabled;
410
+ host->stream_version = STREAMING_PROTOCOL_CURRENT_VERSION;
411
412
if(strcmp(host->hostname, hostname) != 0) {
413
info("Host '%s' has been renamed to '%s'. If this is not intentional it may mean multiple hosts are using the same machine_guid.", host->hostname, hostname);
@@ -1309,7 +1312,9 @@ restart_after_removal:
1312
int rrdhost_set_system_info_variable(struct rrdhost_system_info *system_info, char *name, char *value) {
1313
int res = 0;
1314
1312
- if(!strcmp(name, "NETDATA_CONTAINER_OS_NAME")){
1315
+ if (!strcmp(name, "NETDATA_PROTOCOL_VERSION"))
1316
+ return res;
1317
+ else if(!strcmp(name, "NETDATA_CONTAINER_OS_NAME")){
1318
freez(system_info->container_os_name);
1319
system_info->container_os_name = strdupz(value);
1320
}
streaming/rrdpush.c
+81
-45
@@ -28,6 +28,7 @@
28
#define STREAMING_PROTOCOL_VERSION "1.1"
29
#define START_STREAMING_PROMPT "Hit me baby, push them over..."
30
#define START_STREAMING_PROMPT_V2 "Hit me baby, push them over and bring the host labels..."
31
+#define START_STREAMING_PROMPT_VN "Hit me baby, push them over with the version="
32
33
typedef enum {
34
RRDPUSH_MULTIPLE_CONNECTIONS_ALLOW,
@@ -496,6 +497,11 @@ static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) {
497
}
498
}
499
500
+static inline void rrdpush_set_flags_to_newest_stream(RRDHOST *host) {
501
+ host->labels_flag |= LABEL_FLAG_UPDATE_STREAM;
502
+ host->labels_flag &= ~LABEL_FLAG_STOP_STREAM;
503
+}
504
+
505
//called from client side
506
static int rrdpush_sender_thread_connect_to_master(RRDHOST *host, int default_port, int timeout, size_t *reconnects_counter, char *connected_to, size_t connected_to_size) {
507
struct timeval tv = {
@@ -560,7 +566,7 @@ static int rrdpush_sender_thread_connect_to_master(RRDHOST *host, int default_po
566
#define HTTP_HEADER_SIZE 8192
567
char http[HTTP_HEADER_SIZE + 1];
568
int eol = snprintfz(http, HTTP_HEADER_SIZE,
563
- "STREAM key=%s&hostname=%s®istry_hostname=%s&machine_guid=%s&update_every=%d&os=%s&timezone=%s&tags=%s"
569
+ "STREAM key=%s&hostname=%s®istry_hostname=%s&machine_guid=%s&update_every=%d&os=%s&timezone=%s&tags=%s&ver=%u"
570
"&NETDATA_SYSTEM_OS_NAME=%s"
571
"&NETDATA_SYSTEM_OS_ID=%s"
572
"&NETDATA_SYSTEM_OS_ID_LIKE=%s"
@@ -586,6 +592,7 @@ static int rrdpush_sender_thread_connect_to_master(RRDHOST *host, int default_po
592
, host->os
593
, host->timezone
594
, (host->tags) ? host->tags : ""
595
+ , STREAMING_PROTOCOL_CURRENT_VERSION
596
, (host->system_info->host_os_name) ? host->system_info->host_os_name : ""
597
, (host->system_info->host_os_id) ? host->system_info->host_os_id : ""
598
, (host->system_info->host_os_id_like) ? host->system_info->host_os_id_like : ""
@@ -643,26 +650,44 @@ static int rrdpush_sender_thread_connect_to_master(RRDHOST *host, int default_po
650
651
info("STREAM %s [send to %s]: waiting response from remote netdata...", host->hostname, connected_to);
652
653
+ ssize_t received;
654
#ifdef ENABLE_HTTPS
647
- if(recv_timeout(&host->ssl,host->rrdpush_sender_socket, http, HTTP_HEADER_SIZE, 0, timeout) == -1) {
655
+ received = recv_timeout(&host->ssl,host->rrdpush_sender_socket, http, HTTP_HEADER_SIZE, 0, timeout);
656
+ if(received == -1) {
657
#else
649
- if(recv_timeout(host->rrdpush_sender_socket, http, HTTP_HEADER_SIZE, 0, timeout) == -1) {
658
+ received = recv_timeout(host->rrdpush_sender_socket, http, HTTP_HEADER_SIZE, 0, timeout);
659
+ if(received == -1) {
660
#endif
661
error("STREAM %s [send to %s]: remote netdata does not respond.", host->hostname, connected_to);
662
rrdpush_sender_thread_close_socket(host);
663
return 0;
664
}
665
656
- int answer = strncmp(http, START_STREAMING_PROMPT_V2, strlen(START_STREAMING_PROMPT_V2));
657
- if(!answer) {
658
- host->labels_flag |= LABEL_FLAG_UPDATE_STREAM;
659
- host->labels_flag &= ~LABEL_FLAG_STOP_STREAM;
666
+ http[received] = '\0';
667
+ int answer = -1;
668
+ char *version_start = strchr(http, '=');
669
+ uint32_t version;
670
+ if(version_start) {
671
+ version_start++;
672
+ version = (uint32_t)strtol(version_start, NULL, 10);
673
+ answer = memcmp(http, START_STREAMING_PROMPT_VN, (size_t)(version_start - http));
674
+ if(!answer) {
675
+ rrdpush_set_flags_to_newest_stream(host);
676
+ host->stream_version = version;
677
+ }
678
} else {
661
- answer = strncmp(http, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT));
679
+ answer = memcmp(http, START_STREAMING_PROMPT_V2, strlen(START_STREAMING_PROMPT_V2));
680
if(!answer) {
663
- host->labels_flag |= LABEL_FLAG_STOP_STREAM;
664
- host->labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;
665
- info("STREAM %s [send to %s]: is using an old Netdata.", host->hostname, connected_to);
681
+ version = 1;
682
+ rrdpush_set_flags_to_newest_stream(host);
683
+ }
684
+ else {
685
+ answer = memcmp(http, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT));
686
+ if(!answer) {
687
+ version = 0;
688
+ host->labels_flag |= LABEL_FLAG_STOP_STREAM;
689
+ host->labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;
690
+ }
691
}
692
}
693
@@ -672,7 +697,10 @@ static int rrdpush_sender_thread_connect_to_master(RRDHOST *host, int default_po
697
return 0;
698
}
699
675
- info("STREAM %s [send to %s]: established communication - ready to send metrics...", host->hostname, connected_to);
700
+ info("STREAM %s [send to %s]: established communication with a master using protocol version %u - ready to send metrics..."
701
+ , host->hostname
702
+ , connected_to
703
+ , version);
704
705
if(sock_setnonblock(host->rrdpush_sender_socket) < 0)
706
error("STREAM %s [send to %s]: cannot set non-blocking mode for socket.", host->hostname, connected_to);
@@ -1036,7 +1064,7 @@ static int rrdpush_receive(int fd
1064
, int update_every
1065
, char *client_ip
1066
, char *client_port
1039
- , int stream_flags
1067
+ , uint32_t stream_version
1068
#ifdef ENABLE_HTTPS
1069
, struct netdata_ssl *ssl
1070
#endif
@@ -1153,15 +1181,18 @@ static int rrdpush_receive(int fd
1181
snprintfz(cd.cmd, PLUGINSD_CMD_MAX, "%s:%s", client_ip, client_port);
1182
1183
info("STREAM %s [receive from [%s]:%s]: initializing communication...", host->hostname, client_ip, client_port);
1156
- char *initial_response;
1157
- if (stream_flags & LABEL_FLAG_UPDATE_STREAM) {
1158
- info("STREAM %s [receive from [%s]:%s]: Netdata is using the newest stream protocol.", host->hostname, client_ip, client_port);
1159
- initial_response = START_STREAMING_PROMPT_V2;
1184
+ char initial_response[HTTP_HEADER_SIZE];
1185
+ if (stream_version > 1) {
1186
+ info("STREAM %s [receive from [%s]:%s]: Netdata is using the stream version %u.", host->hostname, client_ip, client_port, stream_version);
1187
+ sprintf(initial_response, "%s%u", START_STREAMING_PROMPT_VN, stream_version);
1188
+ } else if (stream_version == 1) {
1189
+ info("STREAM %s [receive from [%s]:%s]: Netdata is using the stream version %u.", host->hostname, client_ip, client_port, stream_version);
1190
+ sprintf(initial_response, "%s", START_STREAMING_PROMPT_V2);
1191
} else {
1161
- info("STREAM %s [receive from [%s]:%s]: Netdata is using an old protocol.", host->hostname, client_ip, client_port);
1162
- initial_response = START_STREAMING_PROMPT;
1192
+ info("STREAM %s [receive from [%s]:%s]: Netdata is using first stream protocol.", host->hostname, client_ip, client_port);
1193
+ sprintf(initial_response, "%s", START_STREAMING_PROMPT);
1194
}
1164
-#ifdef ENABLE_HTTPS
1195
+ #ifdef ENABLE_HTTPS
1196
host->stream_ssl.conn = ssl->conn;
1197
host->stream_ssl.flags = ssl->flags;
1198
if(send_timeout(ssl, fd, initial_response, strlen(initial_response), 0, 60) != (ssize_t)strlen(initial_response)) {
@@ -1206,7 +1237,7 @@ static int rrdpush_receive(int fd
1237
rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
1238
host->connected_senders++;
1239
host->senders_disconnected_time = 0;
1209
- host->labels_flag = stream_flags;
1240
+ host->labels_flag = (stream_version > 0)?LABEL_FLAG_UPDATE_STREAM:LABEL_FLAG_STOP_STREAM;
1241
1242
if(health_enabled != CONFIG_BOOLEAN_NO) {
1243
if(alarms_delay > 0) {
@@ -1262,7 +1293,7 @@ struct rrdpush_thread {
1293
char *program_version;
1294
struct rrdhost_system_info *system_info;
1295
int update_every;
1265
- int stream_flags;
1296
+ uint32_t stream_version;
1297
#ifdef ENABLE_HTTPS
1298
struct netdata_ssl ssl;
1299
#endif
@@ -1318,7 +1349,7 @@ static void *rrdpush_receiver_thread(void *ptr) {
1349
, rpt->update_every
1350
, rpt->client_ip
1351
, rpt->client_port
1321
- , rpt->stream_flags
1352
+ , rpt->stream_version
1353
#ifdef ENABLE_HTTPS
1354
, &rpt->ssl
1355
#endif
@@ -1367,8 +1398,8 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
1398
1399
char *key = NULL, *hostname = NULL, *registry_hostname = NULL, *machine_guid = NULL, *os = "unknown", *timezone = "unknown", *tags = NULL;
1400
int update_every = default_rrd_update_every;
1401
+ uint32_t stream_version = UINT_MAX;
1402
char buf[GUID_LEN + 1];
1371
- int stream_flags = LABEL_FLAG_STOP_STREAM;
1403
1404
struct rrdhost_system_info *system_info = callocz(1, sizeof(struct rrdhost_system_info));
1405
@@ -1396,31 +1427,36 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
1427
timezone = value;
1428
else if(!strcmp(name, "tags"))
1429
tags = value;
1430
+ else if(!strcmp(name, "ver"))
1431
+ stream_version = MIN((uint32_t) strtoul(value, NULL, 0), STREAMING_PROTOCOL_CURRENT_VERSION);
1432
else {
1400
- if(!strcmp(name, "NETDATA_PROTOCOL_VERSION"))
1401
- stream_flags = LABEL_FLAG_UPDATE_STREAM;
1402
- else {
1403
- // An old Netdata slave does not have a compatible streaming protocol, map to something sane.
1404
- if (!strcmp(name, "NETDATA_SYSTEM_OS_NAME"))
1405
- name = "NETDATA_HOST_OS_NAME";
1406
- else if (!strcmp(name, "NETDATA_SYSTEM_OS_ID"))
1407
- name = "NETDATA_HOST_OS_ID";
1408
- else if (!strcmp(name, "NETDATA_SYSTEM_OS_ID_LIKE"))
1409
- name = "NETDATA_HOST_OS_ID_LIKE";
1410
- else if (!strcmp(name, "NETDATA_SYSTEM_OS_VERSION"))
1411
- name = "NETDATA_HOST_OS_VERSION";
1412
- else if (!strcmp(name, "NETDATA_SYSTEM_OS_VERSION_ID"))
1413
- name = "NETDATA_HOST_OS_VERSION_ID";
1414
- else if (!strcmp(name, "NETDATA_SYSTEM_OS_DETECTION"))
1415
- name = "NETDATA_HOST_OS_DETECTION";
1416
- if (unlikely(rrdhost_set_system_info_variable(system_info, name, value))) {
1417
- info("STREAM [receive from [%s]:%s]: request has parameter '%s' = '%s', which is not used.",
1418
- w->client_ip, w->client_port, key, value);
1419
- }
1433
+ // An old Netdata slave does not have a compatible streaming protocol, map to something sane.
1434
+ if (!strcmp(name, "NETDATA_SYSTEM_OS_NAME"))
1435
+ name = "NETDATA_HOST_OS_NAME";
1436
+ else if (!strcmp(name, "NETDATA_SYSTEM_OS_ID"))
1437
+ name = "NETDATA_HOST_OS_ID";
1438
+ else if (!strcmp(name, "NETDATA_SYSTEM_OS_ID_LIKE"))
1439
+ name = "NETDATA_HOST_OS_ID_LIKE";
1440
+ else if (!strcmp(name, "NETDATA_SYSTEM_OS_VERSION"))
1441
+ name = "NETDATA_HOST_OS_VERSION";
1442
+ else if (!strcmp(name, "NETDATA_SYSTEM_OS_VERSION_ID"))
1443
+ name = "NETDATA_HOST_OS_VERSION_ID";
1444
+ else if (!strcmp(name, "NETDATA_SYSTEM_OS_DETECTION"))
1445
+ name = "NETDATA_HOST_OS_DETECTION";
1446
+ else if(!strcmp(name, "NETDATA_PROTOCOL_VERSION") && stream_version == UINT_MAX) {
1447
+ stream_version = 1;
1448
+ }
1449
+
1450
+ if (unlikely(rrdhost_set_system_info_variable(system_info, name, value))) {
1451
+ info("STREAM [receive from [%s]:%s]: request has parameter '%s' = '%s', which is not used.",
1452
+ w->client_ip, w->client_port, name, value);
1453
}
1454
}
1455
}
1456
1457
+ if (stream_version == UINT_MAX)
1458
+ stream_version = 0;
1459
+
1460
if(!key || !*key) {
1461
rrdhost_system_info_free(system_info);
1462
log_stream_connection(w->client_ip, w->client_port, (key && *key)?key:"-", (machine_guid && *machine_guid)?machine_guid:"-", (hostname && *hostname)?hostname:"-", "ACCESS DENIED - NO KEY");
@@ -1532,7 +1568,7 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
1568
rpt->client_port = strdupz(w->client_port);
1569
rpt->update_every = update_every;
1570
rpt->system_info = system_info;
1535
- rpt->stream_flags = stream_flags;
1571
+ rpt->stream_version = stream_version;
1572
#ifdef ENABLE_HTTPS
1573
rpt->ssl.conn = w->ssl.conn;
1574
rpt->ssl.flags = w->ssl.flags;
streaming/rrdpush.h
+2
@@ -6,6 +6,8 @@
6
#include "web/server/web_client.h"
7
#include "daemon/common.h"
8
9
+#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)2
10
+
11
extern unsigned int default_rrdpush_enabled;
12
extern char *default_rrdpush_destination;
13
extern char *default_rrdpush_api_key;