Regenerate topic base on connect (#9044)
Allow agents to be reclaimed while they are running. Fix a race hazard between claiming and the ACLK. Changes the private key, base topic, username and contents of the LWT. Co-authored-by: <hilari@hilarimoragrega.com>
Andrew Moss committed
May 20, 2020 at 16:28 UTC
53efa359d60683cad6dc73ecf84d0df7ee621303
3 files changed
+117
-68
aclk/agent_cloud_link.c
+79
-57
@@ -24,6 +24,7 @@ static char *global_base_topic = NULL;
24
static int aclk_connecting = 0;
25
int aclk_connected = 0; // Exposed in the web-api
26
int aclk_force_reconnect = 0; // Indication from lower layers
27
+int aclk_kill_link = 0; // Tell the agent to tear down the link
28
usec_t aclk_session_us = 0; // Used by the mqtt layer
29
time_t aclk_session_sec = 0; // Used by the mqtt layer
30
@@ -149,6 +150,9 @@ int cloud_to_agent_parse(JSON_ENTRY *e)
150
static RSA *aclk_private_key = NULL;
151
static int create_private_key()
152
{
153
+ if (aclk_private_key != NULL)
154
+ RSA_free(aclk_private_key);
155
+ aclk_private_key = NULL;
156
char filename[FILENAME_MAX + 1];
157
snprintfz(filename, FILENAME_MAX, "%s/cloud.d/private.pem", netdata_configured_varlib_dir);
158
@@ -429,29 +433,29 @@ struct aclk_query *aclk_queue_pop()
433
434
// This will give the base topic that the agent will publish messages.
435
// subtopics will be sent under the base topic e.g. base_topic/subtopic
432
-// This is called by aclk_init(), to compute the base topic once and have
433
-// it stored internally.
434
-// Need to check if additional logic should be added to make sure that there
435
-// is enough information to determine the base topic at init time
436
+// This is called during the connection, we delete any previous topic
437
+// in-case the user has changed the agent id and reclaimed.
438
439
char *create_publish_base_topic()
440
{
439
- if (unlikely(!is_agent_claimed()))
441
+ char *agent_id = is_agent_claimed();
442
+ if (unlikely(!agent_id))
443
return NULL;
444
445
ACLK_LOCK;
446
444
- if (unlikely(!global_base_topic)) {
445
- char tmp_topic[ACLK_MAX_TOPIC + 1], *tmp;
447
+ if (global_base_topic)
448
+ freez(global_base_topic);
449
+ char tmp_topic[ACLK_MAX_TOPIC + 1], *tmp;
450
447
- snprintf(tmp_topic, ACLK_MAX_TOPIC, ACLK_TOPIC_STRUCTURE, is_agent_claimed());
448
- tmp = strchr(tmp_topic, '\n');
449
- if (unlikely(tmp))
450
- *tmp = '\0';
451
- global_base_topic = strdupz(tmp_topic);
452
- }
451
+ snprintf(tmp_topic, ACLK_MAX_TOPIC, ACLK_TOPIC_STRUCTURE, agent_id);
452
+ tmp = strchr(tmp_topic, '\n');
453
+ if (unlikely(tmp))
454
+ *tmp = '\0';
455
+ global_base_topic = strdupz(tmp_topic);
456
457
ACLK_UNLOCK;
458
+ freez(agent_id);
459
return global_base_topic;
460
}
461
@@ -992,6 +996,39 @@ void *aclk_query_main_thread(void *ptr)
996
return NULL;
997
}
998
999
+static void aclk_graceful_disconnect()
1000
+{
1001
+ size_t write_q, write_q_bytes, read_q;
1002
+ time_t event_loop_timeout;
1003
+
1004
+ // Send a graceful disconnect message
1005
+ BUFFER *b = buffer_create(512);
1006
+ aclk_create_header(b, "disconnect", NULL, 0, 0);
1007
+ buffer_strcat(b, ",\n\t\"payload\": \"graceful\"}\n");
1008
+ aclk_send_message(ACLK_METADATA_TOPIC, (char*)buffer_tostring(b), NULL);
1009
+ buffer_free(b);
1010
+
1011
+ event_loop_timeout = now_realtime_sec() + 5;
1012
+ write_q = 1;
1013
+ while (write_q && event_loop_timeout > now_realtime_sec()) {
1014
+ _link_event_loop();
1015
+ lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
1016
+ }
1017
+
1018
+ aclk_shutting_down = 1;
1019
+ _link_shutdown();
1020
+ aclk_lws_wss_mqtt_layer_disconect_notif();
1021
+
1022
+ write_q = 1;
1023
+ event_loop_timeout = now_realtime_sec() + 5;
1024
+ while (write_q && event_loop_timeout > now_realtime_sec()) {
1025
+ _link_event_loop();
1026
+ lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
1027
+ }
1028
+ aclk_shutting_down = 0;
1029
+}
1030
+
1031
+
1032
// Thread cleanup
1033
static void aclk_main_cleanup(void *ptr)
1034
{
@@ -1000,36 +1037,12 @@ static void aclk_main_cleanup(void *ptr)
1037
1038
info("cleaning up...");
1039
1003
- if (is_agent_claimed() && aclk_connected) {
1004
- size_t write_q, write_q_bytes, read_q;
1005
- time_t event_loop_timeout;
1006
-
1040
+ char *agent_id = is_agent_claimed();
1041
+ if (agent_id && aclk_connected) {
1042
+ freez(agent_id);
1043
// Wakeup thread to cleanup
1044
QUERY_THREAD_WAKEUP;
1009
- // Send a graceful disconnect message
1010
- BUFFER *b = buffer_create(512);
1011
- aclk_create_header(b, "disconnect", NULL, 0, 0);
1012
- buffer_strcat(b, ",\n\t\"payload\": \"graceful\"}\n");
1013
- aclk_send_message(ACLK_METADATA_TOPIC, (char*)buffer_tostring(b), NULL);
1014
- buffer_free(b);
1015
-
1016
- event_loop_timeout = now_realtime_sec() + 5;
1017
- write_q = 1;
1018
- while (write_q && event_loop_timeout > now_realtime_sec()) {
1019
- _link_event_loop();
1020
- lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
1021
- }
1022
-
1023
- aclk_shutting_down = 1;
1024
- _link_shutdown();
1025
- aclk_lws_wss_mqtt_layer_disconect_notif();
1026
-
1027
- write_q = 1;
1028
- event_loop_timeout = now_realtime_sec() + 5;
1029
- while (write_q && event_loop_timeout > now_realtime_sec()) {
1030
- _link_event_loop();
1031
- lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
1032
- }
1045
+ aclk_graceful_disconnect();
1046
}
1047
1048
@@ -1298,23 +1311,32 @@ void aclk_get_challenge(char *aclk_hostname, char *aclk_port)
1311
}
1312
if (aclk_password != NULL )
1313
freez(aclk_password);
1301
- if (aclk_username == NULL)
1302
- aclk_username = strdupz(agent_id);
1314
aclk_password = password.result;
1315
+ if (aclk_username != NULL)
1316
+ freez(aclk_username);
1317
+ aclk_username = agent_id;
1318
+ agent_id = NULL;
1319
1320
CLEANUP:
1321
+ if (agent_id != NULL)
1322
+ freez(agent_id);
1323
freez(data_buffer);
1324
return;
1325
}
1326
1327
static void aclk_try_to_connect(char *hostname, char *port, int port_num)
1328
{
1329
+ if (!aclk_private_key) {
1330
+ error("Cannot try to establish the agent cloud link - no private key available!");
1331
+ return;
1332
+ }
1333
info("Attempting to establish the agent cloud link");
1334
aclk_get_challenge(hostname, port);
1335
if (aclk_password == NULL)
1336
return;
1337
int rc;
1338
aclk_connecting = 1;
1339
+ create_publish_base_topic();
1340
rc = mqtt_attempt_connection(hostname, port_num, aclk_username, aclk_password);
1341
if (unlikely(rc)) {
1342
error("Failed to initialize the agent cloud link library");
@@ -1369,18 +1391,21 @@ void *aclk_main(void *ptr)
1391
uint32_t port_num = 0;
1392
info("Waiting for netdata to be claimed");
1393
while(1) {
1372
- while (likely(!is_agent_claimed())) {
1394
+ char *agent_id = is_agent_claimed();
1395
+ while (likely(!agent_id)) {
1396
sleep_usec(USEC_PER_SEC * 1);
1397
if (netdata_exit)
1398
goto exited;
1399
+ agent_id = is_agent_claimed();
1400
}
1401
+ freez(agent_id);
1402
// The NULL return means the value was never initialised, but this value has been initialized in post_conf_load.
1403
// We trap the impossible NULL here to keep the linter happy without using a fatal() in the code.
1404
char *cloud_base_url = appconfig_get(&cloud_config, CONFIG_SECTION_GLOBAL, "cloud base url", NULL);
1405
if (cloud_base_url == NULL) {
1406
error("Do not move the cloud base url out of post_conf_load!!");
1407
goto exited;
1383
- }
1408
+ }
1409
if (aclk_decode_base_url(cloud_base_url, &aclk_hostname, &aclk_port)) {
1410
error("Agent is claimed but the configuration is invalid, please fix");
1411
}
@@ -1399,14 +1424,19 @@ void *aclk_main(void *ptr)
1424
}
1425
}
1426
1402
- create_publish_base_topic();
1403
-
1427
usec_t reconnect_expiry = 0; // In usecs
1428
1429
while (!netdata_exit) {
1430
static int first_init = 0;
1408
- size_t write_q, write_q_bytes, read_q;
1409
- lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
1431
+ /* size_t write_q, write_q_bytes, read_q;
1432
+ lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);*/
1433
+
1434
+ if (aclk_kill_link) { // User has reloaded the claiming state
1435
+ aclk_kill_link = 0;
1436
+ aclk_graceful_disconnect();
1437
+ create_private_key();
1438
+ continue;
1439
+ }
1440
1441
if (aclk_force_reconnect) {
1442
aclk_lws_wss_destroy_context();
@@ -1577,14 +1607,6 @@ void aclk_disconnect()
1607
aclk_force_reconnect = 1;
1608
}
1609
1580
-void aclk_shutdown()
1581
-{
1582
- info("Shutdown initiated");
1583
- aclk_connected = 0;
1584
- _link_shutdown();
1585
- info("Shutdown complete");
1586
-}
1587
-
1610
inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us)
1611
{
1612
uuid_t uuid;
claim/claim.c
+33
-9
@@ -26,12 +26,19 @@ static char *claiming_errors[] = {
26
"Gateway Timeout", // 16
27
"Service Unavailable" // 17
28
};
29
-
29
+static netdata_mutex_t claim_mutex = NETDATA_MUTEX_INITIALIZER;
30
static char *claimed_id = NULL;
31
32
-char *is_agent_claimed(void)
32
+/* Retrieve the claim id for the agent.
33
+ * Caller owns the string.
34
+*/
35
+char *is_agent_claimed()
36
{
34
- return claimed_id;
37
+ char *result;
38
+ netdata_mutex_lock(&claim_mutex);
39
+ result = (claimed_id == NULL) ? NULL : strdup(claimed_id);
40
+ netdata_mutex_unlock(&claim_mutex);
41
+ return result;
42
}
43
44
#define CLAIMING_COMMAND_LENGTH 16384
@@ -109,12 +116,34 @@ void claim_agent(char *claiming_arguments)
116
#endif
117
}
118
119
+#ifdef ENABLE_ACLK
120
+extern int aclk_connected, aclk_kill_link;
121
+#endif
122
+
123
+/* Change the claimed state of the agent.
124
+ *
125
+ * This only happens when the user has explicitly requested it:
126
+ * - via the cli tool by reloading the claiming state
127
+ * - after spawning the claim because of a command-line argument
128
+ * If this happens with the ACLK active under an old claim then we MUST KILL THE LINK
129
+ */
130
void load_claiming_state(void)
131
{
132
+ // --------------------------------------------------------------------
133
+ // Check if the cloud is enabled
134
+#if defined( DISABLE_CLOUD ) || !defined( ENABLE_ACLK )
135
+ netdata_cloud_setting = 0;
136
+#else
137
+ netdata_mutex_lock(&claim_mutex);
138
if (claimed_id != NULL) {
139
freez(claimed_id);
140
claimed_id = NULL;
141
}
142
+ if (aclk_connected)
143
+ {
144
+ info("Agent was already connected to Cloud - forcing reconnection under new credentials");
145
+ aclk_kill_link = 1;
146
+ }
147
148
// Propagate into aclk and registry. Be kind of atomic...
149
appconfig_get(&cloud_config, CONFIG_SECTION_GLOBAL, "cloud base url", DEFAULT_CLOUD_BASE_URL);
@@ -124,18 +153,13 @@ void load_claiming_state(void)
153
154
long bytes_read;
155
claimed_id = read_by_filename(filename, &bytes_read);
156
+ netdata_mutex_unlock(&claim_mutex); // Only the main thread can call this function, safe to release and then read
157
if (!claimed_id) {
158
info("Unable to load '%s', setting state to AGENT_UNCLAIMED", filename);
159
return;
160
}
161
162
info("File '%s' was found. Setting state to AGENT_CLAIMED.", filename);
133
-
134
- // --------------------------------------------------------------------
135
- // Check if the cloud is enabled
136
-#if defined( DISABLE_CLOUD ) || !defined( ENABLE_ACLK )
137
- netdata_cloud_setting = 0;
138
-#else
163
netdata_cloud_setting = appconfig_get_boolean(&cloud_config, CONFIG_SECTION_GLOBAL, "enabled", 1);
164
#endif
165
}
web/api/web_api_v1.c
+5
-2
@@ -874,10 +874,13 @@ inline int web_client_api_request_v1_info_fill_buffer(RRDHOST *host, BUFFER *wb)
874
#else
875
buffer_strcat(wb, "\t\"cloud-available\": false,\n");
876
#endif
877
- if (is_agent_claimed() == NULL)
877
+ char *agent_id = is_agent_claimed();
878
+ if (agent_id == NULL)
879
buffer_strcat(wb, "\t\"agent-claimed\": false,\n");
879
- else
880
+ else {
881
buffer_strcat(wb, "\t\"agent-claimed\": true,\n");
882
+ freez(agent_id);
883
+ }
884
#ifdef ENABLE_ACLK
885
if (aclk_connected)
886
buffer_strcat(wb, "\t\"aclk-available\": true\n");