lordgamez commented on code in PR #2209: URL: https://github.com/apache/nifi-minifi-cpp/pull/2209#discussion_r3782432271
########## extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp: ########## @@ -0,0 +1,207 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "RocksDbProvenanceRepository.h" + +#include <string> + +#include "core/Resource.h" + +namespace org::apache::nifi::minifi::provenance { + +namespace { +class EventCursor : public ProvenanceRepository::Cursor { +public: + explicit EventCursor(std::string event_id): event_id_(std::move(event_id)) {} + [[nodiscard]] + std::string toString() const override { + return event_id_; + } + ~EventCursor() override = default; + + std::string event_id_; +}; +} // namespace + +static const std::string_view NEXT_EVENT_UUID_KEY = "next_event_uuid"; + +bool RocksDbProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) { + if (!RocksDbRepository::initialize(config)) { + return false; + } + std::string value; + if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_); + if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) { + max_partition_bytes_ = gsl::narrow<int64_t>(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size")); + } + logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_); + if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) { + if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value)) + max_partition_millis_ = *max_partition; + } + logger_->log_debug("MiNiFi Provenance Max Storage Time: [{}]", max_partition_millis_); + + verify_checksums_in_rocksdb_reads_ = (config->get(Configure::nifi_provenance_repository_rocksdb_read_verify_checksums) | utils::andThen(&utils::string::toBool)).value_or(false); + logger_->log_debug("{} checksum verification in RocksDbProvenanceRepository", verify_checksums_in_rocksdb_reads_ ? "Using" : "Not using"); + + auto db_options = [] (minifi::internal::Writable<rocksdb::DBOptions>& db_opts) { + minifi::internal::setCommonRocksDbOptions(db_opts); + }; + + // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after + // This shouldn't go above 16MB and the configured total size of the db should cap it as well + auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) { + int64_t max_buffer_size = 16 << 20; + cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_)); Review Comment: Can we use 16_MB from Literals.h here and omit the max_buffer_size variable? ########## extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp: ########## @@ -0,0 +1,207 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "RocksDbProvenanceRepository.h" + +#include <string> + +#include "core/Resource.h" + +namespace org::apache::nifi::minifi::provenance { + +namespace { +class EventCursor : public ProvenanceRepository::Cursor { +public: + explicit EventCursor(std::string event_id): event_id_(std::move(event_id)) {} + [[nodiscard]] + std::string toString() const override { + return event_id_; + } + ~EventCursor() override = default; + + std::string event_id_; +}; +} // namespace + +static const std::string_view NEXT_EVENT_UUID_KEY = "next_event_uuid"; + +bool RocksDbProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) { + if (!RocksDbRepository::initialize(config)) { + return false; + } + std::string value; + if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_); + if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) { + max_partition_bytes_ = gsl::narrow<int64_t>(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size")); + } + logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_); + if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) { + if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value)) + max_partition_millis_ = *max_partition; Review Comment: nitpick: we should put braces around this line to be consistent with other one line conditional statements ########## extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp: ########## @@ -0,0 +1,207 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "RocksDbProvenanceRepository.h" + +#include <string> + +#include "core/Resource.h" + +namespace org::apache::nifi::minifi::provenance { + +namespace { +class EventCursor : public ProvenanceRepository::Cursor { +public: + explicit EventCursor(std::string event_id): event_id_(std::move(event_id)) {} + [[nodiscard]] + std::string toString() const override { + return event_id_; + } + ~EventCursor() override = default; + + std::string event_id_; +}; +} // namespace + +static const std::string_view NEXT_EVENT_UUID_KEY = "next_event_uuid"; + +bool RocksDbProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) { + if (!RocksDbRepository::initialize(config)) { + return false; + } + std::string value; + if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_); + if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) { + max_partition_bytes_ = gsl::narrow<int64_t>(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size")); + } + logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_); + if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) { + if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value)) + max_partition_millis_ = *max_partition; + } + logger_->log_debug("MiNiFi Provenance Max Storage Time: [{}]", max_partition_millis_); + + verify_checksums_in_rocksdb_reads_ = (config->get(Configure::nifi_provenance_repository_rocksdb_read_verify_checksums) | utils::andThen(&utils::string::toBool)).value_or(false); + logger_->log_debug("{} checksum verification in RocksDbProvenanceRepository", verify_checksums_in_rocksdb_reads_ ? "Using" : "Not using"); + + auto db_options = [] (minifi::internal::Writable<rocksdb::DBOptions>& db_opts) { + minifi::internal::setCommonRocksDbOptions(db_opts); + }; + + // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after + // This shouldn't go above 16MB and the configured total size of the db should cap it as well + auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) { + int64_t max_buffer_size = 16 << 20; + cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_)); + cf_opts.max_write_buffer_number = 4; + cf_opts.min_write_buffer_number_to_merge = 1; + + cf_opts.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO; + cf_opts.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false); + if (max_partition_millis_ > std::chrono::milliseconds(0)) { + cf_opts.ttl = std::chrono::duration_cast<std::chrono::seconds>(max_partition_millis_).count(); + } + }; + + db_ = minifi::internal::RocksDatabase::create(db_options, cf_options, directory_, + minifi::internal::getRocksDbOptionsToOverride(config, Configure::nifi_provenance_repository_rocksdb_options)); + std::string internal_state_db_uri = [&] { + const std::string_view minifidb_scheme = "minifidb://"; + if (directory_.starts_with(minifidb_scheme)) { + return directory_ + "-internal-state"; + } + std::string uri = utils::string::join_pack(minifidb_scheme, directory_); + if (uri.ends_with("/") || uri.ends_with("\\")) { + uri.pop_back(); + } + return uri + "/internal-state"; + }(); + internal_state_db_ = minifi::internal::RocksDatabase::create(db_options, {}, internal_state_db_uri, {}); + if (auto open_state_db = internal_state_db_->open()) { + rocksdb::ReadOptions options; + options.verify_checksums = verify_checksums_in_rocksdb_reads_; + std::string next_event_uuid_str; + if (open_state_db->Get(options, NEXT_EVENT_UUID_KEY, &next_event_uuid_str).ok()) { + next_event_id_ = next_event_uuid_str; + } else { + logger_->log_error("Could not find '{}'", NEXT_EVENT_UUID_KEY); + next_event_id_ = utils::IdGenerator::getIdGenerator()->generate(); + } + logger_->log_trace("Using next event uuid: {}", next_event_id_.to_string()); + } else { + logger_->log_error("Could not open internal state column in provenance repository {}", internal_state_db_uri); + return false; + } + if (db_->open()) { + logger_->log_debug("MiNiFi Provenance Repository database open {} success", directory_); + } else { + logger_->log_error("MiNiFi Provenance Repository database open {} failed", directory_); + return false; + } + + return true; +} + +void RocksDbProvenanceRepository::destroy() { + db_.reset(); +} + +std::unique_ptr<ProvenanceRepository::Cursor> RocksDbProvenanceRepository::cursorFromString(std::optional<std::string> cursor_str) { + if (cursor_str.has_value()) { + return std::make_unique<EventCursor>(cursor_str.value()); + } + return std::make_unique<EventCursor>(""); +} + +std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> RocksDbProvenanceRepository::getEvents(size_t max_size, Cursor* cursor) { + auto* event_cursor = dynamic_cast<EventCursor*>(cursor); + if (cursor && !event_cursor) { + return std::unexpected{"Invalid cursor"}; + } + if (max_size == 0) { + return {}; + } + auto opendb = db_->open(); + if (!opendb) { + return std::unexpected{"Failed to open database"}; + } + std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> records; + rocksdb::ReadOptions options; + options.verify_checksums = verify_checksums_in_rocksdb_reads_; + std::unique_ptr<rocksdb::Iterator> it(opendb->NewIterator(options)); + std::string last_event_id; + if (event_cursor) { + last_event_id = event_cursor->event_id_; + it->Seek(event_cursor->event_id_); + if (it->Valid() && it->key() == event_cursor->event_id_) { + it->Next(); + } + } else { + it->SeekToFirst(); + } + for (; it->Valid(); it->Next()) { + last_event_id = it->key().ToString(); + auto eventRead = ProvenanceEventRecord::create(); + const auto slice = it->value(); + io::BufferStream stream(std::as_bytes(std::span(slice.data(), slice.size()))); + if (eventRead->deserialize(stream)) { + records.push_back(eventRead); + if (--max_size == 0) { + break; + } + } + } + if (event_cursor) { + event_cursor->event_id_ = last_event_id; + } + return records; +} + +std::expected<void, std::string> RocksDbProvenanceRepository::appendEvents(const std::vector<std::shared_ptr<ProvenanceEventRecord>>& events) { + std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>> data; Review Comment: This type is kind of long and repeated quite a lot in the repository domian. It should be named with an alias because its variable is usually also just called "data" which is not really descriptive. ########## libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp: ########## @@ -156,11 +160,36 @@ void SiteToSiteProvenanceReportingTask::onSchedule(core::ProcessContext& context } void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, core::ProcessSession& session) { - std::shared_ptr<core::Repository> repo = context.getProvenanceRepository(); + std::shared_ptr<provenance::ProvenanceRepository> repo = context.getProvenanceRepository(); if (!repo) { throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to retrieve provenance repository"); } - auto records = repo->getElements(batch_size_); + auto* state_manager = context.getStateManager(); + if (!state_manager) { + logger_->log_error("Failed to get StateManager"); + context.yield(); + return; + } + std::optional<std::string> cursor_str; + { + std::unordered_map<std::string, std::string> state_map; + if (state_manager->get(state_map)) { + if (auto it = state_map.find("cursor"); it != state_map.end()) { + cursor_str = it->second; + } + } + } + std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> records; + auto cursor = repo->cursorFromString(cursor_str); + if (cursor_str && !cursor) { + logger_->log_error("Failed to parse cursor, falling back to enumerating from the beginning"); + cursor = repo->cursorFromString(std::nullopt); + } Review Comment: I would extract getting the cursor to a separate function in anonymous namespace. ########## extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp: ########## @@ -0,0 +1,207 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "RocksDbProvenanceRepository.h" + +#include <string> + +#include "core/Resource.h" + +namespace org::apache::nifi::minifi::provenance { + +namespace { +class EventCursor : public ProvenanceRepository::Cursor { +public: + explicit EventCursor(std::string event_id): event_id_(std::move(event_id)) {} + [[nodiscard]] + std::string toString() const override { + return event_id_; + } + ~EventCursor() override = default; + + std::string event_id_; +}; +} // namespace + +static const std::string_view NEXT_EVENT_UUID_KEY = "next_event_uuid"; + +bool RocksDbProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) { + if (!RocksDbRepository::initialize(config)) { + return false; + } + std::string value; + if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_); + if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) { + max_partition_bytes_ = gsl::narrow<int64_t>(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size")); + } + logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_); + if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) { + if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value)) + max_partition_millis_ = *max_partition; + } + logger_->log_debug("MiNiFi Provenance Max Storage Time: [{}]", max_partition_millis_); + + verify_checksums_in_rocksdb_reads_ = (config->get(Configure::nifi_provenance_repository_rocksdb_read_verify_checksums) | utils::andThen(&utils::string::toBool)).value_or(false); + logger_->log_debug("{} checksum verification in RocksDbProvenanceRepository", verify_checksums_in_rocksdb_reads_ ? "Using" : "Not using"); + + auto db_options = [] (minifi::internal::Writable<rocksdb::DBOptions>& db_opts) { + minifi::internal::setCommonRocksDbOptions(db_opts); + }; + + // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after + // This shouldn't go above 16MB and the configured total size of the db should cap it as well + auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) { + int64_t max_buffer_size = 16 << 20; + cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_)); + cf_opts.max_write_buffer_number = 4; + cf_opts.min_write_buffer_number_to_merge = 1; + + cf_opts.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO; + cf_opts.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false); + if (max_partition_millis_ > std::chrono::milliseconds(0)) { + cf_opts.ttl = std::chrono::duration_cast<std::chrono::seconds>(max_partition_millis_).count(); + } + }; + + db_ = minifi::internal::RocksDatabase::create(db_options, cf_options, directory_, + minifi::internal::getRocksDbOptionsToOverride(config, Configure::nifi_provenance_repository_rocksdb_options)); + std::string internal_state_db_uri = [&] { + const std::string_view minifidb_scheme = "minifidb://"; + if (directory_.starts_with(minifidb_scheme)) { + return directory_ + "-internal-state"; + } + std::string uri = utils::string::join_pack(minifidb_scheme, directory_); + if (uri.ends_with("/") || uri.ends_with("\\")) { + uri.pop_back(); + } + return uri + "/internal-state"; + }(); + internal_state_db_ = minifi::internal::RocksDatabase::create(db_options, {}, internal_state_db_uri, {}); + if (auto open_state_db = internal_state_db_->open()) { + rocksdb::ReadOptions options; + options.verify_checksums = verify_checksums_in_rocksdb_reads_; + std::string next_event_uuid_str; + if (open_state_db->Get(options, NEXT_EVENT_UUID_KEY, &next_event_uuid_str).ok()) { + next_event_id_ = next_event_uuid_str; + } else { + logger_->log_error("Could not find '{}'", NEXT_EVENT_UUID_KEY); + next_event_id_ = utils::IdGenerator::getIdGenerator()->generate(); + } + logger_->log_trace("Using next event uuid: {}", next_event_id_.to_string()); + } else { + logger_->log_error("Could not open internal state column in provenance repository {}", internal_state_db_uri); + return false; + } + if (db_->open()) { + logger_->log_debug("MiNiFi Provenance Repository database open {} success", directory_); + } else { + logger_->log_error("MiNiFi Provenance Repository database open {} failed", directory_); + return false; + } + + return true; +} + +void RocksDbProvenanceRepository::destroy() { + db_.reset(); +} + +std::unique_ptr<ProvenanceRepository::Cursor> RocksDbProvenanceRepository::cursorFromString(std::optional<std::string> cursor_str) { + if (cursor_str.has_value()) { + return std::make_unique<EventCursor>(cursor_str.value()); + } + return std::make_unique<EventCursor>(""); +} + +std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> RocksDbProvenanceRepository::getEvents(size_t max_size, Cursor* cursor) { + auto* event_cursor = dynamic_cast<EventCursor*>(cursor); + if (cursor && !event_cursor) { + return std::unexpected{"Invalid cursor"}; + } + if (max_size == 0) { + return {}; + } + auto opendb = db_->open(); + if (!opendb) { + return std::unexpected{"Failed to open database"}; + } + std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> records; + rocksdb::ReadOptions options; + options.verify_checksums = verify_checksums_in_rocksdb_reads_; + std::unique_ptr<rocksdb::Iterator> it(opendb->NewIterator(options)); + std::string last_event_id; + if (event_cursor) { + last_event_id = event_cursor->event_id_; + it->Seek(event_cursor->event_id_); + if (it->Valid() && it->key() == event_cursor->event_id_) { + it->Next(); + } + } else { + it->SeekToFirst(); + } + for (; it->Valid(); it->Next()) { + last_event_id = it->key().ToString(); + auto eventRead = ProvenanceEventRecord::create(); + const auto slice = it->value(); + io::BufferStream stream(std::as_bytes(std::span(slice.data(), slice.size()))); + if (eventRead->deserialize(stream)) { + records.push_back(eventRead); + if (--max_size == 0) { + break; + } + } + } + if (event_cursor) { + event_cursor->event_id_ = last_event_id; + } + return records; +} + +std::expected<void, std::string> RocksDbProvenanceRepository::appendEvents(const std::vector<std::shared_ptr<ProvenanceEventRecord>>& events) { + std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>> data; + data.reserve(events.size()); + std::lock_guard guard(next_event_id_mtx_); + for (auto& event : events) { + event->setUUID(next_event_id_++); + } + { + auto open_state_db = internal_state_db_->open(); + if (!open_state_db) { + return std::unexpected{"Failed to open internal state column in provenance database"}; + } + auto operation = [this, &open_state_db]() { return open_state_db->Put(rocksdb::WriteOptions(), NEXT_EVENT_UUID_KEY, next_event_id_.to_string().view()); }; + if (!ExecuteWithRetry(operation)) { + return std::unexpected{"Failed to update next provenance event id"}; + } + } + for (auto& event : events) { + data.emplace_back(event->getUUIDStr(), std::make_unique<io::BufferStream>()); + event->serialize(*data.back().second); Review Comment: Should we also handle if the event serialization fails? ########## extensions/rocksdb-repos/tests/ProvenanceTests.cpp: ########## @@ -100,11 +100,14 @@ TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::Pro auto sample = 65555ms; - std::shared_ptr<core::Repository> testRepository = std::make_shared<core::repository::VolatileProvenanceRepository>(); + auto testRepository = std::make_shared<core::repository::VolatileProvenanceRepository>(); testRepository->initialize(nullptr); record1->setEventDuration(sample); - testRepository->storeElement(record1); + testRepository->appendEvents({record1}); + + utils::Identifier eventId = record1->getEventId(); Review Comment: This variable is redeclared and does not compile ########## extensions/rocksdb-repos/RocksDbProvenanceRepository.h: ########## @@ -39,22 +40,22 @@ constexpr auto MAX_PROVENANCE_STORAGE_SIZE = 10_MiB; constexpr auto MAX_PROVENANCE_ENTRY_LIFE_TIME = std::chrono::minutes(1); constexpr auto PROVENANCE_PURGE_PERIOD = std::chrono::milliseconds(2500); -class ProvenanceRepository : public core::repository::RocksDbRepository { +class RocksDbProvenanceRepository : public core::repository::RocksDbRepository, public ProvenanceRepository { public: - ProvenanceRepository(std::string_view name, const utils::Identifier& /*uuid*/) - : ProvenanceRepository(name) { + RocksDbProvenanceRepository(std::string_view name, const utils::Identifier& /*uuid*/) + : RocksDbProvenanceRepository(name) { } - explicit ProvenanceRepository(std::string_view repo_name = "", + explicit RocksDbProvenanceRepository(std::string_view repo_name = "", std::string directory = PROVENANCE_DIRECTORY, std::chrono::milliseconds maxPartitionMillis = MAX_PROVENANCE_ENTRY_LIFE_TIME, int64_t maxPartitionBytes = MAX_PROVENANCE_STORAGE_SIZE, std::chrono::milliseconds purgePeriod = PROVENANCE_PURGE_PERIOD) Review Comment: nitpick: the parameters are not indented correctly. ########## extensions/rocksdb-repos/tests/ProvenanceTests.cpp: ########## @@ -127,11 +130,14 @@ TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[Test record1->addChildFlowFile(*ffr1); auto sample = 65555ms; - std::shared_ptr<core::Repository> testRepository = std::make_shared<core::repository::VolatileProvenanceRepository>(); + auto testRepository = std::make_shared<core::repository::VolatileProvenanceRepository>(); testRepository->initialize(nullptr); record1->setEventDuration(sample); - testRepository->storeElement(record1); + testRepository->appendEvents({record1}); + + utils::Identifier eventId = record1->getEventId(); Review Comment: This variable is redeclared and does not compile ########## extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp: ########## @@ -0,0 +1,207 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "RocksDbProvenanceRepository.h" + +#include <string> + +#include "core/Resource.h" + +namespace org::apache::nifi::minifi::provenance { + +namespace { +class EventCursor : public ProvenanceRepository::Cursor { +public: + explicit EventCursor(std::string event_id): event_id_(std::move(event_id)) {} + [[nodiscard]] + std::string toString() const override { + return event_id_; + } + ~EventCursor() override = default; + + std::string event_id_; +}; +} // namespace + +static const std::string_view NEXT_EVENT_UUID_KEY = "next_event_uuid"; + +bool RocksDbProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) { + if (!RocksDbRepository::initialize(config)) { + return false; + } + std::string value; + if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_); + if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) { + max_partition_bytes_ = gsl::narrow<int64_t>(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size")); + } + logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_); + if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) { + if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value)) + max_partition_millis_ = *max_partition; + } + logger_->log_debug("MiNiFi Provenance Max Storage Time: [{}]", max_partition_millis_); + + verify_checksums_in_rocksdb_reads_ = (config->get(Configure::nifi_provenance_repository_rocksdb_read_verify_checksums) | utils::andThen(&utils::string::toBool)).value_or(false); + logger_->log_debug("{} checksum verification in RocksDbProvenanceRepository", verify_checksums_in_rocksdb_reads_ ? "Using" : "Not using"); + + auto db_options = [] (minifi::internal::Writable<rocksdb::DBOptions>& db_opts) { + minifi::internal::setCommonRocksDbOptions(db_opts); + }; + + // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after + // This shouldn't go above 16MB and the configured total size of the db should cap it as well + auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) { + int64_t max_buffer_size = 16 << 20; + cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_)); + cf_opts.max_write_buffer_number = 4; + cf_opts.min_write_buffer_number_to_merge = 1; + + cf_opts.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO; + cf_opts.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false); + if (max_partition_millis_ > std::chrono::milliseconds(0)) { + cf_opts.ttl = std::chrono::duration_cast<std::chrono::seconds>(max_partition_millis_).count(); + } + }; + + db_ = minifi::internal::RocksDatabase::create(db_options, cf_options, directory_, + minifi::internal::getRocksDbOptionsToOverride(config, Configure::nifi_provenance_repository_rocksdb_options)); + std::string internal_state_db_uri = [&] { + const std::string_view minifidb_scheme = "minifidb://"; + if (directory_.starts_with(minifidb_scheme)) { + return directory_ + "-internal-state"; + } + std::string uri = utils::string::join_pack(minifidb_scheme, directory_); + if (uri.ends_with("/") || uri.ends_with("\\")) { + uri.pop_back(); + } + return uri + "/internal-state"; + }(); + internal_state_db_ = minifi::internal::RocksDatabase::create(db_options, {}, internal_state_db_uri, {}); + if (auto open_state_db = internal_state_db_->open()) { + rocksdb::ReadOptions options; + options.verify_checksums = verify_checksums_in_rocksdb_reads_; + std::string next_event_uuid_str; + if (open_state_db->Get(options, NEXT_EVENT_UUID_KEY, &next_event_uuid_str).ok()) { + next_event_id_ = next_event_uuid_str; + } else { + logger_->log_error("Could not find '{}'", NEXT_EVENT_UUID_KEY); + next_event_id_ = utils::IdGenerator::getIdGenerator()->generate(); + } + logger_->log_trace("Using next event uuid: {}", next_event_id_.to_string()); + } else { + logger_->log_error("Could not open internal state column in provenance repository {}", internal_state_db_uri); + return false; + } + if (db_->open()) { + logger_->log_debug("MiNiFi Provenance Repository database open {} success", directory_); + } else { + logger_->log_error("MiNiFi Provenance Repository database open {} failed", directory_); + return false; + } + + return true; +} + +void RocksDbProvenanceRepository::destroy() { + db_.reset(); +} + +std::unique_ptr<ProvenanceRepository::Cursor> RocksDbProvenanceRepository::cursorFromString(std::optional<std::string> cursor_str) { + if (cursor_str.has_value()) { + return std::make_unique<EventCursor>(cursor_str.value()); + } + return std::make_unique<EventCursor>(""); Review Comment: If the std::nullopt behavior is the same as the empty string, do we need to make this optional or can we just use empty string when we want a cursor from the start? Or is it possible in the future that an empty string and nullopt cases would differ? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
