Failed subscription behavior, recovering from a failed subscriptions, monitoring health of subscriptions

[kj4tmp]

I am experiencing an issue where my subscriptions stop reporting data and I cannot tell why. I believe the subscriptions are being removed on the server but I cannot tell when they have been removed.

Here is some example code, in this example I subscribe to data changes but call a sleep that will trigger the lifetime count to be exceeded on the server and the subscription will be deleted. The example uses a public OPCUA example server and a node that reports random data(constantly changing).

import asyncio
from asyncua import Client, Node, ua
import sys
import time
import logging
logging.basicConfig(level=logging.WARNING)

class SubscriptionHandler:
    def __init__(self):
        self.i = 0

    def datachange_notification(self, node: Node, val, data: ua):
        """
        Callback for asyncua Subscription.
        This method will be called when the Client received a data change message from the Server.
        """
        self.i = self.i + 1
        print(self.i, val)

    def event_notification(self, event: ua.EventNotificationList):
        print(event)

    def status_change_notification(self, status: ua.StatusChangeNotification):
        print(status)



async def task(loop):
    client = Client(url='opc.tcp://opcua.demo-this.com:51210/UA/SampleServer')
    async with client:
        print(f'Connected to opcua endpoint')
        dynamic_node = 'ns=2;i=11218'
        static_node = 'ns=2;i=10226'
        node = client.get_node(dynamic_node)

        handler = SubscriptionHandler()
        params = ua.CreateSubscriptionParameters(
            RequestedPublishingInterval=100,
            RequestedLifetimeCount=10,
            RequestedMaxKeepAliveCount=0,
            MaxNotificationsPerPublish=10
        )
        subscription = await client.create_subscription(params, handler)
        await subscription.subscribe_data_change(node, queuesize=0, sampling_interval=-1)
        
        while True:
            try:
                await client.check_connection()
                await asyncio.sleep(1)
                for i in range(12):
                    time.sleep(1)
                    print(f'blocking slept for {i+1}s')
            except Exception as e:
                print(f'Connection lost. Exiting. Error: {e}')
                sys.exit(1)

def main():
    loop = asyncio.get_event_loop()
    loop.set_debug(False)
    loop.run_until_complete(task(loop))
    loop.close()

main()

here is the output:

Connected to opcua endpoint
WARNING:asyncua.client.client:Revised values returned differ from subscription values: CreateSubscriptionResult(SubscriptionId=9, RevisedPublishingInterval=100.0, RevisedLifetimeCount=100, RevisedMaxKeepAliveCount=3)
1 60.0
2 15.0
3 72.0
4 8.0
5 1.0
blocking slept for 1s
blocking slept for 2s
blocking slept for 3s
blocking slept for 4s
blocking slept for 5s
blocking slept for 6s
blocking slept for 7s
blocking slept for 8s
blocking slept for 9s
blocking slept for 10s
blocking slept for 11s
blocking slept for 12s
6 -1.0
WARNING:asyncua.client.ua_client.UASocketProtocol:ServiceFault (BadNoSubscription, diagnostics: DiagnosticInfo(SymbolicId=None, NamespaceURI=None, Locale=None, LocalizedText=None, AdditionalInfo=None, InnerStatusCode=None, InnerDiagnosticInfo=None)) from server received  in response to PublishRequest
blocking slept for 1s
blocking slept for 2s
blocking slept for 3s
blocking slept for 4s
blocking slept for 5s
blocking slept for 6s
blocking slept for 7s
blocking slept for 8s
blocking slept for 9s
blocking slept for 10s
blocking slept for 11s
blocking slept for 12s
blocking slept for 1s
blocking slept for 2s
blocking slept for 3s
blocking slept for 4s
blocking slept for 5s

I never receive any more data and no exceptions or errors are raised. It appears that BadNoSubscription raised but that is caught in

opcua-asyncio/asyncua/client/ua_client.py

Line 598 in daa74c2

except BadNoSubscription: # See Spec. Part 5, 13.8.1

but I cannot do anything about it...

How do I detect and recover from a subscription that has failed server side?

[kj4tmp]

it appears that _publish_loop stops (due to return statement) after a BadNoSubscription in

