Improve ACLK according to results of the smoke-test. (#8358)
* Cleaning up the ACLK part 2 (#8187) * Initial proxy support implementation (#8146) * Implement the ACLK Challenge-Response Authentication (#8317) Co-authored-by: Timo <6674623+underhood@users.noreply.github.com>
Andrew Moss committed
Mar 10, 2020 at 12:21 UTC
4acc880bab388c6ea61a1016dcbc44e387dd932c
18 files changed
+1263
-788
Makefile.am
+2
@@ -472,6 +472,8 @@ CLAIM_PLUGIN_FILES = \
472
$(NULL)
473
474
ACLK_PLUGIN_FILES = \
475
+ aclk/aclk_common.c \
476
+ aclk/aclk_common.h \
477
aclk/agent_cloud_link.c \
478
aclk/agent_cloud_link.h \
479
aclk/mqtt.c \
aclk/aclk_common.c
new
+35
@@ -0,0 +1,35 @@
1
+#include "aclk_common.h"
2
+
3
+struct {
4
+ ACLK_PROXY_TYPE type;
5
+ const char *url_str;
6
+} supported_proxy_types[] = {
7
+ { .type = PROXY_TYPE_SOCKS5, .url_str = "socks5" ACLK_PROXY_PROTO_ADDR_SEPARATOR },
8
+ { .type = PROXY_TYPE_SOCKS5, .url_str = "socks5h" ACLK_PROXY_PROTO_ADDR_SEPARATOR },
9
+ { .type = PROXY_TYPE_UNKNOWN, .url_str = NULL },
10
+};
11
+
12
+static inline ACLK_PROXY_TYPE aclk_find_proxy(const char *string)
13
+{
14
+ int i = 0;
15
+ while( supported_proxy_types[i].url_str ) {
16
+ if(!strncmp(supported_proxy_types[i].url_str, string, strlen(supported_proxy_types[i].url_str)))
17
+ return supported_proxy_types[i].type;
18
+ i++;
19
+ }
20
+ return PROXY_TYPE_UNKNOWN;
21
+}
22
+
23
+ACLK_PROXY_TYPE aclk_verify_proxy(const char *string)
24
+{
25
+ if(!string)
26
+ return PROXY_TYPE_UNKNOWN;
27
+
28
+ while(*string == 0x20)
29
+ string++;
30
+
31
+ if(!*string)
32
+ return PROXY_TYPE_UNKNOWN;
33
+
34
+ return aclk_find_proxy(string);
35
+}
aclk/aclk_common.h
new
+20
@@ -0,0 +1,20 @@
1
+#ifndef ACLK_COMMON_H
2
+#define ACLK_COMMON_H
3
+
4
+#include "libnetdata/libnetdata.h"
5
+
6
+typedef enum aclk_proxy_type {
7
+ PROXY_TYPE_UNKNOWN = 0,
8
+ PROXY_TYPE_SOCKS5,
9
+ PROXY_TYPE_HTTP,
10
+ PROXY_DISABLED,
11
+ PROXY_NOT_SET,
12
+} ACLK_PROXY_TYPE;
13
+
14
+#define ACLK_PROXY_PROTO_ADDR_SEPARATOR "://"
15
+#define ACLK_PROXY_ENV "env"
16
+#define ACLK_PROXY_CONFIG_VAR "proxy"
17
+
18
+ACLK_PROXY_TYPE aclk_verify_proxy(const char *string);
19
+
20
+#endif //ACLK_COMMON_H
aclk/aclk_lws_wss_client.c
+408
-285
@@ -1,382 +1,505 @@
1
#include "aclk_lws_wss_client.h"
2
3
#include "libnetdata/libnetdata.h"
4
+#include "../daemon/common.h"
5
+#include "aclk_common.h"
6
7
static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len);
8
9
struct aclk_lws_wss_perconnect_data {
8
- int todo;
10
+ int todo;
11
};
12
13
struct lws_wss_packet_buffer {
12
- unsigned char* data;
13
- size_t data_size;
14
- struct lws_wss_packet_buffer *next;
14
+ unsigned char *data;
15
+ size_t data_size;
16
+ struct lws_wss_packet_buffer *next;
17
};
18
17
-static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_new(void* data, size_t size)
19
+static struct aclk_lws_wss_engine_instance *engine_instance = NULL;
20
+
21
+static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_new(void *data, size_t size)
22
{
19
- struct lws_wss_packet_buffer *new = callocz(1, sizeof(struct lws_wss_packet_buffer));
20
- if(data) {
21
- new->data = mallocz(LWS_PRE+size);
22
- memcpy(new->data+LWS_PRE, data, size);
23
- new->data_size = size;
24
- }
25
- return new;
23
+ struct lws_wss_packet_buffer *new = callocz(1, sizeof(struct lws_wss_packet_buffer));
24
+ if (data) {
25
+ new->data = mallocz(LWS_PRE + size);
26
+ memcpy(new->data + LWS_PRE, data, size);
27
+ new->data_size = size;
28
+ }
29
+ return new;
30
}
31
32
static inline void lws_wss_packet_buffer_append(struct lws_wss_packet_buffer **list, struct lws_wss_packet_buffer *item)
33
{
30
- struct lws_wss_packet_buffer *tail = *list;
31
- if(!*list) {
32
- *list = item;
33
- return;
34
- }
35
- while(tail->next) {
36
- tail = tail->next;
37
- }
38
- tail->next = item;
34
+ struct lws_wss_packet_buffer *tail = *list;
35
+ if (!*list) {
36
+ *list = item;
37
+ return;
38
+ }
39
+ while (tail->next) {
40
+ tail = tail->next;
41
+ }
42
+ tail->next = item;
43
}
44
45
static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_pop(struct lws_wss_packet_buffer **list)
46
{
43
- struct lws_wss_packet_buffer *ret = *list;
44
- if(ret != NULL)
45
- *list = ret->next;
47
+ struct lws_wss_packet_buffer *ret = *list;
48
+ if (ret != NULL)
49
+ *list = ret->next;
50
47
- return ret;
51
+ return ret;
52
}
53
54
static inline void lws_wss_packet_buffer_free(struct lws_wss_packet_buffer *item)
55
{
52
- freez(item->data);
53
- freez(item);
56
+ freez(item->data);
57
+ freez(item);
58
}
59
60
static inline void _aclk_lws_wss_read_buffer_clear(struct lws_ring *ringbuffer)
61
{
58
- size_t elems = lws_ring_get_count_waiting_elements(ringbuffer, NULL);
59
- if(elems > 0)
60
- lws_ring_consume(ringbuffer, NULL, NULL, elems);
62
+ size_t elems = lws_ring_get_count_waiting_elements(ringbuffer, NULL);
63
+ if (elems > 0)
64
+ lws_ring_consume(ringbuffer, NULL, NULL, elems);
65
}
66
67
static inline void _aclk_lws_wss_write_buffer_clear(struct lws_wss_packet_buffer **list)
68
{
65
- struct lws_wss_packet_buffer *i;
66
- while((i = lws_wss_packet_buffer_pop(list)) != NULL) {
67
- lws_wss_packet_buffer_free(i);
68
- }
69
- *list = NULL;
69
+ struct lws_wss_packet_buffer *i;
70
+ while ((i = lws_wss_packet_buffer_pop(list)) != NULL) {
71
+ lws_wss_packet_buffer_free(i);
72
+ }
73
+ *list = NULL;
74
}
75
72
-static inline void aclk_lws_wss_clear_io_buffers(struct aclk_lws_wss_engine_instance *inst)
76
+static inline void aclk_lws_wss_clear_io_buffers()
77
{
74
- aclk_lws_mutex_lock(&inst->read_buf_mutex);
75
- _aclk_lws_wss_read_buffer_clear(inst->read_ringbuffer);
76
- aclk_lws_mutex_unlock(&inst->read_buf_mutex);
77
- aclk_lws_mutex_lock(&inst->write_buf_mutex);
78
- _aclk_lws_wss_write_buffer_clear(&inst->write_buffer_head);
79
- aclk_lws_mutex_unlock(&inst->write_buf_mutex);
78
+ aclk_lws_mutex_lock(&engine_instance->read_buf_mutex);
79
+ _aclk_lws_wss_read_buffer_clear(engine_instance->read_ringbuffer);
80
+ aclk_lws_mutex_unlock(&engine_instance->read_buf_mutex);
81
+ aclk_lws_mutex_lock(&engine_instance->write_buf_mutex);
82
+ _aclk_lws_wss_write_buffer_clear(&engine_instance->write_buffer_head);
83
+ aclk_lws_mutex_unlock(&engine_instance->write_buf_mutex);
84
}
85
82
-static const struct lws_protocols protocols[] = {
83
- {
84
- "aclk-wss",
85
- aclk_lws_wss_callback,
86
- sizeof(struct aclk_lws_wss_perconnect_data),
87
- 0, 0, 0, 0
88
- },
89
- { NULL, NULL, 0, 0, 0, 0, 0 }
90
-};
86
+static const struct lws_protocols protocols[] = { { "aclk-wss", aclk_lws_wss_callback,
87
+ sizeof(struct aclk_lws_wss_perconnect_data), 0, 0, 0, 0 },
88
+ { NULL, NULL, 0, 0, 0, 0, 0 } };
89
92
-static void aclk_lws_wss_log_divert(int level, const char *line) {
93
- switch(level){
94
- case LLL_ERR:
95
- error("Libwebsockets Error: %s", line);
96
- break;
97
- case LLL_WARN:
98
- debug(D_ACLK, "Libwebsockets Warn: %s", line);
99
- break;
100
- default:
101
- error("Libwebsockets try to log with unknown log level (%d), msg: %s", level, line);
102
- }
90
+static void aclk_lws_wss_log_divert(int level, const char *line)
91
+{
92
+ switch (level) {
93
+ case LLL_ERR:
94
+ error("Libwebsockets Error: %s", line);
95
+ break;
96
+ case LLL_WARN:
97
+ debug(D_ACLK, "Libwebsockets Warn: %s", line);
98
+ break;
99
+ default:
100
+ error("Libwebsockets try to log with unknown log level (%d), msg: %s", level, line);
101
+ }
102
}
103
105
-struct aclk_lws_wss_engine_instance* aclk_lws_wss_client_init (const struct aclk_lws_wss_engine_callbacks *callbacks, const char *target_hostname, int target_port) {
106
- static int lws_logging_initialized = 0;
107
- struct lws_context_creation_info info;
108
- struct aclk_lws_wss_engine_instance *inst;
104
+static int aclk_lws_wss_client_init( char *target_hostname, int target_port)
105
+{
106
+ static int lws_logging_initialized = 0;
107
+ struct lws_context_creation_info info;
108
110
- if(unlikely(!lws_logging_initialized)) {
111
- lws_set_log_level(LLL_ERR | LLL_WARN, aclk_lws_wss_log_divert);
112
- lws_logging_initialized = 1;
113
- }
109
+ if (unlikely(!lws_logging_initialized)) {
110
+ lws_set_log_level(LLL_ERR | LLL_WARN, aclk_lws_wss_log_divert);
111
+ lws_logging_initialized = 1;
112
+ }
113
115
- if(!callbacks || !target_hostname)
116
- return NULL;
114
+ if (!target_hostname)
115
+ return 1;
116
118
- inst = callocz(1, sizeof(struct aclk_lws_wss_engine_instance));
117
+ engine_instance = callocz(1, sizeof(struct aclk_lws_wss_engine_instance));
118
120
- inst->host = target_hostname;
121
- inst->port = target_port;
119
+ engine_instance->host = target_hostname;
120
+ engine_instance->port = target_port;
121
123
- memset(&info, 0, sizeof(struct lws_context_creation_info));
124
- info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
125
- info.port = CONTEXT_PORT_NO_LISTEN;
126
- info.protocols = protocols;
127
- info.user = inst;
128
-
129
- inst->lws_context = lws_create_context(&info);
130
- if(!inst->lws_context)
131
- goto failure_cleanup_2;
122
+ memset(&info, 0, sizeof(struct lws_context_creation_info));
123
+ info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
124
+ info.port = CONTEXT_PORT_NO_LISTEN;
125
+ info.protocols = protocols;
126
133
- inst->callbacks = *callbacks;
127
+ engine_instance->lws_context = lws_create_context(&info);
128
+ if (!engine_instance->lws_context)
129
+ goto failure_cleanup_2;
130
135
- aclk_lws_mutex_init(&inst->write_buf_mutex);
136
- aclk_lws_mutex_init(&inst->read_buf_mutex);
131
+ aclk_lws_mutex_init(&engine_instance->write_buf_mutex);
132
+ aclk_lws_mutex_init(&engine_instance->read_buf_mutex);
133
138
- inst->read_ringbuffer = lws_ring_create(1, ACLK_LWS_WSS_RECV_BUFF_SIZE_BYTES, NULL);
139
- if(!inst->read_ringbuffer)
140
- goto failure_cleanup;
134
+ engine_instance->read_ringbuffer = lws_ring_create(1, ACLK_LWS_WSS_RECV_BUFF_SIZE_BYTES, NULL);
135
+ if (!engine_instance->read_ringbuffer)
136
+ goto failure_cleanup;
137
142
- return inst;
138
+ return 0;
139
140
failure_cleanup:
145
- lws_context_destroy(inst->lws_context);
141
+ lws_context_destroy(engine_instance->lws_context);
142
failure_cleanup_2:
147
- freez(inst);
148
- return NULL;
143
+ freez(engine_instance);
144
+ return 1;
145
}
146
151
-void aclk_lws_wss_client_destroy(struct aclk_lws_wss_engine_instance* inst) {
152
- lws_context_destroy(inst->lws_context);
153
- inst->lws_context = NULL;
154
- inst->lws_wsi = NULL;
147
+void aclk_lws_wss_client_destroy(struct aclk_lws_wss_engine_instance *engine_instance)
148
+{
149
+ if (engine_instance == NULL)
150
+ return;
151
+ lws_context_destroy(engine_instance->lws_context);
152
+ engine_instance->lws_context = NULL;
153
+ engine_instance->lws_wsi = NULL;
154
156
- aclk_lws_wss_clear_io_buffers(inst);
155
+ aclk_lws_wss_clear_io_buffers(engine_instance);
156
157
#ifdef ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED
159
- pthread_mutex_destroy(&inst->write_buf_mutex);
160
- pthread_mutex_destroy(&inst->read_buf_mutex);
158
+ pthread_mutex_destroy(&engine_instance->write_buf_mutex);
159
+ pthread_mutex_destroy(&engine_instance->read_buf_mutex);
160
#endif
161
}
162
164
-void aclk_lws_wss_connect(struct aclk_lws_wss_engine_instance *inst){
163
+static int _aclk_wss_set_socks(struct lws_vhost *vhost, const char *socks)
164
+{
165
+ char *proxy = strstr(socks, ACLK_PROXY_PROTO_ADDR_SEPARATOR);
166
+
167
+ if(!proxy)
168
+ return -1;
169
+
170
+ proxy += strlen(ACLK_PROXY_PROTO_ADDR_SEPARATOR);
171
+
172
+ if(!*proxy)
173
+ return -1;
174
+
175
+ return lws_set_socks(vhost, proxy);
176
+}
177
+
178
+// helper function to censor user&password
179
+// for logging purposes
180
+static void safe_log_proxy_censor(char *proxy) {
181
+ size_t length = strlen(proxy);
182
+ char *auth = proxy+length-1;
183
+ char *cur;
184
+
185
+ while( (auth >= proxy) && (*auth != '@') )
186
+ auth--;
187
+
188
+ //if not found or @ is first char do nothing
189
+ if(auth<=proxy)
190
+ return;
191
+
192
+ cur = strstr(proxy, ACLK_PROXY_PROTO_ADDR_SEPARATOR);
193
+ if(!cur)
194
+ cur = proxy;
195
+ else
196
+ cur += strlen(ACLK_PROXY_PROTO_ADDR_SEPARATOR);
197
+
198
+ while(cur < auth) {
199
+ *cur='X';
200
+ cur++;
201
+ }
202
+}
203
+
204
+static inline void safe_log_proxy_error(char *str, const char *proxy) {
205
+ char *log = strdupz(proxy);
206
+ safe_log_proxy_censor(log);
207
+ error("%s Provided Value:\"%s\"", str, log);
208
+ freez(log);
209
+}
210
+
211
+static inline int check_socks_enviroment(const char **proxy) {
212
+ char *tmp = getenv("socks_proxy");
213
+
214
+ if(!tmp)
215
+ return 1;
216
+
217
+ if(aclk_verify_proxy(tmp) == PROXY_TYPE_SOCKS5) {
218
+ *proxy = tmp;
219
+ return 0;
220
+ }
221
+
222
+ safe_log_proxy_error("Environment var \"socks_proxy\" defined but of unknown format. Supported syntax: \"socks5[h]://[user:pass@]host:ip\".", tmp);
223
+ return 1;
224
+}
225
+
226
+static const char *aclk_lws_wss_get_proxy_setting(ACLK_PROXY_TYPE *type) {
227
+ const char *proxy = config_get(CONFIG_SECTION_ACLK, ACLK_PROXY_CONFIG_VAR, ACLK_PROXY_ENV);
228
+ *type = PROXY_DISABLED;
229
+
230
+ if(strcmp(proxy, "none") == 0)
231
+ return proxy;
232
+
233
+ if(strcmp(proxy, ACLK_PROXY_ENV) == 0) {
234
+ if(check_socks_enviroment(&proxy) == 0)
235
+ *type = PROXY_TYPE_SOCKS5;
236
+ return proxy;
237
+ }
238
+
239
+ *type = aclk_verify_proxy(proxy);
240
+ if(*type == PROXY_TYPE_UNKNOWN) {
241
+ *type = PROXY_DISABLED;
242
+ safe_log_proxy_error("Config var \"" ACLK_PROXY_CONFIG_VAR "\" defined but of unknown format. Supported syntax: \"socks5[h]://[user:pass@]host:ip\".", proxy);
243
+ }
244
+
245
+ return proxy;
246
+}
247
+
248
+// Return code indicates if connection attempt has started async.
249
+int aclk_lws_wss_connect(char *host, int port)
250
+{
251
struct lws_client_connect_info i;
252
+ struct lws_vhost *vhost;
253
+ static const char *proxy = NULL;
254
+ static ACLK_PROXY_TYPE proxy_type = PROXY_NOT_SET;
255
+ char *log;
256
+
257
+ if (!engine_instance) {
258
+ return aclk_lws_wss_client_init(host, port);
259
+ // PROTOCOL_INIT callback will call again.
260
+ }
261
167
- if(inst->lws_wsi) {
262
+ if(proxy_type == PROXY_NOT_SET)
263
+ proxy = aclk_lws_wss_get_proxy_setting(&proxy_type);
264
+
265
+ if (engine_instance->lws_wsi) {
266
error("Already Connected. Only one connection supported at a time.");
169
- return;
267
+ return 0;
268
+ }
269
+
270
+ // from LWS docu:
271
+ // If option LWS_SERVER_OPTION_EXPLICIT_VHOSTS is given, no vhost is
272
+ // created; you're expected to create your own vhosts afterwards using
273
+ // lws_create_vhost(). Otherwise a vhost named "default" is also created
274
+ // using the information in the vhost-related members, for compatibility.
275
+ vhost = lws_get_vhost_by_name(engine_instance->lws_context, "default");
276
+ if(!vhost)
277
+ fatal("Could not find the default LWS vhost.");
278
+
279
+ memset(&i, 0, sizeof(i));
280
+ i.context = engine_instance->lws_context;
281
+ i.port = engine_instance->port;
282
+ i.address = engine_instance->host;
283
+ i.path = "/mqtt";
284
+ i.host = engine_instance->host;
285
+ i.protocol = "mqtt";
286
+
287
+ switch (proxy_type) {
288
+ case PROXY_DISABLED:
289
+ lws_set_socks(vhost, ":");
290
+ lws_set_proxy(vhost, ":");
291
+ break;
292
+ case PROXY_TYPE_SOCKS5:
293
+ log = strdupz(proxy);
294
+ safe_log_proxy_censor(log);
295
+ info("Connecting using SOCKS5 proxy:\"%s\"", log);
296
+ freez(log);
297
+ if(_aclk_wss_set_socks(vhost, proxy))
298
+ error("LWS failed to accept socks proxy.");
299
+ break;
300
+ default:
301
+ error("The proxy could not be set. Unknown proxy type.");
302
}
303
172
- memset(&i, 0, sizeof(i));
173
- i.context = inst->lws_context;
174
- i.port = inst->port;
175
- i.address = inst->host;
176
- i.path = "/mqtt";
177
- i.host = inst->host;
178
- i.protocol = "mqtt";
304
#ifdef ACLK_SSL_ALLOW_SELF_SIGNED
180
- i.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK;
305
+ i.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK;
306
+ info("Disabling SSL certificate checks");
307
#else
182
- i.ssl_connection = LCCSCF_USE_SSL;
308
+ i.ssl_connection = LCCSCF_USE_SSL;
309
#endif
184
- lws_client_connect_via_info(&i);
310
+ lws_client_connect_via_info(&i);
311
+ return 0;
312
}
313
187
-static inline int received_data_to_ringbuff(struct lws_ring *buffer, void* data, size_t len) {
188
- if( lws_ring_insert(buffer, data, len) != len ) {
189
- error("ACLK_LWS_WSS_CLIENT: receive buffer full. Closing connection to prevent flooding.");
190
- return 0;
191
- }
192
- return 1;
314
+static inline int received_data_to_ringbuff(struct lws_ring *buffer, void *data, size_t len)
315
+{
316
+ if (lws_ring_insert(buffer, data, len) != len) {
317
+ error("ACLK_LWS_WSS_CLIENT: receive buffer full. Closing connection to prevent flooding.");
318
+ return 0;
319
+ }
320
+ return 1;
321
}
194
-
322
+
323
static const char *aclk_lws_callback_name(enum lws_callback_reasons reason)
324
{
197
- switch(reason)
198
- {
199
- case LWS_CALLBACK_CLIENT_WRITEABLE:
200
- return "LWS_CALLBACK_CLIENT_WRITEABLE";
201
- case LWS_CALLBACK_CLIENT_RECEIVE:
202
- return "LWS_CALLBACK_CLIENT_RECEIVE";
203
- case LWS_CALLBACK_PROTOCOL_INIT:
204
- return "LWS_CALLBACK_PROTOCOL_INIT";
205
- case LWS_CALLBACK_SERVER_NEW_CLIENT_INSTANTIATED:
206
- return "LWS_CALLBACK_SERVER_NEW_CLIENT_INSTANTIATED";
207
- case LWS_CALLBACK_USER:
208
- return "LWS_CALLBACK_USER";
209
- case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
210
- return "LWS_CALLBACK_CLIENT_CONNECTION_ERROR";
211
- case LWS_CALLBACK_CLIENT_CLOSED:
212
- return "LWS_CALLBACK_CLIENT_CLOSED";
213
- case LWS_CALLBACK_WS_PEER_INITIATED_CLOSE:
214
- return "LWS_CALLBACK_WS_PEER_INITIATED_CLOSE";
215
- case LWS_CALLBACK_WSI_DESTROY:
216
- return "LWS_CALLBACK_WSI_DESTROY";
217
- case LWS_CALLBACK_CLIENT_ESTABLISHED:
218
- return "LWS_CALLBACK_CLIENT_ESTABLISHED";
219
- default:
220
- // Not using an internal buffer here for thread-safety with unknown calling context.
221
- error("Unknown LWS callback %u", reason);
222
- return "unknown";
223
- }
325
+ switch (reason) {
326
+ case LWS_CALLBACK_CLIENT_WRITEABLE:
327
+ return "LWS_CALLBACK_CLIENT_WRITEABLE";
328
+ case LWS_CALLBACK_CLIENT_RECEIVE:
329
+ return "LWS_CALLBACK_CLIENT_RECEIVE";
330
+ case LWS_CALLBACK_PROTOCOL_INIT:
331
+ return "LWS_CALLBACK_PROTOCOL_INIT";
332
+ case LWS_CALLBACK_SERVER_NEW_CLIENT_INSTANTIATED:
333
+ return "LWS_CALLBACK_SERVER_NEW_CLIENT_INSTANTIATED";
334
+ case LWS_CALLBACK_USER:
335
+ return "LWS_CALLBACK_USER";
336
+ case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
337
+ return "LWS_CALLBACK_CLIENT_CONNECTION_ERROR";
338
+ case LWS_CALLBACK_CLIENT_CLOSED:
339
+ return "LWS_CALLBACK_CLIENT_CLOSED";
340
+ case LWS_CALLBACK_WS_PEER_INITIATED_CLOSE:
341
+ return "LWS_CALLBACK_WS_PEER_INITIATED_CLOSE";
342
+ case LWS_CALLBACK_WSI_DESTROY:
343
+ return "LWS_CALLBACK_WSI_DESTROY";
344
+ case LWS_CALLBACK_CLIENT_ESTABLISHED:
345
+ return "LWS_CALLBACK_CLIENT_ESTABLISHED";
346
+ case LWS_CALLBACK_OPENSSL_PERFORM_SERVER_CERT_VERIFICATION:
347
+ return "LWS_CALLBACK_OPENSSL_PERFORM_SERVER_CERT_VERIFICATION";
348
+ default:
349
+ // Not using an internal buffer here for thread-safety with unknown calling context.
350
+ error("Unknown LWS callback %u", reason);
351
+ return "unknown";
352
+ }
353
}
225
-static int
226
-aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason,
227
- void *user, void *in, size_t len)
354
+static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len)
355
{
229
- UNUSED(user);
230
- struct aclk_lws_wss_engine_instance *inst = lws_context_user(lws_get_context(wsi));
231
- struct lws_wss_packet_buffer *data;
232
- int retval = 0;
356
+ UNUSED(user);
357
+ struct lws_wss_packet_buffer *data;
358
+ int retval = 0;
359
+
360
+ // Callback servicing is forced when we are closed from above.
361
+ if (engine_instance->upstream_reconnect_request) {
362
+ error("Closing lws connectino due to libmosquitto error.");
363
+ char *upstream_connection_error = "MQTT protocol error. Closing underlying wss connection.";
364
+ lws_close_reason(
365
+ wsi, LWS_CLOSE_STATUS_PROTOCOL_ERR, (unsigned char *)upstream_connection_error,
366
+ strlen(upstream_connection_error));
367
+ retval = -1;
368
+ engine_instance->upstream_reconnect_request = 0;
369
+ }
370
234
- if( !inst ) {
235
- error("Callback received without any aclk_lws_wss_engine_instance!");
236
- return -1;
237
- }
238
-
239
- // Callback servicing is forced when we are closed from above.
240
- if( inst->upstream_reconnect_request ) {
241
- error("Closing lws connectino due to libmosquitto error.");
242
- char *upstream_connection_error = "MQTT protocol error. Closing underlying wss connection.";
243
- lws_close_reason(wsi, LWS_CLOSE_STATUS_PROTOCOL_ERR, (unsigned char*)upstream_connection_error, strlen(upstream_connection_error));
244
- retval = -1;
245
- inst->upstream_reconnect_request = 0;
246
- }
247
-
248
- // Don't log to info - volume is proportional to message flow on ACLK.
249
- switch (reason) {
250
- case LWS_CALLBACK_CLIENT_WRITEABLE:
251
- aclk_lws_mutex_lock(&inst->write_buf_mutex);
252
- data = lws_wss_packet_buffer_pop(&inst->write_buffer_head);
253
- if(likely(data)) {
254
- lws_write(wsi, data->data + LWS_PRE, data->data_size, LWS_WRITE_BINARY);
255
- lws_wss_packet_buffer_free(data);
256
- if(inst->write_buffer_head)
257
- lws_callback_on_writable(inst->lws_wsi);
258
- }
259
- aclk_lws_mutex_unlock(&inst->write_buf_mutex);
260
- return retval;
261
-
262
- case LWS_CALLBACK_CLIENT_RECEIVE:
263
- aclk_lws_mutex_lock(&inst->read_buf_mutex);
264
- if(!received_data_to_ringbuff(inst->read_ringbuffer, in, len))
265
- retval = 1;
266
- aclk_lws_mutex_unlock(&inst->read_buf_mutex);
267
-
268
- if(likely(inst->callbacks.data_rcvd_callback))
269
- // to future myself -> do not call this while read lock is active as it will eventually
270
- // want to acquire same lock later in aclk_lws_wss_client_read() function
271
- inst->callbacks.data_rcvd_callback();
272
- else
273
- inst->data_to_read = 1; //to inform logic above there is reason to call mosquitto_loop_read
274
- return retval;
275
-
276
- case LWS_CALLBACK_WSI_CREATE:
277
- case LWS_CALLBACK_CLIENT_FILTER_PRE_ESTABLISH:
278
- case LWS_CALLBACK_CLIENT_APPEND_HANDSHAKE_HEADER:
279
- case LWS_CALLBACK_OPENSSL_LOAD_EXTRA_CLIENT_VERIFY_CERTS:
280
- case LWS_CALLBACK_GET_THREAD_ID: // ?
281
- case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
282
- // Expected and safe to ignore.
283
- return retval;
284
-
285
- default:
286
- // Pass to next switch, this case removes compiler warnings.
287
- break;
288
-
289
- }
290
- // Log to info - volume is proportional to connection attempts.
291
- info("Processing callback %s", aclk_lws_callback_name(reason));
292
- switch (reason) {
293
- case LWS_CALLBACK_PROTOCOL_INIT:
294
- aclk_lws_wss_connect(inst); // Makes the outgoing connection
295
- break;
296
- case LWS_CALLBACK_SERVER_NEW_CLIENT_INSTANTIATED:
297
- if (inst->lws_wsi != NULL && inst->lws_wsi != wsi)
298
- error("Multiple connections on same WSI? %p vs %p", inst->lws_wsi, wsi);
299
- inst->lws_wsi = wsi;
300
- break;
301
- case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
302
- error("Could not connect MQTT over WSS server \"%s:%d\". LwsReason:\"%s\"", inst->host, inst->port, (in ? (char*)in : "not given"));
303
- // Fall-through
304
- case LWS_CALLBACK_CLIENT_CLOSED:
305
- case LWS_CALLBACK_WS_PEER_INITIATED_CLOSE:
306
- inst->lws_wsi = NULL; // inside libwebsockets lws_close_free_wsi is called after callback
307
- if (inst->callbacks.connection_closed)
308
- inst->callbacks.connection_closed();
309
- return -1; // the callback response is ignored, hope the above remains true
310
- case LWS_CALLBACK_WSI_DESTROY:
311
- aclk_lws_wss_clear_io_buffers(inst);
312
- inst->lws_wsi = NULL;
313
- inst->websocket_connection_up = 0;
314
- if (inst->callbacks.connection_closed)
315
- inst->callbacks.connection_closed();
316
- break;
317
- case LWS_CALLBACK_CLIENT_ESTABLISHED:
318
- inst->websocket_connection_up = 1;
319
- if(inst->callbacks.connection_established_callback)
320
- inst->callbacks.connection_established_callback();
321
- break;
322
-
323
- default:
324
- error("Unexecpted callback from libwebsockets %s",aclk_lws_callback_name(reason));
325
- break;
326
- }
327
- return retval; //0-OK, other connection should be closed!
371
+ // Don't log to info - volume is proportional to message flow on ACLK.
372
+ switch (reason) {
373
+ case LWS_CALLBACK_CLIENT_WRITEABLE:
374
+ aclk_lws_mutex_lock(&engine_instance->write_buf_mutex);
375
+ data = lws_wss_packet_buffer_pop(&engine_instance->write_buffer_head);
376
+ if (likely(data)) {
377
+ lws_write(wsi, data->data + LWS_PRE, data->data_size, LWS_WRITE_BINARY);
378
+ lws_wss_packet_buffer_free(data);
379
+ if (engine_instance->write_buffer_head)
380
+ lws_callback_on_writable(engine_instance->lws_wsi);
381
+ }
382
+ aclk_lws_mutex_unlock(&engine_instance->write_buf_mutex);
383
+ return retval;
384
+
385
+ case LWS_CALLBACK_CLIENT_RECEIVE:
386
+ aclk_lws_mutex_lock(&engine_instance->read_buf_mutex);
387
+ if (!received_data_to_ringbuff(engine_instance->read_ringbuffer, in, len))
388
+ retval = 1;
389
+ aclk_lws_mutex_unlock(&engine_instance->read_buf_mutex);
390
+
391
+ // to future myself -> do not call this while read lock is active as it will eventually
392
+ // want to acquire same lock later in aclk_lws_wss_client_read() function
393
+ aclk_lws_connection_data_received();
394
+ return retval;
395
+
396
+ case LWS_CALLBACK_WSI_CREATE:
397
+ case LWS_CALLBACK_CLIENT_FILTER_PRE_ESTABLISH:
398
+ case LWS_CALLBACK_CLIENT_APPEND_HANDSHAKE_HEADER:
399
+ case LWS_CALLBACK_OPENSSL_LOAD_EXTRA_CLIENT_VERIFY_CERTS:
400
+ case LWS_CALLBACK_GET_THREAD_ID: // ?
401
+ case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
402
+ case LWS_CALLBACK_OPENSSL_PERFORM_SERVER_CERT_VERIFICATION:
403
+ // Expected and safe to ignore.
404
+ debug(D_ACLK, "Ignoring expected callback from LWS: %s", aclk_lws_callback_name(reason));
405
+ return retval;
406
+
407
+ default:
408
+ // Pass to next switch, this case removes compiler warnings.
409
+ break;
410
+ }
411
+ // Log to info - volume is proportional to connection attempts.
412
+ info("Processing callback %s", aclk_lws_callback_name(reason));
413
+ switch (reason) {
414
+ case LWS_CALLBACK_PROTOCOL_INIT:
415
+ aclk_lws_wss_connect(engine_instance->host, engine_instance->port); // Makes the outgoing connection
416
+ break;
417
+ case LWS_CALLBACK_SERVER_NEW_CLIENT_INSTANTIATED:
418
+ if (engine_instance->lws_wsi != NULL && engine_instance->lws_wsi != wsi)
419
+ error("Multiple connections on same WSI? %p vs %p", engine_instance->lws_wsi, wsi);
420
+ engine_instance->lws_wsi = wsi;
421
+ break;
422
+ case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
423
+ error(
424
+ "Could not connect MQTT over WSS server \"%s:%d\". LwsReason:\"%s\"", engine_instance->host,
425
+ engine_instance->port, (in ? (char *)in : "not given"));
426
+ // Fall-through
427
+ case LWS_CALLBACK_CLIENT_CLOSED:
428
+ case LWS_CALLBACK_WS_PEER_INITIATED_CLOSE:
429
+ engine_instance->lws_wsi = NULL; // inside libwebsockets lws_close_free_wsi is called after callback
430
+ aclk_lws_connection_closed();
431
+ return -1; // the callback response is ignored, hope the above remains true
432
+ case LWS_CALLBACK_WSI_DESTROY:
433
+ aclk_lws_wss_clear_io_buffers(engine_instance);
434
+ engine_instance->lws_wsi = NULL;
435
+ engine_instance->websocket_connection_up = 0;
436
+ aclk_lws_connection_closed();
437
+ break;
438
+ case LWS_CALLBACK_CLIENT_ESTABLISHED:
439
+ engine_instance->websocket_connection_up = 1;
440
+ aclk_lws_connection_established(engine_instance->host, engine_instance->port);
441
+ break;
442
+
443
+ default:
444
+ error("Unexpected callback from libwebsockets %s", aclk_lws_callback_name(reason));
445
+ break;
446
+ }
447
+ return retval; //0-OK, other connection should be closed!
448
}
449
330
-int aclk_lws_wss_client_write(struct aclk_lws_wss_engine_instance *inst, void *buf, size_t count)
450
+int aclk_lws_wss_client_write(void *buf, size_t count)
451
{
332
- if(inst && inst->lws_wsi && inst->websocket_connection_up)
333
- {
334
- aclk_lws_mutex_lock(&inst->write_buf_mutex);
335
- lws_wss_packet_buffer_append(&inst->write_buffer_head, lws_wss_packet_buffer_new(buf, count));
336
- aclk_lws_mutex_unlock(&inst->write_buf_mutex);
337
-
338
- lws_callback_on_writable(inst->lws_wsi);
339
- return count;
340
- }
341
- return 0;
452
+ if (engine_instance && engine_instance->lws_wsi && engine_instance->websocket_connection_up) {
453
+ aclk_lws_mutex_lock(&engine_instance->write_buf_mutex);
454
+ lws_wss_packet_buffer_append(&engine_instance->write_buffer_head, lws_wss_packet_buffer_new(buf, count));
455
+ aclk_lws_mutex_unlock(&engine_instance->write_buf_mutex);
456
+
457
+ lws_callback_on_writable(engine_instance->lws_wsi);
458
+ return count;
459
+ }
460
+ return 0;
461
}
462
344
-int aclk_lws_wss_client_read(struct aclk_lws_wss_engine_instance *inst, void *buf, size_t count)
463
+int aclk_lws_wss_client_read(void *buf, size_t count)
464
{
346
- size_t data_to_be_read = count;
347
-
348
- aclk_lws_mutex_lock(&inst->read_buf_mutex);
349
- size_t readable_byte_count = lws_ring_get_count_waiting_elements(inst->read_ringbuffer, NULL);
350
- if(unlikely(readable_byte_count == 0)) {
351
- errno = EAGAIN;
352
- data_to_be_read = -1;
353
- goto abort;
354
- }
465
+ size_t data_to_be_read = count;
466
+
467
+ aclk_lws_mutex_lock(&engine_instance->read_buf_mutex);
468
+ size_t readable_byte_count = lws_ring_get_count_waiting_elements(engine_instance->read_ringbuffer, NULL);
469
+ if (unlikely(readable_byte_count == 0)) {
470
+ errno = EAGAIN;
471
+ data_to_be_read = -1;
472
+ goto abort;
473
+ }
474
356
- if( readable_byte_count < data_to_be_read )
357
- data_to_be_read = readable_byte_count;
475
+ if (readable_byte_count < data_to_be_read)
476
+ data_to_be_read = readable_byte_count;
477
359
- data_to_be_read = lws_ring_consume(inst->read_ringbuffer, NULL, buf, data_to_be_read);
360
- if(data_to_be_read == readable_byte_count)
361
- inst->data_to_read = 0;
478
+ data_to_be_read = lws_ring_consume(engine_instance->read_ringbuffer, NULL, buf, data_to_be_read);
479
+ if (data_to_be_read == readable_byte_count)
480
+ engine_instance->data_to_read = 0;
481
482
abort:
364
- aclk_lws_mutex_unlock(&inst->read_buf_mutex);
365
- return data_to_be_read;
483
+ aclk_lws_mutex_unlock(&engine_instance->read_buf_mutex);
484
+ return data_to_be_read;
485
}
486
368
-int aclk_lws_wss_service_loop(struct aclk_lws_wss_engine_instance *inst)
487
+void aclk_lws_wss_service_loop()
488
{
370
- return lws_service(inst->lws_context, 0);
489
+ if (engine_instance)
490
+ lws_service(engine_instance->lws_context, 0);
491
}
492
493
// in case the MQTT connection disconnect while lws transport is still operational
494
// we should drop connection and reconnect
495
// this function should be called when that happens to notify lws of that situation
376
-void aclk_lws_wss_mqtt_layer_disconect_notif(struct aclk_lws_wss_engine_instance *inst)
496
+void aclk_lws_wss_mqtt_layer_disconect_notif()
497
{
378
- if(inst->lws_wsi && inst->websocket_connection_up) {
379
- inst->upstream_reconnect_request = 1;
380
- lws_callback_on_writable(inst->lws_wsi); //here we just do it to ensure we get callback called from lws, we don't need any actual data to be written.
381
- }
382
-}
\ No newline at end of file
498
+ if (!engine_instance)
499
+ return;
500
+ if (engine_instance->lws_wsi && engine_instance->websocket_connection_up) {
501
+ engine_instance->upstream_reconnect_request = 1;
502
+ lws_callback_on_writable(
503
+ engine_instance->lws_wsi); //here we just do it to ensure we get callback called from lws, we don't need any actual data to be written.
504
+ }
505
+}
aclk/aclk_lws_wss_client.h
+38
-37
@@ -5,8 +5,6 @@
5
6
#include "libnetdata/libnetdata.h"
7
8
-#define ACLK_LWS_WSS_RECONNECT_TIMEOUT 5
9
-
8
// This is as define because ideally the ACLK at high level
9
// can do mosqitto writes and reads only from one thread
10
// which is cleaner implementation IMHO
@@ -14,64 +12,67 @@
12
// is simpler
13
#define ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED 1
14
17
-#define ACLK_LWS_WSS_RECV_BUFF_SIZE_BYTES 128*1024
15
+#define ACLK_LWS_WSS_RECV_BUFF_SIZE_BYTES (128 * 1024)
16
17
#ifdef ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED
20
- #define aclk_lws_mutex_init(x) netdata_mutex_init(x)
21
- #define aclk_lws_mutex_lock(x) netdata_mutex_lock(x)
22
- #define aclk_lws_mutex_unlock(x) netdata_mutex_unlock(x)
18
+#define aclk_lws_mutex_init(x) netdata_mutex_init(x)
19
+#define aclk_lws_mutex_lock(x) netdata_mutex_lock(x)
20
+#define aclk_lws_mutex_unlock(x) netdata_mutex_unlock(x)
21
#else
24
- #define aclk_lws_mutex_init(x)
25
- #define aclk_lws_mutex_lock(x)
26
- #define aclk_lws_mutex_unlock(x)
22
+#define aclk_lws_mutex_init(x)
23
+#define aclk_lws_mutex_lock(x)
24
+#define aclk_lws_mutex_unlock(x)
25
#endif
26
27
struct aclk_lws_wss_engine_callbacks {
30
- void (*connection_established_callback)();
31
- void (*data_rcvd_callback)();
32
- void (*data_writable_callback)();
28
+ void (*connection_established_callback)();
29
+ void (*data_rcvd_callback)();
30
+ void (*data_writable_callback)();
31
void (*connection_closed)();
32
};
33
34
struct lws_wss_packet_buffer;
35
36
struct aclk_lws_wss_engine_instance {
39
- //target host/port for connection
40
- const char *host;
41
- int port;
37
+ //target host/port for connection
38
+ char *host;
39
+ int port;
40
43
- //internal data
44
- struct lws_context *lws_context;
45
- struct lws *lws_wsi;
41
+ //internal data
42
+ struct lws_context *lws_context;
43
+ struct lws *lws_wsi;
44
45
#ifdef ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED
48
- netdata_mutex_t write_buf_mutex;
49
- netdata_mutex_t read_buf_mutex;
46
+ netdata_mutex_t write_buf_mutex;
47
+ netdata_mutex_t read_buf_mutex;
48
#endif
49
52
- struct lws_wss_packet_buffer *write_buffer_head;
53
- struct lws_ring *read_ringbuffer;
54
-
55
- struct aclk_lws_wss_engine_callbacks callbacks;
50
+ struct lws_wss_packet_buffer *write_buffer_head;
51
+ struct lws_ring *read_ringbuffer;
52
57
- //flags to be readed by engine user
58
- int websocket_connection_up;
53
+ //flags to be readed by engine user
54
+ int websocket_connection_up;
55
60
-// currently this is by default disabled
56
+ // currently this is by default disabled
57
62
- int data_to_read;
63
- int upstream_reconnect_request;
58
+ int data_to_read;
59
+ int upstream_reconnect_request;
60
};
61
66
-struct aclk_lws_wss_engine_instance* aclk_lws_wss_client_init (const struct aclk_lws_wss_engine_callbacks *callbacks, const char *target_hostname, int target_port);
67
-void aclk_lws_wss_client_destroy(struct aclk_lws_wss_engine_instance* inst);
62
+void aclk_lws_wss_client_destroy();
63
+
64
+int aclk_lws_wss_connect(char *host, int port);
65
69
-void aclk_lws_wss_connect(struct aclk_lws_wss_engine_instance *inst);
66
+int aclk_lws_wss_client_write(void *buf, size_t count);
67
+int aclk_lws_wss_client_read(void *buf, size_t count);
68
+void aclk_lws_wss_service_loop();
69
71
-int aclk_lws_wss_client_write(struct aclk_lws_wss_engine_instance *inst, void *buf, size_t count);
72
-int aclk_lws_wss_client_read (struct aclk_lws_wss_engine_instance *inst, void *buf, size_t count);
73
-int aclk_lws_wss_service_loop(struct aclk_lws_wss_engine_instance *inst);
70
+void aclk_lws_wss_mqtt_layer_disconect_notif();
71
75
-void aclk_lws_wss_mqtt_layer_disconect_notif(struct aclk_lws_wss_engine_instance *inst);
72
+// Notifications inside the layer above
73
+void aclk_lws_connection_established();
74
+void aclk_lws_connection_data_received();
75
+void aclk_lws_connection_closed();
76
77
-#endif
\ No newline at end of file
77
+
78
+#endif
aclk/agent_cloud_link.c
+671
-245
@@ -3,21 +3,81 @@
3
#include "libnetdata/libnetdata.h"
4
#include "agent_cloud_link.h"
5
6
-// Read from the config file -- new section [agent_cloud_link]
7
-// Defaults are supplied
6
+// State-machine for the on-connect metadata transmission.
7
+// TODO: The AGENT_STATE should be centralized as it would be useful to control error-logging during the initial
8
+// agent startup phase.
9
+static ACLK_METADATA_STATE aclk_metadata_submitted = ACLK_METADATA_REQUIRED;
10
+static AGENT_STATE agent_state = AGENT_INITIALIZING;
11
+
12
+// Other global state
13
+static int aclk_subscribed = 0;
14
+static int aclk_disable_single_updates = 0;
15
+static time_t last_init_sequence = 0;
16
+static int waiting_init = 1;
17
+static char *aclk_username = NULL;
18
+static char *aclk_password = NULL;
19
+
20
+static char *global_base_topic = NULL;
21
+static int aclk_connecting = 0;
22
+static int aclk_connected = 0;
23
9
-int aclk_port = ACLK_DEFAULT_PORT;
10
-char *aclk_hostname = ACLK_DEFAULT_HOST;
11
-int aclk_subscribed = 0;
12
-int aclk_disable_single_updates = 0;
24
+static netdata_mutex_t aclk_mutex = NETDATA_MUTEX_INITIALIZER;
25
+static netdata_mutex_t query_mutex = NETDATA_MUTEX_INITIALIZER;
26
+static netdata_mutex_t collector_mutex = NETDATA_MUTEX_INITIALIZER;
27
+
28
+#define ACLK_LOCK netdata_mutex_lock(&aclk_mutex)
29
+#define ACLK_UNLOCK netdata_mutex_unlock(&aclk_mutex)
30
14
-int aclk_metadata_submitted = 0;
15
-int agent_state = 0;
16
-time_t last_init_sequence = 0;
17
-int waiting_init = 1;
31
+#define COLLECTOR_LOCK netdata_mutex_lock(&collector_mutex)
32
+#define COLLECTOR_UNLOCK netdata_mutex_unlock(&collector_mutex)
33
+
34
+#define QUERY_LOCK netdata_mutex_lock(&query_mutex)
35
+#define QUERY_UNLOCK netdata_mutex_unlock(&query_mutex)
36
+
37
+pthread_cond_t query_cond_wait = PTHREAD_COND_INITIALIZER;
38
+pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
39
+
40
+#define QUERY_THREAD_LOCK pthread_mutex_lock(&query_lock_wait);
41
+#define QUERY_THREAD_UNLOCK pthread_mutex_unlock(&query_lock_wait)
42
+#define QUERY_THREAD_WAKEUP pthread_cond_signal(&query_cond_wait)
43
+
44
+/*
45
+ * Maintain a list of collectors and chart count
46
+ * If all the charts of a collector are deleted
47
+ * then a new metadata dataset must be send to the cloud
48
+ *
49
+ */
50
+struct _collector {
51
+ time_t created;
52
+ uint32_t count; //chart count
53
+ uint32_t hostname_hash;
54
+ uint32_t plugin_hash;
55
+ uint32_t module_hash;
56
+ char *hostname;
57
+ char *plugin_name;
58
+ char *module_name;
59
+ struct _collector *next;
60
+};
61
19
-char *global_base_topic = NULL;
20
-int aclk_connecting = 0;
62
+struct _collector *collector_list = NULL;
63
+
64
+struct aclk_query {
65
+ time_t created;
66
+ time_t run_after; // Delay run until after this time
67
+ ACLK_CMD cmd; // What command is this
68
+ char *topic; // Topic to respond to
69
+ char *data; // Internal data (NULL if request from the cloud)
70
+ char *msg_id; // msg_id generated by the cloud (NULL if internal)
71
+ char *query; // The actual query
72
+ u_char deleted; // Mark deleted for garbage collect
73
+ struct aclk_query *next;
74
+};
75
+
76
+struct aclk_query_queue {
77
+ struct aclk_query *aclk_query_head;
78
+ struct aclk_query *aclk_query_tail;
79
+ uint64_t count;
80
+} aclk_queue = { .aclk_query_head = NULL, .aclk_query_tail = NULL, .count = 0 };
81
82
char *create_uuid()
83
{
@@ -34,30 +94,30 @@ int cloud_to_agent_parse(JSON_ENTRY *e)
94
{
95
struct aclk_request *data = e->callback_data;
96
37
- switch(e->type) {
97
+ switch (e->type) {
98
case JSON_OBJECT:
99
case JSON_ARRAY:
40
- break;
100
+ break;
101
case JSON_STRING:
42
- if (!strcmp(e->name, ACLK_JSON_IN_MSGID)) {
102
+ if (!strcmp(e->name, "msg-id")) {
103
data->msg_id = strdupz(e->data.string);
104
break;
105
}
46
- if (!strcmp(e->name, ACLK_JSON_IN_TYPE)) {
106
+ if (!strcmp(e->name, "type")) {
107
data->type_id = strdupz(e->data.string);
108
break;
109
}
50
- if (!strcmp(e->name, ACLK_JSON_IN_TOPIC)) {
110
+ if (!strcmp(e->name, "callback-topic")) {
111
data->callback_topic = strdupz(e->data.string);
112
break;
113
}
54
- if (!strcmp(e->name, ACLK_JSON_IN_URL)) {
114
+ if (!strcmp(e->name, "payload")) {
115
data->payload = strdupz(e->data.string);
116
break;
117
}
118
break;
119
case JSON_NUMBER:
60
- if (!strcmp(e->name, ACLK_JSON_IN_VERSION)) {
120
+ if (!strcmp(e->name, "version")) {
121
data->version = atoi(e->original_string);
122
break;
123
}
@@ -73,71 +133,55 @@ int cloud_to_agent_parse(JSON_ENTRY *e)
133
}
134
135
76
-// Set when we have connection up and running from the connection callback
77
-int aclk_connection_initialized = 0;
78
-// TODO modify previous comment if this stays this way
79
-// con_initialized means library is initialized and ready to be used
80
-// acklk_connected means there is actually an established connection
81
-int aclk_mqtt_connected = 0;
82
-
83
-static netdata_mutex_t aclk_mutex = NETDATA_MUTEX_INITIALIZER;
84
-static netdata_mutex_t query_mutex = NETDATA_MUTEX_INITIALIZER;
85
-static netdata_mutex_t collector_mutex = NETDATA_MUTEX_INITIALIZER;
86
-
87
-#define ACLK_LOCK netdata_mutex_lock(&aclk_mutex)
88
-#define ACLK_UNLOCK netdata_mutex_unlock(&aclk_mutex)
89
-
90
-#define COLLECTOR_LOCK netdata_mutex_lock(&collector_mutex)
91
-#define COLLECTOR_UNLOCK netdata_mutex_unlock(&collector_mutex)
92
-
93
-#define QUERY_LOCK netdata_mutex_lock(&query_mutex)
94
-#define QUERY_UNLOCK netdata_mutex_unlock(&query_mutex)
95
-
96
-pthread_cond_t query_cond_wait = PTHREAD_COND_INITIALIZER;
97
-pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
136
+static RSA *aclk_private_key = NULL;
137
+static int create_private_key()
138
+{
139
+ char filename[FILENAME_MAX + 1]; struct stat statbuf;
140
+ snprintfz(filename, FILENAME_MAX, "%s/claim.d/private.pem", netdata_configured_user_config_dir);
141
99
-#define QUERY_THREAD_LOCK pthread_mutex_lock(&query_lock_wait);
100
-#define QUERY_THREAD_UNLOCK pthread_mutex_unlock(&query_lock_wait)
101
-#define QUERY_THREAD_WAKEUP pthread_cond_signal(&query_cond_wait)
142
+ if (lstat(filename, &statbuf) != 0) {
143
+ error("Claimed agent cannot establish ACLK - private key not found '%s' failed.", filename);
144
+ return 1;
145
+ }
146
+ if (unlikely(statbuf.st_size == 0)) {
147
+ info("Claimed agent cannot establish ACLK - private key '%s' is empty.", filename);
148
+ return 1;
149
+ }
150
151
+ FILE *f = fopen(filename, "rt");
152
+ if (unlikely(f == NULL)) {
153
+ error("Claimed agent cannot establish ACLK - unable to open private key '%s'.", filename);
154
+ return 1;
155
+ }
156
104
-/*
105
- * Maintain a list of collectors and chart count
106
- * If all the charts of a collector are deleted
107
- * then a new metadata dataset must be send to the cloud
108
- *
109
- */
110
-struct _collector {
111
- time_t created;
112
- u_int32_t count; //chart count
113
- u_int32_t hostname_hash;
114
- u_int32_t plugin_hash;
115
- u_int32_t module_hash;
116
- char *hostname;
117
- char *plugin_name;
118
- char *module_name;
119
- struct _collector *next;
120
-};
157
+ char *private_key = callocz(1, statbuf.st_size + 1);
158
+ size_t bytes_read = fread(private_key, 1, statbuf.st_size, f);
159
+ private_key[bytes_read] = 0;
160
+ debug(D_ACLK, "Claimed agent loaded private key len=%zu bytes", bytes_read);
161
+ fclose(f);
162
122
-struct _collector *collector_list = NULL;
163
+ BIO *key_bio = BIO_new_mem_buf(private_key, -1);
164
+ if (key_bio==NULL) {
165
+ error("Claimed agent cannot establish ACLK - failed to create BIO for key");
166
+ goto biofailed;
167
+ }
168
124
-struct aclk_query {
125
- time_t created;
126
- time_t run_after; // Delay run until after this time
127
- ACLK_CMD cmd; // What command is this
128
- char *topic; // Topic to respond to
129
- char *data; // Internal data (NULL if request from the cloud)
130
- char *msg_id; // msg_id generated by the cloud (NULL if internal)
131
- char *query; // The actual query
132
- u_char deleted; // Mark deleted for garbage collect
133
- struct aclk_query *next;
134
-};
169
+ aclk_private_key = PEM_read_bio_RSAPrivateKey(key_bio, NULL, NULL, NULL);
170
+ BIO_free(key_bio);
171
+ if (aclk_private_key!=NULL)
172
+ {
173
+ freez(private_key);
174
+ return 0;
175
+ }
176
+ char err[512];
177
+ ERR_error_string_n(ERR_get_error(), err, sizeof(err));
178
+ error("Claimed agent cannot establish ACLK - cannot create private key: %s", err);
179
+ freez(err);
180
136
-struct aclk_query_queue {
137
- struct aclk_query *aclk_query_head;
138
- struct aclk_query *aclk_query_tail;
139
- u_int64_t count;
140
-} aclk_queue = { .aclk_query_head = NULL, .aclk_query_tail = NULL, .count = 0 };
181
+biofailed:
182
+ freez(private_key);
183
+ return 1;
184
+}
185
186
/*
187
* After a connection failure -- delay in milliseconds
@@ -150,12 +194,12 @@ struct aclk_query_queue {
194
*/
195
unsigned long int aclk_reconnect_delay(int mode)
196
{
153
- static int fail = -1;
197
+ static int fail = -1;
198
unsigned long int delay;
199
200
if (!mode || fail == -1) {
201
srandom(time(NULL));
158
- fail = mode-1;
202
+ fail = mode - 1;
203
return 0;
204
}
205
@@ -163,13 +207,12 @@ unsigned long int aclk_reconnect_delay(int mode)
207
208
if (delay >= ACLK_MAX_BACKOFF_DELAY) {
209
delay = ACLK_MAX_BACKOFF_DELAY * 1000;
166
- }
167
- else {
210
+ } else {
211
fail++;
212
delay = (delay * 1000) + (random() % 1000);
213
}
214
172
-// sleep_usec(USEC_PER_MS * delay);
215
+ // sleep_usec(USEC_PER_MS * delay);
216
217
return delay;
218
}
@@ -220,7 +263,8 @@ struct aclk_query *aclk_query_find_position(time_t time_to_run)
263
}
264
265
// Need to have a QUERY lock before calling this
223
-struct aclk_query *aclk_query_find(char *topic, char *data, char *msg_id, char *query, ACLK_CMD cmd, struct aclk_query **last_query)
266
+struct aclk_query *
267
+aclk_query_find(char *topic, char *data, char *msg_id, char *query, ACLK_CMD cmd, struct aclk_query **last_query)
268
{
269
struct aclk_query *tmp_query, *prev_query;
270
UNUSED(cmd);
@@ -232,7 +276,6 @@ struct aclk_query *aclk_query_find(char *topic, char *data, char *msg_id, char *
276
if (strcmp(tmp_query->topic, topic) == 0 && (!query || strcmp(tmp_query->query, query) == 0)) {
277
if ((!data || (data && strcmp(data, tmp_query->data) == 0)) &&
278
(!msg_id || (msg_id && strcmp(msg_id, tmp_query->msg_id) == 0))) {
235
-
279
if (likely(last_query))
280
*last_query = prev_query;
281
return tmp_query;
@@ -258,7 +301,7 @@ int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run
301
return 0;
302
303
// Ignore all commands if agent not stable and reset the last_init_sequence mark
261
- if (agent_state == 0) {
304
+ if (agent_state == AGENT_INITIALIZING) {
305
last_init_sequence = now_realtime_sec();
306
return 0;
307
}
@@ -306,7 +349,7 @@ int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run
349
new_query->created = now_realtime_sec();
350
new_query->run_after = run_after;
351
309
- debug(D_ACLK, "Added query (%s) (%s)", topic, query?query:"");
352
+ debug(D_ACLK, "Added query (%s) (%s)", topic, query ? query : "");
353
354
tmp_query = aclk_query_find_position(run_after);
355
@@ -401,7 +444,6 @@ struct aclk_query *aclk_queue_pop()
444
// Need to check if additional logic should be added to make sure that there
445
// is enough information to determine the base topic at init time
446
404
-
447
char *create_publish_base_topic()
448
{
449
if (unlikely(!is_agent_claimed()))
@@ -446,14 +488,12 @@ char *get_topic(char *sub_topic, char *final_topic, int max_size)
488
return final_topic;
489
}
490
449
-
491
/*
492
* Free a collector structure
493
*/
494
495
static void _free_collector(struct _collector *collector)
496
{
456
-
497
if (likely(collector->plugin_name))
498
freez(collector->plugin_name);
499
@@ -473,8 +513,7 @@ static void _free_collector(struct _collector *collector)
513
#ifdef ACLK_DEBUG
514
static void _dump_connector_list()
515
{
476
-
477
- struct _collector *tmp_collector;
516
+ struct _collector *tmp_collector;
517
518
COLLECTOR_LOCK;
519
@@ -496,7 +535,6 @@ static void _dump_connector_list()
535
tmp_collector->module_name ? tmp_collector->module_name : "", tmp_collector->count);
536
537
tmp_collector = tmp_collector->next;
499
-
538
}
539
info("DUMPING ALL COLLECTORS DONE");
540
COLLECTOR_UNLOCK;
@@ -509,7 +547,7 @@ static void _dump_connector_list()
547
*/
548
static void _reset_connector_list()
549
{
512
- struct _collector *tmp_collector, *next_collector;
550
+ struct _collector *tmp_collector, *next_collector;
551
552
COLLECTOR_LOCK;
553
@@ -519,9 +557,9 @@ static void _reset_connector_list()
557
}
558
559
// Note that the first entry is "dummy"
522
- tmp_collector = collector_list->next;
560
+ tmp_collector = collector_list->next;
561
collector_list->count = 0;
524
- collector_list->next = NULL;
562
+ collector_list->next = NULL;
563
564
// We broke the link; we can unlock
565
COLLECTOR_UNLOCK;
@@ -533,14 +571,14 @@ static void _reset_connector_list()
571
}
572
}
573
536
-
574
/*
575
* Find a collector (if it exists)
576
* Must lock before calling this
577
* If last_collector is not null, it will return the previous collector in the linked
578
* list (used in collector delete)
579
*/
543
-static struct _collector *_find_collector(const char *hostname, const char *plugin_name, const char *module_name, struct _collector **last_collector)
580
+static struct _collector *_find_collector(
581
+ const char *hostname, const char *plugin_name, const char *module_name, struct _collector **last_collector)
582
{
583
struct _collector *tmp_collector, *prev_collector;
584
uint32_t plugin_hash;
@@ -555,21 +593,18 @@ static struct _collector *_find_collector(const char *hostname, const char *plug
593
if (unlikely(!collector_list->next))
594
return NULL;
595
558
- plugin_hash = plugin_name?simple_hash(plugin_name):1;
559
- module_hash = module_name?simple_hash(module_name):1;
596
+ plugin_hash = plugin_name ? simple_hash(plugin_name) : 1;
597
+ module_hash = module_name ? simple_hash(module_name) : 1;
598
hostname_hash = simple_hash(hostname);
599
600
// Note that the first entry is "dummy"
563
- tmp_collector = collector_list->next;
601
+ tmp_collector = collector_list->next;
602
prev_collector = collector_list;
603
while (tmp_collector) {
566
- if (plugin_hash == tmp_collector->plugin_hash &&
567
- module_hash == tmp_collector->module_hash &&
568
- hostname_hash == tmp_collector->hostname_hash &&
569
- (!strcmp(hostname, tmp_collector->hostname)) &&
604
+ if (plugin_hash == tmp_collector->plugin_hash && module_hash == tmp_collector->module_hash &&
605
+ hostname_hash == tmp_collector->hostname_hash && (!strcmp(hostname, tmp_collector->hostname)) &&
606
(!plugin_name || !tmp_collector->plugin_name || !strcmp(plugin_name, tmp_collector->plugin_name)) &&
607
(!module_name || !tmp_collector->module_name || !strcmp(module_name, tmp_collector->module_name))) {
572
-
608
if (unlikely(last_collector))
609
*last_collector = prev_collector;
610
@@ -593,7 +628,7 @@ static struct _collector *_find_collector(const char *hostname, const char *plug
628
*/
629
static struct _collector *_del_collector(const char *hostname, const char *plugin_name, const char *module_name)
630
{
596
- struct _collector *tmp_collector, *prev_collector = NULL;
631
+ struct _collector *tmp_collector, *prev_collector = NULL;
632
633
tmp_collector = _find_collector(hostname, plugin_name, module_name, &prev_collector);
634
@@ -605,35 +640,35 @@ static struct _collector *_del_collector(const char *hostname, const char *plugi
640
return tmp_collector;
641
}
642
608
-
643
/*
644
* Add a new collector (plugin / module) to the list
645
* If it already exists just update the chart count
646
*
647
* Lock before calling
648
*/
615
-static struct _collector *_add_collector(const char *hostname, const char *plugin_name, const char *module_name)
649
+static struct _collector *_add_collector(const char *hostname, const char *plugin_name, const char *module_name)
650
{
617
- struct _collector *tmp_collector;
651
+ struct _collector *tmp_collector;
652
653
tmp_collector = _find_collector(hostname, plugin_name, module_name, NULL);
654
655
if (unlikely(!tmp_collector)) {
622
-
656
tmp_collector = callocz(1, sizeof(struct _collector));
657
tmp_collector->hostname_hash = simple_hash(hostname);
625
- tmp_collector->plugin_hash = plugin_name?simple_hash(plugin_name):1;
626
- tmp_collector->module_hash = module_name?simple_hash(module_name):1;
658
+ tmp_collector->plugin_hash = plugin_name ? simple_hash(plugin_name) : 1;
659
+ tmp_collector->module_hash = module_name ? simple_hash(module_name) : 1;
660
661
tmp_collector->hostname = strdupz(hostname);
629
- tmp_collector->plugin_name = plugin_name?strdupz(plugin_name):NULL;
630
- tmp_collector->module_name = module_name?strdupz(module_name):NULL;
662
+ tmp_collector->plugin_name = plugin_name ? strdupz(plugin_name) : NULL;
663
+ tmp_collector->module_name = module_name ? strdupz(module_name) : NULL;
664
665
tmp_collector->next = collector_list->next;
666
collector_list->next = tmp_collector;
667
}
668
tmp_collector->count++;
636
- debug(D_ACLK, "ADD COLLECTOR %s [%s:%s] -- chart %u", hostname, plugin_name?plugin_name:"*", module_name?module_name:"*", tmp_collector->count);
669
+ debug(
670
+ D_ACLK, "ADD COLLECTOR %s [%s:%s] -- chart %u", hostname, plugin_name ? plugin_name : "*",
671
+ module_name ? module_name : "*", tmp_collector->count);
672
return tmp_collector;
673
}
674
@@ -643,7 +678,7 @@ static struct _collector *_add_collector(const char *hostname, const char *plug
678
*/
679
void aclk_add_collector(const char *hostname, const char *plugin_name, const char *module_name)
680
{
646
- struct _collector *tmp_collector;
681
+ struct _collector *tmp_collector;
682
683
COLLECTOR_LOCK;
684
@@ -669,7 +704,7 @@ void aclk_add_collector(const char *hostname, const char *plugin_name, const cha
704
*/
705
void aclk_del_collector(const char *hostname, const char *plugin_name, const char *module_name)
706
{
672
- struct _collector *tmp_collector;
707
+ struct _collector *tmp_collector;
708
709
COLLECTOR_LOCK;
710
@@ -680,7 +715,9 @@ void aclk_del_collector(const char *hostname, const char *plugin_name, const cha
715
return;
716
}
717
683
- debug(D_ACLK, "DEL COLLECTOR [%s:%s] -- charts %u", plugin_name?plugin_name:"*", module_name?module_name:"*", tmp_collector->count);
718
+ debug(
719
+ D_ACLK, "DEL COLLECTOR [%s:%s] -- charts %u", plugin_name ? plugin_name : "*", module_name ? module_name : "*",
720
+ tmp_collector->count);
721
722
COLLECTOR_UNLOCK;
723
@@ -689,29 +726,6 @@ void aclk_del_collector(const char *hostname, const char *plugin_name, const cha
726
_free_collector(tmp_collector);
727
}
728
692
-
693
-// Wait for ACLK connection to be established
694
-int aclk_wait_for_initialization()
695
-{
696
- if (unlikely(!aclk_connection_initialized)) {
697
- time_t now = now_realtime_sec();
698
-
699
- while (!aclk_connection_initialized && (now_realtime_sec() - now) < ACLK_INITIALIZATION_WAIT) {
700
- sleep_usec(USEC_PER_SEC * ACLK_INITIALIZATION_SLEEP_WAIT);
701
- _link_event_loop(0);
702
-
703
- if (unlikely(!netdata_exit))
704
- return 1;
705
- }
706
-
707
- if (unlikely(!aclk_connection_initialized)) {
708
- error("ACLK connection cannot be established");
709
- return 1;
710
- }
711
- }
712
- return 0;
713
-}
714
-
729
int aclk_execute_query(struct aclk_query *this_query)
730
{
731
if (strncmp(this_query->query, "/api/v1/", 8) == 0) {
@@ -733,18 +747,18 @@ int aclk_execute_query(struct aclk_query *this_query)
747
748
// TODO: handle bad response perhaps in a different way. For now it does to the payload
749
int rc = web_client_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
736
- BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
750
+ BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
751
buffer_flush(local_buffer);
752
local_buffer->contenttype = CT_APPLICATION_JSON;
753
754
aclk_create_header(local_buffer, "http", this_query->msg_id);
755
742
- if (rc != HTTP_RESP_OK || strcmp(mysep?mysep+1:"noop", "badge.svg") == 0)
756
+ if (rc != HTTP_RESP_OK || strcmp(mysep ? mysep + 1 : "noop", "badge.svg") == 0)
757
buffer_sprintf(local_buffer, "\"%s\"", aclk_encode_response(w->response.data)->buffer);
758
else
759
buffer_sprintf(local_buffer, "%s", aclk_encode_response(w->response.data)->buffer);
760
747
- buffer_sprintf(local_buffer,"\n}");
761
+ buffer_sprintf(local_buffer, "\n}");
762
763
aclk_send_message(this_query->topic, local_buffer->buffer, this_query->msg_id);
764
@@ -765,7 +779,7 @@ int aclk_process_query()
779
struct aclk_query *this_query;
780
static long int query_count = 0;
781
768
- if (!aclk_connection_initialized)
782
+ if (!aclk_connected)
783
return 0;
784
785
this_query = aclk_queue_pop();
@@ -781,15 +795,14 @@ int aclk_process_query()
795
query_count++;
796
797
debug(
784
- D_ACLK, "Query #%ld (%s) size=%ld in queue %d seconds", query_count, this_query->topic, this_query->query?strlen(this_query->query):0,
785
- (int)(now_realtime_sec() - this_query->created));
798
+ D_ACLK, "Query #%ld (%s) size=%zu in queue %d seconds", query_count, this_query->topic,
799
+ this_query->query ? strlen(this_query->query) : 0, (int)(now_realtime_sec() - this_query->created));
800
801
switch (this_query->cmd) {
788
-
802
case ACLK_CMD_ONCONNECT:
803
debug(D_ACLK, "EXECUTING on connect metadata command");
804
aclk_send_metadata();
792
- aclk_metadata_submitted = 2;
805
+ aclk_metadata_submitted = ACLK_METADATA_SENT;
806
break;
807
808
case ACLK_CMD_CHART:
@@ -821,8 +834,7 @@ int aclk_process_query()
834
default:
835
break;
836
}
824
- debug(
825
- D_ACLK, "Query #%ld (%s) done", query_count, this_query->topic);
837
+ debug(D_ACLK, "Query #%ld (%s) done", query_count, this_query->topic);
838
839
aclk_query_free(this_query);
840
@@ -837,13 +849,13 @@ int aclk_process_query()
849
850
int aclk_process_queries()
851
{
840
- if (unlikely(netdata_exit || !aclk_connection_initialized))
852
+ if (unlikely(netdata_exit || !aclk_connected))
853
return 0;
854
855
if (likely(!aclk_queue.count))
856
return 0;
857
846
- debug(D_ACLK, "Processing %d queries", (int ) aclk_queue.count);
858
+ debug(D_ACLK, "Processing %d queries", (int)aclk_queue.count);
859
860
//TODO: may consider possible throttling here
861
while (aclk_process_query()) {
@@ -876,20 +888,20 @@ static void aclk_query_thread_cleanup(void *ptr)
888
* On startup wait for the agent collectors to initialize
889
* Expect at least a time of ACLK_STABLE_TIMEOUT seconds
890
* of no new collectors coming in in order to mark the agent
879
- * as stable (set agent_state = 1)
891
+ * as stable (set agent_state = AGENT_STABLE)
892
*/
893
void *aclk_query_main_thread(void *ptr)
894
{
895
netdata_thread_cleanup_push(aclk_query_thread_cleanup, ptr);
896
885
- while (!agent_state && !netdata_exit) {
886
- time_t checkpoint;
897
+ while (agent_state == AGENT_INITIALIZING && !netdata_exit) {
898
+ time_t checkpoint;
899
888
- checkpoint = now_realtime_sec() - last_init_sequence;
900
+ checkpoint = now_realtime_sec() - last_init_sequence;
901
info("Waiting for agent collectors to initialize");
902
sleep_usec(USEC_PER_SEC * ACLK_STABLE_TIMEOUT);
903
if (checkpoint > ACLK_STABLE_TIMEOUT) {
892
- agent_state = 1;
904
+ agent_state = AGENT_STABLE;
905
info("AGENT stable, last collector initialization activity was %ld seconds ago", checkpoint);
906
#ifdef ACLK_DEBUG
907
_dump_connector_list();
@@ -898,9 +910,8 @@ void *aclk_query_main_thread(void *ptr)
910
}
911
912
while (!netdata_exit) {
901
-
913
if (unlikely(!aclk_metadata_submitted)) {
903
- aclk_metadata_submitted = 1;
914
+ aclk_metadata_submitted = ACLK_METADATA_CMD_QUEUED;
915
aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT);
916
}
917
@@ -934,6 +945,412 @@ static void aclk_main_cleanup(void *ptr)
945
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
946
}
947
948
+int send_https_request(char *method, char *host, char *port, char *url, BUFFER *b, char *payload)
949
+{
950
+ struct timeval timeout = { .tv_sec = 30, .tv_usec = 0 };
951
+ int rc=1;
952
+
953
+ size_t payload_len = 0;
954
+ if (payload != NULL)
955
+ payload_len = strlen(payload);
956
+
957
+ buffer_flush(b);
958
+ buffer_sprintf(
959
+ b,
960
+ "%s %s HTTP/1.1\r\nHost: %s\r\nAccept: plain/text\r\nContent-length: %zu\r\nAccept-Language: en-us\r\n"
961
+ "User-Agent: Netdata/rocks\r\n\r\n",
962
+ method, url, host, payload_len);
963
+ if (payload != NULL)
964
+ buffer_strcat(b, payload);
965
+ debug(D_ACLK, "Sending HTTPS req (%zu bytes): '%s'", b->len, buffer_tostring(b));
966
+ int sock = connect_to_this_ip46(IPPROTO_TCP, SOCK_STREAM, host, 0, port, &timeout);
967
+
968
+ if (unlikely(sock == -1)) {
969
+ error("Handshake failed");
970
+ return 1;
971
+ }
972
+
973
+ SSL_CTX *ctx = security_initialize_openssl_client();
974
+ if (ctx==NULL) {
975
+ error("Cannot allocate SSL context");
976
+ goto exit_sock;
977
+ }
978
+ // Certificate chain: not updating the stores - do we need private CA roots?
979
+ // Calls to SSL_CTX_load_verify_locations would go here.
980
+ SSL *ssl = SSL_new(ctx);
981
+ if (ssl==NULL) {
982
+ error("Cannot allocate SSL");
983
+ goto exit_CTX;
984
+ }
985
+ SSL_set_fd(ssl, sock);
986
+ int err = SSL_connect(ssl);
987
+ if (err!=1) {
988
+ error("SSL_connect() failed with err=%d", err);
989
+ goto exit_SSL;
990
+ }
991
+ err = SSL_write(ssl, b->buffer, b->len);
992
+ if (err <= 0)
993
+ {
994
+ error("SSL_write() failed with err=%d", err);
995
+ goto exit_SSL;
996
+ }
997
+ buffer_flush(b);
998
+ int bytes_read = SSL_read(ssl, b->buffer, b->size);
999
+ if (bytes_read >= 0) {
1000
+ debug(D_ACLK, "Received %d bytes in response", bytes_read);
1001
+ b->len = bytes_read;
1002
+ }
1003
+ else {
1004
+ error("No response available - SSL_read()=%d", bytes_read);
1005
+ }
1006
+ SSL_shutdown(ssl);
1007
+ rc = 0;
1008
+exit_SSL:
1009
+ SSL_free(ssl);
1010
+exit_CTX:
1011
+ SSL_CTX_free(ctx);
1012
+exit_sock:
1013
+ close(sock);
1014
+ return rc;
1015
+}
1016
+
1017
+struct dictionary_singleton {
1018
+ char *key;
1019
+ char *result;
1020
+};
1021
+
1022
+int json_extract_singleton(JSON_ENTRY *e)
1023
+{
1024
+ struct dictionary_singleton *data = e->callback_data;
1025
+
1026
+ switch (e->type) {
1027
+ case JSON_OBJECT:
1028
+ case JSON_ARRAY:
1029
+ break;
1030
+ case JSON_STRING:
1031
+ if (!strcmp(e->name, data->key)) {
1032
+ data->result = strdupz(e->data.string);
1033
+ break;
1034
+ }
1035
+ break;
1036
+ case JSON_NUMBER:
1037
+ case JSON_BOOLEAN:
1038
+ case JSON_NULL:
1039
+ break;
1040
+ }
1041
+ return 0;
1042
+}
1043
+
1044
+// Base-64 decoder.
1045
+// Note: This is non-validating, invalid input will be decoded without an error.
1046
+// Challenges are packed into json strings so we don't skip newlines.
1047
+// Size errors (i.e. invalid input size or insufficient output space) are caught.
1048
+size_t base64_decode(unsigned char *input, size_t input_size, unsigned char *output, size_t output_size)
1049
+{
1050
+ static char lookup[256];
1051
+ static int first_time=1;
1052
+ if (first_time)
1053
+ {
1054
+ first_time = 0;
1055
+ for(int i=0; i<256; i++)
1056
+ lookup[i] = -1;
1057
+ for(int i='A'; i<='Z'; i++)
1058
+ lookup[i] = i-'A';
1059
+ for(int i='a'; i<='z'; i++)
1060
+ lookup[i] = i-'a' + 26;
1061
+ for(int i='0'; i<='9'; i++)
1062
+ lookup[i] = i-'0' + 52;
1063
+ lookup['+'] = 62;
1064
+ lookup['/'] = 63;
1065
+ }
1066
+ if ((input_size & 3) != 0)
1067
+ {
1068
+ error("Can't decode base-64 input length %zu", input_size);
1069
+ return 0;
1070
+ }
1071
+ size_t unpadded_size = (input_size/4) * 3;
1072
+ if ( unpadded_size > output_size )
1073
+ {
1074
+ error("Output buffer size %zu is too small to decode %zu into", output_size, input_size);
1075
+ return 0;
1076
+ }
1077
+ // Don't check padding within full quantums
1078
+ for (size_t i = 0 ; i < input_size-4 ; i+=4 )
1079
+ {
1080
+ uint32_t value = (lookup[input[0]] << 18) + (lookup[input[1]] << 12) + (lookup[input[2]] << 6) + lookup[input[3]];
1081
+ output[0] = value >> 16;
1082
+ output[1] = value >> 8;
1083
+ output[2] = value;
1084
+ //error("Decoded %c %c %c %c -> %02x %02x %02x", input[0], input[1], input[2], input[3], output[0], output[1], output[2]);
1085
+ output += 3;
1086
+ input += 4;
1087
+ }
1088
+ // Handle padding only in last quantum
1089
+ if (input[2] == '=') {
1090
+ uint32_t value = (lookup[input[0]] << 6) + lookup[input[1]];
1091
+ output[0] = value >> 4;
1092
+ //error("Decoded %c %c %c %c -> %02x", input[0], input[1], input[2], input[3], output[0]);
1093
+ return unpadded_size-2;
1094
+ }
1095
+ else if (input[3] == '=') {
1096
+ uint32_t value = (lookup[input[0]] << 12) + (lookup[input[1]] << 6) + lookup[input[2]];
1097
+ output[0] = value >> 10;
1098
+ output[1] = value >> 2;
1099
+ //error("Decoded %c %c %c %c -> %02x %02x", input[0], input[1], input[2], input[3], output[0], output[1]);
1100
+ return unpadded_size-1;
1101
+ }
1102
+ else
1103
+ {
1104
+ uint32_t value = (input[0] << 18) + (input[1] << 12) + (input[2]<<6) + input[3];
1105
+ output[0] = value >> 16;
1106
+ output[1] = value >> 8;
1107
+ output[2] = value;
1108
+ //error("Decoded %c %c %c %c -> %02x %02x %02x", input[0], input[1], input[2], input[3], output[0], output[1], output[2]);
1109
+ return unpadded_size;
1110
+ }
1111
+}
1112
+
1113
+size_t base64_encode(unsigned char *input, size_t input_size, char *output, size_t output_size)
1114
+{
1115
+ uint32_t value;
1116
+ static char lookup[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
1117
+ "abcdefghijklmnopqrstuvwxyz"
1118
+ "0123456789+/";
1119
+ if ((input_size/3+1)*4 >= output_size)
1120
+ {
1121
+ error("Output buffer for encoding size=%zu is not large enough for %zu-bytes input", output_size, input_size);
1122
+ return 0;
1123
+ }
1124
+ size_t count = 0;
1125
+ while (input_size>3)
1126
+ {
1127
+ value = ((input[0] << 16) + (input[1] << 8) + input[2]) & 0xffffff;
1128
+ output[0] = lookup[value >> 18];
1129
+ output[1] = lookup[(value >> 12) & 0x3f];
1130
+ output[2] = lookup[(value >> 6) & 0x3f];
1131
+ output[3] = lookup[value & 0x3f];
1132
+ //error("Base-64 encode (%04x) -> %c %c %c %c\n", value, output[0], output[1], output[2], output[3]);
1133
+ output += 4;
1134
+ input += 3;
1135
+ input_size -= 3;
1136
+ count += 4;
1137
+ }
1138
+ switch (input_size)
1139
+ {
1140
+ case 2:
1141
+ value = (input[0] << 10) + (input[1] << 2);
1142
+ output[0] = lookup[(value >> 12) & 0x3f];
1143
+ output[1] = lookup[(value >> 6) & 0x3f];
1144
+ output[2] = lookup[value & 0x3f];
1145
+ output[3] = '=';
1146
+ //error("Base-64 encode (%06x) -> %c %c %c %c\n", (value>>2)&0xffff, output[0], output[1], output[2], output[3]);
1147
+ count += 4;
1148
+ break;
1149
+ case 1:
1150
+ value = input[0] << 4;
1151
+ output[0] = lookup[(value >> 6) & 0x3f];
1152
+ output[1] = lookup[value & 0x3f];
1153
+ output[2] = '=';
1154
+ output[3] = '=';
1155
+ //error("Base-64 encode (%06x) -> %c %c %c %c\n", value, output[0], output[1], output[2], output[3]);
1156
+ count += 4;
1157
+ break;
1158
+ case 0:
1159
+ break;
1160
+ }
1161
+ return count;
1162
+}
1163
+
1164
+
1165
+
1166
+int private_decrypt(unsigned char * enc_data, int data_len, unsigned char *decrypted)
1167
+{
1168
+ int result = RSA_private_decrypt( data_len, enc_data, decrypted, aclk_private_key, RSA_PKCS1_OAEP_PADDING);
1169
+ if (result == -1) {
1170
+ char err[512];
1171
+ ERR_error_string_n(ERR_get_error(), err, sizeof(err));
1172
+ error("Decryption of the challenge failed: %s", err);
1173
+ }
1174
+ return result;
1175
+}
1176
+
1177
+char *extract_payload(BUFFER *b)
1178
+{
1179
+char *s = b->buffer;
1180
+unsigned int line_len=0;
1181
+ for (size_t i=0; i<b->len; i++)
1182
+ {
1183
+ if (*s == 0 )
1184
+ return NULL;
1185
+ if (*s == '\n' ) {
1186
+ if (line_len==0)
1187
+ return s+1;
1188
+ line_len = 0;
1189
+ }
1190
+ else if (*s == '\r') {
1191
+ /* don't count */
1192
+ }
1193
+ else
1194
+ line_len ++;
1195
+ s++;
1196
+ }
1197
+ return NULL;
1198
+}
1199
+
1200
+static int decode_base_url(char *url, char **aclk_hostname, char **aclk_port)
1201
+{
1202
+int pos = 0;
1203
+ if (!strncmp("https://", url, 8))
1204
+ {
1205
+ pos = 8;
1206
+ }
1207
+ else if (!strncmp("http://", url, 7))
1208
+ {
1209
+ error("Cannot connect ACLK over %s -> unencrypted link is not supported", url);
1210
+ return 1;
1211
+ }
1212
+int host_end = pos;
1213
+ while( url[host_end] != 0 && url[host_end] != '/' && url[host_end] != ':' )
1214
+ host_end++;
1215
+ if (url[host_end] == 0)
1216
+ {
1217
+ *aclk_hostname = strdupz(url+pos);
1218
+ *aclk_port = strdupz("443");
1219
+ info("Setting ACLK target host=%s port=%s from %s", *aclk_hostname, *aclk_port, url);
1220
+ return 0;
1221
+ }
1222
+ if (url[host_end] == ':')
1223
+ {
1224
+ *aclk_hostname = callocz(host_end - pos + 1, 1);
1225
+ strncpy(*aclk_hostname, url+pos, host_end - pos);
1226
+ int port_end = host_end + 1;
1227
+ while (url[port_end] >= '0' && url[port_end] <= '9')
1228
+ port_end++;
1229
+ if (port_end - host_end > 6)
1230
+ {
1231
+ error("Port specified in %s is invalid", url);
1232
+ return 0;
1233
+ }
1234
+ *aclk_port = callocz(port_end - host_end + 1, 1);
1235
+ for(int i=host_end + 1; i < port_end; i++)
1236
+ (*aclk_port)[i - host_end - 1] = url[i];
1237
+ }
1238
+ info("Setting ACLK target host=%s port=%s from %s", *aclk_hostname, *aclk_port, url);
1239
+ return 0;
1240
+}
1241
+
1242
+void aclk_get_challenge(char *aclk_hostname, char *aclk_port)
1243
+{
1244
+ debug(D_ACLK, "Performing challenge-response sequence");
1245
+ if (aclk_password != NULL)
1246
+ {
1247
+ freez(aclk_password);
1248
+ aclk_password = NULL;
1249
+ }
1250
+ // curl http://cloud-iam-agent-service:8080/api/v1/auth/node/00000000-0000-0000-0000-000000000000/challenge
1251
+ BUFFER *b = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1252
+ // TODO - target host?
1253
+ char *agent_id = is_agent_claimed();
1254
+ if (agent_id == NULL)
1255
+ {
1256
+ error("Agent was not claimed - cannot perform challenge/response");
1257
+ return;
1258
+ }
1259
+ char url[1024];
1260
+ sprintf(url, "/api/v1/auth/node/%s/challenge", agent_id);
1261
+ info("Retrieving challenge from cloud: %s %s %s", aclk_hostname, aclk_port, url);
1262
+ if(send_https_request("GET", aclk_hostname, aclk_port, url, b, NULL))
1263
+ {
1264
+ error("Challenge failed");
1265
+ return;
1266
+ }
1267
+ struct dictionary_singleton challenge = { .key = "challenge", .result = NULL };
1268
+ // Force null-termination?
1269
+ char *payload = NULL;
1270
+ payload = extract_payload(b);
1271
+ if (payload==NULL) {
1272
+ error("Could not find payload in http response #1 (the challenge):\n%s", b->buffer);
1273
+ return;
1274
+ }
1275
+ debug(D_ACLK, "Challenge response from cloud: %s", payload);
1276
+ if (json_parse(payload, &challenge, json_extract_singleton) != JSON_OK)
1277
+ {
1278
+ freez(challenge.result);
1279
+ error("Could not parse the json response with the challenge: %s", payload);
1280
+ return;
1281
+ }
1282
+ if (challenge.result == NULL ) {
1283
+ error("Could not retrieve challenge from auth response");
1284
+ return;
1285
+ }
1286
+
1287
+
1288
+
1289
+ size_t challenge_len = strlen(challenge.result);
1290
+ unsigned char decoded[512];
1291
+ size_t decoded_len = base64_decode((unsigned char*)challenge.result, challenge_len, decoded, sizeof(decoded));
1292
+
1293
+ unsigned char plaintext[4096]={};
1294
+ int decrypted_length = private_decrypt(decoded, decoded_len, plaintext);
1295
+ freez(challenge.result);
1296
+ char encoded[512];
1297
+ size_t encoded_len = base64_encode(plaintext, decrypted_length, encoded, sizeof(encoded));
1298
+ encoded[encoded_len] = 0;
1299
+ debug(D_ACLK, "Encoded len=%zu Decryption len=%d: '%s'", encoded_len, decrypted_length, encoded);
1300
+
1301
+ char response_json[4096]={};
1302
+ sprintf(response_json, "{\"response\":\"%s\"}", encoded);
1303
+ debug(D_ACLK, "Password phase: %s",response_json);
1304
+ // TODO - host
1305
+ sprintf(url, "/api/v1/auth/node/%s/password", agent_id);
1306
+ if(send_https_request("POST", aclk_hostname, aclk_port, url, b, response_json))
1307
+ {
1308
+ error("Challenge-response failed");
1309
+ return;
1310
+ }
1311
+ payload = extract_payload(b);
1312
+ if (payload==NULL) {
1313
+ error("Could not find payload in http response #2 (the password):\n%s", b->buffer);
1314
+ return;
1315
+ }
1316
+ debug(D_ACLK, "Password response from cloud: %s", payload);
1317
+
1318
+ struct dictionary_singleton password = { .key = "password", .result = NULL };
1319
+ if (json_parse(payload, &password, json_extract_singleton) != JSON_OK)
1320
+ {
1321
+ freez(password.result);
1322
+ error("Could not parse the json response with the password: %s", payload);
1323
+ return;
1324
+ }
1325
+
1326
+ if (password.result == NULL ) {
1327
+ error("Could not retrieve password from auth response");
1328
+ return;
1329
+ }
1330
+ if (aclk_password != NULL )
1331
+ freez(aclk_password);
1332
+ if (aclk_username == NULL)
1333
+ aclk_username = strdupz(agent_id);
1334
+ aclk_password = password.result;
1335
+
1336
+ buffer_free(b);
1337
+}
1338
+
1339
+static void aclk_try_to_connect(char *hostname, char *port, int port_num)
1340
+{
1341
+ info("Attempting to establish the agent cloud link");
1342
+ aclk_get_challenge(hostname, port);
1343
+ if (aclk_password == NULL)
1344
+ return;
1345
+ int rc;
1346
+ rc = mqtt_attempt_connection(hostname, port_num, aclk_username, aclk_password);
1347
+ if (unlikely(rc)) {
1348
+ error("Failed to initialize the agent cloud link library");
1349
+ }
1350
+ aclk_connecting = 1;
1351
+}
1352
+
1353
+
1354
/**
1355
* Main agent cloud link thread
1356
*
@@ -958,29 +1375,42 @@ void *aclk_main(void *ptr)
1375
last_init_sequence = now_realtime_sec();
1376
query_thread = NULL;
1377
961
- aclk_hostname = config_get(CONFIG_SECTION_ACLK, "agent cloud link hostname", ACLK_DEFAULT_HOST);
962
- aclk_port = config_get_number(CONFIG_SECTION_ACLK, "agent cloud link port", ACLK_DEFAULT_PORT);
963
-
1378
965
- // TODO: This may change when we have enough info from the claiming itself to avoid wasting 60 seconds
966
- // TODO: Handle the unclaim command as well -- we may need to shutdown the connection
967
- while(likely(!is_agent_claimed())) {
968
- sleep_usec(USEC_PER_SEC * 5);
969
- if(netdata_exit)
970
- goto exited;
1379
+ char *aclk_hostname = NULL; // Initializers are over-written but prevent gcc complaining about clobbering.
1380
+ char *aclk_port = NULL;
1381
+ uint32_t port_num = 0;
1382
+ char *cloud_base_url = config_get(CONFIG_SECTION_CLOUD, "cloud base url", "https://netdata.cloud");
1383
+ if( decode_base_url(cloud_base_url, &aclk_hostname, &aclk_port))
1384
+ {
1385
+ error("Configuration error - cannot use agent cloud link");
1386
+ return NULL;
1387
+ }
1388
+ port_num = atoi(aclk_port); // SSL library uses the string, MQTT uses the numeric value
1389
+
1390
+ info("Waiting for netdata to be claimed");
1391
+ while(1) {
1392
+ while (likely(!is_agent_claimed())) {
1393
+ sleep_usec(USEC_PER_SEC * 5);
1394
+ if (netdata_exit)
1395
+ goto exited;
1396
+ }
1397
+ if (!create_private_key() && !_mqtt_lib_init())
1398
+ break;
1399
+ sleep_usec(USEC_PER_SEC * 60);
1400
}
1401
create_publish_base_topic();
1402
+ create_private_key();
1403
1404
usec_t reconnect_expiry = 0; // In usecs
1405
1406
while (!netdata_exit) {
1407
static int first_init = 0;
978
- _link_event_loop(ACLK_LOOP_TIMEOUT * 1000);
979
- debug(D_ACLK,"LINK event loop called");
1408
981
- if (unlikely(!aclk_connection_initialized)) {
982
- if (unlikely(first_init)) {
983
- aclk_try_to_connect();
1409
+ info("loop state first_init_%d connected=%d connecting=%d", first_init, aclk_connected, aclk_connecting);
1410
+ sleep_usec(USEC_PER_MS * 500);
1411
+ if (unlikely(!aclk_connected)) {
1412
+ if (unlikely(!first_init)) {
1413
+ aclk_try_to_connect(aclk_hostname, aclk_port, port_num);
1414
first_init = 1;
1415
} else {
1416
if (aclk_connecting == 0) {
@@ -991,33 +1421,44 @@ void *aclk_main(void *ptr)
1421
}
1422
if (now_realtime_usec() >= reconnect_expiry) {
1423
reconnect_expiry = 0;
994
- aclk_connecting = 1;
995
- aclk_try_to_connect();
1424
+ aclk_try_to_connect(aclk_hostname, aclk_port, port_num);
1425
}
1426
sleep_usec(USEC_PER_MS * 100);
1427
}
1428
}
1429
+ if (aclk_connecting) {
1430
+ _link_event_loop();
1431
+ sleep_usec(USEC_PER_MS * 100);
1432
+ }
1433
continue;
1434
}
1435
1003
- if (likely(aclk_mqtt_connected)) {
1436
+ _link_event_loop();
1437
+ sleep_usec(USEC_PER_MS * 100);
1438
1005
- if (unlikely(!aclk_subscribed)) {
1006
- aclk_subscribed = !aclk_subscribe(ACLK_COMMAND_TOPIC, 2);
1007
- }
1439
+ // TODO: Move to on-connect
1440
+ if (unlikely(!aclk_subscribed)) {
1441
+ aclk_subscribed = !aclk_subscribe(ACLK_COMMAND_TOPIC, 2);
1442
+ }
1443
1009
- if (unlikely(!query_thread)) {
1010
- query_thread = callocz(1, sizeof(struct netdata_static_thread));
1011
- query_thread->thread = mallocz(sizeof(netdata_thread_t));
1012
- netdata_thread_create(
1013
- query_thread->thread, ACLK_THREAD_NAME, NETDATA_THREAD_OPTION_DEFAULT, aclk_query_main_thread,
1014
- query_thread);
1015
- }
1444
+ if (unlikely(!query_thread)) {
1445
+ query_thread = callocz(1, sizeof(struct netdata_static_thread));
1446
+ query_thread->thread = mallocz(sizeof(netdata_thread_t));
1447
+ netdata_thread_create(
1448
+ query_thread->thread, ACLK_THREAD_NAME, NETDATA_THREAD_OPTION_DEFAULT, aclk_query_main_thread,
1449
+ query_thread);
1450
}
1451
} // forever
1452
exited:
1453
aclk_shutdown();
1454
1455
+ freez(aclk_username);
1456
+ freez(aclk_password);
1457
+ freez(aclk_hostname);
1458
+ freez(aclk_port);
1459
+ if (aclk_private_key != NULL)
1460
+ RSA_free(aclk_private_key);
1461
+
1462
netdata_thread_cleanup_pop(1);
1463
return NULL;
1464
}
@@ -1037,8 +1478,8 @@ int aclk_send_message(char *sub_topic, char *message, char *msg_id)
1478
1479
UNUSED(msg_id);
1480
1040
- if (unlikely(aclk_wait_for_initialization()))
1041
- return 1;
1481
+ if(!aclk_connected)
1482
+ return 0;
1483
1484
if (unlikely(!message))
1485
return 0;
@@ -1056,7 +1497,6 @@ int aclk_send_message(char *sub_topic, char *message, char *msg_id)
1497
// TODO: link the msg_id with the mid so we can trace it
1498
ACLK_UNLOCK;
1499
1059
-
1500
if (unlikely(rc)) {
1501
errno = 0;
1502
error("Failed to send message, error code %d (%s)", rc, _link_strerror(rc));
@@ -1076,9 +1516,6 @@ int aclk_subscribe(char *sub_topic, int qos)
1516
char topic[ACLK_MAX_TOPIC + 1];
1517
char *final_topic;
1518
1079
- if (unlikely(aclk_wait_for_initialization()))
1080
- return 1;
1081
-
1519
final_topic = get_topic(sub_topic, topic, ACLK_MAX_TOPIC);
1520
if (unlikely(!final_topic)) {
1521
errno = 0;
@@ -1086,6 +1523,11 @@ int aclk_subscribe(char *sub_topic, int qos)
1523
return 1;
1524
}
1525
1526
+ if (!aclk_connected) {
1527
+ error("Cannot subscribe to %s - not connected!", topic);
1528
+ return 1;
1529
+ }
1530
+
1531
ACLK_LOCK;
1532
rc = _link_subscribe(final_topic, qos);
1533
ACLK_UNLOCK;
@@ -1099,13 +1541,11 @@ int aclk_subscribe(char *sub_topic, int qos)
1541
return rc;
1542
}
1543
1102
-
1544
// This is called from a callback when the link goes up
1104
-void aclk_connect(void *ptr)
1545
+void aclk_connect()
1546
{
1106
- UNUSED(ptr);
1107
- info("Connection detected");
1108
- aclk_connection_initialized = 1;
1547
+ info("Connection detected (%"PRIu64" queued queries)", aclk_queue.count);
1548
+ aclk_connected = 1;
1549
waiting_init = 0;
1550
aclk_reconnect_delay(0);
1551
QUERY_THREAD_WAKEUP;
@@ -1113,37 +1553,25 @@ void aclk_connect(void *ptr)
1553
}
1554
1555
// This is called from a callback when the link goes down
1116
-void aclk_disconnect(void *ptr)
1556
+void aclk_disconnect()
1557
{
1118
- UNUSED(ptr);
1119
-
1120
- if (likely(aclk_connection_initialized))
1121
- info("Disconnect detected");
1558
+ if (likely(aclk_connected))
1559
+ info("Disconnect detected (%"PRIu64" queued queries)", aclk_queue.count);
1560
aclk_subscribed = 0;
1123
- aclk_metadata_submitted = 0;
1561
+ aclk_metadata_submitted = ACLK_METADATA_REQUIRED;
1562
waiting_init = 1;
1125
- aclk_connection_initialized = 0;
1563
+ aclk_connected = 0;
1564
aclk_connecting = 0;
1565
}
1566
1567
void aclk_shutdown()
1568
{
1569
info("Shutdown initiated");
1132
- aclk_connection_initialized = 0;
1570
+ aclk_connected = 0;
1571
_link_shutdown();
1572
info("Shutdown complete");
1573
}
1574
1137
-void aclk_try_to_connect()
1138
-{
1139
- int rc;
1140
- rc = _link_lib_init(aclk_hostname, aclk_port, aclk_connect, aclk_disconnect);
1141
- if (unlikely(rc)) {
1142
- error("Failed to initialize the agent cloud link library");
1143
- }
1144
-}
1145
-
1146
-
1575
inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id)
1576
{
1577
uuid_t uuid;
@@ -1234,26 +1662,25 @@ void aclk_send_alarm_metadata()
1662
buffer_flush(local_buffer);
1663
local_buffer->contenttype = CT_APPLICATION_JSON;
1664
1237
- debug(D_ACLK,"Metadata alarms start");
1665
+ debug(D_ACLK, "Metadata alarms start");
1666
1667
aclk_create_header(local_buffer, "connect_alarms", msg_id);
1668
1241
- buffer_sprintf(local_buffer,"{\n\t \"configured-alarms\" : ");
1669
+ buffer_sprintf(local_buffer, "{\n\t \"configured-alarms\" : ");
1670
health_alarms2json(localhost, local_buffer, 1);
1243
- debug(D_ACLK,"Metadata %s with configured alarms has %ld bytes", msg_id, local_buffer->len);
1671
+ debug(D_ACLK, "Metadata %s with configured alarms has %zu bytes", msg_id, local_buffer->len);
1672
1245
- buffer_sprintf(local_buffer,",\n\t \"alarm-log\" : ");
1673
+ buffer_sprintf(local_buffer, ",\n\t \"alarm-log\" : ");
1674
health_alarm_log2json(localhost, local_buffer, 0);
1247
- debug(D_ACLK,"Metadata %s with alarm_log has %ld bytes", msg_id, local_buffer->len);
1675
+ debug(D_ACLK, "Metadata %s with alarm_log has %zu bytes", msg_id, local_buffer->len);
1676
1249
- buffer_sprintf(local_buffer,",\n\t \"alarms-active\" : ");
1677
+ buffer_sprintf(local_buffer, ",\n\t \"alarms-active\" : ");
1678
health_alarms_values2json(localhost, local_buffer, 0);
1251
- debug(D_ACLK,"Metadata %s with alarms_active has %ld bytes", msg_id, local_buffer->len);
1252
-
1679
+ debug(D_ACLK, "Metadata %s with alarms_active has %zu bytes", msg_id, local_buffer->len);
1680
1254
- buffer_sprintf(local_buffer,"\n}\n}");
1681
+ buffer_sprintf(local_buffer, "\n}\n}");
1682
aclk_send_message(ACLK_ALARMS_TOPIC, aclk_encode_response(local_buffer)->buffer, msg_id);
1256
- debug(D_ACLK,"Metadata %s encoded has %ld bytes", msg_id, local_buffer->len);
1683
+ debug(D_ACLK, "Metadata %s encoded has %zu bytes", msg_id, local_buffer->len);
1684
1685
freez(msg_id);
1686
buffer_free(local_buffer);
@@ -1263,24 +1690,24 @@ int aclk_send_info_metadata()
1690
{
1691
BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1692
1266
- debug(D_ACLK,"Metadata /info start");
1693
+ debug(D_ACLK, "Metadata /info start");
1694
1695
char *msg_id = create_uuid();
1696
buffer_flush(local_buffer);
1697
local_buffer->contenttype = CT_APPLICATION_JSON;
1698
1699
aclk_create_header(local_buffer, "connect", msg_id);
1273
- buffer_sprintf(local_buffer,"{\n\t \"info\" : ");
1700
+ buffer_sprintf(local_buffer, "{\n\t \"info\" : ");
1701
web_client_api_request_v1_info_fill_buffer(localhost, local_buffer);
1275
- debug(D_ACLK,"Metadata %s with info has %ld bytes", msg_id, local_buffer->len);
1702
+ debug(D_ACLK, "Metadata %s with info has %zu bytes", msg_id, local_buffer->len);
1703
1277
- buffer_sprintf(local_buffer,", \n\t \"charts\" : ");
1704
+ buffer_sprintf(local_buffer, ", \n\t \"charts\" : ");
1705
charts2json(localhost, local_buffer, 1);
1279
- buffer_sprintf(local_buffer,"\n}\n}");
1280
- debug(D_ACLK,"Metadata %s with chart has %ld bytes", msg_id, local_buffer->len);
1706
+ buffer_sprintf(local_buffer, "\n}\n}");
1707
+ debug(D_ACLK, "Metadata %s with chart has %zu bytes", msg_id, local_buffer->len);
1708
1709
aclk_send_message(ACLK_METADATA_TOPIC, aclk_encode_response(local_buffer)->buffer, msg_id);
1283
- debug(D_ACLK,"Metadata %s encoded has %ld bytes", msg_id, local_buffer->len);
1710
+ debug(D_ACLK, "Metadata %s encoded has %zu bytes", msg_id, local_buffer->len);
1711
freez(msg_id);
1712
1713
buffer_free(local_buffer);
@@ -1291,7 +1718,6 @@ int aclk_send_info_metadata()
1718
// or on request
1719
int aclk_send_metadata()
1720
{
1294
-
1721
aclk_send_info_metadata();
1722
aclk_send_alarm_metadata();
1723
@@ -1311,7 +1737,7 @@ void aclk_single_update_enable()
1737
// Trigged by a health reload, sends the alarm metadata
1738
void aclk_alarm_reload()
1739
{
1314
- if (unlikely(!agent_state))
1740
+ if (unlikely(agent_state != AGENT_STABLE))
1741
return;
1742
1743
aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT);
@@ -1341,7 +1767,7 @@ int aclk_send_single_chart(char *hostname, char *chart)
1767
1768
aclk_create_header(local_buffer, "chart", msg_id);
1769
rrdset2json(st, local_buffer, NULL, NULL, 1);
1344
- buffer_sprintf(local_buffer,"\t\n}");
1770
+ buffer_sprintf(local_buffer, "\t\n}");
1771
1772
aclk_send_message(ACLK_CHART_TOPIC, aclk_encode_response(local_buffer)->buffer, msg_id);
1773
@@ -1350,7 +1776,7 @@ int aclk_send_single_chart(char *hostname, char *chart)
1776
return 0;
1777
}
1778
1353
-int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd)
1779
+int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd)
1780
{
1781
#ifndef ENABLE_ACLK
1782
UNUSED(host);
@@ -1368,15 +1794,14 @@ int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd)
1794
#endif
1795
}
1796
1371
-
1372
-int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1797
+int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1798
{
1799
BUFFER *local_buffer = NULL;
1800
1801
if (host != localhost)
1802
return 0;
1803
1379
- if (agent_state == 0)
1804
+ if (agent_state != AGENT_STABLE)
1805
return 0;
1806
1807
/*
@@ -1399,8 +1824,8 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1824
health_alarm_entry2json_nolock(local_buffer, ae, host);
1825
netdata_rwlock_unlock(&host->health_log.alarm_log_rwlock);
1826
1402
- buffer_sprintf(local_buffer,"\n}");
1403
- aclk_queue_query(ACLK_ALARMS_TOPIC, NULL, msg_id, aclk_encode_response(local_buffer)->buffer , 0, 1, ACLK_CMD_ALARM);
1827
+ buffer_sprintf(local_buffer, "\n}");
1828
+ aclk_queue_query(ACLK_ALARMS_TOPIC, NULL, msg_id, aclk_encode_response(local_buffer)->buffer, 0, 1, ACLK_CMD_ALARM);
1829
1830
freez(msg_id);
1831
buffer_free(local_buffer);
@@ -1413,7 +1838,9 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1838
*/
1839
int aclk_handle_cloud_request(char *payload)
1840
{
1416
- struct aclk_request cloud_to_agent = { .type_id = NULL, .msg_id = NULL, .callback_topic = NULL, .payload = NULL, .version = 0};
1841
+ struct aclk_request cloud_to_agent = {
1842
+ .type_id = NULL, .msg_id = NULL, .callback_topic = NULL, .payload = NULL, .version = 0
1843
+ };
1844
1845
if (unlikely(!payload)) {
1846
debug(D_ACLK, "ACLK incoming message is empty");
@@ -1428,7 +1855,6 @@ int aclk_handle_cloud_request(char *payload)
1855
JSON_OK != rc || !cloud_to_agent.payload || !cloud_to_agent.callback_topic || !cloud_to_agent.msg_id ||
1856
!cloud_to_agent.type_id || cloud_to_agent.version > ACLK_VERSION ||
1857
strcmp(cloud_to_agent.type_id, "http"))) {
1431
-
1858
if (JSON_OK != rc)
1859
error("Malformed json request (%s)", payload);
1860
aclk/agent_cloud_link.h
+31
-37
@@ -8,11 +8,6 @@
8
9
#define ACLK_VERSION 1
10
#define ACLK_THREAD_NAME "ACLK_Query"
11
-#define ACLK_JSON_IN_MSGID "msg-id"
12
-#define ACLK_JSON_IN_TYPE "type"
13
-#define ACLK_JSON_IN_VERSION "version"
14
-#define ACLK_JSON_IN_TOPIC "callback-topic"
15
-#define ACLK_JSON_IN_URL "payload"
11
#define ACLK_CHART_TOPIC "chart"
12
#define ACLK_ALARMS_TOPIC "alarms"
13
#define ACLK_METADATA_TOPIC "meta"
@@ -25,26 +20,23 @@
20
#define ACLK_INITIALIZATION_SLEEP_WAIT 1 // Wait time @ spin lock for MQTT initialization in seconds
21
#define ACLK_QOS 1
22
#define ACLK_PING_INTERVAL 60
28
-#define ACLK_LOOP_TIMEOUT 5 // seconds to wait for operations in the library loop
23
+#define ACLK_LOOP_TIMEOUT 5 // seconds to wait for operations in the library loop
24
30
-#define ACLK_MAX_TOPIC 255
25
+#define ACLK_MAX_TOPIC 255
26
27
#define ACLK_RECONNECT_DELAY 1 // reconnect delay -- with backoff stragegy fow now
28
#define ACLK_STABLE_TIMEOUT 10 // Minimum delay to mark AGENT as stable
34
-#define ACLK_DEFAULT_PORT 9002
29
+#define ACLK_DEFAULT_PORT 9002
30
#define ACLK_DEFAULT_HOST "localhost"
31
37
-#define CONFIG_SECTION_ACLK "agent_cloud_link"
38
-
32
struct aclk_request {
40
- char *type_id;
41
- char *msg_id;
42
- char *callback_topic;
43
- char *payload;
44
- int version;
33
+ char *type_id;
34
+ char *msg_id;
35
+ char *callback_topic;
36
+ char *payload;
37
+ int version;
38
};
39
47
-
40
typedef enum aclk_cmd {
41
ACLK_CMD_CLOUD,
42
ACLK_CMD_ONCONNECT,
@@ -56,24 +48,29 @@ typedef enum aclk_cmd {
48
ACLK_CMD_MAX
49
} ACLK_CMD;
50
59
-typedef enum aclk_init_action {
60
- ACLK_INIT,
61
- ACLK_REINIT
62
-} ACLK_INIT_ACTION;
51
+typedef enum aclk_metadata_state {
52
+ ACLK_METADATA_REQUIRED,
53
+ ACLK_METADATA_CMD_QUEUED,
54
+ ACLK_METADATA_SENT
55
+} ACLK_METADATA_STATE;
56
+
57
+typedef enum agent_state {
58
+ AGENT_INITIALIZING,
59
+ AGENT_STABLE
60
+} AGENT_STATE;
61
62
+typedef enum aclk_init_action { ACLK_INIT, ACLK_REINIT } ACLK_INIT_ACTION;
63
64
void *aclk_main(void *ptr);
65
67
-#define NETDATA_ACLK_HOOK \
68
- { \
69
- .name = "ACLK_Main", \
70
- .config_section = NULL, \
71
- .config_name = NULL, \
72
- .enabled = 1, \
73
- .thread = NULL, \
74
- .init_routine = NULL, \
75
- .start_routine = aclk_main \
76
- },
66
+#define NETDATA_ACLK_HOOK \
67
+ { .name = "ACLK_Main", \
68
+ .config_section = NULL, \
69
+ .config_name = NULL, \
70
+ .enabled = 1, \
71
+ .thread = NULL, \
72
+ .init_routine = NULL, \
73
+ .start_routine = aclk_main },
74
75
extern int aclk_send_message(char *sub_topic, char *message, char *msg_id);
76
@@ -83,23 +80,21 @@ extern int aclk_send_message(char *sub_topic, char *message, char *msg_id);
80
extern char *is_agent_claimed(void);
81
char *create_uuid();
82
86
-
83
// callbacks for agent cloud link
84
int aclk_subscribe(char *topic, int qos);
85
void aclk_shutdown();
86
int cloud_to_agent_parse(JSON_ENTRY *e);
91
-void aclk_disconnect(void *conn);
92
-void aclk_connect(void *conn);
87
+void aclk_disconnect();
88
+void aclk_connect();
89
int aclk_send_metadata();
90
int aclk_send_info_metadata();
91
int aclk_wait_for_initialization();
92
char *create_publish_base_topic();
97
-void aclk_try_to_connect();
93
94
int aclk_send_single_chart(char *host, char *chart);
95
int aclk_queue_query(char *token, char *data, char *msg_type, char *query, int run_after, int internal, ACLK_CMD cmd);
101
-struct aclk_query *aclk_query_find(char *token, char *data, char *msg_id,
102
- char *query, ACLK_CMD cmd, struct aclk_query **last_query);
96
+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);
100
void aclk_create_header(BUFFER *dest, char *type, char *msg_id);
@@ -116,5 +111,4 @@ extern void health_alarm_entry2json_nolock(BUFFER *wb, ALARM_ENTRY *ae, RRDHOST
111
void aclk_single_update_enable();
112
void aclk_single_update_disable();
113
119
-
114
#endif //NETDATA_AGENT_CLOUD_LINK_H
aclk/mqtt.c
+42
-170
@@ -5,61 +5,13 @@
5
#include "mqtt.h"
6
#include "aclk_lws_wss_client.h"
7
8
-void (*_on_connect)(void *ptr) = NULL;
9
-void (*_on_disconnect)(void *ptr) = NULL;
10
-
11
-#ifndef ENABLE_ACLK
12
-
8
inline const char *_link_strerror(int rc)
9
{
15
- UNUSED(rc);
16
- return "no error";
17
-}
18
-
19
-int _link_event_loop(int timeout)
20
-{
21
- UNUSED(timeout);
22
- return 0;
23
-}
24
-
25
-int _link_send_message(char *topic, char *message, int *mid)
26
-{
27
- UNUSED(topic);
28
- UNUSED(message);
29
- UNUSED(mid);
30
- return 0;
31
-}
32
-
33
-int _link_subscribe(char *topic, int qos)
34
-{
35
- UNUSED(topic);
36
- UNUSED(qos);
37
- return 0;
38
-}
39
-
40
-void _link_shutdown()
41
-{
42
- return;
43
-}
44
-
45
-int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *))
46
-{
47
- UNUSED(aclk_hostname);
48
- UNUSED(aclk_port);
49
- UNUSED(on_connect);
50
- UNUSED(on_disconnect);
51
- return 0;
10
+ return mosquitto_strerror(rc);
11
}
12
54
-#else
13
+static struct mosquitto *mosq = NULL;
14
56
-struct mosquitto *mosq = NULL;
57
-
58
-// Get a string description of the error
59
-inline const char *_link_strerror(int rc)
60
-{
61
- return mosquitto_strerror(rc);
62
-}
15
16
void mqtt_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
17
{
@@ -69,12 +21,6 @@ void mqtt_message_callback(struct mosquitto *mosq, void *obj, const struct mosqu
21
aclk_handle_cloud_request(msg->payload);
22
}
23
72
-// This is not define because in future we might want to try plain
73
-// MQTT as fallback ?
74
-// e.g. try 1st MQTT-WSS, 2nd MQTT plain, 3rd https fallback...
75
-int mqtt_over_websockets = 1;
76
-struct aclk_lws_wss_engine_instance *lws_engine_instance = NULL;
77
-
24
void publish_callback(struct mosquitto *mosq, void *obj, int rc)
25
{
26
UNUSED(mosq);
@@ -87,29 +33,26 @@ void publish_callback(struct mosquitto *mosq, void *obj, int rc)
33
34
void connect_callback(struct mosquitto *mosq, void *obj, int rc)
35
{
36
+ UNUSED(mosq);
37
UNUSED(obj);
38
UNUSED(rc);
39
40
info("Connection to cloud estabilished");
94
-
95
- aclk_mqtt_connected = 1;
96
- _on_connect((void *)mosq);
41
+ aclk_connect();
42
43
return;
44
}
45
46
void disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
47
{
48
+ UNUSED(mosq);
49
UNUSED(obj);
50
UNUSED(rc);
51
52
info("Connection to cloud failed");
53
+ aclk_disconnect();
54
108
- aclk_mqtt_connected = 0;
109
- _on_disconnect((void *)mosq);
110
-
111
- if (mqtt_over_websockets && lws_engine_instance)
112
- aclk_lws_wss_mqtt_layer_disconect_notif(lws_engine_instance);
55
+ aclk_lws_wss_mqtt_layer_disconect_notif();
56
57
return;
58
}
@@ -126,24 +69,25 @@ void _show_mqtt_info()
69
70
size_t _mqtt_external_write_hook(void *buf, size_t count)
71
{
129
- return aclk_lws_wss_client_write(lws_engine_instance, buf, count);
72
+ return aclk_lws_wss_client_write(buf, count);
73
}
74
75
size_t _mqtt_external_read_hook(void *buf, size_t count)
76
{
134
- return aclk_lws_wss_client_read(lws_engine_instance, buf, count);
77
+ return aclk_lws_wss_client_read(buf, count);
78
}
79
137
-int _mqtt_lib_init(void (*on_connect)(void *), void (*on_disconnect)(void *))
80
+int _mqtt_lib_init()
81
{
82
int rc;
140
- int libmosq_major, libmosq_minor, libmosq_revision, libmosq_version;
83
+ //int libmosq_major, libmosq_minor, libmosq_revision, libmosq_version;
84
+ /* Commenting out now as it is unused - do not delete, this is needed for the on-prem version.
85
char *ca_crt;
86
char *server_crt;
87
char *server_key;
88
89
// show library info so can have it in the logfile
146
- libmosq_version = mosquitto_lib_version(&libmosq_major, &libmosq_minor, &libmosq_revision);
90
+ //libmosq_version = mosquitto_lib_version(&libmosq_major, &libmosq_minor, &libmosq_revision);
91
ca_crt = config_get(CONFIG_SECTION_ACLK, "agent cloud link cert", "*");
92
server_crt = config_get(CONFIG_SECTION_ACLK, "agent cloud link server cert", "*");
93
server_key = config_get(CONFIG_SECTION_ACLK, "agent cloud link server key", "*");
@@ -162,6 +106,7 @@ int _mqtt_lib_init(void (*on_connect)(void *), void (*on_disconnect)(void *))
106
freez(server_key);
107
server_key = NULL;
108
}
109
+ */
110
111
// info(
112
// "Detected libmosquitto library version %d, %d.%d.%d", libmosq_version, libmosq_major, libmosq_minor,
@@ -172,24 +117,28 @@ int _mqtt_lib_init(void (*on_connect)(void *), void (*on_disconnect)(void *))
117
error("Failed to initialize MQTT (libmosquitto library)");
118
return 1;
119
}
120
+ return 0;
121
+}
122
176
- mosq = mosquitto_new("anon", true, NULL);
123
+static int _mqtt_create_connection(char *username, char *password)
124
+{
125
+ if (mosq != NULL)
126
+ mosquitto_destroy(mosq);
127
+ mosq = mosquitto_new(username, true, NULL);
128
if (unlikely(!mosq)) {
129
mosquitto_lib_cleanup();
130
error("MQTT new structure -- %s", mosquitto_strerror(errno));
180
- return 1;
131
+ return MOSQ_ERR_UNKNOWN;
132
}
133
183
- _on_connect = on_connect;
184
- _on_disconnect = on_disconnect;
185
-
134
mosquitto_connect_callback_set(mosq, connect_callback);
135
mosquitto_disconnect_callback_set(mosq, disconnect_callback);
136
mosquitto_publish_callback_set(mosq, publish_callback);
137
190
- mosquitto_username_pw_set(mosq, NULL, NULL);
138
+ info("Using challenge-response: %s / %s", username, password);
139
+ mosquitto_username_pw_set(mosq, username, password);
140
192
- rc = mosquitto_threaded_set(mosq, 1);
141
+ int rc = mosquitto_threaded_set(mosq, 1);
142
if (unlikely(rc != MOSQ_ERR_SUCCESS))
143
error("Failed to tune the thread model for libmoquitto (%s)", mosquitto_strerror(rc));
144
@@ -202,19 +151,10 @@ int _mqtt_lib_init(void (*on_connect)(void *), void (*on_disconnect)(void *))
151
info("MQTT in flight messages set to 1 -- %s", mosquitto_strerror(rc));
152
#endif
153
205
- if (!mqtt_over_websockets) {
206
- rc = mosquitto_reconnect_delay_set(mosq, ACLK_RECONNECT_DELAY, ACLK_MAX_BACKOFF_DELAY, 1);
207
-
208
- if (unlikely(rc != MOSQ_ERR_SUCCESS))
209
- error("Failed to set the reconnect delay (%d) (%s)", rc, mosquitto_strerror(rc));
210
-
211
- mosquitto_tls_set(mosq, ca_crt, NULL, server_crt, server_key, NULL);
212
- }
213
-
154
return rc;
155
}
156
217
-int _link_mqtt_connect(char *aclk_hostname, int aclk_port)
157
+static int _link_mqtt_connect(char *aclk_hostname, int aclk_port)
158
{
159
int rc;
160
@@ -234,9 +174,6 @@ static inline void _link_mosquitto_write()
174
{
175
int rc;
176
237
- if (!mqtt_over_websockets)
238
- return;
239
-
177
rc = mosquitto_loop_misc(mosq);
178
if (unlikely(rc != MOSQ_ERR_SUCCESS))
179
debug(D_ACLK, "ACLK: failure during mosquitto_loop_misc %s", mosquitto_strerror(rc));
@@ -248,15 +185,13 @@ static inline void _link_mosquitto_write()
185
}
186
}
187
251
-void aclk_lws_connect_notif_callback()
188
+void aclk_lws_connection_established(char *hostname, int port)
189
{
253
- //the connection is done by LWS so this parameters dont matter
254
- //ig MQTT over LWS is used
255
- _link_mqtt_connect(aclk_hostname, aclk_port);
190
+ _link_mqtt_connect(hostname, port); // Parameters only used for logging, lower layer connected.
191
_link_mosquitto_write();
192
}
193
259
-void aclk_lws_data_received_callback()
194
+void aclk_lws_connection_data_received()
195
{
196
int rc = mosquitto_loop_read(mosq, 1);
197
if (rc != MOSQ_ERR_SUCCESS)
@@ -268,87 +203,33 @@ void aclk_lws_connection_closed()
203
aclk_disconnect(NULL);
204
}
205
271
-static const struct aclk_lws_wss_engine_callbacks aclk_lws_engine_callbacks = {
272
- .connection_established_callback = aclk_lws_connect_notif_callback,
273
- .data_rcvd_callback = aclk_lws_data_received_callback,
274
- .data_writable_callback = NULL,
275
- .connection_closed = aclk_lws_connection_closed
276
-};
206
278
-int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *))
207
+int mqtt_attempt_connection(char *aclk_hostname, int aclk_port, char *username, char *password)
208
{
280
- int rc;
209
+ if(aclk_lws_wss_connect(aclk_hostname, aclk_port))
210
+ return MOSQ_ERR_UNKNOWN;
211
+ aclk_lws_wss_service_loop();
212
282
- if (mqtt_over_websockets) {
283
- // we will connect when WebSocket connection is up
284
- // based on callback
285
- if (!lws_engine_instance)
286
- lws_engine_instance = aclk_lws_wss_client_init(&aclk_lws_engine_callbacks, aclk_hostname, aclk_port);
287
- else
288
- aclk_lws_wss_connect(lws_engine_instance);
289
-
290
- aclk_lws_wss_service_loop(lws_engine_instance);
291
- }
292
-
293
- rc = _mqtt_lib_init(on_connect, on_disconnect);
294
- if (rc != MOSQ_ERR_SUCCESS)
213
+ int rc = _mqtt_create_connection(username, password);
214
+ if (rc!= MOSQ_ERR_SUCCESS)
215
return rc;
216
297
- if (mqtt_over_websockets) {
298
- mosquitto_external_callbacks_set(mosq, _mqtt_external_write_hook, _mqtt_external_read_hook);
299
- if (!lws_engine_instance)
300
- return 1;
301
- else
302
- return MOSQ_ERR_SUCCESS;
303
- } else {
304
- // if direct mqtt connection is used
305
- // connect immediatelly
306
- return _link_mqtt_connect(aclk_hostname, aclk_port);
307
- }
217
+ mosquitto_external_callbacks_set(mosq, _mqtt_external_write_hook, _mqtt_external_read_hook);
218
+ return rc;
219
}
220
310
-static inline int _link_event_loop_wss()
221
+inline int _link_event_loop()
222
{
312
- if (unlikely(!lws_engine_instance)) {
313
- return MOSQ_ERR_SUCCESS;
314
- }
223
316
- if (lws_engine_instance && lws_engine_instance->websocket_connection_up)
317
- _link_mosquitto_write();
224
+ // TODO: Check if we need to flush undelivered messages from libmosquitto on new connection attempts (QoS=1).
225
+ _link_mosquitto_write();
226
+ aclk_lws_wss_service_loop();
227
319
- aclk_lws_wss_service_loop(lws_engine_instance);
228
// this is because if use LWS we don't want
229
// mqtt to reconnect by itself
230
return MOSQ_ERR_SUCCESS;
231
}
232
325
-static inline int _link_event_loop_plain_mqtt(int timeout)
326
-{
327
- int rc;
328
-
329
- rc = mosquitto_loop(mosq, timeout, 1);
330
-
331
- if (unlikely(rc != MOSQ_ERR_SUCCESS)) {
332
- errno = 0;
333
- error("Loop error code %d (%s)", rc, mosquitto_strerror(rc));
334
- rc = mosquitto_reconnect(mosq);
335
- if (unlikely(rc != MOSQ_ERR_SUCCESS)) {
336
- error("Reconnect loop error code %d (%s)", rc, mosquitto_strerror(rc));
337
- }
338
- // TBD: Using delay
339
- sleep_usec(USEC_PER_SEC * 10);
340
- }
341
- return rc;
342
-}
343
-
344
-int _link_event_loop(int timeout)
345
-{
346
- if (mqtt_over_websockets)
347
- return _link_event_loop_wss();
348
-
349
- return _link_event_loop_plain_mqtt(timeout);
350
-}
351
-
233
void _link_shutdown()
234
{
235
int rc;
@@ -366,12 +247,7 @@ void _link_shutdown()
247
mosquitto_destroy(mosq);
248
mosq = NULL;
249
369
- if (lws_engine_instance) {
370
- aclk_lws_wss_client_destroy(lws_engine_instance);
371
- lws_engine_instance = NULL;
372
- }
373
-
374
- return;
250
+ aclk_lws_wss_client_destroy();
251
}
252
253
int _link_subscribe(char *topic, int qos)
@@ -391,7 +267,6 @@ int _link_subscribe(char *topic, int qos)
267
}
268
269
_link_mosquitto_write();
394
-
270
return 0;
271
}
272
@@ -418,9 +293,6 @@ int _link_send_message(char *topic, char *message, int *mid)
293
errno = 0;
294
error("MQTT message failed : %s", mosquitto_strerror(rc));
295
}
421
-
296
_link_mosquitto_write();
423
-
297
return rc;
298
}
426
-#endif
\ No newline at end of file
aclk/mqtt.h
+4
-7
@@ -8,18 +8,15 @@
8
#endif
9
10
void _show_mqtt_info();
11
-int _link_event_loop(int timeout);
11
+int _link_event_loop();
12
void _link_shutdown();
13
-int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *));
13
+int mqtt_attempt_connection(char *aclk_hostname, int aclk_port, char *username, char *password);
14
+//int _link_lib_init();
15
+int _mqtt_lib_init();
16
int _link_subscribe(char *topic, int qos);
17
int _link_send_message(char *topic, char *message, int *mid);
18
const char *_link_strerror(int rc);
19
20
int aclk_handle_cloud_request(char *);
21
20
-extern int aclk_connection_initialized;
21
-extern int aclk_mqtt_connected;
22
-extern char *aclk_hostname;
23
-extern int aclk_port;
24
-
22
#endif //NETDATA_MQTT_H
build_external/clean-install-arch-extras.Dockerfile
+1
-1
@@ -19,7 +19,7 @@ RUN pacman --noconfirm --needed -S autoconf \
19
pkgconfig \
20
python \
21
libvirt \
22
- libwebsockets \
22
+ cmake \
23
valgrind
24
25
ARG ACLK=no
build_external/clean-install-arch.Dockerfile
+2
-2
@@ -19,7 +19,7 @@ RUN pacman --noconfirm --needed -S autoconf \
19
pkgconfig \
20
python \
21
libvirt \
22
- libwebsockets
22
+ cmake
23
24
ARG ACLK=no
25
ARG EXTRA_CFLAGS
@@ -51,4 +51,4 @@ RUN ln -sf /dev/stdout /var/log/netdata/access.log
51
RUN ln -sf /dev/stdout /var/log/netdata/debug.log
52
RUN ln -sf /dev/stderr /var/log/netdata/error.log
53
54
-CMD ["/usr/sbin/netdata", "-D"]
\ No newline at end of file
54
+CMD ["/usr/sbin/netdata", "-D"]
build_external/projects/aclk-testing/agent-valgrind-compose.yml
+2
-2
@@ -7,13 +7,13 @@ services:
7
args:
8
- DISTRO=arch
9
- VERSION=extras
10
- image: arch_current_dev:latest
10
+ image: arch_extras_dev:latest
11
command: >
12
sh -c "echo -n 00000000-0000-0000-0000-000000000000 >/etc/netdata/claim.d/claimed_id &&
13
echo '[agent_cloud_link]' >>/etc/netdata/netdata.conf &&
14
echo ' agent cloud link hostname = vernemq' >>/etc/netdata/netdata.conf &&
15
echo ' agent cloud link port = 9002' >>/etc/netdata/netdata.conf &&
16
- /usr/sbin/valgrind --leak-check=full /usr/sbin/netdata -D"
16
+ /usr/sbin/valgrind --leak-check=full /usr/sbin/netdata -D -W debug_flags=0x200000000"
17
ports:
18
- 20000:19999
19
claim/claim.c
+2
@@ -36,8 +36,10 @@ extern struct registry registry;
36
/* rrd_init() must have been called before this function */
37
void claim_agent(char *claiming_arguments)
38
{
39
+#ifndef ENABLE_ACLK
40
info("The claiming feature is under development and still subject to change before the next release");
41
return;
42
+#endif
43
44
int exit_code;
45
pid_t command_pid;
libnetdata/config/appconfig.h
+1
@@ -93,6 +93,7 @@
93
#define CONFIG_SECTION_STREAM "stream"
94
#define CONFIG_SECTION_EXPORTING "exporting:global"
95
#define CONFIG_SECTION_HOST_LABEL "host labels"
96
+#define CONFIG_SECTION_ACLK "agent_cloud_link"
97
#define EXPORTING_CONF "exporting.conf"
98
99
// these are used to limit the configuration names and values lengths
libnetdata/socket/security.c
+1
-1
@@ -84,7 +84,7 @@ void security_openssl_common_options(SSL_CTX *ctx) {
84
*
85
* @return It returns the context on success or NULL otherwise
86
*/
87
-static SSL_CTX * security_initialize_openssl_client() {
87
+SSL_CTX * security_initialize_openssl_client() {
88
SSL_CTX *ctx;
89
#if OPENSSL_VERSION_NUMBER < 0x10100000L
90
ctx = SSL_CTX_new(SSLv23_client_method());
libnetdata/socket/security.h
+1
@@ -41,6 +41,7 @@ void security_clean_openssl();
41
void security_start_ssl(int selector);
42
int security_process_accept(SSL *ssl,int msg);
43
int security_test_certificate(SSL *ssl);
44
+SSL_CTX * security_initialize_openssl_client();
45
46
# endif //ENABLE_HTTPS
47
#endif //NETDATA_SECURITY_H
libnetdata/socket/socket.c
+1
-1
@@ -607,7 +607,7 @@ static inline int connect_to_unix(const char *path, struct timeval *timeout) {
607
// service the service name or port to connect to
608
// timeout the timeout for establishing a connection
609
610
-static inline int connect_to_this_ip46(int protocol, int socktype, const char *host, uint32_t scope_id, const char *service, struct timeval *timeout) {
610
+int connect_to_this_ip46(int protocol, int socktype, const char *host, uint32_t scope_id, const char *service, struct timeval *timeout) {
611
struct addrinfo hints;
612
struct addrinfo *ai_head = NULL, *ai = NULL;
613
libnetdata/socket/socket.h
+1
@@ -56,6 +56,7 @@ extern void listen_sockets_close(LISTEN_SOCKETS *sockets);
56
57
extern int connect_to_this(const char *definition, int default_port, struct timeval *timeout);
58
extern int connect_to_one_of(const char *destination, int default_port, struct timeval *timeout, size_t *reconnects_counter, char *connected_to, size_t connected_to_size);
59
+int connect_to_this_ip46(int protocol, int socktype, const char *host, uint32_t scope_id, const char *service, struct timeval *timeout);
60
61
#ifdef ENABLE_HTTPS
62
extern ssize_t recv_timeout(struct netdata_ssl *ssl,int sockfd, void *buf, size_t len, int flags, int timeout);