diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index bb42090..bb8986b 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -238,7 +238,7 @@ "docs_file_link": "https://github.com/influxdata/influxdb3_plugins/blob/main/influxdata/opcua/README.md", "required_plugins": [], "required_libraries": ["asyncua"], - "last_update": "2026-03-30", + "last_update": "2026-07-28", "trigger_types_supported": ["scheduler"] }, { diff --git a/influxdata/opcua/README.md b/influxdata/opcua/README.md index 14bbf3c..31e52d4 100644 --- a/influxdata/opcua/README.md +++ b/influxdata/opcua/README.md @@ -46,10 +46,10 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor **One of the following is required** (mutually exclusive): -| Parameter | Type | TOML Section | Description | -|---------------|---------|------------------|-----------------------------------------------------------------------------------------------------------| -| `nodes` | string | `[opcua.nodes]` | Space-separated node mappings. Format: `field_name:namespace:identifier[:type]`. See [Node mapping](#node-mapping-format). | -| `browse_root` | string | `[opcua.browse]` | Root node ID for auto-discovery mode (e.g., `ns=2;s=Devices`). See [Browse mode](#browse-mode-parameters). | +| Parameter | Type | TOML Section | Description | +|---------------|---------|------------------|-------------------------------------------------------------------------------------------------------------------------------------| +| `nodes` | string | `[opcua.nodes]` | Space-separated node mappings. Format: `field_name:namespace:identifier[:type]`. See [Node mapping](#node-mapping-format). | +| `browse_root` | string | `[opcua.browse]` | Root node ID for auto-discovery mode (e.g., `ns=2;s=Devices` or `nsu=;s=Devices`). See [Browse mode](#browse-mode-parameters). | ### Namespace parameters @@ -78,16 +78,19 @@ The plugin resolves `nsu=` URIs to numeric `ns=` indexes after connecting to the ### Browse mode parameters -| Parameter | Type | Default | TOML Section | Description | -|--------------------|--------|-----------|------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `browse_root` | string | none | `[opcua.browse]` | Root node ID for auto-discovery (e.g., `ns=2;s=Devices`) | -| `browse_depth` | int | 2 | `[opcua.browse]` | Maximum browse depth. 1 = direct children, 2 = children of children, etc. | -| `path_tags` | list | `[]` | `[opcua.browse]` | Tag names mapping Object hierarchy levels to InfluxDB tags. First entry maps to depth-1 Objects, second to depth-2, etc. Length must be strictly less than `browse_depth`. Objects beyond `path_tags` length become field name prefixes. **Required in TOML** (specify `[]` explicitly for no hierarchy tags); defaults to `[]` in CLI. | -| `filter` | string | none | `[opcua.browse]` | Regex pattern to filter discovered variable names. Only matching variables are included. Matches the original Variable browse name, not the prefixed field name (see [filter in nested hierarchies](#filter-in-nested-hierarchies)). | -| `exclude_branches` | string | none | `[opcua.browse]` | Regex pattern to exclude Object nodes (branches) by browse name. Matched Objects and their entire subtree are skipped. Works at all browse depths. See [Exclude branches](#exclude-branches). | -| `browse_tags` | list | none | `[opcua.browse]` | Variable names that should be stored as InfluxDB tags instead of fields. Values are converted to strings. See [Browse tags](#browse-tags). | -| `name_separator` | string | none | `[opcua.browse]` | Separator for splitting Variable browse names into segments for tag extraction. Required when `name_tags` is set. See [Name tags](#name-tags). | -| `name_tags` | list | none | `[opcua.browse]` | Tag names extracted from leading segments of Variable browse names split by `name_separator`. Remaining segments form the field name (joined with `_`). See [Name tags](#name-tags). | +| Parameter | Type | Default | TOML Section | Description | +|--------------------|--------|-----------|------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `browse_root` | string | none | `[opcua.browse]` | Root node ID for auto-discovery. Accepts a numeric namespace (`ns=2;s=Devices`) or a namespace URI (`nsu=;s=Devices`) resolved at connection time for restart stability. See [Namespace URI resolution](#namespace-uri-resolution). | +| `browse_depth` | int | 2 | `[opcua.browse]` | Maximum browse depth. 1 = direct children, 2 = children of children, etc. | +| `browse_cache_ttl` | int | 3600 | `[opcua.browse]` | Seconds to cache the discovered node set before re-browsing. Discovery runs on the first call and every `browse_cache_ttl` seconds thereafter; cached node IDs are read in between. Set low for a fast-changing address space, high to run discovery infrequently while collecting on a short trigger interval. See [Discovery caching](#discovery-caching). | +| `path_tags` | list | `[]` | `[opcua.browse]` | Tag names mapping Object hierarchy levels to InfluxDB tags. First entry maps to depth-1 Objects, second to depth-2, etc. Length must be strictly less than `browse_depth`. Objects beyond `path_tags` length become field name prefixes. **Required in TOML** (specify `[]` explicitly for no hierarchy tags); defaults to `[]` in CLI. | +| `filter` | string | none | `[opcua.browse]` | Regex pattern to filter discovered variable names. Only matching variables are included. Matches the original Variable browse name, not the prefixed field name (see [filter in nested hierarchies](#filter-in-nested-hierarchies)). | +| `exclude_branches` | string | none | `[opcua.browse]` | Regex pattern to exclude Object nodes (branches) by browse name. Matched Objects and their entire subtree are skipped. Works at all browse depths. See [Exclude branches](#exclude-branches). | +| `browse_tags` | list | none | `[opcua.browse]` | Variable names that should be stored as InfluxDB tags instead of fields. Values are converted to strings. See [Browse tags](#browse-tags). | +| `name_separator` | string | none | `[opcua.browse]` | Separator for splitting Variable browse names into segments for tag extraction. Required when `name_tags` is set. See [Name tags](#name-tags). | +| `name_tags` | list | none | `[opcua.browse]` | Tag names extracted from leading segments of Variable browse names split by `name_separator`. Remaining segments form the field name (joined with `_`). See [Name tags](#name-tags). | + +> **Note:** integer parameters (`browse_depth`, `browse_cache_ttl`) also accept numeric strings, so `browse_depth = "2"` is valid in TOML. `browse_cache_ttl` is capped at 30 days. ### Tag parameters @@ -142,7 +145,8 @@ quality_filter = ["good", "uncertain"] | Parameter | Type | Default | TOML Section | Description | |------------------------|--------|---------|--------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `config_file_path` | string | none | CLI only | Path to TOML config file (absolute or relative to `PLUGIN_DIR`) | -| `disable_config_cache` | bool | false | `[opcua]` | Reload configuration on every call instead of caching for 1 hour. Useful during development. | +| `config_cache_ttl` | int | 3600 | `[opcua]` | Seconds to cache the parsed configuration (max 2592000). Controls how quickly credential, endpoint, table, or filter changes take effect. Independent of `browse_cache_ttl`. | +| `disable_config_cache` | bool | false | `[opcua]` | Reload configuration on every call instead of caching for `config_cache_ttl` seconds. Also disables caching of the discovered browse structure, so in browse mode the address space is re-walked on every call. Useful during development. | | `allow_insecure_auth` | bool | false | `[opcua]` | Permit sending `username`/`password` when `security_policy` is not set (credentials sent in cleartext over an unencrypted connection). Only enable on a trusted network. | | `enable_full_logging` | bool | false | `[opcua]` | When `true`, full exception messages are written to logs. When `false` (default), only the exception type is logged, to avoid leaking sensitive values (credentials, payloads, paths). Enable temporarily for debugging. | @@ -316,16 +320,16 @@ If a relative path is specified and `PLUGIN_DIR` is not set, the plugin will ret ### TOML section reference -| TOML section | Description | -|------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------| -| `[opcua]` | Required. `server_url`, `table_name`, `quality_filter`, `disable_config_cache`, `allow_insecure_auth` | -| `[opcua.default_tags]` | Static tags as `key = "value"` pairs | -| `[opcua.namespaces]` | Namespace alias mappings as `alias = "uri"` pairs. Validated but not used for alias substitution in TOML — use `nsu=;...` directly in node IDs | -| `[opcua.tag_nodes]` | Dynamic tags from OPC UA nodes | -| `[opcua.nodes]` | Explicit node mappings (mutually exclusive with `[opcua.browse]`) | -| `[opcua.browse]` | Browse mode settings: `browse_root`, `browse_depth`, `path_tags`, `filter`, `exclude_branches`, `browse_tags`, `name_separator`, `name_tags` | -| `[opcua.security]` | `security_policy`, `security_mode`, `certificate`, `private_key` | -| `[opcua.auth]` | `username`, `password` | +| TOML section | Description | +|------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `[opcua]` | Required. `server_url`, `table_name`, `quality_filter`, `config_cache_ttl`, `disable_config_cache`, `allow_insecure_auth` | +| `[opcua.default_tags]` | Static tags as `key = "value"` pairs | +| `[opcua.namespaces]` | Namespace alias mappings as `alias = "uri"` pairs. Validated but not used for alias substitution in TOML — use `nsu=;...` directly in node IDs | +| `[opcua.tag_nodes]` | Dynamic tags from OPC UA nodes | +| `[opcua.nodes]` | Explicit node mappings (mutually exclusive with `[opcua.browse]`) | +| `[opcua.browse]` | Browse mode settings: `browse_root`, `browse_depth`, `browse_cache_ttl`, `path_tags`, `filter`, `exclude_branches`, `browse_tags`, `name_separator`, `name_tags` | +| `[opcua.security]` | `security_policy`, `security_mode`, `certificate`, `private_key` | +| `[opcua.auth]` | `username`, `password` | ## Node Mapping Format @@ -773,17 +777,27 @@ Unlike `tag_nodes` which always require "good" quality, `browse_tags` variables If all variables in a device group are listed in `browse_tags` (no regular fields remain), no data point is written for that group. InfluxDB requires at least one field per point — browse_tags alone are not sufficient. -### Config and browse structure caching +### Discovery caching + +Two independent caches, each with its own TTL: -Both the parsed configuration and the browse structure are cached for **1 hour** (controlled by `_CONFIG_CACHE_TTL`). This means: -- Changes to the TOML configuration file will not take effect until the cache expires (or use `disable_config_cache = true` during development) -- New devices added to the OPC UA server will not appear until the browse cache expires +- **Parsed configuration** — cached for `config_cache_ttl` seconds (default **1 hour**). Controls how quickly a change to credentials, endpoint, table, or filters is picked up. +- **Discovered browse structure** — cached for `browse_cache_ttl` seconds (default **1 hour**). Discovery runs on the first call and every `browse_cache_ttl` seconds thereafter; cached node IDs are read in between. + +This means: +- Set a low `browse_cache_ttl` for a fast-changing address space, or set it much larger than the trigger interval so discovery runs rarely while collection stays frequent — the recommended pattern for large or high-latency servers where a full browse is expensive +- New devices added to the OPC UA server appear after at most `browse_cache_ttl` seconds +- Changes to the config take effect after at most `config_cache_ttl` seconds. Changes to the browse-relevant part (`server_url`, `browse_root`, `browse_depth`, `filter`, `exclude_branches`, …) also trigger an immediate re-browse on the next config reload; other changes keep the cached structure +- A failed connection drops the cached config, so a credential or endpoint fix is retried on the next call rather than after `config_cache_ttl` +- `disable_config_cache = true` disables both caches, so config and address space are re-read on every call - Any unhandled error clears all caches, forcing a fresh reload on the next call ### Namespace URI resolution When using `nsu=` node IDs (either from `namespaces` aliases in CLI mode or directly in TOML), the URI-to-index resolution happens **once at connection time**. The resolved numeric indexes are stored in the config object. If the server assigns different namespace indexes after a restart, the plugin will re-resolve them on the next reconnect. +This applies to `nodes`, `tag_nodes`, and `browse_root` — use `browse_root = "nsu=;s=..."` in browse mode for the same restart stability. If the URI is not found on the server, connection fails with a configuration error rather than browsing into an empty result. + ### Namespace aliases in TOML vs CLI In **CLI arguments** mode, you can define short aliases (e.g., `siemens`) via the `namespaces` parameter and use them in `nodes` and `tag_nodes` definitions. In **TOML** mode, aliases are not supported — use the full `nsu=;...` format directly in node IDs. The `[opcua.namespaces]` section in TOML is validated but not used for alias substitution. @@ -872,6 +886,7 @@ When using a security policy, both files are required: #### "Browse discovered no variables" (browse mode) - Verify `browse_root` points to a valid node that contains Object or Variable children +- If `browse_root` uses a numeric `ns=` index, confirm the index is still current — it can shift after a server restart. Use `nsu=;s=...` instead so the plugin resolves the index at connection time - Check that `browse_depth` is sufficient to reach Variable nodes - The root node may contain nodes of other types (e.g., Methods) — only Objects and Variables are traversed - If using `filter`, verify the regex pattern matches your variable names @@ -880,7 +895,7 @@ When using a security policy, both files are required: #### Config changes not taking effect -- Configuration is cached for 1 hour — either wait for expiry or set `disable_config_cache = true` +- Configuration is cached for `config_cache_ttl` seconds (default 1 hour) — either wait for expiry, lower `config_cache_ttl`, or set `disable_config_cache = true` - Any plugin error automatically clears the cache #### "Previous opcua call still running, skipping this tick" diff --git a/influxdata/opcua/manifest.toml b/influxdata/opcua/manifest.toml index 86e0d0d..4ee38bd 100644 --- a/influxdata/opcua/manifest.toml +++ b/influxdata/opcua/manifest.toml @@ -2,7 +2,7 @@ manifest_schema_version = "1.2" [plugin] name = "opcua" -version = "0.2.0" +version = "0.3.0" description = "Enables periodic ingestion of OPC UA node values into InfluxDB 3. Connect to OPC UA servers and read current values with auto-discovery browse mode, data type detection, and namespace resolution." triggers = ["process_scheduled_call"] homepage = "https://www.influxdata.com/" diff --git a/influxdata/opcua/opcua.py b/influxdata/opcua/opcua.py index 74971bf..fb75510 100644 --- a/influxdata/opcua/opcua.py +++ b/influxdata/opcua/opcua.py @@ -35,7 +35,7 @@ { "name": "browse_root", "example": "ns=2;s=Devices", - "description": "Root node ID for auto-discovery mode. The plugin browses the OPC UA address space from this node, discovers all Variable nodes grouped by parent Object (device), and reads their values. Required unless 'nodes' is set.", + "description": "Root node ID for auto-discovery mode. The plugin browses the OPC UA address space from this node, discovers all Variable nodes grouped by parent Object (device), and reads their values. Accepts a numeric namespace ('ns=2;s=Devices') or a namespace URI ('nsu=;s=Devices'), which is resolved to the current numeric index at connection time for stability across server restarts. Required unless 'nodes' is set.", "required": false }, { @@ -44,6 +44,12 @@ "description": "Maximum browse depth for auto-discovery (default: 2). Depth 1 = direct children of browse_root, depth 2 = children of children, etc.", "required": false }, + { + "name": "browse_cache_ttl", + "example": "300", + "description": "Time in seconds to cache the discovered node set before re-browsing the address space (default: 3600, max: 2592000). Discovery runs on the first call and every 'browse_cache_ttl' seconds thereafter; cached node IDs are read in between. Set it much larger than the trigger interval to run discovery rarely while collecting frequently. Independent of 'config_cache_ttl'.", + "required": false + }, { "name": "path_tags", "example": "source zone", @@ -134,10 +140,16 @@ "description": "Space-separated list of OPC UA quality categories to accept: 'good', 'uncertain', 'bad'. Values with non-matching quality are skipped and logged. Default: 'good'.", "required": false }, + { + "name": "config_cache_ttl", + "example": "3600", + "description": "Time in seconds to cache the parsed configuration (default: 3600, max: 2592000). Controls how quickly credential, endpoint, table, or filter changes take effect. Independent of 'browse_cache_ttl'.", + "required": false + }, { "name": "disable_config_cache", "example": "true", - "description": "Disable configuration caching. When set to 'true', the configuration is reloaded from file/arguments on every scheduled call instead of being cached for 1 hour. Useful during development or when the config file changes frequently. Default: false.", + "description": "Disable configuration caching. When set to 'true', the configuration is reloaded from file/arguments on every scheduled call instead of being cached for 'config_cache_ttl' seconds. This also disables caching of the discovered browse structure, so in browse mode the address space is re-walked on every call. Useful during development or when the config file changes frequently. Default: false.", "required": false }, { @@ -157,6 +169,9 @@ """ import asyncio +import copy +import hashlib +import json import os import re import threading @@ -166,7 +181,7 @@ from datetime import datetime from pathlib import Path from typing import Any, Protocol, runtime_checkable -from urllib.parse import urlparse +from urllib.parse import unquote, urlparse from asyncua import Client, ua @@ -178,6 +193,8 @@ # Browse mode defaults _DEFAULT_BROWSE_DEPTH = 2 +_DEFAULT_BROWSE_CACHE_TTL = 60 * 60 # seconds — TTL for cached browse structure +_MAX_CACHE_TTL = 30 * 24 * 60 * 60 # seconds — ceiling for cache TTLs (30 days) # OPC UA VariantType sets for type detection _INT_VARIANT_TYPES = { @@ -213,13 +230,13 @@ _VALID_QUALITY_CATEGORIES = {"good", "uncertain", "bad"} _DEFAULT_QUALITY_FILTER = {"good"} -_CONFIG_CACHE_TTL = 60 * 60 # seconds — TTL for config and browse structure cache +_CONFIG_CACHE_TTL = 60 * 60 # seconds — TTL for the parsed config cache # Allowed URL schemes for 'server_url'. Restricted to OPC UA transport schemes _ALLOWED_OPCUA_SCHEMES = ("opc.tcp", "opc.tls") -# Default True so config/validation errors log in full; set from config -# (default False) once known so runtime errors don't leak values. +# Default True so config-load errors log in full; set from config (default +# False) once known so runtime errors don't leak values. _ENABLE_FULL_LOGGING: bool = True @@ -228,6 +245,32 @@ def _exc(e: BaseException) -> str: return str(e) if _ENABLE_FULL_LOGGING else type(e).__name__ +def _split_nsu_node_id(node_id: str) -> tuple[str, str]: + """Split 'nsu=;' into (uri, identifier). + + Raises ValueError if the ';' part is missing or empty. + """ + parts = node_id[4:].split(";", 1) + if len(parts) != 2 or not parts[0].strip() or not parts[1].strip(): + raise ValueError( + f"Malformed node ID '{node_id}': expected " + f"'nsu=;', e.g. 'nsu=urn:example;s=Devices'" + ) + return unquote(parts[0]), parts[1] + + +def _browse_fingerprint(config: dict[str, Any]) -> str: + """Hash of the browse-relevant config (server URL + browse section). + + Lets the cached browse structure survive a config reload when the address + space to discover has not changed. + """ + payload = {"server_url": config.get("server_url"), "browse": config.get("browse")} + return hashlib.sha256( + json.dumps(payload, sort_keys=True, default=str).encode() + ).hexdigest() + + """ Helper for batching multiple line protocol builders into a single write. """ @@ -579,6 +622,12 @@ def _validate_toml_config(self, config: dict[str, Any]): "Parameter 'opcua.disable_config_cache' must be a boolean or string (true/false)" ) + opcua_config["config_cache_ttl"] = OPCUAConfig._validate_positive_int( + opcua_config.get("config_cache_ttl", _CONFIG_CACHE_TTL), + "opcua.config_cache_ttl", + max_value=_MAX_CACHE_TTL, + ) + # Validate allow_insecure_auth if present (accepts bool or string) allow_insecure_auth = opcua_config.get("allow_insecure_auth") if allow_insecure_auth is not None and not isinstance( @@ -598,6 +647,7 @@ def _validate_toml_nodes(nodes: Any, section_name: str = "opcua.nodes"): if isinstance(node_spec, str): if not node_spec.strip(): raise ValueError(f"Empty node_id for field '{field_name}'") + node_id = node_spec elif isinstance(node_spec, list): if len(node_spec) != 2: raise ValueError( @@ -618,6 +668,39 @@ def _validate_toml_nodes(nodes: Any, section_name: str = "opcua.nodes"): f"Expected string or [node_id, type] list." ) + # nsu= node IDs must be 'nsu=;'. + if node_id.startswith("nsu="): + _split_nsu_node_id(node_id) + + @staticmethod + def _validate_positive_int( + value: Any, param_name: str, max_value: int | None = None + ) -> int: + """Validate a positive-integer parameter, coercing numeric strings. + + Rejects booleans, floats, and non-numeric values with a uniform message. + When max_value is set, also rejects values above that ceiling. + """ + if isinstance(value, bool) or not isinstance(value, (int, str)): + raise ValueError( + f"Parameter '{param_name}' must be a positive integer, got: {value}" + ) + try: + parsed = int(value) + except ValueError: + raise ValueError( + f"Parameter '{param_name}' must be a positive integer, got: {value}" + ) + if parsed < 1: + raise ValueError( + f"Parameter '{param_name}' must be a positive integer, got: {value}" + ) + if max_value is not None and parsed > max_value: + raise ValueError( + f"Parameter '{param_name}' must be at most {max_value}, got: {value}" + ) + return parsed + @staticmethod def _validate_toml_browse(browse: Any): """Validate browse section in TOML config""" @@ -632,12 +715,19 @@ def _validate_toml_browse(browse: Any): browse_root = browse["browse_root"] if not isinstance(browse_root, str) or not browse_root.strip(): raise ValueError("Parameter 'opcua.browse.browse_root' must be a non-empty string") + if browse_root.startswith("nsu="): + _split_nsu_node_id(browse_root) - browse_depth = browse.get("browse_depth", _DEFAULT_BROWSE_DEPTH) - if not isinstance(browse_depth, int) or browse_depth < 1: - raise ValueError( - f"Parameter 'opcua.browse.browse_depth' must be a positive integer, got: {browse_depth}" - ) + browse["browse_depth"] = OPCUAConfig._validate_positive_int( + browse.get("browse_depth", _DEFAULT_BROWSE_DEPTH), "opcua.browse.browse_depth" + ) + browse_depth = browse["browse_depth"] + + browse["browse_cache_ttl"] = OPCUAConfig._validate_positive_int( + browse.get("browse_cache_ttl", _DEFAULT_BROWSE_CACHE_TTL), + "opcua.browse.browse_cache_ttl", + max_value=_MAX_CACHE_TTL, + ) # path_tags: required list of tag names for Object hierarchy levels if "path_tags" not in browse: @@ -785,19 +875,21 @@ def _build_config_from_args(self) -> dict[str, Any]: if not browse_root: raise ValueError("Parameter 'browse_root' must be non-empty") + if browse_root.startswith("nsu="): + _split_nsu_node_id(browse_root) browse_config["browse_root"] = browse_root browse_depth = self.args.get("browse_depth") if browse_depth is not None: - try: - depth_val = int(browse_depth) - if depth_val < 1: - raise ValueError - browse_config["browse_depth"] = depth_val - except ValueError: - raise ValueError( - f"Parameter 'browse_depth' must be a positive integer, got: {browse_depth}" - ) + browse_config["browse_depth"] = self._validate_positive_int( + browse_depth, "browse_depth" + ) + + browse_config["browse_cache_ttl"] = self._validate_positive_int( + self.args.get("browse_cache_ttl", _DEFAULT_BROWSE_CACHE_TTL), + "browse_cache_ttl", + max_value=_MAX_CACHE_TTL, + ) # path_tags: required list of tag names for hierarchy levels path_tags_arg = self.args.get("path_tags") @@ -959,6 +1051,12 @@ def _build_config_from_args(self) -> dict[str, Any]: str(self.args.get("disable_config_cache", "false")).lower() == "true" ) + config_cache_ttl: int = self._validate_positive_int( + self.args.get("config_cache_ttl", _CONFIG_CACHE_TTL), + "config_cache_ttl", + max_value=_MAX_CACHE_TTL, + ) + allow_insecure_auth: bool = ( str(self.args.get("allow_insecure_auth", "false")).lower() == "true" ) @@ -974,6 +1072,7 @@ def _build_config_from_args(self) -> dict[str, Any]: "security": security_config, "auth": auth_config, "disable_config_cache": disable_config_cache, + "config_cache_ttl": config_cache_ttl, "allow_insecure_auth": allow_insecure_auth, "enable_full_logging": enable_full_logging, } @@ -1072,7 +1171,8 @@ class OPCUAConnectionManager: """Manages OPC UA client connection and node reading""" def __init__(self, config: dict[str, Any], influxdb3_local, task_id: str): - self.config: dict[str, Any] = config + self._original_config: dict[str, Any] = config + self.config: dict[str, Any] = config # working copy is set by _resolve_namespace_uris on connect self.influxdb3_local = influxdb3_local self.task_id: str = task_id self.client: Client | None = None @@ -1130,6 +1230,13 @@ async def connect(self) -> bool: return True + except ValueError as e: + self.influxdb3_local.error( + f"[{self.task_id}] OPC UA configuration error: {_exc(e)}" + ) + await self.disconnect_silent() + return False + except Exception: if self.config.get("security", {}).get("security_policy"): self.influxdb3_local.error( @@ -1142,18 +1249,28 @@ async def connect(self) -> bool: f"[{self.task_id}] Error connecting to OPC UA server: " f"connection failed. Check server address and availability." ) + await self.disconnect_silent() return False + async def apply_config(self, config: dict[str, Any]) -> None: + """Replace the source config and resolve nsu= URIs against the live session.""" + self._original_config = config + await self._resolve_namespace_uris() + async def _resolve_namespace_uris(self): """Resolve nsu= prefixed node IDs to ns= using server's namespace array. - Scans all node IDs in 'nodes' and 'tag_nodes' config sections. - Replaces 'nsu=;...' with 'ns=;...' using the server's - namespace array lookup. + Scans all node IDs in the 'nodes' and 'tag_nodes' config sections and + the 'browse.browse_root'. Replaces 'nsu=;...' with 'ns=;...' + using the server's namespace array lookup. """ if not self.client: return + # Resolve from the original (nsu=) config so every reconnect re-resolves + # against the current server namespace array instead of reusing a stale index. + self.config = copy.deepcopy(self._original_config) + # Collect unique URIs to resolve uris_to_resolve: set[str] = set() for config_key in ("nodes", "tag_nodes"): @@ -1161,9 +1278,15 @@ async def _resolve_namespace_uris(self): for node_spec in nodes_config.values(): node_id = node_spec[0] if isinstance(node_spec, list) else node_spec if node_id.startswith("nsu="): - uri = node_id.split(";", 1)[0][4:] # strip "nsu=" + uri, _ = _split_nsu_node_id(node_id) uris_to_resolve.add(uri) + browse_config: dict = self.config.get("browse", {}) + browse_root = browse_config.get("browse_root") + if isinstance(browse_root, str) and browse_root.startswith("nsu="): + uri, _ = _split_nsu_node_id(browse_root) + uris_to_resolve.add(uri) + if not uris_to_resolve: return @@ -1192,7 +1315,7 @@ async def _resolve_namespace_uris(self): node_id = node_spec if node_id.startswith("nsu="): - uri, rest = node_id[4:].split(";", 1) + uri, rest = _split_nsu_node_id(node_id) idx = uri_to_index[uri] resolved = f"ns={idx};{rest}" if isinstance(node_spec, list): @@ -1200,6 +1323,10 @@ async def _resolve_namespace_uris(self): else: nodes_config[field_name] = resolved + if isinstance(browse_root, str) and browse_root.startswith("nsu="): + uri, rest = _split_nsu_node_id(browse_root) + browse_config["browse_root"] = f"ns={uri_to_index[uri]};{rest}" + async def read_nodes(self, config_key: str = "nodes") -> list[dict[str, Any]]: """Batch-read all configured nodes in a single OPC UA Read request. @@ -1785,7 +1912,7 @@ async def _read_with_reconnect(opcua_client, operation, label, influxdb3_local, try: return await operation() except Exception as e: - influxdb3_local.warn(f"[{task_id}] {label} failed ({e}), reconnecting...") + influxdb3_local.warn(f"[{task_id}] {label} failed ({_exc(e)}), reconnecting...") influxdb3_local.cache.delete("opcua_browse_structure") if not await opcua_client.reconnect(): influxdb3_local.cache.delete("opcua_connection") @@ -1808,18 +1935,36 @@ async def _async_scheduled_call( config_from_cache: bool = True cached_config: dict | None = influxdb3_local.cache.get("opcua_config") + disable_config_cache: bool = False if cached_config is None: config_from_cache = False config_loader: OPCUAConfig = OPCUAConfig(influxdb3_local, args, task_id) cached_config = config_loader.get_opcua_config() - if not str(cached_config.get("disable_config_cache", False)).lower() == "true": + disable_config_cache = ( + str(cached_config.get("disable_config_cache", False)).lower() == "true" + ) + + if disable_config_cache: + # Dev mode: reload everything on every call. + influxdb3_local.cache.delete("opcua_browse_structure") + influxdb3_local.cache.delete("opcua_browse_fingerprint") + else: influxdb3_local.cache.put( - "opcua_config", cached_config, _CONFIG_CACHE_TTL + "opcua_config", cached_config, cached_config["config_cache_ttl"] ) - - # Config changed — invalidate cached browse structure - influxdb3_local.cache.delete("opcua_browse_structure") + # Rebrowse only when the browse-relevant config changed; otherwise + # keep the cached structure across config reloads. + new_fp = _browse_fingerprint(cached_config) + if influxdb3_local.cache.get("opcua_browse_fingerprint") != new_fp: + influxdb3_local.cache.delete("opcua_browse_structure") + browse_ttl = (cached_config.get("browse") or {}).get( + "browse_cache_ttl", _DEFAULT_BROWSE_CACHE_TTL + ) + fp_ttl = max(cached_config["config_cache_ttl"], browse_ttl) + influxdb3_local.cache.put( + "opcua_browse_fingerprint", new_fp, fp_ttl + ) browse_config_log = cached_config.get("browse") if browse_config_log: @@ -1871,6 +2016,9 @@ async def _async_scheduled_call( opcua_client = OPCUAConnectionManager(cached_config, influxdb3_local, task_id) if not await opcua_client.connect(): influxdb3_local.error(f"[{task_id}] Failed to connect to OPC UA server") + # Drop the cached config so a credential/endpoint change is + # re-read on the next call instead of after config_cache_ttl. + influxdb3_local.cache.delete("opcua_config") opcua_client = None return influxdb3_local.cache.put("opcua_connection", opcua_client) @@ -1878,7 +2026,24 @@ async def _async_scheduled_call( # Reuse existing connection — update task_id and config for this call opcua_client.task_id = task_id if not config_from_cache: - opcua_client.config = cached_config + # Fresh config has unresolved nsu= URIs — re-resolve on the live connection. + try: + await opcua_client.apply_config(cached_config) + except ValueError as e: + influxdb3_local.error( + f"[{task_id}] Namespace/config error resolving nsu= URIs: {_exc(e)}" + ) + influxdb3_local.cache.delete("opcua_config") + return + except Exception as e: + influxdb3_local.warn( + f"[{task_id}] Namespace resolution on cached connection " + f"failed ({_exc(e)}), reconnecting..." + ) + if not await opcua_client.reconnect(): + influxdb3_local.cache.delete("opcua_connection") + influxdb3_local.error(f"[{task_id}] Reconnect failed") + return dynamic_tags: dict[str, str] = {} if cached_config.get("tag_nodes"): @@ -1918,13 +2083,23 @@ async def _async_scheduled_call( return browse_structure = await opcua_client.browse_nodes() if not browse_structure: + resolved_root = opcua_client.config.get("browse", {}).get("browse_root") influxdb3_local.warn( - f"[{task_id}] Browse still empty after reconnect" + f"[{task_id}] Browse from '{resolved_root}' " + f"returned no variables after reconnect. Likely causes: " + f"browse_root does not exist on the server (wrong or stale " + f"ns= index), 'filter'/'exclude_branches' excludes every " + f"variable, browse_depth is too shallow to reach Variable " + f"nodes, or the subtree genuinely has no variables." ) return - influxdb3_local.cache.put( - "opcua_browse_structure", browse_structure, _CONFIG_CACHE_TTL - ) + # Skip when the config cache is disabled + if not disable_config_cache: + influxdb3_local.cache.put( + "opcua_browse_structure", + browse_structure, + browse_config["browse_cache_ttl"], + ) group_results = await _read_with_reconnect( opcua_client, @@ -2046,6 +2221,7 @@ async def _async_scheduled_call( influxdb3_local.cache.delete("opcua_config") influxdb3_local.cache.delete("opcua_connection") influxdb3_local.cache.delete("opcua_browse_structure") + influxdb3_local.cache.delete("opcua_browse_fingerprint") def process_scheduled_call( diff --git a/influxdata/opcua/opcua_config_example.toml b/influxdata/opcua/opcua_config_example.toml index cdcc687..42e7f68 100644 --- a/influxdata/opcua/opcua_config_example.toml +++ b/influxdata/opcua/opcua_config_example.toml @@ -38,6 +38,11 @@ table_name = "cnc_machine" # exception type is logged, to avoid leaking sensitive values into log output. # enable_full_logging = false +# Seconds to cache the parsed configuration (default: 3600, max: 2592000). +# Controls how quickly credential/endpoint/table/filter changes take effect. +# Independent of browse_cache_ttl. +# config_cache_ttl = 3600 + # Static tags added to every data point [opcua.default_tags] location = "factory_1" @@ -240,6 +245,9 @@ is_running = "ns=2;s=MachineRunning" # # 1 = only direct children of browse_root # # 2 = children of children (device -> variables) # # 3 = deeper nesting (device -> subgroup -> variables) +# browse_cache_ttl = 3600 # Seconds to cache the discovered node set (default: 3600) +# # Lower for a fast-changing address space; higher to +# # discover infrequently while collecting on a short interval # path_tags = ["device"] # Depth-1 Objects -> "device" tag # # filter = "Temperature|Pressure" # Optional: regex to include only matching variable names # # exclude_branches = "debug_.*|test_bench" # Optional: regex to exclude Object branches by name diff --git a/influxdata/opcua/test_opcua.py b/influxdata/opcua/test_opcua.py new file mode 100644 index 0000000..8f5fefb --- /dev/null +++ b/influxdata/opcua/test_opcua.py @@ -0,0 +1,199 @@ +"""Unit tests for the opcua plugin: nsu= parsing, integer validation, and +browse-config normalization. Mock-based, no running OPC UA server. +""" + +import asyncio +import os +import sys + +import pytest + +sys.path.insert(0, os.path.dirname(__file__)) +import opcua +from opcua import OPCUAConfig, _browse_fingerprint, _split_nsu_node_id + + +# --------------------------------------------------------------------------- +# Fakes +# --------------------------------------------------------------------------- + + +class FakeCache: + def __init__(self): + self._d = {} + self._ttls = {} + + def get(self, key, default=None, use_global=None): + return self._d.get(key, default) + + def put(self, key, value, ttl=None, use_global=None): + self._d[key] = value + self._ttls[key] = ttl + + def delete(self, key, use_global=None): + return self._d.pop(key, None) is not None + + +class FakeInfluxdb3Local: + def __init__(self): + self.cache = FakeCache() + self.logs = [] + + def info(self, m): + self.logs.append(("info", m)) + + def warn(self, m): + self.logs.append(("warn", m)) + + def error(self, m): + self.logs.append(("error", m)) + + +# --------------------------------------------------------------------------- +# _split_nsu_node_id +# --------------------------------------------------------------------------- + + +def test_split_nsu_node_id_valid(): + assert _split_nsu_node_id("nsu=urn:example;s=Devices") == ("urn:example", "s=Devices") + + +def test_split_nsu_node_id_percent_decodes_uri(): + # %3B is an escaped ';' inside the URI, decoded before namespace lookup. + assert _split_nsu_node_id("nsu=urn:a%3Bb;s=X") == ("urn:a;b", "s=X") + + +@pytest.mark.parametrize( + "bad", + [ + "nsu=urn:example", # missing ';identifier' + "nsu=urn:example;", # empty identifier + "nsu=urn:example; ", # blank identifier + "nsu=;s=Devices", # empty URI + ], +) +def test_split_nsu_node_id_rejects_malformed(bad): + with pytest.raises(ValueError): + _split_nsu_node_id(bad) + + +# --------------------------------------------------------------------------- +# _validate_positive_int +# --------------------------------------------------------------------------- + + +def test_validate_positive_int_accepts_int_and_numeric_string(): + assert OPCUAConfig._validate_positive_int(5, "p") == 5 + assert OPCUAConfig._validate_positive_int("300", "p") == 300 + + +@pytest.mark.parametrize("bad", [0, -1, "0", "abc", 1.5, True, None]) +def test_validate_positive_int_rejects_invalid(bad): + with pytest.raises(ValueError): + OPCUAConfig._validate_positive_int(bad, "p") + + +def test_validate_positive_int_enforces_max(): + assert OPCUAConfig._validate_positive_int(10, "p", max_value=10) == 10 + with pytest.raises(ValueError): + OPCUAConfig._validate_positive_int(11, "p", max_value=10) + + +# --------------------------------------------------------------------------- +# _validate_toml_browse — normalization + nsu= root +# --------------------------------------------------------------------------- + + +def _browse(**overrides): + browse = {"browse_root": "ns=2;s=Devices", "path_tags": []} + browse.update(overrides) + return browse + + +def test_validate_toml_browse_normalizes_ttl_default(): + browse = _browse() + OPCUAConfig._validate_toml_browse(browse) + assert browse["browse_cache_ttl"] == opcua._DEFAULT_BROWSE_CACHE_TTL + + +def test_validate_toml_browse_coerces_string_ttl(): + browse = _browse(browse_cache_ttl="120") + OPCUAConfig._validate_toml_browse(browse) + assert browse["browse_cache_ttl"] == 120 + + +def test_validate_toml_browse_rejects_ttl_over_ceiling(): + with pytest.raises(ValueError): + OPCUAConfig._validate_toml_browse(_browse(browse_cache_ttl=10**30)) + + +def test_validate_toml_browse_rejects_malformed_nsu_root(): + with pytest.raises(ValueError): + OPCUAConfig._validate_toml_browse(_browse(browse_root="nsu=urn:x")) + + +# --------------------------------------------------------------------------- +# _browse_fingerprint +# --------------------------------------------------------------------------- + + +def test_browse_fingerprint_stable_and_sensitive(): + base = { + "server_url": "opc.tcp://h:4840", + "browse": {"browse_root": "ns=2;s=A", "browse_depth": 2}, + "auth": {"password": "p1"}, + } + # Credential/table changes must not invalidate the browse structure. + creds_changed = {**base, "auth": {"password": "p2"}, "table_name": "other"} + browse_changed = {**base, "browse": {"browse_root": "ns=2;s=B", "browse_depth": 2}} + server_changed = {**base, "server_url": "opc.tcp://h2:4840"} + + assert _browse_fingerprint(base) == _browse_fingerprint(creds_changed) + assert _browse_fingerprint(base) != _browse_fingerprint(browse_changed) + assert _browse_fingerprint(base) != _browse_fingerprint(server_changed) + + +# --------------------------------------------------------------------------- +# Cache TTLs — config put runs before the (failed) connection +# --------------------------------------------------------------------------- + + +def test_config_and_browse_ttls_are_independent(): + fake = FakeInfluxdb3Local() + args = { + "server_url": "opc.tcp://127.0.0.1:1", # refused → connect fails fast + "table_name": "t", + "browse_root": "ns=2;s=Devices", + "path_tags": "", + "config_cache_ttl": "1800", + "browse_cache_ttl": "7200", + } + asyncio.run(opcua._async_scheduled_call(fake, "task", args)) + + # Config TTL follows config_cache_ttl; browse fingerprint follows browse_cache_ttl. + assert fake.cache._ttls.get("opcua_config") == 1800 + assert fake.cache._ttls.get("opcua_browse_fingerprint") == 7200 + + +def test_browse_structure_survives_reload_unless_browse_config_changes(): + fake = FakeInfluxdb3Local() + args = { + "server_url": "opc.tcp://127.0.0.1:1", # refused → connect fails fast + "table_name": "t", + "browse_root": "ns=2;s=Devices", + "path_tags": "", + } + # First call records the browse fingerprint (before the failed connection). + asyncio.run(opcua._async_scheduled_call(fake, "task", args)) + assert fake.cache._d.get("opcua_browse_fingerprint") is not None + + # Config-cache miss (expired) with the same browse config → structure kept. + fake.cache.put("opcua_browse_structure", [("x", [])]) + fake.cache.delete("opcua_config") + asyncio.run(opcua._async_scheduled_call(fake, "task", args)) + assert fake.cache._d.get("opcua_browse_structure") is not None + + # Browse root changed → structure invalidated on reload. + fake.cache.delete("opcua_config") + asyncio.run(opcua._async_scheduled_call(fake, "task", {**args, "browse_root": "ns=2;s=Other"})) + assert fake.cache._d.get("opcua_browse_structure") is None \ No newline at end of file