feat: observe gateway capacity invariants
This commit is contained in:
youfak
parent
af1cd70986
commit
707aff26ea
@ -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)。
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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 是集群设计输入，不是本次验证结果。只有实现上述运行时，并在
|
||||
|
||||
@ -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 公用契约覆盖，跨进程故障场景仍按清单推进 |
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -167,14 +167,31 @@ type Store struct {
|
||||
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,
|
||||
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)
|
||||
}
|
||||
|
||||
@ -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++ {
|
||||
|
||||
@ -7,6 +7,7 @@ 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.
|
||||
@ -14,9 +15,13 @@ import (
|
||||
type GatewayCollector struct {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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%，不等同于生产就绪度。
|
||||
|
||||
Loading…
Reference in New Issue
Block a user