feat: expose checker task metrics

README.md

@@ -39,6 +39,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费
   库存预留，以及 `AVAILABLE -> EXTRACTED` 的一次性独占提取。
 - **Controller 入口**：`proxy-controller` 已装配 Distribution、Admin、Metrics、
   PostgreSQL 迁移与 Redis 活动池，并支持联动优雅停机。
+- **Checker 指标**：Metrics 启用时暴露 Checker 任务下发与 Observation 接受/拒绝计数；
+  标签仅使用固定检查级别和结果，不记录 Proxy、IP、URL 或凭据。
 - **PostgreSQL 管理面**：持久化配置版本、Upstream/Routing 管理状态、Admin
   审计与 Outbox；不保存 Proxy 明细或逐次提取记录。
 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、

docs/configuration/reference.md

@@ -510,8 +510,11 @@ Runtime；指纹已更新但本地源尚未同步时停止旧 Provider，待源
 
 Metrics 启用时 `listen` 必须是合法 `host:port`。该入口固定提供 `/livez`、
 `/readyz` 和 `/metrics`，不复用 Distribution/Admin 的认证边界；外部访问必须由
-网络策略限制。当前 `/metrics` 已包含 Go/进程基础指标，Provider、提取和容量等
+网络策略限制。当前 `/metrics` 包含 Go/进程基础指标，以及 Checker 的
-业务指标仍在后续实施范围。Distribution 启用时 `/readyz` 只以 Redis 活动池为
+`proxy_pool_checker_tasks_dispatched_total{level}` 和
+`proxy_pool_checker_observations_total{level,result}`。`level` 固定为 BASIC、EGRESS、TARGET，
+`result` 固定为 accepted、rejected；绝不包含 Proxy、IP、Checker、Client、URL 或凭据标签。
+Provider、提取和容量等业务指标仍在后续实施范围。Distribution 启用时 `/readyz` 只以 Redis 活动池为
 服务流量门槛，PostgreSQL 故障由 Admin 接口独立报告。Metrics 开关或监听地址
 变更需要重启 Controller。
 

docs/development/implementation-plan.md

@@ -332,6 +332,12 @@ Profile 在启用 Routing 与 Upstream 的组合上才进入调度。
 - [ ] Document backup, recovery, rollout, rollback, capacity, kernel, file descriptor,
   NAT/conntrack, and incident runbooks.
 
+当前进度（2026-08-02）：已接入 Checker 任务流指标
+`proxy_pool_checker_tasks_dispatched_total{level}` 与
+`proxy_pool_checker_observations_total{level,result}`；标签值仅允许固定的
+`BASIC`、`EGRESS`、`TARGET` 级别及 `accepted`、`rejected` 结果。Provider、提取、
+容量指标与密钥安全的结构化日志仍待补齐。
+
 ## Task 14: Documentation, Examples, and Diagrams
 
 **Files:** `docs/**`, `examples/**`, `diagrams/**`

docs/operations/runbook.md

@@ -128,6 +128,14 @@ kubectl -n proxy-pool rollout status deployment/proxy-gateway --timeout=10m
 
 探针不得执行 Provider 请求或完整数据库扫描。
 
+Checker 指标使用固定标签集：
+
+- `proxy_pool_checker_tasks_dispatched_total{level}`：成功写入 Checker 任务流后的任务数。
+- `proxy_pool_checker_observations_total{level,result}`：已解码的 Observation 被 Controller 接受或拒绝的数量。
+
+`level` 仅为 BASIC、EGRESS、TARGET，`result` 仅为 accepted、rejected。不得将
+Proxy ID、IP、Checker ID、目标 URL、Client ID 或凭据加入指标标签。
+
 ### 3.4 发布顺序与兼容性
 
 1. 先做向前兼容数据库迁移。

internal/controller/bootstrap/bootstrap.go

@@ -23,6 +23,7 @@ import (
 	"proxy-pool/internal/controller/worker"
 	"proxy-pool/internal/domain/activitypool"
 	extractionDomain "proxy-pool/internal/domain/extraction"
+	healthDomain "proxy-pool/internal/domain/health"
 	ownershipDomain "proxy-pool/internal/domain/ownership"
 	"proxy-pool/internal/domain/workerruntime"
 	"proxy-pool/internal/platform/admission"
@@ -219,10 +220,16 @@ func runWithWorkerFactory(
 		}
 		dependencies.AdminService = service
 	}
