Janady13/missionmatch / internal/platform/obs/obs.go

1// Package obs is the dependency free observability layer: structured JSON logging, an access log plus 2// metrics middleware, and a minimal Prometheus text /metrics endpoint. It deliberately pulls no metrics 3// library so the FIPS binary stays lean; a richer stack can be added at deploy time behind the same 4// endpoint. It records the chi request id on every line so a log entry and a metric ties back to one 5// request, and it NEVER logs a query string, a header, or a body, so no secret or PII reaches the logs. 6package obs 7 8import ( 9 "fmt" 10 "log/slog" 11 "net/http" 12 "os" 13 "sort" 14 "strconv" 15 "sync" 16 "sync/atomic" 17 "time" 18 19 "github.com/go-chi/chi/v5" 20 "github.com/go-chi/chi/v5/middleware" 21) 22 23// Setup installs a JSON slog logger as the process default at the given level (info unless debug is 24// set). Returns the logger so the caller can also use it directly. 25func Setup(debug bool) *slog.Logger { 26 level := slog.LevelInfo 27 if debug { 28 level = slog.LevelDebug 29 } 30 h := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: level}) 31 l := slog.New(h) 32 slog.SetDefault(l) 33 return l 34} 35 36// histogram buckets in seconds, the conventional Prometheus default ladder. 37var durBuckets = []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10} 38 39// Registry holds the in process HTTP metrics. All counters are atomic so the hot path is lock free. 40type Registry struct { 41 total [6]atomic.Int64 // index 0 = other, 1..5 = 1xx..5xx 42 bucketHits []atomic.Int64 // cumulative le buckets, plus one overflow slot 43 durSum atomic.Uint64 // sum of durations in microseconds (integer, avoids float races) 44 durCount atomic.Int64 45 inFlight atomic.Int64 46 // poolStats, when set, contributes gauge lines (for example pgxpool acquired/idle/total conns). 47 poolStats func() map[string]float64 48 poolMu sync.RWMutex 49} 50 51// NewRegistry builds a metrics registry. 52func NewRegistry() *Registry { 53 return &Registry{bucketHits: make([]atomic.Int64, len(durBuckets)+1)} 54} 55 56// SetPoolStats registers a callback that returns named gauge values (for example database pool stats). 57func (r *Registry) SetPoolStats(f func() map[string]float64) { 58 r.poolMu.Lock() 59 r.poolStats = f 60 r.poolMu.Unlock() 61} 62 63func (r *Registry) observe(status int, d time.Duration) { 64 class := status / 100 65 if class < 1 || class > 5 { 66 class = 0 67 } 68 r.total[class].Add(1) 69 secs := d.Seconds() 70 for i, b := range durBuckets { 71 if secs <= b { 72 r.bucketHits[i].Add(1) 73 } 74 } 75 r.bucketHits[len(durBuckets)].Add(1) // +Inf 76 r.durSum.Add(uint64(d.Microseconds())) 77 r.durCount.Add(1) 78} 79 80// statusRecorder captures the response status and byte count without buffering the body. 81type statusRecorder struct { 82 http.ResponseWriter 83 status int 84 bytes int 85} 86 87func (s *statusRecorder) WriteHeader(code int) { 88 s.status = code 89 s.ResponseWriter.WriteHeader(code) 90} 91 92func (s *statusRecorder) Write(b []byte) (int, error) { 93 if s.status == 0 { 94 s.status = http.StatusOK 95 } 96 n, err := s.ResponseWriter.Write(b) 97 s.bytes += n 98 return n, err 99} 100 101// Middleware times each request, records the metrics, and emits one structured access log line. It logs 102// the method, the route PATTERN (never the raw path or query string, so ids and secrets stay out of the 103// logs), the status, the latency, the bytes, and the chi request id. A 5xx logs at error level. 104func (r *Registry) Middleware(next http.Handler) http.Handler { 105 return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { 106 r.inFlight.Add(1) 107 defer r.inFlight.Add(-1) 108 start := time.Now() 109 rec := &statusRecorder{ResponseWriter: w} 110 next.ServeHTTP(rec, req) 111 dur := time.Since(start) 112 if rec.status == 0 { 113 rec.status = http.StatusOK 114 } 115 r.observe(rec.status, dur) 116 117 pattern := routePattern(req) 118 attrs := []any{ 119 slog.String("request_id", middleware.GetReqID(req.Context())), 120 slog.String("method", req.Method), 121 slog.String("route", pattern), 122 slog.Int("status", rec.status), 123 slog.Int64("latency_ms", dur.Milliseconds()), 124 slog.Int("bytes", rec.bytes), 125 } 126 switch { 127 case rec.status >= 500: 128 slog.Error("request", attrs...) 129 case rec.status >= 400: 130 slog.Warn("request", attrs...) 131 default: 132 slog.Info("request", attrs...) 133 } 134 }) 135} 136 137// routePattern returns the matched chi route pattern (for example /api/candidates/{candidateID}), so the 138// access log and the metrics never carry the concrete id segment. chi populates the RouteContext during 139// matching, so by the time the wrapped handler returns the pattern is set. Falls back to a constant when 140// the route did not match, to avoid an unbounded label set on 404s. 141func routePattern(req *http.Request) string { 142 if rc := chi.RouteContext(req.Context()); rc != nil { 143 if p := rc.RoutePattern(); p != "" { 144 return p 145 } 146 } 147 return "unmatched" 148} 149 150// Handler writes the metrics in Prometheus text exposition format. 151func (r *Registry) Handler() http.Handler { 152 return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { 153 w.Header().Set("Content-Type", "text/plain; version=0.0.4") 154 bw := w 155 156 fmt.Fprintln(bw, "# HELP mm_http_requests_total Total HTTP requests by status class.") 157 fmt.Fprintln(bw, "# TYPE mm_http_requests_total counter") 158 classes := map[int]string{1: "1xx", 2: "2xx", 3: "3xx", 4: "4xx", 5: "5xx", 0: "other"} 159 for class, label := range classes { 160 fmt.Fprintf(bw, "mm_http_requests_total{class=%q} %d\n", label, r.total[class].Load()) 161 } 162 163 fmt.Fprintln(bw, "# HELP mm_http_request_duration_seconds HTTP request latency.") 164 fmt.Fprintln(bw, "# TYPE mm_http_request_duration_seconds histogram") 165 for i, b := range durBuckets { 166 fmt.Fprintf(bw, "mm_http_request_duration_seconds_bucket{le=%q} %d\n", strconv.FormatFloat(b, 'g', -1, 64), r.bucketHits[i].Load()) 167 } 168 fmt.Fprintf(bw, "mm_http_request_duration_seconds_bucket{le=\"+Inf\"} %d\n", r.bucketHits[len(durBuckets)].Load()) 169 fmt.Fprintf(bw, "mm_http_request_duration_seconds_sum %s\n", strconv.FormatFloat(float64(r.durSum.Load())/1e6, 'g', -1, 64)) 170 fmt.Fprintf(bw, "mm_http_request_duration_seconds_count %d\n", r.durCount.Load()) 171 172 fmt.Fprintln(bw, "# HELP mm_http_in_flight_requests In flight HTTP requests.") 173 fmt.Fprintln(bw, "# TYPE mm_http_in_flight_requests gauge") 174 fmt.Fprintf(bw, "mm_http_in_flight_requests %d\n", r.inFlight.Load()) 175 176 r.poolMu.RLock() 177 f := r.poolStats 178 r.poolMu.RUnlock() 179 if f != nil { 180 stats := f() 181 names := make([]string, 0, len(stats)) 182 for k := range stats { 183 names = append(names, k) 184 } 185 sort.Strings(names) 186 fmt.Fprintln(bw, "# HELP mm_db_pool Database pool gauges.") 187 fmt.Fprintln(bw, "# TYPE mm_db_pool gauge") 188 for _, name := range names { 189 fmt.Fprintf(bw, "mm_db_pool{stat=%q} %s\n", name, strconv.FormatFloat(stats[name], 'g', -1, 64)) 190 } 191 } 192 }) 193} 194