feat: persist target health profiles
This commit is contained in:
youfak
parent
0a904a7b6b
commit
19cf3e32fa
@ -273,11 +273,12 @@ Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重
|
||||
- [ ] Ensure target failures affect only the target profile.
|
||||
- [ ] Add fixture target server and deterministic clock/scheduler tests.
|
||||
|
||||
当前进度（2026-07-31）：已新增不依赖 gRPC/Redis 的 `domain/health` Observation、
|
||||
全局/Target Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄接口在
|
||||
Memory 和 Redis Lua 同一原子边界归并，覆盖首次 CHECKING 失败进入 UNHEALTHY、
|
||||
AVAILABLE/SUSPECT 的阈值降级、精确重放、冲突拒绝与成功恢复。TARGET 仍只在纯领域
|
||||
Profile 中归并，尚未接入持久化、调度器、Checker RPC 或独立进程，因此本任务保持未完成。
|
||||
当前进度（2026-07-31）：已新增不依赖 gRPC 的 `domain/health` Observation、全局/Target
|
||||
Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄接口在 Memory 和 Redis
|
||||
Lua 同一原子边界归并，覆盖首次 CHECKING 失败进入 UNHEALTHY、AVAILABLE/SUSPECT 的
|
||||
阈值降级、精确重放、冲突拒绝与成功恢复。TARGET Profile 也已在 Memory 和 Redis 中独立
|
||||
归并，以哈希键保存并随代理 TTL 过期，绝不写入 Proxy 全局状态或选择索引。调度器、
|
||||
Checker RPC 与独立进程尚未接入，因此本任务保持未完成。
|
||||
|
||||
## Task 12: Machine-readable Contracts
|
||||
|
||||
|
||||
@ -58,8 +58,8 @@ Routing 一起纳入 Snapshot checksum，仅在 mTLS 和 Gateway 当前内存 Vi
|
||||
Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum，其运行时消费、
|
||||
Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controller 的 session/sequence/
|
||||
摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 已有全局健康 Reducer
|
||||
与 Memory/Redis 原子状态提交基础，但任务调度、Checker RPC、TARGET Profile 持久化和
|
||||
独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
|
||||
与 Memory/Redis 原子状态提交基础；TARGET Profile 以独立、随代理 TTL 过期的 Redis
|
||||
记录归并，不改写 Proxy 全局状态。任务调度、Checker RPC 和独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
|
||||
`session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller
|
||||
在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用
|
||||
`SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot
|
||||
@ -109,7 +109,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。
|
||||
6. Worker 基础网络快照流、Proxy/Gateway Routing/凭据 Snapshot payload、Gateway Snapshot
|
||||
客户端和进程装配、同版本 Routing 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成；
|
||||
Outcome 上报已完成基础观测链；Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。
|
||||
7. Checker 调度、探测器和健康 reducer。
|
||||
7. Checker 调度、探测器和健康 reducer；其中全局与 TARGET Profile 的 Memory/Redis
|
||||
原子归并已完成。
|
||||
8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。
|
||||
9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。
|
||||
10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS
|
||||
|
||||
@ -79,7 +79,7 @@
|
||||
|
||||
| ID | 最终需求 | 来源 | 验证证据 |
|
||||
|---|---|---|---|
|
||||
| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 持久化和 Routing 消费待实现 |
|
||||
| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 在 Memory 和 Redis 独立、随代理 TTL 归并，不改写 Proxy 全局状态；Routing 消费待实现 |
|
||||
| HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置校验已完成；有界调度器、抖动和分级频率测试待实现 |
|
||||
| HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | 全局 Reducer 已在 Memory/Redis 活动池原子提交连续失败、精确重放和成功恢复；任务调度与 REMOVE 编排待实现 |
|
||||
| SEC-001 | API 认证与 Proxy 认证分离，Secret 统一脱敏 | 7528-8111, 8904-8945 | Config 脱敏、Provider Store -> SecretRef -> Gateway Resolver 跨包测试与格式化泄漏回归测试 |
|
||||
|
||||
@ -110,6 +110,7 @@ func TestNewBuildsClusterSafeKeyspaceAndHashesDynamicTokens(t *testing.T) {
|
||||
dynamicKeys := []string{
|
||||
adapter.keys.idempotency(raw, raw),
|
||||
adapter.keys.operation(raw),
|
||||
adapter.keys.targetHealth(raw),
|
||||
adapter.keys.protocol(raw),
|
||||
adapter.keys.region(raw),
|
||||
adapter.keys.carrier(raw),
|
||||
|
||||
@ -2,7 +2,10 @@ package redisactivity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"proxy-pool/internal/domain/activitypool"
|
||||
healthDomain "proxy-pool/internal/domain/health"
|
||||
@ -10,6 +13,18 @@ import (
|
||||
|
||||
var _ activitypool.HealthStore = (*Adapter)(nil)
|
||||
var _ activitypool.GlobalHealthStore = (*Adapter)(nil)
|
||||
var _ activitypool.TargetHealthStore = (*Adapter)(nil)
|
||||
|
||||
type targetHealthRecord struct {
|
||||
Version int `json:"version"`
|
||||
Status string `json:"status"`
|
||||
ConsecutiveFailures int `json:"consecutiveFailures"`
|
||||
LastHealthTaskID string `json:"lastHealthTaskId,omitempty"`
|
||||
LastHealthDigest string `json:"lastHealthDigest,omitempty"`
|
||||
LastHealthObservedAtMS int64 `json:"lastHealthObservedAtMs,omitempty"`
|
||||
LastSuccessAtMS int64 `json:"lastSuccessAtMs,omitempty"`
|
||||
LatencyNS int64 `json:"latencyNs"`
|
||||
}
|
||||
|
||||
func (a *Adapter) ApplyHealth(ctx context.Context, update activitypool.HealthUpdate) (activitypool.Entry, error) {
|
||||
if ctx == nil {
|
||||
@ -127,3 +142,122 @@ func (a *Adapter) ApplyGlobalObservation(
|
||||
return activitypool.Entry{}, invalidScriptReply("unexpected global health status")
|
||||
}
|
||||
}
|
||||
|
||||
// ApplyTargetObservation reduces a TARGET check into a short-lived Profile
|
||||
// record keyed by (proxy, routing, target URL). It intentionally does not
|
||||
// update the proxy record, global health, or any proxy selection index.
|
||||
func (a *Adapter) ApplyTargetObservation(
|
||||
ctx context.Context,
|
||||
command activitypool.TargetHealthCommand,
|
||||
) (healthDomain.TargetState, error) {
|
||||
if ctx == nil {
|
||||
return healthDomain.TargetState{}, activitypool.ErrInvalidHealthUpdate
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
normalized, err := healthDomain.NormalizeObservation(command.Observation)
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
if a == nil || normalized.Level != healthDomain.LevelTarget || command.MaxConsecutiveFailures <= 0 {
|
||||
return healthDomain.TargetState{}, activitypool.ErrInvalidHealthUpdate
|
||||
}
|
||||
profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{
|
||||
RoutingName: normalized.RoutingName,
|
||||
TargetURL: normalized.TargetURL,
|
||||
})
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
digest, err := healthDomain.ObservationDigest(normalized)
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
operationID, err := newOperationID()
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
success := "0"
|
||||
if normalized.Success {
|
||||
success = "1"
|
||||
}
|
||||
result, err := runScript(ctx, a.client, targetHealthScript, []string{
|
||||
a.keys.records, a.keys.targetHealth(normalized.ProxyID), a.keys.operation(operationID),
|
||||
}, normalized.ObservedAt.UnixMilli(), int64(normalized.Latency), normalized.ProxyID,
|
||||
digestToken(profile.Key()), success, command.MaxConsecutiveFailures, normalized.TaskID,
|
||||
hex.EncodeToString(digest[:]), operationTTLMillis(a.options.OperationTTL))
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
var reply targetHealthScriptReply
|
||||
if err := decodeScriptResult(result, &reply); err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
switch reply.Status {
|
||||
case scriptNotFound:
|
||||
return healthDomain.TargetState{}, activitypool.ErrActivityNotFound
|
||||
case scriptStale:
|
||||
return healthDomain.TargetState{}, healthDomain.ErrStaleObservation
|
||||
case scriptConflict:
|
||||
return healthDomain.TargetState{}, healthDomain.ErrConflictingObservation
|
||||
case scriptInvalid:
|
||||
return healthDomain.TargetState{}, activitypool.ErrInvalidHealthUpdate
|
||||
case scriptOK:
|
||||
if reply.Target == "" {
|
||||
return healthDomain.TargetState{}, invalidScriptReply("target health reply omitted target state")
|
||||
}
|
||||
state, err := decodeTargetHealthState(reply.Target)
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, invalidScriptReply("target health reply contained an invalid target state")
|
||||
}
|
||||
return state, nil
|
||||
default:
|
||||
return healthDomain.TargetState{}, invalidScriptReply("unexpected target health status")
|
||||
}
|
||||
}
|
||||
|
||||
func decodeTargetHealthState(payload string) (healthDomain.TargetState, error) {
|
||||
var record targetHealthRecord
|
||||
if err := decodeJSON(payload, &record); err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
if record.Version != recordVersion || record.ConsecutiveFailures < 0 || record.LatencyNS < 0 {
|
||||
return healthDomain.TargetState{}, ErrInvalidRecord
|
||||
}
|
||||
state := healthDomain.TargetState{
|
||||
Status: healthDomain.TargetStatus(record.Status),
|
||||
ConsecutiveFailures: record.ConsecutiveFailures,
|
||||
LastTaskID: record.LastHealthTaskID,
|
||||
Latency: time.Duration(record.LatencyNS),
|
||||
}
|
||||
switch state.Status {
|
||||
case healthDomain.TargetUnknown, healthDomain.TargetAvailable, healthDomain.TargetSuspect, healthDomain.TargetUnhealthy:
|
||||
default:
|
||||
return healthDomain.TargetState{}, ErrInvalidRecord
|
||||
}
|
||||
if record.LastHealthObservedAtMS == 0 {
|
||||
if state.Status != healthDomain.TargetUnknown || state.LastTaskID != "" || state.ConsecutiveFailures != 0 ||
|
||||
record.LastHealthDigest != "" || record.LastSuccessAtMS != 0 || state.Latency != 0 {
|
||||
return healthDomain.TargetState{}, ErrInvalidRecord
|
||||
}
|
||||
return state, nil
|
||||
}
|
||||
if record.LastHealthObservedAtMS < 0 || record.LastSuccessAtMS < 0 || state.LastTaskID == "" ||
|
||||
len(record.LastHealthDigest) != sha256.Size*2 {
|
||||
return healthDomain.TargetState{}, ErrInvalidRecord
|
||||
}
|
||||
digest, err := hex.DecodeString(record.LastHealthDigest)
|
||||
if err != nil || len(digest) != sha256.Size {
|
||||
return healthDomain.TargetState{}, errors.Join(ErrInvalidRecord, err)
|
||||
}
|
||||
copy(state.LastObservationDigest[:], digest)
|
||||
state.LastObservedAt = time.UnixMilli(record.LastHealthObservedAtMS).UTC()
|
||||
if record.LastSuccessAtMS > 0 {
|
||||
state.LastSuccessAt = time.UnixMilli(record.LastSuccessAtMS).UTC()
|
||||
if state.LastSuccessAt.After(state.LastObservedAt) {
|
||||
return healthDomain.TargetState{}, ErrInvalidRecord
|
||||
}
|
||||
}
|
||||
return state, nil
|
||||
}
|
||||
|
||||
@ -65,6 +65,13 @@ func (keys keyspace) operation(operationID string) string {
|
||||
return keys.prefix + ":op:" + digestToken(operationID)
|
||||
}
|
||||
|
||||
// targetHealth stores short-lived TARGET check state separately from the
|
||||
// proxy record. The proxy ID is hashed so neither endpoint identity nor target
|
||||
// profile information appears in a Redis key name.
|
||||
func (keys keyspace) targetHealth(proxyID string) string {
|
||||
return keys.prefix + ":target-health:" + digestToken(proxyID)
|
||||
}
|
||||
|
||||
func (keys keyspace) protocol(value string) string {
|
||||
return keys.facet("protocol", value)
|
||||
}
|
||||
|
||||
@ -44,6 +44,11 @@ type healthScriptReply struct {
|
||||
Record string `json:"record,omitempty"`
|
||||
}
|
||||
|
||||
type targetHealthScriptReply struct {
|
||||
Status scriptStatus `json:"status"`
|
||||
Target string `json:"target,omitempty"`
|
||||
}
|
||||
|
||||
type extractScriptReply struct {
|
||||
Status scriptStatus `json:"status"`
|
||||
RequestDigest string `json:"requestDigest"`
|
||||
@ -105,6 +110,9 @@ var upsertSource string
|
||||
//go:embed scripts/health.lua
|
||||
var healthSource string
|
||||
|
||||
//go:embed scripts/target_health.lua
|
||||
var targetHealthSource string
|
||||
|
||||
//go:embed scripts/extract.lua
|
||||
var extractSource string
|
||||
|
||||
@ -129,6 +137,7 @@ var workerSnapshotSource string
|
||||
var (
|
||||
upsertScript = redis.NewScript(upsertSource)
|
||||
healthScript = redis.NewScript(healthSource)
|
||||
targetHealthScript = redis.NewScript(targetHealthSource)
|
||||
extractScript = redis.NewScript(extractSource)
|
||||
ownershipScript = redis.NewScript(ownershipSource)
|
||||
sweepScript = redis.NewScript(sweepSource)
|
||||
|
||||
116
internal/adapters/redisactivity/scripts/target_health.lua
Normal file
116
internal/adapters/redisactivity/scripts/target_health.lua
Normal file
@ -0,0 +1,116 @@
|
||||
local records_key = KEYS[1]
|
||||
local target_health_key = KEYS[2]
|
||||
local operation_key = KEYS[3]
|
||||
|
||||
local observed_at_ms = tonumber(ARGV[1])
|
||||
local latency_ns = tonumber(ARGV[2])
|
||||
local proxy_id = ARGV[3]
|
||||
local profile_key = ARGV[4]
|
||||
local success = ARGV[5]
|
||||
local max_consecutive_failures = tonumber(ARGV[6])
|
||||
local task_id = ARGV[7]
|
||||
local observation_digest = ARGV[8]
|
||||
local operation_ttl_ms = tonumber(ARGV[9])
|
||||
|
||||
local committed = redis.call('GET', operation_key)
|
||||
if committed then
|
||||
return committed
|
||||
end
|
||||
|
||||
local function finish(reply, hard_expiry_ms)
|
||||
local encoded = cjson.encode(reply)
|
||||
redis.call('SET', operation_key, encoded, 'PX', operation_ttl_ms)
|
||||
if hard_expiry_ms then
|
||||
local operation_expiry_ms = redis.call('PEXPIRETIME', operation_key)
|
||||
if operation_expiry_ms > hard_expiry_ms then
|
||||
redis.call('PEXPIREAT', operation_key, hard_expiry_ms)
|
||||
end
|
||||
end
|
||||
return encoded
|
||||
end
|
||||
|
||||
if not observed_at_ms or observed_at_ms <= 0 or not latency_ns or latency_ns < 0 or
|
||||
type(proxy_id) ~= 'string' or proxy_id == '' or type(profile_key) ~= 'string' or
|
||||
string.len(profile_key) ~= 64 or (success ~= '0' and success ~= '1') or
|
||||
not max_consecutive_failures or max_consecutive_failures <= 0 or type(task_id) ~= 'string' or
|
||||
task_id == '' or type(observation_digest) ~= 'string' or string.len(observation_digest) ~= 64 or
|
||||
not operation_ttl_ms or operation_ttl_ms <= 0 then
|
||||
return finish({status = 'invalid'})
|
||||
end
|
||||
|
||||
local raw_proxy = redis.call('HGET', records_key, proxy_id)
|
||||
if not raw_proxy then
|
||||
return finish({status = 'not_found'})
|
||||
end
|
||||
local proxy = cjson.decode(raw_proxy)
|
||||
if type(proxy.expiresAtMs) ~= 'number' or proxy.expiresAtMs <= observed_at_ms then
|
||||
redis.call('DEL', target_health_key)
|
||||
return finish({status = 'not_found'})
|
||||
end
|
||||
|
||||
local raw_target = redis.call('HGET', target_health_key, profile_key)
|
||||
local target
|
||||
if raw_target then
|
||||
local decoded, value = pcall(cjson.decode, raw_target)
|
||||
if not decoded or type(value) ~= 'table' then
|
||||
return finish({status = 'invalid'})
|
||||
end
|
||||
target = value
|
||||
else
|
||||
target = {
|
||||
version = 1,
|
||||
status = 'UNKNOWN',
|
||||
consecutiveFailures = 0,
|
||||
latencyNs = 0,
|
||||
}
|
||||
end
|
||||
|
||||
if target.version ~= 1 or type(target.status) ~= 'string' or
|
||||
type(target.consecutiveFailures) ~= 'number' or target.consecutiveFailures < 0 or
|
||||
type(target.latencyNs) ~= 'number' or target.latencyNs < 0 then
|
||||
return finish({status = 'invalid'})
|
||||
end
|
||||
|
||||
local last_task_id = target.lastHealthTaskId or ''
|
||||
local last_digest = target.lastHealthDigest or ''
|
||||
local last_observed_at_ms = tonumber(target.lastHealthObservedAtMs or '0')
|
||||
if not last_observed_at_ms or last_observed_at_ms < 0 then
|
||||
return finish({status = 'invalid'})
|
||||
end
|
||||
if last_observed_at_ms > 0 then
|
||||
if type(last_task_id) ~= 'string' or last_task_id == '' or type(last_digest) ~= 'string' or
|
||||
string.len(last_digest) ~= 64 then
|
||||
return finish({status = 'invalid'})
|
||||
end
|
||||
if last_task_id == task_id then
|
||||
if last_digest == observation_digest then
|
||||
return finish({status = 'ok', target = raw_target}, proxy.expiresAtMs)
|
||||
end
|
||||
return finish({status = 'conflict'})
|
||||
end
|
||||
if observed_at_ms <= last_observed_at_ms then
|
||||
return finish({status = 'stale'})
|
||||
end
|
||||
end
|
||||
|
||||
target.lastHealthTaskId = task_id
|
||||
target.lastHealthDigest = observation_digest
|
||||
target.lastHealthObservedAtMs = observed_at_ms
|
||||
target.latencyNs = latency_ns
|
||||
if success == '1' then
|
||||
target.status = 'AVAILABLE'
|
||||
target.consecutiveFailures = 0
|
||||
target.lastSuccessAtMs = observed_at_ms
|
||||
else
|
||||
target.consecutiveFailures = target.consecutiveFailures + 1
|
||||
if target.consecutiveFailures >= max_consecutive_failures then
|
||||
target.status = 'UNHEALTHY'
|
||||
else
|
||||
target.status = 'SUSPECT'
|
||||
end
|
||||
end
|
||||
|
||||
local encoded = cjson.encode(target)
|
||||
redis.call('HSET', target_health_key, profile_key, encoded)
|
||||
redis.call('PEXPIREAT', target_health_key, proxy.expiresAtMs)
|
||||
return finish({status = 'ok', target = encoded}, proxy.expiresAtMs)
|
||||
@ -0,0 +1,91 @@
|
||||
//go:build integration
|
||||
|
||||
package redisactivity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"proxy-pool/internal/domain/activitypool"
|
||||
healthDomain "proxy-pool/internal/domain/health"
|
||||
proxyDomain "proxy-pool/internal/domain/proxy"
|
||||
)
|
||||
|
||||
func TestRedisKeepsTargetHealthOutsideGlobalProxyState(t *testing.T) {
|
||||
fixture := newRedisTestFixture(t)
|
||||
now := redisTestNow()
|
||||
if _, err := fixture.Adapter.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
|
||||
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
|
||||
Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")},
|
||||
}); err != nil {
|
||||
t.Fatalf("UpsertFetched(): %v", err)
|
||||
}
|
||||
firstObservation := redisTargetObservation("target-task-1", false, now.Add(time.Second))
|
||||
first, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{
|
||||
Observation: firstObservation, MaxConsecutiveFailures: 2,
|
||||
})
|
||||
if err != nil || first.Status != healthDomain.TargetSuspect || first.ConsecutiveFailures != 1 ||
|
||||
first.LastObservedAt != firstObservation.ObservedAt || first.Latency != firstObservation.Latency {
|
||||
t.Fatalf("ApplyTargetObservation(first) = %+v, %v", first, err)
|
||||
}
|
||||
replayed, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{
|
||||
Observation: firstObservation, MaxConsecutiveFailures: 2,
|
||||
})
|
||||
if err != nil || replayed != first {
|
||||
t.Fatalf("ApplyTargetObservation(replay) = %+v, %v", replayed, err)
|
||||
}
|
||||
conflicting := firstObservation
|
||||
conflicting.Success = true
|
||||
conflicting.FailureClass = ""
|
||||
if _, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{
|
||||
Observation: conflicting, MaxConsecutiveFailures: 2,
|
||||
}); !errors.Is(err, healthDomain.ErrConflictingObservation) {
|
||||
t.Fatalf("ApplyTargetObservation(conflict) error = %v, want ErrConflictingObservation", err)
|
||||
}
|
||||
second, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{
|
||||
Observation: redisTargetObservation("target-task-2", false, now.Add(2*time.Second)), MaxConsecutiveFailures: 2,
|
||||
})
|
||||
if err != nil || second.Status != healthDomain.TargetUnhealthy || second.ConsecutiveFailures != 2 {
|
||||
t.Fatalf("ApplyTargetObservation(second failure) = %+v, %v", second, err)
|
||||
}
|
||||
recoveredObservation := redisTargetObservation("target-task-3", true, now.Add(3*time.Second))
|
||||
recovered, err := fixture.Adapter.ApplyTargetObservation(context.Background(), activitypool.TargetHealthCommand{
|
||||
Observation: recoveredObservation, MaxConsecutiveFailures: 2,
|
||||
})
|
||||
if err != nil || recovered.Status != healthDomain.TargetAvailable || recovered.ConsecutiveFailures != 0 ||
|
||||
recovered.LastSuccessAt != recoveredObservation.ObservedAt {
|
||||
t.Fatalf("ApplyTargetObservation(recovery) = %+v, %v", recovered, err)
|
||||
}
|
||||
|
||||
profile := healthDomain.TargetProfile{RoutingName: recoveredObservation.RoutingName, TargetURL: recoveredObservation.TargetURL}
|
||||
profileKey := fixture.Adapter.keys.targetHealth("proxy-a")
|
||||
field := digestToken(profile.Key())
|
||||
if ttl, err := fixture.Client.PTTL(context.Background(), profileKey).Result(); err != nil || ttl <= 0 {
|
||||
t.Fatalf("target health key TTL = (%v, %v), want positive", ttl, err)
|
||||
}
|
||||
if exists, err := fixture.Client.HExists(context.Background(), profileKey, field).Result(); err != nil || !exists {
|
||||
t.Fatalf("target health profile exists = (%v, %v), want true", exists, err)
|
||||
}
|
||||
|
||||
global, err := fixture.Adapter.ApplyHealth(context.Background(), activitypool.HealthUpdate{
|
||||
ProxyID: "proxy-a", CheckedAt: now.Add(4 * time.Second), NextState: proxyDomain.StateChecking,
|
||||
})
|
||||
if err != nil || global.State != proxyDomain.StateChecking || global.GlobalHealth.LastObservedAt != (time.Time{}) ||
|
||||
global.GlobalHealth.ConsecutiveFailures != 0 {
|
||||
t.Fatalf("target observation changed global proxy state: %+v, %v", global, err)
|
||||
}
|
||||
}
|
||||
|
||||
func redisTargetObservation(taskID string, success bool, observedAt time.Time) healthDomain.Observation {
|
||||
observation := healthDomain.Observation{
|
||||
TaskID: taskID, ProxyID: "proxy-a", Level: healthDomain.LevelTarget,
|
||||
RoutingName: "route-a", TargetURL: "https://target.example/check", Success: success,
|
||||
Latency: 20 * time.Millisecond, ObservedAt: observedAt,
|
||||
}
|
||||
if !success {
|
||||
observation.FailureClass = "target_403"
|
||||
}
|
||||
return observation
|
||||
}
|
||||
@ -90,6 +90,17 @@ type GlobalHealthStore interface {
|
||||
ApplyGlobalObservation(context.Context, GlobalHealthCommand) (Entry, error)
|
||||
}
|
||||
|
||||
// TargetHealthCommand carries a TARGET fact. Its result is isolated by Proxy
|
||||
// and TargetProfile and must not modify the activity-pool Proxy state.
|
||||
type TargetHealthCommand struct {
|
||||
Observation healthDomain.Observation
|
||||
MaxConsecutiveFailures int
|
||||
}
|
||||
|
||||
type TargetHealthStore interface {
|
||||
ApplyTargetObservation(context.Context, TargetHealthCommand) (healthDomain.TargetState, error)
|
||||
}
|
||||
|
||||
type InventoryReader interface {
|
||||
Inventory(context.Context, string, time.Time) (Inventory, error)
|
||||
}
|
||||
@ -117,9 +128,15 @@ type MemoryPool struct {
|
||||
keyByID map[string]string
|
||||
idempotent map[string]idempotencyEntry
|
||||
ownership map[string]ownershipDomain.Assignment
|
||||
targets map[targetHealthKey]healthDomain.TargetState
|
||||
nextEpoch uint64
|
||||
}
|
||||
|
||||
type targetHealthKey struct {
|
||||
proxyID string
|
||||
profileKey string
|
||||
}
|
||||
|
||||
type idempotencyEntry struct {
|
||||
command extractionDomain.Command
|
||||
result extractionDomain.Result
|
||||
@ -130,6 +147,7 @@ var (
|
||||
_ Upserter = (*MemoryPool)(nil)
|
||||
_ HealthStore = (*MemoryPool)(nil)
|
||||
_ GlobalHealthStore = (*MemoryPool)(nil)
|
||||
_ TargetHealthStore = (*MemoryPool)(nil)
|
||||
_ InventoryReader = (*MemoryPool)(nil)
|
||||
_ StateInventoryReader = (*MemoryPool)(nil)
|
||||
_ Maintainer = (*MemoryPool)(nil)
|
||||
@ -143,6 +161,7 @@ func NewMemoryPool() *MemoryPool {
|
||||
keyByID: make(map[string]string),
|
||||
idempotent: make(map[string]idempotencyEntry),
|
||||
ownership: make(map[string]ownershipDomain.Assignment),
|
||||
targets: make(map[targetHealthKey]healthDomain.TargetState),
|
||||
}
|
||||
}
|
||||
|
||||
@ -372,6 +391,50 @@ func (p *MemoryPool) ApplyGlobalObservation(
|
||||
return entry, nil
|
||||
}
|
||||
|
||||
// ApplyTargetObservation atomically reduces an isolated TARGET profile. It
|
||||
// intentionally leaves the global Proxy state, health timestamp and indexes
|
||||
// untouched.
|
||||
func (p *MemoryPool) ApplyTargetObservation(
|
||||
ctx context.Context,
|
||||
command TargetHealthCommand,
|
||||
) (healthDomain.TargetState, error) {
|
||||
if ctx == nil {
|
||||
return healthDomain.TargetState{}, ErrInvalidHealthUpdate
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
normalized, err := healthDomain.NormalizeObservation(command.Observation)
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
if normalized.Level != healthDomain.LevelTarget || command.MaxConsecutiveFailures <= 0 || p == nil {
|
||||
return healthDomain.TargetState{}, ErrInvalidHealthUpdate
|
||||
}
|
||||
profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{
|
||||
RoutingName: normalized.RoutingName, TargetURL: normalized.TargetURL,
|
||||
})
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if err := ctx.Err(); err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
p.purgeExpiredLocked(normalized.ObservedAt)
|
||||
if _, exists := p.entryByIDLocked(normalized.ProxyID); !exists {
|
||||
return healthDomain.TargetState{}, ErrActivityNotFound
|
||||
}
|
||||
key := targetHealthKey{proxyID: normalized.ProxyID, profileKey: profile.Key()}
|
||||
next, err := healthDomain.ReduceTarget(p.targets[key], normalized, command.MaxConsecutiveFailures)
|
||||
if err != nil {
|
||||
return healthDomain.TargetState{}, err
|
||||
}
|
||||
p.targets[key] = next
|
||||
return next, nil
|
||||
}
|
||||
|
||||
func (p *MemoryPool) Inventory(ctx context.Context, upstreamID string, now time.Time) (Inventory, error) {
|
||||
result := Inventory{UpstreamID: upstreamID}
|
||||
if ctx == nil {
|
||||
@ -807,6 +870,11 @@ func (p *MemoryPool) removeEntryByIDLocked(proxyID string) {
|
||||
|
||||
func (p *MemoryPool) removeEntryLocked(key string, entry Entry) {
|
||||
delete(p.ownership, entry.Proxy.ID)
|
||||
for target := range p.targets {
|
||||
if target.proxyID == entry.Proxy.ID {
|
||||
delete(p.targets, target)
|
||||
}
|
||||
}
|
||||
delete(p.keyByID, entry.Proxy.ID)
|
||||
delete(p.entries, key)
|
||||
}
|
||||
|
||||
43
internal/domain/activitypool/target_health_test.go
Normal file
43
internal/domain/activitypool/target_health_test.go
Normal file
@ -0,0 +1,43 @@
|
||||
package activitypool
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
healthDomain "proxy-pool/internal/domain/health"
|
||||
proxyDomain "proxy-pool/internal/domain/proxy"
|
||||
)
|
||||
|
||||
func TestMemoryPoolKeepsTargetHealthOutsideGlobalProxyState(t *testing.T) {
|
||||
now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC)
|
||||
pool := seededHealthPool(t, now)
|
||||
before := pool.Snapshot(now)
|
||||
if len(before) != 1 || before[0].State != proxyDomain.StateFetched {
|
||||
t.Fatalf("before = %+v", before)
|
||||
}
|
||||
command := TargetHealthCommand{Observation: healthDomain.Observation{
|
||||
TaskID: "target-task-1", ProxyID: "proxy-a", Level: healthDomain.LevelTarget,
|
||||
RoutingName: "route-a", TargetURL: "https://target.example/check", Success: false,
|
||||
FailureClass: "target_403", Latency: 20 * time.Millisecond, ObservedAt: now.Add(time.Second),
|
||||
}, MaxConsecutiveFailures: 2}
|
||||
first, err := pool.ApplyTargetObservation(context.Background(), command)
|
||||
if err != nil || first.Status != healthDomain.TargetSuspect || first.ConsecutiveFailures != 1 {
|
||||
t.Fatalf("ApplyTargetObservation(first) = %+v, %v", first, err)
|
||||
}
|
||||
after := pool.Snapshot(now.Add(time.Second))
|
||||
if len(after) != 1 || after[0].State != before[0].State || after[0].GlobalHealth != before[0].GlobalHealth {
|
||||
t.Fatalf("target observation changed global entry: before=%+v after=%+v", before[0], after[0])
|
||||
}
|
||||
replayed, err := pool.ApplyTargetObservation(context.Background(), command)
|
||||
if err != nil || replayed != first {
|
||||
t.Fatalf("ApplyTargetObservation(replay) = %+v, %v", replayed, err)
|
||||
}
|
||||
conflicting := command
|
||||
conflicting.Observation.Success = true
|
||||
conflicting.Observation.FailureClass = ""
|
||||
if _, err := pool.ApplyTargetObservation(context.Background(), conflicting); !errors.Is(err, healthDomain.ErrConflictingObservation) {
|
||||
t.Fatalf("ApplyTargetObservation(conflict) error = %v, want ErrConflictingObservation", err)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user