feat(T1.8): implement health checks for Kubernetes deployment
- Add internal/health package with health checker - Implement three endpoints: /health, /health/live, /health/ready - /health returns full JSON report with component status, latency, timestamp - /health/live for K8s liveness probe (service running) - /health/ready for K8s readiness probe (ready to accept traffic) - Temporal connectivity check via GetWorkflow call with timeout - Health check caching (30s interval) to prevent excessive checks - Graceful shutdown: health server stops on SIGINT/SIGTERM - Add --health flag to starter command to run health check - Worker runs health server on port 8081 alongside task queue worker - 10/10 unit tests passing - All verification criteria met Closes T1.8
This commit is contained in:
Test
@@ -0,0 +1,84 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// Handler provides HTTP endpoints for health checks
|
||||
type Handler struct {
|
||||
checker *Checker
|
||||
}
|
||||
|
||||
// NewHandler creates a new HTTP handler for health checks
|
||||
func NewHandler(checker *Checker) *Handler {
|
||||
return &Handler{
|
||||
checker: checker,
|
||||
}
|
||||
}
|
||||
|
||||
// RegisterRoutes registers health check routes on a mux
|
||||
func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("/health", h.handleHealth)
|
||||
mux.HandleFunc("/health/live", h.handleLive)
|
||||
mux.HandleFunc("/health/ready", h.handleReady)
|
||||
}
|
||||
|
||||
// handleHealth returns full health report
|
||||
func (h *Handler) handleHealth(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
report := h.checker.Check(r.Context())
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
|
||||
// Return 200 if healthy, 503 if unhealthy
|
||||
if report.Status != StatusHealthy {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
}
|
||||
|
||||
json.NewEncoder(w).Encode(report)
|
||||
}
|
||||
|
||||
// handleLive is Kubernetes liveness probe endpoint
|
||||
// Returns 200 if the service is running, 503 otherwise
|
||||
func (h *Handler) handleLive(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
if h.checker.temporalClient == nil {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
w.Write([]byte("service not initialized"))
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte("alive"))
|
||||
}
|
||||
|
||||
// handleReady is Kubernetes readiness probe endpoint
|
||||
// Returns 200 if the service is ready to accept traffic, 503 otherwise
|
||||
func (h *Handler) handleReady(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
report := h.checker.Check(r.Context())
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
|
||||
// Service is ready only if healthy
|
||||
if report.Status != StatusHealthy {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
}
|
||||
|
||||
json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"ready": report.Status == StatusHealthy,
|
||||
"components": report.Components,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,167 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.temporal.io/sdk/client"
|
||||
)
|
||||
|
||||
// Status represents the health status of a component
|
||||
type Status string
|
||||
|
||||
const (
|
||||
StatusHealthy Status = "healthy"
|
||||
StatusUnhealthy Status = "unhealthy"
|
||||
StatusUnknown Status = "unknown"
|
||||
)
|
||||
|
||||
// ComponentHealth represents the health of a single system component
|
||||
type ComponentHealth struct {
|
||||
Name string `json:"name"`
|
||||
Status Status `json:"status"`
|
||||
Latency int64 `json:"latency_ms"`
|
||||
LastCheck time.Time `json:"last_check"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// HealthReport is the overall health status of the system
|
||||
type HealthReport struct {
|
||||
Status Status `json:"status"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Components map[string]ComponentHealth `json:"components"`
|
||||
Latency int64 `json:"latency_ms"`
|
||||
}
|
||||
|
||||
// Checker provides health check functionality
|
||||
type Checker struct {
|
||||
temporalClient client.Client
|
||||
mu sync.RWMutex
|
||||
lastReport *HealthReport
|
||||
lastCheckTime time.Time
|
||||
checkInterval time.Duration
|
||||
}
|
||||
|
||||
// NewChecker creates a new health checker
|
||||
func NewChecker(temporalClient client.Client) *Checker {
|
||||
return &Checker{
|
||||
temporalClient: temporalClient,
|
||||
checkInterval: 30 * time.Second,
|
||||
}
|
||||
}
|
||||
|
||||
// Check performs a comprehensive health check
|
||||
func (h *Checker) Check(ctx context.Context) *HealthReport {
|
||||
startTime := time.Now()
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
// Skip check if recently done
|
||||
if time.Since(h.lastCheckTime) < h.checkInterval && h.lastReport != nil {
|
||||
return h.lastReport
|
||||
}
|
||||
|
||||
components := make(map[string]ComponentHealth)
|
||||
|
||||
// Check Temporal connectivity
|
||||
temporalHealth := h.checkTemporal(ctx)
|
||||
components["temporal"] = temporalHealth
|
||||
|
||||
// Determine overall status
|
||||
overallStatus := StatusHealthy
|
||||
for _, comp := range components {
|
||||
if comp.Status == StatusUnhealthy {
|
||||
overallStatus = StatusUnhealthy
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
latency := time.Since(startTime).Milliseconds()
|
||||
report := &HealthReport{
|
||||
Status: overallStatus,
|
||||
Timestamp: time.Now(),
|
||||
Components: components,
|
||||
Latency: latency,
|
||||
}
|
||||
|
||||
h.lastReport = report
|
||||
h.lastCheckTime = time.Now()
|
||||
|
||||
return report
|
||||
}
|
||||
|
||||
// checkTemporal verifies Temporal cluster connectivity
|
||||
func (h *Checker) checkTemporal(ctx context.Context) ComponentHealth {
|
||||
startTime := time.Now()
|
||||
comp := ComponentHealth{
|
||||
Name: "temporal",
|
||||
Status: StatusHealthy,
|
||||
LastCheck: time.Now(),
|
||||
}
|
||||
|
||||
if h.temporalClient == nil {
|
||||
comp.Status = StatusUnhealthy
|
||||
comp.Error = "Temporal client not initialized"
|
||||
comp.Latency = time.Since(startTime).Milliseconds()
|
||||
return comp
|
||||
}
|
||||
|
||||
// Create a short timeout context for the health check
|
||||
checkCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Try to get a workflow to verify connectivity
|
||||
// Use a dummy workflow ID that likely doesn't exist - we're just testing connectivity
|
||||
// If Temporal is unreachable, this will fail; if it's reachable, it will return NotFound error (which is fine)
|
||||
wfRun := h.temporalClient.GetWorkflow(checkCtx, "health-check-dummy-id-"+time.Now().Format("20060102150405"), "")
|
||||
comp.Latency = time.Since(startTime).Milliseconds()
|
||||
|
||||
// We're only checking connectivity, so we try to peek at the result
|
||||
// This will fail if Temporal is unreachable, but return nil error if it just doesn't exist
|
||||
var result interface{}
|
||||
err := wfRun.Get(checkCtx, &result)
|
||||
|
||||
if err != nil {
|
||||
errMsg := err.Error()
|
||||
// NotFound errors mean Temporal responded but workflow doesn't exist - this is healthy
|
||||
if !contains(errMsg, "not found") && !contains(errMsg, "NotFound") {
|
||||
comp.Status = StatusUnhealthy
|
||||
comp.Error = err.Error()
|
||||
}
|
||||
}
|
||||
|
||||
return comp
|
||||
}
|
||||
|
||||
// contains checks if a string contains a substring (case-insensitive)
|
||||
func contains(s, substr string) bool {
|
||||
for i := 0; i <= len(s)-len(substr); i++ {
|
||||
if s[i:i+len(substr)] == substr {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// IsHealthy returns true if the system is healthy
|
||||
func (h *Checker) IsHealthy(ctx context.Context) bool {
|
||||
report := h.Check(ctx)
|
||||
return report.Status == StatusHealthy
|
||||
}
|
||||
|
||||
// GetReport returns the last health report
|
||||
func (h *Checker) GetReport(ctx context.Context) *HealthReport {
|
||||
return h.Check(ctx)
|
||||
}
|
||||
|
||||
// ToJSON converts the health report to JSON
|
||||
func (report *HealthReport) ToJSON() ([]byte, error) {
|
||||
return json.MarshalIndent(report, "", " ")
|
||||
}
|
||||
|
||||
// String returns the health status as a string
|
||||
func (s Status) String() string {
|
||||
return string(s)
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// TestNewChecker tests the creation of a new health checker
|
||||
func TestNewChecker(t *testing.T) {
|
||||
checker := NewChecker(nil)
|
||||
assert.NotNil(t, checker)
|
||||
assert.Equal(t, 30*time.Second, checker.checkInterval)
|
||||
}
|
||||
|
||||
// TestCheckWithNilClient tests health check with nil client
|
||||
func TestCheckWithNilClient(t *testing.T) {
|
||||
checker := NewChecker(nil)
|
||||
report := checker.Check(context.Background())
|
||||
|
||||
assert.NotNil(t, report)
|
||||
assert.Equal(t, StatusUnhealthy, report.Status)
|
||||
assert.Len(t, report.Components, 1)
|
||||
assert.Equal(t, StatusUnhealthy, report.Components["temporal"].Status)
|
||||
assert.Equal(t, "Temporal client not initialized", report.Components["temporal"].Error)
|
||||
}
|
||||
|
||||
// TestIsHealthy tests the IsHealthy method
|
||||
func TestIsHealthy(t *testing.T) {
|
||||
checker := NewChecker(nil)
|
||||
assert.False(t, checker.IsHealthy(context.Background()))
|
||||
}
|
||||
|
||||
// TestGetReport tests the GetReport method
|
||||
func TestGetReport(t *testing.T) {
|
||||
checker := NewChecker(nil)
|
||||
report := checker.GetReport(context.Background())
|
||||
assert.NotNil(t, report)
|
||||
assert.Equal(t, StatusUnhealthy, report.Status)
|
||||
}
|
||||
|
||||
// TestHealthReportJSON tests JSON serialization
|
||||
func TestHealthReportJSON(t *testing.T) {
|
||||
checker := NewChecker(nil)
|
||||
report := checker.Check(context.Background())
|
||||
|
||||
jsonData, err := report.ToJSON()
|
||||
assert.NoError(t, err)
|
||||
assert.NotEmpty(t, jsonData)
|
||||
assert.Contains(t, string(jsonData), "unhealthy")
|
||||
assert.Contains(t, string(jsonData), "temporal")
|
||||
}
|
||||
|
||||
// TestHealthReportTimestamp tests that timestamp is set
|
||||
func TestHealthReportTimestamp(t *testing.T) {
|
||||
checker := NewChecker(nil)
|
||||
before := time.Now()
|
||||
report := checker.Check(context.Background())
|
||||
after := time.Now()
|
||||
|
||||
assert.True(t, report.Timestamp.After(before) || report.Timestamp.Equal(before))
|
||||
assert.True(t, report.Timestamp.Before(after) || report.Timestamp.Equal(after))
|
||||
}
|
||||
|
||||
// TestStatusString tests Status string representation
|
||||
func TestStatusString(t *testing.T) {
|
||||
assert.Equal(t, "healthy", StatusHealthy.String())
|
||||
assert.Equal(t, "unhealthy", StatusUnhealthy.String())
|
||||
assert.Equal(t, "unknown", StatusUnknown.String())
|
||||
}
|
||||
|
||||
// TestComponentHealthLatency tests that latency is recorded
|
||||
func TestComponentHealthLatency(t *testing.T) {
|
||||
checker := NewChecker(nil)
|
||||
report := checker.Check(context.Background())
|
||||
|
||||
assert.NotNil(t, report.Components["temporal"])
|
||||
assert.GreaterOrEqual(t, report.Components["temporal"].Latency, int64(0))
|
||||
}
|
||||
|
||||
// TestHealthCheckCaching tests that recent checks are cached
|
||||
func TestHealthCheckCaching(t *testing.T) {
|
||||
checker := NewChecker(nil)
|
||||
|
||||
// First check
|
||||
report1 := checker.Check(context.Background())
|
||||
time1 := report1.Timestamp
|
||||
|
||||
// Second check immediately (should be cached)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
report2 := checker.Check(context.Background())
|
||||
time2 := report2.Timestamp
|
||||
|
||||
// Timestamps should be the same or very close (cached)
|
||||
assert.Equal(t, time1, time2, "second check should use cached result")
|
||||
}
|
||||
|
||||
// TestComponentHealthDefaults tests default component health values
|
||||
func TestComponentHealthDefaults(t *testing.T) {
|
||||
comp := ComponentHealth{
|
||||
Name: "test",
|
||||
Status: StatusHealthy,
|
||||
}
|
||||
|
||||
assert.Equal(t, "test", comp.Name)
|
||||
assert.Equal(t, StatusHealthy, comp.Status)
|
||||
assert.Empty(t, comp.Error)
|
||||
}
|
||||
Reference in New Issue
Block a user