kaffa/haproxy-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add health check endpoints and environment variable configuration

Browse Source 
New features:
- haproxy_health: System-level health check (MCP, HAProxy, config files)
- haproxy_domain_health: Domain-specific health check with server status

Environment variables support:
- MCP_HOST, MCP_PORT: MCP server binding
- HAPROXY_HOST, HAPROXY_PORT: HAProxy Runtime API connection
- HAPROXY_STATE_FILE, HAPROXY_MAP_FILE, HAPROXY_SERVERS_FILE: File paths
- HAPROXY_POOL_COUNT, HAPROXY_MAX_SLOTS: Pool configuration
- LOG_LEVEL: Logging level (DEBUG, INFO, WARNING, ERROR)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
kaffa
2026-02-01 13:07:22 +00:00
parent 61dd4a69fc
commit 28df45900c
1 changed files with 169 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
mcp/server.py
View File

@@ -4,6 +4,18 @@
This module provides an MCP (Model Context Protocol) server for managing HAProxy
configuration and runtime state. It supports dynamic domain/server management
with HTTP backends (SSL termination at HAProxy frontend).
Environment Variables:
MCP_HOST: Host to bind MCP server (default: 0.0.0.0)
MCP_PORT: Port for MCP server (default: 8000)
HAPROXY_HOST: HAProxy Runtime API host (default: localhost)
HAPROXY_PORT: HAProxy Runtime API port (default: 9999)
HAPROXY_STATE_FILE: Path to server state file
HAPROXY_MAP_FILE: Path to domains.map file
HAPROXY_MAP_FILE_CONTAINER: Container path for domains.map
HAPROXY_SERVERS_FILE: Path to servers.json file
HAPROXY_POOL_COUNT: Number of pool backends (default: 100)
HAPROXY_MAX_SLOTS: Max servers per pool (default: 10)
"""
import socket
@@ -19,23 +31,33 @@ from typing import Any, Dict, Generator, List, Optional, Set, Tuple
from mcp.server.fastmcp import FastMCP
# Configure structured logging
log_level = os.getenv("LOG_LEVEL", "INFO").upper()
logging.basicConfig(
level=logging.INFO,
level=getattr(logging, log_level, logging.INFO),
format='%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)
mcp = FastMCP("haproxy", host="0.0.0.0", port=8000)
# MCP Server configuration
MCP_HOST: str = os.getenv("MCP_HOST", "0.0.0.0")
MCP_PORT: int = int(os.getenv("MCP_PORT", "8000"))
mcp = FastMCP("haproxy", host=MCP_HOST, port=MCP_PORT)
# Constants
HAPROXY_SOCKET: Tuple[str, int] = ("localhost", 9999)
STATE_FILE: str = "/opt/haproxy/data/servers.state"
MAP_FILE: str = "/opt/haproxy/conf/domains.map"
MAP_FILE_CONTAINER: str = "/usr/local/etc/haproxy/domains.map"
SERVERS_FILE: str = "/opt/haproxy/conf/servers.json"
POOL_COUNT: int = 100
MAX_SLOTS: int = 10
# HAProxy Runtime API configuration
HAPROXY_HOST: str = os.getenv("HAPROXY_HOST", "localhost")
HAPROXY_PORT: int = int(os.getenv("HAPROXY_PORT", "9999"))
HAPROXY_SOCKET: Tuple[str, int] = (HAPROXY_HOST, HAPROXY_PORT)
# File paths (configurable via environment)
STATE_FILE: str = os.getenv("HAPROXY_STATE_FILE", "/opt/haproxy/data/servers.state")
MAP_FILE: str = os.getenv("HAPROXY_MAP_FILE", "/opt/haproxy/conf/domains.map")
MAP_FILE_CONTAINER: str = os.getenv("HAPROXY_MAP_FILE_CONTAINER", "/usr/local/etc/haproxy/domains.map")
SERVERS_FILE: str = os.getenv("HAPROXY_SERVERS_FILE", "/opt/haproxy/conf/servers.json")
# Pool configuration
POOL_COUNT: int = int(os.getenv("HAPROXY_POOL_COUNT", "100"))
MAX_SLOTS: int = int(os.getenv("HAPROXY_MAX_SLOTS", "10"))
# Validation patterns - compiled once for performance
DOMAIN_PATTERN = re.compile(
@@ -863,6 +885,143 @@ def haproxy_stats() -> str:
return f"Error: {e}"
@mcp.tool()
def haproxy_health() -> str:
"""Check health of MCP server and HAProxy connectivity.
Returns:
JSON with health status of MCP server, HAProxy, and configuration files
"""
result: Dict[str, Any] = {
"status": "healthy",
"timestamp": time.time(),
"components": {
"mcp": {"status": "ok"},
"haproxy": {"status": "unknown"},
"config_files": {"status": "unknown"}
}
}
# Check HAProxy connectivity
try:
info = haproxy_cmd("show info")
for line in info.split("\n"):
if line.startswith("Version:"):
result["components"]["haproxy"]["version"] = line.split(":", 1)[1].strip()
elif line.startswith("Uptime_sec:"):
result["components"]["haproxy"]["uptime_sec"] = int(line.split(":", 1)[1].strip())
result["components"]["haproxy"]["status"] = "ok"
except HaproxyError as e:
result["components"]["haproxy"]["status"] = "error"
result["components"]["haproxy"]["error"] = str(e)
result["status"] = "degraded"
# Check configuration files
files_ok = True
file_status: Dict[str, str] = {}
for name, path in [("map_file", MAP_FILE), ("servers_file", SERVERS_FILE)]:
if os.path.exists(path):
file_status[name] = "ok"
else:
file_status[name] = "missing"
files_ok = False
result["components"]["config_files"]["files"] = file_status
result["components"]["config_files"]["status"] = "ok" if files_ok else "warning"
if not files_ok:
result["status"] = "degraded"
return json.dumps(result, indent=2)
@mcp.tool()
def haproxy_domain_health(domain: str) -> str:
"""Check health status of backend servers for a specific domain.
Args:
domain: The domain name to check health for
Returns:
JSON with domain health status including each server's state
"""
if not validate_domain(domain):
return json.dumps({"error": "Invalid domain format"})
try:
backend, _ = get_backend_and_prefix(domain)
except ValueError as e:
return json.dumps({"error": str(e)})
result: Dict[str, Any] = {
"domain": domain,
"backend": backend,
"status": "unknown",
"servers": [],
"healthy_count": 0,
"total_count": 0
}
try:
# Get server states
state_output = haproxy_cmd("show servers state")
stat_output = haproxy_cmd("show stat")
# Build status map from stat output (has UP/DOWN/MAINT status)
status_map: Dict[str, Dict[str, str]] = {}
for stat in parse_stat_csv(stat_output):
if stat["pxname"] == backend and stat["svname"] not in ["FRONTEND", "BACKEND"]:
status_map[stat["svname"]] = {
"status": stat["status"],
"check_status": stat["check_status"],
"weight": stat["weight"]
}
# Parse server state for address info
for line in state_output.split("\n"):
parts = line.split()
if len(parts) >= STATE_MIN_COLUMNS and parts[StateField.BE_NAME] == backend:
server_name = parts[StateField.SRV_NAME]
addr = parts[StateField.SRV_ADDR]
port = parts[StateField.SRV_PORT]
# Skip disabled servers (0.0.0.0)
if addr == "0.0.0.0":
continue
server_info: Dict[str, Any] = {
"name": server_name,
"addr": f"{addr}:{port}",
"status": "unknown"
}
# Get status from stat output
if server_name in status_map:
server_info["status"] = status_map[server_name]["status"]
server_info["check_status"] = status_map[server_name]["check_status"]
server_info["weight"] = status_map[server_name]["weight"]
result["servers"].append(server_info)
result["total_count"] += 1
if server_info["status"] == "UP":
result["healthy_count"] += 1
# Determine overall status
if result["total_count"] == 0:
result["status"] = "no_servers"
elif result["healthy_count"] == result["total_count"]:
result["status"] = "healthy"
elif result["healthy_count"] > 0:
result["status"] = "degraded"
else:
result["status"] = "down"
return json.dumps(result, indent=2)
except HaproxyError as e:
return json.dumps({"error": str(e)})
@mcp.tool()
def haproxy_backends() -> str:
"""List all HAProxy backends.