Skip to content

directory_server.health

directory_server.health

Health check and monitoring HTTP server.

Provides endpoints for health checks and status monitoring.

Classes

HealthCheckHandler

Bases: BaseHTTPRequestHandler

Source code in directory_server/src/directory_server/health.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class HealthCheckHandler(BaseHTTPRequestHandler):
    server_instance: DirectoryServer | None = None

    def log_message(self, format: str, *args: Any) -> None:
        pass

    def do_GET(self) -> None:  # noqa: N802
        if self.path == "/health":
            self._handle_health()
        elif self.path == "/status":
            self._handle_status()
        else:
            self.send_error(404)

    def _handle_health(self) -> None:
        if not self.server_instance:
            self.send_error(503)
            return

        try:
            is_healthy = self.server_instance.is_healthy()
            status_code = 200 if is_healthy else 503

            self.send_response(status_code)
            self.send_header("Content-Type", "application/json")
            self.end_headers()

            response = {"status": "healthy" if is_healthy else "unhealthy"}
            self.wfile.write(json.dumps(response).encode())
        except Exception as e:
            logger.error(f"Health check error: {e}")
            self.send_error(500)

    def _handle_status(self) -> None:
        if not self.server_instance:
            self.send_error(503)
            return

        try:
            stats = self.server_instance.get_detailed_stats()

            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()

            self.wfile.write(json.dumps(stats, default=str).encode())
        except Exception as e:
            logger.error(f"Status check error: {e}")
            self.send_error(500)
Attributes
server_instance: DirectoryServer | None = None class-attribute instance-attribute
Functions
do_GET() -> None
Source code in directory_server/src/directory_server/health.py 
26
27
28
29
30
31
32
def do_GET(self) -> None:  # noqa: N802
    if self.path == "/health":
        self._handle_health()
    elif self.path == "/status":
        self._handle_status()
    else:
        self.send_error(404)
log_message(format: str, *args: Any) -> None
Source code in directory_server/src/directory_server/health.py 
23
24
def log_message(self, format: str, *args: Any) -> None:
    pass

HealthCheckServer

Source code in directory_server/src/directory_server/health.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class HealthCheckServer:
    def __init__(self, host: str = "127.0.0.1", port: int = 8080):
        self.host = host
        self.port = port
        self.httpd: HTTPServer | None = None
        self.thread: Thread | None = None

    def start(self, server_instance: DirectoryServer) -> None:
        HealthCheckHandler.server_instance = server_instance

        self.httpd = HTTPServer((self.host, self.port), HealthCheckHandler)
        self.thread = Thread(target=self.httpd.serve_forever, daemon=True)
        self.thread.start()

        logger.info(f"Health check server started on {self.host}:{self.port}")

    def stop(self) -> None:
        if self.httpd:
            self.httpd.shutdown()
            self.httpd.server_close()  # Explicitly close the socket

            # Wait for thread to actually terminate to avoid race conditions
            # This is important in Python 3.12+ where thread cleanup is stricter
            if self.thread and self.thread.is_alive():
                self.thread.join(timeout=5.0)
                if self.thread.is_alive():
                    logger.warning("Health check server thread did not terminate in time")

            # Clear class-level state to avoid leakage between test runs
            HealthCheckHandler.server_instance = None
            self.httpd = None
            self.thread = None
            logger.info("Health check server stopped")
Attributes
host = host instance-attribute
httpd: HTTPServer | None = None instance-attribute
port = port instance-attribute
thread: Thread | None = None instance-attribute
Functions
__init__(host: str = '127.0.0.1', port: int = 8080)
Source code in directory_server/src/directory_server/health.py 
72
73
74
75
76
def __init__(self, host: str = "127.0.0.1", port: int = 8080):
    self.host = host
    self.port = port
    self.httpd: HTTPServer | None = None
    self.thread: Thread | None = None
start(server_instance: DirectoryServer) -> None
Source code in directory_server/src/directory_server/health.py 
78
79
80
81
82
83
84
85
def start(self, server_instance: DirectoryServer) -> None:
    HealthCheckHandler.server_instance = server_instance

    self.httpd = HTTPServer((self.host, self.port), HealthCheckHandler)
    self.thread = Thread(target=self.httpd.serve_forever, daemon=True)
    self.thread.start()

    logger.info(f"Health check server started on {self.host}:{self.port}")
stop() -> None
Source code in directory_server/src/directory_server/health.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def stop(self) -> None:
    if self.httpd:
        self.httpd.shutdown()
        self.httpd.server_close()  # Explicitly close the socket

        # Wait for thread to actually terminate to avoid race conditions
        # This is important in Python 3.12+ where thread cleanup is stricter
        if self.thread and self.thread.is_alive():
            self.thread.join(timeout=5.0)
            if self.thread.is_alive():
                logger.warning("Health check server thread did not terminate in time")

        # Clear class-level state to avoid leakage between test runs
        HealthCheckHandler.server_instance = None
        self.httpd = None
        self.thread = None
        logger.info("Health check server stopped")