[Serve] Add HAProxy support for Ray Serve

[eicherseiji]

Summary

• Introduce HAProxy support to OSS Ray Serve
• HAProxy can be used as an alternative to the default Serve HTTP proxy, providing load balancing with graceful reloads and health checks
• Feature is disabled by default and enabled via RAY_SERVE_ENABLE_HAPROXY=1 environment variable

Changes

• Add haproxy.py (1209 lines) and haproxy_templates.py (132 lines) for HAProxy management
• Add HAProxy-related constants to constants.py (RAY_SERVE_ENABLE_HAPROXY, timeouts, health check settings, etc.)
• Add TARGET_GROUPS to LongPollNamespace for broadcasting target group updates
• Add get_proxy_actor_class() to default_impl.py for proxy selection based on feature flag
• Modify controller.py to support HAProxy mode with target group broadcasting
• Add comprehensive test coverage (4 test files, ~3600 lines)

Test plan

• Verify syntax checks pass for all new/modified files
• Run RAY_SERVE_ENABLE_HAPROXY=1 pytest python/ray/serve/tests/test_haproxy.py -v
• Run RAY_SERVE_ENABLE_HAPROXY=1 pytest python/ray/serve/tests/test_haproxy_api.py -v
• Run RAY_SERVE_ENABLE_HAPROXY=1 pytest python/ray/serve/tests/test_metrics_haproxy.py -v
• Run RAY_SERVE_ENABLE_HAPROXY=1 pytest python/ray/serve/tests/unit/test_controller_haproxy.py -v
• Verify feature isolation: with RAY_SERVE_ENABLE_HAPROXY=0, behavior should be unchanged
• Verify import: python -c "from ray.serve._private.haproxy import HAProxyManager; print('OK')"

[gemini-code-assist]

Code Review

This pull request introduces HAProxy support to Ray Serve, which is a significant and valuable feature. The implementation is comprehensive, covering HAProxy process management, dynamic configuration generation, and extensive testing. The code is well-structured, with a clear separation of concerns. I have a few suggestions to improve robustness and efficiency, such as replacing an external socat dependency with native Python asyncio for socket communication and refining how environment variables are read. Overall, this is a solid contribution.

[gemini-code-assist]

The _send_socket_command method currently uses socat as a subprocess to communicate with the HAProxy admin socket. This introduces a dependency on an external command-line tool (socat) which might not be available in all environments, and is less efficient than using Python's native socket library.

As noted in the TODO on line 748, this can be improved by using Python's asyncio library to interact with the UNIX domain socket directly. This removes the external dependency and is more robust and performant.

Here's a suggested implementation:

async def _send_socket_command(self, command: str) -> str:
    """Send a command to the HAProxy stats socket."""
    try:
        reader, writer = await asyncio.wait_for(
            asyncio.open_unix_connection(self.cfg.socket_path),
            timeout=5.0
        )

        try:
            writer.write(f"{command}\n".encode("utf-8"))
            await writer.drain()

            response_bytes = await asyncio.wait_for(reader.read(), timeout=5.0)
            result = response_bytes.decode("utf-8", errors="ignore")
            logger.debug(f"Socket command '{command}' returned {len(result)} chars.")
            return result
        finally:
            writer.close()
            await writer.wait_closed()

    except FileNotFoundError:
        raise RuntimeError(
            f"HAProxy socket file does not exist: {self.cfg.socket_path}."
        )
    except asyncio.TimeoutError:
        raise RuntimeError(
            f"Timeout while sending command '{command}' to HAProxy socket"
        )
    except Exception as e:
        raise RuntimeError(f"Failed to send socket command '{command}': {e}")

[gemini-code-assist]

This code calls os.environ.get("RAY_SERVE_HAPROXY_TIMEOUT_SERVER_S") twice, which is inefficient. You can use the walrus operator := to store the result of the first call and reuse it, making the code more efficient and readable.

Suggested change

	RAY_SERVE_HAPROXY_TIMEOUT_SERVER_S = (
	int(os.environ.get("RAY_SERVE_HAPROXY_TIMEOUT_SERVER_S"))
	if os.environ.get("RAY_SERVE_HAPROXY_TIMEOUT_SERVER_S")
	else None
	)
	RAY_SERVE_HAPROXY_TIMEOUT_SERVER_S = (
	int(val)
	if (val := os.environ.get("RAY_SERVE_HAPROXY_TIMEOUT_SERVER_S"))
	else None
	)

[gemini-code-assist]

This code calls os.environ.get("RAY_SERVE_HAPROXY_TIMEOUT_CONNECT_S") twice, which is inefficient. You can use the walrus operator := to store the result of the first call and reuse it, making the code more efficient and readable.

Suggested change

	RAY_SERVE_HAPROXY_TIMEOUT_CONNECT_S = (
	int(os.environ.get("RAY_SERVE_HAPROXY_TIMEOUT_CONNECT_S"))
	if os.environ.get("RAY_SERVE_HAPROXY_TIMEOUT_CONNECT_S")
	else None
	)
	RAY_SERVE_HAPROXY_TIMEOUT_CONNECT_S = (
	int(val)
	if (val := os.environ.get("RAY_SERVE_HAPROXY_TIMEOUT_CONNECT_S"))
	else None
	)

[gemini-code-assist]

As this TODO suggests, using an HTTP health check would be a more robust way to verify HAProxy's availability. The current implementation checks if the admin socket is responsive, but an HTTP check would verify that the frontend is up and correctly configured to serve traffic. You could use an HTTP client like httpx to make a request to http://127.0.0.1:{self.cfg.frontend_port}/-/healthz and check for a successful response.