コーディング × インフラ複合 — GKE ヘルスチェック設計

📋 コピー"""GKE向けヘルスチェックエンドポイント。

liveness と readiness を分離し、K8s Probe の仕様に準拠した
ステータスコードを返す。依存サービスは各プロトコルで確認する。
"""

import socket
from http import HTTPStatus

import pymysql
import redis
from flask import Flask, jsonify

app = Flask(__name__)

# 接続設定（実運用では環境変数から取得）
DB_HOST = "10.0.0.1"
DB_PORT = 3306
DB_USER = "health_check"
DB_PASSWORD = "secret"

CACHE_HOST = "10.0.0.2"
CACHE_PORT = 6379

CONNECT_TIMEOUT_SEC = 2


def _check_db() -> bool:
    """MySQLにTCP接続できるかを確認する。"""
    try:
        conn = pymysql.connect(
            host=DB_HOST,
            port=DB_PORT,
            user=DB_USER,
            password=DB_PASSWORD,
            connect_timeout=CONNECT_TIMEOUT_SEC,
        )
        conn.close()
        return True
    except pymysql.Error:
        return False


def _check_cache() -> bool:
    """RedisにPINGを送って応答を確認する。"""
    try:
        client = redis.Redis(
            host=CACHE_HOST,
            port=CACHE_PORT,
            socket_connect_timeout=CONNECT_TIMEOUT_SEC,
        )
        return client.ping()
    except redis.RedisError:
        return False


@app.route("/health/live")
def liveness():
    """livenessProbe 用エンドポイント。

    プロセス自体が応答できるかだけを確認する。
    依存サービスの状態は見ない（依存サービス障害でPodを再起動しても無意味なため）。
    """
    return jsonify({"status": "ok"}), HTTPStatus.OK


@app.route("/health/ready")
def readiness():
    """readinessProbe 用エンドポイント。

    依存サービス（DB・Cache）に接続できる場合のみトラフィックを受け付ける。
    接続不可の場合は 503 を返し、Podをロードバランサから切り離す。
    """
    checks = {
        "db": _check_db(),
        "cache": _check_cache(),
    }
    all_healthy = all(checks.values())
    status_code = HTTPStatus.OK if all_healthy else HTTPStatus.SERVICE_UNAVAILABLE

    return (
        jsonify(
            {
                "status": "ok" if all_healthy else "degraded",
                "checks": checks,
            }
        ),
        status_code,
    )