Channels 4.0 channel worker logging

[pmayostendorp]

Objective

I am building consumers that will help me interact with an OPC-UA server. I have constructed a consumer class that spins up a client that will subscribe to OPC-UA data changes and pass the resulting events/messages into a channel ( opcua-listen). Just doing proof of concept stuff at this point, but the eventual goal will be to capture device states in the application’s back end.

Issue

The problem is none of my consumer log messages appear when I run the associated worker, and it is impossible for me to tell if my consumer’s methods are receiving these messages. In the same project, I've confirmed that I can get logs from a WebsocketConsumer to appear, so it appears that the issue relates only to the AsyncConsumer that I’ve established.

Setup

My consumer class looks like this:

import logging
logger = logging.getLogger("mes.consumers")

class OPCUAListenerConsumer(AsyncConsumer):
    
    def __init__(self, connect_client=True) -> None:
        super().__init__()
        # This sets up an OPCUAClient that subscribes to changes on the OPC-UA server
        self.opcuaclient = OPCUAClient(get_channel_layer())
        if settings.OPCUA_CLIENT_START:
            logger.debug("Initializing OPC-UA client")
            # set up asyncio Task for this here...
        else:
            logger.warn("Will not start OPC-UA client!")

    async def opcua_log(self, event):
        """Log the message and return the message to the client."""
        logger.info(f"Message received: {json.dumps(event['message'], indent=4)}")

The substance of the client class looks more or less like below:

import logging
logger = logging.getLogger("mes.clients")

class OPCUAClient:
    OPCUA_LISTENER_CHANNEL_NAME:str = "opcua-listen"  # the name for our channel
    OPCUA_LOG:str = "opcua-listen.log"

    def __init__(self, channel_layer, opcua_url=None) -> None:
        self.channel_layer = channel_layer
        self.url = f"opc.tcp://{settings.OPCUA_HOST}:{settings.OPCUA_PORT}"
        if opcua_url is not None:
            self.url = opcua_url
        self.client = Client(self.url)
        # Some other housekeeping here... subscribing to points, etc.
        
    async def run(self):
        """Run the client asynchronously and exit cleanly."""
        loop = asyncio.get_event_loop()
        loop.add_signal_handler(signal.SIGINT, self.ask_exit)
        loop.add_signal_handler(signal.SIGTERM, self.ask_exit)
        await self.connect()
        await self.finish()

    # This is a handler for OPC-UA subscriptions
    def datachange_notification(self, node, val, data):
        """
        Notification will be sent to an appropriate handler(s) in the opcua channel.
        """
        node_identifier = node.nodeid.to_string()
        node_label = self.point_subscriptions[node_identifier]
        timestamp = data.monitored_item.Value.SourceTimestamp.timestamp()
        msg = {
            "time": timestamp,
            "node_identifier": node_identifier,
            "node_label": node_label,
            "data": val
        }
        event = {
            "type": self.OPCUA_LOG,
            "message": msg
        }
        # This is where we relay messages into the channel, proxying for a protocol server
        logger.debug(f"Sending: {json.dumps(event, indent=4)}")
        async_to_sync(self.channel_layer.send(self.OPCUA_LISTENER_CHANNEL_NAME, event))

My Django logging setup is:

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
        },
    },
    "root": {
        "handlers": ["console"],
        "level": "DEBUG",
    },
    "loggers": {
        "django.channels.worker": {
            "level": "DEBUG",
            "handlers": ["console"],
        },
        "mes.clients": {
            "level": "DEBUG",
            "handlers": ["console"],
        },
        "mes.consumers": {
            "level": "DEBUG",
            "handlers": ["console"],
        },
        "asgiref.server": {
            "level": "DEBUG",
            "handlers": ["console"]
        }
    },
}

In stdout from runserver and runworker, I find no trace of log statements for this consumer or the associated client that it spins up. I do see logs for other consumer types (from HTTP and Websocket protocol stuff), but not from named channels.

• Your OS and runtime environment, and browser if applicable
  python:3.9.0-buster base docker image

• A pip freeze output showing your package versions
  freeze.txt

• What you expected to happen vs. what actually happened
  Expecting to see debug log statements per other consumers, Django views, etc. Seems very similar to Logging not working #1625 Logging not working #1648.

• How you're running Channels (runserver? daphne/runworker? Nginx/Apache in front?)
  manage.py runserver and manage.py runworker

[pmayostendorp]

I've since set up a separate management command and spun up these clients in a totally different worker thread from the channels and their consumers. This seems to have resolved the issue.

My $0.02: there is very limited documentation on the right way to create or attach new protocol servers or on how to proxy such a server through a small client, such as I've done above. One of the purported benefits of channels, straight from the docs:

Channels wraps Django’s native asynchronous view support, allowing Django projects to handle not only HTTP, but protocols that require long-running connections too - WebSockets, MQTT, chatbots, amateur radio, and more.

But as far as solid examples and guidance on best practices for going beyond HTTP, Websockets and the natively supported stuff, not much exists. Are there community projects, perhaps, that could be drawn on to augment the docs and enable such workflows (namely, a workflow that bridges protocols like MQTT, OPC-UA, etc. with channels)?

[carltongibson]

Yep, that's the problem. Async is still the Wild West, so folks are still working out the patterns. Writing about your experiences is the best way to help build a critical mass of community knowledge and experience.