Connectors/opcua/high load

build_latest_docker.sh

@@ -0,0 +1,33 @@
+#!/bin/sh
+#     Copyright 2024. ThingsBoard
+#
+#     Licensed 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.
+
+set -e # exit on any error
+
+# Fetch the current branch name and latest commit ID
+BRANCH_NAME=$(git rev-parse --abbrev-ref HEAD | sed 's/[\/]/-/g')
+COMMIT_ID=$(git rev-parse --short HEAD)
+
+# Combine them to create a version tag
+VERSION_TAG="${BRANCH_NAME}-${COMMIT_ID}"
+
+echo "$(date) Building project with version tag $VERSION_TAG ..."
+set -x
+
+# multi arch
+ DOCKER_CLI_EXPERIMENTAL=enabled \
+ docker buildx build . -t tb-gateway:$VERSION_TAG -f docker/Dockerfile --platform=linux/amd64,linux/arm64 -o type=registry
+
+set +x
+echo "$(date) Done."

docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM --platform=$TARGETPLATFORM python:3.11-slim AS build
+FROM --platform=$TARGETPLATFORM python:3.12-slim AS build
 
 ARG TARGETPLATFORM
 ARG BUILDPLATFORM
@@ -29,8 +29,11 @@ RUN mkdir -p /default-config/config /default-config/extensions/ &&  \
             echo "Unsupported platform detected. Trying to use default value...";; \
         esac && \
     curl https://sh.rustup.rs -sSf | sh -s -- -y --default-host=$DEFAULT_HOST --profile minimal && \
+    apt-get remove --purge -y \
+            gcc python3-dev build-essential libssl-dev libffi-dev zlib1g-dev pkg-config && \
+    apt-get autoremove -y && \
     apt-get clean && \
