diff --git a/internal/nonkube/controller/collector_handler.go b/internal/nonkube/controller/collector_handler.go index 8d59249ae..5fa264dd3 100644 --- a/internal/nonkube/controller/collector_handler.go +++ b/internal/nonkube/controller/collector_handler.go @@ -25,9 +25,17 @@ type CollectorLifecycleHandler struct { logger *slog.Logger running bool mux sync.Mutex - startMethod func() + startMethod func(ctx context.Context) + // lastDone is closed once the most recently started generation has + // returned from startMethod. The next generation waits on it so that two + // collectors never overlap on the shared status ConfigMap. + lastDone <-chan struct{} } +// Start launches a collector generation unless one is already running. The +// new generation waits for the previous one to finish before it starts, so +// two collectors never share the status ConfigMap at the same time. Start +// itself returns immediately. Closing stopCh stops the running generation. func (c *CollectorLifecycleHandler) Start(stopCh <-chan struct{}) { c.mux.Lock() defer c.mux.Unlock() @@ -40,17 +48,34 @@ func (c *CollectorLifecycleHandler) Start(stopCh <-chan struct{}) { if c.startMethod == nil { c.startMethod = c.startCollectorLite } - go c.startMethod() - go c.handleShutdown(stopCh) + // Capture ctx by value while the lock is held. Goroutines must not read + // c.ctx through the struct pointer once the lock is dropped: a later + // Start() call writes a new c.ctx under the lock, which would race with + // a goroutine reading the old value via c.ctx.Done(). + ctx := c.ctx + start := c.startMethod + prev := c.lastDone + done := make(chan struct{}) + c.lastDone = done + go func() { + defer close(done) + if prev != nil { + <-prev + } + start(ctx) + }() + go c.handleShutdown(ctx, stopCh) } -func (c *CollectorLifecycleHandler) handleShutdown(stopCh <-chan struct{}) { +// handleShutdown stops the handler when stopCh closes, and returns without +// stopping anything when ctx is canceled first (for example by Stop). +func (c *CollectorLifecycleHandler) handleShutdown(ctx context.Context, stopCh <-chan struct{}) { const stopMsg = "Stopping collector lifecycle handler" select { case <-stopCh: c.logger.Info(stopMsg) c.Stop() - case <-c.ctx.Done(): + case <-ctx.Done(): c.logger.Info(stopMsg) } } @@ -72,9 +97,17 @@ func (c *CollectorLifecycleHandler) Id() string { return "collector.handler" } -func (c *CollectorLifecycleHandler) startCollectorLite() { +// startCollectorLite runs the collector for one generation and blocks until +// ctx is canceled. It returns immediately if ctx is already canceled, so a +// generation that was stopped before it was scheduled never touches the +// shared status ConfigMap. +func (c *CollectorLifecycleHandler) startCollectorLite(ctx context.Context) { + if err := ctx.Err(); err != nil { + c.logger.Info("Skipping stale collector startup", slog.Any("reason", err)) + return + } c.logger.Info("Starting collector lite") - err := flow.StartCollector(c.ctx, c.namespace) + err := flow.StartCollector(ctx, c.namespace) if err != nil { c.logger.Error("error starting collector", slog.Any("error", err.Error())) return diff --git a/internal/nonkube/controller/collector_handler_generation_test.go b/internal/nonkube/controller/collector_handler_generation_test.go new file mode 100644 index 000000000..73a5622dd --- /dev/null +++ b/internal/nonkube/controller/collector_handler_generation_test.go @@ -0,0 +1,174 @@ +package controller + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/skupperproject/skupper/pkg/nonkube/api" +) + +const generationTestNamespace = "collector-generations" + +// collectorDataHome is created once for the package. Collector goroutines can +// still be reading api.DefaultRootDataHome after a test returns, so it must +// not be rewritten per test. +var ( + collectorDataHomeOnce sync.Once + collectorDataHome string +) + +// setupCollectorFS points the nonkube filesystem helpers at a temp directory +// and creates the minimum inputs flow.StartCollector needs (platform file, +// RouterAccess and client certificates). It returns the path of the shared +// network status ConfigMap file written by every collector generation. No +// router is listening, so the status sync itself never gets a session. +func setupCollectorFS(t *testing.T) string { + t.Helper() + collectorDataHomeOnce.Do(func() { + dir, err := os.MkdirTemp("", "collector-generations") + if err != nil { + t.Fatal(err) + } + collectorDataHome = dir + // root uses DefaultRootDataHome, other users use XDG_DATA_HOME + api.DefaultRootDataHome = dir + os.Setenv("XDG_DATA_HOME", dir) + }) + nsDir := filepath.Join(api.GetDataHome(), "namespaces", generationTestNamespace) + if err := os.RemoveAll(nsDir); err != nil { + t.Fatal(err) + } + write := func(p, content string) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, []byte(content), 0o600); err != nil { + t.Fatal(err) + } + } + write(filepath.Join(nsDir, "internal", "platform.yaml"), "platform: podman\n") + write(filepath.Join(nsDir, string(api.RuntimeSiteStatePath), "RouterAccess-skupper-local.yaml"), + "apiVersion: skupper.io/v2alpha1\nkind: RouterAccess\nmetadata:\n name: skupper-local\nspec:\n roles:\n - name: normal\n port: 1\n") + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "collector-generations"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + BasicConstraintsValid: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + keyDER, err := x509.MarshalECPrivateKey(key) + if err != nil { + t.Fatal(err) + } + certPEM := string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})) + keyPEM := string(pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})) + certDir := filepath.Join(nsDir, string(api.CertificatesPath), "skupper-local-client") + write(filepath.Join(certDir, "ca.crt"), certPEM) + write(filepath.Join(certDir, "tls.crt"), certPEM) + write(filepath.Join(certDir, "tls.key"), keyPEM) + + return filepath.Join(nsDir, string(api.RuntimeSiteStatePath), "ConfigMap-skupper-network-status.yaml") +} + +// fileExists reports whether p exists. +func fileExists(p string) bool { + _, err := os.Stat(p) + return err == nil +} + +// waitFor polls cond until it returns true or timeout elapses, and reports +// whether cond became true. +func waitFor(cond func() bool, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return true + } + time.Sleep(time.Millisecond) + } + return cond() +} + +// lastGeneration returns the done channel of the most recently started +// generation. It is closed once that generation has returned from its start +// method, which for the real collector means its status ConfigMap is gone. +func lastGeneration(h *CollectorLifecycleHandler) <-chan struct{} { + h.mux.Lock() + defer h.mux.Unlock() + return h.lastDone +} + +// TestCollectorHandlerRestartKeepsStatus restarts the real collector back to +// back, both before and after the first generation has written its status +// ConfigMap. It waits for the first generation to finish, which removes its +// ConfigMap, and then requires the second generation to create its own and +// keep it until it is stopped. A stale generation's cleanup must never remove +// the current generation's ConfigMap. +func TestCollectorHandlerRestartKeepsStatus(t *testing.T) { + cmPath := setupCollectorFS(t) + const ( + rounds = 20 + timeout = 5 * time.Second + ) + for i := 0; i < rounds; i++ { + h := NewCollectorLifecycleHandler(generationTestNamespace) + stopCh := make(chan struct{}) + + h.Start(stopCh) + gen1Done := lastGeneration(h) + if i%2 == 0 { + // let generation 1 get as far as writing the ConfigMap + if !waitFor(func() bool { return fileExists(cmPath) }, timeout) { + t.Fatalf("round %d: gen1 never created %s", i, cmPath) + } + } + h.Stop() + h.Start(stopCh) // generation 2, immediately after Stop + gen2Done := lastGeneration(h) + + // Generation 1 removes its ConfigMap before it finishes, so after + // this wait the only file that can exist belongs to generation 2. + select { + case <-gen1Done: + case <-time.After(timeout): + t.Fatalf("round %d: gen1 did not finish after Stop", i) + } + if !waitFor(func() bool { return fileExists(cmPath) }, timeout) { + t.Fatalf("round %d: gen2 never created %s after gen1 finished", i, cmPath) + } + + h.Stop() + select { + case <-gen2Done: + case <-time.After(timeout): + t.Fatalf("round %d: gen2 did not finish after Stop", i) + } + if fileExists(cmPath) { + t.Fatalf("round %d: ConfigMap left behind after gen2 finished", i) + } + close(stopCh) + } +} diff --git a/internal/nonkube/controller/collector_handler_test.go b/internal/nonkube/controller/collector_handler_test.go index 4b8eecbdb..04be783bd 100644 --- a/internal/nonkube/controller/collector_handler_test.go +++ b/internal/nonkube/controller/collector_handler_test.go @@ -1,22 +1,25 @@ package controller import ( + "context" "testing" "time" "gotest.tools/v3/assert" ) +// TestCollectorHandler checks the start, stop, restart and parent-stop +// lifecycle of CollectorLifecycleHandler using a fake start method. func TestCollectorHandler(t *testing.T) { started := make(chan struct{}) done := make(chan struct{}) stopCh := make(chan struct{}) ch := NewCollectorLifecycleHandler("sample-namespace") - fakeStart := func() { + fakeStart := func(ctx context.Context) { close(started) go func() { - <-ch.ctx.Done() + <-ctx.Done() close(done) }() } diff --git a/internal/nonkube/flow/collector.go b/internal/nonkube/flow/collector.go index e134b1b8b..572eb9285 100644 --- a/internal/nonkube/flow/collector.go +++ b/internal/nonkube/flow/collector.go @@ -40,6 +40,9 @@ func (s *StatusSyncClient) Update(ctx context.Context, latest *corev1.ConfigMap) return s.client.Update(*latest, true) } +// siteCollector creates the network status ConfigMap, runs the status sync +// against the local router and blocks until ctx is canceled and the sync has +// stopped. The ConfigMap is removed on return, including when setup fails. func siteCollector(ctx context.Context, namespace string) error { platformLoader := &common.NamespacePlatformLoader{} platform, err := platformLoader.Load(namespace) @@ -65,6 +68,10 @@ func siteCollector(ctx context.Context, namespace string) error { if err := client.Add(cm, true); err != nil { return err } + // Remove the status ConfigMap only once this collector has fully stopped, + // including on setup errors below. The caller serializes generations on + // this function returning, so a later collector never sees this delete. + defer func() { _ = client.Delete(cm.Name, true) }() address, err := runtime.GetLocalRouterAddress(namespace) if err != nil { @@ -84,11 +91,13 @@ func siteCollector(ctx context.Context, namespace string) error { namespace: namespace, } statusSync := flow.NewStatusSync(factory, nil, statusSyncClient, types.NetworkStatusConfigMapName) - go statusSync.Run(ctx) + runDone := make(chan struct{}) go func() { - <-ctx.Done() - _ = client.Delete(cm.Name, true) + defer close(runDone) + statusSync.Run(ctx) }() + <-ctx.Done() + <-runDone return nil } @@ -101,6 +110,9 @@ func getLocalTLSConfig(namespace string) (*tls.Config, error) { return config, err } +// StartCollector runs the site collector for namespace. It blocks until ctx +// is canceled and the collector has stopped and removed its status +// ConfigMap, or until setup fails. func StartCollector(ctx context.Context, namespace string) error { slog.Info("COLLECTOR: Starting site collection", slog.String("namespace", namespace)) if err := siteCollector(ctx, namespace); err != nil {