409 lines
13 KiB
Go
409 lines
13 KiB
Go
package bootstrap
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"proxy-pool/internal/config"
|
|
"proxy-pool/internal/controller/admin"
|
|
"proxy-pool/internal/domain/adminstate"
|
|
"proxy-pool/internal/platform/lifecycle"
|
|
)
|
|
|
|
func TestProviderSupervisorAppliesDisableAndConfigurationReplacement(t *testing.T) {
|
|
configuration, err := config.Load(strings.NewReader(bootstrapTestConfig))
|
|
if err != nil {
|
|
t.Fatalf("config.Load(): %v", err)
|
|
}
|
|
store, err := config.NewStore(configuration)
|
|
if err != nil {
|
|
t.Fatalf("config.NewStore(): %v", err)
|
|
}
|
|
state := &mutableProviderState{enabled: map[string]bool{"provider-a": true, "provider-b": true}}
|
|
started := make(chan providerRuntimeEvent, 4)
|
|
stopped := make(chan providerRuntimeEvent, 4)
|
|
supervisor, err := newProviderSupervisor(store, state, func(name string, upstream config.Upstream) (lifecycle.Runner, error) {
|
|
event := providerRuntimeEvent{name: name, url: upstream.API.URL}
|
|
return supervisorRunnerFunc(func(ctx context.Context) error {
|
|
started <- event
|
|
<-ctx.Done()
|
|
stopped <- event
|
|
return ctx.Err()
|
|
}), nil
|
|
}, nil, nil, nil, bootstrapTestFingerprintKey, time.Hour)
|
|
if err != nil {
|
|
t.Fatalf("newProviderSupervisor(): %v", err)
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- supervisor.Run(ctx) }()
|
|
waitForRuntimeEvents(t, started, 2)
|
|
|
|
state.set("provider-a", false)
|
|
supervisor.Notify()
|
|
if event := waitForRuntimeEvent(t, stopped); event.name != "provider-a" {
|
|
t.Fatalf("stopped Provider = %s, want provider-a", event.name)
|
|
}
|
|
|
|
updated := store.Current()
|
|
providerB := updated.Upstreams["provider-b"]
|
|
providerB.API.URL = "https://replacement.invalid/proxies"
|
|
updated.Upstreams["provider-b"] = providerB
|
|
if !store.PublishRevision(updated, 1) {
|
|
t.Fatal("PublishRevision() rejected updated configuration")
|
|
}
|
|
supervisor.Notify()
|
|
if event := waitForRuntimeEvent(t, stopped); event.name != "provider-b" {
|
|
t.Fatalf("replaced Provider = %s, want provider-b", event.name)
|
|
}
|
|
if event := waitForRuntimeEvent(t, started); event.name != "provider-b" || event.url != providerB.API.URL {
|
|
t.Fatalf("replacement Provider = %+v", event)
|
|
}
|
|
|
|
cancel()
|
|
if err := <-done; !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("Run() error = %v, want context cancellation", err)
|
|
}
|
|
}
|
|
|
|
func TestProviderSupervisorKeepsStaticallyDisabledUpstreamStopped(t *testing.T) {
|
|
configuration, err := config.Load(strings.NewReader(bootstrapTestConfig))
|
|
if err != nil {
|
|
t.Fatalf("config.Load(): %v", err)
|
|
}
|
|
disabled := configuration.Upstreams["provider-b"]
|
|
disabled.Enabled = false
|
|
configuration.Upstreams["provider-b"] = disabled
|
|
store, err := config.NewStore(configuration)
|
|
if err != nil {
|
|
t.Fatalf("config.NewStore(): %v", err)
|
|
}
|
|
supervisor, err := newProviderSupervisor(
|
|
store,
|
|
&mutableProviderState{enabled: map[string]bool{"provider-a": true, "provider-b": true}},
|
|
func(string, config.Upstream) (lifecycle.Runner, error) {
|
|
return supervisorRunnerFunc(func(context.Context) error { return nil }), nil
|
|
},
|
|
nil, nil, nil, bootstrapTestFingerprintKey, time.Hour,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("newProviderSupervisor() error = %v", err)
|
|
}
|
|
desired, err := supervisor.desired(context.Background())
|
|
if err != nil || len(desired) != 1 {
|
|
t.Fatalf("desired() = %+v, %v", desired, err)
|
|
}
|
|
if _, exists := desired["provider-a"]; !exists {
|
|
t.Fatalf("desired() = %+v, provider-a missing", desired)
|
|
}
|
|
}
|
|
|
|
func TestProviderSupervisorPropagatesUnexpectedRuntimeFailure(t *testing.T) {
|
|
configuration, err := config.Load(strings.NewReader(bootstrapTestConfig))
|
|
if err != nil {
|
|
t.Fatalf("config.Load(): %v", err)
|
|
}
|
|
store, err := config.NewStore(configuration)
|
|
if err != nil {
|
|
t.Fatalf("config.NewStore(): %v", err)
|
|
}
|
|
wantErr := errors.New("coordination stopped")
|
|
supervisor, err := newProviderSupervisor(store, nil, func(string, config.Upstream) (lifecycle.Runner, error) {
|
|
return supervisorRunnerFunc(func(context.Context) error { return wantErr }), nil
|
|
}, nil, nil, nil, bootstrapTestFingerprintKey, time.Hour)
|
|
if err != nil {
|
|
t.Fatalf("newProviderSupervisor(): %v", err)
|
|
}
|
|
if err := supervisor.Run(context.Background()); !errors.Is(err, wantErr) {
|
|
t.Fatalf("Run() error = %v, want %v", err, wantErr)
|
|
}
|
|
}
|
|
|
|
func TestProviderSupervisorRetainsRuntimesWhileManagementStateIsUnavailable(t *testing.T) {
|
|
configuration, err := config.Load(strings.NewReader(bootstrapTestConfig))
|
|
if err != nil {
|
|
t.Fatalf("config.Load(): %v", err)
|
|
}
|
|
store, err := config.NewStore(configuration)
|
|
if err != nil {
|
|
t.Fatalf("config.NewStore(): %v", err)
|
|
}
|
|
state := &mutableProviderState{enabled: map[string]bool{"provider-a": true, "provider-b": true}}
|
|
started := make(chan providerRuntimeEvent, 2)
|
|
stopped := make(chan providerRuntimeEvent, 2)
|
|
supervisor, err := newProviderSupervisor(store, state, func(name string, upstream config.Upstream) (lifecycle.Runner, error) {
|
|
event := providerRuntimeEvent{name: name, url: upstream.API.URL}
|
|
return supervisorRunnerFunc(func(ctx context.Context) error {
|
|
started <- event
|
|
<-ctx.Done()
|
|
stopped <- event
|
|
return ctx.Err()
|
|
}), nil
|
|
}, nil, nil, nil, bootstrapTestFingerprintKey, time.Hour)
|
|
if err != nil {
|
|
t.Fatalf("newProviderSupervisor(): %v", err)
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- supervisor.Run(ctx) }()
|
|
waitForRuntimeEvents(t, started, 2)
|
|
|
|
state.setError(errors.New("PostgreSQL temporarily unavailable"))
|
|
supervisor.Notify()
|
|
select {
|
|
case err := <-done:
|
|
t.Fatalf("Supervisor stopped during transient state failure: %v", err)
|
|
case event := <-stopped:
|
|
t.Fatalf("Provider stopped during transient state failure: %+v", event)
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
|
|
state.setError(nil)
|
|
state.set("provider-a", false)
|
|
supervisor.Notify()
|
|
if event := waitForRuntimeEvent(t, stopped); event.name != "provider-a" {
|
|
t.Fatalf("stopped Provider = %s, want provider-a", event.name)
|
|
}
|
|
|
|
cancel()
|
|
if err := <-done; !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("Run() error = %v, want context cancellation", err)
|
|
}
|
|
}
|
|
|
|
func TestProviderSupervisorStopsStaleRuntimesAndLoadsAuthoritativeConfiguration(t *testing.T) {
|
|
initial, err := config.Load(strings.NewReader(bootstrapTestConfig))
|
|
if err != nil {
|
|
t.Fatalf("config.Load(): %v", err)
|
|
}
|
|
store, err := config.NewStore(initial)
|
|
if err != nil {
|
|
t.Fatalf("config.NewStore(): %v", err)
|
|
}
|
|
initialChecksum, err := config.Fingerprint(initial, bootstrapTestFingerprintKey)
|
|
if err != nil {
|
|
t.Fatalf("config.Fingerprint(initial): %v", err)
|
|
}
|
|
state := &mutableProviderState{
|
|
enabled: map[string]bool{"provider-a": true, "provider-b": true},
|
|
checksum: initialChecksum,
|
|
revision: 1,
|
|
}
|
|
source := &mutableProviderConfigurationSource{configuration: initial}
|
|
started := make(chan providerRuntimeEvent, 4)
|
|
stopped := make(chan providerRuntimeEvent, 4)
|
|
supervisor, err := newProviderSupervisor(store, state, func(name string, upstream config.Upstream) (lifecycle.Runner, error) {
|
|
event := providerRuntimeEvent{name: name, url: upstream.API.URL}
|
|
return supervisorRunnerFunc(func(ctx context.Context) error {
|
|
started <- event
|
|
<-ctx.Done()
|
|
stopped <- event
|
|
return ctx.Err()
|
|
}), nil
|
|
}, nil, nil, source, bootstrapTestFingerprintKey, time.Hour)
|
|
if err != nil {
|
|
t.Fatalf("newProviderSupervisor(): %v", err)
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- supervisor.Run(ctx) }()
|
|
waitForRuntimeEvents(t, started, 2)
|
|
|
|
updated := store.Current()
|
|
providerA := updated.Upstreams["provider-a"]
|
|
providerA.API.URL = "https://replacement.invalid/proxies"
|
|
updated.Upstreams["provider-a"] = providerA
|
|
updatedChecksum, err := config.Fingerprint(updated, bootstrapTestFingerprintKey)
|
|
if err != nil {
|
|
t.Fatalf("config.Fingerprint(updated): %v", err)
|
|
}
|
|
state.setChecksum(updatedChecksum)
|
|
supervisor.Notify()
|
|
waitForRuntimeEvents(t, stopped, 2)
|
|
select {
|
|
case err := <-done:
|
|
t.Fatalf("Supervisor stopped for stale local configuration: %v", err)
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
|
|
source.set(updated)
|
|
supervisor.Notify()
|
|
events := []providerRuntimeEvent{waitForRuntimeEvent(t, started), waitForRuntimeEvent(t, started)}
|
|
foundReplacement := false
|
|
for _, event := range events {
|
|
if event.name == "provider-a" && event.url == providerA.API.URL {
|
|
foundReplacement = true
|
|
}
|
|
}
|
|
if !foundReplacement {
|
|
t.Fatalf("started Provider runtimes = %+v, replacement missing", events)
|
|
}
|
|
if currentChecksum, err := config.Fingerprint(store.Current(), bootstrapTestFingerprintKey); err != nil || currentChecksum != updatedChecksum {
|
|
t.Fatalf("published checksum = %q, %v; want %q", currentChecksum, err, updatedChecksum)
|
|
}
|
|
|
|
cancel()
|
|
if err := <-done; !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("Run() error = %v, want context cancellation", err)
|
|
}
|
|
}
|
|
|
|
func TestProviderSupervisorDoesNotPublishConfigurationOlderThanLocalRevision(t *testing.T) {
|
|
initial, err := config.Load(strings.NewReader(bootstrapTestConfig))
|
|
if err != nil {
|
|
t.Fatalf("config.Load(): %v", err)
|
|
}
|
|
store, err := config.NewStore(initial)
|
|
if err != nil {
|
|
t.Fatalf("config.NewStore(): %v", err)
|
|
}
|
|
store.PublishRevision(initial, 1)
|
|
|
|
candidate := store.Current()
|
|
providerA := candidate.Upstreams["provider-a"]
|
|
providerA.API.URL = "https://candidate.invalid/proxies"
|
|
candidate.Upstreams["provider-a"] = providerA
|
|
candidateChecksum, err := config.Fingerprint(candidate, bootstrapTestFingerprintKey)
|
|
if err != nil {
|
|
t.Fatalf("config.Fingerprint(candidate): %v", err)
|
|
}
|
|
newest := store.Current()
|
|
providerA = newest.Upstreams["provider-a"]
|
|
providerA.API.URL = "https://newest.invalid/proxies"
|
|
newest.Upstreams["provider-a"] = providerA
|
|
|
|
source := providerConfigurationSourceFunc(func(context.Context) (admin.LoadedConfiguration, error) {
|
|
if !store.PublishRevision(newest, 3) {
|
|
t.Fatal("failed to publish simulated concurrent revision")
|
|
}
|
|
return admin.LoadedConfiguration{Value: candidate, Source: "controller.yaml"}, nil
|
|
})
|
|
supervisor, err := newProviderSupervisor(
|
|
store,
|
|
&mutableProviderState{},
|
|
func(string, config.Upstream) (lifecycle.Runner, error) {
|
|
return supervisorRunnerFunc(func(context.Context) error { return nil }), nil
|
|
},
|
|
nil,
|
|
nil,
|
|
source,
|
|
bootstrapTestFingerprintKey,
|
|
time.Hour,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("newProviderSupervisor(): %v", err)
|
|
}
|
|
|
|
_, err = supervisor.synchronizeConfiguration(context.Background(), initial, adminstate.Snapshot{
|
|
Config: &adminstate.ConfigRevision{Revision: 2, Checksum: candidateChecksum},
|
|
})
|
|
if !errors.Is(err, errProviderManagementSnapshotStale) {
|
|
t.Fatalf("synchronizeConfiguration() error = %v, want stale snapshot", err)
|
|
}
|
|
if got := store.Revision(); got != 3 {
|
|
t.Fatalf("configuration revision = %d, want 3", got)
|
|
}
|
|
}
|
|
|
|
type providerRuntimeEvent struct {
|
|
name string
|
|
url string
|
|
}
|
|
|
|
type supervisorRunnerFunc func(context.Context) error
|
|
|
|
func (run supervisorRunnerFunc) Run(ctx context.Context) error { return run(ctx) }
|
|
|
|
type mutableProviderState struct {
|
|
mu sync.Mutex
|
|
enabled map[string]bool
|
|
err error
|
|
checksum string
|
|
revision uint64
|
|
}
|
|
|
|
func (state *mutableProviderState) Snapshot(context.Context) (adminstate.Snapshot, error) {
|
|
state.mu.Lock()
|
|
defer state.mu.Unlock()
|
|
if state.err != nil {
|
|
return adminstate.Snapshot{}, state.err
|
|
}
|
|
snapshot := adminstate.Snapshot{Upstreams: make([]adminstate.UpstreamState, 0, len(state.enabled))}
|
|
if state.checksum != "" {
|
|
snapshot.Config = &adminstate.ConfigRevision{
|
|
Revision: state.revision, ConfigVersion: "cfg-" + state.checksum, Checksum: state.checksum,
|
|
}
|
|
}
|
|
for name, enabled := range state.enabled {
|
|
snapshot.Upstreams = append(snapshot.Upstreams, adminstate.UpstreamState{Name: name, Enabled: enabled})
|
|
}
|
|
return snapshot, nil
|
|
}
|
|
|
|
func (state *mutableProviderState) setError(err error) {
|
|
state.mu.Lock()
|
|
defer state.mu.Unlock()
|
|
state.err = err
|
|
}
|
|
|
|
func (state *mutableProviderState) setChecksum(checksum string) {
|
|
state.mu.Lock()
|
|
defer state.mu.Unlock()
|
|
state.checksum = checksum
|
|
state.revision++
|
|
}
|
|
|
|
type mutableProviderConfigurationSource struct {
|
|
mu sync.Mutex
|
|
configuration *config.Config
|
|
}
|
|
|
|
type providerConfigurationSourceFunc func(context.Context) (admin.LoadedConfiguration, error)
|
|
|
|
func (source providerConfigurationSourceFunc) LoadConfiguration(ctx context.Context) (admin.LoadedConfiguration, error) {
|
|
return source(ctx)
|
|
}
|
|
|
|
func (source *mutableProviderConfigurationSource) LoadConfiguration(context.Context) (admin.LoadedConfiguration, error) {
|
|
source.mu.Lock()
|
|
defer source.mu.Unlock()
|
|
return admin.LoadedConfiguration{Value: source.configuration, Source: "controller.yaml"}, nil
|
|
}
|
|
|
|
func (source *mutableProviderConfigurationSource) set(configuration *config.Config) {
|
|
source.mu.Lock()
|
|
defer source.mu.Unlock()
|
|
source.configuration = configuration
|
|
}
|
|
|
|
func (state *mutableProviderState) set(name string, enabled bool) {
|
|
state.mu.Lock()
|
|
defer state.mu.Unlock()
|
|
state.enabled[name] = enabled
|
|
}
|
|
|
|
func waitForRuntimeEvents(t *testing.T, events <-chan providerRuntimeEvent, count int) {
|
|
t.Helper()
|
|
for range count {
|
|
_ = waitForRuntimeEvent(t, events)
|
|
}
|
|
}
|
|
|
|
func waitForRuntimeEvent(t *testing.T, events <-chan providerRuntimeEvent) providerRuntimeEvent {
|
|
t.Helper()
|
|
select {
|
|
case event := <-events:
|
|
return event
|
|
case <-time.After(time.Second):
|
|
t.Fatal("timed out waiting for Provider runtime event")
|
|
return providerRuntimeEvent{}
|
|
}
|
|
}
|