Adds claimed_id streaming (#9804)
* streams claimed_id of child nodes to parents * adds this information into /api/v1/info
Timotej S committed
Aug 26, 2020 at 14:50 UTC
ab7ff3131f3698710e0bd9fa3c66d31a3a194725
9 files changed
+129
-22
claim/claim.c
+16
-11
@@ -27,8 +27,6 @@ static char *claiming_errors[] = {
27
"Service Unavailable", // 17
28
"Agent Unique Id Not Readable" // 18
29
};
30
-static netdata_mutex_t claim_mutex = NETDATA_MUTEX_INITIALIZER;
31
-static char *claimed_id = NULL;
30
31
/* Retrieve the claim id for the agent.
32
* Caller owns the string.
@@ -36,9 +34,9 @@ static char *claimed_id = NULL;
34
char *is_agent_claimed()
35
{
36
char *result;
39
- netdata_mutex_lock(&claim_mutex);
40
- result = (claimed_id == NULL) ? NULL : strdup(claimed_id);
41
- netdata_mutex_unlock(&claim_mutex);
37
+ netdata_mutex_lock(&localhost->claimed_id_lock);
38
+ result = (localhost->claimed_id == NULL) ? NULL : strdupz(localhost->claimed_id);
39
+ netdata_mutex_unlock(&localhost->claimed_id_lock);
40
return result;
41
}
42
@@ -135,10 +133,11 @@ void load_claiming_state(void)
133
#if defined( DISABLE_CLOUD ) || !defined( ENABLE_ACLK )
134
netdata_cloud_setting = 0;
135
#else
138
- netdata_mutex_lock(&claim_mutex);
139
- if (claimed_id != NULL) {
140
- freez(claimed_id);
141
- claimed_id = NULL;
136
+ uuid_t uuid;
137
+ netdata_mutex_lock(&localhost->claimed_id_lock);
138
+ if (localhost->claimed_id) {
139
+ freez(localhost->claimed_id);
140
+ localhost->claimed_id = NULL;
141
}
142
if (aclk_connected)
143
{
@@ -153,8 +152,14 @@ void load_claiming_state(void)
152
snprintfz(filename, FILENAME_MAX, "%s/cloud.d/claimed_id", netdata_configured_varlib_dir);
153
154
long bytes_read;
156
- claimed_id = read_by_filename(filename, &bytes_read);
157
- netdata_mutex_unlock(&claim_mutex); // Only the main thread can call this function, safe to release and then read
155
+ char *claimed_id = read_by_filename(filename, &bytes_read);
156
+ if(claimed_id && uuid_parse(claimed_id, uuid)) {
157
+ error("claimed_id \"%s\" doesn't look like valid UUID", claimed_id);
158
+ freez(claimed_id);
159
+ claimed_id = NULL;
160
+ }
161
+ localhost->claimed_id = claimed_id;
162
+ netdata_mutex_unlock(&localhost->claimed_id_lock);
163
if (!claimed_id) {
164
info("Unable to load '%s', setting state to AGENT_UNCLAIMED", filename);
165
return;
daemon/commands.c
+1
@@ -202,6 +202,7 @@ static cmd_status_t cmd_reload_claiming_state_execute(char *args, char **message
202
info("COMMAND: Reloading Agent Claiming configuration.");
203
load_claiming_state();
204
registry_update_cloud_base_url();
205
+ rrdpush_claimed_id(localhost);
206
error_log_limit_reset();
207
return CMD_STATUS_SUCCESS;
208
}
database/rrd.h
+3
@@ -817,6 +817,9 @@ struct rrdhost {
817
struct netdata_ssl stream_ssl; //Structure used to encrypt the stream
818
#endif
819
820
+ netdata_mutex_t claimed_id_lock;
821
+ char *claimed_id; // Claimed ID if host has one otherwise NULL
822
+
823
struct rrdhost *next;
824
};
825
extern RRDHOST *localhost;
database/rrdhost.c
+4
@@ -167,6 +167,8 @@ RRDHOST *rrdhost_create(const char *hostname,
167
netdata_rwlock_init(&host->rrdhost_rwlock);
168
netdata_rwlock_init(&host->labels_rwlock);
169
170
+ netdata_mutex_init(&host->claimed_id_lock);
171
+
172
rrdhost_init_hostname(host, hostname);
173
rrdhost_init_machine_guid(host, guid);
174
@@ -858,6 +860,8 @@ void rrdhost_free(RRDHOST *host) {
860
// ------------------------------------------------------------------------
861
// free it
862
863
+ pthread_mutex_destroy(&host->claimed_id_lock);
864
+ freez(host->claimed_id);
865
freez((void *)host->tags);
866
free_host_labels(host->labels);
867
freez((void *)host->os);
streaming/receiver.c
+43
@@ -95,6 +95,48 @@ PARSER_RC streaming_timestamp(char **words, void *user, PLUGINSD_ACTION *plugins
95
return PARSER_RC_ERROR;
96
}
97
98
+#define CLAIMED_ID_MIN_WORDS 3
99
+PARSER_RC streaming_claimed_id(char **words, void *user, PLUGINSD_ACTION *plugins_action)
100
+{
101
+ UNUSED(plugins_action);
102
+
103
+ int i;
104
+ uuid_t uuid;
105
+ RRDHOST *host = ((PARSER_USER_OBJECT *)user)->host;
106
+
107
+ for (i = 0; words[i]; i++) ;
108
+ if (i != CLAIMED_ID_MIN_WORDS) {
109
+ error("Command CLAIMED_ID came malformed %d parameters are expected but %d received", CLAIMED_ID_MIN_WORDS - 1, i - 1);
110
+ return PARSER_RC_ERROR;
111
+ }
112
+
113
+ // We don't need the parsed UUID
114
+ // just do it to check the format
115
+ if(uuid_parse(words[1], uuid)) {
116
+ error("1st parameter (host GUID) to CLAIMED_ID command is not valid GUID. Received: \"%s\".", words[1]);
117
+ return PARSER_RC_ERROR;
118
+ }
119
+ if(uuid_parse(words[2], uuid) && strcmp(words[2], "NULL")) {
120
+ error("2nd parameter (Claim ID) to CLAIMED_ID command is not valid GUID. Received: \"%s\".", words[2]);
121
+ return PARSER_RC_ERROR;
122
+ }
123
+
124
+ if(strcmp(words[1], host->machine_guid)) {
125
+ error("Claim ID is for host \"%s\" but it came over connection for \"%s\"", words[1], host->machine_guid);
126
+ return PARSER_RC_OK; //the message is OK problem must be somewehere else
127
+ }
128
+
129
+ netdata_mutex_lock(&host->claimed_id_lock);
130
+ if (host->claimed_id)
131
+ freez(host->claimed_id);
132
+ host->claimed_id = strcmp(words[2], "NULL") ? strdupz(words[2]) : NULL;
133
+ netdata_mutex_unlock(&host->claimed_id_lock);
134
+
135
+ rrdpush_claimed_id(host);
136
+
137
+ return PARSER_RC_OK;
138
+}
139
+
140
/* The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing.
141
*/
142
static int receiver_read(struct receiver_state *r, FILE *fp) {
@@ -156,6 +198,7 @@ size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp
198
199
PARSER *parser = parser_init(rpt->host, user, fp, PARSER_INPUT_SPLIT);
200
parser_add_keyword(parser, "TIMESTAMP", streaming_timestamp);
201
+ parser_add_keyword(parser, "CLAIMED_ID", streaming_claimed_id);
202
203
if (unlikely(!parser)) {
204
error("Failed to initialize parser");
streaming/rrdpush.c
+19
@@ -364,6 +364,25 @@ void rrdpush_send_labels(RRDHOST *host) {
364
365
host->labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;
366
}
367
+
368
+void rrdpush_claimed_id(RRDHOST *host)
369
+{
370
+ if(unlikely(!host->rrdpush_send_enabled || !host->rrdpush_sender_connected))
371
+ return;
372
+
373
+ sender_start(host->sender);
374
+ netdata_mutex_lock(&host->claimed_id_lock);
375
+
376
+ buffer_sprintf(host->sender->build, "CLAIMED_ID %s %s\n", host->machine_guid, (host->claimed_id ? host->claimed_id : "NULL") );
377
+
378
+ netdata_mutex_unlock(&host->claimed_id_lock);
379
+ sender_commit(host->sender);
380
+
381
+ // signal the sender there are more data
382
+ if(host->rrdpush_sender_pipe[PIPE_WRITE] != -1 && write(host->rrdpush_sender_pipe[PIPE_WRITE], " ", 1) == -1)
383
+ error("STREAM %s [send]: cannot write to internal pipe", host->hostname);
384
+}
385
+
386
// ----------------------------------------------------------------------------
387
// rrdpush sender thread
388
streaming/rrdpush.h
+5
-3
@@ -10,9 +10,10 @@
10
11
#define CONNECTED_TO_SIZE 100
12
13
-// #define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)3 Gap-filling
14
-#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)2
15
-#define VERSION_GAP_FILLING 3
13
+// #define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)4 Gap-filling
14
+#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)3
15
+#define VERSION_GAP_FILLING 4
16
+#define STREAM_VERSION_CLAIM 3
17
18
#define STREAMING_PROTOCOL_VERSION "1.1"
19
#define START_STREAMING_PROMPT "Hit me baby, push them over..."
@@ -106,6 +107,7 @@ extern void rrdset_done_push(RRDSET *st);
107
extern void rrdset_push_chart_definition_now(RRDSET *st);
108
extern void *rrdpush_sender_thread(void *ptr);
109
extern void rrdpush_send_labels(RRDHOST *host);
110
+extern void rrdpush_claimed_id(RRDHOST *host);
111
112
extern int rrdpush_receiver_thread_spawn(struct web_client *w, char *url);
113
extern void rrdpush_sender_thread_stop(RRDHOST *host);
streaming/sender.c
+2
@@ -622,6 +622,8 @@ void *rrdpush_sender_thread(void *ptr) {
622
buffer_sprintf(s->build, "TIMESTAMP %ld", now);
623
sender_commit(s);
624
}
625
+ if (s->version >= STREAM_VERSION_CLAIM)
626
+ rrdpush_claimed_id(s->host);
627
continue;
628
}
629
web/api/web_api_v1.c
+36
-8
@@ -774,18 +774,48 @@ static inline void web_client_api_request_v1_info_summary_alarm_statuses(RRDHOST
774
}
775
776
static inline void web_client_api_request_v1_info_mirrored_hosts(BUFFER *wb) {
777
- RRDHOST *rc;
777
+ RRDHOST *host;
778
int count = 0;
779
+
780
+ buffer_strcat(wb, "\t\"mirrored_hosts\": [\n");
781
rrd_rdlock();
780
- rrdhost_foreach_read(rc) {
781
- if (rrdhost_flag_check(rc, RRDHOST_FLAG_ARCHIVED))
782
+ rrdhost_foreach_read(host) {
783
+ if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED))
784
continue;
783
- if(count > 0) buffer_strcat(wb, ",\n");
784
- buffer_sprintf(wb, "\t\t\"%s\"", rc->hostname);
785
+ if (count > 0)
786
+ buffer_strcat(wb, ",\n");
787
+
788
+ buffer_sprintf(wb, "\t\t\"%s\"", host->hostname);
789
+ count++;
790
+ }
791
+
792
+ buffer_strcat(wb, "\n\t],\n\t\"mirrored_hosts_status\": [\n");
793
+ count = 0;
794
+ rrdhost_foreach_read(host)
795
+ {
796
+ if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED))
797
+ continue;
798
+ if (count > 0)
799
+ buffer_strcat(wb, ",\n");
800
+
801
+ netdata_mutex_lock(&host->receiver_lock);
802
+ buffer_sprintf(
803
+ wb, "\t\t{ \"guid\": \"%s\", \"reachable\": %s, \"claim_id\": ", host->machine_guid,
804
+ (host->receiver || host == localhost) ? "true" : "false");
805
+ netdata_mutex_unlock(&host->receiver_lock);
806
+
807
+ netdata_mutex_lock(&host->claimed_id_lock);
808
+ if (host->claimed_id)
809
+ buffer_sprintf(wb, "\"%s\" }", host->claimed_id);
810
+ else
811
+ buffer_strcat(wb, "null }");
812
+ netdata_mutex_unlock(&host->claimed_id_lock);
813
+
814
count++;
815
}
787
- buffer_strcat(wb, "\n");
816
rrd_unlock();
817
+
818
+ buffer_strcat(wb, "\n\t],\n");
819
}
820
821
inline void host_labels2json(RRDHOST *host, BUFFER *wb, size_t indentation) {
@@ -825,9 +855,7 @@ inline int web_client_api_request_v1_info_fill_buffer(RRDHOST *host, BUFFER *wb)
855
buffer_sprintf(wb, "\t\"version\": \"%s\",\n", host->program_version);
856
buffer_sprintf(wb, "\t\"uid\": \"%s\",\n", host->machine_guid);
857
828
- buffer_strcat(wb, "\t\"mirrored_hosts\": [\n");
858
web_client_api_request_v1_info_mirrored_hosts(wb);
830
- buffer_strcat(wb, "\t],\n");
859
860
buffer_strcat(wb, "\t\"alarms\": {\n");
861
web_client_api_request_v1_info_summary_alarm_statuses(host, wb);