feat: expose controller drain metrics

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、重试、超时、

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 公用契约覆盖，跨进程故障场景仍按清单推进 |

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 {

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
 }

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)
+	}
 }

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
 }
 

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) {

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 (

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
+	}
+}

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)
+}

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

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 指标。后续补齐更完整的运行态可观测闭环。
 
 ## 串并行关系