344 lines
12 KiB
Go
344 lines
12 KiB
Go
//go:build integration
|
|
|
|
package bootstrap
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
|
|
"proxy-pool/internal/adapters/redisactivity"
|
|
"proxy-pool/internal/config"
|
|
"proxy-pool/internal/controller/admin"
|
|
"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/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) {
|
|
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")
|
|
}
|
|
var providerCalls atomic.Int64
|
|
providerServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
|
|
providerCalls.Add(1)
|
|
_, _ = writer.Write([]byte("http://192.0.2.10:8080"))
|
|
}))
|
|
defer providerServer.Close()
|
|
namespace := "controller-it-" + strconv.FormatInt(time.Now().UnixNano(), 10)
|
|
inventory := newIntegrationInventoryReader(t, redisURL, namespace)
|
|
|
|
source := strings.ReplaceAll(bootstrapTestConfig, "postgres://fixture", postgresURL)
|
|
source = strings.ReplaceAll(source, "redis://fixture", redisURL)
|
|
source = strings.ReplaceAll(source, "https://provider.invalid/proxies", providerServer.URL)
|
|
resolver := &memoryResolver{files: map[string][]byte{"controller.yaml": []byte(source)}}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
providerResults := make(chan provider.Result, 8)
|
|
factory := &integrationRuntimeFactory{
|
|
cancel: cancel, inventory: inventory, providerResults: providerResults,
|
|
}
|
|
infrastructure := &integrationInfrastructure{
|
|
productionInfrastructure: productionInfrastructure{namespace: namespace},
|
|
results: providerResultRecorder(func(result provider.Result) { providerResults <- result }),
|
|
}
|
|
|
|
err := run(ctx, Options{
|
|
ConfigPath: "controller.yaml", Resolver: resolver, Now: func() time.Time {
|
|
return time.Date(2026, 7, 30, 13, 0, 0, 0, time.UTC)
|
|
},
|
|
FingerprintKey: bootstrapTestFingerprintKey,
|
|
}, infrastructure, factory)
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("run() error = %v, want context cancellation", err)
|
|
}
|
|
if factory.status.ConfigVersion == "" || len(factory.status.Upstreams) != 2 {
|
|
t.Fatalf("Admin Status = %+v", factory.status)
|
|
}
|
|
if factory.status.Upstreams[0].Name != "provider-a" || factory.status.Upstreams[0].Available != 0 ||
|
|
factory.status.Upstreams[1].Name != "provider-b" {
|
|
t.Fatalf("Admin Status upstreams = %+v", factory.status.Upstreams)
|
|
}
|
|
if factory.readyStatus != http.StatusOK || factory.metricsStatus != http.StatusOK ||
|
|
!strings.Contains(factory.metricsBody, "go_") {
|
|
t.Fatalf("Metrics probes = ready:%d metrics:%d body:%q", factory.readyStatus, factory.metricsStatus, factory.metricsBody)
|
|
}
|
|
if providerCalls.Load() == 0 {
|
|
t.Fatal("production Provider HTTP adapter was not called")
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
stream, err := client.WatchSnapshots(requestCtx, &controlplanev1.WatchSnapshotsRequest{
|
|
WorkerId: "worker-a", SessionId: registration.GetSessionId(),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("WatchSnapshots(): %v", err)
|
|
}
|
|
issued, err := stream.Recv()
|
|
if err != nil || issued.GetFull() == nil || issued.GetFull().GetOwnershipEpoch() != registration.GetOwnershipEpoch() {
|
|
t.Fatalf("WatchSnapshots.Recv() = %+v, %v", issued, err)
|
|
}
|
|
if _, err := client.AcknowledgeSnapshot(requestCtx, &controlplanev1.AcknowledgeSnapshotRequest{
|
|
WorkerId: "worker-a", SessionId: registration.GetSessionId(), Version: issued.GetFull().GetVersion(),
|
|
OwnershipEpoch: issued.GetFull().GetOwnershipEpoch(), Checksum: issued.GetFull().GetChecksum(), Applied: true,
|
|
}); err != nil {
|
|
t.Fatalf("AcknowledgeSnapshot(): %v", err)
|
|
}
|
|
runtime, err := client.ReportRuntime(requestCtx, &controlplanev1.ReportRuntimeRequest{
|
|
WorkerId: "worker-a", SessionId: registration.GetSessionId(), SnapshotVersion: issued.GetFull().GetVersion(),
|
|
OwnershipEpoch: issued.GetFull().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
|
|
metricsStatus int
|
|
metricsBody string
|
|
cancel context.CancelFunc
|
|
inventory pool.InventoryReader
|
|
providerResults <-chan provider.Result
|
|
}
|
|
|
|
func (factory *integrationRuntimeFactory) New(
|
|
_ *config.Config,
|
|
dependencies controllerRuntime.Dependencies,
|
|
_ controllerRuntime.Options,
|
|
) (controllerRunner, error) {
|
|
return integrationRunner{run: func(ctx context.Context) error {
|
|
if err := dependencies.Readiness.Ready(ctx); err != nil {
|
|
return err
|
|
}
|
|
ready := httptest.NewRecorder()
|
|
dependencies.MetricsHandler.ServeHTTP(ready, httptest.NewRequest(http.MethodGet, "/readyz", nil))
|
|
factory.readyStatus = ready.Code
|
|
metrics := httptest.NewRecorder()
|
|
dependencies.MetricsHandler.ServeHTTP(metrics, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
|
factory.metricsStatus = metrics.Code
|
|
factory.metricsBody = metrics.Body.String()
|
|
status, err := dependencies.AdminService.Status(ctx)
|
|
factory.status = status
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := waitForProviderInventory(
|
|
ctx,
|
|
factory.inventory,
|
|
factory.providerResults,
|
|
[]string{"provider-a", "provider-b"},
|
|
); err != nil {
|
|
return err
|
|
}
|
|
factory.cancel()
|
|
<-ctx.Done()
|
|
return ctx.Err()
|
|
}}, nil
|
|
}
|
|
|
|
func newIntegrationInventoryReader(t *testing.T, redisURL, namespace string) pool.InventoryReader {
|
|
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(200)
|
|
if err != nil {
|
|
t.Fatalf("credentials.NewMemoryStore(): %v", err)
|
|
}
|
|
reader, err := redisactivity.New(client, redisactivity.Options{
|
|
Namespace: namespace, Credentials: credentialStore,
|
|
OperationTTL: redisOperationTTL, MaxCandidateScan: redisMinimumScan,
|
|
MaxRuntimeCounters: 200, MaxInventoryScan: 200, CleanupLimit: redisCleanupLimit,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("redisactivity.New(): %v", err)
|
|
}
|
|
return reader
|
|
}
|
|
|
|
func waitForProviderInventory(
|
|
ctx context.Context,
|
|
inventory pool.InventoryReader,
|
|
results <-chan provider.Result,
|
|
upstreamIDs []string,
|
|
) error {
|
|
deadline, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
defer cancel()
|
|
ticker := time.NewTicker(10 * time.Millisecond)
|
|
defer ticker.Stop()
|
|
var latestResult provider.Result
|
|
for {
|
|
var inventoryErr error
|
|
for _, upstreamID := range upstreamIDs {
|
|
snapshot, err := inventory.ReadInventory(deadline, upstreamID, 0)
|
|
inventoryErr = errors.Join(inventoryErr, err)
|
|
if err == nil && snapshot.Managed > 0 {
|
|
return nil
|
|
}
|
|
}
|
|
select {
|
|
case <-deadline.Done():
|
|
return errors.Join(
|
|
errors.New("wait for Provider Redis inventory"),
|
|
deadline.Err(),
|
|
inventoryErr,
|
|
latestResult.Err,
|
|
)
|
|
case result := <-results:
|
|
latestResult = result
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
type integrationInfrastructure struct {
|
|
productionInfrastructure
|
|
results provider.ResultRecorder
|
|
}
|
|
|
|
func (infrastructure *integrationInfrastructure) Open(
|
|
ctx context.Context,
|
|
configuration *config.Config,
|
|
) (ports, error) {
|
|
opened, err := infrastructure.productionInfrastructure.Open(ctx, configuration)
|
|
opened.providerResults = infrastructure.results
|
|
return opened, err
|
|
}
|
|
|
|
type providerResultRecorder func(provider.Result)
|
|
|
|
func (record providerResultRecorder) Record(result provider.Result) { record(result) }
|
|
|
|
type integrationRunner struct {
|
|
run func(context.Context) error
|
|
}
|
|
|
|
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,
|
|
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, options)
|
|
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
|
|
}
|