Health Aggregator Service

mkdir health-aggregator
cd health-aggregator
go mod init github.com/yourorg/health-aggregator
go get github.com/redis/go-redis/v9@latest

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/redis/go-redis/v9"
)

// --- Health Check Types ---

type Status string

const (
	StatusUp       Status = "up"
	StatusDown     Status = "down"
	StatusDegraded Status = "degraded"
)

type CheckResult struct {
	Name       string `json:"name"`
	Status     Status `json:"status"`
	DurationMs int64  `json:"duration_ms"`
	Error      string `json:"error,omitempty"`
}

type HealthResponse struct {
	Status  Status        `json:"status"`
	Checks  []CheckResult `json:"checks,omitempty"`
	Version string        `json:"version,omitempty"`
}

type Check struct {
	Name    string
	Fn      func() error
	Timeout time.Duration
}

type Checker struct {
	checks []Check
	mu     sync.RWMutex
}

// NewChecker creates an empty Checker. Register checks with AddCheck before calling RunChecks.
func NewChecker() *Checker {
	return &Checker{}
}

// AddCheck registers a named health check function with a per-check timeout.
func (c *Checker) AddCheck(name string, fn func() error, timeout time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.checks = append(c.checks, Check{Name: name, Fn: fn, Timeout: timeout})
}

// RunChecks executes all registered checks concurrently and returns an aggregated response.
// Overall status is "down" if any check fails, "up" if all pass.
func (c *Checker) RunChecks() HealthResponse {
	c.mu.RLock()
	checks := make([]Check, len(c.checks))
	copy(checks, c.checks)
	c.mu.RUnlock()

	results := make([]CheckResult, len(checks))
	var wg sync.WaitGroup

	for i, check := range checks {
		wg.Add(1)
		go func(idx int, chk Check) {
			defer wg.Done()
			start := time.Now()

			errCh := make(chan error, 1)
			go func() { errCh <- chk.Fn() }()

			select {
			case err := <-errCh:
				result := CheckResult{
					Name:       chk.Name,
					DurationMs: time.Since(start).Milliseconds(),
					Status:     StatusUp,
				}
				if err != nil {
					result.Status = StatusDown
					result.Error = err.Error()
				}
				results[idx] = result
			case <-time.After(chk.Timeout):
				results[idx] = CheckResult{
					Name:       chk.Name,
					Status:     StatusDown,
					DurationMs: chk.Timeout.Milliseconds(),
					Error:      "timeout",
				}
			}
		}(i, check)
	}

	wg.Wait()

	overall := StatusUp
	for _, r := range results {
		if r.Status == StatusDown {
			overall = StatusDown
			break
		}
	}

	return HealthResponse{Status: overall, Checks: results}
}

// RunChecksWithDegradation runs all checks and distinguishes critical from non-critical failures.
// A failed critical check returns "down"; a failed non-critical check returns "degraded".
func (c *Checker) RunChecksWithDegradation(critical []string) HealthResponse {
	resp := c.RunChecks()

	criticalSet := make(map[string]bool)
	for _, name := range critical {
		criticalSet[name] = true
	}

	overall := StatusUp
	for _, check := range resp.Checks {
		if check.Status == StatusDown {
			if criticalSet[check.Name] {
				overall = StatusDown
				break
			}
			overall = StatusDegraded
		}
	}
	resp.Status = overall
	return resp
}

// --- Dependency Checks ---

// DatabaseCheck returns a check function that pings the database with a 2s timeout.
func DatabaseCheck(db *sql.DB) func() error {
	return func() error {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()
		return db.PingContext(ctx)
	}
}

// RedisCheck returns a check function that pings the Redis client with a 2s timeout.
func RedisCheck(client *redis.Client) func() error {
	return func() error {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()
		return client.Ping(ctx).Err()
	}
}

// HTTPCheck returns a check function that GETs url and expects expectedStatus in response.
func HTTPCheck(url string, expectedStatus int) func() error {
	client := &http.Client{Timeout: 5 * time.Second}
	return func() error {
		resp, err := client.Get(url)
		if err != nil {
			return err
		}
		defer resp.Body.Close()
		if resp.StatusCode != expectedStatus {
			return fmt.Errorf("expected %d, got %d", expectedStatus, resp.StatusCode)
		}
		return nil
	}
}

// --- Circuit Breaker ---

type State int

const (
	Closed   State = iota // Normal operation: requests pass through
	Open                  // Failing: requests blocked
	HalfOpen              // Testing: one request allowed through
)

var ErrCircuitOpen = errors.New("circuit breaker is open")

