Add support for acquire/release operations on RRDSETs (#14945)
* Add acquire/release support for RRDSET * Release/acquire RRDSET when training. * Fix function name in log message. * Use proper function name to get the hostname. * Add acquire/release for hosts and skip training orphan/obsolete hosts/charts. * Fix variable name
vkalintiris committed
Apr 26, 2023 at 11:37 UTC
a58101434cabf862e661a2e4e72471c05b6bd09c
6 files changed
+95
-28
database/rrd.h
+9
@@ -1260,6 +1260,10 @@ extern RRDHOST *localhost;
1260
extern DICTIONARY *rrdhost_root_index;
1261
size_t rrdhost_hosts_available(void);
1262
1263
+RRDHOST_ACQUIRED *rrdhost_find_and_acquire(const char *machine_guid);
1264
+RRDHOST *rrdhost_acquired_to_rrdhost(RRDHOST_ACQUIRED *rha);
1265
+void rrdhost_acquired_release(RRDHOST_ACQUIRED *rha);
1266
+
1267
// ----------------------------------------------------------------------------
1268
1269
#define rrdhost_foreach_read(var) \
@@ -1368,6 +1372,11 @@ void rrdset_update_heterogeneous_flag(RRDSET *st);
1372
time_t rrdset_set_update_every_s(RRDSET *st, time_t update_every_s);
1373
1374
RRDSET *rrdset_find(RRDHOST *host, const char *id);
1375
+
1376
+RRDSET_ACQUIRED *rrdset_find_and_acquire(RRDHOST *host, const char *id);
1377
+RRDSET *rrdset_acquired_to_rrdset(RRDSET_ACQUIRED *rsa);
1378
+void rrdset_acquired_release(RRDSET_ACQUIRED *rsa);
1379
+
1380
#define rrdset_find_localhost(id) rrdset_find(localhost, id)
1381
/* This will not return charts that are archived */
1382
static inline RRDSET *rrdset_find_active_localhost(const char *id)
database/rrddim.c
+1
-1
@@ -320,7 +320,7 @@ inline RRDDIM *rrddim_find(RRDSET *st, const char *id) {
320
}
321
322
inline RRDDIM_ACQUIRED *rrddim_find_and_acquire(RRDSET *st, const char *id) {
323
- debug(D_RRD_CALLS, "rrddim_find() for chart %s, dimension %s", rrdset_name(st), id);
323
+ debug(D_RRD_CALLS, "rrddim_find_and_acquire() for chart %s, dimension %s", rrdset_name(st), id);
324
325
return (RRDDIM_ACQUIRED *)dictionary_get_and_acquire_item(st->rrddim_root_index, id);
326
}
database/rrdhost.c
+20
@@ -78,6 +78,26 @@ static inline void rrdhost_init() {
78
}
79
}
80
81
+RRDHOST_ACQUIRED *rrdhost_find_and_acquire(const char *machine_guid) {
82
+ debug(D_RRD_CALLS, "rrdhost_find_and_acquire() host %s", machine_guid);
83
+
84
+ return (RRDHOST_ACQUIRED *)dictionary_get_and_acquire_item(rrdhost_root_index, machine_guid);
85
+}
86
+
87
+RRDHOST *rrdhost_acquired_to_rrdhost(RRDHOST_ACQUIRED *rha) {
88
+ if(unlikely(!rha))
89
+ return NULL;
90
+
91
+ return (RRDHOST *) dictionary_acquired_item_value((const DICTIONARY_ITEM *)rha);
92
+}
93
+
94
+void rrdhost_acquired_release(RRDHOST_ACQUIRED *rha) {
95
+ if(unlikely(!rha))
96
+ return;
97
+
98
+ dictionary_acquired_item_release(rrdhost_root_index, (const DICTIONARY_ITEM *)rha);
99
+}
100
+
101
// ----------------------------------------------------------------------------
102
// RRDHOST index by UUID
103
database/rrdset.c
+21
@@ -479,6 +479,27 @@ inline RRDSET *rrdset_find_byname(RRDHOST *host, const char *name) {
479
return(st);
480
}
481
482
+RRDSET_ACQUIRED *rrdset_find_and_acquire(RRDHOST *host, const char *id) {
483
+ debug(D_RRD_CALLS, "rrdset_find_and_acquire() for host %s, chart %s", rrdhost_hostname(host), id);
484
+
485
+ return (RRDSET_ACQUIRED *)dictionary_get_and_acquire_item(host->rrdset_root_index, id);
486
+}
487
+
488
+RRDSET *rrdset_acquired_to_rrdset(RRDSET_ACQUIRED *rsa) {
489
+ if(unlikely(!rsa))
490
+ return NULL;
491
+
492
+ return (RRDSET *) dictionary_acquired_item_value((const DICTIONARY_ITEM *)rsa);
493
+}
494
+
495
+void rrdset_acquired_release(RRDSET_ACQUIRED *rsa) {
496
+ if(unlikely(!rsa))
497
+ return;
498
+
499
+ RRDSET *rs = rrdset_acquired_to_rrdset(rsa);
500
+ dictionary_acquired_item_release(rs->rrdhost->rrdset_root_index, (const DICTIONARY_ITEM *)rsa);
501
+}
502
+
503
// ----------------------------------------------------------------------------
504
// RRDSET - rename charts
505
ml/ml-private.h
+1
-1
@@ -124,7 +124,7 @@ enum ml_training_result {
124
125
typedef struct {
126
// Chart/dimension we want to train
127
- STRING *host_id;
127
+ char machine_guid[GUID_LEN + 1];
128
STRING *chart_id;
129
STRING *dimension_id;
130
ml/ml.cc
+43
-26
@@ -268,7 +268,7 @@ ml_queue_pop(ml_queue_t *q)
268
netdata_mutex_lock(&q->mutex);
269
270
ml_training_request_t req = {
271
- NULL, // host_id
271
+ {'\0'}, // machine_guid
272
NULL, // chart id
273
NULL, // dimension id
274
0, // current time
@@ -783,14 +783,14 @@ ml_dimension_schedule_for_training(ml_dimension_t *dim, time_t curr_time)
783
}
784
785
if (schedule_for_training) {
786
- ml_training_request_t req = {
787
- string_dup(dim->rd->rrdset->rrdhost->hostname),
788
- string_dup(dim->rd->rrdset->id),
789
- string_dup(dim->rd->id),
790
- curr_time,
791
- rrddim_first_entry_s(dim->rd),
792
- rrddim_last_entry_s(dim->rd),
793
- };
786
+ ml_training_request_t req;
787
+
788
+ memcpy(req.machine_guid, dim->rd->rrdset->rrdhost->machine_guid, GUID_LEN + 1);
789
+ req.chart_id = string_dup(dim->rd->rrdset->id);
790
+ req.dimension_id = string_dup(dim->rd->id);
791
+ req.request_time = curr_time;
792
+ req.first_entry_on_request = rrddim_first_entry_s(dim->rd);
793
+ req.last_entry_on_request = rrddim_last_entry_s(dim->rd);
794
795
ml_host_t *host = (ml_host_t *) dim->rd->rrdset->rrdhost->ml_host;
796
ml_queue_push(host->training_queue, req);
@@ -1022,31 +1022,45 @@ ml_host_detect_once(ml_host_t *host)
1022
}
1023
1024
typedef struct {
1025
+ RRDHOST_ACQUIRED *acq_rh;
1026
+ RRDSET_ACQUIRED *acq_rs;
1027
RRDDIM_ACQUIRED *acq_rd;
1028
ml_dimension_t *dim;
1029
} ml_acquired_dimension_t;
1030
1031
static ml_acquired_dimension_t
1030
-ml_acquired_dimension_get(STRING *host_id, STRING *chart_id, STRING *dimension_id)
1032
+ml_acquired_dimension_get(char *machine_guid, STRING *chart_id, STRING *dimension_id)
1033
{
1034
+ RRDHOST_ACQUIRED *acq_rh = NULL;
1035
+ RRDSET_ACQUIRED *acq_rs = NULL;
1036
RRDDIM_ACQUIRED *acq_rd = NULL;
1037
ml_dimension_t *dim = NULL;
1038
1035
- RRDHOST *rh = rrdhost_find_by_hostname(string2str(host_id));
1036
- if (rh) {
1037
- RRDSET *rs = rrdset_find(rh, string2str(chart_id));
1038
- if (rs) {
1039
- acq_rd = rrddim_find_and_acquire(rs, string2str(dimension_id));
1040
- if (acq_rd) {
1041
- RRDDIM *rd = rrddim_acquired_to_rrddim(acq_rd);
1042
- if (rd)
1043
- dim = (ml_dimension_t *) rd->ml_dimension;
1039
+ rrd_rdlock();
1040
+
1041
+ acq_rh = rrdhost_find_and_acquire(machine_guid);
1042
+ if (acq_rh) {
1043
+ RRDHOST *rh = rrdhost_acquired_to_rrdhost(acq_rh);
1044
+ if (rh && !rrdhost_flag_check(rh, RRDHOST_FLAG_ORPHAN | RRDHOST_FLAG_ARCHIVED)) {
1045
+ acq_rs = rrdset_find_and_acquire(rh, string2str(chart_id));
1046
+ if (acq_rs) {
1047
+ RRDSET *rs = rrdset_acquired_to_rrdset(acq_rs);
1048
+ if (rs && !rrdset_flag_check(rs, RRDSET_FLAG_ARCHIVED | RRDSET_FLAG_OBSOLETE)) {
1049
+ acq_rd = rrddim_find_and_acquire(rs, string2str(dimension_id));
1050
+ if (acq_rd) {
1051
+ RRDDIM *rd = rrddim_acquired_to_rrddim(acq_rd);
1052
+ if (rd)
1053
+ dim = (ml_dimension_t *) rd->ml_dimension;
1054
+ }
1055
+ }
1056
}
1057
}
1058
}
1059
1060
+ rrd_unlock();
1061
+
1062
ml_acquired_dimension_t acq_dim = {
1049
- acq_rd, dim
1063
+ acq_rh, acq_rs, acq_rd, dim
1064
};
1065
1066
return acq_dim;
@@ -1055,10 +1069,14 @@ ml_acquired_dimension_get(STRING *host_id, STRING *chart_id, STRING *dimension_i
1069
static void
1070
ml_acquired_dimension_release(ml_acquired_dimension_t acq_dim)
1071
{
1058
- if (!acq_dim.acq_rd)
1059
- return;
1072
+ if (acq_dim.acq_rd)
1073
+ rrddim_acquired_release(acq_dim.acq_rd);
1074
+
1075
+ if (acq_dim.acq_rs)
1076
+ rrdset_acquired_release(acq_dim.acq_rs);
1077
1061
- rrddim_acquired_release(acq_dim.acq_rd);
1078
+ if (acq_dim.acq_rh)
1079
+ rrdhost_acquired_release(acq_dim.acq_rh);
1080
}
1081
1082
static enum ml_training_result
@@ -1385,7 +1403,7 @@ static void *ml_train_main(void *arg) {
1403
1404
// we know this thread has been cancelled, when the queue starts
1405
// returning "null" requests without blocking on queue's pop().
1388
- if (training_req.host_id == NULL)
1406
+ if (training_req.chart_id == NULL)
1407
break;
1408
1409
size_t queue_size = ml_queue_size(training_thread->training_queue) + 1;
@@ -1400,13 +1418,12 @@ static void *ml_train_main(void *arg) {
1418
{
1419
worker_is_busy(WORKER_TRAIN_ACQUIRE_DIMENSION);
1420
ml_acquired_dimension_t acq_dim = ml_acquired_dimension_get(
1403
- training_req.host_id,
1421
+ training_req.machine_guid,
1422
training_req.chart_id,
1423
training_req.dimension_id);
1424
1425
training_res = ml_acquired_dimension_train(training_thread, acq_dim, training_req);
1426
1409
- string_freez(training_req.host_id);
1427
string_freez(training_req.chart_id);
1428
string_freez(training_req.dimension_id);
1429