Copilot commented on code in PR #2203: URL: https://github.com/apache/nifi-minifi-cpp/pull/2203#discussion_r3773844479
########## extensions/lmdb/LmdbWrapper.cpp: ########## @@ -0,0 +1,267 @@ +/** + * + * 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 "LmdbWrapper.h" + +#include <filesystem> + +#include "minifi-cpp/utils/gsl.h" + +namespace org::apache::nifi::minifi::extensions::lmdb { + +bool LmdbWrapper::initialize(const std::string& directory, size_t max_db_size) { + if (const auto rc = mdb_env_create(&lmdb_env_); rc != MDB_SUCCESS) { + logger_->log_error("Failed to create LMDB environment: {}", mdb_strerror(rc)); + return false; + } + + logger_->log_info("Setting LMDB max DB size to {} bytes", max_db_size); + if (const auto rc = mdb_env_set_mapsize(lmdb_env_, max_db_size); rc != MDB_SUCCESS) { + logger_->log_error("Failed to set LMDB map size: {}", mdb_strerror(rc)); + mdb_env_close(lmdb_env_); + lmdb_env_ = nullptr; + return false; + } + + if (std::filesystem::exists(directory)) { + logger_->log_info("Using existing LMDB Repository directory at {}", directory); + } else { + logger_->log_info("Creating LMDB Repository directory at {}", directory); + if (!std::filesystem::create_directories(directory)) { + logger_->log_error("Failed to create LMDB Repository directory at {}", directory); + return false; + } Review Comment: This failure branch leaves the newly created LMDB environment open and `lmdb_env_` non-null, unlike the other initialization failures below. The wrapper remains partially initialized until destruction, and a retry would overwrite the pointer and leak the environment. Close and reset the environment before returning. ########## extensions/lmdb/LmdbFlowFileRepository.cpp: ########## @@ -0,0 +1,243 @@ +/** + * 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 "LmdbFlowFileRepository.h" +#include "core/Resource.h" +#include "minifi-cpp/FlowFileRecord.h" + +using namespace std::literals::chrono_literals; + +namespace org::apache::nifi::minifi::extensions::lmdb { + +namespace { +bool getRepositoryCheckHealth(const Configure& configure) { + std::string check_health_str; + configure.get(Configure::nifi_flow_file_repository_check_health, check_health_str); + return utils::string::toBool(check_health_str).value_or(true); +} +} // namespace + +bool LmdbFlowFileRepository::initialize(const std::shared_ptr<Configure> &configure) { + std::string value; + + if (configure->get(Configure::nifi_flowfile_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + check_flowfile_content_size_ = getRepositoryCheckHealth(*configure); + logger_->log_debug("NiFi LMDB FlowFile Repository Directory {}", directory_); + + // Reserve virtual address space for the DB file (max size it can grow to) + const auto max_db_size = configure->get(Configure::nifi_flowfile_repository_lmdb_max_db_size) | utils::andThen([](auto max_db_size_str) -> std::optional<uint64_t> { + if (max_db_size_str.empty()) { return std::nullopt; } + return parsing::parseDataSize(max_db_size_str) | utils::orThrow(fmt::format("{} was set to invalid value: '{}'", Configure::nifi_flowfile_repository_lmdb_max_db_size, max_db_size_str)); + }) | utils::orElse([] { + return std::make_optional<uint64_t>(MAX_FLOWFILE_REPOSITORY_STORAGE_SIZE); + }); + + if (!max_db_size) { + logger_->log_error("Invalid max DB size configuration for LMDB FlowFile Repository"); + return false; + } + + logger_->log_info("Using LMDB FlowFile Repository directory '{}'", directory_); + return lmdb_wrapper_.initialize(directory_, *max_db_size); +} + +bool LmdbFlowFileRepository::Delete(const std::string& key) { + keys_to_delete_.enqueue({.key = key}); + return true; +} + +bool LmdbFlowFileRepository::Delete(const std::shared_ptr<core::CoreComponent>& item) { + if (auto ff = std::dynamic_pointer_cast<core::FlowFile>(item)) { + keys_to_delete_.enqueue({.key = item->getUUIDStr(), .content = ff->getResourceClaim()}); + } else { + keys_to_delete_.enqueue({.key = item->getUUIDStr()}); + } + return true; +} + +bool LmdbFlowFileRepository::Put(const std::string& key, const uint8_t* buf, size_t bufLen) { + return lmdb_wrapper_.putValue(key, std::string(reinterpret_cast<const char*>(buf), bufLen)); Review Comment: `Put` is explicitly expected to accept `(nullptr, 0)`, but constructing `std::string` from a null pointer does not satisfy the constructor's valid-range precondition even when the length is zero; standard-library debug modes may assert and the behavior is not portable. Special-case the empty value and reject a null pointer with a nonzero length. ########## minifi-api/include/minifi-cpp/properties/Configuration.h: ########## @@ -79,6 +79,7 @@ class Configuration : public virtual Properties { static constexpr const char *nifi_dbcontent_optimize_for_small_db_cache_size = "nifi.database.content.repository.optimize.for.small.db.cache.size"; static constexpr const char *nifi_content_repository_lmdb_max_db_size = "nifi.content.repository.lmdb.max.db.size"; + static constexpr const char *nifi_flowfile_repository_lmdb_max_db_size = "nifi.flowfile.repository.lmdb.max.db.size"; Review Comment: The new public backend and configuration key are not documented: `CONFIGURE.md:623` still lists only `NoOpRepository` as the FlowFile repository alternative, while its LMDB section documents only the content repository. Add the `nifi.flowfile.repository.class.name=LmdbFlowFileRepository` setup, this key's default, and the relevant LMDB sizing caveats so users can configure the feature safely. ########## extensions/lmdb/LmdbFlowFileRepository.cpp: ########## @@ -0,0 +1,243 @@ +/** + * 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 "LmdbFlowFileRepository.h" +#include "core/Resource.h" +#include "minifi-cpp/FlowFileRecord.h" + +using namespace std::literals::chrono_literals; + +namespace org::apache::nifi::minifi::extensions::lmdb { + +namespace { +bool getRepositoryCheckHealth(const Configure& configure) { + std::string check_health_str; + configure.get(Configure::nifi_flow_file_repository_check_health, check_health_str); + return utils::string::toBool(check_health_str).value_or(true); +} +} // namespace + +bool LmdbFlowFileRepository::initialize(const std::shared_ptr<Configure> &configure) { + std::string value; + + if (configure->get(Configure::nifi_flowfile_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + check_flowfile_content_size_ = getRepositoryCheckHealth(*configure); + logger_->log_debug("NiFi LMDB FlowFile Repository Directory {}", directory_); + + // Reserve virtual address space for the DB file (max size it can grow to) + const auto max_db_size = configure->get(Configure::nifi_flowfile_repository_lmdb_max_db_size) | utils::andThen([](auto max_db_size_str) -> std::optional<uint64_t> { + if (max_db_size_str.empty()) { return std::nullopt; } + return parsing::parseDataSize(max_db_size_str) | utils::orThrow(fmt::format("{} was set to invalid value: '{}'", Configure::nifi_flowfile_repository_lmdb_max_db_size, max_db_size_str)); + }) | utils::orElse([] { + return std::make_optional<uint64_t>(MAX_FLOWFILE_REPOSITORY_STORAGE_SIZE); + }); + + if (!max_db_size) { + logger_->log_error("Invalid max DB size configuration for LMDB FlowFile Repository"); + return false; + } + + logger_->log_info("Using LMDB FlowFile Repository directory '{}'", directory_); + return lmdb_wrapper_.initialize(directory_, *max_db_size); +} + +bool LmdbFlowFileRepository::Delete(const std::string& key) { + keys_to_delete_.enqueue({.key = key}); + return true; +} + +bool LmdbFlowFileRepository::Delete(const std::shared_ptr<core::CoreComponent>& item) { + if (auto ff = std::dynamic_pointer_cast<core::FlowFile>(item)) { + keys_to_delete_.enqueue({.key = item->getUUIDStr(), .content = ff->getResourceClaim()}); + } else { + keys_to_delete_.enqueue({.key = item->getUUIDStr()}); + } + return true; +} + +bool LmdbFlowFileRepository::Put(const std::string& key, const uint8_t* buf, size_t bufLen) { + return lmdb_wrapper_.putValue(key, std::string(reinterpret_cast<const char*>(buf), bufLen)); +} + +bool LmdbFlowFileRepository::MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) { + return lmdb_wrapper_.putValues(data); +} + +bool LmdbFlowFileRepository::Get(const std::string& key, std::string& value) { + auto result = lmdb_wrapper_.getValue(key); + if (result) { + value = std::move(*result); + return true; + } + return false; +} + +uint64_t LmdbFlowFileRepository::getRepositorySize() const { + const auto stat = lmdb_wrapper_.getDbStat(); + return stat.ms_psize * (stat.ms_branch_pages + stat.ms_leaf_pages + stat.ms_overflow_pages); +} + +uint64_t LmdbFlowFileRepository::getRepositoryEntryCount() const { + return lmdb_wrapper_.getDbStat().ms_entries; +} + +void LmdbFlowFileRepository::flush() { + std::list<ExpiredFlowFileInfo> flow_files; + + while (keys_to_delete_.size_approx() > 0) { + ExpiredFlowFileInfo info; + if (keys_to_delete_.try_dequeue(info)) { + flow_files.push_back(std::move(info)); + } + } + + deserializeFlowFilesWithNoContentClaim(flow_files); + + std::vector<std::string> flow_file_keys; + for (auto& ff : flow_files) { + flow_file_keys.push_back(ff.key); + logger_->log_debug("Issuing batch delete, including {}, Content path {}", ff.key, ff.content ? ff.content->getContentFullPath() : "null"); + } + + if (!lmdb_wrapper_.removeKeys(flow_file_keys)) { + for (auto&& ff : flow_files) { + keys_to_delete_.enqueue(std::move(ff)); + } + return; // Stop here - don't delete from content repo while we have records in FF repo + } + + if (content_repo_) { + for (auto& ff : flow_files) { + if (ff.content) { + ff.content->decreaseFlowFileRecordOwnedCount(); + } + } + } +} + +void LmdbFlowFileRepository::deserializeFlowFilesWithNoContentClaim(std::list<ExpiredFlowFileInfo>& flow_files) { + std::vector<std::string> keys; + std::vector<std::list<ExpiredFlowFileInfo>::iterator> key_positions; + for (auto it = flow_files.begin(); it != flow_files.end(); ++it) { + if (!it->content) { + keys.push_back(it->key); + key_positions.push_back(it); + } + } + if (keys.empty()) { + return; + } + std::vector<std::optional<std::string>> values; + values.reserve(keys.size()); + for (const auto& key : keys) { + values.push_back(lmdb_wrapper_.getValue(key)); + } + + gsl_Expects(keys.size() == values.size()); + + for (size_t i = 0; i < keys.size(); ++i) { + if (!values[i]) { + logger_->log_error("Failed to read key from LMDB: {}! DB is most probably in an inconsistent state!", keys[i].data()); + flow_files.erase(key_positions.at(i)); + continue; + } + + utils::Identifier container_id; + auto flow_file = FlowFileRecord::DeSerialize(std::as_bytes(std::span(*values[i])), content_repo_, container_id); + if (flow_file) { + gsl_Expects(flow_file->getUUIDStr() == key_positions.at(i)->key); + key_positions.at(i)->content = flow_file->getResourceClaim(); + } else { + logger_->log_error("Could not deserialize flow file {}", key_positions.at(i)->key); + } + } +} + +void LmdbFlowFileRepository::run() { + while (isRunning()) { + std::this_thread::sleep_for(purge_period_); + flush(); + } + flush(); +} + +bool LmdbFlowFileRepository::contentSizeIsAmpleForFlowFile(const core::FlowFile& flow_file_record, const std::shared_ptr<ResourceClaim>& resource_claim) const { + const auto stream_size = resource_claim ? content_repo_->size(*resource_claim) : 0; + const auto required_size = flow_file_record.getOffset() + flow_file_record.getSize(); + return stream_size >= required_size; +} + +core::Connectable* LmdbFlowFileRepository::getContainer(const std::string& container_id) { + auto container = containers_.find(container_id); + if (container != containers_.end()) + return container->second; + // for backward compatibility + container = connection_map_.find(container_id); + if (container != connection_map_.end()) + return container->second; + return nullptr; +} + +void LmdbFlowFileRepository::initialize_repository() { + gsl_Expects(content_repo_); + logger_->log_info("Reading existing flow files from database"); + + lmdb_wrapper_.forEach([this](const MDB_val& key, const MDB_val& value) { Review Comment: The result of `forEach` is ignored. If LMDB cannot start/open the cursor or iteration fails partway through, this method still reaches `content_repo_->clearOrphans()`. Because this scan is what repopulates the content repository's claim counts, content belonging to every unvisited FlowFile can then be classified as orphaned and deleted. Abort loading (and especially orphan cleanup) when `forEach` returns false. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
