diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 76a3491..acf1375 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -278,8 +278,9 @@ Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄 Lua 同一原子边界归并，覆盖首次 CHECKING 失败进入 UNHEALTHY、AVAILABLE/SUSPECT 的 阈值降级、精确重放、冲突拒绝与成功恢复。TARGET Profile 也已在 Memory 和 Redis 中独立 归并，以哈希键保存并随代理 TTL 过期，绝不写入 Proxy 全局状态或选择索引。调度器、 -Controller 到 Checker 的 RPC、任务调度和独立进程尚未接入；Controller 已新增公用 Reducer -作为 Observation 到原子 Store 的唯一归并边界，因此本任务保持未完成。 +Checker Observation 上报 RPC 已复用既有控制面监听接入 Controller：SPIFFE `checker` 身份、 +每批上限、配置阈值解析、代理归属查询与 Reducer 均已闭环。`StreamCheckTasks` 的租约调度、 +Checker 独立进程与探测器尚未实现，因此本任务保持未完成。 ## Task 12: Machine-readable Contracts diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 3308321..33a8972 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -60,7 +60,8 @@ Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controll 摘要 Redis 栅栏；原始事件不落 Redis 或 PostgreSQL。Checker 已有全局健康 Reducer 与 Memory/Redis 原子状态提交基础；TARGET Profile 以独立、随代理 TTL 过期的 Redis 记录归并，不改写 Proxy 全局状态。Controller 公用 Reducer 已作为 Observation 的唯一状态 -归并边界；任务调度、Checker RPC 和独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 +归并边界，Checker Observation RPC 已在同一控制面监听以独立 SPIFFE 身份接入；任务调度和 +独立执行进程尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller 在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用 `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot @@ -110,8 +111,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；其中全局与 TARGET Profile 的 Memory/Redis - 原子归并已完成。 +7. Checker 调度与探测器；全局与 TARGET Profile 的 Memory/Redis 原子归并、Controller + Reducer 和 Observation 上报 RPC 已完成。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index f370d27..e1480d8 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -7,7 +7,7 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Gateway 已将它们编译为同版本动态 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物待实现 | +| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Gateway 已将它们编译为同版本动态 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物与任务流待实现 | | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；Outcome 仅写入有界非阻塞本地队列，代表性性能剖析待完成 | | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 | | ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 | @@ -80,7 +80,7 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| | 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-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置有效合并、URL 校验和 Observation 上报已完成；有界调度器、抖动和分级频率测试待实现 | | HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Controller 公用 Reducer 已通过 Memory/Redis 活动池原子提交全局连续失败、精确重放和成功恢复；任务调度与 REMOVE 编排待实现 | | SEC-001 | API 认证与 Proxy 认证分离，Secret 统一脱敏 | 7528-8111, 8904-8945 | Config 脱敏、Provider Store -> SecretRef -> Gateway Resolver 跨包测试与格式化泄漏回归测试 | | SEC-002 | 非回环监听无保护时严格模式启动失败 | 8112-8441 | 配置校验测试 | diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index 79ebf1f..842686c 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -49,6 +49,11 @@ type targetHealthScriptReply struct { Target string `json:"target,omitempty"` } +type upstreamLookupScriptReply struct { + Status scriptStatus `json:"status"` + Upstream string `json:"upstream,omitempty"` +} + type extractScriptReply struct { Status scriptStatus `json:"status"` RequestDigest string `json:"requestDigest"` @@ -113,6 +118,9 @@ var healthSource string //go:embed scripts/target_health.lua var targetHealthSource string +//go:embed scripts/upstream_lookup.lua +var upstreamLookupSource string + //go:embed scripts/extract.lua var extractSource string @@ -138,6 +146,7 @@ var ( upsertScript = redis.NewScript(upsertSource) healthScript = redis.NewScript(healthSource) targetHealthScript = redis.NewScript(targetHealthSource) + upstreamLookupScript = redis.NewScript(upstreamLookupSource) extractScript = redis.NewScript(extractSource) ownershipScript = redis.NewScript(ownershipSource) sweepScript = redis.NewScript(sweepSource) diff --git a/internal/adapters/redisactivity/scripts/upstream_lookup.lua b/internal/adapters/redisactivity/scripts/upstream_lookup.lua new file mode 100644 index 0000000..6f15d93 --- /dev/null +++ b/internal/adapters/redisactivity/scripts/upstream_lookup.lua @@ -0,0 +1,22 @@ +local records_key = KEYS[1] + +local now_ms = tonumber(ARGV[1]) +local proxy_id = ARGV[2] + +if not now_ms or now_ms <= 0 or type(proxy_id) ~= 'string' or proxy_id == '' then + return cjson.encode({status = 'invalid'}) +end + +local raw = redis.call('HGET', records_key, proxy_id) +if not raw then + return cjson.encode({status = 'not_found'}) +end +local decoded, record = pcall(cjson.decode, raw) +if not decoded or type(record) ~= 'table' or type(record.sourceUpstream) ~= 'string' or + record.sourceUpstream == '' or type(record.expiresAtMs) ~= 'number' then + return cjson.encode({status = 'invalid'}) +end +if record.expiresAtMs <= now_ms then + return cjson.encode({status = 'not_found'}) +end +return cjson.encode({status = 'ok', upstream = record.sourceUpstream}) diff --git a/internal/adapters/redisactivity/upstream_lookup.go b/internal/adapters/redisactivity/upstream_lookup.go new file mode 100644 index 0000000..ab93c26 --- /dev/null +++ b/internal/adapters/redisactivity/upstream_lookup.go @@ -0,0 +1,45 @@ +package redisactivity + +import ( + "context" + "time" + + "proxy-pool/internal/domain/activitypool" +) + +var _ activitypool.ProxyUpstreamReader = (*Adapter)(nil) + +// UpstreamForProxy is a Controller-only metadata lookup. It returns neither +// the proxy record nor its credentials and is never used by the Gateway path. +func (a *Adapter) UpstreamForProxy(ctx context.Context, proxyID string, now time.Time) (string, error) { + if ctx == nil { + return "", activitypool.ErrInvalidProxyLookup + } + if err := ctx.Err(); err != nil { + return "", err + } + if a == nil || proxyID == "" || now.IsZero() { + return "", activitypool.ErrInvalidProxyLookup + } + result, err := runScript(ctx, a.client, upstreamLookupScript, []string{a.keys.records}, now.UnixMilli(), proxyID) + if err != nil { + return "", err + } + var reply upstreamLookupScriptReply + if err := decodeScriptResult(result, &reply); err != nil { + return "", err + } + switch reply.Status { + case scriptOK: + if reply.Upstream == "" { + return "", invalidScriptReply("upstream lookup reply omitted upstream") + } + return reply.Upstream, nil + case scriptNotFound: + return "", activitypool.ErrActivityNotFound + case scriptInvalid: + return "", activitypool.ErrInvalidProxyLookup + default: + return "", invalidScriptReply("unexpected upstream lookup status") + } +} diff --git a/internal/adapters/redisactivity/upstream_lookup_integration_test.go b/internal/adapters/redisactivity/upstream_lookup_integration_test.go new file mode 100644 index 0000000..1cc01bd --- /dev/null +++ b/internal/adapters/redisactivity/upstream_lookup_integration_test.go @@ -0,0 +1,31 @@ +//go:build integration + +package redisactivity + +import ( + "context" + "errors" + "testing" + "time" + + "proxy-pool/internal/domain/activitypool" + proxyDomain "proxy-pool/internal/domain/proxy" +) + +func TestRedisLooksUpOnlyLiveProxyUpstream(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) + } + upstream, err := fixture.Adapter.UpstreamForProxy(context.Background(), "proxy-a", now.Add(time.Second)) + if err != nil || upstream != "provider-a" { + t.Fatalf("UpstreamForProxy(live) = (%q, %v)", upstream, err) + } + if _, err := fixture.Adapter.UpstreamForProxy(context.Background(), "proxy-a", now.Add(2*time.Minute)); !errors.Is(err, activitypool.ErrActivityNotFound) { + t.Fatalf("UpstreamForProxy(expired) error = %v, want ErrActivityNotFound", err) + } +} diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index 949368b..2b23fe9 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -15,6 +15,7 @@ import ( "proxy-pool/internal/controller/admin" "proxy-pool/internal/controller/distribution" "proxy-pool/internal/controller/extraction" + controllerHealth "proxy-pool/internal/controller/health" "proxy-pool/internal/controller/operations" "proxy-pool/internal/controller/pool" "proxy-pool/internal/controller/provider" @@ -49,6 +50,9 @@ type Options struct { type activityStore interface { extractionDomain.Store activitypool.Upserter + activitypool.GlobalHealthStore + activitypool.TargetHealthStore + activitypool.ProxyUpstreamReader pool.InventoryReader activitypool.StateInventoryReader } @@ -79,7 +83,7 @@ type runtimeFactory interface { } type workerRuntimeFactory interface { - New(config.ControlPlane, worker.Service) (controllerRunner, error) + New(config.ControlPlane, worker.Service, worker.ServerOptions) (controllerRunner, error) } func Run(ctx context.Context, options Options) error { @@ -230,7 +234,7 @@ func runWithWorkerFactory( runners = append(runners, runner) } if loaded.Value.ControlPlane.Enabled { - if nilInterface(opened.workerStore) { + if nilInterface(opened.workerStore) || nilInterface(opened.activity) { return errors.Join(ErrStartup, ErrInvalidOptions) } var routingSources []worker.RoutingStateReader @@ -262,7 +266,25 @@ func runWithWorkerFactory( if serviceErr != nil { return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr) } - runner, runnerErr := workerFactory.New(loaded.Value.ControlPlane, service) + threshold, thresholdErr := controllerHealth.NewConfiguredFailureThresholdResolver(configurationStore, opened.activity, options.Now) + if thresholdErr != nil { + return fmt.Errorf("%w: build Checker failure threshold resolver: %w", ErrStartup, thresholdErr) + } + reducer, reducerErr := controllerHealth.NewReducer(opened.activity, opened.activity, threshold) + if reducerErr != nil { + return fmt.Errorf("%w: build Checker health reducer: %w", ErrStartup, reducerErr) + } + checkerIdentity, identityErr := newCheckerIdentity(loaded.Value.ControlPlane) + if identityErr != nil { + return fmt.Errorf("%w: build Checker identity authorizer: %w", ErrStartup, identityErr) + } + checkerHandler, handlerErr := controllerHealth.NewGRPCHandler(reducer, checkerIdentity, controllerHealth.DefaultGRPCHandlerOptions()) + if handlerErr != nil { + return fmt.Errorf("%w: build Checker control handler: %w", ErrStartup, handlerErr) + } + serverOptions := worker.DefaultServerOptions() + serverOptions.Checker = checkerHandler + runner, runnerErr := workerFactory.New(loaded.Value.ControlPlane, service, serverOptions) if runnerErr != nil { return fmt.Errorf("%w: build Worker control server: %w", ErrStartup, runnerErr) } @@ -279,6 +301,17 @@ func runWithWorkerFactory( return group.Run(ctx) } +func newCheckerIdentity(controlPlane config.ControlPlane) (controllerHealth.CheckerIdentityAuthorizer, error) { + switch controlPlane.TLS.Mode { + case "disabled": + return worker.AllowLoopbackIdentity{}, nil + case "mtls": + return worker.NewSPIFFEIdentityAuthorizer(controlPlane.TLS.TrustDomain, controlPlane.TLS.Environment) + default: + return nil, ErrInvalidOptions + } +} + func prepareProviderConfiguration( ctx context.Context, configuration *config.Config, @@ -361,6 +394,10 @@ func (productionRuntimeFactory) New( type productionWorkerRuntimeFactory struct{} -func (productionWorkerRuntimeFactory) New(controlPlane config.ControlPlane, service worker.Service) (controllerRunner, error) { - return worker.NewServer(controlPlane, service, worker.DefaultServerOptions()) +func (productionWorkerRuntimeFactory) New( + controlPlane config.ControlPlane, + service worker.Service, + options worker.ServerOptions, +) (controllerRunner, error) { + return worker.NewServer(controlPlane, service, options) } diff --git a/internal/controller/bootstrap/bootstrap_integration_test.go b/internal/controller/bootstrap/bootstrap_integration_test.go index 8eb51ff..72648f8 100644 --- a/internal/controller/bootstrap/bootstrap_integration_test.go +++ b/internal/controller/bootstrap/bootstrap_integration_test.go @@ -326,12 +326,13 @@ type integrationWorkerRuntimeFactory struct { func (factory *integrationWorkerRuntimeFactory) New( controlPlane config.ControlPlane, service controllerWorker.Service, + options controllerWorker.ServerOptions, ) (controllerRunner, error) { listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { return nil, err } - server, err := controllerWorker.NewServer(controlPlane, service, controllerWorker.DefaultServerOptions()) + server, err := controllerWorker.NewServer(controlPlane, service, options) if err != nil { _ = listener.Close() return nil, err diff --git a/internal/controller/bootstrap/bootstrap_test.go b/internal/controller/bootstrap/bootstrap_test.go index 9ed7a7c..0ad52d7 100644 --- a/internal/controller/bootstrap/bootstrap_test.go +++ b/internal/controller/bootstrap/bootstrap_test.go @@ -16,6 +16,7 @@ import ( "proxy-pool/internal/domain/activitypool" "proxy-pool/internal/domain/adminstate" extractionDomain "proxy-pool/internal/domain/extraction" + healthDomain "proxy-pool/internal/domain/health" "proxy-pool/internal/domain/upstream" "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/platform/admission" @@ -213,8 +214,8 @@ func TestRunStartsWorkerControlPlaneWithoutHTTPRuntime(t *testing.T) { if !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("run() error = %v, want context deadline exceeded", err) } - if workerFactory.controlPlane.Listen != "127.0.0.1:0" || workerFactory.service == nil { - t.Fatalf("worker runtime factory = controlPlane:%+v service:%T", workerFactory.controlPlane, workerFactory.service) + if workerFactory.controlPlane.Listen != "127.0.0.1:0" || workerFactory.service == nil || workerFactory.options.Checker == nil { + t.Fatalf("worker runtime factory = controlPlane:%+v service:%T checker:%T", workerFactory.controlPlane, workerFactory.service, workerFactory.options.Checker) } } @@ -383,6 +384,7 @@ func (factory runtimeFactoryFunc) New( type recordingWorkerRuntimeFactory struct { controlPlane config.ControlPlane service controllerWorker.Service + options controllerWorker.ServerOptions runner controllerRunner err error } @@ -390,9 +392,11 @@ type recordingWorkerRuntimeFactory struct { func (factory *recordingWorkerRuntimeFactory) New( controlPlane config.ControlPlane, service controllerWorker.Service, + options controllerWorker.ServerOptions, ) (controllerRunner, error) { factory.controlPlane = controlPlane factory.service = service + factory.options = options return factory.runner, factory.err } @@ -426,6 +430,24 @@ func (*stubActivityStore) UpsertFetched( return activitypool.UpsertResult{Accepted: len(batch.Proxies), Inserted: len(batch.Proxies)}, nil } +func (*stubActivityStore) ApplyGlobalObservation( + context.Context, + activitypool.GlobalHealthCommand, +) (activitypool.Entry, error) { + return activitypool.Entry{}, nil +} + +func (*stubActivityStore) ApplyTargetObservation( + context.Context, + activitypool.TargetHealthCommand, +) (healthDomain.TargetState, error) { + return healthDomain.TargetState{Status: healthDomain.TargetUnknown}, nil +} + +func (*stubActivityStore) UpstreamForProxy(context.Context, string, time.Time) (string, error) { + return "provider-a", nil +} + func (*stubActivityStore) ReadInventory( context.Context, string, diff --git a/internal/controller/health/grpc_handler.go b/internal/controller/health/grpc_handler.go index 7c7db11..d89dda0 100644 --- a/internal/controller/health/grpc_handler.go +++ b/internal/controller/health/grpc_handler.go @@ -120,7 +120,7 @@ func rejectedObservationError(err error) bool { return errors.Is(err, healthDomain.ErrInvalidObservation) || errors.Is(err, healthDomain.ErrInvalidFailureThreshold) || errors.Is(err, healthDomain.ErrStaleObservation) || errors.Is(err, healthDomain.ErrConflictingObservation) || errors.Is(err, activitypool.ErrInvalidHealthUpdate) || errors.Is(err, activitypool.ErrActivityNotFound) || - errors.Is(err, ErrInvalidThreshold) + errors.Is(err, ErrInvalidThreshold) || errors.Is(err, ErrUnconfiguredUpstream) } func healthGRPCError(err error) error { diff --git a/internal/controller/health/policy.go b/internal/controller/health/policy.go new file mode 100644 index 0000000..a6b5f5b --- /dev/null +++ b/internal/controller/health/policy.go @@ -0,0 +1,65 @@ +package health + +import ( + "context" + "errors" + "time" + + "proxy-pool/internal/config" + "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" +) + +var ( + ErrInvalidPolicyResolver = errors.New("invalid health policy resolver") + ErrUnconfiguredUpstream = errors.New("health observation references an unconfigured upstream") +) + +// ConfigurationSource returns a detached, validated configuration snapshot. +// config.Store implements this interface without exposing its publication +// mechanism to the health package. +type ConfigurationSource interface { + Current() *config.Config +} + +// NewConfiguredFailureThresholdResolver resolves the policy against the +// proxy's authoritative upstream at apply time. It keeps Checker input from +// choosing its own failure threshold and follows Controller config reloads. +func NewConfiguredFailureThresholdResolver( + configuration ConfigurationSource, + proxies activitypool.ProxyUpstreamReader, + now func() time.Time, +) (FailureThresholdResolver, error) { + if nilInterface(configuration) || nilInterface(proxies) || now == nil { + return nil, ErrInvalidPolicyResolver + } + return func(ctx context.Context, observation healthDomain.Observation) (int, error) { + if ctx == nil { + return 0, ErrInvalidPolicyResolver + } + if err := ctx.Err(); err != nil { + return 0, err + } + current := configuration.Current() + if current == nil { + return 0, ErrInvalidPolicyResolver + } + currentTime := now() + if currentTime.IsZero() { + return 0, ErrInvalidPolicyResolver + } + upstreamID, err := proxies.UpstreamForProxy(ctx, observation.ProxyID, currentTime.UTC()) + if err != nil { + return 0, err + } + upstream, exists := current.Upstreams[upstreamID] + if !exists || !upstream.Enabled { + return 0, ErrUnconfiguredUpstream + } + check := config.EffectiveCheck(current.Defaults.Check, upstream.Check) + if check.MaxConsecutiveFailures <= 0 { + return 0, ErrInvalidThreshold + } + return check.MaxConsecutiveFailures, nil + }, nil +} diff --git a/internal/controller/health/policy_test.go b/internal/controller/health/policy_test.go new file mode 100644 index 0000000..83d0eb0 --- /dev/null +++ b/internal/controller/health/policy_test.go @@ -0,0 +1,85 @@ +package health + +import ( + "context" + "errors" + "testing" + "time" + + "proxy-pool/internal/config" + "proxy-pool/internal/domain/activitypool" + healthDomain "proxy-pool/internal/domain/health" +) + +func TestConfiguredFailureThresholdResolverUsesCurrentProxyUpstream(t *testing.T) { + now := time.Date(2026, 7, 31, 16, 0, 0, 0, time.UTC) + source := &configurationSourceStub{configuration: &config.Config{ + Defaults: config.Defaults{Check: config.Check{MaxConsecutiveFailures: 2}}, + Upstreams: map[string]config.Upstream{ + "provider-a": {Enabled: true, Check: config.Check{MaxConsecutiveFailures: 4}}, + "provider-b": {Enabled: true}, + }, + }} + proxies := &proxyUpstreamReaderStub{upstreams: map[string]string{"proxy-a": "provider-a", "proxy-b": "provider-b"}} + resolver, err := NewConfiguredFailureThresholdResolver(source, proxies, func() time.Time { return now }) + if err != nil { + t.Fatalf("NewConfiguredFailureThresholdResolver(): %v", err) + } + if threshold, err := resolver(context.Background(), reducerObservation(healthDomain.LevelBasic, now)); err != nil || threshold != 4 { + t.Fatalf("resolver(provider-a) = (%d, %v)", threshold, err) + } + fact := reducerObservation(healthDomain.LevelTarget, now) + fact.ProxyID = "proxy-b" + if threshold, err := resolver(context.Background(), fact); err != nil || threshold != 2 { + t.Fatalf("resolver(provider-b default) = (%d, %v)", threshold, err) + } + if proxies.now != now { + t.Fatalf("lookup time = %v, want %v", proxies.now, now) + } +} + +func TestConfiguredFailureThresholdResolverRejectsUnknownAndInvalidInputs(t *testing.T) { + now := time.Date(2026, 7, 31, 16, 0, 0, 0, time.UTC) + source := &configurationSourceStub{configuration: &config.Config{Upstreams: map[string]config.Upstream{}}} + proxies := &proxyUpstreamReaderStub{upstreams: map[string]string{"proxy-a": "missing"}} + resolver, err := NewConfiguredFailureThresholdResolver(source, proxies, func() time.Time { return now }) + if err != nil { + t.Fatalf("NewConfiguredFailureThresholdResolver(): %v", err) + } + if _, err := resolver(context.Background(), reducerObservation(healthDomain.LevelBasic, now)); !errors.Is(err, ErrUnconfiguredUpstream) { + t.Fatalf("resolver(unknown) error = %v, want ErrUnconfiguredUpstream", err) + } + if resolver, err := NewConfiguredFailureThresholdResolver(nil, proxies, func() time.Time { return now }); err == nil || resolver != nil { + t.Fatalf("NewConfiguredFailureThresholdResolver(nil configuration) = (%v, %v)", resolver, err) + } + var typedNilReader *proxyUpstreamReaderStub + if resolver, err := NewConfiguredFailureThresholdResolver(source, typedNilReader, func() time.Time { return now }); err == nil || resolver != nil { + t.Fatalf("NewConfiguredFailureThresholdResolver(typed nil reader) = (%v, %v)", resolver, err) + } +} + +type configurationSourceStub struct { + configuration *config.Config +} + +func (source *configurationSourceStub) Current() *config.Config { + return source.configuration +} + +type proxyUpstreamReaderStub struct { + upstreams map[string]string + now time.Time + err error +} + +func (reader *proxyUpstreamReaderStub) UpstreamForProxy(_ context.Context, proxyID string, now time.Time) (string, error) { + reader.now = now + if reader.err != nil { + return "", reader.err + } + upstream, exists := reader.upstreams[proxyID] + if !exists { + return "", activitypool.ErrActivityNotFound + } + return upstream, nil +} diff --git a/internal/controller/worker/identity.go b/internal/controller/worker/identity.go index 4a213de..3e49ddc 100644 --- a/internal/controller/worker/identity.go +++ b/internal/controller/worker/identity.go @@ -10,7 +10,7 @@ import ( "google.golang.org/grpc/peer" ) -var ErrUnauthorizedIdentity = errors.New("worker identity is not authorized") +var ErrUnauthorizedIdentity = errors.New("control-plane identity is not authorized") // AllowLoopbackIdentity is used only by the validated loopback plaintext mode. // It deliberately does not inspect transport credentials because that mode has no TLS peer. @@ -20,6 +20,10 @@ func (AllowLoopbackIdentity) Authorize(context.Context, string) error { return nil } +func (AllowLoopbackIdentity) AuthorizeChecker(context.Context, string) error { + return nil +} + type SPIFFEIdentityAuthorizer struct { trustDomain string environment string @@ -33,7 +37,15 @@ func NewSPIFFEIdentityAuthorizer(trustDomain, environment string) (*SPIFFEIdenti } func (authorizer *SPIFFEIdentityAuthorizer) Authorize(ctx context.Context, workerID string) error { - if authorizer == nil || workerID == "" { + return authorizer.authorize(ctx, "worker", workerID) +} + +func (authorizer *SPIFFEIdentityAuthorizer) AuthorizeChecker(ctx context.Context, checkerID string) error { + return authorizer.authorize(ctx, "checker", checkerID) +} + +func (authorizer *SPIFFEIdentityAuthorizer) authorize(ctx context.Context, resourceType, resourceID string) error { + if authorizer == nil || (resourceType != "worker" && resourceType != "checker") || resourceID == "" { return ErrUnauthorizedIdentity } peerInfo, ok := peer.FromContext(ctx) @@ -45,8 +57,8 @@ func (authorizer *SPIFFEIdentityAuthorizer) Authorize(ctx context.Context, worke return ErrUnauthorizedIdentity } - expectedPath := "/" + authorizer.environment + "/worker/" + workerID - workerIdentityCount := 0 + expectedPath := "/" + authorizer.environment + "/" + resourceType + "/" + resourceID + identityCount := 0 for _, chain := range tlsInfo.State.VerifiedChains { if len(chain) == 0 || chain[0] == nil { continue @@ -56,16 +68,16 @@ func (authorizer *SPIFFEIdentityAuthorizer) Authorize(ctx context.Context, worke continue } segments := strings.Split(strings.Trim(uri.Path, "/"), "/") - if len(segments) != 3 || segments[1] != "worker" { + if len(segments) != 3 || segments[1] != resourceType { continue } - workerIdentityCount++ - if workerIdentityCount > 1 || uri.RawQuery != "" || uri.Fragment != "" || uri.Path != expectedPath { + identityCount++ + if identityCount > 1 || uri.RawQuery != "" || uri.Fragment != "" || uri.Path != expectedPath { return ErrUnauthorizedIdentity } } } - if workerIdentityCount != 1 { + if identityCount != 1 { return ErrUnauthorizedIdentity } return nil diff --git a/internal/controller/worker/identity_test.go b/internal/controller/worker/identity_test.go index 8a01ae1..b434be1 100644 --- a/internal/controller/worker/identity_test.go +++ b/internal/controller/worker/identity_test.go @@ -66,10 +66,29 @@ func TestSPIFFEIdentityAuthorizerRejectsMultipleWorkerURIs(t *testing.T) { } } +func TestSPIFFEIdentityAuthorizerAuthorizesCheckerOnlyForMatchingResource(t *testing.T) { + authorizer, err := NewSPIFFEIdentityAuthorizer("proxy.example", "prod") + if err != nil { + t.Fatalf("NewSPIFFEIdentityAuthorizer(): %v", err) + } + if err := authorizer.AuthorizeChecker(tlsPeerContext(t, "spiffe://proxy.example/prod/checker/checker-a"), "checker-a"); err != nil { + t.Fatalf("AuthorizeChecker(matching checker) = %v", err) + } + if err := authorizer.AuthorizeChecker(tlsPeerContext(t, "spiffe://proxy.example/prod/worker/checker-a"), "checker-a"); err == nil { + t.Fatal("AuthorizeChecker(worker identity) error = nil") + } + if err := authorizer.Authorize(tlsPeerContext(t, "spiffe://proxy.example/prod/checker/checker-a"), "checker-a"); err == nil { + t.Fatal("Authorize(checker identity) error = nil") + } +} + func TestAllowLoopbackIdentity(t *testing.T) { if err := (AllowLoopbackIdentity{}).Authorize(context.Background(), "worker-a"); err != nil { t.Fatalf("Authorize(): %v", err) } + if err := (AllowLoopbackIdentity{}).AuthorizeChecker(context.Background(), "checker-a"); err != nil { + t.Fatalf("AuthorizeChecker(): %v", err) + } } func tlsPeerContext(t *testing.T, identityURI string) context.Context { diff --git a/internal/controller/worker/server.go b/internal/controller/worker/server.go index 07bfe0c..19b7793 100644 --- a/internal/controller/worker/server.go +++ b/internal/controller/worker/server.go @@ -8,6 +8,7 @@ import ( "fmt" "net" "os" + "reflect" "strings" "time" @@ -24,6 +25,7 @@ var ErrInvalidServer = errors.New("invalid worker control server configuration") type ServerOptions struct { ShutdownTimeout time.Duration Snapshots SnapshotSource + Checker controlplanev1.CheckerControlPlaneServer } func DefaultServerOptions() ServerOptions { @@ -74,9 +76,25 @@ func NewServer(controlPlane config.ControlPlane, service Service, options Server ) grpcServer := grpc.NewServer(serverOptions...) controlplanev1.RegisterWorkerControlPlaneServer(grpcServer, NewGRPCHandler(service, identity, snapshots)) + if !nilService(options.Checker) { + controlplanev1.RegisterCheckerControlPlaneServer(grpcServer, options.Checker) + } return &Server{listen: controlPlane.Listen, grpcServer: grpcServer, shutdownTimeout: options.ShutdownTimeout}, nil } +func nilService(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} + func (server *Server) Run(ctx context.Context) error { if server == nil || server.grpcServer == nil || server.listen == "" { return ErrInvalidServer diff --git a/internal/controller/worker/server_test.go b/internal/controller/worker/server_test.go index fe55df1..d7f22d8 100644 --- a/internal/controller/worker/server_test.go +++ b/internal/controller/worker/server_test.go @@ -70,6 +70,56 @@ func TestServerServesAndStopsOnContextCancellation(t *testing.T) { } } +func TestServerRegistersCheckerServiceOnTheExistingControlPlaneListener(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("net.Listen(): %v", err) + } + controlPlane := validServerControlPlane() + controlPlane.Listen = listener.Addr().String() + checker := &checkerServiceStub{} + server, err := NewServer(controlPlane, &grpcServiceStub{}, ServerOptions{ShutdownTimeout: time.Second, Checker: checker}) + if err != nil { + t.Fatalf("NewServer(): %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- server.Serve(ctx, listener) }() + + connection, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatalf("grpc.NewClient(): %v", err) + } + response, err := controlplanev1.NewCheckerControlPlaneClient(connection).ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ + CheckerId: "checker-a", + }) + if err != nil || response.GetAccepted() != 1 || checker.checkerID != "checker-a" { + t.Fatalf("ReportObservations() = (%+v, %v); checker=%q", response, err, checker.checkerID) + } + if err := connection.Close(); err != nil { + t.Fatalf("connection.Close(): %v", err) + } + cancel() + select { + case err := <-result: + if err != nil { + t.Fatalf("Serve() error = %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Serve() did not stop after context cancellation") + } +} + +type checkerServiceStub struct { + controlplanev1.UnimplementedCheckerControlPlaneServer + checkerID string +} + +func (stub *checkerServiceStub) ReportObservations(_ context.Context, request *controlplanev1.ObservationBatch) (*controlplanev1.ReportObservationsResponse, error) { + stub.checkerID = request.GetCheckerId() + return &controlplanev1.ReportObservationsResponse{Accepted: 1}, nil +} + func validServerControlPlane() config.ControlPlane { return config.ControlPlane{ Enabled: true, diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index 8206a34..38132b4 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -20,6 +20,7 @@ const defaultIdempotencyTTL = 5 * time.Minute var ( ErrInvalidBatch = errors.New("invalid activity pool batch") ErrInvalidHealthUpdate = errors.New("invalid activity pool health update") + ErrInvalidProxyLookup = errors.New("invalid activity pool proxy lookup") ErrActivityNotFound = errors.New("activity pool proxy not found") ErrStaleHealthUpdate = errors.New("stale activity pool health update") ErrInvalidInventory = errors.New("invalid activity pool inventory query") @@ -101,6 +102,13 @@ type TargetHealthStore interface { ApplyTargetObservation(context.Context, TargetHealthCommand) (healthDomain.TargetState, error) } +// ProxyUpstreamReader returns only the authoritative upstream that owns a +// still-live proxy. It intentionally does not expose proxy endpoints or +// credentials to Controller policy resolution. +type ProxyUpstreamReader interface { + UpstreamForProxy(context.Context, string, time.Time) (string, error) +} + type InventoryReader interface { Inventory(context.Context, string, time.Time) (Inventory, error) } @@ -148,6 +156,7 @@ var ( _ HealthStore = (*MemoryPool)(nil) _ GlobalHealthStore = (*MemoryPool)(nil) _ TargetHealthStore = (*MemoryPool)(nil) + _ ProxyUpstreamReader = (*MemoryPool)(nil) _ InventoryReader = (*MemoryPool)(nil) _ StateInventoryReader = (*MemoryPool)(nil) _ Maintainer = (*MemoryPool)(nil) @@ -435,6 +444,26 @@ func (p *MemoryPool) ApplyTargetObservation( return next, nil } +func (p *MemoryPool) UpstreamForProxy(ctx context.Context, proxyID string, now time.Time) (string, error) { + if ctx == nil { + return "", ErrInvalidProxyLookup + } + if err := ctx.Err(); err != nil { + return "", err + } + if p == nil || proxyID == "" || now.IsZero() { + return "", ErrInvalidProxyLookup + } + p.mu.Lock() + defer p.mu.Unlock() + p.purgeExpiredLocked(now) + entry, exists := p.entryByIDLocked(proxyID) + if !exists || entry.Proxy.SourceUpstream == "" { + return "", ErrActivityNotFound + } + return entry.Proxy.SourceUpstream, nil +} + func (p *MemoryPool) Inventory(ctx context.Context, upstreamID string, now time.Time) (Inventory, error) { result := Inventory{UpstreamID: upstreamID} if ctx == nil { diff --git a/internal/domain/activitypool/upstream_lookup_test.go b/internal/domain/activitypool/upstream_lookup_test.go new file mode 100644 index 0000000..da0efca --- /dev/null +++ b/internal/domain/activitypool/upstream_lookup_test.go @@ -0,0 +1,23 @@ +package activitypool + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestMemoryPoolLooksUpOnlyLiveProxyUpstream(t *testing.T) { + now := time.Date(2026, 7, 31, 15, 0, 0, 0, time.UTC) + pool := seededHealthPool(t, now) + upstream, err := pool.UpstreamForProxy(context.Background(), "proxy-a", now.Add(time.Second)) + if err != nil || upstream != "provider-a" { + t.Fatalf("UpstreamForProxy(live) = (%q, %v)", upstream, err) + } + if _, err := pool.UpstreamForProxy(context.Background(), "proxy-a", now.Add(2*time.Minute)); !errors.Is(err, ErrActivityNotFound) { + t.Fatalf("UpstreamForProxy(expired) error = %v, want ErrActivityNotFound", err) + } + if _, err := pool.UpstreamForProxy(context.Background(), "", now); !errors.Is(err, ErrInvalidProxyLookup) { + t.Fatalf("UpstreamForProxy(empty) error = %v, want ErrInvalidProxyLookup", err) + } +}