selfheal-lab-apps/demo-app/main.go

// demo-app api: the "container layer" service of the self-healing lab (experiment design 3.1).
// GET  /healthz            process liveness
// GET  /readyz             DB ping + legacy-api /healthz (readiness)
// GET  /api/orders         recent orders (DB read)
// POST /api/orders         {"item":"apple","qty":2} (DB write)
// GET  /api/price/{item}   proxied from legacy-api via the ILB (LEGACY_URL, LEGACY_TIMEOUT)
// GET  /metrics            Prometheus exposition (demo_http_*)
// Traces go to OTEL_EXPORTER_OTLP_ENDPOINT (Alloy) when set.
package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"io"
	"log"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"syscall"
	"time"

	_ "github.com/jackc/pgx/v5/stdlib"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
	"go.opentelemetry.io/otel/sdk/resource"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
)

var (
	reqTotal = prometheus.NewCounterVec(prometheus.CounterOpts{Name: "demo_http_requests_total", Help: "HTTP requests by route and status"}, []string{"route", "status"})
	reqDur   = prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: "demo_http_request_duration_seconds", Help: "HTTP request latency",
		Buckets: []float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5}}, []string{"route"})
	legacyDur = prometheus.NewHistogram(prometheus.HistogramOpts{Name: "demo_legacy_request_duration_seconds", Help: "Upstream legacy-api latency",
		Buckets: []float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5}})
)

type statusWriter struct {
	http.ResponseWriter
	code int
}

func (w *statusWriter) WriteHeader(c int) { w.code = c; w.ResponseWriter.WriteHeader(c) }

func instrument(route string, h http.HandlerFunc) http.Handler {
	return otelhttp.NewHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		sw := &statusWriter{ResponseWriter: w, code: 200}
		start := time.Now()
		h(sw, r)
		reqDur.WithLabelValues(route).Observe(time.Since(start).Seconds())
		reqTotal.WithLabelValues(route, strconv.Itoa(sw.code)).Inc()
	}), route)
}

func writeJSON(w http.ResponseWriter, code int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	_ = json.NewEncoder(w).Encode(v)
}

func env(k, d string) string {
	if v := os.Getenv(k); v != "" {
		return v
	}
	return d
}

func setupTracing(ctx context.Context) func(context.Context) error {
	if os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") == "" {
		return func(context.Context) error { return nil }
	}
	exp, err := otlptracegrpc.New(ctx, otlptracegrpc.WithInsecure())
	if err != nil {
		log.Printf("otlp exporter: %v (tracing disabled)", err)
		return func(context.Context) error { return nil }
	}
	res, _ := resource.Merge(resource.Default(), resource.NewWithAttributes(semconv.SchemaURL, semconv.ServiceName("demo-api")))
	tp := sdktrace.NewTracerProvider(sdktrace.WithBatcher(exp), sdktrace.WithResource(res))
	otel.SetTracerProvider(tp)
	return tp.Shutdown
}