+	var checkerMetrics healthDomain.TaskMetricsObserver
 	if loaded.Value.Metrics.Enabled {
 		if nilInterface(opened.metricsReadiness) {
 			return errors.Join(ErrStartup, ErrInvalidOptions)
 		}
+		collector, collectorErr := platformMetrics.NewCheckerCollector(prometheus.DefaultRegisterer)
+		if collectorErr != nil {
+			return fmt.Errorf("%w: build Checker metrics: %w", ErrStartup, collectorErr)
+		}
+		checkerMetrics = collector
 		handler, handlerErr := platformMetrics.NewHandler(platformMetrics.Dependencies{
 			Gatherer: prometheus.DefaultGatherer, Readiness: opened.metricsReadiness,
 		})
@@ -289,6 +296,7 @@ func runWithWorkerFactory(
 			return fmt.Errorf("%w: build Checker identity authorizer: %w", ErrStartup, identityErr)
 		}
 		checkerOptions := controllerHealth.DefaultGRPCHandlerOptions()
+		checkerOptions.Metrics = checkerMetrics
 		if tasks, ok := opened.activity.(healthTaskRuntime); ok && !nilInterface(tasks) {
 			checkerOptions.TaskBroker = tasks
 		}

internal/controller/health/grpc_handler.go

@@ -31,6 +31,7 @@ type GRPCHandlerOptions struct {
 	MaxObservationsPerBatch int
 	MaxTasksPerClaim        int
 	TaskBroker              TaskBroker
+	Metrics                 healthDomain.TaskMetricsObserver
 	Now                     func() time.Time
 }
 
@@ -122,6 +123,9 @@ func (handler *GRPCHandler) StreamCheckTasks(
 		if sendErr := stream.Send(wire); sendErr != nil {
 			return sendErr
 		}
+		if handler.options.Metrics != nil {
+			handler.options.Metrics.ObserveTaskDispatch(task.Level, 1)
+		}
 	}
 	return nil
 }
@@ -160,10 +164,16 @@ func (handler *GRPCHandler) ReportObservations(
 		}
 		if err == nil {
 			response.Accepted++
+			if handler.options.Metrics != nil {
+				handler.options.Metrics.ObserveObservation(observation.Level, healthDomain.ObservationMetricAccepted)
+			}
 			continue
 		}
 		if rejectedObservationError(err) {
 			response.Rejected++
+			if handler.options.Metrics != nil && observation.Level != "" {
+				handler.options.Metrics.ObserveObservation(observation.Level, healthDomain.ObservationMetricRejected)
+			}
 			continue
 		}
 		return nil, healthGRPCError(err)

internal/controller/health/grpc_handler_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"io"
 	"net"
+	"sync"
 	"testing"
 	"time"
 
@@ -127,8 +128,9 @@ func TestGRPCHandlerStreamsLeasedTasksAndFencesReportedFacts(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewReducer(): %v", err)
 	}