-    rm -rf /var/lib/apt/lists/* &&  \
+    rm -rf /var/lib/apt/lists/* /tmp/* && \
     echo '#!/bin/sh\n\
 # Main start script\n\
 CONF_FOLDER="/thingsboard_gateway/config"\n\

thingsboard_gateway/config/tb_gateway.json

@@ -12,8 +12,8 @@
       "enable": false,
       "filterFile": "list.json"
     },
-    "maxPayloadSizeBytes": 1024,
-    "minPackSendDelayMS": 200,
+    "maxPayloadSizeBytes": 8196,
+    "minPackSendDelayMS": 50,
     "minPackSizeToSend": 500,
     "checkConnectorsConfigurationInSeconds": 60,
     "handleDeviceRenaming": true,

thingsboard_gateway/connectors/opcua/backward_compatibility_adapter.py

@@ -1,4 +1,4 @@
-from copy import copy
+from copy import deepcopy
 import re
 
 
@@ -11,13 +11,16 @@ class BackwardCompatibilityAdapter:
     }
 
     def __init__(self, config, logger):
-        self._config = copy(config)
+        self._config = deepcopy(config)
         self._log = logger
 
     def convert(self):
-        for node_config in self._config.get('server', {}).get('mapping', []):
+        mapping_configuration = deepcopy(self._config.get('server', {}).get('mapping', []))
+        if not mapping_configuration:
+            return self._config
+        for node_config in mapping_configuration:
             try:
-                node_config['deviceNodeSource'] = self.get_value_source(node_config['deviceNodePattern'])
+                node_config['deviceNodeSource'] = self.get_value_source(node_config['deviceNodePattern'], False)
 
                 device_type_pattern = node_config.pop('deviceTypePattern', 'default')
                 device_name_pattern = node_config.pop('deviceNamePattern', None)
@@ -60,16 +63,18 @@ def convert(self):
                 self._log.error('Error during conversion: ', e)
                 self._log.info('Config: ', node_config)
 
-        mapping = self._config.get('server', {}).pop('mapping', [])
-        self._config['mapping'] = mapping
+        # Removing old mapping section
+        self._config['server'].pop('mapping')
+        # Adding new mapping section
+        self._config['mapping'] = mapping_configuration
 
         return self._config
 
     @staticmethod
-    def get_value_source(value):
+    def get_value_source(value, possible_constant=True):
         if re.search(r"(ns=\d+;[isgb]=[^}]+)", value):
             return 'identifier'
-        elif re.search(r"\${([A-Za-z.:\\\d]+)}", value):
+        elif re.search(r"\${([A-Za-z.:\\\d]+)}", value) or not possible_constant:
             return 'path'
         else:
             return 'constant'

thingsboard_gateway/connectors/opcua/device.py

@@ -27,6 +27,7 @@ def __init__(self, path, name, config, converter, converter_for_sub, logger):
             'timeseries': [],
             'attributes': []
         }
+        self.nodes = []
 
         self.load_values()
 

thingsboard_gateway/connectors/opcua/opcua_connector.py

@@ -102,7 +102,8 @@ def __init__(self, gateway, config, connector_type):
         self.daemon = True
 
         self.__device_nodes = []
-        self.__last_poll = 0
+        self.__next_poll = 0
+        self.__next_scan = 0
 
     def open(self):
         self.__stopped = False
@@ -203,6 +204,7 @@ async def start_client(self):
             try:
                 self.__client = asyncua.Client(url=self.__opcua_url,
                                                timeout=self.__server_conf.get('timeoutInMillis', 4000) / 1000)
+                self.__client.session_timeout = self.__server_conf.get('sessionTimeoutInMillis', 3600000)
 
                 if self.__server_conf["identity"].get("type") == "cert.PEM":
                     await self.__set_auth_settings_by_cert()
@@ -220,41 +222,44 @@ async def start_client(self):
                 except Exception as e:
                     self.__log.error("Error on loading type definitions:\n %s", e)
 
-                scan_period = self.__server_conf.get('scanPeriodInMillis', 5000) / 1000
+                poll_period = int(self.__server_conf.get('pollPeriodInMillis', 5000) / 1000)
+                scan_period = int(self.__server_conf.get('scanPeriodInMillis', 3600000) / 1000)
+
                 while not self.__stopped:
-                    if monotonic() - self.__last_poll >= scan_period:
+                    if monotonic() >= self.__next_scan:
+                        self.__next_scan = monotonic() + scan_period
                         await self.__scan_device_nodes()
+
+                    if monotonic() >= self.__next_poll:
+                        self.__next_poll = monotonic() + poll_period
                         await self.__poll_nodes()
-                        self.__last_poll = monotonic()
 
-                    if not scan_period < 0.2:
-                        await asyncio.sleep(0.2)
-                    else:
-                        await asyncio.sleep(scan_period)
+                    current_time = monotonic()
+                    time_to_sleep = min(self.__next_poll - current_time, self.__next_scan - current_time)
+                    if time_to_sleep > 0:
+                        await asyncio.sleep(time_to_sleep)
             except (ConnectionError, BadSessionClosed):
                 self.__log.warning('Connection lost for %s', self.get_name())
             except asyncio.exceptions.TimeoutError:
                 self.__log.warning('Failed to connect %s', self.get_name())
             except asyncio.CancelledError as e:
                 if self.__stopped:
                     self.__log.debug('Task was cancelled due to connector stop: %s', e.__str__())
-                    break
                 else:
                     self.__log.exception('Task was cancelled: %s', e.__str__())
             except UaStatusCodeError as e:
                 self.__log.error('Error in main loop, trying to reconnect: %s', exc_info=e)
                 if self.__connected:
                     await self.__client.disconnect()
                     self.__connected = False
-                break
             except Exception as e:
                 self.__log.exception("Error in main loop: %s", e)
-                break
             finally:
-                if self.__connected:
-                    await self.__client.disconnect()
-                self.__connected = False
-                await asyncio.sleep(1)
+                if self.__stopped:
+                    if self.__connected:
+                        await self.__client.disconnect()
+                    self.__connected = False
+                await asyncio.sleep(.5)
 
     async def __set_auth_settings_by_cert(self):
         try:
@@ -340,40 +345,6 @@ async def find_node_name_space_index(self, path):
         unresolved = path[resolved_level:]
         return await self.__find_nodes(unresolved, current_parent_node=parent_node, nodes=resolved)
 
-    async def __scan_device_nodes(self):
-        existing_devices = list(map(lambda dev: dev.name, self.__device_nodes))
-
-        scanned_devices = []
-        for device in self.__config.get('mapping', []):
-            nodes = await self.find_nodes(device['deviceNodePattern'])
-            self.__log.debug('Found devices: %s', nodes)
-
-            device_names = await self._get_device_info_by_pattern(
-                device.get('deviceInfo', {}).get('deviceNameExpression'))
-
-            for device_name in device_names:
-                scanned_devices.append(device_name)
-                if device_name not in existing_devices:
-                    for node in nodes:
-                        converter = self.__load_converter(device)
-                        device_type = await self._get_device_info_by_pattern(
-                            device.get('deviceInfo', {}).get('deviceProfileExpression', 'default'),
-                            get_first=True)
-                        device_config = {**device, 'device_name': device_name, 'device_type': device_type}
-                        self.__device_nodes.append(
-                            Device(path=node, name=device_name, config=device_config,
-                                   converter=converter(device_config, self.__log),
-                                   converter_for_sub=converter(device_config, self.__log) if not self.__server_conf.get(
-                                       'disableSubscriptions',
-                                       False) else None, logger=self.__log))
-                        self.__log.info('Added device node: %s', device_name)
-
-        for device_name in existing_devices:
-            if device_name not in scanned_devices:
-                await self.__reset_nodes(device_name)
-
-        self.__log.debug('Device nodes: %s', self.__device_nodes)
-
     async def _get_device_info_by_pattern(self, pattern, get_first=False):
         result = []
 
@@ -411,17 +382,56 @@ def __convert_sub_data(self):
                     for section in ('attributes', 'timeseries'):
                         for node in device.values.get(section, []):
                             if node.get('id') == sub_node.__str__():
-                                device.converter_for_sub.convert(config={'section': section, 'key': node['key']},
-                                                                 val=data.monitored_item.Value)
+                                device.converter_for_sub.convert({'section': section, 'key': node['key']},
+                                                                 data.monitored_item.Value)
                                 converter_data = device.converter_for_sub.get_data()
 
                                 if converter_data:
                                     self.__data_to_send.put(*converter_data)
                                     device.converter_for_sub.clear_data()
             else:
-                sleep(.2)
+                sleep(.05)
 
-    async def __poll_nodes(self):
+    async def __scan_device_nodes(self):
+        await self._create_new_devices()
+        await self._load_devices_nodes()
+
+    async def _create_new_devices(self):
+        existing_devices = list(map(lambda dev: dev.name, self.__device_nodes))
+
+        scanned_devices = []
+        for device in self.__config.get('mapping', []):
+            nodes = await self.find_nodes(device['deviceNodePattern'])
+            self.__log.debug('Found devices: %s', nodes)
+
+            device_names = await self._get_device_info_by_pattern(
+                device.get('deviceInfo', {}).get('deviceNameExpression'))
+
+            for device_name in device_names:
+                scanned_devices.append(device_name)
+                if device_name not in existing_devices:
+                    for node in nodes:
+                        converter = self.__load_converter(device)
+                        device_type = await self._get_device_info_by_pattern(
+                            device.get('deviceInfo', {}).get('deviceProfileExpression', 'default'),
+                            get_first=True)
+                        device_config = {**device, 'device_name': device_name, 'device_type': device_type}
+                        self.__device_nodes.append(
+                            Device(path=node, name=device_name, config=device_config,
+                                   converter=converter(device_config, self.__log),
+                                   converter_for_sub=converter(device_config, self.__log) if not self.__server_conf.get(
+                                       'disableSubscriptions',
+                                       False) else None, logger=self.__log))
+
+                        self.__log.info('Added device node: %s', device_name)
+
+        for device_name in existing_devices:
+            if device_name not in scanned_devices:
+                await self.__reset_nodes(device_name)
+
+        self.__log.debug('Device nodes: %s', self.__device_nodes)
+
+    async def _load_devices_nodes(self):
         for device in self.__device_nodes:
             for section in ('attributes', 'timeseries'):
                 for node in device.values.get(section, []):
@@ -434,24 +444,25 @@ async def __poll_nodes(self):
                                 qualified_path = await self.find_node_name_space_index(path)
                                 if len(qualified_path) == 0:
                                     if node.get('valid', True):
-                                        self.__log.warning('Node not found; device: %s, key: %s, path: %s', device.name,
+                                        self.__log.warning('Node not found; device: %s, key: %s, path: %s',
+                                                           device.name,
                                                            node['key'], node['path'])
                                         await self.__reset_node(node)
                                     continue
                                 elif len(qualified_path) > 1:
                                     self.__log.warning(
-                                        'Multiple matching nodes found; device: %s, key: %s, path: %s; %s', device.name,
+                                        'Multiple matching nodes found; device: %s, key: %s, path: %s; %s',
+                                        device.name,
                                         node['key'], node['path'], qualified_path)
                                 node['qualified_path'] = qualified_path[0]
                                 path = qualified_path[0]
 
                             var = await self.__client.nodes.root.get_child(path)
 
+                        device.nodes.append({'var': var, 'key': node['key'], 'section': section})
+
                         if (node.get('valid') is None or
                                 (node.get('valid') and self.__server_conf.get('disableSubscriptions', False))):
-                            value = await var.read_data_value()
-                            device.converter.convert(config={'section': section, 'key': node['key']}, val=value)
-
                             if (not self.__server_conf.get('disableSubscriptions', False)
                                     and not node.get('sub_on', False)
                                     and not self.__stopped):
@@ -462,14 +473,16 @@ async def __poll_nodes(self):
                                 node['subscription'] = handle
                                 node['sub_on'] = True
                                 node['id'] = var.nodeid.to_string()
-                                self.__log.info("Subscribed on data change; device: %s, key: %s, path: %s", device.name,
+                                self.__log.info("Subscribed on data change; device: %s, key: %s, path: %s",
+                                                device.name,
                                                 node['key'], node['id'])
 
                             node['valid'] = True
                     except ConnectionError:
                         raise
                     except (BadNodeIdUnknown, BadConnectionClosed, BadInvalidState, BadAttributeIdInvalid,
-                            BadCommunicationError, BadOutOfService, BadNoMatch, BadUnexpectedError, UaStatusCodeErrors,
+                            BadCommunicationError, BadOutOfService, BadNoMatch, BadUnexpectedError,
+                            UaStatusCodeErrors,
                             BadWaitingForInitialData):
                         if node.get('valid', True):
                             self.__log.warning('Node not found (2); device: %s, key: %s, path: %s', device.name,
@@ -484,23 +497,38 @@ async def __poll_nodes(self):
                             self.__log.exception(e)
                             await self.__reset_node(node)
 
-            converter_data = device.converter.get_data()
-            if converter_data:
-                self.__data_to_send.put(*converter_data)
+    async def __poll_nodes(self):
+        all_nodes = [node_config['var'] for device in self.__device_nodes for node_config in device.nodes]
+
+        if len(all_nodes) > 0:
+            values = await self.__client.read_attributes(all_nodes)
+
+            converted_nodes_count = 0
+            for device in self.__device_nodes:
+                nodes_count = len(device.nodes)
+                device_values = values[converted_nodes_count:converted_nodes_count + nodes_count]
+                converted_nodes_count += nodes_count
+                device.converter.convert(device.nodes, device_values)
+                converter_data = device.converter.get_data()
+                if converter_data:
+                    self.__data_to_send.put(*converter_data)
 
-                device.converter.clear_data()
+                    device.converter.clear_data()
+
+            self.__log.debug('Converted nodes values count: %s', converted_nodes_count)
+        else:
+            self.__log.info('No nodes to poll')
 
     def __send_data(self):
         while not self.__stopped:
             if not self.__data_to_send.empty():
                 data = self.__data_to_send.get()
                 self.statistics['MessagesReceived'] = self.statistics['MessagesReceived'] + 1
-                self.__log.debug(data)
                 self.__gateway.send_to_storage(self.get_name(), self.get_id(), data)
                 self.statistics['MessagesSent'] = self.statistics['MessagesSent'] + 1
-                self.__log.debug('Data to ThingsBoard %s', data)
+                self.__log.debug('Count data packs to ThingsBoard: %s', self.statistics['MessagesSent'])
             else:
-                sleep(.2)
+                sleep(.05)
 
     async def get_shared_attr_node_id(self, path, result={}):
         try: