// SPDX-License-Identifier: GPL-3.0-or-later #include #include "metadatalog.h" #include "metalogpluginsd.h" void generate_metadata_logfile_path(struct metadata_logfile *metalogfile, char *str, size_t maxlen) { (void) snprintfz(str, maxlen, "%s/" METALOG_PREFIX METALOG_FILE_NUMBER_PRINT_TMPL METALOG_EXTENSION, metalogfile->ctx->rrdeng_ctx->dbfiles_path, metalogfile->starting_fileno, metalogfile->fileno); } void metadata_logfile_init(struct metadata_logfile *metalogfile, struct metalog_instance *ctx, unsigned starting_fileno, unsigned fileno) { metalogfile->starting_fileno = starting_fileno; metalogfile->fileno = fileno; metalogfile->file = (uv_file)0; metalogfile->pos = 0; metalogfile->next = NULL; metalogfile->ctx = ctx; } int rename_metadata_logfile(struct metadata_logfile *metalogfile, unsigned new_starting_fileno, unsigned new_fileno) { //struct metalog_instance *ctx = metalogfile->ctx; uv_fs_t req; int ret; char oldpath[RRDENG_PATH_MAX], newpath[RRDENG_PATH_MAX]; unsigned backup_starting_fileno, backup_fileno; backup_starting_fileno = metalogfile->starting_fileno; backup_fileno = metalogfile->fileno; generate_metadata_logfile_path(metalogfile, oldpath, sizeof(oldpath)); metalogfile->starting_fileno = new_starting_fileno; metalogfile->fileno = new_fileno; generate_metadata_logfile_path(metalogfile, newpath, sizeof(newpath)); info("Renaming metadata log file \"%s\" to \"%s\".", oldpath, newpath); ret = uv_fs_rename(NULL, &req, oldpath, newpath, NULL); if (ret < 0) { error("uv_fs_rename(%s): %s", oldpath, uv_strerror(ret)); //++ctx->stats.fs_errors; /* this is racy, may miss some errors */ rrd_stat_atomic_add(&global_fs_errors, 1); /* restore previous values */ metalogfile->starting_fileno = backup_starting_fileno; metalogfile->fileno = backup_fileno; } uv_fs_req_cleanup(&req); return ret; } int unlink_metadata_logfile(struct metadata_logfile *metalogfile) { //struct metalog_instance *ctx = metalogfile->ctx; uv_fs_t req; int ret; char path[RRDENG_PATH_MAX]; generate_metadata_logfile_path(metalogfile, path, sizeof(path)); ret = uv_fs_unlink(NULL, &req, path, NULL); if (ret < 0) { error("uv_fs_fsunlink(%s): %s", path, uv_strerror(ret)); // ++ctx->stats.fs_errors; rrd_stat_atomic_add(&global_fs_errors, 1); } uv_fs_req_cleanup(&req); return ret; } static int check_metadata_logfile_superblock(uv_file file) { int ret; struct rrdeng_metalog_sb *superblock; uv_buf_t iov; uv_fs_t req; ret = posix_memalign((void *)&superblock, RRDFILE_ALIGNMENT, sizeof(*superblock)); if (unlikely(ret)) { fatal("posix_memalign:%s", strerror(ret)); } iov = uv_buf_init((void *)superblock, sizeof(*superblock)); ret = uv_fs_read(NULL, &req, file, &iov, 1, 0, NULL); if (ret < 0) { error("uv_fs_read: %s", uv_strerror(ret)); uv_fs_req_cleanup(&req); goto error; } fatal_assert(req.result >= 0); uv_fs_req_cleanup(&req); if (strncmp(superblock->magic_number, RRDENG_METALOG_MAGIC, RRDENG_MAGIC_SZ)) { error("File has invalid superblock."); ret = UV_EINVAL; } else { ret = 0; } if (superblock->version > RRDENG_METALOG_VER) { error("File has unknown version %"PRIu16". Compatibility is not guaranteed.", superblock->version); } error: free(superblock); return ret; } void replay_record(struct metadata_logfile *metalogfile, struct rrdeng_metalog_record_header *header, void *payload) { struct metalog_instance *ctx = metalogfile->ctx; char *line, *nextline, *record_end; int ret; debug(D_METADATALOG, "RECORD contents: %.*s", (int)header->payload_length, (char *)payload); record_end = (char *)payload + header->payload_length - 1; *record_end = '\0'; for (line = payload ; line ; line = nextline) { nextline = strchr(line, '\n'); if (nextline) { *nextline++ = '\0'; } ret = parser_action(ctx->metalog_parser_object->parser, line); debug(D_METADATALOG, "parser_action ret:%d", ret); if (ret) return; /* skip record due to error */ }; } /* This function only works with buffered I/O */ static inline int metalogfile_read(struct metadata_logfile *metalogfile, void *buf, size_t len, uint64_t offset) { // struct metalog_instance *ctx; uv_file file; uv_buf_t iov; uv_fs_t req; int ret; // ctx = metalogfile->ctx; file = metalogfile->file; iov = uv_buf_init(buf, len); ret = uv_fs_read(NULL, &req, file, &iov, 1, offset, NULL); if (unlikely(ret < 0 && ret != req.result)) { fatal("uv_fs_read: %s", uv_strerror(ret)); } if (req.result < 0) { // ++ctx->stats.io_errors; rrd_stat_atomic_add(&global_io_errors, 1); error("%s: uv_fs_read - %s - record at offset %"PRIu64"(%u) in metadata logfile %u-%u.", __func__, uv_strerror((int)req.result), offset, (unsigned)len, metalogfile->starting_fileno, metalogfile->fileno); } uv_fs_req_cleanup(&req); // ctx->stats.io_read_bytes += len; // ++ctx->stats.io_read_requests; return ret; } /* Return 0 on success */ static int metadata_record_integrity_check(void *record) { int ret; uint32_t data_size; struct rrdeng_metalog_record_header *header; struct rrdeng_metalog_record_trailer *trailer; uLong crc; header = record; data_size = header->header_length + header->payload_length; trailer = record + data_size; crc = crc32(0L, Z_NULL, 0); crc = crc32(crc, record, data_size); ret = crc32cmp(trailer->checksum, crc); return ret; } #define MAX_READ_BYTES (RRDENG_BLOCK_SIZE * 32) /* no record should be over 128KiB in this version */ /* * Iterates metadata log file records and creates database objects (host/chart/dimension) */ static void iterate_records(struct metadata_logfile *metalogfile) { uint32_t file_size, pos, bytes_remaining, record_size; void *buf; struct rrdeng_metalog_record_header *header; struct metalog_instance *ctx = metalogfile->ctx; struct metalog_pluginsd_state *state = ctx->metalog_parser_object->private; const size_t min_header_size = offsetof(struct rrdeng_metalog_record_header, header_length) + sizeof(header->header_length); file_size = metalogfile->pos; state->metalogfile = metalogfile; buf = mallocz(MAX_READ_BYTES); for (pos = sizeof(struct rrdeng_metalog_sb) ; pos < file_size ; pos += record_size) { bytes_remaining = file_size - pos; if (bytes_remaining < min_header_size) { error("%s: unexpected end of file in metadata logfile %u-%u.", __func__, metalogfile->starting_fileno, metalogfile->fileno); break; } if (metalogfile_read(metalogfile, buf, min_header_size, pos) < 0) break; header = (struct rrdeng_metalog_record_header *)buf; if (METALOG_STORE_PADDING == header->type) { info("%s: Skipping padding in metadata logfile %u-%u.", __func__, metalogfile->starting_fileno, metalogfile->fileno); record_size = ALIGN_BYTES_FLOOR(pos + RRDENG_BLOCK_SIZE) - pos; continue; } if (metalogfile_read(metalogfile, buf + min_header_size, sizeof(*header) - min_header_size, pos + min_header_size) < 0) break; record_size = header->header_length + header->payload_length + sizeof(struct rrdeng_metalog_record_trailer); if (header->header_length < min_header_size || record_size > bytes_remaining) { error("%s: Corrupted record in metadata logfile %u-%u.", __func__, metalogfile->starting_fileno, metalogfile->fileno); break; } if (record_size > MAX_READ_BYTES) { error("%s: Record is too long (%u bytes) in metadata logfile %u-%u.", __func__, record_size, metalogfile->starting_fileno, metalogfile->fileno); continue; } if (metalogfile_read(metalogfile, buf + sizeof(*header), record_size - sizeof(*header), pos + sizeof(*header)) < 0) break; if (metadata_record_integrity_check(buf)) { error("%s: Record at offset %"PRIu32" was read from disk. CRC32 check: FAILED", __func__, pos); continue; } debug(D_METADATALOG, "%s: Record at offset %"PRIu32" was read from disk. CRC32 check: SUCCEEDED", __func__, pos); replay_record(metalogfile, header, buf + header->header_length); } freez(buf); } int load_metadata_logfile(struct metalog_instance *ctx, struct metadata_logfile *metalogfile) { UNUSED(ctx); uv_fs_t req; uv_file file; int ret, fd, error; uint64_t file_size; char path[RRDENG_PATH_MAX]; generate_metadata_logfile_path(metalogfile, path, sizeof(path)); if (file_is_migrated(path)) return 0; fd = open_file_buffered_io(path, O_RDWR, &file); if (fd < 0) { // ++ctx->stats.fs_errors; rrd_stat_atomic_add(&global_fs_errors, 1); return fd; } info("Loading metadata log \"%s\".", path); ret = check_file_properties(file, &file_size, sizeof(struct rrdeng_metalog_sb)); if (ret) goto error; ret = check_metadata_logfile_superblock(file); if (ret) goto error; // ctx->stats.io_read_bytes += sizeof(struct rrdeng_jf_sb); // ++ctx->stats.io_read_requests; metalogfile->file = file; metalogfile->pos = file_size; iterate_records(metalogfile); info("Metadata log \"%s\" migrated to the database (size:%"PRIu64").", path, file_size); add_migrated_file(path, file_size); return 0; error: error = ret; ret = uv_fs_close(NULL, &req, file, NULL); if (ret < 0) { error("uv_fs_close(%s): %s", path, uv_strerror(ret)); // ++ctx->stats.fs_errors; rrd_stat_atomic_add(&global_fs_errors, 1); } uv_fs_req_cleanup(&req); return error; } static int scan_metalog_files_cmp(const void *a, const void *b) { struct metadata_logfile *file1, *file2; char path1[RRDENG_PATH_MAX], path2[RRDENG_PATH_MAX]; file1 = *(struct metadata_logfile **)a; file2 = *(struct metadata_logfile **)b; generate_metadata_logfile_path(file1, path1, sizeof(path1)); generate_metadata_logfile_path(file2, path2, sizeof(path2)); return strcmp(path1, path2); } /* Returns number of metadata logfiles that were loaded or < 0 on error */ static int scan_metalog_files(struct metalog_instance *ctx) { int ret; unsigned starting_no, no, matched_files, i, failed_to_load; static uv_fs_t req; uv_dirent_t dent; struct metadata_logfile **metalogfiles, *metalogfile; char *dbfiles_path = ctx->rrdeng_ctx->dbfiles_path; ret = uv_fs_scandir(NULL, &req, dbfiles_path, 0, NULL); if (ret < 0) { fatal_assert(req.result < 0); uv_fs_req_cleanup(&req); error("uv_fs_scandir(%s): %s", dbfiles_path, uv_strerror(ret)); // ++ctx->stats.fs_errors; rrd_stat_atomic_add(&global_fs_errors, 1); return ret; } info("Found %d files in path %s", ret, dbfiles_path); metalogfiles = callocz(MIN(ret, MAX_DATAFILES), sizeof(*metalogfiles)); for (matched_files = 0 ; UV_EOF != uv_fs_scandir_next(&req, &dent) && matched_files < MAX_DATAFILES ; ) { info("Scanning file \"%s/%s\"", dbfiles_path, dent.name); ret = sscanf(dent.name, METALOG_PREFIX METALOG_FILE_NUMBER_SCAN_TMPL METALOG_EXTENSION, &starting_no, &no); if (2 == ret) { info("Matched file \"%s/%s\"", dbfiles_path, dent.name); metalogfile = mallocz(sizeof(*metalogfile)); metadata_logfile_init(metalogfile, ctx, starting_no, no); metalogfiles[matched_files++] = metalogfile; } } uv_fs_req_cleanup(&req); if (0 == matched_files) { freez(metalogfiles); return 0; } if (matched_files == MAX_DATAFILES) { error("Warning: hit maximum database engine file limit of %d files", MAX_DATAFILES); } qsort(metalogfiles, matched_files, sizeof(*metalogfiles), scan_metalog_files_cmp); ret = compaction_failure_recovery(ctx, metalogfiles, &matched_files); if (ret) { /* If the files are corrupted fail */ for (i = 0 ; i < matched_files ; ++i) { freez(metalogfiles[i]); } freez(metalogfiles); return UV_EINVAL; } //ctx->last_fileno = metalogfiles[matched_files - 1]->fileno; struct plugind cd = { .enabled = 1, .update_every = 0, .pid = 0, .serial_failures = 0, .successful_collections = 0, .obsolete = 0, .started_t = INVALID_TIME, .next = NULL, .version = 0, }; struct metalog_pluginsd_state metalog_parser_state; metalog_pluginsd_state_init(&metalog_parser_state, ctx); PARSER_USER_OBJECT metalog_parser_object = { .enabled = cd.enabled, .host = ctx->rrdeng_ctx->host, .cd = &cd, .trust_durations = 0, .private = &metalog_parser_state }; PARSER *parser = parser_init(metalog_parser_object.host, &metalog_parser_object, NULL, PARSER_INPUT_SPLIT); parser_add_keyword(parser, PLUGINSD_KEYWORD_HOST, metalog_pluginsd_host); parser_add_keyword(parser, PLUGINSD_KEYWORD_GUID, pluginsd_guid); parser_add_keyword(parser, PLUGINSD_KEYWORD_CONTEXT, pluginsd_context); parser_add_keyword(parser, PLUGINSD_KEYWORD_TOMBSTONE, pluginsd_tombstone); parser->plugins_action->dimension_action = &metalog_pluginsd_dimension_action; parser->plugins_action->chart_action = &metalog_pluginsd_chart_action; parser->plugins_action->guid_action = &metalog_pluginsd_guid_action; parser->plugins_action->context_action = &metalog_pluginsd_context_action; parser->plugins_action->tombstone_action = &metalog_pluginsd_tombstone_action; parser->plugins_action->host_action = &metalog_pluginsd_host_action; metalog_parser_object.parser = parser; ctx->metalog_parser_object = &metalog_parser_object; for (failed_to_load = 0, i = 0 ; i < matched_files ; ++i) { metalogfile = metalogfiles[i]; db_lock(); db_execute("BEGIN TRANSACTION;"); ret = load_metadata_logfile(ctx, metalogfile); if (0 != ret) { error("Deleting invalid metadata log file \"%s/"METALOG_PREFIX METALOG_FILE_NUMBER_PRINT_TMPL METALOG_EXTENSION"\"", dbfiles_path, metalogfile->starting_fileno, metalogfile->fileno); unlink_metadata_logfile(metalogfile); ++failed_to_load; db_execute("ROLLBACK TRANSACTION;"); } else db_execute("COMMIT TRANSACTION;"); db_unlock(); freez(metalogfile); } matched_files -= failed_to_load; debug(D_METADATALOG, "PARSER ended"); parser_destroy(parser); size_t count __maybe_unused = metalog_parser_object.count; debug(D_METADATALOG, "Parsing count=%u", (unsigned)count); freez(metalogfiles); return matched_files; } /* Return 0 on success. */ int init_metalog_files(struct metalog_instance *ctx) { int ret; char *dbfiles_path = ctx->rrdeng_ctx->dbfiles_path; ret = scan_metalog_files(ctx); if (ret < 0) { error("Failed to scan path \"%s\".", dbfiles_path); return ret; }/* else if (0 == ret) { ctx->last_fileno = 1; }*/ return 0; }