feat: add demo-app api service
실험·앱 계층 설계서 3.1의 demo-app api(컨테이너 계층). /healthz, /readyz(DB ping +
legacy /healthz), /api/orders GET/POST(app.orders), /api/price/{item}(legacy-api via ILB,
LEGACY_TIMEOUT 기본 2s), /metrics(demo_http_*). OTLP 트레이스는 OTEL_EXPORTER_OTLP_ENDPOINT
설정 시 전송(otelhttp). 이미지는 공개 Artifact Registry selfheal-public(노드 익명 pull).
This commit is contained in:
seon committed
1 parent
7186ef0c49
commit
8f5378b6ea
5 files changed
+397
-1
No files matched your search
@@ -0,0 +1,242 @@
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user