From e310f77a37cd3afb16bed08e1ff03c43394156bc Mon Sep 17 00:00:00 2001 From: Adam Debreceni Date: Mon, 6 Jul 2026 11:13:01 +0200 Subject: [PATCH 1/3] 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 -- .../rocksdb-repos/ProvenanceRepository.cpp | 109 --------- .../RocksDbProvenanceRepository.cpp | 207 ++++++++++++++++++ ...sitory.h => RocksDbProvenanceRepository.h} | 30 ++- .../tests/DBProvenanceRepositoryTests.cpp | 142 +++++++++++- .../rocksdb-repos/tests/ProvenanceTests.cpp | 26 ++- extensions/rocksdb-repos/tests/RepoTests.cpp | 12 +- .../tests/integration/VerifyInvokeHTTP.h | 2 +- .../tests/unit/ProcessorTests.cpp | 7 +- libminifi/include/CronDrivenSchedulingAgent.h | 2 +- .../include/EventDrivenSchedulingAgent.h | 2 +- libminifi/include/FlowController.h | 6 +- libminifi/include/SchedulingAgent.h | 4 +- libminifi/include/ThreadedSchedulingAgent.h | 2 +- .../include/TimerDrivenSchedulingAgent.h | 2 +- libminifi/include/core/ProcessContextImpl.h | 8 +- .../SiteToSiteProvenanceReportingTask.h | 2 +- .../core/repository/NoOpThreadedRepository.h | 18 +- .../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 | 53 ++++- libminifi/src/provenance/Provenance.cpp | 12 +- libminifi/test/flow-tests/SessionTests.cpp | 2 +- .../test/integration/C2PauseResumeTest.cpp | 2 +- .../ControllerServiceIntegrationTests.cpp | 2 +- .../libtest/integration/IntegrationBase.cpp | 2 +- .../test/libtest/unit/ProvenanceTestHelper.h | 27 ++- libminifi/test/libtest/unit/TestBase.cpp | 4 +- libminifi/test/libtest/unit/TestBase.h | 6 +- .../libtest/unit/TestControllerWithFlow.cpp | 2 +- .../persistence-tests/PersistenceTests.cpp | 6 +- libminifi/test/unit/SchedulingAgentTests.cpp | 2 +- .../common/include/minifi-cpp/utils/Id.h | 3 + .../include/minifi-cpp/core/ProcessContext.h | 3 +- .../include/minifi-cpp/core/Repository.h | 4 - .../minifi-cpp/provenance/Provenance.h | 5 +- .../provenance/ProvenanceRepository.h | 44 ++++ minifi_main/MiNiFiMain.cpp | 2 +- 44 files changed, 663 insertions(+), 232 deletions(-) create mode 100644 cmake/FetchSQLite.cmake delete mode 100644 extensions/rocksdb-repos/ProvenanceRepository.cpp create mode 100644 extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp rename extensions/rocksdb-repos/{ProvenanceRepository.h => RocksDbProvenanceRepository.h} (69%) create mode 100644 minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h diff --git a/cmake/FetchSQLite.cmake b/cmake/FetchSQLite.cmake new file mode 100644 index 0000000000..65a9fb62ff --- /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 d401d0a5e0..09bdee3eb3 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 7a817ae2ff..1615a20b82 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> getElements(size_t /*max_size*/) override { - return {}; - } - - bool storeElement(const std::shared_ptr& element) override; - void loadComponent(const std::shared_ptr& /*content_repo*/) override { } diff --git a/core-framework/src/core/Repository.cpp b/core-framework/src/core/Repository.cpp index 722e82e7a6..7004e05c0d 100644 --- a/core-framework/src/core/Repository.cpp +++ b/core-framework/src/core/Repository.cpp @@ -27,20 +27,4 @@ bool RepositoryImpl::Delete(std::vector& element) { - if (!element) { - return false; - } - - org::apache::nifi::minifi::io::BufferStream stream; - - element->serialize(stream); - - if (!Put(element->getUUIDStr(), reinterpret_cast(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 2eb186067f..0000000000 --- a/extensions/rocksdb-repos/ProvenanceRepository.cpp +++ /dev/null @@ -1,109 +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 - -#include "core/Resource.h" -#include "minifi-cpp/Exception.h" - -namespace org::apache::nifi::minifi::provenance { - -bool ProvenanceRepository::initialize(const std::shared_ptr &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(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(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& 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(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(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> ProvenanceRepository::getElements(size_t max_size) { - if (max_size == 0) { - return {}; - } - auto opendb = db_->open(); - if (!opendb) { - throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to open provenance repository"); - } - std::vector> records; - rocksdb::ReadOptions options; - options.verify_checksums = verify_checksums_in_rocksdb_reads_; - std::unique_ptr 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 0000000000..f3397ea626 --- /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 + +#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 &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(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(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& 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(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(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 RocksDbProvenanceRepository::cursorFromString(std::optional cursor_str) { + if (cursor_str.has_value()) { + return std::make_unique(cursor_str.value()); + } + return std::make_unique(""); +} + +std::expected>, std::string> RocksDbProvenanceRepository::getEvents(size_t max_size, Cursor* cursor) { + auto* event_cursor = dynamic_cast(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> records; + rocksdb::ReadOptions options; + options.verify_checksums = verify_checksums_in_rocksdb_reads_; + std::unique_ptr 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 RocksDbProvenanceRepository::appendEvents(const std::vector>& events) { + std::vector>> 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()); + 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 950ecd31cf..d841fecd89 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(), - directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory::getLogger()) { + : RocksDbRepository(repo_name.length() > 0 ? repo_name : core::className(), + directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory::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> 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 cursorFromString(std::optional cursor_str) override; + + std::expected>, std::string> getEvents(size_t max_size, Cursor* cursor) override; + + std::expected appendEvents(const std::vector>& events) override; private: // Run function for the thread void run() override {}; + + std::unique_ptr 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 8f6677c354..154a9961d8 100644 --- a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp +++ b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp @@ -21,7 +21,7 @@ #include #include -#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 serializeEvent(minifi::provenance::ProvenanceEventRecord& event) { + minifi::io::BufferStream stream; + event.serialize(stream); + return stream.moveBuffer(); +} + +template +void appendAll(std::vector& sink, const std::vector& 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(); 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(); 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(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb.initialize(configuration)); + + std::vector> 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> 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(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb.initialize(configuration)); + + std::vector> 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> 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("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + auto configuration = std::make_shared(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb->initialize(configuration)); + + std::vector> events; + for (size_t i = 0; i < 4; ++i) { + events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + } + + REQUIRE(provdb->appendEvents(events)); + + provdb = std::make_unique("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + REQUIRE(provdb->initialize(configuration)); + + std::vector> 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> 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 4047914488..3677743bb2 100644 --- a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp +++ b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp @@ -50,10 +50,10 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv record1->setDetails(smileyface); auto sample = 65555ms; - std::shared_ptr testRepository = std::make_shared(); + auto testRepository = std::make_shared(); record1->setEventDuration(sample); - testRepository->storeElement(record1); + testRepository->appendEvents({record1}); auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); @@ -76,10 +76,10 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") { record1->addChildFlowFile(*ffr1); auto sample = 65555ms; - std::shared_ptr testRepository = std::make_shared(); + auto testRepository = std::make_shared(); record1->setEventDuration(sample); - testRepository->storeElement(record1); + testRepository->appendEvents({record1}); auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); @@ -100,11 +100,14 @@ TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::Pro auto sample = 65555ms; - std::shared_ptr testRepository = std::make_shared(); + auto testRepository = std::make_shared(); 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); @@ -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 testRepository = std::make_shared(); + auto testRepository = std::make_shared(); 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); @@ -152,11 +158,11 @@ TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::Provena auto sample = 65555ms; - std::shared_ptr testRepository = core::createRepository("nooprepository"); + std::shared_ptr testRepository = utils::dynamic_unique_cast(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 f3a8fd76f8..771c8f2a87 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 prov_repo = std::make_shared(); + auto prov_repo = std::make_shared(); auto ff_repository = std::make_shared("flowFileRepository"); std::shared_ptr content_repo = std::make_shared(); 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("ff", dir.string(), 0ms, 0, 1ms); + SECTION("RocksDbProvenanceRepository") { + repository = std::make_shared("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("ff", dir.string(), 0ms, 0, 1ms); } - SECTION("ProvenanceRepository") { - repository = std::make_shared("ff", dir.string(), 0ms, 0, 1ms); + SECTION("RocksDbProvenanceRepository") { + repository = std::make_shared("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 0f6c74a032..499e5e41b3 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 test_repo = std::make_shared(); + auto test_repo = std::make_shared(); std::shared_ptr test_flow_repo = std::make_shared(); 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 4ce6c206e6..10c2533694 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 &, const std::shared_ptr&)> verifyReporter = [&](const std::shared_ptr &context, const std::shared_ptr &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 d08ac5a4f0..46b0d68c00 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 controller_service_provider, - std::shared_ptr repo, + std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr configuration, diff --git a/libminifi/include/EventDrivenSchedulingAgent.h b/libminifi/include/EventDrivenSchedulingAgent.h index 060fff5887..d5ccd54b2e 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 controller_service_provider, std::shared_ptr repo, + EventDrivenSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr 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 00ff3f756d..f3f654c1a2 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 provenance_repo, std::shared_ptr flow_file_repo, + FlowController(std::shared_ptr provenance_repo, std::shared_ptr flow_file_repo, std::shared_ptr configure, std::shared_ptr flow_configuration, std::shared_ptr content_repo, std::unique_ptr metrics_publisher_store = nullptr, std::shared_ptr filesystem = std::make_shared(), std::function request_restart = []{}, @@ -74,7 +74,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi ~FlowController() override; - virtual std::shared_ptr getProvenanceRepository() { + virtual std::shared_ptr 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 configuration_; - std::shared_ptr provenance_repo_; + std::shared_ptr provenance_repo_; std::shared_ptr flow_file_repo_; std::shared_ptr content_repo_; std::shared_ptr flow_configuration_; diff --git a/libminifi/include/SchedulingAgent.h b/libminifi/include/SchedulingAgent.h index d5cab7af55..03e5111bda 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 controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, + SchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr configuration, utils::ThreadPool& thread_pool) : admin_yield_duration_(), bored_yield_duration_(0), @@ -122,7 +122,7 @@ class SchedulingAgent { std::shared_ptr configure_; - std::shared_ptr repo_; + std::shared_ptr repo_; std::shared_ptr flow_repo_; diff --git a/libminifi/include/ThreadedSchedulingAgent.h b/libminifi/include/ThreadedSchedulingAgent.h index 7a3ad2299e..2668bab0d4 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 controller_service_provider, std::shared_ptr repo, + ThreadedSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr 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 99073ce997..055a81eb78 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 controller_service_provider, std::shared_ptr repo, + TimerDrivenSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr 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 7bbf3ccf5f..a03aa6d97e 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& state_storage, - const std::shared_ptr& repo, const std::shared_ptr& flow_repo, + const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& content_repo = repository::createFileSystemRepository()); ProcessContextImpl(Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, - const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& configuration, + const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& configuration, const std::shared_ptr& 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 getProvenanceRepository() override { return repo_; } + std::shared_ptr getProvenanceRepository() override { return repo_; } /** * Returns a reference to the content repository for the running instance. @@ -199,7 +199,7 @@ class ProcessContextImpl : public core::VariableRegistryImpl, public virtual Pro std::shared_ptr logger_; controller::ControllerServiceProvider* controller_service_provider_; std::shared_ptr state_storage_; - std::shared_ptr repo_; + std::shared_ptr repo_; std::shared_ptr flow_repo_; std::shared_ptr content_repo_; Processor& processor_; diff --git a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h index 5bf817347f..7a1931c4db 100644 --- a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h +++ b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h @@ -44,7 +44,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> &records); // NOLINT + static std::string getJsonReport(core::ProcessContext& context, core::ProcessSession& session, const std::vector> &records); // NOLINT void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& session_factory) override; 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 df23a1b32e..06bc01f37d 100644 --- a/libminifi/include/core/repository/NoOpThreadedRepository.h +++ b/libminifi/include/core/repository/NoOpThreadedRepository.h @@ -21,10 +21,11 @@ #include #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 appendEvents(const std::vector>& /*events*/) override { + return {}; + } + + std::unique_ptr cursorFromString(std::optional /*cursor_str*/) override { + return nullptr; + } + + std::expected>, 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 da5df86fc9..278f0b8f15 100644 --- a/libminifi/include/core/repository/VolatileProvenanceRepository.h +++ b/libminifi/include/core/repository/VolatileProvenanceRepository.h @@ -21,10 +21,11 @@ #include #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) override { + if (!VolatileRepository::initialize(configure)) { + return false; + } + next_event_id_ = utils::IdGenerator::getIdGenerator()->generate(); + return true; + } + + std::expected appendEvents(const std::vector>& events) override { + std::vector>> 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()); + event->serialize(*data.back().second); + } + MultiPut(data); + + return {}; + } + + std::unique_ptr cursorFromString(std::optional /*cursor_str*/) override { + return nullptr; + } + + std::expected>, 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 5539dc5c2a..91d4b91dd0 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 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 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 repo, utils::Identifier component_id, std::string component_type) + ProvenanceReporterImpl(std::shared_ptr repo, utils::Identifier component_id, std::string component_type) : component_id_(component_id), component_type_(std::move(component_type)), logger_(core::logging::LoggerFactory::getLogger()), @@ -266,16 +277,12 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter { clear(); } - std::set> getEvents() const override { + std::vector> getEvents() const override { return events_; } void add(const std::shared_ptr &event) override { - events_.insert(event); - } - - void remove(const std::shared_ptr &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 logger_; - std::set> events_; - std::shared_ptr repo_; + std::vector> events_; + std::shared_ptr repo_; }; } // namespace org::apache::nifi::minifi::provenance diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp index 916299805b..107c44b37f 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 provenance_repo, std::shared_ptr flow_file_repo, +FlowController::FlowController(std::shared_ptr provenance_repo, std::shared_ptr flow_file_repo, std::shared_ptr configure, std::shared_ptr flow_configuration, std::shared_ptr content_repo, std::unique_ptr metrics_publisher_store, std::shared_ptr filesystem, std::function request_restart, diff --git a/libminifi/src/core/ProcessContextImpl.cpp b/libminifi/src/core/ProcessContextImpl.cpp index c732f32bca..4686e4bc74 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& state_storage, const std::shared_ptr& repo, + Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& content_repo) : VariableRegistryImpl(static_cast>(minifi::Configure::create())), logger_(logging::LoggerFactory::getLogger()), @@ -55,7 +55,7 @@ ProcessContextImpl::ProcessContextImpl( info_(std::make_unique(processor)) {} ProcessContextImpl::ProcessContextImpl( - Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, const std::shared_ptr& repo, + Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& configuration, const std::shared_ptr& content_repo) : VariableRegistryImpl(configuration), diff --git a/libminifi/src/core/flow/StructuredConfiguration.cpp b/libminifi/src/core/flow/StructuredConfiguration.cpp index 04c1231418..7e3c242ee3 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 363c81f800..15fd21c955 100644 --- a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp +++ b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp @@ -88,7 +88,7 @@ void appendJsonStr(const utils::SmallString& value, rapidjson::Value& parent, parent.PushBack(valueVal, alloc); } -std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, const std::vector> &records) { +std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, const std::vector> &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); @@ -156,11 +160,36 @@ void SiteToSiteProvenanceReportingTask::onSchedule(core::ProcessContext& context } void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, core::ProcessSession& session) { - std::shared_ptr repo = context.getProvenanceRepository(); + std::shared_ptr 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 cursor_str; + { + std::unordered_map 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> 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()); + } if (records.empty()) { logger_->log_debug("No new provenance records"); return; @@ -188,8 +217,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 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> 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 b3b4a83314..0ca80643d5 100644 --- a/libminifi/src/provenance/Provenance.cpp +++ b/libminifi/src/provenance/Provenance.cpp @@ -454,15 +454,7 @@ void ProvenanceReporterImpl::commit() { return; } - std::vector>> flowData; - - for (auto& event : events_) { - auto stramptr = std::make_unique(); - 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 533e3bf567..876a22375f 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 prov_repo = core::createRepository("nooprepository"); + std::shared_ptr prov_repo = utils::dynamic_unique_cast(core::createRepository("nooprepository")); std::shared_ptr ff_repository = std::make_shared("flowFileRepository"); std::shared_ptr content_repo; SECTION("VolatileContentRepository") { diff --git a/libminifi/test/integration/C2PauseResumeTest.cpp b/libminifi/test/integration/C2PauseResumeTest.cpp index b9e6efe153..39afd2bef2 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 test_repo = std::make_shared(); + auto test_repo = std::make_shared(); std::shared_ptr test_flow_repo = std::make_shared(); std::shared_ptr configuration = std::make_shared(); 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 d5586ca735..99f80b4015 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 configuration = std::make_shared(); - std::shared_ptr test_repo = std::make_shared(); + auto test_repo = std::make_shared(); std::shared_ptr test_flow_repo = std::make_shared(); 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 8a767546a6..b70ffa0667 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 test_repo = std::make_shared(); + auto test_repo = std::make_shared(); std::shared_ptr test_flow_repo = std::make_shared(); if (flow_config_path_.config_path) { diff --git a/libminifi/test/libtest/unit/ProvenanceTestHelper.h b/libminifi/test/libtest/unit/ProvenanceTestHelper.h index 03d6e0c328..1901c707a9 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 -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> getElements(size_t max_size) override { + std::expected>, 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> store; + std::vector> store; std::lock_guard 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 appendEvents(const std::vector>& events) override { + std::vector>> data; + data.reserve(events.size()); + for (auto& event : events) { + data.emplace_back(event->getUUIDStr(), std::make_unique()); + event->serialize(*data.back().second); + } + if (!MultiPut(data)) { + return std::unexpected{"Failed to store provenance events"}; + } + + return {}; + } + + std::unique_ptr cursorFromString(std::optional /*cursor_str*/) override { + return nullptr; + } + protected: mutable std::mutex repository_results_mutex_; std::map repository_results_; diff --git a/libminifi/test/libtest/unit/TestBase.cpp b/libminifi/test/libtest/unit/TestBase.cpp index 23a967ae9a..c5477e67ce 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 content_repo, std::shared_ptr flow_repo, std::shared_ptr prov_repo, +TestPlan::TestPlan(std::shared_ptr content_repo, std::shared_ptr flow_repo, std::shared_ptr prov_repo, std::shared_ptr flow_version, std::shared_ptr configuration, const char* state_dir) : configuration_(std::move(configuration)), content_repo_(std::move(content_repo)), @@ -637,7 +637,7 @@ std::shared_ptr TestPlan::getFlowFileProducedByCurrentPr return nullptr; } -std::set> TestPlan::getProvenanceRecords() { +std::vector> 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 9e55571944..b2d20cde88 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 content_repo, std::shared_ptr flow_repo, std::shared_ptr prov_repo, + explicit TestPlan(std::shared_ptr content_repo, std::shared_ptr flow_repo, std::shared_ptr prov_repo, std::shared_ptr flow_version, std::shared_ptr configuration, const char* state_dir); virtual ~TestPlan(); @@ -305,7 +305,7 @@ class TestPlan { bool runCurrentProcessor(); bool runCurrentProcessorUntilFlowfileIsProduced(std::chrono::milliseconds wait_duration); - std::set> getProvenanceRecords(); + std::vector> getProvenanceRecords(); std::shared_ptr getCurrentFlowFile(); std::vector getProcessorOutboundConnections(minifi::core::Processor* processor); @@ -357,7 +357,7 @@ class TestPlan { std::shared_ptr content_repo_; std::shared_ptr flow_repo_; - std::shared_ptr prov_repo_; + std::shared_ptr prov_repo_; std::shared_ptr controller_services_provider_; diff --git a/libminifi/test/libtest/unit/TestControllerWithFlow.cpp b/libminifi/test/libtest/unit/TestControllerWithFlow.cpp index 26f5408b43..4d8b6988a9 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 prov_repo = std::make_shared(); + auto prov_repo = std::make_shared(); std::shared_ptr ff_repo = std::make_shared(); std::shared_ptr content_repo = std::make_shared(); diff --git a/libminifi/test/persistence-tests/PersistenceTests.cpp b/libminifi/test/persistence-tests/PersistenceTests.cpp index 11538a8f09..70642304d8 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& ff_repository, const std::shared_ptr& content_repo, const std::shared_ptr& prov_repo, + TestFlow(const std::shared_ptr& ff_repository, const std::shared_ptr& content_repo, const std::shared_ptr& prov_repo, const std::function(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 prov_repo = std::make_shared(); + auto prov_repo = std::make_shared(); auto ff_repository = std::make_shared("flowFileRepository"); std::shared_ptr content_repo = std::make_shared(); 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 prov_repo = std::make_shared(); + auto prov_repo = std::make_shared(); std::shared_ptr ff_repository = std::make_shared("flowFileRepository"); std::shared_ptr content_repo; SECTION("VolatileContentRepository") { diff --git a/libminifi/test/unit/SchedulingAgentTests.cpp b/libminifi/test/unit/SchedulingAgentTests.cpp index 0e087cf1a2..8c2f0fee4d 100644 --- a/libminifi/test/unit/SchedulingAgentTests.cpp +++ b/libminifi/test/unit/SchedulingAgentTests.cpp @@ -78,7 +78,7 @@ class SchedulingAgentTestFixture { } protected: - std::shared_ptr test_repo_ = std::make_shared(); + std::shared_ptr test_repo_ = std::make_shared(); std::shared_ptr content_repo_ = std::make_shared(); 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 5c435dcf92..95a6759850 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 7cecd92f1b..a2dafc45c5 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 getProvenanceRepository() = 0; + virtual std::shared_ptr getProvenanceRepository() = 0; virtual std::shared_ptr getContentRepository() const = 0; virtual std::shared_ptr getFlowFileRepository() const = 0; diff --git a/minifi-api/include/minifi-cpp/core/Repository.h b/minifi-api/include/minifi-cpp/core/Repository.h index 33b5c3747a..6158d9d510 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> getElements(size_t max_size) = 0; - - virtual bool storeElement(const std::shared_ptr& element) = 0; - virtual void loadComponent(const std::shared_ptr& /*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 ef396958a0..4cf5f19240 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 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 getAttributes() const = 0; @@ -192,9 +194,8 @@ class ProvenanceReporter { public: virtual ~ProvenanceReporter() = default; - virtual std::set> getEvents() const = 0; + virtual std::vector> getEvents() const = 0; virtual void add(const std::shared_ptr &event) = 0; - virtual void remove(const std::shared_ptr &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 0000000000..cfebebe353 --- /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 appendEvents(const std::vector>& events) = 0; + + // if @cursor_str is nullopt it creates a cursor to the beginning if the underlying repository supports it + virtual std::unique_ptr cursorFromString(std::optional cursor_str) = 0; + + virtual std::expected>, 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 fcc2ba9d6a..ea7e44fdcf 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(core::createRepository(prov_repo_class, "provenance")); if (!prov_repo || !prov_repo->initialize(configure)) { logger->log_error("Provenance repository failed to initialize, exiting.."); From acfd34072826dd3e9025466b403d1823fc734b5b Mon Sep 17 00:00:00 2001 From: Adam Debreceni Date: Mon, 6 Jul 2026 11:31:30 +0200 Subject: [PATCH 2/3] MINIFICPP-2845 - Remove extra file --- cmake/FetchSQLite.cmake | 29 ----------------------------- 1 file changed, 29 deletions(-) delete mode 100644 cmake/FetchSQLite.cmake diff --git a/cmake/FetchSQLite.cmake b/cmake/FetchSQLite.cmake deleted file mode 100644 index 65a9fb62ff..0000000000 --- a/cmake/FetchSQLite.cmake +++ /dev/null @@ -1,29 +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(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}) From af3f98ac40e37b8a5754b26fd4bfd4c274d494b6 Mon Sep 17 00:00:00 2001 From: Adam Debreceni <64783590+adamdebreceni@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:41:07 +0200 Subject: [PATCH 3/3] Update core-framework/common/src/utils/Id.cpp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Márton Szász --- core-framework/common/src/utils/Id.cpp | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/core-framework/common/src/utils/Id.cpp b/core-framework/common/src/utils/Id.cpp index 09bdee3eb3..810a10d68e 100644 --- a/core-framework/common/src/utils/Id.cpp +++ b/core-framework/common/src/utils/Id.cpp @@ -44,7 +44,13 @@ bool Identifier::isNil() const { } Identifier& Identifier::operator++() { - for (int byte_idx = 15; byte_idx >= 0 && ++data_[byte_idx] == 0; --byte_idx) {} + // increment with carry + for (int byte_idx = 15; byte_idx >= 0; --byte_idx) { + ++data_[byte_idx]; + if (data_[byte_idx] != 0) { + break; + } + } return *this; }