This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 746c4cfeb MINIFICPP-2845 - Monotonic provenance event identifiers,
iterations support, refactor (#2209)
746c4cfeb is described below
commit 746c4cfebe01fb9d714fb7f58bf92d2771b78d31
Author: Adam Debreceni <[email protected]>
AuthorDate: Mon Nov 9 17:12:39 2026 +0100
MINIFICPP-2845 - Monotonic provenance event identifiers, iterations
support, refactor (#2209)
---
core-framework/include/core/Repository.h | 8 +-
core-framework/src/core/Repository.cpp | 16 --
extensions/rocksdb-repos/ProvenanceRepository.cpp | 109 ---------
.../rocksdb-repos/RocksDbProvenanceRepository.cpp | 244 ++++++++++++++++++++
...eRepository.h => RocksDbProvenanceRepository.h} | 38 ++--
.../rocksdb-repos/database/RocksDbInstance.cpp | 6 +
.../tests/DBProvenanceRepositoryTests.cpp | 253 ++++++++++++++++++++-
extensions/rocksdb-repos/tests/ProvenanceTests.cpp | 136 ++++++++---
extensions/rocksdb-repos/tests/RepoTests.cpp | 12 +-
.../tests/integration/VerifyInvokeHTTP.h | 2 +-
.../tests/unit/ProcessorTests.cpp | 9 +-
libminifi/include/CronDrivenSchedulingAgent.h | 4 +-
libminifi/include/EventDrivenSchedulingAgent.h | 4 +-
libminifi/include/FlowController.h | 6 +-
libminifi/include/SchedulingAgent.h | 8 +-
libminifi/include/ThreadedSchedulingAgent.h | 4 +-
libminifi/include/TimerDrivenSchedulingAgent.h | 4 +-
libminifi/include/core/ProcessContextImpl.h | 8 +-
.../reporting/SiteToSiteProvenanceReportingTask.h | 2 +-
.../include/core/repository/AtomicRepoEntries.h | 18 ++
.../core/repository/NoOpThreadedRepository.h | 18 +-
.../core/repository/VolatileProvenanceRepository.h | 34 ++-
.../include/core/repository/VolatileRepository.h | 4 +-
libminifi/include/provenance/Provenance.h | 32 +--
libminifi/src/FlowController.cpp | 2 +-
libminifi/src/ThreadedSchedulingAgent.cpp | 4 +-
libminifi/src/core/ProcessContextImpl.cpp | 12 +-
libminifi/src/core/ProcessSession.cpp | 2 +-
.../src/core/flow/StructuredConfiguration.cpp | 4 +
.../SiteToSiteProvenanceReportingTask.cpp | 95 ++++++--
.../repository/VolatileProvenanceRepository.cpp | 112 +++++++++
.../src/core/repository/VolatileRepository.cpp | 2 +-
libminifi/src/provenance/Provenance.cpp | 75 ++----
libminifi/src/sitetosite/SiteToSiteClient.cpp | 2 +-
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 | 33 ++-
libminifi/test/libtest/unit/TestBase.cpp | 7 +-
libminifi/test/libtest/unit/TestBase.h | 9 +-
.../test/libtest/unit/TestControllerWithFlow.cpp | 2 +-
.../test/persistence-tests/PersistenceTests.cpp | 9 +-
libminifi/test/unit/LogMetricsPublisherTests.cpp | 11 +-
libminifi/test/unit/SchedulingAgentTests.cpp | 2 +-
.../unit/VolatileProvenanceRepositoryTests.cpp | 169 ++++++++++++++
.../include/minifi-cpp/core/ProcessContext.h | 3 +-
minifi-api/include/minifi-cpp/core/Repository.h | 8 +-
.../include/minifi-cpp/provenance/Provenance.h | 8 +-
.../minifi-cpp/provenance/ProvenanceRepository.h | 44 ++++
minifi_main/MiNiFiMain.cpp | 2 +-
51 files changed, 1256 insertions(+), 348 deletions(-)
diff --git a/core-framework/include/core/Repository.h
b/core-framework/include/core/Repository.h
index 7a817ae2f..37e289489 100644
--- a/core-framework/include/core/Repository.h
+++ b/core-framework/include/core/Repository.h
@@ -80,7 +80,7 @@ class RepositoryImpl : public core::CoreComponentImpl, public
core::RepositoryMe
return true;
}
- bool MultiPut(const std::vector<std::pair<std::string,
std::unique_ptr<io::BufferStream>>>& /*data*/) override {
+ bool MultiPut(const EntryStreams& /*data*/) override {
return true;
}
@@ -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 2eb186067..000000000
--- 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 <string>
-
-#include "core/Resource.h"
-#include "minifi-cpp/Exception.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) {
- throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to
open provenance repository");
- }
- 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..12d46e1bb
--- /dev/null
+++ b/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp
@@ -0,0 +1,244 @@
+/**
+ * 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 <optional>
+#include <string>
+
+#include "core/Resource.h"
+#include "minifi-cpp/utils/Literals.h"
+
+namespace org::apache::nifi::minifi::provenance {
+
+namespace {
+// Events are keyed by their ordinal, and the keys are iterated in bytewise
order, so the
+// encoding must be fixed width, otherwise key order would not match ordinal
order
+// (e.g. "10" < "2") and an iteration could skip events. 20 digits fit any
uint64_t.
+std::string eventOrdinalToKey(uint64_t event_ordinal) {
+ return fmt::format("{:020}", event_ordinal);
+}
+
+// also accepts keys written before the fixed width encoding was introduced
+std::optional<uint64_t> keyToEventOrdinal(std::string_view key) {
+ if (auto event_ordinal = parsing::parseIntegral<uint64_t>(key)) {
+ return *event_ordinal;
+ }
+ return std::nullopt;
+}
+
+class EventCursor : public ProvenanceRepository::Cursor {
+ public:
+ explicit EventCursor(uint64_t last_event_ordinal):
last_event_ordinal_(last_event_ordinal) {}
+ EventCursor(const EventCursor&) = default;
+ EventCursor(EventCursor&&) = default;
+ EventCursor& operator=(const EventCursor&) = default;
+ EventCursor& operator=(EventCursor&&) = default;
+
+ // the ordinal of the last event observed, 0 if no event has been read yet
+ [[nodiscard]]
+ std::string toString() const override {
+ return std::to_string(last_event_ordinal_);
+ }
+ ~EventCursor() override = default;
+
+ uint64_t last_event_ordinal_;
+};
+} // namespace
+
+static const std::string_view NEXT_EVENT_KEY = "next_event_key";
+
+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) {
+ cf_opts.write_buffer_size = std::min<size_t>(16_MiB, 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_key_str;
+ if (open_state_db->Get(options, NEXT_EVENT_KEY, &next_event_key_str).ok())
{
+ next_event_key_ = std::stoull(next_event_key_str);
+ } else {
+ logger_->log_debug("Could not find '{}', resetting event key",
NEXT_EVENT_KEY);
+ next_event_key_ = 1;
+ }
+ logger_->log_trace("Using next event key: {}", next_event_key_);
+ } 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::string_view cursor_str) {
+ if (cursor_str.empty()) {
+ return std::make_unique<EventCursor>(0);
+ }
+ if (auto last_event_ordinal = keyToEventOrdinal(cursor_str)) {
+ return std::make_unique<EventCursor>(*last_event_ordinal);
+ }
+ logger_->log_warn("Could not interpret provenance cursor '{}', reading from
the first event", cursor_str);
+ return std::make_unique<EventCursor>(0);
+}
+
+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));
+ uint64_t last_event_ordinal = 0;
+ if (event_cursor) {
+ last_event_ordinal = event_cursor->last_event_ordinal_;
+ const auto last_event_key = eventOrdinalToKey(last_event_ordinal);
+ it->Seek(last_event_key);
+ if (it->Valid() && it->key() == last_event_key) {
+ it->Next();
+ }
+ } else {
+ it->SeekToFirst();
+ }
+ for (; it->Valid(); it->Next()) {
+ // the ordinal is taken from the key, not from the event, so that an event
that fails to
+ // deserialize does not stall the cursor
+ if (auto event_ordinal =
keyToEventOrdinal(std::string_view{it->key().data(), it->key().size()})) {
+ last_event_ordinal = *event_ordinal;
+ } else {
+ logger_->log_warn("Skipping provenance entry with unexpected key '{}'",
it->key().ToString());
+ continue;
+ }
+ 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->last_event_ordinal_ = last_event_ordinal;
+ }
+ return records;
+}
+
+std::expected<void, std::string>
RocksDbProvenanceRepository::appendEvents(const
std::vector<std::shared_ptr<ProvenanceEventRecord>>& events) {
+ EntryStreams data;
+ data.reserve(events.size());
+ std::lock_guard guard(next_event_key_mtx_);
+ for (auto& event : events) {
+ event->setEventOrdinal(next_event_key_++);
+ }
+ {
+ 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_KEY,
std::to_string(next_event_key_)); };
+ if (!ExecuteWithRetry(operation)) {
+ return std::unexpected{"Failed to update next provenance event key"};
+ }
+ }
+ for (auto& event : events) {
+ data.emplace_back(eventOrdinalToKey(event->getEventOrdinal()),
std::make_unique<io::BufferStream>());
+ if (!event->serialize(*data.back().second)) {
+ return std::unexpected{fmt::format("Failed to serialize provenance event
'{}'", event->getUUIDStr())};
+ }
+ }
+ if (MultiPut(data)) {
+ return {};
+ }
+
+ return std::unexpected{"Failed to append provenance events"};
+}
+
+REGISTER_RESOURCE_AS(RocksDbProvenanceRepository, InternalResource,
("RocksDbProvenanceRepository", "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 61%
rename from extensions/rocksdb-repos/ProvenanceRepository.h
rename to extensions/rocksdb-repos/RocksDbProvenanceRepository.h
index 950ecd31c..34a5872b9 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 = "",
- 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()) {
+ 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<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::string_view 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_key_mtx_;
+ uint64_t next_event_key_;
};
} // namespace org::apache::nifi::minifi::provenance
diff --git a/extensions/rocksdb-repos/database/RocksDbInstance.cpp
b/extensions/rocksdb-repos/database/RocksDbInstance.cpp
index 3c6de0c03..2aca1d8c4 100644
--- a/extensions/rocksdb-repos/database/RocksDbInstance.cpp
+++ b/extensions/rocksdb-repos/database/RocksDbInstance.cpp
@@ -180,6 +180,12 @@ std::optional<OpenRocksDb> RocksDbInstance::open(const
std::string& column) {
logger_->log_error("Couldn't query database '{}' for options: '{}'",
db_name_, latest_option_status.ToString());
return std::nullopt;
}
+ if (mode_ == RocksDbMode::ReadWrite) {
+ // The column families are taken from the persisted options, but we
create them on demand
+ // (see getOrCreateColumnFamily), so one that the options mention while
the database itself
+ // does not have it yet must not fail the open of the whole database.
+ db_options_.create_missing_column_families = true;
+ }
std::vector<rocksdb::ColumnFamilyHandle*> column_handles;
switch (mode_) {
case RocksDbMode::ReadWrite:
diff --git a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp
b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp
index 8f6677c35..8b93991ad 100644
--- a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp
+++ b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp
@@ -18,10 +18,13 @@
#include <array>
#include <chrono>
+#include <filesystem>
+#include <memory>
#include <random>
+#include <string>
#include <vector>
-#include "ProvenanceRepository.h"
+#include "RocksDbProvenanceRepository.h"
#include "unit/TestBase.h"
#include "unit/Catch.h"
@@ -61,13 +64,33 @@ 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());
+}
+
+std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>>
createEvents(size_t count) {
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>>
events;
+ events.reserve(count);
+ for (size_t i = 0; i < count; ++i) {
+ events.push_back(minifi::provenance::ProvenanceEventRecord::create());
+ }
+ return events;
+}
+
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 +110,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 +137,227 @@ 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));
+
+ auto events = createEvents(8);
+
+ REQUIRE(provdb.appendEvents(events));
+
+ auto cursor = provdb.cursorFromString("");
+ 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);
+
+ for (size_t i = 0; i < queried_events.size(); ++i) {
+ REQUIRE(queried_events.at(i)->getEventOrdinal() == i + 1);
+ 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));
+
+ auto events = createEvents(8);
+
+ REQUIRE(provdb.appendEvents(events));
+
+ auto cursor = provdb.cursorFromString("");
+ 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);
+
+ // the cursor is persisted as the ordinal of the last event read
+ REQUIRE(cursor->toString() == "3");
+
+ cursor = provdb.cursorFromString(cursor->toString());
+ REQUIRE(cursor);
+
+ appendAll(queried_events, provdb.getEvents(3, cursor.get()).value());
+ REQUIRE(queried_events.size() == 6);
+
+ for (size_t i = 0; i < queried_events.size(); ++i) {
+ REQUIRE(queried_events.at(i)->getEventOrdinal() == i + 1);
+ REQUIRE(serializeEvent(*events.at(i)) ==
serializeEvent(*queried_events.at(i)));
+ }
+}
+
+TEST_CASE("Test appendEvents assigns consecutive ordinals",
"[eventOrdinalTest]") {
+ 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));
+
+ auto first_batch = createEvents(5);
+ REQUIRE(provdb.appendEvents(first_batch));
+ for (size_t i = 0; i < first_batch.size(); ++i) {
+ REQUIRE(first_batch.at(i)->getEventOrdinal() == i + 1);
+ }
+
+ auto second_batch = createEvents(3);
+ REQUIRE(provdb.appendEvents(second_batch));
+ for (size_t i = 0; i < second_batch.size(); ++i) {
+ REQUIRE(second_batch.at(i)->getEventOrdinal() == first_batch.size() + i +
1);
+ }
+}
+
+TEST_CASE("Test querying events whose ordinals have different number of
digits", "[eventOrdinalTest]") {
+ 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));
+
+ auto events = createEvents(12);
+ REQUIRE(provdb.appendEvents(events));
+
+ auto queried_events = provdb.getEvents(12, nullptr).value();
+ REQUIRE(queried_events.size() == 12);
+
+ for (size_t i = 0; i < queried_events.size(); ++i) {
+ REQUIRE(queried_events.at(i)->getEventOrdinal() == i + 1);
+ REQUIRE(serializeEvent(*events.at(i)) ==
serializeEvent(*queried_events.at(i)));
+ }
+}
+
+TEST_CASE("Test cursor observes events appended with more digits in their
ordinal", "[eventOrdinalTest]") {
+ 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));
+
+ REQUIRE(provdb.appendEvents(createEvents(9)));
+
+ auto cursor = provdb.cursorFromString("");
+ REQUIRE(cursor);
+ REQUIRE(provdb.getEvents(9, cursor.get()).value().size() == 9);
+ REQUIRE(cursor->toString() == "9");
+
+ REQUIRE(provdb.appendEvents(createEvents(3)));
+
+ auto queried_events = provdb.getEvents(9, cursor.get()).value();
+ REQUIRE(queried_events.size() == 3);
+ for (size_t i = 0; i < queried_events.size(); ++i) {
+ REQUIRE(queried_events.at(i)->getEventOrdinal() == i + 10);
+ }
+}
+
+TEST_CASE("Test opening a database whose options mention the internal state
column but does not have it", "[eventOrdinalTest]") {
+ TestController testController;
+ auto temp_dir = testController.createTempDirectory();
+ REQUIRE(!temp_dir.empty());
+
+ auto configuration =
std::make_shared<org::apache::nifi::minifi::ConfigureImpl>();
+
configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default,
temp_dir.string());
+
+ {
+ minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo",
temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s);
+ REQUIRE(provdb.initialize(configuration));
+ REQUIRE(provdb.appendEvents(createEvents(1)));
+ }
+
+ // Leave the directory in the state an interrupted deletion produces: the
persisted options still
+ // list the internal state column, while the database itself is gone. The
column is created on
+ // demand, so this must not stop the repository from opening.
+ bool options_file_kept = false;
+ for (const auto& entry : std::filesystem::directory_iterator{temp_dir}) {
+ const auto filename = entry.path().filename().string();
+ if (filename.starts_with("CURRENT") || filename.starts_with("MANIFEST") ||
filename.ends_with(".log")) {
+ std::filesystem::remove(entry.path());
+ } else if (filename.starts_with("OPTIONS")) {
+ options_file_kept = true;
+ }
+ }
+ // the scenario is only reproduced as long as the persisted options are the
ones left behind
+ REQUIRE(options_file_kept);
+
+ minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo",
temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s);
+ REQUIRE(provdb.initialize(configuration));
+ REQUIRE(provdb.getRocksDbStats());
+ REQUIRE(provdb.appendEvents(createEvents(1)));
+}
+
+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));
+
+ auto events = createEvents(4);
+
+ 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));
+
+ auto new_events = createEvents(4);
+ appendAll(events, new_events);
+
+ REQUIRE(provdb->appendEvents(new_events));
+
+ // the counter continues where the previous instance left off
+ for (size_t i = 0; i < new_events.size(); ++i) {
+ REQUIRE(new_events.at(i)->getEventOrdinal() == i + 5);
+ }
+
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>>
queried_events = provdb->getEvents(8, nullptr).value();
+ REQUIRE(queried_events.size() == 8);
+
+ for (size_t i = 0; i < queried_events.size(); ++i) {
+ REQUIRE(queried_events.at(i)->getEventOrdinal() == i + 1);
+ 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 404791448..8b314bc96 100644
--- a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
+++ b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
@@ -16,12 +16,17 @@
* limitations under the License.
*/
+#include <cstddef>
+#include <limits>
#include <map>
#include <memory>
+#include <span>
#include <utility>
#include <string>
+#include <vector>
#include "core/Core.h"
+#include "io/BufferStream.h"
#include "core/repository/AtomicRepoEntries.h"
#include "core/repository/VolatileProvenanceRepository.h"
#include "core/RepositoryFactory.h"
@@ -40,35 +45,94 @@ TEST_CASE("Test Provenance record create",
"[Testprovenance::ProvenanceEventReco
REQUIRE(record1->getAlternateIdentifierUri().empty());
}
+TEST_CASE("Test Provenance event ordinal serialization round trip",
"[Testprovenance::ProvenanceEventRecordSerializeDeser]") {
+ auto record1 =
std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE,
+
utils::Identifier::parse("00000000-0000-0000-0000-000000000007").value(),
"componenttype");
+ record1->setDetails("details");
+
+ uint64_t event_ordinal = 0;
+ SECTION("Unset ordinal") {
+ event_ordinal = 0;
+ }
+ SECTION("Small ordinal") {
+ event_ordinal = 42;
+ }
+ SECTION("Largest ordinal") {
+ event_ordinal = std::numeric_limits<uint64_t>::max();
+ }
+ record1->setEventOrdinal(event_ordinal);
+
+ minifi::io::BufferStream stream;
+ REQUIRE(record1->serialize(stream));
+
+ auto record2 = provenance::ProvenanceEventRecord::create();
+ REQUIRE(record2->deserialize(stream));
+ REQUIRE(record2->getEventOrdinal() == event_ordinal);
+ // the ordinal is serialized last, verify that the preceding fields are
intact as well
+ REQUIRE(record2->getEventId() == record1->getEventId());
+ REQUIRE(record2->getEventType() == record1->getEventType());
+ REQUIRE(record2->getDetails() == record1->getDetails());
+}
+
+TEST_CASE("Test Provenance event serialized before the ordinal was
introduced", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") {
+ auto record1 =
std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE,
+
utils::Identifier::parse("00000000-0000-0000-0000-000000000008").value(),
"componenttype");
+ record1->setDetails("details");
+ record1->setEventOrdinal(42);
+
+ minifi::io::BufferStream stream;
+ REQUIRE(record1->serialize(stream));
+ const auto serialized_event = stream.moveBuffer();
+ REQUIRE(serialized_event.size() > sizeof(uint64_t));
+
+ // the ordinal is written last, an event stored before it was introduced
simply ends earlier
+ std::ptrdiff_t missing_bytes = 0;
+ SECTION("The ordinal is missing entirely") {
+ missing_bytes = sizeof(uint64_t);
+ }
+ SECTION("The ordinal is truncated") {
+ missing_bytes = sizeof(uint64_t) / 2;
+ }
+ const std::vector<std::byte> old_format_event{serialized_event.begin(),
serialized_event.end() - missing_bytes};
+
+ minifi::io::BufferStream old_format_stream{std::span<const
std::byte>{old_format_event}};
+ auto record2 = provenance::ProvenanceEventRecord::create();
+ REQUIRE(record2->deserialize(old_format_stream));
+ REQUIRE(record2->getEventOrdinal() == 0);
+ REQUIRE(record2->getEventId() == record1->getEventId());
+ REQUIRE(record2->getDetails() == record1->getDetails());
+}
+
TEST_CASE("Test Provenance record serialization",
"[Testprovenance::ProvenanceEventRecordSerializeDeser]") {
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();
-
std::string smileyface = ":)";
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);
- auto record2 = provenance::ProvenanceEventRecord::create();
- record2->setEventId(eventId);
- REQUIRE(record2->loadFromRepository(testRepository) == true);
+ REQUIRE(testRepository->appendEvents({record1}));
+ REQUIRE(record1->getEventOrdinal() == 1);
+
+ auto events = testRepository->getEvents(1, nullptr);
+ REQUIRE(events);
+ REQUIRE(events->size() == 1);
+ const auto& record2 = events->at(0);
REQUIRE(record2->getEventId() == record1->getEventId());
REQUIRE(record2->getComponentId() == record1->getComponentId());
REQUIRE(record2->getComponentType() == record1->getComponentType());
REQUIRE(record2->getDetails() == record1->getDetails());
REQUIRE(record2->getDetails() == smileyface);
REQUIRE(record2->getEventDuration() == sample);
+ REQUIRE(record2->getEventOrdinal() == record1->getEventOrdinal());
}
TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") {
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");
ffr1->setAttribute("tomato", "tomatoe");
@@ -76,13 +140,15 @@ 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);
- auto record2 = provenance::ProvenanceEventRecord::create();
- record2->setEventId(eventId);
- REQUIRE(record2->loadFromRepository(testRepository) == true);
+ REQUIRE(testRepository->appendEvents({record1}));
+
+ auto events = testRepository->getEvents(1, nullptr);
+ REQUIRE(events);
+ REQUIRE(events->size() == 1);
+ const auto& record2 = events->at(0);
REQUIRE(record1->getChildrenUuids().size() == 1);
REQUIRE(record2->getChildrenUuids().size() == 1);
utils::Identifier childId = record2->getChildrenUuids().at(0);
@@ -93,33 +159,34 @@ TEST_CASE("Test Provenance record serialization Volatile",
"[Testprovenance::Pro
auto record1 =
std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE,
utils::Identifier::parse("00000000-0000-0000-0000-000000000004").value(),
"componenttype");
- utils::Identifier eventId = record1->getEventId();
-
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);
- auto record2 = provenance::ProvenanceEventRecord::create();
- record2->setEventId(eventId);
- REQUIRE(record2->loadFromRepository(testRepository) == true);
+ REQUIRE(testRepository->appendEvents({record1}));
+ REQUIRE(record1->getEventOrdinal() == 1);
+
+ auto events = testRepository->getEvents(1, nullptr);
+ REQUIRE(events);
+ REQUIRE(events->size() == 1);
+ const auto& record2 = events->at(0);
REQUIRE(record2->getEventId() == record1->getEventId());
REQUIRE(record2->getComponentId() == record1->getComponentId());
REQUIRE(record2->getComponentType() == record1->getComponentType());
REQUIRE(record2->getDetails() == record1->getDetails());
REQUIRE(record2->getDetails() == smileyface);
REQUIRE(record2->getEventDuration() == sample);
+ REQUIRE(record2->getEventOrdinal() == record1->getEventOrdinal());
}
TEST_CASE("Test Flowfile record added to provenance using Volatile Repo",
"[TestFlowAndProv1]") {
auto record1 =
std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE,
utils::Identifier::parse("00000000-0000-0000-0000-000000000005").value(),
"componenttype");
- utils::Identifier eventId = record1->getEventId();
std::shared_ptr<minifi::FlowFileRecord> ffr1 =
std::make_shared<minifi::FlowFileRecordImpl>();
ffr1->setAttribute("potato", "potatoe");
ffr1->setAttribute("tomato", "tomatoe");
@@ -127,14 +194,16 @@ 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);
- auto record2 = provenance::ProvenanceEventRecord::create();
- record2->setEventId(eventId);
- REQUIRE(record2->loadFromRepository(testRepository) == true);
+ REQUIRE(testRepository->appendEvents({record1}));
+
+ auto events = testRepository->getEvents(1, nullptr);
+ REQUIRE(events);
+ REQUIRE(events->size() == 1);
+ const auto& record2 = events->at(0);
REQUIRE(record1->getChildrenUuids().size() == 1);
REQUIRE(record2->getChildrenUuids().size() == 1);
utils::Identifier childId = record2->getChildrenUuids().at(0);
@@ -145,19 +214,20 @@ TEST_CASE("Test Provenance record serialization NoOp",
"[Testprovenance::Provena
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();
-
std::string smileyface = ":)";
record1->setDetails(smileyface);
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));
- auto record2 = provenance::ProvenanceEventRecord::create();
- record2->setEventId(eventId);
- REQUIRE(record2->loadFromRepository(testRepository) == false);
+ REQUIRE(testRepository->appendEvents({record1}));
+ // the event is not stored, so it is not assigned an ordinal either
+ REQUIRE(record1->getEventOrdinal() == 0);
+
+ auto events = testRepository->getEvents(1, nullptr);
+ REQUIRE(events);
+ REQUIRE(events->empty());
}
diff --git a/extensions/rocksdb-repos/tests/RepoTests.cpp
b/extensions/rocksdb-repos/tests/RepoTests.cpp
index 79b887d18..4f0ad3ea9 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);
@@ -554,8 +554,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;
}
@@ -712,8 +712,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 4589f1ac9..503f4421b 100644
--- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp
+++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
@@ -468,13 +468,16 @@ 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);
+ REQUIRE(recordsReport->at(0)->getEventOrdinal() == 1);
+ REQUIRE(json_str.find("\"eventOrdinal\": 1") != std::string::npos);
};
testController.runSession(plan, false, verifyReporter);
diff --git a/libminifi/include/CronDrivenSchedulingAgent.h
b/libminifi/include/CronDrivenSchedulingAgent.h
index d08ac5a4f..aed7d30e5 100644
--- a/libminifi/include/CronDrivenSchedulingAgent.h
+++ b/libminifi/include/CronDrivenSchedulingAgent.h
@@ -37,12 +37,12 @@ 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>
provenance_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, std::move(repo),
std::move(flow_repo), std::move(content_repo), std::move(configuration),
thread_pool) {
+ : ThreadedSchedulingAgent(controller_service_provider,
std::move(provenance_repo), std::move(flow_repo), std::move(content_repo),
std::move(configuration), thread_pool) {
}
~CronDrivenSchedulingAgent() override = default;
diff --git a/libminifi/include/EventDrivenSchedulingAgent.h
b/libminifi/include/EventDrivenSchedulingAgent.h
index 060fff588..5da7d7e65 100644
--- a/libminifi/include/EventDrivenSchedulingAgent.h
+++ b/libminifi/include/EventDrivenSchedulingAgent.h
@@ -35,10 +35,10 @@ 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>
provenance_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) {
+ : ThreadedSchedulingAgent(controller_service_provider, provenance_repo,
flow_repo, content_repo, configuration, thread_pool) {
using namespace std::literals::chrono_literals;
time_slice_ =
configuration->get(Configure::nifi_flow_engine_event_driven_time_slice)
diff --git a/libminifi/include/FlowController.h
b/libminifi/include/FlowController.h
index 5177e2957..0b9efcbfd 100644
--- a/libminifi/include/FlowController.h
+++ b/libminifi/include/FlowController.h
@@ -67,7 +67,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>(),
@@ -76,7 +76,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_;
}
@@ -189,7 +189,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..6c8b12f15 100644
--- a/libminifi/include/SchedulingAgent.h
+++ b/libminifi/include/SchedulingAgent.h
@@ -50,8 +50,8 @@ 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,
- std::shared_ptr<core::ContentRepository> content_repo,
std::shared_ptr<Configure> configuration, utils::ThreadPool& thread_pool)
+ SchedulingAgent(const
gsl::not_null<core::controller::ControllerServiceProvider*>
controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository>
provenance_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),
configure_(configuration),
@@ -60,7 +60,7 @@ class SchedulingAgent {
controller_service_provider_(controller_service_provider),
logger_(core::logging::LoggerFactory<SchedulingAgent>::getLogger()) {
running_ = false;
- repo_ = repo;
+ provenance_repo_ = provenance_repo;
flow_repo_ = flow_repo;
alert_time_ = configuration->get(Configure::nifi_flow_engine_alert_period)
@@ -122,7 +122,7 @@ class SchedulingAgent {
std::shared_ptr<Configure> configure_;
- std::shared_ptr<core::Repository> repo_;
+ std::shared_ptr<provenance::ProvenanceRepository> provenance_repo_;
std::shared_ptr<core::Repository> flow_repo_;
diff --git a/libminifi/include/ThreadedSchedulingAgent.h
b/libminifi/include/ThreadedSchedulingAgent.h
index 7a3ad2299..5de546779 100644
--- a/libminifi/include/ThreadedSchedulingAgent.h
+++ b/libminifi/include/ThreadedSchedulingAgent.h
@@ -38,10 +38,10 @@ 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>
provenance_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) {
+ : SchedulingAgent(controller_service_provider, provenance_repo,
flow_repo, content_repo, configuration, thread_pool) {
}
~ThreadedSchedulingAgent() override = default;
diff --git a/libminifi/include/TimerDrivenSchedulingAgent.h
b/libminifi/include/TimerDrivenSchedulingAgent.h
index 99073ce99..936d60c74 100644
--- a/libminifi/include/TimerDrivenSchedulingAgent.h
+++ b/libminifi/include/TimerDrivenSchedulingAgent.h
@@ -30,10 +30,10 @@
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>
provenance_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) {
+ : ThreadedSchedulingAgent(controller_service_provider, provenance_repo,
flow_repo, content_repo, configure, thread_pool) {
}
utils::TaskRescheduleInfo run(core::Processor* processor, const
std::shared_ptr<core::ProcessContext> &processContext,
diff --git a/libminifi/include/core/ProcessContextImpl.h
b/libminifi/include/core/ProcessContextImpl.h
index 7bbf3ccf5..2115a598c 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 provenance_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<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> provenance_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 055cd867d..83c4900ca 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<std::shared_ptr<core::SerializableComponent>> &records);
+ static std::string getJsonReport(core::ProcessContext& context,
core::ProcessSession& session, const
std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> &records);
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/AtomicRepoEntries.h
b/libminifi/include/core/repository/AtomicRepoEntries.h
index 317bd0eea..4156db192 100644
--- a/libminifi/include/core/repository/AtomicRepoEntries.h
+++ b/libminifi/include/core/repository/AtomicRepoEntries.h
@@ -278,6 +278,24 @@ class AtomicEntry {
return true;
}
+ /**
+ * Calls @visitor with the key and the buffer of the contained value,
without consuming it.
+ * The visitor is called while this entry is locked, so it should not block.
+ * @return false if this entry is empty
+ */
+ template<typename Visitor>
+ bool visitValue(Visitor&& visitor) {
+ try_lock();
+ if (!has_value_) {
+ try_unlock();
+ return false;
+ }
+ const T& key = value_.getKey();
+ std::forward<Visitor>(visitor)(key, value_.getBuffer());
+ try_unlock();
+ return true;
+ }
+
void decrementOwnership() {
try_lock();
if (!has_value_) {
diff --git a/libminifi/include/core/repository/NoOpThreadedRepository.h
b/libminifi/include/core/repository/NoOpThreadedRepository.h
index df23a1b32..3b7b5354c 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::string_view /*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..5877bb892 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,35 @@ class VolatileProvenanceRepository : public
VolatileRepository {
stop();
}
+ bool initialize(const std::shared_ptr<Configure> &configure) override {
+ if (!VolatileRepository::initialize(configure)) {
+ return false;
+ }
+ next_event_key_ = 1;
+ return true;
+ }
+
+ std::expected<void, std::string> appendEvents(const
std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>& events)
override {
+ EntryStreams data;
+ data.reserve(events.size());
+ std::lock_guard guard(next_event_key_mtx_);
+ for (auto& event : events) {
+ event->setEventOrdinal(next_event_key_++);
+ data.emplace_back(std::to_string(event->getEventOrdinal()),
std::make_unique<io::BufferStream>());
+ event->serialize(*data.back().second);
+ }
+ MultiPut(data);
+
+ return {};
+ }
+
+ std::unique_ptr<ProvenanceRepository::Cursor>
cursorFromString(std::string_view cursor_str) override;
+
+ // Returns the events following the cursor in ordinal order. As this
repository can overwrite
+ // its oldest entries, a cursor guarantees that no event is returned twice,
not that no event
+ // is missed.
+
std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>,
std::string> getEvents(size_t max_size, Cursor* cursor) override;
+
private:
void run() override {
}
@@ -51,6 +81,8 @@ class VolatileProvenanceRepository : public
VolatileRepository {
}
std::thread thread_;
+ std::mutex next_event_key_mtx_;
+ uint64_t next_event_key_;
};
} // namespace org::apache::nifi::minifi::core::repository
diff --git a/libminifi/include/core/repository/VolatileRepository.h
b/libminifi/include/core/repository/VolatileRepository.h
index c552601cd..5273cd26e 100644
--- a/libminifi/include/core/repository/VolatileRepository.h
+++ b/libminifi/include/core/repository/VolatileRepository.h
@@ -65,7 +65,7 @@ class VolatileRepository : public
core::ThreadedRepositoryImpl {
* Places new objects into the volatile memory area
* @param data the key-value pairs to add to the repository
**/
- bool MultiPut(const std::vector<std::pair<std::string,
std::unique_ptr<io::BufferStream>>>& data) override;
+ bool MultiPut(const EntryStreams& data) override;
/**
* Deletes the key
@@ -106,8 +106,6 @@ class VolatileRepository : public
core::ThreadedRepositoryImpl {
std::atomic<uint32_t> current_index_;
std::mutex purge_mutex_;
std::vector<std::string> purge_list_;
-
- private:
std::shared_ptr<logging::Logger> logger_;
};
diff --git a/libminifi/include/provenance/Provenance.h
b/libminifi/include/provenance/Provenance.h
index d891581de..3f433409b 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;
+ 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();
}
@@ -214,10 +223,11 @@ class ProvenanceEventRecordImpl : public
core::SerializableComponentImpl, public
bool serialize(io::OutputStream& output_stream) override;
bool deserialize(io::InputStream &input_stream) override;
- bool loadFromRepository(const std::shared_ptr<core::Repository> &repo)
override;
protected:
ProvenanceEventType event_type_;
+ // the index of the event
+ uint64_t event_ordinal_{0};
// Date at which the event was created
std::chrono::system_clock::time_point event_time_{};
// Date at which the flow file entered the flow
@@ -243,15 +253,11 @@ class ProvenanceEventRecordImpl : public
core::SerializableComponentImpl, public
std::string source_queue_identifier_;
std::string relationship_;
std::string alternate_identifier_uri_;
-
- private:
- static std::shared_ptr<core::logging::Logger> logger_;
- static std::shared_ptr<utils::IdGenerator> id_generator_;
};
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 +272,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 {
@@ -290,7 +292,7 @@ class ProvenanceReporterImpl : public virtual
ProvenanceReporter {
void clone(const core::FlowFile& parent, const core::FlowFile& child)
override;
void expire(const core::FlowFile& flow_file, const std::string& detail)
override;
void drop(const core::FlowFile& flow_file, const std::string& reason)
override;
- void send(const core::FlowFile& flow_file, const std::string& transitUri,
const std::string& detail, std::chrono::milliseconds processingDuration, bool
force) override;
+ void send(const core::FlowFile& flow_file, const std::string& transitUri,
const std::string& detail, std::chrono::milliseconds processingDuration)
override;
void fetch(const core::FlowFile& flow_file, const std::string& transitUri,
const std::string& detail, std::chrono::milliseconds processingDuration)
override;
void receive(const core::FlowFile& flow_file, const std::string& transitUri,
const std::string& sourceSystemFlowFileIdentifier, const std::string&
detail, std::chrono::milliseconds processingDuration) override;
@@ -313,8 +315,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 e315ae3b5..040196cec 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, RequestShutdownCallback request_shutdown,
diff --git a/libminifi/src/ThreadedSchedulingAgent.cpp
b/libminifi/src/ThreadedSchedulingAgent.cpp
index 8750af781..9b298b235 100644
--- a/libminifi/src/ThreadedSchedulingAgent.cpp
+++ b/libminifi/src/ThreadedSchedulingAgent.cpp
@@ -72,7 +72,7 @@ void ThreadedSchedulingAgent::schedule(core::Processor*
processor) {
}
auto state_storage = core::ProcessContextImpl::getStateStorage(logger_,
controller_service_provider_, configure_);
- auto process_context =
std::make_shared<core::ProcessContextImpl>(*processor,
controller_service_provider_, state_storage, repo_, flow_repo_, configure_,
content_repo_);
+ auto process_context =
std::make_shared<core::ProcessContextImpl>(*processor,
controller_service_provider_, state_storage, provenance_repo_, flow_repo_,
configure_, content_repo_);
auto session_factory =
std::make_shared<core::ProcessSessionFactoryImpl>(process_context);
@@ -81,7 +81,7 @@ void ThreadedSchedulingAgent::schedule(core::Processor*
processor) {
ThreadedSchedulingAgent *agent = this;
for (uint8_t i = 0; i < processor->getMaxConcurrentTasks(); i++) {
processor->incrementActiveTasks();
- auto thread_process_context =
std::make_shared<core::ProcessContextImpl>(*processor,
controller_service_provider_, state_storage, repo_, flow_repo_, configure_,
content_repo_);
+ auto thread_process_context =
std::make_shared<core::ProcessContextImpl>(*processor,
controller_service_provider_, state_storage, provenance_repo_, flow_repo_,
configure_, content_repo_);
auto thread_session_factory =
std::make_shared<core::ProcessSessionFactoryImpl>(thread_process_context);
std::function<utils::TaskRescheduleInfo()> f_ex = [agent, processor,
thread_process_context, thread_session_factory] () {
return agent->run(processor, thread_process_context,
thread_session_factory);
diff --git a/libminifi/src/core/ProcessContextImpl.cpp
b/libminifi/src/core/ProcessContextImpl.cpp
index c732f32bc..bb8399645 100644
--- a/libminifi/src/core/ProcessContextImpl.cpp
+++ b/libminifi/src/core/ProcessContextImpl.cpp
@@ -41,13 +41,13 @@ 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,
- const std::shared_ptr<core::Repository>& flow_repo, const
std::shared_ptr<core::ContentRepository>& content_repo)
+ Processor& processor, controller::ControllerServiceProvider*
controller_service_provider, const std::shared_ptr<core::StateStorage>&
state_storage,
+ const std::shared_ptr<provenance::ProvenanceRepository>& provenance_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()),
controller_service_provider_(controller_service_provider),
state_storage_(state_storage),
- repo_(repo),
+ provenance_repo_(provenance_repo),
flow_repo_(flow_repo),
content_repo_(content_repo),
processor_(processor),
@@ -55,14 +55,14 @@ 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,
- const std::shared_ptr<core::Repository>& flow_repo, const
std::shared_ptr<minifi::Configure>& configuration,
+ Processor& processor, controller::ControllerServiceProvider*
controller_service_provider, const std::shared_ptr<core::StateStorage>&
state_storage,
+ const std::shared_ptr<provenance::ProvenanceRepository>& provenance_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),
logger_(logging::LoggerFactory<ProcessContext>::getLogger()),
controller_service_provider_(controller_service_provider),
state_storage_(state_storage),
- repo_(repo),
+ provenance_repo_(provenance_repo),
flow_repo_(flow_repo),
content_repo_(content_repo),
processor_(processor),
diff --git a/libminifi/src/core/ProcessSession.cpp
b/libminifi/src/core/ProcessSession.cpp
index 4352f7b41..b6d4fce1a 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -1065,7 +1065,7 @@ void ProcessSessionImpl::persistFlowFilesBeforeTransfer(
return;
}
- std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>>
flowData;
+ Repository::EntryStreams flowData;
enum class Type {
Dropped, Transferred
diff --git a/libminifi/src/core/flow/StructuredConfiguration.cpp
b/libminifi/src/core/flow/StructuredConfiguration.cpp
index 47324cb62..ef68db941 100644
--- a/libminifi/src/core/flow/StructuredConfiguration.cpp
+++ b/libminifi/src/core/flow/StructuredConfiguration.cpp
@@ -587,6 +587,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 363c81f80..c7c004695 100644
--- a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
+++ b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
@@ -17,35 +17,59 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include <vector>
-#include <queue>
+#include "core/reporting/SiteToSiteProvenanceReportingTask.h"
+
+#include <functional>
+#include <iostream>
#include <map>
-#include <set>
-#include <string>
#include <memory>
+#include <queue>
+#include <set>
#include <sstream>
-#include <functional>
-#include <iostream>
+#include <string>
#include <utility>
+#include <vector>
-#include "rapidjson/document.h"
-#include "rapidjson/writer.h"
-#include "rapidjson/stringbuffer.h"
-#include "rapidjson/prettywriter.h"
-
-#include "minifi-cpp/core/Repository.h"
-#include "core/reporting/SiteToSiteProvenanceReportingTask.h"
-#include "utils/TimeUtil.h"
-#include "minifi-cpp/core/ProcessContext.h"
+#include "FlowController.h"
#include "core/ProcessSession.h"
+#include "minifi-cpp/core/ProcessContext.h"
+#include "minifi-cpp/core/Repository.h"
#include "minifi-cpp/provenance/Provenance.h"
-#include "FlowController.h"
#include "minifi-cpp/utils/gsl.h"
+#include "rapidjson/document.h"
+#include "rapidjson/prettywriter.h"
+#include "rapidjson/stringbuffer.h"
+#include "rapidjson/writer.h"
+#include "utils/TimeUtil.h"
+#include "utils/net/NetworkInterfaceInfo.h"
namespace org::apache::nifi::minifi::core::reporting {
const char *SiteToSiteProvenanceReportingTask::ProvenanceAppStr = "MiNiFi
Flow";
+namespace {
+
+std::unique_ptr<provenance::ProvenanceRepository::Cursor>
getCursor(StateManager& state_manager, provenance::ProvenanceRepository&
provenance_repository, logging::Logger& logger) {
+ 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 = provenance_repository.cursorFromString(cursor_str);
+ if (!cursor_str.empty() && !cursor) {
+ logger.log_error("Failed to parse cursor, falling back to enumerating from
the beginning");
+ cursor = provenance_repository.cursorFromString("");
+ }
+ return cursor;
+}
+
+} // namespace
+
void SiteToSiteProvenanceReportingTask::initialize() {
RemoteProcessGroupPort::initialize();
}
@@ -88,7 +112,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 +135,8 @@ std::string
SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContex
recordJson.AddMember("entityType", "org.apache.nifi.flowfile.FlowFile",
alloc);
+ recordJson.AddMember("eventOrdinal", record->getEventOrdinal(), 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 +182,23 @@ void
SiteToSiteProvenanceReportingTask::onSchedule(core::ProcessContext& context
}
void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext&
context, core::ProcessSession& session) {
- std::shared_ptr<core::Repository> repo = context.getProvenanceRepository();
+ std::shared_ptr<provenance::ProvenanceRepository> repo =
context.getProvenanceRepository();
if (!repo) {
throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to
retrieve provenance repository");
}
- auto records = repo->getElements(batch_size_);
+ auto* state_manager = context.getStateManager();
+ if (!state_manager) {
+ logger_->log_error("Failed to get StateManager");
+ context.yield();
+ return;
+ }
+ auto cursor = getCursor(*state_manager, *repo, *logger_);
+ std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> records;
+ 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;
@@ -182,14 +220,29 @@ void
SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context,
std::map<std::string, std::string> attributes;
if (!protocol_->transmitPayload(context, jsonStr, attributes)) {
context.yield();
+ return;
}
} catch (...) {
// if transfer bytes failed, return instead of purge the provenance records
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/core/repository/VolatileProvenanceRepository.cpp
b/libminifi/src/core/repository/VolatileProvenanceRepository.cpp
new file mode 100644
index 000000000..31adcbf54
--- /dev/null
+++ b/libminifi/src/core/repository/VolatileProvenanceRepository.cpp
@@ -0,0 +1,112 @@
+/**
+ *
+ * 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 "core/repository/VolatileProvenanceRepository.h"
+
+#include <algorithm>
+#include <memory>
+#include <span>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "io/BufferStream.h"
+#include "utils/ParsingUtils.h"
+
+namespace org::apache::nifi::minifi::core::repository {
+
+namespace {
+class EventCursor : public provenance::ProvenanceRepository::Cursor {
+ public:
+ explicit EventCursor(uint64_t last_event_ordinal):
last_event_ordinal_(last_event_ordinal) {}
+
+ // the ordinal of the last event observed, 0 if no event has been read yet
+ [[nodiscard]]
+ std::string toString() const override {
+ return std::to_string(last_event_ordinal_);
+ }
+
+ uint64_t last_event_ordinal_;
+};
+} // namespace
+
+std::unique_ptr<provenance::ProvenanceRepository::Cursor>
VolatileProvenanceRepository::cursorFromString(std::string_view cursor_str) {
+ if (cursor_str.empty()) {
+ return std::make_unique<EventCursor>(0);
+ }
+ if (auto last_event_ordinal = parsing::parseIntegral<uint64_t>(cursor_str)) {
+ return std::make_unique<EventCursor>(*last_event_ordinal);
+ }
+ logger_->log_warn("Could not interpret provenance cursor '{}', reading from
the first event", cursor_str);
+ return std::make_unique<EventCursor>(0);
+}
+
+std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>,
std::string> VolatileProvenanceRepository::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 {};
+ }
+ const uint64_t first_event_ordinal = event_cursor ?
event_cursor->last_event_ordinal_ + 1 : 0;
+
+ // The entries are stored in no particular order, but each of them is keyed
by the ordinal of
+ // the event it holds, so the events to be returned can be selected without
deserializing all
+ // of them.
+ std::vector<std::pair<uint64_t, AtomicEntry<std::string>*>> candidates;
+ const auto event_ordinal_of = [] (const std::pair<uint64_t,
AtomicEntry<std::string>*>& candidate) { return candidate.first; };
+ for (auto* entry : repo_data_.value_vector) {
+ entry->visitValue([&] (const std::string& key, std::span<const std::byte>
/*buffer*/) {
+ if (auto event_ordinal = parsing::parseIntegral<uint64_t>(key);
event_ordinal && *event_ordinal >= first_event_ordinal) {
+ candidates.emplace_back(*event_ordinal, entry);
+ }
+ });
+ }
+ std::ranges::sort(candidates, std::less<>{}, event_ordinal_of);
+ if (candidates.size() > max_size) {
+ candidates.resize(max_size);
+ }
+
+ std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> records;
+ records.reserve(candidates.size());
+ for (const auto& [event_ordinal, entry] : candidates) {
+ auto event = provenance::ProvenanceEventRecord::create();
+ bool deserialized = false;
+ entry->visitValue([&] (const std::string& key, std::span<const std::byte>
buffer) {
+ if (key != std::to_string(event_ordinal)) {
+ // the entry has been overwritten since it was selected
+ return;
+ }
+ io::BufferStream stream{buffer};
+ deserialized = event->deserialize(stream);
+ });
+ if (deserialized) {
+ records.push_back(std::move(event));
+ } else {
+ logger_->log_warn("Could not read provenance event '{}'", event_ordinal);
+ }
+ }
+ if (event_cursor && !candidates.empty()) {
+ // advanced even for the events we failed to read, otherwise they would
stall the iteration
+ event_cursor->last_event_ordinal_ = candidates.back().first;
+ }
+
+ return records;
+}
+
+} // namespace org::apache::nifi::minifi::core::repository
diff --git a/libminifi/src/core/repository/VolatileRepository.cpp
b/libminifi/src/core/repository/VolatileRepository.cpp
index f08cad67a..fc237a1b9 100644
--- a/libminifi/src/core/repository/VolatileRepository.cpp
+++ b/libminifi/src/core/repository/VolatileRepository.cpp
@@ -78,7 +78,7 @@ bool VolatileRepository::Put(const std::string& key, const
uint8_t *buf, size_t
return true;
}
-bool VolatileRepository::MultiPut(const std::vector<std::pair<std::string,
std::unique_ptr<io::BufferStream>>>& data) {
+bool VolatileRepository::MultiPut(const EntryStreams& data) {
for (const auto& item : data) {
if (!Put(item.first, reinterpret_cast<const
uint8_t*>(item.second->getBuffer().data()), item.second->size())) {
return false;
diff --git a/libminifi/src/provenance/Provenance.cpp
b/libminifi/src/provenance/Provenance.cpp
index 7a01638a3..7d0e80916 100644
--- a/libminifi/src/provenance/Provenance.cpp
+++ b/libminifi/src/provenance/Provenance.cpp
@@ -26,7 +26,6 @@
#include <utility>
#include "minifi-cpp/core/Repository.h"
-#include "io/BufferStream.h"
#include "minifi-cpp/core/logging/Logger.h"
#include "core/Relationship.h"
#include "FlowController.h"
@@ -36,9 +35,6 @@ namespace org::apache::nifi::minifi::provenance {
constexpr auto MAX_COMPONENT_NAME_LENGTH = 1_KiB;
-std::shared_ptr<utils::IdGenerator> ProvenanceEventRecordImpl::id_generator_ =
utils::IdGenerator::getIdGenerator();
-std::shared_ptr<core::logging::Logger> ProvenanceEventRecordImpl::logger_ =
core::logging::LoggerFactory<ProvenanceEventRecord>::getLogger();
-
const char *ProvenanceEventRecord::ProvenanceEventTypeStr[REPLAY + 1] = {
"CREATE", "RECEIVE", "FETCH", "SEND", "DOWNLOAD", //
NOLINT(cppcoreguidelines-avoid-c-arrays)
"DROP", "EXPIRE", "FORK", "JOIN", "CLONE", "CONTENT_MODIFIED",
"ATTRIBUTES_MODIFIED", "ROUTE", "ADDINFO", "REPLAY" };
@@ -50,36 +46,6 @@
ProvenanceEventRecordImpl::ProvenanceEventRecordImpl(ProvenanceEventRecord::Prov
component_type_(std::move(component_type)) {
}
-bool ProvenanceEventRecordImpl::loadFromRepository(const
std::shared_ptr<core::Repository> &repo) {
- std::string value;
- bool ret = false;
-
- if (nullptr == repo || uuid_.isNil()) {
- logger_->log_error("Repo could not be assigned");
- return false;
- }
- ret = repo->Get(getUUIDStr(), value);
-
- if (!ret) {
- logger_->log_error("NiFi Provenance Store event {} can not be found",
getUUIDStr());
- return false;
- } else {
- logger_->log_debug("NiFi Provenance Read event {}", getUUIDStr());
- }
-
- org::apache::nifi::minifi::io::BufferStream stream(value);
-
- ret = deserialize(stream);
-
- if (ret) {
- logger_->log_debug("NiFi Provenance retrieve event {} size {} eventType {}
success", getUUIDStr(), stream.size(), magic_enum::enum_name(event_type_));
- } else {
- logger_->log_debug("NiFi Provenance retrieve event {} size {} eventType {}
fail", getUUIDStr(), stream.size(), magic_enum::enum_name(event_type_));
- }
-
- return ret;
-}
-
bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
{
const auto ret = output_stream.write(uuid_);
@@ -240,6 +206,13 @@ bool
ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
}
}
+ {
+ const auto ret = output_stream.write(event_ordinal_);
+ if (ret != 8) {
+ return false;
+ }
+ }
+
return true;
}
@@ -443,6 +416,17 @@ bool
ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
}
}
+ {
+ uint64_t event_ordinal = 0;
+ const auto ret = input_stream.read(event_ordinal);
+ if (ret != 8) {
+ // backwards compatibility to be able to deserialize older provenance
events
+ event_ordinal_ = 0;
+ } else {
+ event_ordinal_ = event_ordinal;
+ }
+ }
+
return true;
}
@@ -452,19 +436,15 @@ void ProvenanceReporterImpl::commit() {
}
if (repo_->isFull()) {
- logger_->log_debug("Provenance Repository is full");
+ logger_->log_error("Provenance Repository is full");
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));
+ if (auto append_result = repo_->appendEvents(events_); !append_result) {
+ // do not let failed provenance repo stop all processing
+ logger_->log_error("Failed to append provenance events: {}",
append_result);
+ return;
}
- repo_->MultiPut(flowData);
}
void ProvenanceReporterImpl::create(const core::FlowFile& flow_file, const
std::string& detail) {
@@ -534,19 +514,14 @@ void ProvenanceReporterImpl::drop(const core::FlowFile&
flow_file, const std::st
}
}
-void ProvenanceReporterImpl::send(const core::FlowFile& flow_file, const
std::string& transitUri, const std::string& detail, std::chrono::milliseconds
processingDuration, bool force) {
+void ProvenanceReporterImpl::send(const core::FlowFile& flow_file, const
std::string& transitUri, const std::string& detail, std::chrono::milliseconds
processingDuration) {
auto event = allocate(ProvenanceEventRecord::SEND, flow_file);
if (event) {
event->setTransitUri(transitUri);
event->setDetails(detail);
event->setEventDuration(processingDuration);
- if (!force) {
- add(event);
- } else {
- if (!repo_->isFull())
- repo_->storeElement(event);
- }
+ add(event);
}
}
diff --git a/libminifi/src/sitetosite/SiteToSiteClient.cpp
b/libminifi/src/sitetosite/SiteToSiteClient.cpp
index 3871e666f..4586329d2 100644
--- a/libminifi/src/sitetosite/SiteToSiteClient.cpp
+++ b/libminifi/src/sitetosite/SiteToSiteClient.cpp
@@ -152,7 +152,7 @@ bool
SiteToSiteClient::transferFlowFiles(core::ProcessContext& context, core::Pr
auto end_time = std::chrono::steady_clock::now();
std::string transit_uri = peer_->getURL() + "/" + flow->getUUIDStr();
std::string details = "urn:nifi:" + flow->getUUIDStr() + "Remote Host="
+ peer_->getHostName();
- session.getProvenanceReporter()->send(*flow, transit_uri, details,
std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time),
false);
+ session.getProvenanceReporter()->send(*flow, transit_uri, details,
std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time));
session.remove(flow);
std::chrono::nanoseconds transfer_duration =
std::chrono::high_resolution_clock::now() - transaction_started_at;
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 c825ef6d6..28c7bf6eb 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..c1df15a1f 100644
--- a/libminifi/test/libtest/unit/ProvenanceTestHelper.h
+++ b/libminifi/test/libtest/unit/ProvenanceTestHelper.h
@@ -27,6 +27,7 @@
#include <string>
#include <utility>
#include <vector>
+#include "fmt/format.h"
#include "core/repository/VolatileContentRepository.h"
#include "core/Processor.h"
#include "core/ThreadedRepository.h"
@@ -41,7 +42,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 +90,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,9 +117,32 @@ 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());
+ std::lock_guard guard{next_event_key_mtx_};
+ for (auto& event : events) {
+ event->setEventOrdinal(next_event_key_++);
+ // zero padded so that the keys of the underlying map are ordered by the
event ordinal
+ data.emplace_back(fmt::format("{:020}", event->getEventOrdinal()),
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::string_view /*cursor_str*/) override {
+ return nullptr;
+ }
+
protected:
mutable std::mutex repository_results_mutex_;
std::map<std::string, std::string> repository_results_;
+ std::mutex next_event_key_mtx_;
+ uint64_t next_event_key_{1};
};
class TestRepository : public
TestRepositoryBase<org::apache::nifi::minifi::core::RepositoryImpl> {
diff --git a/libminifi/test/libtest/unit/TestBase.cpp
b/libminifi/test/libtest/unit/TestBase.cpp
index 23a967ae9..e69fecbb9 100644
--- a/libminifi/test/libtest/unit/TestBase.cpp
+++ b/libminifi/test/libtest/unit/TestBase.cpp
@@ -208,8 +208,9 @@ 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,
- std::shared_ptr<minifi::state::response::FlowVersion>
flow_version, std::shared_ptr<minifi::Configure> configuration, const char*
state_dir)
+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)),
flow_repo_(std::move(flow_repo)),
@@ -637,7 +638,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 d1e87f743..3bf8a31a1 100644
--- a/libminifi/test/libtest/unit/TestBase.h
+++ b/libminifi/test/libtest/unit/TestBase.h
@@ -243,8 +243,9 @@ 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,
- std::shared_ptr<minifi::state::response::FlowVersion>
flow_version, std::shared_ptr<minifi::Configure> configuration, const char*
state_dir);
+ 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 +306,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 +358,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..2dc659fde 100644
--- a/libminifi/test/persistence-tests/PersistenceTests.cpp
+++ b/libminifi/test/persistence-tests/PersistenceTests.cpp
@@ -60,8 +60,9 @@ 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,
- const
std::function<std::unique_ptr<core::Processor>(utils::Identifier&)>&
processorGenerator, const core::Relationship& relationshipToOutput)
+ 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
auto processor = processorGenerator(mainProcUUID());
@@ -178,7 +179,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 +293,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/LogMetricsPublisherTests.cpp
b/libminifi/test/unit/LogMetricsPublisherTests.cpp
index aa9765d64..429c8447c 100644
--- a/libminifi/test/unit/LogMetricsPublisherTests.cpp
+++ b/libminifi/test/unit/LogMetricsPublisherTests.cpp
@@ -50,8 +50,15 @@ class LogPublisherTestFixture {
~LogPublisherTestFixture() {
publisher_.reset(); // explicit because LogTestController should outlive
the thread in publisher_
- minifi::utils::file::delete_dir(provenance_repo_->getDirectory());
- minifi::utils::file::delete_dir(flow_file_repo_->getDirectory());
+ const auto provenance_repo_directory = provenance_repo_->getDirectory();
+ const auto flow_file_repo_directory = flow_file_repo_->getDirectory();
+ // the repositories have to be closed before their directories are
removed, as the files of an
+ // open database cannot be deleted on every platform
+ response_node_loader_.reset();
+ provenance_repo_.reset();
+ flow_file_repo_.reset();
+ minifi::utils::file::delete_dir(provenance_repo_directory);
+ minifi::utils::file::delete_dir(flow_file_repo_directory);
LogTestController::getInstance().reset();
}
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/libminifi/test/unit/VolatileProvenanceRepositoryTests.cpp
b/libminifi/test/unit/VolatileProvenanceRepositoryTests.cpp
new file mode 100644
index 000000000..6a4f0566c
--- /dev/null
+++ b/libminifi/test/unit/VolatileProvenanceRepositoryTests.cpp
@@ -0,0 +1,169 @@
+/**
+ *
+ * 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 <algorithm>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "core/repository/VolatileProvenanceRepository.h"
+#include "properties/Configure.h"
+#include "utils/StringUtils.h"
+#include "unit/TestBase.h"
+#include "unit/Catch.h"
+
+namespace org::apache::nifi::minifi::core::repository::test {
+
+std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>
createEvents(size_t count) {
+ std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> events;
+ events.reserve(count);
+ for (size_t i = 0; i < count; ++i) {
+ events.push_back(provenance::ProvenanceEventRecord::create());
+ }
+ return events;
+}
+
+std::vector<uint64_t> getEventOrdinals(const
std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>& events) {
+ std::vector<uint64_t> ordinals;
+ ordinals.reserve(events.size());
+ for (const auto& event : events) {
+ ordinals.push_back(event->getEventOrdinal());
+ }
+ return ordinals;
+}
+
+std::shared_ptr<VolatileProvenanceRepository> createRepository(uint32_t
max_count = 0) {
+ auto repository = std::make_shared<VolatileProvenanceRepository>();
+ std::shared_ptr<Configure> configuration;
+ if (max_count > 0) {
+ configuration = std::make_shared<ConfigureImpl>();
+
configuration->set(utils::string::join_pack(Configure::nifi_volatile_repository_options,
repository->getName(), ".", VOLATILE_REPO_MAX_COUNT),
std::to_string(max_count));
+ }
+ REQUIRE(repository->initialize(configuration));
+ return repository;
+}
+
+TEST_CASE("VolatileProvenanceRepository assigns consecutive ordinals",
"[volatileProvenanceRepository]") {
+ auto repository = createRepository();
+
+ auto first_batch = createEvents(3);
+ REQUIRE(repository->appendEvents(first_batch));
+ REQUIRE(getEventOrdinals(first_batch) == std::vector<uint64_t>{1, 2, 3});
+
+ auto second_batch = createEvents(2);
+ REQUIRE(repository->appendEvents(second_batch));
+ REQUIRE(getEventOrdinals(second_batch) == std::vector<uint64_t>{4, 5});
+}
+
+TEST_CASE("VolatileProvenanceRepository restarts the ordinals in a new
instance", "[volatileProvenanceRepository]") {
+ auto events = createEvents(3);
+ {
+ auto repository = createRepository();
+ REQUIRE(repository->appendEvents(events));
+ REQUIRE(events.back()->getEventOrdinal() == 3);
+ }
+
+ // unlike the persistent repository, this one does not remember the ordinals
it handed out
+ auto new_events = createEvents(1);
+ auto new_repository = createRepository();
+ REQUIRE(new_repository->appendEvents(new_events));
+ REQUIRE(new_events.front()->getEventOrdinal() == 1);
+}
+
+TEST_CASE("VolatileProvenanceRepository returns the events in ordinal order",
"[volatileProvenanceRepository]") {
+ auto repository = createRepository();
+
+ // more than nine events, so that the ordinals do not all have the same
number of digits
+ auto events = createEvents(12);
+ REQUIRE(repository->appendEvents(events));
+
+ auto cursor = repository->cursorFromString("");
+ REQUIRE(cursor);
+
+ auto queried_events = repository->getEvents(12, cursor.get());
+ REQUIRE(queried_events);
+ REQUIRE(queried_events->size() == 12);
+ REQUIRE(getEventOrdinals(*queried_events) == std::vector<uint64_t>{1, 2, 3,
4, 5, 6, 7, 8, 9, 10, 11, 12});
+ for (size_t i = 0; i < queried_events->size(); ++i) {
+ REQUIRE(queried_events->at(i)->getEventId() == events.at(i)->getEventId());
+ }
+}
+
+TEST_CASE("VolatileProvenanceRepository continues the iteration from the
cursor", "[volatileProvenanceRepository]") {
+ auto repository = createRepository();
+
+ REQUIRE(repository->appendEvents(createEvents(8)));
+
+ auto cursor = repository->cursorFromString("");
+ REQUIRE(cursor);
+
+ auto first_page = repository->getEvents(3, cursor.get());
+ REQUIRE(first_page);
+ REQUIRE(getEventOrdinals(*first_page) == std::vector<uint64_t>{1, 2, 3});
+ REQUIRE(cursor->toString() == "3");
+
+ // the cursor can be persisted and restored
+ cursor = repository->cursorFromString(cursor->toString());
+ REQUIRE(cursor);
+
+ auto second_page = repository->getEvents(3, cursor.get());
+ REQUIRE(second_page);
+ REQUIRE(getEventOrdinals(*second_page) == std::vector<uint64_t>{4, 5, 6});
+
+ // events appended after the cursor was created are picked up as well
+ REQUIRE(repository->appendEvents(createEvents(2)));
+
+ auto third_page = repository->getEvents(10, cursor.get());
+ REQUIRE(third_page);
+ REQUIRE(getEventOrdinals(*third_page) == std::vector<uint64_t>{7, 8, 9, 10});
+
+ auto fourth_page = repository->getEvents(10, cursor.get());
+ REQUIRE(fourth_page);
+ REQUIRE(fourth_page->empty());
+}
+
+TEST_CASE("VolatileProvenanceRepository does not consume the events it
returns", "[volatileProvenanceRepository]") {
+ auto repository = createRepository();
+
+ REQUIRE(repository->appendEvents(createEvents(3)));
+
+ auto cursor = repository->cursorFromString("");
+ REQUIRE(cursor);
+ REQUIRE(getEventOrdinals(repository->getEvents(3, cursor.get()).value()) ==
std::vector<uint64_t>{1, 2, 3});
+
+ auto new_cursor = repository->cursorFromString("");
+ REQUIRE(new_cursor);
+ REQUIRE(getEventOrdinals(repository->getEvents(3, new_cursor.get()).value())
== std::vector<uint64_t>{1, 2, 3});
+}
+
+TEST_CASE("VolatileProvenanceRepository returns the events that survived an
overflow", "[volatileProvenanceRepository]") {
+ auto repository = createRepository(4);
+
+ REQUIRE(repository->appendEvents(createEvents(6)));
+
+ auto queried_events = repository->getEvents(10, nullptr);
+ REQUIRE(queried_events);
+ // the oldest events have been overwritten, the remaining ones are still
ordered
+ REQUIRE(queried_events->size() <= 4);
+ REQUIRE(!queried_events->empty());
+ auto ordinals = getEventOrdinals(*queried_events);
+ REQUIRE(std::ranges::is_sorted(ordinals));
+ REQUIRE(ordinals.back() == 6);
+}
+
+} // namespace org::apache::nifi::minifi::core::repository::test
diff --git a/minifi-api/include/minifi-cpp/core/ProcessContext.h
b/minifi-api/include/minifi-cpp/core/ProcessContext.h
index 7cecd92f1..a2dafc45c 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..accfc695c 100644
--- a/minifi-api/include/minifi-cpp/core/Repository.h
+++ b/minifi-api/include/minifi-cpp/core/Repository.h
@@ -36,6 +36,8 @@ namespace org::apache::nifi::minifi::core {
class Repository : public virtual core::CoreComponent, public virtual
core::RepositoryMetricsSource {
public:
+ using EntryStreams = std::vector<std::pair<std::string,
std::unique_ptr<io::BufferStream>>>;
+
virtual bool initialize(const std::shared_ptr<Configure> &configure) = 0;
virtual bool start() = 0;
virtual bool stop() = 0;
@@ -43,7 +45,7 @@ class Repository : public virtual core::CoreComponent, public
virtual core::Repo
virtual void flush() = 0;
virtual bool Put(const std::string& /*key*/, const uint8_t* /*buf*/, size_t
/*bufLen*/) = 0;
- virtual bool MultiPut(const std::vector<std::pair<std::string,
std::unique_ptr<io::BufferStream>>>& /*data*/) = 0;
+ virtual bool MultiPut(const EntryStreams& /*data*/) = 0;
virtual bool Delete(const std::string& /*key*/) = 0;
virtual bool Delete(const std::shared_ptr<core::CoreComponent>& item) = 0;
virtual bool
Delete(std::vector<std::shared_ptr<core::SerializableComponent>> &storedValues)
= 0;
@@ -54,10 +56,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..66d88d510 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 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;
@@ -183,7 +185,6 @@ class ProvenanceEventRecord : public virtual
core::SerializableComponent {
virtual std::string getSourceQueueIdentifier() const = 0;
virtual void setSourceQueueIdentifier(const std::string& identifier) = 0;
virtual void fromFlowFile(const core::FlowFile& flow_file) = 0;
- virtual bool loadFromRepository(const std::shared_ptr<core::Repository>
&repo) = 0;
static std::shared_ptr<ProvenanceEventRecord> create();
};
@@ -192,9 +193,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;
@@ -205,7 +205,7 @@ class ProvenanceReporter {
virtual void clone(const core::FlowFile& parent, const core::FlowFile&
child) = 0;
virtual void expire(const core::FlowFile& flow_file, const std::string&
detail) = 0;
virtual void drop(const core::FlowFile& flow_file, const std::string&
reason) = 0;
- virtual void send(const core::FlowFile& flow_file, const std::string&
transitUri, const std::string& detail, std::chrono::milliseconds
processingDuration, bool force) = 0;
+ virtual void send(const core::FlowFile& flow_file, const std::string&
transitUri, const std::string& detail, std::chrono::milliseconds
processingDuration) = 0;
virtual void fetch(const core::FlowFile& flow_file, const std::string&
transitUri, const std::string& detail, std::chrono::milliseconds
processingDuration) = 0;
virtual void receive(const core::FlowFile& flow_file, const std::string&
transitUri,
const std::string& sourceSystemFlowFileIdentifier, const std::string&
detail, std::chrono::milliseconds processingDuration) = 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..bf436b90c
--- /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 the empty string it creates a cursor to the beginning
if the underlying repository supports it
+ virtual std::unique_ptr<Cursor> cursorFromString(std::string_view
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 1516a22f8..53fa79d82 100644
--- a/minifi_main/MiNiFiMain.cpp
+++ b/minifi_main/MiNiFiMain.cpp
@@ -382,7 +382,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..");
