szaszm commented on code in PR #2265: URL: https://github.com/apache/nifi-minifi-cpp/pull/2265#discussion_r4559045154
########## extensions/opc/src/FetchOPCHistory.cpp: ########## @@ -0,0 +1,418 @@ +/** + * 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 <vector> + +#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; Review Comment: I'd replace this with fmt::format/std::format or similar, to avoid O(n²) allocations, but it's not super important. ########## behave_framework/src/minifi_behave/steps/checking_steps.py: ########## @@ -284,39 +302,62 @@ def verify_files_with_contents_in_directory(context: MinifiTestContext, director @then( - 'files with at least these contents "{contents}" are placed in the "{directory}" directory in less than {timeout}' + 'in the "{container_name}" container files with at least these contents "{contents}" are placed in the "{directory}" directory in less than {timeout}' ) -def verify_files_with_at_least_contents_in_directory( - context: MinifiTestContext, directory: str, timeout: str, contents: str +def verify_files_with_at_least_contents_in_directory_in_container( + context: MinifiTestContext, container_name: str, directory: str, timeout: str, contents: str ): timeout_seconds = humanfriendly.parse_timespan(timeout) new_contents = contents.replace("\\n", "\n") contents_arr = new_contents.split(",") assert wait_for_condition( condition=lambda: all( - context.containers[DEFAULT_MINIFI_CONTAINER_NAME].directory_contains_file_with_content(directory, content) + context.containers[container_name].directory_contains_file_with_content(directory, content) for content in contents_arr ), timeout_seconds=timeout_seconds, - bail_condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].exited, + bail_condition=lambda: context.containers[container_name].exited, context=context, ) -@then('a file with the JSON content "{content}" is placed in the "{directory}" directory in less than {duration}') -@then("a file with the JSON content '{content}' is placed in the '{directory}' directory in less than {duration}") -def verify_file_with_json_content_in_directory(context: MinifiTestContext, content: str, directory: str, duration: str): +@then( + 'files with at least these contents "{contents}" are placed in the "{directory}" directory in less than {timeout}' +) Review Comment: I don't like when we blow up these to 3 lines, two of which are almost empty. The step wording is also unnecessarily verbose. ```suggestion @then('files containing "{contents}" appear in "{directory}" in {timeout}') ``` ########## extensions/opc/src/FetchOPCHistory.cpp: ########## @@ -0,0 +1,418 @@ +/** + * 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 <vector> + +#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, opc::BinaryEncoding binary_encoding, + 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, binary_encoding); + } 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 int64_t new_timestamp = new_entries.back().source_timestamp; + const auto new_timestamp_str = std::to_string(new_timestamp); + + auto& stored_timestamp = state_map[LAST_FETCHED_TIMESTAMP_KEY]; + auto& fingerprints = state_map[LAST_FETCHED_FINGERPRINT_KEY]; + if (stored_timestamp != new_timestamp_str) { + stored_timestamp = new_timestamp_str; + fingerprints.clear(); + } + + for (const auto& entry : new_entries) { + if (entry.source_timestamp == new_timestamp) { + if (!fingerprints.empty()) { + fingerprints += ","; + } + fingerprints += entryFingerprint(entry, has_modification_info); + } + } +} + +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); + + const auto binary_encoding = opc_history_context->record_set_writer ? opc::BinaryEncoding::Base64 : opc::BinaryEncoding::Raw; + auto batch = extractHistoryBatch(data, binary_encoding, 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); + } + } + + 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 toUaDateTime(std::chrono::system_clock::time_point tp) { + // UA_DateTime counts 100 ns ticks since 1601; UA_DATETIME_USEC ticks make up one microsecond. Converting at microsecond + // resolution (rather than truncating to whole seconds) preserves the sub-second precision of user-provided timestamps. + const auto usec_since_epoch = std::chrono::duration_cast<std::chrono::microseconds>(tp.time_since_epoch()).count(); + return UA_DATETIME_UNIX_EPOCH + usec_since_epoch * UA_DATETIME_USEC; +} + +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()) { + return toUaDateTime(*start_timestamp); + } + return UA_DateTime_fromUnixTime(0); +} + +UA_DateTime calculateEndTime(const std::optional<std::chrono::system_clock::time_point>& end_timestamp) { + if (end_timestamp.has_value()) { + return toUaDateTime(*end_timestamp); + } + 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_ = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_idx_, node_id_.c_str())}; + break; + case opc::OPCNodeIDType::Int: + node_ = opc::NodeId{UA_NODEID_NUMERIC(namespace_idx_, std::stoi(node_id_))}; + break; + case opc::OPCNodeIDType::Guid: { + UA_Guid guid; + if (UA_Guid_parse(&guid, UA_STRING(const_cast<char*>(node_id_.c_str()))) != UA_STATUSCODE_GOOD) { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("{} cannot be used as a GUID type node ID", node_id_)); + } + node_ = opc::NodeId{UA_NODEID_GUID(namespace_idx_, guid)}; + break; + } + case opc::OPCNodeIDType::Path: + path_reference_types_.clear(); + readPathReferenceTypes(context, node_id_); + path_node_id_resolved_ = false; + break; + default: + throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Unsupported Node ID type: {}", magic_enum::enum_name(id_type_))); + } + + history_type_ = utils::parseEnumProperty<opc::HistoryReadTypeOption>(context, HistoryReadType); + auto start_timestamp_str = utils::parseOptionalProperty(context, StartTimestamp); + if (start_timestamp_str) { + start_timestamp_ = utils::timeutils::parseRfc3339(start_timestamp_str.value()); + if (!start_timestamp_) { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Invalid StartTimestamp: {}", start_timestamp_str.value())); + } + } else { + start_timestamp_ = std::nullopt; + } + + auto end_timestamp_str = utils::parseOptionalProperty(context, EndTimestamp); + if (end_timestamp_str) { + end_timestamp_ = utils::timeutils::parseRfc3339(end_timestamp_str.value()); + if (!end_timestamp_) { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Invalid EndTimestamp: {}", end_timestamp_str.value())); + } + } else { + end_timestamp_ = std::nullopt; + } Review Comment: The property descriptions need to specify that the required format is RFC3339. This is not a timestamp format that's natural to most humans. ########## extensions/opc/tests/FetchOPCProcessorTests.cpp: ########## @@ -43,7 +43,7 @@ TEST_CASE("Test fetching using path node id", "[fetchopcprocessor]") { CHECK(flow_file->getAttribute("Datasize") == "4"); CHECK(flow_file->getAttribute("Full path") == "Simulator/Default/Device1/INT" + std::to_string(i + 1)); CHECK(flow_file->getAttribute("NodeID")); - CHECK(flow_file->getAttribute("NodeID type") == "numeric"); + CHECK(flow_file->getAttribute("NodeID type") == "string"); Review Comment: why did this change? ########## extensions/opc/src/FetchOPCHistory.cpp: ########## @@ -0,0 +1,418 @@ +/** + * 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 <vector> + +#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) { Review Comment: could a string_view work here? ########## extensions/opc/tests/OpcUaTestServer.h: ########## @@ -136,17 +223,171 @@ 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[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: We already have this logic in Identifier, can we reuse that? ########## extensions/opc/include/OPCCommon.h: ########## @@ -118,6 +134,8 @@ class Client { template<typename T> UA_StatusCode add_node(const UA_NodeId parent_node_id, const UA_NodeId target_node_id, const UA_UInt32 ref_type_id, std::string_view browse_name, T value, UA_NodeId *received_node_id); + UA_StatusCode readHistory(HistoryReadTypeOption history_type, const UA_NodeId& node_id, const HistoryCallback callback, UA_DateTime start_time, UA_DateTime end_time, + void *callback_context); Review Comment: I'd keep the related parameters (callback, callback_context) next to each other if possible ########## extensions/opc/src/FetchOPCHistory.cpp: ########## @@ -0,0 +1,418 @@ +/** + * 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 <vector> + +#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, opc::BinaryEncoding binary_encoding, + 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, binary_encoding); + } 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 int64_t new_timestamp = new_entries.back().source_timestamp; + const auto new_timestamp_str = std::to_string(new_timestamp); + + auto& stored_timestamp = state_map[LAST_FETCHED_TIMESTAMP_KEY]; + auto& fingerprints = state_map[LAST_FETCHED_FINGERPRINT_KEY]; + if (stored_timestamp != new_timestamp_str) { + stored_timestamp = new_timestamp_str; + fingerprints.clear(); + } + + for (const auto& entry : new_entries) { + if (entry.source_timestamp == new_timestamp) { + if (!fingerprints.empty()) { + fingerprints += ","; + } + fingerprints += entryFingerprint(entry, has_modification_info); + } + } +} + +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); + + const auto binary_encoding = opc_history_context->record_set_writer ? opc::BinaryEncoding::Base64 : opc::BinaryEncoding::Raw; + auto batch = extractHistoryBatch(data, binary_encoding, 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); + } + } + + 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 toUaDateTime(std::chrono::system_clock::time_point tp) { + // UA_DateTime counts 100 ns ticks since 1601; UA_DATETIME_USEC ticks make up one microsecond. Converting at microsecond + // resolution (rather than truncating to whole seconds) preserves the sub-second precision of user-provided timestamps. + const auto usec_since_epoch = std::chrono::duration_cast<std::chrono::microseconds>(tp.time_since_epoch()).count(); + return UA_DATETIME_UNIX_EPOCH + usec_since_epoch * UA_DATETIME_USEC; +} + +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()) { + return toUaDateTime(*start_timestamp); + } + return UA_DateTime_fromUnixTime(0); +} + +UA_DateTime calculateEndTime(const std::optional<std::chrono::system_clock::time_point>& end_timestamp) { + if (end_timestamp.has_value()) { + return toUaDateTime(*end_timestamp); + } + return UA_DateTime_now(); Review Comment: Is this a special value that specifies no end time limit, or does this call query the system for the current time and return that? The latter case may be vulnerable to TOCTOU problems. ########## PROCESSORS.md: ########## @@ -1089,6 +1090,53 @@ In the list below, the names of required properties appear in bold. Any other pr | failure | An error occurred processing | +## FetchOPCHistory + +### 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. + +### Properties + +In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. + +| Name | Default Value | Allowable Values | Description | +|---------------------------------|---------------|----------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| **OPC server endpoint** | | | Specifies the address, port and relative path of an OPC endpoint | +| Application URI | | | Application URI of the client in the format 'urn:unconfigured:application'. Mandatory, if using Secure Channel and must match the URI included in the certificate's Subject Alternative Names. | +| Username | | | Username to log in with. | +| Password | | | Password to log in with.<br/>**Sensitive Property: true** | +| Certificate path | | | Path to the DER-encoded cert file | +| Key path | | | Path to the DER-encoded key file | +| Trusted server certificate path | | | Comma separated list of paths to the DER-encoded trusted server certificates | +| Path reference types | | | Specify the reference types between nodes in the path if Path Node ID type is used. If not provided, all reference types are assumed to be Organizes. The format is 'referenceType1/referenceType2/.../referenceTypeN' and the supported reference types are Organizes, HasComponent, HasProperty, and HasSubtype. | +| **Node ID type** | | Path<br/>Int<br/>String<br/>Guid | Specifies the type of the provided node ID | +| **Node ID** | | | Specifies the ID of the root node to fetch history for. In case of a Path Node ID Type, the path should be provided in the format of 'path/to/node'. | +| **Namespace index** | 0 | | The index of the namespace. | +| Start timestamp | | | Timestamp after which the events should be returned. If not specified entries are returned from the beginning of the history. | +| End timestamp | | | Timestamp before which the events should be returned. If not specified entries are returned until the current time. | +| Batch size | | | Maximum number entries to read and return in a single batch. If set to zero or empty all available entries are returned. | +| **History read type** | Raw | Raw<br/>Audit | Whether to fetch raw historical values or the audit trail of modifications to historical values | +| Record set writer | | | Specifies the Controller Service to use for writing results to a FlowFile instead of using the default output format. | + +### Relationships + +| Name | Description | +|---------|----------------------------------------------------| +| success | Successfully retrieved OPC-UA node history entries | + +### Output Attributes + +| Attribute | Relationship | Description | +|------------------------|--------------|-----------------------------------------------------------------------------------------| +| NodeID | success | ID of the node. | +| Namespace index | success | Namespace index of the node. | +| Sourcetimestamp | success | The timestamp of when the node was created in the server as 'YYYY-MM-DDTHH:MM:SS.sssZ'. | +| ModificationUsername | success | Username of the user who modified the node. | +| ModificationTime | success | Timestamp of when the node was modified. | +| ModificationUpdateType | success | Type of modification performed on the node. | Review Comment: attributes are generally all lowercase and separated with dots. I would not deviate from that convention here. ########## extensions/opc/src/FetchOPCHistory.cpp: ########## @@ -0,0 +1,418 @@ +/** + * 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 <vector> + +#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, opc::BinaryEncoding binary_encoding, + 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, binary_encoding); + } 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 int64_t new_timestamp = new_entries.back().source_timestamp; + const auto new_timestamp_str = std::to_string(new_timestamp); + + auto& stored_timestamp = state_map[LAST_FETCHED_TIMESTAMP_KEY]; + auto& fingerprints = state_map[LAST_FETCHED_FINGERPRINT_KEY]; + if (stored_timestamp != new_timestamp_str) { + stored_timestamp = new_timestamp_str; + fingerprints.clear(); + } + + for (const auto& entry : new_entries) { + if (entry.source_timestamp == new_timestamp) { + if (!fingerprints.empty()) { + fingerprints += ","; + } + fingerprints += entryFingerprint(entry, has_modification_info); + } + } +} + +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); + + const auto binary_encoding = opc_history_context->record_set_writer ? opc::BinaryEncoding::Base64 : opc::BinaryEncoding::Raw; Review Comment: Why do we change the format based on whether it goes in records or flow file contents? -- 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]