func main() {
	prometheus.MustRegister(reqTotal, reqDur, legacyDur)
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()
	shutdownTracing := setupTracing(ctx)
	defer func() { _ = shutdownTracing(context.Background()) }()

	dsn := os.Getenv("DB_DSN")
	var db *sql.DB
	if dsn != "" {
		var err error
		if db, err = sql.Open("pgx", dsn); err != nil {
			log.Printf("db open: %v", err)
			db = nil
		} else {
			db.SetMaxOpenConns(10)
			db.SetConnMaxLifetime(5 * time.Minute)
		}
	} else {
		log.Printf("DB_DSN not set")
	}
	legacyURL := env("LEGACY_URL", "http://10.20.0.100:8080")
	legacyTimeout, err := time.ParseDuration(env("LEGACY_TIMEOUT", "2s"))
	if err != nil {
		legacyTimeout = 2 * time.Second
	}
	legacy := &http.Client{Timeout: legacyTimeout, Transport: otelhttp.NewTransport(http.DefaultTransport)}

	mux := http.NewServeMux()
	mux.Handle("/healthz", instrument("healthz", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/plain")
		_, _ = w.Write([]byte("ok\n"))
	}))
	mux.Handle("/readyz", instrument("readyz", func(w http.ResponseWriter, r *http.Request) {
		ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
		defer cancel()
		problems := map[string]string{}
		if db == nil {
			problems["db"] = "not configured"
		} else if err := db.PingContext(ctx); err != nil {
			problems["db"] = err.Error()
		}
		req, _ := http.NewRequestWithContext(ctx, http.MethodGet, legacyURL+"/healthz", nil)
		if resp, err := legacy.Do(req); err != nil {
			problems["legacy"] = err.Error()
		} else {
			_, _ = io.Copy(io.Discard, resp.Body)
			resp.Body.Close()
			if resp.StatusCode != 200 {
				problems["legacy"] = "status " + strconv.Itoa(resp.StatusCode)
			}
		}
		if len(problems) > 0 {
			writeJSON(w, 503, map[string]any{"ready": false, "problems": problems})
			return
		}
		writeJSON(w, 200, map[string]any{"ready": true})
	}))
	mux.Handle("/api/orders", instrument("orders", func(w http.ResponseWriter, r *http.Request) {
		if db == nil {
			writeJSON(w, 503, map[string]string{"error": "db unavailable"})
			return
		}
		ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
		defer cancel()
		switch r.Method {
		case http.MethodGet:
			rows, err := db.QueryContext(ctx, "SELECT id, item, qty, status, created_at FROM app.orders ORDER BY id DESC LIMIT 20")
			if err != nil {
				log.Printf("orders query: %v", err)
				writeJSON(w, 503, map[string]string{"error": "db unavailable"})
				return
			}
			defer rows.Close()
			type order struct {
				ID        int64     `json:"id"`
				Item      string    `json:"item"`
				Qty       int       `json:"qty"`
				Status    string    `json:"status"`
				CreatedAt time.Time `json:"created_at"`
			}
			out := []order{}
			for rows.Next() {
				var o order
				if err := rows.Scan(&o.ID, &o.Item, &o.Qty, &o.Status, &o.CreatedAt); err != nil {
					writeJSON(w, 500, map[string]string{"error": err.Error()})
					return
				}
				out = append(out, o)
			}
			writeJSON(w, 200, map[string]any{"orders": out, "count": len(out)})
		case http.MethodPost:
			var in struct {
				Item string `json:"item"`
				Qty  int    `json:"qty"`
			}
			if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&in); err != nil || in.Item == "" || in.Qty <= 0 {
				writeJSON(w, 400, map[string]string{"error": "item and positive qty required"})
				return
			}
			var id int64
			err := db.QueryRowContext(ctx, "INSERT INTO app.orders (item, qty) VALUES ($1, $2) RETURNING id", in.Item, in.Qty).Scan(&id)
			if err != nil {
				if strings.Contains(err.Error(), "violates foreign key") {
					writeJSON(w, 404, map[string]string{"error": "unknown item", "item": in.Item})
					return
				}
				log.Printf("orders insert: %v", err)
				writeJSON(w, 503, map[string]string{"error": "db unavailable"})
				return
			}
			writeJSON(w, 201, map[string]any{"id": id, "item": in.Item, "qty": in.Qty, "status": "new"})
		default:
			writeJSON(w, 405, map[string]string{"error": "method not allowed"})
		}
	}))
	mux.Handle("/api/price/", instrument("price", func(w http.ResponseWriter, r *http.Request) {
		item := strings.TrimPrefix(r.URL.Path, "/api/price/")
		if item == "" || strings.Contains(item, "/") {
			writeJSON(w, 404, map[string]string{"error": "item required"})
			return
		}
		req, _ := http.NewRequestWithContext(r.Context(), http.MethodGet, legacyURL+"/v1/price/"+item, nil)
		start := time.Now()
		resp, err := legacy.Do(req)
		legacyDur.Observe(time.Since(start).Seconds())
		if err != nil {
			code := 502
			if errors.Is(err, context.DeadlineExceeded) || os.IsTimeout(err) {
				code = 504
			}
			writeJSON(w, code, map[string]string{"error": "legacy-api unreachable", "detail": err.Error()})
			return
		}
		defer resp.Body.Close()
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(resp.StatusCode)
		_, _ = io.Copy(w, resp.Body)
	}))
	mux.Handle("/metrics", promhttp.Handler())

	srv := &http.Server{Addr: ":" + env("PORT", "8080"), Handler: mux, ReadHeaderTimeout: 5 * time.Second}
	go func() {
		<-ctx.Done()
		sctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_ = srv.Shutdown(sctx)
	}()
	log.Printf("demo-api listening on %s (legacy=%s timeout=%s)", srv.Addr, legacyURL, legacyTimeout)
	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		log.Fatal(err)
	}
}