Add a new UnixDomainSocketHandler to logging.handlers for modern containerized environments

[Yocote0111]

Feature or enhancement

Proposal:

Background:
In modern infrastructure architectures, specifically when using Kubernetes, containers frequently communicate with sidecar agents (like Fluent Bit, Vector, or OpenTelemetry Collector) for logging and data streaming.
While TCP/IP sockets are available, Unix Domain Sockets (UDS) are highly preferred for communication within the same Pod because:

1. Performance: They have lower overhead by avoiding the TCP/IP stack overhead, operating directly in kernel memory.
2. Security: They can be constrained by file system permissions within the shared Pod volume (emptyDir).
3. Reliability: They are not susceptible to network failures within the node.
   Currently, Python's logging.handlers provides SocketHandler (TCP) and DatagramHandler (UDP), but lacks a dedicated handler for Unix Domain Sockets. Users must implement a custom handler by subclassing SocketHandler, which is a common point of friction.
   Proposal:
   Introduce a new handler class named UnixDomainSocketHandler in logging.handlers that natively supports AF_UNIX sockets.
   Implementation Example (Draft):

import logging
import socket
import os

class UnixDomainSocketHandler(logging.Handler):
    """
    A handler class which writes logging records directly to a 
    Unix Domain Socket as raw bytes.
    
    Usage:
        import logging
        from logging.handlers import UnixDomainSocketHandler

        # 1. Setup the logger
        logger = logging.getLogger("MyUDSLogger")
        
        # 2. Configure the handler with the socket path created by the receiver
        uds_handler = UnixDomainSocketHandler("/var/run/vector/vector.sock")
        
        # 3. Optional: Set a formatter to control the output format
        # For raw binary data, use a simple formatter that doesn't add text decorations
        formatter = logging.Formatter('%(message)s')
        uds_handler.setFormatter(formatter)
        
        logger.addHandler(uds_handler)
        
        # 4. Log data as raw bytes (if formatted correctly)
        # Assuming Vector is configured to decode as 'bytes'
        logger.info(b"\x00\x01\x02\x03\xff")
    """
    def __init__(self, address, mode=socket.SOCK_STREAM):
        """
        Initialize the handler with the UDS address and socket type.
        """
        super().__init__()
        self.address = address
        self.mode = mode
        self.sock = None

    def _connect(self):
        """
        Create and connect to the Unix Domain Socket if not already connected.
        """
        if self.sock is None:
            self.sock = socket.socket(socket.AF_UNIX, self.mode)
            try:
                self.sock.connect(self.address)
            except socket.error:
                self.sock.close()
                self.sock = None
                raise

    def emit(self, record):
        """
        Emit a record by sending the formatted record to the socket.
        """
        try:
            self._connect()
            # Format the record and convert to bytes
            msg = self.format(record)
            if isinstance(msg, str):
                msg = msg.encode('utf-8')
            
            # Send raw bytes to the socket
            if self.mode == socket.SOCK_STREAM:
                self.sock.sendall(msg)
            else:
                self.sock.sendto(msg, self.address)
        except Exception:
            self.handleError(record)

    def close(self):
        """
        Closes the socket.
        """
        if self.sock:
            self.sock.close()
            self.sock = None
        super().close()

Benefits:
• Provides first-class support for UDS in the standard library.
• Simplifies code for developers deploying Python applications to Kubernetes/containers.
• Improves performance and reliability for high-throughput logging requirements.

Has this already been discussed elsewhere?

No response given

Links to previous discussion of this feature:

No response