diff --git a/core-framework/common/src/utils/Id.cpp b/core-framework/common/src/utils/Id.cpp index d401d0a5e0..09bdee3eb3 100644 --- a/core-framework/common/src/utils/Id.cpp +++ b/core-framework/common/src/utils/Id.cpp @@ -43,6 +43,18 @@ bool Identifier::isNil() const { return *this == Identifier{}; } +Identifier& Identifier::operator++() { + for (int byte_idx = 15; byte_idx >= 0 && ++data_[byte_idx] == 0; --byte_idx) {} + return *this; +} + +Identifier Identifier::operator++(int) { + auto result = *this; + ++*this; + return result; +} + + bool Identifier::operator!=(const Identifier& other) const { return !(*this == other); } diff --git a/core-framework/include/core/Repository.h b/core-framework/include/core/Repository.h index ee2d3a0403..1615a20b82 100644 --- a/core-framework/include/core/Repository.h +++ b/core-framework/include/core/Repository.h @@ -106,12 +106,6 @@ class RepositoryImpl : public core::CoreComponentImpl, public core::RepositoryMe return false; } - bool getElements(std::vector>& /*store*/, size_t& /*max_size*/) override { - return true; - } - - bool storeElement(const std::shared_ptr& element) override; - void loadComponent(const std::shared_ptr& /*content_repo*/) override { } diff --git a/core-framework/src/core/Repository.cpp b/core-framework/src/core/Repository.cpp index 722e82e7a6..7004e05c0d 100644 --- a/core-framework/src/core/Repository.cpp +++ b/core-framework/src/core/Repository.cpp @@ -27,20 +27,4 @@ bool RepositoryImpl::Delete(std::vector& element) { - if (!element) { - return false; - } - - org::apache::nifi::minifi::io::BufferStream stream; - - element->serialize(stream); - - if (!Put(element->getUUIDStr(), reinterpret_cast(stream.getBuffer().data()), stream.size())) { - logger_->log_error("NiFi Provenance Store event {} size {} fail", element->getUUIDStr(), stream.size()); - return false; - } - return true; -} - } // namespace org::apache::nifi::minifi::core diff --git a/extensions/grafana-loki/PushGrafanaLokiGrpc.cpp b/extensions/grafana-loki/PushGrafanaLokiGrpc.cpp index 53428988f7..100d60d52d 100644 --- a/extensions/grafana-loki/PushGrafanaLokiGrpc.cpp +++ b/extensions/grafana-loki/PushGrafanaLokiGrpc.cpp @@ -116,7 +116,7 @@ std::expected PushGrafanaLokiGrpc::submitRequest(const std::v for (const auto& flow_file : batched_flow_files) { logproto::EntryAdapter *entry = stream->add_entries(); - auto timestamp_str = std::to_string(flow_file->getlineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1)); + auto timestamp_str = std::to_string(flow_file->getLineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1)); auto timestamp_nanos = std::stoll(timestamp_str); *entry->mutable_timestamp() = google::protobuf::util::TimeUtil::NanosecondsToTimestamp(timestamp_nanos); diff --git a/extensions/grafana-loki/PushGrafanaLokiREST.cpp b/extensions/grafana-loki/PushGrafanaLokiREST.cpp index fd01e15d4c..1a95918882 100644 --- a/extensions/grafana-loki/PushGrafanaLokiREST.cpp +++ b/extensions/grafana-loki/PushGrafanaLokiREST.cpp @@ -133,7 +133,7 @@ std::string PushGrafanaLokiREST::createLokiJson(const std::vectorgetlineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1)); + auto timestamp_str = std::to_string(flow_file->getLineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1)); rapidjson::Value timestamp; timestamp.SetString(timestamp_str.c_str(), gsl::narrow(timestamp_str.length()), allocator); rapidjson::Value log_line_value; diff --git a/extensions/rocksdb-repos/ProvenanceRepository.cpp b/extensions/rocksdb-repos/ProvenanceRepository.cpp deleted file mode 100644 index e14a97299e..0000000000 --- a/extensions/rocksdb-repos/ProvenanceRepository.cpp +++ /dev/null @@ -1,106 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "ProvenanceRepository.h" - -#include - -#include "core/Resource.h" - -namespace org::apache::nifi::minifi::provenance { - -bool ProvenanceRepository::initialize(const std::shared_ptr &config) { - std::string value; - if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) { - directory_ = value; - } - logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_); - if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) { - max_partition_bytes_ = gsl::narrow(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size")); - } - logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_); - if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) { - if (auto max_partition = utils::timeutils::StringToDuration(value)) - max_partition_millis_ = *max_partition; - } - logger_->log_debug("MiNiFi Provenance Max Storage Time: [{}]", max_partition_millis_); - - verify_checksums_in_rocksdb_reads_ = (config->get(Configure::nifi_provenance_repository_rocksdb_read_verify_checksums) | utils::andThen(&utils::string::toBool)).value_or(false); - logger_->log_debug("{} checksum verification in ProvenanceRepository", verify_checksums_in_rocksdb_reads_ ? "Using" : "Not using"); - - auto db_options = [] (minifi::internal::Writable& db_opts) { - minifi::internal::setCommonRocksDbOptions(db_opts); - }; - - // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after - // This shouldn't go above 16MB and the configured total size of the db should cap it as well - auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) { - int64_t max_buffer_size = 16 << 20; - cf_opts.write_buffer_size = gsl::narrow(std::min(max_buffer_size, max_partition_bytes_)); - cf_opts.max_write_buffer_number = 4; - cf_opts.min_write_buffer_number_to_merge = 1; - - cf_opts.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO; - cf_opts.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false); - if (max_partition_millis_ > std::chrono::milliseconds(0)) { - cf_opts.ttl = std::chrono::duration_cast(max_partition_millis_).count(); - } - }; - - db_ = minifi::internal::RocksDatabase::create(db_options, cf_options, directory_, - minifi::internal::getRocksDbOptionsToOverride(config, Configure::nifi_provenance_repository_rocksdb_options)); - if (db_->open()) { - logger_->log_debug("MiNiFi Provenance Repository database open {} success", directory_); - } else { - logger_->log_error("MiNiFi Provenance Repository database open {} failed", directory_); - return false; - } - - return true; -} - -bool ProvenanceRepository::getElements(std::vector> &records, size_t &max_size) { - auto opendb = db_->open(); - if (!opendb) { - return false; - } - rocksdb::ReadOptions options; - options.verify_checksums = verify_checksums_in_rocksdb_reads_; - std::unique_ptr it(opendb->NewIterator(options)); - size_t requested_batch = max_size; - max_size = 0; - for (it->SeekToFirst(); it->Valid(); it->Next()) { - if (max_size >= requested_batch) - break; - 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)) { - max_size++; - records.push_back(eventRead); - } - } - return max_size > 0; -} - -void ProvenanceRepository::destroy() { - db_.reset(); -} - -REGISTER_RESOURCE_AS(ProvenanceRepository, InternalResource, ("ProvenanceRepository", "provenancerepository")); - -} // namespace org::apache::nifi::minifi::provenance diff --git a/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp b/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp new file mode 100644 index 0000000000..f3397ea626 --- /dev/null +++ b/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp @@ -0,0 +1,207 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "RocksDbProvenanceRepository.h" + +#include + +#include "core/Resource.h" + +namespace org::apache::nifi::minifi::provenance { + +namespace { +class EventCursor : public ProvenanceRepository::Cursor { +public: + explicit EventCursor(std::string event_id): event_id_(std::move(event_id)) {} + [[nodiscard]] + std::string toString() const override { + return event_id_; + } + ~EventCursor() override = default; + + std::string event_id_; +}; +} // namespace + +static const std::string_view NEXT_EVENT_UUID_KEY = "next_event_uuid"; + +bool RocksDbProvenanceRepository::initialize(const std::shared_ptr &config) { + if (!RocksDbRepository::initialize(config)) { + return false; + } + std::string value; + if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_); + if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) { + max_partition_bytes_ = gsl::narrow(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size")); + } + logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_); + if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) { + if (auto max_partition = utils::timeutils::StringToDuration(value)) + max_partition_millis_ = *max_partition; + } + logger_->log_debug("MiNiFi Provenance Max Storage Time: [{}]", max_partition_millis_); + + verify_checksums_in_rocksdb_reads_ = (config->get(Configure::nifi_provenance_repository_rocksdb_read_verify_checksums) | utils::andThen(&utils::string::toBool)).value_or(false); + logger_->log_debug("{} checksum verification in RocksDbProvenanceRepository", verify_checksums_in_rocksdb_reads_ ? "Using" : "Not using"); + + auto db_options = [] (minifi::internal::Writable& db_opts) { + minifi::internal::setCommonRocksDbOptions(db_opts); + }; + + // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after + // This shouldn't go above 16MB and the configured total size of the db should cap it as well + auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) { + int64_t max_buffer_size = 16 << 20; + cf_opts.write_buffer_size = gsl::narrow(std::min(max_buffer_size, max_partition_bytes_)); + cf_opts.max_write_buffer_number = 4; + cf_opts.min_write_buffer_number_to_merge = 1; + + cf_opts.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO; + cf_opts.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false); + if (max_partition_millis_ > std::chrono::milliseconds(0)) { + cf_opts.ttl = std::chrono::duration_cast(max_partition_millis_).count(); + } + }; + + db_ = minifi::internal::RocksDatabase::create(db_options, cf_options, directory_, + minifi::internal::getRocksDbOptionsToOverride(config, Configure::nifi_provenance_repository_rocksdb_options)); + std::string internal_state_db_uri = [&] { + const std::string_view minifidb_scheme = "minifidb://"; + if (directory_.starts_with(minifidb_scheme)) { + return directory_ + "-internal-state"; + } + std::string uri = utils::string::join_pack(minifidb_scheme, directory_); + if (uri.ends_with("/") || uri.ends_with("\\")) { + uri.pop_back(); + } + return uri + "/internal-state"; + }(); + internal_state_db_ = minifi::internal::RocksDatabase::create(db_options, {}, internal_state_db_uri, {}); + if (auto open_state_db = internal_state_db_->open()) { + rocksdb::ReadOptions options; + options.verify_checksums = verify_checksums_in_rocksdb_reads_; + std::string next_event_uuid_str; + if (open_state_db->Get(options, NEXT_EVENT_UUID_KEY, &next_event_uuid_str).ok()) { + next_event_id_ = next_event_uuid_str; + } else { + logger_->log_error("Could not find '{}'", NEXT_EVENT_UUID_KEY); + next_event_id_ = utils::IdGenerator::getIdGenerator()->generate(); + } + logger_->log_trace("Using next event uuid: {}", next_event_id_.to_string()); + } else { + logger_->log_error("Could not open internal state column in provenance repository {}", internal_state_db_uri); + return false; + } + if (db_->open()) { + logger_->log_debug("MiNiFi Provenance Repository database open {} success", directory_); + } else { + logger_->log_error("MiNiFi Provenance Repository database open {} failed", directory_); + return false; + } + + return true; +} + +void RocksDbProvenanceRepository::destroy() { + db_.reset(); +} + +std::unique_ptr RocksDbProvenanceRepository::cursorFromString(std::optional cursor_str) { + if (cursor_str.has_value()) { + return std::make_unique(cursor_str.value()); + } + return std::make_unique(""); +} + +std::expected>, std::string> RocksDbProvenanceRepository::getEvents(size_t max_size, Cursor* cursor) { + auto* event_cursor = dynamic_cast(cursor); + if (cursor && !event_cursor) { + return std::unexpected{"Invalid cursor"}; + } + if (max_size == 0) { + return {}; + } + auto opendb = db_->open(); + if (!opendb) { + return std::unexpected{"Failed to open database"}; + } + std::vector> records; + rocksdb::ReadOptions options; + options.verify_checksums = verify_checksums_in_rocksdb_reads_; + std::unique_ptr it(opendb->NewIterator(options)); + std::string last_event_id; + if (event_cursor) { + last_event_id = event_cursor->event_id_; + it->Seek(event_cursor->event_id_); + if (it->Valid() && it->key() == event_cursor->event_id_) { + it->Next(); + } + } else { + it->SeekToFirst(); + } + for (; it->Valid(); it->Next()) { + last_event_id = it->key().ToString(); + auto eventRead = ProvenanceEventRecord::create(); + const auto slice = it->value(); + io::BufferStream stream(std::as_bytes(std::span(slice.data(), slice.size()))); + if (eventRead->deserialize(stream)) { + records.push_back(eventRead); + if (--max_size == 0) { + break; + } + } + } + if (event_cursor) { + event_cursor->event_id_ = last_event_id; + } + return records; +} + +std::expected RocksDbProvenanceRepository::appendEvents(const std::vector>& events) { + std::vector>> data; + data.reserve(events.size()); + std::lock_guard guard(next_event_id_mtx_); + for (auto& event : events) { + event->setUUID(next_event_id_++); + } + { + auto open_state_db = internal_state_db_->open(); + if (!open_state_db) { + return std::unexpected{"Failed to open internal state column in provenance database"}; + } + auto operation = [this, &open_state_db]() { return open_state_db->Put(rocksdb::WriteOptions(), NEXT_EVENT_UUID_KEY, next_event_id_.to_string().view()); }; + if (!ExecuteWithRetry(operation)) { + return std::unexpected{"Failed to update next provenance event id"}; + } + } + for (auto& event : events) { + data.emplace_back(event->getUUIDStr(), std::make_unique()); + event->serialize(*data.back().second); + } + if (MultiPut(data)) { + return {}; + } + + return std::unexpected{"Failed to append provenance events"}; +} + +REGISTER_RESOURCE_AS(RocksDbProvenanceRepository, InternalResource, ("RocksDbProvenanceRepository", "ProvenanceRepository", "provenancerepository")); + +} // namespace org::apache::nifi::minifi::provenance diff --git a/extensions/rocksdb-repos/ProvenanceRepository.h b/extensions/rocksdb-repos/RocksDbProvenanceRepository.h similarity index 69% rename from extensions/rocksdb-repos/ProvenanceRepository.h rename to extensions/rocksdb-repos/RocksDbProvenanceRepository.h index 3984c9911f..d841fecd89 100644 --- a/extensions/rocksdb-repos/ProvenanceRepository.h +++ b/extensions/rocksdb-repos/RocksDbProvenanceRepository.h @@ -29,6 +29,7 @@ #include "core/Core.h" #include "core/logging/LoggerFactory.h" #include "minifi-cpp/provenance/Provenance.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" #include "minifi-cpp/utils/Literals.h" #include "RocksDbRepository.h" @@ -39,22 +40,22 @@ constexpr auto MAX_PROVENANCE_STORAGE_SIZE = 10_MiB; constexpr auto MAX_PROVENANCE_ENTRY_LIFE_TIME = std::chrono::minutes(1); constexpr auto PROVENANCE_PURGE_PERIOD = std::chrono::milliseconds(2500); -class ProvenanceRepository : public core::repository::RocksDbRepository { +class RocksDbProvenanceRepository : public core::repository::RocksDbRepository, public ProvenanceRepository { public: - ProvenanceRepository(std::string_view name, const utils::Identifier& /*uuid*/) - : ProvenanceRepository(name) { + RocksDbProvenanceRepository(std::string_view name, const utils::Identifier& /*uuid*/) + : RocksDbProvenanceRepository(name) { } - explicit ProvenanceRepository(std::string_view repo_name = "", + explicit RocksDbProvenanceRepository(std::string_view repo_name = "", std::string directory = PROVENANCE_DIRECTORY, std::chrono::milliseconds maxPartitionMillis = MAX_PROVENANCE_ENTRY_LIFE_TIME, int64_t maxPartitionBytes = MAX_PROVENANCE_STORAGE_SIZE, std::chrono::milliseconds purgePeriod = PROVENANCE_PURGE_PERIOD) - : RocksDbRepository(repo_name.length() > 0 ? repo_name : core::className(), - directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory::getLogger()) { + : RocksDbRepository(repo_name.length() > 0 ? repo_name : core::className(), + directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory::getLogger()) { } - ~ProvenanceRepository() override { + ~RocksDbProvenanceRepository() override { stop(); } @@ -72,17 +73,26 @@ class ProvenanceRepository : public core::repository::RocksDbRepository { // The repo is cleaned up by itself, there is no need to delete items. return true; } - bool getElements(std::vector> &records, size_t &max_size) override; void destroy(); - ProvenanceRepository(const ProvenanceRepository &parent) = delete; + RocksDbProvenanceRepository(const RocksDbProvenanceRepository &parent) = delete; - ProvenanceRepository &operator=(const ProvenanceRepository &parent) = delete; + RocksDbProvenanceRepository &operator=(const RocksDbProvenanceRepository &parent) = delete; + + std::unique_ptr cursorFromString(std::optional cursor_str) override; + + std::expected>, std::string> getEvents(size_t max_size, Cursor* cursor) override; + + std::expected appendEvents(const std::vector>& events) override; private: // Run function for the thread void run() override {}; + + std::unique_ptr internal_state_db_; + std::mutex next_event_id_mtx_; + utils::Identifier next_event_id_; }; } // namespace org::apache::nifi::minifi::provenance diff --git a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp index 8f6677c354..154a9961d8 100644 --- a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp +++ b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp @@ -21,7 +21,7 @@ #include #include -#include "ProvenanceRepository.h" +#include "RocksDbProvenanceRepository.h" #include "unit/TestBase.h" #include "unit/Catch.h" @@ -61,13 +61,24 @@ void verifyMaxKeyCount(const minifi::provenance::ProvenanceRepository& repo, uin REQUIRE(k < keyCount); } +std::vector serializeEvent(minifi::provenance::ProvenanceEventRecord& event) { + minifi::io::BufferStream stream; + event.serialize(stream); + return stream.moveBuffer(); +} + +template +void appendAll(std::vector& sink, const std::vector& source) { + sink.insert(sink.end(), source.begin(), source.end()); +} + TEST_CASE("Test size limit", "[sizeLimitTest]") { TestController testController; auto temp_dir = testController.createTempDirectory(); REQUIRE(!temp_dir.empty()); // 60 sec, 100 KB - going to exceed the size limit - minifi::provenance::ProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1min, TEST_PROVENANCE_STORAGE_SIZE, 1s); + minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1min, TEST_PROVENANCE_STORAGE_SIZE, 1s); auto configuration = std::make_shared(); configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); @@ -87,7 +98,7 @@ TEST_CASE("Test time limit", "[timeLimitTest]") { REQUIRE(!temp_dir.empty()); // 1 sec, 100 MB - going to exceed TTL - minifi::provenance::ProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); auto configuration = std::make_shared(); configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); @@ -114,3 +125,128 @@ TEST_CASE("Test time limit", "[timeLimitTest]") { verifyMaxKeyCount(provdb, 400); } + +TEST_CASE("Test query elements after cursor", "[iterationTest]") { + TestController testController; + auto temp_dir = testController.createTempDirectory(); + REQUIRE(!temp_dir.empty()); + + minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + auto configuration = std::make_shared(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb.initialize(configuration)); + + std::vector> events; + for (size_t i = 0; i < 8; ++i) { + events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + } + + REQUIRE(provdb.appendEvents(events)); + + auto cursor = provdb.cursorFromString(std::nullopt); + REQUIRE(cursor); + + std::vector> queried_events; + + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 3); + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 6); + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 8); + + std::string last_event_id; + + for (size_t i = 0; i < queried_events.size(); ++i) { + REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()}); + last_event_id = queried_events.at(i)->getUUIDStr(); + REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i))); + } +} + +TEST_CASE("Test loading cursor from string", "[cursorSerializationTest]") { + TestController testController; + auto temp_dir = testController.createTempDirectory(); + REQUIRE(!temp_dir.empty()); + + minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + auto configuration = std::make_shared(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb.initialize(configuration)); + + std::vector> events; + for (size_t i = 0; i < 8; ++i) { + events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + } + + REQUIRE(provdb.appendEvents(events)); + + auto cursor = provdb.cursorFromString(std::nullopt); + REQUIRE(cursor); + + std::vector> queried_events; + + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 3); + + cursor = provdb.cursorFromString(cursor->toString()); + REQUIRE(cursor); + + appendAll(queried_events, provdb.getEvents(3, cursor.get()).value()); + REQUIRE(queried_events.size() == 6); + + std::string last_event_id; + + for (size_t i = 0; i < queried_events.size(); ++i) { + REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()}); + last_event_id = queried_events.at(i)->getUUIDStr(); + REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i))); + } +} + +TEST_CASE("Test opening existing database loads monotonic counter", "[eventUuidMonotonicTest]") { + TestController testController; + auto temp_dir = testController.createTempDirectory(); + REQUIRE(!temp_dir.empty()); + + auto provdb = std::make_unique("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + auto configuration = std::make_shared(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string()); + + REQUIRE(provdb->initialize(configuration)); + + std::vector> events; + for (size_t i = 0; i < 4; ++i) { + events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + } + + REQUIRE(provdb->appendEvents(events)); + + provdb = std::make_unique("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s); + + REQUIRE(provdb->initialize(configuration)); + + std::vector> new_events; + for (size_t i = 0; i < 4; ++i) { + new_events.push_back(minifi::provenance::ProvenanceEventRecord::create()); + events.push_back(new_events.back()); + } + + REQUIRE(provdb->appendEvents(new_events)); + + std::vector> queried_events = provdb->getEvents(8, nullptr).value(); + REQUIRE(queried_events.size() == 8); + + std::string last_event_id; + + for (size_t i = 0; i < queried_events.size(); ++i) { + REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()}); + last_event_id = queried_events.at(i)->getUUIDStr(); + REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i))); + } +} diff --git a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp index 95f1298835..bc40ebe190 100644 --- a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp +++ b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp @@ -34,13 +34,13 @@ namespace provenance = minifi::provenance; using namespace std::literals::chrono_literals; TEST_CASE("Test Provenance record create", "[Testprovenance::ProvenanceEventRecord]") { - auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "blah", "blahblah"); + auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000001").value(), "blahblah"); REQUIRE(record1->getAttributes().empty()); REQUIRE(record1->getAlternateIdentifierUri().empty()); } TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") { - auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype"); + auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000002").value(), "componenttype"); utils::Identifier eventId = record1->getEventId(); @@ -48,11 +48,11 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv record1->setDetails(smileyface); auto sample = 65555ms; - std::shared_ptr testRepository = std::make_shared(); + auto testRepository = std::make_shared(); record1->setEventDuration(sample); - testRepository->storeElement(record1); - auto record2 = std::make_shared(); + testRepository->appendEvents({record1}); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); REQUIRE(record2->getEventId() == record1->getEventId()); @@ -64,7 +64,7 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv } TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") { - auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, "componentid", "componenttype"); + auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, utils::Identifier::parse("00000000-0000-0000-0000-000000000003").value(), "componenttype"); utils::Identifier eventId = record1->getEventId(); std::shared_ptr ffr1 = std::make_shared(); ffr1->setAttribute("potato", "potatoe"); @@ -73,37 +73,36 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") { record1->addChildFlowFile(*ffr1); auto sample = 65555ms; - std::shared_ptr testRepository = std::make_shared(); + auto testRepository = std::make_shared(); record1->setEventDuration(sample); - testRepository->storeElement(record1); - auto record2 = std::make_shared(); + testRepository->appendEvents({record1}); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); REQUIRE(record1->getChildrenUuids().size() == 1); REQUIRE(record2->getChildrenUuids().size() == 1); utils::Identifier childId = record2->getChildrenUuids().at(0); REQUIRE(childId == ffr1->getUUID()); - record2->removeChildUuid(childId); - REQUIRE(record2->getChildrenUuids().empty()); } TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") { - auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype"); - - utils::Identifier eventId = record1->getEventId(); + auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000004").value(), "componenttype"); std::string smileyface = ":)"; record1->setDetails(smileyface); auto sample = 65555ms; - std::shared_ptr testRepository = std::make_shared(); + auto testRepository = std::make_shared(); testRepository->initialize(nullptr); record1->setEventDuration(sample); - testRepository->storeElement(record1); - auto record2 = std::make_shared(); + testRepository->appendEvents({record1}); + + utils::Identifier eventId = record1->getEventId(); + + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); REQUIRE(record2->getEventId() == record1->getEventId()); @@ -115,8 +114,7 @@ TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::Pro } TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[TestFlowAndProv1]") { - auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, "componentid", "componenttype"); - utils::Identifier eventId = record1->getEventId(); + auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, utils::Identifier::parse("00000000-0000-0000-0000-000000000005").value(), "componenttype"); std::shared_ptr ffr1 = std::make_shared(); ffr1->setAttribute("potato", "potatoe"); ffr1->setAttribute("tomato", "tomatoe"); @@ -124,24 +122,25 @@ TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[Test record1->addChildFlowFile(*ffr1); auto sample = 65555ms; - std::shared_ptr testRepository = std::make_shared(); + auto testRepository = std::make_shared(); testRepository->initialize(nullptr); record1->setEventDuration(sample); - testRepository->storeElement(record1); - auto record2 = std::make_shared(); + testRepository->appendEvents({record1}); + + utils::Identifier eventId = record1->getEventId(); + + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); REQUIRE(record1->getChildrenUuids().size() == 1); REQUIRE(record2->getChildrenUuids().size() == 1); utils::Identifier childId = record2->getChildrenUuids().at(0); REQUIRE(childId == ffr1->getUUID()); - record2->removeChildUuid(childId); - REQUIRE(record2->getChildrenUuids().empty()); } TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") { - auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype"); + auto record1 = std::make_shared(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000006").value(), "componenttype"); utils::Identifier eventId = record1->getEventId(); @@ -150,12 +149,12 @@ TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::Provena auto sample = 65555ms; - std::shared_ptr testRepository = core::createRepository("nooprepository"); + std::shared_ptr testRepository = utils::dynamic_unique_cast(core::createRepository("nooprepository")); testRepository->initialize(nullptr); record1->setEventDuration(sample); - REQUIRE(testRepository->storeElement(record1)); - auto record2 = std::make_shared(); + REQUIRE(testRepository->appendEvents({record1})); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == false); } diff --git a/extensions/rocksdb-repos/tests/RepoTests.cpp b/extensions/rocksdb-repos/tests/RepoTests.cpp index f3a8fd76f8..771c8f2a87 100644 --- a/extensions/rocksdb-repos/tests/RepoTests.cpp +++ b/extensions/rocksdb-repos/tests/RepoTests.cpp @@ -28,7 +28,7 @@ #include "core/RepositoryFactory.h" #include "FlowFileRecord.h" #include "FlowFileRepository.h" -#include "ProvenanceRepository.h" +#include "RocksDbProvenanceRepository.h" #include "properties/Configure.h" #include "unit/ProvenanceTestHelper.h" #include "unit/TestBase.h" @@ -280,7 +280,7 @@ TEST_CASE("Test FlowFile Restore", "[TestFFR6]") { config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string()); config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string()); - std::shared_ptr prov_repo = std::make_shared(); + auto prov_repo = std::make_shared(); auto ff_repository = std::make_shared("flowFileRepository"); std::shared_ptr content_repo = std::make_shared(); ff_repository->initialize(config); @@ -549,8 +549,8 @@ TEST_CASE("Test getting flow file repository size properties", "[TestGettingRepo expected_rocksdb_stats = true; } - SECTION("ProvenanceRepository") { - repository = std::make_shared("ff", dir.string(), 0ms, 0, 1ms); + SECTION("RocksDbProvenanceRepository") { + repository = std::make_shared("ff", dir.string(), 0ms, 0, 1ms); expected_rocksdb_stats = true; } @@ -707,8 +707,8 @@ TEST_CASE("Flow file repositories can be stopped", "[TestRepoIsRunning]") { repository = std::make_shared("ff", dir.string(), 0ms, 0, 1ms); } - SECTION("ProvenanceRepository") { - repository = std::make_shared("ff", dir.string(), 0ms, 0, 1ms); + SECTION("RocksDbProvenanceRepository") { + repository = std::make_shared("ff", dir.string(), 0ms, 0, 1ms); } SECTION("VolatileProvenanceRepository") { diff --git a/extensions/standard-processors/processors/LogAttribute.cpp b/extensions/standard-processors/processors/LogAttribute.cpp index 66cd7c040a..4d8170569a 100644 --- a/extensions/standard-processors/processors/LogAttribute.cpp +++ b/extensions/standard-processors/processors/LogAttribute.cpp @@ -82,7 +82,7 @@ std::string LogAttribute::generateLogMessage(core::ProcessSession& session, cons message << "\nStandard FlowFile Attributes"; message << "\n" << "UUID:" << flow_file->getUUIDStr(); message << "\n" << "EntryDate:" << utils::timeutils::getTimeStr(flow_file->getEntryDate()); - message << "\n" << "lineageStartDate:" << utils::timeutils::getTimeStr(flow_file->getlineageStartDate()); + message << "\n" << "lineageStartDate:" << utils::timeutils::getTimeStr(flow_file->getLineageStartDate()); message << "\n" << "Size:" << flow_file->getSize() << " Offset:" << flow_file->getOffset(); message << "\nFlowFile Attributes Map Content"; for (const auto& [attr_key, attr_value] : flow_file->getAttributes()) { diff --git a/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h b/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h index 0f6c74a032..499e5e41b3 100644 --- a/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h +++ b/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h @@ -85,7 +85,7 @@ class VerifyInvokeHTTP : public HTTPIntegrationBase { virtual void setupFlow() { testSetup(); - std::shared_ptr test_repo = std::make_shared(); + auto test_repo = std::make_shared(); std::shared_ptr test_flow_repo = std::make_shared(); if (flow_config_path_.config_path) { diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp b/extensions/standard-processors/tests/unit/ProcessorTests.cpp index aa0e048ca5..3f3c861b0f 100644 --- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp +++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp @@ -444,18 +444,18 @@ TEST_CASE("Test Find file", "[getfileCreate3]") { REQUIRE(2 == repo->getRepoMap().size()); for (auto entry : repo->getRepoMap()) { - minifi::provenance::ProvenanceEventRecordImpl newRecord; + auto newRecord = minifi::provenance::ProvenanceEventRecordImpl::create(); minifi::io::BufferStream stream(std::as_bytes(std::span(entry.second))); - newRecord.deserialize(stream); + newRecord->deserialize(stream); bool found = false; for (const auto& provRec : records) { - if (provRec->getEventId() == newRecord.getEventId()) { - REQUIRE(provRec->getEventId() == newRecord.getEventId()); - REQUIRE(provRec->getComponentId() == newRecord.getComponentId()); - REQUIRE(provRec->getComponentType() == newRecord.getComponentType()); - REQUIRE(provRec->getDetails() == newRecord.getDetails()); - REQUIRE(provRec->getEventDuration() == newRecord.getEventDuration()); + if (provRec->getEventId() == newRecord->getEventId()) { + REQUIRE(provRec->getEventId() == newRecord->getEventId()); + REQUIRE(provRec->getComponentId() == newRecord->getComponentId()); + REQUIRE(provRec->getComponentType() == newRecord->getComponentType()); + REQUIRE(provRec->getDetails() == newRecord->getDetails()); + REQUIRE(provRec->getEventDuration() == newRecord->getEventDuration()); found = true; break; } @@ -466,19 +466,16 @@ TEST_CASE("Test Find file", "[getfileCreate3]") { } auto taskReport = &processorReport.get(); taskReport->setBatchSize(1); - std::vector> recordsReport; - recordsReport.push_back(std::make_shared()); processorReport->incrementActiveTasks(); processorReport->setScheduledState(core::ScheduledState::RUNNING); - std::string jsonStr; - std::size_t deserialized = 0; - repo->getElements(recordsReport, deserialized); + auto recordsReport = repo->getEvents(1, nullptr); + REQUIRE(recordsReport); std::function &, const std::shared_ptr&)> verifyReporter = [&](const std::shared_ptr &context, const std::shared_ptr &session) { - taskReport->getJsonReport(*context, *session, recordsReport, jsonStr); - 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(jsonStr.find("\"componentType\": \"getfileCreate2\"") != std::string::npos); + REQUIRE(json_str.find("\"componentType\": \"getfileCreate2\"") != std::string::npos); }; testController.runSession(plan, false, verifyReporter); diff --git a/libminifi/include/CronDrivenSchedulingAgent.h b/libminifi/include/CronDrivenSchedulingAgent.h index d08ac5a4f0..46b0d68c00 100644 --- a/libminifi/include/CronDrivenSchedulingAgent.h +++ b/libminifi/include/CronDrivenSchedulingAgent.h @@ -37,7 +37,7 @@ namespace org::apache::nifi::minifi { class CronDrivenSchedulingAgent : public ThreadedSchedulingAgent { public: CronDrivenSchedulingAgent(const gsl::not_null controller_service_provider, - std::shared_ptr repo, + std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr configuration, diff --git a/libminifi/include/EventDrivenSchedulingAgent.h b/libminifi/include/EventDrivenSchedulingAgent.h index 060fff5887..d5ccd54b2e 100644 --- a/libminifi/include/EventDrivenSchedulingAgent.h +++ b/libminifi/include/EventDrivenSchedulingAgent.h @@ -35,7 +35,7 @@ namespace org::apache::nifi::minifi { class EventDrivenSchedulingAgent : public ThreadedSchedulingAgent { public: - EventDrivenSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, + EventDrivenSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr configuration, utils::ThreadPool &thread_pool) : ThreadedSchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configuration, thread_pool) { diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h index 00ff3f756d..f3f654c1a2 100644 --- a/libminifi/include/FlowController.h +++ b/libminifi/include/FlowController.h @@ -66,7 +66,7 @@ class ProcessorController; class FlowController : public core::controller::ForwardingControllerServiceProvider, public state::StateMonitor { public: - FlowController(std::shared_ptr provenance_repo, std::shared_ptr flow_file_repo, + FlowController(std::shared_ptr provenance_repo, std::shared_ptr flow_file_repo, std::shared_ptr configure, std::shared_ptr flow_configuration, std::shared_ptr content_repo, std::unique_ptr metrics_publisher_store = nullptr, std::shared_ptr filesystem = std::make_shared(), std::function request_restart = []{}, @@ -74,7 +74,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi ~FlowController() override; - virtual std::shared_ptr getProvenanceRepository() { + virtual std::shared_ptr getProvenanceRepository() { return this->provenance_repo_; } @@ -187,7 +187,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi // Thread pool for schedulers utils::ThreadPool thread_pool_; std::shared_ptr configuration_; - std::shared_ptr provenance_repo_; + std::shared_ptr provenance_repo_; std::shared_ptr flow_file_repo_; std::shared_ptr content_repo_; std::shared_ptr flow_configuration_; diff --git a/libminifi/include/SchedulingAgent.h b/libminifi/include/SchedulingAgent.h index d5cab7af55..03e5111bda 100644 --- a/libminifi/include/SchedulingAgent.h +++ b/libminifi/include/SchedulingAgent.h @@ -50,7 +50,7 @@ namespace org::apache::nifi::minifi { class SchedulingAgent { public: - SchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, + SchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr configuration, utils::ThreadPool& thread_pool) : admin_yield_duration_(), bored_yield_duration_(0), @@ -122,7 +122,7 @@ class SchedulingAgent { std::shared_ptr configure_; - std::shared_ptr repo_; + std::shared_ptr repo_; std::shared_ptr flow_repo_; diff --git a/libminifi/include/ThreadedSchedulingAgent.h b/libminifi/include/ThreadedSchedulingAgent.h index 7a3ad2299e..2668bab0d4 100644 --- a/libminifi/include/ThreadedSchedulingAgent.h +++ b/libminifi/include/ThreadedSchedulingAgent.h @@ -38,7 +38,7 @@ namespace org::apache::nifi::minifi { */ class ThreadedSchedulingAgent : public SchedulingAgent { public: - ThreadedSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, + ThreadedSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr configuration, utils::ThreadPool &thread_pool) : SchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configuration, thread_pool) { diff --git a/libminifi/include/TimerDrivenSchedulingAgent.h b/libminifi/include/TimerDrivenSchedulingAgent.h index 99073ce997..055a81eb78 100644 --- a/libminifi/include/TimerDrivenSchedulingAgent.h +++ b/libminifi/include/TimerDrivenSchedulingAgent.h @@ -30,7 +30,7 @@ namespace org::apache::nifi::minifi { class TimerDrivenSchedulingAgent : public ThreadedSchedulingAgent { public: - TimerDrivenSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, + TimerDrivenSchedulingAgent(const gsl::not_null controller_service_provider, std::shared_ptr repo, std::shared_ptr flow_repo, std::shared_ptr content_repo, std::shared_ptr configure, utils::ThreadPool &thread_pool) : ThreadedSchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configure, thread_pool) { diff --git a/libminifi/include/core/FlowFile.h b/libminifi/include/core/FlowFile.h index 72f1aea2f1..931bb10559 100644 --- a/libminifi/include/core/FlowFile.h +++ b/libminifi/include/core/FlowFile.h @@ -115,7 +115,7 @@ class FlowFileImpl : public CoreComponentImpl, public ReferenceContainerImpl, pu * Get lineage start date * @return lineage start date uint64_t */ - [[nodiscard]] std::chrono::system_clock::time_point getlineageStartDate() const override; + [[nodiscard]] std::chrono::system_clock::time_point getLineageStartDate() const override; /** * Sets the lineage start date diff --git a/libminifi/include/core/ProcessContextImpl.h b/libminifi/include/core/ProcessContextImpl.h index 9466e69dfb..0f761013d3 100644 --- a/libminifi/include/core/ProcessContextImpl.h +++ b/libminifi/include/core/ProcessContextImpl.h @@ -50,11 +50,11 @@ class Processor; class ProcessContextImpl : public core::VariableRegistryImpl, public virtual ProcessContext { public: ProcessContextImpl(Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, - const std::shared_ptr& repo, const std::shared_ptr& flow_repo, + const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& content_repo = repository::createFileSystemRepository()); ProcessContextImpl(Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, - const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& configuration, + const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& configuration, const std::shared_ptr& content_repo = repository::createFileSystemRepository()); // Get Processor associated with the Process Context @@ -85,7 +85,7 @@ class ProcessContextImpl : public core::VariableRegistryImpl, public virtual Pro void yield() override; - std::shared_ptr getProvenanceRepository() override { return repo_; } + std::shared_ptr getProvenanceRepository() override { return repo_; } /** * Returns a reference to the content repository for the running instance. @@ -198,7 +198,7 @@ class ProcessContextImpl : public core::VariableRegistryImpl, public virtual Pro std::shared_ptr logger_; controller::ControllerServiceProvider* controller_service_provider_; std::shared_ptr state_storage_; - std::shared_ptr repo_; + std::shared_ptr repo_; std::shared_ptr flow_repo_; std::shared_ptr content_repo_; Processor& processor_; diff --git a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h index 182e6f15d3..b53a78013a 100644 --- a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h +++ b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h @@ -45,9 +45,8 @@ class SiteToSiteProvenanceReportingTask : public minifi::RemoteProcessGroupPort static constexpr char const* ReportTaskName = "SiteToSiteProvenanceReportingTask"; static const char *ProvenanceAppStr; - static void getJsonReport(core::ProcessContext& context, core::ProcessSession& session, std::vector> &records, std::string &report); // NOLINT + static std::string getJsonReport(core::ProcessContext& context, core::ProcessSession& session, const std::vector> &records); // NOLINT - void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& session_factory) override; void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override; void initialize() override; diff --git a/libminifi/include/core/repository/NoOpThreadedRepository.h b/libminifi/include/core/repository/NoOpThreadedRepository.h index df23a1b32e..06bc01f37d 100644 --- a/libminifi/include/core/repository/NoOpThreadedRepository.h +++ b/libminifi/include/core/repository/NoOpThreadedRepository.h @@ -21,10 +21,11 @@ #include #include "core/ThreadedRepository.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" namespace org::apache::nifi::minifi::core::repository { -class NoOpThreadedRepository : public core::ThreadedRepositoryImpl { +class NoOpThreadedRepository : public core::ThreadedRepositoryImpl, public provenance::ProvenanceRepository { public: explicit NoOpThreadedRepository(std::string_view repo_name) : ThreadedRepositoryImpl(repo_name) { @@ -47,6 +48,21 @@ class NoOpThreadedRepository : public core::ThreadedRepositoryImpl { return 0; } + std::expected appendEvents(const std::vector>& /*events*/) override { + return {}; + } + + std::unique_ptr cursorFromString(std::optional /*cursor_str*/) override { + return nullptr; + } + + std::expected>, std::string> getEvents(size_t /*max_size*/, Cursor* cursor) override { + if (cursor) { + return std::unexpected{"Cursor based query is not supported"}; + } + return {}; + } + private: void run() override { } diff --git a/libminifi/include/core/repository/VolatileProvenanceRepository.h b/libminifi/include/core/repository/VolatileProvenanceRepository.h index da5df86fc9..278f0b8f15 100644 --- a/libminifi/include/core/repository/VolatileProvenanceRepository.h +++ b/libminifi/include/core/repository/VolatileProvenanceRepository.h @@ -21,10 +21,11 @@ #include #include "VolatileRepository.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" namespace org::apache::nifi::minifi::core::repository { -class VolatileProvenanceRepository : public VolatileRepository { +class VolatileProvenanceRepository : public VolatileRepository, public provenance::ProvenanceRepository { public: explicit VolatileProvenanceRepository(std::string_view repo_name = "", std::string /*dir*/ = REPOSITORY_DIRECTORY, @@ -38,6 +39,36 @@ class VolatileProvenanceRepository : public VolatileRepository { stop(); } + bool initialize(const std::shared_ptr &configure) override { + if (!VolatileRepository::initialize(configure)) { + return false; + } + next_event_id_ = utils::IdGenerator::getIdGenerator()->generate(); + return true; + } + + std::expected appendEvents(const std::vector>& events) override { + std::vector>> data; + data.reserve(events.size()); + std::lock_guard guard(next_event_id_mtx_); + for (auto& event : events) { + event->setUUID(next_event_id_++); + data.emplace_back(event->getUUIDStr(), std::make_unique()); + event->serialize(*data.back().second); + } + MultiPut(data); + + return {}; + } + + std::unique_ptr cursorFromString(std::optional /*cursor_str*/) override { + return nullptr; + } + + std::expected>, std::string> getEvents(size_t /*max_size*/, Cursor* /*cursor*/) override { + return std::unexpected{"Querying events is not yet supported"}; + } + private: void run() override { } @@ -51,6 +82,8 @@ class VolatileProvenanceRepository : public VolatileRepository { } std::thread thread_; + std::mutex next_event_id_mtx_; + utils::Identifier next_event_id_; }; } // namespace org::apache::nifi::minifi::core::repository diff --git a/libminifi/include/provenance/Provenance.h b/libminifi/include/provenance/Provenance.h index 37e24f6809..91d4b91dd0 100644 --- a/libminifi/include/provenance/Provenance.h +++ b/libminifi/include/provenance/Provenance.h @@ -38,6 +38,7 @@ #include "utils/Id.h" #include "utils/TimeUtil.h" #include "minifi-cpp/provenance/Provenance.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" namespace org::apache::nifi::minifi::provenance { @@ -45,15 +46,23 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public public: static const char *ProvenanceEventTypeStr[REPLAY + 1]; - ProvenanceEventRecordImpl(ProvenanceEventType event, std::string componentId, std::string componentType); + ProvenanceEventRecordImpl(ProvenanceEventType event, utils::Identifier component_id, std::string component_type); - ProvenanceEventRecordImpl() - : core::SerializableComponentImpl(core::className()) { - _eventTime = std::chrono::system_clock::now(); - } + ProvenanceEventRecordImpl(const ProvenanceEventRecordImpl&) = delete; + ProvenanceEventRecordImpl(ProvenanceEventRecordImpl&&) = delete; + ProvenanceEventRecordImpl& operator=(const ProvenanceEventRecordImpl&) = delete; + ProvenanceEventRecordImpl& operator=(ProvenanceEventRecordImpl&&) = delete; ~ProvenanceEventRecordImpl() override = default; + std::optional getEventOrdinal() const override { + return event_ordinal_; + } + + void setEventOrdinal(uint64_t event_ordinal) override { + event_ordinal_ = event_ordinal; + } + utils::Identifier getEventId() const override { return getUUID(); } @@ -63,47 +72,47 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public } std::map getAttributes() const override { - return _attributes; + return attributes_; } uint64_t getFileSize() const override { - return _size; + return size_; } uint64_t getFileOffset() const override { - return _offset; + return offset_; } std::chrono::system_clock::time_point getFlowFileEntryDate() const override { - return _entryDate; + return entry_date_; } - std::chrono::system_clock::time_point getlineageStartDate() const override { - return _lineageStartDate; + std::chrono::system_clock::time_point getLineageStartDate() const override { + return lineage_start_date_; } std::chrono::system_clock::time_point getEventTime() const override { - return _eventTime; + return event_time_; } std::chrono::milliseconds getEventDuration() const override { - return _eventDuration; + return event_duration_; } void setEventDuration(std::chrono::milliseconds duration) override { - _eventDuration = duration; + event_duration_ = duration; } ProvenanceEventType getEventType() const override { - return _eventType; + return event_type_; } std::string getComponentId() const override { - return _componentId; + return component_id_; } std::string getComponentType() const override { - return _componentType; + return component_type_; } utils::Identifier getFlowFileUuid() const override { @@ -111,69 +120,61 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public } std::string getContentFullPath() const override { - return _contentFullPath; + return content_full_path; } std::vector getLineageIdentifiers() const override { - return _lineageIdentifiers; + return lineage_identifiers; } std::string getDetails() const override { - return _details; + return details_; } void setDetails(const std::string& details) override { - _details = details; + details_ = details; } std::string getTransitUri() override { - return _transitUri; + return transit_uri_; } void setTransitUri(const std::string& uri) override { - _transitUri = uri; + transit_uri_ = uri; } std::string getSourceSystemFlowFileIdentifier() const override { - return _sourceSystemFlowFileIdentifier; + return source_system_flow_file_identifier_; } void setSourceSystemFlowFileIdentifier(const std::string& identifier) override { - _sourceSystemFlowFileIdentifier = identifier; + source_system_flow_file_identifier_ = identifier; } std::vector getParentUuids() const override { - return _parentUuids; + return parent_uuids_; } void addParentUuid(const utils::Identifier& uuid) override { - if (std::find(_parentUuids.begin(), _parentUuids.end(), uuid) != _parentUuids.end()) + if (std::find(parent_uuids_.begin(), parent_uuids_.end(), uuid) != parent_uuids_.end()) return; else - _parentUuids.push_back(uuid); + parent_uuids_.push_back(uuid); } void addParentFlowFile(const core::FlowFile& flow_file) override { addParentUuid(flow_file.getUUID()); } - void removeParentUuid(const utils::Identifier& uuid) override { - _parentUuids.erase(std::remove(_parentUuids.begin(), _parentUuids.end(), uuid), _parentUuids.end()); - } - - void removeParentFlowFile(const core::FlowFile& flow_file) override { - removeParentUuid(flow_file.getUUID()); - } - std::vector getChildrenUuids() const override { - return _childrenUuids; + return children_uuids_; } void addChildUuid(const utils::Identifier& uuid) override { - if (std::find(_childrenUuids.begin(), _childrenUuids.end(), uuid) != _childrenUuids.end()) + if (std::find(children_uuids_.begin(), children_uuids_.end(), uuid) != children_uuids_.end()) return; else - _childrenUuids.push_back(uuid); + children_uuids_.push_back(uuid); } void addChildFlowFile(const core::FlowFile& flow_file) override { @@ -181,50 +182,42 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public return; } - void removeChildUuid(const utils::Identifier& uuid) override { - _childrenUuids.erase(std::remove(_childrenUuids.begin(), _childrenUuids.end(), uuid), _childrenUuids.end()); - } - - void removeChildFlowFile(const core::FlowFile& flow_file) override { - removeChildUuid(flow_file.getUUID()); - } - std::string getAlternateIdentifierUri() const override { - return _alternateIdentifierUri; + return alternate_identifier_uri_; } void setAlternateIdentifierUri(const std::string& uri) override { - _alternateIdentifierUri = uri; + alternate_identifier_uri_ = uri; } std::string getRelationship() const override { - return _relationship; + return relationship_; } void setRelationship(const std::string& relation) override { - _relationship = relation; + relationship_ = relation; } std::string getSourceQueueIdentifier() const override { - return _sourceQueueIdentifier; + return source_queue_identifier_; } void setSourceQueueIdentifier(const std::string& identifier) override { - _sourceQueueIdentifier = identifier; + source_queue_identifier_ = identifier; } void fromFlowFile(const core::FlowFile& flow_file) override { - _entryDate = flow_file.getEntryDate(); - _lineageStartDate = flow_file.getlineageStartDate(); - _lineageIdentifiers = flow_file.getlineageIdentifiers(); + entry_date_ = flow_file.getEntryDate(); + lineage_start_date_ = flow_file.getLineageStartDate(); + lineage_identifiers = flow_file.getlineageIdentifiers(); flow_uuid_ = flow_file.getUUID(); - _attributes = flow_file.getAttributes(); - _size = flow_file.getSize(); - _offset = flow_file.getOffset(); + attributes_ = flow_file.getAttributes(); + size_ = flow_file.getSize(); + offset_ = flow_file.getOffset(); if (flow_file.getConnection()) - _sourceQueueIdentifier = flow_file.getConnection()->getName(); + source_queue_identifier_ = flow_file.getConnection()->getName(); if (flow_file.getResourceClaim()) { - _contentFullPath = flow_file.getResourceClaim()->getContentFullPath(); + content_full_path = flow_file.getResourceClaim()->getContentFullPath(); } } @@ -233,69 +226,67 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public bool loadFromRepository(const std::shared_ptr &repo) override; protected: - ProvenanceEventType _eventType; + ProvenanceEventType event_type_; + // the index of the event + std::optional event_ordinal_; // Date at which the event was created - std::chrono::system_clock::time_point _eventTime{}; + std::chrono::system_clock::time_point event_time_{}; // Date at which the flow file entered the flow - std::chrono::system_clock::time_point _entryDate{}; + std::chrono::system_clock::time_point entry_date_{}; // Date at which the origin of this flow file entered the flow - std::chrono::system_clock::time_point _lineageStartDate{}; - std::chrono::milliseconds _eventDuration{}; - std::string _componentId; - std::string _componentType; + std::chrono::system_clock::time_point lineage_start_date_{}; + std::chrono::milliseconds event_duration_{}; + std::string component_id_; + std::string component_type_; // Size in bytes of the data corresponding to this flow file - uint64_t _size = 0; + uint64_t size_ = 0; utils::Identifier flow_uuid_; - uint64_t _offset = 0; - std::string _contentFullPath; - std::map _attributes; + uint64_t offset_ = 0; + std::string content_full_path; + std::map attributes_; // UUID string for all parents - std::vector _lineageIdentifiers; - std::string _transitUri; - std::string _sourceSystemFlowFileIdentifier; - std::vector _parentUuids; - std::vector _childrenUuids; - std::string _details; - std::string _sourceQueueIdentifier; - std::string _relationship; - std::string _alternateIdentifierUri; + std::vector lineage_identifiers; + std::string transit_uri_; + std::string source_system_flow_file_identifier_; + std::vector parent_uuids_; + std::vector children_uuids_; + std::string details_; + std::string source_queue_identifier_; + std::string relationship_; + std::string alternate_identifier_uri_; private: - ProvenanceEventRecordImpl(const ProvenanceEventRecordImpl &parent); - ProvenanceEventRecordImpl &operator=(const ProvenanceEventRecordImpl &parent); static std::shared_ptr logger_; static std::shared_ptr id_generator_; }; class ProvenanceReporterImpl : public virtual ProvenanceReporter { public: - ProvenanceReporterImpl(std::shared_ptr repo, std::string componentId, std::string componentType) - : logger_(core::logging::LoggerFactory::getLogger()) { - _componentId = componentId; - _componentType = componentType; - repo_ = repo; - } + ProvenanceReporterImpl(std::shared_ptr repo, utils::Identifier component_id, std::string component_type) + : component_id_(component_id), + component_type_(std::move(component_type)), + logger_(core::logging::LoggerFactory::getLogger()), + repo_(std::move(repo)) {} + + ProvenanceReporterImpl(const ProvenanceReporterImpl&) = delete; + ProvenanceReporterImpl(ProvenanceReporterImpl&&) = delete; + ProvenanceReporterImpl& operator=(const ProvenanceReporterImpl&) = delete; + ProvenanceReporterImpl& operator=(ProvenanceReporterImpl&&) = delete; ~ProvenanceReporterImpl() override { clear(); } - std::set> getEvents() const override { - return _events; + std::vector> getEvents() const override { + return events_; } void add(const std::shared_ptr &event) override { - _events.insert(event); + events_.push_back(event); } - void remove(const std::shared_ptr &event) override { - if (_events.find(event) != _events.end()) { - _events.erase(event); - } - } - - void clear() override { - _events.clear(); + void clear() final { + events_.clear(); } void commit() override; @@ -317,23 +308,20 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter { return nullptr; } - auto event = std::make_shared(eventType, _componentId, _componentType); + auto event = std::make_shared(eventType, component_id_, component_type_); if (event) event->fromFlowFile(flow_file); return event; } - std::string _componentId; - std::string _componentType; + utils::Identifier component_id_; + std::string component_type_; private: std::shared_ptr logger_; - std::set> _events; - std::shared_ptr repo_; - - ProvenanceReporterImpl(const ProvenanceReporterImpl &parent); - ProvenanceReporterImpl &operator=(const ProvenanceReporterImpl &parent); + std::vector> events_; + std::shared_ptr repo_; }; } // namespace org::apache::nifi::minifi::provenance diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp index 916299805b..107c44b37f 100644 --- a/libminifi/src/FlowController.cpp +++ b/libminifi/src/FlowController.cpp @@ -46,7 +46,7 @@ namespace org::apache::nifi::minifi { -FlowController::FlowController(std::shared_ptr provenance_repo, std::shared_ptr flow_file_repo, +FlowController::FlowController(std::shared_ptr provenance_repo, std::shared_ptr flow_file_repo, std::shared_ptr configure, std::shared_ptr flow_configuration, std::shared_ptr content_repo, std::unique_ptr metrics_publisher_store, std::shared_ptr filesystem, std::function request_restart, diff --git a/libminifi/src/core/FlowFile.cpp b/libminifi/src/core/FlowFile.cpp index 7f0819a28b..0a19365a33 100644 --- a/libminifi/src/core/FlowFile.cpp +++ b/libminifi/src/core/FlowFile.cpp @@ -131,7 +131,7 @@ std::chrono::system_clock::time_point FlowFileImpl::getEventTime() const { return event_time_; } // ! Get Lineage Start Date -std::chrono::system_clock::time_point FlowFileImpl::getlineageStartDate() const { +std::chrono::system_clock::time_point FlowFileImpl::getLineageStartDate() const { return lineage_start_date_; } diff --git a/libminifi/src/core/ProcessContextImpl.cpp b/libminifi/src/core/ProcessContextImpl.cpp index 841b74ca51..6ac0372cf9 100644 --- a/libminifi/src/core/ProcessContextImpl.cpp +++ b/libminifi/src/core/ProcessContextImpl.cpp @@ -41,7 +41,7 @@ class StandardProcessorInfo : public ProcessorInfo { } // namespace ProcessContextImpl::ProcessContextImpl( - Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, const std::shared_ptr& repo, + Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& content_repo) : VariableRegistryImpl(static_cast>(minifi::Configure::create())), logger_(logging::LoggerFactory::getLogger()), @@ -55,7 +55,7 @@ ProcessContextImpl::ProcessContextImpl( info_(std::make_unique(processor)) {} ProcessContextImpl::ProcessContextImpl( - Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, const std::shared_ptr& repo, + Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr& state_storage, const std::shared_ptr& repo, const std::shared_ptr& flow_repo, const std::shared_ptr& configuration, const std::shared_ptr& content_repo) : VariableRegistryImpl(configuration), diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp index bd59eeaaa3..31c6a6b29b 100644 --- a/libminifi/src/core/ProcessSession.cpp +++ b/libminifi/src/core/ProcessSession.cpp @@ -69,7 +69,7 @@ ProcessSessionImpl::ProcessSessionImpl(std::shared_ptr processCo stateManager_(process_context_->getStateManager()) { logger_->log_trace("ProcessSession created for {}", process_context_->getProcessor().getName()); auto repo = process_context_->getProvenanceRepository(); - provenance_report_ = std::make_shared(repo, process_context_->getProcessor().getName(), process_context_->getProcessor().getName()); + provenance_report_ = std::make_shared(repo, process_context_->getProcessor().getUUID(), process_context_->getProcessor().getName()); content_session_ = process_context_->getContentRepository()->createSession(); if (stateManager_ && !stateManager_->beginTransaction()) { @@ -113,7 +113,7 @@ std::shared_ptr ProcessSessionImpl::create(const core::FlowFile* } record->setAttribute(attribute.first, attribute.second); } - record->setLineageStartDate(parent->getlineageStartDate()); + record->setLineageStartDate(parent->getLineageStartDate()); record->setLineageIdentifiers(parent->getlineageIdentifiers()); record->getlineageIdentifiers().push_back(parent->getUUID()); } @@ -163,7 +163,7 @@ std::shared_ptr ProcessSessionImpl::cloneDuringTransfer(const co } record->setAttribute(attribute.first, attribute.second); } - record->setLineageStartDate(parent.getlineageStartDate()); + record->setLineageStartDate(parent.getLineageStartDate()); record->setLineageIdentifiers(parent.getlineageIdentifiers()); record->getlineageIdentifiers().push_back(parent.getUUID()); diff --git a/libminifi/src/core/flow/FlowSchema.cpp b/libminifi/src/core/flow/FlowSchema.cpp index 06ad9fe112..76d48f4270 100644 --- a/libminifi/src/core/flow/FlowSchema.cpp +++ b/libminifi/src/core/flow/FlowSchema.cpp @@ -135,9 +135,9 @@ FlowSchema FlowSchema::getNiFiFlowJson() { .controller_services = {"controllerServices"}, .controller_service_properties = {"properties"}, .remote_process_group = {"remoteProcessGroups"}, - .provenance_reporting = {}, - .provenance_reporting_port_uuid = {}, - .provenance_reporting_batch_size = {}, + .provenance_reporting = {"provenanceReporting"}, + .provenance_reporting_port_uuid = {"portUuid"}, + .provenance_reporting_batch_size = {"batchSize"}, .funnels = {"funnels"}, .input_ports = {"inputPorts"}, .output_ports = {"outputPorts"}, diff --git a/libminifi/src/core/flow/StructuredConfiguration.cpp b/libminifi/src/core/flow/StructuredConfiguration.cpp index e64b5a34d7..d6a94a3306 100644 --- a/libminifi/src/core/flow/StructuredConfiguration.cpp +++ b/libminifi/src/core/flow/StructuredConfiguration.cpp @@ -586,6 +586,10 @@ void StructuredConfiguration::parseProvenanceReporting(const Node& node, core::P auto report_task = createProvenanceReportTask(); + utils::Identifier task_uuid; + task_uuid = getOrGenerateId(node); + report_task->setUUID(task_uuid); + checkRequiredField(node, schema_.scheduling_strategy); auto schedulingStrategyStr = node[schema_.scheduling_strategy].getString().value(); checkRequiredField(node, schema_.scheduling_period); diff --git a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp index 363fddede1..38bc15d646 100644 --- a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp +++ b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp @@ -88,8 +88,7 @@ void appendJsonStr(const utils::SmallString& value, rapidjson::Value& parent, parent.PushBack(valueVal, alloc); } -void SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, - std::vector> &records, std::string &report) { +std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, const std::vector> &records) { rapidjson::Document array(rapidjson::kArrayType); rapidjson::Document::AllocatorType &alloc = array.GetAllocator(); @@ -106,12 +105,16 @@ void SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, cor recordJson.AddMember("timestampMillis", int64_t{std::chrono::duration_cast(record->getEventTime().time_since_epoch()).count()}, alloc); recordJson.AddMember("durationMillis", int64_t{record->getEventDuration().count()}, alloc); - recordJson.AddMember("lineageStart", int64_t{std::chrono::duration_cast(record->getlineageStartDate().time_since_epoch()).count()}, alloc); + recordJson.AddMember("lineageStart", int64_t{std::chrono::duration_cast(record->getLineageStartDate().time_since_epoch()).count()}, alloc); recordJson.AddMember("entitySize", record->getFileSize(), alloc); recordJson.AddMember("entityOffset", record->getFileOffset(), alloc); recordJson.AddMember("entityType", "org.apache.nifi.flowfile.FlowFile", alloc); + if (auto event_ordinal = record->getEventOrdinal()) { + recordJson.AddMember("eventOrdinal", event_ordinal.value(), alloc); + } + recordJson.AddMember("eventId", getStringValue(record->getEventId().to_string(), alloc), alloc); recordJson.AddMember("eventType", getStringValue(provenance::ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()], alloc), alloc); recordJson.AddMember("details", getStringValue(record->getDetails(), alloc), alloc); @@ -148,24 +151,46 @@ void SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, cor rapidjson::PrettyWriter writer(buffer); array.Accept(writer); - report = buffer.GetString(); -} - -void SiteToSiteProvenanceReportingTask::onSchedule(core::ProcessContext&, core::ProcessSessionFactory&) { + return buffer.GetString(); } void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, core::ProcessSession& session) { - logger_->log_debug("SiteToSiteProvenanceReportingTask -- onTrigger"); - std::vector> records; - logger_->log_debug("batch size {} records", batch_size_); - size_t deserialized = batch_size_; - std::shared_ptr repo = context.getProvenanceRepository(); - if (!repo->getElements(records, deserialized) && deserialized == 0) { + std::shared_ptr repo = context.getProvenanceRepository(); + if (!repo) { + throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to retrieve provenance repository"); + } + auto* state_manager = context.getStateManager(); + if (!state_manager) { + logger_->log_error("Failed to get StateManager"); + context.yield(); return; } - logger_->log_debug("Captured {} records", deserialized); - std::string jsonStr; - this->getJsonReport(context, session, records, jsonStr); + std::optional cursor_str; + { + std::unordered_map state_map; + if (state_manager->get(state_map)) { + if (auto it = state_map.find("cursor"); it != state_map.end()) { + cursor_str = it->second; + } + } + } + std::vector> records; + auto cursor = repo->cursorFromString(cursor_str); + if (cursor_str && !cursor) { + logger_->log_error("Failed to parse cursor, falling back to enumerating from the beginning"); + cursor = repo->cursorFromString(std::nullopt); + } + if (auto result = repo->getEvents(batch_size_, cursor.get())) { + records = std::move(result.value()); + } else { + throw minifi::Exception(GENERAL_EXCEPTION, "Failed to retrieve records: " + result.error()); + } + if (records.empty()) { + logger_->log_debug("No new provenance records"); + return; + } + logger_->log_debug("Captured {} records", records.size()); + std::string jsonStr = getJsonReport(context, session, records); if (jsonStr.empty()) { return; } @@ -187,8 +212,22 @@ void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, return; } - // we transfer the record, purge the record from DB - repo->Delete(records); + if (cursor) { + // no need to delete just update the state + std::unordered_map state_map; + state_map["cursor"] = cursor->toString(); + if (!state_manager->set(state_map)) { + logger_->log_error("Failed to update cursor state"); + } + } else { + // we transfer the record, purge the record from DB + std::vector> entries; + entries.reserve(records.size()); + for (const auto& record : records) { + entries.push_back(record); + } + repo->Delete(entries); + } returnProtocol(context, std::move(protocol_)); } diff --git a/libminifi/src/provenance/Provenance.cpp b/libminifi/src/provenance/Provenance.cpp index d8ee5e7239..8fa9486429 100644 --- a/libminifi/src/provenance/Provenance.cpp +++ b/libminifi/src/provenance/Provenance.cpp @@ -40,12 +40,12 @@ std::shared_ptr ProvenanceEventRecordImpl::logger_ = core 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" }; -ProvenanceEventRecordImpl::ProvenanceEventRecordImpl(ProvenanceEventRecord::ProvenanceEventType event, std::string componentId, std::string componentType) +ProvenanceEventRecordImpl::ProvenanceEventRecordImpl(ProvenanceEventRecord::ProvenanceEventType event, utils::Identifier component_id, std::string component_type) : core::SerializableComponentImpl(core::className()), - _eventType(event), - _eventTime(std::chrono::system_clock::now()), - _componentId(std::move(componentId)), - _componentType(std::move(componentType)) { + event_type_(event), + event_time_(std::chrono::system_clock::now()), + component_id_(std::move(component_id.to_string())), + component_type_(std::move(component_type)) { } bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptr &repo) { @@ -70,9 +70,9 @@ bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptrlog_debug("NiFi Provenance retrieve event {} size {} eventType {} success", getUUIDStr(), stream.size(), magic_enum::enum_name(_eventType)); + 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(_eventType)); + logger_->log_debug("NiFi Provenance retrieve event {} size {} eventType {} fail", getUUIDStr(), stream.size(), magic_enum::enum_name(event_type_)); } return ret; @@ -80,79 +80,79 @@ bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptruuid_); + const auto ret = output_stream.write(uuid_); if (ret == 0 || io::isError(ret)) { return false; } } { - uint32_t eventType = this->_eventType; + uint32_t eventType = event_type_; const auto ret = output_stream.write(eventType); if (ret != 4) { return false; } } { - uint64_t event_time_ms = std::chrono::duration_cast(_eventTime.time_since_epoch()).count(); + uint64_t event_time_ms = std::chrono::duration_cast(event_time_.time_since_epoch()).count(); const auto ret = output_stream.write(event_time_ms); if (ret != 8) { return false; } } { - uint64_t entry_date_ms = std::chrono::duration_cast(_entryDate.time_since_epoch()).count(); + uint64_t entry_date_ms = std::chrono::duration_cast(entry_date_.time_since_epoch()).count(); const auto ret = output_stream.write(entry_date_ms); if (ret != 8) { return false; } } { - uint64_t event_duration_ms = this->_eventDuration.count(); + uint64_t event_duration_ms = event_duration_.count(); const auto ret = output_stream.write(event_duration_ms); if (ret != 8) { return false; } } { - uint64_t lineage_start_date_ms = std::chrono::duration_cast(_lineageStartDate.time_since_epoch()).count(); + uint64_t lineage_start_date_ms = std::chrono::duration_cast(lineage_start_date_.time_since_epoch()).count(); const auto ret = output_stream.write(lineage_start_date_ms); if (ret != 8) { return false; } } { - const auto ret = output_stream.write(this->_componentId); + const auto ret = output_stream.write(component_id_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->_componentType); + const auto ret = output_stream.write(component_type_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->flow_uuid_); + const auto ret = output_stream.write(flow_uuid_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->_details); + const auto ret = output_stream.write(details_); if (ret == 0 || io::isError(ret)) { return false; } } // write flow attributes { - const auto numAttributes = gsl::narrow(this->_attributes.size()); + const auto numAttributes = gsl::narrow(attributes_.size()); const auto ret = output_stream.write(numAttributes); if (ret != 4) { return false; } } - for (const auto& itAttribute : _attributes) { + for (const auto& itAttribute : attributes_) { { const auto ret = output_stream.write(itAttribute.first); if (ret == 0 || io::isError(ret)) { @@ -167,71 +167,71 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) { } } { - const auto ret = output_stream.write(this->_contentFullPath); + const auto ret = output_stream.write(content_full_path); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->_size); + const auto ret = output_stream.write(size_); if (ret != 8) { return false; } } { - const auto ret = output_stream.write(this->_offset); + const auto ret = output_stream.write(offset_); if (ret != 8) { return false; } } { - const auto ret = output_stream.write(this->_sourceQueueIdentifier); + const auto ret = output_stream.write(source_queue_identifier_); if (ret == 0 || io::isError(ret)) { return false; } } - if (this->_eventType == ProvenanceEventRecord::FORK || this->_eventType == ProvenanceEventRecord::CLONE || this->_eventType == ProvenanceEventRecord::JOIN) { + if (event_type_ == ProvenanceEventRecord::FORK || event_type_ == ProvenanceEventRecord::CLONE || event_type_ == ProvenanceEventRecord::JOIN) { // write UUIDs { - const auto parent_uuids_count = gsl::narrow(this->_parentUuids.size()); + const auto parent_uuids_count = gsl::narrow(parent_uuids_.size()); const auto ret = output_stream.write(parent_uuids_count); if (ret != 4) { return false; } } - for (const auto& parentUUID : _parentUuids) { + for (const auto& parentUUID : parent_uuids_) { const auto ret = output_stream.write(parentUUID); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto children_uuids_count = gsl::narrow(this->_childrenUuids.size()); + const auto children_uuids_count = gsl::narrow(children_uuids_.size()); const auto ret = output_stream.write(children_uuids_count); if (ret != 4) { return false; } } - for (const auto& childUUID : _childrenUuids) { + for (const auto& childUUID : children_uuids_) { const auto ret = output_stream.write(childUUID); if (ret == 0 || io::isError(ret)) { return false; } } - } else if (this->_eventType == ProvenanceEventRecord::SEND || this->_eventType == ProvenanceEventRecord::FETCH) { - const auto ret = output_stream.write(this->_transitUri); + } else if (event_type_ == ProvenanceEventRecord::SEND || event_type_ == ProvenanceEventRecord::FETCH) { + const auto ret = output_stream.write(transit_uri_); if (ret == 0 || io::isError(ret)) { return false; } - } else if (this->_eventType == ProvenanceEventRecord::RECEIVE) { + } else if (event_type_ == ProvenanceEventRecord::RECEIVE) { { - const auto ret = output_stream.write(this->_transitUri); + const auto ret = output_stream.write(transit_uri_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->_sourceSystemFlowFileIdentifier); + const auto ret = output_stream.write(source_system_flow_file_identifier_); if (ret == 0 || io::isError(ret)) { return false; } @@ -258,7 +258,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { } if (auto event_type_opt = magic_enum::enum_cast(eventType)) { - _eventType = *event_type_opt; + event_type_ = *event_type_opt; } else { return false; } @@ -269,7 +269,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { if (ret != 8) { return false; } - _eventTime = std::chrono::system_clock::time_point() + std::chrono::milliseconds(event_time_in_ms); + event_time_ = std::chrono::system_clock::time_point() + std::chrono::milliseconds(event_time_in_ms); } { @@ -278,7 +278,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { if (ret != 8) { return false; } - _entryDate = std::chrono::system_clock::time_point() + std::chrono::milliseconds(entry_date_in_ms); + entry_date_ = std::chrono::system_clock::time_point() + std::chrono::milliseconds(entry_date_in_ms); } { @@ -287,7 +287,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { if (ret != 8) { return false; } - _eventDuration = std::chrono::milliseconds(event_duration_ms); + event_duration_ = std::chrono::milliseconds(event_duration_ms); } { @@ -296,32 +296,32 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { if (ret != 8) { return false; } - _lineageStartDate = std::chrono::system_clock::time_point() + std::chrono::milliseconds(lineage_start_date_in_ms); + lineage_start_date_ = std::chrono::system_clock::time_point() + std::chrono::milliseconds(lineage_start_date_in_ms); } { - const auto ret = input_stream.read(this->_componentId); + const auto ret = input_stream.read(component_id_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->_componentType); + const auto ret = input_stream.read(component_type_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->flow_uuid_); + const auto ret = input_stream.read(flow_uuid_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->_details); + const auto ret = input_stream.read(details_); if (ret == 0 || io::isError(ret)) { return false; } @@ -351,38 +351,38 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { return false; } } - this->_attributes[key] = value; + attributes_[key] = value; } { - const auto ret = input_stream.read(this->_contentFullPath); + const auto ret = input_stream.read(content_full_path); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->_size); + const auto ret = input_stream.read(size_); if (ret != 8) { return false; } } { - const auto ret = input_stream.read(this->_offset); + const auto ret = input_stream.read(offset_); if (ret != 8) { return false; } } { - const auto ret = input_stream.read(this->_sourceQueueIdentifier); + const auto ret = input_stream.read(source_queue_identifier_); if (ret == 0 || io::isError(ret)) { return false; } } - if (this->_eventType == ProvenanceEventRecord::FORK || this->_eventType == ProvenanceEventRecord::CLONE || this->_eventType == ProvenanceEventRecord::JOIN) { + if (event_type_ == ProvenanceEventRecord::FORK || event_type_ == ProvenanceEventRecord::CLONE || event_type_ == ProvenanceEventRecord::JOIN) { // read UUIDs uint32_t number = 0; { @@ -400,7 +400,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { return false; } } - this->addParentUuid(parentUUID); + addParentUuid(parentUUID); } number = 0; { @@ -417,24 +417,24 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { return false; } } - this->addChildUuid(childUUID); + addChildUuid(childUUID); } - } else if (this->_eventType == ProvenanceEventRecord::SEND || this->_eventType == ProvenanceEventRecord::FETCH) { + } else if (event_type_ == ProvenanceEventRecord::SEND || event_type_ == ProvenanceEventRecord::FETCH) { { - const auto ret = input_stream.read(this->_transitUri); + const auto ret = input_stream.read(transit_uri_); if (ret == 0 || io::isError(ret)) { return false; } } - } else if (this->_eventType == ProvenanceEventRecord::RECEIVE) { + } else if (event_type_ == ProvenanceEventRecord::RECEIVE) { { - const auto ret = input_stream.read(this->_transitUri); + const auto ret = input_stream.read(transit_uri_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->_sourceSystemFlowFileIdentifier); + const auto ret = input_stream.read(source_system_flow_file_identifier_); if (ret == 0 || io::isError(ret)) { return false; } @@ -454,15 +454,7 @@ void ProvenanceReporterImpl::commit() { return; } - std::vector>> flowData; - - for (auto& event : _events) { - auto stramptr = std::make_unique(); - event->serialize(*stramptr); - - flowData.emplace_back(event->getUUIDStr(), std::move(stramptr)); - } - repo_->MultiPut(flowData); + repo_->appendEvents(events_); } void ProvenanceReporterImpl::create(const core::FlowFile& flow_file, const std::string& detail) { @@ -509,7 +501,6 @@ void ProvenanceReporterImpl::clone(const core::FlowFile& parent, const core::Flo if (event) { event->addChildFlowFile(child); - event->addParentFlowFile(parent); add(event); } } @@ -544,7 +535,7 @@ void ProvenanceReporterImpl::send(const core::FlowFile& flow_file, const std::st add(event); } else { if (!repo_->isFull()) - repo_->storeElement(event); + repo_->appendEvents({event}); } } } @@ -577,7 +568,7 @@ void ProvenanceReporterImpl::fetch(const core::FlowFile& flow_file, const std::s } std::shared_ptr ProvenanceEventRecord::create() { - return std::make_shared(); + return std::make_shared(ProvenanceEventType::CLONE, utils::Identifier{}, ""); } } // namespace org::apache::nifi::minifi::provenance diff --git a/libminifi/test/flow-tests/SessionTests.cpp b/libminifi/test/flow-tests/SessionTests.cpp index 533e3bf567..876a22375f 100644 --- a/libminifi/test/flow-tests/SessionTests.cpp +++ b/libminifi/test/flow-tests/SessionTests.cpp @@ -62,7 +62,7 @@ TEST_CASE("Import null data") { config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string()); config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string()); - std::shared_ptr prov_repo = core::createRepository("nooprepository"); + std::shared_ptr prov_repo = utils::dynamic_unique_cast(core::createRepository("nooprepository")); std::shared_ptr ff_repository = std::make_shared("flowFileRepository"); std::shared_ptr content_repo; SECTION("VolatileContentRepository") { diff --git a/libminifi/test/integration/C2PauseResumeTest.cpp b/libminifi/test/integration/C2PauseResumeTest.cpp index b9e6efe153..39afd2bef2 100644 --- a/libminifi/test/integration/C2PauseResumeTest.cpp +++ b/libminifi/test/integration/C2PauseResumeTest.cpp @@ -122,7 +122,7 @@ TEST_CASE("C2PauseResumeTest", "[c2test]") { VerifyC2PauseResume harness{test_file_path, flow_resumed_successfully}; PauseResumeHandler responder{flow_resumed_successfully, harness.getConfiguration()}; - std::shared_ptr test_repo = std::make_shared(); + auto test_repo = std::make_shared(); std::shared_ptr test_flow_repo = std::make_shared(); std::shared_ptr configuration = std::make_shared(); configuration->set(minifi::Configure::nifi_flow_configuration_file, test_file_path.string()); diff --git a/libminifi/test/integration/ControllerServiceIntegrationTests.cpp b/libminifi/test/integration/ControllerServiceIntegrationTests.cpp index d5586ca735..99f80b4015 100644 --- a/libminifi/test/integration/ControllerServiceIntegrationTests.cpp +++ b/libminifi/test/integration/ControllerServiceIntegrationTests.cpp @@ -52,7 +52,7 @@ TEST_CASE("ControllerServiceIntegrationTests", "[controller]") { using org::apache::nifi::minifi::test::utils::verifyEventHappenedInPollTime; std::shared_ptr configuration = std::make_shared(); - std::shared_ptr test_repo = std::make_shared(); + auto test_repo = std::make_shared(); std::shared_ptr test_flow_repo = std::make_shared(); const auto test_file_path = std::filesystem::path(TEST_RESOURCES) / "TestControllerServices.yml"; diff --git a/libminifi/test/libtest/integration/IntegrationBase.cpp b/libminifi/test/libtest/integration/IntegrationBase.cpp index 8a767546a6..b70ffa0667 100644 --- a/libminifi/test/libtest/integration/IntegrationBase.cpp +++ b/libminifi/test/libtest/integration/IntegrationBase.cpp @@ -59,7 +59,7 @@ void IntegrationBase::run() { using namespace std::literals::chrono_literals; testSetup(); - std::shared_ptr test_repo = std::make_shared(); + auto test_repo = std::make_shared(); std::shared_ptr test_flow_repo = std::make_shared(); if (flow_config_path_.config_path) { diff --git a/libminifi/test/libtest/unit/ProvenanceTestHelper.h b/libminifi/test/libtest/unit/ProvenanceTestHelper.h index 0d0407a9cd..1901c707a9 100644 --- a/libminifi/test/libtest/unit/ProvenanceTestHelper.h +++ b/libminifi/test/libtest/unit/ProvenanceTestHelper.h @@ -34,13 +34,14 @@ #include "properties/Configure.h" #include "minifi-cpp/SwapManager.h" #include "io/BufferStream.h" +#include "minifi-cpp/provenance/Provenance.h" using namespace std::literals::chrono_literals; const int64_t TEST_MAX_REPOSITORY_STORAGE_SIZE = 100; template -class TestRepositoryBase : public T_BaseRepository { +class TestRepositoryBase : public T_BaseRepository, public org::apache::nifi::minifi::provenance::ProvenanceRepository { public: TestRepositoryBase() : T_BaseRepository("repo_name", "./dir", 1s, TEST_MAX_REPOSITORY_STORAGE_SIZE, 0ms) { @@ -88,19 +89,26 @@ class TestRepositoryBase : public T_BaseRepository { } } - bool getElements(std::vector> &store, size_t &max_size) override { + std::expected>, std::string> getEvents(size_t max_size, Cursor* cursor) override { + if (cursor) { + return std::unexpected{"Cursor based query is not supported"}; + } + if (max_size == 0) { + return {}; + } + std::vector> store; std::lock_guard lock{repository_results_mutex_}; - max_size = 0; for (const auto &entry : repository_results_) { - if (max_size >= store.size()) { - break; - } - const auto eventRead = store.at(max_size); + const auto eventRead = org::apache::nifi::minifi::provenance::ProvenanceEventRecord::create(); org::apache::nifi::minifi::io::BufferStream stream(std::as_bytes(std::span(entry.second))); - eventRead->deserialize(stream); - ++max_size; + if (eventRead->deserialize(stream)) { + store.push_back(std::move(eventRead)); + if (--max_size == 0) { + break; + } + } } - return true; + return store; } std::map getRepoMap() const { @@ -108,6 +116,24 @@ class TestRepositoryBase : public T_BaseRepository { return repository_results_; } + std::expected appendEvents(const std::vector>& events) override { + std::vector>> data; + data.reserve(events.size()); + for (auto& event : events) { + data.emplace_back(event->getUUIDStr(), std::make_unique()); + event->serialize(*data.back().second); + } + if (!MultiPut(data)) { + return std::unexpected{"Failed to store provenance events"}; + } + + return {}; + } + + std::unique_ptr cursorFromString(std::optional /*cursor_str*/) override { + return nullptr; + } + protected: mutable std::mutex repository_results_mutex_; std::map repository_results_; diff --git a/libminifi/test/libtest/unit/TestBase.cpp b/libminifi/test/libtest/unit/TestBase.cpp index 23a967ae9a..c5477e67ce 100644 --- a/libminifi/test/libtest/unit/TestBase.cpp +++ b/libminifi/test/libtest/unit/TestBase.cpp @@ -208,7 +208,7 @@ LogTestController::LogTestController(const std::shared_ptr content_repo, std::shared_ptr flow_repo, std::shared_ptr prov_repo, +TestPlan::TestPlan(std::shared_ptr content_repo, std::shared_ptr flow_repo, std::shared_ptr prov_repo, std::shared_ptr flow_version, std::shared_ptr configuration, const char* state_dir) : configuration_(std::move(configuration)), content_repo_(std::move(content_repo)), @@ -637,7 +637,7 @@ std::shared_ptr TestPlan::getFlowFileProducedByCurrentPr return nullptr; } -std::set> TestPlan::getProvenanceRecords() { +std::vector> TestPlan::getProvenanceRecords() { return process_sessions_.at(location)->getProvenanceReporter()->getEvents(); } diff --git a/libminifi/test/libtest/unit/TestBase.h b/libminifi/test/libtest/unit/TestBase.h index 9e55571944..b2d20cde88 100644 --- a/libminifi/test/libtest/unit/TestBase.h +++ b/libminifi/test/libtest/unit/TestBase.h @@ -243,7 +243,7 @@ class TypedProcessorWrapper { class TestPlan { public: - explicit TestPlan(std::shared_ptr content_repo, std::shared_ptr flow_repo, std::shared_ptr prov_repo, + explicit TestPlan(std::shared_ptr content_repo, std::shared_ptr flow_repo, std::shared_ptr prov_repo, std::shared_ptr flow_version, std::shared_ptr configuration, const char* state_dir); virtual ~TestPlan(); @@ -305,7 +305,7 @@ class TestPlan { bool runCurrentProcessor(); bool runCurrentProcessorUntilFlowfileIsProduced(std::chrono::milliseconds wait_duration); - std::set> getProvenanceRecords(); + std::vector> getProvenanceRecords(); std::shared_ptr getCurrentFlowFile(); std::vector getProcessorOutboundConnections(minifi::core::Processor* processor); @@ -357,7 +357,7 @@ class TestPlan { std::shared_ptr content_repo_; std::shared_ptr flow_repo_; - std::shared_ptr prov_repo_; + std::shared_ptr prov_repo_; std::shared_ptr controller_services_provider_; diff --git a/libminifi/test/libtest/unit/TestControllerWithFlow.cpp b/libminifi/test/libtest/unit/TestControllerWithFlow.cpp index 26f5408b43..4d8b6988a9 100644 --- a/libminifi/test/libtest/unit/TestControllerWithFlow.cpp +++ b/libminifi/test/libtest/unit/TestControllerWithFlow.cpp @@ -55,7 +55,7 @@ TestControllerWithFlow::TestControllerWithFlow(const char* yamlConfigContent, bo } void TestControllerWithFlow::setupFlow() { - std::shared_ptr prov_repo = std::make_shared(); + auto prov_repo = std::make_shared(); std::shared_ptr ff_repo = std::make_shared(); std::shared_ptr content_repo = std::make_shared(); diff --git a/libminifi/test/persistence-tests/PersistenceTests.cpp b/libminifi/test/persistence-tests/PersistenceTests.cpp index 11538a8f09..70642304d8 100644 --- a/libminifi/test/persistence-tests/PersistenceTests.cpp +++ b/libminifi/test/persistence-tests/PersistenceTests.cpp @@ -60,7 +60,7 @@ class TestProcessor : public minifi::core::ProcessorImpl { }; struct TestFlow{ - TestFlow(const std::shared_ptr& ff_repository, const std::shared_ptr& content_repo, const std::shared_ptr& prov_repo, + TestFlow(const std::shared_ptr& ff_repository, const std::shared_ptr& content_repo, const std::shared_ptr& prov_repo, const std::function(utils::Identifier&)>& processorGenerator, const core::Relationship& relationshipToOutput) : ff_repository(ff_repository), content_repo(content_repo), prov_repo(prov_repo) { // setup processor @@ -178,7 +178,7 @@ TEST_CASE("Processors Can Store FlowFiles", "[TestP1]") { config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string()); config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string()); - std::shared_ptr prov_repo = std::make_shared(); + auto prov_repo = std::make_shared(); auto ff_repository = std::make_shared("flowFileRepository"); std::shared_ptr content_repo = std::make_shared(); ff_repository->initialize(config); @@ -292,7 +292,7 @@ TEST_CASE("Persisted flowFiles are updated on modification", "[TestP1]") { config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string()); config->set(minifi::Configure::nifi_dbcontent_repository_purge_period, "0 s"); - std::shared_ptr prov_repo = std::make_shared(); + auto prov_repo = std::make_shared(); std::shared_ptr ff_repository = std::make_shared("flowFileRepository"); std::shared_ptr content_repo; SECTION("VolatileContentRepository") { diff --git a/libminifi/test/unit/SchedulingAgentTests.cpp b/libminifi/test/unit/SchedulingAgentTests.cpp index 0e087cf1a2..8c2f0fee4d 100644 --- a/libminifi/test/unit/SchedulingAgentTests.cpp +++ b/libminifi/test/unit/SchedulingAgentTests.cpp @@ -78,7 +78,7 @@ class SchedulingAgentTestFixture { } protected: - std::shared_ptr test_repo_ = std::make_shared(); + std::shared_ptr test_repo_ = std::make_shared(); std::shared_ptr content_repo_ = std::make_shared(); TestController test_controller_; diff --git a/minifi-api/common/include/minifi-cpp/utils/Id.h b/minifi-api/common/include/minifi-cpp/utils/Id.h index 5c435dcf92..95a6759850 100644 --- a/minifi-api/common/include/minifi-cpp/utils/Id.h +++ b/minifi-api/common/include/minifi-cpp/utils/Id.h @@ -46,6 +46,9 @@ class Identifier { return !isNil(); } + Identifier& operator++(); + Identifier operator++(int); + bool operator!=(const Identifier& other) const; bool operator==(const Identifier& other) const; bool operator<(const Identifier& other) const; diff --git a/minifi-api/include/minifi-cpp/core/FlowFile.h b/minifi-api/include/minifi-cpp/core/FlowFile.h index 811f9e8a48..951829b45f 100644 --- a/minifi-api/include/minifi-cpp/core/FlowFile.h +++ b/minifi-api/include/minifi-cpp/core/FlowFile.h @@ -54,7 +54,7 @@ class FlowFile : public virtual CoreComponent, public virtual ReferenceContainer virtual void setDeleted(bool deleted) = 0; [[nodiscard]] virtual std::chrono::system_clock::time_point getEntryDate() const = 0; [[nodiscard]] virtual std::chrono::system_clock::time_point getEventTime() const = 0; - [[nodiscard]] virtual std::chrono::system_clock::time_point getlineageStartDate() const = 0; + [[nodiscard]] virtual std::chrono::system_clock::time_point getLineageStartDate() const = 0; virtual void setLineageStartDate(std::chrono::system_clock::time_point date) = 0; virtual void setLineageIdentifiers(const std::vector& lineage_Identifiers) = 0; virtual void setAttribute(std::string_view name, std::string value) = 0; diff --git a/minifi-api/include/minifi-cpp/core/ProcessContext.h b/minifi-api/include/minifi-cpp/core/ProcessContext.h index 6fa663a7ad..dd7b086e34 100644 --- a/minifi-api/include/minifi-cpp/core/ProcessContext.h +++ b/minifi-api/include/minifi-cpp/core/ProcessContext.h @@ -29,6 +29,7 @@ #include "minifi-cpp/core/StateStorage.h" #include "minifi-cpp/core/VariableRegistry.h" #include "minifi-cpp/core/controller/ControllerServiceHandle.h" +#include "minifi-cpp/provenance/ProvenanceRepository.h" namespace org::apache::nifi::minifi::core { @@ -79,7 +80,7 @@ class ProcessContext : public virtual core::VariableRegistry, public virtual uti virtual bool isRunning() const = 0; virtual bool isAutoTerminated(Relationship relationship) const = 0; virtual uint8_t getMaxConcurrentTasks() const = 0; - virtual std::shared_ptr getProvenanceRepository() = 0; + virtual std::shared_ptr getProvenanceRepository() = 0; virtual std::shared_ptr getContentRepository() const = 0; virtual std::shared_ptr getFlowFileRepository() const = 0; diff --git a/minifi-api/include/minifi-cpp/core/Repository.h b/minifi-api/include/minifi-cpp/core/Repository.h index f2177f00c7..6158d9d510 100644 --- a/minifi-api/include/minifi-cpp/core/Repository.h +++ b/minifi-api/include/minifi-cpp/core/Repository.h @@ -54,10 +54,6 @@ class Repository : public virtual core::CoreComponent, public virtual core::Repo virtual bool Get(const std::string& /*key*/, std::string& /*value*/) = 0; - virtual bool getElements(std::vector>& /*store*/, size_t& /*max_size*/) = 0; - - virtual bool storeElement(const std::shared_ptr& element) = 0; - virtual void loadComponent(const std::shared_ptr& /*content_repo*/) = 0; virtual std::string getDirectory() const = 0; diff --git a/minifi-api/include/minifi-cpp/provenance/Provenance.h b/minifi-api/include/minifi-cpp/provenance/Provenance.h index ef5c5a9b5f..4cf5f19240 100644 --- a/minifi-api/include/minifi-cpp/provenance/Provenance.h +++ b/minifi-api/include/minifi-cpp/provenance/Provenance.h @@ -148,13 +148,15 @@ class ProvenanceEventRecord : public virtual core::SerializableComponent { ~ProvenanceEventRecord() override = default; + virtual std::optional getEventOrdinal() const = 0; + virtual void setEventOrdinal(uint64_t value) = 0; virtual utils::Identifier getEventId() const = 0; virtual void setEventId(const utils::Identifier &id) = 0; virtual std::map getAttributes() const = 0; virtual uint64_t getFileSize() const = 0; virtual uint64_t getFileOffset() const = 0; virtual std::chrono::system_clock::time_point getFlowFileEntryDate() const = 0; - virtual std::chrono::system_clock::time_point getlineageStartDate() const = 0; + virtual std::chrono::system_clock::time_point getLineageStartDate() const = 0; virtual std::chrono::system_clock::time_point getEventTime() const = 0; virtual std::chrono::milliseconds getEventDuration() const = 0; virtual void setEventDuration(std::chrono::milliseconds duration) = 0; @@ -173,13 +175,9 @@ class ProvenanceEventRecord : public virtual core::SerializableComponent { virtual std::vector getParentUuids() const = 0; virtual void addParentUuid(const utils::Identifier& uuid) = 0; virtual void addParentFlowFile(const core::FlowFile& flow_file) = 0; - virtual void removeParentUuid(const utils::Identifier& uuid) = 0; - virtual void removeParentFlowFile(const core::FlowFile& flow_file) = 0; virtual std::vector getChildrenUuids() const = 0; virtual void addChildUuid(const utils::Identifier& uuid) = 0; virtual void addChildFlowFile(const core::FlowFile& flow_file) = 0; - virtual void removeChildUuid(const utils::Identifier& uuid) = 0; - virtual void removeChildFlowFile(const core::FlowFile& flow_file) = 0; virtual std::string getAlternateIdentifierUri() const = 0; virtual void setAlternateIdentifierUri(const std::string& uri) = 0; virtual std::string getRelationship() const = 0; @@ -196,9 +194,8 @@ class ProvenanceReporter { public: virtual ~ProvenanceReporter() = default; - virtual std::set> getEvents() const = 0; + virtual std::vector> getEvents() const = 0; virtual void add(const std::shared_ptr &event) = 0; - virtual void remove(const std::shared_ptr &event) = 0; virtual void clear() = 0; virtual void commit() = 0; diff --git a/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h b/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h new file mode 100644 index 0000000000..cfebebe353 --- /dev/null +++ b/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h @@ -0,0 +1,44 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "minifi-cpp/core/Repository.h" +#include "utils/Id.h" +#include "minifi-cpp/provenance/Provenance.h" + +namespace org::apache::nifi::minifi::provenance { + +class ProvenanceRepository : public virtual core::Repository { + public: + class Cursor { + public: + [[nodiscard]] + virtual std::string toString() const = 0; + virtual ~Cursor() = default; + }; + + virtual std::expected appendEvents(const std::vector>& events) = 0; + + // if @cursor_str is nullopt it creates a cursor to the beginning if the underlying repository supports it + virtual std::unique_ptr cursorFromString(std::optional cursor_str) = 0; + + virtual std::expected>, std::string> getEvents(size_t max_size, Cursor* cursor) = 0; +}; + +} // namespace org::apache::nifi::minifi::provenance diff --git a/minifi_main/MiNiFiMain.cpp b/minifi_main/MiNiFiMain.cpp index fcc2ba9d6a..ea7e44fdcf 100644 --- a/minifi_main/MiNiFiMain.cpp +++ b/minifi_main/MiNiFiMain.cpp @@ -343,7 +343,7 @@ int main(int argc, char **argv) { configure->get(minifi::Configure::nifi_provenance_repository_class_name, prov_repo_class); // Create repos for flow record and provenance - std::shared_ptr prov_repo = core::createRepository(prov_repo_class, "provenance"); + std::shared_ptr prov_repo = utils::dynamic_unique_cast(core::createRepository(prov_repo_class, "provenance")); if (!prov_repo || !prov_repo->initialize(configure)) { logger->log_error("Provenance repository failed to initialize, exiting..");