From 04a9fa89a0f620957085cd25e0fb16223e963856 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 7 Aug 2026 17:12:57 +0800 Subject: [PATCH] feat: observe gateway request durations --- deploy/README.md | 2 +- deploy/compose_test.go | 6 ++- .../dashboards/proxy-pool-overview.json | 4 +- internal/gateway/server/handler.go | 13 +++++- internal/gateway/server/handler_test.go | 32 +++++++++++--- internal/platform/metrics/gateway.go | 44 +++++++++++++++++++ internal/platform/metrics/gateway_test.go | 25 +++++++++++ progress.md | 2 + 8 files changed, 117 insertions(+), 11 deletions(-) diff --git a/deploy/README.md b/deploy/README.md index d917b83..dcdd437 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -36,6 +36,6 @@ PostgreSQL Adapter 和对应执行脚本完成前，不把数据库契约记为 并完成代表性集群容量验证。 Grafana Overview 与 Prometheus 规则只引用代码已注册的低基数指标。它们覆盖 Gateway -请求/Outcome、Controller 容量、Provider、Extraction、Checker 和 Drain；不按 Proxy、IP、 +请求/端到端时延/Outcome、Controller 容量、Provider、Extraction、Checker 和 Drain；不按 Proxy、IP、 Client、Upstream、Worker、Session 或完整 URL 聚合。`go test ./deploy` 会解析两类资产并拒绝 不存在的指标与禁止标签，指标改名或新增面板时必须同步更新该契约。 diff --git a/deploy/compose_test.go b/deploy/compose_test.go index bea2cd8..f30072d 100644 --- a/deploy/compose_test.go +++ b/deploy/compose_test.go @@ -33,6 +33,7 @@ var observableMetricNames = map[string]struct{}{ "proxy_pool_gateway_capacity_invariant_violations_total": {}, "proxy_pool_gateway_outcome_queue_dropped_total": {}, "proxy_pool_gateway_outcomes_total": {}, + "proxy_pool_gateway_request_duration_seconds": {}, "proxy_pool_gateway_requests_in_flight": {}, "proxy_pool_gateway_requests_total": {}, } @@ -305,7 +306,8 @@ func TestObservabilityAssetsUseRegisteredLowCardinalityMetrics(t *testing.T) { forbiddenLabel := regexp.MustCompile(`(?:by\s*\([^)]*\b(?:upstream|worker)\b|\{[^}]*\b(?:upstream|worker)\s*=)`) for _, expression := range expressions { for _, name := range metricPattern.FindAllString(expression, -1) { - if _, exists := observableMetricNames[name]; !exists { + baseName := strings.TrimSuffix(strings.TrimSuffix(strings.TrimSuffix(name, "_bucket"), "_count"), "_sum") + if _, exists := observableMetricNames[baseName]; !exists { t.Errorf("observability expression references unregistered metric %q: %s", name, expression) } } @@ -314,7 +316,7 @@ func TestObservabilityAssetsUseRegisteredLowCardinalityMetrics(t *testing.T) { } } for _, required := range []string{ - "proxy_pool_gateway_requests_total", "proxy_pool_controller_capacity_available_slots", + "proxy_pool_gateway_requests_total", "proxy_pool_gateway_request_duration_seconds", "proxy_pool_controller_capacity_available_slots", "proxy_pool_controller_provider_fetch_results_total", "proxy_pool_controller_extraction_requests_total", "proxy_pool_checker_observations_total", } { diff --git a/deploy/grafana/dashboards/proxy-pool-overview.json b/deploy/grafana/dashboards/proxy-pool-overview.json index baca390..1195709 100644 --- a/deploy/grafana/dashboards/proxy-pool-overview.json +++ b/deploy/grafana/dashboards/proxy-pool-overview.json @@ -4,8 +4,8 @@ "graphTooltip": 1, "panels": [ {"type":"timeseries","title":"Gateway QPS","gridPos":{"h":8,"w":8,"x":0,"y":0},"targets":[{"expr":"sum by (protocol) (rate(proxy_pool_gateway_requests_total[1m]))","legendFormat":"{{protocol}}"}]}, - {"type":"timeseries","title":"Gateway In Flight","gridPos":{"h":8,"w":8,"x":8,"y":0},"targets":[{"expr":"sum by (protocol) (proxy_pool_gateway_requests_in_flight)","legendFormat":"{{protocol}}"}]}, - {"type":"timeseries","title":"Active CONNECT Tunnels","gridPos":{"h":8,"w":8,"x":16,"y":0},"targets":[{"expr":"sum(proxy_pool_gateway_active_tunnels)","legendFormat":"tunnels"}]}, + {"type":"timeseries","title":"Gateway p99","gridPos":{"h":8,"w":8,"x":8,"y":0},"targets":[{"expr":"histogram_quantile(0.99, sum by (le, protocol) (rate(proxy_pool_gateway_request_duration_seconds_bucket[5m])))","legendFormat":"{{protocol}}"}]}, + {"type":"timeseries","title":"Gateway In Flight and Tunnels","gridPos":{"h":8,"w":8,"x":16,"y":0},"targets":[{"expr":"sum by (protocol) (proxy_pool_gateway_requests_in_flight)","legendFormat":"in-flight {{protocol}}"},{"expr":"sum(proxy_pool_gateway_active_tunnels)","legendFormat":"tunnels"}]}, {"type":"timeseries","title":"Gateway Outcome Failures","gridPos":{"h":8,"w":12,"x":0,"y":8},"targets":[{"expr":"sum by (stage) (rate(proxy_pool_gateway_outcomes_total{result=\"failure\"}[5m]))","legendFormat":"{{stage}}"}]}, {"type":"timeseries","title":"Gateway Outcome Queue Drops","gridPos":{"h":8,"w":12,"x":12,"y":8},"targets":[{"expr":"sum(rate(proxy_pool_gateway_outcome_queue_dropped_total[5m]))","legendFormat":"drops/s"}]}, {"type":"timeseries","title":"Controller Capacity","gridPos":{"h":8,"w":12,"x":0,"y":16},"targets":[{"expr":"sum(proxy_pool_controller_capacity_available_slots)","legendFormat":"available slots"},{"expr":"sum(proxy_pool_controller_capacity_effective_slots)","legendFormat":"effective slots"},{"expr":"sum(proxy_pool_controller_capacity_pending_expected_proxies)","legendFormat":"pending proxies"}]}, diff --git a/internal/gateway/server/handler.go b/internal/gateway/server/handler.go index d341849..d6b9280 100644 --- a/internal/gateway/server/handler.go +++ b/internal/gateway/server/handler.go @@ -91,6 +91,7 @@ type OutcomeRecorder interface { type RequestMetricsObserver interface { ObserveRequestStarted(protocol string) ObserveRequestFinished(protocol string) + ObserveRequestDuration(protocol string, duration time.Duration) ObserveTunnelOpened() ObserveTunnelClosed() } @@ -182,8 +183,12 @@ func (handler *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Requ } defer handler.finishRequest() protocol := requestMetricProtocol(request) + started := time.Now() handler.observeRequestStarted(protocol) - defer handler.observeRequestFinished(protocol) + defer func() { + handler.observeRequestFinished(protocol) + handler.observeRequestDuration(protocol, time.Since(started)) + }() if handler.inFlight != nil { select { case handler.inFlight <- struct{}{}: @@ -495,6 +500,12 @@ func (handler *Handler) observeRequestFinished(protocol string) { } } +func (handler *Handler) observeRequestDuration(protocol string, duration time.Duration) { + if handler.metrics != nil { + handler.metrics.ObserveRequestDuration(protocol, duration) + } +} + func requestMetricProtocol(request *http.Request) string { if request != nil && request.Method == http.MethodConnect { return "CONNECT" diff --git a/internal/gateway/server/handler_test.go b/internal/gateway/server/handler_test.go index e28638b..7ac63f9 100644 --- a/internal/gateway/server/handler_test.go +++ b/internal/gateway/server/handler_test.go @@ -107,12 +107,18 @@ func TestHandlerReportsLocalRequestAndTunnelLifecycles(t *testing.T) { if got := metrics.finished("HTTP"); got != 1 { t.Fatalf("HTTP request finishes = %d, want 1", got) } + if got := metrics.durationCount("HTTP"); got != 1 { + t.Fatalf("HTTP request durations = %d, want 1", got) + } if got := metrics.started("CONNECT"); got != 1 { t.Fatalf("CONNECT request starts = %d, want 1", got) } if got := metrics.finished("CONNECT"); got != 1 { t.Fatalf("CONNECT request finishes = %d, want 1", got) } + if got := metrics.durationCount("CONNECT"); got != 1 { + t.Fatalf("CONNECT request durations = %d, want 1", got) + } if metrics.opened != 1 || metrics.closed != 1 { t.Fatalf("tunnel lifecycle = opened:%d closed:%d, want 1:1", metrics.opened, metrics.closed) } @@ -960,11 +966,12 @@ type fakeTransport struct { } type recordingRequestMetrics struct { - mu sync.Mutex - starts map[string]int - finishes map[string]int - opened int - closed int + mu sync.Mutex + starts map[string]int + finishes map[string]int + durations map[string][]time.Duration + opened int + closed int } func (metrics *recordingRequestMetrics) ObserveRequestStarted(protocol string) { @@ -985,6 +992,15 @@ func (metrics *recordingRequestMetrics) ObserveRequestFinished(protocol string) metrics.finishes[protocol]++ } +func (metrics *recordingRequestMetrics) ObserveRequestDuration(protocol string, duration time.Duration) { + metrics.mu.Lock() + defer metrics.mu.Unlock() + if metrics.durations == nil { + metrics.durations = make(map[string][]time.Duration) + } + metrics.durations[protocol] = append(metrics.durations[protocol], duration) +} + func (metrics *recordingRequestMetrics) ObserveTunnelOpened() { metrics.mu.Lock() defer metrics.mu.Unlock() @@ -1009,6 +1025,12 @@ func (metrics *recordingRequestMetrics) finished(protocol string) int { return metrics.finishes[protocol] } +func (metrics *recordingRequestMetrics) durationCount(protocol string) int { + metrics.mu.Lock() + defer metrics.mu.Unlock() + return len(metrics.durations[protocol]) +} + type outcomeRecorder struct { mu sync.Mutex events []outcomeDomain.Event diff --git a/internal/platform/metrics/gateway.go b/internal/platform/metrics/gateway.go index 018a27c..7d7e213 100644 --- a/internal/platform/metrics/gateway.go +++ b/internal/platform/metrics/gateway.go @@ -3,6 +3,7 @@ package metrics import ( "errors" "fmt" + "time" "github.com/prometheus/client_golang/prometheus" @@ -20,6 +21,8 @@ type GatewayCollector struct { connectRequests prometheus.Counter httpInFlight prometheus.Gauge connectInFlight prometheus.Gauge + httpDuration prometheus.Observer + connectDuration prometheus.Observer activeTunnels prometheus.Gauge } @@ -67,6 +70,14 @@ func NewGatewayCollector(registerer prometheus.Registerer) (*GatewayCollector, e if err != nil { return nil, err } + durations, err := registerHistogramVec(registerer, prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "proxy_pool", Subsystem: "gateway", Name: "request_duration_seconds", + Help: "Gateway request duration from admission to completion by fixed proxy protocol.", + Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 300}, + }, []string{"protocol"})) + if err != nil { + return nil, err + } activeTunnels, err := registerGauge(registerer, prometheus.NewGauge(prometheus.GaugeOpts{ Namespace: "proxy_pool", Subsystem: "gateway", Name: "active_tunnels", Help: "Number of established CONNECT tunnels currently relaying through this Gateway Worker.", @@ -78,6 +89,7 @@ func NewGatewayCollector(registerer prometheus.Registerer) (*GatewayCollector, e outcomes: outcomes, dropped: dropped, invariants: invariants, httpRequests: requests.WithLabelValues("HTTP"), connectRequests: requests.WithLabelValues("CONNECT"), httpInFlight: inFlight.WithLabelValues("HTTP"), connectInFlight: inFlight.WithLabelValues("CONNECT"), + httpDuration: durations.WithLabelValues("HTTP"), connectDuration: durations.WithLabelValues("CONNECT"), activeTunnels: activeTunnels, }, nil } @@ -145,6 +157,22 @@ func (collector *GatewayCollector) ObserveRequestFinished(protocol string) { } } +func (collector *GatewayCollector) ObserveRequestDuration(protocol string, duration time.Duration) { + if collector == nil || duration < 0 { + return + } + switch protocol { + case "HTTP": + if collector.httpDuration != nil { + collector.httpDuration.Observe(duration.Seconds()) + } + case "CONNECT": + if collector.connectDuration != nil { + collector.connectDuration.Observe(duration.Seconds()) + } + } +} + func (collector *GatewayCollector) ObserveTunnelOpened() { if collector != nil && collector.activeTunnels != nil { collector.activeTunnels.Inc() @@ -205,6 +233,22 @@ func registerGaugeVec(registerer prometheus.Registerer, candidate *prometheus.Ga } } +func registerHistogramVec(registerer prometheus.Registerer, candidate *prometheus.HistogramVec) (*prometheus.HistogramVec, error) { + if err := registerer.Register(candidate); err == nil { + return candidate, nil + } else { + var registered prometheus.AlreadyRegisteredError + if !errors.As(err, ®istered) { + return nil, fmt.Errorf("register histogram vector: %w", err) + } + existing, ok := registered.ExistingCollector.(*prometheus.HistogramVec) + if !ok { + return nil, fmt.Errorf("register histogram vector: existing collector has unexpected type") + } + return existing, nil + } +} + func validGatewayStage(stage outcomeDomain.Stage) bool { switch stage { case outcomeDomain.StageDial, outcomeDomain.StageProxyHandshake, diff --git a/internal/platform/metrics/gateway_test.go b/internal/platform/metrics/gateway_test.go index 2719080..db04243 100644 --- a/internal/platform/metrics/gateway_test.go +++ b/internal/platform/metrics/gateway_test.go @@ -2,6 +2,7 @@ package metrics import ( "testing" + "time" "github.com/prometheus/client_golang/prometheus" @@ -27,6 +28,9 @@ func TestGatewayCollectorRecordsOnlyFixedDimensions(t *testing.T) { collector.ObserveRequestStarted("CONNECT") collector.ObserveRequestFinished("HTTP") collector.ObserveRequestFinished("CONNECT") + collector.ObserveRequestDuration("HTTP", 25*time.Millisecond) + collector.ObserveRequestDuration("CONNECT", 2*time.Second) + collector.ObserveRequestDuration("unknown", time.Second) collector.ObserveTunnelOpened() collector.ObserveTunnelClosed() collector.ObserveRequestStarted("unknown") @@ -54,6 +58,8 @@ func TestGatewayCollectorRecordsOnlyFixedDimensions(t *testing.T) { "protocol": "CONNECT", }, 0) assertMetricValue(t, registry, "proxy_pool_gateway_active_tunnels", nil, 0) + assertHistogramSampleCount(t, registry, "proxy_pool_gateway_request_duration_seconds", map[string]string{"protocol": "HTTP"}, 1) + assertHistogramSampleCount(t, registry, "proxy_pool_gateway_request_duration_seconds", map[string]string{"protocol": "CONNECT"}, 1) } func TestNewGatewayCollectorReusesRegisteredCollectors(t *testing.T) { @@ -70,3 +76,22 @@ func TestNewGatewayCollectorReusesRegisteredCollectors(t *testing.T) { second.ObserveDropped() assertMetricValue(t, registry, "proxy_pool_gateway_outcome_queue_dropped_total", nil, 2) } + +func assertHistogramSampleCount(t *testing.T, registry *prometheus.Registry, name string, labels map[string]string, want uint64) { + t.Helper() + metrics, err := registry.Gather() + if err != nil { + t.Fatalf("Gather() = %v", err) + } + for _, family := range metrics { + if family.GetName() != name { + continue + } + for _, metric := range family.GetMetric() { + if metricLabelsMatch(metric.GetLabel(), labels) && metric.GetHistogram().GetSampleCount() == want { + return + } + } + } + t.Fatalf("histogram %s labels=%v count=%d was not found", name, labels, want) +} diff --git a/progress.md b/progress.md index c5c1b97..b787534 100644 --- a/progress.md +++ b/progress.md @@ -20,6 +20,8 @@ - Grafana Overview 与 Prometheus 告警已从早期失效指标迁移到当前代码注册的低基数指标，覆盖 Gateway、Controller 容量、Provider、Extraction、Checker 与 Drain。部署测试会解析仪表盘和 规则，拒绝未注册指标以及 `upstream`/`worker` 聚合或 selector 标签，防止观测资产再次漂移。 +- Gateway 新增按固定 `HTTP`/`CONNECT` protocol 标签聚合的请求总耗时 Histogram，覆盖从准入到 + HTTP 响应或 CONNECT 隧道结束的完整生命周期；Grafana Overview 已恢复该真实指标的 p99 面板。 - 全仓 `go test -count=1 -timeout 60s ./...`、`go vet ./...`、`go build ./...`、 Protobuf descriptor、Kustomize Base 渲染及开发证书 SAN/SPIFFE 校验均通过。Compose 容器端到端启动在拉取 Dockerfile 前端与监控镜像时受 Docker Desktop HTTPS 代理缺失阻断，