From 162fd4b7b8eea3f3da188bb813b52c5fb7376ecf Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 13:55:23 +0800 Subject: [PATCH] feat: expose provider fetch metrics --- README.md | 5 ++ docs/development/implementation-plan.md | 9 ++- docs/requirements/traceability.md | 2 +- findings.md | 7 +++ internal/controller/bootstrap/bootstrap.go | 7 +++ internal/controller/provider/ports.go | 12 ++++ internal/controller/provider/stats.go | 60 ++++++++++++++++--- internal/controller/provider/stats_test.go | 27 +++++++++ internal/platform/metrics/provider.go | 68 +++++++++++++++++++++ internal/platform/metrics/provider_test.go | 69 ++++++++++++++++++++++ progress.md | 6 ++ 11 files changed, 261 insertions(+), 11 deletions(-) create mode 100644 internal/platform/metrics/provider.go create mode 100644 internal/platform/metrics/provider_test.go diff --git a/README.md b/README.md index b0a041c..cf08038 100644 --- a/README.md +++ b/README.md @@ -46,6 +46,11 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 - **Drain 指标**：Controller 暴露 `proxy_pool_controller_drain_candidates_total` 与 `proxy_pool_controller_drains_started_total`；`reason` 仅有 `unhealthy` 与 `upstream_disabled`，不包含 Proxy、Worker、Upstream、会话或地址。 +- **Provider 指标**：Controller 暴露 + `proxy_pool_controller_provider_fetch_results_total{class}`、 + `proxy_pool_controller_provider_valid_candidates_total` 与 + `proxy_pool_controller_provider_new_proxies_total`；`class` 仅有 `valid`、 + `empty`、`duplicate_only`、`error`，不包含 Upstream、Proxy 或错误文本标签。 - **PostgreSQL 管理面**：持久化配置版本、Upstream/Routing 管理状态、Admin 审计与 Outbox；不保存 Proxy 明细或逐次提取记录。 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index cb3820b..4f09a80 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -205,8 +205,8 @@ Distribution/Admin 服务构造、错误合并和资源关闭。生产 Provider 停机；Admin disable 会取消 Runtime，reload 在提交前预检并在发布后替换运行实例。 组合 fixture 已验证隔离 Redis namespace 下的选主、Provider HTTP 调用、模板解析 和活动池写入。Controller Metrics 独立入口现已提供 `/livez`、`/readyz` 与基础 -Prometheus 运行时指标，三监听器隔离已通过测试；业务指标仍待 -实现。双存储 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 组合 fixture，覆盖 +Prometheus 运行时指标，三监听器隔离已通过测试；Checker、Gateway、Drain 与 +Provider 业务指标已接入，提取、容量指标和密钥安全的结构化日志仍待实现。双存储 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 组合 fixture，覆盖 迁移、启动配置提交、Readiness、Admin Status 和 Metrics 探针。 WorkerControlPlane 现已接入 Controller 生命周期：Register、ACK 和 Runtime @@ -351,7 +351,10 @@ Supervisor 也改为同时服从静态配置与管理态，消除两条启停消 `BASIC`、`EGRESS`、`TARGET` 级别及 `accepted`、`rejected` 结果。Gateway 还暴露 `proxy_pool_gateway_outcomes_total{stage,result}` 和 `proxy_pool_gateway_outcome_queue_dropped_total`，其中阶段和结果均为固定枚举。Provider、 -提取、容量指标与密钥安全的结构化日志仍待补齐。 +拉取已暴露 `proxy_pool_controller_provider_fetch_results_total{class}`、 +`proxy_pool_controller_provider_valid_candidates_total` 与 +`proxy_pool_controller_provider_new_proxies_total`；`class` 仅允许 `valid`、`empty`、 +`duplicate_only`、`error`。Provider 指标已完成，提取、容量指标与密钥安全的结构化日志仍待补齐。 ## Task 14: Documentation, Examples, and Diagrams diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index d460089..a162bdb 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -87,5 +87,5 @@ | OPS-001 | 配置校验后构建不可变快照并原子替换 | 8959-8999 | 100k 索引、版本/epoch 与并发 Apply/Acquire 测试 | | OPS-002 | 优雅停机停止新请求/Fetch，等待现有流量后超时关闭 | 8981-9000 | Provider Run 收敛与 `Handler.Shutdown` HTTP 排空、Hijacked CONNECT 超时关闭测试 | | OPS-003 | PostgreSQL 只保存管理修订、Upstream/Routing 状态、Admin 审计与 Outbox | 当前会话 | ADR-006、`adminstate` 公用契约和六表 Schema 边界测试；真实 PostgreSQL 契约待完成 | -| OBS-001 | 指标禁止 Proxy IP、session、Client、完整 URL 高基数标签 | 9001-9029 | Controller Prometheus/探针模块已实现；Checker Collector 暴露固定等级和结果，Gateway Collector 暴露 `proxy_pool_gateway_outcomes_total{stage,result}` 与 `proxy_pool_gateway_outcome_queue_dropped_total`，Drain Collector 暴露 `proxy_pool_controller_drain_candidates_total{reason}` 与 `proxy_pool_controller_drains_started_total{reason}`，其中 reason 固定为 unhealthy/upstream_disabled；注册表测试锁定标签集。Provider、提取和容量业务指标待实现 | +| OBS-001 | 指标禁止 Proxy IP、session、Client、完整 URL 高基数标签 | 9001-9029 | Controller Prometheus/探针模块已实现；Checker Collector 暴露固定等级和结果，Gateway Collector 暴露 `proxy_pool_gateway_outcomes_total{stage,result}` 与 `proxy_pool_gateway_outcome_queue_dropped_total`，Drain Collector 暴露 `proxy_pool_controller_drain_candidates_total{reason}` 与 `proxy_pool_controller_drains_started_total{reason}`，其中 reason 固定为 unhealthy/upstream_disabled；Provider Collector 暴露 `proxy_pool_controller_provider_fetch_results_total{class}`、`proxy_pool_controller_provider_valid_candidates_total` 与 `proxy_pool_controller_provider_new_proxies_total`，其中 class 固定为 valid/empty/duplicate_only/error；注册表测试锁定标签集。Provider 指标已完成，提取、容量指标和密钥安全的结构化日志仍待实现 | | TEST-001 | 覆盖对话中列出的 11 个关键并发与故障场景 | 9030-9082 | 测试清单；Redis 活动池由 Memory/Redis 公用契约覆盖，跨进程故障场景仍按清单推进 | diff --git a/findings.md b/findings.md index de432bb..bdc4631 100644 --- a/findings.md +++ b/findings.md @@ -49,6 +49,13 @@ Extract 请求使用哪些 Upstream。 - Gateway、Distribution、Admin、Metrics 使用独立监听和认证/访问控制。 - 非回环监听且无认证、无 CIDR 保护时，严格模式必须拒绝启动。 +## 可观测性边界 + +- Provider 拉取结果只按固定 `class` 聚合：`valid`、`empty`、`duplicate_only`、`error`。 + 指标分别记录拉取结果次数、合法候选数和新入池代理数，不按 Upstream、Proxy、IP、错误 + 文本或凭据拆分。 +- Provider 指标已完成；提取、容量指标和密钥安全的结构化日志仍待实现。 + ## 集群与性能 - 用户补充：高峰可能达到 100,000 请求/秒。 diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index e9e0358..d563edf 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -237,6 +237,13 @@ func runWithWorkerFactory( return fmt.Errorf("%w: build Drain metrics: %w", ErrStartup, drainCollectorErr) } drainMetrics = drainCollector + providerCollector, providerCollectorErr := platformMetrics.NewProviderCollector(prometheus.DefaultRegisterer) + if providerCollectorErr != nil { + return fmt.Errorf("%w: build Provider metrics: %w", ErrStartup, providerCollectorErr) + } + if registrar, ok := opened.providerResults.(provider.ResultObserverRegistrar); ok && !nilInterface(registrar) { + registrar.AddResultObserver(providerCollector) + } handler, handlerErr := platformMetrics.NewHandler(platformMetrics.Dependencies{ Gatherer: prometheus.DefaultGatherer, Readiness: opened.metricsReadiness, }) diff --git a/internal/controller/provider/ports.go b/internal/controller/provider/ports.go index a4921b6..4b77992 100644 --- a/internal/controller/provider/ports.go +++ b/internal/controller/provider/ports.go @@ -49,6 +49,18 @@ type ResultRecorder interface { Record(Result) } +// ResultObserver receives one validated Provider fetch result. Observers must +// not retain sensitive response data or block the fetch reconciliation loop. +type ResultObserver interface { + ObserveProviderResult(Result) +} + +// ResultObserverRegistrar lets bootstrap attach low-cardinality observers to +// a shared result recorder without changing the Provider runtime contract. +type ResultObserverRegistrar interface { + AddResultObserver(ResultObserver) +} + type Ports struct { Adapter ProviderAdapter Parser Parser diff --git a/internal/controller/provider/stats.go b/internal/controller/provider/stats.go index eaf4687..580c583 100644 --- a/internal/controller/provider/stats.go +++ b/internal/controller/provider/stats.go @@ -3,6 +3,7 @@ package provider import ( "errors" "math" + "reflect" "sync" "proxy-pool/internal/domain/upstream" @@ -25,9 +26,10 @@ type StatsRetainer interface { } type StatsRecorder struct { - mu sync.Mutex - maximum int - byID map[string]Stats + mu sync.Mutex + maximum int + byID map[string]Stats + observers []ResultObserver } func NewStatsRecorder(maximum int) (*StatsRecorder, error) { @@ -38,14 +40,16 @@ func NewStatsRecorder(maximum int) (*StatsRecorder, error) { } func (recorder *StatsRecorder) Record(result Result) { - if recorder == nil || result.UpstreamID == "" { + if recorder == nil || result.UpstreamID == "" || !validResultClass(result.Class) { return } recorder.mu.Lock() - defer recorder.mu.Unlock() + observers := append([]ResultObserver(nil), recorder.observers...) stats, exists := recorder.byID[result.UpstreamID] if !exists { if len(recorder.byID) >= recorder.maximum { + recorder.mu.Unlock() + recorder.observe(observers, result) return } stats.UpstreamID = result.UpstreamID @@ -61,10 +65,50 @@ func (recorder *StatsRecorder) Record(result Result) { if stats.FetchErrorCount < math.MaxInt64 { stats.FetchErrorCount++ } - default: - return } recorder.byID[result.UpstreamID] = stats + recorder.mu.Unlock() + recorder.observe(observers, result) +} + +// AddResultObserver appends one process-local observer. Duplicate instances +// are ignored so repeated bootstrap assembly cannot double-count metrics. +func (recorder *StatsRecorder) AddResultObserver(observer ResultObserver) { + if recorder == nil || observer == nil { + return + } + recorder.mu.Lock() + defer recorder.mu.Unlock() + for _, existing := range recorder.observers { + if sameResultObserver(existing, observer) { + return + } + } + recorder.observers = append(recorder.observers, observer) +} + +func (recorder *StatsRecorder) observe(observers []ResultObserver, result Result) { + for _, observer := range observers { + observer.ObserveProviderResult(result) + } +} + +func validResultClass(class upstream.FetchClass) bool { + switch class { + case upstream.FetchValid, upstream.FetchEmpty, upstream.FetchDuplicateOnly, upstream.FetchError: + return true + default: + return false + } +} + +func sameResultObserver(left, right ResultObserver) bool { + leftValue := reflect.ValueOf(left) + rightValue := reflect.ValueOf(right) + if !leftValue.IsValid() || !rightValue.IsValid() || leftValue.Type() != rightValue.Type() || !leftValue.Type().Comparable() { + return false + } + return leftValue.Interface() == rightValue.Interface() } func (recorder *StatsRecorder) ReadProviderStats(upstreamIDs []string) []Stats { @@ -101,3 +145,5 @@ func (recorder *StatsRecorder) RetainProviderStats(upstreamIDs []string) { } } } + +var _ ResultObserverRegistrar = (*StatsRecorder)(nil) diff --git a/internal/controller/provider/stats_test.go b/internal/controller/provider/stats_test.go index ff1abc2..f4e009c 100644 --- a/internal/controller/provider/stats_test.go +++ b/internal/controller/provider/stats_test.go @@ -81,3 +81,30 @@ func TestStatsRecorderRetainsConfiguredProvidersAndReusesCapacity(t *testing.T) t.Fatalf("stats after retention = %+v", stats) } } + +func TestStatsRecorderObservesValidatedResultsOutsideBoundedStatusStorage(t *testing.T) { + recorder, err := NewStatsRecorder(1) + if err != nil { + t.Fatalf("NewStatsRecorder() = %v", err) + } + observer := &providerResultObserverStub{} + recorder.AddResultObserver(observer) + recorder.AddResultObserver(observer) + recorder.Record(Result{UpstreamID: "provider-a", Class: upstream.FetchEmpty}) + recorder.Record(Result{UpstreamID: "provider-b", Class: upstream.FetchValid, ValidCount: 3, NewCount: 2}) + recorder.Record(Result{UpstreamID: "provider-c", Class: "invalid"}) + + if len(observer.results) != 2 || observer.results[0].UpstreamID != "provider-a" || observer.results[1].UpstreamID != "provider-b" { + t.Fatalf("observed results = %+v", observer.results) + } + stats := recorder.ReadProviderStats([]string{"provider-a", "provider-b"}) + if stats[0].ConsecutiveEmptyFetch != 1 || stats[1].ConsecutiveEmptyFetch != 0 || stats[1].FetchErrorCount != 0 { + t.Fatalf("bounded status stats = %+v", stats) + } +} + +type providerResultObserverStub struct{ results []Result } + +func (observer *providerResultObserverStub) ObserveProviderResult(result Result) { + observer.results = append(observer.results, result) +} diff --git a/internal/platform/metrics/provider.go b/internal/platform/metrics/provider.go new file mode 100644 index 0000000..cb727d6 --- /dev/null +++ b/internal/platform/metrics/provider.go @@ -0,0 +1,68 @@ +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + + controllerProvider "proxy-pool/internal/controller/provider" + "proxy-pool/internal/domain/upstream" +) + +// ProviderCollector exposes bounded Provider fetch outcomes. It never uses +// upstream names, proxy addresses, response bodies or error strings as labels. +type ProviderCollector struct { + results *prometheus.CounterVec + candidates prometheus.Counter + inserted prometheus.Counter +} + +var _ controllerProvider.ResultObserver = (*ProviderCollector)(nil) + +func NewProviderCollector(registerer prometheus.Registerer) (*ProviderCollector, error) { + if registerer == nil { + return nil, ErrInvalidDependencies + } + results, err := registerCounterVec(registerer, prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "proxy_pool", Subsystem: "controller", Name: "provider_fetch_results_total", + Help: "Number of completed Provider fetch reconciliations by fixed result class.", + }, []string{"class"})) + if err != nil { + return nil, err + } + candidates, err := registerCounter(registerer, prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: "proxy_pool", Subsystem: "controller", Name: "provider_valid_candidates_total", + Help: "Number of valid Provider proxy candidates parsed before deduplication.", + })) + if err != nil { + return nil, err + } + inserted, err := registerCounter(registerer, prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: "proxy_pool", Subsystem: "controller", Name: "provider_new_proxies_total", + Help: "Number of Provider proxy candidates inserted into the activity pool.", + })) + if err != nil { + return nil, err + } + return &ProviderCollector{results: results, candidates: candidates, inserted: inserted}, nil +} + +func (collector *ProviderCollector) ObserveProviderResult(result controllerProvider.Result) { + if collector == nil || collector.results == nil || !validProviderFetchClass(result.Class) { + return + } + collector.results.WithLabelValues(string(result.Class)).Inc() + if result.ValidCount > 0 && collector.candidates != nil { + collector.candidates.Add(float64(result.ValidCount)) + } + if result.NewCount > 0 && collector.inserted != nil { + collector.inserted.Add(float64(result.NewCount)) + } +} + +func validProviderFetchClass(class upstream.FetchClass) bool { + switch class { + case upstream.FetchValid, upstream.FetchEmpty, upstream.FetchDuplicateOnly, upstream.FetchError: + return true + default: + return false + } +} diff --git a/internal/platform/metrics/provider_test.go b/internal/platform/metrics/provider_test.go new file mode 100644 index 0000000..666b70a --- /dev/null +++ b/internal/platform/metrics/provider_test.go @@ -0,0 +1,69 @@ +package metrics + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + + controllerProvider "proxy-pool/internal/controller/provider" + "proxy-pool/internal/domain/upstream" +) + +func TestProviderCollectorRecordsOnlyFixedResultClasses(t *testing.T) { + registry := prometheus.NewRegistry() + collector, err := NewProviderCollector(registry) + if err != nil { + t.Fatalf("NewProviderCollector() error = %v", err) + } + collector.ObserveProviderResult(controllerProvider.Result{Class: upstream.FetchValid, ValidCount: 5, NewCount: 2}) + collector.ObserveProviderResult(controllerProvider.Result{Class: upstream.FetchDuplicateOnly, ValidCount: 3}) + collector.ObserveProviderResult(controllerProvider.Result{Class: upstream.FetchEmpty}) + collector.ObserveProviderResult(controllerProvider.Result{Class: upstream.FetchError}) + collector.ObserveProviderResult(controllerProvider.Result{Class: "invalid", ValidCount: 9, NewCount: 9}) + + assertMetricValue(t, registry, "proxy_pool_controller_provider_fetch_results_total", map[string]string{"class": "valid"}, 1) + assertMetricValue(t, registry, "proxy_pool_controller_provider_fetch_results_total", map[string]string{"class": "duplicate_only"}, 1) + assertMetricValue(t, registry, "proxy_pool_controller_provider_fetch_results_total", map[string]string{"class": "empty"}, 1) + assertMetricValue(t, registry, "proxy_pool_controller_provider_fetch_results_total", map[string]string{"class": "error"}, 1) + assertMetricValue(t, registry, "proxy_pool_controller_provider_valid_candidates_total", nil, 8) + assertMetricValue(t, registry, "proxy_pool_controller_provider_new_proxies_total", nil, 2) +} + +func TestNewProviderCollectorReusesRegisteredCollectors(t *testing.T) { + registry := prometheus.NewRegistry() + first, err := NewProviderCollector(registry) + if err != nil { + t.Fatalf("first NewProviderCollector() error = %v", err) + } + second, err := NewProviderCollector(registry) + if err != nil { + t.Fatalf("second NewProviderCollector() error = %v", err) + } + first.ObserveProviderResult(controllerProvider.Result{Class: upstream.FetchValid, ValidCount: 1, NewCount: 1}) + second.ObserveProviderResult(controllerProvider.Result{Class: upstream.FetchValid, ValidCount: 2, NewCount: 2}) + assertMetricValue(t, registry, "proxy_pool_controller_provider_fetch_results_total", map[string]string{"class": "valid"}, 2) + assertMetricValue(t, registry, "proxy_pool_controller_provider_valid_candidates_total", nil, 3) + assertMetricValue(t, registry, "proxy_pool_controller_provider_new_proxies_total", nil, 3) +} + +func TestProviderCollectorObservesSharedStatsRecorder(t *testing.T) { + registry := prometheus.NewRegistry() + collector, err := NewProviderCollector(registry) + if err != nil { + t.Fatalf("NewProviderCollector() error = %v", err) + } + recorder, err := controllerProvider.NewStatsRecorder(1) + if err != nil { + t.Fatalf("NewStatsRecorder() error = %v", err) + } + recorder.AddResultObserver(collector) + recorder.Record(controllerProvider.Result{UpstreamID: "provider-a", Class: upstream.FetchValid, ValidCount: 2, NewCount: 1}) + // The bounded Admin-status map has no room for provider-b, but metrics must + // retain the process-wide fetch observation. + recorder.Record(controllerProvider.Result{UpstreamID: "provider-b", Class: upstream.FetchEmpty}) + + assertMetricValue(t, registry, "proxy_pool_controller_provider_fetch_results_total", map[string]string{"class": "valid"}, 1) + assertMetricValue(t, registry, "proxy_pool_controller_provider_fetch_results_total", map[string]string{"class": "empty"}, 1) + assertMetricValue(t, registry, "proxy_pool_controller_provider_valid_candidates_total", nil, 2) + assertMetricValue(t, registry, "proxy_pool_controller_provider_new_proxies_total", nil, 1) +} diff --git a/progress.md b/progress.md index 7f30466..121d10b 100644 --- a/progress.md +++ b/progress.md @@ -2,6 +2,12 @@ ## 2026-08-02 +- Provider 拉取可观测性已接入公用 `ResultObserver`：Controller 暴露 + `proxy_pool_controller_provider_fetch_results_total{class}`、 + `proxy_pool_controller_provider_valid_candidates_total` 与 + `proxy_pool_controller_provider_new_proxies_total`。`class` 仅允许 `valid`、`empty`、 + `duplicate_only`、`error`；指标不包含 Upstream、Proxy、IP、错误文本或凭据标签。 + Provider 指标已完成，提取、容量指标和密钥安全的结构化日志仍待实现。 - 已实现管理态驱动的即时完整快照刷新：`SnapshotRefreshBroker` 以每个 Worker 流一格的 非阻塞通道广播，连续变更自动合并。Admin 成功提交 Upstream 启停、Routing 切换和配置 发布时触发通知；Worker 保留原有定时刷新作为失效保护。Routing 切换不按旧 Upstream