From bcd34a9075c654a3dc886b2f398680c7517315f6 Mon Sep 17 00:00:00 2001 From: Gabor Gyimesi Date: Thu, 27 Aug 2026 16:18:46 +0200 Subject: [PATCH] MINIFICPP-2894 Add FetchOPCHistory processor - Additionally changed the other OPC UA processors to use ISO-8601 timestamp format in attributes --- .../ubuntu_22_04_clang_arm_manifest.json | 179 ++++++++ PROCESSORS.md | 66 ++- README.md | 2 +- .../src/minifi_behave/steps/checking_steps.py | 81 +++- .../steps/flow_building_steps.py | 9 + extensions/opc/include/FetchOPCHistory.h | 158 +++++++ extensions/opc/include/FetchOPCProcessor.h | 6 +- extensions/opc/include/OPCCommon.h | 17 +- extensions/opc/src/FetchOPCHistory.cpp | 398 +++++++++++++++++ extensions/opc/src/OPCCommon.cpp | 135 +++--- extensions/opc/tests/FetchOPCHistoryTests.cpp | 405 ++++++++++++++++++ .../opc/tests/FetchOPCProcessorTests.cpp | 16 +- extensions/opc/tests/OpcUaTestServer.h | 357 +++++++++++++-- extensions/opc/tests/features/opcua.feature | 94 ++++ extensions/opc/tests/features/steps/steps.py | 7 + 15 files changed, 1756 insertions(+), 174 deletions(-) create mode 100644 extensions/opc/include/FetchOPCHistory.h create mode 100644 extensions/opc/src/FetchOPCHistory.cpp create mode 100644 extensions/opc/tests/FetchOPCHistoryTests.cpp diff --git a/.github/references/ubuntu_22_04_clang_arm_manifest.json b/.github/references/ubuntu_22_04_clang_arm_manifest.json index 139b3854d7..eb54a92cf8 100644 --- a/.github/references/ubuntu_22_04_clang_arm_manifest.json +++ b/.github/references/ubuntu_22_04_clang_arm_manifest.json @@ -6506,6 +6506,185 @@ "bundles": { "componentManifest": { "processors": [ + { + "propertyDescriptors": { + "Application URI": { + "name": "Application URI", + "description": "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.", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Batch Size": { + "name": "Batch Size", + "description": "Maximum number entries to read and return in a single batch. If set to zero or empty all available entries are returned.", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Certificate path": { + "name": "Certificate path", + "description": "Path to the DER-encoded cert file", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "End timestamp": { + "name": "End timestamp", + "description": "Timestamp before which the events should be returned. If not specified entries are returned until the current time.", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "History Read Type": { + "name": "History Read Type", + "description": "Whether to fetch raw historical values or the audit trail of modifications to historical values", + "validator": "VALID", + "required": "true", + "sensitive": "false", + "expressionLanguageScope": "NONE", + "defaultValue": "Raw", + "allowableValues": [ + { + "value": "Raw", + "displayName": "Raw" + }, + { + "value": "Audit", + "displayName": "Audit" + } + ] + }, + "Key path": { + "name": "Key path", + "description": "Path to the DER-encoded key file", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Namespace index": { + "name": "Namespace index", + "description": "The index of the namespace.", + "validator": "INTEGER_VALIDATOR", + "required": "true", + "sensitive": "false", + "expressionLanguageScope": "NONE", + "defaultValue": "0" + }, + "Node ID": { + "name": "Node ID", + "description": "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'.", + "validator": "VALID", + "required": "true", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Node ID type": { + "name": "Node ID type", + "description": "Specifies the type of the provided node ID", + "validator": "VALID", + "required": "true", + "sensitive": "false", + "expressionLanguageScope": "NONE", + "allowableValues": [ + { + "value": "Path", + "displayName": "Path" + }, + { + "value": "Int", + "displayName": "Int" + }, + { + "value": "String", + "displayName": "String" + }, + { + "value": "Guid", + "displayName": "Guid" + } + ] + }, + "OPC server endpoint": { + "name": "OPC server endpoint", + "description": "Specifies the address, port and relative path of an OPC endpoint", + "validator": "VALID", + "required": "true", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Password": { + "name": "Password", + "description": "Password to log in with.", + "validator": "VALID", + "required": "false", + "sensitive": "true", + "expressionLanguageScope": "NONE" + }, + "Path reference types": { + "name": "Path reference types", + "description": "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.", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Record Set Writer": { + "typeProvidedByValue": { + "type": "org.apache.nifi.minifi.core.RecordSetWriter", + "group": "org.apache.nifi.minifi", + "artifact": "minifi-system" + }, + "name": "Record Set Writer", + "description": "Specifies the Controller Service to use for writing results to a FlowFile instead of using the default output format.", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Start timestamp": { + "name": "Start timestamp", + "description": "Timestamp after which the events should be returned. If not specified entries are returned from the beginning of the history.", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Trusted server certificate path": { + "name": "Trusted server certificate path", + "description": "Comma separated list of paths to the DER-encoded trusted server certificates", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Username": { + "name": "Username", + "description": "Username to log in with.", + "validator": "VALID", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_FORBIDDEN", + "isSingleThreaded": "true", + "supportedRelationships": [ + { + "name": "success", + "description": "Successfully retrieved OPC-UA node history entries" + } + ], + "typeDescription": "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.", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "org.apache.nifi.minifi.processors.FetchOPCHistory" + }, { "propertyDescriptors": { "Application URI": { diff --git a/PROCESSORS.md b/PROCESSORS.md index 4c7705b2a6..5c18e0081e 100644 --- a/PROCESSORS.md +++ b/PROCESSORS.md @@ -41,6 +41,7 @@ limitations under the License. - [FetchFile](#FetchFile) - [FetchGCSObject](#FetchGCSObject) - [FetchModbusTcp](#FetchModbusTcp) +- [FetchOPCHistory](#FetchOPCHistory) - [FetchOPCProcessor](#FetchOPCProcessor) - [FetchS3Object](#FetchS3Object) - [FetchSFTP](#FetchSFTP) @@ -887,6 +888,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.
**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
Int
String
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
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. | + + ## FetchOPCProcessor ### Description @@ -922,15 +970,15 @@ In the list below, the names of required properties appear in bold. Any other pr ### Output Attributes -| Attribute | Relationship | Description | -|-----------------|--------------|----------------------------------------------------------------------------------------| -| NodeID | success | ID of the node. | -| NodeID type | success | Type of the node ID. | -| Browsename | success | The browse name of the node. | -| Full path | success | The full path of the node. | -| Sourcetimestamp | success | The timestamp of when the node was created in the server as 'MM-dd-yyyy HH:mm:ss.mmm'. | -| Typename | success | The type name of the node data. | -| Datasize | success | The size of the node data. | +| Attribute | Relationship | Description | +|-----------------|--------------|-----------------------------------------------------------------------------------------| +| NodeID | success | ID of the node. | +| NodeID type | success | Type of the node ID. | +| Browsename | success | The browse name of the node. | +| Full path | success | The full path of the node. | +| Sourcetimestamp | success | The timestamp of when the node was created in the server as 'YYYY-MM-DDTHH:MM:SS.sssZ'. | +| Typename | success | The type name of the node data. | +| Datasize | success | The size of the node data. | ## FetchS3Object diff --git a/README.md b/README.md index a26a995c5e..3c1642ea1a 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,7 @@ The next table outlines CMAKE flags that correspond with MiNiFi extensions. Exte | LlamaCpp | [RunLlamaCppInference](PROCESSORS.md#runllamacppinference) | -DENABLE_LLAMACPP=ON | | Lua Scripting | [ExecuteScript](PROCESSORS.md#executescript) | -DENABLE_LUA_SCRIPTING=ON | | MQTT | [ConsumeMQTT](PROCESSORS.md#consumemqtt)
[PublishMQTT](PROCESSORS.md#publishmqtt) | -DENABLE_MQTT=ON | -| OPC | [FetchOPCProcessor](PROCESSORS.md#fetchopcprocessor)
[PutOPCProcessor](PROCESSORS.md#putopcprocessor) | -DENABLE_OPC=ON | +| OPC | [FetchOPCHistory](PROCESSORS.md#fetchopchistory)
[FetchOPCProcessor](PROCESSORS.md#fetchopcprocessor)
[PutOPCProcessor](PROCESSORS.md#putopcprocessor) | -DENABLE_OPC=ON | | PDH (Windows) | [PerformanceDataMonitor](PROCESSORS.md#performancedatamonitor) | -DENABLE_PDH=ON | | ProcFs (Linux) | [ProcFsMonitor](PROCESSORS.md#procfsmonitor) | -DENABLE_PROCFS=ON | | Python Scripting | [ExecuteScript](PROCESSORS.md#executescript)
[**Custom Python Processors**](extensions/python/PYTHON.md) | -DENABLE_PYTHON_SCRIPTING=ON | diff --git a/behave_framework/src/minifi_behave/steps/checking_steps.py b/behave_framework/src/minifi_behave/steps/checking_steps.py index 83ff073b7d..31c9e7650a 100644 --- a/behave_framework/src/minifi_behave/steps/checking_steps.py +++ b/behave_framework/src/minifi_behave/steps/checking_steps.py @@ -73,6 +73,9 @@ def verify_single_file_content_in_directory(context: MinifiTestContext, content: ) +@then( + "in the '{container_name}' container at least one file with the content '{content}' is placed in the '{directory}' directory in less than {duration}" +) @then( 'in the "{container_name}" container at least one file with the content "{content}" is placed in the "{directory}" directory in less than {duration}' ) @@ -223,24 +226,39 @@ def verify_at_least_number_of_files_in_directory(context: MinifiTestContext, num ) -@then('at least one file in "{directory}" content match the following regex: "{regex_str}" in less than {duration}') @then( - "the content of at least one file in the \"{directory}\" directory matches the '{regex_str}' regex in less than {duration}" + 'in the "{container_name}" container at least one file in "{directory}" content match the following regex: "{regex_str}" in less than {duration}' ) -def verify_file_content_matches_regex_in_directory( - context: MinifiTestContext, directory: str, regex_str: str, duration: str +@then( + "in the '{container_name}' container at least one file in '{directory}' content match the following regex: '{regex_str}' in less than {duration}" +) +@then( + "in the '{container_name}' container the content of at least one file in the \"{directory}\" directory matches the '{regex_str}' regex in less than {duration}" +) +def verify_file_content_matches_regex_in_directory_in_container( + context: MinifiTestContext, container_name: str, directory: str, regex_str: str, duration: str ): duration_seconds = humanfriendly.parse_timespan(duration) assert wait_for_condition( - condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].directory_contains_file_with_regex( - directory, regex_str - ), + condition=lambda: context.containers[container_name].directory_contains_file_with_regex(directory, regex_str), timeout_seconds=duration_seconds, - bail_condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].exited, + bail_condition=lambda: context.containers[container_name].exited, context=context, ) +@then('at least one file in "{directory}" content match the following regex: "{regex_str}" in less than {duration}') +@then( + "the content of at least one file in the \"{directory}\" directory matches the '{regex_str}' regex in less than {duration}" +) +def verify_file_content_matches_regex_in_directory( + context: MinifiTestContext, directory: str, regex_str: str, duration: str +): + context.execute_steps( + f'then in the "{DEFAULT_MINIFI_CONTAINER_NAME}" container at least one file in "{directory}" content match the following regex: "{regex_str}" in less than {duration}' + ) + + @then( 'files with contents "{content_one}" and "{content_two}" are placed in the "{directory}" directory in less than {timeout}' ) @@ -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}' +) +def verify_files_with_at_least_contents_in_directory( + context: MinifiTestContext, directory: str, timeout: str, contents: str +): + context.execute_steps( + f'then in the "{DEFAULT_MINIFI_CONTAINER_NAME}" container files with at least these contents "{contents}" are placed in the "{directory}" directory in less than {timeout}' + ) + + +@then( + 'in the "{container_name}" container a file with the JSON content "{content}" is placed in the "{directory}" directory in less than {duration}' +) +@then( + "in the '{container_name}' container 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_in_container( + context: MinifiTestContext, container_name: str, content: str, directory: str, duration: str +): timeout_in_seconds = humanfriendly.parse_timespan(duration) assert wait_for_condition( - condition=lambda: context.containers[DEFAULT_MINIFI_CONTAINER_NAME].verify_path_with_json_content( - directory, content - ), + condition=lambda: context.containers[container_name].verify_path_with_json_content(directory, content), timeout_seconds=timeout_in_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): + context.execute_steps( + f'then in the "{DEFAULT_MINIFI_CONTAINER_NAME}" container a file with the JSON content "{content}" is placed in the "{directory}" directory in less than {duration}' + ) + + @then("MiNiFi's memory usage does not increase by more than {max_increase} after {duration}") def verify_minifi_memory_usage_increase(context: MinifiTestContext, max_increase: str, duration: str): time_in_seconds = humanfriendly.parse_timespan(duration) diff --git a/behave_framework/src/minifi_behave/steps/flow_building_steps.py b/behave_framework/src/minifi_behave/steps/flow_building_steps.py index 29b8a6876a..615c5e9ed8 100644 --- a/behave_framework/src/minifi_behave/steps/flow_building_steps.py +++ b/behave_framework/src/minifi_behave/steps/flow_building_steps.py @@ -457,6 +457,15 @@ def set_processor_timer_driven(context: MinifiTestContext, processor_name: str, processor.scheduling_period = scheduling_period +@given('a {service_name} controller service is set up in the "{minifi_container_name}" flow') +@given('an {service_name} controller service is set up in the "{minifi_container_name}" flow') +def setup_controller_service_in_minifi_flow(context: MinifiTestContext, service_name: str, minifi_container_name: str): + controller_service = ControllerService(class_name=service_name, service_name=service_name) + context.get_or_create_minifi_container(minifi_container_name).flow_definition.controller_services.append( + controller_service + ) + + @given("a {service_name} controller service is set up") @given("an {service_name} controller service is set up") def setup_controller_service(context: MinifiTestContext, service_name: str): diff --git a/extensions/opc/include/FetchOPCHistory.h b/extensions/opc/include/FetchOPCHistory.h new file mode 100644 index 0000000000..e487d043f5 --- /dev/null +++ b/extensions/opc/include/FetchOPCHistory.h @@ -0,0 +1,158 @@ +/** + * 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 +#include +#include +#include +#include +#include +#include +#include +#include + +#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 fingerprints; +}; + +struct FetchOPCHistoryContext { + core::ProcessSession& session; + std::shared_ptr record_set_writer; + std::unordered_map& state_map; + size_t& entries_transferred; + const uint64_t batch_size; + const std::string& node_id; + const int32_t namespace_index; + const std::optional fetched_state; + std::shared_ptr 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()>::createProperty("Node ID type") + .withDescription("Specifies the type of the provided node ID") + .isRequired(true) + .withAllowedValues(magic_enum::enum_names()) + .build(); + EXTENSIONAPI static constexpr auto NodeID = + core::PropertyDefinitionBuilder<>::createProperty("Node ID") + .withDescription( + "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'.") + .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()>::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()) + .withDefaultValue(magic_enum::enum_name()) + .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() + .build(); + EXTENSIONAPI static constexpr auto Properties = utils::array_cat(BaseOPCProcessor::Properties, + std::to_array( + {NodeIDType, NodeID, NameSpaceIndex, StartTimestamp, EndTimestamp, BatchSize, HistoryReadType, RecordSetWriter})); + + EXTENSIONAPI static constexpr auto Success = core::RelationshipDefinition{"success", "Successfully retrieved OPC-UA node history entries"}; + EXTENSIONAPI static constexpr auto Relationships = std::array{Success}; + + EXTENSIONAPI static constexpr auto NodeIDAttr = core::OutputAttributeDefinition<>{"NodeID", {Success}, "ID of the node."}; + EXTENSIONAPI static constexpr auto NamespaceIndexAttr = core::OutputAttributeDefinition<>{ + "Namespace index", {Success}, "Namespace index of the node."}; + EXTENSIONAPI static constexpr auto SourcetimestampAttr = core::OutputAttributeDefinition<>{ + "Sourcetimestamp", {Success}, "The timestamp of when the node was created in the server as 'YYYY-MM-DDTHH:MM:SS.sssZ'."}; + EXTENSIONAPI static constexpr auto ModificationUsernameAttr = core::OutputAttributeDefinition<>{ + "ModificationUsername", {Success}, "Username of the user who modified the node."}; + EXTENSIONAPI static constexpr auto ModificationTimeAttr = core::OutputAttributeDefinition<>{ + "ModificationTime", {Success}, "Timestamp of when the node was modified."}; + EXTENSIONAPI static constexpr auto ModificationUpdateTypeAttr = core::OutputAttributeDefinition<>{ + "ModificationUpdateType", {Success}, "Type of modification performed on the node."}; + + EXTENSIONAPI static constexpr auto OutputAttributes = std::array{ + NodeIDAttr, NamespaceIndexAttr, SourcetimestampAttr, ModificationUsernameAttr, ModificationTimeAttr, ModificationUpdateTypeAttr}; + + EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false; + EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false; + EXTENSIONAPI static constexpr core::annotation::Input InputRequirement = core::annotation::Input::INPUT_FORBIDDEN; + EXTENSIONAPI static constexpr bool IsSingleThreaded = true; + + ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS + + void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& factory) override; + void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override; + void initialize() override; + + private: + std::optional parseFetchedState(const std::unordered_map& state_map); + + opc::HistoryReadTypeOption history_type_ = opc::HistoryReadTypeOption::Raw; + std::optional start_timestamp_; + std::optional end_timestamp_; + uint64_t batch_size_ = 0; + std::shared_ptr record_set_writer_; + opc::NodeId node_; + bool path_node_id_resolved_ = false; +}; + +} // namespace org::apache::nifi::minifi::processors diff --git a/extensions/opc/include/FetchOPCProcessor.h b/extensions/opc/include/FetchOPCProcessor.h index 0bf5abd3ab..5ae0d0ba96 100644 --- a/extensions/opc/include/FetchOPCProcessor.h +++ b/extensions/opc/include/FetchOPCProcessor.h @@ -73,10 +73,10 @@ class FetchOPCProcessor final : public BaseOPCProcessor { EXTENSIONAPI static constexpr const char* Description = "Fetches OPC-UA node"; - EXTENSIONAPI static constexpr auto NodeIDType = core::PropertyDefinitionBuilder()>::createProperty("Node ID type") + EXTENSIONAPI static constexpr auto NodeIDType = core::PropertyDefinitionBuilder<3>::createProperty("Node ID type") .withDescription("Specifies the type of the provided node ID") .isRequired(true) - .withAllowedValues(magic_enum::enum_names()) + .withAllowedValues({"Path", "Int", "String"}) .build(); EXTENSIONAPI static constexpr auto NodeID = core::PropertyDefinitionBuilder<>::createProperty("Node ID") .withDescription("Specifies the ID of the root node to traverse. In case of a Path Node ID Type, the path should be provided in the format of 'path/to/node'.") @@ -119,7 +119,7 @@ class FetchOPCProcessor final : public BaseOPCProcessor { EXTENSIONAPI static constexpr auto BrowsenameAttr = core::OutputAttributeDefinition<>{"Browsename", { Success }, "The browse name of the node."}; EXTENSIONAPI static constexpr auto FullPathAttr = core::OutputAttributeDefinition<>{"Full path", { Success }, "The full path of the node."}; EXTENSIONAPI static constexpr auto SourcetimestampAttr = core::OutputAttributeDefinition<>{"Sourcetimestamp", { Success }, - "The timestamp of when the node was created in the server as 'MM-dd-yyyy HH:mm:ss.mmm'."}; + "The timestamp of when the node was created in the server as 'YYYY-MM-DDTHH:MM:SS.sssZ'."}; EXTENSIONAPI static constexpr auto TypenameAttr = core::OutputAttributeDefinition<>{"Typename", { Success }, "The type name of the node data."}; EXTENSIONAPI static constexpr auto DatasizeAttr = core::OutputAttributeDefinition<>{"Datasize", { Success }, "The size of the node data."}; diff --git a/extensions/opc/include/OPCCommon.h b/extensions/opc/include/OPCCommon.h index df780de4be..acc90b815e 100644 --- a/extensions/opc/include/OPCCommon.h +++ b/extensions/opc/include/OPCCommon.h @@ -34,6 +34,11 @@ namespace org::apache::nifi::minifi::opc { +typedef UA_Boolean +(*HistoryCallback)( + UA_Client *client, const UA_NodeId *nodeId, UA_Boolean moreDataAvailable, + const UA_ExtensionObject *data, void *callbackContext); + class OPCException : public minifi::Exception { public: OPCException(ExceptionType type, std::string &&error_msg) @@ -44,7 +49,8 @@ class OPCException : public minifi::Exception { enum class OPCNodeIDType{ Path, Int, - String + String, + Guid }; enum class OPCNodeDataType{ @@ -58,6 +64,11 @@ enum class OPCNodeDataType{ String }; +enum class HistoryReadTypeOption { + Raw, + Audit +}; + // RAII owner for a UA_NodeId that calls UA_NodeId_clear to free node id allocation class NodeId { public: @@ -118,6 +129,8 @@ class Client { template 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); static std::unique_ptr createClient(const std::shared_ptr& logger, const std::string& application_uri, const std::vector& cert_buffer, const std::vector& key_buffer, @@ -176,6 +189,8 @@ struct NodeData { std::string nodeValue2String(const NodeData& nd); +std::string variantToString(const UA_Variant& variant); + std::string OPCDateTime2String(UA_DateTime raw_date); void logFunc(void *context, UA_LogLevel level, UA_LogCategory category, const char *msg, va_list args); diff --git a/extensions/opc/src/FetchOPCHistory.cpp b/extensions/opc/src/FetchOPCHistory.cpp new file mode 100644 index 0000000000..15e7648663 --- /dev/null +++ b/extensions/opc/src/FetchOPCHistory.cpp @@ -0,0 +1,398 @@ +/** + * 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 +#include +#include + +#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(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 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 extractHistoryBatch(const UA_ExtensionObject* data, const std::shared_ptr& 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(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(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 selectNewEntries(std::vector entries, const std::optional& last_fetched, + bool has_modification_info, std::optional max_entries) { + std::vector 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& 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& entries) { + for (const auto& entry : entries) { + auto flow_file = context.session.create(); + context.session.write(flow_file, [&entry](const std::shared_ptr& output_stream) -> io::IoResult { + output_stream->write(reinterpret_cast(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& state_map, const std::vector& 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(ctx); + + auto batch = extractHistoryBatch(data, opc_history_context->logger); + if (batch && !batch->entries.empty()) { + const std::optional remaining = opc_history_context->batch_size != 0 + ? std::optional(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(tp.time_since_epoch()).count(); + return UA_DATETIME_UNIX_EPOCH + usec_since_epoch * UA_DATETIME_USEC; +} + +UA_DateTime calculateStartTime(const std::optional& fetched_state, + const std::optional& 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& 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(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(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: + 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(context, HistoryReadType); + start_timestamp_ = utils::parseOptionalProperty(context, StartTimestamp) | utils::andThen(utils::timeutils::parseRfc3339); + end_timestamp_ = utils::parseOptionalProperty(context, EndTimestamp) | utils::andThen(utils::timeutils::parseRfc3339); + batch_size_ = utils::parseOptionalU64Property(context, BatchSize).value_or(0); + const auto record_set_writer_name = context.getProperty(RecordSetWriter).value_or(""); + auto controller_service = context.getControllerService(record_set_writer_name, getUUID()); + if (!record_set_writer_name.empty() && !controller_service) { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Controller service '{}' not found", record_set_writer_name)); + } + record_set_writer_ = std::dynamic_pointer_cast(controller_service); +} + +std::optional FetchOPCHistory::parseFetchedState(const std::unordered_map& state_map) { + const auto timestamp_it = state_map.find(LAST_FETCHED_TIMESTAMP_KEY); + const auto fingerprints_it = state_map.find(LAST_FETCHED_FINGERPRINT_KEY); + if (timestamp_it == state_map.end() || fingerprints_it == state_map.end()) { + return std::nullopt; + } + + FetchedState state; + try { + state.timestamp = std::stoll(timestamp_it->second); + } catch (const std::exception&) { + logger_->log_error("Failed to parse timestamp from state map: {}", timestamp_it->second); + return std::nullopt; + } + + for (auto& fingerprint : utils::string::split(fingerprints_it->second, ",")) { + if (!fingerprint.empty()) { + state.fingerprints.insert(std::move(fingerprint)); + } + } + return state; +} + +void FetchOPCHistory::onTrigger(core::ProcessContext& context, core::ProcessSession& session) { + logger_->log_trace("FetchOPCHistory::onTrigger"); + + if (!reconnect()) { + context.yield(); + return; + } + + if (id_type_ == opc::OPCNodeIDType::Path && !path_node_id_resolved_) { + std::vector translated_node_ids; + auto sc = connection_->translateBrowsePathsToNodeIdsRequest(node_id_, translated_node_ids, namespace_idx_, path_reference_types_, logger_); + if (sc != UA_STATUSCODE_GOOD) { + logger_->log_error("Failed to translate path '{}' to a node id: {}", node_id_, UA_StatusCode_name(sc)); + context.yield(); + return; + } + if (translated_node_ids.size() != 1) { + logger_->log_error("Path '{}' resolved to {} node ids; exactly one is required to fetch history", node_id_, translated_node_ids.size()); + context.yield(); + return; + } + node_ = std::move(translated_node_ids[0]); + path_node_id_resolved_ = true; + } + + auto* state_manager = context.getStateManager(); + std::unordered_map state_map; + + state_manager->get(state_map); + const auto fetched_state = parseFetchedState(state_map); + + size_t entries_transferred = 0; + FetchOPCHistoryContext + history_context{session, record_set_writer_, state_map, entries_transferred, batch_size_, node_id_, namespace_idx_, fetched_state, logger_}; + + auto retval = connection_->readHistory(history_type_, + node_, + &historyReadCallback, + calculateStartTime(fetched_state, start_timestamp_), + calculateEndTime(end_timestamp_), + static_cast(&history_context)); + + if (retval != UA_STATUSCODE_GOOD) { + logger_->log_error("Failed to read OPC UA node history, status code: {}", UA_StatusCode_name(retval)); + context.yield(); + return; + } + + if (!state_manager->set(state_map)) { + logger_->log_warn("Failed to persist FetchOPCHistory state, entries may be re-fetched on the next trigger"); + } +} + +REGISTER_RESOURCE(FetchOPCHistory, Processor); + +} // namespace org::apache::nifi::minifi::processors diff --git a/extensions/opc/src/OPCCommon.cpp b/extensions/opc/src/OPCCommon.cpp index 8bf836a4a9..8a1beadd37 100644 --- a/extensions/opc/src/OPCCommon.cpp +++ b/extensions/opc/src/OPCCommon.cpp @@ -493,107 +493,64 @@ template UA_StatusCode Client::add_node(const UA_NodeId parent_nod template UA_StatusCode Client::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, std::string value, UA_NodeId *received_node_id); -std::string nodeValue2String(const NodeData& nd) { - std::string ret_val; - switch (nd.data_type_id) { +std::string variantToString(const UA_Variant& variant) { + if (variant.type == nullptr || variant.data == nullptr) { + throw OPCException(GENERAL_EXCEPTION, "Cannot convert an empty variant to string"); + } + switch (variant.type->typeKind) { case UA_DATATYPEKIND_STRING: case UA_DATATYPEKIND_LOCALIZEDTEXT: case UA_DATATYPEKIND_BYTESTRING: { - UA_String value = *reinterpret_cast(nd.var_->data); - ret_val = std::string(reinterpret_cast(value.data), value.length); - break; - } - case UA_DATATYPEKIND_BOOLEAN: { - bool b = false; - memcpy(&b, nd.data.data(), sizeof(bool)); - ret_val = b ? "True" : "False"; - break; - } - case UA_DATATYPEKIND_SBYTE: { - int8_t i8t = 0; - memcpy(&i8t, nd.data.data(), sizeof(i8t)); - ret_val = std::to_string(i8t); - break; - } - case UA_DATATYPEKIND_BYTE: { - uint8_t ui8t = 0; - memcpy(&ui8t, nd.data.data(), sizeof(ui8t)); - ret_val = std::to_string(ui8t); - break; - } - case UA_DATATYPEKIND_INT16: { - int16_t i16t = 0; - memcpy(&i16t, nd.data.data(), sizeof(i16t)); - ret_val = std::to_string(i16t); - break; + const auto *value = static_cast(variant.data); + return {reinterpret_cast(value->data), value->length}; } - case UA_DATATYPEKIND_UINT16: { - uint16_t ui16t = 0; - memcpy(&ui16t, nd.data.data(), sizeof(ui16t)); - ret_val = std::to_string(ui16t); - break; - } - case UA_DATATYPEKIND_INT32: { - int32_t i32t = 0; - memcpy(&i32t, nd.data.data(), sizeof(i32t)); - ret_val = std::to_string(i32t); - break; - } - case UA_DATATYPEKIND_UINT32: { - uint32_t ui32t = 0; - memcpy(&ui32t, nd.data.data(), sizeof(ui32t)); - ret_val = std::to_string(ui32t); - break; - } - case UA_DATATYPEKIND_INT64: { - int64_t i64t = 0; - memcpy(&i64t, nd.data.data(), sizeof(i64t)); - ret_val = std::to_string(i64t); - break; - } - case UA_DATATYPEKIND_UINT64: { - uint64_t ui64t = 0; - memcpy(&ui64t, nd.data.data(), sizeof(ui64t)); - ret_val = std::to_string(ui64t); - break; - } - case UA_DATATYPEKIND_FLOAT: { + case UA_DATATYPEKIND_BOOLEAN: + return *static_cast(variant.data) ? "True" : "False"; + case UA_DATATYPEKIND_SBYTE: + return std::to_string(*static_cast(variant.data)); + case UA_DATATYPEKIND_BYTE: + return std::to_string(*static_cast(variant.data)); + case UA_DATATYPEKIND_INT16: + return std::to_string(*static_cast(variant.data)); + case UA_DATATYPEKIND_UINT16: + return std::to_string(*static_cast(variant.data)); + case UA_DATATYPEKIND_INT32: + return std::to_string(*static_cast(variant.data)); + case UA_DATATYPEKIND_UINT32: + return std::to_string(*static_cast(variant.data)); + case UA_DATATYPEKIND_INT64: + return std::to_string(*static_cast(variant.data)); + case UA_DATATYPEKIND_UINT64: + return std::to_string(*static_cast(variant.data)); + case UA_DATATYPEKIND_FLOAT: if (sizeof(float) == 4 && std::numeric_limits::is_iec559) { - float f = 0; - memcpy(&f, nd.data.data(), sizeof(float)); - ret_val = std::to_string(f); - } else { - throw OPCException(GENERAL_EXCEPTION, "Float is non-standard on this system, OPC data cannot be extracted!"); + return std::to_string(*static_cast(variant.data)); } - break; - } - case UA_DATATYPEKIND_DOUBLE: { + throw OPCException(GENERAL_EXCEPTION, "Float is non-standard on this system, OPC data cannot be extracted!"); + case UA_DATATYPEKIND_DOUBLE: if (sizeof(double) == 8 && std::numeric_limits::is_iec559) { - double d = 0; - memcpy(&d, nd.data.data(), sizeof(double)); - ret_val = std::to_string(d); - } else { - throw OPCException(GENERAL_EXCEPTION, "Double is non-standard on this system, OPC data cannot be extracted!"); + return std::to_string(*static_cast(variant.data)); } - break; - } - case UA_DATATYPEKIND_DATETIME: { - UA_DateTime dt = 0; - memcpy(&dt, nd.data.data(), sizeof(UA_DateTime)); - ret_val = opc::OPCDateTime2String(dt); - break; - } + throw OPCException(GENERAL_EXCEPTION, "Double is non-standard on this system, OPC data cannot be extracted!"); + case UA_DATATYPEKIND_DATETIME: + return opc::OPCDateTime2String(*static_cast(variant.data)); default: - throw OPCException(GENERAL_EXCEPTION, "Data type is not supported "); + throw OPCException(GENERAL_EXCEPTION, "Data type is not supported: " + std::string(variant.type->typeName)); } - return ret_val; +} + +std::string nodeValue2String(const NodeData& nd) { + if (nd.var_ == nullptr) { + throw OPCException(GENERAL_EXCEPTION, "Node has no value to convert"); + } + return variantToString(*nd.var_); } std::string OPCDateTime2String(UA_DateTime raw_date) { UA_DateTimeStruct dts = UA_DateTime_toStruct(raw_date); std::array charBuf{}; - int sz = snprintf(charBuf.data(), charBuf.size(), "%02hu-%02hu-%04hu %02hu:%02hu:%02hu.%03hu", dts.day, dts.month, dts.year, dts.hour, dts.min, dts.sec, dts.milliSec); + int sz = snprintf(charBuf.data(), charBuf.size(), "%04hu-%02hu-%02huT%02hu:%02hu:%02hu.%03huZ", dts.year, dts.month, dts.day, dts.hour, dts.min, dts.sec, dts.milliSec); return {charBuf.data(), gsl::narrow(sz)}; } @@ -619,4 +576,12 @@ std::optional mapOpcReferenceType(const std::string& ref_type) { return std::nullopt; } +UA_StatusCode Client::readHistory(HistoryReadTypeOption history_type, const UA_NodeId& node_id, const HistoryCallback callback, UA_DateTime start_time, UA_DateTime end_time, + void* callback_context) { + if (history_type == HistoryReadTypeOption::Audit) { + return UA_Client_HistoryRead_modified(client_, &node_id, callback, start_time, end_time, UA_STRING_NULL, false, 0, UA_TIMESTAMPSTORETURN_SOURCE, callback_context); + } + return UA_Client_HistoryRead_raw(client_, &node_id, callback, start_time, end_time, UA_STRING_NULL, false, 0, UA_TIMESTAMPSTORETURN_SOURCE, callback_context); +} + } // namespace org::apache::nifi::minifi::opc diff --git a/extensions/opc/tests/FetchOPCHistoryTests.cpp b/extensions/opc/tests/FetchOPCHistoryTests.cpp new file mode 100644 index 0000000000..c1015b4c01 --- /dev/null +++ b/extensions/opc/tests/FetchOPCHistoryTests.cpp @@ -0,0 +1,405 @@ +/** + * + * 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 +#include + +#include "OpcUaTestServer.h" +#include "catch2/generators/catch_generators.hpp" +#include "include/FetchOPCHistory.h" +#include "rapidjson/document.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("FetchOPCHistory")), + processor_(controller_.getProcessor()) { + LogTestController::getInstance().setDebug(); + LogTestController::getInstance().setDebug(); + LogTestController::getInstance().setTrace(); + LogTestController::getInstance().setDebug(); + } + + 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:4842/")); + 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& 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); + } + + static void checkModificationAttributes(const std::shared_ptr& 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) const { + 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); + } + + protected: + minifi::test::SingleProcessorTestController controller_; + core::Processor* processor_; + OpcUaTestServer server_{4842}; +}; + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test fetching history of node with a single entry", "[fetchopchistory]") { + server_.start(); + setupProcessor("String", "INT1"); + const auto contains_modification_attributes = GENERATE(true, false); + if (contains_modification_attributes) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + const auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 1); + auto flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "1", "INT1", "2024-06-15T10:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "test_user", "Replace", "2024-06-15T10:30:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test fetching history of node with a single integer nodeid entry", "[fetchopchistory]") { + server_.start(); + setupProcessor("Int", "666"); + const auto contains_modification_attributes = GENERATE(true, false); + if (contains_modification_attributes) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + const auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 1); + auto flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "256", "666", "2001-01-01T22:22:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "integer_user", "Insert", "2001-01-01T22:22:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test fetching history after a specific timestamp", "[fetchopchistory]") { + server_.start(); + setupProcessor("String", "INT2"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::StartTimestamp.name, "2025-10-01T00:00:00Z")); + + const auto contains_modification_attributes = GENERATE(true, false); + if (contains_modification_attributes) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + const auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 2); + auto flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "3", "INT2", "2025-11-11T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "admin_user", "Update", "2025-11-11T11:30:00.000Z"); + + flow_file = results.at(processors::FetchOPCHistory::Success)[1]; + checkFlowFile(flow_file, "4", "INT2", "2026-03-11T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "test_user", "Replace", "2026-03-11T11:30:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test fetching history before a specific timestamp", "[fetchopchistory]") { + server_.start(); + setupProcessor("String", "INT2"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::EndTimestamp.name, "2025-11-12T00:00:00Z")); + + const auto contains_modification_attributes = GENERATE(true, false); + if (contains_modification_attributes) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + const auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 2); + auto flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "2", "INT2", "2021-03-15T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "admin_user", "Insert", "2021-03-15T11:30:00.000Z"); + + flow_file = results.at(processors::FetchOPCHistory::Success)[1]; + checkFlowFile(flow_file, "3", "INT2", "2025-11-11T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "admin_user", "Update", "2025-11-11T11:30:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test batch size limit", "[fetchopchistory]") { + server_.start(); + setupProcessor("String", "INT2"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::BatchSize.name, "2")); + + const auto contains_modification_attributes = GENERATE(true, false); + if (contains_modification_attributes) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + const auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 2); + auto flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "2", "INT2", "2021-03-15T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "admin_user", "Insert", "2021-03-15T11:30:00.000Z"); + + flow_file = results.at(processors::FetchOPCHistory::Success)[1]; + checkFlowFile(flow_file, "3", "INT2", "2025-11-11T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "admin_user", "Update", "2025-11-11T11:30:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test batch size of zero returns all available entries", "[fetchopchistory]") { + server_.start(); + setupProcessor("String", "INT2"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::BatchSize.name, "0")); + + const auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 3); + checkFlowFile(results.at(processors::FetchOPCHistory::Success)[0], "2", "INT2", "2021-03-15T11:30:00.000Z"); + checkFlowFile(results.at(processors::FetchOPCHistory::Success)[1], "3", "INT2", "2025-11-11T11:30:00.000Z"); + checkFlowFile(results.at(processors::FetchOPCHistory::Success)[2], "4", "INT2", "2026-03-11T11:30:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test triggering again after all entries have been fetched returns nothing", "[fetchopchistory]") { + server_.start(); + setupProcessor("String", "INT2"); + + auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 3); + + results = controller_.trigger(); + CHECK(results.at(processors::FetchOPCHistory::Success).empty()); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test fetching history of a non-existing node yields no flow files", "[fetchopchistory]") { + server_.start(); + setupProcessor("String", "NONEXISTENT"); + + const auto results = controller_.trigger(); + CHECK(results.at(processors::FetchOPCHistory::Success).empty()); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test using a non-numeric node ID with the Int node ID type throws on schedule", + "[fetchopchistory]") { + server_.start(); + setupProcessor("Int", "not_a_number"); + + REQUIRE_THROWS_WITH(controller_.trigger(), "Process Schedule Operation: not_a_number cannot be used as an int type node ID"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test entries sharing a timestamp are all delivered once and then deduplicated", + "[fetchopchistory]") { + const auto shared_time = OpcUaTestServer::makeDateTime(2023, 1, 1, 0, 0, 0, 0); + server_.setHistory("INT3", + {HistoryModificationRecord{.value = 10, .username = "user", .update_type = UA_HISTORYUPDATETYPE_INSERT, .modification_time = shared_time}, + HistoryModificationRecord{.value = 20, .username = "user", .update_type = UA_HISTORYUPDATETYPE_INSERT, .modification_time = shared_time}}); + server_.start(); + setupProcessor("String", "INT3"); + + auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 2); + checkFlowFile(results.at(processors::FetchOPCHistory::Success)[0], "10", "INT3", "2023-01-01T00:00:00.000Z"); + checkFlowFile(results.at(processors::FetchOPCHistory::Success)[1], "20", "INT3", "2023-01-01T00:00:00.000Z"); + + results = controller_.trigger(); + CHECK(results.at(processors::FetchOPCHistory::Success).empty()); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, + "Test entries sharing the boundary timestamp are not re-fetched when the history is paginated across callbacks", "[fetchopchistory]") { + const auto shared_time = OpcUaTestServer::makeDateTime(2023, 1, 1, 0, 0, 0, 0); + server_.setHistory("INT3", + {HistoryModificationRecord{.value = 10, .username = "user", .update_type = UA_HISTORYUPDATETYPE_INSERT, .modification_time = shared_time}, + HistoryModificationRecord{.value = 20, .username = "user", .update_type = UA_HISTORYUPDATETYPE_INSERT, .modification_time = shared_time}}); + server_.setHistoryPageSize(1); + server_.start(); + setupProcessor("String", "INT3"); + + auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 2); + checkFlowFile(results.at(processors::FetchOPCHistory::Success)[0], "10", "INT3", "2023-01-01T00:00:00.000Z"); + checkFlowFile(results.at(processors::FetchOPCHistory::Success)[1], "20", "INT3", "2023-01-01T00:00:00.000Z"); + + results = controller_.trigger(); + CHECK(results.at(processors::FetchOPCHistory::Success).empty()); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test non-existing record set writer", "[fetchopchistory]") { + server_.start(); + auto json_record_set_writer = controller_.plan->addController("JsonRecordSetWriter", "JsonRecordSetWriter"); + REQUIRE(controller_.plan->setProperty(json_record_set_writer, "Output Grouping", "One Line Per Object")); + setupProcessor("String", "INT1"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::RecordSetWriter.name, "InvalidRecordSetWriter")); + + REQUIRE_THROWS_WITH(controller_.trigger(), "Process Schedule Operation: Controller service 'InvalidRecordSetWriter' not found"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test RecordSetWriter with JSON output format", "[fetchopchistory]") { + server_.start(); + auto json_record_set_writer = controller_.plan->addController("JsonRecordSetWriter", "JsonRecordSetWriter"); + REQUIRE(controller_.plan->setProperty(json_record_set_writer, "Output Grouping", "One Line Per Object")); + setupProcessor("String", "INT1"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::RecordSetWriter.name, "JsonRecordSetWriter")); + + std::string expected_json_content; + SECTION("Fetch full history") { + expected_json_content = R"({"Value":"1","Sourcetimestamp":"2024-06-15T10:30:00.000Z","NodeID":"INT1","Namespace index":")" + + std::to_string(server_.getNamespaceIndex()) + + "\"," + R"("ModificationUsername":"test_user","ModificationUpdateType":"Replace","ModificationTime":"2024-06-15T10:30:00.000Z"})"; + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + SECTION("Fetch raw history") { + expected_json_content = R"({"Value":"1","Sourcetimestamp":"2024-06-15T10:30:00.000Z","NodeID":"INT1","Namespace index":")" + + std::to_string(server_.getNamespaceIndex()) + "\"}"; + } + + const auto results = controller_.trigger(); + verifyResults(results, expected_json_content); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test RecordSetWriter with JSON output format with multiple values", "[fetchopchistory]") { + server_.start(); + auto json_record_set_writer = controller_.plan->addController("JsonRecordSetWriter", "JsonRecordSetWriter"); + setupProcessor("String", "INT2"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::RecordSetWriter.name, "JsonRecordSetWriter")); + + std::string expected_json_content; + SECTION("Fetch full history") { + expected_json_content = R"([{"Value":"2","Sourcetimestamp":"2021-03-15T11:30:00.000Z","NodeID":"INT2","Namespace index":")" + + std::to_string(server_.getNamespaceIndex()) + + "\"," + R"("ModificationUsername":"admin_user","ModificationUpdateType":"Insert","ModificationTime":"2021-03-15T11:30:00.000Z"}, )" + R"({"Value":"3","Sourcetimestamp":"2025-11-11T11:30:00.000Z","NodeID":"INT2","Namespace index":")" + + std::to_string(server_.getNamespaceIndex()) + + "\"," + R"("ModificationUsername":"admin_user","ModificationUpdateType":"Update","ModificationTime":"2025-11-11T11:30:00.000Z"}, )" + R"({"Value":"4","Sourcetimestamp":"2026-03-11T11:30:00.000Z","NodeID":"INT2","Namespace index":")" + + std::to_string(server_.getNamespaceIndex()) + + "\"," + R"("ModificationUsername":"test_user","ModificationUpdateType":"Replace","ModificationTime":"2026-03-11T11:30:00.000Z"}])"; + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + SECTION("Fetch raw history") { + expected_json_content = R"([{"Value":"2","Sourcetimestamp":"2021-03-15T11:30:00.000Z","NodeID":"INT2","Namespace index":")" + + std::to_string(server_.getNamespaceIndex()) + + "\"}," + R"({"Value":"3","Sourcetimestamp":"2025-11-11T11:30:00.000Z","NodeID":"INT2","Namespace index":")" + + std::to_string(server_.getNamespaceIndex()) + + "\"}," + R"({"Value":"4","Sourcetimestamp":"2026-03-11T11:30:00.000Z","NodeID":"INT2","Namespace index":")" + + std::to_string(server_.getNamespaceIndex()) + "\"}]"; + } + + const auto results = controller_.trigger(); + verifyResults(results, expected_json_content); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test multiple triggers with state kept in state manager", "[fetchopchistory]") { + server_.start(); + setupProcessor("String", "INT2"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::BatchSize.name, "1")); + + const auto contains_modification_attributes = GENERATE(true, false); + if (contains_modification_attributes) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 1); + auto flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "2", "INT2", "2021-03-15T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "admin_user", "Insert", "2021-03-15T11:30:00.000Z"); + + results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 1); + flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "3", "INT2", "2025-11-11T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "admin_user", "Update", "2025-11-11T11:30:00.000Z"); + + results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 1); + flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "4", "INT2", "2026-03-11T11:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "test_user", "Replace", "2026-03-11T11:30:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test fetching history of a node with a GUID node ID type", "[fetchopchistory]") { + server_.start(); + setupProcessor("Guid", "72962b91-fa75-4ae6-8d28-b404dc7daf63"); + const auto contains_modification_attributes = GENERATE(true, false); + if (contains_modification_attributes) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + const auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 1); + auto flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "7", "72962b91-fa75-4ae6-8d28-b404dc7daf63", "2020-05-20T12:00:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "guid_user", "Insert", "2020-05-20T12:00:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test using an invalid GUID with the Guid node ID type throws on schedule", "[fetchopchistory]") { + server_.start(); + setupProcessor("Guid", "not-a-guid"); + + REQUIRE_THROWS_WITH(controller_.trigger(), "Process Schedule Operation: not-a-guid cannot be used as a GUID type node ID"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test fetching history of a node resolved from a path node ID type", "[fetchopchistory]") { + server_.start(); + setupProcessor("Path", "Simulator/Default/Device1/INT1"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::PathReferenceTypes.name, "Organizes/Organizes/HasComponent")); + const auto contains_modification_attributes = GENERATE(true, false); + if (contains_modification_attributes) { + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::HistoryReadType.name, "Audit")); + } + + const auto results = controller_.trigger(); + REQUIRE(results.at(processors::FetchOPCHistory::Success).size() == 1); + auto flow_file = results.at(processors::FetchOPCHistory::Success)[0]; + checkFlowFile(flow_file, "1", "Simulator/Default/Device1/INT1", "2024-06-15T10:30:00.000Z"); + checkModificationAttributes(flow_file, contains_modification_attributes, "test_user", "Replace", "2024-06-15T10:30:00.000Z"); +} + +TEST_CASE_METHOD(FetchOPCHistoryTestController, "Test fetching history of a non-existing path yields no flow files", "[fetchopchistory]") { + server_.start(); + setupProcessor("Path", "Simulator/Default/Nope"); + REQUIRE(processor_->setProperty(processors::FetchOPCHistory::PathReferenceTypes.name, "Organizes/Organizes")); + + const auto results = controller_.trigger(); + CHECK(results.at(processors::FetchOPCHistory::Success).empty()); +} + +} // namespace org::apache::nifi::minifi::test diff --git a/extensions/opc/tests/FetchOPCProcessorTests.cpp b/extensions/opc/tests/FetchOPCProcessorTests.cpp index 98c335e35b..753ec4688f 100644 --- a/extensions/opc/tests/FetchOPCProcessorTests.cpp +++ b/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"); CHECK(flow_file->getAttribute("Typename") == "Int32"); CHECK(flow_file->getAttribute("Sourcetimestamp")); CHECK(controller.plan->getContent(flow_file) == std::to_string(i + 1)); @@ -54,7 +54,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/INT3/INT4"); CHECK(flow_file->getAttribute("NodeID")); - CHECK(flow_file->getAttribute("NodeID type") == "numeric"); + CHECK(flow_file->getAttribute("NodeID type") == "string"); CHECK(flow_file->getAttribute("Typename") == "Int32"); CHECK(flow_file->getAttribute("Sourcetimestamp")); CHECK(controller.plan->getContent(flow_file) == "4"); @@ -79,7 +79,7 @@ TEST_CASE("Test fetching using custom reference type id path", "[fetchopcprocess CHECK(flow_file->getAttribute("Datasize") == "4"); CHECK(flow_file->getAttribute("Full path") == "Simulator/Default/Device1/INT3"); CHECK(flow_file->getAttribute("NodeID")); - CHECK(flow_file->getAttribute("NodeID type") == "numeric"); + CHECK(flow_file->getAttribute("NodeID type") == "string"); CHECK(flow_file->getAttribute("Typename") == "Int32"); CHECK(flow_file->getAttribute("Sourcetimestamp")); CHECK(controller.plan->getContent(flow_file) == "3"); @@ -88,7 +88,7 @@ TEST_CASE("Test fetching using custom reference type id path", "[fetchopcprocess CHECK(flow_file->getAttribute("Datasize") == "4"); CHECK(flow_file->getAttribute("Full path") == "Simulator/Default/Device1/INT3/INT4"); CHECK(flow_file->getAttribute("NodeID")); - CHECK(flow_file->getAttribute("NodeID type") == "numeric"); + CHECK(flow_file->getAttribute("NodeID type") == "string"); CHECK(flow_file->getAttribute("Typename") == "Int32"); CHECK(flow_file->getAttribute("Sourcetimestamp")); CHECK(controller.plan->getContent(flow_file) == "4"); @@ -242,7 +242,7 @@ TEST_CASE("Test fetch for nodes with changed timestamps with lazy mode", "[fetch REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 4); - server.updateNodeTimestamp("Simulator/Default/Device1/INT3"); + server.updateNodeTimestamp("INT3"); results = controller.trigger(); REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 1); @@ -266,7 +266,7 @@ TEST_CASE("Test no fetch result using lazy new value mode when no values are cha REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 4); - server.updateNodeTimestamp("Simulator/Default/Device1/INT3"); + server.updateNodeTimestamp("INT3"); results = controller.trigger(); REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).empty()); @@ -289,8 +289,8 @@ TEST_CASE("Test fetching new values using lazy new value mode", "[fetchopcproces REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 4); - server.updateNodeTimestamp("Simulator/Default/Device1/INT3"); - server.updateNodeValue("Simulator/Default/Device1/INT2", 42); + server.updateNodeTimestamp("INT3"); + server.updateNodeValue("INT2", 42); results = controller.trigger(); REQUIRE(results.at(processors::FetchOPCProcessor::Failure).empty()); REQUIRE(results.at(processors::FetchOPCProcessor::Success).size() == 1); diff --git a/extensions/opc/tests/OpcUaTestServer.h b/extensions/opc/tests/OpcUaTestServer.h index 191e806ccc..b25f73c398 100644 --- a/extensions/opc/tests/OpcUaTestServer.h +++ b/extensions/opc/tests/OpcUaTestServer.h @@ -16,13 +16,23 @@ */ #pragma once +#include #include #include -#include -#include + #include -#include "unit/TestUtils.h" +#include +#include +#include +#include +#include +#include +#include +#include + +#include "include/OPCCommon.h" #include "unit/Catch.h" +#include "unit/TestUtils.h" extern "C" int mp_vsnprintf(char* s, size_t count, const char* format, va_list arg); @@ -30,6 +40,13 @@ using namespace std::literals::chrono_literals; namespace org::apache::nifi::minifi::test { +struct HistoryModificationRecord { + int32_t value = 0; + std::string username; + UA_HistoryUpdateType update_type = UA_HISTORYUPDATETYPE_INSERT; + UA_DateTime modification_time = 0; +}; + class OpcUaTestServer { public: explicit OpcUaTestServer(UA_UInt16 port = 4840) : server_(UA_Server_new()) { @@ -37,19 +54,32 @@ class OpcUaTestServer { auto config = UA_Server_getConfig(server_); UA_ServerConfig_setMinimal(config, port, nullptr); - config->logging->log = [] (void *log_context, UA_LogLevel level, UA_LogCategory /*category*/, const char *msg, va_list args) { + config->logging->log = [](void* log_context, UA_LogLevel level, UA_LogCategory /*category*/, const char* msg, va_list args) { char buffer[1024]; mp_vsnprintf(buffer, sizeof(buffer), msg, args); std::string level_str; switch (level) { - case UA_LOGLEVEL_TRACE: return; - case UA_LOGLEVEL_DEBUG: level_str = "DEBUG"; break; - case UA_LOGLEVEL_INFO: level_str = "INFO"; break; - case UA_LOGLEVEL_WARNING: level_str = "WARNING"; break; - case UA_LOGLEVEL_ERROR: level_str = "ERROR"; break; - case UA_LOGLEVEL_FATAL: level_str = "FATAL"; break; - default: level_str = "UNKNOWN"; break; + case UA_LOGLEVEL_TRACE: + return; + case UA_LOGLEVEL_DEBUG: + level_str = "DEBUG"; + break; + case UA_LOGLEVEL_INFO: + level_str = "INFO"; + break; + case UA_LOGLEVEL_WARNING: + level_str = "WARNING"; + break; + case UA_LOGLEVEL_ERROR: + level_str = "ERROR"; + break; + case UA_LOGLEVEL_FATAL: + level_str = "FATAL"; + break; + default: + level_str = "UNKNOWN"; + break; } std::string log_message = "[" + level_str + "] " + buffer + "\n"; @@ -59,30 +89,62 @@ class OpcUaTestServer { config->logging->context = this; + UA_HistoryDatabase history_database; + memset(&history_database, 0, sizeof(history_database)); + history_database.context = this; + history_database.readRaw = &OpcUaTestServer::readRawCallback; + history_database.readModified = &OpcUaTestServer::readModifiedCallback; + config->historyDatabase = history_database; + ns_index_ = UA_Server_addNamespace(server_, "custom.namespace"); UA_NodeId simulator_node = addObject("Simulator", UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER)); UA_NodeId default_node = addObject("Default", simulator_node); UA_NodeId device1_node = addObject("Device1", default_node); + UA_NodeId device2_node = addObject("Device2", default_node); - auto int1_node = addIntVariable("INT1", device1_node, 1); - node_ids_["Simulator/Default/Device1/INT1"] = int1_node; - auto int2_node = addIntVariable("INT2", device1_node, 2); - node_ids_["Simulator/Default/Device1/INT2"] = int2_node; - auto int3_node = addIntVariable("INT3", device1_node, 3); - node_ids_["Simulator/Default/Device1/INT3"] = int3_node; - auto int4_node = addIntVariable("INT4", int3_node, 4); - node_ids_["Simulator/Default/Device1/INT4"] = int4_node; - + addIntVariable("INT1", opc::OPCNodeIDType::String, device1_node, 1); + addIntVariable("INT2", opc::OPCNodeIDType::String, device1_node, 2); + UA_NodeId int3_node = addIntVariable("INT3", opc::OPCNodeIDType::String, device1_node, 3); + addIntVariable("INT4", opc::OPCNodeIDType::String, int3_node, 4); addStringVariable("StringNode", "the.answer.node", UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER), 42); + addIntVariable("666", opc::OPCNodeIDType::Int, device2_node, 256); + addIntVariable("72962b91-fa75-4ae6-8d28-b404dc7daf63", opc::OPCNodeIDType::Guid, device2_node, 7); + + setHistory("INT1", + {HistoryModificationRecord{.value = 1, + .username = "test_user", + .update_type = UA_HISTORYUPDATETYPE_REPLACE, + .modification_time = makeDateTime(2024, 6, 15, 10, 30, 0, 0)}}); + setHistory("INT2", + {HistoryModificationRecord{.value = 2, + .username = "admin_user", + .update_type = UA_HISTORYUPDATETYPE_INSERT, + .modification_time = makeDateTime(2021, 3, 15, 11, 30, 0, 0)}, + HistoryModificationRecord{.value = 3, + .username = "admin_user", + .update_type = UA_HISTORYUPDATETYPE_UPDATE, + .modification_time = makeDateTime(2025, 11, 11, 11, 30, 0, 0)}, + HistoryModificationRecord{.value = 4, + .username = "test_user", + .update_type = UA_HISTORYUPDATETYPE_REPLACE, + .modification_time = makeDateTime(2026, 3, 11, 11, 30, 0, 0)}}); + setHistory("666", + {HistoryModificationRecord{.value = 256, + .username = "integer_user", + .update_type = UA_HISTORYUPDATETYPE_INSERT, + .modification_time = makeDateTime(2001, 1, 1, 22, 22, 0, 0)}}); + setHistory("72962b91-fa75-4ae6-8d28-b404dc7daf63", + {HistoryModificationRecord{.value = 7, + .username = "guid_user", + .update_type = UA_HISTORYUPDATETYPE_INSERT, + .modification_time = makeDateTime(2020, 5, 20, 12, 0, 0, 0)}}); } void start() { std::lock_guard lock(mutex_); running_ = true; - server_thread_ = std::thread([this]() { - UA_Server_run(server_, &running_); - }); + server_thread_ = std::thread([this]() { UA_Server_run(server_, &running_); }); ensureConnection(); } @@ -106,6 +168,30 @@ class OpcUaTestServer { return ns_index_; } + void setHistory(const std::string& node_id, std::vector records) { + std::lock_guard lock(history_mutex_); + history_records_[node_id] = std::move(records); + } + + // When set to a non-zero value the server returns at most this many entries per history read response and hands out a + // continuation point for the rest, forcing the client to drive the read across multiple callback invocations. + void setHistoryPageSize(size_t page_size) { + std::lock_guard lock(history_mutex_); + history_page_size_ = page_size; + } + + static UA_DateTime makeDateTime(uint16_t year, uint16_t month, uint16_t day, uint16_t hour, uint16_t min, uint16_t sec, uint16_t milli) { + UA_DateTimeStruct dts{}; + dts.year = year; + dts.month = month; + dts.day = day; + dts.hour = hour; + dts.min = min; + dts.sec = sec; + dts.milliSec = milli; + return UA_DateTime_fromStruct(dts); + } + void addLog(const std::string& log) { std::lock_guard lock(server_logs_mutex_); server_logs_.push_back(log); @@ -136,17 +222,163 @@ 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(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(id.identifier.guid.data4[2]) << 40) | (gsl::narrow(id.identifier.guid.data4[3]) << 32) | + (gsl::narrow(id.identifier.guid.data4[4]) << 24) | (gsl::narrow(id.identifier.guid.data4[5]) << 16) | + (gsl::narrow(id.identifier.guid.data4[6]) << 8) | gsl::narrow(id.identifier.guid.data4[7])); + return std::string(guid_str); + } + return {}; + } + + static std::vector selectRecords(const std::vector& records, UA_DateTime start_time, + UA_DateTime end_time, size_t offset, size_t page_size, bool& has_more_data) { + has_more_data = false; + std::vector matching; + for (const auto& record : records) { + if (record.modification_time >= start_time && record.modification_time < end_time) { + matching.push_back(&record); + } + } + if (offset >= matching.size()) { + return {}; + } + std::vector selected(matching.begin() + offset, matching.end()); + if (page_size > 0 && selected.size() > page_size) { + selected.resize(page_size); + has_more_data = true; + } + return selected; + } + + static size_t readContinuationOffset(const UA_ByteString& continuation_point) { + size_t offset = 0; + if (continuation_point.length == sizeof(size_t)) { + memcpy(&offset, continuation_point.data, sizeof(size_t)); + } + return offset; + } + + static void writeContinuationPoint(UA_HistoryReadResult& result, size_t next_offset) { + UA_ByteString_allocBuffer(&result.continuationPoint, sizeof(size_t)); + memcpy(result.continuationPoint.data, &next_offset, sizeof(size_t)); + } + + static UA_DataValue* makeDataValues(const std::vector& records) { + auto* values = static_cast(UA_Array_new(records.size(), &UA_TYPES[UA_TYPES_DATAVALUE])); + for (size_t i = 0; i < records.size(); ++i) { + UA_Int32 value = records[i]->value; + UA_Variant_setScalarCopy(&values[i].value, &value, &UA_TYPES[UA_TYPES_INT32]); + values[i].hasValue = true; + values[i].hasSourceTimestamp = true; + values[i].sourceTimestamp = records[i]->modification_time; + } + return values; + } + + static UA_ModificationInfo* makeModificationInfos(const std::vector& records) { + auto* mods = static_cast(UA_Array_new(records.size(), &UA_TYPES[UA_TYPES_MODIFICATIONINFO])); + for (size_t i = 0; i < records.size(); ++i) { + mods[i].updateType = records[i]->update_type; + mods[i].modificationTime = records[i]->modification_time; + mods[i].userName = UA_STRING_ALLOC(records[i]->username.c_str()); + } + return mods; + } + + std::optional> readNodeHistory(const UA_HistoryReadValueId& node_to_read, + const UA_ReadRawModifiedDetails& details, UA_HistoryReadResult& result, size_t offset, bool& has_more_data) { + auto node_id_str = nodeIdToString(node_to_read.nodeId); + auto it = history_records_.find(node_id_str); + auto node_id_it = node_ids_.find(node_id_str); + if (it == history_records_.end() || node_id_it == node_ids_.end() || node_to_read.nodeId.identifierType != node_id_it->second.get().identifierType) { + result.statusCode = UA_STATUSCODE_BADNODEIDUNKNOWN; + return std::nullopt; + } + return selectRecords(it->second, details.startTime, details.endTime, offset, history_page_size_, has_more_data); + } + + static void readRawCallback(UA_Server* /*server*/, void* hdb_context, const UA_NodeId* /*sessionId*/, void* /*session_context*/, + const UA_RequestHeader* /*request_header*/, const UA_ReadRawModifiedDetails* history_read_details, + UA_TimestampsToReturn /*timestamps_to_return*/, UA_Boolean /*release_continuation_points*/, size_t nodes_to_read_size, + const UA_HistoryReadValueId* nodes_to_read, UA_HistoryReadResponse* response, UA_HistoryData* const* const history_data) { + auto* self = static_cast(hdb_context); + std::lock_guard lock(self->history_mutex_); + + for (size_t i = 0; i < nodes_to_read_size; ++i) { + const size_t offset = readContinuationOffset(nodes_to_read[i].continuationPoint); + bool has_more_data = false; + auto records = self->readNodeHistory(nodes_to_read[i], *history_read_details, response->results[i], offset, has_more_data); + if (!records) { + continue; + } + + history_data[i]->dataValues = makeDataValues(*records); + history_data[i]->dataValuesSize = records->size(); + if (has_more_data) { + writeContinuationPoint(response->results[i], offset + records->size()); + } + response->results[i].statusCode = UA_STATUSCODE_GOOD; + } + + response->responseHeader.serviceResult = UA_STATUSCODE_GOOD; + } + + static void readModifiedCallback(UA_Server* /*server*/, void* hdb_context, const UA_NodeId* /*sessionId*/, void* /*session_context*/, + const UA_RequestHeader* /*request_header*/, const UA_ReadRawModifiedDetails* history_read_details, + UA_TimestampsToReturn /*timestamps_to_return*/, UA_Boolean /*release_continuation_points*/, size_t nodes_to_read_size, + const UA_HistoryReadValueId* nodes_to_read, UA_HistoryReadResponse* response, UA_HistoryModifiedData* const* const history_data) { + auto* self = static_cast(hdb_context); + std::lock_guard lock(self->history_mutex_); + + for (size_t i = 0; i < nodes_to_read_size; ++i) { + const size_t offset = readContinuationOffset(nodes_to_read[i].continuationPoint); + bool has_more_data = false; + auto records = self->readNodeHistory(nodes_to_read[i], *history_read_details, response->results[i], offset, has_more_data); + if (!records) { + continue; + } + + history_data[i]->dataValues = makeDataValues(*records); + history_data[i]->dataValuesSize = records->size(); + history_data[i]->modificationInfos = makeModificationInfos(*records); + history_data[i]->modificationInfosSize = records->size(); + if (has_more_data) { + writeContinuationPoint(response->results[i], offset + records->size()); + } + response->results[i].statusCode = UA_STATUSCODE_GOOD; + } + + response->responseHeader.serviceResult = UA_STATUSCODE_GOOD; + } + + UA_NodeId addObject(const char* name, UA_NodeId parent) { UA_NodeId object_id; UA_ObjectAttributes attr = UA_ObjectAttributes_default; attr.displayName = UA_LOCALIZEDTEXT_ALLOC("en-US", name); - auto status = UA_Server_addObjectNode( - server_, UA_NODEID_NULL, parent, - UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES), - UA_QUALIFIEDNAME(ns_index_, const_cast(name)), - UA_NODEID_NUMERIC(0, UA_NS0ID_BASEOBJECTTYPE), - attr, nullptr, &object_id); + auto status = UA_Server_addObjectNode(server_, + UA_NODEID_NULL, + parent, + UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES), + UA_QUALIFIEDNAME(ns_index_, const_cast(name)), + UA_NODEID_NUMERIC(0, UA_NS0ID_BASEOBJECTTYPE), + attr, + nullptr, + &object_id); if (status != UA_STATUSCODE_GOOD) { UA_LocalizedText_clear(&attr.displayName); @@ -157,20 +389,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(browse_name.c_str())); + opc::NodeId node_id; + + switch (type) { + case opc::OPCNodeIDType::Int: + node_id = opc::NodeId{UA_NODEID_NUMERIC(ns_index_, std::stoi(browse_name))}; + break; + case opc::OPCNodeIDType::String: + node_id = opc::NodeId{UA_NODEID_STRING_ALLOC(ns_index_, browse_name.c_str())}; + break; + case opc::OPCNodeIDType::Guid: + node_id = opc::NodeId{UA_NODEID_GUID(ns_index_, UA_GUID(browse_name.c_str()))}; + break; + default: + return UA_STATUSCODE_BADNOTSUPPORTED; + } + + return UA_Server_addVariableNode(server_, + node_id, + parent_node_id, + UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT), + qname, + UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE), + attr, + nullptr, + &target_node_id); + } + + UA_NodeId addIntVariable(const std::string& name, opc::OPCNodeIDType type, UA_NodeId parent, UA_Int32 value) { UA_VariableAttributes attr = UA_VariableAttributes_default; - attr.displayName = UA_LOCALIZEDTEXT_ALLOC("en-US", name); - attr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE; + attr.displayName = UA_LOCALIZEDTEXT_ALLOC("en-US", name.c_str()); + attr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE | UA_ACCESSLEVELMASK_HISTORYREAD; UA_Variant_setScalar(&attr.value, &value, &UA_TYPES[UA_TYPES_INT32]); UA_NodeId node_id; - auto status = UA_Server_addVariableNode( - server_, UA_NODEID_NULL, parent, - UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT), - UA_QUALIFIEDNAME(ns_index_, const_cast(name)), - UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE), - attr, nullptr, &node_id); + auto status = addNode(parent, node_id, type, name, attr); if (status != UA_STATUSCODE_GOOD) { UA_LocalizedText_clear(&attr.displayName); @@ -178,7 +435,8 @@ class OpcUaTestServer { } UA_LocalizedText_clear(&attr.displayName); - return node_id; + node_ids_[name] = opc::NodeId{node_id}; + return node_ids_[name].get(); } void addStringVariable(const char* name, const char* node_id_str, UA_NodeId parent, UA_Int32 value) { @@ -211,12 +469,14 @@ class OpcUaTestServer { void ensureConnection() { REQUIRE(utils::verifyEventHappenedInPollTime( - 5s, - [&]() { - auto logs = getLogs(); - return std::find_if(logs.begin(), logs.end(), [](const std::string& message) { return message.find("New DiscoveryUrl added") != std::string::npos;}) != logs.end(); - }, - 100ms)); + 5s, + [&]() { + auto logs = getLogs(); + return std::find_if(logs.begin(), logs.end(), [](const std::string& message) { + return message.find("New DiscoveryUrl added") != std::string::npos; + }) != logs.end(); + }, + 100ms)); } UA_Server* server_; @@ -226,7 +486,10 @@ class OpcUaTestServer { std::thread server_thread_; mutable std::mutex server_logs_mutex_; std::vector server_logs_; - std::unordered_map node_ids_; + std::unordered_map node_ids_; + std::mutex history_mutex_; + std::unordered_map> history_records_; + size_t history_page_size_ = 0; }; } // namespace org::apache::nifi::minifi::test diff --git a/extensions/opc/tests/features/opcua.feature b/extensions/opc/tests/features/opcua.feature index 0b6c8966d8..9be4fe9d09 100644 --- a/extensions/opc/tests/features/opcua.feature +++ b/extensions/opc/tests/features/opcua.feature @@ -244,3 +244,97 @@ Feature: Putting and fetching data to OPC UA server Then in the "fetch-opc-ua-node" container at least one file with the content "Test" is placed in the "/tmp/output" directory in less than 60 seconds And the logs of the "fetch-opc-ua-node" container do not contain the following message: "Username/password authentication is used without encryption, which is not secure. Please consider configuring encryption for better security." after 0 seconds And the logs of the "create-opc-ua-node" container do not contain the following message: "Username/password authentication is used without encryption, which is not secure. Please consider configuring encryption for better security." after 0 seconds + + Scenario: Update and fetch historical data from an OPC UA node + Given a GenerateFlowFile processor with the "Unique FlowFiles" property set to "false" in the "update-opc-ua-node" flow + And the "Custom Text" property of the GenerateFlowFile processor is set to "${nextInt()}" in the "update-opc-ua-node" flow + And the "Data Format" property of the GenerateFlowFile processor is set to "Text" in the "update-opc-ua-node" flow + And the scheduling period of the GenerateFlowFile processor is set to "1 sec" in the "update-opc-ua-node" flow + And a PutOPCProcessor processor in the "update-opc-ua-node" flow + And PutOPCProcessor is EVENT_DRIVEN in the "update-opc-ua-node" flow + And PutOPCProcessor's success relationship is auto-terminated in the "update-opc-ua-node" flow + And a FetchOPCHistory processor in the "fetch-opc-ua-node-history" flow + And the scheduling period of the FetchOPCHistory processor is set to "10 sec" in the "fetch-opc-ua-node-history" flow + And a PutFile processor with the "Directory" property set to "/tmp/output" in the "fetch-opc-ua-node-history" flow + And PutFile's success relationship is auto-terminated in the "fetch-opc-ua-node-history" flow + And PutFile is EVENT_DRIVEN in the "fetch-opc-ua-node-history" flow + And a LogAttribute processor in the "fetch-opc-ua-node-history" flow + And LogAttribute is EVENT_DRIVEN in the "fetch-opc-ua-node-history" flow + And the "Log Payload" property of the LogAttribute processor is set to "true" in the "fetch-opc-ua-node-history" flow + And LogAttribute's success relationship is auto-terminated in the "fetch-opc-ua-node-history" flow + And these processor properties are set in the "update-opc-ua-node" flow + | processor name | property name | property value | + | PutOPCProcessor | Parent node ID | 85 | + | PutOPCProcessor | Parent node ID type | Int | + | PutOPCProcessor | Target node ID | myUintValue | + | PutOPCProcessor | Target node ID type | String | + | PutOPCProcessor | Target node namespace index | 1 | + | PutOPCProcessor | Value type | UInt32 | + | PutOPCProcessor | OPC server endpoint | opc.tcp://opcua-server-${scenario_id}:4840/ | + | PutOPCProcessor | Target node browse name | testnodename | + And these processor properties are set in the "fetch-opc-ua-node-history" flow + | processor name | property name | property value | + | FetchOPCHistory | Node ID | myUintValue | + | FetchOPCHistory | Node ID type | String | + | FetchOPCHistory | Namespace index | 1 | + | FetchOPCHistory | OPC server endpoint | opc.tcp://opcua-server-${scenario_id}:4840/ | + | FetchOPCHistory | Batch Size | 3 | + | FetchOPCHistory | History Read Type | Raw | + + And in the "update-opc-ua-node" flow the "success" relationship of the GenerateFlowFile processor is connected to the PutOPCProcessor + And in the "fetch-opc-ua-node-history" flow the "success" relationship of the FetchOPCHistory processor is connected to the PutFile + And in the "fetch-opc-ua-node-history" flow the "success" relationship of the PutFile processor is connected to the LogAttribute + + And an OPC UA server is set up with historical data support + + When all instances start up + Then in the "fetch-opc-ua-node-history" container files with at least these contents "0,1,2,3,4" are placed in the "/tmp/output" directory in less than 30 seconds + And the logs of the "fetch-opc-ua-node-history" container contain the following message: "key:Namespace index value:1" in less than 10 seconds + And the logs of the "fetch-opc-ua-node-history" container contain the following message: "key:NodeID value:myUintValue" in less than 10 seconds + And the logs of the "fetch-opc-ua-node-history" container contain the following message: "key:Sourcetimestamp value:" in less than 10 seconds + + Scenario: Update and fetch historical data from an OPC UA node in JSON format + Given a JsonRecordSetWriter controller service is set up in the "fetch-opc-ua-node-history" flow + And a GenerateFlowFile processor with the "Unique FlowFiles" property set to "false" in the "update-opc-ua-node" flow + And the "Custom Text" property of the GenerateFlowFile processor is set to "${nextInt()}" in the "update-opc-ua-node" flow + And the "Data Format" property of the GenerateFlowFile processor is set to "Text" in the "update-opc-ua-node" flow + And the scheduling period of the GenerateFlowFile processor is set to "1 sec" in the "update-opc-ua-node" flow + And a PutOPCProcessor processor in the "update-opc-ua-node" flow + And PutOPCProcessor is EVENT_DRIVEN in the "update-opc-ua-node" flow + And PutOPCProcessor's success relationship is auto-terminated in the "update-opc-ua-node" flow + And a FetchOPCHistory processor in the "fetch-opc-ua-node-history" flow + And the scheduling period of the FetchOPCHistory processor is set to "10 sec" in the "fetch-opc-ua-node-history" flow + And a PutFile processor with the "Directory" property set to "/tmp/output" in the "fetch-opc-ua-node-history" flow + And PutFile's success relationship is auto-terminated in the "fetch-opc-ua-node-history" flow + And PutFile is EVENT_DRIVEN in the "fetch-opc-ua-node-history" flow + And a LogAttribute processor in the "fetch-opc-ua-node-history" flow + And LogAttribute is EVENT_DRIVEN in the "fetch-opc-ua-node-history" flow + And the "Log Payload" property of the LogAttribute processor is set to "true" in the "fetch-opc-ua-node-history" flow + And LogAttribute's success relationship is auto-terminated in the "fetch-opc-ua-node-history" flow + And these processor properties are set in the "update-opc-ua-node" flow + | processor name | property name | property value | + | PutOPCProcessor | Parent node ID | 85 | + | PutOPCProcessor | Parent node ID type | Int | + | PutOPCProcessor | Target node ID | myUintValue | + | PutOPCProcessor | Target node ID type | String | + | PutOPCProcessor | Target node namespace index | 1 | + | PutOPCProcessor | Value type | UInt32 | + | PutOPCProcessor | OPC server endpoint | opc.tcp://opcua-server-${scenario_id}:4840/ | + | PutOPCProcessor | Target node browse name | testnodename | + And these processor properties are set in the "fetch-opc-ua-node-history" flow + | processor name | property name | property value | + | FetchOPCHistory | Node ID | myUintValue | + | FetchOPCHistory | Node ID type | String | + | FetchOPCHistory | Namespace index | 1 | + | FetchOPCHistory | OPC server endpoint | opc.tcp://opcua-server-${scenario_id}:4840/ | + | FetchOPCHistory | Batch Size | 3 | + | FetchOPCHistory | Record Set Writer | JsonRecordSetWriter | + + And in the "update-opc-ua-node" flow the "success" relationship of the GenerateFlowFile processor is connected to the PutOPCProcessor + And in the "fetch-opc-ua-node-history" flow the "success" relationship of the FetchOPCHistory processor is connected to the PutFile + And in the "fetch-opc-ua-node-history" flow the "success" relationship of the PutFile processor is connected to the LogAttribute + + And an OPC UA server is set up with historical data support + + When all instances start up + Then in the 'fetch-opc-ua-node-history' container at least one file in '/tmp/output' content match the following regex: '\[\{"Sourcetimestamp":".+","NodeID":"myUintValue","Namespace index":"1","Value":"0"\},\{"Sourcetimestamp":".+","NodeID":"myUintValue","Namespace index":"1","Value":"1"\},\{"Sourcetimestamp":".+","NodeID":"myUintValue","Namespace index":"1","Value":"2"\}\]' in less than 30 seconds diff --git a/extensions/opc/tests/features/steps/steps.py b/extensions/opc/tests/features/steps/steps.py index 7e39976ad6..50540b78ce 100644 --- a/extensions/opc/tests/features/steps/steps.py +++ b/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"] + ) + + @then('the OPC UA server logs contain the following message: "{log_message}" in less than {duration}') def verify_opcua_server_logs_contain_message(context, log_message, duration): timeout_seconds = humanfriendly.parse_timespan(duration)