@cryptotaxi247 / netdata-1 / commits / ffe402f02

Fix RRDDIM_MEM storage engine index (#19830)

* add more logging to rrddim fatal incorrect pointer returned from index * increase status file version to 13 * remove reuse of old uuidmap ids * debug info to find leaked refcount * fix strings leak in streaming; count delayed dictionaries * switch rd pointer when rrddim_mem finds the same uuidmap id on a different rrddim * fix the rrdset-name cleanup * cleanup

Costa Tsaousis committed Mar 12, 2025 at 21:15 UTC ffe402f023d0c726e2a11aba3c272cb4f14529c6

13 files changed +108 -87

src/daemon/daemon-shutdown-watcher.c

+3 -3

@@ -35,7 +35,7 @@ static void watcher_wait_for_step(const watcher_step_id_t step_id, usec_t shutdo

35 watcher_steps[step_id].msg);

37 #if defined(FSANITIZE_ADDRESS)

38 - fprintf(stderr, " > shutdown step: [%d/%d] - {at %s} started '%s'...\n",

38 + fprintf(stdout, " > shutdown step: [%d/%d] - {at %s} started '%s'...\n",

39 (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

40 watcher_steps[step_id].msg);

41 #endif

@@ -69,7 +69,7 @@ static void watcher_wait_for_step(const watcher_step_id_t step_id, usec_t shutdo

69 watcher_steps[step_id].msg, step_duration_txt);

71 #if defined(FSANITIZE_ADDRESS)

72 - fprintf(stderr, " > shutdown step: [%d/%d] - {at %s} finished '%s' in %s\n",

72 + fprintf(stdout, " > shutdown step: [%d/%d] - {at %s} finished '%s' in %s\n",

73 (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

74 watcher_steps[step_id].msg, step_duration_txt);

75 #endif

@@ -81,7 +81,7 @@ static void watcher_wait_for_step(const watcher_step_id_t step_id, usec_t shutdo

81 watcher_steps[step_id].msg, step_duration_txt);

83 #if defined(FSANITIZE_ADDRESS)

84 - fprintf(stderr, "shutdown step: [%d/%d] - {at %s} timeout '%s' takes too long (%s) - giving up...\n",

84 + fprintf(stdout, "shutdown step: [%d/%d] - {at %s} timeout '%s' takes too long (%s) - giving up...\n",

85 (int)step_id + 1, (int)WATCHER_STEP_ID_MAX, start_duration_txt,

86 watcher_steps[step_id].msg, step_duration_txt);

87 #endif

src/daemon/daemon-shutdown.c

+4 -2

@@ -330,8 +330,10 @@ void netdata_cleanup_and_exit(EXIT_REASON reason, const char *action, const char

330 rrdhost_free_all();

331

332 fprintf(stderr, "Cleaning up destroyed dictionaries...\n");

333 - if(cleanup_destroyed_dictionaries())

334 - fprintf(stderr, "WARNING: There are still dictionaries with references in them, that cannot be destroyed.\n");

333 + size_t dictionaries_referenced = cleanup_destroyed_dictionaries();

334 + if(dictionaries_referenced)

335 + fprintf(stderr, "WARNING: There are %zu dictionaries with references in them, that cannot be destroyed.\n",

336 + dictionaries_referenced);

337

338 // destroy the caches in reverse order (extent and open depend on main cache)

339 fprintf(stderr, "Destroying extent cache (PGC)...\n");

src/daemon/main.c

+1 -1

@@ -185,7 +185,7 @@ int help(int exitcode) {

185 */

186

187 #if defined(FSANITIZE_ADDRESS)

188 -#define LOG_TO_STDERR(...) fprintf(stderr, __VA_ARGS__)

188 +#define LOG_TO_STDERR(...) fprintf(stdout, __VA_ARGS__)

189 #else

190 #define LOG_TO_STDERR(...)

191 #endif

src/daemon/service.c

@@ -152,6 +152,8 @@ static inline void svc_rrdhost_cleanup_charts_marked_obsolete(RRDHOST *host) {

152 }

153 rrdset_foreach_done(st);

154

155 + dictionary_garbage_collect(host->rrdset_root_index);

156 +

157 if(partial_archives != partial_candidates)

158 rrdhost_flag_set(host, RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS);

159

src/database/ram/rrddim_mem.c

+8 -3

@@ -70,8 +70,13 @@ STORAGE_METRIC_HANDLE *rrddim_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE

70 netdata_rwlock_wrunlock(&rrddim_Judy_rwlock);

71 }

73 - if(unlikely(mh->rd != rd))

74 - fatal("DB_RAM_ALLOC: incorrect pointer returned from index.");

73 + if(unlikely(mh->rd != rd)) {

74 + // this can happen when the old RRDDIM is being deleted,

75 + // but the dictionary has not yet run the destructors

76 + netdata_rwlock_wrlock(&rrddim_Judy_rwlock);

77 + mh->rd = rd;

78 + netdata_rwlock_wrunlock(&rrddim_Judy_rwlock);

79 + }

81 return (STORAGE_METRIC_HANDLE *)mh;

82 }

@@ -112,7 +117,7 @@ STORAGE_METRIC_HANDLE *rrddim_metric_dup(STORAGE_METRIC_HANDLE *smh) {

117 return smh;

118 }

119

115 -void rrddim_metric_release(STORAGE_METRIC_HANDLE *smh __maybe_unused) {

120 +void rrddim_metric_release(STORAGE_METRIC_HANDLE *smh) {

121 struct mem_metric_handle *mh = (struct mem_metric_handle *)smh;

122

123 if(refcount_release_and_acquire_for_deletion(&mh->refcount)) {

src/database/rrdset-index-name.c

+15 -4

@@ -80,8 +80,14 @@ void rrdset_index_add_name(RRDHOST *host, RRDSET *st) {

80 const DICTIONARY_ITEM *sta = dictionary_get_and_acquire_item(host->rrdset_root_index, rrdset_id(st));

81 if(sta) {

82 const DICTIONARY_ITEM *sta2 = dictionary_view_set_and_acquire_item(host->rrdset_root_index_name, rrdset_name(st), sta);

83 - if(sta2 && dictionary_acquired_item_value(sta2) == st)

84 - rrdset_flag_set(st, RRDSET_FLAG_INDEXED_NAME);

83 + if(sta2) {

84 + if(dictionary_acquired_item_value(sta2) == st)

85 + rrdset_flag_set(st, RRDSET_FLAG_INDEXED_NAME);

86 +

87 + dictionary_acquired_item_release(host->rrdset_root_index_name, sta2);

88 + }

89 +

90 + dictionary_acquired_item_release(host->rrdset_root_index, sta);

91 }

92 }

@@ -89,8 +95,13 @@ void rrdset_index_del_name(RRDHOST *host, RRDSET *st) {

95 if(rrdset_flag_check(st, RRDSET_FLAG_INDEXED_NAME)) {

96 const DICTIONARY_ITEM *sta = dictionary_get_and_acquire_item(host->rrdset_root_index_name, rrdset_name(st));

92 - if(sta && dictionary_acquired_item_value(sta) == st)

93 - dictionary_del(host->rrdset_root_index_name, rrdset_name(st));

98 + if(sta) {

99 + if(dictionary_acquired_item_value(sta) == st)

100 + dictionary_del(host->rrdset_root_index_name, rrdset_name(st));

101 +

102 + dictionary_acquired_item_release(host->rrdset_root_index_name, sta);

103 + dictionary_garbage_collect(host->rrdset_root_index_name);

104 + }

105

106 rrdset_flag_clear(st, RRDSET_FLAG_INDEXED_NAME);

107 }

src/libnetdata/dictionary/dictionary.c

+9 -6

@@ -330,13 +330,15 @@ static void dictionary_queue_for_destruction(DICTIONARY *dict) {

330 netdata_mutex_unlock(&dictionaries_waiting_to_be_destroyed_mutex);

331 }

332

333 -bool cleanup_destroyed_dictionaries(void) {

333 +size_t cleanup_destroyed_dictionaries(void) {

334 netdata_mutex_lock(&dictionaries_waiting_to_be_destroyed_mutex);

335 if (!dictionaries_waiting_to_be_destroyed) {

336 netdata_mutex_unlock(&dictionaries_waiting_to_be_destroyed_mutex);

337 - return false;

337 + return 0;

338 }

339

340 + size_t remaining = 0;

341 +

342 DICTIONARY *dict, *last = NULL, *next = NULL;

343 for(dict = dictionaries_waiting_to_be_destroyed; dict ; dict = next) {

344 next = dict->next;

@@ -353,7 +355,7 @@ bool cleanup_destroyed_dictionaries(void) {

355

356 internal_error(

357 true,

356 - "DICTIONARY: freed dictionary with delayed destruction, created from %s() %zu@%s pid %d.",

358 + "DICTIONARY DELAYED: freed dict created from %s() %zu@%s pid %d.",

359 function, line, file, pid);

360

361 if(last) last->next = next;

@@ -363,18 +365,19 @@ bool cleanup_destroyed_dictionaries(void) {

365

366 internal_error(

367 true,

366 - "DICTIONARY: cannot free dictionary with delayed destruction, created from %s() %zu@%s pid %d.",

368 + "DICTIONARY DELAYED %zu: %zu referenced in dict created from %s() %zu@%s pid %d.",

369 + remaining + 1, dictionary_referenced_items(dict),

370 function, line, file, pid);

371

372 DICTIONARY_STATS_DICT_DESTROY_QUEUED_PLUS1(dict);

373 last = dict;

374 + remaining++;

375 }

376 }

377

374 - bool ret = dictionaries_waiting_to_be_destroyed != NULL;

378 netdata_mutex_unlock(&dictionaries_waiting_to_be_destroyed_mutex);

379

377 - return ret;

380 + return remaining;

381 }

382

383 // ----------------------------------------------------------------------------

src/libnetdata/dictionary/dictionary.h

+1 -1

@@ -168,7 +168,7 @@ void dictionary_version_increment(DICTIONARY *dict);

168

169 void dictionary_garbage_collect(DICTIONARY *dict);

170

171 -bool cleanup_destroyed_dictionaries(void);

171 +size_t cleanup_destroyed_dictionaries(void);

172

173 // ----------------------------------------------------------------------------

174 // Set an item in the dictionary

src/libnetdata/uuid/uuidmap.c

+8 -41

@@ -2,8 +2,6 @@

3 #include "uuidmap.h"

5 -#define UUIDMAP_REUSE_GAP 1000

6 -

5 struct uuidmap_entry {

6 nd_uuid_t uuid;

7 REFCOUNT refcount;

@@ -12,9 +10,7 @@ struct uuidmap_entry {

10 struct uuidmap_partition {

11 Pvoid_t uuid_to_id; // JudyL: UUID string -> ID

12 Pvoid_t id_to_uuid; // JudyL: ID -> UUID binary

15 - Pvoid_t freed_ids; // JudyL: the freed IDs

13 UUIDMAP_ID next_id; // Only use lower bits

17 - UUIDMAP_ID next_free_id; // fifo id when reusing freed ids

14 RW_SPINLOCK spinlock;

16 int64_t memory;

@@ -64,33 +60,13 @@ static void uuidmap_init_aral(void) {

60 }

62 static UUIDMAP_ID get_next_id_unsafe(struct uuidmap_partition *partition) {

67 - UUIDMAP_ID id = 0;

68 -

69 - // Try to get a freed ID first if we have enough gap

70 - Pvoid_t *PValue;

71 - Word_t Index = 0;

72 -

73 - PValue = JudyLFirst(partition->freed_ids, &Index, PJE0);

74 - if (PValue && PValue != PJERR && *PValue) {

75 - // Check if the stored ID is old enough to be reused

76 - UUIDMAP_ID stored_id = *(UUIDMAP_ID *)PValue;

77 - UUIDMAP_ID current_next_id = uuidmap_make_id(partition - uuid_map.p, partition->next_id);

78 -

79 - if ((current_next_id - stored_id) >= UUIDMAP_REUSE_GAP) {

80 - id = stored_id;

81 - // Remove this entry from freed_ids since we're reusing it

82 - int rc = JudyLDel(&partition->freed_ids, Index, PJE0);

83 - if (unlikely(!rc))

84 - fatal("UUIDMAP: cannot delete ID from freed_ids JudyL");

85 - }

86 - }

63 + // Check if we've reached the maximum ID value

64 + if (partition->next_id >= 0x1FFFFFFF)

65 + fatal("UUIDMAP: Maximum ID limit reached for partition %u. UUIDs exhausted.",

66 + (unsigned int)(partition - uuid_map.p));

88 - if (id == 0) {

89 - // No reusable IDs available, get next sequential ID

90 - id = uuidmap_make_id(partition - uuid_map.p, ++partition->next_id);

91 - }

92 -

93 - return id;

68 + // Simply increment and return the next ID

69 + return uuidmap_make_id(partition - uuid_map.p, ++partition->next_id);

70 }

72 static inline UUIDMAP_ID uuidmap_acquire_by_uuid(const nd_uuid_t uuid) {

@@ -220,13 +196,6 @@ void uuidmap_free(UUIDMAP_ID id) {

196 if(unlikely(!rc))

197 fatal("UUIDMAP: cannot delete ID from JudyL");

198

223 - // Add the freed ID to the freed_ids JudyL using next_free_id as index

224 - Pvoid_t *PValue = JudyLIns(&uuid_map.p[partition].freed_ids, ++uuid_map.p[partition].next_free_id, PJE0);

225 - if (!PValue || PValue == PJERR)

226 - fatal("UUIDMAP: corrupted freed_ids JudyL array");

227 -

228 - *(UUIDMAP_ID *)PValue = id; // Store the actual METRIC_ID as the value

229 -

199 uuid_map.p[partition].memory -= sizeof(*ue);

200 uuid_map.p[partition].entries--;

201

@@ -288,7 +257,6 @@ size_t uuidmap_destroy(void) {

257

258 Pvoid_t uuid_to_id = uuid_map.p[partition].uuid_to_id;

259 Pvoid_t id_to_uuid = uuid_map.p[partition].id_to_uuid;

291 - Pvoid_t freed_ids = uuid_map.p[partition].freed_ids;

260

261 // Process all entries in the id_to_uuid map

262 Word_t id_index = 0;

@@ -313,7 +281,6 @@ size_t uuidmap_destroy(void) {

281 // Free all Judy arrays

282 JudyHSFreeArray(&uuid_to_id, PJE0);

283 JudyLFreeArray(&id_to_uuid, PJE0);

316 - JudyLFreeArray(&freed_ids, PJE0);

284

285 // Reset partition data

286 memset(&uuid_map.p[partition], 0, sizeof(uuid_map.p[partition]));

@@ -617,7 +584,7 @@ int uuidmap_unittest(void) {

584 double ops = (double)successful / secs;

585

586 fprintf(stderr, "uuidmap_uuid_ptr() : %.2f ops/sec (%.2f usec/op)\n",

620 - ops, (double)(end_ut - start_ut) / successful);

587 + ops, (double)(end_ut - start_ut) / (double)successful);

588

589 // Second benchmark: uuidmap_get_by_uuid()

590 successful = 0;

@@ -636,7 +603,7 @@ int uuidmap_unittest(void) {

603 ops = (double)successful / secs;

604

605 fprintf(stderr, "uuidmap_acquire_by_uuid(): %.2f ops/sec (%.2f usec/op)\n",

639 - ops, (double)(end_ut - start_ut) / successful);

606 + ops, (double)(end_ut - start_ut) / (double)successful);

607 }

608

609 // Phase 2: Delete everything

src/streaming/stream-connector.c

+35

@@ -629,6 +629,41 @@ static void *stream_connector_thread(void *ptr) {

629 return NULL;

630 }

631

632 +void stream_connector_remove_host(RRDHOST *host) {

633 + if(!host || !host->sender) return;

634 +

635 + struct connector *sc = stream_connector_get(host->sender);

636 +

637 + spinlock_lock(&sc->queue.spinlock);

638 + Word_t idx = 0;

639 + for(struct sender_state *s = SENDERS_FIRST(&sc->queue.senders, &idx);

640 + s;

641 + s = SENDERS_NEXT(&sc->queue.senders, &idx)) {

642 +

643 + if(s != host->sender)

644 + continue;

645 +

646 + ND_LOG_STACK lgs[] = {

647 + ND_LOG_FIELD_STR(NDF_NIDL_NODE, s->host->hostname),

648 + ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_to_parent_msgid),

649 + ND_LOG_FIELD_END(),

650 + };

651 + ND_LOG_STACK_PUSH(lgs);

652 +

653 + SENDERS_DEL(&sc->queue.senders, idx);

654 + spinlock_unlock(&sc->queue.spinlock);

655 +

656 + // do not have the connector lock when calling these

657 + stream_sender_on_disconnect(s);

658 + stream_sender_remove(s, s->exit.reason);

659 +

660 + spinlock_lock(&sc->queue.spinlock);

661 + break;

662 + }

663 +

664 + spinlock_unlock(&sc->queue.spinlock);

665 +}

666 +

667 bool stream_connector_init(struct sender_state *s) {

668 static SPINLOCK spinlock = SPINLOCK_INITIALIZER;

669 if(!s) return false;

src/streaming/stream-receiver.c

+18 -25

@@ -441,7 +441,19 @@ void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct re

441 "Failed to add receiver socket to nd_poll()",

442 sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port);

443

444 - // put the client IP and port into the buffers used by plugins.d

444 + rpt->thread.compressed.start = 0;

445 + rpt->thread.compressed.used = 0;

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

447 + buffered_reader_init(&rpt->thread.uncompressed);

448 +

449 + rpt->thread.line_buffer = buffer_create(sizeof(rpt->thread.uncompressed.read_buffer), NULL);

450 +

451 + // help preferred_sender_buffer() select the right buffer

452 + rpt->host->stream.snd.commit.receiver_tid = gettid_cached();

453 +

454 + rpt->replication.last_progress_ut = now_monotonic_usec();

455 +

456 + PARSER *parser = NULL;

457 {

458 char buf[CONFIG_MAX_NAME];

459 snprintfz(buf, sizeof(buf), "[%s]:%s", rpt->remote_ip, rpt->remote_port);

@@ -456,31 +468,12 @@ void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct re

468

469 string_freez(rpt->thread.cd.cmd);

470 rpt->thread.cd.cmd = NULL;

459 - }

471

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

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

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

464 - buffered_reader_init(&rpt->thread.uncompressed);

465 -

466 - rpt->thread.line_buffer = buffer_create(sizeof(rpt->thread.uncompressed.read_buffer), NULL);

467 -

468 - // help preferred_sender_buffer() select the right buffer

469 - rpt->host->stream.snd.commit.receiver_tid = gettid_cached();

470 -

471 - rpt->replication.last_progress_ut = now_monotonic_usec();

472 -

473 - PARSER *parser = NULL;

474 - {

475 - rpt->thread.cd = (struct plugind){

476 - .update_every = nd_profile.update_every,

477 - .unsafe = {

478 - .spinlock = SPINLOCK_INITIALIZER,

479 - .running = true,

480 - .enabled = true,

481 - },

482 - .started_t = now_realtime_sec(),

483 - };

472 + rpt->thread.cd.update_every = (int)nd_profile.update_every;

473 + spinlock_init(&rpt->thread.cd.unsafe.spinlock);

474 + rpt->thread.cd.unsafe.running = true;

475 + rpt->thread.cd.unsafe.enabled = true;

476 + rpt->thread.cd.started_t = now_realtime_sec();

477

478 PARSER_USER_OBJECT user = {

479 .enabled = plugin_is_enabled(&rpt->thread.cd),

src/streaming/stream-sender-api.c

+3 -1

@@ -134,6 +134,8 @@ void stream_sender_signal_to_stop_and_wait(struct rrdhost *host, STREAM_HANDSHAK

134

135 stream_sender_send_opcode(host->sender, msg);

136

137 - while(wait && rrdhost_flag_check(host, RRDHOST_FLAG_STREAM_SENDER_ADDED))

137 + while(wait && rrdhost_flag_check(host, RRDHOST_FLAG_STREAM_SENDER_ADDED)) {

138 sleep_usec(10 * USEC_PER_MS);

139 + stream_connector_remove_host(host);

140 + }

141 }

src/streaming/stream.h

@@ -19,6 +19,7 @@ struct receiver_state;

19 void *stream_sender_start_localhost(void *ptr);

20 void stream_sender_start_host(struct rrdhost *host);

21 void stream_sender_signal_to_stop_and_wait(struct rrdhost *host, STREAM_HANDSHAKE reason, bool wait);

22 +void stream_connector_remove_host(RRDHOST *host);

24 // managing host sender structures

25 void stream_sender_structures_init(struct rrdhost *host, bool stream, STRING *parents, STRING *api_key, STRING *send_charts_matching);