+	metrics := &recordingTaskMetrics{}
 	handler, err := NewGRPCHandler(reducer, &recordingCheckerIdentity{}, GRPCHandlerOptions{
-		MaxObservationsPerBatch: 2, MaxTasksPerClaim: 2, TaskBroker: broker, Now: func() time.Time { return now },
+		MaxObservationsPerBatch: 2, MaxTasksPerClaim: 2, TaskBroker: broker, Metrics: metrics, Now: func() time.Time { return now },
 	})
 	if err != nil {
 		t.Fatalf("NewGRPCHandler(): %v", err)
@@ -171,6 +173,11 @@ func TestGRPCHandlerStreamsLeasedTasksAndFencesReportedFacts(t *testing.T) {
 	if err != nil || rejected.GetRejected() != 1 || len(global.commands) != 1 {
 		t.Fatalf("ReportObservations(other checker) = (%+v, %v), global=%d", rejected, err, len(global.commands))
 	}
+	if metrics.dispatches(healthDomain.LevelBasic) != 1 ||
+		metrics.observations(healthDomain.LevelBasic, healthDomain.ObservationMetricAccepted) != 1 ||
+		metrics.observations(healthDomain.LevelBasic, healthDomain.ObservationMetricRejected) != 1 {
+		t.Fatalf("Checker metrics = %+v", metrics)
+	}
 }
 
 func TestGRPCHandlerStreamsEgressProbeURLAndAcceptsGlobalFact(t *testing.T) {
@@ -265,6 +272,42 @@ type recordingCheckerIdentity struct {
 	err       error
 }
 
+type recordingTaskMetrics struct {
+	mu          sync.Mutex
+	dispatch    map[healthDomain.Level]int
+	observation map[string]int
+}
+
+func (metrics *recordingTaskMetrics) ObserveTaskDispatch(level healthDomain.Level, count int) {
+	metrics.mu.Lock()
+	defer metrics.mu.Unlock()
+	if metrics.dispatch == nil {
+		metrics.dispatch = make(map[healthDomain.Level]int)
+	}
+	metrics.dispatch[level] += count
+}
+
+func (metrics *recordingTaskMetrics) ObserveObservation(level healthDomain.Level, result healthDomain.ObservationMetricResult) {
+	metrics.mu.Lock()
+	defer metrics.mu.Unlock()
+	if metrics.observation == nil {
+		metrics.observation = make(map[string]int)
+	}
+	metrics.observation[string(level)+"\x00"+string(result)]++
+}
+
+func (metrics *recordingTaskMetrics) dispatches(level healthDomain.Level) int {
+	metrics.mu.Lock()
+	defer metrics.mu.Unlock()
+	return metrics.dispatch[level]
+}
+
+func (metrics *recordingTaskMetrics) observations(level healthDomain.Level, result healthDomain.ObservationMetricResult) int {
+	metrics.mu.Lock()
+	defer metrics.mu.Unlock()
+	return metrics.observation[string(level)+"\x00"+string(result)]
+}
+
 func (identity *recordingCheckerIdentity) AuthorizeChecker(_ context.Context, checkerID string) error {
 	identity.checkerID = checkerID
 	return identity.err

internal/domain/health/metrics.go

@@ -0,0 +1,16 @@
+package health
+
+// TaskMetricsObserver accepts only the low-cardinality dimensions emitted by
+// Checker task processing. Implementations must not add task, proxy, checker,
+// endpoint, target URL, or credential values as metric labels.
+type TaskMetricsObserver interface {
+	ObserveTaskDispatch(Level, int)
+	ObserveObservation(Level, ObservationMetricResult)
+}
+
+type ObservationMetricResult string
+
+const (
+	ObservationMetricAccepted ObservationMetricResult = "accepted"
+	ObservationMetricRejected ObservationMetricResult = "rejected"
+)

internal/platform/metrics/checker.go

@@ -0,0 +1,80 @@
+package metrics
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	healthDomain "proxy-pool/internal/domain/health"
+)
+
+// CheckerCollector exposes bounded Checker task-flow counters. The only
+// labels are the three fixed health levels and accepted/rejected outcome.
+type CheckerCollector struct {
+	tasksDispatched *prometheus.CounterVec
+	observations    *prometheus.CounterVec
+}
+
+var _ healthDomain.TaskMetricsObserver = (*CheckerCollector)(nil)
+
+func NewCheckerCollector(registerer prometheus.Registerer) (*CheckerCollector, error) {
+	if registerer == nil {
+		return nil, ErrInvalidDependencies
+	}
+	tasks, err := registerCounterVec(registerer, prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "proxy_pool", Subsystem: "checker", Name: "tasks_dispatched_total",
+		Help: "Number of checker tasks dispatched after a successful task-stream send.",
+	}, []string{"level"}))
+	if err != nil {
+		return nil, err
+	}
+	observations, err := registerCounterVec(registerer, prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "proxy_pool", Subsystem: "checker", Name: "observations_total",
+		Help: "Number of checker observations accepted or rejected after decoding a known level.",
+	}, []string{"level", "result"}))
+	if err != nil {
+		return nil, err
+	}
+	return &CheckerCollector{tasksDispatched: tasks, observations: observations}, nil
+}
+
+func (collector *CheckerCollector) ObserveTaskDispatch(level healthDomain.Level, count int) {
+	if collector == nil || collector.tasksDispatched == nil || count <= 0 || !validCheckerLevel(level) {
+		return
+	}
+	collector.tasksDispatched.WithLabelValues(string(level)).Add(float64(count))
+}
+
+func (collector *CheckerCollector) ObserveObservation(level healthDomain.Level, result healthDomain.ObservationMetricResult) {
+	if collector == nil || collector.observations == nil || !validCheckerLevel(level) ||
+		(result != healthDomain.ObservationMetricAccepted && result != healthDomain.ObservationMetricRejected) {
+		return
+	}
+	collector.observations.WithLabelValues(string(level), string(result)).Inc()
+}
+
+func registerCounterVec(registerer prometheus.Registerer, candidate *prometheus.CounterVec) (*prometheus.CounterVec, error) {
+	if err := registerer.Register(candidate); err == nil {
+		return candidate, nil
+	} else {
+		var registered prometheus.AlreadyRegisteredError
+		if !errors.As(err, &registered) {
+			return nil, fmt.Errorf("register checker metrics: %w", err)
+		}
+		existing, ok := registered.ExistingCollector.(*prometheus.CounterVec)
+		if !ok {
+			return nil, fmt.Errorf("register checker metrics: existing collector has unexpected type")
+		}
+		return existing, nil
+	}
+}
+
+func validCheckerLevel(level healthDomain.Level) bool {
+	switch level {
+	case healthDomain.LevelBasic, healthDomain.LevelEgress, healthDomain.LevelTarget:
+		return true
+	default:
+		return false
+	}
+}

