This is an automated email from the ASF dual-hosted git repository. adamdebreceni pushed a commit to branch MINIFICPP-2845 in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3685104d65e786c9d86378a5a4101fa25c539efa Author: Adam Debreceni <[email protected]> AuthorDate: Mon Jul 6 11:13:01 2026 +0200 MINIFICPP-2845 - Monotonic provenance event ids, iteration support, refactor --- cmake/FetchSQLite.cmake | 29 +++ core-framework/common/src/utils/Id.cpp | 12 ++ core-framework/include/core/Repository.h | 6 - core-framework/src/core/Repository.cpp | 16 -- extensions/rocksdb-repos/ProvenanceRepository.cpp | 108 ----------- .../rocksdb-repos/RocksDbProvenanceRepository.cpp | 207 +++++++++++++++++++++ ...eRepository.h => RocksDbProvenanceRepository.h} | 30 ++- .../tests/DBProvenanceRepositoryTests.cpp | 142 +++++++++++++- extensions/rocksdb-repos/tests/ProvenanceTests.cpp | 41 ++-- extensions/rocksdb-repos/tests/RepoTests.cpp | 12 +- .../tests/integration/VerifyInvokeHTTP.h | 2 +- .../tests/unit/ProcessorTests.cpp | 7 +- libminifi/include/CronDrivenSchedulingAgent.h | 2 +- libminifi/include/EventDrivenSchedulingAgent.h | 2 +- libminifi/include/FlowController.h | 6 +- libminifi/include/SchedulingAgent.h | 4 +- libminifi/include/ThreadedSchedulingAgent.h | 2 +- libminifi/include/TimerDrivenSchedulingAgent.h | 2 +- libminifi/include/core/ProcessContextImpl.h | 8 +- .../reporting/SiteToSiteProvenanceReportingTask.h | 2 +- .../core/repository/NoOpThreadedRepository.h | 18 +- .../core/repository/VolatileProvenanceRepository.h | 35 +++- libminifi/include/provenance/Provenance.h | 25 ++- libminifi/src/FlowController.cpp | 2 +- libminifi/src/core/ProcessContextImpl.cpp | 4 +- .../src/core/flow/StructuredConfiguration.cpp | 4 + .../SiteToSiteProvenanceReportingTask.cpp | 51 ++++- libminifi/src/provenance/Provenance.cpp | 12 +- libminifi/test/flow-tests/SessionTests.cpp | 2 +- libminifi/test/integration/C2PauseResumeTest.cpp | 2 +- .../ControllerServiceIntegrationTests.cpp | 2 +- .../test/libtest/integration/IntegrationBase.cpp | 2 +- libminifi/test/libtest/unit/ProvenanceTestHelper.h | 27 ++- libminifi/test/libtest/unit/TestBase.cpp | 4 +- libminifi/test/libtest/unit/TestBase.h | 6 +- .../test/libtest/unit/TestControllerWithFlow.cpp | 2 +- .../test/persistence-tests/PersistenceTests.cpp | 6 +- libminifi/test/unit/SchedulingAgentTests.cpp | 2 +- minifi-api/common/include/minifi-cpp/utils/Id.h | 3 + .../include/minifi-cpp/core/ProcessContext.h | 3 +- minifi-api/include/minifi-cpp/core/Repository.h | 4 - .../include/minifi-cpp/provenance/Provenance.h | 5 +- .../minifi-cpp/provenance/ProvenanceRepository.h | 44 +++++ minifi_main/MiNiFiMain.cpp | 2 +- 44 files changed, 665 insertions(+), 242 deletions(-) diff --git a/cmake/FetchSQLite.cmake b/cmake/FetchSQLite.cmake new file mode 100644 index 000000000..65a9fb62f --- /dev/null +++ b/cmake/FetchSQLite.cmake @@ -0,0 +1,29 @@ +# 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(FetchContent) + +FetchContent_Declare( + sqlite3 + URL "https://www.sqlite.org/2026/sqlite-amalgamation-3530200.zip"; + URL_HASH "SHA256=8a310d0a16c7a90cacd4c884e70faa51c902afed2a89f63aaa0126ab83558a32" +) + +FetchContent_MakeAvailable(sqlite3) + +add_library(sqlite3 STATIC ${sqlite3_SOURCE_DIR}/sqlite3.c) +target_include_directories(sqlite3 PUBLIC ${sqlite3_SOURCE_DIR}) diff --git a/core-framework/common/src/utils/Id.cpp b/core-framework/common/src/utils/Id.cpp index d401d0a5e..09bdee3eb 100644 --- a/core-framework/common/src/utils/Id.cpp +++ b/core-framework/common/src/utils/Id.cpp @@ -43,6 +43,18 @@ bool Identifier::isNil() const { return *this == Identifier{}; } +Identifier& Identifier::operator++() { + for (int byte_idx = 15; byte_idx >= 0 && ++data_[byte_idx] == 0; --byte_idx) {} + return *this; +} + +Identifier Identifier::operator++(int) { + auto result = *this; + ++*this; + return result; +} + + bool Identifier::operator!=(const Identifier& other) const { return !(*this == other); } diff --git a/core-framework/include/core/Repository.h b/core-framework/include/core/Repository.h index 7a817ae2f..1615a20b8 100644 --- a/core-framework/include/core/Repository.h +++ b/core-framework/include/core/Repository.h @@ -106,12 +106,6 @@ class RepositoryImpl : public core::CoreComponentImpl, public core::RepositoryMe return false; } - std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t /*max_size*/) override { - return {}; - } - - bool storeElement(const std::shared_ptr<core::SerializableComponent>& element) override; - void loadComponent(const std::shared_ptr<core::ContentRepository>& /*content_repo*/) override { } diff --git a/core-framework/src/core/Repository.cpp b/core-framework/src/core/Repository.cpp index 722e82e7a..7004e05c0 100644 --- a/core-framework/src/core/Repository.cpp +++ b/core-framework/src/core/Repository.cpp @@ -27,20 +27,4 @@ bool RepositoryImpl::Delete(std::vector<std::shared_ptr<core::SerializableCompon return found; } -bool RepositoryImpl::storeElement(const std::shared_ptr<core::SerializableComponent>& element) { - if (!element) { - return false; - } - - org::apache::nifi::minifi::io::BufferStream stream; - - element->serialize(stream); - - if (!Put(element->getUUIDStr(), reinterpret_cast<const uint8_t*>(stream.getBuffer().data()), stream.size())) { - logger_->log_error("NiFi Provenance Store event {} size {} fail", element->getUUIDStr(), stream.size()); - return false; - } - return true; -} - } // namespace org::apache::nifi::minifi::core diff --git a/extensions/rocksdb-repos/ProvenanceRepository.cpp b/extensions/rocksdb-repos/ProvenanceRepository.cpp deleted file mode 100644 index 5c380545f..000000000 --- a/extensions/rocksdb-repos/ProvenanceRepository.cpp +++ /dev/null @@ -1,108 +0,0 @@ -/** - * 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 "ProvenanceRepository.h" - -#include <string> - -#include "core/Resource.h" - -namespace org::apache::nifi::minifi::provenance { - -bool ProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) { - 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 ProvenanceRepository", 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)); - 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; -} - -std::vector<std::shared_ptr<core::SerializableComponent>> ProvenanceRepository::getElements(size_t max_size) { - if (max_size == 0) { - return {}; - } - auto opendb = db_->open(); - if (!opendb) { - return {}; - } - std::vector<std::shared_ptr<core::SerializableComponent>> records; - rocksdb::ReadOptions options; - options.verify_checksums = verify_checksums_in_rocksdb_reads_; - std::unique_ptr<rocksdb::Iterator> it(opendb->NewIterator(options)); - for (it->SeekToFirst(); it->Valid(); it->Next()) { - 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; - } - } - } - return records; -} - -void ProvenanceRepository::destroy() { - db_.reset(); -} - -REGISTER_RESOURCE_AS(ProvenanceRepository, InternalResource, ("ProvenanceRepository", "provenancerepository")); - -} // namespace org::apache::nifi::minifi::provenance diff --git a/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp b/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp new file mode 100644 index 000000000..f3397ea62 --- /dev/null +++ b/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); + } + if (MultiPut(data)) { + return {}; + } + + return std::unexpected{"Failed to append provenance events"}; +} + +REGISTER_RESOURCE_AS(RocksDbProvenanceRepository, InternalResource, ("RocksDbProvenanceRepository", "ProvenanceRepository", "provenancerepository")); + +} // namespace org::apache::nifi::minifi::provenance diff --git a/extensions/rocksdb-repos/ProvenanceRepository.h b/extensions/rocksdb-repos/RocksDbProvenanceRepository.h similarity index 69% rename from extensions/rocksdb-repos/ProvenanceRepository.h rename to extensions/rocksdb-repos/RocksDbProvenanceRepository.h index 950ecd31c..d841fecd8 100644 --- a/extensions/rocksdb-repos/ProvenanceRepository.h +++ b/extensions/rocksdb-repos/RocksDbProvenanceRepository.h @@ -29,6 +29,7 @@ #include "core/Core.h" #include "core/logging/LoggerFactory.h" #include "minifi-cpp/provenance/Provenance.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" #include "minifi-cpp/utils/Literals.h" #include "RocksDbRepository.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) - : RocksDbRepository(repo_name.length() > 0 ? repo_name : core::className<ProvenanceRepository>(), - directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory<ProvenanceRepository>::getLogger()) { + : RocksDbRepository(repo_name.length() > 0 ? repo_name : core::className<RocksDbProvenanceRepository>(), + directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory<RocksDbProvenanceRepository>::getLogger()) { } - ~ProvenanceRepository() override { + ~RocksDbProvenanceRepository() override { stop(); } @@ -72,17 +73,26 @@ class ProvenanceRepository : public core::repository::RocksDbRepository { // The repo is cleaned up by itself, there is no need to delete items. return true; } - std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t max_size) override; void destroy(); - ProvenanceRepository(const ProvenanceRepository &parent) = delete; + RocksDbProvenanceRepository(const RocksDbProvenanceRepository &parent) = delete; - ProvenanceRepository &operator=(const ProvenanceRepository &parent) = delete; + RocksDbProvenanceRepository &operator=(const RocksDbProvenanceRepository &parent) = delete; + + std::unique_ptr<Cursor> cursorFromString(std::optional<std::string> cursor_str) override; + + std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t max_size, Cursor* cursor) override; + + std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<ProvenanceEventRecord>>& events) override; private: // Run function for the thread void run() override {}; + + std::unique_ptr<minifi::internal::RocksDatabase> internal_state_db_; + std::mutex next_event_id_mtx_; + utils::Identifier next_event_id_; }; } // namespace org::apache::nifi::minifi::provenance diff --git a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp index 8f6677c35..154a9961d 100644 --- a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp +++ b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp @@ -21,7 +21,7 @@ #include <random> #include <vector> -#include "ProvenanceRepository.h" +#include "RocksDbProvenanceRepository.h" #include "unit/TestBase.h" #include "unit/Catch.h" @@ -61,13 +61,24 @@ void verifyMaxKeyCount(const minifi::provenance::ProvenanceRepository& repo, uin REQUIRE(k < keyCount); } +std::vector<std::byte> serializeEvent(minifi::provenance::ProvenanceEventRecord& event) { + minifi::io::BufferStream stream; + event.serialize(stream); + return stream.moveBuffer(); +} + +template<typename T> +void appendAll(std::vector<T>& sink, const std::vector<T>& source) { + sink.insert(sink.end(), source.begin(), source.end()); +} + TEST_CASE("Test size limit", "[sizeLimitTest]") { TestController testController; auto temp_dir = testController.createTempDirectory(); REQUIRE(!temp_dir.empty()); // 60 sec, 100 KB - going to exceed the size limit - minifi::provenance::ProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1min, TEST_PROVENANCE_STORAGE_SIZE, 1s); + minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1min, TEST_PROVENANCE_STORAGE_SIZE, 1s); auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>(); configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); @@ -87,7 +98,7 @@ TEST_CASE("Test time limit", "[timeLimitTest]") { REQUIRE(!temp_dir.empty()); // 1 sec, 100 MB - going to exceed TTL - minifi::provenance::ProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>(); configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); @@ -114,3 +125,128 @@ TEST_CASE("Test time limit", "[timeLimitTest]") { verifyMaxKeyCount(provdb, 400); } + +TEST_CASE("Test query elements after cursor", "[iterationTest]") { + TestController testController; + auto temp_dir = testController.createTempDirectory(); + REQUIRE(!temp_dir.empty()); + + minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb.initialize(configuration)); + + std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> events; + for (size_t i = 0; i < 8; ++i) { + events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + } + + REQUIRE(provdb.appendEvents(events)); + + auto cursor = provdb.cursorFromString(std::nullopt); + REQUIRE(cursor); + + std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> queried_events; + + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 3); + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 6); + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 8); + + std::string last_event_id; + + for (size_t i = 0; i < queried_events.size(); ++i) { + REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()}); + last_event_id = queried_events.at(i)->getUUIDStr(); + REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i))); + } +} + +TEST_CASE("Test loading cursor from string", "[cursorSerializationTest]") { + TestController testController; + auto temp_dir = testController.createTempDirectory(); + REQUIRE(!temp_dir.empty()); + + minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb.initialize(configuration)); + + std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> events; + for (size_t i = 0; i < 8; ++i) { + events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + } + + REQUIRE(provdb.appendEvents(events)); + + auto cursor = provdb.cursorFromString(std::nullopt); + REQUIRE(cursor); + + std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> queried_events; + + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 3); + + cursor = provdb.cursorFromString(cursor->toString()); + REQUIRE(cursor); + + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 6); + + std::string last_event_id; + + for (size_t i = 0; i < queried_events.size(); ++i) { + REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()}); + last_event_id = queried_events.at(i)->getUUIDStr(); + REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i))); + } +} + +TEST_CASE("Test opening existing database loads monotonic counter", "[eventUuidMonotonicTest]") { + TestController testController; + auto temp_dir = testController.createTempDirectory(); + REQUIRE(!temp_dir.empty()); + + auto provdb = std::make_unique<minifi::provenance::RocksDbProvenanceRepository>("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb->initialize(configuration)); + + std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> events; + for (size_t i = 0; i < 4; ++i) { + events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + } + + REQUIRE(provdb->appendEvents(events)); + + provdb = std::make_unique<minifi::provenance::RocksDbProvenanceRepository>("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + REQUIRE(provdb->initialize(configuration)); + + std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> new_events; + for (size_t i = 0; i < 4; ++i) { + new_events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + events.push_back(new_events.back()); + } + + REQUIRE(provdb->appendEvents(new_events)); + + std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> queried_events = provdb->getEvents(8, nullptr).value(); + REQUIRE(queried_events.size() == 8); + + std::string last_event_id; + + for (size_t i = 0; i < queried_events.size(); ++i) { + REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()}); + last_event_id = queried_events.at(i)->getUUIDStr(); + REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i))); + } +} diff --git a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp index fca1926bb..bc40ebe19 100644 --- a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp +++ b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp @@ -34,13 +34,13 @@ namespace provenance = minifi::provenance; using namespace std::literals::chrono_literals; TEST_CASE("Test Provenance record create", "[Testprovenance::ProvenanceEventRecord]") { - auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "blah", "blahblah"); + auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000001").value(), "blahblah"); REQUIRE(record1->getAttributes().empty()); REQUIRE(record1->getAlternateIdentifierUri().empty()); } TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") { - auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype"); + auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000002").value(), "componenttype"); utils::Identifier eventId = record1->getEventId(); @@ -48,10 +48,10 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv record1->setDetails(smileyface); auto sample = 65555ms; - std::shared_ptr<core::Repository> testRepository = std::make_shared<TestRepository>(); + auto testRepository = std::make_shared<TestRepository>(); record1->setEventDuration(sample); - testRepository->storeElement(record1); + testRepository->appendEvents({record1}); auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); @@ -64,7 +64,7 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv } TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") { - auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, "componentid", "componenttype"); + auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, utils::Identifier::parse("00000000-0000-0000-0000-000000000003").value(), "componenttype"); utils::Identifier eventId = record1->getEventId(); std::shared_ptr<minifi::FlowFileRecord> ffr1 = std::make_shared<minifi::FlowFileRecordImpl>(); ffr1->setAttribute("potato", "potatoe"); @@ -73,10 +73,10 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") { record1->addChildFlowFile(*ffr1); auto sample = 65555ms; - std::shared_ptr<core::Repository> testRepository = std::make_shared<TestRepository>(); + auto testRepository = std::make_shared<TestRepository>(); record1->setEventDuration(sample); - testRepository->storeElement(record1); + testRepository->appendEvents({record1}); auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); @@ -87,20 +87,21 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") { } TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") { - auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype"); - - utils::Identifier eventId = record1->getEventId(); + auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000004").value(), "componenttype"); std::string smileyface = ":)"; record1->setDetails(smileyface); 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(); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); @@ -113,8 +114,7 @@ TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::Pro } TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[TestFlowAndProv1]") { - auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, "componentid", "componenttype"); - utils::Identifier eventId = record1->getEventId(); + auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, utils::Identifier::parse("00000000-0000-0000-0000-000000000005").value(), "componenttype"); std::shared_ptr<minifi::FlowFileRecord> ffr1 = std::make_shared<minifi::FlowFileRecordImpl>(); ffr1->setAttribute("potato", "potatoe"); ffr1->setAttribute("tomato", "tomatoe"); @@ -122,11 +122,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(); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); @@ -137,7 +140,7 @@ TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[Test } TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") { - auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype"); + auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000006").value(), "componenttype"); utils::Identifier eventId = record1->getEventId(); @@ -146,11 +149,11 @@ TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::Provena auto sample = 65555ms; - std::shared_ptr<core::Repository> testRepository = core::createRepository("nooprepository"); + std::shared_ptr testRepository = utils::dynamic_unique_cast<minifi::provenance::ProvenanceRepository>(core::createRepository("nooprepository")); testRepository->initialize(nullptr); record1->setEventDuration(sample); - REQUIRE(testRepository->storeElement(record1)); + REQUIRE(testRepository->appendEvents({record1})); auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == false); diff --git a/extensions/rocksdb-repos/tests/RepoTests.cpp b/extensions/rocksdb-repos/tests/RepoTests.cpp index f3a8fd76f..771c8f2a8 100644 --- a/extensions/rocksdb-repos/tests/RepoTests.cpp +++ b/extensions/rocksdb-repos/tests/RepoTests.cpp @@ -28,7 +28,7 @@ #include "core/RepositoryFactory.h" #include "FlowFileRecord.h" #include "FlowFileRepository.h" -#include "ProvenanceRepository.h" +#include "RocksDbProvenanceRepository.h" #include "properties/Configure.h" #include "unit/ProvenanceTestHelper.h" #include "unit/TestBase.h" @@ -280,7 +280,7 @@ TEST_CASE("Test FlowFile Restore", "[TestFFR6]") { config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string()); config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string()); - std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestThreadedRepository>(); + auto prov_repo = std::make_shared<TestThreadedRepository>(); auto ff_repository = std::make_shared<core::repository::FlowFileRepository>("flowFileRepository"); std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::FileSystemRepository>(); ff_repository->initialize(config); @@ -549,8 +549,8 @@ TEST_CASE("Test getting flow file repository size properties", "[TestGettingRepo expected_rocksdb_stats = true; } - SECTION("ProvenanceRepository") { - repository = std::make_shared<minifi::provenance::ProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms); + SECTION("RocksDbProvenanceRepository") { + repository = std::make_shared<minifi::provenance::RocksDbProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms); expected_rocksdb_stats = true; } @@ -707,8 +707,8 @@ TEST_CASE("Flow file repositories can be stopped", "[TestRepoIsRunning]") { repository = std::make_shared<core::repository::FlowFileRepository>("ff", dir.string(), 0ms, 0, 1ms); } - SECTION("ProvenanceRepository") { - repository = std::make_shared<minifi::provenance::ProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms); + SECTION("RocksDbProvenanceRepository") { + repository = std::make_shared<minifi::provenance::RocksDbProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms); } SECTION("VolatileProvenanceRepository") { diff --git a/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h b/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h index 0f6c74a03..499e5e41b 100644 --- a/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h +++ b/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h @@ -85,7 +85,7 @@ class VerifyInvokeHTTP : public HTTPIntegrationBase { virtual void setupFlow() { testSetup(); - std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>(); + auto test_repo = std::make_shared<TestThreadedRepository>(); std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>(); if (flow_config_path_.config_path) { diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp b/extensions/standard-processors/tests/unit/ProcessorTests.cpp index 6dd308c7c..3f3c861b0 100644 --- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp +++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp @@ -468,11 +468,12 @@ TEST_CASE("Test Find file", "[getfileCreate3]") { taskReport->setBatchSize(1); processorReport->incrementActiveTasks(); processorReport->setScheduledState(core::ScheduledState::RUNNING); - auto recordsReport = repo->getElements(1); + auto recordsReport = repo->getEvents(1, nullptr); + REQUIRE(recordsReport); std::function<void(const std::shared_ptr<core::ProcessContext> &, const std::shared_ptr<core::ProcessSession>&)> verifyReporter = [&](const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) { - auto json_str = taskReport->getJsonReport(*context, *session, recordsReport); - REQUIRE(recordsReport.size() == 1); + auto json_str = taskReport->getJsonReport(*context, *session, recordsReport.value()); + REQUIRE(recordsReport->size() == 1); REQUIRE(taskReport->getName() == std::string(minifi::core::reporting::SiteToSiteProvenanceReportingTask::ReportTaskName)); REQUIRE(json_str.find("\"componentType\": \"getfileCreate2\"") != std::string::npos); }; diff --git a/libminifi/include/CronDrivenSchedulingAgent.h b/libminifi/include/CronDrivenSchedulingAgent.h index d08ac5a4f..46b0d68c0 100644 --- a/libminifi/include/CronDrivenSchedulingAgent.h +++ b/libminifi/include/CronDrivenSchedulingAgent.h @@ -37,7 +37,7 @@ namespace org::apache::nifi::minifi { class CronDrivenSchedulingAgent : public ThreadedSchedulingAgent { public: CronDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, - std::shared_ptr<core::Repository> repo, + std::shared_ptr<provenance::ProvenanceRepository> repo, std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration, diff --git a/libminifi/include/EventDrivenSchedulingAgent.h b/libminifi/include/EventDrivenSchedulingAgent.h index 060fff588..d5ccd54b2 100644 --- a/libminifi/include/EventDrivenSchedulingAgent.h +++ b/libminifi/include/EventDrivenSchedulingAgent.h @@ -35,7 +35,7 @@ namespace org::apache::nifi::minifi { class EventDrivenSchedulingAgent : public ThreadedSchedulingAgent { public: - EventDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo, + EventDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository> repo, std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration, utils::ThreadPool &thread_pool) : ThreadedSchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configuration, thread_pool) { diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h index 00ff3f756..f3f654c1a 100644 --- a/libminifi/include/FlowController.h +++ b/libminifi/include/FlowController.h @@ -66,7 +66,7 @@ class ProcessorController; class FlowController : public core::controller::ForwardingControllerServiceProvider, public state::StateMonitor { public: - FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo, + FlowController(std::shared_ptr<provenance::ProvenanceRepository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<Configure> configure, std::shared_ptr<core::FlowConfiguration> flow_configuration, std::shared_ptr<core::ContentRepository> content_repo, std::unique_ptr<state::MetricsPublisherStore> metrics_publisher_store = nullptr, std::shared_ptr<utils::file::FileSystem> filesystem = std::make_shared<utils::file::FileSystem>(), std::function<void()> request_restart = []{}, @@ -74,7 +74,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi ~FlowController() override; - virtual std::shared_ptr<core::Repository> getProvenanceRepository() { + virtual std::shared_ptr<provenance::ProvenanceRepository> getProvenanceRepository() { return this->provenance_repo_; } @@ -187,7 +187,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi // Thread pool for schedulers utils::ThreadPool thread_pool_; std::shared_ptr<Configure> configuration_; - std::shared_ptr<core::Repository> provenance_repo_; + std::shared_ptr<provenance::ProvenanceRepository> provenance_repo_; std::shared_ptr<core::Repository> flow_file_repo_; std::shared_ptr<core::ContentRepository> content_repo_; std::shared_ptr<core::FlowConfiguration> flow_configuration_; diff --git a/libminifi/include/SchedulingAgent.h b/libminifi/include/SchedulingAgent.h index d5cab7af5..03e5111bd 100644 --- a/libminifi/include/SchedulingAgent.h +++ b/libminifi/include/SchedulingAgent.h @@ -50,7 +50,7 @@ namespace org::apache::nifi::minifi { class SchedulingAgent { public: - SchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_repo, + SchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository> repo, std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration, utils::ThreadPool& thread_pool) : admin_yield_duration_(), bored_yield_duration_(0), @@ -122,7 +122,7 @@ class SchedulingAgent { std::shared_ptr<Configure> configure_; - std::shared_ptr<core::Repository> repo_; + std::shared_ptr<provenance::ProvenanceRepository> repo_; std::shared_ptr<core::Repository> flow_repo_; diff --git a/libminifi/include/ThreadedSchedulingAgent.h b/libminifi/include/ThreadedSchedulingAgent.h index 7a3ad2299..2668bab0d 100644 --- a/libminifi/include/ThreadedSchedulingAgent.h +++ b/libminifi/include/ThreadedSchedulingAgent.h @@ -38,7 +38,7 @@ namespace org::apache::nifi::minifi { */ class ThreadedSchedulingAgent : public SchedulingAgent { public: - ThreadedSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo, + ThreadedSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository> repo, std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration, utils::ThreadPool &thread_pool) : SchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configuration, thread_pool) { diff --git a/libminifi/include/TimerDrivenSchedulingAgent.h b/libminifi/include/TimerDrivenSchedulingAgent.h index 99073ce99..055a81eb7 100644 --- a/libminifi/include/TimerDrivenSchedulingAgent.h +++ b/libminifi/include/TimerDrivenSchedulingAgent.h @@ -30,7 +30,7 @@ namespace org::apache::nifi::minifi { class TimerDrivenSchedulingAgent : public ThreadedSchedulingAgent { public: - TimerDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo, + TimerDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository> repo, std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configure, utils::ThreadPool &thread_pool) : ThreadedSchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configure, thread_pool) { diff --git a/libminifi/include/core/ProcessContextImpl.h b/libminifi/include/core/ProcessContextImpl.h index 9466e69df..0f761013d 100644 --- a/libminifi/include/core/ProcessContextImpl.h +++ b/libminifi/include/core/ProcessContextImpl.h @@ -50,11 +50,11 @@ class Processor; class ProcessContextImpl : public core::VariableRegistryImpl, public virtual ProcessContext { public: ProcessContextImpl(Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, - const std::shared_ptr<core::Repository>& repo, const std::shared_ptr<core::Repository>& flow_repo, + const std::shared_ptr<provenance::ProvenanceRepository>& repo, const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<core::ContentRepository>& content_repo = repository::createFileSystemRepository()); ProcessContextImpl(Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, - const std::shared_ptr<core::Repository>& repo, const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<minifi::Configure>& configuration, + const std::shared_ptr<provenance::ProvenanceRepository>& repo, const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<minifi::Configure>& configuration, const std::shared_ptr<core::ContentRepository>& content_repo = repository::createFileSystemRepository()); // Get Processor associated with the Process Context @@ -85,7 +85,7 @@ class ProcessContextImpl : public core::VariableRegistryImpl, public virtual Pro void yield() override; - std::shared_ptr<core::Repository> getProvenanceRepository() override { return repo_; } + std::shared_ptr<provenance::ProvenanceRepository> getProvenanceRepository() override { return repo_; } /** * Returns a reference to the content repository for the running instance. @@ -198,7 +198,7 @@ class ProcessContextImpl : public core::VariableRegistryImpl, public virtual Pro std::shared_ptr<logging::Logger> logger_; controller::ControllerServiceProvider* controller_service_provider_; std::shared_ptr<core::StateStorage> state_storage_; - std::shared_ptr<core::Repository> repo_; + std::shared_ptr<provenance::ProvenanceRepository> repo_; std::shared_ptr<core::Repository> flow_repo_; std::shared_ptr<core::ContentRepository> content_repo_; Processor& processor_; diff --git a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h index 29767ad60..b53a78013 100644 --- a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h +++ b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h @@ -45,7 +45,7 @@ class SiteToSiteProvenanceReportingTask : public minifi::RemoteProcessGroupPort static constexpr char const* ReportTaskName = "SiteToSiteProvenanceReportingTask"; static const char *ProvenanceAppStr; - static std::string getJsonReport(core::ProcessContext& context, core::ProcessSession& session, const std::vector<std::shared_ptr<core::SerializableComponent>> &records); // NOLINT + static std::string getJsonReport(core::ProcessContext& context, core::ProcessSession& session, const std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> &records); // NOLINT void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override; diff --git a/libminifi/include/core/repository/NoOpThreadedRepository.h b/libminifi/include/core/repository/NoOpThreadedRepository.h index df23a1b32..06bc01f37 100644 --- a/libminifi/include/core/repository/NoOpThreadedRepository.h +++ b/libminifi/include/core/repository/NoOpThreadedRepository.h @@ -21,10 +21,11 @@ #include <thread> #include "core/ThreadedRepository.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" namespace org::apache::nifi::minifi::core::repository { -class NoOpThreadedRepository : public core::ThreadedRepositoryImpl { +class NoOpThreadedRepository : public core::ThreadedRepositoryImpl, public provenance::ProvenanceRepository { public: explicit NoOpThreadedRepository(std::string_view repo_name) : ThreadedRepositoryImpl(repo_name) { @@ -47,6 +48,21 @@ class NoOpThreadedRepository : public core::ThreadedRepositoryImpl { return 0; } + std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>& /*events*/) override { + return {}; + } + + std::unique_ptr<ProvenanceRepository::Cursor> cursorFromString(std::optional<std::string> /*cursor_str*/) override { + return nullptr; + } + + std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t /*max_size*/, Cursor* cursor) override { + if (cursor) { + return std::unexpected{"Cursor based query is not supported"}; + } + return {}; + } + private: void run() override { } diff --git a/libminifi/include/core/repository/VolatileProvenanceRepository.h b/libminifi/include/core/repository/VolatileProvenanceRepository.h index da5df86fc..278f0b8f1 100644 --- a/libminifi/include/core/repository/VolatileProvenanceRepository.h +++ b/libminifi/include/core/repository/VolatileProvenanceRepository.h @@ -21,10 +21,11 @@ #include <string_view> #include "VolatileRepository.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" namespace org::apache::nifi::minifi::core::repository { -class VolatileProvenanceRepository : public VolatileRepository { +class VolatileProvenanceRepository : public VolatileRepository, public provenance::ProvenanceRepository { public: explicit VolatileProvenanceRepository(std::string_view repo_name = "", std::string /*dir*/ = REPOSITORY_DIRECTORY, @@ -38,6 +39,36 @@ class VolatileProvenanceRepository : public VolatileRepository { stop(); } + bool initialize(const std::shared_ptr<Configure> &configure) override { + if (!VolatileRepository::initialize(configure)) { + return false; + } + next_event_id_ = utils::IdGenerator::getIdGenerator()->generate(); + return true; + } + + std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>& events) override { + 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_++); + data.emplace_back(event->getUUIDStr(), std::make_unique<io::BufferStream>()); + event->serialize(*data.back().second); + } + MultiPut(data); + + return {}; + } + + std::unique_ptr<ProvenanceRepository::Cursor> cursorFromString(std::optional<std::string> /*cursor_str*/) override { + return nullptr; + } + + std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t /*max_size*/, Cursor* /*cursor*/) override { + return std::unexpected{"Querying events is not yet supported"}; + } + private: void run() override { } @@ -51,6 +82,8 @@ class VolatileProvenanceRepository : public VolatileRepository { } std::thread thread_; + std::mutex next_event_id_mtx_; + utils::Identifier next_event_id_; }; } // namespace org::apache::nifi::minifi::core::repository diff --git a/libminifi/include/provenance/Provenance.h b/libminifi/include/provenance/Provenance.h index 5539dc5c2..91d4b91dd 100644 --- a/libminifi/include/provenance/Provenance.h +++ b/libminifi/include/provenance/Provenance.h @@ -38,6 +38,7 @@ #include "utils/Id.h" #include "utils/TimeUtil.h" #include "minifi-cpp/provenance/Provenance.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" namespace org::apache::nifi::minifi::provenance { @@ -54,6 +55,14 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public ~ProvenanceEventRecordImpl() override = default; + std::optional<uint64_t> getEventOrdinal() const override { + return event_ordinal_; + } + + void setEventOrdinal(uint64_t event_ordinal) override { + event_ordinal_ = event_ordinal; + } + utils::Identifier getEventId() const override { return getUUID(); } @@ -218,6 +227,8 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public protected: ProvenanceEventType event_type_; + // the index of the event + std::optional<uint64_t> event_ordinal_; // Date at which the event was created std::chrono::system_clock::time_point event_time_{}; // Date at which the flow file entered the flow @@ -251,7 +262,7 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public class ProvenanceReporterImpl : public virtual ProvenanceReporter { public: - ProvenanceReporterImpl(std::shared_ptr<core::Repository> repo, utils::Identifier component_id, std::string component_type) + ProvenanceReporterImpl(std::shared_ptr<provenance::ProvenanceRepository> repo, utils::Identifier component_id, std::string component_type) : component_id_(component_id), component_type_(std::move(component_type)), logger_(core::logging::LoggerFactory<ProvenanceReporter>::getLogger()), @@ -266,16 +277,12 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter { clear(); } - std::set<std::shared_ptr<ProvenanceEventRecord>> getEvents() const override { + std::vector<std::shared_ptr<ProvenanceEventRecord>> getEvents() const override { return events_; } void add(const std::shared_ptr<ProvenanceEventRecord> &event) override { - events_.insert(event); - } - - void remove(const std::shared_ptr<ProvenanceEventRecord> &event) override { - events_.erase(event); + events_.push_back(event); } void clear() final { @@ -313,8 +320,8 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter { private: std::shared_ptr<core::logging::Logger> logger_; - std::set<std::shared_ptr<ProvenanceEventRecord>> events_; - std::shared_ptr<core::Repository> repo_; + std::vector<std::shared_ptr<ProvenanceEventRecord>> events_; + std::shared_ptr<provenance::ProvenanceRepository> repo_; }; } // namespace org::apache::nifi::minifi::provenance diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp index 916299805..107c44b37 100644 --- a/libminifi/src/FlowController.cpp +++ b/libminifi/src/FlowController.cpp @@ -46,7 +46,7 @@ namespace org::apache::nifi::minifi { -FlowController::FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo, +FlowController::FlowController(std::shared_ptr<provenance::ProvenanceRepository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<Configure> configure, std::shared_ptr<core::FlowConfiguration> flow_configuration, std::shared_ptr<core::ContentRepository> content_repo, std::unique_ptr<state::MetricsPublisherStore> metrics_publisher_store, std::shared_ptr<utils::file::FileSystem> filesystem, std::function<void()> request_restart, diff --git a/libminifi/src/core/ProcessContextImpl.cpp b/libminifi/src/core/ProcessContextImpl.cpp index 841b74ca5..6ac0372cf 100644 --- a/libminifi/src/core/ProcessContextImpl.cpp +++ b/libminifi/src/core/ProcessContextImpl.cpp @@ -41,7 +41,7 @@ class StandardProcessorInfo : public ProcessorInfo { } // namespace ProcessContextImpl::ProcessContextImpl( - Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, const std::shared_ptr<core::Repository>& repo, + Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, const std::shared_ptr<provenance::ProvenanceRepository>& repo, const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<core::ContentRepository>& content_repo) : VariableRegistryImpl(static_cast<std::shared_ptr<Configure>>(minifi::Configure::create())), logger_(logging::LoggerFactory<ProcessContext>::getLogger()), @@ -55,7 +55,7 @@ ProcessContextImpl::ProcessContextImpl( info_(std::make_unique<StandardProcessorInfo>(processor)) {} ProcessContextImpl::ProcessContextImpl( - Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, const std::shared_ptr<core::Repository>& repo, + Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, const std::shared_ptr<provenance::ProvenanceRepository>& repo, const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<minifi::Configure>& configuration, const std::shared_ptr<core::ContentRepository>& content_repo) : VariableRegistryImpl(configuration), diff --git a/libminifi/src/core/flow/StructuredConfiguration.cpp b/libminifi/src/core/flow/StructuredConfiguration.cpp index e64b5a34d..d6a94a330 100644 --- a/libminifi/src/core/flow/StructuredConfiguration.cpp +++ b/libminifi/src/core/flow/StructuredConfiguration.cpp @@ -586,6 +586,10 @@ void StructuredConfiguration::parseProvenanceReporting(const Node& node, core::P auto report_task = createProvenanceReportTask(); + utils::Identifier task_uuid; + task_uuid = getOrGenerateId(node); + report_task->setUUID(task_uuid); + checkRequiredField(node, schema_.scheduling_strategy); auto schedulingStrategyStr = node[schema_.scheduling_strategy].getString().value(); checkRequiredField(node, schema_.scheduling_period); diff --git a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp index 6fb3d5d3c..38bc15d64 100644 --- a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp +++ b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp @@ -88,7 +88,7 @@ void appendJsonStr(const utils::SmallString<N>& value, rapidjson::Value& parent, parent.PushBack(valueVal, alloc); } -std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, const std::vector<std::shared_ptr<core::SerializableComponent>> &records) { +std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, const std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> &records) { rapidjson::Document array(rapidjson::kArrayType); rapidjson::Document::AllocatorType &alloc = array.GetAllocator(); @@ -111,6 +111,10 @@ std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContex recordJson.AddMember("entityType", "org.apache.nifi.flowfile.FlowFile", alloc); + if (auto event_ordinal = record->getEventOrdinal()) { + recordJson.AddMember("eventOrdinal", event_ordinal.value(), alloc); + } + recordJson.AddMember("eventId", getStringValue(record->getEventId().to_string(), alloc), alloc); recordJson.AddMember("eventType", getStringValue(provenance::ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()], alloc), alloc); recordJson.AddMember("details", getStringValue(record->getDetails(), alloc), alloc); @@ -151,17 +155,36 @@ std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContex } void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, core::ProcessSession& session) { + std::shared_ptr<provenance::ProvenanceRepository> repo = context.getProvenanceRepository(); + if (!repo) { + throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to retrieve provenance repository"); + } auto* state_manager = context.getStateManager(); if (!state_manager) { logger_->log_error("Failed to get StateManager"); context.yield(); return; } - std::shared_ptr<core::Repository> repo = context.getProvenanceRepository(); - if (!repo) { - throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to retrieve provenance repository"); + 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); + } + if (auto result = repo->getEvents(batch_size_, cursor.get())) { + records = std::move(result.value()); + } else { + throw minifi::Exception(GENERAL_EXCEPTION, "Failed to retrieve records: " + result.error()); } - auto records = repo->getElements(batch_size_); if (records.empty()) { logger_->log_debug("No new provenance records"); return; @@ -189,8 +212,22 @@ void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, return; } - // we transfer the record, purge the record from DB - repo->Delete(records); + if (cursor) { + // no need to delete just update the state + std::unordered_map<std::string, std::string> state_map; + state_map["cursor"] = cursor->toString(); + if (!state_manager->set(state_map)) { + logger_->log_error("Failed to update cursor state"); + } + } else { + // we transfer the record, purge the record from DB + std::vector<std::shared_ptr<core::SerializableComponent>> entries; + entries.reserve(records.size()); + for (const auto& record : records) { + entries.push_back(record); + } + repo->Delete(entries); + } returnProtocol(context, std::move(protocol_)); } diff --git a/libminifi/src/provenance/Provenance.cpp b/libminifi/src/provenance/Provenance.cpp index 66bcff976..8fa948642 100644 --- a/libminifi/src/provenance/Provenance.cpp +++ b/libminifi/src/provenance/Provenance.cpp @@ -454,15 +454,7 @@ void ProvenanceReporterImpl::commit() { return; } - std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>> flowData; - - for (auto& event : events_) { - auto stramptr = std::make_unique<io::BufferStream>(); - event->serialize(*stramptr); - - flowData.emplace_back(event->getUUIDStr(), std::move(stramptr)); - } - repo_->MultiPut(flowData); + repo_->appendEvents(events_); } void ProvenanceReporterImpl::create(const core::FlowFile& flow_file, const std::string& detail) { @@ -543,7 +535,7 @@ void ProvenanceReporterImpl::send(const core::FlowFile& flow_file, const std::st add(event); } else { if (!repo_->isFull()) - repo_->storeElement(event); + repo_->appendEvents({event}); } } } diff --git a/libminifi/test/flow-tests/SessionTests.cpp b/libminifi/test/flow-tests/SessionTests.cpp index 533e3bf56..876a22375 100644 --- a/libminifi/test/flow-tests/SessionTests.cpp +++ b/libminifi/test/flow-tests/SessionTests.cpp @@ -62,7 +62,7 @@ TEST_CASE("Import null data") { config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string()); config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string()); - std::shared_ptr<core::Repository> prov_repo = core::createRepository("nooprepository"); + std::shared_ptr prov_repo = utils::dynamic_unique_cast<minifi::provenance::ProvenanceRepository>(core::createRepository("nooprepository")); std::shared_ptr<core::Repository> ff_repository = std::make_shared<core::repository::FlowFileRepository>("flowFileRepository"); std::shared_ptr<core::ContentRepository> content_repo; SECTION("VolatileContentRepository") { diff --git a/libminifi/test/integration/C2PauseResumeTest.cpp b/libminifi/test/integration/C2PauseResumeTest.cpp index b9e6efe15..39afd2bef 100644 --- a/libminifi/test/integration/C2PauseResumeTest.cpp +++ b/libminifi/test/integration/C2PauseResumeTest.cpp @@ -122,7 +122,7 @@ TEST_CASE("C2PauseResumeTest", "[c2test]") { VerifyC2PauseResume harness{test_file_path, flow_resumed_successfully}; PauseResumeHandler responder{flow_resumed_successfully, harness.getConfiguration()}; - std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>(); + auto test_repo = std::make_shared<TestThreadedRepository>(); std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>(); std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::ConfigureImpl>(); configuration->set(minifi::Configure::nifi_flow_configuration_file, test_file_path.string()); diff --git a/libminifi/test/integration/ControllerServiceIntegrationTests.cpp b/libminifi/test/integration/ControllerServiceIntegrationTests.cpp index d5586ca73..99f80b401 100644 --- a/libminifi/test/integration/ControllerServiceIntegrationTests.cpp +++ b/libminifi/test/integration/ControllerServiceIntegrationTests.cpp @@ -52,7 +52,7 @@ TEST_CASE("ControllerServiceIntegrationTests", "[controller]") { using org::apache::nifi::minifi::test::utils::verifyEventHappenedInPollTime; std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::ConfigureImpl>(); - std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>(); + auto test_repo = std::make_shared<TestThreadedRepository>(); std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>(); const auto test_file_path = std::filesystem::path(TEST_RESOURCES) / "TestControllerServices.yml"; diff --git a/libminifi/test/libtest/integration/IntegrationBase.cpp b/libminifi/test/libtest/integration/IntegrationBase.cpp index 8a767546a..b70ffa066 100644 --- a/libminifi/test/libtest/integration/IntegrationBase.cpp +++ b/libminifi/test/libtest/integration/IntegrationBase.cpp @@ -59,7 +59,7 @@ void IntegrationBase::run() { using namespace std::literals::chrono_literals; testSetup(); - std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>(); + auto test_repo = std::make_shared<TestThreadedRepository>(); std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>(); if (flow_config_path_.config_path) { diff --git a/libminifi/test/libtest/unit/ProvenanceTestHelper.h b/libminifi/test/libtest/unit/ProvenanceTestHelper.h index 03d6e0c32..1901c707a 100644 --- a/libminifi/test/libtest/unit/ProvenanceTestHelper.h +++ b/libminifi/test/libtest/unit/ProvenanceTestHelper.h @@ -41,7 +41,7 @@ using namespace std::literals::chrono_literals; const int64_t TEST_MAX_REPOSITORY_STORAGE_SIZE = 100; template <typename T_BaseRepository> -class TestRepositoryBase : public T_BaseRepository { +class TestRepositoryBase : public T_BaseRepository, public org::apache::nifi::minifi::provenance::ProvenanceRepository { public: TestRepositoryBase() : T_BaseRepository("repo_name", "./dir", 1s, TEST_MAX_REPOSITORY_STORAGE_SIZE, 0ms) { @@ -89,11 +89,14 @@ class TestRepositoryBase : public T_BaseRepository { } } - std::vector<std::shared_ptr<org::apache::nifi::minifi::core::SerializableComponent>> getElements(size_t max_size) override { + std::expected<std::vector<std::shared_ptr<org::apache::nifi::minifi::provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t max_size, Cursor* cursor) override { + if (cursor) { + return std::unexpected{"Cursor based query is not supported"}; + } if (max_size == 0) { return {}; } - std::vector<std::shared_ptr<org::apache::nifi::minifi::core::SerializableComponent>> store; + std::vector<std::shared_ptr<org::apache::nifi::minifi::provenance::ProvenanceEventRecord>> store; std::lock_guard<std::mutex> lock{repository_results_mutex_}; for (const auto &entry : repository_results_) { const auto eventRead = org::apache::nifi::minifi::provenance::ProvenanceEventRecord::create(); @@ -113,6 +116,24 @@ class TestRepositoryBase : public T_BaseRepository { return repository_results_; } + std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<org::apache::nifi::minifi::provenance::ProvenanceEventRecord>>& events) override { + std::vector<std::pair<std::string, std::unique_ptr<org::apache::nifi::minifi::io::BufferStream>>> data; + data.reserve(events.size()); + for (auto& event : events) { + data.emplace_back(event->getUUIDStr(), std::make_unique<org::apache::nifi::minifi::io::BufferStream>()); + event->serialize(*data.back().second); + } + if (!MultiPut(data)) { + return std::unexpected{"Failed to store provenance events"}; + } + + return {}; + } + + std::unique_ptr<ProvenanceRepository::Cursor> cursorFromString(std::optional<std::string> /*cursor_str*/) override { + return nullptr; + } + protected: mutable std::mutex repository_results_mutex_; std::map<std::string, std::string> repository_results_; diff --git a/libminifi/test/libtest/unit/TestBase.cpp b/libminifi/test/libtest/unit/TestBase.cpp index 23a967ae9..c5477e67c 100644 --- a/libminifi/test/libtest/unit/TestBase.cpp +++ b/libminifi/test/libtest/unit/TestBase.cpp @@ -208,7 +208,7 @@ LogTestController::LogTestController(const std::shared_ptr<logging::LoggerProper init(loggerProps); } -TestPlan::TestPlan(std::shared_ptr<minifi::core::ContentRepository> content_repo, std::shared_ptr<minifi::core::Repository> flow_repo, std::shared_ptr<minifi::core::Repository> prov_repo, +TestPlan::TestPlan(std::shared_ptr<minifi::core::ContentRepository> content_repo, std::shared_ptr<minifi::core::Repository> flow_repo, std::shared_ptr<minifi::provenance::ProvenanceRepository> prov_repo, std::shared_ptr<minifi::state::response::FlowVersion> flow_version, std::shared_ptr<minifi::Configure> configuration, const char* state_dir) : configuration_(std::move(configuration)), content_repo_(std::move(content_repo)), @@ -637,7 +637,7 @@ std::shared_ptr<minifi::core::FlowFile> TestPlan::getFlowFileProducedByCurrentPr return nullptr; } -std::set<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> TestPlan::getProvenanceRecords() { +std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> TestPlan::getProvenanceRecords() { return process_sessions_.at(location)->getProvenanceReporter()->getEvents(); } diff --git a/libminifi/test/libtest/unit/TestBase.h b/libminifi/test/libtest/unit/TestBase.h index 9e5557194..b2d20cde8 100644 --- a/libminifi/test/libtest/unit/TestBase.h +++ b/libminifi/test/libtest/unit/TestBase.h @@ -243,7 +243,7 @@ class TypedProcessorWrapper { class TestPlan { public: - explicit TestPlan(std::shared_ptr<minifi::core::ContentRepository> content_repo, std::shared_ptr<minifi::core::Repository> flow_repo, std::shared_ptr<minifi::core::Repository> prov_repo, + explicit TestPlan(std::shared_ptr<minifi::core::ContentRepository> content_repo, std::shared_ptr<minifi::core::Repository> flow_repo, std::shared_ptr<minifi::provenance::ProvenanceRepository> prov_repo, std::shared_ptr<minifi::state::response::FlowVersion> flow_version, std::shared_ptr<minifi::Configure> configuration, const char* state_dir); virtual ~TestPlan(); @@ -305,7 +305,7 @@ class TestPlan { bool runCurrentProcessor(); bool runCurrentProcessorUntilFlowfileIsProduced(std::chrono::milliseconds wait_duration); - std::set<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> getProvenanceRecords(); + std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> getProvenanceRecords(); std::shared_ptr<minifi::core::FlowFile> getCurrentFlowFile(); std::vector<minifi::Connection*> getProcessorOutboundConnections(minifi::core::Processor* processor); @@ -357,7 +357,7 @@ class TestPlan { std::shared_ptr<minifi::core::ContentRepository> content_repo_; std::shared_ptr<minifi::core::Repository> flow_repo_; - std::shared_ptr<minifi::core::Repository> prov_repo_; + std::shared_ptr<minifi::provenance::ProvenanceRepository> prov_repo_; std::shared_ptr<minifi::core::controller::ControllerServiceProvider> controller_services_provider_; diff --git a/libminifi/test/libtest/unit/TestControllerWithFlow.cpp b/libminifi/test/libtest/unit/TestControllerWithFlow.cpp index 26f5408b4..4d8b6988a 100644 --- a/libminifi/test/libtest/unit/TestControllerWithFlow.cpp +++ b/libminifi/test/libtest/unit/TestControllerWithFlow.cpp @@ -55,7 +55,7 @@ TestControllerWithFlow::TestControllerWithFlow(const char* yamlConfigContent, bo } void TestControllerWithFlow::setupFlow() { - std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestThreadedRepository>(); + auto prov_repo = std::make_shared<TestThreadedRepository>(); std::shared_ptr<core::Repository> ff_repo = std::make_shared<TestFlowRepository>(); std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>(); diff --git a/libminifi/test/persistence-tests/PersistenceTests.cpp b/libminifi/test/persistence-tests/PersistenceTests.cpp index 11538a8f0..70642304d 100644 --- a/libminifi/test/persistence-tests/PersistenceTests.cpp +++ b/libminifi/test/persistence-tests/PersistenceTests.cpp @@ -60,7 +60,7 @@ class TestProcessor : public minifi::core::ProcessorImpl { }; struct TestFlow{ - TestFlow(const std::shared_ptr<core::Repository>& ff_repository, const std::shared_ptr<core::ContentRepository>& content_repo, const std::shared_ptr<core::Repository>& prov_repo, + TestFlow(const std::shared_ptr<core::Repository>& ff_repository, const std::shared_ptr<core::ContentRepository>& content_repo, const std::shared_ptr<minifi::provenance::ProvenanceRepository>& prov_repo, const std::function<std::unique_ptr<core::Processor>(utils::Identifier&)>& processorGenerator, const core::Relationship& relationshipToOutput) : ff_repository(ff_repository), content_repo(content_repo), prov_repo(prov_repo) { // setup processor @@ -178,7 +178,7 @@ TEST_CASE("Processors Can Store FlowFiles", "[TestP1]") { config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string()); config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string()); - std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestThreadedRepository>(); + auto prov_repo = std::make_shared<TestThreadedRepository>(); auto ff_repository = std::make_shared<core::repository::FlowFileRepository>("flowFileRepository"); std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::FileSystemRepository>(); ff_repository->initialize(config); @@ -292,7 +292,7 @@ TEST_CASE("Persisted flowFiles are updated on modification", "[TestP1]") { config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string()); config->set(minifi::Configure::nifi_dbcontent_repository_purge_period, "0 s"); - std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestThreadedRepository>(); + auto prov_repo = std::make_shared<TestThreadedRepository>(); std::shared_ptr<core::Repository> ff_repository = std::make_shared<core::repository::FlowFileRepository>("flowFileRepository"); std::shared_ptr<core::ContentRepository> content_repo; SECTION("VolatileContentRepository") { diff --git a/libminifi/test/unit/SchedulingAgentTests.cpp b/libminifi/test/unit/SchedulingAgentTests.cpp index 0e087cf1a..8c2f0fee4 100644 --- a/libminifi/test/unit/SchedulingAgentTests.cpp +++ b/libminifi/test/unit/SchedulingAgentTests.cpp @@ -78,7 +78,7 @@ class SchedulingAgentTestFixture { } protected: - std::shared_ptr<core::Repository> test_repo_ = std::make_shared<TestThreadedRepository>(); + std::shared_ptr<provenance::ProvenanceRepository> test_repo_ = std::make_shared<TestThreadedRepository>(); std::shared_ptr<core::ContentRepository> content_repo_ = std::make_shared<core::repository::VolatileContentRepository>(); TestController test_controller_; diff --git a/minifi-api/common/include/minifi-cpp/utils/Id.h b/minifi-api/common/include/minifi-cpp/utils/Id.h index 5c435dcf9..95a675985 100644 --- a/minifi-api/common/include/minifi-cpp/utils/Id.h +++ b/minifi-api/common/include/minifi-cpp/utils/Id.h @@ -46,6 +46,9 @@ class Identifier { return !isNil(); } + Identifier& operator++(); + Identifier operator++(int); + bool operator!=(const Identifier& other) const; bool operator==(const Identifier& other) const; bool operator<(const Identifier& other) const; diff --git a/minifi-api/include/minifi-cpp/core/ProcessContext.h b/minifi-api/include/minifi-cpp/core/ProcessContext.h index 6fa663a7a..dd7b086e3 100644 --- a/minifi-api/include/minifi-cpp/core/ProcessContext.h +++ b/minifi-api/include/minifi-cpp/core/ProcessContext.h @@ -29,6 +29,7 @@ #include "minifi-cpp/core/StateStorage.h" #include "minifi-cpp/core/VariableRegistry.h" #include "minifi-cpp/core/controller/ControllerServiceHandle.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" namespace org::apache::nifi::minifi::core { @@ -79,7 +80,7 @@ class ProcessContext : public virtual core::VariableRegistry, public virtual uti virtual bool isRunning() const = 0; virtual bool isAutoTerminated(Relationship relationship) const = 0; virtual uint8_t getMaxConcurrentTasks() const = 0; - virtual std::shared_ptr<core::Repository> getProvenanceRepository() = 0; + virtual std::shared_ptr<provenance::ProvenanceRepository> getProvenanceRepository() = 0; virtual std::shared_ptr<core::ContentRepository> getContentRepository() const = 0; virtual std::shared_ptr<core::Repository> getFlowFileRepository() const = 0; diff --git a/minifi-api/include/minifi-cpp/core/Repository.h b/minifi-api/include/minifi-cpp/core/Repository.h index 33b5c3747..6158d9d51 100644 --- a/minifi-api/include/minifi-cpp/core/Repository.h +++ b/minifi-api/include/minifi-cpp/core/Repository.h @@ -54,10 +54,6 @@ class Repository : public virtual core::CoreComponent, public virtual core::Repo virtual bool Get(const std::string& /*key*/, std::string& /*value*/) = 0; - virtual std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t max_size) = 0; - - virtual bool storeElement(const std::shared_ptr<core::SerializableComponent>& element) = 0; - virtual void loadComponent(const std::shared_ptr<core::ContentRepository>& /*content_repo*/) = 0; virtual std::string getDirectory() const = 0; diff --git a/minifi-api/include/minifi-cpp/provenance/Provenance.h b/minifi-api/include/minifi-cpp/provenance/Provenance.h index ef396958a..4cf5f1924 100644 --- a/minifi-api/include/minifi-cpp/provenance/Provenance.h +++ b/minifi-api/include/minifi-cpp/provenance/Provenance.h @@ -148,6 +148,8 @@ class ProvenanceEventRecord : public virtual core::SerializableComponent { ~ProvenanceEventRecord() override = default; + virtual std::optional<uint64_t> getEventOrdinal() const = 0; + virtual void setEventOrdinal(uint64_t value) = 0; virtual utils::Identifier getEventId() const = 0; virtual void setEventId(const utils::Identifier &id) = 0; virtual std::map<std::string, std::string> getAttributes() const = 0; @@ -192,9 +194,8 @@ class ProvenanceReporter { public: virtual ~ProvenanceReporter() = default; - virtual std::set<std::shared_ptr<ProvenanceEventRecord>> getEvents() const = 0; + virtual std::vector<std::shared_ptr<ProvenanceEventRecord>> getEvents() const = 0; virtual void add(const std::shared_ptr<ProvenanceEventRecord> &event) = 0; - virtual void remove(const std::shared_ptr<ProvenanceEventRecord> &event) = 0; virtual void clear() = 0; virtual void commit() = 0; diff --git a/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h b/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h new file mode 100644 index 000000000..cfebebe35 --- /dev/null +++ b/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h @@ -0,0 +1,44 @@ +/** + * + * 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. + */ + +#pragma once + +#include "minifi-cpp/core/Repository.h" +#include "utils/Id.h" +#include "minifi-cpp/provenance/Provenance.h" + +namespace org::apache::nifi::minifi::provenance { + +class ProvenanceRepository : public virtual core::Repository { + public: + class Cursor { + public: + [[nodiscard]] + virtual std::string toString() const = 0; + virtual ~Cursor() = default; + }; + + virtual std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<ProvenanceEventRecord>>& events) = 0; + + // if @cursor_str is nullopt it creates a cursor to the beginning if the underlying repository supports it + virtual std::unique_ptr<Cursor> cursorFromString(std::optional<std::string> cursor_str) = 0; + + virtual std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t max_size, Cursor* cursor) = 0; +}; + +} // namespace org::apache::nifi::minifi::provenance diff --git a/minifi_main/MiNiFiMain.cpp b/minifi_main/MiNiFiMain.cpp index fcc2ba9d6..ea7e44fdc 100644 --- a/minifi_main/MiNiFiMain.cpp +++ b/minifi_main/MiNiFiMain.cpp @@ -343,7 +343,7 @@ int main(int argc, char **argv) { configure->get(minifi::Configure::nifi_provenance_repository_class_name, prov_repo_class); // Create repos for flow record and provenance - std::shared_ptr prov_repo = core::createRepository(prov_repo_class, "provenance"); + std::shared_ptr prov_repo = utils::dynamic_unique_cast<minifi::provenance::ProvenanceRepository>(core::createRepository(prov_repo_class, "provenance")); if (!prov_repo || !prov_repo->initialize(configure)) { logger->log_error("Provenance repository failed to initialize, exiting..");
