ACLK Child Availability Messages (#9918)
* new ACLK messages for Claiming MVP1
Timotej S committed
Nov 26, 2020 at 17:26 UTC
f1db235a36d85f81c5a19a7087567405771874f7
13 files changed
+402
-130
aclk/aclk_common.c
-3
@@ -12,9 +12,6 @@ int aclk_disable_runtime = 0;
12
int aclk_kill_link = 0;
13
14
struct aclk_shared_state aclk_shared_state = {
15
- .metadata_submitted = ACLK_METADATA_REQUIRED,
16
- .agent_state = AGENT_INITIALIZING,
17
- .last_popcorn_interrupt = 0,
15
.version_neg = 0,
16
.version_neg_wait_till = 0
17
};
aclk/aclk_common.h
+31
-9
@@ -10,7 +10,7 @@ extern netdata_mutex_t aclk_shared_state_mutex;
10
// minimum and maximum supported version of ACLK
11
// in this version of agent
12
#define ACLK_VERSION_MIN 2
13
-#define ACLK_VERSION_MAX 2
13
+#define ACLK_VERSION_MAX 3
14
15
// Version negotiation messages have they own versioning
16
// this is also used for LWT message as we set that up
@@ -26,7 +26,8 @@ extern netdata_mutex_t aclk_shared_state_mutex;
26
#endif
27
28
// Define ACLK Feature Version Boundaries Here
29
-#define ACLK_V_COMPRESSION 2
29
+#define ACLK_V_COMPRESSION 2
30
+#define ACLK_V_CHILDRENSTATE 3
31
32
typedef enum aclk_cmd {
33
ACLK_CMD_CLOUD,
@@ -35,7 +36,9 @@ typedef enum aclk_cmd {
36
ACLK_CMD_CHART,
37
ACLK_CMD_CHARTDEL,
38
ACLK_CMD_ALARM,
38
- ACLK_CMD_CLOUD_QUERY_2
39
+ ACLK_CMD_CLOUD_QUERY_2,
40
+ ACLK_CMD_CHILD_CONNECT,
41
+ ACLK_CMD_CHILD_DISCONNECT
42
} ACLK_CMD;
43
44
typedef enum aclk_metadata_state {
@@ -45,13 +48,32 @@ typedef enum aclk_metadata_state {
48
} ACLK_METADATA_STATE;
49
50
typedef enum aclk_agent_state {
48
- AGENT_INITIALIZING,
49
- AGENT_STABLE
50
-} ACLK_AGENT_STATE;
51
+ ACLK_HOST_INITIALIZING,
52
+ ACLK_HOST_STABLE
53
+} ACLK_POPCORNING_STATE;
54
+
55
+typedef struct aclk_rrdhost_state {
56
+ char *claimed_id; // Claimed ID if host has one otherwise NULL
57
+
58
+#ifdef ENABLE_ACLK
59
+ // per child popcorning
60
+ ACLK_POPCORNING_STATE state;
61
+ ACLK_METADATA_STATE metadata;
62
+
63
+ time_t timestamp_created;
64
+ time_t t_last_popcorn_update;
65
+#endif
66
+} aclk_rrdhost_state;
67
+
68
+#define ACLK_IS_HOST_INITIALIZING(host) (host->aclk_state.state == ACLK_HOST_INITIALIZING)
69
+#define ACLK_IS_HOST_POPCORNING(host) (ACLK_IS_HOST_INITIALIZING(host) && host->aclk_state.t_last_popcorn_update)
70
+
71
+typedef struct rrdhost RRDHOST;
72
+
73
extern struct aclk_shared_state {
52
- ACLK_METADATA_STATE metadata_submitted;
53
- ACLK_AGENT_STATE agent_state;
54
- time_t last_popcorn_interrupt;
74
+ // optimization to avoid looping trough hosts
75
+ // every time Query Thread wakes up
76
+ RRDHOST *next_popcorn_host;
77
78
// read only while ACLK connected
79
// protect by lock otherwise
aclk/aclk_query.c
+92
-32
@@ -512,6 +512,13 @@ cleanup:
512
return retval;
513
}
514
515
+#define ACLK_HOST_PTR_COMPULSORY(x) \
516
+ if (unlikely(!host)) { \
517
+ errno = 0; \
518
+ error(x " needs host pointer"); \
519
+ break; \
520
+ }
521
+
522
/*
523
* This function will fetch the next pending command and process it
524
*
@@ -546,25 +553,40 @@ static int aclk_process_query(struct aclk_query_thread *t_info)
553
554
switch (this_query->cmd) {
555
case ACLK_CMD_ONCONNECT:
549
- debug(D_ACLK, "EXECUTING on connect metadata command");
550
- ACLK_SHARED_STATE_LOCK;
551
- meta_state = aclk_shared_state.metadata_submitted;
552
- aclk_shared_state.metadata_submitted = ACLK_METADATA_SENT;
553
- ACLK_SHARED_STATE_UNLOCK;
554
- aclk_send_metadata(meta_state);
556
+ ACLK_HOST_PTR_COMPULSORY("ACLK_CMD_ONCONNECT");
557
+#if ACLK_VERSION_MIN < ACLK_V_CHILDRENSTATE
558
+ if (host != localhost && aclk_shared_state.version_neg < ACLK_V_CHILDRENSTATE) {
559
+ error("We are not allowed to send connect message in ACLK version before %d", ACLK_V_CHILDRENSTATE);
560
+ break;
561
+ }
562
+#else
563
+#warning "This check became unnecessary. Remove"
564
+#endif
565
+
566
+ debug(D_ACLK, "EXECUTING on connect metadata command for host \"%s\" GUID \"%s\"",
567
+ host->hostname,
568
+ host->machine_guid);
569
+
570
+ rrdhost_aclk_state_lock(host);
571
+ meta_state = host->aclk_state.metadata;
572
+ host->aclk_state.metadata = ACLK_METADATA_SENT;
573
+ rrdhost_aclk_state_unlock(host);
574
+ aclk_send_metadata(meta_state, host);
575
break;
576
577
case ACLK_CMD_CHART:
578
+ ACLK_HOST_PTR_COMPULSORY("ACLK_CMD_CHART");
579
+
580
debug(D_ACLK, "EXECUTING a chart update command");
559
- if (!host)
560
- fatal("Pointer to host compulsory");
561
- aclk_send_single_chart(host->hostname, this_query->query);
581
+ aclk_send_single_chart(host, this_query->query);
582
break;
583
584
case ACLK_CMD_CHARTDEL:
585
+ ACLK_HOST_PTR_COMPULSORY("ACLK_CMD_CHARTDEL");
586
+
587
debug(D_ACLK, "EXECUTING a chart delete command");
588
//TODO: This send the info metadata for now
567
- aclk_send_info_metadata(ACLK_METADATA_SENT);
589
+ aclk_send_info_metadata(ACLK_METADATA_SENT, host);
590
break;
591
592
case ACLK_CMD_ALARM:
@@ -581,7 +603,19 @@ static int aclk_process_query(struct aclk_query_thread *t_info)
603
aclk_execute_query_v2(this_query);
604
break;
605
606
+ case ACLK_CMD_CHILD_CONNECT:
607
+ case ACLK_CMD_CHILD_DISCONNECT:
608
+ ACLK_HOST_PTR_COMPULSORY("ACLK_CMD_CHILD_CONNECT/ACLK_CMD_CHILD_DISCONNECT");
609
+
610
+ debug(
611
+ D_ACLK, "Execution Child %s command",
612
+ this_query->cmd == ACLK_CMD_CHILD_CONNECT ? "connect" : "disconnect");
613
+ aclk_send_info_child_connection(host, this_query->cmd);
614
+ break;
615
+
616
default:
617
+ errno = 0;
618
+ error("Unknown ACLK Query Command");
619
break;
620
}
621
debug(D_ACLK, "Query #%ld (%s) done", query_count, this_query->topic);
@@ -633,6 +667,39 @@ void aclk_query_threads_start(struct aclk_query_threads *query_threads)
667
}
668
}
669
670
+/**
671
+ * Checks and updates popcorning state of rrdhost
672
+ * returns actual/updated popcorning state
673
+ */
674
+
675
+ACLK_POPCORNING_STATE aclk_host_popcorn_check(RRDHOST *host)
676
+{
677
+ rrdhost_aclk_state_lock(host);
678
+ ACLK_POPCORNING_STATE ret = host->aclk_state.state;
679
+ if (host->aclk_state.state != ACLK_HOST_INITIALIZING){
680
+ rrdhost_aclk_state_unlock(host);
681
+ return ret;
682
+ }
683
+
684
+ if (!host->aclk_state.t_last_popcorn_update){
685
+ rrdhost_aclk_state_unlock(host);
686
+ return ret;
687
+ }
688
+
689
+ time_t t_diff = now_monotonic_sec() - host->aclk_state.t_last_popcorn_update;
690
+
691
+ if (t_diff >= ACLK_STABLE_TIMEOUT) {
692
+ host->aclk_state.state = ACLK_HOST_STABLE;
693
+ host->aclk_state.t_last_popcorn_update = 0;
694
+ rrdhost_aclk_state_unlock(host);
695
+ info("Host \"%s\" stable, ACLK popcorning finished. Last interrupt was %ld seconds ago", host->hostname, t_diff);
696
+ return ACLK_HOST_STABLE;
697
+ }
698
+
699
+ rrdhost_aclk_state_unlock(host);
700
+ return ret;
701
+}
702
+
703
/**
704
* Main query processing thread
705
*
@@ -644,32 +711,14 @@ void aclk_query_threads_start(struct aclk_query_threads *query_threads)
711
void *aclk_query_main_thread(void *ptr)
712
{
713
struct aclk_query_thread *info = ptr;
647
- time_t previous_popcorn_interrupt = 0;
714
715
while (!netdata_exit) {
650
- ACLK_SHARED_STATE_LOCK;
651
- if (aclk_shared_state.agent_state != AGENT_INITIALIZING) {
652
- ACLK_SHARED_STATE_UNLOCK;
653
- break;
654
- }
655
-
656
- time_t checkpoint = now_realtime_sec() - aclk_shared_state.last_popcorn_interrupt;
657
-
658
- if (checkpoint > ACLK_STABLE_TIMEOUT) {
659
- aclk_shared_state.agent_state = AGENT_STABLE;
660
- ACLK_SHARED_STATE_UNLOCK;
661
- info("AGENT stable, last collector initialization activity was %ld seconds ago", checkpoint);
716
+ if(aclk_host_popcorn_check(localhost) == ACLK_HOST_STABLE) {
717
#ifdef ACLK_DEBUG
718
_dump_collector_list();
719
#endif
720
break;
721
}
667
-
668
- if (previous_popcorn_interrupt != aclk_shared_state.last_popcorn_interrupt) {
669
- info("Waiting %ds from this moment for agent collectors to initialize." , ACLK_STABLE_TIMEOUT);
670
- previous_popcorn_interrupt = aclk_shared_state.last_popcorn_interrupt;
671
- }
672
- ACLK_SHARED_STATE_UNLOCK;
722
sleep_usec(USEC_PER_SEC * 1);
723
}
724
@@ -692,15 +741,26 @@ void *aclk_query_main_thread(void *ptr)
741
aclk_shared_state.version_neg = ACLK_VERSION_MIN;
742
aclk_set_rx_handlers(aclk_shared_state.version_neg);
743
}
695
- if (unlikely(aclk_shared_state.metadata_submitted == ACLK_METADATA_REQUIRED)) {
696
- if (unlikely(aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
744
+ ACLK_SHARED_STATE_UNLOCK;
745
+
746
+ rrdhost_aclk_state_lock(localhost);
747
+ if (unlikely(localhost->aclk_state.metadata == ACLK_METADATA_REQUIRED)) {
748
+ if (unlikely(aclk_queue_query("on_connect", localhost, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
749
ACLK_SHARED_STATE_UNLOCK;
750
errno = 0;
751
error("ACLK failed to queue on_connect command");
752
sleep(1);
753
continue;
754
}
703
- aclk_shared_state.metadata_submitted = ACLK_METADATA_CMD_QUEUED;
755
+ localhost->aclk_state.metadata = ACLK_METADATA_CMD_QUEUED;
756
+ }
757
+ rrdhost_aclk_state_unlock(localhost);
758
+
759
+ ACLK_SHARED_STATE_LOCK;
760
+ if (aclk_shared_state.next_popcorn_host && aclk_host_popcorn_check(aclk_shared_state.next_popcorn_host) == ACLK_HOST_STABLE) {
761
+ aclk_queue_query("on_connect", aclk_shared_state.next_popcorn_host, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT);
762
+ aclk_shared_state.next_popcorn_host = NULL;
763
+ aclk_update_next_child_to_popcorn();
764
}
765
ACLK_SHARED_STATE_UNLOCK;
766
aclk/aclk_rx_msgs.c
+4
-4
@@ -38,13 +38,13 @@ static inline int aclk_v2_payload_get_query(const char *payload, struct aclk_req
38
return 0;
39
}
40
41
-#define HTTP_CHECK_AGENT_INITIALIZED() ACLK_SHARED_STATE_LOCK;\
42
- if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {\
41
+#define HTTP_CHECK_AGENT_INITIALIZED() rrdhost_aclk_state_lock(localhost);\
42
+ if (unlikely(localhost->aclk_state.state == ACLK_HOST_INITIALIZING)) {\
43
debug(D_ACLK, "Ignoring \"http\" cloud request; agent not in stable state");\
44
- ACLK_SHARED_STATE_UNLOCK;\
44
+ rrdhost_aclk_state_unlock(localhost);\
45
return 1;\
46
}\
47
- ACLK_SHARED_STATE_UNLOCK;
47
+ rrdhost_aclk_state_unlock(localhost);
48
49
/*
50
* Parse the incoming payload and queue a command if valid
aclk/agent_cloud_link.c
+218
-46
@@ -438,23 +438,136 @@ static struct _collector *_add_collector(const char *hostname, const char *plugi
438
#pragma endregion
439
#endif
440
441
-inline static int aclk_popcorn_check_bump()
441
+/* Avoids the need to scan trough all RRDHOSTS
442
+ * every time any Query Thread Wakes Up
443
+ * (every time we need to check child popcorn expiry)
444
+ * call with ACLK_SHARED_STATE_LOCK held
445
+ */
446
+void aclk_update_next_child_to_popcorn(void)
447
{
448
+ RRDHOST *host;
449
+ int any = 0;
450
+
451
+ rrd_rdlock();
452
+ rrdhost_foreach_read(host) {
453
+ if (unlikely(host == localhost || rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)))
454
+ continue;
455
+
456
+ rrdhost_aclk_state_lock(host);
457
+ if (!ACLK_IS_HOST_POPCORNING(host)) {
458
+ rrdhost_aclk_state_unlock(host);
459
+ continue;
460
+ }
461
+
462
+ any = 1;
463
+
464
+ if (unlikely(!aclk_shared_state.next_popcorn_host)) {
465
+ aclk_shared_state.next_popcorn_host = host;
466
+ rrdhost_aclk_state_unlock(host);
467
+ continue;
468
+ }
469
+
470
+ if (aclk_shared_state.next_popcorn_host->aclk_state.t_last_popcorn_update > host->aclk_state.t_last_popcorn_update)
471
+ aclk_shared_state.next_popcorn_host = host;
472
+
473
+ rrdhost_aclk_state_unlock(host);
474
+ }
475
+ if(!any)
476
+ aclk_shared_state.next_popcorn_host = NULL;
477
+
478
+ rrd_unlock();
479
+}
480
+
481
+/* If popcorning bump timer.
482
+ * If popcorning or initializing (host not stable) return 1
483
+ * Otherwise return 0
484
+ */
485
+static int aclk_popcorn_check_bump(RRDHOST *host)
486
+{
487
+ time_t now = now_monotonic_sec();
488
+ int updated = 0, ret;
489
ACLK_SHARED_STATE_LOCK;
444
- if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
445
- aclk_shared_state.last_popcorn_interrupt = now_realtime_sec();
490
+ rrdhost_aclk_state_lock(host);
491
+
492
+ ret = ACLK_IS_HOST_INITIALIZING(host);
493
+ if (unlikely(ACLK_IS_HOST_POPCORNING(host))) {
494
+ if(now != host->aclk_state.t_last_popcorn_update) {
495
+ updated = 1;
496
+ info("Restarting ACLK popcorn timer for host \"%s\" with GUID \"%s\"", host->hostname, host->machine_guid);
497
+ }
498
+ host->aclk_state.t_last_popcorn_update = now;
499
+ rrdhost_aclk_state_unlock(host);
500
+
501
+ if (host != localhost && updated)
502
+ aclk_update_next_child_to_popcorn();
503
+
504
ACLK_SHARED_STATE_UNLOCK;
447
- return 1;
505
+ return ret;
506
+ }
507
+
508
+ rrdhost_aclk_state_unlock(host);
509
+ ACLK_SHARED_STATE_UNLOCK;
510
+ return ret;
511
+}
512
+
513
+inline static int aclk_host_initializing(RRDHOST *host)
514
+{
515
+ rrdhost_aclk_state_lock(host);
516
+ int ret = ACLK_IS_HOST_INITIALIZING(host);
517
+ rrdhost_aclk_state_unlock(host);
518
+ return ret;
519
+}
520
+
521
+static void aclk_start_host_popcorning(RRDHOST *host)
522
+{
523
+ usec_t now = now_monotonic_sec();
524
+ info("Starting ACLK popcorn timer for host \"%s\" with GUID \"%s\"", host->hostname, host->machine_guid);
525
+ ACLK_SHARED_STATE_LOCK;
526
+ rrdhost_aclk_state_lock(host);
527
+ if (host == localhost && !ACLK_IS_HOST_INITIALIZING(host)) {
528
+ errno = 0;
529
+ error("Localhost is allowed to do popcorning only once after startup!");
530
+ rrdhost_aclk_state_unlock(host);
531
+ ACLK_SHARED_STATE_UNLOCK;
532
+ return;
533
+ }
534
+
535
+ host->aclk_state.state = ACLK_HOST_INITIALIZING;
536
+ host->aclk_state.metadata = ACLK_METADATA_REQUIRED;
537
+ host->aclk_state.t_last_popcorn_update = now;
538
+ rrdhost_aclk_state_unlock(host);
539
+ if (host != localhost)
540
+ aclk_update_next_child_to_popcorn();
541
+ ACLK_SHARED_STATE_UNLOCK;
542
+}
543
+
544
+static void aclk_stop_host_popcorning(RRDHOST *host)
545
+{
546
+ ACLK_SHARED_STATE_LOCK;
547
+ rrdhost_aclk_state_lock(host);
548
+ if (!ACLK_IS_HOST_POPCORNING(host)) {
549
+ rrdhost_aclk_state_unlock(host);
550
+ ACLK_SHARED_STATE_UNLOCK;
551
+ return;
552
+ }
553
+
554
+ info("Host Disconnected before ACLK popcorning finished. Canceling. Host \"%s\" GUID:\"%s\"", host->hostname, host->machine_guid);
555
+ host->aclk_state.t_last_popcorn_update = 0;
556
+ host->aclk_state.metadata = ACLK_METADATA_REQUIRED;
557
+ rrdhost_aclk_state_unlock(host);
558
+
559
+ if(host == aclk_shared_state.next_popcorn_host) {
560
+ aclk_shared_state.next_popcorn_host = NULL;
561
+ aclk_update_next_child_to_popcorn();
562
}
563
ACLK_SHARED_STATE_UNLOCK;
450
- return 0;
564
}
565
566
/*
567
* Add a new collector to the list
568
* If it exists, update the chart count
569
*/
457
-void aclk_add_collector(const char *hostname, const char *plugin_name, const char *module_name)
570
+void aclk_add_collector(RRDHOST *host, const char *plugin_name, const char *module_name)
571
{
572
struct _collector *tmp_collector;
573
if (unlikely(!netdata_ready)) {
@@ -463,7 +576,7 @@ void aclk_add_collector(const char *hostname, const char *plugin_name, const cha
576
577
COLLECTOR_LOCK;
578
466
- tmp_collector = _add_collector(hostname, plugin_name, module_name);
579
+ tmp_collector = _add_collector(host->hostname, plugin_name, module_name);
580
581
if (unlikely(tmp_collector->count != 1)) {
582
COLLECTOR_UNLOCK;
@@ -472,10 +585,10 @@ void aclk_add_collector(const char *hostname, const char *plugin_name, const cha
585
586
COLLECTOR_UNLOCK;
587
475
- if(aclk_popcorn_check_bump())
588
+ if(aclk_popcorn_check_bump(host))
589
return;
590
478
- if (unlikely(aclk_queue_query("collector", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT)))
591
+ if (unlikely(aclk_queue_query("collector", host, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT)))
592
debug(D_ACLK, "ACLK failed to queue on_connect command on collector addition");
593
}
594
@@ -487,7 +600,7 @@ void aclk_add_collector(const char *hostname, const char *plugin_name, const cha
600
* This function will release the memory used and schedule
601
* a cloud update
602
*/
490
-void aclk_del_collector(const char *hostname, const char *plugin_name, const char *module_name)
603
+void aclk_del_collector(RRDHOST *host, const char *plugin_name, const char *module_name)
604
{
605
struct _collector *tmp_collector;
606
if (unlikely(!netdata_ready)) {
@@ -496,7 +609,7 @@ void aclk_del_collector(const char *hostname, const char *plugin_name, const cha
609
610
COLLECTOR_LOCK;
611
499
- tmp_collector = _del_collector(hostname, plugin_name, module_name);
612
+ tmp_collector = _del_collector(host->hostname, plugin_name, module_name);
613
614
if (unlikely(!tmp_collector || tmp_collector->count)) {
615
COLLECTOR_UNLOCK;
@@ -511,10 +624,10 @@ void aclk_del_collector(const char *hostname, const char *plugin_name, const cha
624
625
_free_collector(tmp_collector);
626
514
- if (aclk_popcorn_check_bump())
627
+ if (aclk_popcorn_check_bump(host))
628
return;
629
517
- if (unlikely(aclk_queue_query("collector", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT)))
630
+ if (unlikely(aclk_queue_query("collector", host, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT)))
631
debug(D_ACLK, "ACLK failed to queue on_connect command on collector deletion");
632
}
633
@@ -895,6 +1008,7 @@ void *aclk_main(void *ptr)
1008
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
1009
struct aclk_query_threads query_threads;
1010
struct aclk_stats_thread *stats_thread = NULL;
1011
+ time_t last_periodic_query_wakeup = 0;
1012
1013
query_threads.thread_list = NULL;
1014
@@ -941,7 +1055,8 @@ void *aclk_main(void *ptr)
1055
config_set_number(CONFIG_SECTION_CLOUD, "query thread count", query_threads.count);
1056
}
1057
944
- aclk_shared_state.last_popcorn_interrupt = now_realtime_sec(); // without mutex here because threads are not yet started
1058
+ //start localhost popcorning
1059
+ aclk_start_host_popcorning(localhost);
1060
1061
aclk_stats_enabled = config_get_boolean(CONFIG_SECTION_CLOUD, "statistics", CONFIG_BOOLEAN_YES);
1062
if (aclk_stats_enabled) {
@@ -1051,6 +1166,14 @@ void *aclk_main(void *ptr)
1166
if (unlikely(!query_threads.thread_list)) {
1167
aclk_query_threads_start(&query_threads);
1168
}
1169
+
1170
+ time_t now = now_monotonic_sec();
1171
+ if(aclk_connected && last_periodic_query_wakeup < now) {
1172
+ // to make `aclk_queue_query()` param `run_after` work
1173
+ // also makes per child popcorning work
1174
+ last_periodic_query_wakeup = now;
1175
+ QUERY_THREAD_WAKEUP;
1176
+ }
1177
} // forever
1178
exited:
1179
// Wakeup query thread to cleanup
@@ -1200,9 +1323,9 @@ void aclk_disconnect()
1323
aclk_stats_upd_online(0);
1324
1325
aclk_subscribed = 0;
1203
- ACLK_SHARED_STATE_LOCK;
1204
- aclk_shared_state.metadata_submitted = ACLK_METADATA_REQUIRED;
1205
- ACLK_SHARED_STATE_UNLOCK;
1326
+ rrdhost_aclk_state_lock(localhost);
1327
+ localhost->aclk_state.metadata = ACLK_METADATA_REQUIRED;
1328
+ rrdhost_aclk_state_unlock(localhost);
1329
aclk_connected = 0;
1330
aclk_connecting = 0;
1331
aclk_force_reconnect = 1;
@@ -1294,7 +1417,7 @@ void aclk_send_alarm_metadata(ACLK_METADATA_STATE metadata_submitted)
1417
* /api/v1/info
1418
* charts
1419
*/
1297
-int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted)
1420
+int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted, RRDHOST *host)
1421
{
1422
BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1423
@@ -1315,11 +1438,11 @@ int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted)
1438
buffer_strcat(local_buffer, ",\n\t\"payload\": ");
1439
1440
buffer_sprintf(local_buffer, "{\n\t \"info\" : ");
1318
- web_client_api_request_v1_info_fill_buffer(localhost, local_buffer);
1441
+ web_client_api_request_v1_info_fill_buffer(host, local_buffer);
1442
debug(D_ACLK, "Metadata %s with info has %zu bytes", msg_id, local_buffer->len);
1443
1444
buffer_sprintf(local_buffer, ", \n\t \"charts\" : ");
1322
- charts2json(localhost, local_buffer, 1, 0);
1445
+ charts2json(host, local_buffer, 1, 0);
1446
buffer_sprintf(local_buffer, "\n}\n}");
1447
debug(D_ACLK, "Metadata %s with chart has %zu bytes", msg_id, local_buffer->len);
1448
@@ -1330,6 +1453,66 @@ int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted)
1453
return 0;
1454
}
1455
1456
+int aclk_send_info_child_connection(RRDHOST *host, ACLK_CMD cmd)
1457
+{
1458
+ BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1459
+ local_buffer->contenttype = CT_APPLICATION_JSON;
1460
+
1461
+ if(aclk_shared_state.version_neg < ACLK_V_CHILDRENSTATE)
1462
+ fatal("This function should not be called if ACLK version is less than %d (current %d)", ACLK_V_CHILDRENSTATE, aclk_shared_state.version_neg);
1463
+
1464
+ debug(D_ACLK, "Sending Child Disconnect");
1465
+
1466
+ char *msg_id = create_uuid();
1467
+
1468
+ aclk_create_header(local_buffer, cmd == ACLK_CMD_CHILD_CONNECT ? "child_connect" : "child_disconnect", msg_id, 0, 0, aclk_shared_state.version_neg);
1469
+
1470
+ buffer_strcat(local_buffer, ",\"payload\":");
1471
+
1472
+ buffer_sprintf(local_buffer, "{\"guid\":\"%s\",\"claim_id\":", host->machine_guid);
1473
+ rrdhost_aclk_state_lock(host);
1474
+ if(host->aclk_state.claimed_id)
1475
+ buffer_sprintf(local_buffer, "\"%s\"}}", host->aclk_state.claimed_id);
1476
+ else
1477
+ buffer_strcat(local_buffer, "null}}");
1478
+
1479
+ rrdhost_aclk_state_unlock(host);
1480
+
1481
+ aclk_send_message(ACLK_METADATA_TOPIC, local_buffer->buffer, msg_id);
1482
+
1483
+ freez(msg_id);
1484
+ buffer_free(local_buffer);
1485
+ return 0;
1486
+}
1487
+
1488
+void aclk_host_state_update(RRDHOST *host, ACLK_CMD cmd)
1489
+{
1490
+#if ACLK_VERSION_MIN < ACLK_V_CHILDRENSTATE
1491
+ if (aclk_shared_state.version_neg < ACLK_V_CHILDRENSTATE)
1492
+ return;
1493
+#else
1494
+#warning "This check became unnecessary. Remove"
1495
+#endif
1496
+
1497
+ if (unlikely(aclk_host_initializing(localhost)))
1498
+ return;
1499
+
1500
+ switch (cmd) {
1501
+ case ACLK_CMD_CHILD_CONNECT:
1502
+ debug(D_ACLK, "Child Connected %s %s.", host->hostname, host->machine_guid);
1503
+ aclk_start_host_popcorning(host);
1504
+ aclk_queue_query("add_child", host, NULL, NULL, 0, 1, ACLK_CMD_CHILD_CONNECT);
1505
+ break;
1506
+ case ACLK_CMD_CHILD_DISCONNECT:
1507
+ debug(D_ACLK, "Child Disconnected %s %s.", host->hostname, host->machine_guid);
1508
+ aclk_stop_host_popcorning(host);
1509
+ aclk_queue_query("del_child", host, NULL, NULL, 0, 1, ACLK_CMD_CHILD_DISCONNECT);
1510
+ break;
1511
+ default:
1512
+ error("Unknown command for aclk_host_state_update %d.", (int)cmd);
1513
+ }
1514
+}
1515
+
1516
void aclk_send_stress_test(size_t size)
1517
{
1518
char *buffer = mallocz(size);
@@ -1351,11 +1534,12 @@ void aclk_send_stress_test(size_t size)
1534
1535
// Send info metadata message to the cloud if the link is established
1536
// or on request
1354
-int aclk_send_metadata(ACLK_METADATA_STATE state)
1537
+int aclk_send_metadata(ACLK_METADATA_STATE state, RRDHOST *host)
1538
{
1539
+ aclk_send_info_metadata(state, host);
1540
1357
- aclk_send_info_metadata(state);
1358
- aclk_send_alarm_metadata(state);
1541
+ if(host == localhost)
1542
+ aclk_send_alarm_metadata(state);
1543
1544
return 0;
1545
}
@@ -1373,15 +1557,10 @@ void aclk_single_update_enable()
1557
// Trigged by a health reload, sends the alarm metadata
1558
void aclk_alarm_reload()
1559
{
1376
-
1377
- ACLK_SHARED_STATE_LOCK;
1378
- if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
1379
- ACLK_SHARED_STATE_UNLOCK;
1560
+ if (unlikely(aclk_host_initializing(localhost)))
1561
return;
1381
- }
1382
- ACLK_SHARED_STATE_UNLOCK;
1562
1384
- if (unlikely(aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
1563
+ if (unlikely(aclk_queue_query("on_connect", localhost, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
1564
if (likely(aclk_connected)) {
1565
errno = 0;
1566
error("ACLK failed to queue on_connect command on alarm reload");
@@ -1390,17 +1569,11 @@ void aclk_alarm_reload()
1569
}
1570
//rrd_stats_api_v1_chart(RRDSET *st, BUFFER *buf)
1571
1393
-int aclk_send_single_chart(char *hostname, char *chart)
1572
+int aclk_send_single_chart(RRDHOST *host, char *chart)
1573
{
1395
- RRDHOST *target_host;
1396
-
1397
- target_host = rrdhost_find_by_hostname(hostname, 0);
1398
- if (!target_host)
1399
- return 1;
1400
-
1401
- RRDSET *st = rrdset_find(target_host, chart);
1574
+ RRDSET *st = rrdset_find(host, chart);
1575
if (!st)
1403
- st = rrdset_find_byname(target_host, chart);
1576
+ st = rrdset_find_byname(host, chart);
1577
if (!st) {
1578
info("FAILED to find chart %s", chart);
1579
return 1;
@@ -1437,13 +1610,16 @@ int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd)
1610
if (!netdata_cloud_setting)
1611
return 0;
1612
1440
- if (host != localhost)
1613
+ if (aclk_shared_state.version_neg < ACLK_V_CHILDRENSTATE && host != localhost)
1614
+ return 0;
1615
+
1616
+ if (aclk_host_initializing(localhost))
1617
return 0;
1618
1619
if (unlikely(aclk_disable_single_updates))
1620
return 0;
1621
1446
- if (aclk_popcorn_check_bump())
1622
+ if (aclk_popcorn_check_bump(host))
1623
return 0;
1624
1625
if (unlikely(aclk_queue_query("_chart", host, NULL, chart_name, 0, 1, aclk_cmd))) {
@@ -1467,12 +1643,8 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1643
if (host != localhost)
1644
return 0;
1645
1470
- ACLK_SHARED_STATE_LOCK;
1471
- if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
1472
- ACLK_SHARED_STATE_UNLOCK;
1646
+ if(unlikely(aclk_host_initializing(localhost)))
1647
return 0;
1474
- }
1475
- ACLK_SHARED_STATE_UNLOCK;
1648
1649
/*
1650
* Check if individual updates have been disabled
aclk/agent_cloud_link.h
+9
-5
@@ -66,24 +66,28 @@ int cloud_to_agent_parse(JSON_ENTRY *e);
66
void aclk_disconnect();
67
void aclk_connect();
68
69
-int aclk_send_metadata(ACLK_METADATA_STATE state);
70
-int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted);
69
+int aclk_send_metadata(ACLK_METADATA_STATE state, RRDHOST *host);
70
+int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted, RRDHOST *host);
71
void aclk_send_alarm_metadata(ACLK_METADATA_STATE metadata_submitted);
72
73
int aclk_wait_for_initialization();
74
char *create_publish_base_topic();
75
76
-int aclk_send_single_chart(char *host, char *chart);
76
+int aclk_send_single_chart(RRDHOST *host, char *chart);
77
int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd);
78
int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae);
79
void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us, int version);
80
int aclk_handle_cloud_message(char *payload);
81
-void aclk_add_collector(const char *hostname, const char *plugin_name, const char *module_name);
82
-void aclk_del_collector(const char *hostname, const char *plugin_name, const char *module_name);
81
+void aclk_add_collector(RRDHOST *host, const char *plugin_name, const char *module_name);
82
+void aclk_del_collector(RRDHOST *host, const char *plugin_name, const char *module_name);
83
void aclk_alarm_reload();
84
unsigned long int aclk_reconnect_delay(int mode);
85
extern void health_alarm_entry2json_nolock(BUFFER *wb, ALARM_ENTRY *ae, RRDHOST *host);
86
void aclk_single_update_enable();
87
void aclk_single_update_disable();
88
89
+void aclk_host_state_update(RRDHOST *host, ACLK_CMD cmd);
90
+int aclk_send_info_child_connection(RRDHOST *host, ACLK_CMD cmd);
91
+void aclk_update_next_child_to_popcorn(void);
92
+
93
#endif //NETDATA_AGENT_CLOUD_LINK_H
claim/claim.c
+9
-9
@@ -34,9 +34,9 @@ static char *claiming_errors[] = {
34
char *is_agent_claimed()
35
{
36
char *result;
37
- netdata_mutex_lock(&localhost->claimed_id_lock);
38
- result = (localhost->claimed_id == NULL) ? NULL : strdupz(localhost->claimed_id);
39
- netdata_mutex_unlock(&localhost->claimed_id_lock);
37
+ rrdhost_aclk_state_lock(localhost);
38
+ result = (localhost->aclk_state.claimed_id == NULL) ? NULL : strdupz(localhost->aclk_state.claimed_id);
39
+ rrdhost_aclk_state_unlock(localhost);
40
return result;
41
}
42
@@ -134,10 +134,10 @@ void load_claiming_state(void)
134
netdata_cloud_setting = 0;
135
#else
136
uuid_t uuid;
137
- netdata_mutex_lock(&localhost->claimed_id_lock);
138
- if (localhost->claimed_id) {
139
- freez(localhost->claimed_id);
140
- localhost->claimed_id = NULL;
137
+ rrdhost_aclk_state_lock(localhost);
138
+ if (localhost->aclk_state.claimed_id) {
139
+ freez(localhost->aclk_state.claimed_id);
140
+ localhost->aclk_state.claimed_id = NULL;
141
}
142
if (aclk_connected)
143
{
@@ -159,8 +159,8 @@ void load_claiming_state(void)
159
freez(claimed_id);
160
claimed_id = NULL;
161
}
162
- localhost->claimed_id = claimed_id;
163
- netdata_mutex_unlock(&localhost->claimed_id_lock);
162
+ localhost->aclk_state.claimed_id = claimed_id;
163
+ rrdhost_aclk_state_unlock(localhost);
164
if (!claimed_id) {
165
info("Unable to load '%s', setting state to AGENT_UNCLAIMED", filename);
166
return;
database/rrd.h
+6
-2
@@ -33,6 +33,7 @@ struct pg_cache_page_index;
33
#include "rrdcalc.h"
34
#include "rrdcalctemplate.h"
35
#include "../streaming/rrdpush.h"
36
+#include "../aclk/aclk_common.h"
37
38
struct context_param {
39
RRDDIM *rd;
@@ -825,8 +826,8 @@ struct rrdhost {
826
struct netdata_ssl stream_ssl; //Structure used to encrypt the stream
827
#endif
828
828
- netdata_mutex_t claimed_id_lock;
829
- char *claimed_id; // Claimed ID if host has one otherwise NULL
829
+ netdata_mutex_t aclk_state_lock;
830
+ aclk_rrdhost_state aclk_state;
831
832
struct rrdhost *next;
833
};
@@ -836,6 +837,9 @@ extern RRDHOST *localhost;
837
#define rrdhost_wrlock(host) netdata_rwlock_wrlock(&((host)->rrdhost_rwlock))
838
#define rrdhost_unlock(host) netdata_rwlock_unlock(&((host)->rrdhost_rwlock))
839
840
+#define rrdhost_aclk_state_lock(host) netdata_mutex_lock(&((host)->aclk_state_lock))
841
+#define rrdhost_aclk_state_unlock(host) netdata_mutex_unlock(&((host)->aclk_state_lock))
842
+
843
// ----------------------------------------------------------------------------
844
// these loop macros make sure the linked list is accessed with the right lock
845
database/rrdhost.c
+3
-3
@@ -189,7 +189,7 @@ RRDHOST *rrdhost_create(const char *hostname,
189
netdata_rwlock_init(&host->rrdhost_rwlock);
190
netdata_rwlock_init(&host->labels_rwlock);
191
192
- netdata_mutex_init(&host->claimed_id_lock);
192
+ netdata_mutex_init(&host->aclk_state_lock);
193
194
host->system_info = system_info;
195
@@ -870,8 +870,8 @@ void rrdhost_free(RRDHOST *host) {
870
// ------------------------------------------------------------------------
871
// free it
872
873
- pthread_mutex_destroy(&host->claimed_id_lock);
874
- freez(host->claimed_id);
873
+ pthread_mutex_destroy(&host->aclk_state_lock);
874
+ freez(host->aclk_state.claimed_id);
875
freez((void *)host->tags);
876
free_host_labels(host->labels);
877
freez((void *)host->os);
database/rrdset.c
+5
-5
@@ -442,7 +442,7 @@ void rrdset_delete_custom(RRDSET *st, int db_rotated) {
442
recursively_delete_dir(st->cache_dir, "left-over chart");
443
#ifdef ENABLE_ACLK
444
if ((netdata_cloud_setting) && (db_rotated || RRD_MEMORY_MODE_DBENGINE != st->rrd_memory_mode)) {
445
- aclk_del_collector(st->rrdhost->hostname, st->plugin_name, st->module_name);
445
+ aclk_del_collector(st->rrdhost, st->plugin_name, st->module_name);
446
aclk_update_chart(st->rrdhost, st->id, ACLK_CMD_CHARTDEL);
447
}
448
#endif
@@ -626,14 +626,14 @@ RRDSET *rrdset_create_custom(
626
#ifdef ENABLE_ACLK
627
if (netdata_cloud_setting) {
628
if (mark_rebuild & META_CHART_ACTIVATED) {
629
- aclk_add_collector(host->hostname, st->plugin_name, st->module_name);
629
+ aclk_add_collector(host, st->plugin_name, st->module_name);
630
}
631
else {
632
if (mark_rebuild & (META_PLUGIN_UPDATED | META_MODULE_UPDATED)) {
633
aclk_del_collector(
634
- host->hostname, mark_rebuild & META_PLUGIN_UPDATED ? old_plugin : st->plugin_name,
634
+ host, mark_rebuild & META_PLUGIN_UPDATED ? old_plugin : st->plugin_name,
635
mark_rebuild & META_MODULE_UPDATED ? old_module : st->module_name);
636
- aclk_add_collector(host->hostname, st->plugin_name, st->module_name);
636
+ aclk_add_collector(host, st->plugin_name, st->module_name);
637
}
638
}
639
aclk_update_chart(host, st->id, ACLK_CMD_CHART);
@@ -933,7 +933,7 @@ RRDSET *rrdset_create_custom(
933
rrdhost_unlock(host);
934
#ifdef ENABLE_ACLK
935
if (netdata_cloud_setting) {
936
- aclk_add_collector(host->hostname, plugin, module);
936
+ aclk_add_collector(host, plugin, module);
937
aclk_update_chart(host, st->id, ACLK_CMD_CHART);
938
}
939
#endif
streaming/receiver.c
+18
-5
@@ -126,11 +126,11 @@ PARSER_RC streaming_claimed_id(char **words, void *user, PLUGINSD_ACTION *plugin
126
return PARSER_RC_OK; //the message is OK problem must be somewehere else
127
}
128
129
- netdata_mutex_lock(&host->claimed_id_lock);
130
- if (host->claimed_id)
131
- freez(host->claimed_id);
132
- host->claimed_id = strcmp(words[2], "NULL") ? strdupz(words[2]) : NULL;
133
- netdata_mutex_unlock(&host->claimed_id_lock);
129
+ rrdhost_aclk_state_lock(host);
130
+ if (host->aclk_state.claimed_id)
131
+ freez(host->aclk_state.claimed_id);
132
+ host->aclk_state.claimed_id = strcmp(words[2], "NULL") ? strdupz(words[2]) : NULL;
133
+ rrdhost_aclk_state_unlock(host);
134
135
rrdpush_claimed_id(host);
136
@@ -440,6 +440,12 @@ static int rrdpush_receive(struct receiver_state *rpt)
440
441
cd.version = rpt->stream_version;
442
443
+#ifdef ENABLE_ACLK
444
+ // in case we have cloud connection we inform cloud
445
+ // new slave connected
446
+ if (netdata_cloud_setting)
447
+ aclk_host_state_update(rpt->host, ACLK_CMD_CHILD_CONNECT);
448
+#endif
449
450
size_t count = streaming_parser(rpt, &cd, fp);
451
@@ -448,6 +454,13 @@ static int rrdpush_receive(struct receiver_state *rpt)
454
error("STREAM %s [receive from [%s]:%s]: disconnected (completed %zu updates).", rpt->hostname, rpt->client_ip,
455
rpt->client_port, count);
456
457
+#ifdef ENABLE_ACLK
458
+ // in case we have cloud connection we inform cloud
459
+ // new slave connected
460
+ if (netdata_cloud_setting)
461
+ aclk_host_state_update(rpt->host, ACLK_CMD_CHILD_DISCONNECT);
462
+#endif
463
+
464
// During a shutdown there is cleanup code in rrdhost that will cancel the sender thread
465
if (!netdata_exit && rpt->host) {
466
rrd_rdlock();
streaming/rrdpush.c
+3
-3
@@ -374,11 +374,11 @@ void rrdpush_claimed_id(RRDHOST *host)
374
return;
375
376
sender_start(host->sender);
377
- netdata_mutex_lock(&host->claimed_id_lock);
377
+ rrdhost_aclk_state_lock(host);
378
379
- buffer_sprintf(host->sender->build, "CLAIMED_ID %s %s\n", host->machine_guid, (host->claimed_id ? host->claimed_id : "NULL") );
379
+ buffer_sprintf(host->sender->build, "CLAIMED_ID %s %s\n", host->machine_guid, (host->aclk_state.claimed_id ? host->aclk_state.claimed_id : "NULL") );
380
381
- netdata_mutex_unlock(&host->claimed_id_lock);
381
+ rrdhost_aclk_state_unlock(host);
382
sender_commit(host->sender);
383
384
// signal the sender there are more data
web/api/web_api_v1.c
+4
-4
@@ -841,12 +841,12 @@ static inline void web_client_api_request_v1_info_mirrored_hosts(BUFFER *wb) {
841
(host->receiver || host == localhost) ? "true" : "false");
842
netdata_mutex_unlock(&host->receiver_lock);
843
844
- netdata_mutex_lock(&host->claimed_id_lock);
845
- if (host->claimed_id)
846
- buffer_sprintf(wb, "\"%s\" }", host->claimed_id);
844
+ rrdhost_aclk_state_lock(host);
845
+ if (host->aclk_state.claimed_id)
846
+ buffer_sprintf(wb, "\"%s\" }", host->aclk_state.claimed_id);
847
else
848
buffer_strcat(wb, "null }");
849
- netdata_mutex_unlock(&host->claimed_id_lock);
849
+ rrdhost_aclk_state_unlock(host);
850
851
count++;
852
}