Count currently streaming senders on the localhost (#13755)
rebased
Emmanuel Vasilakis committed
Oct 13, 2022 at 11:21 UTC
fb86ddd70d14f7943e0314b02235ce3cbca5cb2d
5 files changed
+33
-3
aclk/aclk.c
+7
@@ -1182,3 +1182,10 @@ void add_aclk_host_labels(void) {
1182
rrdlabels_add(labels, "_aclk_available", "false", RRDLABEL_SRC_AUTO|RRDLABEL_SRC_ACLK);
1183
#endif
1184
}
1185
+
1186
+void aclk_queue_node_info(RRDHOST *host) {
1187
+ struct aclk_database_worker_config *wc = (struct aclk_database_worker_config *) host->dbsync_worker;
1188
+ if (likely(wc)) {
1189
+ wc->node_info_send = 1;
1190
+ }
1191
+}
aclk/aclk.h
+1
@@ -54,5 +54,6 @@ void aclk_send_bin_msg(char *msg, size_t msg_len, enum aclk_topics subtopic, con
54
char *aclk_state(void);
55
char *aclk_state_json(void);
56
void add_aclk_host_labels(void);
57
+void aclk_queue_node_info(RRDHOST *host);
58
59
#endif /* ACLK_H */
database/rrd.h
+2
-1
@@ -245,7 +245,7 @@ void rrdlabels_copy(DICTIONARY *dst, DICTIONARY *src);
245
void reload_host_labels(void);
246
void rrdset_update_rrdlabels(RRDSET *st, DICTIONARY *new_rrdlabels);
247
void rrdset_save_rrdlabels_to_sql(RRDSET *st);
248
-
248
+void rrdhost_set_is_parent_label(int count);
249
int rrdlabels_unittest(void);
250
251
// unfortunately this break when defined in exporting_engine.h
@@ -934,6 +934,7 @@ struct rrdhost {
934
time_t senders_connect_time; // the time the last sender was connected
935
time_t senders_last_chart_command; // the time of the last CHART streaming command
936
time_t senders_disconnected_time; // the time the last sender was disconnected
937
+ int senders_count; // number of senders currently streaming
938
939
struct receiver_state *receiver;
940
netdata_mutex_t receiver_lock;
database/rrdhost.c
+17
-1
@@ -1306,12 +1306,28 @@ static void rrdhost_load_auto_labels(void) {
1306
health_add_host_labels();
1307
1308
rrdlabels_add(
1309
- labels, "_is_parent", (rrdhost_hosts_available() > 1 || configured_as_parent()) ? "true" : "false", RRDLABEL_SRC_AUTO);
1309
+ labels, "_is_parent", (localhost->senders_count > 0) ? "true" : "false", RRDLABEL_SRC_AUTO);
1310
1311
if (localhost->rrdpush_send_destination)
1312
rrdlabels_add(labels, "_streams_to", localhost->rrdpush_send_destination, RRDLABEL_SRC_AUTO);
1313
}
1314
1315
+void rrdhost_set_is_parent_label(int count) {
1316
+ DICTIONARY *labels = localhost->rrdlabels;
1317
+
1318
+ if (count == 0 || count == 1) {
1319
+ rrdlabels_add(
1320
+ labels, "_is_parent", (count) ? "true" : "false", RRDLABEL_SRC_AUTO);
1321
+
1322
+ //queue a node info
1323
+#ifdef ENABLE_ACLK
1324
+ if (netdata_cloud_setting) {
1325
+ aclk_queue_node_info(localhost);
1326
+ }
1327
+#endif
1328
+ }
1329
+}
1330
+
1331
static void rrdhost_load_config_labels(void) {
1332
int status = config_load(NULL, 1, CONFIG_SECTION_HOST_LABEL);
1333
if(!status) {
streaming/receiver.c
+6
-1
@@ -744,6 +744,7 @@ static int rrdpush_receive(struct receiver_state *rpt)
744
rpt->host->senders_connect_time = now_realtime_sec();
745
rpt->host->senders_last_chart_command = 0;
746
rpt->host->trigger_chart_obsoletion_check = 1;
747
+
748
rrdhost_unlock(rpt->host);
749
750
// call the plugins.d processor to receive the metrics
@@ -759,6 +760,8 @@ static int rrdpush_receive(struct receiver_state *rpt)
760
aclk_host_state_update(rpt->host, 1);
761
#endif
762
763
+ rrdhost_set_is_parent_label(++localhost->senders_count);
764
+
765
rrdcontext_host_child_connected(rpt->host);
766
767
size_t count = streaming_parser(rpt, &cd, fp_in, fp_out);
@@ -772,11 +775,13 @@ static int rrdpush_receive(struct receiver_state *rpt)
775
776
#ifdef ENABLE_ACLK
777
// in case we have cloud connection we inform cloud
775
- // new child connected
778
+ // a child disconnected
779
if (netdata_cloud_setting)
780
aclk_host_state_update(rpt->host, 0);
781
#endif
782
783
+ rrdhost_set_is_parent_label(--localhost->senders_count);
784
+
785
// During a shutdown there is cleanup code in rrdhost that will cancel the sender thread
786
if (!netdata_exit && rpt->host) {
787
rrd_rdlock();