From b24d71fb90df4643a1e86310eccbef725aa3f683 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 14:21:18 +0800 Subject: [PATCH] feat: expose capacity metrics --- README.md | 4 + docs/configuration/reference.md | 6 +- docs/development/implementation-plan.md | 5 +- docs/requirements/completion-audit.md | 4 +- docs/requirements/traceability.md | 2 +- findings.md | 9 +- internal/controller/bootstrap/bootstrap.go | 6 + internal/controller/bootstrap/provider.go | 1 + internal/controller/pool/capacity_metrics.go | 29 ++++ .../controller/provider/upstream_runtime.go | 28 ++- .../provider/upstream_runtime_test.go | 131 ++++++++++++++ internal/platform/metrics/capacity.go | 160 ++++++++++++++++++ internal/platform/metrics/capacity_test.go | 115 +++++++++++++ progress.md | 9 +- 14 files changed, 497 insertions(+), 12 deletions(-) create mode 100644 internal/controller/pool/capacity_metrics.go create mode 100644 internal/platform/metrics/capacity.go create mode 100644 internal/platform/metrics/capacity_test.go diff --git a/README.md b/README.md index 22e9136..7be00af 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,10 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 `proxy_pool_controller_extraction_returned_proxies_total`；`result` 仅有完成、 部分、空、库存不足、幂等冲突、限流、不可用、无效与内部错误等固定枚举， 不包含 Client、请求、过滤条件、Upstream、Proxy 或错误文本标签。 +- **容量指标**：Controller 在既有 Provider 库存对账周期聚合托管 Proxy、可用/有效 + Slot、待拉取数量和活跃上游数，并暴露 + `proxy_pool_controller_capacity_inventory_reads_total{result}`；`result` 仅有 + `success` 和 `error`，不按 Upstream、Worker、Proxy、会话或地址拆分。 - **PostgreSQL 管理面**：持久化配置版本、Upstream/Routing 管理状态、Admin 审计与 Outbox；不保存 Proxy 明细或逐次提取记录。 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index f7e1ef5..a080dd8 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -526,8 +526,10 @@ Metrics 启用时 `listen` 必须是合法 `host:port`。该入口固定提供 ` `result` 固定为 accepted、rejected。Gateway 另暴露 `proxy_pool_gateway_outcomes_total{stage,result}` 与 `proxy_pool_gateway_outcome_queue_dropped_total`；`stage` 固定为 DIAL、PROXY_HANDSHAKE、 -RESPONSE_HEADERS、TUNNEL，`result` 固定为 success、failure。绝不包含 Proxy、IP、Checker、 -Client、路由、目标 URL 或凭据标签。Provider、提取和容量等业务指标仍在后续实施范围。Distribution 启用时 `/readyz` 只以 Redis 活动池为 +RESPONSE_HEADERS、TUNNEL，`result` 固定为 success、failure。Provider、Extraction、 +Capacity 和 Drain 业务指标也已接入，所有标签均为固定枚举；绝不包含 Proxy、IP、Checker、 +Client、路由、目标 URL 或凭据标签。Capacity 从既有 Provider 库存对账循环聚合， +不进入 Gateway 热路径。Distribution 启用时 `/readyz` 只以 Redis 活动池为 服务流量门槛，PostgreSQL 故障由 Admin 接口独立报告。Metrics 开关或监听地址 变更需要重启 Controller。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 8cfb2c4..ad58465 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -206,7 +206,7 @@ Distribution/Admin 服务构造、错误合并和资源关闭。生产 Provider 组合 fixture 已验证隔离 Redis namespace 下的选主、Provider HTTP 调用、模板解析 和活动池写入。Controller Metrics 独立入口现已提供 `/livez`、`/readyz` 与基础 Prometheus 运行时指标，三监听器隔离已通过测试；Checker、Gateway、Drain、 -Provider 与 Extraction 业务指标已接入，容量指标和密钥安全的结构化日志仍待实现。双存储 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 组合 fixture，覆盖 +Provider、Extraction 与容量业务指标已接入，密钥安全的结构化日志仍待实现。双存储 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 组合 fixture，覆盖 迁移、启动配置提交、Readiness、Admin Status 和 Metrics 探针。 WorkerControlPlane 现已接入 Controller 生命周期：Register、ACK 和 Runtime @@ -355,7 +355,8 @@ Supervisor 也改为同时服从静态配置与管理态，消除两条启停消 `proxy_pool_controller_provider_valid_candidates_total` 与 `proxy_pool_controller_provider_new_proxies_total`；`class` 仅允许 `valid`、`empty`、 `duplicate_only`、`error`。Extraction 还暴露固定 `result` 的请求次数、请求数与响应 -交付数，幂等重放按响应交付统计。容量指标与密钥安全的结构化日志仍待补齐。 +交付数，幂等重放按响应交付统计。容量指标由既有 Provider 库存对账周期聚合，不进入 +Gateway 热路径；仅密钥安全的结构化日志仍待补齐。 ## Task 14: Documentation, Examples, and Diagrams diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 42117b6..b0e527e 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -104,8 +104,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 响应校验场景；`proxy-checker` 的 BASIC 任务进程已经完成， `proxy-controller` 已完成 Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，`proxy-gateway` 已完成 - HTTP/Metrics 与控制面 Session 装配；Checker 与 Gateway Outcome 的低基数业务指标 - 已接入，但 Provider、提取和容量指标链未闭环。 + HTTP/Metrics 与控制面 Session 装配；Checker、Gateway Outcome、Drain、Provider、 + Extraction 与 Capacity 的低基数业务指标已闭环，密钥安全的结构化日志仍待补齐。 2. Gateway 的生产连接池调优与代表性流量压测。 3. Provider 分布式 singleflight/Leader、长期凭据回收和累计额度执行器。 4. Controller 的 PostgreSQL 连接池、迁移和 pgx Adapter 启动装配已完成； diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 9cfc7ae..4f8282b 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 Collector 暴露固定 class 的拉取次数与候选计数；Extraction Collector 暴露 `proxy_pool_controller_extraction_requests_total{result}`、请求数与交付数，result 固定为 complete/partial/empty/insufficient/idempotency_conflict/rate_limited/unavailable/invalid/error。注册表测试锁定标签集；容量指标和密钥安全的结构化日志仍待实现 | +| 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 暴露固定 class 的拉取次数与候选计数；Extraction Collector 暴露固定 result 的请求、请求数与交付数；Capacity Collector 按 Controller 进程聚合库存 Gauge，并只暴露固定 success/error 的库存读取结果，不把 Upstream 写成标签。注册表测试锁定标签集；密钥安全的结构化日志仍待实现 | | TEST-001 | 覆盖对话中列出的 11 个关键并发与故障场景 | 9030-9082 | 测试清单；Redis 活动池由 Memory/Redis 公用契约覆盖，跨进程故障场景仍按清单推进 | diff --git a/findings.md b/findings.md index 6cef519..7cdc54f 100644 --- a/findings.md +++ b/findings.md @@ -54,12 +54,17 @@ Extract 请求使用哪些 Upstream。 - Provider 拉取结果只按固定 `class` 聚合：`valid`、`empty`、`duplicate_only`、`error`。 指标分别记录拉取结果次数、合法候选数和新入池代理数，不按 Upstream、Proxy、IP、错误 文本或凭据拆分。 -- Provider 指标已完成；提取、容量指标和密钥安全的结构化日志仍待实现。 +- Provider 指标已完成；密钥安全的结构化日志仍待实现。 - Extraction 只按固定 `result` 聚合请求：`complete`、`partial`、`empty`、 `insufficient`、`idempotency_conflict`、`rate_limited`、`unavailable`、`invalid`、 `error`。计数只记录请求数量和响应交付数量；幂等重放不被误记为新的 Proxy 消费。 -- Extraction 指标已完成；容量指标和密钥安全的结构化日志仍待实现。 +- Extraction 指标已完成；密钥安全的结构化日志仍待实现。 + +- 容量指标由 Provider 库存对账循环采样，聚合 managed、available、effective、pending + 与 active Upstream Gauge。Upstream ID 仅保留在 Controller 进程内用于聚合替换和 + 任期结束清理，不输出为 Prometheus 标签；读取结果只允许 `success` 或 `error`。 +- 容量指标已完成；密钥安全的结构化日志仍待实现。 ## 集群与性能 diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index b032401..a8e2fa5 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -77,6 +77,7 @@ type ports struct { coordinator provider.Coordinator credentials credentials.Store providerResults provider.ResultRecorder + capacityMetrics pool.CapacityObserver workerStore workerruntime.ControlStore close func() error } @@ -203,6 +204,11 @@ func runWithWorkerFactory( if registrar, ok := opened.providerResults.(provider.ResultObserverRegistrar); ok && !nilInterface(registrar) { registrar.AddResultObserver(providerCollector) } + capacityCollector, capacityCollectorErr := platformMetrics.NewCapacityCollector(prometheus.DefaultRegisterer) + if capacityCollectorErr != nil { + return fmt.Errorf("%w: build capacity metrics: %w", ErrStartup, capacityCollectorErr) + } + opened.capacityMetrics = capacityCollector extractionCollector, extractionCollectorErr := platformMetrics.NewExtractionCollector(prometheus.DefaultRegisterer) if extractionCollectorErr != nil { return fmt.Errorf("%w: build extraction metrics: %w", ErrStartup, extractionCollectorErr) diff --git a/internal/controller/bootstrap/provider.go b/internal/controller/bootstrap/provider.go index 6f9f0df..95dd276 100644 --- a/internal/controller/bootstrap/provider.go +++ b/internal/controller/bootstrap/provider.go @@ -75,6 +75,7 @@ func providerRuntimeBuilder(opened ports) (buildUpstreamRuntime, error) { Parser: parser, Activity: opened.activity, Results: results, + Capacity: opened.capacityMetrics, }) }, nil } diff --git a/internal/controller/pool/capacity_metrics.go b/internal/controller/pool/capacity_metrics.go new file mode 100644 index 0000000..70b9f25 --- /dev/null +++ b/internal/controller/pool/capacity_metrics.go @@ -0,0 +1,29 @@ +package pool + +// CapacityReadResult is the fixed outcome of one authoritative inventory read. +type CapacityReadResult string + +const ( + CapacityReadSuccess CapacityReadResult = "success" + CapacityReadError CapacityReadResult = "error" +) + +// CapacityObservation contains one Controller-side inventory sample. Its +// UpstreamID and SourceID are for process-local aggregation only and must not +// be exposed as Prometheus labels. +type CapacityObservation struct { + UpstreamID string + SourceID string + Result CapacityReadResult + Managed int + AvailableSlots int64 + EffectiveSlots int64 + PendingExpected int +} + +// CapacityObserver receives cold-path Provider inventory samples. It must +// return promptly and must not issue another inventory read or persistent I/O. +type CapacityObserver interface { + ObserveCapacity(CapacityObservation) + RemoveCapacityUpstream(upstreamID, sourceID string) +} diff --git a/internal/controller/provider/upstream_runtime.go b/internal/controller/provider/upstream_runtime.go index be30d21..7c6e1b5 100644 --- a/internal/controller/provider/upstream_runtime.go +++ b/internal/controller/provider/upstream_runtime.go @@ -6,6 +6,7 @@ import ( "fmt" "hash/fnv" "reflect" + "sync/atomic" "time" "proxy-pool/internal/controller/pool" @@ -30,6 +31,7 @@ type UpstreamRuntimeDependencies struct { Parser Parser Activity activitypool.Upserter Results ResultRecorder + Capacity pool.CapacityObserver } // UpstreamRuntime owns every leader-scoped object for one Upstream. A fresh @@ -38,6 +40,7 @@ type UpstreamRuntime struct { config UpstreamRuntimeConfig dependencies UpstreamRuntimeDependencies sleeper Sleeper + termSequence atomic.Uint64 } func NewUpstreamRuntime( @@ -126,26 +129,33 @@ func (runtime *UpstreamRuntime) newLeaderTerm() (*upstreamLeaderTerm, error) { } return &upstreamLeaderTerm{ upstreamID: runtime.config.Provider.UpstreamID, + sourceID: fmt.Sprintf("%p:%d", runtime, runtime.termSequence.Add(1)), interval: runtime.config.ReconcileInterval, safetyMargin: runtime.config.ReconcilePolicy.SafetyMargin, inventory: runtime.dependencies.Inventory, pool: poolReconciler, provider: providerReconciler, + capacity: runtime.dependencies.Capacity, sleeper: runtime.sleeper, }, nil } type upstreamLeaderTerm struct { upstreamID string + sourceID string interval time.Duration safetyMargin time.Duration inventory pool.InventoryReader pool *pool.Reconciler provider *Reconciler + capacity pool.CapacityObserver sleeper Sleeper } func (term *upstreamLeaderTerm) run(ctx context.Context, session LeaderSession) error { + if !nilRuntimeDependency(term.capacity) { + defer term.capacity.RemoveCapacityUpstream(term.upstreamID, term.sourceID) + } termCtx, cancel := context.WithCancel(ctx) defer cancel() done := make(chan error, 2) @@ -183,7 +193,16 @@ func (term *upstreamLeaderTerm) reconcileInventory(ctx context.Context) error { term.safetyMargin, ) if err == nil && inventory.Managed >= 0 && inventory.AvailableSlots >= 0 { - term.pool.ReconcileSnapshot(inventory) + decision := term.pool.ReconcileSnapshot(inventory) + term.observeCapacity(pool.CapacityObservation{ + UpstreamID: term.upstreamID, SourceID: term.sourceID, Result: pool.CapacityReadSuccess, + Managed: inventory.Managed, AvailableSlots: inventory.AvailableSlots, + EffectiveSlots: decision.EffectiveSlots, PendingExpected: decision.PendingExpected, + }) + } else { + term.observeCapacity(pool.CapacityObservation{ + UpstreamID: term.upstreamID, SourceID: term.sourceID, Result: pool.CapacityReadError, + }) } if err := term.sleeper.Sleep(ctx, term.interval); err != nil { if ctx.Err() != nil { @@ -195,6 +214,13 @@ func (term *upstreamLeaderTerm) reconcileInventory(ctx context.Context) error { return nil } +func (term *upstreamLeaderTerm) observeCapacity(observation pool.CapacityObservation) { + if term == nil || nilRuntimeDependency(term.capacity) { + return + } + term.capacity.ObserveCapacity(observation) +} + func initialReconcileDelay(upstreamID string, interval time.Duration) time.Duration { window := min(interval/4, 250*time.Millisecond) if window <= 1 { diff --git a/internal/controller/provider/upstream_runtime_test.go b/internal/controller/provider/upstream_runtime_test.go index 495a14d..7775038 100644 --- a/internal/controller/provider/upstream_runtime_test.go +++ b/internal/controller/provider/upstream_runtime_test.go @@ -126,6 +126,112 @@ func TestUpstreamRuntimeFailsClosedUntilInventoryReadRecovers(t *testing.T) { } } +func TestUpstreamRuntimeReportsCapacityReadsAndClearsTermState(t *testing.T) { + observer := newCapacityObserverStub() + runtime, err := NewUpstreamRuntime(UpstreamRuntimeConfig{ + Provider: runtimeProviderConfig("provider-a"), + ReconcilePolicy: runtimeReconcilePolicy(), + ReconcileInterval: 10 * time.Millisecond, + }, UpstreamRuntimeDependencies{ + Coordinator: coordinatorFunc(func( + ctx context.Context, + _ string, + _ CoordinationLimits, + work func(context.Context, LeaderSession) error, + ) error { + return work(ctx, unlimitedLeaderSession{}) + }), + Inventory: inventoryReaderFunc(func(context.Context, string, time.Duration) (pool.InventorySnapshot, error) { + return pool.InventorySnapshot{Managed: 3, AvailableSlots: 7}, nil + }), + Adapter: adapterFunc(func(context.Context) (FetchResponse, error) { return FetchResponse{}, nil }), + Parser: parserFunc(func(context.Context, []byte) ([]proxyDomain.Proxy, error) { return nil, nil }), + Activity: activitySinkFunc(func(context.Context, string, activitypool.FetchedBatch) (activitypool.UpsertResult, error) { + return activitypool.UpsertResult{}, nil + }), + Results: resultRecorderFunc(func(Result) {}), + Capacity: observer, + }) + if err != nil { + t.Fatalf("NewUpstreamRuntime() error = %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- runtime.Run(ctx) }() + select { + case observed := <-observer.observations: + if observed.UpstreamID != "provider-a" || observed.SourceID == "" || observed.Result != pool.CapacityReadSuccess || + observed.Managed != 3 || observed.AvailableSlots != 7 || observed.EffectiveSlots != 7 || observed.PendingExpected != 0 { + t.Fatalf("capacity observation = %+v", observed) + } + case <-time.After(time.Second): + cancel() + <-done + t.Fatal("timed out waiting for capacity observation") + } + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v, want context cancellation", err) + } + select { + case removal := <-observer.removed: + if removal.upstreamID != "provider-a" || removal.sourceID == "" { + t.Fatalf("capacity removal = %+v", removal) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for capacity cleanup") + } +} + +func TestUpstreamRuntimeReportsFailedCapacityReads(t *testing.T) { + observer := newCapacityObserverStub() + runtime, err := NewUpstreamRuntime(UpstreamRuntimeConfig{ + Provider: runtimeProviderConfig("provider-a"), + ReconcilePolicy: runtimeReconcilePolicy(), + ReconcileInterval: 10 * time.Millisecond, + }, UpstreamRuntimeDependencies{ + Coordinator: coordinatorFunc(func( + ctx context.Context, + _ string, + _ CoordinationLimits, + work func(context.Context, LeaderSession) error, + ) error { + return work(ctx, unlimitedLeaderSession{}) + }), + Inventory: inventoryReaderFunc(func(context.Context, string, time.Duration) (pool.InventorySnapshot, error) { + return pool.InventorySnapshot{}, errors.New("redis unavailable") + }), + Adapter: adapterFunc(func(context.Context) (FetchResponse, error) { return FetchResponse{}, nil }), + Parser: parserFunc(func(context.Context, []byte) ([]proxyDomain.Proxy, error) { return nil, nil }), + Activity: activitySinkFunc(func(context.Context, string, activitypool.FetchedBatch) (activitypool.UpsertResult, error) { + return activitypool.UpsertResult{}, nil + }), + Results: resultRecorderFunc(func(Result) {}), + Capacity: observer, + }) + if err != nil { + t.Fatalf("NewUpstreamRuntime() error = %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- runtime.Run(ctx) }() + select { + case observed := <-observer.observations: + if observed.UpstreamID != "provider-a" || observed.SourceID == "" || observed.Result != pool.CapacityReadError || + observed.Managed != 0 || observed.AvailableSlots != 0 || observed.EffectiveSlots != 0 || observed.PendingExpected != 0 { + t.Fatalf("capacity observation = %+v", observed) + } + case <-time.After(time.Second): + cancel() + <-done + t.Fatal("timed out waiting for failed capacity observation") + } + cancel() + <-done +} + func TestUpstreamRuntimeDelegatesRequestIntervalOnlyToCoordinator(t *testing.T) { providerConfig := runtimeProviderConfig("provider-a") providerConfig.RequestInterval = 500 * time.Millisecond @@ -245,3 +351,28 @@ func (f inventoryReaderFunc) ReadInventory( ) (pool.InventorySnapshot, error) { return f(ctx, upstreamID, safetyMargin) } + +type capacityObserverStub struct { + observations chan pool.CapacityObservation + removed chan capacityRemoval +} + +type capacityRemoval struct { + upstreamID string + sourceID string +} + +func newCapacityObserverStub() *capacityObserverStub { + return &capacityObserverStub{ + observations: make(chan pool.CapacityObservation, 4), + removed: make(chan capacityRemoval, 2), + } +} + +func (observer *capacityObserverStub) ObserveCapacity(observation pool.CapacityObservation) { + observer.observations <- observation +} + +func (observer *capacityObserverStub) RemoveCapacityUpstream(upstreamID, sourceID string) { + observer.removed <- capacityRemoval{upstreamID: upstreamID, sourceID: sourceID} +} diff --git a/internal/platform/metrics/capacity.go b/internal/platform/metrics/capacity.go new file mode 100644 index 0000000..5ca9ba7 --- /dev/null +++ b/internal/platform/metrics/capacity.go @@ -0,0 +1,160 @@ +package metrics + +import ( + "errors" + "fmt" + "sync" + + "github.com/prometheus/client_golang/prometheus" + + controllerPool "proxy-pool/internal/controller/pool" +) + +// CapacityCollector aggregates Controller leader-term inventory samples. An +// upstream ID is retained only in process memory for replacement and removal; +// it is never emitted as a Prometheus label. +type CapacityCollector struct { + mu sync.Mutex + byUpstream map[string]capacityState + + reads *prometheus.CounterVec + managed *prometheus.Desc + available *prometheus.Desc + effective *prometheus.Desc + pending *prometheus.Desc + activeUpstreams *prometheus.Desc +} + +type capacityState struct { + sourceID string + managed int + available int64 + effective int64 + pending int +} + +var ( + _ controllerPool.CapacityObserver = (*CapacityCollector)(nil) + _ prometheus.Collector = (*CapacityCollector)(nil) +) + +func NewCapacityCollector(registerer prometheus.Registerer) (*CapacityCollector, error) { + if registerer == nil { + return nil, ErrInvalidDependencies + } + reads, err := registerCounterVec(registerer, prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "proxy_pool", Subsystem: "controller", Name: "capacity_inventory_reads_total", + Help: "Number of Provider inventory reads by fixed result.", + }, []string{"result"})) + if err != nil { + return nil, err + } + candidate := &CapacityCollector{ + byUpstream: map[string]capacityState{}, reads: reads, + managed: prometheus.NewDesc( + "proxy_pool_controller_capacity_managed_proxies", + "Managed non-extracted proxies from the latest successful samples of active Controller Provider terms.", nil, nil, + ), + available: prometheus.NewDesc( + "proxy_pool_controller_capacity_available_slots", + "Available proxy concurrency slots from the latest successful samples of active Controller Provider terms.", nil, nil, + ), + effective: prometheus.NewDesc( + "proxy_pool_controller_capacity_effective_slots", + "Available slots plus expected pending fetch capacity from the latest successful active Controller samples.", nil, nil, + ), + pending: prometheus.NewDesc( + "proxy_pool_controller_capacity_pending_expected_proxies", + "Expected proxy candidates reserved by in-flight fetches in the latest successful active Controller samples.", nil, nil, + ), + activeUpstreams: prometheus.NewDesc( + "proxy_pool_controller_capacity_active_upstreams", + "Number of Upstreams with a successful inventory sample in this Controller process.", nil, nil, + ), + } + return registerCapacityCollector(registerer, candidate) +} + +func (collector *CapacityCollector) ObserveCapacity(observation controllerPool.CapacityObservation) { + if collector == nil || collector.reads == nil || observation.UpstreamID == "" || observation.SourceID == "" { + return + } + switch observation.Result { + case controllerPool.CapacityReadError: + collector.reads.WithLabelValues(string(observation.Result)).Inc() + return + case controllerPool.CapacityReadSuccess: + if observation.Managed < 0 || observation.AvailableSlots < 0 || observation.EffectiveSlots < 0 || observation.PendingExpected < 0 { + return + } + collector.reads.WithLabelValues(string(observation.Result)).Inc() + default: + return + } + collector.mu.Lock() + collector.byUpstream[observation.UpstreamID] = capacityState{ + sourceID: observation.SourceID, + managed: observation.Managed, available: observation.AvailableSlots, + effective: observation.EffectiveSlots, pending: observation.PendingExpected, + } + collector.mu.Unlock() +} + +func (collector *CapacityCollector) RemoveCapacityUpstream(upstreamID, sourceID string) { + if collector == nil || upstreamID == "" || sourceID == "" { + return + } + collector.mu.Lock() + if state, exists := collector.byUpstream[upstreamID]; exists && state.sourceID == sourceID { + delete(collector.byUpstream, upstreamID) + } + collector.mu.Unlock() +} + +func (collector *CapacityCollector) Describe(descriptions chan<- *prometheus.Desc) { + if collector == nil { + return + } + descriptions <- collector.managed + descriptions <- collector.available + descriptions <- collector.effective + descriptions <- collector.pending + descriptions <- collector.activeUpstreams +} + +func (collector *CapacityCollector) Collect(metrics chan<- prometheus.Metric) { + if collector == nil { + return + } + collector.mu.Lock() + var managed, available, effective, pending int64 + for _, state := range collector.byUpstream { + managed += int64(state.managed) + available += state.available + effective += state.effective + pending += int64(state.pending) + } + activeUpstreams := len(collector.byUpstream) + collector.mu.Unlock() + metrics <- prometheus.MustNewConstMetric(collector.managed, prometheus.GaugeValue, float64(managed)) + metrics <- prometheus.MustNewConstMetric(collector.available, prometheus.GaugeValue, float64(available)) + metrics <- prometheus.MustNewConstMetric(collector.effective, prometheus.GaugeValue, float64(effective)) + metrics <- prometheus.MustNewConstMetric(collector.pending, prometheus.GaugeValue, float64(pending)) + metrics <- prometheus.MustNewConstMetric(collector.activeUpstreams, prometheus.GaugeValue, float64(activeUpstreams)) +} + +func registerCapacityCollector(registerer prometheus.Registerer, candidate *CapacityCollector) (*CapacityCollector, 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 capacity collector: %w", err) + } + existing, ok := registered.ExistingCollector.(*CapacityCollector) + if !ok { + return nil, fmt.Errorf("register capacity collector: existing collector has unexpected type") + } + return existing, nil + } +} diff --git a/internal/platform/metrics/capacity_test.go b/internal/platform/metrics/capacity_test.go new file mode 100644 index 0000000..36655c0 --- /dev/null +++ b/internal/platform/metrics/capacity_test.go @@ -0,0 +1,115 @@ +package metrics + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + + controllerPool "proxy-pool/internal/controller/pool" +) + +func TestCapacityCollectorAggregatesActiveUpstreamsWithoutLabels(t *testing.T) { + registry := prometheus.NewRegistry() + collector, err := NewCapacityCollector(registry) + if err != nil { + t.Fatalf("NewCapacityCollector() error = %v", err) + } + collector.ObserveCapacity(controllerPool.CapacityObservation{ + UpstreamID: "provider-a", SourceID: "term-a", Result: controllerPool.CapacityReadSuccess, + Managed: 2, AvailableSlots: 5, EffectiveSlots: 7, PendingExpected: 2, + }) + collector.ObserveCapacity(controllerPool.CapacityObservation{ + UpstreamID: "provider-b", SourceID: "term-b", Result: controllerPool.CapacityReadSuccess, + Managed: 3, AvailableSlots: 4, EffectiveSlots: 6, PendingExpected: 1, + }) + collector.ObserveCapacity(controllerPool.CapacityObservation{ + UpstreamID: "provider-a", SourceID: "term-a", Result: controllerPool.CapacityReadError, + }) + collector.ObserveCapacity(controllerPool.CapacityObservation{ + UpstreamID: "client-a", SourceID: "term-client", Result: "invalid", Managed: 100, + }) + + assertMetricValue(t, registry, "proxy_pool_controller_capacity_inventory_reads_total", map[string]string{"result": "success"}, 2) + assertMetricValue(t, registry, "proxy_pool_controller_capacity_inventory_reads_total", map[string]string{"result": "error"}, 1) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_managed_proxies", 5) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_available_slots", 9) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_effective_slots", 13) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_pending_expected_proxies", 3) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_active_upstreams", 2) + + collector.RemoveCapacityUpstream("provider-a", "term-a") + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_managed_proxies", 3) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_available_slots", 4) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_effective_slots", 6) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_pending_expected_proxies", 1) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_active_upstreams", 1) +} + +func TestNewCapacityCollectorReusesRegisteredCollectors(t *testing.T) { + registry := prometheus.NewRegistry() + first, err := NewCapacityCollector(registry) + if err != nil { + t.Fatalf("first NewCapacityCollector() error = %v", err) + } + second, err := NewCapacityCollector(registry) + if err != nil { + t.Fatalf("second NewCapacityCollector() error = %v", err) + } + first.ObserveCapacity(controllerPool.CapacityObservation{ + UpstreamID: "provider-a", SourceID: "term-a", Result: controllerPool.CapacityReadSuccess, + Managed: 1, AvailableSlots: 2, EffectiveSlots: 3, + }) + second.ObserveCapacity(controllerPool.CapacityObservation{ + UpstreamID: "provider-b", SourceID: "term-b", Result: controllerPool.CapacityReadSuccess, + Managed: 2, AvailableSlots: 3, EffectiveSlots: 5, PendingExpected: 1, + }) + + assertMetricValue(t, registry, "proxy_pool_controller_capacity_inventory_reads_total", map[string]string{"result": "success"}, 2) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_managed_proxies", 3) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_available_slots", 5) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_effective_slots", 8) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_pending_expected_proxies", 1) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_active_upstreams", 2) +} + +func TestCapacityCollectorKeepsNewTermWhenOldTermStops(t *testing.T) { + registry := prometheus.NewRegistry() + collector, err := NewCapacityCollector(registry) + if err != nil { + t.Fatalf("NewCapacityCollector() error = %v", err) + } + collector.ObserveCapacity(controllerPool.CapacityObservation{ + UpstreamID: "provider-a", SourceID: "old-term", Result: controllerPool.CapacityReadSuccess, + Managed: 1, AvailableSlots: 2, EffectiveSlots: 3, + }) + collector.ObserveCapacity(controllerPool.CapacityObservation{ + UpstreamID: "provider-a", SourceID: "new-term", Result: controllerPool.CapacityReadSuccess, + Managed: 4, AvailableSlots: 5, EffectiveSlots: 6, PendingExpected: 1, + }) + collector.RemoveCapacityUpstream("provider-a", "old-term") + + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_managed_proxies", 4) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_available_slots", 5) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_effective_slots", 6) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_pending_expected_proxies", 1) + assertGaugeValue(t, registry, "proxy_pool_controller_capacity_active_upstreams", 1) +} + +func assertGaugeValue(t *testing.T, registry *prometheus.Registry, name string, want float64) { + 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 len(metric.GetLabel()) == 0 && metric.GetGauge().GetValue() == want { + return + } + } + } + t.Fatalf("gauge %s value=%v was not found", name, want) +} diff --git a/progress.md b/progress.md index 6d30a50..16007fe 100644 --- a/progress.md +++ b/progress.md @@ -2,17 +2,22 @@ ## 2026-08-02 +- 容量可观测性已接入 Provider 补池对账循环：每个成功 Redis 库存读数更新 Controller + 进程级 managed/available/effective/pending/active Gauge，失败只增加固定 + `success`/`error` 的读取计数。Upstream 仅用于进程内聚合，领导任期结束即清除， + 不成为 Prometheus 标签；不进入 Gateway 10 万 QPS 容量热路径，也不增加 Redis 查询。 + 密钥安全的结构化日志仍待实现。 - Extraction 可观测性已接入公用 `MetricsObserver`：Controller 暴露固定 `result` 的请求计数、请求 Proxy 总数和响应交付 Proxy 总数。事件在服务单一出口分类，覆盖 complete/partial/empty/insufficient/idempotency_conflict/rate_limited/unavailable/ invalid/error；不带 Client、Request、Filter、Upstream、Proxy 或错误文本标签。 - 幂等重放统计响应交付而不误计为新的 Redis 消费。容量指标和密钥安全结构化日志仍待实现。 + 幂等重放统计响应交付而不误计为新的 Redis 消费。密钥安全结构化日志仍待实现。 - 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 指标已完成，提取、容量指标和密钥安全的结构化日志仍待实现。 + Provider 指标已完成；密钥安全的结构化日志仍待实现。 - 已实现管理态驱动的即时完整快照刷新：`SnapshotRefreshBroker` 以每个 Worker 流一格的 非阻塞通道广播，连续变更自动合并。Admin 成功提交 Upstream 启停、Routing 切换和配置 发布时触发通知；Worker 保留原有定时刷新作为失效保护。Routing 切换不按旧 Upstream