health check
**Health Checks for ML Services**
**Types of Health Checks**
| Check | Purpose | Kubernetes |
|-------|---------|------------|
| Liveness | Is the process alive? | livenessProbe |
| Readiness | Can it accept traffic? | readinessProbe |
| Startup | Has it fully started? | startupProbe |
**Implementation**
**Basic Health Endpoint**
```python
from fastapi import FastAPI
app = FastAPI()
@app.get("/health")
def health():
return {"status": "healthy"}
@app.get("/ready")
def ready():
# Check dependencies
if not model_loaded:
return {"status": "not ready"}, 503
if not can_connect_to_db():
return {"status": "not ready"}, 503
return {"status": "ready"}
```
**Model Health Check**
```python
@app.get("/health/model")
async def model_health():
try:
# Run inference on test input
start = time.time()
result = model.predict(test_input)
latency = time.time() - start
return {
"status": "healthy",
"model_loaded": True,
"inference_latency_ms": latency * 1000
}
except Exception as e:
return {"status": "unhealthy", "error": str(e)}, 503
```
**Kubernetes Configuration**
```yaml
apiVersion: apps/v1
kind: Deployment
spec:
template:
spec:
containers:
- name: llm-server
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
failureThreshold: 3
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 60 # Model loading time
periodSeconds: 5
failureThreshold: 2
startupProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 0
periodSeconds: 10
failureThreshold: 30 # 5 minutes to start
```
**ML-Specific Considerations**
| Check | What to Verify |
|-------|----------------|
| Model loaded | Model weights in memory |
| GPU available | CUDA device accessible |
| Warm-up complete | First inference done |
| Dependencies | Vector DB, Redis connected |
**Deep Health Check**
```python
@app.get("/health/deep")
async def deep_health():
checks = {
"model": await check_model_health(),
"gpu": await check_gpu_health(),
"vector_db": await check_vector_db(),
"cache": await check_redis(),
}
all_healthy = all(c["healthy"] for c in checks.values())
status_code = 200 if all_healthy else 503
return checks, status_code
```
**Best Practices**
- Keep liveness probes simple and fast
- Use readiness to control traffic
- Set appropriate timeouts
- Log health check failures
- Monitor health endpoints