Announce proto capability and enable if cloud supports (#11476)
The time has come to push the button.
Timotej S committed
Nov 11, 2021 at 13:58 UTC
e20c0237fdb73483e678eb49d72e3f6adeb108c9
5 files changed
+49
-13
aclk/aclk.c
+34
-10
@@ -30,8 +30,6 @@ int aclk_alert_reloaded = 1; //1 on startup, and again on health_reload
30
31
time_t aclk_block_until = 0;
32
33
-aclk_env_t *aclk_env = NULL;
34
-
33
mqtt_wss_client mqttwss_client;
34
35
netdata_mutex_t aclk_shared_state_mutex = NETDATA_MUTEX_INITIALIZER;
@@ -183,7 +181,7 @@ void aclk_mqtt_wss_log_cb(mqtt_wss_log_type_t log_type, const char* str)
181
182
//TODO prevent big buffer on stack
183
#define RX_MSGLEN_MAX 4096
186
-static void msg_callback(const char *topic, const void *msg, size_t msglen, int qos)
184
+static void msg_callback_old_protocol(const char *topic, const void *msg, size_t msglen, int qos)
185
{
186
char cmsg[RX_MSGLEN_MAX];
187
size_t len = (msglen < RX_MSGLEN_MAX - 1) ? msglen : (RX_MSGLEN_MAX - 1);
@@ -227,7 +225,7 @@ static void msg_callback(const char *topic, const void *msg, size_t msglen, int
225
}
226
227
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
230
-static void msg_callback_new(const char *topic, const void *msg, size_t msglen, int qos)
228
+static void msg_callback_new_protocol(const char *topic, const void *msg, size_t msglen, int qos)
229
{
230
if (msglen > RX_MSGLEN_MAX)
231
error("Incoming ACLK message was bigger than MAX of %d and got truncated.", RX_MSGLEN_MAX);
@@ -264,7 +262,14 @@ static void msg_callback_new(const char *topic, const void *msg, size_t msglen,
262
263
aclk_handle_new_cloud_msg(msgtype, msg, msglen);
264
}
267
-#endif
265
+
266
+static inline void msg_callback(const char *topic, const void *msg, size_t msglen, int qos) {
267
+ if (aclk_use_new_cloud_arch)
268
+ msg_callback_new_protocol(topic, msg, msglen, qos);
269
+ else
270
+ msg_callback_old_protocol(topic, msg, msglen, qos);
271
+}
272
+#endif /* ENABLE_NEW_CLOUD_PROTOCOL */
273
274
static void puback_callback(uint16_t packet_id)
275
{
@@ -600,6 +605,13 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
605
.drop_on_publish_fail = 1
606
};
607
608
+#if defined(ENABLE_NEW_CLOUD_PROTOCOL) && defined(ACLK_NEWARCH_DEVMODE)
609
+ aclk_use_new_cloud_arch = 1;
610
+ info("Switching ACLK to new protobuf protocol. Due to #define ACLK_NEWARCH_DEVMODE.");
611
+#else
612
+ aclk_use_new_cloud_arch = 0;
613
+#endif
614
+
615
#ifndef ACLK_DISABLE_CHALLENGE
616
if (aclk_env) {
617
aclk_env_t_destroy(aclk_env);
@@ -618,6 +630,21 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
630
if (netdata_exit)
631
return 1;
632
633
+#ifndef ACLK_NEWARCH_DEVMODE
634
+ if (aclk_env->encoding == ACLK_ENC_PROTO) {
635
+#ifndef ENABLE_NEW_CLOUD_PROTOCOL
636
+ error("Cloud requested New Cloud Protocol to be used but this agent cannot support it!");
637
+ continue;
638
+#endif
639
+ if (!aclk_env_has_capa("proto")) {
640
+ error ("Can't encoding=proto without at least \"proto\" capability.");
641
+ continue;
642
+ }
643
+ info("Switching ACLK to new protobuf protocol. Due to /env response.");
644
+ aclk_use_new_cloud_arch = 1;
645
+ }
646
+#endif
647
+
648
memset(&auth_url, 0, sizeof(url_t));
649
if (url_parse(aclk_env->auth_endpoint, &auth_url)) {
650
error("Parsing URL returned by env endpoint for authentication failed. \"%s\"", aclk_env->auth_endpoint);
@@ -716,9 +743,6 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
743
*/
744
void *aclk_main(void *ptr)
745
{
719
-#if defined(ENABLE_NEW_CLOUD_PROTOCOL) && defined(ACLK_NEWARCH_DEVMODE)
720
- aclk_use_new_cloud_arch = 1;
721
-#endif
746
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
747
748
struct aclk_stats_thread *stats_thread = NULL;
@@ -753,9 +777,9 @@ void *aclk_main(void *ptr)
777
goto exit;
778
779
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
756
- if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, (aclk_use_new_cloud_arch ? msg_callback_new : msg_callback), puback_callback))) {
757
-#else
780
if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, msg_callback, puback_callback))) {
781
+#else
782
+ if (!(mqttwss_client = mqtt_wss_new("mqtt_wss", aclk_mqtt_wss_log_cb, msg_callback_old_protocol, puback_callback))) {
783
#endif
784
error("Couldn't initialize MQTT_WSS network library");
785
goto exit;
aclk/aclk.h
-2
@@ -14,8 +14,6 @@ extern time_t aclk_block_until;
14
15
extern int disconnect_req;
16
17
-extern aclk_env_t *aclk_env;
18
-
17
void *aclk_main(void *ptr);
18
19
extern netdata_mutex_t aclk_shared_state_mutex;
aclk/aclk_otp.c
+1
-1
@@ -842,7 +842,7 @@ int aclk_get_env(aclk_env_t *env, const char* aclk_hostname, int aclk_port) {
842
return 1;
843
}
844
845
- buffer_sprintf(buf, "/api/v1/env?v=%s&cap=json&claim_id=%s", &(VERSION[1]) /* skip 'v' at beginning */, agent_id);
845
+ buffer_sprintf(buf, "/api/v1/env?v=%s&cap=json,proto&claim_id=%s", &(VERSION[1]) /* skip 'v' at beginning */, agent_id);
846
freez(agent_id);
847
848
req.host = (char*)aclk_hostname;
aclk/aclk_util.c
+11
@@ -7,6 +7,8 @@
7
int aclk_use_new_cloud_arch = 0;
8
usec_t aclk_session_newarch = 0;
9
10
+aclk_env_t *aclk_env = NULL;
11
+
12
int chart_batch_id;
13
14
aclk_encoding_type_t aclk_encoding_type_t_from_str(const char *str) {
@@ -51,6 +53,15 @@ void aclk_env_t_destroy(aclk_env_t *env) {
53
}
54
}
55
56
+int aclk_env_has_capa(const char *capa)
57
+{
58
+ for (int i = 0; i < aclk_env->capability_count; i++) {
59
+ if (!strcasecmp(capa, aclk_env->capabilities[i]))
60
+ return 1;
61
+ }
62
+ return 0;
63
+}
64
+
65
#ifdef ACLK_LOG_CONVERSATION_DIR
66
volatile int aclk_conversation_log_counter = 0;
67
#if !defined(HAVE_C___ATOMIC) || defined(NETDATA_NO_ATOMIC_INSTRUCTIONS)
aclk/aclk_util.h
+3
@@ -49,11 +49,14 @@ typedef struct {
49
aclk_backoff_t backoff;
50
} aclk_env_t;
51
52
+extern aclk_env_t *aclk_env;
53
+
54
aclk_encoding_type_t aclk_encoding_type_t_from_str(const char *str);
55
aclk_transport_type_t aclk_transport_type_t_from_str(const char *str);
56
57
void aclk_transport_desc_t_destroy(aclk_transport_desc_t *trp_desc);
58
void aclk_env_t_destroy(aclk_env_t *env);
59
+int aclk_env_has_capa(const char *capa);
60
61
enum aclk_topics {
62
ACLK_TOPICID_UNKNOWN = 0,