ACLK new cloud architecture new TBEB (#10941)
* new TBEB impl. honoring new cloud architecture requirements * handle error cases during env/passwd/challenge as per spec of new cloud architecture
Timotej S committed
Apr 26, 2021 at 10:32 UTC
690df2de3b2c8631b5734290dff1c8958c0711f7
5 files changed
+218
-23
aclk/aclk.c
+32
-5
@@ -29,6 +29,8 @@ int aclk_kill_link = 0;
29
30
int aclk_pubacks_per_conn = 0; // How many PubAcks we got since MQTT conn est.
31
32
+time_t aclk_block_until = 0;
33
+
34
usec_t aclk_session_us = 0; // Used by the mqtt layer
35
time_t aclk_session_sec = 0; // Used by the mqtt layer
36
@@ -241,7 +243,7 @@ static void msg_callback(const char *topic, const void *msg, size_t msglen, int
243
static void puback_callback(uint16_t packet_id)
244
{
245
if (++aclk_pubacks_per_conn == ACLK_PUBACKS_CONN_STABLE)
244
- aclk_reconnect_delay(0);
246
+ aclk_tbeb_reset();
247
248
#ifdef NETDATA_INTERNAL_CHECKS
249
aclk_stats_msg_puback(packet_id);
@@ -404,16 +406,41 @@ void aclk_graceful_disconnect(mqtt_wss_client client)
406
mqtt_wss_disconnect(client, 1000);
407
}
408
409
+static unsigned long aclk_reconnect_delay() {
410
+ unsigned long recon_delay;
411
+ time_t now;
412
+
413
+ if (aclk_disable_runtime) {
414
+ aclk_tbeb_reset();
415
+ return 60 * MSEC_PER_SEC;
416
+ }
417
+
418
+ now = now_monotonic_sec();
419
+ if (aclk_block_until) {
420
+ if (now < aclk_block_until) {
421
+ recon_delay = aclk_block_until - now;
422
+ recon_delay *= MSEC_PER_SEC;
423
+ aclk_block_until = 0;
424
+ aclk_tbeb_reset();
425
+ return recon_delay;
426
+ }
427
+ aclk_block_until = 0;
428
+ }
429
+
430
+ if (!aclk_env || !aclk_env->backoff.base)
431
+ return aclk_tbeb_delay(0, 2, 0, 1024);
432
+
433
+ return aclk_tbeb_delay(0, aclk_env->backoff.base, aclk_env->backoff.min_s, aclk_env->backoff.max_s);
434
+}
435
+
436
/* Block till aclk_reconnect_delay is satisifed or netdata_exit is signalled
437
* @return 0 - Go ahead and connect (delay expired)
438
* 1 - netdata_exit
439
*/
440
#define NETDATA_EXIT_POLL_MS (MSEC_PER_SEC/4)
441
static int aclk_block_till_recon_allowed() {
413
- // Handle reconnect exponential backoff
414
- // fnc aclk_reconnect_delay comes from ACLK Legacy @amoss
415
- // but has been modifed slightly (more randomness)
416
- unsigned long recon_delay = aclk_reconnect_delay(1);
442
+ unsigned long recon_delay = aclk_reconnect_delay();
443
+
444
info("Wait before attempting to reconnect in %.3f seconds\n", recon_delay / (float)MSEC_PER_SEC);
445
// we want to wake up from time to time to check netdata_exit
446
while (recon_delay)
aclk/aclk.h
+2
@@ -25,6 +25,8 @@ extern int aclk_disable_single_updates;
25
extern int aclk_kill_link;
26
extern int aclk_connected;
27
28
+extern time_t aclk_block_until;
29
+
30
extern usec_t aclk_session_us;
31
extern time_t aclk_session_sec;
32
aclk/aclk_otp.c
+159
@@ -275,6 +275,161 @@ exit:
275
return rc;
276
}
277
278
+#define JSON_KEY_ERTRY "errorNonRetryable"
279
+#define JSON_KEY_EDELAY "errorRetryDelaySeconds"
280
+#define JSON_KEY_EEC "errorCode"
281
+#define JSON_KEY_EMSGKEY "errorMsgKey"
282
+#define JSON_KEY_EMSG "errorMessage"
283
+#if JSON_C_MINOR_VERSION >= 13
284
+static const char *get_json_str_by_path(json_object *json, const char *path) {
285
+ json_object *ptr;
286
+ if (json_pointer_get(json, path, &ptr)) {
287
+ error("Missing compulsory key \"%s\" in error response", path);
288
+ return NULL;
289
+ }
290
+ if (json_object_get_type(ptr) != json_type_string) {
291
+ error("Value of Key \"%s\" in error response should be string", path);
292
+ return NULL;
293
+ }
294
+ return json_object_get_string(ptr);
295
+}
296
+
297
+static int aclk_parse_otp_error(const char *json_str) {
298
+ int rc = 1;
299
+ json_object *json, *ptr;
300
+ const char *ec;
301
+ const char *ek;
302
+ const char *emsg;
303
+ int block_retry = -1, backoff = -1;
304
+
305
+
306
+ json = json_tokener_parse(json_str);
307
+ if (!json) {
308
+ error("JSON-C failed to parse the payload of http response of /env endpoint");
309
+ return 1;
310
+ }
311
+
312
+ if ((ec = get_json_str_by_path(json, "/" JSON_KEY_EEC)) == NULL)
313
+ goto exit;
314
+
315
+ if ((ek = get_json_str_by_path(json, "/" JSON_KEY_EMSGKEY)) == NULL)
316
+ goto exit;
317
+
318
+ if ((emsg = get_json_str_by_path(json, "/" JSON_KEY_EMSG)) == NULL)
319
+ goto exit;
320
+
321
+ // optional field
322
+ if (!json_pointer_get(json, "/" JSON_KEY_ERTRY, &ptr)) {
323
+ if (json_object_get_type(ptr) != json_type_boolean) {
324
+ error("Error response Key " "/" JSON_KEY_ERTRY " should be of boolean type");
325
+ goto exit;
326
+ }
327
+ block_retry = json_object_get_boolean(ptr);
328
+ }
329
+
330
+ // optional field
331
+ if (!json_pointer_get(json, "/" JSON_KEY_EDELAY, &ptr)) {
332
+ if (json_object_get_type(ptr) != json_type_int) {
333
+ error("Error response Key " "/" JSON_KEY_EDELAY " should be of integer type");
334
+ goto exit;
335
+ }
336
+ backoff = json_object_get_int(ptr);
337
+ }
338
+
339
+ if (block_retry > 0)
340
+ aclk_disable_runtime = 1;
341
+
342
+ if (backoff > 0)
343
+ aclk_block_until = now_monotonic_sec() + backoff;
344
+
345
+ error("Cloud returned EC=\"%s\", Msg-Key:\"%s\", Msg:\"%s\", BlockRetry:%s, Backoff:%ds (-1 unset by cloud)", ec, ek, emsg, block_retry > 0 ? "true" : "false", backoff);
346
+ rc = 0;
347
+exit:
348
+ json_object_put(json);
349
+ return rc;
350
+}
351
+#else
352
+static int aclk_parse_otp_error(const char *json_str) {
353
+ int rc = 1;
354
+ int block_retry = -1, backoff = -1;
355
+
356
+ const char *ec = NULL;
357
+ const char *ek = NULL;
358
+ const char *emsg = NULL;
359
+
360
+ json_object *json;
361
+ struct json_object_iterator it;
362
+ struct json_object_iterator itEnd;
363
+
364
+ json = json_tokener_parse(json_str);
365
+ if (!json) {
366
+ error("JSON-C failed to parse the payload of http respons of /env endpoint");
367
+ return 1;
368
+ }
369
+
370
+ it = json_object_iter_begin(json);
371
+ itEnd = json_object_iter_end(json);
372
+
373
+ while (!json_object_iter_equal(&it, &itEnd)) {
374
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_EMSG)) {
375
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_string, JSON_KEY_EMSG)
376
+
377
+ emsg = json_object_get_string(json_object_iter_peek_value(&it));
378
+ json_object_iter_next(&it);
379
+ continue;
380
+ }
381
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_EMSGKEY)) {
382
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_string, JSON_KEY_EMSGKEY)
383
+
384
+ ek = json_object_get_string(json_object_iter_peek_value(&it));
385
+ json_object_iter_next(&it);
386
+ continue;
387
+ }
388
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_EEC)) {
389
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_string, JSON_KEY_EEC)
390
+
391
+ ec = strdupz(json_object_get_string(json_object_iter_peek_value(&it)));
392
+ json_object_iter_next(&it);
393
+ continue;
394
+ }
395
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_EDELAY)) {
396
+ if (json_object_get_type(json_object_iter_peek_value(&it)) != json_type_int) {
397
+ error("value of key " JSON_KEY_EDELAY " should be integer");
398
+ goto exit;
399
+ }
400
+
401
+ backoff = json_object_get_int(json_object_iter_peek_value(&it));
402
+ json_object_iter_next(&it);
403
+ continue;
404
+ }
405
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_ERTRY)) {
406
+ if (json_object_get_type(json_object_iter_peek_value(&it)) != json_type_boolean) {
407
+ error("value of key " JSON_KEY_ERTRY " should be integer");
408
+ goto exit;
409
+ }
410
+
411
+ block_retry = json_object_get_boolean(json_object_iter_peek_value(&it));
412
+ json_object_iter_next(&it);
413
+ continue;
414
+ }
415
+ error("Unknown key \"%s\" in error response payload. Ignoring", json_object_iter_peek_name(&it));
416
+ json_object_iter_next(&it);
417
+ }
418
+
419
+ if (block_retry > 0)
420
+ aclk_disable_runtime = 1;
421
+
422
+ if (backoff > 0)
423
+ aclk_block_until = now_monotonic_sec() + backoff;
424
+
425
+ error("Cloud returned EC=\"%s\", Msg-Key:\"%s\", Msg:\"%s\", BlockRetry:%s, Backoff:%ds (-1 unset by cloud)", ec, ek, emsg, block_retry > 0 ? "true" : "false", backoff);
426
+ rc = 0;
427
+exit:
428
+ json_object_put(json);
429
+ return rc;
430
+}
431
+#endif
432
+
433
#define OTP_URL_PREFIX "/api/v1/auth/node/"
434
int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target) {
435
// TODO this fnc will be rewritten and simplified in following PRs
@@ -304,6 +459,8 @@ int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_p
459
}
460
if (resp.http_code != 200) {
461
error ("ACLK_OTP Challenge HTTP code not 200 OK (got %d)", resp.http_code);
462
+ if (resp.payload_size)
463
+ aclk_parse_otp_error(resp.payload);
464
goto cleanup_resp;
465
}
466
info ("ACLK_OTP Got Challenge from Cloud");
@@ -355,6 +512,8 @@ int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_p
512
}
513
if (resp.http_code != 201) {
514
error ("ACLK_OTP Password HTTP code not 201 Created (got %d)", resp.http_code);
515
+ if (resp.payload_size)
516
+ aclk_parse_otp_error(resp.payload);
517
goto cleanup_resp;
518
}
519
info ("ACLK_OTP Got Password from Cloud");
aclk/aclk_util.c
+23
-17
@@ -279,33 +279,39 @@ const char *aclk_get_topic(enum aclk_topics topic)
279
/*
280
* TBEB with randomness
281
*
282
- * @param mode 0 - to reset the delay,
283
- * 1 - to advance a step and calculate sleep time [0 .. ACLK_MAX_BACKOFF_DELAY * 1000] ms
282
+ * @param reset 1 - to reset the delay,
283
+ * 0 - to advance a step and calculate sleep time in ms
284
+ * @param min, max in seconds
285
* @returns delay in ms
286
*
287
*/
287
-#define ACLK_MAX_BACKOFF_DELAY 1024
288
-unsigned long int aclk_reconnect_delay(int mode)
289
-{
290
- static int fail = -1;
291
- unsigned long int delay;
288
293
- if (!mode || fail == -1) {
294
- srandom(time(NULL));
295
- fail = mode - 1;
289
+unsigned long int aclk_tbeb_delay(int reset, int base, unsigned long int min, unsigned long int max) {
290
+ static int attempt = -1;
291
+
292
+ if (reset) {
293
+ attempt = -1;
294
return 0;
295
}
296
299
- delay = (1 << fail);
297
+ attempt++;
298
301
- if (delay >= ACLK_MAX_BACKOFF_DELAY) {
302
- delay = ACLK_MAX_BACKOFF_DELAY * 1000;
303
- } else {
304
- fail++;
305
- delay *= 1000;
306
- delay += (random() % (MAX(1000, delay/2)));
299
+ if (attempt == 0) {
300
+ srandom(time(NULL));
301
+ return 0;
302
}
303
304
+ unsigned long int delay = pow(base, attempt - 1);
305
+ delay *= MSEC_PER_SEC;
306
+
307
+ delay += (random() % (MAX(1000, delay/2)));
308
+
309
+ if (delay <= min * MSEC_PER_SEC)
310
+ return min;
311
+
312
+ if (delay >= max * MSEC_PER_SEC)
313
+ return max;
314
+
315
return delay;
316
}
317
aclk/aclk_util.h
+2
-1
@@ -75,7 +75,8 @@ int aclk_get_conv_log_next();
75
#endif
76
#endif
77
78
-unsigned long int aclk_reconnect_delay(int mode);
78
+unsigned long int aclk_tbeb_delay(int reset, int base, unsigned long int min, unsigned long int max);
79
+#define aclk_tbeb_reset(x) aclk_tbeb_delay(1, 0, 0, 0)
80
81
typedef enum aclk_proxy_type {
82
PROXY_TYPE_UNKNOWN = 0,