Stream with labels (#7549)
This commit enables streaming host labels
thiagoftsm committed
Jan 14, 2020 at 10:27 UTC
ef2b11fcb4d56ec946f6dc24929ba6ec0b54d0f2
7 files changed
+150
-17
collectors/plugins.d/plugins_d.c
+40
@@ -234,9 +234,12 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp, int
234
uint32_t DIMENSION_HASH = simple_hash(PLUGINSD_KEYWORD_DIMENSION);
235
uint32_t DISABLE_HASH = simple_hash(PLUGINSD_KEYWORD_DISABLE);
236
uint32_t VARIABLE_HASH = simple_hash(PLUGINSD_KEYWORD_VARIABLE);
237
+ uint32_t LABEL_HASH = simple_hash(PLUGINSD_KEYWORD_LABEL);
238
+ uint32_t OVERWRITE_HASH = simple_hash(PLUGINSD_KEYWORD_OVERWRITE);
239
240
RRDSET *st = NULL;
241
uint32_t hash;
242
+ struct label *new_labels = NULL;
243
244
errno = 0;
245
clearerr(fp);
@@ -283,6 +286,7 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp, int
286
#else
287
r = fgets(line, PLUGINSD_LINE_MAX, fp);
288
#endif
289
+
290
if(unlikely(!r)) {
291
if(feof(fp))
292
error("read failed: end of file");
@@ -616,6 +620,39 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp, int
620
enabled = 0;
621
break;
622
}
623
+ else if(likely(hash == LABEL_HASH && !strcmp(s, PLUGINSD_KEYWORD_LABEL))) {
624
+ debug(D_PLUGINSD, "requested a LABEL CHANGE");
625
+ char *store;
626
+ if(!words[4])
627
+ store = words[3];
628
+ else {
629
+ store = callocz(PLUGINSD_LINE_MAX + 1, sizeof(char));
630
+ char *move = store;
631
+ int i = 3;
632
+ while (i < w) {
633
+ size_t length = strlen(words[i]);
634
+ memcpy(move, words[i], length);
635
+ move += length;
636
+ *move++ = ' ';
637
+
638
+ i++;
639
+ if(!words[i])
640
+ break;
641
+ }
642
+ }
643
+
644
+ new_labels = add_label_to_list(new_labels, words[1], store, strtol(words[2], NULL, 10));
645
+ }
646
+ else if(likely(hash == OVERWRITE_HASH && !strcmp(s, PLUGINSD_KEYWORD_OVERWRITE))) {
647
+ debug(D_PLUGINSD, "requested a OVERWITE a variable");
648
+ if(!host->labels) {
649
+ host->labels = new_labels;
650
+ } else {
651
+ replace_label_list(host, new_labels);
652
+ }
653
+
654
+ new_labels = NULL;
655
+ }
656
else {
657
error("sent command '%s' which is not known by netdata, for host '%s'. Disabling it.", s, host->hostname);
658
enabled = 0;
@@ -626,6 +663,9 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp, int
663
cleanup:
664
cd->enabled = enabled;
665
666
+ if(new_labels)
667
+ free_host_labels(new_labels);
668
+
669
if(likely(count)) {
670
cd->successful_collections += count;
671
cd->serial_failures = 0;
collectors/plugins.d/plugins_d.h
+2
@@ -29,6 +29,8 @@
29
#define PLUGINSD_KEYWORD_FLUSH "FLUSH"
30
#define PLUGINSD_KEYWORD_DISABLE "DISABLE"
31
#define PLUGINSD_KEYWORD_VARIABLE "VARIABLE"
32
+#define PLUGINSD_KEYWORD_LABEL "LABEL"
33
+#define PLUGINSD_KEYWORD_OVERWRITE "OVERWRITE"
34
35
#define PLUGINSD_LINE_MAX 1024
36
#define PLUGINSD_LINE_MAX_SSL_READ 512
database/rrd.h
+6
@@ -156,6 +156,9 @@ typedef enum label_source {
156
LABEL_SOURCE_KUBERNETES = 4
157
} LABEL_SOURCE;
158
159
+#define LABEL_FLAG_UPDATE_STREAM 1
160
+#define LABEL_FLAG_STOP_STREAM 2
161
+
162
struct label {
163
char *key, *value;
164
uint32_t key_hash;
@@ -166,6 +169,8 @@ struct label {
169
char *translate_label_source(LABEL_SOURCE l);
170
struct label *create_label(char *key, char *value, LABEL_SOURCE label_source);
171
struct label *add_label_to_list(struct label *l, char *key, char *value, LABEL_SOURCE label_source);
172
+extern void replace_label_list(RRDHOST *host, struct label *new_labels);
173
+extern void free_host_labels(struct label *labels);
174
void reload_host_labels();
175
176
// ----------------------------------------------------------------------------
@@ -749,6 +754,7 @@ struct rrdhost {
754
// Support for host-level labels
755
struct label *labels;
756
netdata_rwlock_t labels_rwlock; // lock for the label list
757
+ uint32_t labels_flag; //Flags for labels
758
759
// ------------------------------------------------------------------------
760
// indexes
database/rrdhost.c
+24
-9
@@ -890,6 +890,26 @@ struct label *create_label(char *key, char *value, LABEL_SOURCE label_source)
890
return result;
891
}
892
893
+void free_host_labels(struct label *labels)
894
+{
895
+ while (labels != NULL)
896
+ {
897
+ struct label *current = labels;
898
+ labels = labels->next;
899
+ freez(current);
900
+ }
901
+}
902
+
903
+void replace_label_list(RRDHOST *host, struct label *new_labels)
904
+{
905
+ netdata_rwlock_wrlock(&host->labels_rwlock);
906
+ struct label *old_labels = host->labels;
907
+ host->labels = new_labels;
908
+ netdata_rwlock_unlock(&host->labels_rwlock);
909
+
910
+ free_host_labels(old_labels);
911
+}
912
+
913
struct label *add_label_to_list(struct label *l, char *key, char *value, LABEL_SOURCE label_source)
914
{
915
struct label *lab = create_label(key, value, label_source);
@@ -937,16 +957,11 @@ void reload_host_labels()
957
struct label *new_labels = merge_label_lists(from_auto, from_k8s);
958
new_labels = merge_label_lists(new_labels, from_config);
959
940
- netdata_rwlock_wrlock(&localhost->labels_rwlock);
941
- struct label *old_labels = localhost->labels;
942
- localhost->labels = new_labels;
943
- netdata_rwlock_unlock(&localhost->labels_rwlock);
960
+ replace_label_list(localhost, new_labels);
961
945
- while (old_labels != NULL)
946
- {
947
- struct label *current = old_labels;
948
- old_labels = old_labels->next;
949
- freez(current);
962
+ if(localhost->rrdpush_send_enabled && localhost->rrdpush_sender_buffer){
963
+ localhost->labels_flag |= LABEL_FLAG_UPDATE_STREAM;
964
+ rrdpush_send_labels(localhost);
965
}
966
967
health_reload();
streaming/rrdpush.c
+77
-7
@@ -25,7 +25,9 @@
25
*
26
*/
27
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
32
typedef enum {
33
RRDPUSH_MULTIPLE_CONNECTIONS_ALLOW,
@@ -79,6 +81,7 @@ int rrdpush_init() {
81
default_rrdpush_send_charts_matching = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "send charts matching", "*");
82
rrdhost_free_orphan_time = config_get_number(CONFIG_SECTION_GLOBAL, "cleanup orphan hosts after seconds", rrdhost_free_orphan_time);
83
84
+
85
if(default_rrdpush_enabled && (!default_rrdpush_destination || !*default_rrdpush_destination || !default_rrdpush_api_key || !*default_rrdpush_api_key)) {
86
error("STREAM [send]: cannot enable sending thread - information is missing.");
87
default_rrdpush_enabled = 0;
@@ -338,6 +341,36 @@ void rrdset_done_push(RRDSET *st) {
341
rrdpush_buffer_unlock(host);
342
}
343
344
+// labels
345
+void rrdpush_send_labels(RRDHOST *host) {
346
+ if (!host->labels || !(host->labels_flag & LABEL_FLAG_UPDATE_STREAM) || (host->labels_flag & LABEL_FLAG_STOP_STREAM))
347
+ return;
348
+
349
+ rrdpush_buffer_lock(host);
350
+ netdata_rwlock_rdlock(&host->labels_rwlock);
351
+
352
+ struct label *labels = host->labels;
353
+ while(labels) {
354
+ buffer_sprintf(host->rrdpush_sender_buffer
355
+ , "LABEL \"%s\" = %d %s\n"
356
+ , labels->key
357
+ , (int)labels->label_source
358
+ , labels->value);
359
+
360
+ labels = labels->next;
361
+ }
362
+
363
+ buffer_sprintf(host->rrdpush_sender_buffer
364
+ , "OVERWRITE %s\n", "labels");
365
+
366
+ netdata_rwlock_unlock(&host->labels_rwlock);
367
+
368
+ if(host->rrdpush_sender_pipe[PIPE_WRITE] != -1 && write(host->rrdpush_sender_pipe[PIPE_WRITE], " ", 1) == -1)
369
+ error("STREAM %s [send]: cannot write to internal pipe", host->hostname);
370
+
371
+ rrdpush_buffer_unlock(host);
372
+ host->labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;
373
+}
374
// ----------------------------------------------------------------------------
375
// rrdpush sender thread
376
@@ -536,6 +569,7 @@ static int rrdpush_sender_thread_connect_to_master(RRDHOST *host, int default_po
569
"&NETDATA_SYSTEM_VIRT_DETECTION=%s"
570
"&NETDATA_SYSTEM_CONTAINER=%s"
571
"&NETDATA_SYSTEM_CONTAINER_DETECTION=%s"
572
+ "&NETDATA_PROTOCOL_VERSION=%s"
573
" HTTP/1.1\r\n"
574
"User-Agent: %s/%s\r\n"
575
"Accept: */*\r\n\r\n"
@@ -560,6 +594,7 @@ static int rrdpush_sender_thread_connect_to_master(RRDHOST *host, int default_po
594
, (host->system_info->virt_detection) ? host->system_info->virt_detection : ""
595
, (host->system_info->container) ? host->system_info->container : ""
596
, (host->system_info->container_detection) ? host->system_info->container_detection : ""
597
+ , STREAMING_PROTOCOL_VERSION
598
, host->program_name
599
, host->program_version
600
);
@@ -613,7 +648,20 @@ static int rrdpush_sender_thread_connect_to_master(RRDHOST *host, int default_po
648
return 0;
649
}
650
616
- if(strncmp(http, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT)) != 0) {
651
+ int answer = strncmp(http, START_STREAMING_PROMPT_V2, strlen(START_STREAMING_PROMPT_V2));
652
+ if(!answer) {
653
+ host->labels_flag |= LABEL_FLAG_UPDATE_STREAM;
654
+ host->labels_flag &= ~LABEL_FLAG_STOP_STREAM;
655
+ } else {
656
+ answer = strncmp(http, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT));
657
+ if(!answer) {
658
+ host->labels_flag |= LABEL_FLAG_STOP_STREAM;
659
+ host->labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;
660
+ info("STREAM %s [send to %s]: is using an old Netdata.", host->hostname, connected_to);
661
+ }
662
+ }
663
+
664
+ if(answer != 0) {
665
error("STREAM %s [send to %s]: server is not replying properly (is it a netdata?).", host->hostname, connected_to);
666
rrdpush_sender_thread_close_socket(host);
667
return 0;
@@ -814,6 +862,8 @@ void *rrdpush_sender_thread(void *ptr) {
862
}
863
864
if (ofd->revents & POLLOUT) {
865
+ rrdpush_send_labels(host);
866
+
867
if (begin < buffer_strlen(host->rrdpush_sender_buffer)) {
868
debug(D_STREAM, "STREAM: Sending data (current buffer length %zu bytes, begin = %zu)...", buffer_strlen(host->rrdpush_sender_buffer), begin);
869
@@ -981,6 +1031,7 @@ static int rrdpush_receive(int fd
1031
, int update_every
1032
, char *client_ip
1033
, char *client_port
1034
+ , int stream_flags
1035
#ifdef ENABLE_HTTPS
1036
, struct netdata_ssl *ssl
1037
#endif
@@ -1097,12 +1148,20 @@ static int rrdpush_receive(int fd
1148
snprintfz(cd.cmd, PLUGINSD_CMD_MAX, "%s:%s", client_ip, client_port);
1149
1150
info("STREAM %s [receive from [%s]:%s]: initializing communication...", host->hostname, client_ip, client_port);
1151
+ char *initial_response;
1152
+ if (stream_flags & LABEL_FLAG_UPDATE_STREAM) {
1153
+ info("STREAM %s [receive from [%s]:%s]: Netdata is using the newest stream protocol.", host->hostname, client_ip, client_port);
1154
+ initial_response = START_STREAMING_PROMPT_V2;
1155
+ } else {
1156
+ info("STREAM %s [receive from [%s]:%s]: Netdata is using an old protocol.", host->hostname, client_ip, client_port);
1157
+ initial_response = START_STREAMING_PROMPT;
1158
+ }
1159
#ifdef ENABLE_HTTPS
1160
host->stream_ssl.conn = ssl->conn;
1161
host->stream_ssl.flags = ssl->flags;
1103
- if(send_timeout(ssl,fd, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT), 0, 60) != strlen(START_STREAMING_PROMPT)) {
1162
+ if(send_timeout(ssl, fd, initial_response, strlen(initial_response), 0, 60) != (ssize_t)strlen(initial_response)) {
1163
#else
1105
- if(send_timeout(fd, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT), 0, 60) != strlen(START_STREAMING_PROMPT)) {
1164
+ if(send_timeout(fd, initial_response, strlen(initial_response), 0, 60) != strlen(initial_response)) {
1165
#endif
1166
log_stream_connection(client_ip, client_port, key, host->machine_guid, host->hostname, "FAILED - CANNOT REPLY");
1167
error("STREAM %s [receive from [%s]:%s]: cannot send ready command.", host->hostname, client_ip, client_port);
@@ -1142,6 +1201,8 @@ static int rrdpush_receive(int fd
1201
rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
1202
host->connected_senders++;
1203
host->senders_disconnected_time = 0;
1204
+ host->labels_flag = stream_flags;
1205
+
1206
if(health_enabled != CONFIG_BOOLEAN_NO) {
1207
if(alarms_delay > 0) {
1208
host->health_delay_up_to = now_realtime_sec() + alarms_delay;
@@ -1196,6 +1257,7 @@ struct rrdpush_thread {
1257
char *program_version;
1258
struct rrdhost_system_info *system_info;
1259
int update_every;
1260
+ int stream_flags;
1261
#ifdef ENABLE_HTTPS
1262
struct netdata_ssl ssl;
1263
#endif
@@ -1251,6 +1313,7 @@ static void *rrdpush_receiver_thread(void *ptr) {
1313
, rpt->update_every
1314
, rpt->client_ip
1315
, rpt->client_port
1316
+ , rpt->stream_flags
1317
#ifdef ENABLE_HTTPS
1318
, &rpt->ssl
1319
#endif
@@ -1300,6 +1363,7 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
1363
char *key = NULL, *hostname = NULL, *registry_hostname = NULL, *machine_guid = NULL, *os = "unknown", *timezone = "unknown", *tags = NULL;
1364
int update_every = default_rrd_update_every;
1365
char buf[GUID_LEN + 1];
1366
+ int stream_flags = LABEL_FLAG_STOP_STREAM;
1367
1368
struct rrdhost_system_info *system_info = callocz(1, sizeof(struct rrdhost_system_info));
1369
@@ -1327,10 +1391,15 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
1391
timezone = value;
1392
else if(!strcmp(name, "tags"))
1393
tags = value;
1330
- else
1331
- if(unlikely(rrdhost_set_system_info_variable(system_info, name, value))) {
1332
- info("STREAM [receive from [%s]:%s]: request has parameter '%s' = '%s', which is not used.", w->client_ip, w->client_port, key, value);
1333
- }
1394
+ else {
1395
+ if(!strcmp(name, "NETDATA_PROTOCOL_VERSION"))
1396
+ stream_flags = LABEL_FLAG_UPDATE_STREAM;
1397
+ else
1398
+ if (unlikely(rrdhost_set_system_info_variable(system_info, name, value))) {
1399
+ info("STREAM [receive from [%s]:%s]: request has parameter '%s' = '%s', which is not used.",
1400
+ w->client_ip, w->client_port, key, value);
1401
+ }
1402
+ }
1403
}
1404
1405
if(!key || !*key) {
@@ -1444,6 +1513,7 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
1513
rpt->client_port = strdupz(w->client_port);
1514
rpt->update_every = update_every;
1515
rpt->system_info = system_info;
1516
+ rpt->stream_flags = stream_flags;
1517
#ifdef ENABLE_HTTPS
1518
rpt->ssl.conn = w->ssl.conn;
1519
rpt->ssl.flags = w->ssl.flags;
streaming/rrdpush.h
+1
@@ -17,6 +17,7 @@ extern int configured_as_master();
17
extern void rrdset_done_push(RRDSET *st);
18
extern void rrdset_push_chart_definition_now(RRDSET *st);
19
extern void *rrdpush_sender_thread(void *ptr);
20
+extern void rrdpush_send_labels(RRDHOST *host);
21
22
extern int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url);
23
extern void rrdpush_sender_thread_stop(RRDHOST *host);
streaming/stream.conf
-1
@@ -87,7 +87,6 @@
87
# Sync the clock of the charts for that many iterations, when starting.
88
initial clock resync iterations = 60
89
90
-
90
# -----------------------------------------------------------------------------
91
# 2. ON MASTER NETDATA - THE ONE THAT WILL BE RECEIVING METRICS
92