diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index db5fd6f..7da6f92 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -19,8 +19,10 @@ import ( "proxy-pool/internal/controller/pool" "proxy-pool/internal/controller/provider" controllerRuntime "proxy-pool/internal/controller/runtime" + "proxy-pool/internal/controller/worker" "proxy-pool/internal/domain/activitypool" extractionDomain "proxy-pool/internal/domain/extraction" + "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/platform/admission" "proxy-pool/internal/platform/credentials" "proxy-pool/internal/platform/httpserver" @@ -59,6 +61,7 @@ type ports struct { coordinator provider.Coordinator credentials credentials.Store providerResults provider.ResultRecorder + workerStore workerruntime.ControlStore close func() error } @@ -74,15 +77,29 @@ type runtimeFactory interface { New(*config.Config, controllerRuntime.Dependencies, controllerRuntime.Options) (controllerRunner, error) } +type workerRuntimeFactory interface { + New(config.ControlPlane, worker.Service) (controllerRunner, error) +} + func Run(ctx context.Context, options Options) error { - return run(ctx, options, &productionInfrastructure{ + return runWithWorkerFactory(ctx, options, &productionInfrastructure{ holderID: options.HolderID, namespace: options.RedisNamespace, - }, productionRuntimeFactory{}) + }, productionRuntimeFactory{}, productionWorkerRuntimeFactory{}) } func run(ctx context.Context, options Options, infrastructure infrastructure, factory runtimeFactory) (resultErr error) { + return runWithWorkerFactory(ctx, options, infrastructure, factory, productionWorkerRuntimeFactory{}) +} + +func runWithWorkerFactory( + ctx context.Context, + options Options, + infrastructure infrastructure, + factory runtimeFactory, + workerFactory workerRuntimeFactory, +) (resultErr error) { if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" || - nilInterface(options.Resolver) || nilInterface(infrastructure) || nilInterface(factory) { + nilInterface(options.Resolver) || nilInterface(infrastructure) || nilInterface(factory) || nilInterface(workerFactory) { return ErrInvalidOptions } if err := ctx.Err(); err != nil { @@ -200,7 +217,7 @@ func run(ctx context.Context, options Options, infrastructure infrastructure, fa dependencies.MetricsHandler = handler } - runners := make([]lifecycle.Runner, 0, 2) + runners := make([]lifecycle.Runner, 0, 3) if hasHTTPRuntime(loaded.Value) { runner, err := factory.New(configurationStore.Current(), dependencies, controllerRuntime.Options{HTTP: options.HTTP}) if err != nil { @@ -211,6 +228,29 @@ func run(ctx context.Context, options Options, infrastructure infrastructure, fa } runners = append(runners, runner) } + if loaded.Value.ControlPlane.Enabled { + if nilInterface(opened.workerStore) { + return errors.Join(ErrStartup, ErrInvalidOptions) + } + service, serviceErr := worker.NewService(opened.workerStore, worker.Options{ + ProtocolVersion: loaded.Value.ControlPlane.ProtocolVersion, + HeartbeatInterval: loaded.Value.ControlPlane.HeartbeatInterval.Value(), + SessionTTL: loaded.Value.ControlPlane.SessionTTL.Value(), + MaxStaleAge: loaded.Value.ControlPlane.MaxStaleAge.Value(), + MaxRuntimeCounters: loaded.Value.ControlPlane.MaxRuntimeCounters, + }) + if serviceErr != nil { + return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr) + } + runner, runnerErr := workerFactory.New(loaded.Value.ControlPlane, service) + if runnerErr != nil { + return fmt.Errorf("%w: build Worker control server: %w", ErrStartup, runnerErr) + } + if nilInterface(runner) { + return errors.Join(ErrStartup, ErrInvalidOptions) + } + runners = append(runners, runner) + } runners = append(runners, supervisor) group, err := lifecycle.NewGroup(runners...) if err != nil { @@ -298,3 +338,9 @@ func (productionRuntimeFactory) New( ) (controllerRunner, error) { return controllerRuntime.New(configuration, dependencies, options) } + +type productionWorkerRuntimeFactory struct{} + +func (productionWorkerRuntimeFactory) New(controlPlane config.ControlPlane, service worker.Service) (controllerRunner, error) { + return worker.NewServer(controlPlane, service, worker.DefaultServerOptions()) +} diff --git a/internal/controller/bootstrap/bootstrap_test.go b/internal/controller/bootstrap/bootstrap_test.go index fd15bf1..9ed7a7c 100644 --- a/internal/controller/bootstrap/bootstrap_test.go +++ b/internal/controller/bootstrap/bootstrap_test.go @@ -12,10 +12,12 @@ import ( "proxy-pool/internal/controller/pool" "proxy-pool/internal/controller/provider" controllerRuntime "proxy-pool/internal/controller/runtime" + controllerWorker "proxy-pool/internal/controller/worker" "proxy-pool/internal/domain/activitypool" "proxy-pool/internal/domain/adminstate" extractionDomain "proxy-pool/internal/domain/extraction" "proxy-pool/internal/domain/upstream" + "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/platform/admission" "proxy-pool/internal/platform/credentials" ) @@ -170,6 +172,52 @@ func TestRunSupportsProviderOnlyConfigurationWithoutHTTPRuntime(t *testing.T) { } } +func TestRunRequiresWorkerControlStoreWhenControlPlaneEnabled(t *testing.T) { + source := bootstrapControlPlaneConfig() + credentialStore, err := credentials.NewMemoryStore(10) + if err != nil { + t.Fatalf("NewMemoryStore(): %v", err) + } + infrastructure := &stubInfrastructure{ports: ports{ + activity: &stubActivityStore{}, coordinator: coordinatorStub{}, credentials: credentialStore, close: func() error { return nil }, + }} + err = run(context.Background(), Options{ + ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(source)}}, Now: time.Now, + }, infrastructure, &recordingRuntimeFactory{}) + if !errors.Is(err, ErrStartup) || !errors.Is(err, ErrInvalidOptions) { + t.Fatalf("run() error = %v, want startup invalid options", err) + } +} + +func TestRunStartsWorkerControlPlaneWithoutHTTPRuntime(t *testing.T) { + store, err := workerruntime.NewMemoryStore(time.Now) + if err != nil { + t.Fatalf("NewMemoryStore(): %v", err) + } + credentialStore, err := credentials.NewMemoryStore(10) + if err != nil { + t.Fatalf("NewMemoryStore(): %v", err) + } + infrastructure := &stubInfrastructure{ports: ports{ + activity: &stubActivityStore{}, workerStore: store, coordinator: coordinatorStub{}, credentials: credentialStore, close: func() error { return nil }, + }} + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + workerFactory := &recordingWorkerRuntimeFactory{runner: runnerFunc(func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() + })} + err = runWithWorkerFactory(ctx, Options{ + ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(bootstrapControlPlaneConfig())}}, Now: time.Now, + }, infrastructure, &recordingRuntimeFactory{}, workerFactory) + 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) + } +} + func TestRunAdminDisableStopsActiveProviderRuntime(t *testing.T) { state := adminstate.NewMemoryStore() credentialStore, err := credentials.NewMemoryStore(200) @@ -332,6 +380,22 @@ func (factory runtimeFactoryFunc) New( return factory(configuration, dependencies, options) } +type recordingWorkerRuntimeFactory struct { + controlPlane config.ControlPlane + service controllerWorker.Service + runner controllerRunner + err error +} + +func (factory *recordingWorkerRuntimeFactory) New( + controlPlane config.ControlPlane, + service controllerWorker.Service, +) (controllerRunner, error) { + factory.controlPlane = controlPlane + factory.service = service + return factory.runner, factory.err +} + type readyStub struct{} func (readyStub) Ready(context.Context) error { return nil } @@ -453,3 +517,22 @@ upstreams: urls: [https://example.invalid/health] provider-b: *upstream ` + +func bootstrapControlPlaneConfig() string { + source := strings.ReplaceAll(bootstrapTestConfig, "distribution:\n enabled: true", "distribution:\n enabled: false") + source = strings.ReplaceAll(source, "admin:\n enabled: true", "admin:\n enabled: false") + source = strings.ReplaceAll(source, "metrics:\n enabled: true", "metrics:\n enabled: false") + return source + ` +controlPlane: + enabled: true + listen: 127.0.0.1:0 + protocolVersion: 1 + heartbeatInterval: 10s + sessionTTL: 30s + maxStaleAge: 10s + maxMessageBytes: 1048576 + maxRuntimeCounters: 100 + maxConcurrentStreams: 10 + tls: {mode: disabled} +` +} diff --git a/internal/controller/bootstrap/infrastructure.go b/internal/controller/bootstrap/infrastructure.go index 8c2501f..6a4bf4f 100644 --- a/internal/controller/bootstrap/infrastructure.go +++ b/internal/controller/bootstrap/infrastructure.go @@ -102,7 +102,7 @@ func (infrastructure *productionInfrastructure) Open( } providersEnabled := hasEnabledUpstream(configuration) - if configuration.Distribution.Enabled || configuration.Admin.Enabled || providersEnabled { + if configuration.Distribution.Enabled || configuration.Admin.Enabled || providersEnabled || configuration.ControlPlane.Enabled { if strings.TrimSpace(configuration.Storage.RedisURL) == "" { return ports{}, ErrRedisConfiguration } @@ -123,7 +123,7 @@ func (infrastructure *productionInfrastructure) Open( Credentials: credentialStore, OperationTTL: redisOperationTTL, MaxCandidateScan: candidateScan(configuration), - MaxRuntimeCounters: credentialCapacity(configuration), + MaxRuntimeCounters: runtimeCounterCapacity(configuration), MaxInventoryScan: maxInventoryScan(configuration), CleanupLimit: redisCleanupLimit, }) @@ -131,6 +131,7 @@ func (infrastructure *productionInfrastructure) Open( return ports{}, err } opened.activity = adapter + opened.workerStore = adapter opened.readiness = redisReadiness{client: redisClient} opened.credentials = credentialStore if configuration.Distribution.Enabled { @@ -203,7 +204,7 @@ func selectMetricsReadiness( configuration *config.Config, admin, activity platformMetrics.ReadinessChecker, ) platformMetrics.ReadinessChecker { - if configuration.Distribution.Enabled || hasEnabledUpstream(configuration) { + if configuration.Distribution.Enabled || configuration.ControlPlane.Enabled || hasEnabledUpstream(configuration) { return activity } if configuration.Admin.Enabled { @@ -275,6 +276,14 @@ func credentialCapacity(configuration *config.Config) int { return capacity } +func runtimeCounterCapacity(configuration *config.Config) int { + capacity := credentialCapacity(configuration) + if configuration != nil && configuration.ControlPlane.Enabled && configuration.ControlPlane.MaxRuntimeCounters > capacity { + return configuration.ControlPlane.MaxRuntimeCounters + } + return capacity +} + func providerCredentialCapacity(configuration *config.Config) int { capacity := 0 maximum := int(^uint(0) >> 1) diff --git a/internal/controller/bootstrap/infrastructure_test.go b/internal/controller/bootstrap/infrastructure_test.go index f9be8dc..40a4e97 100644 --- a/internal/controller/bootstrap/infrastructure_test.go +++ b/internal/controller/bootstrap/infrastructure_test.go @@ -50,6 +50,13 @@ func TestProductionInfrastructureRejectsInvalidStorageWithoutLeakingURLs(t *test if !errors.Is(err, ErrRedisConfiguration) || strings.Contains(err.Error(), redisSecret) { t.Fatalf("Open(invalid Redis) error = %v", err) } + _, err = (&productionInfrastructure{}).Open(context.Background(), &config.Config{ + ControlPlane: config.ControlPlane{Enabled: true}, + Storage: config.Storage{RedisURL: "redis://user:" + redisSecret + "@%zz"}, + }) + if !errors.Is(err, ErrRedisConfiguration) || strings.Contains(err.Error(), redisSecret) { + t.Fatalf("Open(control plane invalid Redis) error = %v", err) + } } func TestNewDistributionAdmitterPassesConfiguredLimits(t *testing.T) { @@ -124,6 +131,19 @@ func TestSelectMetricsReadinessUsesAdminStoresWithoutDistribution(t *testing.T) } } +func TestSelectMetricsReadinessUsesRedisForControlPlane(t *testing.T) { + t.Parallel() + wantErr := errors.New("redis unavailable") + selected := selectMetricsReadiness( + &config.Config{ControlPlane: config.ControlPlane{Enabled: true}}, + readinessFunc(func(context.Context) error { return nil }), + readinessFunc(func(context.Context) error { return wantErr }), + ) + if err := selected.Ready(context.Background()); !errors.Is(err, wantErr) { + t.Fatalf("Ready() error = %v, want %v", err, wantErr) + } +} + type readinessFunc func(context.Context) error func (function readinessFunc) Ready(ctx context.Context) error { return function(ctx) } @@ -143,6 +163,14 @@ func TestProductionRedisSizingUsesConfigurationBounds(t *testing.T) { if got := credentialCapacity(configuration); got != 5_000 { t.Fatalf("credentialCapacity() = %d, want 5000", got) } + configuration.ControlPlane = config.ControlPlane{Enabled: true, MaxRuntimeCounters: 100_000} + if got := runtimeCounterCapacity(configuration); got != 100_000 { + t.Fatalf("runtimeCounterCapacity(control plane) = %d, want 100000", got) + } + configuration.ControlPlane.Enabled = false + if got := runtimeCounterCapacity(configuration); got != 5_000 { + t.Fatalf("runtimeCounterCapacity(disabled control plane) = %d, want 5000", got) + } if got := providerCredentialCapacity(configuration); got != 158_000 { t.Fatalf("providerCredentialCapacity() = %d, want 158000", got) }