From d4e77a64a9d991682d2d2d1262768037fca4bf84 Mon Sep 17 00:00:00 2001 From: youfak Date: Thu, 30 Jul 2026 12:30:42 +0800 Subject: [PATCH] feat: add controller metrics probes --- go.mod | 12 +- go.sum | 40 ++++-- internal/config/config_test.go | 25 ++++ internal/config/validate.go | 24 +++- internal/controller/bootstrap/bootstrap.go | 24 +++- .../bootstrap/bootstrap_integration_test.go | 18 ++- .../controller/bootstrap/bootstrap_test.go | 8 +- .../controller/bootstrap/infrastructure.go | 52 ++++++++ .../bootstrap/infrastructure_test.go | 42 ++++++ internal/controller/runtime/runtime.go | 39 +++++- internal/controller/runtime/runtime_test.go | 51 ++++++++ internal/platform/metrics/handler.go | 70 ++++++++++ internal/platform/metrics/handler_test.go | 121 ++++++++++++++++++ 13 files changed, 500 insertions(+), 26 deletions(-) create mode 100644 internal/platform/metrics/handler.go create mode 100644 internal/platform/metrics/handler_test.go diff --git a/go.mod b/go.mod index 9be3ad6..9cf3003 100644 --- a/go.mod +++ b/go.mod @@ -4,17 +4,25 @@ go 1.26.0 require ( github.com/jackc/pgx/v5 v5.6.0 + github.com/prometheus/client_golang v1.24.1 github.com/redis/go-redis/v9 v9.19.0 go.yaml.in/yaml/v4 v4.0.0-rc.3 ) require ( + github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect github.com/jackc/puddle/v2 v2.2.1 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.70.1 // indirect + github.com/prometheus/procfs v0.21.1 // indirect go.uber.org/atomic v1.11.0 // indirect golang.org/x/crypto v0.17.0 // indirect - golang.org/x/sync v0.1.0 // indirect - golang.org/x/text v0.14.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.40.0 // indirect + google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/go.sum b/go.sum index 10fd021..36873af 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= @@ -7,6 +9,8 @@ github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= @@ -15,31 +19,51 @@ github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY= github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw= github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk= github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= +github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= +github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY= +github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc= +github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI= +github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= github.com/redis/go-redis/v9 v9.19.0 h1:XPVaaPSnG6RhYf7p+rmSa9zZfeVAnWsH5h3lxthOm/k= github.com/redis/go-redis/v9 v9.19.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= -github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v4 v4.0.0-rc.3 h1:3h1fjsh1CTAPjW7q/EMe+C8shx5d8ctzZTrLcs/j8Go= go.yaml.in/yaml/v4 v4.0.0-rc.3/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k= golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4= -golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= -golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= -golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= -golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= +golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 99122e2..031d831 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -318,6 +318,31 @@ func TestValidateAcceptsBearerListenerAuthentication(t *testing.T) { } } +func TestValidateMetricsListener(t *testing.T) { + t.Parallel() + cfg := mustLoadValidConfig(t) + cfg.Metrics = Metrics{Enabled: true} + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "metrics listen") { + t.Fatalf("Validate(metrics without listen) error = %v", err) + } + cfg.Metrics.Listen = "not-an-address" + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "metrics listen") { + t.Fatalf("Validate(invalid metrics listen) error = %v", err) + } + cfg.Metrics.Listen = "127.0.0.1:70000" + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "metrics listen") { + t.Fatalf("Validate(out-of-range metrics port) error = %v", err) + } + cfg.Metrics.Listen = "0.0.0.0:9090" + if err := Validate(cfg); err != nil { + t.Fatalf("Validate(public metrics listener) error = %v", err) + } + cfg.Metrics = Metrics{Enabled: false, Listen: "not-an-address"} + if err := Validate(cfg); err != nil { + t.Fatalf("Validate(disabled metrics listener) error = %v", err) + } +} + func TestValidateRejectsInvalidConfigurationMatrix(t *testing.T) { tests := []struct { name string diff --git a/internal/config/validate.go b/internal/config/validate.go index 5211a5a..ab5df72 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -5,6 +5,7 @@ import ( "net" "net/url" "regexp" + "strconv" "strings" ) @@ -28,6 +29,14 @@ func Validate(cfg *Config) error { return err } } + if cfg.Metrics.Enabled { + if cfg.Metrics.Listen == "" { + return fmt.Errorf("validate metrics listen: address is required") + } + if _, err := validateListenAddress("metrics", cfg.Metrics.Listen); err != nil { + return err + } + } if fetchConfigured(cfg.Defaults.Fetch) { if err := validateFetch("defaults.fetch", cfg.Defaults.Fetch); err != nil { return err @@ -110,9 +119,9 @@ func validateListener(name string, listener Listener, security Security) error { return fmt.Errorf("validate %s limits.%s: must be non-negative", name, limit.name) } } - host, _, err := net.SplitHostPort(listener.Listen) + host, err := validateListenAddress(name, listener.Listen) if err != nil { - return fmt.Errorf("validate %s listen: %w", name, err) + return err } if security.RequireProtectionOnPublicListen && isPublicHost(host) && listener.Auth.Mode == "none" && len(listener.Access.AllowCIDRs) == 0 { return fmt.Errorf("validate %s: unprotected public listener is forbidden", name) @@ -139,6 +148,17 @@ func validateListener(name string, listener Listener, security Security) error { return nil } +func validateListenAddress(name, address string) (string, error) { + host, port, err := net.SplitHostPort(address) + if err != nil { + return "", fmt.Errorf("validate %s listen: %w", name, err) + } + if _, err = strconv.ParseUint(port, 10, 16); err != nil { + return "", fmt.Errorf("validate %s listen: invalid port", name) + } + return host, nil +} + func validateRouting(index int, route Routing, upstreams map[string]Upstream, seen map[string]struct{}) error { if route.Name == "" { return fmt.Errorf("validate routing[%d]: name is required", index) diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index 2031609..57a3fb5 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -8,6 +8,8 @@ import ( "strings" "time" + "github.com/prometheus/client_golang/prometheus" + "proxy-pool/internal/config" "proxy-pool/internal/controller/admin" "proxy-pool/internal/controller/distribution" @@ -18,6 +20,7 @@ import ( extractionDomain "proxy-pool/internal/domain/extraction" "proxy-pool/internal/platform/admission" "proxy-pool/internal/platform/httpserver" + platformMetrics "proxy-pool/internal/platform/metrics" ) var ( @@ -38,10 +41,11 @@ type activityStore interface { } type ports struct { - state admin.StateRepository - activity activityStore - readiness distribution.ReadinessChecker - close func() error + state admin.StateRepository + activity activityStore + readiness distribution.ReadinessChecker + metricsReadiness platformMetrics.ReadinessChecker + close func() error } type infrastructure interface { @@ -129,6 +133,18 @@ func run(ctx context.Context, options Options, infrastructure infrastructure, fa } dependencies.AdminService = service } + if loaded.Value.Metrics.Enabled { + if nilInterface(opened.metricsReadiness) { + return errors.Join(ErrStartup, ErrInvalidOptions) + } + handler, handlerErr := platformMetrics.NewHandler(platformMetrics.Dependencies{ + Gatherer: prometheus.DefaultGatherer, Readiness: opened.metricsReadiness, + }) + if handlerErr != nil { + return fmt.Errorf("%w: build metrics handler: %w", ErrStartup, handlerErr) + } + dependencies.MetricsHandler = handler + } runner, err := factory.New(configurationStore.Current(), dependencies, controllerRuntime.Options{HTTP: options.HTTP}) if err != nil { diff --git a/internal/controller/bootstrap/bootstrap_integration_test.go b/internal/controller/bootstrap/bootstrap_integration_test.go index f9de8bb..fddb0fb 100644 --- a/internal/controller/bootstrap/bootstrap_integration_test.go +++ b/internal/controller/bootstrap/bootstrap_integration_test.go @@ -4,6 +4,8 @@ package bootstrap import ( "context" + "net/http" + "net/http/httptest" "os" "strings" "testing" @@ -40,10 +42,17 @@ func TestProductionBootstrapOpensBothStoresAndCommitsStartupConfiguration(t *tes factory.status.Upstreams[1].Name != "provider-b" { t.Fatalf("Admin Status upstreams = %+v", factory.status.Upstreams) } + if factory.readyStatus != http.StatusOK || factory.metricsStatus != http.StatusOK || + !strings.Contains(factory.metricsBody, "go_") { + t.Fatalf("Metrics probes = ready:%d metrics:%d body:%q", factory.readyStatus, factory.metricsStatus, factory.metricsBody) + } } type integrationRuntimeFactory struct { - status admin.Status + status admin.Status + readyStatus int + metricsStatus int + metricsBody string } func (factory *integrationRuntimeFactory) New( @@ -55,6 +64,13 @@ func (factory *integrationRuntimeFactory) New( if err := dependencies.Readiness.Ready(ctx); err != nil { return err } + ready := httptest.NewRecorder() + dependencies.MetricsHandler.ServeHTTP(ready, httptest.NewRequest(http.MethodGet, "/readyz", nil)) + factory.readyStatus = ready.Code + metrics := httptest.NewRecorder() + dependencies.MetricsHandler.ServeHTTP(metrics, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + factory.metricsStatus = metrics.Code + factory.metricsBody = metrics.Body.String() status, err := dependencies.AdminService.Status(ctx) factory.status = status return err diff --git a/internal/controller/bootstrap/bootstrap_test.go b/internal/controller/bootstrap/bootstrap_test.go index 2b11883..4cabe30 100644 --- a/internal/controller/bootstrap/bootstrap_test.go +++ b/internal/controller/bootstrap/bootstrap_test.go @@ -20,7 +20,7 @@ func TestRunLoadsOneSnapshotCommitsItAndClosesInfrastructure(t *testing.T) { activity := &stubActivityStore{} closeErr := errors.New("close failed") infrastructure := &stubInfrastructure{ports: ports{ - state: state, activity: activity, readiness: readyStub{}, + state: state, activity: activity, readiness: readyStub{}, metricsReadiness: readyStub{}, close: func() error { return closeErr }, }} runErr := errors.New("runtime failed") @@ -44,7 +44,8 @@ func TestRunLoadsOneSnapshotCommitsItAndClosesInfrastructure(t *testing.T) { t.Fatalf("management snapshot = %+v, %v", snapshot, snapshotErr) } if factory.configuration == nil || factory.dependencies.Extractor == nil || - factory.dependencies.Readiness == nil || factory.dependencies.AdminService == nil { + factory.dependencies.Readiness == nil || factory.dependencies.AdminService == nil || + factory.dependencies.MetricsHandler == nil { t.Fatalf("runtime assembly = config:%p dependencies:%+v", factory.configuration, factory.dependencies) } } @@ -169,6 +170,9 @@ admin: enabled: true listen: 127.0.0.1:0 auth: {mode: none} +metrics: + enabled: true + listen: 127.0.0.1:0 storage: postgresURL: postgres://fixture redisURL: redis://fixture diff --git a/internal/controller/bootstrap/infrastructure.go b/internal/controller/bootstrap/infrastructure.go index d851dab..56d3a2e 100644 --- a/internal/controller/bootstrap/infrastructure.go +++ b/internal/controller/bootstrap/infrastructure.go @@ -13,6 +13,7 @@ import ( "proxy-pool/internal/adapters/redisactivity" "proxy-pool/internal/config" "proxy-pool/internal/platform/credentials" + platformMetrics "proxy-pool/internal/platform/metrics" ) const ( @@ -111,9 +112,60 @@ func (*productionInfrastructure) Open( opened.activity = adapter opened.readiness = redisReadiness{client: redisClient} } + if configuration.Metrics.Enabled { + opened.metricsReadiness = selectMetricsReadiness( + configuration, + storeReadiness{postgres: postgresPool, redis: redisClient}, + redisReadiness{client: redisClient}, + ) + } return opened, nil } +func selectMetricsReadiness( + configuration *config.Config, + admin, activity platformMetrics.ReadinessChecker, +) platformMetrics.ReadinessChecker { + if configuration.Distribution.Enabled { + return activity + } + if configuration.Admin.Enabled { + return admin + } + return alwaysReady{} +} + +type alwaysReady struct{} + +func (alwaysReady) Ready(ctx context.Context) error { + if ctx == nil { + return ErrInvalidOptions + } + return ctx.Err() +} + +type storeReadiness struct { + postgres *pgxpool.Pool + redis *redis.Client +} + +func (readiness storeReadiness) Ready(ctx context.Context) error { + if ctx == nil { + return ErrInvalidOptions + } + if readiness.postgres != nil { + if err := readiness.postgres.Ping(ctx); err != nil { + return contextOr(ctx, ErrPostgresUnavailable) + } + } + if readiness.redis != nil { + if err := readiness.redis.Ping(ctx).Err(); err != nil { + return contextOr(ctx, ErrRedisUnavailable) + } + } + return nil +} + type redisReadiness struct { client *redis.Client } diff --git a/internal/controller/bootstrap/infrastructure_test.go b/internal/controller/bootstrap/infrastructure_test.go index eadff90..e16e4fe 100644 --- a/internal/controller/bootstrap/infrastructure_test.go +++ b/internal/controller/bootstrap/infrastructure_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "strings" + "sync/atomic" "testing" "proxy-pool/internal/config" @@ -30,6 +31,47 @@ func TestProductionInfrastructureRejectsInvalidStorageWithoutLeakingURLs(t *test } } +func TestSelectMetricsReadinessPreservesDistributionWhenAdminStoreFails(t *testing.T) { + t.Parallel() + adminCalls := &atomic.Int64{} + activityCalls := &atomic.Int64{} + adminReady := readinessFunc(func(context.Context) error { + adminCalls.Add(1) + return ErrPostgresUnavailable + }) + activityReady := readinessFunc(func(context.Context) error { + activityCalls.Add(1) + return nil + }) + selected := selectMetricsReadiness(&config.Config{ + Admin: config.Listener{Enabled: true}, + Distribution: config.Distribution{Listener: config.Listener{Enabled: true}}, + }, adminReady, activityReady) + if err := selected.Ready(context.Background()); err != nil { + t.Fatalf("Ready() error = %v", err) + } + if adminCalls.Load() != 0 || activityCalls.Load() != 1 { + t.Fatalf("readiness calls = admin:%d activity:%d", adminCalls.Load(), activityCalls.Load()) + } +} + +func TestSelectMetricsReadinessUsesAdminStoresWithoutDistribution(t *testing.T) { + t.Parallel() + wantErr := errors.New("admin unavailable") + selected := selectMetricsReadiness( + &config.Config{Admin: config.Listener{Enabled: true}}, + readinessFunc(func(context.Context) error { return wantErr }), + readinessFunc(func(context.Context) error { return nil }), + ) + if err := selected.Ready(context.Background()); !errors.Is(err, wantErr) { + t.Fatalf("Ready() error = %v, want %v", err, wantErr) + } +} + +type readinessFunc func(context.Context) error + +func (function readinessFunc) Ready(ctx context.Context) error { return function(ctx) } + func TestProductionRedisSizingUsesConfigurationBounds(t *testing.T) { t.Parallel() configuration := &config.Config{ diff --git a/internal/controller/runtime/runtime.go b/internal/controller/runtime/runtime.go index 013eab7..f1fc46a 100644 --- a/internal/controller/runtime/runtime.go +++ b/internal/controller/runtime/runtime.go @@ -23,9 +23,10 @@ const ( var ErrInvalidRuntime = errors.New("invalid controller HTTP runtime") type Dependencies struct { - Extractor distribution.Extractor - Readiness distribution.ReadinessChecker - AdminService admin.Service + Extractor distribution.Extractor + Readiness distribution.ReadinessChecker + AdminService admin.Service + MetricsHandler http.Handler } type Options struct { @@ -37,6 +38,7 @@ type Options struct { type Listeners struct { Distribution net.Listener Admin net.Listener + Metrics net.Listener } type Runtime struct { @@ -46,11 +48,14 @@ type Runtime struct { adminEnabled bool adminAddress string adminHandler http.Handler + metricsEnabled bool + metricsAddress string + metricsHandler http.Handler httpOptions httpserver.Options } func New(cfg *config.Config, dependencies Dependencies, options Options) (*Runtime, error) { - if cfg == nil || (!cfg.Distribution.Enabled && !cfg.Admin.Enabled) { + if cfg == nil || (!cfg.Distribution.Enabled && !cfg.Admin.Enabled && !cfg.Metrics.Enabled) { return nil, ErrInvalidRuntime } distributionBodyLimit, adminBodyLimit, err := resolveBodyLimits(options) @@ -104,6 +109,15 @@ func New(cfg *config.Config, dependencies Dependencies, options Options) (*Runti result.adminAddress = cfg.Admin.Listen result.adminHandler = handler } + + if cfg.Metrics.Enabled { + if strings.TrimSpace(cfg.Metrics.Listen) == "" || dependencies.MetricsHandler == nil { + return nil, ErrInvalidRuntime + } + result.metricsEnabled = true + result.metricsAddress = cfg.Metrics.Listen + result.metricsHandler = dependencies.MetricsHandler + } return result, nil } @@ -111,7 +125,7 @@ func (runtime *Runtime) Run(ctx context.Context) error { if runtime == nil || ctx == nil { return ErrInvalidRuntime } - bindings := make([]httpserver.Binding, 0, 2) + bindings := make([]httpserver.Binding, 0, 3) if runtime.distributionEnabled { bindings = append(bindings, httpserver.Binding{ Name: "distribution", Address: runtime.distributionAddress, Handler: runtime.distributionHandler, @@ -122,6 +136,11 @@ func (runtime *Runtime) Run(ctx context.Context) error { Name: "admin", Address: runtime.adminAddress, Handler: runtime.adminHandler, }) } + if runtime.metricsEnabled { + bindings = append(bindings, httpserver.Binding{ + Name: "metrics", Address: runtime.metricsAddress, Handler: runtime.metricsHandler, + }) + } if err := httpserver.ListenAndServe(ctx, runtime.httpOptions, bindings...); err != nil { return fmt.Errorf("run controller HTTP runtime: %w", err) } @@ -131,10 +150,11 @@ func (runtime *Runtime) Run(ctx context.Context) error { func (runtime *Runtime) Serve(ctx context.Context, listeners Listeners) error { if runtime == nil || ctx == nil || runtime.distributionEnabled != (listeners.Distribution != nil) || - runtime.adminEnabled != (listeners.Admin != nil) { + runtime.adminEnabled != (listeners.Admin != nil) || + runtime.metricsEnabled != (listeners.Metrics != nil) { return ErrInvalidRuntime } - endpoints := make([]httpserver.Endpoint, 0, 2) + endpoints := make([]httpserver.Endpoint, 0, 3) if runtime.distributionEnabled { endpoints = append(endpoints, httpserver.Endpoint{ Name: "distribution", Listener: listeners.Distribution, Handler: runtime.distributionHandler, @@ -145,6 +165,11 @@ func (runtime *Runtime) Serve(ctx context.Context, listeners Listeners) error { Name: "admin", Listener: listeners.Admin, Handler: runtime.adminHandler, }) } + if runtime.metricsEnabled { + endpoints = append(endpoints, httpserver.Endpoint{ + Name: "metrics", Listener: listeners.Metrics, Handler: runtime.metricsHandler, + }) + } if err := httpserver.Serve(ctx, runtime.httpOptions, endpoints...); err != nil { return fmt.Errorf("serve controller HTTP runtime: %w", err) } diff --git a/internal/controller/runtime/runtime_test.go b/internal/controller/runtime/runtime_test.go index 79bf936..25f7dde 100644 --- a/internal/controller/runtime/runtime_test.go +++ b/internal/controller/runtime/runtime_test.go @@ -19,34 +19,47 @@ import ( func TestRuntimeServesDistributionAndAdminOnIndependentListeners(t *testing.T) { t.Parallel() cfg := runtimeConfig() + cfg.Metrics = config.Metrics{Enabled: true, Listen: "127.0.0.1:0"} adminService := &stubAdminService{status: admin.Status{ConfigVersion: "cfg-7", SnapshotVersion: 11}} runtime, err := New(cfg, Dependencies{ Extractor: stubExtractor{}, Readiness: stubReadiness{}, AdminService: adminService, + MetricsHandler: http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/readyz" { + http.NotFound(response, request) + return + } + response.WriteHeader(http.StatusOK) + }), }, Options{HTTP: testHTTPOptions()}) if err != nil { t.Fatalf("New() error = %v", err) } distributionListener := mustListen(t) adminListener := mustListen(t) + metricsListener := mustListen(t) ctx, cancel := context.WithCancel(context.Background()) result := make(chan error, 1) go func() { result <- runtime.Serve(ctx, Listeners{ Distribution: distributionListener, Admin: adminListener, + Metrics: metricsListener, }) }() distributionURL := "http://" + distributionListener.Addr().String() adminURL := "http://" + adminListener.Addr().String() + metricsURL := "http://" + metricsListener.Addr().String() assertStatus(t, http.MethodGet, distributionURL+"/health/live", nil, http.StatusOK) assertStatus(t, http.MethodGet, distributionURL+"/api/v1/status", nil, http.StatusNotFound) assertStatus(t, http.MethodGet, adminURL+"/api/v1/status", nil, http.StatusUnauthorized) adminHeaders := http.Header{"Authorization": []string{"Bearer admin-token"}} assertStatus(t, http.MethodGet, adminURL+"/api/v1/status", adminHeaders, http.StatusOK) assertStatus(t, http.MethodGet, adminURL+"/health/live", adminHeaders, http.StatusNotFound) + assertStatus(t, http.MethodGet, metricsURL+"/readyz", nil, http.StatusOK) + assertStatus(t, http.MethodGet, metricsURL+"/api/v1/status", nil, http.StatusNotFound) if adminService.statusCalls.Load() != 1 { t.Fatalf("admin status calls = %d, want 1", adminService.statusCalls.Load()) } @@ -107,6 +120,44 @@ func TestServeRequiresExactlyTheEnabledListeners(t *testing.T) { } } +func TestRuntimeServesMetricsOnIndependentListener(t *testing.T) { + t.Parallel() + cfg := &config.Config{Metrics: config.Metrics{Enabled: true, Listen: "127.0.0.1:0"}} + metricsHandler := http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/readyz" { + http.NotFound(response, request) + return + } + response.WriteHeader(http.StatusOK) + }) + runtime, err := New(cfg, Dependencies{MetricsHandler: metricsHandler}, Options{HTTP: testHTTPOptions()}) + if err != nil { + t.Fatalf("New(metrics-only) error = %v", err) + } + listener := mustListen(t) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- runtime.Serve(ctx, Listeners{Metrics: listener}) }() + assertStatus(t, http.MethodGet, "http://"+listener.Addr().String()+"/readyz", nil, http.StatusOK) + cancel() + select { + case err := <-result: + if err != nil { + t.Fatalf("Serve(metrics-only) error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Serve(metrics-only) did not stop") + } +} + +func TestNewRequiresMetricsHandlerWhenEnabled(t *testing.T) { + t.Parallel() + cfg := &config.Config{Metrics: config.Metrics{Enabled: true, Listen: "127.0.0.1:0"}} + if _, err := New(cfg, Dependencies{}, Options{}); !errors.Is(err, ErrInvalidRuntime) { + t.Fatalf("New(metrics without handler) error = %v", err) + } +} + func runtimeConfig() *config.Config { return &config.Config{ Distribution: config.Distribution{ diff --git a/internal/platform/metrics/handler.go b/internal/platform/metrics/handler.go new file mode 100644 index 0000000..037315d --- /dev/null +++ b/internal/platform/metrics/handler.go @@ -0,0 +1,70 @@ +package metrics + +import ( + "context" + "errors" + "net/http" + "reflect" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +var ErrInvalidDependencies = errors.New("invalid metrics dependencies") + +type ReadinessChecker interface { + Ready(context.Context) error +} + +type Dependencies struct { + Gatherer prometheus.Gatherer + Readiness ReadinessChecker +} + +func NewHandler(dependencies Dependencies) (http.Handler, error) { + if nilInterface(dependencies.Gatherer) || nilInterface(dependencies.Readiness) { + return nil, ErrInvalidDependencies + } + metricsHandler := promhttp.HandlerFor(dependencies.Gatherer, promhttp.HandlerOpts{}) + return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.Method != http.MethodGet { + response.Header().Set("Allow", http.MethodGet) + http.Error(response, "method not allowed", http.StatusMethodNotAllowed) + return + } + switch request.URL.Path { + case "/livez": + writeStatus(response, http.StatusOK, "live") + case "/readyz": + if err := dependencies.Readiness.Ready(request.Context()); err != nil { + writeStatus(response, http.StatusServiceUnavailable, "unavailable") + return + } + writeStatus(response, http.StatusOK, "ready") + case "/metrics": + metricsHandler.ServeHTTP(response, request) + default: + http.NotFound(response, request) + } + }), nil +} + +func writeStatus(response http.ResponseWriter, status int, value string) { + response.Header().Set("Content-Type", "text/plain; charset=utf-8") + response.Header().Set("X-Content-Type-Options", "nosniff") + response.WriteHeader(status) + _, _ = response.Write([]byte(value + "\n")) +} + +func nilInterface(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} diff --git a/internal/platform/metrics/handler_test.go b/internal/platform/metrics/handler_test.go new file mode 100644 index 0000000..29a1219 --- /dev/null +++ b/internal/platform/metrics/handler_test.go @@ -0,0 +1,121 @@ +package metrics + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "github.com/prometheus/client_golang/prometheus" +) + +func TestHandlerServesMetricsAndIndependentProbes(t *testing.T) { + t.Parallel() + registry := prometheus.NewRegistry() + gauge := prometheus.NewGauge(prometheus.GaugeOpts{Name: "proxy_pool_test_inventory"}) + gauge.Set(7) + registry.MustRegister(gauge) + readiness := &recordingReadiness{} + handler, err := NewHandler(Dependencies{Gatherer: registry, Readiness: readiness}) + if err != nil { + t.Fatalf("NewHandler() error = %v", err) + } + + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + assertProbe(t, http.MethodGet, server.URL+"/livez", http.StatusOK, "live") + if readiness.calls.Load() != 0 { + t.Fatalf("livez readiness calls = %d, want 0", readiness.calls.Load()) + } + assertProbe(t, http.MethodGet, server.URL+"/readyz", http.StatusOK, "ready") + if readiness.calls.Load() != 1 { + t.Fatalf("readyz readiness calls = %d, want 1", readiness.calls.Load()) + } + + response, err := http.Get(server.URL + "/metrics") + if err != nil { + t.Fatalf("GET /metrics: %v", err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("ReadAll(/metrics): %v", err) + } + if response.StatusCode != http.StatusOK || !strings.Contains(string(body), "proxy_pool_test_inventory 7") { + t.Fatalf("GET /metrics = %d %q", response.StatusCode, body) + } +} + +func TestHandlerFailsReadyWithoutLeakingDependencyError(t *testing.T) { + t.Parallel() + const secret = "redis://user:secret@redis:6379" + handler, err := NewHandler(Dependencies{ + Gatherer: prometheus.NewRegistry(), + Readiness: &recordingReadiness{err: errors.New(secret)}, + }) + if err != nil { + t.Fatalf("NewHandler() error = %v", err) + } + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil)) + if recorder.Code != http.StatusServiceUnavailable || strings.Contains(recorder.Body.String(), secret) { + t.Fatalf("GET /readyz = %d %q", recorder.Code, recorder.Body.String()) + } +} + +func TestHandlerRejectsInvalidDependenciesMethodsAndPaths(t *testing.T) { + t.Parallel() + if _, err := NewHandler(Dependencies{}); !errors.Is(err, ErrInvalidDependencies) { + t.Fatalf("NewHandler(empty) error = %v", err) + } + handler, err := NewHandler(Dependencies{ + Gatherer: prometheus.NewRegistry(), Readiness: &recordingReadiness{}, + }) + if err != nil { + t.Fatalf("NewHandler() error = %v", err) + } + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/livez", nil)) + if recorder.Code != http.StatusMethodNotAllowed || recorder.Header().Get("Allow") != http.MethodGet { + t.Fatalf("POST /livez = %d Allow=%q", recorder.Code, recorder.Header().Get("Allow")) + } + recorder = httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/missing", nil)) + if recorder.Code != http.StatusNotFound { + t.Fatalf("GET /missing = %d", recorder.Code) + } +} + +type recordingReadiness struct { + calls atomic.Int64 + err error +} + +func (readiness *recordingReadiness) Ready(context.Context) error { + readiness.calls.Add(1) + return readiness.err +} + +func assertProbe(t *testing.T, method, target string, wantStatus int, wantBody string) { + t.Helper() + request, err := http.NewRequest(method, target, nil) + if err != nil { + t.Fatalf("NewRequest(): %v", err) + } + response, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatalf("Do(%s): %v", target, err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("ReadAll(%s): %v", target, err) + } + if response.StatusCode != wantStatus || strings.TrimSpace(string(body)) != wantBody { + t.Fatalf("%s = %d %q", target, response.StatusCode, body) + } +}