ACLK Version Negotiation (#9819)
* implements version negotiation for ACLK
Timotej S committed
Sep 8, 2020 at 11:14 UTC
ae7a9aa7ed8109eda7f681fa7583dbb22dce0172
8 files changed
+261
-71
aclk/aclk_common.c
+5
-1
@@ -4,10 +4,14 @@
4
5
netdata_mutex_t aclk_shared_state_mutex = NETDATA_MUTEX_INITIALIZER;
6
7
+int aclk_disable_runtime = 0;
8
+
9
struct aclk_shared_state aclk_shared_state = {
10
.metadata_submitted = ACLK_METADATA_REQUIRED,
11
.agent_state = AGENT_INITIALIZING,
10
- .last_popcorn_interrupt = 0
12
+ .last_popcorn_interrupt = 0,
13
+ .version_neg = 0,
14
+ .version_neg_wait_till = 0
15
};
16
17
struct {
aclk/aclk_common.h
+25
@@ -7,6 +7,24 @@ extern netdata_mutex_t aclk_shared_state_mutex;
7
#define ACLK_SHARED_STATE_LOCK netdata_mutex_lock(&aclk_shared_state_mutex)
8
#define ACLK_SHARED_STATE_UNLOCK netdata_mutex_unlock(&aclk_shared_state_mutex)
9
10
+// minimum and maximum supported version of ACLK
11
+// in this version of agent
12
+#define ACLK_VERSION_MIN 1
13
+#define ACLK_VERSION_MAX 1
14
+
15
+// Version negotiation messages have they own versioning
16
+// this is also used for LWT message as we set that up
17
+// before version negotiation
18
+#define ACLK_VERSION_NEG_VERSION 1
19
+
20
+// Maximum time to wait for version negotiation before aborting
21
+// and defaulting to oldest supported version
22
+#define VERSION_NEG_TIMEOUT 3
23
+
24
+#if ACLK_VERSION_MIN > ACLK_VERSION_MAX
25
+#error "ACLK_VERSION_MAX must be >= than ACLK_VERSION_MIN"
26
+#endif
27
+
28
typedef enum aclk_cmd {
29
ACLK_CMD_CLOUD,
30
ACLK_CMD_ONCONNECT,
@@ -31,6 +49,11 @@ extern struct aclk_shared_state {
49
ACLK_METADATA_STATE metadata_submitted;
50
ACLK_AGENT_STATE agent_state;
51
time_t last_popcorn_interrupt;
52
+
53
+ // read only while ACLK connected
54
+ // protect by lock otherwise
55
+ int version_neg;
56
+ usec_t version_neg_wait_till;
57
} aclk_shared_state;
58
59
typedef enum aclk_proxy_type {
@@ -53,4 +76,6 @@ void safe_log_proxy_censor(char *proxy);
76
int aclk_decode_base_url(char *url, char **aclk_hostname, char **aclk_port);
77
const char *aclk_get_proxy(ACLK_PROXY_TYPE *type);
78
79
+extern int aclk_disable_runtime;
80
+
81
#endif //ACLK_COMMON_H
aclk/aclk_query.c
+19
-4
@@ -348,7 +348,7 @@ static int aclk_execute_query(struct aclk_query *this_query)
348
buffer_flush(local_buffer);
349
local_buffer->contenttype = CT_APPLICATION_JSON;
350
351
- aclk_create_header(local_buffer, "http", this_query->msg_id, 0, 0);
351
+ aclk_create_header(local_buffer, "http", this_query->msg_id, 0, 0, aclk_shared_state.version_neg);
352
buffer_strcat(local_buffer, ",\n\t\"payload\": ");
353
char *encoded_response = aclk_encode_response(w->response.data->buffer, w->response.data->len, 0);
354
char *encoded_header = aclk_encode_response(w->response.header_output->buffer, w->response.header_output->len, 1);
@@ -537,16 +537,31 @@ void *aclk_query_main_thread(void *ptr)
537
}
538
539
while (!netdata_exit) {
540
+ if(aclk_disable_runtime) {
541
+ sleep(1);
542
+ continue;
543
+ }
544
ACLK_SHARED_STATE_LOCK;
541
- if (unlikely(!aclk_shared_state.metadata_submitted)) {
542
- ACLK_SHARED_STATE_UNLOCK;
545
+ if (unlikely(!aclk_shared_state.version_neg)) {
546
+ if (!aclk_shared_state.version_neg_wait_till || aclk_shared_state.version_neg_wait_till > now_monotonic_usec()) {
547
+ ACLK_SHARED_STATE_UNLOCK;
548
+ info("Waiting for ACLK Version Negotiation message from Cloud");
549
+ sleep(1);
550
+ continue;
551
+ }
552
+ errno = 0;
553
+ error("ACLK version negotiation failed. No reply to \"hello\" with \"version\" from cloud in time of %ds."
554
+ " Reverting to default ACLK version of %d.", VERSION_NEG_TIMEOUT, ACLK_VERSION_MIN);
555
+ aclk_shared_state.version_neg = ACLK_VERSION_MIN;
556
+ }
557
+ if (unlikely(aclk_shared_state.metadata_submitted == ACLK_METADATA_REQUIRED)) {
558
if (unlikely(aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
559
+ ACLK_SHARED_STATE_UNLOCK;
560
errno = 0;
561
error("ACLK failed to queue on_connect command");
562
sleep(1);
563
continue;
564
}
549
- ACLK_SHARED_STATE_LOCK;
565
aclk_shared_state.metadata_submitted = ACLK_METADATA_CMD_QUEUED;
566
}
567
ACLK_SHARED_STATE_UNLOCK;
aclk/agent_cloud_link.c
+203
-59
@@ -100,6 +100,15 @@ int cloud_to_agent_parse(JSON_ENTRY *e)
100
data->version = e->data.number;
101
break;
102
}
103
+ if (!strcmp(e->name, "min-version")) {
104
+ data->min_version = e->data.number;
105
+ break;
106
+ }
107
+ if (!strcmp(e->name, "max-version")) {
108
+ data->max_version = e->data.number;
109
+ break;
110
+ }
111
+
112
break;
113
114
case JSON_BOOLEAN:
@@ -513,7 +522,7 @@ static void aclk_graceful_disconnect()
522
523
// Send a graceful disconnect message
524
BUFFER *b = buffer_create(512);
516
- aclk_create_header(b, "disconnect", NULL, 0, 0);
525
+ aclk_create_header(b, "disconnect", NULL, 0, 0, aclk_shared_state.version_neg);
526
buffer_strcat(b, ",\n\t\"payload\": \"graceful\"}\n");
527
aclk_send_message(ACLK_METADATA_TOPIC, (char*)buffer_tostring(b), NULL);
528
buffer_free(b);
@@ -820,12 +829,36 @@ static void aclk_try_to_connect(char *hostname, char *port, int port_num)
829
int rc;
830
aclk_connecting = 1;
831
create_publish_base_topic();
832
+ ACLK_SHARED_STATE_LOCK;
833
+ aclk_shared_state.version_neg = 0;
834
+ aclk_shared_state.version_neg_wait_till = 0;
835
+ ACLK_SHARED_STATE_UNLOCK;
836
rc = mqtt_attempt_connection(hostname, port_num, aclk_username, aclk_password);
837
if (unlikely(rc)) {
838
error("Failed to initialize the agent cloud link library");
839
}
840
}
841
842
+// Sends "hello" message to negotiate ACLK version with cloud
843
+static inline void aclk_hello_msg()
844
+{
845
+ BUFFER *buf = buffer_create(NETDATA_WEB_RESPONSE_HEADER_SIZE);
846
+
847
+ char *msg_id = create_uuid();
848
+
849
+ ACLK_SHARED_STATE_LOCK;
850
+ aclk_shared_state.version_neg = 0;
851
+ aclk_shared_state.version_neg_wait_till = now_monotonic_usec() + USEC_PER_SEC * VERSION_NEG_TIMEOUT;
852
+ ACLK_SHARED_STATE_UNLOCK;
853
+
854
+ //Hello message is versioned separatelly from the rest of the protocol
855
+ aclk_create_header(buf, "hello", msg_id, 0, 0, ACLK_VERSION_NEG_VERSION);
856
+ buffer_sprintf(buf, ",\"min-version\":%d,\"max-version\":%d}", ACLK_VERSION_MIN, ACLK_VERSION_MAX);
857
+ aclk_send_message(ACLK_METADATA_TOPIC, buf->buffer, msg_id);
858
+ freez(msg_id);
859
+ buffer_free(buf);
860
+}
861
+
862
/**
863
* Main agent cloud link thread
864
*
@@ -932,6 +965,11 @@ void *aclk_main(void *ptr)
965
/* size_t write_q, write_q_bytes, read_q;
966
lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);*/
967
968
+ if (aclk_disable_runtime && !aclk_connected) {
969
+ sleep(1);
970
+ continue;
971
+ }
972
+
973
if (aclk_kill_link) { // User has reloaded the claiming state
974
aclk_kill_link = 0;
975
aclk_graceful_disconnect();
@@ -978,9 +1016,9 @@ void *aclk_main(void *ptr)
1016
stress_counter = 0;
1017
}*/
1018
981
- // TODO: Move to on-connect
1019
if (unlikely(!aclk_subscribed)) {
1020
aclk_subscribed = !aclk_subscribe(ACLK_COMMAND_TOPIC, 1);
1021
+ aclk_hello_msg();
1022
}
1023
1024
if (unlikely(!query_threads.thread_list)) {
@@ -1117,6 +1155,7 @@ void aclk_connect()
1155
1156
aclk_connected = 1;
1157
aclk_reconnect_delay(0);
1158
+
1159
QUERY_THREAD_WAKEUP;
1160
return;
1161
}
@@ -1138,7 +1177,7 @@ void aclk_disconnect()
1177
aclk_force_reconnect = 1;
1178
}
1179
1141
-inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us)
1180
+inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us, int version)
1181
{
1182
uuid_t uuid;
1183
char uuid_str[36 + 1];
@@ -1164,9 +1203,9 @@ inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts
1203
"\t\"connect\": %ld,\n"
1204
"\t\"connect-offset-usec\": %llu,\n"
1205
"\t\"version\": %d",
1167
- type, msg_id, ts_secs, ts_us, aclk_session_sec, aclk_session_us, ACLK_VERSION);
1206
+ type, msg_id, ts_secs, ts_us, aclk_session_sec, aclk_session_us, version);
1207
1169
- debug(D_ACLK, "Sending v%d msgid [%s] type [%s] time [%ld]", ACLK_VERSION, msg_id, type, ts_secs);
1208
+ debug(D_ACLK, "Sending v%d msgid [%s] type [%s] time [%ld]", version, msg_id, type, ts_secs);
1209
}
1210
1211
@@ -1194,9 +1233,9 @@ void aclk_send_alarm_metadata(ACLK_METADATA_STATE metadata_submitted)
1233
// session.
1234
1235
if (metadata_submitted == ACLK_METADATA_SENT)
1197
- aclk_create_header(local_buffer, "connect_alarms", msg_id, 0, 0);
1236
+ aclk_create_header(local_buffer, "connect_alarms", msg_id, 0, 0, aclk_shared_state.version_neg);
1237
else
1199
- aclk_create_header(local_buffer, "connect_alarms", msg_id, aclk_session_sec, aclk_session_us);
1238
+ aclk_create_header(local_buffer, "connect_alarms", msg_id, aclk_session_sec, aclk_session_us, aclk_shared_state.version_neg);
1239
buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1240
1241
@@ -1239,9 +1278,9 @@ int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted)
1278
// a fake on_connect message then use the real timestamp to indicate it is within the existing
1279
// session.
1280
if (metadata_submitted == ACLK_METADATA_SENT)
1242
- aclk_create_header(local_buffer, "update", msg_id, 0, 0);
1281
+ aclk_create_header(local_buffer, "update", msg_id, 0, 0, aclk_shared_state.version_neg);
1282
else
1244
- aclk_create_header(local_buffer, "connect", msg_id, aclk_session_sec, aclk_session_us);
1283
+ aclk_create_header(local_buffer, "connect", msg_id, aclk_session_sec, aclk_session_us, aclk_shared_state.version_neg);
1284
buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1285
1286
buffer_sprintf(local_buffer, "{\n\t \"info\" : ");
@@ -1341,7 +1380,7 @@ int aclk_send_single_chart(char *hostname, char *chart)
1380
buffer_flush(local_buffer);
1381
local_buffer->contenttype = CT_APPLICATION_JSON;
1382
1344
- aclk_create_header(local_buffer, "chart", msg_id, 0, 0);
1383
+ aclk_create_header(local_buffer, "chart", msg_id, 0, 0, aclk_shared_state.version_neg);
1384
buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1385
1386
rrdset2json(st, local_buffer, NULL, NULL, 1);
@@ -1418,7 +1457,7 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1457
char *msg_id = create_uuid();
1458
1459
buffer_flush(local_buffer);
1421
- aclk_create_header(local_buffer, "status-change", msg_id, 0, 0);
1460
+ aclk_create_header(local_buffer, "status-change", msg_id, 0, 0, aclk_shared_state.version_neg);
1461
buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1462
1463
netdata_rwlock_rdlock(&host->health_log.alarm_log_rwlock);
@@ -1443,75 +1482,180 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1482
/*
1483
* Parse the incoming payload and queue a command if valid
1484
*/
1446
-int aclk_handle_cloud_request(char *payload)
1485
+static int aclk_handle_cloud_request(struct aclk_request *cloud_to_agent)
1486
{
1448
- struct aclk_request cloud_to_agent = {
1449
- .type_id = NULL, .msg_id = NULL, .callback_topic = NULL, .payload = NULL, .version = 0
1450
- };
1451
-
1452
- if (aclk_stats_enabled) {
1453
- ACLK_STATS_LOCK;
1454
- aclk_metrics_per_sample.cloud_req_recvd++;
1455
- ACLK_STATS_UNLOCK;
1456
- }
1457
-
1487
+ errno = 0;
1488
ACLK_SHARED_STATE_LOCK;
1489
if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
1460
- debug(D_ACLK, "Ignoring cloud request; agent not in stable state");
1490
+ debug(D_ACLK, "Ignoring \"http\" cloud request; agent not in stable state");
1491
ACLK_SHARED_STATE_UNLOCK;
1462
- return 0;
1492
+ return 1;
1493
}
1494
ACLK_SHARED_STATE_UNLOCK;
1495
1466
- if (unlikely(!payload)) {
1467
- debug(D_ACLK, "ACLK incoming message is empty");
1468
- return 0;
1496
+ if (unlikely(cloud_to_agent->version != aclk_shared_state.version_neg)) {
1497
+ error("Received \"http\" message from Cloud with version %d, but ACLK version %d is used", cloud_to_agent->version, aclk_shared_state.version_neg);
1498
+ return 1;
1499
}
1500
1471
- debug(D_ACLK, "ACLK incoming message (%s)", payload);
1501
+ if (unlikely(!cloud_to_agent->payload)) {
1502
+ error("payload missing");
1503
+ return 1;
1504
+ }
1505
+
1506
+ if (unlikely(!cloud_to_agent->callback_topic)) {
1507
+ error("callback_topic missing");
1508
+ return 1;
1509
+ }
1510
+
1511
+ if (unlikely(!cloud_to_agent->msg_id)) {
1512
+ error("msg_id missing");
1513
+ return 1;
1514
+ }
1515
1473
- int rc = json_parse(payload, &cloud_to_agent, cloud_to_agent_parse);
1516
+ if (unlikely(aclk_queue_query(cloud_to_agent->callback_topic, NULL, cloud_to_agent->msg_id, cloud_to_agent->payload, 0, 0, ACLK_CMD_CLOUD)))
1517
+ debug(D_ACLK, "ACLK failed to queue incoming \"http\" message");
1518
+
1519
+ // Note: the payload comes from the callback and it will be automatically freed
1520
+ return 0;
1521
+}
1522
+
1523
+// This handles `version` message from cloud used to negotiate
1524
+// protocol version we will use
1525
+static int aclk_handle_version_response(struct aclk_request *cloud_to_agent)
1526
+{
1527
+ int version = -1;
1528
+ errno = 0;
1529
+
1530
+ if(unlikely(cloud_to_agent->version != ACLK_VERSION_NEG_VERSION)) {
1531
+ error("Unsuported version of \"version\" message from cloud. Expected %d, Got %d", ACLK_VERSION_NEG_VERSION, cloud_to_agent->version);
1532
+ return 1;
1533
+ }
1534
+ if(unlikely(!cloud_to_agent->min_version)) {
1535
+ error("Min version missing or 0");
1536
+ return 1;
1537
+ }
1538
+ if(unlikely(!cloud_to_agent->max_version)) {
1539
+ error("Max version missing or 0");
1540
+ return 1;
1541
+ }
1542
+ if(unlikely(cloud_to_agent->max_version < cloud_to_agent->min_version)) {
1543
+ error("Max version (%d) must be >= than min version (%d)", cloud_to_agent->max_version, cloud_to_agent->min_version);
1544
+ return 1;
1545
+ }
1546
1475
- if (unlikely(
1476
- JSON_OK != rc || !cloud_to_agent.payload || !cloud_to_agent.callback_topic || !cloud_to_agent.msg_id ||
1477
- !cloud_to_agent.type_id || cloud_to_agent.version > ACLK_VERSION ||
1478
- strcmp(cloud_to_agent.type_id, "http"))) {
1479
- if (JSON_OK != rc)
1480
- error("Malformed json request (%s)", payload);
1547
+ if(unlikely(cloud_to_agent->min_version > ACLK_VERSION_MAX)) {
1548
+ error("Agent too old for this cloud. Minimum version required by cloud %d. Maximum version supported by this agent %d.", cloud_to_agent->min_version, ACLK_VERSION_MAX);
1549
+ aclk_kill_link = 1;
1550
+ aclk_disable_runtime = 1;
1551
+ return 1;
1552
+ }
1553
+ if(unlikely(cloud_to_agent->max_version < ACLK_VERSION_MIN)) {
1554
+ error("Cloud version is too old for this agent. Maximum version supported by cloud %d. Minimum (oldest) version supported by this agent %d.", cloud_to_agent->max_version, ACLK_VERSION_MIN);
1555
+ aclk_kill_link = 1;
1556
+ return 1;
1557
+ }
1558
+
1559
+ version = MIN(cloud_to_agent->max_version, ACLK_VERSION_MAX);
1560
1482
- if (cloud_to_agent.version > ACLK_VERSION)
1483
- error("Unsupported version in JSON request %d", cloud_to_agent.version);
1561
+ ACLK_SHARED_STATE_LOCK;
1562
+ if (unlikely(now_monotonic_usec() > aclk_shared_state.version_neg_wait_till)) {
1563
+ errno = 0;
1564
+ error("The \"version\" message came too late ignoring.");
1565
+ goto err_cleanup;
1566
+ }
1567
+ if (unlikely(aclk_shared_state.version_neg)) {
1568
+ errno = 0;
1569
+ error("Version has already been set to %d", aclk_shared_state.version_neg);
1570
+ goto err_cleanup;
1571
+ }
1572
+ aclk_shared_state.version_neg = version;
1573
+ ACLK_SHARED_STATE_UNLOCK;
1574
1485
- if (cloud_to_agent.payload)
1486
- freez(cloud_to_agent.payload);
1575
+ info("Choosing version %d of ACLK", version);
1576
1488
- if (cloud_to_agent.type_id)
1489
- freez(cloud_to_agent.type_id);
1577
+ return 0;
1578
1491
- if (cloud_to_agent.msg_id)
1492
- freez(cloud_to_agent.msg_id);
1579
+err_cleanup:
1580
+ ACLK_SHARED_STATE_UNLOCK;
1581
+ return 1;
1582
+}
1583
1494
- if (cloud_to_agent.callback_topic)
1495
- freez(cloud_to_agent.callback_topic);
1584
+struct {
1585
+ char *name;
1586
+ int(*fnc)(struct aclk_request *cloud_to_agent);
1587
+} aclk_incoming_msg_types[] = {
1588
+ { .name = "http", .fnc = aclk_handle_cloud_request },
1589
+ { .name = "version", .fnc = aclk_handle_version_response },
1590
+ { .name = NULL, .fnc = NULL }
1591
+};
1592
1497
- if (aclk_stats_enabled) {
1498
- ACLK_STATS_LOCK;
1499
- aclk_metrics_per_sample.cloud_req_err++;
1500
- ACLK_STATS_UNLOCK;
1501
- }
1593
+int aclk_handle_cloud_message(char *payload)
1594
+{
1595
+ struct aclk_request cloud_to_agent;
1596
+ memset(&cloud_to_agent, 0, sizeof(struct aclk_request));
1597
1503
- return 1;
1598
+ if (aclk_stats_enabled) {
1599
+ ACLK_STATS_LOCK;
1600
+ aclk_metrics_per_sample.cloud_req_recvd++;
1601
+ ACLK_STATS_UNLOCK;
1602
}
1603
1506
- // Checked to be "http", not needed anymore
1507
- if (likely(cloud_to_agent.type_id)) {
1508
- freez(cloud_to_agent.type_id);
1509
- cloud_to_agent.type_id = NULL;
1604
+ if (unlikely(!payload)) {
1605
+ errno = 0;
1606
+ error("ACLK incoming message is empty");
1607
+ goto err_cleanup_nojson;
1608
}
1609
1512
- if (unlikely(aclk_queue_query(cloud_to_agent.callback_topic, NULL, cloud_to_agent.msg_id, cloud_to_agent.payload, 0, 0, ACLK_CMD_CLOUD)))
1513
- debug(D_ACLK, "ACLK failed to queue incoming message (%s)", payload);
1610
+ debug(D_ACLK, "ACLK incoming message (%s)", payload);
1611
1515
- // Note: the payload comes from the callback and it will be automatically freed
1516
- return 0;
1612
+ int rc = json_parse(payload, &cloud_to_agent, cloud_to_agent_parse);
1613
+
1614
+ if (unlikely(rc != JSON_OK)) {
1615
+ errno = 0;
1616
+ error("Malformed json request (%s)", payload);
1617
+ goto err_cleanup;
1618
+ }
1619
+
1620
+ if (!cloud_to_agent.type_id) {
1621
+ errno = 0;
1622
+ error("Cloud message is missing compulsory key \"type\"");
1623
+ goto err_cleanup;
1624
+ }
1625
+
1626
+ for (int i = 0; aclk_incoming_msg_types[i].name; i++) {
1627
+ if (strcmp(cloud_to_agent.type_id, aclk_incoming_msg_types[i].name) == 0) {
1628
+ if (likely(!aclk_incoming_msg_types[i].fnc(&cloud_to_agent))) {
1629
+ // in case of success handler is supposed to clean up after itself
1630
+ // or as in the case of aclk_handle_cloud_request take
1631
+ // ownership of the pointers (done to avoid copying)
1632
+ // see what `aclk_queue_query` parameter `internal` does
1633
+ freez(cloud_to_agent.type_id);
1634
+ return 0;
1635
+ }
1636
+ goto err_cleanup;
1637
+ }
1638
+ }
1639
+
1640
+ errno = 0;
1641
+ error("Unknown message type from Cloud \"%s\"", cloud_to_agent.type_id);
1642
+
1643
+err_cleanup:
1644
+ if (cloud_to_agent.payload)
1645
+ freez(cloud_to_agent.payload);
1646
+ if (cloud_to_agent.type_id)
1647
+ freez(cloud_to_agent.type_id);
1648
+ if (cloud_to_agent.msg_id)
1649
+ freez(cloud_to_agent.msg_id);
1650
+ if (cloud_to_agent.callback_topic)
1651
+ freez(cloud_to_agent.callback_topic);
1652
+
1653
+err_cleanup_nojson:
1654
+ if (aclk_stats_enabled) {
1655
+ ACLK_STATS_LOCK;
1656
+ aclk_metrics_per_sample.cloud_req_err++;
1657
+ ACLK_STATS_UNLOCK;
1658
+ }
1659
+
1660
+ return 1;
1661
}
aclk/agent_cloud_link.h
+4
-3
@@ -7,7 +7,6 @@
7
#include "mqtt.h"
8
#include "aclk_common.h"
9
10
-#define ACLK_VERSION 1
10
#define ACLK_THREAD_NAME "ACLK_Query"
11
#define ACLK_CHART_TOPIC "outbound/meta"
12
#define ACLK_ALARMS_TOPIC "outbound/alarms"
@@ -35,6 +34,8 @@ struct aclk_request {
34
char *callback_topic;
35
char *payload;
36
int version;
37
+ int min_version;
38
+ int max_version;
39
};
40
41
typedef enum aclk_init_action { ACLK_INIT, ACLK_REINIT } ACLK_INIT_ACTION;
@@ -72,8 +73,8 @@ char *create_publish_base_topic();
73
int aclk_send_single_chart(char *host, char *chart);
74
int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd);
75
int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae);
75
-void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us);
76
-int aclk_handle_cloud_request(char *payload);
76
+void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us, int version);
77
+int aclk_handle_cloud_message(char *payload);
78
void aclk_add_collector(const char *hostname, const char *plugin_name, const char *module_name);
79
void aclk_del_collector(const char *hostname, const char *plugin_name, const char *module_name);
80
void aclk_alarm_reload();
aclk/mqtt.c
+2
-2
@@ -26,7 +26,7 @@ void mqtt_message_callback(struct mosquitto *mosq, void *obj, const struct mosqu
26
UNUSED(mosq);
27
UNUSED(obj);
28
29
- aclk_handle_cloud_request(msg->payload);
29
+ aclk_handle_cloud_message(msg->payload);
30
}
31
32
void publish_callback(struct mosquitto *mosq, void *obj, int rc)
@@ -306,7 +306,7 @@ int _link_set_lwt(char *sub_topic, int qos)
306
307
usec_t lwt_time = aclk_session_sec * USEC_PER_SEC + aclk_session_us + 1;
308
BUFFER *b = buffer_create(512);
309
- aclk_create_header(b, "disconnect", NULL, lwt_time / USEC_PER_SEC, lwt_time % USEC_PER_SEC);
309
+ aclk_create_header(b, "disconnect", NULL, lwt_time / USEC_PER_SEC, lwt_time % USEC_PER_SEC, ACLK_VERSION_NEG_VERSION);
310
buffer_strcat(b, ", \"payload\": \"unexpected\" }");
311
rc = mosquitto_will_set(mosq, topic, buffer_strlen(b), buffer_tostring(b), qos, 0);
312
buffer_free(b);
aclk/mqtt.h
+1
-1
@@ -19,7 +19,7 @@ const char *_link_strerror(int rc);
19
int _link_set_lwt(char *topic, int qos);
20
21
22
-int aclk_handle_cloud_request(char *);
22
+int aclk_handle_cloud_message(char *);
23
extern char *get_topic(char *sub_topic, char *final_topic, int max_size);
24
25
#endif //NETDATA_MQTT_H
claim/claim.c
+2
-1
@@ -116,7 +116,7 @@ void claim_agent(char *claiming_arguments)
116
}
117
118
#ifdef ENABLE_ACLK
119
-extern int aclk_connected, aclk_kill_link;
119
+extern int aclk_connected, aclk_kill_link, aclk_disable_runtime;
120
#endif
121
122
/* Change the claimed state of the agent.
@@ -144,6 +144,7 @@ void load_claiming_state(void)
144
info("Agent was already connected to Cloud - forcing reconnection under new credentials");
145
aclk_kill_link = 1;
146
}
147
+ aclk_disable_runtime = 0;
148
149
// Propagate into aclk and registry. Be kind of atomic...
150
appconfig_get(&cloud_config, CONFIG_SECTION_GLOBAL, "cloud base url", DEFAULT_CLOUD_BASE_URL);