ACLK: Implemented Last Will and Testament (#8410)
* Added support for Last Will and Testament to the ACLK * On normal agent shutdown an alternate "graceful shutdown" message is published
Stelios Fragkakis committed
Mar 18, 2020 at 21:37 UTC
2c716dc31ee231372e10a705118703f9f23f4885
5 files changed
+103
-8
aclk/aclk_lws_wss_client.c
+11
@@ -6,6 +6,8 @@
6
#include "../daemon/common.h"
7
#include "aclk_common.h"
8
9
+extern int aclk_shutting_down;
10
+
11
static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len);
12
13
struct aclk_lws_wss_perconnect_data {
@@ -320,6 +322,8 @@ static const char *aclk_lws_callback_name(enum lws_callback_reasons reason)
322
return "LWS_CALLBACK_CLIENT_ESTABLISHED";
323
case LWS_CALLBACK_OPENSSL_PERFORM_SERVER_CERT_VERIFICATION:
324
return "LWS_CALLBACK_OPENSSL_PERFORM_SERVER_CERT_VERIFICATION";
325
+ case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
326
+ return "LWS_CALLBACK_EVENT_WAIT_CANCELLED";
327
default:
328
// Not using an internal buffer here for thread-safety with unknown calling context.
329
error("Unknown LWS callback %u", reason);
@@ -331,6 +335,13 @@ static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reas
335
UNUSED(user);
336
struct lws_wss_packet_buffer *data;
337
int retval = 0;
338
+ static int lws_shutting_down = 0;
339
+
340
+ if (unlikely(aclk_shutting_down && !lws_shutting_down)) {
341
+ lws_shutting_down = 1;
342
+ retval = -1;
343
+ engine_instance->upstream_reconnect_request = 0;
344
+ }
345
346
// Callback servicing is forced when we are closed from above.
347
if (engine_instance->upstream_reconnect_request) {
aclk/agent_cloud_link.c
+48
-5
@@ -5,6 +5,7 @@
5
#include "aclk_lws_https_client.h"
6
#include "aclk_common.h"
7
8
+int aclk_shutting_down = 0;
9
// State-machine for the on-connect metadata transmission.
10
// TODO: The AGENT_STATE should be centralized as it would be useful to control error-logging during the initial
11
// agent startup phase.
@@ -43,6 +44,8 @@ pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
44
#define QUERY_THREAD_UNLOCK pthread_mutex_unlock(&query_lock_wait)
45
#define QUERY_THREAD_WAKEUP pthread_cond_signal(&query_cond_wait)
46
47
+void lws_wss_check_queues(size_t *write_len, size_t *write_len_bytes, size_t *read_len);
48
+
49
/*
50
* Maintain a list of collectors and chart count
51
* If all the charts of a collector are deleted
@@ -936,13 +939,54 @@ void *aclk_query_main_thread(void *ptr)
939
// Thread cleanup
940
static void aclk_main_cleanup(void *ptr)
941
{
942
+ char payload[512];
943
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
944
static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
945
946
info("cleaning up...");
947
944
- // Wakeup thread to cleanup
945
- QUERY_THREAD_WAKEUP;
948
+ if (is_agent_claimed() && aclk_connected) {
949
+ size_t write_q, write_q_bytes, read_q;
950
+ time_t event_loop_timeout;
951
+
952
+ // Wakeup thread to cleanup
953
+ QUERY_THREAD_WAKEUP;
954
+ // Send a graceful disconnect message
955
+ time_t time_created = now_realtime_sec();
956
+ char *msg_id = create_uuid();
957
+
958
+ snprintfz(
959
+ payload, 511,
960
+ "{ \"type\": \"disconnect\","
961
+ " \"msg-id\": \"%s\","
962
+ " \"timestamp\": %ld,"
963
+ " \"version\": %d,"
964
+ " \"payload\": \"graceful\" }",
965
+ msg_id, time_created, ACLK_VERSION);
966
+
967
+ aclk_send_message(ACLK_METADATA_TOPIC, payload, msg_id);
968
+ freez(msg_id);
969
+
970
+ event_loop_timeout = now_realtime_sec() + 5;
971
+ write_q = 1;
972
+ while (write_q && event_loop_timeout > now_realtime_sec()) {
973
+ _link_event_loop();
974
+ lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
975
+ }
976
+
977
+ aclk_shutting_down = 1;
978
+ _link_shutdown();
979
+ aclk_lws_wss_mqtt_layer_disconect_notif();
980
+
981
+ write_q = 1;
982
+ event_loop_timeout = now_realtime_sec() + 5;
983
+ while (write_q && event_loop_timeout > now_realtime_sec()) {
984
+ _link_event_loop();
985
+ lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
986
+ }
987
+ }
988
+
989
+ info("Disconnected");
990
991
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
992
}
@@ -1243,7 +1287,6 @@ static void aclk_try_to_connect(char *hostname, char *port, int port_num)
1287
*
1288
* @return It always returns NULL
1289
*/
1246
-void lws_wss_check_queues(size_t *write_len, size_t *write_len_bytes, size_t *read_len);
1290
void *aclk_main(void *ptr)
1291
{
1292
struct netdata_static_thread *query_thread;
@@ -1295,8 +1338,8 @@ void *aclk_main(void *ptr)
1338
size_t write_q, write_q_bytes, read_q;
1339
lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
1340
//info("loop state first_init_%d connected=%d connecting=%d wq=%zu (%zu-bytes) rq=%zu",
1298
- // first_init, aclk_connected, aclk_connecting, write_q, write_q_bytes, read_q);
1299
- if (unlikely(!aclk_connected)) {
1341
+ // first_init, aclk_connected, aclk_connecting, write_q, write_q_bytes, read_q);
1342
+ if (unlikely(!netdata_exit && !aclk_connected)) {
1343
if (unlikely(!first_init)) {
1344
aclk_try_to_connect(aclk_hostname, aclk_port, port_num);
1345
first_init = 1;
aclk/agent_cloud_link.h
+1
@@ -78,6 +78,7 @@ extern int aclk_send_message(char *sub_topic, char *message, char *msg_id);
78
//char *get_base_topic();
79
80
extern char *is_agent_claimed(void);
81
+extern void aclk_lws_wss_mqtt_layer_disconect_notif();
82
char *create_uuid();
83
84
// callbacks for agent cloud link
aclk/mqtt.c
+40
-3
@@ -131,6 +131,8 @@ static int _mqtt_create_connection(char *username, char *password)
131
return MOSQ_ERR_UNKNOWN;
132
}
133
134
+ _link_set_lwt("outbound/meta", 2);
135
+
136
mosquitto_connect_callback_set(mosq, connect_callback);
137
mosquitto_disconnect_callback_set(mosq, disconnect_callback);
138
mosquitto_publish_callback_set(mosq, publish_callback);
@@ -174,6 +176,10 @@ static inline void _link_mosquitto_write()
176
{
177
int rc;
178
179
+ if (unlikely(!mosq)) {
180
+ return;
181
+ }
182
+
183
rc = mosquitto_loop_misc(mosq);
184
if (unlikely(rc != MOSQ_ERR_SUCCESS))
185
debug(D_ACLK, "ACLK: failure during mosquitto_loop_misc %s", mosquitto_strerror(rc));
@@ -234,6 +240,9 @@ void _link_shutdown()
240
{
241
int rc;
242
243
+ if (likely(!mosq))
244
+ return;
245
+
246
rc = mosquitto_disconnect(mosq);
247
switch (rc) {
248
case MOSQ_ERR_SUCCESS:
@@ -243,11 +252,39 @@ void _link_shutdown()
252
info("MQTT invalid structure");
253
break;
254
};
255
+}
256
+
257
247
- mosquitto_destroy(mosq);
248
- mosq = NULL;
258
+int _link_set_lwt(char *sub_topic, int qos)
259
+{
260
+ int rc;
261
+ char topic[ACLK_MAX_TOPIC + 1];
262
+ char payload[512];
263
+ char *final_topic;
264
250
- aclk_lws_wss_client_destroy();
265
+ final_topic = get_topic(sub_topic, topic, ACLK_MAX_TOPIC);
266
+ if (unlikely(!final_topic)) {
267
+ errno = 0;
268
+ error("Unable to build outgoing topic; truncated?");
269
+ return 1;
270
+ }
271
+
272
+ time_t time_created = now_realtime_sec();
273
+ char *msg_id = create_uuid();
274
+
275
+ snprintfz(
276
+ payload, 511,
277
+ "{ \"type\": \"disconnect\","
278
+ " \"msg-id\": \"%s\","
279
+ " \"timestamp\": %ld,"
280
+ " \"version\": %d,"
281
+ " \"payload\": \"unexpected\" }",
282
+ msg_id, time_created, ACLK_VERSION);
283
+
284
+ freez(msg_id);
285
+
286
+ rc = mosquitto_will_set(mosq, topic, strlen(payload), (const void *) payload, qos, 0);
287
+ return rc;
288
}
289
290
int _link_subscribe(char *topic, int qos)
aclk/mqtt.h
+3
@@ -16,7 +16,10 @@ int _mqtt_lib_init();
16
int _link_subscribe(char *topic, int qos);
17
int _link_send_message(char *topic, unsigned char *message, int *mid);
18
const char *_link_strerror(int rc);
19
+int _link_set_lwt(char *topic, int qos);
20
+
21
22
int aclk_handle_cloud_request(char *);
23
+extern char *get_topic(char *sub_topic, char *final_topic, int max_size);
24
25
#endif //NETDATA_MQTT_H