diff --git a/README.md b/README.md index f7305d4..a135dd3 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,9 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 标签仅使用固定检查级别和结果，不记录 Proxy、IP、URL 或凭据。 - **Gateway 指标**：Metrics 启用时暴露代理尝试的固定阶段成功/失败计数，以及本地 Outcome 队列满后的丢弃计数；不记录 Proxy、路由、目标、客户端或凭据。 +- **Drain 指标**：Controller 暴露 `proxy_pool_controller_drain_candidates_total` 与 + `proxy_pool_controller_drains_started_total`；`reason` 仅有 `unhealthy` 与 + `upstream_disabled`，不包含 Proxy、Worker、Upstream、会话或地址。 - **PostgreSQL 管理面**：持久化配置版本、Upstream/Routing 管理状态、Admin 审计与 Outbox；不保存 Proxy 明细或逐次提取记录。 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 7c97f67..18b9505 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`，均以注册表测试固定标签集。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、提取和容量业务指标待实现 | | TEST-001 | 覆盖对话中列出的 11 个关键并发与故障场景 | 9030-9082 | 测试清单；Redis 活动池由 Memory/Redis 公用契约覆盖，跨进程故障场景仍按清单推进 | diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index 2ef906b..5fba07f 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -221,6 +221,7 @@ func runWithWorkerFactory( dependencies.AdminService = service } var checkerMetrics healthDomain.TaskMetricsObserver + var drainMetrics healthDomain.DrainMetricsObserver if loaded.Value.Metrics.Enabled { if nilInterface(opened.metricsReadiness) { return errors.Join(ErrStartup, ErrInvalidOptions) @@ -230,6 +231,11 @@ func runWithWorkerFactory( return fmt.Errorf("%w: build Checker metrics: %w", ErrStartup, collectorErr) } checkerMetrics = collector + drainCollector, drainCollectorErr := platformMetrics.NewDrainCollector(prometheus.DefaultRegisterer) + if drainCollectorErr != nil { + return fmt.Errorf("%w: build Drain metrics: %w", ErrStartup, drainCollectorErr) + } + drainMetrics = drainCollector handler, handlerErr := platformMetrics.NewHandler(platformMetrics.Dependencies{ Gatherer: prometheus.DefaultGatherer, Readiness: opened.metricsReadiness, }) @@ -331,7 +337,7 @@ func runWithWorkerFactory( configurationStore, unhealthy, controllerHealth.UnhealthyReaperOptions{ - PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: options.Now, + PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: options.Now, Metrics: drainMetrics, }, ) if reaperErr != nil { @@ -355,7 +361,7 @@ func runWithWorkerFactory( policySource, drainStore, controllerHealth.UpstreamDrainReaperOptions{ - PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: options.Now, + PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: options.Now, Metrics: drainMetrics, }, ) if reaperErr != nil { diff --git a/internal/controller/health/unhealthy_reaper.go b/internal/controller/health/unhealthy_reaper.go index 684d081..69cc105 100644 --- a/internal/controller/health/unhealthy_reaper.go +++ b/internal/controller/health/unhealthy_reaper.go @@ -7,6 +7,7 @@ import ( "proxy-pool/internal/config" "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" ) var ErrInvalidUnhealthyReaper = errors.New("invalid unhealthy reaper") @@ -15,6 +16,7 @@ type UnhealthyReaperOptions struct { PollInterval time.Duration BatchSize int Now func() time.Time + Metrics healthDomain.DrainMetricsObserver } // ConfiguredUnhealthyReaper applies the current upstream grace policies to a @@ -62,12 +64,23 @@ func (reaper *ConfiguredUnhealthyReaper) Tick(ctx context.Context) (activitypool } drainer, supported := reaper.store.(activitypool.UnhealthyDrainStarter) if !supported || nilInterface(drainer) { + if reaper.options.Metrics != nil { + reaper.options.Metrics.ObserveDrain(healthDomain.DrainReasonUnhealthy, len(result.DrainCandidates), 0) + } return result, nil } + started := 0 for _, candidate := range result.DrainCandidates { - if _, err := drainer.BeginUnhealthyDrain(ctx, utcNow, candidate); err != nil { + didStart, err := drainer.BeginUnhealthyDrain(ctx, utcNow, candidate) + if err != nil { return result, err } + if didStart { + started++ + } + } + if reaper.options.Metrics != nil { + reaper.options.Metrics.ObserveDrain(healthDomain.DrainReasonUnhealthy, len(result.DrainCandidates), started) } return result, nil } diff --git a/internal/controller/health/unhealthy_reaper_test.go b/internal/controller/health/unhealthy_reaper_test.go index 2d13db9..43296c0 100644 --- a/internal/controller/health/unhealthy_reaper_test.go +++ b/internal/controller/health/unhealthy_reaper_test.go @@ -7,6 +7,7 @@ import ( "proxy-pool/internal/config" "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" ) type unhealthyRemoverStub struct { @@ -85,11 +86,12 @@ func TestConfiguredUnhealthyReaperStartsBoundedDrainCandidates(t *testing.T) { store := &unhealthyRemoverStub{result: activitypool.UnhealthySweepResult{ DeferredOwned: 1, DrainCandidates: []activitypool.UnhealthyDrainCandidate{candidate}, }} + metrics := &recordingDrainMetrics{} reaper, err := NewConfiguredUnhealthyReaper(&configurationSourceStub{configuration: &config.Config{ Upstreams: map[string]config.Upstream{ "provider-a": {Enabled: true, Check: config.Check{UnhealthyRemoveAfter: config.Duration(time.Minute)}}, }, - }}, store, UnhealthyReaperOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }}) + }}, store, UnhealthyReaperOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }, Metrics: metrics}) if err != nil { t.Fatalf("NewConfiguredUnhealthyReaper() = %v", err) } @@ -100,4 +102,9 @@ func TestConfiguredUnhealthyReaperStartsBoundedDrainCandidates(t *testing.T) { if store.drainCalls != 1 || !store.drainNow.Equal(now) || len(store.candidates) != 1 || store.candidates[0] != candidate { t.Fatalf("BeginUnhealthyDrain() calls=%d now=%v candidates=%+v", store.drainCalls, store.drainNow, store.candidates) } + if len(metrics.records) != 1 || metrics.records[0] != (drainMetricRecord{ + reason: healthDomain.DrainReasonUnhealthy, candidates: 1, started: 1, + }) { + t.Fatalf("Drain metrics = %+v", metrics.records) + } } diff --git a/internal/controller/health/upstream_drain_reaper.go b/internal/controller/health/upstream_drain_reaper.go index 00f24a0..4432132 100644 --- a/internal/controller/health/upstream_drain_reaper.go +++ b/internal/controller/health/upstream_drain_reaper.go @@ -9,6 +9,7 @@ import ( "proxy-pool/internal/config" "proxy-pool/internal/domain/activitypool" "proxy-pool/internal/domain/adminstate" + healthDomain "proxy-pool/internal/domain/health" ) var ErrInvalidUpstreamDrainReaper = errors.New("invalid upstream drain reaper") @@ -120,6 +121,7 @@ type UpstreamDrainReaperOptions struct { PollInterval time.Duration BatchSize int Now func() time.Time + Metrics healthDomain.DrainMetricsObserver } type UpstreamDrainResult struct { @@ -192,6 +194,9 @@ func (reaper *ConfiguredUpstreamDrainReaper) Tick(ctx context.Context) (Upstream } remaining -= len(candidates) } + if reaper.options.Metrics != nil { + reaper.options.Metrics.ObserveDrain(healthDomain.DrainReasonUpstreamDisabled, result.Candidates, result.Started) + } return result, nil } diff --git a/internal/controller/health/upstream_drain_reaper_test.go b/internal/controller/health/upstream_drain_reaper_test.go index b11a3d2..edc2b75 100644 --- a/internal/controller/health/upstream_drain_reaper_test.go +++ b/internal/controller/health/upstream_drain_reaper_test.go @@ -8,6 +8,7 @@ import ( "proxy-pool/internal/config" "proxy-pool/internal/domain/activitypool" "proxy-pool/internal/domain/adminstate" + healthDomain "proxy-pool/internal/domain/health" proxyDomain "proxy-pool/internal/domain/proxy" ) @@ -31,6 +32,18 @@ func (source upstreamDrainStateStub) Snapshot(context.Context) (adminstate.Snaps return source.snapshot, nil } +type drainMetricRecord struct { + reason healthDomain.DrainReason + candidates int + started int +} + +type recordingDrainMetrics struct{ records []drainMetricRecord } + +func (metrics *recordingDrainMetrics) ObserveDrain(reason healthDomain.DrainReason, candidates, started int) { + metrics.records = append(metrics.records, drainMetricRecord{reason: reason, candidates: candidates, started: started}) +} + func (source upstreamDrainConfigurationStub) Snapshot() (*config.Config, uint64) { return source.configuration, source.revision } @@ -51,8 +64,9 @@ func TestConfiguredUpstreamDrainReaperStartsBoundedDisabledAssignments(t *testin t.Fatalf("Assign() error = %v", err) } policy := activitypool.UpstreamDrainPolicy{UpstreamID: "provider-a", Revision: 21} + metrics := &recordingDrainMetrics{} reaper, err := NewConfiguredUpstreamDrainReaper(upstreamDrainPolicySourceStub{policies: []activitypool.UpstreamDrainPolicy{policy}}, pool, - UpstreamDrainReaperOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }}) + UpstreamDrainReaperOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }, Metrics: metrics}) if err != nil { t.Fatalf("NewConfiguredUpstreamDrainReaper() error = %v", err) } @@ -64,6 +78,11 @@ func TestConfiguredUpstreamDrainReaperStartsBoundedDisabledAssignments(t *testin current.Epoch != assignment.Epoch || !current.Draining { t.Fatalf("Get(after drain) = %+v, %t, %v", current, found, err) } + if len(metrics.records) != 1 || metrics.records[0] != (drainMetricRecord{ + reason: healthDomain.DrainReasonUpstreamDisabled, candidates: 1, started: 1, + }) { + t.Fatalf("Drain metrics = %+v", metrics.records) + } } func TestEffectiveUpstreamDrainPolicySourceUsesStableConfigOnlyRevision(t *testing.T) { diff --git a/internal/domain/health/metrics.go b/internal/domain/health/metrics.go index 746c10b..2dacef3 100644 --- a/internal/domain/health/metrics.go +++ b/internal/domain/health/metrics.go @@ -8,6 +8,22 @@ type TaskMetricsObserver interface { ObserveObservation(Level, ObservationMetricResult) } +// DrainReason is intentionally finite so Controller metrics cannot expose +// upstream, proxy, Worker, session or endpoint identity as labels. +type DrainReason string + +const ( + DrainReasonUnhealthy DrainReason = "unhealthy" + DrainReasonUpstreamDisabled DrainReason = "upstream_disabled" +) + +// DrainMetricsObserver records bounded Controller-side Drain attempts after a +// complete sweep. Candidate and started counts are separate because a fenced +// candidate may legitimately become stale before the ownership transition. +type DrainMetricsObserver interface { + ObserveDrain(reason DrainReason, candidates, started int) +} + type ObservationMetricResult string const ( diff --git a/internal/platform/metrics/drain.go b/internal/platform/metrics/drain.go new file mode 100644 index 0000000..50d06e1 --- /dev/null +++ b/internal/platform/metrics/drain.go @@ -0,0 +1,58 @@ +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + + healthDomain "proxy-pool/internal/domain/health" +) + +// DrainCollector exposes only fixed Controller Drain reasons. It never uses +// proxy, Upstream, Worker, session, routing or endpoint values as labels. +type DrainCollector struct { + candidates *prometheus.CounterVec + started *prometheus.CounterVec +} + +var _ healthDomain.DrainMetricsObserver = (*DrainCollector)(nil) + +func NewDrainCollector(registerer prometheus.Registerer) (*DrainCollector, error) { + if registerer == nil { + return nil, ErrInvalidDependencies + } + candidates, err := registerCounterVec(registerer, prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "proxy_pool", Subsystem: "controller", Name: "drain_candidates_total", + Help: "Number of bounded Controller Drain candidates selected by reason.", + }, []string{"reason"})) + if err != nil { + return nil, err + } + started, err := registerCounterVec(registerer, prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "proxy_pool", Subsystem: "controller", Name: "drains_started_total", + Help: "Number of Controller Drain tickets started by reason.", + }, []string{"reason"})) + if err != nil { + return nil, err + } + return &DrainCollector{candidates: candidates, started: started}, nil +} + +func (collector *DrainCollector) ObserveDrain(reason healthDomain.DrainReason, candidates, started int) { + if collector == nil || !validDrainReason(reason) || candidates < 0 || started < 0 || started > candidates { + return + } + if candidates > 0 && collector.candidates != nil { + collector.candidates.WithLabelValues(string(reason)).Add(float64(candidates)) + } + if started > 0 && collector.started != nil { + collector.started.WithLabelValues(string(reason)).Add(float64(started)) + } +} + +func validDrainReason(reason healthDomain.DrainReason) bool { + switch reason { + case healthDomain.DrainReasonUnhealthy, healthDomain.DrainReasonUpstreamDisabled: + return true + default: + return false + } +} diff --git a/internal/platform/metrics/drain_test.go b/internal/platform/metrics/drain_test.go new file mode 100644 index 0000000..64326c7 --- /dev/null +++ b/internal/platform/metrics/drain_test.go @@ -0,0 +1,42 @@ +package metrics + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + + healthDomain "proxy-pool/internal/domain/health" +) + +func TestDrainCollectorRecordsOnlyFixedReasons(t *testing.T) { + registry := prometheus.NewRegistry() + collector, err := NewDrainCollector(registry) + if err != nil { + t.Fatalf("NewDrainCollector() error = %v", err) + } + collector.ObserveDrain(healthDomain.DrainReasonUnhealthy, 3, 2) + collector.ObserveDrain(healthDomain.DrainReasonUpstreamDisabled, 2, 1) + collector.ObserveDrain("invalid", 9, 9) + collector.ObserveDrain(healthDomain.DrainReasonUnhealthy, 1, 2) + + assertMetricValue(t, registry, "proxy_pool_controller_drain_candidates_total", map[string]string{"reason": "unhealthy"}, 3) + assertMetricValue(t, registry, "proxy_pool_controller_drains_started_total", map[string]string{"reason": "unhealthy"}, 2) + assertMetricValue(t, registry, "proxy_pool_controller_drain_candidates_total", map[string]string{"reason": "upstream_disabled"}, 2) + assertMetricValue(t, registry, "proxy_pool_controller_drains_started_total", map[string]string{"reason": "upstream_disabled"}, 1) +} + +func TestNewDrainCollectorReusesRegisteredCollectors(t *testing.T) { + registry := prometheus.NewRegistry() + first, err := NewDrainCollector(registry) + if err != nil { + t.Fatalf("first NewDrainCollector() error = %v", err) + } + second, err := NewDrainCollector(registry) + if err != nil { + t.Fatalf("second NewDrainCollector() error = %v", err) + } + first.ObserveDrain(healthDomain.DrainReasonUnhealthy, 1, 1) + second.ObserveDrain(healthDomain.DrainReasonUnhealthy, 2, 2) + assertMetricValue(t, registry, "proxy_pool_controller_drain_candidates_total", map[string]string{"reason": "unhealthy"}, 3) + assertMetricValue(t, registry, "proxy_pool_controller_drains_started_total", map[string]string{"reason": "unhealthy"}, 3) +} diff --git a/progress.md b/progress.md index b23535a..68e0aec 100644 --- a/progress.md +++ b/progress.md @@ -2,6 +2,10 @@ ## 2026-08-02 +- Drain 可观测性已落地：`proxy_pool_controller_drain_candidates_total` 与 + `proxy_pool_controller_drains_started_total` 只允许 `unhealthy` 和 + `upstream_disabled` 两种 reason。健康异常与上游停用 Reaper 复用领域接口上报， + 不产生 Proxy、Worker、上游或地址标签；Collector 注册复用与两类 Reaper 都有单测。 - 已实现上游停用的自动 Drain：有效策略由静态配置与 PostgreSQL Admin 管理态共同计算， Controller 持续发布到 Redis；每个候选绑定上游 revision，只从 `owned(upstream)` 有界索引 读取。Redis 在创建 Drain Ticket 前原子复核策略 revision、Proxy 上游、Worker、assignment diff --git a/task_plan.md b/task_plan.md index ba92213..5cb7817 100644 --- a/task_plan.md +++ b/task_plan.md @@ -44,7 +44,8 @@ 刷新，以及 Drain 发起时的持久化 Ticket、Worker 待绑定索引、ownership epoch 推进和 完整排除 Snapshot 屏障绑定；已在 ACK 与 Runtime 零计数之间完成 Redis 原子自动收尾。 配置与管理态共同决定的上游停用已接入同一 `BeginDrain` 编排：候选绑定上游 revision， - Redis 原子复核策略、归属与 assignment epoch。后续补齐可观测闭环。 + Redis 原子复核策略、归属与 assignment epoch；两类自动 Drain 均已接入低基数 + Prometheus 指标。后续补齐更完整的运行态可观测闭环。 ## 串并行关系