diff --git a/go.mod b/go.mod index e7f12ba..49e4c05 100644 --- a/go.mod +++ b/go.mod @@ -109,8 +109,8 @@ require ( github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect - github.com/prometheus/client_model v0.6.2 // indirect - github.com/prometheus/common v0.67.5 // indirect + github.com/prometheus/client_model v0.6.2 + github.com/prometheus/common v0.67.5 github.com/prometheus/procfs v0.20.1 // indirect github.com/puzpuzpuz/xsync/v3 v3.5.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect diff --git a/pkg/metrics/README.md b/pkg/metrics/README.md index e63603b..078785a 100644 --- a/pkg/metrics/README.md +++ b/pkg/metrics/README.md @@ -48,11 +48,59 @@ metrics.SetProvider(provider) | `Namespace` | `string` | `""` | Prefix for all metric names | | `HTTPRequestBuckets` | `[]float64` | See below | Histogram buckets for HTTP duration (seconds) | | `DBQueryBuckets` | `[]float64` | See below | Histogram buckets for DB query duration (seconds) | +| `HTTPMaxPaths` | `int` | `1024` | Max distinct `path` label values; extras become `"other"` (negative disables) | +| `HTTPPathNormalizer` | `func(*http.Request) string` | `nil` | Custom request → `path` label mapping (return `""` to use the default) | + +**HTTP `path` label:** the middleware uses, in order: `HTTPPathNormalizer`, the matched `http.ServeMux` pattern (`r.Pattern`, e.g. `/users/{id}`), then the raw path with numeric/UUID/hex/opaque-token segments replaced by `:id`. For routers other than `ServeMux`, supply `HTTPPathNormalizer` with your route template. The `HTTPMaxPaths` cap applies on top. **Default HTTP Request Buckets:** `[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10]` **Default DB Query Buckets:** `[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5]` +### Enabled flag and JSON pull + +`Config.Enabled` is honoured: a disabled provider records nothing, `Middleware` passes requests straight through, `Handler()`/`JSONHandler()` answer 404, and push loops are not started (manual pushes return an error). Note a `&metrics.Config{}` literal has `Enabled: false`; use `DefaultConfig()` or set `Enabled: true`. `NewPrometheusProvider(nil)` is enabled. + +`provider.JSONHandler()` serves the same JSON as the push `json` format on `GET`/`HEAD`: + +```go +http.Handle("/metrics", provider.Handler()) // Prometheus text +http.Handle("/metrics.json", provider.JSONHandler()) // JSON +``` + +### Resetting Stats + +- `provider.Reset()` clears counters, histograms and the cache-size gauge (live gauges such as in-flight requests are kept). Package-level `metrics.Reset()` does the same for the current provider if it implements `metrics.Resetter`. +- `provider.PushAndReset()` pushes to the Pushgateway and resets only if the push succeeded (errors if no Pushgateway is configured). +- `Config.PushgatewayResetOnPush: true` makes the automatic push loop do this on every tick. +- `provider.ResetHandler()` is a `POST`-only endpoint (`?push=true` to push first). It has no auth: mount it on an internal route. + +```go +http.Handle("/metrics/reset", provider.ResetHandler()) +``` + +Note: the normal `/metrics` scrape is read-only and never clears anything. Observations recorded between a push and its reset are lost. Prometheus handles the counter drop as a reset, but if you reset often, prefer `increase()`/`rate()` over raw counter values. + +### Custom Push Endpoint (Optional) + +POST metrics to your own server, optionally clearing local stats after a 2xx reply: + +```go +provider := metrics.NewPrometheusProvider(&metrics.Config{ + PushEndpointURL: "https://collector.example.com/metrics", + PushEndpointFormat: "json", // or "text" (Prometheus exposition, default) + PushEndpointHeaders: map[string]string{"Authorization": "Bearer token"}, + PushEndpointInterval: 30, // seconds; 0 = manual only + PushEndpointTimeout: 10, // seconds (default 10) + PushEndpointResetOnSuccess: true, // clear local stats after a 2xx +}) + +err := provider.PushToEndpoint(ctx) // manual push; also honours ResetOnSuccess +provider.StopAutoPush() // stops the Pushgateway and endpoint loops +``` + +The `json` body is a list of `{name, help, type, metrics:[{labels, value | count, sum, buckets}]}`. Failures (non-2xx, network, timeout) are logged and never reset stats, so the next tick retries with the accumulated data. The payload covers everything in the default Prometheus registry, including Go runtime metrics. + ### Pushgateway Configuration (Optional) For batch jobs, cron tasks, or short-lived processes, you can push metrics to Prometheus Pushgateway: @@ -457,10 +505,9 @@ scrape_configs: - ✅ Good: `method`, `status_code` - ❌ Bad: `user_id`, `timestamp` -2. **Path Normalization**: Normalize dynamic paths +2. **Path Normalization**: Done automatically for the `path` label (see Configuration Options) ```go - // Instead of /api/users/123 - // Use /api/users/:id + // /api/users/123 is recorded as /api/users/:id ``` 3. **Metric Naming**: Follow Prometheus conventions diff --git a/pkg/metrics/config.go b/pkg/metrics/config.go index 8ccfdb6..db9f03c 100644 --- a/pkg/metrics/config.go +++ b/pkg/metrics/config.go @@ -1,5 +1,7 @@ package metrics +import "net/http" + // Config holds configuration for the metrics provider type Config struct { // Enabled determines whether metrics collection is enabled @@ -19,6 +21,17 @@ type Config struct { // Default: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5] DBQueryBuckets []float64 `mapstructure:"db_query_buckets"` + // HTTPMaxPaths caps the number of distinct values of the "path" label on HTTP + // metrics. Paths beyond the cap are reported as "other". Paths are already + // normalized (route pattern, or dynamic segments replaced with ":id"). + // Default: 1024. Set to a negative value to disable the cap. + HTTPMaxPaths int `mapstructure:"http_max_paths"` + + // HTTPPathNormalizer optionally maps a request to its "path" label (e.g. the + // matched route template of your router). Return "" to fall back to the + // default behaviour (ServeMux pattern, then generic ID normalization). + HTTPPathNormalizer func(*http.Request) string `mapstructure:"-"` + // PushgatewayURL is the URL of the Prometheus Pushgateway (optional) // If set, metrics will be pushed to this gateway instead of only being scraped // Example: "http://pushgateway:9091" @@ -32,6 +45,34 @@ type Config struct { // Only used if PushgatewayURL is set. If 0, automatic pushing is disabled. // Default: 0 (no automatic pushing) PushgatewayInterval int `mapstructure:"pushgateway_interval"` + + // PushEndpointURL is a custom HTTP endpoint that metrics are POSTed to + // (independent of Pushgateway). Example: "https://collector.example.com/metrics" + PushEndpointURL string `mapstructure:"push_endpoint_url"` + + // PushEndpointFormat is the request body format: "text" (Prometheus text + // exposition, Content-Type text/plain; version=0.0.4) or "json". + // Default: "text" + PushEndpointFormat string `mapstructure:"push_endpoint_format"` + + // PushEndpointHeaders are extra headers sent with each POST (e.g. Authorization). + PushEndpointHeaders map[string]string `mapstructure:"push_endpoint_headers"` + + // PushEndpointInterval is the interval in seconds for automatic POSTs. + // If 0, automatic posting is disabled (PushToEndpoint can still be called manually). + PushEndpointInterval int `mapstructure:"push_endpoint_interval"` + + // PushEndpointTimeout is the per-request timeout in seconds. Default: 10 + PushEndpointTimeout int `mapstructure:"push_endpoint_timeout"` + + // PushEndpointResetOnSuccess clears local counters and histograms after the + // endpoint answers with a 2xx status. Default: false. + PushEndpointResetOnSuccess bool `mapstructure:"push_endpoint_reset_on_success"` + + // PushgatewayResetOnPush clears the local counters and histograms after each + // successful push (automatic or via PushAndReset), so each push carries only + // the activity since the previous one. Default: false. + PushgatewayResetOnPush bool `mapstructure:"pushgateway_reset_on_push"` } // DefaultConfig returns a Config with sensible defaults @@ -43,6 +84,7 @@ func DefaultConfig() *Config { HTTPRequestBuckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, // DB queries are usually faster DBQueryBuckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5}, + HTTPMaxPaths: defaultHTTPMaxPaths, } } @@ -57,6 +99,17 @@ func (c *Config) ApplyDefaults() { if len(c.DBQueryBuckets) == 0 { c.DBQueryBuckets = []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5} } + if c.PushEndpointURL != "" { + if c.PushEndpointFormat == "" { + c.PushEndpointFormat = "text" + } + if c.PushEndpointTimeout <= 0 { + c.PushEndpointTimeout = 10 + } + } + if c.HTTPMaxPaths == 0 { + c.HTTPMaxPaths = defaultHTTPMaxPaths + } // Set default job name if pushgateway is configured but job name is empty if c.PushgatewayURL != "" && c.PushgatewayJobName == "" { c.PushgatewayJobName = "resolvespec" diff --git a/pkg/metrics/endpoint.go b/pkg/metrics/endpoint.go new file mode 100644 index 0000000..73dada9 --- /dev/null +++ b/pkg/metrics/endpoint.go @@ -0,0 +1,189 @@ +package metrics + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "sync" + "time" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/expfmt" + + "github.com/bitechdev/ResolveSpec/pkg/logger" +) + +const textContentType = "text/plain; version=0.0.4; charset=utf-8" + +// endpointPusher POSTs gathered metrics to a user-configured HTTP endpoint. +type endpointPusher struct { + url string + format string + headers map[string]string + client *http.Client + resetOnOK bool + provider *PrometheusProvider + gatherer prometheus.Gatherer + stopOnce sync.Once + stopCh chan struct{} + startedMu sync.Mutex + started bool +} + +func newEndpointPusher(cfg *Config, p *PrometheusProvider) *endpointPusher { + return &endpointPusher{ + url: cfg.PushEndpointURL, + format: cfg.PushEndpointFormat, + headers: cfg.PushEndpointHeaders, + client: &http.Client{Timeout: time.Duration(cfg.PushEndpointTimeout) * time.Second}, + resetOnOK: cfg.PushEndpointResetOnSuccess, + provider: p, + gatherer: prometheus.DefaultGatherer, + stopCh: make(chan struct{}), + } +} + +func (e *endpointPusher) start(interval time.Duration) { + e.startedMu.Lock() + defer e.startedMu.Unlock() + if e.started { + return + } + e.started = true + go func() { + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-t.C: + if err := e.push(context.Background()); err != nil { + logger.Warn("Failed to push metrics to endpoint %s: %v", e.url, err) + } + case <-e.stopCh: + return + } + } + }() +} + +func (e *endpointPusher) stop() { + e.stopOnce.Do(func() { close(e.stopCh) }) +} + +func (e *endpointPusher) push(ctx context.Context) error { + mfs, err := e.gatherer.Gather() + if err != nil && len(mfs) == 0 { + return fmt.Errorf("gather metrics: %w", err) + } + + body, contentType, err := encodeMetrics(mfs, e.format) + if err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, e.url, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", contentType) + for k, v := range e.headers { + req.Header.Set(k, v) + } + + resp, err := e.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + if resp.StatusCode < 200 || resp.StatusCode > 299 { + return fmt.Errorf("endpoint returned %s: %s", resp.Status, bytes.TrimSpace(snippet)) + } + + if e.resetOnOK { + e.provider.Reset() + } + return nil +} + +func encodeMetrics(mfs []*dto.MetricFamily, format string) (body []byte, contentType string, err error) { + switch format { + case "json": + b, err := json.Marshal(toJSONFamilies(mfs)) + return b, "application/json", err + case "", "text": + var buf bytes.Buffer + enc := expfmt.NewEncoder(&buf, expfmt.NewFormat(expfmt.TypeTextPlain)) + for _, mf := range mfs { + if err := enc.Encode(mf); err != nil { + return nil, "", err + } + } + return buf.Bytes(), textContentType, nil + default: + return nil, "", fmt.Errorf("unsupported push endpoint format %q", format) + } +} + +type jsonFamily struct { + Name string `json:"name"` + Help string `json:"help,omitempty"` + Type string `json:"type"` + Metrics []jsonMetric `json:"metrics"` +} + +type jsonMetric struct { + Labels map[string]string `json:"labels,omitempty"` + Value *float64 `json:"value,omitempty"` + Count *uint64 `json:"count,omitempty"` + Sum *float64 `json:"sum,omitempty"` + Buckets []jsonBucket `json:"buckets,omitempty"` +} + +type jsonBucket struct { + UpperBound float64 `json:"le"` + Count uint64 `json:"count"` +} + +func toJSONFamilies(mfs []*dto.MetricFamily) []jsonFamily { + out := make([]jsonFamily, 0, len(mfs)) + for _, mf := range mfs { + f := jsonFamily{Name: mf.GetName(), Help: mf.GetHelp(), Type: mf.GetType().String()} + for _, m := range mf.GetMetric() { + jm := jsonMetric{} + if len(m.GetLabel()) > 0 { + jm.Labels = make(map[string]string, len(m.GetLabel())) + for _, l := range m.GetLabel() { + jm.Labels[l.GetName()] = l.GetValue() + } + } + switch { + case m.Counter != nil: + v := m.Counter.GetValue() + jm.Value = &v + case m.Gauge != nil: + v := m.Gauge.GetValue() + jm.Value = &v + case m.Untyped != nil: + v := m.Untyped.GetValue() + jm.Value = &v + case m.Histogram != nil: + c, s := m.Histogram.GetSampleCount(), m.Histogram.GetSampleSum() + jm.Count, jm.Sum = &c, &s + for _, b := range m.Histogram.GetBucket() { + jm.Buckets = append(jm.Buckets, jsonBucket{UpperBound: b.GetUpperBound(), Count: b.GetCumulativeCount()}) + } + case m.Summary != nil: + c, s := m.Summary.GetSampleCount(), m.Summary.GetSampleSum() + jm.Count, jm.Sum = &c, &s + } + f.Metrics = append(f.Metrics, jm) + } + out = append(out, f) + } + return out +} diff --git a/pkg/metrics/interfaces.go b/pkg/metrics/interfaces.go index c040297..5c0cdb1 100644 --- a/pkg/metrics/interfaces.go +++ b/pkg/metrics/interfaces.go @@ -47,6 +47,21 @@ type Provider interface { Handler() http.Handler } +// Resetter is optionally implemented by providers that can clear their recorded stats. +type Resetter interface { + Reset() +} + +// Reset clears the current provider's stats if it supports resetting. +// It returns false if the provider does not implement Resetter. +func Reset() bool { + if r, ok := GetProvider().(Resetter); ok { + r.Reset() + return true + } + return false +} + // globalProvider is the global metrics provider, protected by globalProviderMu. var ( globalProviderMu sync.RWMutex diff --git a/pkg/metrics/path.go b/pkg/metrics/path.go new file mode 100644 index 0000000..6bce8d9 --- /dev/null +++ b/pkg/metrics/path.go @@ -0,0 +1,175 @@ +package metrics + +import ( + "net/http" + "strings" + "sync" +) + +const ( + // defaultHTTPMaxPaths is the default cap on distinct values of the "path" label. + defaultHTTPMaxPaths = 1024 + + // overflowPathLabel is used once the cap on distinct path labels is reached. + overflowPathLabel = "other" +) + +// routeLabel returns the low-cardinality path label for a request, preferring +// (in order): the custom normalizer, the matched ServeMux pattern, and finally +// the generic normalization of the raw URL path. +func routeLabel(r *http.Request, custom func(*http.Request) string) string { + if custom != nil { + if p := custom(r); p != "" { + return p + } + } + if r.Pattern != "" { + return stripPatternMethod(r.Pattern) + } + return NormalizePath(r.URL.Path) +} + +// stripPatternMethod removes the optional "METHOD " prefix (and host) from a +// Go 1.22+ ServeMux pattern, e.g. "GET /users/{id}" -> "/users/{id}". +func stripPatternMethod(pattern string) string { + if i := strings.IndexByte(pattern, ' '); i >= 0 { + pattern = strings.TrimLeft(pattern[i+1:], " ") + } + if i := strings.IndexByte(pattern, '/'); i > 0 { + pattern = pattern[i:] // drop host part + } + return pattern +} + +// NormalizePath replaces dynamic-looking path segments (numeric IDs, UUIDs, +// long hex strings and other long opaque tokens) with ":id" so that +// /users/123 and /users/456 share one label value. +func NormalizePath(path string) string { + if path == "" { + return "/" + } + if !strings.Contains(path, "/") { + return path + } + segs := strings.Split(path, "/") + for i, s := range segs { + if isDynamicSegment(s) { + segs[i] = ":id" + } + } + return strings.Join(segs, "/") +} + +func isDynamicSegment(s string) bool { + if s == "" { + return false + } + if allDigits(s) { + return true + } + if isUUID(s) { + return true + } + // Long hex strings (hashes, object IDs) + if len(s) >= 16 && allHex(s) { + return true + } + // Long opaque tokens containing digits (base64/ULID-like) + if len(s) >= 24 && hasDigit(s) && !strings.ContainsAny(s, ".") { + return true + } + return false +} + +func allDigits(s string) bool { + for i := 0; i < len(s); i++ { + if s[i] < '0' || s[i] > '9' { + return false + } + } + return true +} + +func hasDigit(s string) bool { + for i := 0; i < len(s); i++ { + if s[i] >= '0' && s[i] <= '9' { + return true + } + } + return false +} + +func allHex(s string) bool { + for i := 0; i < len(s); i++ { + c := s[i] + if !isHexByte(c) { + return false + } + } + return true +} + +func isUUID(s string) bool { + if len(s) != 36 { + return false + } + for i := 0; i < len(s); i++ { + c := s[i] + switch i { + case 8, 13, 18, 23: + if c != '-' { + return false + } + default: + if !isHexByte(c) { + return false + } + } + } + return true +} + +// pathLimiter bounds the number of distinct path label values. Once the cap is +// reached, unseen paths are reported as "other". +type pathLimiter struct { + mu sync.RWMutex + max int // <= 0 disables the cap + seen map[string]struct{} +} + +func newPathLimiter(limit int) *pathLimiter { + return &pathLimiter{max: limit, seen: make(map[string]struct{})} +} + +func (l *pathLimiter) label(path string) string { + if l.max <= 0 { + return path + } + l.mu.RLock() + _, ok := l.seen[path] + l.mu.RUnlock() + if ok { + return path + } + + l.mu.Lock() + defer l.mu.Unlock() + if _, ok := l.seen[path]; ok { + return path + } + if len(l.seen) >= l.max { + return overflowPathLabel + } + l.seen[path] = struct{}{} + return path +} + +func (l *pathLimiter) reset() { + l.mu.Lock() + l.seen = make(map[string]struct{}) + l.mu.Unlock() +} + +func isHexByte(c byte) bool { + return c >= '0' && c <= '9' || c >= 'a' && c <= 'f' || c >= 'A' && c <= 'F' +} diff --git a/pkg/metrics/path_test.go b/pkg/metrics/path_test.go new file mode 100644 index 0000000..f0233c3 --- /dev/null +++ b/pkg/metrics/path_test.go @@ -0,0 +1,237 @@ +package metrics + +import ( + "context" + "encoding/json" + "github.com/prometheus/client_golang/prometheus" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestNormalizePath(t *testing.T) { + cases := map[string]string{ + "": "/", + "/": "/", + "/users": "/users", + "/users/123": "/users/:id", + "/users/123/orders/9": "/users/:id/orders/:id", + "/x/550e8400-e29b-41d4-a716-446655440000": "/x/:id", + "/x/507f1f77bcf86cd799439011": "/x/:id", + "/api/public/users": "/api/public/users", + "/files/report.v2": "/files/report.v2", + } + for in, want := range cases { + if got := NormalizePath(in); got != want { + t.Errorf("NormalizePath(%q) = %q, want %q", in, got, want) + } + } +} + +func TestRouteLabel(t *testing.T) { + r := httptest.NewRequest("GET", "/users/42", nil) + if got := routeLabel(r, nil); got != "/users/:id" { + t.Errorf("fallback = %q", got) + } + r.Pattern = "GET /users/{id}" + if got := routeLabel(r, nil); got != "/users/{id}" { + t.Errorf("pattern = %q", got) + } + got := routeLabel(r, func(*http.Request) string { return "/custom" }) + if got != "/custom" { + t.Errorf("custom = %q", got) + } +} + +func TestPathLimiter(t *testing.T) { + l := newPathLimiter(2) + for _, p := range []string{"/a", "/b", "/a"} { + if got := l.label(p); got != p { + t.Errorf("label(%q) = %q", p, got) + } + } + if got := l.label("/c"); got != overflowPathLabel { + t.Errorf("overflow = %q", got) + } + if got := newPathLimiter(-1).label("/z"); got != "/z" { + t.Errorf("disabled = %q", got) + } +} + +func TestMiddlewareUsesPattern(t *testing.T) { + p := NewPrometheusProvider(&Config{Enabled: true, Namespace: "pathtest"}) + mux := http.NewServeMux() + mux.HandleFunc("GET /users/{id}", func(w http.ResponseWriter, r *http.Request) {}) + h := p.Middleware(mux) + for _, id := range []string{"1", "2", "abc"} { + h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", "/users/"+id, nil)) + } + if n := len(p.pathLimiter.seen); n != 1 { + t.Errorf("distinct paths = %d, want 1", n) + } + if _, ok := p.pathLimiter.seen["/users/{id}"]; !ok { + t.Errorf("seen = %v", p.pathLimiter.seen) + } +} + +func TestResetAndHandler(t *testing.T) { + p := NewPrometheusProvider(&Config{Enabled: true, Namespace: "resettest"}) + p.RecordHTTPRequest("GET", "/a/1", "200", 0) + p.RecordDBQuery("SELECT", "s", "e", "t", 0, nil) + p.IncRequestsInFlight() + + count := func() int { + mfs, _ := prometheus.DefaultGatherer.Gather() + n := 0 + for _, mf := range mfs { + if strings.HasPrefix(mf.GetName(), "resettest_") && mf.GetName() != "resettest_http_requests_in_flight" && mf.GetName() != "resettest_event_queue_size" { + n += len(mf.GetMetric()) + } + } + return n + } + if count() == 0 { + t.Fatal("expected recorded series") + } + + rec := httptest.NewRecorder() + p.ResetHandler().ServeHTTP(rec, httptest.NewRequest("GET", "/reset", nil)) + if rec.Code != http.StatusMethodNotAllowed || count() == 0 { + t.Fatalf("GET should be rejected, code=%d", rec.Code) + } + + rec = httptest.NewRecorder() + p.ResetHandler().ServeHTTP(rec, httptest.NewRequest("POST", "/reset", nil)) + if rec.Code != http.StatusNoContent || count() != 0 { + t.Fatalf("reset failed, code=%d series=%d", rec.Code, count()) + } + if len(p.pathLimiter.seen) != 0 { + t.Error("path limiter not reset") + } + + // push=true without a pushgateway must fail and not be silent + rec = httptest.NewRecorder() + p.ResetHandler().ServeHTTP(rec, httptest.NewRequest("POST", "/reset?push=true", nil)) + if rec.Code != http.StatusBadGateway { + t.Errorf("push without gateway code=%d", rec.Code) + } +} + +func TestPushAndResetKeepsStatsOnFailure(t *testing.T) { + p := NewPrometheusProvider(&Config{Enabled: true, Namespace: "pushfail", PushgatewayURL: "http://127.0.0.1:1"}) + p.RecordHTTPRequest("GET", "/a", "200", 0) + if err := p.PushAndReset(); err == nil { + t.Fatal("expected push error") + } + if len(p.pathLimiter.seen) != 1 { + t.Error("stats were reset despite failed push") + } +} + +func TestPushToEndpoint(t *testing.T) { + for _, format := range []string{"text", "json"} { + var gotCT, gotAuth string + var gotBody []byte + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("method = %s", r.Method) + } + gotCT, gotAuth = r.Header.Get("Content-Type"), r.Header.Get("Authorization") + gotBody, _ = io.ReadAll(r.Body) + })) + + ns := "ep" + format + p := NewPrometheusProvider(&Config{ + Enabled: true, + Namespace: ns, + PushEndpointURL: srv.URL, + PushEndpointFormat: format, + PushEndpointHeaders: map[string]string{"Authorization": "Bearer x"}, + PushEndpointResetOnSuccess: true, + }) + p.RecordHTTPRequest("GET", "/a", "200", time.Millisecond) + + if err := p.PushToEndpoint(context.Background()); err != nil { + t.Fatalf("%s: %v", format, err) + } + srv.Close() + if gotAuth != "Bearer x" || !strings.Contains(string(gotBody), ns+"_http_requests_total") { + t.Errorf("%s: auth=%q body=%.200s", format, gotAuth, gotBody) + } + if format == "json" && gotCT != "application/json" || format == "text" && !strings.HasPrefix(gotCT, "text/plain") { + t.Errorf("%s: content-type %q", format, gotCT) + } + if len(p.pathLimiter.seen) != 0 { + t.Errorf("%s: stats not reset after success", format) + } + } +} + +func TestPushToEndpointFailureKeepsStats(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "nope", http.StatusInternalServerError) + })) + defer srv.Close() + p := NewPrometheusProvider(&Config{Enabled: true, Namespace: "epfail", PushEndpointURL: srv.URL, PushEndpointResetOnSuccess: true}) + p.RecordHTTPRequest("GET", "/a", "200", 0) + if err := p.PushToEndpoint(context.Background()); err == nil { + t.Fatal("expected error on 500") + } + if len(p.pathLimiter.seen) != 1 { + t.Error("stats reset despite failure") + } + if err := NewPrometheusProvider(&Config{Enabled: true, Namespace: "epnone"}).PushToEndpoint(context.Background()); err == nil { + t.Error("expected error without endpoint") + } +} + +func TestDisabledProvider(t *testing.T) { + p := NewPrometheusProvider(&Config{Namespace: "disabled", PushEndpointURL: "http://127.0.0.1:1", PushEndpointInterval: 1}) + p.RecordHTTPRequest("GET", "/a", "200", 0) + if len(p.pathLimiter.seen) != 0 { + t.Error("disabled provider recorded") + } + for name, h := range map[string]http.Handler{"handler": p.Handler(), "json": p.JSONHandler()} { + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest("GET", "/m", nil)) + if rec.Code != http.StatusNotFound { + t.Errorf("%s code=%d", name, rec.Code) + } + } + if p.endpoint != nil || p.PushToEndpoint(context.Background()) == nil || p.Push() == nil { + t.Error("disabled provider must not push") + } +} + +func TestJSONHandler(t *testing.T) { + p := NewPrometheusProvider(&Config{Enabled: true, Namespace: "jsonpull"}) + p.RecordHTTPRequest("GET", "/a", "200", time.Millisecond) + + rec := httptest.NewRecorder() + p.JSONHandler().ServeHTTP(rec, httptest.NewRequest("GET", "/m", nil)) + if rec.Code != 200 || rec.Header().Get("Content-Type") != "application/json" { + t.Fatalf("code=%d ct=%q", rec.Code, rec.Header().Get("Content-Type")) + } + var fams []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &fams); err != nil { + t.Fatal(err) + } + found := false + for _, f := range fams { + if f["name"] == "jsonpull_http_requests_total" { + found = true + } + } + if !found { + t.Error("metric family missing from JSON") + } + + rec = httptest.NewRecorder() + p.JSONHandler().ServeHTTP(rec, httptest.NewRequest("POST", "/m", nil)) + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("POST code=%d", rec.Code) + } +} diff --git a/pkg/metrics/prometheus.go b/pkg/metrics/prometheus.go index 8f08b8e..6c54db9 100644 --- a/pkg/metrics/prometheus.go +++ b/pkg/metrics/prometheus.go @@ -1,6 +1,8 @@ package metrics import ( + "context" + "errors" "net/http" "strconv" "time" @@ -9,8 +11,12 @@ import ( "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/prometheus/client_golang/prometheus/push" + + "github.com/bitechdev/ResolveSpec/pkg/logger" ) +var errMetricsDisabled = errors.New("metrics: disabled") + // PrometheusProvider implements the Provider interface using Prometheus type PrometheusProvider struct { requestDuration *prometheus.HistogramVec @@ -27,9 +33,16 @@ type PrometheusProvider struct { eventQueueSize prometheus.Gauge panicsTotal *prometheus.CounterVec + pathLimiter *pathLimiter + pathNormalizer func(*http.Request) string + + enabled bool + endpoint *endpointPusher + // Pushgateway fields (optional) pushgatewayURL string pushgatewayJobName string + resetOnPush bool pusher *push.Pusher pushTicker *time.Ticker pushStop chan bool @@ -55,6 +68,7 @@ func NewPrometheusProvider(cfg *Config) *PrometheusProvider { } p := &PrometheusProvider{ + enabled: cfg.Enabled, requestDuration: promauto.NewHistogramVec( prometheus.HistogramOpts{ Name: metricName("http_request_duration_seconds"), @@ -149,12 +163,17 @@ func NewPrometheusProvider(cfg *Config) *PrometheusProvider { []string{"method"}, ), + pathLimiter: newPathLimiter(cfg.HTTPMaxPaths), + pathNormalizer: cfg.HTTPPathNormalizer, + pushgatewayURL: cfg.PushgatewayURL, pushgatewayJobName: cfg.PushgatewayJobName, + resetOnPush: cfg.PushgatewayResetOnPush, } // Initialize pushgateway if configured - if cfg.PushgatewayURL != "" { + // Pushing is never started for a disabled provider + if cfg.PushgatewayURL != "" && cfg.Enabled { p.pusher = push.New(cfg.PushgatewayURL, cfg.PushgatewayJobName). Gatherer(prometheus.DefaultGatherer) @@ -166,6 +185,13 @@ func NewPrometheusProvider(cfg *Config) *PrometheusProvider { } } + if cfg.PushEndpointURL != "" && cfg.Enabled { + p.endpoint = newEndpointPusher(cfg, p) + if cfg.PushEndpointInterval > 0 { + p.endpoint.start(time.Duration(cfg.PushEndpointInterval) * time.Second) + } + } + return p } @@ -188,23 +214,37 @@ func (rw *ResponseWriter) WriteHeader(code int) { } // RecordHTTPRequest implements Provider interface +// The path is normalized and capped to keep label cardinality bounded. func (p *PrometheusProvider) RecordHTTPRequest(method, path, status string, duration time.Duration) { + if !p.enabled { + return + } + path = p.pathLimiter.label(NormalizePath(path)) p.requestDuration.WithLabelValues(method, path, status).Observe(duration.Seconds()) p.requestTotal.WithLabelValues(method, path, status).Inc() } // IncRequestsInFlight implements Provider interface func (p *PrometheusProvider) IncRequestsInFlight() { + if !p.enabled { + return + } p.requestsInFlight.Inc() } // DecRequestsInFlight implements Provider interface func (p *PrometheusProvider) DecRequestsInFlight() { + if !p.enabled { + return + } p.requestsInFlight.Dec() } // RecordDBQuery implements Provider interface func (p *PrometheusProvider) RecordDBQuery(operation, schema, entity, table string, duration time.Duration, err error) { + if !p.enabled { + return + } status := "success" if err != nil { status = "error" @@ -215,47 +255,115 @@ func (p *PrometheusProvider) RecordDBQuery(operation, schema, entity, table stri // RecordCacheHit implements Provider interface func (p *PrometheusProvider) RecordCacheHit(provider string) { + if !p.enabled { + return + } p.cacheHits.WithLabelValues(provider).Inc() } // RecordCacheMiss implements Provider interface func (p *PrometheusProvider) RecordCacheMiss(provider string) { + if !p.enabled { + return + } p.cacheMisses.WithLabelValues(provider).Inc() } // UpdateCacheSize implements Provider interface func (p *PrometheusProvider) UpdateCacheSize(provider string, size int64) { + if !p.enabled { + return + } p.cacheSize.WithLabelValues(provider).Set(float64(size)) } // RecordEventPublished implements Provider interface func (p *PrometheusProvider) RecordEventPublished(source, eventType string) { + if !p.enabled { + return + } p.eventPublished.WithLabelValues(source, eventType).Inc() } // RecordEventProcessed implements Provider interface func (p *PrometheusProvider) RecordEventProcessed(source, eventType, status string, duration time.Duration) { + if !p.enabled { + return + } p.eventProcessed.WithLabelValues(source, eventType, status).Inc() p.eventDuration.WithLabelValues(source, eventType).Observe(duration.Seconds()) } // UpdateEventQueueSize implements Provider interface func (p *PrometheusProvider) UpdateEventQueueSize(size int64) { + if !p.enabled { + return + } p.eventQueueSize.Set(float64(size)) } // RecordPanic implements the Provider interface func (p *PrometheusProvider) RecordPanic(methodName string) { + if !p.enabled { + return + } p.panicsTotal.WithLabelValues(methodName).Inc() } // Handler implements Provider interface +// It responds 404 when metrics are disabled. func (p *PrometheusProvider) Handler() http.Handler { + if !p.enabled { + return disabledHandler() + } return promhttp.Handler() } +// JSONHandler returns an HTTP handler serving the current metrics as JSON +// (same shape as the "json" push endpoint format). Only GET and HEAD are +// accepted, and it responds 404 when metrics are disabled. It performs no +// authentication; mount it on an internal/protected route. +func (p *PrometheusProvider) JSONHandler() http.Handler { + if !p.enabled { + return disabledHandler() + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + w.Header().Set("Allow", "GET, HEAD") + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + mfs, err := prometheus.DefaultGatherer.Gather() + if err != nil && len(mfs) == 0 { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + body, contentType, err := encodeMetrics(mfs, "json") + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", contentType) + if r.Method == http.MethodGet { + if _, err := w.Write(body); err != nil { + logger.Warn("Failed to write metrics JSON: %v", err) + } + } + }) +} + +func disabledHandler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "metrics disabled", http.StatusNotFound) + }) +} + // Middleware returns an HTTP middleware that collects metrics +// When metrics are disabled it returns next unchanged. func (p *PrometheusProvider) Middleware(next http.Handler) http.Handler { + if !p.enabled { + return next + } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { start := time.Now() @@ -273,13 +381,17 @@ func (p *PrometheusProvider) Middleware(next http.Handler) http.Handler { duration := time.Since(start) status := strconv.Itoa(rw.statusCode) - p.RecordHTTPRequest(r.Method, r.URL.Path, status, duration) + // Read the label after next has run so the router has set r.Pattern. + p.RecordHTTPRequest(r.Method, routeLabel(r, p.pathNormalizer), status, duration) }) } // Push manually pushes metrics to the configured Pushgateway // Returns an error if pushing fails or if Pushgateway is not configured func (p *PrometheusProvider) Push() error { + if !p.enabled { + return errMetricsDisabled + } if p.pusher == nil { return nil // Pushgateway not configured, silently skip } @@ -291,10 +403,15 @@ func (p *PrometheusProvider) startAutoPush() { for { select { case <-p.pushTicker.C: - if err := p.Push(); err != nil { - // Log error but continue pushing - // Note: In production, you might want to use a proper logger - _ = err + var err error + if p.resetOnPush { + err = p.PushAndReset() + } else { + err = p.Push() + } + if err != nil { + // Log and keep going; the next tick retries (and nothing was reset) + logger.Warn("Failed to push metrics to Pushgateway: %v", err) } case <-p.pushStop: p.pushTicker.Stop() @@ -303,10 +420,87 @@ func (p *PrometheusProvider) startAutoPush() { } } +// Reset clears all recorded counters, histograms and labelled gauges (cache size) +// and forgets the tracked HTTP path labels. Live gauges (requests in flight, +// event queue size) are left untouched since they reflect current state. +// Prometheus treats the drop in counters as a counter reset, so rate() and +// increase() keep working on the scraper side. +func (p *PrometheusProvider) Reset() { + p.requestDuration.Reset() + p.requestTotal.Reset() + p.dbQueryDuration.Reset() + p.dbQueryTotal.Reset() + p.cacheHits.Reset() + p.cacheMisses.Reset() + p.cacheSize.Reset() + p.eventPublished.Reset() + p.eventProcessed.Reset() + p.eventDuration.Reset() + p.panicsTotal.Reset() + p.pathLimiter.reset() +} + +// PushAndReset pushes metrics to the Pushgateway and, only if the push +// succeeded, clears the local stats. Returns an error if Pushgateway is not +// configured, so stats are never discarded without being delivered. Observations +// recorded between the push and the reset are lost. +func (p *PrometheusProvider) PushAndReset() error { + if !p.enabled { + return errMetricsDisabled + } + if p.pusher == nil { + return errors.New("metrics: pushgateway not configured, refusing to reset") + } + if err := p.pusher.Push(); err != nil { + return err + } + p.Reset() + return nil +} + +// PushToEndpoint POSTs the current metrics to the configured PushEndpointURL. +// If PushEndpointResetOnSuccess is set, local stats are cleared after a 2xx reply. +// Returns an error if no endpoint is configured. +func (p *PrometheusProvider) PushToEndpoint(ctx context.Context) error { + if !p.enabled { + return errMetricsDisabled + } + if p.endpoint == nil { + return errors.New("metrics: push endpoint not configured") + } + return p.endpoint.push(ctx) +} + +// ResetHandler returns an HTTP handler that clears local stats on POST. +// With ?push=true it first pushes to the Pushgateway and only resets on success. +// The handler performs no authentication; mount it on an internal/protected route. +func (p *PrometheusProvider) ResetHandler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if r.URL.Query().Get("push") == "true" { + if err := p.PushAndReset(); err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + } else { + p.Reset() + } + w.WriteHeader(http.StatusNoContent) + }) +} + // StopAutoPush stops the automatic push goroutine // This should be called when shutting down the application func (p *PrometheusProvider) StopAutoPush() { if p.pushStop != nil { close(p.pushStop) + p.pushStop = nil + } + if p.endpoint != nil { + p.endpoint.stop() } }