internal/platform/metrics/checker_test.go

@@ -0,0 +1,79 @@
+package metrics
+
+import (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+
+	healthDomain "proxy-pool/internal/domain/health"
+)
+
+func TestCheckerCollectorRecordsOnlyFixedDimensions(t *testing.T) {
+	registry := prometheus.NewRegistry()
+	collector, err := NewCheckerCollector(registry)
+	if err != nil {
+		t.Fatalf("NewCheckerCollector() = %v", err)
+	}
+	collector.ObserveTaskDispatch(healthDomain.LevelBasic, 2)
+	collector.ObserveTaskDispatch(healthDomain.LevelTarget, 1)
+	collector.ObserveTaskDispatch("invalid", 99)
+	collector.ObserveObservation(healthDomain.LevelBasic, healthDomain.ObservationMetricAccepted)
+	collector.ObserveObservation(healthDomain.LevelTarget, healthDomain.ObservationMetricRejected)
+	collector.ObserveObservation("invalid", healthDomain.ObservationMetricAccepted)
+
+	assertMetricValue(t, registry, "proxy_pool_checker_tasks_dispatched_total", map[string]string{"level": "BASIC"}, 2)
+	assertMetricValue(t, registry, "proxy_pool_checker_tasks_dispatched_total", map[string]string{"level": "TARGET"}, 1)
+	assertMetricValue(t, registry, "proxy_pool_checker_observations_total", map[string]string{
+		"level": "BASIC", "result": "accepted",
+	}, 1)
+	assertMetricValue(t, registry, "proxy_pool_checker_observations_total", map[string]string{
+		"level": "TARGET", "result": "rejected",
+	}, 1)
+}
+
+func TestNewCheckerCollectorReusesRegisteredCollectors(t *testing.T) {
+	registry := prometheus.NewRegistry()
+	first, err := NewCheckerCollector(registry)
+	if err != nil {
+		t.Fatalf("first NewCheckerCollector() = %v", err)
+	}
+	second, err := NewCheckerCollector(registry)
+	if err != nil {
+		t.Fatalf("second NewCheckerCollector() = %v", err)
+	}
+	first.ObserveTaskDispatch(healthDomain.LevelEgress, 1)
+	second.ObserveTaskDispatch(healthDomain.LevelEgress, 1)
+	assertMetricValue(t, registry, "proxy_pool_checker_tasks_dispatched_total", map[string]string{"level": "EGRESS"}, 2)
+}
+
+func assertMetricValue(t *testing.T, registry *prometheus.Registry, name string, labels map[string]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 metricLabelsMatch(metric.GetLabel(), labels) && metric.GetCounter().GetValue() == want {
+				return
+			}
+		}
+	}
+	t.Fatalf("metric %s labels=%v value=%v was not found", name, labels, want)
+}
+
+func metricLabelsMatch(labels []*dto.LabelPair, want map[string]string) bool {
+	if len(labels) != len(want) {
+		return false
+	}
+	for _, label := range labels {
+		if want[label.GetName()] != label.GetValue() {
+			return false
+		}
+	}
+	return true
+}