more platform stuff
This commit is contained in:
1 parent
4b9a882018
commit
fc5eefbe5e
7 files changed
+756
No files matched your search
Vendored
+87
@@ -0,0 +1,87 @@
|
|||||||
|
// Package cache provides a small, concurrency-safe TTL + LRU cache. It's a
|
||||||
|
// generic building block for sharing an expensive computation across requests
|
||||||
|
// (e.g. the reader shares one relevance match between /api/stream and
|
||||||
|
// /api/facets): a hit returns the stored value, a miss recomputes. The TTL
|
||||||
|
// bounds staleness; the capacity bounds memory.
|
||||||
|
package cache
|
||||||
|
|
||||||
|
import (
|
||||||
|
"container/list"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
type entry[K comparable, V any] struct {
|
||||||
|
key K
|
||||||
|
val V
|
||||||
|
at time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// TTLLRU is a fixed-capacity cache that evicts the least-recently-used entry on
|
||||||
|
// overflow and treats entries older than ttl as misses. The zero value is not
|
||||||
|
// usable — construct with New. All methods are safe for concurrent use.
|
||||||
|
type TTLLRU[K comparable, V any] struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
ll *list.List // front = most-recently-used
|
||||||
|
m map[K]*list.Element
|
||||||
|
cap int
|
||||||
|
ttl time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
// New returns a cache holding up to capacity entries, each valid for ttl.
|
||||||
|
func New[K comparable, V any](capacity int, ttl time.Duration) *TTLLRU[K, V] {
|
||||||
|
if capacity < 1 {
|
||||||
|
capacity = 1
|
||||||
|
}
|
||||||
|
return &TTLLRU[K, V]{
|
||||||
|
ll: list.New(),
|
||||||
|
m: make(map[K]*list.Element, capacity),
|
||||||
|
cap: capacity,
|
||||||
|
ttl: ttl,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get returns the cached value for key. The boolean is false on a miss or when
|
||||||
|
// the entry has expired (in which case it is also evicted). A hit refreshes the
|
||||||
|
// entry's recency.
|
||||||
|
func (c *TTLLRU[K, V]) Get(key K) (V, bool) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
el, ok := c.m[key]
|
||||||
|
if !ok {
|
||||||
|
var zero V
|
||||||
|
return zero, false
|
||||||
|
}
|
||||||
|
ent := el.Value.(*entry[K, V])
|
||||||
|
if time.Since(ent.at) > c.ttl {
|
||||||
|
c.ll.Remove(el)
|
||||||
|
delete(c.m, key)
|
||||||
|
var zero V
|
||||||
|
return zero, false
|
||||||
|
}
|
||||||
|
c.ll.MoveToFront(el)
|
||||||
|
return ent.val, true
|
||||||
|
}
|
||||||
|
|
||||||
|
// Put stores val under key, refreshing its timestamp and recency, evicting the
|
||||||
|
// least-recently-used entry if that exceeds the capacity.
|
||||||
|
func (c *TTLLRU[K, V]) Put(key K, val V) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
if el, ok := c.m[key]; ok {
|
||||||
|
ent := el.Value.(*entry[K, V])
|
||||||
|
ent.val = val
|
||||||
|
ent.at = time.Now()
|
||||||
|
c.ll.MoveToFront(el)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
c.m[key] = c.ll.PushFront(&entry[K, V]{key: key, val: val, at: time.Now()})
|
||||||
|
for c.ll.Len() > c.cap {
|
||||||
|
back := c.ll.Back()
|
||||||
|
if back == nil {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
c.ll.Remove(back)
|
||||||
|
delete(c.m, back.Value.(*entry[K, V]).key)
|
||||||
|
}
|
||||||
|
}
|
||||||
Vendored
+47
@@ -0,0 +1,47 @@
|
|||||||
|
package cache
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestGetPut(t *testing.T) {
|
||||||
|
c := New[string, int](2, time.Minute)
|
||||||
|
if _, ok := c.Get("a"); ok {
|
||||||
|
t.Fatal("empty cache returned a hit")
|
||||||
|
}
|
||||||
|
c.Put("a", 1)
|
||||||
|
got, ok := c.Get("a")
|
||||||
|
if !ok || got != 1 {
|
||||||
|
t.Fatalf("get after put: ok=%v val=%v", ok, got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTTL(t *testing.T) {
|
||||||
|
c := New[string, int](8, 10*time.Millisecond)
|
||||||
|
c.Put("a", 1)
|
||||||
|
if _, ok := c.Get("a"); !ok {
|
||||||
|
t.Fatal("fresh entry missing")
|
||||||
|
}
|
||||||
|
time.Sleep(20 * time.Millisecond)
|
||||||
|
if _, ok := c.Get("a"); ok {
|
||||||
|
t.Fatal("expired entry still served")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLRUEviction(t *testing.T) {
|
||||||
|
c := New[string, int](2, time.Minute)
|
||||||
|
c.Put("a", 1)
|
||||||
|
c.Put("b", 2)
|
||||||
|
_, _ = c.Get("a") // make "a" most-recently-used so "b" is the eviction victim
|
||||||
|
c.Put("c", 3) // exceeds cap → evict LRU ("b")
|
||||||
|
if _, ok := c.Get("b"); ok {
|
||||||
|
t.Fatal("LRU victim 'b' not evicted")
|
||||||
|
}
|
||||||
|
if _, ok := c.Get("a"); !ok {
|
||||||
|
t.Fatal("recently-used 'a' was evicted")
|
||||||
|
}
|
||||||
|
if _, ok := c.Get("c"); !ok {
|
||||||
|
t.Fatal("freshly-put 'c' missing")
|
||||||
|
}
|
||||||
|
}
|
||||||
+140
@@ -0,0 +1,140 @@
|
|||||||
|
package conv
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Uint32 coerces an interface value to a uint32.
|
||||||
|
// It returns the uint32 value and true on success, or 0 and false if coercion fails.
|
||||||
|
func Uint32(v any) (uint32, bool) {
|
||||||
|
if v == nil {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
switch n := v.(type) {
|
||||||
|
case uint32:
|
||||||
|
return n, true
|
||||||
|
case float64:
|
||||||
|
if n < 0 || n > 4294967295 {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return uint32(n), true
|
||||||
|
case float32:
|
||||||
|
if n < 0 || n > 4294967295 {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return uint32(n), true
|
||||||
|
case int:
|
||||||
|
if n < 0 || n > 4294967295 {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return uint32(n), true
|
||||||
|
case int64:
|
||||||
|
if n < 0 || n > 4294967295 {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return uint32(n), true
|
||||||
|
case uint64:
|
||||||
|
if n > 4294967295 {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return uint32(n), true
|
||||||
|
case json.Number:
|
||||||
|
u, err := strconv.ParseUint(n.String(), 10, 32)
|
||||||
|
if err != nil {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return uint32(u), true
|
||||||
|
case string:
|
||||||
|
s := strings.TrimSpace(n)
|
||||||
|
if s == "" {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
u, err := strconv.ParseUint(s, 10, 32)
|
||||||
|
if err != nil {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return uint32(u), true
|
||||||
|
}
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
|
||||||
|
// String coerces an interface value to a string.
|
||||||
|
// It handles common numeric types and falls back to fmt.Sprintf.
|
||||||
|
func String(v any) string {
|
||||||
|
if v == nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
switch n := v.(type) {
|
||||||
|
case string:
|
||||||
|
return n
|
||||||
|
case int:
|
||||||
|
return strconv.Itoa(n)
|
||||||
|
case int64:
|
||||||
|
return strconv.FormatInt(n, 10)
|
||||||
|
case uint32:
|
||||||
|
return strconv.FormatUint(uint64(n), 10)
|
||||||
|
case uint64:
|
||||||
|
return strconv.FormatUint(n, 10)
|
||||||
|
case float64:
|
||||||
|
// If it's a whole number, represent it as an integer to avoid scientific notation
|
||||||
|
if n == float64(int64(n)) {
|
||||||
|
return strconv.FormatInt(int64(n), 10)
|
||||||
|
}
|
||||||
|
return strconv.FormatFloat(n, 'f', -1, 64)
|
||||||
|
case float32:
|
||||||
|
if n == float32(int32(n)) {
|
||||||
|
return strconv.Itoa(int(n))
|
||||||
|
}
|
||||||
|
return strconv.FormatFloat(float64(n), 'f', -1, 32)
|
||||||
|
case json.Number:
|
||||||
|
return n.String()
|
||||||
|
case bool:
|
||||||
|
if n {
|
||||||
|
return "true"
|
||||||
|
}
|
||||||
|
return "false"
|
||||||
|
default:
|
||||||
|
return fmt.Sprintf("%v", v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Float64 coerces an interface value to a float64.
|
||||||
|
func Float64(v any) (float64, bool) {
|
||||||
|
if v == nil {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
switch n := v.(type) {
|
||||||
|
case float64:
|
||||||
|
return n, true
|
||||||
|
case float32:
|
||||||
|
return float64(n), true
|
||||||
|
case int:
|
||||||
|
return float64(n), true
|
||||||
|
case int64:
|
||||||
|
return float64(n), true
|
||||||
|
case uint32:
|
||||||
|
return float64(n), true
|
||||||
|
case uint64:
|
||||||
|
return float64(n), true
|
||||||
|
case string:
|
||||||
|
s := strings.TrimSpace(n)
|
||||||
|
if s == "" {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
f, err := strconv.ParseFloat(s, 64)
|
||||||
|
if err != nil {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return f, true
|
||||||
|
case json.Number:
|
||||||
|
f, err := n.Float64()
|
||||||
|
if err != nil {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
return f, true
|
||||||
|
}
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
@@ -0,0 +1,90 @@
|
|||||||
|
package conv
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestUint32(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
input any
|
||||||
|
expected uint32
|
||||||
|
ok bool
|
||||||
|
}{
|
||||||
|
{uint32(123), 123, true},
|
||||||
|
{float64(456.0), 456, true},
|
||||||
|
{float64(-1.0), 0, false},
|
||||||
|
{float32(789.0), 789, true},
|
||||||
|
{100, 100, true},
|
||||||
|
{int64(200), 200, true},
|
||||||
|
{uint64(300), 300, true},
|
||||||
|
{uint64(0xFFFFFFFF + 1), 0, false},
|
||||||
|
{json.Number("42"), 42, true},
|
||||||
|
{json.Number("invalid"), 0, false},
|
||||||
|
{"500", 500, true},
|
||||||
|
{" 600 ", 600, true},
|
||||||
|
{"invalid", 0, false},
|
||||||
|
{nil, 0, false},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
val, ok := Uint32(tt.input)
|
||||||
|
if val != tt.expected || ok != tt.ok {
|
||||||
|
t.Errorf("Uint32(%v) = (%d, %t), expected (%d, %t)", tt.input, val, ok, tt.expected, tt.ok)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestString(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
input any
|
||||||
|
expected string
|
||||||
|
}{
|
||||||
|
{"hello", "hello"},
|
||||||
|
{123, "123"},
|
||||||
|
{int64(456), "456"},
|
||||||
|
{uint32(789), "789"},
|
||||||
|
{uint64(100), "100"},
|
||||||
|
{float64(12.34), "12.34"},
|
||||||
|
{float64(56.0), "56"},
|
||||||
|
{float32(78.0), "78"},
|
||||||
|
{json.Number("42.2"), "42.2"},
|
||||||
|
{true, "true"},
|
||||||
|
{false, "false"},
|
||||||
|
{nil, ""},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
val := String(tt.input)
|
||||||
|
if val != tt.expected {
|
||||||
|
t.Errorf("String(%v) = %q, expected %q", tt.input, val, tt.expected)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestFloat64(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
input any
|
||||||
|
expected float64
|
||||||
|
ok bool
|
||||||
|
}{
|
||||||
|
{float64(12.34), 12.34, true},
|
||||||
|
{float32(12.5), 12.5, true},
|
||||||
|
{100, 100.0, true},
|
||||||
|
{int64(200), 200.0, true},
|
||||||
|
{uint32(300), 300.0, true},
|
||||||
|
{uint64(400), 400.0, true},
|
||||||
|
{"12.34", 12.34, true},
|
||||||
|
{" 56.78 ", 56.78, true},
|
||||||
|
{"invalid", 0.0, false},
|
||||||
|
{json.Number("42.2"), 42.2, true},
|
||||||
|
{nil, 0.0, false},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
val, ok := Float64(tt.input)
|
||||||
|
if val != tt.expected || ok != tt.ok {
|
||||||
|
t.Errorf("Float64(%v) = (%f, %t), expected (%f, %t)", tt.input, val, ok, tt.expected, tt.ok)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,38 @@
|
|||||||
|
// Package httpclient provides shared HTTP client helpers used across the
|
||||||
|
// monorepo: client IP extraction, CORS middleware, and related utilities.
|
||||||
|
package httpclient
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
"strings"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ClientIP extracts the best-effort client IP from a request by checking
|
||||||
|
// common reverse-proxy and CDN headers in priority order, falling back to
|
||||||
|
// the TCP remote address when no header is present. It does NOT respect
|
||||||
|
// query-parameter overrides (those belong at the handler level — see
|
||||||
|
// slask-finder/cmd/stores for the ?ip= debug pattern).
|
||||||
|
//
|
||||||
|
// Priority order: CF-Connecting-IP → X-Real-IP → X-Forwarded-For (first
|
||||||
|
// entry) → RemoteAddr (with port stripped when present).
|
||||||
|
func ClientIP(r *http.Request) string {
|
||||||
|
for _, h := range []string{"CF-Connecting-IP", "X-Real-IP", "X-Forwarded-For"} {
|
||||||
|
v := r.Header.Get(h)
|
||||||
|
if v == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// X-Forwarded-For may be a comma-separated list; take the first.
|
||||||
|
if h == "X-Forwarded-For" {
|
||||||
|
if i := strings.IndexByte(v, ','); i >= 0 {
|
||||||
|
v = v[:i]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return strings.TrimSpace(v)
|
||||||
|
}
|
||||||
|
// Fall back to the TCP remote address, stripping the port when present.
|
||||||
|
if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil {
|
||||||
|
return host
|
||||||
|
}
|
||||||
|
return r.RemoteAddr
|
||||||
|
}
|
||||||
@@ -0,0 +1,101 @@
|
|||||||
|
package httpclient
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/http"
|
||||||
|
)
|
||||||
|
|
||||||
|
type corsConfig struct {
|
||||||
|
origin string // "" = mirror request Origin; "*" = wildcard (no credentials)
|
||||||
|
methods string
|
||||||
|
headers string
|
||||||
|
exposeHeaders string
|
||||||
|
}
|
||||||
|
|
||||||
|
// CORSOption customises the CORS middleware.
|
||||||
|
type CORSOption func(*corsConfig)
|
||||||
|
|
||||||
|
// WithOrigin sets the Access-Control-Allow-Origin value. Use "*" for public
|
||||||
|
// APIs that don't need credentials. The default (empty string) mirrors the
|
||||||
|
// request's Origin header and enables credentials.
|
||||||
|
func WithOrigin(origin string) CORSOption {
|
||||||
|
return func(c *corsConfig) { c.origin = origin }
|
||||||
|
}
|
||||||
|
|
||||||
|
// WithMethods sets Access-Control-Allow-Methods. Default: "GET, POST, PUT, OPTIONS".
|
||||||
|
func WithMethods(methods string) CORSOption {
|
||||||
|
return func(c *corsConfig) { c.methods = methods }
|
||||||
|
}
|
||||||
|
|
||||||
|
// WithHeaders sets Access-Control-Allow-Headers. Default: "*".
|
||||||
|
func WithHeaders(headers string) CORSOption {
|
||||||
|
return func(c *corsConfig) { c.headers = headers }
|
||||||
|
}
|
||||||
|
|
||||||
|
// WithExposeHeaders sets Access-Control-Expose-Headers. Default: empty.
|
||||||
|
func WithExposeHeaders(headers string) CORSOption {
|
||||||
|
return func(c *corsConfig) { c.exposeHeaders = headers }
|
||||||
|
}
|
||||||
|
|
||||||
|
// CORS returns middleware that sets CORS response headers and handles
|
||||||
|
// preflight (OPTIONS) requests. By default it mirrors the request Origin,
|
||||||
|
// enables credentials, and allows GET/POST/PUT/OPTIONS with any request
|
||||||
|
// headers.
|
||||||
|
//
|
||||||
|
// Use WithOrigin("*") for public APIs that don't need credentials (per the
|
||||||
|
// CORS spec, credentials and wildcard origin are mutually exclusive).
|
||||||
|
//
|
||||||
|
// Common patterns:
|
||||||
|
//
|
||||||
|
// // Public API (wildcard origin, no credentials):
|
||||||
|
// mux := http.NewServeMux()
|
||||||
|
// handler := httpclient.CORS(
|
||||||
|
// httpclient.WithOrigin("*"),
|
||||||
|
// httpclient.WithMethods("GET, POST, OPTIONS"),
|
||||||
|
// )(mux)
|
||||||
|
//
|
||||||
|
// // Authenticated API (origin reflection, credentials):
|
||||||
|
// handler := httpclient.CORS()(mux)
|
||||||
|
func CORS(opts ...CORSOption) func(http.Handler) http.Handler {
|
||||||
|
cfg := corsConfig{
|
||||||
|
methods: "GET, POST, PUT, OPTIONS",
|
||||||
|
headers: "*",
|
||||||
|
}
|
||||||
|
for _, o := range opts {
|
||||||
|
o(&cfg)
|
||||||
|
}
|
||||||
|
return func(next http.Handler) http.Handler {
|
||||||
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
setCORSHeaders(w, r, cfg)
|
||||||
|
if r.Method == http.MethodOptions {
|
||||||
|
w.Header().Set("Access-Control-Max-Age", "86400")
|
||||||
|
w.WriteHeader(http.StatusNoContent)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
next.ServeHTTP(w, r)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func setCORSHeaders(w http.ResponseWriter, r *http.Request, cfg corsConfig) {
|
||||||
|
origin := cfg.origin
|
||||||
|
if origin == "" {
|
||||||
|
origin = r.Header.Get("Origin")
|
||||||
|
}
|
||||||
|
if origin == "" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.Header().Set("Access-Control-Allow-Origin", origin)
|
||||||
|
// Per spec, credentials can't be used with wildcard origin.
|
||||||
|
if origin != "*" {
|
||||||
|
w.Header().Set("Access-Control-Allow-Credentials", "true")
|
||||||
|
}
|
||||||
|
if cfg.exposeHeaders != "" {
|
||||||
|
w.Header().Set("Access-Control-Expose-Headers", cfg.exposeHeaders)
|
||||||
|
}
|
||||||
|
if cfg.methods != "" {
|
||||||
|
w.Header().Set("Access-Control-Allow-Methods", cfg.methods)
|
||||||
|
}
|
||||||
|
if cfg.headers != "" {
|
||||||
|
w.Header().Set("Access-Control-Allow-Headers", cfg.headers)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,253 @@
|
|||||||
|
// Package httpserver holds the shared HTTP server lifecycle helpers used by
|
||||||
|
// every service in the monorepo: graceful shutdown with ordered hooks,
|
||||||
|
// timeout configuration, and standard probe/debug mux registration.
|
||||||
|
//
|
||||||
|
// This package consolidates the ~40-line signal+server+shutdown block that
|
||||||
|
// was hand-rolled in 12+ service main.go files (go-redirector,
|
||||||
|
// go-form-submissions, go-order-manager, go-cart-actor cmds, go-graph-cms-v2,
|
||||||
|
// slask-finder cmds, backoffice). It is promoted from
|
||||||
|
// slask-finder/pkg/common/graceful.go, which was the first implementation.
|
||||||
|
//
|
||||||
|
// # Usage
|
||||||
|
//
|
||||||
|
// Simplest (no custom readiness, no shutdown hooks):
|
||||||
|
//
|
||||||
|
// mux := http.NewServeMux()
|
||||||
|
// // … register handlers …
|
||||||
|
// httpserver.RegisterProbes(mux, nil) // /healthz + /readyz (always ready)
|
||||||
|
//
|
||||||
|
// server := &http.Server{Addr: ":8080", Handler: mux, ReadHeaderTimeout: 5*time.Second}
|
||||||
|
// if err := httpserver.RunServerWithShutdown(server, "my-service", 15*time.Second, 5*time.Second); err != nil {
|
||||||
|
// log.Fatalf("server: %v", err)
|
||||||
|
// }
|
||||||
|
//
|
||||||
|
// With custom readiness and shutdown hooks:
|
||||||
|
//
|
||||||
|
// httpserver.RegisterProbes(mux, func() bool { return store.Loaded() })
|
||||||
|
//
|
||||||
|
// if err := httpserver.RunServerWithShutdown(server, "my-service", 15*time.Second, 5*time.Second,
|
||||||
|
// func(ctx context.Context) error { return store.Flush() },
|
||||||
|
// func(ctx context.Context) error { return cache.Close() },
|
||||||
|
// ); err != nil {
|
||||||
|
// log.Fatalf("server: %v", err)
|
||||||
|
// }
|
||||||
|
//
|
||||||
|
// With a separate debug mux (metrics + pprof on a second port):
|
||||||
|
//
|
||||||
|
// debugMux := http.NewServeMux()
|
||||||
|
// httpserver.RegisterDebug(debugMux, promhttp.Handler())
|
||||||
|
// go http.ListenAndServe(":8081", debugMux)
|
||||||
|
package httpserver
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"log"
|
||||||
|
"net/http"
|
||||||
|
"net/http/pprof"
|
||||||
|
"os"
|
||||||
|
"os/signal"
|
||||||
|
"strconv"
|
||||||
|
"syscall"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ShutdownHook is a function executed after a termination signal is received
|
||||||
|
// but before the HTTP server begins its graceful shutdown. If a hook returns
|
||||||
|
// an error it will be logged; shutdown continues regardless.
|
||||||
|
type ShutdownHook func(ctx context.Context) error
|
||||||
|
|
||||||
|
// RunServerWithShutdown starts the provided *http.Server and blocks until either
|
||||||
|
// a listener error occurs or a termination signal (SIGINT or SIGTERM) is
|
||||||
|
// received. It then runs any provided hooks (in order) with a context that
|
||||||
|
// shares the overall shutdown deadline, and finally gracefully shuts down the
|
||||||
|
// server. A per-hook timeout can be specified; if zero a default is used.
|
||||||
|
//
|
||||||
|
// Returns nil if the server shut down cleanly after a signal. Returns a
|
||||||
|
// wrapped listener error if ListenAndServe failed (the composition root then
|
||||||
|
// decides retry vs exit).
|
||||||
|
//
|
||||||
|
// Parameters:
|
||||||
|
//
|
||||||
|
// server: configured *http.Server (Addr, Handler, timeouts etc.)
|
||||||
|
// startupLog: message to log when the server starts listening
|
||||||
|
// shutdownTimeout: total time allowed for all hooks + server shutdown
|
||||||
|
// hookTimeout: individual timeout for each hook (if <=0 defaults to 5s)
|
||||||
|
// hooks: optional ordered list of shutdown hooks
|
||||||
|
func RunServerWithShutdown(server *http.Server, startupLog string, shutdownTimeout, hookTimeout time.Duration, hooks ...ShutdownHook) error {
|
||||||
|
if hookTimeout <= 0 {
|
||||||
|
hookTimeout = 5 * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
|
// serveErr is buffered so the goroutine never blocks. nil means clean stop.
|
||||||
|
serveErr := make(chan error, 1)
|
||||||
|
|
||||||
|
// Start server in goroutine
|
||||||
|
go func() {
|
||||||
|
log.Printf("starting %s on %s", startupLog, server.Addr)
|
||||||
|
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||||
|
log.Printf("%s listen error: %v", startupLog, err)
|
||||||
|
serveErr <- err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
serveErr <- nil
|
||||||
|
}()
|
||||||
|
|
||||||
|
// Wait for either a signal or a listener error.
|
||||||
|
stop := make(chan os.Signal, 1)
|
||||||
|
signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
|
||||||
|
|
||||||
|
defer signal.Stop(stop)
|
||||||
|
|
||||||
|
var listenerErr error
|
||||||
|
select {
|
||||||
|
case <-stop:
|
||||||
|
log.Printf("shutdown signal received for %s", startupLog)
|
||||||
|
case err := <-serveErr:
|
||||||
|
// listener died before any signal — return wrapped error to the caller
|
||||||
|
// instead of crashing the process from inside this helper.
|
||||||
|
if err != nil {
|
||||||
|
listenerErr = fmt.Errorf("%s listen error: %w", startupLog, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
// Run hooks sequentially with individual timeouts
|
||||||
|
for i, h := range hooks {
|
||||||
|
if h == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
hCtx, hCancel := context.WithTimeout(ctx, hookTimeout)
|
||||||
|
if err := h(hCtx); err != nil {
|
||||||
|
log.Printf("shutdown hook %d failed: %v", i, err)
|
||||||
|
}
|
||||||
|
hCancel()
|
||||||
|
if err := hCtx.Err(); err == context.DeadlineExceeded {
|
||||||
|
log.Printf("shutdown hook %d timed out", i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Finally shut down the HTTP server (idempotent if listener already died).
|
||||||
|
if err := server.Shutdown(ctx); err != nil {
|
||||||
|
log.Printf("graceful shutdown failed: %v", err)
|
||||||
|
} else if listenerErr == nil {
|
||||||
|
log.Printf("%s shutdown complete", startupLog)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Drain the serve goroutine result if it raced.
|
||||||
|
select {
|
||||||
|
case <-serveErr:
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
|
if listenerErr != nil {
|
||||||
|
return listenerErr
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// TimeoutConfig holds server and shutdown related timeouts (all durations).
|
||||||
|
type TimeoutConfig struct {
|
||||||
|
ReadHeader time.Duration
|
||||||
|
Read time.Duration
|
||||||
|
Write time.Duration
|
||||||
|
Idle time.Duration
|
||||||
|
Shutdown time.Duration
|
||||||
|
Hook time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
// LoadTimeoutConfig reads environment variables (if present) to override defaults.
|
||||||
|
// Each env var is parsed as an integer number of seconds. If parsing fails or value <=0,
|
||||||
|
// the provided default is retained.
|
||||||
|
// Env variables:
|
||||||
|
//
|
||||||
|
// READ_HEADER_TIMEOUT
|
||||||
|
// READ_TIMEOUT
|
||||||
|
// WRITE_TIMEOUT
|
||||||
|
// IDLE_TIMEOUT
|
||||||
|
// SHUTDOWN_TIMEOUT
|
||||||
|
// HOOK_TIMEOUT
|
||||||
|
func LoadTimeoutConfig(defaults TimeoutConfig) TimeoutConfig {
|
||||||
|
apply := func(curr *time.Duration, env string) {
|
||||||
|
if v := os.Getenv(env); v != "" {
|
||||||
|
if n, err := strconv.Atoi(v); err == nil && n > 0 {
|
||||||
|
*curr = time.Duration(n) * time.Second
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
apply(&defaults.ReadHeader, "READ_HEADER_TIMEOUT")
|
||||||
|
apply(&defaults.Read, "READ_TIMEOUT")
|
||||||
|
apply(&defaults.Write, "WRITE_TIMEOUT")
|
||||||
|
apply(&defaults.Idle, "IDLE_TIMEOUT")
|
||||||
|
apply(&defaults.Shutdown, "SHUTDOWN_TIMEOUT")
|
||||||
|
apply(&defaults.Hook, "HOOK_TIMEOUT")
|
||||||
|
return defaults
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewServerWithTimeouts attaches timeout settings to an existing *http.Server or creates a new one if nil.
|
||||||
|
func NewServerWithTimeouts(base *http.Server, cfg TimeoutConfig) *http.Server {
|
||||||
|
if base == nil {
|
||||||
|
return &http.Server{
|
||||||
|
ReadHeaderTimeout: cfg.ReadHeader,
|
||||||
|
ReadTimeout: cfg.Read,
|
||||||
|
WriteTimeout: cfg.Write,
|
||||||
|
IdleTimeout: cfg.Idle,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
base.ReadHeaderTimeout = cfg.ReadHeader
|
||||||
|
base.ReadTimeout = cfg.Read
|
||||||
|
base.WriteTimeout = cfg.Write
|
||||||
|
base.IdleTimeout = cfg.Idle
|
||||||
|
return base
|
||||||
|
}
|
||||||
|
|
||||||
|
// RegisterProbes registers /healthz and /readyz on the given mux.
|
||||||
|
//
|
||||||
|
// /healthz always returns 200 OK (liveness — the process is running).
|
||||||
|
// /readyz returns 200 if ready() returns true (or if ready is nil), 503
|
||||||
|
// otherwise (readiness — the service can handle requests).
|
||||||
|
//
|
||||||
|
// Both endpoints are registered with the "GET " method prefix so they work
|
||||||
|
// with Go 1.22+ pattern-based mux routing. The paths are pod-local ops
|
||||||
|
// endpoints (never exposed at the edge); register them first so they take
|
||||||
|
// precedence over catch-all handlers.
|
||||||
|
func RegisterProbes(mux *http.ServeMux, ready func() bool) {
|
||||||
|
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
_, _ = w.Write([]byte("ok"))
|
||||||
|
})
|
||||||
|
mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
if ready != nil && !ready() {
|
||||||
|
w.WriteHeader(http.StatusServiceUnavailable)
|
||||||
|
_, _ = w.Write([]byte("not ready"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
_, _ = w.Write([]byte("ready"))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// RegisterDebug registers /metrics and /debug/pprof/* on the given mux for a
|
||||||
|
// separate debug/admin port. The metricsHandler is typically
|
||||||
|
// prometheus/promhttp.Handler(); it is passed as a parameter so this package
|
||||||
|
// does not depend on the prometheus client library. The pprof endpoints are
|
||||||
|
// registered from net/http/pprof.
|
||||||
|
//
|
||||||
|
// Use this on a separate mux served on a different port (e.g. :8081) so the
|
||||||
|
// debug surface is not exposed at the edge:
|
||||||
|
//
|
||||||
|
// debugMux := http.NewServeMux()
|
||||||
|
// httpserver.RegisterDebug(debugMux, promhttp.Handler())
|
||||||
|
// go http.ListenAndServe(":8081", debugMux)
|
||||||
|
func RegisterDebug(mux *http.ServeMux, metricsHandler http.Handler) {
|
||||||
|
if metricsHandler != nil {
|
||||||
|
mux.Handle("/metrics", metricsHandler)
|
||||||
|
}
|
||||||
|
mux.HandleFunc("/debug/pprof/", pprof.Index)
|
||||||
|
mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
|
||||||
|
mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
|
||||||
|
mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
|
||||||
|
mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
||||||
|
}
|
||||||
Reference in new issue
Block a user