Copilot commented on code in PR #2259: URL: https://github.com/apache/nifi-minifi-cpp/pull/2259#discussion_r3979233972
########## extensions/opc/src/FetchOPCHistory.cpp: ########## @@ -0,0 +1,364 @@ +/** + * 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 "FetchOPCHistory.h" + +#include <optional> +#include <string> +#include <unordered_set> + +#include "core/ProcessSession.h" +#include "core/Resource.h" +#include "minifi-cpp/core/ProcessContext.h" +#include "utils/ProcessorConfigUtils.h" +#include "utils/StringUtils.h" + +namespace org::apache::nifi::minifi::processors { + +namespace { + +constexpr const char* LAST_FETCHED_TIMESTAMP_KEY = "last_fetched_timestamp"; +constexpr const char* LAST_FETCHED_FINGERPRINT_KEY = "last_fetched_fingerprint"; + +std::string updateTypeToString(UA_HistoryUpdateType type) { + switch (type) { + case UA_HISTORYUPDATETYPE_INSERT: + return "Insert"; + case UA_HISTORYUPDATETYPE_REPLACE: + return "Replace"; + case UA_HISTORYUPDATETYPE_UPDATE: + return "Update"; + case UA_HISTORYUPDATETYPE_DELETE: + return "Delete"; + default: + return "Unknown"; + } +} + +std::string uaStringToString(const UA_String& str) { + return {reinterpret_cast<const char*>(str.data), str.length}; +} + +struct HistoryEntry { + std::string value; + int64_t source_timestamp = 0; + const UA_ModificationInfo* modification_info = nullptr; + + [[nodiscard]] int64_t modificationTime() const { + return modification_info ? modification_info->modificationTime : UA_DateTime_fromUnixTime(0); + } +}; + +struct HistoryBatch { + std::vector<HistoryEntry> entries; + bool has_modification_info = false; +}; + +std::string entryFingerprint(const HistoryEntry& entry, bool has_modification_info) { + // The fingerprint deduplicates entries sharing the boundary source timestamp across triggers. For raw value history a + // duplicated (value, source timestamp) pair represents no change in the history, so losing one to deduplication is harmless. + // For audit (modification) history the modification time and update type distinguish otherwise-identical entries, so they + // are included to make the fingerprint more unique and reduce the chance of dropping a distinct modification. + std::string raw = ":" + entry.value; + if (has_modification_info) { + const auto update_type = entry.modification_info ? updateTypeToString(entry.modification_info->updateType) : ""; + raw = std::to_string(entry.modificationTime()) + ":" + update_type + raw; + } + return utils::string::to_hex(raw); +} + +// NOLINTBEGIN(cppcoreguidelines-pro-type-union-access) +std::optional<HistoryBatch> extractHistoryBatch(const UA_ExtensionObject* data, const std::shared_ptr<core::logging::Logger>& logger) { + const UA_DataValue* data_values = nullptr; + size_t data_value_size = 0; + const UA_ModificationInfo* modification_infos = nullptr; + size_t modification_infos_size = 0; + + if (data->content.decoded.type == &UA_TYPES[UA_TYPES_HISTORYDATA]) { + const auto* history_data = static_cast<const UA_HistoryData*>(data->content.decoded.data); + data_values = history_data->dataValues; + data_value_size = history_data->dataValuesSize; + } else if (data->content.decoded.type == &UA_TYPES[UA_TYPES_HISTORYMODIFIEDDATA]) { + const auto* modified_data = static_cast<const UA_HistoryModifiedData*>(data->content.decoded.data); + data_values = modified_data->dataValues; + data_value_size = modified_data->dataValuesSize; + modification_infos = modified_data->modificationInfos; + modification_infos_size = modified_data->modificationInfosSize; + } else { + logger->log_error("Unexpected data type received in the history read callback: {}", data->content.decoded.type->typeName); + return std::nullopt; + } + + HistoryBatch batch; + batch.has_modification_info = modification_infos != nullptr; + batch.entries.reserve(data_value_size); + for (size_t i = 0; i < data_value_size; ++i) { + HistoryEntry entry; + try { + entry.value = opc::variantToString(data_values[i].value); + } catch (const opc::OPCException& ex) { + logger->log_warn("Failed to convert value at index {} to string, skipping entry: {}", i, ex.what()); + continue; + } + entry.source_timestamp = data_values[i].sourceTimestamp; + entry.modification_info = (modification_infos && i < modification_infos_size) ? &modification_infos[i] : nullptr; + batch.entries.push_back(std::move(entry)); + } + return batch; +} +// NOLINTEND(cppcoreguidelines-pro-type-union-access) + +std::vector<HistoryEntry> selectNewEntries(std::vector<HistoryEntry> entries, const std::optional<FetchedState>& last_fetched, + bool has_modification_info, std::optional<size_t> max_entries) { + std::vector<HistoryEntry> new_entries; + new_entries.reserve(max_entries ? std::min(*max_entries, entries.size()) : entries.size()); + for (auto& entry : entries) { + if (max_entries && new_entries.size() >= *max_entries) { + break; + } + if (last_fetched && entry.source_timestamp == last_fetched->timestamp && + last_fetched->fingerprints.contains(entryFingerprint(entry, has_modification_info))) { + continue; + } + new_entries.push_back(std::move(entry)); + } + return new_entries; +} + +void addModificationInfo(core::Record& record, const UA_ModificationInfo& modification_info) { + if (modification_info.userName.length > 0) { + record.emplace("ModificationUsername", core::RecordField(uaStringToString(modification_info.userName))); + } + record.emplace("ModificationTime", core::RecordField(opc::OPCDateTime2String(modification_info.modificationTime))); + record.emplace("ModificationUpdateType", core::RecordField(updateTypeToString(modification_info.updateType))); +} + +void addModificationInfo(core::FlowFile& flow_file, const UA_ModificationInfo& modification_info) { + if (modification_info.userName.length > 0) { + flow_file.addAttribute("ModificationUsername", uaStringToString(modification_info.userName)); + } + flow_file.addAttribute("ModificationTime", opc::OPCDateTime2String(modification_info.modificationTime)); + flow_file.addAttribute("ModificationUpdateType", updateTypeToString(modification_info.updateType)); +} + +core::Record toRecord(const std::string& node_id, const int32_t namespace_index, const HistoryEntry& entry) { + core::Record record; + record.emplace("Value", core::RecordField(entry.value)); + record.emplace("NodeID", core::RecordField(node_id)); + record.emplace("Namespace index", core::RecordField(std::to_string(namespace_index))); + record.emplace("Sourcetimestamp", core::RecordField(opc::OPCDateTime2String(entry.source_timestamp))); + if (entry.modification_info) { + addModificationInfo(record, *entry.modification_info); + } + return record; +} + +void writeAsRecordSet(FetchOPCHistoryContext& context, const std::vector<HistoryEntry>& entries) { + core::RecordSet record_set; + for (const auto& entry : entries) { + record_set.push_back(toRecord(context.node_id, context.namespace_index, entry)); + } + + auto flow_file = context.session.create(); + context.record_set_writer->write(record_set, flow_file, context.session); + context.session.transfer(flow_file, FetchOPCHistory::Success); + context.entries_transferred += entries.size(); +} + +void writeAsFlowFiles(FetchOPCHistoryContext& context, const std::vector<HistoryEntry>& entries) { + for (const auto& entry : entries) { + auto flow_file = context.session.create(); + context.session.write(flow_file, [&entry](const std::shared_ptr<io::OutputStream>& output_stream) -> io::IoResult { + output_stream->write(reinterpret_cast<const uint8_t*>(entry.value.data()), entry.value.size()); + return io::IoResult::from(entry.value.size()); + }); + flow_file->addAttribute("NodeID", context.node_id); + flow_file->addAttribute("Namespace index", std::to_string(context.namespace_index)); + flow_file->addAttribute("Sourcetimestamp", opc::OPCDateTime2String(entry.source_timestamp)); + if (entry.modification_info) { + addModificationInfo(*flow_file, *entry.modification_info); + } + context.session.transfer(flow_file, FetchOPCHistory::Success); + ++context.entries_transferred; + } +} + +void updateState(std::unordered_map<std::string, std::string>& state_map, const std::vector<HistoryEntry>& new_entries, bool has_modification_info, + const std::optional<FetchedState>& fetched_state) { + const int64_t new_timestamp = new_entries.back().source_timestamp; + + std::unordered_set<std::string> fingerprints; + if (fetched_state && fetched_state->timestamp == new_timestamp) { + fingerprints = fetched_state->fingerprints; + } + for (const auto& entry : new_entries) { + if (entry.source_timestamp == new_timestamp) { + fingerprints.insert(entryFingerprint(entry, has_modification_info)); + } + } + + state_map[LAST_FETCHED_TIMESTAMP_KEY] = std::to_string(new_timestamp); + state_map[LAST_FETCHED_FINGERPRINT_KEY] = utils::string::join(",", fingerprints); +} + +UA_Boolean historyReadCallback(UA_Client* /*client*/, const UA_NodeId* /*node_id*/, UA_Boolean more_data_available, const UA_ExtensionObject* data, + void* ctx) { + auto* opc_history_context = static_cast<FetchOPCHistoryContext*>(ctx); + + auto batch = extractHistoryBatch(data, opc_history_context->logger); + if (batch && !batch->entries.empty()) { + const std::optional<size_t> remaining = opc_history_context->batch_size != 0 + ? std::optional<size_t>(opc_history_context->batch_size - opc_history_context->entries_transferred) + : std::nullopt; + auto new_entries = selectNewEntries(std::move(batch->entries), opc_history_context->fetched_state, batch->has_modification_info, remaining); + + if (!new_entries.empty()) { + if (opc_history_context->record_set_writer) { + writeAsRecordSet(*opc_history_context, new_entries); + } else { + writeAsFlowFiles(*opc_history_context, new_entries); + } + updateState(opc_history_context->state_map, new_entries, batch->has_modification_info, opc_history_context->fetched_state); + } + } + + const bool batch_limit_reached = opc_history_context->batch_size != 0 && + opc_history_context->entries_transferred >= opc_history_context->batch_size; + return more_data_available && !batch_limit_reached; +} + +UA_DateTime calculateStartTime(const std::optional<FetchedState>& fetched_state, + const std::optional<std::chrono::system_clock::time_point>& start_timestamp) { + if (fetched_state && fetched_state->timestamp != 0) { + return fetched_state->timestamp; + } else if (start_timestamp.has_value()) { + uint64_t start_time_seconds = std::chrono::duration_cast<std::chrono::seconds>(start_timestamp->time_since_epoch()).count(); + return UA_DateTime_fromUnixTime(gsl::narrow<UA_Int64>(start_time_seconds)); + } + return UA_DateTime_fromUnixTime(0); +} + +UA_DateTime calculateEndTime(const std::optional<std::chrono::system_clock::time_point>& end_timestamp) { + if (end_timestamp.has_value()) { + uint64_t end_time_seconds = std::chrono::duration_cast<std::chrono::seconds>(end_timestamp->time_since_epoch()).count(); + return UA_DateTime_fromUnixTime(gsl::narrow<UA_Int64>(end_time_seconds)); + } + return UA_DateTime_now(); +} + +} // namespace + +void FetchOPCHistory::initialize() { + setSupportedProperties(Properties); + setSupportedRelationships(Relationships); +} + +void FetchOPCHistory::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& factory) { + logger_->log_trace("FetchOPCHistory::onSchedule"); + BaseOPCProcessor::onSchedule(context, factory); + node_id_ = utils::parseProperty(context, NodeID); + parseIdType(context, NodeIDType); + namespace_idx_ = gsl::narrow<int32_t>(utils::parseI64Property(context, NameSpaceIndex)); + + switch (id_type_) { + case opc::OPCNodeIDType::String: + node_ = UA_NODEID_STRING(namespace_idx_, const_cast<char*>(node_id_.c_str())); + break; + case opc::OPCNodeIDType::Int: + node_ = UA_NODEID_NUMERIC(namespace_idx_, std::stoi(node_id_)); + break; + case opc::OPCNodeIDType::Guid: + node_ = UA_NODEID_GUID(namespace_idx_, UA_GUID(node_id_.c_str())); + break; Review Comment: Same issue as in the test server: `UA_GUID(node_id_.c_str())` is not a GUID string parser in open62541. This makes the `Guid` Node ID type unusable. Parse `node_id_` into a `UA_Guid` via the proper open62541 parsing function for your version, then pass the parsed `UA_Guid` to `UA_NODEID_GUID(...)`. ########## extensions/opc/tests/FetchOPCHistoryTests.cpp: ########## @@ -0,0 +1,338 @@ +/** + * + * 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 <optional> +#include <string> + +#include "OpcUaTestServer.h" +#include "catch2/generators/catch_generators.hpp" +#include "include/FetchOPCHistory.h" +#include "unit/Catch.h" +#include "unit/SingleProcessorTestController.h" +#include "unit/TestBase.h" +#include "unit/TestUtils.h" + +namespace org::apache::nifi::minifi::test { + +class FetchOPCHistoryTestController { + public: + FetchOPCHistoryTestController() + : controller_(minifi::test::utils::make_processor<processors::FetchOPCHistory>("FetchOPCHistory")), + processor_(controller_.getProcessor()) { + LogTestController::getInstance().setDebug<TestPlan>(); + LogTestController::getInstance().setDebug<minifi::core::Processor>(); + LogTestController::getInstance().setTrace<minifi::core::ProcessSession>(); + LogTestController::getInstance().setDebug<processors::FetchOPCHistory>(); + } + + void setupProcessor(const std::string& node_id_type, const std::string& node_id) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::OPCServerEndPoint.name, "opc.tcp://127.0.0.1:4841/")); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::NodeIDType.name, node_id_type)); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::NodeID.name, node_id)); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::NameSpaceIndex.name, std::to_string(server_.getNamespaceIndex()))); + } + + void checkFlowFile(const std::shared_ptr<core::FlowFile>& flow_file, const std::string& content, const std::string& node_id, + const std::string& source_timestamp) { + CHECK(controller_.plan->getContent(flow_file) == content); + CHECK(flow_file->getAttribute("NodeID") == node_id); + CHECK(flow_file->getAttribute("Namespace index") == std::to_string(server_.getNamespaceIndex())); + CHECK(flow_file->getAttribute("Sourcetimestamp") == source_timestamp); + } + + void checkModificationAttributes(const std::shared_ptr<core::FlowFile>& flow_file, bool present, const std::string& username = "", + const std::string& update_type = "", const std::string& modification_time = "") { + if (present) { + CHECK(flow_file->getAttribute("ModificationUsername") == username); + CHECK(flow_file->getAttribute("ModificationUpdateType") == update_type); + CHECK(flow_file->getAttribute("ModificationTime") == modification_time); + } else { + CHECK(flow_file->getAttribute("ModificationUsername") == std::nullopt); + CHECK(flow_file->getAttribute("ModificationUpdateType") == std::nullopt); + CHECK(flow_file->getAttribute("ModificationTime") == std::nullopt); + } + } + + void verifyResults(const ProcessorTriggerResult& results, const std::string& expected_contents) { + auto& fetch_results = results.at(processors::FetchOPCHistory::Success); + REQUIRE(fetch_results.size() == 1); + rapidjson::Document result_document; + result_document.Parse(controller_.plan->getContent(fetch_results[0]).c_str()); + rapidjson::Document expected_document; + expected_document.Parse(expected_contents.c_str()); + REQUIRE(result_document == expected_document); + } Review Comment: This test uses `rapidjson::Document` without including the RapidJSON headers in this file. Unless another include guarantees it (not shown in this diff), this will fail to compile. Add the appropriate include (e.g., RapidJSON document header) in `FetchOPCHistoryTests.cpp`. ########## extensions/opc/include/FetchOPCHistory.h: ########## @@ -0,0 +1,155 @@ +/** + * 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 <chrono> +#include <cstdint> +#include <memory> +#include <optional> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <utility> +#include <vector> + +#include "BaseOPCProcessor.h" +#include "OPCCommon.h" +#include "core/ProcessSession.h" +#include "core/PropertyDefinitionBuilder.h" +#include "core/logging/LoggerFactory.h" +#include "minifi-cpp/controllers/RecordSetWriter.h" +#include "minifi-cpp/core/Property.h" +#include "minifi-cpp/core/PropertyValidator.h" +#include "minifi-cpp/core/RelationshipDefinition.h" +#include "minifi-cpp/core/StateManager.h" + +namespace org::apache::nifi::minifi::processors { + +struct FetchedState { + int64_t timestamp = 0; + std::unordered_set<std::string> fingerprints; +}; + +struct FetchOPCHistoryContext { + core::ProcessSession& session; + std::shared_ptr<core::RecordSetWriter> record_set_writer; + std::unordered_map<std::string, std::string>& state_map; + size_t& entries_transferred; + const uint64_t batch_size; + const std::string& node_id; + const int32_t namespace_index; + const std::optional<FetchedState> fetched_state; + std::shared_ptr<core::logging::Logger> logger; +}; + +class FetchOPCHistory final : public BaseOPCProcessor { + public: + using BaseOPCProcessor::BaseOPCProcessor; + + EXTENSIONAPI static constexpr const char* Description = + "Fetches OPC-UA node history between the start and end timestamps. " + "A history entry is only fetched once, on every trigger only the not yet fetched entries are returned."; + + EXTENSIONAPI static constexpr auto NodeIDType = + core::PropertyDefinitionBuilder<3>::createProperty("Node ID type") + .withDescription("Specifies the type of the provided node ID") + .isRequired(true) + .withAllowedValues({"String", "Int", "Guid"}) + .build(); + EXTENSIONAPI static constexpr auto NodeID = + core::PropertyDefinitionBuilder<>::createProperty("Node ID") + .withDescription("Specifies the ID of the root node to fetch history for.") + .isRequired(true) + .build(); + EXTENSIONAPI static constexpr auto NameSpaceIndex = + core::PropertyDefinitionBuilder<>::createProperty("Namespace index") + .withDescription("The index of the namespace.") + .withValidator(core::StandardPropertyValidators::INTEGER_VALIDATOR) + .withDefaultValue("0") + .isRequired(true) + .build(); + EXTENSIONAPI static constexpr auto StartTimestamp = + core::PropertyDefinitionBuilder<>::createProperty("Start timestamp") + .withDescription( + "Timestamp after which the events should be returned. If not specified entries are returned from the beginning of the history.") + .build(); + EXTENSIONAPI static constexpr auto EndTimestamp = + core::PropertyDefinitionBuilder<>::createProperty("End timestamp") + .withDescription("Timestamp before which the events should be returned. If not specified entries are returned until the current time.") + .build(); + EXTENSIONAPI static constexpr auto BatchSize = + core::PropertyDefinitionBuilder<>::createProperty("Batch Size") + .withDescription("Maximum number entries to read and return in a single batch. If set to zero or empty all available entries are returned.") + .withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR) + .build(); + EXTENSIONAPI static constexpr auto HistoryReadType = + core::PropertyDefinitionBuilder<magic_enum::enum_count<opc::HistoryReadTypeOption>()>::createProperty("History Read Type") + .withDescription("Whether to fetch raw historical values or the audit trail of modifications to historical values") + .isRequired(true) + .withAllowedValues(magic_enum::enum_names<opc::HistoryReadTypeOption>()) + .withDefaultValue(magic_enum::enum_name<opc::HistoryReadTypeOption::Raw>()) + .build(); + EXTENSIONAPI static constexpr auto RecordSetWriter = + core::PropertyDefinitionBuilder<>::createProperty("Record Set Writer") + .withDescription("Specifies the Controller Service to use for writing results to a FlowFile instead of using the default output format.") + .withAllowedTypes<core::RecordSetWriter>() + .build(); + EXTENSIONAPI static constexpr auto Properties = utils::array_cat(BaseOPCProcessor::Properties, + std::to_array<core::PropertyReference>( + {NodeIDType, NodeID, NameSpaceIndex, StartTimestamp, EndTimestamp, BatchSize, HistoryReadType, RecordSetWriter})); Review Comment: `FetchOPCHistory` exposes `Node ID type` values `{String, Int, Guid}`, but it inherits all `BaseOPCProcessor::Properties`, which (per existing OPC processors/docs in this PR) likely includes path-specific properties like “Path reference types”. If those inherited properties are irrelevant for history reads, they become confusing/no-op configuration surface. Consider either (a) excluding path-specific base properties from this processor’s `Properties`, or (b) making `Node ID type` support `Path` if history over path traversal is actually intended. ########## extensions/opc/tests/OpcUaTestServer.h: ########## @@ -134,17 +212,148 @@ class OpcUaTestServer { } private: - UA_NodeId addObject(const char *name, UA_NodeId parent) { + static std::string nodeIdToString(const UA_NodeId& id) { + if (id.identifierType == UA_NODEIDTYPE_STRING) { + return std::string(reinterpret_cast<const char*>(id.identifier.string.data), id.identifier.string.length); + } else if (id.identifierType == UA_NODEIDTYPE_NUMERIC) { + return std::to_string(id.identifier.numeric); + } else if (id.identifierType == UA_NODEIDTYPE_GUID) { + char guid_str[37]; + snprintf(guid_str, + sizeof(guid_str), + "%08x-%04x-%04x-%04x-%012" PRIx64, + id.identifier.guid.data1, + id.identifier.guid.data2, + id.identifier.guid.data3, + (id.identifier.guid.data4[0] << 8) | id.identifier.guid.data4[1], + (gsl::narrow<uint64_t>(id.identifier.guid.data4[2]) << 40) | (gsl::narrow<uint64_t>(id.identifier.guid.data4[3]) << 32) | + (gsl::narrow<uint64_t>(id.identifier.guid.data4[2]) << 40) | (gsl::narrow<uint64_t>(id.identifier.guid.data4[3]) << 32) | + (gsl::narrow<uint64_t>(id.identifier.guid.data4[4]) << 24) | (gsl::narrow<uint64_t>(id.identifier.guid.data4[5]) << 16) | + (gsl::narrow<uint64_t>(id.identifier.guid.data4[6]) << 8) | gsl::narrow<uint64_t>(id.identifier.guid.data4[7])); + } + return {}; + } + + static std::vector<const HistoryModificationRecord*> selectRecords(const std::vector<HistoryModificationRecord>& records, UA_DateTime start_time, + UA_DateTime end_time, size_t num_values_per_node, bool& has_more_data) { + has_more_data = false; + std::vector<const HistoryModificationRecord*> selected; + for (const auto& record : records) { + if (record.modification_time >= start_time && record.modification_time < end_time) { + if (num_values_per_node > 0 && selected.size() >= num_values_per_node) { + // A qualifying record exists beyond the requested window: the server must signal this with a continuation point. + has_more_data = true; + break; + } + selected.push_back(&record); + } + } + return selected; + } + + static void readRawCallback(UA_Server* /*server*/, void* hdbContext, const UA_NodeId* /*sessionId*/, void* /*sessionContext*/, + const UA_RequestHeader* /*requestHeader*/, const UA_ReadRawModifiedDetails* historyReadDetails, UA_TimestampsToReturn /*timestampsToReturn*/, + UA_Boolean /*releaseContinuationPoints*/, size_t nodesToReadSize, const UA_HistoryReadValueId* nodesToRead, UA_HistoryReadResponse* response, + UA_HistoryData* const* const historyData) { + auto* self = static_cast<OpcUaTestServer*>(hdbContext); + std::lock_guard<std::mutex> lock(self->history_mutex_); + + for (size_t i = 0; i < nodesToReadSize; ++i) { + auto node_id_str = nodeIdToString(nodesToRead[i].nodeId); + auto it = self->history_records_.find(node_id_str); + if (it == self->history_records_.end() || nodesToRead[i].nodeId.identifierType != self->node_ids_[node_id_str].identifierType) { Review Comment: Using `self->node_ids_[node_id_str]` will insert a default `UA_NodeId` entry when `node_id_str` is missing, mutating `node_ids_` during reads and potentially masking errors. Use `find()` on `node_ids_` and treat missing entries as `BADNODEIDUNKNOWN` without inserting. ########## extensions/opc/tests/OpcUaTestServer.h: ########## @@ -134,17 +212,148 @@ class OpcUaTestServer { } private: - UA_NodeId addObject(const char *name, UA_NodeId parent) { + static std::string nodeIdToString(const UA_NodeId& id) { + if (id.identifierType == UA_NODEIDTYPE_STRING) { + return std::string(reinterpret_cast<const char*>(id.identifier.string.data), id.identifier.string.length); + } else if (id.identifierType == UA_NODEIDTYPE_NUMERIC) { + return std::to_string(id.identifier.numeric); + } else if (id.identifierType == UA_NODEIDTYPE_GUID) { + char guid_str[37]; + snprintf(guid_str, + sizeof(guid_str), + "%08x-%04x-%04x-%04x-%012" PRIx64, + id.identifier.guid.data1, + id.identifier.guid.data2, + id.identifier.guid.data3, + (id.identifier.guid.data4[0] << 8) | id.identifier.guid.data4[1], + (gsl::narrow<uint64_t>(id.identifier.guid.data4[2]) << 40) | (gsl::narrow<uint64_t>(id.identifier.guid.data4[3]) << 32) | + (gsl::narrow<uint64_t>(id.identifier.guid.data4[2]) << 40) | (gsl::narrow<uint64_t>(id.identifier.guid.data4[3]) << 32) | + (gsl::narrow<uint64_t>(id.identifier.guid.data4[4]) << 24) | (gsl::narrow<uint64_t>(id.identifier.guid.data4[5]) << 16) | + (gsl::narrow<uint64_t>(id.identifier.guid.data4[6]) << 8) | gsl::narrow<uint64_t>(id.identifier.guid.data4[7])); Review Comment: `nodeIdToString()` never returns the formatted GUID string (it falls through to `return {}`), and the last 48-bit composition duplicates `data4[2]`/`data4[3]` shifts. This will break lookups (GUID node IDs become empty strings) and produces incorrect GUID rendering. Return `std::string(guid_str)` from the GUID branch and fix the `data4[2..7]` packing to include each byte exactly once. ########## extensions/opc/tests/OpcUaTestServer.h: ########## @@ -155,20 +364,45 @@ class OpcUaTestServer { return object_id; } - UA_NodeId addIntVariable(const char *name, UA_NodeId parent, UA_Int32 value) { + UA_StatusCode addNode(UA_NodeId parent_node_id, UA_NodeId& target_node_id, opc::OPCNodeIDType type, const std::string& browse_name, + const UA_VariableAttributes& attr) { + UA_QualifiedName qname = UA_QUALIFIEDNAME(ns_index_, const_cast<char*>(browse_name.c_str())); + UA_NodeId node_id; + + switch (type) { + case opc::OPCNodeIDType::Int: + node_id = UA_NODEID_NUMERIC(ns_index_, std::stoi(browse_name)); + break; + case opc::OPCNodeIDType::String: + node_id = UA_NODEID_STRING_ALLOC(ns_index_, browse_name.c_str()); + break; + case opc::OPCNodeIDType::Guid: + node_id = UA_NODEID_GUID(ns_index_, UA_GUID(browse_name.c_str())); + break; Review Comment: `UA_GUID(...)` in open62541 is not a GUID string parser; it’s typically a macro/constructor expecting GUID components, so passing `browse_name.c_str()` is invalid and will not work (often won’t compile). Parse the GUID string into a `UA_Guid` using the appropriate open62541 helper (e.g., a `UA_Guid_fromString(...)`-style API for your version) before building the `UA_NodeId`. ########## extensions/opc/tests/features/steps/steps.py: ########## @@ -48,6 +48,13 @@ def setup_opcua_server_with_access_control(context: MinifiTestContext): ) +@step("an OPC UA server is set up with historical data support") +def setup_opcua_server_with_historical_data(context: MinifiTestContext): + context.containers["opcua-server-historical"] = OPCUAServerContainer( + context, command=["/opt/open62541/examples/tutorial_server_historicaldata"] + ) Review Comment: The feature scenarios configure endpoints using the hostname pattern `opcua-server-${scenario_id}`, but this step registers the container under `opcua-server-historical`. If the Behave framework derives the container hostname from the dictionary key (as it commonly does), the configured endpoint won’t resolve to the started container. Align the container key with the expected hostname used in the feature (or update the feature endpoints to match this container name). ########## extensions/opc/src/FetchOPCHistory.cpp: ########## @@ -0,0 +1,364 @@ +/** + * 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 "FetchOPCHistory.h" + +#include <optional> +#include <string> +#include <unordered_set> + +#include "core/ProcessSession.h" +#include "core/Resource.h" +#include "minifi-cpp/core/ProcessContext.h" +#include "utils/ProcessorConfigUtils.h" +#include "utils/StringUtils.h" + +namespace org::apache::nifi::minifi::processors { + +namespace { + +constexpr const char* LAST_FETCHED_TIMESTAMP_KEY = "last_fetched_timestamp"; +constexpr const char* LAST_FETCHED_FINGERPRINT_KEY = "last_fetched_fingerprint"; + +std::string updateTypeToString(UA_HistoryUpdateType type) { + switch (type) { + case UA_HISTORYUPDATETYPE_INSERT: + return "Insert"; + case UA_HISTORYUPDATETYPE_REPLACE: + return "Replace"; + case UA_HISTORYUPDATETYPE_UPDATE: + return "Update"; + case UA_HISTORYUPDATETYPE_DELETE: + return "Delete"; + default: + return "Unknown"; + } +} + +std::string uaStringToString(const UA_String& str) { + return {reinterpret_cast<const char*>(str.data), str.length}; +} + +struct HistoryEntry { + std::string value; + int64_t source_timestamp = 0; + const UA_ModificationInfo* modification_info = nullptr; + + [[nodiscard]] int64_t modificationTime() const { + return modification_info ? modification_info->modificationTime : UA_DateTime_fromUnixTime(0); + } +}; + +struct HistoryBatch { + std::vector<HistoryEntry> entries; + bool has_modification_info = false; +}; + +std::string entryFingerprint(const HistoryEntry& entry, bool has_modification_info) { + // The fingerprint deduplicates entries sharing the boundary source timestamp across triggers. For raw value history a + // duplicated (value, source timestamp) pair represents no change in the history, so losing one to deduplication is harmless. + // For audit (modification) history the modification time and update type distinguish otherwise-identical entries, so they + // are included to make the fingerprint more unique and reduce the chance of dropping a distinct modification. + std::string raw = ":" + entry.value; + if (has_modification_info) { + const auto update_type = entry.modification_info ? updateTypeToString(entry.modification_info->updateType) : ""; + raw = std::to_string(entry.modificationTime()) + ":" + update_type + raw; + } + return utils::string::to_hex(raw); +} + +// NOLINTBEGIN(cppcoreguidelines-pro-type-union-access) +std::optional<HistoryBatch> extractHistoryBatch(const UA_ExtensionObject* data, const std::shared_ptr<core::logging::Logger>& logger) { + const UA_DataValue* data_values = nullptr; + size_t data_value_size = 0; + const UA_ModificationInfo* modification_infos = nullptr; + size_t modification_infos_size = 0; + + if (data->content.decoded.type == &UA_TYPES[UA_TYPES_HISTORYDATA]) { + const auto* history_data = static_cast<const UA_HistoryData*>(data->content.decoded.data); + data_values = history_data->dataValues; + data_value_size = history_data->dataValuesSize; + } else if (data->content.decoded.type == &UA_TYPES[UA_TYPES_HISTORYMODIFIEDDATA]) { + const auto* modified_data = static_cast<const UA_HistoryModifiedData*>(data->content.decoded.data); + data_values = modified_data->dataValues; + data_value_size = modified_data->dataValuesSize; + modification_infos = modified_data->modificationInfos; + modification_infos_size = modified_data->modificationInfosSize; + } else { + logger->log_error("Unexpected data type received in the history read callback: {}", data->content.decoded.type->typeName); + return std::nullopt; + } + + HistoryBatch batch; + batch.has_modification_info = modification_infos != nullptr; + batch.entries.reserve(data_value_size); + for (size_t i = 0; i < data_value_size; ++i) { + HistoryEntry entry; + try { + entry.value = opc::variantToString(data_values[i].value); + } catch (const opc::OPCException& ex) { + logger->log_warn("Failed to convert value at index {} to string, skipping entry: {}", i, ex.what()); + continue; + } + entry.source_timestamp = data_values[i].sourceTimestamp; + entry.modification_info = (modification_infos && i < modification_infos_size) ? &modification_infos[i] : nullptr; + batch.entries.push_back(std::move(entry)); + } + return batch; +} +// NOLINTEND(cppcoreguidelines-pro-type-union-access) + +std::vector<HistoryEntry> selectNewEntries(std::vector<HistoryEntry> entries, const std::optional<FetchedState>& last_fetched, + bool has_modification_info, std::optional<size_t> max_entries) { + std::vector<HistoryEntry> new_entries; + new_entries.reserve(max_entries ? std::min(*max_entries, entries.size()) : entries.size()); + for (auto& entry : entries) { + if (max_entries && new_entries.size() >= *max_entries) { + break; + } + if (last_fetched && entry.source_timestamp == last_fetched->timestamp && + last_fetched->fingerprints.contains(entryFingerprint(entry, has_modification_info))) { + continue; + } + new_entries.push_back(std::move(entry)); + } + return new_entries; +} + +void addModificationInfo(core::Record& record, const UA_ModificationInfo& modification_info) { + if (modification_info.userName.length > 0) { + record.emplace("ModificationUsername", core::RecordField(uaStringToString(modification_info.userName))); + } + record.emplace("ModificationTime", core::RecordField(opc::OPCDateTime2String(modification_info.modificationTime))); + record.emplace("ModificationUpdateType", core::RecordField(updateTypeToString(modification_info.updateType))); +} + +void addModificationInfo(core::FlowFile& flow_file, const UA_ModificationInfo& modification_info) { + if (modification_info.userName.length > 0) { + flow_file.addAttribute("ModificationUsername", uaStringToString(modification_info.userName)); + } + flow_file.addAttribute("ModificationTime", opc::OPCDateTime2String(modification_info.modificationTime)); + flow_file.addAttribute("ModificationUpdateType", updateTypeToString(modification_info.updateType)); +} + +core::Record toRecord(const std::string& node_id, const int32_t namespace_index, const HistoryEntry& entry) { + core::Record record; + record.emplace("Value", core::RecordField(entry.value)); + record.emplace("NodeID", core::RecordField(node_id)); + record.emplace("Namespace index", core::RecordField(std::to_string(namespace_index))); + record.emplace("Sourcetimestamp", core::RecordField(opc::OPCDateTime2String(entry.source_timestamp))); + if (entry.modification_info) { + addModificationInfo(record, *entry.modification_info); + } + return record; +} + +void writeAsRecordSet(FetchOPCHistoryContext& context, const std::vector<HistoryEntry>& entries) { + core::RecordSet record_set; + for (const auto& entry : entries) { + record_set.push_back(toRecord(context.node_id, context.namespace_index, entry)); + } + + auto flow_file = context.session.create(); + context.record_set_writer->write(record_set, flow_file, context.session); + context.session.transfer(flow_file, FetchOPCHistory::Success); + context.entries_transferred += entries.size(); +} + +void writeAsFlowFiles(FetchOPCHistoryContext& context, const std::vector<HistoryEntry>& entries) { + for (const auto& entry : entries) { + auto flow_file = context.session.create(); + context.session.write(flow_file, [&entry](const std::shared_ptr<io::OutputStream>& output_stream) -> io::IoResult { + output_stream->write(reinterpret_cast<const uint8_t*>(entry.value.data()), entry.value.size()); + return io::IoResult::from(entry.value.size()); + }); + flow_file->addAttribute("NodeID", context.node_id); + flow_file->addAttribute("Namespace index", std::to_string(context.namespace_index)); + flow_file->addAttribute("Sourcetimestamp", opc::OPCDateTime2String(entry.source_timestamp)); + if (entry.modification_info) { + addModificationInfo(*flow_file, *entry.modification_info); + } + context.session.transfer(flow_file, FetchOPCHistory::Success); + ++context.entries_transferred; + } +} + +void updateState(std::unordered_map<std::string, std::string>& state_map, const std::vector<HistoryEntry>& new_entries, bool has_modification_info, + const std::optional<FetchedState>& fetched_state) { + const int64_t new_timestamp = new_entries.back().source_timestamp; + + std::unordered_set<std::string> fingerprints; + if (fetched_state && fetched_state->timestamp == new_timestamp) { + fingerprints = fetched_state->fingerprints; + } + for (const auto& entry : new_entries) { + if (entry.source_timestamp == new_timestamp) { + fingerprints.insert(entryFingerprint(entry, has_modification_info)); + } + } + + state_map[LAST_FETCHED_TIMESTAMP_KEY] = std::to_string(new_timestamp); + state_map[LAST_FETCHED_FINGERPRINT_KEY] = utils::string::join(",", fingerprints); +} + +UA_Boolean historyReadCallback(UA_Client* /*client*/, const UA_NodeId* /*node_id*/, UA_Boolean more_data_available, const UA_ExtensionObject* data, + void* ctx) { + auto* opc_history_context = static_cast<FetchOPCHistoryContext*>(ctx); + + auto batch = extractHistoryBatch(data, opc_history_context->logger); + if (batch && !batch->entries.empty()) { + const std::optional<size_t> remaining = opc_history_context->batch_size != 0 + ? std::optional<size_t>(opc_history_context->batch_size - opc_history_context->entries_transferred) + : std::nullopt; + auto new_entries = selectNewEntries(std::move(batch->entries), opc_history_context->fetched_state, batch->has_modification_info, remaining); + + if (!new_entries.empty()) { + if (opc_history_context->record_set_writer) { + writeAsRecordSet(*opc_history_context, new_entries); + } else { + writeAsFlowFiles(*opc_history_context, new_entries); + } + updateState(opc_history_context->state_map, new_entries, batch->has_modification_info, opc_history_context->fetched_state); + } + } + + const bool batch_limit_reached = opc_history_context->batch_size != 0 && + opc_history_context->entries_transferred >= opc_history_context->batch_size; + return more_data_available && !batch_limit_reached; +} + +UA_DateTime calculateStartTime(const std::optional<FetchedState>& fetched_state, + const std::optional<std::chrono::system_clock::time_point>& start_timestamp) { + if (fetched_state && fetched_state->timestamp != 0) { + return fetched_state->timestamp; + } else if (start_timestamp.has_value()) { + uint64_t start_time_seconds = std::chrono::duration_cast<std::chrono::seconds>(start_timestamp->time_since_epoch()).count(); + return UA_DateTime_fromUnixTime(gsl::narrow<UA_Int64>(start_time_seconds)); + } + return UA_DateTime_fromUnixTime(0); +} + +UA_DateTime calculateEndTime(const std::optional<std::chrono::system_clock::time_point>& end_timestamp) { + if (end_timestamp.has_value()) { + uint64_t end_time_seconds = std::chrono::duration_cast<std::chrono::seconds>(end_timestamp->time_since_epoch()).count(); + return UA_DateTime_fromUnixTime(gsl::narrow<UA_Int64>(end_time_seconds)); Review Comment: Start/end timestamps are truncated to whole seconds. If a user provides ISO-8601 timestamps with sub-second precision, the effective query window will be widened and can re-fetch/skip boundary entries incorrectly. Prefer converting to milliseconds (or directly building a `UA_DateTime` with sub-second precision) rather than `duration_cast<std::chrono::seconds>`. -- 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]
