Method _query_dirty_workers

gunicorn/ctl/handlers.py:516–559

Query the dirty arbiter for worker information. Connects to the dirty arbiter socket and sends a status request. Returns: List of dirty worker info dicts, or empty list on error

(self)

        }

    def _query_dirty_workers(self) -> list:
        """
        Query the dirty arbiter for worker information.

        Connects to the dirty arbiter socket and sends a status request.

        Returns:
            List of dirty worker info dicts, or empty list on error
        """
        # Get socket path from arbiter object or environment
        dirty_socket_path = None
        if hasattr(self.arbiter, 'dirty_arbiter') and self.arbiter.dirty_arbiter:
            dirty_socket_path = getattr(self.arbiter.dirty_arbiter, 'socket_path', None)
        if not dirty_socket_path:
            dirty_socket_path = os.environ.get('GUNICORN_DIRTY_SOCKET')
        if not dirty_socket_path:
            return []

        try:
            from gunicorn.dirty.protocol import DirtyProtocol

            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            sock.settimeout(2.0)
            sock.connect(dirty_socket_path)

            # Send status request
            request = {
                "type": DirtyProtocol.MSG_TYPE_STATUS,
                "id": "ctl-status-1",
            }
            DirtyProtocol.write_message(sock, request)

            # Read response
            response = DirtyProtocol.read_message(sock)
            sock.close()

            if response.get("type") == DirtyProtocol.MSG_TYPE_RESPONSE:
                result = response.get("result", {})
                return result.get("workers", [])

        except Exception:
            pass

        return []

    def help(self) -> dict:
        """