46. Project: an async service health checker

Goal: check a list of endpoints concurrently every N seconds: DNS time, TCP connect time, TLS handshake time and certificate expiry, HTTP status, and total latency. Print a table and exit non-zero if anything is down, so cron or CI can alert. It reuses every chapter in this Part.

Only point it at systems you own or are allowed to test.

# labs/healthcheck.py
import asyncio
import socket
import ssl
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from urllib.parse import urlsplit

@dataclass
class Result:
    url: str
    ok: bool = False
    dns_ms: float = 0
    tcp_ms: float = 0
    tls_ms: float = 0
    http_ms: float = 0
    status: int | None = None
    cert_days: int | None = None
    error: str = ""

async def check(url: str, timeout: float = 5.0) -> Result:
    res = Result(url)
    u = urlsplit(url)
    host, https = u.hostname or "", u.scheme == "https"
    port = u.port or (443 if https else 80)
    loop = asyncio.get_running_loop()
    try:
        async with asyncio.timeout(timeout):
            t0 = time.perf_counter()
            infos = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
            ip = infos[0][4][0]
            t1 = time.perf_counter(); res.dns_ms = (t1 - t0) * 1000

            reader, writer = await asyncio.open_connection(ip, port)
            t2 = time.perf_counter(); res.tcp_ms = (t2 - t1) * 1000

            if https:
                ctx = ssl.create_default_context()
                await writer.start_tls(ctx, server_hostname=host)     # TLS on the open connection
                t3 = time.perf_counter(); res.tls_ms = (t3 - t2) * 1000
                cert = writer.get_extra_info("peercert")
                expires = datetime.fromtimestamp(ssl.cert_time_to_seconds(cert["notAfter"]), timezone.utc)
                res.cert_days = (expires - datetime.now(timezone.utc)).days
            else:
                t3 = t2

            path = u.path or "/"
            writer.write(f"GET {path} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: healthcheck/0.1\r\n"
                         "Connection: close\r\n\r\n".encode())
            await writer.drain()
            status_line = await reader.readline()
            res.http_ms = (time.perf_counter() - t3) * 1000
            res.status = int(status_line.split()[1])
            res.ok = res.status < 500 and (res.cert_days is None or res.cert_days > 14)
            if res.cert_days is not None and res.cert_days <= 14:
                res.error = f"certificate expires in {res.cert_days} days"
            writer.close()
    except TimeoutError:
        res.error = "timeout"
    except (OSError, ssl.SSLError, ValueError, IndexError) as e:
        res.error = f"{type(e).__name__}: {e}"
    return res

async def main(urls: list[str]) -> int:
    results = await asyncio.gather(*(check(u) for u in urls))
    print(f"{'':2} {'URL':<34} {'DNS':>6} {'TCP':>6} {'TLS':>6} {'HTTP':>6} {'CODE':>4} {'CERT':>5}  NOTE")
    for r in results:
        mark = "✅" if r.ok else "❌"
        cert = f"{r.cert_days}d" if r.cert_days is not None else "-"
        print(f"{mark} {r.url:<34} {r.dns_ms:6.0f} {r.tcp_ms:6.0f} {r.tls_ms:6.0f} {r.http_ms:6.0f} "
              f"{r.status or '-':>4} {cert:>5}  {r.error}")
    return 0 if all(r.ok for r in results) else 1

if __name__ == "__main__":
    urls = sys.argv[1:] or ["https://example.com/", "https://www.python.org/", "http://neverssl.com/",
                            "https://expired.badssl.com/", "https://nonexistent.invalid/"]
    raise SystemExit(asyncio.run(main(urls)))

   URL                                   DNS    TCP    TLS   HTTP CODE  CERT  NOTE
✅ https://example.com/                    4     12     25     14  200  60d
✅ https://www.python.org/                 3      9     21     18  200  70d
✅ http://neverssl.com/                    5     90      0     95  200     -
❌ https://expired.badssl.com/             3    160      0      0    -     -  SSLCertVerificationError: ...
❌ https://nonexistent.invalid/            0      0      0      0    -     -  gaierror: ...

The phases match the timeline in the HTTPS lifecycle guide: DNS, then TCP, then TLS, then the HTTP exchange. When one endpoint is slow, the column tells you which layer to debug, the method from Net Ch 34.

Extend it: run in a loop and export Prometheus metrics; add JSON output for a dashboard; store a week of results and flag endpoints whose p95 latency drifts, which is a first step toward the anomaly detection in Part VII.