From 707aff26ea377d91382a4f96e84fd41e247cc809 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 14:45:47 +0800 Subject: [PATCH] feat: observe gateway capacity invariants --- README.md | 2 +- docs/development/implementation-plan.md | 10 ++- docs/requirements/completion-audit.md | 6 +- docs/requirements/traceability.md | 4 +- internal/domain/proxy/capacity.go | 73 +++++++++++++++-- internal/domain/proxy/capacity_test.go | 90 +++++++++++++++++++++ internal/gateway/bootstrap/bootstrap.go | 7 +- internal/gateway/snapshot/store.go | 41 +++++++--- internal/gateway/snapshot/store_test.go | 99 +++++++++++++++++++++++ internal/platform/metrics/gateway.go | 40 ++++++++- internal/platform/metrics/gateway_test.go | 8 ++ task_plan.md | 4 +- 12 files changed, 345 insertions(+), 39 deletions(-) diff --git a/README.md b/README.md index 6b96de7..8ecb782 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,7 @@ flowchart LR ## 当前完成度 -截至 **2026-08-02**，实施计划中可直接勾选的检查项为 **58 / 74（78.4%）**。详情见 +截至 **2026-08-02**，实施计划中可直接勾选的检查项为 **59 / 74（79.7%）**。详情见 [实施计划](docs/development/implementation-plan.md)和 [交付完成度审计](docs/requirements/completion-audit.md)。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index c725e40..a4184e3 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -82,12 +82,16 @@ test/{fixtures,integration,e2e,load}/ - [x] Implement state transitions and reject illegal transitions. - [ ] Implement sharded runtime counters with CAS Reserve, Commit, Cancel, Release. - [x] Prove with 1,000 concurrent goroutines that effective capacity is never exceeded. -- [ ] Add race coverage and duplicate-release invariant metrics hook. +- [x] Add race coverage and duplicate-release invariant metrics hook. -当前进度（2026-07-29）：固定 Max 下的每 Proxy 打包 CAS、Cancel/Commit/Release +当前进度（2026-08-02）：固定 Max 下的每 Proxy 打包 CAS、Cancel/Commit/Release 生命周期、重复终结、错误顺序和同一 Reservation 并发终结已通过领域测试；已退出 Proxy 会停止新预留、保留非零 Drain 计数，并在归零后由后续 Snapshot Apply 回收。 -动态降容契约、低基数不变量指标和 Linux race 证据仍待完成。 +Snapshot 降低 `maxConcurrency` 时会保留既有 Active 工作、禁止新预留，直到运行计数 +低于新上限；运行态复用、降容和恢复预留均有回归测试。领域的 +`CapacityInvariantObserver` 只发出固定生命周期违规枚举，Gateway 指标 +`proxy_pool_gateway_capacity_invariant_violations_total{operation}` 不含 Proxy、请求或 +Worker 维度；该测试纳入 Linux CI 的 race 范围，本机因 `CGO_ENABLED=0` 未执行 race。 ## Task 4: Routing and Sequential Switching diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 9be25b9..6f4ab6b 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -105,7 +105,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 `proxy-controller` 已完成 Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，`proxy-gateway` 已完成 HTTP/Metrics 与控制面 Session 装配；Checker、Gateway Outcome、Drain、Provider、 - Extraction 与 Capacity 的低基数业务指标已闭环；Controller、Gateway、Checker 的 + Extraction 与 Capacity 的低基数业务指标已闭环；Gateway Capacity 的动态降容保留 + 已有连接并在其排空前停止新增预留，非法 Reservation 终结汇总为固定枚举指标；Controller、Gateway、Checker 的 进程级致命错误现使用统一 JSON 脱敏日志出口，不输出错误原文。 2. Gateway 的生产连接池调优与代表性流量压测。 3. Provider 分布式 singleflight/Leader、长期凭据回收和累计额度执行器。 @@ -125,9 +126,6 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS 和 disabled candidate 语义。 -11. 补齐 Proxy Capacity 动态降容契约、Reservation 全生命周期观测；短 TTL - Proxy 的零计数运行态已在后续 Snapshot Apply 中回收。 - ## 4. 容量结论 100,000 QPS 是集群设计输入，不是本次验证结果。只有实现上述运行时，并在 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index a1fd3d5..538d3e9 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -45,7 +45,7 @@ | PROXY-001 | Proxy 保存协议、地址、凭据引用、来源、TTL、健康、容量和标签 | 71-105, 8605-8678 | Domain 类型与序列化测试 | | PROXY-002 | 唯一键包含 scheme、host、port、username、credentialVersion | 6655-6727, 8605-8678 | 去重单测 | | PROXY-003 | TTL 来源优先级明确并统一 UTC | 681-747, 8655-8678 | TTL 表驱动测试 | -| CAP-001 | Gateway 分配使用 Reserved -> Active 原子转换 | 1203-1467, 8530-8597 | 固定 Max 下打包 CAS、1,000 并发不超卖、退役 Proxy 停止新预留及零计数历史运行态回收已完成；动态降容和完整生命周期证据待完成 | +| CAP-001 | Gateway 分配使用 Reserved -> Active 原子转换 | 1203-1467, 8530-8597 | 固定 Max 下打包 CAS、1,000 并发不超卖、退役 Proxy 停止新预留及零计数历史运行态回收已完成；Snapshot 动态降容会保留既有 Active、在计数低于新上限前拒绝新预留，运行态复用与恢复预留均有回归测试。非法 Commit/Cancel/Release 经固定枚举观察器进入低基数指标 | | CAP-002 | 补池依据 Available Slots，不只看 Proxy 数量 | 1203-1402, 8530-8597 | `AvailableSlots`、显式 minimum/target 水位、pending 槽位和迟滞 Reconciler 已测试；Worker session/ACK/sequence/TTL/ownership fence、基础 Snapshot 流及 Gateway 会话组件已通过内存、Redis 8.2 和 gRPC 测试；权威 payload、目标健康和 Gateway reserve 策略仍待完成 | | CAP-003 | pool.maxSize 包括 FETCHED/CHECKING/AVAILABLE/SUSPECT/DRAINING 与 pending expected | 3001-3533, 6642-6680 | `FetchBudget` 100 并发额度预占测试 | | CAP-004 | TTL safety margin 内禁止新分配 | 173-220, 6728-6741 | 时钟测试 | @@ -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 暴露固定 result 的请求、请求数与交付数；Capacity Collector 按 Controller 进程聚合库存 Gauge，并只暴露固定 success/error 的库存读取结果，不把 Upstream 写成标签。`platform/logging` 使用 JSON `slog`，敏感字段、URL 用户信息、查询 Secret 和错误对象写出前均被脱敏；回归测试锁定标签集与 Secret 不泄露 | +| 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` 与 `proxy_pool_gateway_capacity_invariant_violations_total{operation}`，后者仅允许固定的 Reservation 生命周期违规枚举。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 写成标签。`platform/logging` 使用 JSON `slog`，敏感字段、URL 用户信息、查询 Secret 和错误对象写出前均被脱敏；回归测试锁定标签集与 Secret 不泄露 | | TEST-001 | 覆盖对话中列出的 11 个关键并发与故障场景 | 9030-9082 | 测试清单；Redis 活动池由 Memory/Redis 公用契约覆盖，跨进程故障场景仍按清单推进 | diff --git a/internal/domain/proxy/capacity.go b/internal/domain/proxy/capacity.go index 2913027..42a583c 100644 --- a/internal/domain/proxy/capacity.go +++ b/internal/domain/proxy/capacity.go @@ -12,6 +12,30 @@ var ( ErrReservationFinished = errors.New("reservation is already finished") ) +// CapacityInvariantViolation is a fixed lifecycle violation category. It +// deliberately excludes Proxy and request identifiers so observers can export +// bounded metrics safely. +type CapacityInvariantViolation string + +const ( + CapacityInvariantCommitAlreadyCommitted CapacityInvariantViolation = "commit_already_committed" + CapacityInvariantCommitFinished CapacityInvariantViolation = "commit_finished" + CapacityInvariantCancelFinished CapacityInvariantViolation = "cancel_finished" + CapacityInvariantReleaseBeforeCommit CapacityInvariantViolation = "release_before_commit" + CapacityInvariantReleaseFinished CapacityInvariantViolation = "release_finished" +) + +type CapacityInvariant struct { + Violation CapacityInvariantViolation +} + +// CapacityInvariantObserver is a transport-free hook for lifecycle misuse. +// Implementations must tolerate concurrent calls and must not block request +// handling. +type CapacityInvariantObserver interface { + ObserveCapacityInvariant(CapacityInvariant) +} + type Capacity struct { max atomic.Uint32 counters atomic.Uint64 @@ -19,25 +43,38 @@ type Capacity struct { reserving atomic.Int64 configuredObserver *activityObserver observer atomic.Pointer[activityObserver] + invariants CapacityInvariantObserver } type activityObserver struct{ notify func(bool) } func NewCapacity(max int64) *Capacity { - return NewCapacityWithActivityObserver(max, nil) + return NewCapacityWithObservers(max, nil, nil) } // NewCapacityWithActivityObserver reports successful zero-to-nonzero and // nonzero-to-zero transitions. The observer must tolerate concurrent calls. func NewCapacityWithActivityObserver(max int64, observer func(nonzero bool)) *Capacity { + return NewCapacityWithObservers(max, observer, nil) +} + +// NewCapacityWithObservers creates an atomic capacity with optional local +// activity and invariant observers. The invariant observer is only called on +// invalid reservation lifecycle transitions. +func NewCapacityWithObservers( + max int64, + activity func(nonzero bool), + invariants CapacityInvariantObserver, +) *Capacity { capacity := &Capacity{} if max < 0 || max > int64(counterMask) { max = 0 } capacity.max.Store(uint32(max)) capacity.reservationEnabled.Store(true) - if observer != nil { - capacity.configuredObserver = &activityObserver{notify: observer} + capacity.invariants = invariants + if activity != nil { + capacity.configuredObserver = &activityObserver{notify: activity} capacity.observer.Store(capacity.configuredObserver) } return capacity @@ -197,6 +234,13 @@ func (c *Capacity) release() { } } +func (c *Capacity) observeInvariant(violation CapacityInvariantViolation) { + if c == nil || c.invariants == nil { + return + } + c.invariants.ObserveCapacityInvariant(CapacityInvariant{Violation: violation}) +} + func pack(active, reserved uint32) uint64 { return uint64(reserved)<<32 | uint64(active) } @@ -210,11 +254,20 @@ type Reservation struct { state atomic.Uint32 } +const ( + reservationOpen uint32 = iota + reservationCommitted + reservationCanceled + reservationReleased +) + func (r *Reservation) Commit() error { - if !r.state.CompareAndSwap(0, 1) { - if r.state.Load() == 1 { + if !r.state.CompareAndSwap(reservationOpen, reservationCommitted) { + if r.state.Load() == reservationCommitted { + r.capacity.observeInvariant(CapacityInvariantCommitAlreadyCommitted) return ErrReservationCommitted } + r.capacity.observeInvariant(CapacityInvariantCommitFinished) return ErrReservationFinished } r.capacity.commit() @@ -222,7 +275,8 @@ func (r *Reservation) Commit() error { } func (r *Reservation) Cancel() error { - if !r.state.CompareAndSwap(0, 2) { + if !r.state.CompareAndSwap(reservationOpen, reservationCanceled) { + r.capacity.observeInvariant(CapacityInvariantCancelFinished) return ErrReservationFinished } r.capacity.cancel() @@ -230,7 +284,12 @@ func (r *Reservation) Cancel() error { } func (r *Reservation) Release() error { - if !r.state.CompareAndSwap(1, 2) { + if !r.state.CompareAndSwap(reservationCommitted, reservationReleased) { + if r.state.Load() == reservationOpen { + r.capacity.observeInvariant(CapacityInvariantReleaseBeforeCommit) + } else { + r.capacity.observeInvariant(CapacityInvariantReleaseFinished) + } return ErrReservationFinished } r.capacity.release() diff --git a/internal/domain/proxy/capacity_test.go b/internal/domain/proxy/capacity_test.go index f7e3a3a..c4c903c 100644 --- a/internal/domain/proxy/capacity_test.go +++ b/internal/domain/proxy/capacity_test.go @@ -250,11 +250,101 @@ func TestConcurrentReleaseSucceedsOnce(t *testing.T) { assertCapacityCounters(t, capacity, 0, 0) } +func TestCapacityLoweringPreservesExistingWorkAndBlocksNewReservations(t *testing.T) { + capacity := NewCapacity(3) + reservations := make([]*Reservation, 0, 3) + for range 3 { + reservation, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() = false") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit() = %v", err) + } + reservations = append(reservations, reservation) + } + + if !capacity.SetMax(1) { + t.Fatal("SetMax(1) = false") + } + if active, reserved, maximum := capacity.Counters(); active != 3 || reserved != 0 || maximum != 1 { + t.Fatalf("Counters() after lowering = (%d, %d, %d), want (3, 0, 1)", active, reserved, maximum) + } + if _, ok := capacity.Reserve(); ok { + t.Fatal("Reserve() while active work exceeds lowered maximum = true") + } + if err := reservations[0].Release(); err != nil { + t.Fatalf("Release(first) = %v", err) + } + if err := reservations[1].Release(); err != nil { + t.Fatalf("Release(second) = %v", err) + } + if _, ok := capacity.Reserve(); ok { + t.Fatal("Reserve() at lowered maximum = true") + } + if err := reservations[2].Release(); err != nil { + t.Fatalf("Release(third) = %v", err) + } + + reused, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() after draining to lowered maximum = false") + } + if err := reused.Cancel(); err != nil { + t.Fatalf("Cancel(reused) = %v", err) + } +} + +func TestCapacityReportsFixedLifecycleInvariantViolations(t *testing.T) { + observer := &recordingInvariantObserver{} + capacity := NewCapacityWithObservers(1, nil, observer) + reservation, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() = false") + } + if err := reservation.Release(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("Release() before Commit error = %v, want ErrReservationFinished", err) + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit() = %v", err) + } + if err := reservation.Commit(); !errors.Is(err, ErrReservationCommitted) { + t.Fatalf("second Commit() error = %v, want ErrReservationCommitted", err) + } + if err := reservation.Cancel(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("Cancel() after Commit error = %v, want ErrReservationFinished", err) + } + if err := reservation.Release(); err != nil { + t.Fatalf("Release() = %v", err) + } + if err := reservation.Release(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("second Release() error = %v, want ErrReservationFinished", err) + } + + want := []CapacityInvariantViolation{ + CapacityInvariantReleaseBeforeCommit, + CapacityInvariantCommitAlreadyCommitted, + CapacityInvariantCancelFinished, + CapacityInvariantReleaseFinished, + } + if !reflect.DeepEqual(observer.violations, want) { + t.Fatalf("violations = %v, want %v", observer.violations, want) + } +} + type terminationResult struct { operation string err error } +type recordingInvariantObserver struct { + violations []CapacityInvariantViolation +} + +func (observer *recordingInvariantObserver) ObserveCapacityInvariant(event CapacityInvariant) { + observer.violations = append(observer.violations, event.Violation) +} + func assertCapacityCounters(t *testing.T, capacity *Capacity, active, reserved int64) { t.Helper() if got := capacity.Active(); got != active { diff --git a/internal/gateway/bootstrap/bootstrap.go b/internal/gateway/bootstrap/bootstrap.go index 210d6bc..8669c1f 100644 --- a/internal/gateway/bootstrap/bootstrap.go +++ b/internal/gateway/bootstrap/bootstrap.go @@ -137,7 +137,6 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio if ctx == nil || configuration == nil { return nil, ErrInvalidOptions } - store := snapshot.NewStore(options.ClusterID, options.WorkerID) protection, err := server.BuildProtection(configuration.Gateway) if err != nil { return nil, fmt.Errorf("build gateway protections: %w", err) @@ -146,16 +145,18 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio if err != nil { return nil, fmt.Errorf("build gateway target policy: %w", err) } - proxyTransport := transport.New(transport.Config{}, snapshotCredentialResolver{store: store}) var outcomeMetrics outcomeDomain.MetricsObserver + storeOptions := snapshot.StoreOptions{} if configuration.Metrics.Enabled { collector, err := platformMetrics.NewGatewayCollector(prometheus.DefaultRegisterer) if err != nil { - proxyTransport.CloseIdleConnections() return nil, fmt.Errorf("build gateway outcome metrics: %w", err) } outcomeMetrics = collector + storeOptions.CapacityInvariantObserver = collector } + store := snapshot.NewStoreWithOptions(options.ClusterID, options.WorkerID, storeOptions) + proxyTransport := transport.New(transport.Config{}, snapshotCredentialResolver{store: store}) outcomes, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{ Capacity: defaultOutcomeQueueCapacity, MaxBatch: min(defaultOutcomeBatchSize, configuration.ControlPlane.MaxRuntimeCounters), Metrics: outcomeMetrics, diff --git a/internal/gateway/snapshot/store.go b/internal/gateway/snapshot/store.go index 14f7a69..6d74040 100644 --- a/internal/gateway/snapshot/store.go +++ b/internal/gateway/snapshot/store.go @@ -163,18 +163,35 @@ type Store struct { workerID string current atomic.Pointer[View] - mu sync.Mutex - runtimes map[string]*runtimeRegistration - active activeRuntimeIndex - limit int + mu sync.Mutex + runtimes map[string]*runtimeRegistration + active activeRuntimeIndex + limit int + invariantObserver proxyDomain.CapacityInvariantObserver +} + +// StoreOptions configures bounded process-local runtime state. Observers only +// receive fixed lifecycle violation categories and never Proxy identifiers. +type StoreOptions struct { + RuntimeLimit int + CapacityInvariantObserver proxyDomain.CapacityInvariantObserver } func NewStore(clusterID, workerID string) *Store { + return NewStoreWithOptions(clusterID, workerID, StoreOptions{}) +} + +func NewStoreWithOptions(clusterID, workerID string, options StoreOptions) *Store { + limit := defaultRuntimeLimit + if options.RuntimeLimit > 0 { + limit = options.RuntimeLimit + } return &Store{ - clusterID: clusterID, - workerID: workerID, - runtimes: make(map[string]*runtimeRegistration), - limit: defaultRuntimeLimit, + clusterID: clusterID, + workerID: workerID, + runtimes: make(map[string]*runtimeRegistration), + limit: limit, + invariantObserver: options.CapacityInvariantObserver, } } @@ -182,9 +199,7 @@ func NewStoreWithRuntimeLimit(clusterID, workerID string, limit int) (*Store, er if limit <= 0 { return nil, ErrInvalidRuntimeLimit } - store := NewStore(clusterID, workerID) - store.limit = limit - return store, nil + return NewStoreWithOptions(clusterID, workerID, StoreOptions{RuntimeLimit: limit}), nil } func (s *Store) Current() *View { @@ -315,12 +330,12 @@ func (s *Store) Apply(envelope Envelope) error { if registration == nil { proxyID := descriptor.ID registration = &runtimeRegistration{} - registration.capacity = proxyDomain.NewCapacityWithActivityObserver(descriptor.MaxConcurrency, func(nonzero bool) { + registration.capacity = proxyDomain.NewCapacityWithObservers(descriptor.MaxConcurrency, func(nonzero bool) { if registration.current.Load() { return } s.active.track(proxyID, registration.capacity, nonzero) - }) + }, s.invariantObserver) } else { registration.capacity.SetMax(descriptor.MaxConcurrency) } diff --git a/internal/gateway/snapshot/store_test.go b/internal/gateway/snapshot/store_test.go index be1638a..6ad9f2d 100644 --- a/internal/gateway/snapshot/store_test.go +++ b/internal/gateway/snapshot/store_test.go @@ -299,6 +299,97 @@ func TestStoreApplyReusesCapacityAcrossVersionsAndEpochs(t *testing.T) { } } +func TestStoreApplyLoweredCapacityDrainsExistingWorkBeforeReusingSlot(t *testing.T) { + store := NewStore("cluster-a", "worker-a") + proxy := proxyDomain.Proxy{ + ID: "stable", Scheme: proxyDomain.SchemeHTTP, State: proxyDomain.StateAvailable, MaxConcurrency: 3, + } + first := Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, + Proxies: []proxyDomain.Proxy{proxy}, + } + first.Checksum = Checksum(first.Proxies) + if err := store.Apply(first); err != nil { + t.Fatalf("Apply(first) = %v", err) + } + runtime := store.Current().Entries[0].Runtime + reservations := make([]*proxyDomain.Reservation, 0, 3) + for range 3 { + reservation, ok := runtime.Reserve() + if !ok { + t.Fatal("Reserve() = false") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit() = %v", err) + } + reservations = append(reservations, reservation) + } + + lowered := first + lowered.Version = 2 + lowered.Proxies[0].MaxConcurrency = 1 + lowered.Checksum = Checksum(lowered.Proxies) + if err := store.Apply(lowered); err != nil { + t.Fatalf("Apply(lowered) = %v", err) + } + if current := store.Current().Entries[0].Runtime; current != runtime || current.Max() != 1 || current.Active() != 3 { + t.Fatalf("runtime after lowering = %p max=%d active=%d, want reused max=1 active=3", current, current.Max(), current.Active()) + } + if _, ok := runtime.Reserve(); ok { + t.Fatal("Reserve() while active work exceeds lowered maximum = true") + } + for _, reservation := range reservations[:2] { + if err := reservation.Release(); err != nil { + t.Fatalf("Release() = %v", err) + } + } + if _, ok := runtime.Reserve(); ok { + t.Fatal("Reserve() at lowered maximum = true") + } + if err := reservations[2].Release(); err != nil { + t.Fatalf("Release(final) = %v", err) + } + reused, ok := runtime.Reserve() + if !ok { + t.Fatal("Reserve() after draining lowered runtime = false") + } + if err := reused.Cancel(); err != nil { + t.Fatalf("Cancel(reused) = %v", err) + } +} + +func TestStorePropagatesCapacityInvariantObserverToRuntime(t *testing.T) { + observer := &capacityInvariantRecorder{} + store := NewStoreWithOptions("cluster-a", "worker-a", StoreOptions{CapacityInvariantObserver: observer}) + proxy := proxyDomain.Proxy{ + ID: "stable", Scheme: proxyDomain.SchemeHTTP, State: proxyDomain.StateAvailable, MaxConcurrency: 1, + } + envelope := Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, + Proxies: []proxyDomain.Proxy{proxy}, + } + envelope.Checksum = Checksum(envelope.Proxies) + if err := store.Apply(envelope); err != nil { + t.Fatalf("Apply() = %v", err) + } + reservation, ok := store.Current().Entries[0].Runtime.Reserve() + if !ok { + t.Fatal("Reserve() = false") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit() = %v", err) + } + if err := reservation.Release(); err != nil { + t.Fatalf("Release() = %v", err) + } + if err := reservation.Release(); !errors.Is(err, proxyDomain.ErrReservationFinished) { + t.Fatalf("second Release() = %v, want ErrReservationFinished", err) + } + if got, want := observer.violations, []proxyDomain.CapacityInvariantViolation{proxyDomain.CapacityInvariantReleaseFinished}; !reflect.DeepEqual(got, want) { + t.Fatalf("violations = %v, want %v", got, want) + } +} + func TestStoreReusesRuntimeWhenProxyDisappearsAndReappears(t *testing.T) { store := NewStore("cluster-a", "worker-a") proxy := proxyDomain.Proxy{ @@ -559,6 +650,14 @@ func (s *Store) activeRuntimeCount() int { return count } +type capacityInvariantRecorder struct { + violations []proxyDomain.CapacityInvariantViolation +} + +func (recorder *capacityInvariantRecorder) ObserveCapacityInvariant(event proxyDomain.CapacityInvariant) { + recorder.violations = append(recorder.violations, event.Violation) +} + func collectSelectionIDs(selection Selection) []string { ids := make([]string, 0, selection.Len()) for index := 0; index < selection.Len(); index++ { diff --git a/internal/platform/metrics/gateway.go b/internal/platform/metrics/gateway.go index 4148efa..2e3256f 100644 --- a/internal/platform/metrics/gateway.go +++ b/internal/platform/metrics/gateway.go @@ -7,16 +7,21 @@ import ( "github.com/prometheus/client_golang/prometheus" outcomeDomain "proxy-pool/internal/domain/outcome" + proxyDomain "proxy-pool/internal/domain/proxy" ) // GatewayCollector exposes fixed-cardinality request-path outcome metrics. // Proxy, route, destination, client and credential values are never labels. type GatewayCollector struct { - outcomes *prometheus.CounterVec - dropped prometheus.Counter + outcomes *prometheus.CounterVec + dropped prometheus.Counter + invariants *prometheus.CounterVec } -var _ outcomeDomain.MetricsObserver = (*GatewayCollector)(nil) +var ( + _ outcomeDomain.MetricsObserver = (*GatewayCollector)(nil) + _ proxyDomain.CapacityInvariantObserver = (*GatewayCollector)(nil) +) func NewGatewayCollector(registerer prometheus.Registerer) (*GatewayCollector, error) { if registerer == nil { @@ -36,7 +41,14 @@ func NewGatewayCollector(registerer prometheus.Registerer) (*GatewayCollector, e if err != nil { return nil, err } - return &GatewayCollector{outcomes: outcomes, dropped: dropped}, nil + invariants, err := registerCounterVec(registerer, prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "proxy_pool", Subsystem: "gateway", Name: "capacity_invariant_violations_total", + Help: "Number of invalid Gateway local proxy capacity lifecycle transitions.", + }, []string{"operation"})) + if err != nil { + return nil, err + } + return &GatewayCollector{outcomes: outcomes, dropped: dropped, invariants: invariants}, nil } func (collector *GatewayCollector) Observe(event outcomeDomain.Event) { @@ -57,6 +69,13 @@ func (collector *GatewayCollector) ObserveDropped() { collector.dropped.Inc() } +func (collector *GatewayCollector) ObserveCapacityInvariant(event proxyDomain.CapacityInvariant) { + if collector == nil || collector.invariants == nil || !validCapacityInvariantViolation(event.Violation) { + return + } + collector.invariants.WithLabelValues(string(event.Violation)).Inc() +} + func registerCounter(registerer prometheus.Registerer, candidate prometheus.Counter) (prometheus.Counter, error) { if err := registerer.Register(candidate); err == nil { return candidate, nil @@ -97,3 +116,16 @@ func gatewayStageLabel(stage outcomeDomain.Stage) string { return "" } } + +func validCapacityInvariantViolation(violation proxyDomain.CapacityInvariantViolation) bool { + switch violation { + case proxyDomain.CapacityInvariantCommitAlreadyCommitted, + proxyDomain.CapacityInvariantCommitFinished, + proxyDomain.CapacityInvariantCancelFinished, + proxyDomain.CapacityInvariantReleaseBeforeCommit, + proxyDomain.CapacityInvariantReleaseFinished: + return true + default: + return false + } +} diff --git a/internal/platform/metrics/gateway_test.go b/internal/platform/metrics/gateway_test.go index a5a5b0c..8d9f84f 100644 --- a/internal/platform/metrics/gateway_test.go +++ b/internal/platform/metrics/gateway_test.go @@ -6,6 +6,7 @@ import ( "github.com/prometheus/client_golang/prometheus" outcomeDomain "proxy-pool/internal/domain/outcome" + proxyDomain "proxy-pool/internal/domain/proxy" ) func TestGatewayCollectorRecordsOnlyFixedDimensions(t *testing.T) { @@ -18,6 +19,10 @@ func TestGatewayCollectorRecordsOnlyFixedDimensions(t *testing.T) { collector.Observe(outcomeDomain.Event{Stage: outcomeDomain.StageTunnel, Success: false}) collector.Observe(outcomeDomain.Event{Stage: outcomeDomain.StageUnspecified, Success: true}) collector.ObserveDropped() + collector.ObserveCapacityInvariant(proxyDomain.CapacityInvariant{ + Violation: proxyDomain.CapacityInvariantReleaseFinished, + }) + collector.ObserveCapacityInvariant(proxyDomain.CapacityInvariant{Violation: "unknown"}) assertMetricValue(t, registry, "proxy_pool_gateway_outcomes_total", map[string]string{ "stage": "DIAL", "result": "success", @@ -26,6 +31,9 @@ func TestGatewayCollectorRecordsOnlyFixedDimensions(t *testing.T) { "stage": "TUNNEL", "result": "failure", }, 1) assertMetricValue(t, registry, "proxy_pool_gateway_outcome_queue_dropped_total", nil, 1) + assertMetricValue(t, registry, "proxy_pool_gateway_capacity_invariant_violations_total", map[string]string{ + "operation": "release_finished", + }, 1) } func TestNewGatewayCollectorReusesRegisteredCollectors(t *testing.T) { diff --git a/task_plan.md b/task_plan.md index 9f33dc7..7902502 100644 --- a/task_plan.md +++ b/task_plan.md @@ -72,5 +72,5 @@ Provider 分布式协调和 Worker 运行态 Redis 原语已完成，但 WorkerControlPlane 接收端、Provider Fleet、Gateway、Checker、Loadgen、业务指标、Redis 故障 转移验证与代表性集群压测属于后续实施范围。 -- `implementation-plan.md` 当前按 73 个验收项统计；已校正为 51 项完成， - 验收项完成率约 69.9%，不等同于生产就绪度。 +- `implementation-plan.md` 当前按 74 个验收项统计；已校正为 59 项完成， + 验收项完成率约 79.7%，不等同于生产就绪度。