opcua-asyncio/asyncua/client/ua_client.py

Line 613 in daa74c2

return

perhaps I need to somehow monitor the execution of the publish loop?

[schroeder-]

If you call sleep from time or any other blocking code. You will stop all opc ua code from working. If the client doens't answer in a time Intervall, the subscription gets deleted. Try to move you blocking code in thread for example.

      while True:
           try:
               await client.check_connection()
               await asyncio.sleep(1)
               def blocking_call(): 
                   for i in range(12):
                       time.sleep(1)
                       print(f'blocking slept for {i+1}s')
                await asyncio.to_thread(blocking_call)
           except Exception as e:
               print(f'Connection lost. Exiting. Error: {e}')
               sys.exit(1)

[kj4tmp]

Sleep in this case was only used to trigger the subscription deletion server side as an example.

In my real application, it may be difficult for me to fully ensure this non-blocking behavior 100% of the time. Is there a way for me to detect and recover from this scenario? Somehow check the status of my subscription?

[schroeder-]

You can increase the RequestedLifetimeCount and maybe set RequestedMaxKeepAliveCount to a value other than 0.
RequestedLifetimeCount should be at least 3 times of RequestedMaxKeepAliveCount.

LifetimeCounter

A value that contains the number of consecutive publishing timer expirations without Client activity before the Subscription is terminated.

[kj4tmp]

Forgive me but this does not answer my question.

The opcua server I am working with (codesys) is killing my subscriptions for some reason and it is difficult for me to know why. My lifetime count and and max keep alive count are extremely high (in the thousands) and my requested publishing interval is only 50 ms.

How can I check if my subscription still exists?

[kj4tmp]

i was able to detect the dead publish task and recover with:

if not client.uaclient._publish_task or client.uaclient._publish_task.done():
      print('detected dead publish task, rebuilding...')
      subscription = await client.create_subscription(params, handler)
      await subscription.subscribe_data_change(node, queuesize=0, sampling_interval=-1)

[kj4tmp]

for anyone else's reference, here is the revised example with the fix. the dead subscription is detected and a new subscription is created.

import asyncio
from asyncua import Client, Node, ua
import sys
import time
import logging
logging.basicConfig(level=logging.WARNING)

class SubscriptionHandler:
    def __init__(self):
        self.i = 0

    def datachange_notification(self, node: Node, val, data: ua):
        """
        Callback for asyncua Subscription.
        This method will be called when the Client received a data change message from the Server.
        """
        self.i = self.i + 1
        print(self.i, val)

    def event_notification(self, event: ua.EventNotificationList):
        print(event)

    def status_change_notification(self, status: ua.StatusChangeNotification):
        print(status)



async def task(loop):
    client = Client(url='opc.tcp://opcua.demo-this.com:51210/UA/SampleServer')
    async with client:
        print(f'Connected to opcua endpoint')
        dynamic_node = 'ns=2;i=11218'
        static_node = 'ns=2;i=10226'
        node = client.get_node(dynamic_node)

        handler = SubscriptionHandler()
        params = ua.CreateSubscriptionParameters(
            RequestedPublishingInterval=100,
            RequestedLifetimeCount=10,
            RequestedMaxKeepAliveCount=0,
            MaxNotificationsPerPublish=10
        )
        subscription = await client.create_subscription(params, handler)
        await subscription.subscribe_data_change(node, queuesize=0, sampling_interval=-1)
        
        while True:
            try:
                await client.check_connection()
                await asyncio.sleep(1)
                for i in range(12):
                    time.sleep(1)
                    print(f'blocking slept for {i+1}s')
                if not client.uaclient._publish_task or client.uaclient._publish_task.done():
                    print('detected dead publish task, rebuilding...')
                    subscription = await client.create_subscription(params, handler)
                    await subscription.subscribe_data_change(node, queuesize=0, sampling_interval=-1)
            except Exception as e:
                print(f'Connection lost. Exiting. Error: {e}')
                sys.exit(1)

def main():
    loop = asyncio.get_event_loop()
    loop.set_debug(False)
    loop.run_until_complete(task(loop))
    loop.close()

main()