Merging the feature branch for the ACLK in the previous sprint. (#8179)
* ACLK connection and protocol improvements (#8139) * Adding ACLK retry on connection failure (#8147) * Fixed reconnect issues on the ACLK. (#8163) * Cleaning up ACLK - part 1 (#8167) Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>
Andrew Moss committed
Feb 24, 2020 at 12:10 UTC
c6d945200f201b05c2b019fa862cdf080a39a9d4
25 files changed
+1286
-679
aclk/aclk_lws_wss_client.c
+98
-53
@@ -161,8 +161,13 @@ void aclk_lws_wss_client_destroy(struct aclk_lws_wss_engine_instance* inst) {
161
#endif
162
}
163
164
-void _aclk_wss_connect(struct aclk_lws_wss_engine_instance *inst){
165
- struct lws_client_connect_info i;
164
+void aclk_lws_wss_connect(struct aclk_lws_wss_engine_instance *inst){
165
+ struct lws_client_connect_info i;
166
+
167
+ if(inst->lws_wsi) {
168
+ error("Already Connected. Only one connection supported at a time.");
169
+ return;
170
+ }
171
172
memset(&i, 0, sizeof(i));
173
i.context = inst->lws_context;
@@ -186,7 +191,37 @@ static inline int received_data_to_ringbuff(struct lws_ring *buffer, void* data,
191
}
192
return 1;
193
}
189
-
194
+
195
+static const char *aclk_lws_callback_name(enum lws_callback_reasons reason)
196
+{
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
+ }
224
+}
225
static int
226
aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason,
227
void *user, void *in, size_t len)
@@ -201,6 +236,7 @@ aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason,
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.";
@@ -209,74 +245,83 @@ aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason,
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) {
213
- case LWS_CALLBACK_CLIENT_WRITEABLE:
214
- aclk_lws_mutex_lock(&inst->write_buf_mutex);
215
- data = lws_wss_packet_buffer_pop(&inst->write_buffer_head);
216
- if(likely(data)) {
217
- lws_write(wsi, data->data + LWS_PRE, data->data_size, LWS_WRITE_BINARY);
218
- lws_wss_packet_buffer_free(data);
219
- if(inst->write_buffer_head)
220
- lws_callback_on_writable(inst->lws_wsi);
221
- }
222
- aclk_lws_mutex_unlock(&inst->write_buf_mutex);
223
- break;
224
- case LWS_CALLBACK_CLIENT_RECEIVE:
225
- aclk_lws_mutex_lock(&inst->read_buf_mutex);
226
- if(!received_data_to_ringbuff(inst->read_ringbuffer, in, len))
227
- retval = 1;
228
- aclk_lws_mutex_unlock(&inst->read_buf_mutex);
229
-
230
- if(likely(inst->callbacks.data_rcvd_callback))
231
- // to future myself -> do not call this while read lock is active as it will eventually
232
- // want to acquire same lock later in aclk_lws_wss_client_read() function
233
- inst->callbacks.data_rcvd_callback();
234
- else
235
- inst->data_to_read = 1; //to inform logic above there is reason to call mosquitto_loop_read
236
- break;
293
case LWS_CALLBACK_PROTOCOL_INIT:
238
- //initial connection here
239
- //later we will reconnect with delay od ACLK_LWS_WSS_RECONNECT_TIMEOUT
240
- //in case this connection fails or drops
241
- _aclk_wss_connect(inst);
242
- break;
243
- case LWS_CALLBACK_SERVER_NEW_CLIENT_INSTANTIATED:
244
- //TODO if already active make some error noise
245
- //currently we expect only one connection per netdata
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;
248
-#ifdef AUTO_RECONNECT_ON_LWS_LAYER
249
- case LWS_CALLBACK_USER:
250
- inst->reconnect_timeout_running = 0;
251
- _aclk_wss_connect(inst);
252
- break;
253
-#endif
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"));
256
- /* FALLTHRU */
303
+ // Fall-through
304
case LWS_CALLBACK_CLIENT_CLOSED:
305
case LWS_CALLBACK_WS_PEER_INITIATED_CLOSE:
259
-#ifdef AUTO_RECONNECT_ON_LWS_LAYER
260
- if(!inst->reconnect_timeout_running) {
261
- lws_timed_callback_vh_protocol(lws_get_vhost(wsi),
262
- lws_get_protocol(wsi),
263
- LWS_CALLBACK_USER, ACLK_LWS_WSS_RECONNECT_TIMEOUT);
264
- inst->reconnect_timeout_running = 1;
265
- }
266
- /* FALLTHRU */
267
-#endif
268
- //no break here on purpose we want to continue with LWS_CALLBACK_WSI_DESTROY
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;
273
- break;
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!
aclk/aclk_lws_wss_client.h
+4
-8
@@ -30,6 +30,7 @@ struct aclk_lws_wss_engine_callbacks {
30
void (*connection_established_callback)();
31
void (*data_rcvd_callback)();
32
void (*data_writable_callback)();
33
+ void (*connection_closed)();
34
};
35
36
struct lws_wss_packet_buffer;
@@ -57,14 +58,7 @@ struct aclk_lws_wss_engine_instance {
58
int websocket_connection_up;
59
60
// currently this is by default disabled
60
-// as decision has been made that reconnection
61
-// will have to be done from top layer
62
-// (after getting the new MQTT auth data)
63
-// for now i keep it here as it is usefull for
64
-// some of my internall testing
65
-#ifdef AUTO_RECONNECT_ON_LWS_LAYER
66
- int reconnect_timeout_running;
67
-#endif
61
+
62
int data_to_read;
63
int upstream_reconnect_request;
64
};
@@ -72,6 +66,8 @@ struct aclk_lws_wss_engine_instance {
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);
68
69
+void aclk_lws_wss_connect(struct aclk_lws_wss_engine_instance *inst);
70
+
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);
aclk/agent_cloud_link.c
+820
-362
@@ -5,19 +5,30 @@
5
6
// Read from the config file -- new section [agent_cloud_link]
7
// Defaults are supplied
8
-int aclk_recv_maximum = 0; // default 20
9
-int aclk_send_maximum = 0; // default 20
8
11
-int aclk_port = 0; // default 1883
12
-char *aclk_hostname = NULL; //default localhost
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;
13
14
int aclk_metadata_submitted = 0;
15
+int agent_state = 0;
16
+time_t last_init_sequence = 0;
17
int waiting_init = 1;
17
-int cmdpause = 0; // Used to pause query processing
18
19
-BUFFER *aclk_buffer = NULL;
19
char *global_base_topic = NULL;
20
+int aclk_connecting = 0;
21
+
22
+char *create_uuid()
23
+{
24
+ uuid_t uuid;
25
+ char *uuid_str = mallocz(36 + 1);
26
+
27
+ uuid_generate(uuid);
28
+ uuid_unparse(uuid, uuid_str);
29
+
30
+ return uuid_str;
31
+}
32
33
int cloud_to_agent_parse(JSON_ENTRY *e)
34
{
@@ -25,11 +36,8 @@ int cloud_to_agent_parse(JSON_ENTRY *e)
36
37
switch(e->type) {
38
case JSON_OBJECT:
28
- e->callback_function = cloud_to_agent_parse;
29
- break;
39
case JSON_ARRAY:
31
- e->callback_function = cloud_to_agent_parse;
32
- break;
40
+ break;
41
case JSON_STRING:
42
if (!strcmp(e->name, ACLK_JSON_IN_MSGID)) {
43
data->msg_id = strdupz(e->data.string);
@@ -40,17 +48,17 @@ int cloud_to_agent_parse(JSON_ENTRY *e)
48
break;
49
}
50
if (!strcmp(e->name, ACLK_JSON_IN_TOPIC)) {
43
- data->topic = strdupz(e->data.string);
51
+ data->callback_topic = strdupz(e->data.string);
52
break;
53
}
54
if (!strcmp(e->name, ACLK_JSON_IN_URL)) {
47
- data->url = strdupz(e->data.string);
55
+ data->payload = strdupz(e->data.string);
56
break;
57
}
58
break;
59
case JSON_NUMBER:
60
if (!strcmp(e->name, ACLK_JSON_IN_VERSION)) {
53
- data->version = atol(e->data.string);
61
+ data->version = atoi(e->original_string);
62
break;
63
}
64
break;
@@ -64,33 +72,6 @@ int cloud_to_agent_parse(JSON_ENTRY *e)
72
return 0;
73
}
74
67
-//char *send_http_request(char *host, char *port, char *url, BUFFER *b)
68
-//{
69
-// struct timeval timeout = { .tv_sec = 30, .tv_usec = 0 };
70
-//
71
-// buffer_flush(b);
72
-// buffer_sprintf(
73
-// b,
74
-// "GET %s HTTP/1.1\r\nHost: %s\r\nAccept: plain/text\r\nAccept-Language: en-us\r\nUser-Agent: Netdata/rocks\r\n\r\n",
75
-// url, host);
76
-// int sock = connect_to_this_ip46(IPPROTO_TCP, SOCK_STREAM, host, 0, "443", &timeout);
77
-//
78
-// if (unlikely(sock == -1)) {
79
-// error("Handshake failed");
80
-// return NULL;
81
-// }
82
-//
83
-// SSL_CTX *ctx = security_initialize_openssl_client();
84
-// // Certificate chain: not updating the stores - do we need private CA roots?
85
-// // Calls to SSL_CTX_load_verify_locations would go here.
86
-// SSL *ssl = SSL_new(ctx);
87
-// SSL_set_fd(ssl, sock);
88
-// int err = SSL_connect(ssl);
89
-// SSL_write(ssl, b->buffer, b->len); // Timeout options?
90
-// int bytes_read = SSL_read(ssl, b->buffer, b->len);
91
-// SSL_shutdown(ssl);
92
-// close(sock);
93
-//}
75
76
// Set when we have connection up and running from the connection callback
77
int aclk_connection_initialized = 0;
@@ -101,10 +82,14 @@ 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
@@ -116,14 +101,35 @@ pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
101
#define QUERY_THREAD_WAKEUP pthread_cond_signal(&query_cond_wait)
102
103
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
+};
121
+
122
+struct _collector *collector_list = NULL;
123
+
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
126
- u_char deleted; // Mark deleted for garbage collect
132
+ u_char deleted; // Mark deleted for garbage collect
133
struct aclk_query *next;
134
};
135
@@ -133,6 +139,41 @@ struct aclk_query_queue {
139
u_int64_t count;
140
} aclk_queue = { .aclk_query_head = NULL, .aclk_query_tail = NULL, .count = 0 };
141
142
+/*
143
+ * After a connection failure -- delay in milliseconds
144
+ * When a connection is established, the delay function
145
+ * should be called with
146
+ *
147
+ * mode 0 to reset the delay
148
+ * mode 1 to sleep for the calculated amount of time [0 .. ACLK_MAX_BACKOFF_DELAY * 1000] ms
149
+ *
150
+ */
151
+unsigned long int aclk_reconnect_delay(int mode)
152
+{
153
+ static int fail = -1;
154
+ unsigned long int delay;
155
+
156
+ if (!mode || fail == -1) {
157
+ srandom(time(NULL));
158
+ fail = mode-1;
159
+ return 0;
160
+ }
161
+
162
+ delay = (1 << fail);
163
+
164
+ if (delay >= ACLK_MAX_BACKOFF_DELAY) {
165
+ delay = ACLK_MAX_BACKOFF_DELAY * 1000;
166
+ }
167
+ else {
168
+ fail++;
169
+ delay = (delay * 1000) + (random() % 1000);
170
+ }
171
+
172
+// sleep_usec(USEC_PER_MS * delay);
173
+
174
+ return delay;
175
+}
176
+
177
/*
178
* Free a query structure when done
179
*/
@@ -143,23 +184,29 @@ void aclk_query_free(struct aclk_query *this_query)
184
return;
185
186
freez(this_query->topic);
146
- freez(this_query->query);
147
- if (this_query->data)
187
+ if (likely(this_query->query))
188
+ freez(this_query->query);
189
+ if (likely(this_query->data))
190
freez(this_query->data);
149
- if (this_query->msg_id)
191
+ if (likely(this_query->msg_id))
192
freez(this_query->msg_id);
193
freez(this_query);
152
- return;
194
}
195
196
// Returns the entry after which we need to create a new entry to run at the specified time
197
// If NULL is returned we need to add to HEAD
157
-// Called with locked entries
198
+// Need to have a QUERY lock before calling this
199
200
struct aclk_query *aclk_query_find_position(time_t time_to_run)
201
{
202
struct aclk_query *tmp_query, *last_query;
203
204
+ // Quick check if we will add to the end
205
+ if (likely(aclk_queue.aclk_query_tail)) {
206
+ if (aclk_queue.aclk_query_tail->run_after <= time_to_run)
207
+ return aclk_queue.aclk_query_tail;
208
+ }
209
+
210
last_query = NULL;
211
tmp_query = aclk_queue.aclk_query_head;
212
@@ -172,21 +219,27 @@ struct aclk_query *aclk_query_find_position(time_t time_to_run)
219
return last_query;
220
}
221
175
-// Need to have a lock before calling this
176
-struct aclk_query *aclk_query_find(char *topic, char *data, char *msg_id, char *query)
222
+// 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)
224
{
178
- struct aclk_query *tmp_query;
225
+ struct aclk_query *tmp_query, *prev_query;
226
+ UNUSED(cmd);
227
228
tmp_query = aclk_queue.aclk_query_head;
181
-
229
+ prev_query = NULL;
230
while (tmp_query) {
231
if (likely(!tmp_query->deleted)) {
184
- if (strcmp(tmp_query->topic, topic) == 0 && (strcmp(tmp_query->query, query) == 0)) {
232
+ if (strcmp(tmp_query->topic, topic) == 0 && (!query || strcmp(tmp_query->query, query) == 0)) {
233
if ((!data || (data && strcmp(data, tmp_query->data) == 0)) &&
186
- (!msg_id || (msg_id && strcmp(msg_id, tmp_query->msg_id) == 0)))
234
+ (!msg_id || (msg_id && strcmp(msg_id, tmp_query->msg_id) == 0))) {
235
+
236
+ if (likely(last_query))
237
+ *last_query = prev_query;
238
return tmp_query;
239
+ }
240
}
241
}
242
+ prev_query = tmp_query;
243
tmp_query = tmp_query->next;
244
}
245
return NULL;
@@ -196,7 +249,7 @@ struct aclk_query *aclk_query_find(char *topic, char *data, char *msg_id, char *
249
* Add a query to execute, the result will be send to the specified topic
250
*/
251
199
-int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run_after, int internal)
252
+int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run_after, int internal, ACLK_CMD aclk_cmd)
253
{
254
struct aclk_query *new_query, *tmp_query;
255
@@ -204,23 +257,42 @@ int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run
257
if (unlikely(waiting_init))
258
return 0;
259
260
+ // Ignore all commands if agent not stable and reset the last_init_sequence mark
261
+ if (agent_state == 0) {
262
+ last_init_sequence = now_realtime_sec();
263
+ return 0;
264
+ }
265
+
266
run_after = now_realtime_sec() + run_after;
267
268
QUERY_LOCK;
210
- tmp_query = aclk_query_find(topic, data, msg_id, query);
269
+ struct aclk_query *last_query = NULL;
270
+
271
+ //last_query = NULL;
272
+ tmp_query = aclk_query_find(topic, data, msg_id, query, aclk_cmd, &last_query);
273
if (unlikely(tmp_query)) {
274
if (tmp_query->run_after == run_after) {
275
QUERY_UNLOCK;
276
QUERY_THREAD_WAKEUP;
277
return 0;
278
}
217
- tmp_query->deleted = 1;
279
+
280
+ if (last_query)
281
+ last_query->next = tmp_query->next;
282
+ else
283
+ aclk_queue.aclk_query_head = tmp_query->next;
284
+
285
+ debug(D_ACLK, "Removing double entry");
286
+ aclk_query_free(tmp_query);
287
+ aclk_queue.count--;
288
}
289
290
new_query = callocz(1, sizeof(struct aclk_query));
291
+ new_query->cmd = aclk_cmd;
292
if (internal) {
293
new_query->topic = strdupz(topic);
223
- new_query->query = strdupz(query);
294
+ if (likely(query))
295
+ new_query->query = strdupz(query);
296
} else {
297
new_query->topic = topic;
298
new_query->query = query;
@@ -234,7 +306,7 @@ int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run
306
new_query->created = now_realtime_sec();
307
new_query->run_after = run_after;
308
237
- info("Added query (%s) (%s)", topic, query);
309
+ debug(D_ACLK, "Added query (%s) (%s)", topic, query?query:"");
310
311
tmp_query = aclk_query_find_position(run_after);
312
@@ -256,29 +328,11 @@ int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run
328
QUERY_UNLOCK;
329
QUERY_THREAD_WAKEUP;
330
return 0;
259
-
260
-// if (likely(aclk_queue.aclk_query_tail)) {
261
-// aclk_queue.aclk_query_tail->next = new_query;
262
-// aclk_queue.aclk_query_tail = new_query;
263
-// aclk_queue.count++;
264
-// QUERY_UNLOCK;
265
-// return 0;
266
-// }
267
-//
268
-// if (likely(!aclk_queue.aclk_query_head)) {
269
-// aclk_queue.aclk_query_head = new_query;
270
-// aclk_queue.aclk_query_tail = new_query;
271
-// aclk_queue.count++;
272
-// QUERY_UNLOCK;
273
-// return 0;
274
-// }
275
-// QUERY_UNLOCK;
276
-// return 0;
331
}
332
333
inline int aclk_submit_request(struct aclk_request *request)
334
{
281
- return aclk_queue_query(request->topic, NULL, request->msg_id, request->url, 0, 0);
335
+ return aclk_queue_query(request->callback_topic, NULL, request->msg_id, request->payload, 0, 0, ACLK_CMD_CLOUD);
336
}
337
338
/*
@@ -303,7 +357,27 @@ struct aclk_query *aclk_queue_pop()
357
358
this_query = aclk_queue.aclk_query_head;
359
306
- if (this_query->run_after > now_realtime_sec()) {
360
+ // Get rid of the deleted entries
361
+ while (this_query && this_query->deleted) {
362
+ aclk_queue.count--;
363
+
364
+ aclk_queue.aclk_query_head = aclk_queue.aclk_query_head->next;
365
+
366
+ if (likely(!aclk_queue.aclk_query_head)) {
367
+ aclk_queue.aclk_query_tail = NULL;
368
+ }
369
+
370
+ aclk_query_free(this_query);
371
+
372
+ this_query = aclk_queue.aclk_query_head;
373
+ }
374
+
375
+ if (likely(!this_query)) {
376
+ QUERY_UNLOCK;
377
+ return NULL;
378
+ }
379
+
380
+ if (!this_query->deleted && this_query->run_after > now_realtime_sec()) {
381
info("Query %s will run in %ld seconds", this_query->query, this_query->run_after - now_realtime_sec());
382
QUERY_UNLOCK;
383
return NULL;
@@ -327,59 +401,295 @@ struct aclk_query *aclk_queue_pop()
401
// Need to check if additional logic should be added to make sure that there
402
// is enough information to determine the base topic at init time
403
330
-// TODO: Locking may be needed, depends on the calculation of the base topic and also if we need to switch
331
-// that on the fly
404
333
-char *get_publish_base_topic(PUBLISH_TOPIC_ACTION action)
405
+char *create_publish_base_topic()
406
{
335
- static char *topic = NULL;
336
-
407
if (unlikely(!is_agent_claimed()))
408
return NULL;
409
410
ACLK_LOCK;
411
342
- if (unlikely(action == PUBLICH_TOPIC_FREE)) {
343
- if (likely(topic)) {
344
- freez(topic);
345
- topic = NULL;
346
- }
412
+ if (unlikely(!global_base_topic)) {
413
+ char tmp_topic[ACLK_MAX_TOPIC + 1], *tmp;
414
+
415
+ snprintf(tmp_topic, ACLK_MAX_TOPIC, ACLK_TOPIC_STRUCTURE, is_agent_claimed());
416
+ tmp = strchr(tmp_topic, '\n');
417
+ if (unlikely(tmp))
418
+ *tmp = '\0';
419
+ global_base_topic = strdupz(tmp_topic);
420
+ }
421
+
422
+ ACLK_UNLOCK;
423
+ return global_base_topic;
424
+}
425
+
426
+/*
427
+ * Build a topic based on sub_topic and final_topic
428
+ * if the sub topic starts with / assume that is an absolute topic
429
+ *
430
+ */
431
+
432
+char *get_topic(char *sub_topic, char *final_topic, int max_size)
433
+{
434
+ int rc;
435
+
436
+ if (likely(sub_topic && sub_topic[0] == '/'))
437
+ return sub_topic;
438
+
439
+ if (unlikely(!global_base_topic))
440
+ return sub_topic;
441
348
- ACLK_UNLOCK;
442
+ rc = snprintf(final_topic, max_size, "%s/%s", global_base_topic, sub_topic);
443
+ if (unlikely(rc >= max_size))
444
+ debug(D_ACLK, "Topic has been truncated to [%s] instead of [%s/%s]", final_topic, global_base_topic, sub_topic);
445
+
446
+ return final_topic;
447
+}
448
+
449
+
450
+/*
451
+ * Free a collector structure
452
+ */
453
+
454
+static void _free_collector(struct _collector *collector)
455
+{
456
+
457
+ if (likely(collector->plugin_name))
458
+ freez(collector->plugin_name);
459
+
460
+ if (likely(collector->module_name))
461
+ freez(collector->module_name);
462
+
463
+ if (likely(collector->hostname))
464
+ freez(collector->hostname);
465
+
466
+ freez(collector);
467
+}
468
+
469
+/*
470
+ * This will report the collector list
471
+ *
472
+ */
473
+#ifdef ACLK_DEBUG
474
+static void _dump_connector_list()
475
+{
476
+
477
+ struct _collector *tmp_collector;
478
+
479
+ COLLECTOR_LOCK;
480
+
481
+ info("DUMPING ALL COLLECTORS");
482
+
483
+ if (unlikely(!collector_list || !collector_list->next)) {
484
+ COLLECTOR_UNLOCK;
485
+ info("DUMPING ALL COLLECTORS -- nothing found");
486
+ return;
487
+ }
488
489
+ // Note that the first entry is "dummy"
490
+ tmp_collector = collector_list->next;
491
+
492
+ while (tmp_collector) {
493
+ info(
494
+ "COLLECTOR %s : [%s:%s] count = %u", tmp_collector->hostname,
495
+ tmp_collector->plugin_name ? tmp_collector->plugin_name : "",
496
+ tmp_collector->module_name ? tmp_collector->module_name : "", tmp_collector->count);
497
+
498
+ tmp_collector = tmp_collector->next;
499
+
500
+ }
501
+ info("DUMPING ALL COLLECTORS DONE");
502
+ COLLECTOR_UNLOCK;
503
+}
504
+#endif
505
+
506
+/*
507
+ * This will cleanup the collector list
508
+ *
509
+ */
510
+static void _reset_connector_list()
511
+{
512
+ struct _collector *tmp_collector, *next_collector;
513
+
514
+ COLLECTOR_LOCK;
515
+
516
+ if (unlikely(!collector_list || !collector_list->next)) {
517
+ COLLECTOR_UNLOCK;
518
+ return;
519
+ }
520
+
521
+ // Note that the first entry is "dummy"
522
+ tmp_collector = collector_list->next;
523
+ collector_list->count = 0;
524
+ collector_list->next = NULL;
525
+
526
+ // We broke the link; we can unlock
527
+ COLLECTOR_UNLOCK;
528
+
529
+ while (tmp_collector) {
530
+ next_collector = tmp_collector->next;
531
+ _free_collector(tmp_collector);
532
+ tmp_collector = next_collector;
533
+ }
534
+}
535
+
536
+
537
+/*
538
+ * Find a collector (if it exists)
539
+ * Must lock before calling this
540
+ * If last_collector is not null, it will return the previous collector in the linked
541
+ * list (used in collector delete)
542
+ */
543
+static struct _collector *_find_collector(const char *hostname, const char *plugin_name, const char *module_name, struct _collector **last_collector)
544
+{
545
+ struct _collector *tmp_collector, *prev_collector;
546
+ uint32_t plugin_hash;
547
+ uint32_t module_hash;
548
+ uint32_t hostname_hash;
549
+
550
+ if (unlikely(!collector_list)) {
551
+ collector_list = callocz(1, sizeof(struct _collector));
552
return NULL;
553
}
554
353
- if (unlikely(action == PUBLICH_TOPIC_REBUILD)) {
354
- ACLK_UNLOCK;
355
- get_publish_base_topic(PUBLICH_TOPIC_FREE);
356
- return get_publish_base_topic(PUBLICH_TOPIC_GET);
555
+ if (unlikely(!collector_list->next))
556
+ return NULL;
557
+
558
+ plugin_hash = plugin_name?simple_hash(plugin_name):1;
559
+ module_hash = module_name?simple_hash(module_name):1;
560
+ hostname_hash = simple_hash(hostname);
561
+
562
+ // Note that the first entry is "dummy"
563
+ tmp_collector = collector_list->next;
564
+ prev_collector = collector_list;
565
+ 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)) &&
570
+ (!plugin_name || !tmp_collector->plugin_name || !strcmp(plugin_name, tmp_collector->plugin_name)) &&
571
+ (!module_name || !tmp_collector->module_name || !strcmp(module_name, tmp_collector->module_name))) {
572
+
573
+ if (unlikely(last_collector))
574
+ *last_collector = prev_collector;
575
+
576
+ return tmp_collector;
577
+ }
578
+
579
+ prev_collector = tmp_collector;
580
+ tmp_collector = tmp_collector->next;
581
}
582
359
- if (unlikely(!topic)) {
360
- char tmp_topic[ACLK_MAX_TOPIC + 1];
583
+ return tmp_collector;
584
+}
585
362
- sprintf(tmp_topic, ACLK_TOPIC_STRUCTURE, is_agent_claimed());
363
- topic = strdupz(tmp_topic);
586
+/*
587
+ * Called to delete a collector
588
+ * It will reduce the count (chart_count) and will remove it
589
+ * from the linked list if the count reaches zero
590
+ * The structure will be returned to the caller to free
591
+ * the resources
592
+ *
593
+ */
594
+static struct _collector *_del_collector(const char *hostname, const char *plugin_name, const char *module_name)
595
+{
596
+ struct _collector *tmp_collector, *prev_collector = NULL;
597
+
598
+ tmp_collector = _find_collector(hostname, plugin_name, module_name, &prev_collector);
599
+
600
+ if (likely(tmp_collector)) {
601
+ --tmp_collector->count;
602
+ if (unlikely(!tmp_collector->count))
603
+ prev_collector->next = tmp_collector->next;
604
}
605
+ return tmp_collector;
606
+}
607
366
- ACLK_UNLOCK;
367
- return topic;
608
+
609
+/*
610
+ * Add a new collector (plugin / module) to the list
611
+ * If it already exists just update the chart count
612
+ *
613
+ * Lock before calling
614
+ */
615
+static struct _collector *_add_collector(const char *hostname, const char *plugin_name, const char *module_name)
616
+{
617
+ struct _collector *tmp_collector;
618
+
619
+ tmp_collector = _find_collector(hostname, plugin_name, module_name, NULL);
620
+
621
+ if (unlikely(!tmp_collector)) {
622
+
623
+ tmp_collector = callocz(1, sizeof(struct _collector));
624
+ 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;
627
+
628
+ 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;
631
+
632
+ tmp_collector->next = collector_list->next;
633
+ collector_list->next = tmp_collector;
634
+ }
635
+ 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);
637
+ return tmp_collector;
638
}
639
370
-char *get_topic(char *sub_topic, char *final_topic, int max_size)
640
+/*
641
+ * Add a new collector to the list
642
+ * If it exists, update the chart count
643
+ */
644
+void aclk_add_collector(const char *hostname, const char *plugin_name, const char *module_name)
645
{
372
- if (unlikely(!global_base_topic))
373
- global_base_topic = GET_PUBLISH_BASE_TOPIC;
646
+ struct _collector *tmp_collector;
647
375
- if (unlikely(!global_base_topic))
376
- return sub_topic;
648
+ COLLECTOR_LOCK;
649
378
- snprintfz(final_topic, max_size, "%s/%s", global_base_topic, sub_topic);
650
+ tmp_collector = _add_collector(hostname, plugin_name, module_name);
651
380
- return final_topic;
652
+ if (unlikely(tmp_collector->count != 1)) {
653
+ COLLECTOR_UNLOCK;
654
+ return;
655
+ }
656
+
657
+ aclk_queue_query("connector", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT);
658
+
659
+ COLLECTOR_UNLOCK;
660
+}
661
+
662
+/*
663
+ * Delete a collector from the list
664
+ * If the chart count reaches zero the collector will be removed
665
+ * from the list by calling del_collector.
666
+ *
667
+ * This function will release the memory used and schedule
668
+ * a cloud update
669
+ */
670
+void aclk_del_collector(const char *hostname, const char *plugin_name, const char *module_name)
671
+{
672
+ struct _collector *tmp_collector;
673
+
674
+ COLLECTOR_LOCK;
675
+
676
+ tmp_collector = _del_collector(hostname, plugin_name, module_name);
677
+
678
+ if (unlikely(!tmp_collector || tmp_collector->count)) {
679
+ COLLECTOR_UNLOCK;
680
+ return;
681
+ }
682
+
683
+ debug(D_ACLK, "DEL COLLECTOR [%s:%s] -- charts %u", plugin_name?plugin_name:"*", module_name?module_name:"*", tmp_collector->count);
684
+
685
+ COLLECTOR_UNLOCK;
686
+
687
+ aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT);
688
+
689
+ _free_collector(tmp_collector);
690
}
691
692
+
693
// Wait for ACLK connection to be established
694
int aclk_wait_for_initialization()
695
{
@@ -389,6 +699,9 @@ int aclk_wait_for_initialization()
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)) {
@@ -399,6 +712,50 @@ int aclk_wait_for_initialization()
712
return 0;
713
}
714
715
+int aclk_execute_query(struct aclk_query *this_query)
716
+{
717
+ if (strncmp(this_query->query, "/api/v1/", 8) == 0) {
718
+ struct web_client *w = (struct web_client *)callocz(1, sizeof(struct web_client));
719
+ w->response.data = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
720
+ strcpy(w->origin, "*"); // Simulate web_client_create_on_fd()
721
+ w->cookie1[0] = 0; // Simulate web_client_create_on_fd()
722
+ w->cookie2[0] = 0; // Simulate web_client_create_on_fd()
723
+ w->acl = 0x1f;
724
+
725
+ char *mysep = strchr(this_query->query, '?');
726
+ if (mysep) {
727
+ strncpyz(w->decoded_query_string, mysep, NETDATA_WEB_REQUEST_URL_SIZE);
728
+ *mysep = '\0';
729
+ } else
730
+ strncpyz(w->decoded_query_string, this_query->query, NETDATA_WEB_REQUEST_URL_SIZE);
731
+
732
+ mysep = strrchr(this_query->query, '/');
733
+
734
+ // TODO: handle bad response perhaps in a different way. For now it does to the payload
735
+ int rc = web_client_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
736
+ BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
737
+ buffer_flush(local_buffer);
738
+ local_buffer->contenttype = CT_APPLICATION_JSON;
739
+
740
+ aclk_create_header(local_buffer, "http", this_query->msg_id);
741
+
742
+ if (rc != HTTP_RESP_OK || strcmp(mysep?mysep+1:"noop", "badge.svg") == 0)
743
+ buffer_sprintf(local_buffer, "\"%s\"", aclk_encode_response(w->response.data)->buffer);
744
+ else
745
+ buffer_sprintf(local_buffer, "%s", aclk_encode_response(w->response.data)->buffer);
746
+
747
+ buffer_sprintf(local_buffer,"\n}");
748
+
749
+ aclk_send_message(this_query->topic, local_buffer->buffer, this_query->msg_id);
750
+
751
+ buffer_free(w->response.data);
752
+ freez(w);
753
+ buffer_free(local_buffer);
754
+ return 0;
755
+ }
756
+ return 1;
757
+}
758
+
759
/*
760
* This function will fetch the next pending command and process it
761
*
@@ -406,79 +763,72 @@ int aclk_wait_for_initialization()
763
int aclk_process_query()
764
{
765
struct aclk_query *this_query;
409
- static u_int64_t query_count = 0;
410
- //int rc;
411
-
412
- if (unlikely(cmdpause))
413
- return 0;
766
+ static long int query_count = 0;
767
768
if (!aclk_connection_initialized)
769
return 0;
770
771
this_query = aclk_queue_pop();
772
if (likely(!this_query)) {
420
- //info("No pending queries");
773
return 0;
774
}
775
776
if (unlikely(this_query->deleted)) {
425
- info("Garbage collect query %s:%s", this_query->topic, this_query->query);
777
+ debug(D_ACLK, "Garbage collect query %s:%s", this_query->topic, this_query->query);
778
aclk_query_free(this_query);
779
return 1;
780
}
429
-
781
query_count++;
431
- info(
432
- "Query #%d (%s) (%s) in queue %d seconds", (int) query_count, this_query->topic, this_query->query,
433
- (int) (now_realtime_sec() - this_query->created));
782
435
- if (strncmp((char *)this_query->query, "/api/v1/", 8) == 0) {
436
- struct web_client *w = (struct web_client *)callocz(1, sizeof(struct web_client));
437
- w->response.data = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
438
- strcpy(w->origin, "*"); // Simulate web_client_create_on_fd()
439
- w->cookie1[0] = 0; // Simulate web_client_create_on_fd()
440
- w->cookie2[0] = 0; // Simulate web_client_create_on_fd()
441
- w->acl = 0x1f;
783
+ 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));
786
443
- char *mysep = strchr(this_query->query, '?');
444
- if (mysep) {
445
- strncpyz(w->decoded_query_string, mysep, NETDATA_WEB_REQUEST_URL_SIZE);
446
- *mysep = '\0';
447
- } else
448
- strncpyz(w->decoded_query_string, this_query->query, NETDATA_WEB_REQUEST_URL_SIZE);
787
+ switch (this_query->cmd) {
788
450
- mysep = strrchr(this_query->query, '/');
789
+ case ACLK_CMD_ONCONNECT:
790
+ debug(D_ACLK, "EXECUTING on connect metadata command");
791
+ aclk_send_metadata();
792
+ aclk_metadata_submitted = 2;
793
+ break;
794
452
- // TODO: ignore return code for now
453
- web_client_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
795
+ case ACLK_CMD_CHART:
796
+ debug(D_ACLK, "EXECUTING a chart update command");
797
+ aclk_send_single_chart(this_query->data, this_query->query);
798
+ break;
799
+
800
+ case ACLK_CMD_CHARTDEL:
801
+ debug(D_ACLK, "EXECUTING a chart delete command");
802
+ //TODO: This send the info metadata for now
803
+ aclk_send_info_metadata();
804
+ break;
805
455
- //TODO: handle bad response perhaps in a different way. For now it does to the payload
456
- //if (rc == HTTP_RESP_OK || 1) {
457
- buffer_flush(aclk_buffer);
806
+ case ACLK_CMD_ALARM:
807
+ debug(D_ACLK, "EXECUTING an alarm update command");
808
+ aclk_send_message(this_query->topic, this_query->query, this_query->msg_id);
809
+ break;
810
459
- aclk_create_metadata_message(aclk_buffer, mysep ? mysep + 1 : "noop", this_query->msg_id, w->response.data);
460
- aclk_buffer->contenttype = CT_APPLICATION_JSON;
461
- aclk_send_message(this_query->topic, aclk_buffer->buffer);
462
- //} else
463
- // error("Query RESP: %s", w->response.data->buffer);
811
+ case ACLK_CMD_ALARMS:
812
+ debug(D_ACLK, "EXECUTING an alarms update command");
813
+ aclk_send_alarm_metadata();
814
+ break;
815
465
- buffer_free(w->response.data);
466
- freez(w);
467
- aclk_query_free(this_query);
468
- return 1;
469
- }
816
+ case ACLK_CMD_CLOUD:
817
+ debug(D_ACLK, "EXECUTING a cloud command");
818
+ aclk_execute_query(this_query);
819
+ break;
820
471
- if (strcmp((char *)this_query->topic, "_chart") == 0) {
472
- aclk_send_single_chart(this_query->data, this_query->query);
821
+ default:
822
+ break;
823
}
824
+ debug(
825
+ D_ACLK, "Query #%ld (%s) done", query_count, this_query->topic);
826
827
aclk_query_free(this_query);
828
829
return 1;
830
}
831
480
-// Launch a query processing thread
481
-
832
/*
833
* Process all pending queries
834
* Return 0 if no queries were processed, 1 otherwise
@@ -487,17 +837,17 @@ int aclk_process_query()
837
838
int aclk_process_queries()
839
{
490
- if (unlikely(cmdpause))
840
+ if (unlikely(netdata_exit || !aclk_connection_initialized))
841
return 0;
842
493
- // Return if no queries pending
843
if (likely(!aclk_queue.count))
844
return 0;
845
497
- info("Processing %d queries", (int ) aclk_queue.count);
846
+ debug(D_ACLK, "Processing %d queries", (int ) aclk_queue.count);
847
848
+ //TODO: may consider possible throttling here
849
while (aclk_process_query()) {
500
- //rc = _link_event_loop(0);
850
+ // Process all commands
851
};
852
853
return 1;
@@ -510,38 +860,58 @@ static void aclk_query_thread_cleanup(void *ptr)
860
861
info("cleaning up...");
862
863
+ COLLECTOR_LOCK;
864
+
865
+ _reset_connector_list();
866
+ freez(collector_list);
867
+
868
+ COLLECTOR_UNLOCK;
869
+
870
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
871
}
872
873
/**
517
- * MAin query processing thread
874
+ * Main query processing thread
875
*
876
+ * On startup wait for the agent collectors to initialize
877
+ * Expect at least a time of ACLK_STABLE_TIMEOUT seconds
878
+ * of no new collectors coming in in order to mark the agent
879
+ * as stable (set agent_state = 1)
880
*/
881
void *aclk_query_main_thread(void *ptr)
882
{
883
netdata_thread_cleanup_push(aclk_query_thread_cleanup, ptr);
884
524
- while (!netdata_exit) {
885
+ while (!agent_state && !netdata_exit) {
886
+ time_t checkpoint;
887
+
888
+ checkpoint = now_realtime_sec() - last_init_sequence;
889
+ info("Waiting for agent collectors to initialize");
890
+ sleep_usec(USEC_PER_SEC * ACLK_STABLE_TIMEOUT);
891
+ if (checkpoint > ACLK_STABLE_TIMEOUT) {
892
+ agent_state = 1;
893
+ info("AGENT stable, last collector initialization activity was %ld seconds ago", checkpoint);
894
+#ifdef ACLK_DEBUG
895
+ _dump_connector_list();
896
+#endif
897
+ }
898
+ }
899
526
- QUERY_THREAD_LOCK;
900
+ while (!netdata_exit) {
901
902
if (unlikely(!aclk_metadata_submitted)) {
529
- aclk_send_metadata();
903
aclk_metadata_submitted = 1;
904
+ aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT);
905
}
906
907
+ aclk_process_queries();
908
+
909
+ QUERY_THREAD_LOCK;
910
+
911
+ // TODO: Need to check if there are queries awaiting already
912
if (unlikely(pthread_cond_wait(&query_cond_wait, &query_lock_wait)))
913
sleep_usec(USEC_PER_SEC * 1);
914
536
- if (likely(aclk_connection_initialized && !netdata_exit)) {
537
- while (aclk_process_queries()) {
538
- // Sleep for a few ms and retry maybe we have something to process
539
- // before going to sleep
540
- // TODO: This needs improvement to avoid missed queries
541
- sleep_usec(USEC_PER_MS * 100);
542
- }
543
- }
544
-
915
QUERY_THREAD_UNLOCK;
916
917
} // forever
@@ -558,6 +928,7 @@ static void aclk_main_cleanup(void *ptr)
928
929
info("cleaning up...");
930
931
+ // Wakeup thread to cleanup
932
QUERY_THREAD_WAKEUP;
933
934
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
@@ -575,73 +946,76 @@ static void aclk_main_cleanup(void *ptr)
946
*/
947
void *aclk_main(void *ptr)
948
{
578
- //netdata_thread_t *query_thread;
579
- struct netdata_static_thread query_thread;
580
-
581
- memset(&query_thread, 0, sizeof(query_thread));
949
+ struct netdata_static_thread *query_thread;
950
951
netdata_thread_cleanup_push(aclk_main_cleanup, ptr);
952
585
- if (unlikely(!aclk_buffer))
586
- aclk_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
587
-
588
- assert(aclk_buffer != NULL);
589
-
590
- //netdata_thread_cleanup_push(aclk_query_thread_cleanup, ptr);
591
- //netdata_thread_create(&query_thread.thread , "ACLKQ", NETDATA_THREAD_OPTION_DEFAULT, aclk_query_main_thread, &query_thread);
953
info("Waiting for netdata to be ready");
954
while (!netdata_ready) {
955
sleep_usec(USEC_PER_MS * 300);
956
}
596
- info("Waiting %d seconds for the agent to initialize", ACLK_STARTUP_WAIT);
597
- sleep_usec(USEC_PER_SEC * ACLK_STARTUP_WAIT);
957
599
- // Ok mark we are ready to accept incoming requests
600
- waiting_init = 0;
958
+ last_init_sequence = now_realtime_sec();
959
+ query_thread = NULL;
960
+
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
+
964
+
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;
971
+ }
972
+ create_publish_base_topic();
973
+
974
+ usec_t reconnect_expiry = 0; // In usecs
975
976
while (!netdata_exit) {
603
- // TODO: This may change when we have enough info from the claiming itself to avoid wasting 60 seconds
604
- // TODO: Handle the unclaim command as well -- we may need to shutdown the connection
605
- if (likely(!is_agent_claimed())) {
606
- sleep_usec(USEC_PER_SEC * 60);
607
- info("Checking agent claiming status");
608
- continue;
609
- }
977
+ static int first_init = 0;
978
+ _link_event_loop(ACLK_LOOP_TIMEOUT * 1000);
979
+ debug(D_ACLK,"LINK event loop called");
980
981
if (unlikely(!aclk_connection_initialized)) {
612
- static int initializing = 0;
613
-
614
- if (likely(initializing)) {
615
- _link_event_loop(ACLK_LOOP_TIMEOUT * 1000);
616
- continue;
617
- }
618
- initializing = 1;
619
- info("Initializing connection");
620
- //send_http_request(aclk_hostname, "443", "/auth/challenge?id=blah", aclk_buffer);
621
- if (unlikely(aclk_init(ACLK_INIT))) {
622
- // TODO: TBD how to handle. We are claimed and we cant init the connection. For now keep trying.
623
- sleep_usec(USEC_PER_SEC * 60);
624
- continue;
982
+ if (unlikely(first_init)) {
983
+ aclk_try_to_connect();
984
+ first_init = 1;
985
} else {
626
- sleep_usec(USEC_PER_SEC * 1);
986
+ if (aclk_connecting == 0) {
987
+ if (reconnect_expiry == 0) {
988
+ unsigned long int delay = aclk_reconnect_delay(1);
989
+ reconnect_expiry = now_realtime_usec() + delay * 1000;
990
+ info("Retrying to establish the ACLK connection in %.3f seconds", delay / 1000.0);
991
+ }
992
+ if (now_realtime_usec() >= reconnect_expiry) {
993
+ reconnect_expiry = 0;
994
+ aclk_connecting = 1;
995
+ aclk_try_to_connect();
996
+ }
997
+ sleep_usec(USEC_PER_MS * 100);
998
+ }
999
}
628
- _link_event_loop(ACLK_LOOP_TIMEOUT * 1000);
1000
continue;
1001
}
1002
632
- if (unlikely(!aclk_subscribed) && aclk_mqtt_connected) {
633
- aclk_subscribed = !aclk_subscribe(ACLK_COMMAND_TOPIC, 2);
634
- }
635
- if (unlikely(!query_thread.thread && aclk_mqtt_connected)) {
636
- query_thread.thread = mallocz(sizeof(netdata_thread_t));
637
- netdata_thread_create(
638
- query_thread.thread, "ACLKQ", NETDATA_THREAD_OPTION_DEFAULT, aclk_query_main_thread, &query_thread);
639
- }
1003
+ if (likely(aclk_mqtt_connected)) {
1004
641
- //TODO: Check if there is a return code
642
- _link_event_loop(ACLK_LOOP_TIMEOUT * 1000);
1005
+ if (unlikely(!aclk_subscribed)) {
1006
+ aclk_subscribed = !aclk_subscribe(ACLK_COMMAND_TOPIC, 2);
1007
+ }
1008
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
+ }
1016
+ }
1017
} // forever
1018
+exited:
1019
aclk_shutdown();
1020
1021
netdata_thread_cleanup_pop(1);
@@ -654,41 +1028,39 @@ void *aclk_main(void *ptr)
1028
* If base_topic is missing then the global_base_topic will be used (if available)
1029
*
1030
*/
657
-int aclk_send_message(char *sub_topic, char *message)
1031
+int aclk_send_message(char *sub_topic, char *message, char *msg_id)
1032
{
1033
int rc;
660
- static int skip_due_to_shutdown = 0;
1034
+ int mid;
1035
char topic[ACLK_MAX_TOPIC + 1];
1036
char *final_topic;
1037
664
- if (!aclk_connection_initialized)
665
- return 0;
666
-
667
- if (unlikely(netdata_exit)) {
668
- if (unlikely(!aclk_connection_initialized))
669
- return 1;
1038
+ UNUSED(msg_id);
1039
671
- ++skip_due_to_shutdown;
672
- if (unlikely(!(skip_due_to_shutdown % 100)))
673
- info("%d messages not sent -- shutdown in progress", skip_due_to_shutdown);
1040
+ if (unlikely(aclk_wait_for_initialization()))
1041
return 1;
675
- }
1042
1043
if (unlikely(!message))
1044
return 0;
1045
680
- if (unlikely(aclk_wait_for_initialization()))
681
- return 1;
682
-
1046
final_topic = get_topic(sub_topic, topic, ACLK_MAX_TOPIC);
1047
1048
+ if (unlikely(!final_topic)) {
1049
+ errno = 0;
1050
+ error("Unable to build outgoing topic; truncated?");
1051
+ return 1;
1052
+ }
1053
+
1054
ACLK_LOCK;
686
- rc = _link_send_message(final_topic, message);
1055
+ rc = _link_send_message(final_topic, message, &mid);
1056
+ // TODO: link the msg_id with the mid so we can trace it
1057
ACLK_UNLOCK;
1058
689
- // TODO: Add better handling -- error will flood the logfile here
690
- if (unlikely(rc))
1059
+
1060
+ if (unlikely(rc)) {
1061
+ errno = 0;
1062
error("Failed to send message, error code %d (%s)", rc, _link_strerror(rc));
1063
+ }
1064
1065
return rc;
1066
}
@@ -701,48 +1073,57 @@ int aclk_send_message(char *sub_topic, char *message)
1073
int aclk_subscribe(char *sub_topic, int qos)
1074
{
1075
int rc;
704
- //static char *global_base_topic = NULL;
1076
char topic[ACLK_MAX_TOPIC + 1];
1077
char *final_topic;
1078
708
- if (!aclk_connection_initialized)
709
- return 0;
710
-
711
- if (unlikely(netdata_exit)) {
712
- return 1;
713
- }
714
-
1079
if (unlikely(aclk_wait_for_initialization()))
1080
return 1;
1081
1082
final_topic = get_topic(sub_topic, topic, ACLK_MAX_TOPIC);
1083
+ if (unlikely(!final_topic)) {
1084
+ errno = 0;
1085
+ error("Unable to build outgoing topic; truncated?");
1086
+ return 1;
1087
+ }
1088
1089
ACLK_LOCK;
1090
rc = _link_subscribe(final_topic, qos);
1091
ACLK_UNLOCK;
1092
1093
// TODO: Add better handling -- error will flood the logfile here
725
- if (unlikely(rc))
726
- error("Failed to send message, error code %d (%s)", rc, _link_strerror(rc));
1094
+ if (unlikely(rc)) {
1095
+ errno = 0;
1096
+ error("Failed subscribe to command topic %d (%s)", rc, _link_strerror(rc));
1097
+ }
1098
1099
return rc;
1100
}
1101
1102
+
1103
// This is called from a callback when the link goes up
1104
void aclk_connect(void *ptr)
1105
{
734
- (void) ptr;
1106
+ UNUSED(ptr);
1107
info("Connection detected");
1108
+ aclk_connection_initialized = 1;
1109
+ waiting_init = 0;
1110
+ aclk_reconnect_delay(0);
1111
+ QUERY_THREAD_WAKEUP;
1112
return;
1113
}
1114
1115
// This is called from a callback when the link goes down
1116
void aclk_disconnect(void *ptr)
1117
{
742
- (void) ptr;
743
- info("Disconnect detected");
1118
+ UNUSED(ptr);
1119
+
1120
+ if (likely(aclk_connection_initialized))
1121
+ info("Disconnect detected");
1122
aclk_subscribed = 0;
1123
aclk_metadata_submitted = 0;
1124
+ waiting_init = 1;
1125
+ aclk_connection_initialized = 0;
1126
+ aclk_connecting = 0;
1127
}
1128
1129
void aclk_shutdown()
@@ -753,46 +1134,20 @@ void aclk_shutdown()
1134
info("Shutdown complete");
1135
}
1136
756
-int aclk_init(ACLK_INIT_ACTION action)
1137
+void aclk_try_to_connect()
1138
{
758
- (void) action;
759
-
760
- static int init = 0;
1139
int rc;
762
-
763
- if (likely(init))
764
- return 0;
765
-
766
- aclk_send_maximum = config_get_number(CONFIG_SECTION_ACLK, "agent cloud link send maximum", 20);
767
- aclk_recv_maximum = config_get_number(CONFIG_SECTION_ACLK, "agent cloud link receive maximum", 20);
768
-
769
- aclk_hostname = config_get(CONFIG_SECTION_ACLK, "agent cloud link hostname", "localhost");
770
- aclk_port = config_get_number(CONFIG_SECTION_ACLK, "agent cloud link port", 9002);
771
-
772
- info("Maximum parallel outgoing messages %d", aclk_send_maximum);
773
- info("Maximum parallel incoming messages %d", aclk_recv_maximum);
774
-
775
- // This will setup the base publish topic internally
776
- //get_publish_base_topic(PUBLICH_TOPIC_GET);
777
-
778
- // initialize the low level link to the cloud
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");
782
- return 1;
1143
}
784
- global_base_topic = GET_PUBLISH_BASE_TOPIC;
785
- init = 1;
786
-
787
- return 0;
1144
}
1145
790
-// Use this to disable encoding of quotes and newlines so that
791
-// MQTT subscriber can display more readable data on screen
1146
793
-void aclk_create_header(BUFFER *dest, char *type, char *msg_id)
1147
+inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id)
1148
{
1149
uuid_t uuid;
1150
+ time_t time_created;
1151
char uuid_str[36 + 1];
1152
1153
if (unlikely(!msg_id)) {
@@ -801,50 +1156,56 @@ void aclk_create_header(BUFFER *dest, char *type, char *msg_id)
1156
msg_id = uuid_str;
1157
}
1158
1159
+ time_created = now_realtime_sec();
1160
+
1161
buffer_sprintf(
1162
dest,
1163
"\t{\"type\": \"%s\",\n"
1164
"\t\"msg-id\": \"%s\",\n"
808
- "\t\"version\": %s,\n"
1165
+ "\t\"timestamp\": %ld,\n"
1166
+ "\t\"version\": %d,\n"
1167
"\t\"payload\": ",
810
- type, msg_id, ACLK_VERSION);
1168
+ type, msg_id, time_created, ACLK_VERSION);
1169
+
1170
+ debug(D_ACLK, "Sending v%d msgid [%s] type [%s] time [%ld]", ACLK_VERSION, msg_id, type, time_created);
1171
}
1172
813
-#define EYE_FRIENDLY 1
1173
+//#define EYE_FRIENDLY
1174
815
-// encapsulate contents into metadata message as per ACLK documentation
816
-void aclk_create_metadata_message(BUFFER *dest, char *type, char *msg_id, BUFFER *contents)
1175
+/*
1176
+ * Take a buffer, encode it and rewrite it
1177
+ *
1178
+ */
1179
+
1180
+BUFFER *aclk_encode_response(BUFFER *contents)
1181
{
818
-#ifndef EYE_FRIENDLY
819
- char *tmp_buffer = mallocz(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
820
- char *src, *dst;
821
-#endif
1182
+#ifdef EYE_FRIENDLY
1183
823
- buffer_sprintf(
824
- dest,
825
- "\t{\"type\": \"%s\",\n"
826
- "\t\"msg-id\": \"%s\",\n"
827
- "\t\"payload\": %s\n\t}",
828
- type, msg_id ? msg_id : "", contents->buffer);
1184
+ return contents;
1185
+#else
1186
+ char *tmp_buffer = mallocz(contents->len * 2);
1187
+ char *src, *dst;
1188
830
-#ifndef EYE_FRIENDLY
831
- //TODO: this is the initial escaping, It will expanded
832
- src = dest->buffer;
1189
+ src = contents->buffer;
1190
dst = tmp_buffer;
1191
while (*src) {
1192
switch (*src) {
836
- case '0x0a':
1193
case '\n':
1194
*dst++ = '\\';
1195
*dst++ = 'n';
1196
break;
841
- case '\"':
1197
+ case 0x01 ... 0x09:
1198
+ case 0x0b ... 0x1F:
1199
*dst++ = '\\';
843
- *dst++ = '\"';
1200
+ *dst++ = '0';
1201
+ *dst++ = '0';
1202
+ *dst++ = (*src < 0x0F) ? '0' : '1';
1203
+ *dst++ = to_hex(*src);
1204
break;
1205
+ case '\"':
1206
case '\'':
1207
*dst++ = '\\';
847
- *dst++ = '\"';
1208
+ *dst++ = *src;
1209
break;
1210
default:
1211
*dst++ = *src;
@@ -853,147 +1214,244 @@ void aclk_create_metadata_message(BUFFER *dest, char *type, char *msg_id, BUFFER
1214
}
1215
*dst = '\0';
1216
856
- buffer_flush(dest);
857
- buffer_sprintf(dest, "%s", tmp_buffer);
1217
+ buffer_flush(contents);
1218
+ buffer_sprintf(contents, "%s", tmp_buffer);
1219
1220
freez(tmp_buffer);
1221
+ return contents;
1222
#endif
861
- return;
1223
}
1224
864
-//TODO: this has been changed in the latest specs. We need to pack the data in one MQTT
865
-//message with a payload and has a list of json objects
866
-int aclk_send_alarm_metadata()
1225
+/*
1226
+ * This will send the alarms configuration
1227
+ * and
1228
+ */
1229
+void aclk_send_alarm_metadata()
1230
{
868
- //TODO: improve locking on the buffer -- same lock is used for the message send
869
- //improve error handling
870
- ACLK_LOCK;
871
- buffer_flush(aclk_buffer);
872
- // Alarms configuration
873
- aclk_create_header(aclk_buffer, "alarms", NULL);
874
- health_alarms2json(localhost, aclk_buffer, 1);
875
- buffer_sprintf(aclk_buffer,"\n}");
876
- ACLK_UNLOCK;
877
- aclk_send_message(ACLK_ALARMS_TOPIC, aclk_buffer->buffer);
1231
+ BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1232
879
- // Alarms log
880
- ACLK_LOCK;
881
- buffer_flush(aclk_buffer);
882
- aclk_create_header(aclk_buffer, "alarms_log", NULL);
883
- health_alarm_log2json(localhost, aclk_buffer, 0);
884
- buffer_sprintf(aclk_buffer,"\n}");
885
- ACLK_UNLOCK;
886
- aclk_send_message(ACLK_ALARMS_TOPIC, aclk_buffer->buffer);
1233
+ char *msg_id = create_uuid();
1234
+ buffer_flush(local_buffer);
1235
+ local_buffer->contenttype = CT_APPLICATION_JSON;
1236
888
- return 0;
1237
+ debug(D_ACLK,"Metadata alarms start");
1238
+
1239
+ aclk_create_header(local_buffer, "connect_alarms", msg_id);
1240
+
1241
+ buffer_sprintf(local_buffer,"{\n\t \"configured-alarms\" : ");
1242
+ health_alarms2json(localhost, local_buffer, 1);
1243
+ debug(D_ACLK,"Metadata %s with configured alarms has %ld bytes", msg_id, local_buffer->len);
1244
+
1245
+ buffer_sprintf(local_buffer,",\n\t \"alarm-log\" : ");
1246
+ health_alarm_log2json(localhost, local_buffer, 0);
1247
+ debug(D_ACLK,"Metadata %s with alarm_log has %ld bytes", msg_id, local_buffer->len);
1248
+
1249
+ buffer_sprintf(local_buffer,",\n\t \"alarms-active\" : ");
1250
+ 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
+
1253
+
1254
+ buffer_sprintf(local_buffer,"\n}\n}");
1255
+ 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);
1257
+
1258
+ freez(msg_id);
1259
+ buffer_free(local_buffer);
1260
}
1261
1262
+int aclk_send_info_metadata()
1263
+{
1264
+ BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1265
+
1266
+ debug(D_ACLK,"Metadata /info start");
1267
+
1268
+ char *msg_id = create_uuid();
1269
+ buffer_flush(local_buffer);
1270
+ local_buffer->contenttype = CT_APPLICATION_JSON;
1271
+
1272
+ aclk_create_header(local_buffer, "connect", msg_id);
1273
+ buffer_sprintf(local_buffer,"{\n\t \"info\" : ");
1274
+ 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);
1276
+
1277
+ buffer_sprintf(local_buffer,", \n\t \"charts\" : ");
1278
+ 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);
1281
+
1282
+ 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);
1284
+ freez(msg_id);
1285
+
1286
+ buffer_free(local_buffer);
1287
+ return 0;
1288
+}
1289
1290
// Send info metadata message to the cloud if the link is established
1291
// or on request
1292
int aclk_send_metadata()
1293
{
896
- ACLK_LOCK;
1294
898
- buffer_flush(aclk_buffer);
1295
+ aclk_send_info_metadata();
1296
+ aclk_send_alarm_metadata();
1297
900
- aclk_create_header(aclk_buffer, "connect", NULL);
901
- buffer_sprintf(aclk_buffer,"{\n\t \"info\" : ");
902
- web_client_api_request_v1_info_fill_buffer(localhost, aclk_buffer);
903
- buffer_sprintf(aclk_buffer,", \n\t \"charts\" : ");
904
- charts2json(localhost, aclk_buffer);
905
- buffer_sprintf(aclk_buffer,"\n}\n}");
906
- aclk_buffer->contenttype = CT_APPLICATION_JSON;
1298
+ return 0;
1299
+}
1300
908
- ACLK_UNLOCK;
1301
+void aclk_single_update_disable()
1302
+{
1303
+ aclk_disable_single_updates = 1;
1304
+}
1305
910
- aclk_send_message(ACLK_METADATA_TOPIC, aclk_buffer->buffer);
1306
+void aclk_single_update_enable()
1307
+{
1308
+ aclk_disable_single_updates = 0;
1309
+}
1310
912
- aclk_send_alarm_metadata();
1311
+// Trigged by a health reload, sends the alarm metadata
1312
+void aclk_alarm_reload()
1313
+{
1314
+ if (unlikely(!agent_state))
1315
+ return;
1316
914
- return 0;
1317
+ aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT);
1318
}
916
-
1319
//rrd_stats_api_v1_chart(RRDSET *st, BUFFER *buf)
1320
1321
int aclk_send_single_chart(char *hostname, char *chart)
1322
{
1323
RRDHOST *target_host;
922
- ACLK_LOCK;
923
-
924
- buffer_flush(aclk_buffer);
1324
1325
target_host = rrdhost_find_by_hostname(hostname, 0);
1326
if (!target_host)
1327
return 1;
1328
1329
RRDSET *st = rrdset_find(target_host, chart);
931
-
1330
if (!st)
1331
st = rrdset_find_byname(target_host, chart);
934
-
1332
if (!st) {
1333
info("FAILED to find chart %s", chart);
1334
return 1;
1335
}
1336
940
- aclk_buffer->contenttype = CT_APPLICATION_JSON;
1337
+ BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1338
+ char *msg_id = create_uuid();
1339
+ buffer_flush(local_buffer);
1340
+ local_buffer->contenttype = CT_APPLICATION_JSON;
1341
942
- buffer_flush(aclk_buffer);
1342
+ aclk_create_header(local_buffer, "chart", msg_id);
1343
+ rrdset2json(st, local_buffer, NULL, NULL, 1);
1344
+ buffer_sprintf(local_buffer,"\t\n}");
1345
944
- aclk_create_header(aclk_buffer, "chart", NULL);
1346
+ aclk_send_message(ACLK_CHART_TOPIC, aclk_encode_response(local_buffer)->buffer, msg_id);
1347
946
- rrdset2json(st, aclk_buffer, NULL, NULL);
947
- buffer_sprintf(aclk_buffer,"\n}\n}");
948
-
949
-
950
- ACLK_UNLOCK;
951
- aclk_send_message(ACLK_METADATA_TOPIC, aclk_buffer->buffer);
1348
+ freez(msg_id);
1349
+ buffer_free(local_buffer);
1350
return 0;
1351
}
1352
955
-int aclk_update_chart(RRDHOST *host, char *chart_name)
1353
+int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd)
1354
{
957
- (void) host;
958
- (void) chart_name;
1355
#ifndef ENABLE_ACLK
1356
+ UNUSED(host);
1357
+ UNUSED(chart_name);
1358
return 0;
1359
#else
1360
if (host != localhost)
1361
return 0;
1362
965
- aclk_queue_query("_chart", host->hostname, NULL, chart_name, 2, 1);
1363
+ if (unlikely(aclk_disable_single_updates))
1364
+ return 0;
1365
+
1366
+ aclk_queue_query("_chart", host->hostname, NULL, chart_name, 0, 1, aclk_cmd);
1367
return 0;
1368
#endif
1369
}
1370
970
-int aclk_update_alarm(RRDHOST *host, char *alarm_name)
1371
+
1372
+int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1373
{
1374
+ BUFFER *local_buffer = NULL;
1375
+
1376
if (host != localhost)
1377
return 0;
1378
975
- aclk_queue_query("_alarm", host->hostname, NULL, alarm_name, 2, 1);
1379
+ if (agent_state == 0)
1380
+ return 0;
1381
+
1382
+ /*
1383
+ * Check if individual updates have been disabled
1384
+ * This will be the case when we do health reload
1385
+ * and all the alarms will be dropped and recreated.
1386
+ * At the end of the health reload the complete alarm metadata
1387
+ * info will be sent
1388
+ */
1389
+ if (unlikely(aclk_disable_single_updates))
1390
+ return 0;
1391
+
1392
+ local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1393
+ char *msg_id = create_uuid();
1394
+
1395
+ buffer_flush(local_buffer);
1396
+ aclk_create_header(local_buffer, "alarms", msg_id);
1397
+
1398
+ netdata_rwlock_rdlock(&host->health_log.alarm_log_rwlock);
1399
+ health_alarm_entry2json_nolock(local_buffer, ae, host);
1400
+ netdata_rwlock_unlock(&host->health_log.alarm_log_rwlock);
1401
+
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);
1404
+
1405
+ freez(msg_id);
1406
+ buffer_free(local_buffer);
1407
+
1408
return 0;
1409
}
1410
979
-
980
-//TODO: add and check the incoming type e.g http
1411
+/*
1412
+ * Parse the incoming payload and queue a command if valid
1413
+ */
1414
int aclk_handle_cloud_request(char *payload)
1415
{
983
- struct aclk_request cloud_to_agent = { .msg_id = NULL, .topic = NULL, .url = NULL, .version = 1};
1416
+ struct aclk_request cloud_to_agent = { .type_id = NULL, .msg_id = NULL, .callback_topic = NULL, .payload = NULL, .version = 0};
1417
+
1418
+ if (unlikely(!payload)) {
1419
+ debug(D_ACLK, "ACLK incoming message is empty");
1420
+ return 0;
1421
+ }
1422
+
1423
+ debug(D_ACLK, "ACLK incoming message [%s]", payload);
1424
1425
int rc = json_parse(payload, &cloud_to_agent, cloud_to_agent_parse);
1426
987
- if (unlikely(JSON_OK != rc)) {
988
- error("Malformed json request (%s)", payload);
989
- return 1;
990
- }
1427
+ if (unlikely(
1428
+ JSON_OK != rc || !cloud_to_agent.payload || !cloud_to_agent.callback_topic || !cloud_to_agent.msg_id ||
1429
+ !cloud_to_agent.type_id || cloud_to_agent.version > ACLK_VERSION ||
1430
+ strcmp(cloud_to_agent.type_id, "http"))) {
1431
+
1432
+ if (JSON_OK != rc)
1433
+ error("Malformed json request (%s)", payload);
1434
+
1435
+ if (cloud_to_agent.version > ACLK_VERSION)
1436
+ error("Unsupported version in JSON request %d", cloud_to_agent.version);
1437
+
1438
+ if (cloud_to_agent.payload)
1439
+ freez(cloud_to_agent.payload);
1440
+
1441
+ if (cloud_to_agent.type_id)
1442
+ freez(cloud_to_agent.type_id);
1443
+
1444
+ if (cloud_to_agent.msg_id)
1445
+ freez(cloud_to_agent.msg_id);
1446
+
1447
+ if (cloud_to_agent.callback_topic)
1448
+ freez(cloud_to_agent.callback_topic);
1449
992
- if (unlikely(!cloud_to_agent.url || !cloud_to_agent.topic)) {
1450
return 1;
1451
}
1452
1453
aclk_submit_request(&cloud_to_agent);
1454
1455
+ // Note: the payload comes from the callback and it will be automatically freed
1456
return 0;
1457
}
aclk/agent_cloud_link.h
+48
-31
@@ -6,20 +6,21 @@
6
#include "../daemon/common.h"
7
#include "mqtt.h"
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"
14
-
15
-
16
-#define ACLK_MSG_TYPE_CHART "chart"
16
+#define ACLK_CHART_TOPIC "chart"
17
#define ACLK_ALARMS_TOPIC "alarms"
18
#define ACLK_METADATA_TOPIC "meta"
19
#define ACLK_COMMAND_TOPIC "cmd"
20
#define ACLK_TOPIC_STRUCTURE "/agent/%s"
21
22
-#define ACLK_STARTUP_WAIT 30 // Seconds to wait before establishing initialization process
22
+#define ACLK_MAX_BACKOFF_DELAY 1024 // maximum backoff delay in seconds
23
+
24
#define ACLK_INITIALIZATION_WAIT 60 // Wait for link to initialize in seconds (per msg)
25
#define ACLK_INITIALIZATION_SLEEP_WAIT 1 // Wait time @ spin lock for MQTT initialization in seconds
26
#define ACLK_QOS 1
@@ -28,26 +29,32 @@
29
30
#define ACLK_MAX_TOPIC 255
31
31
-#define ACLK_RECONNECT_DELAY 1 // reconnect delay -- with backoff stragegy fow now
32
-#define ACLK_MAX_RECONNECT_DELAY 120
33
-#define ACLK_VERSION "1"
32
+#define ACLK_RECONNECT_DELAY 1 // reconnect delay -- with backoff stragegy fow now
33
+#define ACLK_STABLE_TIMEOUT 10 // Minimum delay to mark AGENT as stable
34
+#define ACLK_DEFAULT_PORT 9002
35
+#define ACLK_DEFAULT_HOST "localhost"
36
37
#define CONFIG_SECTION_ACLK "agent_cloud_link"
38
39
struct aclk_request {
40
char *type_id;
41
char *msg_id;
40
- char *topic;
41
- char *url;
42
+ char *callback_topic;
43
+ char *payload;
44
int version;
45
};
46
47
46
-typedef enum publish_topic_action {
47
- PUBLICH_TOPIC_GET,
48
- PUBLICH_TOPIC_FREE,
49
- PUBLICH_TOPIC_REBUILD
50
-} PUBLISH_TOPIC_ACTION;
48
+typedef enum aclk_cmd {
49
+ ACLK_CMD_CLOUD,
50
+ ACLK_CMD_ONCONNECT,
51
+ ACLK_CMD_INFO,
52
+ ACLK_CMD_CHART,
53
+ ACLK_CMD_CHARTDEL,
54
+ ACLK_CMD_ALARM,
55
+ ACLK_CMD_ALARMS,
56
+ ACLK_CMD_MAX
57
+} ACLK_CMD;
58
59
typedef enum aclk_init_action {
60
ACLK_INIT,
@@ -55,10 +62,6 @@ typedef enum aclk_init_action {
62
} ACLK_INIT_ACTION;
63
64
58
-#define GET_PUBLISH_BASE_TOPIC get_publish_base_topic(0)
59
-#define FREE_PUBLISH_BASE_TOPIC get_publish_base_topic(1)
60
-#define REBUILD_PUBLISH_BASE_TOPIC get_publish_base_topic(2)
61
-
65
void *aclk_main(void *ptr);
66
67
#define NETDATA_ACLK_HOOK \
@@ -72,32 +75,46 @@ void *aclk_main(void *ptr);
75
.start_routine = aclk_main \
76
},
77
75
-extern int aclk_send_message(char *sub_topic, char *message);
78
+extern int aclk_send_message(char *sub_topic, char *message, char *msg_id);
79
77
-int aclk_init();
78
-char *get_base_topic();
80
+//int aclk_init();
81
+//char *get_base_topic();
82
83
extern char *is_agent_claimed(void);
84
+char *create_uuid();
85
+
86
87
// callbacks for agent cloud link
83
-int aclk_subscribe(char *topic, int qos);
88
+int aclk_subscribe(char *topic, int qos);
89
void aclk_shutdown();
85
-//void aclk_message_callback(struct mosquitto *moqs, void *obj, const struct mosquitto_message *msg);
86
-
90
int cloud_to_agent_parse(JSON_ENTRY *e);
91
void aclk_disconnect(void *conn);
92
void aclk_connect(void *conn);
90
-void aclk_create_metadata_message(BUFFER *dest, char *type, char *msg_id, BUFFER *contents);
93
int aclk_send_metadata();
94
+int aclk_send_info_metadata();
95
int aclk_wait_for_initialization();
93
-//int aclk_send_charts(RRDHOST *host, BUFFER *wb);
96
+char *create_publish_base_topic();
97
+void aclk_try_to_connect();
98
+
99
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);
96
-struct aclk_query *aclk_query_find(char *token, char *data, char *msg_id, char *query);
97
-//void aclk_rrdset2json(RRDSET *st, BUFFER *wb, char *hostname, int is_slave);
98
-int aclk_update_chart(RRDHOST *host, char *chart_name);
99
-int aclk_update_alarm(RRDHOST *host, char *alarm_name);
100
+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);
103
+int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd);
104
+int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae);
105
void aclk_create_header(BUFFER *dest, char *type, char *msg_id);
106
int aclk_handle_cloud_request(char *payload);
107
int aclk_submit_request(struct aclk_request *);
108
+void aclk_add_collector(const char *hostname, const char *plugin_name, const char *module_name);
109
+void aclk_del_collector(const char *hostname, const char *plugin_name, const char *module_name);
110
+void aclk_alarm_reload();
111
+void aclk_send_alarm_metadata();
112
+int aclk_execute_query(struct aclk_query *query);
113
+BUFFER *aclk_encode_response(BUFFER *contents);
114
+unsigned long int aclk_reconnect_delay(int mode);
115
+extern void health_alarm_entry2json_nolock(BUFFER *wb, ALARM_ENTRY *ae, RRDHOST *host);
116
+void aclk_single_update_enable();
117
+void aclk_single_update_disable();
118
+
119
+
120
#endif //NETDATA_AGENT_CLOUD_LINK_H
aclk/mqtt.c
+91
-111
@@ -7,34 +7,33 @@
7
8
void (*_on_connect)(void *ptr) = NULL;
9
void (*_on_disconnect)(void *ptr) = NULL;
10
-extern int cmdpause;
11
-
10
11
#ifndef ENABLE_ACLK
12
13
inline const char *_link_strerror(int rc)
14
{
17
- (void) rc;
15
+ UNUSED(rc);
16
return "no error";
17
}
18
19
int _link_event_loop(int timeout)
20
{
23
- (void) timeout;
21
+ UNUSED(timeout);
22
return 0;
23
}
24
27
-int _link_send_message(char *topic, char *message)
25
+int _link_send_message(char *topic, char *message, int *mid)
26
{
29
- (void) topic;
30
- (void) message;
27
+ UNUSED(topic);
28
+ UNUSED(message);
29
+ UNUSED(mid);
30
return 0;
31
}
32
33
int _link_subscribe(char *topic, int qos)
34
{
36
- (void) topic;
37
- (void) qos;
35
+ UNUSED(topic);
36
+ UNUSED(qos);
37
return 0;
38
}
39
@@ -45,115 +44,84 @@ void _link_shutdown()
44
45
int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *))
46
{
48
- (void) aclk_hostname;
49
- (void) aclk_port;
50
- (void) on_connect;
51
- (void) on_disconnect;
47
+ UNUSED(aclk_hostname);
48
+ UNUSED(aclk_port);
49
+ UNUSED(on_connect);
50
+ UNUSED(on_disconnect);
51
return 0;
52
}
53
54
#else
56
-/*
57
- * Just report the library info in the logfile for reference when issues arise
58
- *
59
- */
55
56
struct mosquitto *mosq = NULL;
57
58
// Get a string description of the error
64
-
59
inline const char *_link_strerror(int rc)
60
{
61
return mosquitto_strerror(rc);
62
}
63
70
-
71
-void mqtt_message_callback(
72
- struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
64
+void mqtt_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
65
{
74
- (void) mosq;
75
- (void) obj;
76
-
77
- // TODO: handle commands in a more efficient way, if we have many
78
-
79
- if (strcmp((char *)msg->payload, "pause") == 0) {
80
- cmdpause = 1;
81
- return;
82
- }
83
-
84
- if (strcmp((char *)msg->payload, "resume") == 0) {
85
- cmdpause = 0;
86
- return;
87
- }
88
-
89
- if (strcmp((char *)msg->payload, "reload") == 0) {
90
- error_log_limit_unlimited();
91
- info("Reloading health configuration");
92
- health_reload();
93
- error_log_limit_reset();
94
- return;
95
- }
96
-
97
- if (strcmp((char *)msg->payload, "info") == 0) {
98
- aclk_send_metadata();
99
- return;
100
- }
66
+ UNUSED(mosq);
67
+ UNUSED(obj);
68
69
aclk_handle_cloud_request(msg->payload);
103
-
104
- //info("Received type=[%s], msg-id=[%s], topic=[%s], url=[%s]",cloud_to_agent.type_id, cloud_to_agent.msg_id, cloud_to_agent.topic, cloud_to_agent.url);
105
-
70
}
71
108
-int lws_wss_client_initialized = 0;
109
-
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
78
+void publish_callback(struct mosquitto *mosq, void *obj, int rc)
79
+{
80
+ UNUSED(mosq);
81
+ UNUSED(obj);
82
+ UNUSED(rc);
83
+
84
+ // TODO: link this with a msg_id so it can be traced
85
+ return;
86
+}
87
+
88
void connect_callback(struct mosquitto *mosq, void *obj, int rc)
89
{
118
- (void) obj;
119
- (void) rc;
90
+ UNUSED(obj);
91
+ UNUSED(rc);
92
93
info("Connection to cloud estabilished");
94
123
- aclk_connection_initialized = 1;
95
aclk_mqtt_connected = 1;
125
- _on_connect((void *) mosq);
96
+ _on_connect((void *)mosq);
97
98
return;
99
}
100
130
-
101
void disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
102
{
133
- (void) obj;
134
- (void) rc;
103
+ UNUSED(obj);
104
+ UNUSED(rc);
105
106
info("Connection to cloud failed");
137
- // TODO: Keep the connection "alive" for now. The library will reconnect.
107
139
- //mqtt_connection_initialized = 0;
108
aclk_mqtt_connected = 0;
141
- _on_disconnect((void *) mosq);
109
+ _on_disconnect((void *)mosq);
110
143
- if(mqtt_over_websockets && lws_engine_instance)
111
+ if (mqtt_over_websockets && lws_engine_instance)
112
aclk_lws_wss_mqtt_layer_disconect_notif(lws_engine_instance);
113
146
- //sleep_usec(USEC_PER_SEC * 5);
114
return;
115
}
116
150
-
117
void _show_mqtt_info()
118
{
119
int libmosq_major, libmosq_minor, libmosq_revision, libmosq_version;
154
- libmosq_version = mosquitto_lib_version(&libmosq_major, &libmosq_minor, &libmosq_revision);
120
+ libmosq_version = mosquitto_lib_version(&libmosq_major, &libmosq_minor, &libmosq_revision);
121
156
- info("Detected libmosquitto library version %d, %d.%d.%d",libmosq_version, libmosq_major, libmosq_minor, libmosq_revision);
122
+ info(
123
+ "Detected libmosquitto library version %d, %d.%d.%d", libmosq_version, libmosq_major, libmosq_minor,
124
+ libmosq_revision);
125
}
126
127
size_t _mqtt_external_write_hook(void *buf, size_t count)
@@ -166,7 +134,7 @@ size_t _mqtt_external_read_hook(void *buf, size_t count)
134
return aclk_lws_wss_client_read(lws_engine_instance, buf, count);
135
}
136
169
-int _mqtt_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *))
137
+int _mqtt_lib_init(void (*on_connect)(void *), void (*on_disconnect)(void *))
138
{
139
int rc;
140
int libmosq_major, libmosq_minor, libmosq_revision, libmosq_version;
@@ -174,13 +142,12 @@ int _mqtt_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *
142
char *server_crt;
143
char *server_key;
144
177
- // show library info so can have in in the logfile
145
+ // show library info so can have it in the logfile
146
libmosq_version = mosquitto_lib_version(&libmosq_major, &libmosq_minor, &libmosq_revision);
147
ca_crt = config_get(CONFIG_SECTION_ACLK, "agent cloud link cert", "*");
148
server_crt = config_get(CONFIG_SECTION_ACLK, "agent cloud link server cert", "*");
149
server_key = config_get(CONFIG_SECTION_ACLK, "agent cloud link server key", "*");
150
183
-
151
if (ca_crt[0] == '*') {
152
freez(ca_crt);
153
ca_crt = NULL;
@@ -196,9 +163,9 @@ int _mqtt_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *
163
server_key = NULL;
164
}
165
199
- info(
200
- "Detected libmosquitto library version %d, %d.%d.%d", libmosq_version, libmosq_major, libmosq_minor,
201
- libmosq_revision);
166
+ // info(
167
+ // "Detected libmosquitto library version %d, %d.%d.%d", libmosq_version, libmosq_major, libmosq_minor,
168
+ // libmosq_revision);
169
170
rc = mosquitto_lib_init();
171
if (unlikely(rc != MOSQ_ERR_SUCCESS)) {
@@ -218,6 +185,7 @@ int _mqtt_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *
185
186
mosquitto_connect_callback_set(mosq, connect_callback);
187
mosquitto_disconnect_callback_set(mosq, disconnect_callback);
188
+ mosquitto_publish_callback_set(mosq, publish_callback);
189
190
mosquitto_username_pw_set(mosq, NULL, NULL);
191
@@ -234,8 +202,8 @@ int _mqtt_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *
202
info("MQTT in flight messages set to 1 -- %s", mosquitto_strerror(rc));
203
#endif
204
237
- if(!mqtt_over_websockets) {
238
- rc = mosquitto_reconnect_delay_set(mosq, ACLK_RECONNECT_DELAY, ACLK_MAX_RECONNECT_DELAY, 1);
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));
@@ -253,9 +221,11 @@ int _link_mqtt_connect(char *aclk_hostname, int aclk_port)
221
rc = mosquitto_connect_async(mosq, aclk_hostname, aclk_port, ACLK_PING_INTERVAL);
222
223
if (unlikely(rc != MOSQ_ERR_SUCCESS))
256
- error("Connect %s MQTT status = %d (%s)", aclk_hostname, rc, mosquitto_strerror(rc));
224
+ error(
225
+ "Failed to establish link to [%s:%d] MQTT status = %d (%s)", aclk_hostname, aclk_port, rc,
226
+ mosquitto_strerror(rc));
227
else
258
- info("Establishing MQTT link to %s", aclk_hostname);
228
+ info("Establishing MQTT link to [%s:%d]", aclk_hostname, aclk_port);
229
230
return rc;
231
}
@@ -264,60 +234,72 @@ static inline void _link_mosquitto_write()
234
{
235
int rc;
236
267
- if(!mqtt_over_websockets)
237
+ if (!mqtt_over_websockets)
238
return;
239
240
rc = mosquitto_loop_misc(mosq);
271
- if(unlikely( rc != MOSQ_ERR_SUCCESS ))
241
+ if (unlikely(rc != MOSQ_ERR_SUCCESS))
242
debug(D_ACLK, "ACLK: failure during mosquitto_loop_misc %s", mosquitto_strerror(rc));
243
274
- if(likely( mosquitto_want_write(mosq) )) {
244
+ if (likely(mosquitto_want_write(mosq))) {
245
rc = mosquitto_loop_write(mosq, 1);
276
- if( rc != MOSQ_ERR_SUCCESS )
246
+ if (rc != MOSQ_ERR_SUCCESS)
247
debug(D_ACLK, "ACLK: failure during mosquitto_loop_write %s", mosquitto_strerror(rc));
248
}
249
}
250
281
-void aclk_lws_connect_notif_callback(){
251
+void aclk_lws_connect_notif_callback()
252
+{
253
//the connection is done by LWS so this parameters dont matter
254
//ig MQTT over LWS is used
284
- _link_mqtt_connect("doesntmatter", 12345);
255
+ _link_mqtt_connect(aclk_hostname, aclk_port);
256
_link_mosquitto_write();
257
}
258
288
-void aclk_lws_data_received_callback(){
259
+void aclk_lws_data_received_callback()
260
+{
261
int rc = mosquitto_loop_read(mosq, 1);
290
- if(rc != MOSQ_ERR_SUCCESS)
291
- debug(D_ACLK, "ACLK: failure during mosquitto_loop_read %s", mosquitto_strerror(rc));
262
+ if (rc != MOSQ_ERR_SUCCESS)
263
+ debug(D_ACLK, "ACLK: failure during mosquitto_loop_read %s", mosquitto_strerror(rc));
264
+}
265
+
266
+void aclk_lws_connection_closed()
267
+{
268
+ aclk_disconnect(NULL);
269
}
270
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,
297
- .data_writable_callback = NULL
274
+ .data_writable_callback = NULL,
275
+ .connection_closed = aclk_lws_connection_closed
276
};
277
278
int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *))
279
{
280
int rc;
281
304
- if(mqtt_over_websockets) {
282
+ if (mqtt_over_websockets) {
283
// we will connect when WebSocket connection is up
284
// based on callback
307
- if(!lws_wss_client_initialized) {
285
+ if (!lws_engine_instance)
286
lws_engine_instance = aclk_lws_wss_client_init(&aclk_lws_engine_callbacks, aclk_hostname, aclk_port);
309
- aclk_lws_wss_service_loop(lws_engine_instance);
310
- lws_wss_client_initialized = 1;
311
- }
287
+ else
288
+ aclk_lws_wss_connect(lws_engine_instance);
289
+
290
+ aclk_lws_wss_service_loop(lws_engine_instance);
291
}
292
314
- rc = _mqtt_lib_init(aclk_hostname, aclk_port, on_connect, on_disconnect);
315
- if(rc != MOSQ_ERR_SUCCESS)
293
+ rc = _mqtt_lib_init(on_connect, on_disconnect);
294
+ if (rc != MOSQ_ERR_SUCCESS)
295
return rc;
296
318
- if(mqtt_over_websockets) {
297
+ if (mqtt_over_websockets) {
298
mosquitto_external_callbacks_set(mosq, _mqtt_external_write_hook, _mqtt_external_read_hook);
320
- return MOSQ_ERR_SUCCESS;
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
@@ -327,7 +309,11 @@ int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *
309
310
static inline int _link_event_loop_wss()
311
{
330
- if(lws_engine_instance && lws_engine_instance->websocket_connection_up)
312
+ if (unlikely(!lws_engine_instance)) {
313
+ return MOSQ_ERR_SUCCESS;
314
+ }
315
+
316
+ if (lws_engine_instance && lws_engine_instance->websocket_connection_up)
317
_link_mosquitto_write();
318
319
aclk_lws_wss_service_loop(lws_engine_instance);
@@ -357,9 +343,9 @@ static inline int _link_event_loop_plain_mqtt(int timeout)
343
344
int _link_event_loop(int timeout)
345
{
360
- if(mqtt_over_websockets)
346
+ if (mqtt_over_websockets)
347
return _link_event_loop_wss();
362
-
348
+
349
return _link_event_loop_plain_mqtt(timeout);
350
}
351
@@ -380,7 +366,7 @@ void _link_shutdown()
366
mosquitto_destroy(mosq);
367
mosq = NULL;
368
383
- if(lws_engine_instance) {
369
+ if (lws_engine_instance) {
370
aclk_lws_wss_client_destroy(lws_engine_instance);
371
lws_engine_instance = NULL;
372
}
@@ -388,8 +374,7 @@ void _link_shutdown()
374
return;
375
}
376
391
-
392
-int _link_subscribe(char *topic, int qos)
377
+int _link_subscribe(char *topic, int qos)
378
{
379
int rc;
380
@@ -410,13 +395,12 @@ int _link_subscribe(char *topic, int qos)
395
return 0;
396
}
397
413
-
398
/*
399
* Send a message to the cloud to specific topic
400
*
401
*/
402
419
-int _link_send_message(char *topic, char *message)
403
+int _link_send_message(char *topic, char *message, int *mid)
404
{
405
int rc;
406
@@ -427,15 +411,11 @@ int _link_send_message(char *topic, char *message)
411
412
int msg_len = strlen(message);
413
430
- // TODO: handle encoding validation -- the message should be UFT8 encoded by the sender
431
- //rc = mosquitto_validate_utf8(message, msg_len);
432
- //if (unlikely(rc != MOSQ_ERR_SUCCESS))
433
- // return rc;
434
-
435
- rc = mosquitto_publish(mosq, NULL, topic, msg_len, message, ACLK_QOS, 0);
414
+ rc = mosquitto_publish(mosq, mid, topic, msg_len, message, ACLK_QOS, 0);
415
416
// TODO: Add better handling -- error will flood the logfile here
417
if (unlikely(rc != MOSQ_ERR_SUCCESS)) {
418
+ errno = 0;
419
error("MQTT message failed : %s", mosquitto_strerror(rc));
420
}
421
aclk/mqtt.h
+3
-1
@@ -12,12 +12,14 @@ int _link_event_loop(int timeout);
12
void _link_shutdown();
13
int _link_lib_init(char *aclk_hostname, int aclk_port, void (*on_connect)(void *), void (*on_disconnect)(void *));
14
int _link_subscribe(char *topic, int qos);
15
-int _link_send_message(char *topic, char *message);
15
+int _link_send_message(char *topic, char *message, int *mid);
16
const char *_link_strerror(int rc);
17
18
int aclk_handle_cloud_request(char *);
19
20
extern int aclk_connection_initialized;
21
extern int aclk_mqtt_connected;
22
+extern char *aclk_hostname;
23
+extern int aclk_port;
24
25
#endif //NETDATA_MQTT_H
build_external/bin/clean-install.sh
+5
@@ -42,6 +42,11 @@ else
42
--build-arg "DISTRO=$DISTRO" --build-arg "VERSION=$VERSION" --build-arg ACLK=yes \
43
--build-arg EXTRA_CFLAGS="-DACLK_SSL_ALLOW_SELF_SIGNED"
44
;;
45
+ arch-extras) # Add valgrind to the container
46
+ docker build -f "$BuildBase/clean-install-arch-extras.Dockerfile" -t "${DISTRO}_${VERSION}_dev" "$BuildBase/.." \
47
+ --build-arg "DISTRO=$DISTRO" --build-arg "VERSION=$VERSION" --build-arg ACLK=yes \
48
+ --build-arg EXTRA_CFLAGS="-DACLK_SSL_ALLOW_SELF_SIGNED"
49
+ ;;
50
*)
51
echo "Unknown $DISTRO-$VERSION"
52
;;
build_external/clean-install-arch-extras.Dockerfile
new
+56
@@ -0,0 +1,56 @@
1
+FROM archlinux/base:latest
2
+
3
+# There is some redundancy between this file and the archlinux Dockerfile in the helper images
4
+# repo and also with the clean-install.Dockefile. Once the help image is availabled on Docker
5
+# Hub this file can be deleted.
6
+
7
+RUN pacman -Syyu --noconfirm
8
+RUN pacman --noconfirm --needed -S autoconf \
9
+ autoconf-archive \
10
+ autogen \
11
+ automake \
12
+ gcc \
13
+ make \
14
+ git \
15
+ libuv \
16
+ lz4 \
17
+ netcat \
18
+ openssl \
19
+ pkgconfig \
20
+ python \
21
+ libvirt \
22
+ libwebsockets \
23
+ valgrind
24
+
25
+ARG ACLK=no
26
+ARG EXTRA_CFLAGS
27
+COPY . /opt/netdata/source
28
+WORKDIR /opt/netdata/source
29
+
30
+RUN git config --global user.email "root@container"
31
+RUN git config --global user.name "Fake root"
32
+
33
+# RUN make distclean -> not safe if tree state changed on host since last config
34
+# Kill everything that is not in .gitignore preserving any fresh changes, i.e. untracked changes will be
35
+# deleted but local changes to tracked files will be preserved.
36
+RUN if git status --porcelain | grep '^[MADRC]'; then \
37
+ git stash && git clean -dxf && (git stash apply || true) \
38
+ else \
39
+ git clean -dxf ; \
40
+ fi
41
+
42
+# Not everybody is updating distclean properly - fix.
43
+RUN find . -name '*.Po' -exec rm \{\} \;
44
+RUN rm -rf autom4te.cache
45
+RUN rm -rf .git/
46
+RUN find . -type f >/opt/netdata/manifest
47
+
48
+RUN CFLAGS="-O1 -ggdb -Wall -Wextra -Wformat-signedness -fstack-protector-all -DNETDATA_INTERNAL_CHECKS=1\
49
+ -D_FORTIFY_SOURCE=2 -DNETDATA_VERIFY_LOCKS=1 ${EXTRA_CFLAGS}" ./netdata-installer.sh --disable-lto
50
+
51
+RUN ln -sf /dev/stdout /var/log/netdata/access.log
52
+RUN ln -sf /dev/stdout /var/log/netdata/debug.log
53
+RUN ln -sf /dev/stderr /var/log/netdata/error.log
54
+
55
+CMD ["/usr/sbin/valgrind", "--leak-check=full", "/usr/sbin/netdata", "-D"]
56
+
build_external/clean-install-arch.Dockerfile
+2
@@ -50,3 +50,5 @@ RUN CFLAGS="-O1 -ggdb -Wall -Wextra -Wformat-signedness -fstack-protector-all -D
50
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
build_external/projects/aclk-testing/agent-valgrind-compose.yml
new
+19
@@ -0,0 +1,19 @@
1
+version: '3.3'
2
+services:
3
+ agent_master:
4
+ build:
5
+ context: ../../..
6
+ dockerfile: build_external/make-install.Dockerfile
7
+ args:
8
+ - DISTRO=arch
9
+ - VERSION=extras
10
+ image: arch_current_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"
17
+ ports:
18
+ - 20000:19999
19
+
build_external/projects/aclk-testing/docker-compose.yml
deleted
-51
@@ -1,51 +0,0 @@
1
-version: '3.3'
2
-services:
3
- agent_master:
4
- build:
5
- context: ../../..
6
- dockerfile: build_external/make_install_fedora_30.Dockerfile
7
- args:
8
- ACLK: "yes"
9
- image: fedora_30_dev
10
- command: >
11
- sh -c "echo 00000000-0000-0000-0000-000000000000 >/etc/netdata/claim.d/claimed_id &&
12
- echo '[agent_cloud_link]' >>/etc/netdata/netdata.conf &&
13
- echo ' agent cloud link hostname = 172.22.0.100' >>/etc/netdata/netdata.conf &&
14
- echo ' agent cloud link port = 8080' >>/etc/netdata/netdata.conf &&
15
- /usr/sbin/netdata -D"
16
- ports:
17
- - 20000:19999
18
- networks:
19
- service1_net:
20
- ipv4_address: 172.22.0.99
21
- #volumes:
22
- #- ./master_stream.conf:/etc/netdata/stream.conf:ro
23
-# agent_slave1:
24
-# image: debian_buster_dev
25
-# command: /usr/sbin/netdata -D
26
-# ports:
27
-# - 20001:19999
28
-# volumes:
29
-# - ./slave_stream.conf:/etc/netdata/stream.conf:ro
30
-# agent_slave2:
31
-# image: ubuntu_2004_dev
32
-# command: /usr/sbin/netdata -D
33
-# ports:
34
-# - 20002:19999
35
-# volumes:
36
-# - ./slave_stream.conf:/etc/netdata/stream.conf:ro
37
- vernemq:
38
- build:
39
- dockerfile: configureVerneMQ.Dockerfile
40
- context: .
41
- networks:
42
- service1_net:
43
- ipv4_address: 172.22.0.100
44
-
45
-networks:
46
- service1_net:
47
- ipam:
48
- driver: default
49
- config:
50
- - subnet: 172.22.0.0/16
51
-
build_external/projects/aclk-testing/paho-compose.yml
new
+6
@@ -0,0 +1,6 @@
1
+version: '3.3'
2
+services:
3
+ paho_inspect:
4
+ build:
5
+ context: .
6
+ dockerfile: paho.Dockerfile
\ No newline at end of file
build_external/projects/aclk-testing/paho-inspection.py
new
+33
@@ -0,0 +1,33 @@
1
+import ssl
2
+import paho.mqtt.client as mqtt
3
+
4
+def on_connect(mqttc, obj, flags, rc):
5
+ print("connected rc: "+str(rc), flush=True)
6
+ mqttc.subscribe("/agent/#",0)
7
+def on_disconnect(mqttc, obj, flags, rc):
8
+ print("disconnected rc: "+str(rc), flush=True)
9
+def on_message(mqttc, obj, msg):
10
+ print(msg.topic+" "+str(msg.qos)+" "+str(msg.payload), flush=True)
11
+def on_publish(mqttc, obj, mid):
12
+ print("mid: "+str(mid), flush=True)
13
+def on_subscribe(mqttc, obj, mid, granted_qos):
14
+ print("Subscribed: "+str(mid)+" "+str(granted_qos), flush=True)
15
+def on_log(mqttc, obj, level, string):
16
+ print(string)
17
+print("Starting paho-inspection", flush=True)
18
+mqttc = mqtt.Client(transport='websockets')
19
+#mqttc.tls_set(certfile="server.crt", keyfile="server.key", cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS, ciphers=None)
20
+#mqttc.tls_set(ca_certs="server.crt", cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS, ciphers=None)
21
+mqttc.tls_set(cert_reqs=ssl.CERT_NONE, tls_version=ssl.PROTOCOL_TLS, ciphers=None)
22
+mqttc.tls_insecure_set(True)
23
+mqttc.on_message = on_message
24
+mqttc.on_connect = on_connect
25
+mqttc.on_disconnect = on_disconnect
26
+mqttc.on_publish = on_publish
27
+mqttc.on_subscribe = on_subscribe
28
+mqttc.connect("vernemq", 9002, 60)
29
+
30
+#mqttc.publish("/agent/mine","Test1")
31
+#mqttc.subscribe("$SYS/#", 0)
32
+print("Connected succesfully, monitoring /agent/#", flush=True)
33
+mqttc.loop_forever()
build_external/projects/aclk-testing/paho.Dockerfile
new
+12
@@ -0,0 +1,12 @@
1
+FROM archlinux/base:latest
2
+
3
+RUN pacman -Syyu --noconfirm
4
+RUN pacman --noconfirm --needed -S python-pip
5
+
6
+RUN pip install paho-mqtt
7
+
8
+RUN mkdir -p /opt/paho
9
+COPY paho-inspection.py /opt/paho/
10
+
11
+WORKDIR /opt/paho
12
+CMD ["/usr/sbin/python", "paho-inspection.py"]
\ No newline at end of file
database/rrddim.c
+7
-7
@@ -184,7 +184,7 @@ void rrdcalc_link_to_rrddim(RRDDIM *rd, RRDSET *st, RRDHOST *host) {
184
}
185
}
186
#ifdef ENABLE_ACLK
187
- aclk_update_chart(st->rrdhost, st->id);
187
+ aclk_update_chart(st->rrdhost, st->id, ACLK_CMD_CHART);
188
#endif
189
}
190
@@ -428,7 +428,7 @@ RRDDIM *rrddim_add_custom(RRDSET *st, const char *id, const char *name, collecte
428
}
429
rrdset_unlock(st);
430
#ifdef ENABLE_ACLK
431
- aclk_update_chart(host, st->id);
431
+ aclk_update_chart(host, st->id, ACLK_CMD_CHART);
432
#endif
433
return(rd);
434
}
@@ -484,7 +484,7 @@ void rrddim_free(RRDSET *st, RRDDIM *rd)
484
break;
485
}
486
#ifdef ENABLE_ACLK
487
- aclk_update_chart(st->rrdhost, st->id);
487
+ aclk_update_chart(st->rrdhost, st->id, ACLK_CMD_CHART);
488
#endif
489
}
490
@@ -505,7 +505,7 @@ int rrddim_hide(RRDSET *st, const char *id) {
505
506
rrddim_flag_set(rd, RRDDIM_FLAG_HIDDEN);
507
#ifdef ENABLE_ACLK
508
- aclk_update_chart(st->rrdhost, st->id);
508
+ aclk_update_chart(st->rrdhost, st->id, ACLK_CMD_CHART);
509
#endif
510
return 0;
511
}
@@ -522,7 +522,7 @@ int rrddim_unhide(RRDSET *st, const char *id) {
522
523
rrddim_flag_clear(rd, RRDDIM_FLAG_HIDDEN);
524
#ifdef ENABLE_ACLK
525
- aclk_update_chart(st->rrdhost, st->id);
525
+ aclk_update_chart(st->rrdhost, st->id, ACLK_CMD_CHART);
526
#endif
527
return 0;
528
}
@@ -533,7 +533,7 @@ inline void rrddim_is_obsolete(RRDSET *st, RRDDIM *rd) {
533
rrddim_flag_set(rd, RRDDIM_FLAG_OBSOLETE);
534
rrdset_flag_set(st, RRDSET_FLAG_OBSOLETE_DIMENSIONS);
535
#ifdef ENABLE_ACLK
536
- aclk_update_chart(st->rrdhost, st->id);
536
+ aclk_update_chart(st->rrdhost, st->id, ACLK_CMD_CHART);
537
#endif
538
}
539
@@ -542,7 +542,7 @@ inline void rrddim_isnot_obsolete(RRDSET *st __maybe_unused, RRDDIM *rd) {
542
543
rrddim_flag_clear(rd, RRDDIM_FLAG_OBSOLETE);
544
#ifdef ENABLE_ACLK
545
- aclk_update_chart(st->rrdhost, st->id);
545
+ aclk_update_chart(st->rrdhost, st->id, ACLK_CMD_CHART);
546
#endif
547
}
548
database/rrdset.c
+6
-1
@@ -424,6 +424,10 @@ void rrdset_delete(RRDSET *st) {
424
}
425
426
recursively_delete_dir(st->cache_dir, "left-over chart");
427
+#ifdef ENABLE_ACLK
428
+ aclk_del_collector(st->rrdhost->hostname, st->plugin_name, st->module_name);
429
+ aclk_update_chart(st->rrdhost, st->id, ACLK_CMD_CHARTDEL);
430
+#endif
431
}
432
433
void rrdset_delete_obsolete_dimensions(RRDSET *st) {
@@ -762,7 +766,8 @@ RRDSET *rrdset_create_custom(
766
767
rrdhost_unlock(host);
768
#ifdef ENABLE_ACLK
765
- aclk_update_chart(host, st->id);
769
+ aclk_add_collector(host->hostname, plugin, module);
770
+ aclk_update_chart(host, st->id, ACLK_CMD_CHART);
771
#endif
772
return(st);
773
}
health/health.c
+7
-1
@@ -179,7 +179,9 @@ void health_reload_host(RRDHOST *host) {
179
* Reload the host configuration for all hosts.
180
*/
181
void health_reload(void) {
182
-
182
+#ifdef ENABLE_ACLK
183
+ aclk_single_update_disable();
184
+#endif
185
rrd_rdlock();
186
187
RRDHOST *host;
@@ -187,6 +189,10 @@ void health_reload(void) {
189
health_reload_host(host);
190
191
rrd_unlock();
192
+#ifdef ENABLE_ACLK
193
+ aclk_single_update_enable();
194
+ aclk_alarm_reload();
195
+#endif
196
}
197
198
// ----------------------------------------------------------------------------
health/health_json.c
+1
-1
@@ -13,7 +13,7 @@ static inline void health_string2json(BUFFER *wb, const char *prefix, const char
13
buffer_sprintf(wb, "%s\"%s\":null%s", prefix, label, suffix);
14
}
15
16
-static inline void health_alarm_entry2json_nolock(BUFFER *wb, ALARM_ENTRY *ae, RRDHOST *host) {
16
+inline void health_alarm_entry2json_nolock(BUFFER *wb, ALARM_ENTRY *ae, RRDHOST *host) {
17
buffer_sprintf(wb,
18
"\n\t{\n"
19
"\t\t\"hostname\": \"%s\",\n"
health/health_log.c
+3
@@ -152,6 +152,9 @@ inline void health_alarm_log_save(RRDHOST *host, ALARM_ENTRY *ae) {
152
host->health_log_entries_written++;
153
}
154
}
155
+#ifdef ENABLE_ACLK
156
+ aclk_update_alarm(host, ae);
157
+#endif
158
}
159
160
inline ssize_t health_alarm_log_read(RRDHOST *host, FILE *fp, const char *filename) {
web/api/formatters/charts2json.c
+2
-2
@@ -36,7 +36,7 @@ static inline const char* get_release_channel() {
36
return (use_stable)?"stable":"nightly";
37
}
38
39
-void charts2json(RRDHOST *host, BUFFER *wb) {
39
+void charts2json(RRDHOST *host, BUFFER *wb, int skip_volatile) {
40
static char *custom_dashboard_info_js_filename = NULL;
41
size_t c, dimensions = 0, memory = 0, alarms = 0;
42
RRDSET *st;
@@ -76,7 +76,7 @@ void charts2json(RRDHOST *host, BUFFER *wb) {
76
buffer_strcat(wb, "\n\t\t\"");
77
buffer_strcat(wb, st->id);
78
buffer_strcat(wb, "\": ");
79
- rrdset2json(st, wb, &dimensions, &memory);
79
+ rrdset2json(st, wb, &dimensions, &memory, skip_volatile);
80
81
c++;
82
st->last_accessed_time = now;
web/api/formatters/charts2json.h
+1
-1
@@ -5,7 +5,7 @@
5
6
#include "rrd2json.h"
7
8
-extern void charts2json(RRDHOST *host, BUFFER *wb);
8
+extern void charts2json(RRDHOST *host, BUFFER *wb, int skip_volatile);
9
extern void chartcollectors2json(RRDHOST *host, BUFFER *wb);
10
11
#endif //NETDATA_API_FORMATTER_CHARTS2JSON_H
web/api/formatters/rrd2json.c
+1
-1
@@ -3,7 +3,7 @@
3
#include "web/api/web_api_v1.h"
4
5
void rrd_stats_api_v1_chart(RRDSET *st, BUFFER *wb) {
6
- rrdset2json(st, wb, NULL, NULL);
6
+ rrdset2json(st, wb, NULL, NULL, 0);
7
}
8
9
void rrdr_buffer_print_format(BUFFER *wb, uint32_t format) {
web/api/formatters/rrdset2json.c
+59
-46
@@ -4,7 +4,7 @@
4
5
// generate JSON for the /api/v1/chart API call
6
7
-void rrdset2json(RRDSET *st, BUFFER *wb, size_t *dimensions_count, size_t *memory_used) {
7
+void rrdset2json(RRDSET *st, BUFFER *wb, size_t *dimensions_count, size_t *memory_used, int skip_volatile) {
8
rrdset_rdlock(st);
9
10
time_t first_entry_t = rrdset_first_entry_t(st);
@@ -25,30 +25,44 @@ void rrdset2json(RRDSET *st, BUFFER *wb, size_t *dimensions_count, size_t *memor
25
"\t\t\t\"units\": \"%s\",\n"
26
"\t\t\t\"data_url\": \"/api/v1/data?chart=%s\",\n"
27
"\t\t\t\"chart_type\": \"%s\",\n"
28
- "\t\t\t\"duration\": %ld,\n"
29
- "\t\t\t\"first_entry\": %ld,\n"
30
- "\t\t\t\"last_entry\": %ld,\n"
31
- "\t\t\t\"update_every\": %d,\n"
32
- "\t\t\t\"dimensions\": {\n"
33
- , st->id
34
- , st->name
35
- , st->type
36
- , st->family
37
- , st->context
38
- , st->title, st->name
39
- , st->priority
40
- , st->plugin_name?st->plugin_name:""
41
- , st->module_name?st->module_name:""
42
- , rrdset_flag_check(st, RRDSET_FLAG_ENABLED)?"true":"false"
43
- , st->units
44
- , st->name
45
- , rrdset_type_name(st->chart_type)
46
- , last_entry_t - first_entry_t + st->update_every//st->entries * st->update_every
47
- , first_entry_t//rrdset_first_entry_t(st)
48
- , last_entry_t//rrdset_last_entry_t(st)
49
- , st->update_every
28
+ , st->id
29
+ , st->name
30
+ , st->type
31
+ , st->family
32
+ , st->context
33
+ , st->title, st->name
34
+ , st->priority
35
+ , st->plugin_name?st->plugin_name:""
36
+ , st->module_name?st->module_name:""
37
+ , rrdset_flag_check(st, RRDSET_FLAG_ENABLED)?"true":"false"
38
+ , st->units
39
+ , st->name
40
+ , rrdset_type_name(st->chart_type)
41
);
42
43
+ if (likely(!skip_volatile))
44
+ buffer_sprintf(wb,
45
+ "\t\t\t\"duration\": %ld,\n"
46
+ , last_entry_t - first_entry_t + st->update_every//st->entries * st->update_every
47
+ );
48
+
49
+ buffer_sprintf(wb,
50
+ "\t\t\t\"first_entry\": %ld,\n"
51
+ , first_entry_t //rrdset_first_entry_t(st)
52
+ );
53
+
54
+ if (likely(!skip_volatile))
55
+ buffer_sprintf(wb,
56
+ "\t\t\t\"last_entry\": %ld,\n"
57
+ , last_entry_t//rrdset_last_entry_t(st)
58
+ );
59
+
60
+ buffer_sprintf(wb,
61
+ "\t\t\t\"update_every\": %d,\n"
62
+ "\t\t\t\"dimensions\": {\n"
63
+ , st->update_every
64
+ );
65
+
66
unsigned long memory = st->memsize;
67
68
size_t dimensions = 0;
@@ -81,33 +95,32 @@ void rrdset2json(RRDSET *st, BUFFER *wb, size_t *dimensions_count, size_t *memor
95
buffer_strcat(wb, ",\n\t\t\t\"red\": ");
96
buffer_rrd_value(wb, st->red);
97
84
- buffer_strcat(wb, ",\n\t\t\t\"alarms\": {\n");
85
- size_t alarms = 0;
86
- RRDCALC *rc;
87
- for(rc = st->alarms; rc ; rc = rc->rrdset_next) {
88
-
89
- buffer_sprintf(
90
- wb
91
- , "%s"
92
- "\t\t\t\t\"%s\": {\n"
93
- "\t\t\t\t\t\"id\": %u,\n"
94
- "\t\t\t\t\t\"status\": \"%s\",\n"
95
- "\t\t\t\t\t\"units\": \"%s\",\n"
96
- "\t\t\t\t\t\"update_every\": %d\n"
97
- "\t\t\t\t}"
98
- , (alarms) ? ",\n" : ""
99
- , rc->name
100
- , rc->id
101
- , rrdcalc_status2string(rc->status)
102
- , rc->units
103
- , rc->update_every
98
+ if (likely(!skip_volatile)) {
99
+ buffer_strcat(wb, ",\n\t\t\t\"alarms\": {\n");
100
+ size_t alarms = 0;
101
+ RRDCALC *rc;
102
+ for (rc = st->alarms; rc; rc = rc->rrdset_next) {
103
+ buffer_sprintf(
104
+ wb,
105
+ "%s"
106
+ "\t\t\t\t\"%s\": {\n"
107
+ "\t\t\t\t\t\"id\": %u,\n"
108
+ "\t\t\t\t\t\"status\": \"%s\",\n"
109
+ "\t\t\t\t\t\"units\": \"%s\",\n"
110
+ "\t\t\t\t\t\"update_every\": %d\n"
111
+ "\t\t\t\t}",
112
+ (alarms) ? ",\n" : "", rc->name, rc->id, rrdcalc_status2string(rc->status), rc->units,
113
+ rc->update_every);
114
+
115
+ alarms++;
116
+ }
117
+ buffer_sprintf(wb,
118
+ "\n\t\t\t}"
119
);
105
-
106
- alarms++;
120
}
121
122
buffer_sprintf(wb,
110
- "\n\t\t\t}\n\t\t}"
123
+ "\n\t\t}"
124
);
125
126
rrdset_unlock(st);
web/api/formatters/rrdset2json.h
+1
-1
@@ -5,6 +5,6 @@
5
6
#include "rrd2json.h"
7
8
-extern void rrdset2json(RRDSET *st, BUFFER *wb, size_t *dimensions_count, size_t *memory_used);
8
+extern void rrdset2json(RRDSET *st, BUFFER *wb, size_t *dimensions_count, size_t *memory_used, int skip_volatile);
9
10
#endif //NETDATA_API_FORMATTER_RRDSET2JSON_H
web/api/web_api_v1.c
+1
-1
@@ -349,7 +349,7 @@ inline int web_client_api_request_v1_charts(RRDHOST *host, struct web_client *w,
349
350
buffer_flush(w->response.data);
351
w->response.data->contenttype = CT_APPLICATION_JSON;
352
- charts2json(host, w->response.data);
352
+ charts2json(host, w->response.data, 0);
353
return HTTP_RESP_OK;
354
}
355