diff --git a/internal/controller/bootstrap/bootstrap_integration_test.go b/internal/controller/bootstrap/bootstrap_integration_test.go index a72010b..b89c7a6 100644 --- a/internal/controller/bootstrap/bootstrap_integration_test.go +++ b/internal/controller/bootstrap/bootstrap_integration_test.go @@ -4,7 +4,9 @@ package bootstrap import ( "context" + "crypto/sha256" "errors" + "net" "net/http" "net/http/httptest" "os" @@ -22,7 +24,14 @@ 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/workerruntime" "proxy-pool/internal/platform/credentials" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/types/known/timestamppb" + controlplanev1 "proxy-pool/gen/controlplane/v1" ) func TestProductionBootstrapOpensBothStoresAndCommitsStartupConfiguration(t *testing.T) { @@ -80,6 +89,95 @@ func TestProductionBootstrapOpensBothStoresAndCommitsStartupConfiguration(t *tes } } +func TestProductionBootstrapServesWorkerControlPlane(t *testing.T) { + postgresURL := os.Getenv("PROXY_POOL_TEST_POSTGRES_URL") + redisURL := os.Getenv("PROXY_POOL_TEST_REDIS_URL") + if postgresURL == "" || redisURL == "" { + t.Skip("PROXY_POOL_TEST_POSTGRES_URL and PROXY_POOL_TEST_REDIS_URL are required") + } + providerServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { + _, _ = writer.Write([]byte("http://192.0.2.10:8080")) + })) + defer providerServer.Close() + namespace := "controller-worker-it-" + strconv.FormatInt(time.Now().UnixNano(), 10) + source := strings.ReplaceAll(bootstrapTestConfig, "postgres://fixture", postgresURL) + source = strings.ReplaceAll(source, "redis://fixture", redisURL) + source = strings.ReplaceAll(source, "https://provider.invalid/proxies", providerServer.URL) + 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} +` + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + workerFactory := &integrationWorkerRuntimeFactory{ready: make(chan struct{})} + result := make(chan error, 1) + go func() { + result <- runWithWorkerFactory(ctx, Options{ + ConfigPath: "controller.yaml", Resolver: &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(source)}}, + Now: time.Now, FingerprintKey: bootstrapTestFingerprintKey, + }, &productionInfrastructure{namespace: namespace}, integrationBlockingRuntimeFactory{}, workerFactory) + }() + select { + case <-workerFactory.ready: + case <-time.After(5 * time.Second): + t.Fatal("worker control plane did not start") + } + + connection, err := grpc.NewClient(workerFactory.address, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatalf("grpc.NewClient(): %v", err) + } + defer connection.Close() + client := controlplanev1.NewWorkerControlPlaneClient(connection) + requestCtx, requestCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer requestCancel() + registration, err := client.RegisterWorker(requestCtx, &controlplanev1.RegisterWorkerRequest{ + WorkerId: "worker-a", InstanceId: "instance-a", Zone: "zone-a", SupportedProtocolVersion: 1, + }) + if err != nil { + t.Fatalf("RegisterWorker(): %v", err) + } + store := newIntegrationWorkerStore(t, redisURL, namespace) + checksum := sha256.Sum256([]byte("snapshot-1")) + reference := workerruntime.SnapshotReference{ + WorkerID: "worker-a", Version: 1, OwnershipEpoch: registration.GetOwnershipEpoch(), Checksum: checksum, + } + if err := store.RecordIssuedSnapshot(requestCtx, reference, time.Minute); err != nil { + t.Fatalf("RecordIssuedSnapshot(): %v", err) + } + if _, err := client.AcknowledgeSnapshot(requestCtx, &controlplanev1.AcknowledgeSnapshotRequest{ + WorkerId: "worker-a", SessionId: registration.GetSessionId(), Version: 1, + OwnershipEpoch: registration.GetOwnershipEpoch(), Checksum: checksum[:], Applied: true, + }); err != nil { + t.Fatalf("AcknowledgeSnapshot(): %v", err) + } + runtime, err := client.ReportRuntime(requestCtx, &controlplanev1.ReportRuntimeRequest{ + WorkerId: "worker-a", SessionId: registration.GetSessionId(), SnapshotVersion: 1, + OwnershipEpoch: registration.GetOwnershipEpoch(), ReportSequence: 1, ObservedAt: timestamppb.Now(), + }) + if err != nil || runtime.GetRequireFullSnapshot() || runtime.GetAcceptedOwnershipEpoch() != registration.GetOwnershipEpoch() { + t.Fatalf("ReportRuntime() = %+v, %v", runtime, err) + } + cancel() + select { + case err := <-result: + if !errors.Is(err, context.Canceled) { + t.Fatalf("run() error = %v, want context cancellation", err) + } + case <-time.After(5 * time.Second): + t.Fatal("controller did not stop") + } +} + type integrationRuntimeFactory struct { status admin.Status readyStatus int @@ -206,3 +304,62 @@ type integrationRunner struct { } func (runner integrationRunner) Run(ctx context.Context) error { return runner.run(ctx) } + +type integrationBlockingRuntimeFactory struct{} + +func (integrationBlockingRuntimeFactory) New( + _ *config.Config, + _ controllerRuntime.Dependencies, + _ controllerRuntime.Options, +) (controllerRunner, error) { + return integrationRunner{run: func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() + }}, nil +} + +type integrationWorkerRuntimeFactory struct { + address string + ready chan struct{} +} + +func (factory *integrationWorkerRuntimeFactory) New( + controlPlane config.ControlPlane, + service controllerWorker.Service, +) (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()) + if err != nil { + _ = listener.Close() + return nil, err + } + factory.address = listener.Addr().String() + close(factory.ready) + return integrationRunner{run: func(ctx context.Context) error { return server.Serve(ctx, listener) }}, nil +} + +func newIntegrationWorkerStore(t *testing.T, redisURL, namespace string) *redisactivity.Adapter { + t.Helper() + options, err := redis.ParseURL(redisURL) + if err != nil { + t.Fatalf("redis.ParseURL(): %v", err) + } + client := redis.NewClient(options) + t.Cleanup(func() { _ = client.Close() }) + credentialStore, err := credentials.NewMemoryStore(100) + if err != nil { + t.Fatalf("credentials.NewMemoryStore(): %v", err) + } + store, err := redisactivity.New(client, redisactivity.Options{ + Namespace: namespace, Credentials: credentialStore, OperationTTL: redisOperationTTL, + MaxCandidateScan: redisMinimumScan, MaxRuntimeCounters: 100, MaxInventoryScan: 100, + CleanupLimit: redisCleanupLimit, + }) + if err != nil { + t.Fatalf("redisactivity.New(): %v", err) + } + return store +}