1
// SPDX-License-Identifier: GPL-3.0-or-later
2
#include "rrdengine.h"
3
4
+void queue_journalfile_v2_migration(struct rrdengine_worker_config *wc)
5
+{
6
+ struct rrdeng_work *work_request;
7
+
8
+ if (unlikely(wc->running_journal_migration))
9
+ return;
10
+
11
+ work_request = mallocz(sizeof(*work_request));
12
+ work_request->req.data = work_request;
13
+ work_request->wc = wc;
14
+ work_request->count = 0;
15
+ work_request->rerun = false;
16
+ wc->running_journal_migration = 1;
17
+ if (unlikely(uv_queue_work(wc->loop, &work_request->req, start_journal_indexing, after_journal_indexing))) {
18
+ freez(work_request);
19
+ wc->running_journal_migration = 0;
20
+ }
21
+}
22
+
23
static void flush_transaction_buffer_cb(uv_fs_t* req)
24
{
25
struct generic_io_descriptor *io_descr = req->data;
66
io_descr->bytes = size;
67
io_descr->pos = journalfile->pos;
68
io_descr->req.data = io_descr;
69
+ io_descr->data = journalfile;
70
io_descr->completion = NULL;
71
72
io_descr->iov = uv_buf_init((void *)io_descr->buf, size);
113
return ctx->commit_log.buf + buf_pos;
114
}
115
116
+void generate_journalfilepath_v2(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
117
+{
118
+ (void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION_V2,
119
+ datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno);
120
+}
121
+
122
void generate_journalfilepath(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
123
{
124
(void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION,
130
journalfile->file = (uv_file)0;
131
journalfile->pos = 0;
132
journalfile->datafile = datafile;
133
+ journalfile->journal_data = NULL;
134
+ journalfile->journal_data_size = 0;
135
+ journalfile->JudyL_array = (Pvoid_t) NULL;
136
+ journalfile->data = NULL;
137
+ journalfile->file_index = 0;
138
+ journalfile->last_access = 0;
139
}
140
109
-int close_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
141
+static int close_uv_file(struct rrdengine_datafile *datafile, uv_file file)
142
{
111
- struct rrdengine_instance *ctx = datafile->ctx;
112
- uv_fs_t req;
143
int ret;
144
char path[RRDENG_PATH_MAX];
145
116
- generate_journalfilepath(datafile, path, sizeof(path));
117
-
118
- ret = uv_fs_close(NULL, &req, journalfile->file, NULL);
146
+ uv_fs_t req;
147
+ ret = uv_fs_close(NULL, &req, file, NULL);
148
if (ret < 0) {
149
+ generate_journalfilepath(datafile, path, sizeof(path));
150
error("uv_fs_close(%s): %s", path, uv_strerror(ret));
121
- ++ctx->stats.fs_errors;
151
+ ++datafile->ctx->stats.fs_errors;
152
rrd_stat_atomic_add(&global_fs_errors, 1);
153
}
154
uv_fs_req_cleanup(&req);
125
-
155
return ret;
156
}
157
158
+int close_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
159
+{
160
+ struct rrdengine_instance *ctx = datafile->ctx;
161
+ char path[RRDENG_PATH_MAX];
162
+
163
+ if (likely(journalfile->journal_data)) {
164
+ if (munmap(journalfile->journal_data, journalfile->journal_data_size)) {
165
+ generate_journalfilepath_v2(datafile, path, sizeof(path));
166
+ error("Failed to unmap journal index file for %s", path);
167
+ ++ctx->stats.fs_errors;
168
+ rrd_stat_atomic_add(&global_fs_errors, 1);
169
+ }
170
+ journalfile->journal_data = NULL;
171
+ journalfile->journal_data_size = 0;
172
+ return 0;
173
+ }
174
+
175
+ return close_uv_file(datafile, journalfile->file);
176
+}
177
+
178
int unlink_journal_file(struct rrdengine_journalfile *journalfile)
179
{
180
struct rrdengine_datafile *datafile = journalfile->datafile;
198
return ret;
199
}
200
152
-int destroy_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
201
+int destroy_journal_file_unsafe(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
202
{
203
struct rrdengine_instance *ctx = datafile->ctx;
204
uv_fs_t req;
205
int ret;
206
char path[RRDENG_PATH_MAX];
207
+ char path_v2[RRDENG_PATH_MAX];
208
209
generate_journalfilepath(datafile, path, sizeof(path));
210
+ generate_journalfilepath_v2(datafile, path_v2, sizeof(path));
211
212
ret = uv_fs_ftruncate(NULL, &req, journalfile->file, 0, NULL);
213
if (ret < 0) {
217
}
218
uv_fs_req_cleanup(&req);
219
169
- ret = uv_fs_close(NULL, &req, journalfile->file, NULL);
220
+ (void) close_uv_file(datafile, journalfile->file);
221
+
222
+ // This is the new journal v2 index file
223
+ ret = uv_fs_unlink(NULL, &req, path_v2, NULL);
224
if (ret < 0) {
171
- error("uv_fs_close(%s): %s", path, uv_strerror(ret));
225
+ error("uv_fs_fsunlink(%s): %s", path, uv_strerror(ret));
226
++ctx->stats.fs_errors;
227
rrd_stat_atomic_add(&global_fs_errors, 1);
228
}
236
}
237
uv_fs_req_cleanup(&req);
238
239
+ ++ctx->stats.journalfile_deletions;
240
++ctx->stats.journalfile_deletions;
241
242
+ if (journalfile->journal_data) {
243
+ if (munmap(journalfile->journal_data, journalfile->journal_data_size)) {
244
+ error("Failed to unmap index file %s", path_v2);
245
+ }
246
+ }
247
+
248
return ret;
249
}
250
288
uv_fs_req_cleanup(&req);
289
posix_memfree(superblock);
290
if (ret < 0) {
230
- destroy_journal_file(journalfile, datafile);
291
+ destroy_journal_file_unsafe(journalfile, datafile);
292
return ret;
293
}
294
427
uv_rwlock_wrlock(&pg_cache->metrics_index.lock);
428
PValue = JudyHSIns(&pg_cache->metrics_index.JudyHS_array, temp_id, sizeof(uuid_t), PJE0);
429
fatal_assert(NULL == *PValue); /* TODO: figure out concurrency model */
369
- *PValue = page_index = create_page_index(temp_id, ctx);
370
- page_index->prev = pg_cache->metrics_index.last_page_index;
371
- pg_cache->metrics_index.last_page_index = page_index;
430
+ *PValue = page_index = create_page_index(temp_id, ctx);
431
+ page_index->prev = pg_cache->metrics_index.last_page_index;
432
+ pg_cache->metrics_index.last_page_index = page_index;
433
uv_rwlock_wrunlock(&pg_cache->metrics_index.lock);
434
}
435
375
- descr = pg_cache_create_descr();
436
+ // Lookup this descriptor
437
+ Word_t p_time = start_time_ut / USEC_PER_SEC;
438
+ uv_rwlock_rdlock(&page_index->lock);
439
+ PValue = JudyLFirst(page_index->JudyL_array, &p_time, PJE0);
440
+ descr = (NULL == PValue) ? NULL: *PValue;
441
+ uv_rwlock_rdunlock(&page_index->lock);
442
+
443
+ bool descr_found = false;
444
+ if (unlikely(descr && descr->start_time_ut == start_time_ut)) {
445
+ // We have this descriptor already
446
+ descr_found = true;
447
+
448
+#ifdef NETDATA_INTERNAL_CHECKS
449
+ char uuid_str[UUID_STR_LEN];
450
+ uuid_unparse_lower(page_index->id, uuid_str);
451
+ internal_error(true, "REMOVING UUID %s with %lu, %lu length=%u (fileno=%u), extent database offset = %lu, size = %u",
452
+ uuid_str, start_time_ut, end_time_ut, jf_metric_data->descr[i].page_length, descr->extent->datafile->fileno,
453
+ descr->extent->offset, descr->extent->size);
454
+ internal_error(true, "APPLYING UUID %s with %lu, %lu length=%u (fileno=%u), extent database offset = %lu, size = %u",
455
+ uuid_str, start_time_ut, end_time_ut, jf_metric_data->descr[i].page_length, extent->datafile->fileno,
456
+ extent->offset, extent->size);
457
+#endif
458
+ // Remove entry from previous extent
459
+ unlink_descriptor_extent_unsafe(descr);
460
+ internal_error(true, "REMOVING UUID %s with %lu OK", uuid_str, start_time_ut);
461
+ }
462
+ else {
463
+ descr = pg_cache_create_descr();
464
+ descr->id = &page_index->id;
465
+ }
466
+
467
descr->page_length = jf_metric_data->descr[i].page_length;
468
descr->start_time_ut = start_time_ut;
469
descr->end_time_ut = end_time_ut;
470
descr->update_every_s = (update_every_s > 0) ? (uint32_t)update_every_s : (page_index->latest_update_every_s);
380
- descr->id = &page_index->id;
471
+
472
descr->extent = extent;
473
descr->type = page_type;
474
extent->pages[valid_pages++] = descr;
384
- pg_cache_insert(ctx, page_index, descr);
475
+ if (likely(!descr_found))
476
+ (void)pg_cache_insert(ctx, page_index, descr, true);
477
386
- if(page_index->latest_time_ut == descr->end_time_ut)
478
+ if (page_index->latest_time_ut == descr->end_time_ut)
479
page_index->latest_update_every_s = descr->update_every_s;
480
481
if(descr->update_every_s == 0)
390
- fatal(
391
- "DBENGINE: page descriptor update every is zero, end_time_ut = %llu, start_time_ut = %llu, entries = %zu",
482
+ fatal("DBENGINE: page descriptor update every is zero, end_time_ut = %llu, start_time_ut = %llu, entries = %zu",
483
(unsigned long long)end_time_ut, (unsigned long long)start_time_ut, entries);
484
}
485
486
extent->number_of_pages = valid_pages;
487
397
- if (likely(valid_pages))
488
+ if (likely(valid_pages)) {
489
df_extent_insert(extent);
490
+ }
491
else {
492
freez(extent);
493
ctx->load_errors[LOAD_ERRORS_DROPPED_EXTENT].counter++;
536
return size_bytes;
537
}
538
switch (jf_header->type) {
447
- case STORE_DATA:
448
- debug(D_RRDENGINE, "Replaying transaction %"PRIu64"", jf_header->id);
449
- restore_extent_metadata(ctx, journalfile, buf + sizeof(*jf_header), payload_length);
450
- break;
451
- default:
452
- error("Unknown transaction type. Skipping record.");
453
- break;
539
+ case STORE_DATA:
540
+ debug(D_RRDENGINE, "Replaying transaction %"PRIu64"", jf_header->id);
541
+ restore_extent_metadata(ctx, journalfile, buf + sizeof(*jf_header), payload_length);
542
+ break;
543
+ default:
544
+ error("Unknown transaction type. Skipping record.");
545
+ break;
546
}
547
548
return size_bytes;
610
if (likely(journal_is_mmapped))
611
buf += size_bytes;
612
}
521
-skip_file:
613
+ skip_file:
614
if (unlikely(!journal_is_mmapped))
615
posix_memfree(buf);
616
return max_id;
617
}
618
619
+bool unlink_descriptor_extent_unsafe(struct rrdeng_page_descr *descr)
620
+{
621
+ if (unlikely(!descr || !descr->extent))
622
+ return true;
623
+
624
+ struct extent_info *extent = descr->extent;
625
+ for (uint8_t index = 0; index < extent->number_of_pages; index++) {
626
+ if (extent->pages[index] == descr) {
627
+ extent->pages[index] = NULL;
628
+ return true;
629
+ }
630
+ }
631
+ return false;
632
+}
633
+
634
+// Checks that the extent list checksum is valid
635
+static int check_journal_v2_extent_list (void *data_start, size_t file_size)
636
+{
637
+ UNUSED(file_size);
638
+ uLong crc;
639
+
640
+ struct journal_v2_header *j2_header = (void *) data_start;
641
+ struct journal_v2_block_trailer *journal_v2_trailer;
642
+
643
+ journal_v2_trailer = (struct journal_v2_block_trailer *) ((uint8_t *) data_start + j2_header->extent_trailer_offset);
644
+ crc = crc32(0L, Z_NULL, 0);
645
+ crc = crc32(crc, (uint8_t *) data_start + j2_header->extent_offset, j2_header->extent_count * sizeof(struct journal_extent_list));
646
+ if (unlikely(crc32cmp(journal_v2_trailer->checksum, crc))) {
647
+ error("Extent list CRC32 check: FAILED");
648
+ return 1;
649
+ }
650
+
651
+ return 0;
652
+}
653
+
654
+// Checks that the metric list (UUIDs) checksum is valid
655
+static int check_journal_v2_metric_list(void *data_start, size_t file_size)
656
+{
657
+ UNUSED(file_size);
658
+ uLong crc;
659
+
660
+ struct journal_v2_header *j2_header = (void *) data_start;
661
+ struct journal_v2_block_trailer *journal_v2_trailer;
662
+
663
+ journal_v2_trailer = (struct journal_v2_block_trailer *) ((uint8_t *) data_start + j2_header->metric_trailer_offset);
664
+ crc = crc32(0L, Z_NULL, 0);
665
+ crc = crc32(crc, (uint8_t *) data_start + j2_header->metric_offset, j2_header->metric_count * sizeof(struct journal_metric_list));
666
+ if (unlikely(crc32cmp(journal_v2_trailer->checksum, crc))) {
667
+ error("Metric list CRC32 check: FAILED");
668
+ return 1;
669
+ }
670
+ return 0;
671
+}
672
+
673
+static int check_journal_v2_file(void *data_start, size_t file_size, uint32_t original_size)
674
+{
675
+ int rc;
676
+ uLong crc;
677
+
678
+ struct journal_v2_header *j2_header = (void *) data_start;
679
+ struct journal_v2_block_trailer *journal_v2_trailer;
680
+
681
+ if (j2_header->magic == JOURVAL_V2_REBUILD_MAGIC)
682
+ return 2;
683
+
684
+ // Magic failure
685
+ if (j2_header->magic != JOURVAL_V2_MAGIC)
686
+ return 1;
687
+
688
+ if (j2_header->total_file_size != file_size)
689
+ return 1;
690
+
691
+ if (original_size && j2_header->original_file_size != original_size)
692
+ return 1;
693
+
694
+ journal_v2_trailer = (struct journal_v2_block_trailer *) ((uint8_t *) data_start + file_size - sizeof(*journal_v2_trailer));
695
+
696
+ crc = crc32(0L, Z_NULL, 0);
697
+ crc = crc32(crc, (void *) j2_header, sizeof(*j2_header));
698
+
699
+ rc = crc32cmp(journal_v2_trailer->checksum, crc);
700
+ if (unlikely(rc)) {
701
+ error("File CRC32 check: FAILED");
702
+ return 1;
703
+ }
704
+
705
+ rc = check_journal_v2_extent_list(data_start, file_size);
706
+ if (rc) return 1;
707
+
708
+ rc = check_journal_v2_metric_list(data_start, file_size);
709
+ if (rc) return 1;
710
+
711
+ if (!db_engine_journal_check)
712
+ return 0;
713
+
714
+ // Verify complete UUID chain
715
+
716
+ struct journal_metric_list *metric = (void *) (data_start + j2_header->metric_offset);
717
+
718
+ unsigned verified = 0;
719
+ unsigned entries;
720
+ unsigned total_pages = 0;
721
+
722
+ info("Checking %u metrics that exist in the journal", j2_header->metric_count);
723
+ for (entries = 0; entries < j2_header->metric_count; entries++) {
724
+
725
+ char uuid_str[UUID_STR_LEN];
726
+ uuid_unparse_lower(metric->uuid, uuid_str);
727
+ struct journal_page_header *metric_list_header = (void *) (data_start + metric->page_offset);
728
+ struct journal_page_header local_metric_list_header = *metric_list_header;
729
+
730
+ local_metric_list_header.crc = JOURVAL_V2_MAGIC;
731
+
732
+ crc = crc32(0L, Z_NULL, 0);
733
+ crc = crc32(crc, (void *) &local_metric_list_header, sizeof(local_metric_list_header));
734
+ rc = crc32cmp(metric_list_header->checksum, crc);
735
+
736
+ internal_error(true, "Index %u : %s entries %u at offset %u (%llu -- %llu) verified, HEADER CRC computed %lu, stored %u", entries, uuid_str, metric->entries, metric->page_offset,
737
+ j2_header->start_time_ut + (usec_t) metric->delta_start * USEC_PER_SEC, j2_header->start_time_ut + (usec_t) metric->delta_end * USEC_PER_SEC,
738
+ crc, metric_list_header->crc);
739
+ if (!rc) {
740
+ struct journal_v2_block_trailer *journal_trailer =
741
+ (void *) data_start + metric->page_offset + sizeof(struct journal_page_header) + (metric_list_header->entries * sizeof(struct journal_page_list));
742
+
743
+ crc = crc32(0L, Z_NULL, 0);
744
+ crc = crc32(crc, (uint8_t *) metric_list_header + sizeof(struct journal_page_header), metric_list_header->entries * sizeof(struct journal_page_list));
745
+ rc = crc32cmp(journal_trailer->checksum, crc);
746
+ internal_error(rc, "Index %u : %s entries %u at offset %u verified, DATA CRC computed %lu, stored %u", entries, uuid_str, metric->entries, metric->page_offset,
747
+ crc, metric_list_header->crc);
748
+ if (!rc) {
749
+ total_pages += metric_list_header->entries;
750
+ verified++;
751
+ }
752
+ }
753
+
754
+ metric++;
755
+ if (((uint8_t *) metric - (uint8_t *) data_start) > (uint32_t) file_size) {
756
+ info("Verification failed EOF reached -- total entries %u, verified %u", entries, verified);
757
+ return 1;
758
+ }
759
+ }
760
+
761
+ if (entries != verified) {
762
+ info("Verification failed -- total entries %u, verified %u", entries, verified);
763
+ return 1;
764
+ }
765
+ info("Verification succeeded -- total entries %u, verified %u (%u total pages)", entries, verified, total_pages);
766
+
767
+ return 0;
768
+}
769
+
770
+int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
771
+{
772
+ struct page_cache *pg_cache = &ctx->pg_cache;
773
+ struct pg_cache_page_index *page_index;
774
+ int ret, fd;
775
+ uint64_t file_size;
776
+ char path[RRDENG_PATH_MAX];
777
+ struct stat statbuf;
778
+ uint32_t original_file_size = 0;
779
+
780
+ generate_journalfilepath(datafile, path, sizeof(path));
781
+ ret = stat(path, &statbuf);
782
+ if (!ret)
783
+ original_file_size = (uint32_t)statbuf.st_size;
784
+
785
+ generate_journalfilepath_v2(datafile, path, sizeof(path));
786
+
787
+ fd = open(path, O_RDONLY);
788
+ if (fd < 0) {
789
+ if (errno == ENOENT)
790
+ return 1;
791
+ ++ctx->stats.fs_errors;
792
+ rrd_stat_atomic_add(&global_fs_errors, 1);
793
+ error("Failed to open %s", path);
794
+ return 1;
795
+ }
796
+
797
+ ret = fstat(fd, &statbuf);
798
+ if (ret) {
799
+ error("Failed to get file information for %s", path);
800
+ close(fd);
801
+ return 1;
802
+ }
803
+
804
+ file_size = (size_t)statbuf.st_size;
805
+
806
+ if (file_size < sizeof(struct journal_v2_header)) {
807
+ error_report("Invalid file %s. Not the expected size", path);
808
+ close(fd);
809
+ return 1;
810
+ }
811
+
812
+ usec_t start_loading = now_realtime_usec();
813
+ uint8_t *data_start = mmap(NULL, file_size, PROT_READ, MAP_SHARED, fd, 0);
814
+ if (data_start == MAP_FAILED) {
815
+ close(fd);
816
+ return 1;
817
+ }
818
+ close(fd);
819
+
820
+ info("Checking integrity of %s", path);
821
+ int rc = check_journal_v2_file(data_start, file_size, original_file_size);
822
+ if (rc) {
823
+ if (rc == 2)
824
+ error_report("File %s needs to be rebuilt", path);
825
+ else
826
+ error_report("File %s is invalid", path);
827
+ if (unlikely(munmap(data_start, file_size)))
828
+ error("Failed to unmap %s", path);
829
+ return 1;
830
+ }
831
+
832
+ struct journal_v2_header *j2_header = (void *) data_start;
833
+
834
+ size_t entries = j2_header->metric_count;
835
+
836
+ if (!entries) {
837
+ if (unlikely(munmap(data_start, file_size)))
838
+ error("Failed to unmap %s", path);
839
+ return 1;
840
+ }
841
+
842
+ rc = madvise(data_start, file_size, MADV_DONTFORK);
843
+ if (rc)
844
+ error("MADV_DONTFORK: setting failed");
845
+
846
+ rc = madvise(data_start, file_size, MADV_DONTDUMP);
847
+ if (rc)
848
+ error("MADV_DONTDUMP: setting failed");
849
+
850
+ struct journal_metric_list *metric = (struct journal_metric_list *) (data_start + j2_header->metric_offset);
851
+
852
+ uv_rwlock_wrlock(&pg_cache->metrics_index.lock);
853
+
854
+ // Initialize the journal file to be able to access the data
855
+ journalfile->journal_data = data_start;
856
+ journalfile->journal_data_size = file_size;
857
+
858
+ usec_t header_start_time = j2_header->start_time_ut;
859
+ for (size_t i=0; i < entries; i++) {
860
+ Pvoid_t *PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, metric->uuid, sizeof(uuid_t));
861
+ if (likely(NULL != PValue)) {
862
+ page_index = *PValue;
863
+ }
864
+ else {
865
+ PValue = JudyHSIns(&pg_cache->metrics_index.JudyHS_array, metric->uuid, sizeof(uuid_t), PJE0);
866
+ fatal_assert(NULL == *PValue);
867
+ *PValue = page_index = create_page_index(&metric->uuid, ctx);
868
+ page_index->oldest_time_ut = LLONG_MAX;
869
+ page_index->latest_time_ut = 0;
870
+ page_index->prev = pg_cache->metrics_index.last_page_index;
871
+ pg_cache->metrics_index.last_page_index = page_index;
872
+ }
873
+
874
+ usec_t metric_start_ut = header_start_time + (usec_t ) metric->delta_start * USEC_PER_SEC;
875
+ usec_t metric_end_ut = header_start_time + (usec_t ) metric->delta_end * USEC_PER_SEC;
876
+
877
+ if (page_index->oldest_time_ut > metric_start_ut)
878
+ page_index->oldest_time_ut = metric_start_ut;
879
+
880
+ if (page_index->latest_time_ut < metric_end_ut)
881
+ page_index->latest_time_ut = metric_end_ut;
882
+
883
+ ++page_index->page_count;
884
+ ++pg_cache->page_descriptors;
885
+ metric++;
886
+ }
887
+
888
+ uv_rwlock_wrunlock(&pg_cache->metrics_index.lock);
889
+
890
+ info("Journal file \"%s\" loaded (size:%"PRIu64") with %lu metrics in %d ms",
891
+ path, file_size, entries,
892
+ (int) ((now_realtime_usec() - start_loading) / USEC_PER_MS));
893
+ return 0;
894
+}
895
+
896
+
897
+struct metric_info_s {
898
+ uuid_t *id;
899
+ struct pg_cache_page_index *page_index;
900
+ uint32_t entries;
901
+ time_t min_index_time_s;
902
+ time_t max_index_time_s;
903
+ usec_t min_time_ut;
904
+ usec_t max_time_ut;
905
+ uint32_t page_list_header;
906
+ Pvoid_t JudyL_array;
907
+};
908
+
909
+struct journal_metric_list_to_sort {
910
+ struct metric_info_s *metric_info;
911
+};
912
+
913
+static int journal_metric_compare (const void *item1, const void *item2)
914
+{
915
+ const struct metric_info_s *metric1 = ((struct journal_metric_list_to_sort *) item1)->metric_info;
916
+ const struct metric_info_s *metric2 = ((struct journal_metric_list_to_sort *) item2)->metric_info;
917
+
918
+ return uuid_compare(*(metric1->id), *(metric2->id));
919
+}
920
+
921
+
922
+// Write list of extents for the journalfile
923
+void *journal_v2_write_extent_list(struct rrdengine_journalfile *journalfile, void *data)
924
+{
925
+ struct extent_info *extent = journalfile->datafile->extents.first;
926
+ struct journal_extent_list *j2_extent = (void *) data;
927
+ while (extent) {
928
+ j2_extent->datafile_offset = extent->offset;
929
+ j2_extent->datafile_size = extent->size;
930
+ j2_extent->pages = extent->number_of_pages;
931
+ j2_extent->file_index = journalfile->file_index;
932
+ j2_extent++;
933
+ extent = extent->next;
934
+ };
935
+ return j2_extent;
936
+}
937
+
938
+static int verify_journal_space(struct journal_v2_header *j2_header, void *data, uint32_t bytes)
939
+{
940
+ if ((unsigned long)(((uint8_t *) data - (uint8_t *) j2_header->data) + bytes) > (j2_header->total_file_size - sizeof(struct journal_v2_block_trailer)))
941
+ return 1;
942
+
943
+ return 0;
944
+}
945
+
946
+void *journal_v2_write_metric_page(struct journal_v2_header *j2_header, void *data, struct metric_info_s *metric_info, uint32_t pages_offset)
947
+{
948
+ struct journal_metric_list *metric = (void *) data;
949
+
950
+ if (verify_journal_space(j2_header, data, sizeof(*metric)))
951
+ return NULL;
952
+
953
+ uuid_copy(metric->uuid, *metric_info->id);
954
+ metric->entries = metric_info->entries;
955
+ metric->page_offset = pages_offset;
956
+ metric->delta_start = (metric_info->min_time_ut - j2_header->start_time_ut) / USEC_PER_SEC;
957
+ metric->delta_end = (metric_info->max_time_ut - j2_header->start_time_ut) / USEC_PER_SEC;
958
+
959
+ return ++metric;
960
+}
961
+
962
+void *journal_v2_write_data_page_header(struct journal_v2_header *j2_header __maybe_unused, void *data, struct metric_info_s *metric_info, uint32_t uuid_offset)
963
+{
964
+ struct journal_page_header *data_page_header = (void *) data;
965
+ uLong crc;
966
+
967
+ uuid_copy(data_page_header->uuid, *metric_info->id);
968
+ data_page_header->entries = metric_info->entries;
969
+ data_page_header->uuid_offset = uuid_offset; // data header OFFSET poings to METRIC in the directory
970
+ data_page_header->crc = JOURVAL_V2_MAGIC;
971
+ crc = crc32(0L, Z_NULL, 0);
972
+ crc = crc32(crc, (void *) data_page_header, sizeof(*data_page_header));
973
+ crc32set(data_page_header->checksum, crc);
974
+ return ++data_page_header;
975
+}
976
+
977
+void *journal_v2_write_data_page_trailer(struct journal_v2_header *j2_header __maybe_unused, void *data, void *page_header)
978
+{
979
+ struct journal_page_header *data_page_header = (void *) page_header;
980
+ struct journal_v2_block_trailer *journal_trailer = (void *) data;
981
+ uLong crc;
982
+
983
+ crc = crc32(0L, Z_NULL, 0);
984
+ crc = crc32(crc, (uint8_t *) page_header + sizeof(struct journal_page_header), data_page_header->entries * sizeof(struct journal_page_list));
985
+ crc32set(journal_trailer->checksum, crc);
986
+ return ++journal_trailer;
987
+}
988
+
989
+void *journal_v2_write_data_page(struct journal_v2_header *j2_header, void *data, struct rrdeng_page_descr *descr)
990
+{
991
+ if (unlikely(!descr))
992
+ return data;
993
+
994
+ struct journal_page_list *data_page = data;
995
+
996
+ // verify that we can write number of bytes
997
+ if (verify_journal_space(j2_header, data, sizeof(*data_page)))
998
+ return NULL;
999
+
1000
+ fatal_assert(descr->extent != NULL);
1001
+
1002
+ uint32_t extent_index = unlikely(NULL == descr->extent) ? UINT32_MAX : descr->extent->index;
1003
+
1004
+ data_page->delta_start_s = (descr->start_time_ut - j2_header->start_time_ut) / USEC_PER_SEC;
1005
+ data_page->delta_end_s = (descr->end_time_ut - j2_header->start_time_ut) / USEC_PER_SEC;
1006
+ data_page->extent_index = extent_index;
1007
+ data_page->update_every_s = (uint16_t) descr->update_every_s;
1008
+ data_page->page_length = descr->page_length;
1009
+ data_page->type = descr->type;
1010
+
1011
+ // Rebuild on start to resolve unknown entry
1012
+ if (unlikely(UINT32_MAX == extent_index))
1013
+ j2_header->magic = JOURVAL_V2_REBUILD_MAGIC;
1014
+
1015
+ return ++data_page;
1016
+}
1017
+
1018
+// For a page_index write all descr @ time entries
1019
+// Must be recorded in metric_info->entries
1020
+void *journal_v2_write_descriptors(struct journal_v2_header *j2_header, void *data, struct metric_info_s *metric_info, struct rrdengine_journalfile *journalfile)
1021
+{
1022
+ struct rrdeng_page_descr *descr;
1023
+ Pvoid_t *PValue;
1024
+
1025
+ struct journal_page_list *data_page = (void *)data;
1026
+ struct page_cache *pg_cache = &journalfile->datafile->ctx->pg_cache;
1027
+ struct pg_cache_page_index *page_index;
1028
+
1029
+ uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
1030
+ PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, metric_info->id, sizeof(uuid_t));
1031
+ page_index = (NULL == PValue) ? NULL : *PValue;
1032
+ uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
1033
+
1034
+ if (page_index == NULL)
1035
+ return data_page;
1036
+
1037
+ // We need to write all descriptors with index metric_info->min_index_time_s, metric_info->max_index_time_s
1038
+ // that belong to this journal file
1039
+
1040
+ Word_t index_time = metric_info->min_index_time_s;
1041
+ unsigned entries = 0;
1042
+
1043
+ uv_rwlock_rdlock(&page_index->lock);
1044
+
1045
+ Pvoid_t JudyL_array = metric_info->JudyL_array ? metric_info->JudyL_array : page_index->JudyL_array;
1046
+
1047
+ // Need page_index lock if running live
1048
+ for (PValue = JudyLFirst(JudyL_array, &index_time, PJE0),
1049
+ descr = unlikely(NULL == PValue) ? NULL : *PValue;
1050
+ descr != NULL;
1051
+ PValue = JudyLNext(JudyL_array, &index_time, PJE0),
1052
+ descr = unlikely(NULL == PValue) ? NULL : *PValue) {
1053
+
1054
+ if (unlikely((time_t) index_time > metric_info->max_index_time_s) || entries == metric_info->entries)
1055
+ break;
1056
+
1057
+ // Write one descriptor and return the next data page location
1058
+ data_page = journal_v2_write_data_page(j2_header, (void *)data_page, descr);
1059
+
1060
+ entries++;
1061
+ if (unlikely(!data_page))
1062
+ break;
1063
+ }
1064
+
1065
+ uv_rwlock_rdunlock(&page_index->lock);
1066
+ return data_page;
1067
+}
1068
+
1069
+static void journal_v2_remove_active_descriptors(struct rrdengine_journalfile *journalfile, struct metric_info_s *metric_info, bool startup)
1070
+{
1071
+ if (true == startup) {
1072
+ // This is during startup, so we are the only ones accessing the structures
1073
+ // thats why we can safely remote the entire page_index->JudyL_array
1074
+ struct rrdeng_page_descr *descr;
1075
+ Word_t index_time;
1076
+ Pvoid_t *PValue;
1077
+ struct pg_cache_page_index *page_index;
1078
+
1079
+ page_index = metric_info->page_index;
1080
+
1081
+ for (index_time = 0, PValue = JudyLFirst(page_index->JudyL_array, &index_time, PJE0),
1082
+ descr = unlikely(NULL == PValue) ? NULL : *PValue; descr != NULL;
1083
+ PValue = JudyLNext(page_index->JudyL_array, &index_time, PJE0),
1084
+ descr = unlikely(NULL == PValue) ? NULL : *PValue) {
1085
+
1086
+ rrdeng_page_descr_freez(descr);
1087
+ }
1088
+ (void)JudyLFreeArray(&page_index->JudyL_array, PJE0);
1089
+ }
1090
+ else {
1091
+ // This is during runtime
1092
+ struct rrdeng_page_descr *descr;
1093
+ Pvoid_t *PValue;
1094
+ struct pg_cache_page_index *page_index = metric_info->page_index;
1095
+ struct page_cache *pg_cache = &page_index->ctx->pg_cache;
1096
+ struct rrdengine_instance *ctx = page_index->ctx;
1097
+
1098
+ Word_t index_time = metric_info->min_index_time_s;
1099
+ uint32_t metric_info_offset = metric_info->page_list_header;
1100
+
1101
+ struct journal_page_header *page_list_header = (struct journal_page_header *) ((uint8_t *) journalfile->journal_data + metric_info_offset);
1102
+ struct journal_v2_header *journal_header = (struct journal_v2_header *) journalfile->journal_data;
1103
+ // Sanity check that we refer to the same UUID
1104
+ fatal_assert(uuid_compare(page_list_header->uuid, *metric_info->id) == 0);
1105
+
1106
+ struct journal_page_list *page_list = (struct journal_page_list *)((uint8_t *) page_list_header + sizeof(*page_list_header));
1107
+ struct journal_extent_list *extent_list = (void *)((uint8_t *)journal_header + journal_header->extent_offset);
1108
+
1109
+ uint32_t index = 0;
1110
+ uint32_t entries = page_list_header->entries;
1111
+
1112
+ uv_rwlock_rdlock(&page_index->lock);
1113
+
1114
+ bool mark_journalfile_for_expiration_check = false;
1115
+ for (PValue = JudyLFirst(metric_info->JudyL_array, &index_time, PJE0),
1116
+ descr = unlikely(NULL == PValue) ? NULL : *PValue;
1117
+ descr != NULL;
1118
+ PValue = JudyLNext(metric_info->JudyL_array, &index_time, PJE0),
1119
+ descr = unlikely(NULL == PValue) ? NULL : *PValue) {
1120
+
1121
+ if (unlikely((time_t) index_time > metric_info->max_index_time_s) || index == entries)
1122
+ break;
1123
+
1124
+ if (descr->extent_entry || (!descr->extent_entry && descr->extent && descr->extent->datafile->journalfile != journalfile))
1125
+ continue;
1126
+
1127
+ struct journal_page_list *page_entry = &page_list[index++];
1128
+
1129
+ if (likely(page_entry->extent_index != UINT32_MAX)) {
1130
+
1131
+ fatal_assert(descr->extent->offset == extent_list[page_entry->extent_index].datafile_offset);
1132
+ fatal_assert(descr->extent->size == extent_list[page_entry->extent_index].datafile_size);
1133
+
1134
+ rrdeng_page_descr_mutex_lock(ctx, descr);
1135
+ while (!pg_cache_try_get_unsafe(descr, 1)) {
1136
+ pg_cache_wait_event_unsafe(descr);
1137
+ }
1138
+
1139
+ descr->extent_entry = &extent_list[page_entry->extent_index];
1140
+ descr->extent = NULL;
1141
+ descr->file = journalfile->datafile->file;
1142
+ ++pg_cache->active_descriptors;
1143
+ pg_cache_put_unsafe(descr);
1144
+ rrdeng_try_deallocate_pg_cache_descr(ctx, descr);
1145
+ rrdeng_page_descr_mutex_unlock(ctx, descr);
1146
+ mark_journalfile_for_expiration_check = true;
1147
+ }
1148
+ }
1149
+
1150
+ if (mark_journalfile_for_expiration_check) {
1151
+ uint32_t page_offset = (uint8_t *)page_list_header - (uint8_t *)journalfile->journal_data;
1152
+ mark_journalfile_descriptor(pg_cache, journalfile, page_offset, 1);
1153
+ }
1154
+
1155
+ uv_rwlock_rdunlock(&page_index->lock);
1156
+ }
1157
+}
1158
+
1159
+bool descriptor_is_corrupted(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr)
1160
+{
1161
+ struct page_cache *pg_cache = &ctx->pg_cache;
1162
+ struct pg_cache_page_index *page_index = get_page_index(pg_cache, descr->id);
1163
+
1164
+ if (unlikely(!page_index))
1165
+ return true;
1166
+
1167
+ time_t index_time_s = (time_t) (descr->start_time_ut / USEC_PER_SEC);
1168
+ struct rrdeng_page_descr *idx_descr = get_descriptor(page_index, index_time_s);
1169
+
1170
+ bool is_corrupted = (idx_descr != descr);
1171
+
1172
+#ifdef NETDATA_INTERNAL_CHECKS
1173
+ char uuid_str[UUID_STR_LEN];
1174
+ uuid_unparse_lower(page_index->id, uuid_str);
1175
+ internal_error(is_corrupted, "Descriptor corrupted (Extent %p Judy %p) @ %ld", descr, idx_descr, index_time_s);
1176
+#endif
1177
+
1178
+ return is_corrupted;
1179
+}
1180
+
1181
+static bool journalfile_ready_to_index(struct rrdengine_datafile *datafile)
1182
+{
1183
+ struct extent_info *extent = datafile->extents.first;
1184
+ while (extent) {
1185
+ uint8_t extent_pages = extent->number_of_pages;
1186
+ for (uint8_t index = 0; index < extent_pages; index++) {
1187
+ struct rrdeng_page_descr *descr = extent->pages[index];
1188
+ if (unlikely(!descr))
1189
+ continue;
1190
+ if (unlikely(!descr->extent))
1191
+ return false;
1192
+ }
1193
+ extent = extent->next;
1194
+ }
1195
+ return true;
1196
+}
1197
+
1198
+// Migrate the journalfile pointed by datafile
1199
+// activate : make the new file active immediately
1200
+// journafile data will be set and descriptors (if deleted) will be repopulated as needed
1201
+// startup : if the migration is done during agent startup
1202
+// this will allow us to optimize certain things
1203
+void migrate_journal_file_v2(struct rrdengine_datafile *datafile, bool activate, bool startup)
1204
+{
1205
+ char path[RRDENG_PATH_MAX];
1206
+ size_t number_of_extents = 0; // Number of extents
1207
+ size_t number_of_metrics = 0; // Number of unique metrics (UUIDS)
1208
+ size_t number_of_pages = 0; // Total number of descriptors @ time
1209
+ Pvoid_t *PValue;
1210
+ Pvoid_t metrics_JudyL_array = NULL;
1211
+ Pvoid_t metrics_JudyHS_array = NULL;
1212
+ struct rrdengine_instance *ctx = datafile->ctx;
1213
+ struct rrdengine_journalfile *journalfile = datafile->journalfile;
1214
+ usec_t min_time_ut = LLONG_MAX;
1215
+ usec_t max_time_ut = 0;
1216
+ struct metric_info_s *metric_info;
1217
+
1218
+ // Do nothing if we already have a mmaped file
1219
+ if (unlikely(journalfile->journal_data))
1220
+ return;
1221
+
1222
+ generate_journalfilepath_v2(datafile, path, sizeof(path));
1223
+ info("Indexing file %s", path);
1224
+
1225
+#ifdef NETDATA_INTERNAL_CHECKS
1226
+ usec_t start_loading = now_realtime_usec();
1227
+#endif
1228
+
1229
+ if (false == startup)
1230
+ uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1231
+
1232
+ struct extent_info *extent = datafile->extents.first;
1233
+ while (extent) {
1234
+ uint8_t extent_pages = extent->number_of_pages;
1235
+ for (uint8_t index = 0; index < extent_pages; index++) {
1236
+ struct rrdeng_page_descr *descr = extent->pages[index];
1237
+
1238
+ if (unlikely(!descr))
1239
+ continue;
1240
+
1241
+ if (false == startup) {
1242
+ if (unlikely(descriptor_is_corrupted(ctx, descr))) {
1243
+ extent->pages[index] = NULL;
1244
+ continue;
1245
+ }
1246
+ }
1247
+
1248
+ PValue = JudyHSGet(metrics_JudyHS_array, descr->id, sizeof(uuid_t));
1249
+ if (likely(NULL != PValue)) {
1250
+ metric_info = *PValue;
1251
+ }
1252
+ else {
1253
+ PValue = JudyHSIns(&metrics_JudyHS_array, descr->id, sizeof(uuid_t), PJE0);
1254
+ *PValue = metric_info = mallocz(sizeof(*metric_info));
1255
+
1256
+ metric_info->entries =0;
1257
+ metric_info->min_time_ut = LLONG_MAX;
1258
+ metric_info->max_time_ut = 0;
1259
+ metric_info->min_index_time_s = LLONG_MAX;
1260
+ metric_info->max_index_time_s = 0;
1261
+ metric_info->id = descr->id;
1262
+ metric_info->page_index = NULL;
1263
+ metric_info->page_list_header = 0;
1264
+ metric_info->JudyL_array = (Pvoid_t) NULL;
1265
+
1266
+ PValue = JudyLIns(&metrics_JudyL_array,number_of_metrics, PJE0);
1267
+ *PValue = metric_info;
1268
+ number_of_metrics++;
1269
+ }
1270
+ time_t current_index_time_s = (time_t) (descr->start_time_ut / USEC_PER_SEC);
1271
+
1272
+ if (metric_info->min_time_ut > descr->start_time_ut) {
1273
+ metric_info->min_time_ut = descr->start_time_ut;
1274
+ metric_info->min_index_time_s = current_index_time_s;
1275
+ }
1276
+
1277
+ metric_info->max_index_time_s= MAX(metric_info->max_index_time_s, current_index_time_s);
1278
+ metric_info->max_time_ut = MAX(metric_info->max_time_ut , descr->end_time_ut);
1279
+
1280
+ if (false == startup) {
1281
+ PValue = JudyLIns(&metric_info->JudyL_array, current_index_time_s, PJE0);
1282
+ fatal_assert(NULL != PValue && NULL == *PValue);
1283
+ *PValue = descr;
1284
+ }
1285
+
1286
+ metric_info->entries++;
1287
+ number_of_pages++;
1288
+
1289
+ // Maintain the min max times to add to the journal header
1290
+ min_time_ut = MIN(min_time_ut, descr->start_time_ut);
1291
+ max_time_ut = MAX(max_time_ut, descr->end_time_ut);
1292
+ }
1293
+ extent->index = number_of_extents++;
1294
+ extent = extent->next;
1295
+ }
1296
+ info("File %s has %lu extents", path, number_of_extents);
1297
+
1298
+ if (false == startup)
1299
+ uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1300
+
1301
+ internal_error(true, "Scan and extbuild metric %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1302
+
1303
+ // Calculate total jourval file size
1304
+ size_t total_file_size = 0;
1305
+ total_file_size += (sizeof(struct journal_v2_header) + JOURNAL_V2_HEADER_PADDING_SZ);
1306
+
1307
+ // Extents will start here
1308
+ uint32_t extent_offset = total_file_size;
1309
+ total_file_size += (number_of_extents * sizeof(struct journal_extent_list));
1310
+
1311
+ uint32_t extent_offset_trailer = total_file_size;
1312
+ total_file_size += sizeof(struct journal_v2_block_trailer);
1313
+
1314
+ // UUID list will start here
1315
+ uint32_t metrics_offset = total_file_size;
1316
+ total_file_size += (number_of_metrics * sizeof(struct journal_metric_list));
1317
+
1318
+ // UUID list trailer
1319
+ uint32_t metric_offset_trailer = total_file_size;
1320
+ total_file_size += sizeof(struct journal_v2_block_trailer);
1321
+
1322
+ // descr @ time will start here
1323
+ uint32_t pages_offset = total_file_size;
1324
+ total_file_size += (number_of_pages * (sizeof(struct journal_page_list) + sizeof(struct journal_page_header) + sizeof(struct journal_v2_block_trailer)));
1325
+
1326
+ // File trailer
1327
+ uint32_t trailer_offset = total_file_size;
1328
+ total_file_size += sizeof(struct journal_v2_block_trailer);
1329
+
1330
+ uint8_t *data_start = netdata_mmap(path, total_file_size, MAP_SHARED, 0, false);
1331
+ uint8_t *data = data_start;
1332
+
1333
+ memset(data_start, 0, extent_offset);
1334
+
1335
+ // Write header
1336
+ struct journal_v2_header j2_header;
1337
+ memset(&j2_header, 0, sizeof(j2_header));
1338
+
1339
+ j2_header.magic = JOURVAL_V2_MAGIC;
1340
+ j2_header.start_time_ut = min_time_ut;
1341
+ j2_header.end_time_ut = max_time_ut;
1342
+ j2_header.extent_count = number_of_extents;
1343
+ j2_header.extent_offset = extent_offset;
1344
+ j2_header.metric_count = number_of_metrics;
1345
+ j2_header.metric_offset = metrics_offset;
1346
+ j2_header.page_count = number_of_pages;
1347
+ j2_header.page_offset = pages_offset;
1348
+ j2_header.extent_trailer_offset = extent_offset_trailer;
1349
+ j2_header.metric_trailer_offset = metric_offset_trailer;
1350
+ j2_header.total_file_size = total_file_size;
1351
+ j2_header.original_file_size = (uint32_t) journalfile->pos;
1352
+ j2_header.data = data_start; // Used during migration
1353
+
1354
+ struct journal_v2_block_trailer *journal_v2_trailer;
1355
+
1356
+ // Write all the extents we have
1357
+ data = journal_v2_write_extent_list(journalfile, data_start + extent_offset);
1358
+ internal_error(true, "Write extent list so far %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1359
+
1360
+ fatal_assert(data == data_start + extent_offset_trailer);
1361
+
1362
+ // Calculate CRC for extents
1363
+ journal_v2_trailer = (struct journal_v2_block_trailer *) (data_start + extent_offset_trailer);
1364
+ uLong crc;
1365
+ crc = crc32(0L, Z_NULL, 0);
1366
+ crc = crc32(crc, (uint8_t *) data_start + extent_offset, number_of_extents * sizeof(struct journal_extent_list));
1367
+ crc32set(journal_v2_trailer->checksum, crc);
1368
+
1369
+ internal_error(true, "CALCULATE CRC FOR EXTENT %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1370
+ // Skip the trailer, point to the metrics off
1371
+ data += sizeof(struct journal_v2_block_trailer);
1372
+
1373
+ // Sanity check -- we must be at the metrics_offset
1374
+ fatal_assert(data == data_start + metrics_offset);
1375
+
1376
+ // Allocate array to sort UUIDs and keep them sorted in the journal because we want to do binary search when we do lookups
1377
+ struct journal_metric_list_to_sort *uuid_list = mallocz(number_of_metrics * sizeof(struct journal_metric_list_to_sort));
1378
+
1379
+ Word_t Index;
1380
+ struct page_cache *pg_cache = &ctx->pg_cache;
1381
+ struct pg_cache_page_index *page_index;
1382
+ for (Index = 0, PValue = JudyLFirst(metrics_JudyL_array, &Index, PJE0),
1383
+ metric_info = unlikely(NULL == PValue) ? NULL : *PValue;
1384
+ metric_info != NULL;
1385
+ PValue = JudyLNext(metrics_JudyL_array, &Index, PJE0),
1386
+ metric_info = unlikely(NULL == PValue) ? NULL : *PValue) {
1387
+
1388
+ fatal_assert(Index < number_of_metrics);
1389
+ uuid_list[Index].metric_info = metric_info;
1390
+
1391
+ if (false == startup)
1392
+ uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
1393
+
1394
+ PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, metric_info->id, sizeof(uuid_t));
1395
+ page_index = (NULL == PValue) ? NULL : *PValue;
1396
+
1397
+ if (false == startup)
1398
+ uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
1399
+
1400
+ fatal_assert(NULL != page_index);
1401
+
1402
+ metric_info->page_index = page_index;
1403
+
1404
+ }
1405
+ // Cleanup judy arrays we no longer need
1406
+ JudyLFreeArray(&metrics_JudyL_array, PJE0);
1407
+ JudyHSFreeArray(&metrics_JudyHS_array, PJE0);
1408
+
1409
+ qsort(&uuid_list[0], number_of_metrics, sizeof(struct journal_metric_list_to_sort), journal_metric_compare);
1410
+ internal_error(true, "Traverse and qsort UUID %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1411
+ // Write sorted UUID LIST
1412
+ // The loop will write a single UUID entry then
1413
+ // Write all entries (descr @ time) for that UUID at the proper location (header, number of entries, trailer)
1414
+ // Move on to write the next UUID
1415
+ // Write trailer after the UUID list
1416
+ for (Index = 0; Index < number_of_metrics; Index++) {
1417
+ metric_info = uuid_list[Index].metric_info;
1418
+
1419
+ // Calculate current UUID offset from start of file. We will store this in the data page header
1420
+ uint32_t uuid_offset = data - data_start;
1421
+
1422
+ // Write the UUID we are processing
1423
+ data = (void *) journal_v2_write_metric_page(&j2_header, data, metric_info, pages_offset);
1424
+ if (unlikely(!data))
1425
+ break;
1426
+
1427
+ // Next we will write
1428
+ // Header
1429
+ // Detailed entries (descr @ time)
1430
+ // Trailer (checksum)
1431
+
1432
+ // Keep the page_list_header, to be used for migration when where agent is running
1433
+ metric_info->page_list_header = pages_offset;
1434
+ // Write page header
1435
+ void *metric_page = journal_v2_write_data_page_header(&j2_header, data_start + pages_offset, metric_info, uuid_offset);
1436
+
1437
+ // Start writing descr @ time
1438
+ void *page_trailer = journal_v2_write_descriptors(&j2_header, metric_page, metric_info, journalfile);
1439
+ if (unlikely(!page_trailer))
1440
+ break;
1441
+
1442
+ // Trailer (checksum)
1443
+ uint8_t *next_page_address = journal_v2_write_data_page_trailer(&j2_header, page_trailer, data_start + pages_offset);
1444
+
1445
+ // Calculate start of the pages start for next descriptor
1446
+ pages_offset += (metric_info->entries * (sizeof(struct journal_page_list)) + sizeof(struct journal_page_header) + sizeof(struct journal_v2_block_trailer));
1447
+ // Verify we are at the right location
1448
+ fatal_assert(pages_offset == (next_page_address - data_start));
1449
+ }
1450
+ // Data should be at the UUID trailer offset
1451
+ fatal_assert(data == data_start + metric_offset_trailer);
1452
+
1453
+ internal_error(true, "WRITE METRICS AND PAGES %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1454
+
1455
+ // Calculate CRC for metrics
1456
+ journal_v2_trailer = (struct journal_v2_block_trailer *) (data_start + metric_offset_trailer);
1457
+ crc = crc32(0L, Z_NULL, 0);
1458
+ crc = crc32(crc, (uint8_t *) data_start + metrics_offset, number_of_metrics * sizeof(struct journal_metric_list));
1459
+ crc32set(journal_v2_trailer->checksum, crc);
1460
+ internal_error(true, "CALCULATE CRC FOR UUIDs %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1461
+
1462
+ // Prepare to write checksum for the file
1463
+ j2_header.data = NULL;
1464
+ journal_v2_trailer = (struct journal_v2_block_trailer *) (data_start + trailer_offset);
1465
+ crc = crc32(0L, Z_NULL, 0);
1466
+ crc = crc32(crc, (void *) &j2_header, sizeof(j2_header));
1467
+ crc32set(journal_v2_trailer->checksum, crc);
1468
+
1469
+ // Write header to the file
1470
+ memcpy(data_start, &j2_header, sizeof(j2_header));
1471
+
1472
+ internal_error(true, "FILE COMPLETED --------> %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1473
+
1474
+ info("Migrated journal file %s, File size %lu", path, total_file_size);
1475
+
1476
+ if (activate) {
1477
+ journalfile->journal_data = data_start;
1478
+ journalfile->journal_data_size = total_file_size;
1479
+
1480
+ // HERE we need to remove old descriptors and activate the new ones
1481
+ {
1482
+ for (Index = 0; Index < number_of_metrics; Index++) {
1483
+ journal_v2_remove_active_descriptors(journalfile, uuid_list[Index].metric_info, startup);
1484
+
1485
+ if (false == startup)
1486
+ JudyLFreeArray(&uuid_list[Index].metric_info->JudyL_array, PJE0);
1487
+
1488
+ freez(uuid_list[Index].metric_info);
1489
+ }
1490
+ internal_error(true, "ACTIVATING NEW INDEX JNL %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1491
+
1492
+ if (false == startup) {
1493
+ uv_rwlock_wrlock(&journalfile->datafile->extent_rwlock);
1494
+ df_extent_delete_all_unsafe(journalfile->datafile);
1495
+ uv_rwlock_wrunlock(&journalfile->datafile->extent_rwlock);
1496
+ }
1497
+ else
1498
+ df_extent_delete_all_unsafe(journalfile->datafile);
1499
+ }
1500
+
1501
+ ctx->disk_space += total_file_size;
1502
+ }
1503
+ else {
1504
+ // If we failed and didnt process the entire list, free the rest
1505
+ for (Index = 0; Index < number_of_metrics; Index++)
1506
+ freez(uuid_list[Index].metric_info);
1507
+ netdata_munmap(data_start, total_file_size);
1508
+ }
1509
+ freez(uuid_list);
1510
+}
1511
+
1512
int load_journal_file(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
1513
struct rrdengine_datafile *datafile)
1514
{
1517
int ret, fd, error;
1518
uint64_t file_size, max_id;
1519
char path[RRDENG_PATH_MAX];
1520
+ int should_try_migration = 0;
1521
+
1522
+ // Do not try to load the latest file (always rebuild and live migrate)
1523
+ if (datafile->fileno != ctx->last_fileno && db_engine_journal_indexing) {
1524
+ if (!(should_try_migration = load_journal_file_v2(ctx, journalfile, datafile))) {
1525
+ return 0;
1526
+ }
1527
+ }
1528
1529
generate_journalfilepath(datafile, path, sizeof(path));
537
- fd = open_file_direct_io(path, O_RDWR, &file);
1530
+
1531
+ // If it is not the last file, open read only
1532
+ fd = open_file_direct_io(path, datafile->fileno == ctx->last_fileno ? O_RDWR : O_RDONLY, &file);
1533
if (fd < 0) {
1534
++ctx->stats.fs_errors;
1535
rrd_stat_atomic_add(&global_fs_errors, 1);
1536
return fd;
1537
}
543
- info("Loading journal file \"%s\".", path);
1538
1539
ret = check_file_properties(file, &file_size, sizeof(struct rrdeng_df_sb));
1540
if (ret)
1542
file_size = ALIGN_BYTES_FLOOR(file_size);
1543
1544
ret = check_journal_file_superblock(file);
551
- if (ret)
1545
+ if (ret) {
1546
+ info("Invalid journal file \"%s\" ; superblock check failed.", path);
1547
goto error;
1548
+ }
1549
ctx->stats.io_read_bytes += sizeof(struct rrdeng_jf_sb);
1550
++ctx->stats.io_read_requests;
1551
1552
journalfile->file = file;
1553
journalfile->pos = file_size;
558
- journalfile->data = netdata_mmap(path, file_size, MAP_SHARED, 0);
1554
+
1555
+ journalfile->data = netdata_mmap(path, file_size, MAP_SHARED, 0, !(datafile->fileno == ctx->last_fileno));
1556
info("Loading journal file \"%s\" using %s.", path, journalfile->data?"MMAP":"uv_fs_read");
1557
1558
max_id = iterate_transactions(ctx, journalfile);
1562
info("Journal file \"%s\" loaded (size:%"PRIu64").", path, file_size);
1563
if (likely(journalfile->data))
1564
netdata_munmap(journalfile->data, file_size);
1565
+
1566
+ // Don't Index the last file
1567
+ if (ctx->last_fileno == journalfile->datafile->fileno || !db_engine_journal_indexing)
1568
+ return 0;
1569
+
1570
+ if (should_try_migration == 1)
1571
+ migrate_journal_file_v2(datafile, true, true);
1572
+ else
1573
+ error_report("File %s cannot be migrated to the new journal format. Index will be allocated in memory", path);
1574
+
1575
return 0;
1576
1577
error:
1586
return error;
1587
}
1588
1589
+void after_journal_indexing(uv_work_t *req, int status)
1590
+{
1591
+ struct rrdeng_work *work_request = req->data;
1592
+ struct rrdengine_worker_config *wc = work_request->wc;
1593
+
1594
+ if (likely(status != UV_ECANCELED)) {
1595
+ errno = 0;
1596
+ if (likely(work_request->count))
1597
+ internal_error(true, "Journal indexing done; %u files processed", work_request->count);
1598
+ }
1599
+ wc->running_journal_migration = 0;
1600
+ wc->run_indexing= work_request->rerun;
1601
+ freez(work_request);
1602
+}
1603
+
1604
+#define MAX_RETRIES_TO_START_INDEX (100)
1605
+void start_journal_indexing(uv_work_t *req)
1606
+{
1607
+ struct rrdeng_work *work_request = req->data;
1608
+ struct rrdengine_worker_config *wc = work_request->wc;
1609
+ struct rrdengine_instance *ctx = wc->ctx;
1610
+
1611
+ unsigned count = 0;
1612
+ while ((wc->now_deleting_files || wc->now_deleting_descriptors) && count++ < MAX_RETRIES_TO_START_INDEX)
1613
+ sleep_usec(100 * USEC_PER_MS);
1614
+
1615
+ if (count == MAX_RETRIES_TO_START_INDEX)
1616
+ return;
1617
+
1618
+ struct rrdengine_datafile *datafile = ctx->datafiles.first;
1619
+
1620
+ while (datafile && datafile->fileno != ctx->last_fileno) {
1621
+ if (unlikely(!datafile->journalfile->journal_data)) {
1622
+ bool ready_to_index = journalfile_ready_to_index(datafile);
1623
+ if (ready_to_index) {
1624
+ info("Journal file %u is ready to be indexed", datafile->fileno);
1625
+ migrate_journal_file_v2(datafile, true, false);
1626
+ ++work_request->count;
1627
+ }
1628
+ else {
1629
+ info("Journal file %u is not ready to be indexed", datafile->fileno);
1630
+ work_request->rerun = true;
1631
+ sleep_usec(100 * USEC_PER_MS);
1632
+ }
1633
+ }
1634
+ datafile = datafile->next;
1635
+ if (unlikely(NO_QUIESCE != ctx->quiesce))
1636
+ break;
1637
+ }
1638
+}
1639
+
1640
void init_commit_log(struct rrdengine_instance *ctx)
1641
{
1642
ctx->commit_log.buf = NULL;