bf4a3a55f · cryptotaxi247/netdata-1

CMakeLists.txt

+2

@@ -1431,6 +1431,8 @@ set(RRD_PLUGIN_FILES

1431 src/database/rrdcollector-internals.h

1432 src/database/rrd-database-mode.h

1433 src/database/rrd-database-mode.c

1434 + src/database/rrdhost-state-id.c

1435 + src/database/rrdhost-state-id.h

1436 )

1437

1438 if(ENABLE_DBENGINE)

src/collectors/ebpf.plugin/ebpf_socket.c

-1

@@ -1759,7 +1759,6 @@ end_socket_loop: ; // the empty statement is here to allow code to be compiled b

1759 else {

1760 ebpf_release_pid_data(local_pid, fd, key.pid, EBPF_MODULE_SOCKET_IDX);

1761 ebpf_socket_release_publish(curr);

1762 - local_pid->socket = NULL;

1762 }

1763 memset(values, 0, length);

1764 memcpy(&key, &next_key, sizeof(key));

src/database/rrd.h

+6 -1

@@ -11,6 +11,7 @@ extern "C" {

11 #include "rrd-database-mode.h"

12 #include "streaming/stream-traffic-types.h"

13 #include "streaming/stream-sender-commit.h"

14 +#include "rrdhost-state-id.h"

15

16 // non-existing structs instead of voids

17 // to enable type checking at compile time

@@ -1161,6 +1162,11 @@ struct rrdhost {

1162 STRING *program_name; // the program name that collects metrics for this host

1163 STRING *program_version; // the program version that collects metrics for this host

1164

1165 + REFCOUNT state_refcount;

1166 + RRDHOST_STATE state_id; // every time data collection (stream receiver) (dis)connects,

1167 + // this gets incremented - it is used to detect stale functions,

1168 + // stale backfilling requests, etc.

1169 +

1170 int32_t utc_offset; // the offset in seconds from utc

1171

1172 RRDHOST_OPTIONS options; // configuration option for this RRDHOST (no atomics on this)

@@ -1238,7 +1244,6 @@ struct rrdhost {

1244

1245 struct {

1246 pid_t tid;

1241 - uint32_t state_id; // every time the receiver connects/disconnects, this is incremented

1247

1248 time_t last_connected; // the time the last sender was connected

1249 time_t last_disconnected; // the time the last sender was disconnected

src/database/rrdfunctions-internals.h

+1

@@ -29,6 +29,7 @@ struct rrd_host_function {

29 rrd_function_execute_cb_t execute_cb;

30 void *execute_cb_data;

31

32 + RRDHOST_STATE rrdhost_state_id;

33 struct rrd_collector *collector;

34 };

35

src/database/rrdfunctions.c

+30 -3

@@ -15,11 +15,12 @@

15 // ----------------------------------------------------------------------------

16

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

18 - RRDHOST *host = rrdhost; (void)host;

18 + RRDHOST *host = rrdhost;

19 struct rrd_host_function *rdcf = func;

20

21 rrd_collector_started();

22 rdcf->collector = rrd_collector_acquire_current_thread();

23 + rdcf->rrdhost_state_id = rrdhost_state_id(host);

24

25 if(!rdcf->priority)

26 rdcf->priority = RRDFUNCTIONS_PRIORITY_DEFAULT;

@@ -57,6 +58,17 @@ static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_

58 changed = true;

59 }

60

61 + if(rdcf->rrdhost_state_id != rrdhost_state_id(host)) {

62 + nd_log(NDLS_DAEMON, NDLP_DEBUG,

63 + "FUNCTIONS: function '%s' of host '%s' changed state id from %u to %u",

64 + dictionary_acquired_item_name(item), rrdhost_hostname(host),

65 + rdcf->rrdhost_state_id,

66 + rrdhost_state_id(host));

67 +

68 + rdcf->rrdhost_state_id = rrdhost_state_id(host);

69 + changed = true;

70 + }

71 +

72 if(rdcf->execute_cb != new_rdcf->execute_cb) {

73 nd_log(NDLS_DAEMON, NDLP_DEBUG,

74 "FUNCTIONS: function '%s' of host '%s' changed execute callback",

@@ -260,6 +272,8 @@ int rrd_functions_find_by_name(RRDHOST *host, BUFFER *wb, const char *name, size

272 strncpyz(buffer, name, sizeof(buffer) - 1);

273 char *s = NULL;

274

275 + RRDHOST_STATE state_id = rrdhost_state_id(host);

276 +

277 bool found = false;

278 *item = NULL;

279 if(host->functions) {

@@ -268,10 +282,23 @@ int rrd_functions_find_by_name(RRDHOST *host, BUFFER *wb, const char *name, size

282 found = true;

283

284 struct rrd_host_function *rdcf = dictionary_acquired_item_value(*item);

271 - if(rrd_collector_running(rdcf->collector)) {

285 + if(rrd_collector_running(rdcf->collector) && rdcf->rrdhost_state_id == state_id) {

286 break;

287 }

288 else {

289 +

290 + nd_log(NDLS_DAEMON, NDLP_DEBUG,

291 + "Function '%s' is not available. "

292 + "host '%s', collector = { tid: %d, running: %s }, host tid { rcv: %d, snd: %d }, host state { id: %u, expected %u }, hops: %d",

293 + name,

294 + rrdhost_hostname(host),

295 + rrd_collector_tid(rdcf->collector),

296 + rrd_collector_running(rdcf->collector) ? "yes" : "no",

297 + host->stream.rcv.status.tid, host->stream.snd.status.tid,

298 + state_id, rdcf->rrdhost_state_id,

299 + host->system_info->hops

300 + );

301 +

302 dictionary_acquired_item_release(host->functions, *item);

303 *item = NULL;

304 }

@@ -314,7 +341,7 @@ bool rrd_function_available(RRDHOST *host, const char *function) {

341 const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(host->functions, function);

342 if(item) {

343 struct rrd_host_function *rdcf = dictionary_acquired_item_value(item);

317 - if(rrd_collector_running(rdcf->collector))

344 + if(rrd_collector_running(rdcf->collector) && rdcf->rrdhost_state_id == rrdhost_state_id(host))

345 ret = true;

346

347 dictionary_acquired_item_release(host->functions, item);

src/database/rrdhost-state-id.c new

+73

@@ -0,0 +1,73 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "rrdhost-state-id.h"

4 +#include "rrd.h"

5 +

6 +RRDHOST_STATE rrdhost_state_id(struct rrdhost *host) {

7 + return __atomic_load_n(&host->state_id, __ATOMIC_RELAXED);

8 +}

9 +

10 +bool rrdhost_state_connected(RRDHOST *host) {

11 + __atomic_add_fetch(&host->state_id, 1, __ATOMIC_RELAXED);

12 +

13 + int32_t expected = __atomic_load_n(&host->state_refcount, __ATOMIC_RELAXED);

14 + int32_t desired;

15 +

16 + do {

17 + if(expected >= 0) {

18 + internal_fatal(true, "Cannot get the node connected");

19 + return false;

20 + }

21 +

22 + desired = 0;

23 +

24 + } while(!__atomic_compare_exchange_n(

25 + &host->state_refcount, &expected, desired, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));

26 +

27 + return true;

28 +}

29 +

30 +bool rrdhost_state_disconnected(RRDHOST *host) {

31 + __atomic_add_fetch(&host->state_id, 1, __ATOMIC_RELAXED);

32 +

33 + int32_t expected = __atomic_load_n(&host->state_refcount, __ATOMIC_RELAXED);

34 + int32_t desired;

35 +

36 + do {

37 + if(expected < 0) {

38 + internal_fatal(true, "Cannot get the node disconnected");

39 + return false;

40 + }

41 +

42 + desired = -1;

43 +

44 + } while(!__atomic_compare_exchange_n(

45 + &host->state_refcount, &expected, desired, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));

46 +

47 + return true;

48 +}

49 +

50 +bool rrdhost_state_acquire(RRDHOST *host, RRDHOST_STATE wanted_state_id) {

51 + int32_t expected = __atomic_load_n(&host->state_refcount, __ATOMIC_RELAXED);

52 + int32_t desired;

53 +

54 + do {

55 + if(expected < 0)

56 + return false;

57 +

58 + desired = expected + 1;

59 +

60 + } while(!__atomic_compare_exchange_n(

61 + &host->state_refcount, &expected, desired, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));

62 +

63 + if(rrdhost_state_id(host) != wanted_state_id) {

64 + rrdhost_state_release(host);

65 + return false;

66 + }

67 +

68 + return true;

69 +}

70 +

71 +void rrdhost_state_release(RRDHOST *host) {

72 + __atomic_sub_fetch(&host->state_refcount, 1, __ATOMIC_RELAXED);

73 +}

src/database/rrdhost-state-id.h new

+19

@@ -0,0 +1,19 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#ifndef NETDATA_RRDHOST_STATE_ID_H

4 +#define NETDATA_RRDHOST_STATE_ID_H

5 +

6 +#include "libnetdata/libnetdata.h"

7 +

8 +typedef uint32_t RRDHOST_STATE;

9 +

10 +struct rrdhost;

11 +RRDHOST_STATE rrdhost_state_id(struct rrdhost *host);

12 +

13 +bool rrdhost_state_connected(struct rrdhost *host);

14 +bool rrdhost_state_disconnected(struct rrdhost *host);

15 +

16 +bool rrdhost_state_acquire(struct rrdhost *host, RRDHOST_STATE wanted_state_id);

17 +void rrdhost_state_release(struct rrdhost *host);

18 +

19 +#endif //NETDATA_RRDHOST_STATE_ID_H

src/database/rrdhost.c

+3

@@ -327,6 +327,8 @@ static RRDHOST *rrdhost_create(

327 }

328

329 RRDHOST *host = callocz(1, sizeof(RRDHOST));

330 + host->state_refcount = -1;

331 +

332 __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_allocations_size, sizeof(RRDHOST), __ATOMIC_RELAXED);

333

334 strncpyz(host->machine_guid, guid, GUID_LEN + 1);

@@ -840,6 +842,7 @@ int rrd_init(const char *hostname, struct rrdhost_system_info *system_info, bool

842 return 1;

843

844 rrdhost_flag_set(localhost, RRDHOST_FLAG_COLLECTOR_ONLINE);

845 + rrdhost_state_connected(localhost);

846

847 ml_host_start(localhost);

848 dyncfg_host_init(localhost);

src/libnetdata/common.h

+4

@@ -400,6 +400,10 @@ typedef uint32_t uid_t;

400

401 // --------------------------------------------------------------------------------------------------------------------

402

403 +typedef int32_t REFCOUNT;

404 +

405 +// --------------------------------------------------------------------------------------------------------------------

406 +

407 #if defined(OS_WINDOWS)

408 #include <windows.h>

409 #include <wctype.h>

src/plugins.d/pluginsd_parser.c

+15 -12

@@ -205,6 +205,7 @@ static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, si

205

206 rrdhost_option_set(host, RRDHOST_OPTION_VIRTUAL_HOST);

207 rrdhost_flag_set(host, RRDHOST_FLAG_COLLECTOR_ONLINE);

208 + rrdhost_state_connected(host);

209 ml_host_start(host);

210 dyncfg_host_init(host);

211

@@ -376,16 +377,19 @@ static inline PARSER_RC pluginsd_chart(char **words, size_t num_words, PARSER *p

377 }

378

379 static void backfill_callback(size_t successful_dims __maybe_unused, size_t failed_dims __maybe_unused, struct backfill_request_data *brd) {

379 - if (brd->rrdhost_receiver_state_id == __atomic_load_n(&brd->host->stream.rcv.status.state_id, __ATOMIC_RELAXED)) {

380 - if (!replicate_chart_request(send_to_plugin, brd->parser, brd->host, brd->st,

381 - brd->first_entry_child, brd->last_entry_child, brd->child_wall_clock_time,

382 - 0, 0)) {

383 - netdata_log_error(

384 - "PLUGINSD: 'host:%s' failed to initiate replication for 'chart:%s'",

385 - rrdhost_hostname(brd->host),

386 - rrdset_id(brd->st));

387 - }

380 + if(!rrdhost_state_acquire(brd->host, brd->rrdhost_receiver_state_id))

381 + return;

382 +

383 + if (!replicate_chart_request(send_to_plugin, brd->parser, brd->host, brd->st,

384 + brd->first_entry_child, brd->last_entry_child, brd->child_wall_clock_time,

385 + 0, 0)) {

386 + netdata_log_error(

387 + "PLUGINSD: 'host:%s' failed to initiate replication for 'chart:%s'",

388 + rrdhost_hostname(brd->host),

389 + rrdset_id(brd->st));

390 }

391 +

392 + rrdhost_state_release(brd->host);

393 }

394

395 static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, PARSER *parser) {

@@ -417,7 +421,7 @@ static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_w

421 rrdhost_receiver_replicating_charts_plus_one(st->rrdhost);

422

423 struct backfill_request_data brd = {

420 - .rrdhost_receiver_state_id =__atomic_load_n(&host->stream.rcv.status.state_id, __ATOMIC_RELAXED),

424 + .rrdhost_receiver_state_id = rrdhost_state_id(host),

425 .parser = parser,

426 .host = host,

427 .st = st,

@@ -1173,8 +1177,6 @@ void pluginsd_process_cleanup(PARSER *parser) {

1177 pluginsd_cleanup_v2(parser);

1178 pluginsd_host_define_cleanup(parser);

1179

1176 - rrd_collector_finished();

1177 -

1180 #ifdef NETDATA_LOG_STREAM_RECEIVE

1181 if(parser->user.stream_log_fp) {

1182 fclose(parser->user.stream_log_fp);

@@ -1188,6 +1190,7 @@ void pluginsd_process_cleanup(PARSER *parser) {

1190 void pluginsd_process_thread_cleanup(void *pptr) {

1191 PARSER *parser = CLEANUP_FUNCTION_GET_PTR(pptr);

1192 pluginsd_process_cleanup(parser);

1193 + rrd_collector_finished();

1194 }

1195

1196 bool parser_reconstruct_node(BUFFER *wb, void *ptr) {

src/plugins.d/pluginsd_parser.h

+1 -1

@@ -5,7 +5,7 @@

5

6 #include "daemon/common.h"

7

8 -#define WORKER_PARSER_FIRST_JOB 35

8 +#define WORKER_PARSER_FIRST_JOB 36

9

10 // this has to be in-sync with the same at stream-thread.c

11 #define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION 25

src/streaming/stream-receiver.c

+4 -5

@@ -358,8 +358,6 @@ static void streaming_parser_init(struct receiver_state *rpt) {

358

359 pluginsd_keywords_init(parser, PARSER_INIT_STREAMING);

360

361 - rrd_collector_started();

362 -

361 rpt->thread.compressed.start = 0;

362 rpt->thread.compressed.used = 0;

363 rpt->thread.compressed.enabled = stream_decompression_initialize(rpt);

@@ -446,6 +444,7 @@ void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct re

444 sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);

445

446 stream_receive_log_database_gap(rpt);

447 + rrdhost_state_connected(rpt->host);

448

449 // keep this last, since it sends commands back to the child

450 streaming_parser_init(rpt);

@@ -477,6 +476,8 @@ static void stream_receiver_remove(struct stream_thread *sth, struct receiver_st

476 , rpt->client_port ? rpt->client_port : "-"

477 , why ? why : "");

478

479 + rrdhost_state_disconnected(rpt->host);

480 +

481 internal_fatal(META_GET(&sth->run.meta, (Word_t)&rpt->thread.meta) == NULL, "Receiver to be removed is not found in the list of receivers");

482 META_DEL(&sth->run.meta, (Word_t)&rpt->thread.meta);

483

@@ -806,8 +807,6 @@ bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {

807 rrdhost_receiver_lock(host);

808

809 if (!host->receiver) {

809 - __atomic_add_fetch(&host->stream.rcv.status.state_id, 1, __ATOMIC_RELAXED);

810 -

810 rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);

811

812 host->stream.rcv.status.connections++;

@@ -870,8 +869,8 @@ void rrdhost_clear_receiver(struct receiver_state *rpt) {

869 // Make sure that we detach this thread and don't kill a freshly arriving receiver

870

871 if (host->receiver == rpt) {

873 - __atomic_add_fetch(&host->stream.rcv.status.state_id, 1, __ATOMIC_RELAXED);

872 rrdhost_flag_clear(host, RRDHOST_FLAG_COLLECTOR_ONLINE);

873 +

874 rrdhost_receiver_unlock(host);

875 {

876 // run all these without having the receiver lock

src/streaming/stream-sender.c

+3 -1

@@ -540,8 +540,10 @@ bool stream_sender_process_poll_events(struct stream_thread *sth, struct sender_

540 return false;

541 }

542 }

543 - else

543 + else {

544 + sth->snd.send_misses++;

545 break;

546 + }

547 }

548 }

549

src/streaming/stream-thread.c

+12 -2

@@ -450,10 +450,15 @@ void *stream_thread(void *ptr) {

450 "ops processed", "messages",

451 WORKER_METRIC_INCREMENTAL_TOTAL);

452

453 - worker_register_job_custom_metric(WORKER_SENDER_JOB_RECEIVERS_WAITING_LIST_SIZE,

453 + worker_register_job_custom_metric(WORKER_STREAM_JOB_RECEIVERS_WAITING_LIST_SIZE,

454 "receivers waiting to be added", "nodes",

455 WORKER_METRIC_ABSOLUTE);

456

457 + worker_register_job_custom_metric(WORKER_STREAM_JOB_SEND_MISSES,

458 + "send misses", "misses",

459 + WORKER_METRIC_INCREMENTAL_TOTAL);

460 +

461 +

462 if(pipe(sth->pipe.fds) != 0) {

463 nd_log(NDLS_DAEMON, NDLP_ERR, "STREAM THREAD[%zu]: cannot create required pipe.", sth->id);

464 sth->pipe.fds[PIPE_READ] = -1;

@@ -487,6 +492,8 @@ void *stream_thread(void *ptr) {

492 sth->snd.bytes_received = 0;

493 sth->snd.bytes_sent = 0;

494

495 + rrd_collector_started();

496 +

497 while(!exit_thread && !nd_thread_signaled_to_cancel() && service_running(SERVICE_STREAMING)) {

498 usec_t now_ut = now_monotonic_usec();

499

@@ -521,7 +528,8 @@ void *stream_thread(void *ptr) {

528 worker_set_metric(WORKER_SENDER_JOB_BYTES_SENT, (NETDATA_DOUBLE)sth->snd.bytes_sent);

529 worker_set_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, (NETDATA_DOUBLE)replay_entries);

530

524 - worker_set_metric(WORKER_SENDER_JOB_RECEIVERS_WAITING_LIST_SIZE, (NETDATA_DOUBLE)receivers_waiting);

531 + worker_set_metric(WORKER_STREAM_JOB_RECEIVERS_WAITING_LIST_SIZE, (NETDATA_DOUBLE)receivers_waiting);

532 + worker_set_metric(WORKER_STREAM_JOB_SEND_MISSES, (NETDATA_DOUBLE)sth->snd.send_misses);

533 replay_entries = 0;

534 sth->snd.bytes_received = 0;

535 sth->snd.bytes_sent = 0;

@@ -597,6 +605,8 @@ void *stream_thread(void *ptr) {

605

606 worker_unregister();

607

608 + rrd_collector_finished();

609 +

610 return NULL;

611 }

612

src/streaming/stream-thread.h

+3 -1

@@ -83,7 +83,8 @@ struct stream_opcode {

83 #define WORKER_SENDER_JOB_BYTES_COMPRESSION_RATIO 32

84 #define WORKER_SENDER_JOB_REPLAY_DICT_SIZE 33

85 #define WORKER_SENDER_JOB_MESSAGES 34

86 -#define WORKER_SENDER_JOB_RECEIVERS_WAITING_LIST_SIZE 35

86 +#define WORKER_STREAM_JOB_RECEIVERS_WAITING_LIST_SIZE 35

87 +#define WORKER_STREAM_JOB_SEND_MISSES 36

88

89 // IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly

90

@@ -125,6 +126,7 @@ struct stream_thread {

126 struct {

127 size_t bytes_received;

128 size_t bytes_sent;

129 + size_t send_misses;

130 } snd;

131

132 struct {

src/web/api/queries/backfill.c

+6 -5

@@ -48,7 +48,7 @@ bool backfill_request_add(RRDSET *st, backfill_callback_t cb, struct backfill_re

48 if(backfill_globals.running) {

49 struct backfill_request *br = aral_mallocz(backfill_globals.ar_br);

50 br->data = *data;

51 - br->rrdhost_receiver_state_id =__atomic_load_n(&st->rrdhost->stream.rcv.status.state_id, __ATOMIC_RELAXED);

51 + br->rrdhost_receiver_state_id = rrdhost_state_id(st->rrdhost);

52 br->rsa = rrdset_find_and_acquire(st->rrdhost, string2str(st->id));

53 if(br->rsa) {

54 br->cb = cb;

@@ -95,19 +95,20 @@ bool backfill_request_add(RRDSET *st, backfill_callback_t cb, struct backfill_re

95 bool backfill_execute(struct backfill_dim_work *bdm) {

96 RRDSET *st = rrdset_acquired_to_rrdset(bdm->br->rsa);

97

98 - if(bdm->br->rrdhost_receiver_state_id !=__atomic_load_n(&st->rrdhost->stream.rcv.status.state_id, __ATOMIC_RELAXED))

98 + if(!rrdhost_state_acquire(st->rrdhost, bdm->br->rrdhost_receiver_state_id))

99 return false;

100

101 + size_t success = 0;

102 RRDDIM *rd = rrddim_acquired_to_rrddim(bdm->rda);

103

103 - size_t success = 0;

104 for (size_t tier = 1; tier < storage_tiers; tier++)

105 - if(backfill_tier_from_smaller_tiers(rd, tier, now_realtime_sec()))

105 + if (backfill_tier_from_smaller_tiers(rd, tier, now_realtime_sec()))

106 success++;

107

108 - if(success > 0)

108 + if (success > 0)

109 rrddim_option_set(rd, RRDDIM_OPTION_BACKFILLED_HIGH_TIERS);

110

111 + rrdhost_state_release(st->rrdhost);

112 return success > 0;

113 }

114

Streaming Improvements No 5 (#19193)