@cryptotaxi247 / netdata-1 / commits / 11de4e4ab

Functions: allow collectors to be restarted (#15983)

Costa Tsaousis committed Sep 16, 2023 at 16:00 UTC 11de4e4ab77177bc1a4f9b6358151adf525f2ca0

5 files changed +80 -29

database/rrd.h

@@ -931,6 +931,8 @@ typedef enum __attribute__ ((__packed__)) rrdhost_flags {

931

932 RRDHOST_FLAG_METADATA_CLAIMID = (1 << 28), // metadata needs to be stored in the database

933 RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED = (1 << 29), // set when the receiver part is disconnected

934 +

935 + RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED = (1 << 30), // set when the host has updated global functions

936 } RRDHOST_FLAGS;

937

938 #define rrdhost_flag_check(host, flag) (__atomic_load_n(&((host)->flags), __ATOMIC_SEQ_CST) & (flag))

database/rrdfunctions.c

+69 -29

@@ -310,8 +310,11 @@ struct rrd_collector {

310 static __thread struct rrd_collector *thread_rrd_collector = NULL;

311

312 static void rrd_collector_free(struct rrd_collector *rdc) {

313 + if(rdc->running)

314 + return;

315 +

316 int32_t expected = 0;

314 - if(likely(!__atomic_compare_exchange_n(&rdc->refcount, &expected, -1, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST))) {

317 + if(!__atomic_compare_exchange_n(&rdc->refcount, &expected, -1, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {

318 // the collector is still referenced by charts.

319 // leave it hanging there, the last chart will actually free it.

320 return;

@@ -323,9 +326,9 @@ static void rrd_collector_free(struct rrd_collector *rdc) {

326

327 // called once per collector

328 void rrd_collector_started(void) {

326 - if(likely(thread_rrd_collector)) return;

329 + if(!thread_rrd_collector)

330 + thread_rrd_collector = callocz(1, sizeof(struct rrd_collector));

331

328 - thread_rrd_collector = callocz(1, sizeof(struct rrd_collector));

332 thread_rrd_collector->tid = gettid();

333 thread_rrd_collector->running = true;

334 }

@@ -341,43 +344,70 @@ void rrd_collector_finished(void) {

344 }

345

346 static struct rrd_collector *rrd_collector_acquire(void) {

344 - __atomic_add_fetch(&thread_rrd_collector->refcount, 1, __ATOMIC_SEQ_CST);

347 + rrd_collector_started();

348 +

349 + int32_t expected = __atomic_load_n(&thread_rrd_collector->refcount, __ATOMIC_RELAXED), wanted = 0;

350 + do {

351 + if(expected < 0 || !thread_rrd_collector->running) {

352 + internal_fatal(true, "FUNCTIONS: Trying to acquire a collector that is exiting.");

353 + return thread_rrd_collector;

354 + }

355 +

356 + wanted = expected + 1;

357 +

358 + } while(!__atomic_compare_exchange_n(&thread_rrd_collector->refcount, &expected, wanted, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED));

359 +

360 return thread_rrd_collector;

361 }

362

363 static void rrd_collector_release(struct rrd_collector *rdc) {

364 if(unlikely(!rdc)) return;

365

351 - int32_t refcount = __atomic_sub_fetch(&rdc->refcount, 1, __ATOMIC_SEQ_CST);

352 - if(refcount == 0 && !rdc->running)

366 + int32_t expected = __atomic_load_n(&rdc->refcount, __ATOMIC_RELAXED), wanted = 0;

367 + do {

368 + if(expected < 0) {

369 + internal_fatal(true, "FUNCTIONS: Trying to release a collector that is exiting.");

370 + return;

371 + }

372 +

373 + if(expected == 0) {

374 + internal_fatal(true, "FUNCTIONS: Trying to release a collector that is not acquired.");

375 + return;

376 + }

377 +

378 + wanted = expected - 1;

379 +

380 + } while(!__atomic_compare_exchange_n(&rdc->refcount, &expected, wanted, false, __ATOMIC_RELEASE, __ATOMIC_RELAXED));

381 +

382 + if(wanted == 0)

383 rrd_collector_free(rdc);

384 }

385

356 -static void rrd_functions_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func __maybe_unused,

357 - void *rrdhost __maybe_unused) {

386 +static void rrd_functions_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func, void *rrdhost) {

387 + RRDHOST *host = rrdhost; (void)host;

388 struct rrd_collector_function *rdcf = func;

389

360 - if(!thread_rrd_collector)

361 - fatal("RRDSET_COLLECTOR: called %s() for function '%s' without calling rrd_collector_started() first.",

362 - __FUNCTION__, dictionary_acquired_item_name(item));

363 -

390 + rrd_collector_started();

391 rdcf->collector = rrd_collector_acquire();

392 +

393 +// internal_error(true, "FUNCTIONS: adding function '%s' on host '%s', collection tid %d, %s",

394 +// dictionary_acquired_item_name(item), rrdhost_hostname(host),

395 +// rdcf->collector->tid, rdcf->collector->running ? "running" : "NOT running");

396 }

397

367 -static void rrd_functions_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func __maybe_unused,

398 +static void rrd_functions_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func,

399 void *rrdhost __maybe_unused) {

400 struct rrd_collector_function *rdcf = func;

401 rrd_collector_release(rdcf->collector);

402 }

403

373 -static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func __maybe_unused,

374 - void *new_func __maybe_unused, void *rrdhost __maybe_unused) {

404 +static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func,

405 + void *new_func, void *rrdhost) {

406 + RRDHOST *host = rrdhost; (void)host;

407 struct rrd_collector_function *rdcf = func;

408 struct rrd_collector_function *new_rdcf = new_func;

409

378 - if(!thread_rrd_collector)

379 - fatal("RRDSET_COLLECTOR: called %s() for function '%s' without calling rrd_collector_started() first.",

380 - __FUNCTION__, dictionary_acquired_item_name(item));

410 + rrd_collector_started();

411

412 bool changed = false;

413

@@ -417,6 +447,10 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

447 changed = true;

448 }

449

450 +// internal_error(true, "FUNCTIONS: adding function '%s' on host '%s', collection tid %d, %s",

451 +// dictionary_acquired_item_name(item), rrdhost_hostname(host),

452 +// rdcf->collector->tid, rdcf->collector->running ? "running" : "NOT running");

453 +

454 return changed;

455 }

456

@@ -460,6 +494,8 @@ void rrd_collector_add_function(RRDHOST *host, RRDSET *st, const char *name, int

494

495 if(st)

496 dictionary_view_set(st->functions_view, key, item);

497 + else

498 + rrdhost_flag_set(host, RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED);

499

500 dictionary_acquired_item_release(host->functions, item);

501 }

@@ -481,6 +517,8 @@ void rrd_functions_expose_rrdpush(RRDSET *st, BUFFER *wb) {

517 }

518

519 void rrd_functions_expose_global_rrdpush(RRDHOST *host, BUFFER *wb) {

520 + rrdhost_flag_clear(host, RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED);

521 +

522 struct rrd_collector_function *tmp;

523 dfe_start_read(host->functions, tmp) {

524 if(!(tmp->options & RRD_FUNCTION_GLOBAL))

@@ -565,20 +603,22 @@ static int rrd_call_function_find(RRDHOST *host, BUFFER *wb, const char *name, s

603 char *s = NULL;

604

605 *rdcf = NULL;

568 - while(!(*rdcf) && buffer[0]) {

569 - *rdcf = dictionary_get(host->functions, buffer);

570 - if(*rdcf) break;

606 + if(host->functions) {

607 + while (!(*rdcf) && buffer[0]) {

608 + *rdcf = dictionary_get(host->functions, buffer);

609 + if (*rdcf) break;

610

572 - // if s == NULL, set it to the end of the buffer

573 - // this should happen only the first time

574 - if(unlikely(!s))

575 - s = &buffer[key_length - 1];

611 + // if s == NULL, set it to the end of the buffer

612 + // this should happen only the first time

613 + if (unlikely(!s))

614 + s = &buffer[key_length - 1];

615

577 - // skip a word from the end

578 - while(s >= buffer && !isspace(*s)) *s-- = '\0';

616 + // skip a word from the end

617 + while (s >= buffer && !isspace(*s)) *s-- = '\0';

618

580 - // skip all spaces

581 - while(s >= buffer && isspace(*s)) *s-- = '\0';

619 + // skip all spaces

620 + while (s >= buffer && isspace(*s)) *s-- = '\0';

621 + }

622 }

623

624 buffer_flush(wb);

libnetdata/required_dummies.h

@@ -37,6 +37,7 @@ void rrdset_thread_rda_free(void){};

37 void sender_thread_buffer_free(void){};

38 void query_target_free(void){};

39 void service_exits(void){};

40 +void rrd_collector_finished(void){};

42 // required by get_system_cpus()

43 char *netdata_configured_host_prefix = "";

libnetdata/threads/threads.c

@@ -178,6 +178,7 @@ void rrdset_thread_rda_free(void);

178 void sender_thread_buffer_free(void);

179 void query_target_free(void);

180 void service_exits(void);

181 +void rrd_collector_finished(void);

182

183 static void thread_cleanup(void *ptr) {

184 if(netdata_thread != ptr) {

@@ -188,6 +189,7 @@ static void thread_cleanup(void *ptr) {

189 if(!(netdata_thread->options & NETDATA_THREAD_OPTION_DONT_LOG_CLEANUP))

190 netdata_log_info("thread with task id %d finished", gettid());

191

192 + rrd_collector_finished();

193 sender_thread_buffer_free();

194 rrdset_thread_rda_free();

195 query_target_free();

streaming/rrdpush.c

@@ -489,6 +489,12 @@ RRDSET_STREAM_BUFFER rrdset_push_metric_initialize(RRDSET *st, time_t wall_clock

489 rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_LOGGED_STATUS);

490 }

491

492 + if(unlikely(host_flags & RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED)) {

493 + BUFFER *wb = sender_start(host->sender);

494 + rrd_functions_expose_global_rrdpush(host, wb);

495 + sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);

496 + }

497 +

498 RRDSET_FLAGS rrdset_flags = __atomic_load_n(&st->flags, __ATOMIC_SEQ_CST);

499 bool exposed_upstream = (rrdset_flags & RRDSET_FLAG_UPSTREAM_EXPOSED);

500 bool replication_in_progress = !(rrdset_flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED);