lordgamez commented on code in PR #2259: URL: https://github.com/apache/nifi-minifi-cpp/pull/2259#discussion_r3980762267
########## 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: Fixed in https://github.com/apache/nifi-minifi-cpp/pull/2259/commits/4744af4d3d02f1d747ede09b61156aec440f3998 ########## 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: Updated in https://github.com/apache/nifi-minifi-cpp/pull/2259/commits/4744af4d3d02f1d747ede09b61156aec440f3998 -- 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]