type CircuitBreaker struct {
	mu           sync.Mutex
	state        State
	failures     int
	threshold    int
	lastFailure  time.Time
	resetTimeout time.Duration
}

// NewCircuitBreaker creates a circuit breaker that opens after threshold consecutive
// failures and attempts recovery after resetTimeout.
func NewCircuitBreaker(threshold int, resetTimeout time.Duration) *CircuitBreaker {
	return &CircuitBreaker{
		state:        Closed,
		threshold:    threshold,
		resetTimeout: resetTimeout,
	}
}

// Execute runs fn if the circuit is closed or half-open. Returns ErrCircuitOpen
// immediately if the circuit is open and the reset timeout has not elapsed.
func (cb *CircuitBreaker) Execute(fn func() error) error {
	cb.mu.Lock()

	switch cb.state {
	case Open:
		if time.Since(cb.lastFailure) > cb.resetTimeout {
			cb.state = HalfOpen
			cb.mu.Unlock()
			return cb.tryExec(fn)
		}
		cb.mu.Unlock()
		return ErrCircuitOpen

	case HalfOpen:
		cb.mu.Unlock()
		return cb.tryExec(fn)

	default: // Closed
		cb.mu.Unlock()
		return cb.tryExec(fn)
	}
}

func (cb *CircuitBreaker) tryExec(fn func() error) error {
	err := fn()

	cb.mu.Lock()
	defer cb.mu.Unlock()

	if err != nil {
		cb.failures++
		cb.lastFailure = time.Now()
		if cb.failures >= cb.threshold {
			cb.state = Open
		}
		return err
	}

	// Success: reset
	cb.failures = 0
	cb.state = Closed
	return nil
}

// --- Health Aggregation ---

type ServiceHealth struct {
	Name   string `json:"name"`
	URL    string `json:"-"`
	Status Status `json:"status"`
	Error  string `json:"error,omitempty"`
}

// AggregateHealth returns "up" if all services are up, "down" if all are down,
// and "degraded" for any partial failure or if any service is degraded.
func AggregateHealth(services []ServiceHealth) Status {
	downCount := 0
	degradedCount := 0
	for _, svc := range services {
		switch svc.Status {
		case StatusDown:
			downCount++
		case StatusDegraded:
			degradedCount++
		}
	}

	switch {
	case downCount == len(services):
		return StatusDown
	case downCount > 0 || degradedCount > 0:
		return StatusDegraded
	default:
		return StatusUp
	}
}

// --- Main: Wire It All Together ---

func main() {
	checker := NewChecker()

	// Wrap database check with circuit breaker. In production, replace the stub
	// with a real *sql.DB obtained from sql.Open and passed in as a parameter.
	dbBreaker := NewCircuitBreaker(3, 30*time.Second)
	checker.AddCheck("postgres", func() error {
		return dbBreaker.Execute(func() error {
			// Production: return db.PingContext(ctx)
			return nil
		})
	}, 3*time.Second)

	mux := http.NewServeMux()

	// Liveness: is the process alive?
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte(`{"status":"up"}`))
	})

	// Readiness: can we serve traffic?
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
		resp := checker.RunChecksWithDegradation([]string{"postgres"})
		status := http.StatusOK
		if resp.Status == StatusDown {
			status = http.StatusServiceUnavailable
		}
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(status)
		json.NewEncoder(w).Encode(resp)
	})

	// Aggregation endpoint for multi-service monitoring
	mux.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) {
		services := []ServiceHealth{
			{Name: "api", Status: StatusUp},
			{Name: "worker", Status: StatusUp},
			{Name: "scheduler", Status: StatusDegraded},
		}
		overall := AggregateHealth(services)
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string]any{
			"status":   overall,
			"services": services,
		})
	})

	fmt.Println("Health service listening on :8080")
	if err := http.ListenAndServe(":8080", mux); err != nil {
		log.Fatal(err)
	}
}

go run main.go
# Health service listening on :8080

# Liveness probe
curl -s http://localhost:8080/healthz
# {"status":"up"}

# Readiness probe (postgres stub always passes)
curl -s http://localhost:8080/readyz | jq .
# {
#   "status": "up",
#   "checks": [
#     { "name": "postgres", "status": "up", "duration_ms": 0 }
#   ]
# }

# Multi-service aggregation (scheduler is degraded, so overall is degraded)
curl -s http://localhost:8080/status | jq .
# {
#   "status": "degraded",
#   "services": [
#     { "name": "api",       "status": "up" },
#     { "name": "worker",    "status": "up" },
#     { "name": "scheduler", "status": "degraded" }
#   ]
# }