Add session-id using connect timestamp (#8633)
Added a session-id to the ACLK messages to overcome a problem with the LWT timestamp being out of sequence with the rest of the message flow.
Andrew Moss committed
Apr 8, 2020 at 19:34 UTC
6b63ba8fe52b490f83c1983645c16f1f49591149
3 files changed
+59
-59
aclk/agent_cloud_link.c
+44
-38
@@ -23,6 +23,8 @@ static char *aclk_password = NULL;
23
static char *global_base_topic = NULL;
24
static int aclk_connecting = 0;
25
int aclk_connected = 0; // Exposed in the web-api
26
+usec_t aclk_session_us = 0; // Used by the mqtt layer
27
+time_t aclk_session_sec = 0; // Used by the mqtt layer
28
29
static netdata_mutex_t aclk_mutex = NETDATA_MUTEX_INITIALIZER;
30
static netdata_mutex_t query_mutex = NETDATA_MUTEX_INITIALIZER;
@@ -750,8 +752,8 @@ int aclk_execute_query(struct aclk_query *this_query)
752
buffer_flush(local_buffer);
753
local_buffer->contenttype = CT_APPLICATION_JSON;
754
753
- aclk_create_header(local_buffer, "http", this_query->msg_id);
754
-
755
+ aclk_create_header(local_buffer, "http", this_query->msg_id, 0, 0);
756
+ buffer_strcat(local_buffer, ",\n\t\"payload\": ");
757
char *encoded_response = aclk_encode_response(w->response.data);
758
759
buffer_sprintf(
@@ -821,11 +823,6 @@ int aclk_process_query()
823
aclk_send_message(this_query->topic, this_query->query, this_query->msg_id);
824
break;
825
824
- case ACLK_CMD_ALARMS:
825
- debug(D_ACLK, "EXECUTING an alarms update command");
826
- aclk_send_alarm_metadata();
827
- break;
828
-
826
case ACLK_CMD_CLOUD:
827
debug(D_ACLK, "EXECUTING a cloud command");
828
aclk_execute_query(this_query);
@@ -939,7 +936,6 @@ void *aclk_query_main_thread(void *ptr)
936
// Thread cleanup
937
static void aclk_main_cleanup(void *ptr)
938
{
942
- char payload[512];
939
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
940
static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
941
@@ -952,24 +948,11 @@ static void aclk_main_cleanup(void *ptr)
948
// Wakeup thread to cleanup
949
QUERY_THREAD_WAKEUP;
950
// Send a graceful disconnect message
955
- char *msg_id = create_uuid();
956
-
957
- usec_t time_created_offset_usec = now_realtime_usec();
958
- time_t time_created = time_created_offset_usec / USEC_PER_SEC;
959
- time_created_offset_usec = time_created_offset_usec % USEC_PER_SEC;
960
-
961
- snprintfz(
962
- payload, 511,
963
- "{ \"type\": \"disconnect\","
964
- " \"msg-id\": \"%s\","
965
- " \"timestamp\": %ld,"
966
- " \"timestamp-offset-usec\": %llu,"
967
- " \"version\": %d,"
968
- " \"payload\": \"graceful\" }",
969
- msg_id, time_created, time_created_offset_usec, ACLK_VERSION);
970
-
971
- aclk_send_message(ACLK_METADATA_TOPIC, payload, msg_id);
972
- freez(msg_id);
951
+ BUFFER *b = buffer_create(512);
952
+ aclk_create_header(b, "disconnect", NULL, 0, 0);
953
+ buffer_strcat(b, ",\n\t\"payload\": \"graceful\"}\n");
954
+ aclk_send_message(ACLK_METADATA_TOPIC, (char*)buffer_tostring(b), NULL);
955
+ buffer_free(b);
956
957
event_loop_timeout = now_realtime_sec() + 5;
958
write_q = 1;
@@ -1514,7 +1497,7 @@ void aclk_shutdown()
1497
info("Shutdown complete");
1498
}
1499
1517
-inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id)
1500
+inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us)
1501
{
1502
uuid_t uuid;
1503
char uuid_str[36 + 1];
@@ -1525,9 +1508,11 @@ inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id)
1508
msg_id = uuid_str;
1509
}
1510
1528
- usec_t time_created_offset_usec = now_realtime_usec();
1529
- time_t time_created = time_created_offset_usec / USEC_PER_SEC;
1530
- time_created_offset_usec = time_created_offset_usec % USEC_PER_SEC;
1511
+ if (ts_secs == 0) {
1512
+ ts_us = now_realtime_usec();
1513
+ ts_secs = ts_us / USEC_PER_SEC;
1514
+ ts_us = ts_us % USEC_PER_SEC;
1515
+ }
1516
1517
buffer_sprintf(
1518
dest,
@@ -1535,11 +1520,12 @@ inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id)
1520
"\t\"msg-id\": \"%s\",\n"
1521
"\t\"timestamp\": %ld,\n"
1522
"\t\"timestamp-offset-usec\": %llu,\n"
1538
- "\t\"version\": %d,\n"
1539
- "\t\"payload\": ",
1540
- type, msg_id, time_created, time_created_offset_usec, ACLK_VERSION);
1523
+ "\t\"connect\": %ld,\n"
1524
+ "\t\"connect-offset-usec\": %llu,\n"
1525
+ "\t\"version\": %d",
1526
+ type, msg_id, ts_secs, ts_us, aclk_session_sec, aclk_session_us, ACLK_VERSION);
1527
1542
- debug(D_ACLK, "Sending v%d msgid [%s] type [%s] time [%ld]", ACLK_VERSION, msg_id, type, time_created);
1528
+ debug(D_ACLK, "Sending v%d msgid [%s] type [%s] time [%ld]", ACLK_VERSION, msg_id, type, ts_secs);
1529
}
1530
1531
/*
@@ -1599,7 +1585,15 @@ void aclk_send_alarm_metadata()
1585
1586
debug(D_ACLK, "Metadata alarms start");
1587
1602
- aclk_create_header(local_buffer, "connect_alarms", msg_id);
1588
+ // on_connect messages are sent on a health reload, if the on_connect message is real then we
1589
+ // use the session time as the fake timestamp to indicate that it starts the session. If it is
1590
+ // a fake on_connect message then use the real timestamp to indicate it is within the existing
1591
+ // session.
1592
+ if (aclk_metadata_submitted == ACLK_METADATA_SENT)
1593
+ aclk_create_header(local_buffer, "connect_alarms", msg_id, 0, 0);
1594
+ else
1595
+ aclk_create_header(local_buffer, "connect_alarms", msg_id, aclk_session_sec, aclk_session_us);
1596
+ buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1597
1598
buffer_sprintf(local_buffer, "{\n\t \"configured-alarms\" : ");
1599
health_alarms2json(localhost, local_buffer, 1);
@@ -1635,7 +1629,16 @@ int aclk_send_info_metadata()
1629
buffer_flush(local_buffer);
1630
local_buffer->contenttype = CT_APPLICATION_JSON;
1631
1638
- aclk_create_header(local_buffer, "connect", msg_id);
1632
+ // on_connect messages are sent on a health reload, if the on_connect message is real then we
1633
+ // use the session time as the fake timestamp to indicate that it starts the session. If it is
1634
+ // a fake on_connect message then use the real timestamp to indicate it is within the existing
1635
+ // session.
1636
+ if (aclk_metadata_submitted == ACLK_METADATA_SENT)
1637
+ aclk_create_header(local_buffer, "connect", msg_id, 0, 0);
1638
+ else
1639
+ aclk_create_header(local_buffer, "connect", msg_id, aclk_session_sec, aclk_session_us);
1640
+ buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1641
+
1642
buffer_sprintf(local_buffer, "{\n\t \"info\" : ");
1643
web_client_api_request_v1_info_fill_buffer(localhost, local_buffer);
1644
debug(D_ACLK, "Metadata %s with info has %zu bytes", msg_id, local_buffer->len);
@@ -1728,7 +1731,9 @@ int aclk_send_single_chart(char *hostname, char *chart)
1731
buffer_flush(local_buffer);
1732
local_buffer->contenttype = CT_APPLICATION_JSON;
1733
1731
- aclk_create_header(local_buffer, "chart", msg_id);
1734
+ aclk_create_header(local_buffer, "chart", msg_id, 0, 0);
1735
+ buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1736
+
1737
rrdset2json(st, local_buffer, NULL, NULL, 1);
1738
buffer_sprintf(local_buffer, "\t\n}");
1739
@@ -1793,7 +1798,8 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1798
char *msg_id = create_uuid();
1799
1800
buffer_flush(local_buffer);
1796
- aclk_create_header(local_buffer, "status-change", msg_id);
1801
+ aclk_create_header(local_buffer, "status-change", msg_id, 0, 0);
1802
+ buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1803
1804
netdata_rwlock_rdlock(&host->health_log.alarm_log_rwlock);
1805
health_alarm_entry2json_nolock(local_buffer, ae, host);
aclk/agent_cloud_link.h
+1
-2
@@ -44,7 +44,6 @@ typedef enum aclk_cmd {
44
ACLK_CMD_CHART,
45
ACLK_CMD_CHARTDEL,
46
ACLK_CMD_ALARM,
47
- ACLK_CMD_ALARMS,
47
ACLK_CMD_MAX
48
} ACLK_CMD;
49
@@ -98,7 +97,7 @@ struct aclk_query *
97
aclk_query_find(char *token, char *data, char *msg_id, char *query, ACLK_CMD cmd, struct aclk_query **last_query);
98
int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd);
99
int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae);
101
-void aclk_create_header(BUFFER *dest, char *type, char *msg_id);
100
+void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us);
101
int aclk_handle_cloud_request(char *payload);
102
int aclk_submit_request(struct aclk_request *);
103
void aclk_add_collector(const char *hostname, const char *plugin_name, const char *module_name);
aclk/mqtt.c
+14
-19
@@ -5,6 +5,9 @@
5
#include "mqtt.h"
6
#include "aclk_lws_wss_client.h"
7
8
+extern usec_t aclk_session_us;
9
+extern time_t aclk_session_sec;
10
+
11
inline const char *_link_strerror(int rc)
12
{
13
return mosquitto_strerror(rc);
@@ -131,6 +134,11 @@ static int _mqtt_create_connection(char *username, char *password)
134
return MOSQ_ERR_UNKNOWN;
135
}
136
137
+ // Record the session start time to allow a nominal LWT timestamp
138
+ usec_t now = now_realtime_usec();
139
+ aclk_session_sec = now / USEC_PER_SEC;
140
+ aclk_session_us = now % USEC_PER_SEC;
141
+
142
_link_set_lwt("outbound/meta", 2);
143
144
mosquitto_connect_callback_set(mosq, connect_callback);
@@ -259,7 +267,6 @@ int _link_set_lwt(char *sub_topic, int qos)
267
{
268
int rc;
269
char topic[ACLK_MAX_TOPIC + 1];
262
- char payload[512];
270
char *final_topic;
271
272
final_topic = get_topic(sub_topic, topic, ACLK_MAX_TOPIC);
@@ -269,25 +276,13 @@ int _link_set_lwt(char *sub_topic, int qos)
276
return 1;
277
}
278
272
- usec_t time_created_offset_usec = now_realtime_usec();
273
- time_t time_created = time_created_offset_usec / USEC_PER_SEC;
274
- time_created_offset_usec = time_created_offset_usec % USEC_PER_SEC;
275
-
276
- char *msg_id = create_uuid();
277
-
278
- snprintfz(
279
- payload, 511,
280
- "{ \"type\": \"disconnect\","
281
- " \"msg-id\": \"%s\","
282
- " \"timestamp\": %ld,"
283
- " \"timestamp-offset-usec\": %llu,"
284
- " \"version\": %d,"
285
- " \"payload\": \"unexpected\" }",
286
- msg_id, time_created, time_created_offset_usec, ACLK_VERSION);
287
-
288
- freez(msg_id);
279
+ usec_t lwt_time = aclk_session_sec * USEC_PER_SEC + aclk_session_us + 1;
280
+ BUFFER *b = buffer_create(512);
281
+ aclk_create_header(b, "disconnect", NULL, lwt_time / USEC_PER_SEC, lwt_time % USEC_PER_SEC);
282
+ buffer_strcat(b, ", \"payload\": \"unexpected\" }");
283
+ rc = mosquitto_will_set(mosq, topic, buffer_strlen(b), buffer_tostring(b), qos, 0);
284
+ buffer_free(b);
285
290
- rc = mosquitto_will_set(mosq, topic, strlen(payload), (const void *) payload, qos, 0);
286
return rc;
287
}
288