From 5f53767e905a0b08911c5cecd2b75b6fe9ac734f Mon Sep 17 00:00:00 2001 From: Hongkai Liu Date: Tue, 9 Jun 2026 08:26:30 -0400 Subject: [PATCH] Add thread-safe AddSchemes() to fix concurrent map writes Fixes "fatal error: concurrent map writes" in e2e-agnostic-operator [1, 2]. The issue occurred because multiple tests (e.g., controller_test.go and start_integration_test.go) calling addSchemes() would modified the global scheme.Scheme map without synchronization. Changes: - Created pkg/internal/schema.go with AddSchemes() using sync.Once - Moved addSchemes() from pkg/start/start.go to avoid import cycles (proposal <- cvo <- start creates cycle when proposal imports start) - Updated pkg/start/start.go to call internal.AddSchemes() - Updated pkg/proposal/controller_test.go to call internal.AddSchemes() - Added pkg/internal/schema_test.go with concurrent safety test The sync.Once pattern ensures scheme registration executes exactly once across all concurrent callers, preventing race conditions. [1]. https://prow.ci.openshift.org/view/gs/test-platform-results/pr-logs/pull/openshift_cluster-version-operator/1386/pull-ci-openshift-cluster-version-operator-main-e2e-agnostic-operator/2063953653860929536 [2]. https://prow.ci.openshift.org/view/gs/test-platform-results/pr-logs/pull/openshift_cluster-version-operator/1399/pull-ci-openshift-cluster-version-operator-main-e2e-agnostic-operator/2064194902241054720 Co-Authored-By: Claude Sonnet 4.5 --- pkg/internal/schema.go | 26 ++++++++++++++++++++++++ pkg/internal/schema_test.go | 35 +++++++++++++++++++++++++++++++++ pkg/proposal/controller_test.go | 6 +++--- pkg/start/start.go | 10 +--------- 4 files changed, 65 insertions(+), 12 deletions(-) create mode 100644 pkg/internal/schema.go create mode 100644 pkg/internal/schema_test.go diff --git a/pkg/internal/schema.go b/pkg/internal/schema.go new file mode 100644 index 000000000..0c84fd5f0 --- /dev/null +++ b/pkg/internal/schema.go @@ -0,0 +1,26 @@ +package internal + +import ( + "fmt" + "sync" + + "k8s.io/client-go/kubernetes/scheme" + + proposalv1alpha1 "github.com/openshift/lightspeed-agentic-operator/api/v1alpha1" +) + +var ( + addSchemesOnce sync.Once + addSchemesErr error +) + +// AddSchemes registers the proposalv1alpha1 scheme with the global Kubernetes scheme. +// This function is safe to call concurrently and will only execute once. +func AddSchemes() error { + addSchemesOnce.Do(func() { + if err := proposalv1alpha1.AddToScheme(scheme.Scheme); err != nil { + addSchemesErr = fmt.Errorf("failed to add proposalv1alpha1 to scheme: %w", err) + } + }) + return addSchemesErr +} diff --git a/pkg/internal/schema_test.go b/pkg/internal/schema_test.go new file mode 100644 index 000000000..7d59ecada --- /dev/null +++ b/pkg/internal/schema_test.go @@ -0,0 +1,35 @@ +package internal + +import ( + "sync" + "testing" +) + +// TestAddSchemesConcurrent verifies that AddSchemes can be called concurrently +// without causing race conditions or panics. This is a regression test for +// "fatal error: concurrent map writes" when integration tests run in parallel. +func TestAddSchemesConcurrent(t *testing.T) { + const numGoroutines = 10 + + var wg sync.WaitGroup + errChan := make(chan error, numGoroutines) + + // Call AddSchemes concurrently from multiple goroutines + for i := 0; i < numGoroutines; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if err := AddSchemes(); err != nil { + errChan <- err + } + }() + } + + wg.Wait() + close(errChan) + + // Check if any goroutine encountered an error + for err := range errChan { + t.Errorf("AddSchemes failed: %v", err) + } +} diff --git a/pkg/proposal/controller_test.go b/pkg/proposal/controller_test.go index 9c9d58563..c60f8758b 100644 --- a/pkg/proposal/controller_test.go +++ b/pkg/proposal/controller_test.go @@ -17,15 +17,15 @@ import ( kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" kutilerrors "k8s.io/apimachinery/pkg/util/errors" - "k8s.io/client-go/kubernetes/scheme" configv1 "github.com/openshift/api/config/v1" proposalv1alpha1 "github.com/openshift/lightspeed-agentic-operator/api/v1alpha1" + + "github.com/openshift/cluster-version-operator/pkg/internal" ) func init() { - err := proposalv1alpha1.AddToScheme(scheme.Scheme) - if err != nil { + if err := internal.AddSchemes(); err != nil { panic(err) } } diff --git a/pkg/start/start.go b/pkg/start/start.go index b1a6291e3..62c6989db 100644 --- a/pkg/start/start.go +++ b/pkg/start/start.go @@ -40,7 +40,6 @@ import ( operatorinformers "github.com/openshift/client-go/operator/informers/externalversions" "github.com/openshift/library-go/pkg/config/clusterstatus" libgoleaderelection "github.com/openshift/library-go/pkg/config/leaderelection" - proposalv1alpha1 "github.com/openshift/lightspeed-agentic-operator/api/v1alpha1" "github.com/openshift/cluster-version-operator/pkg/autoupdate" "github.com/openshift/cluster-version-operator/pkg/clusterconditions" @@ -615,13 +614,6 @@ type Context struct { OperatorInformerFactory operatorinformers.SharedInformerFactory } -func addSchemes() error { - if err := proposalv1alpha1.AddToScheme(scheme.Scheme); err != nil { - return fmt.Errorf("failed to add proposalv1alpha1 to scheme: %w", err) - } - return nil -} - // NewControllerContext initializes the default Context for the current Options. It does // not start any background processes. func (o *Options) NewControllerContext( @@ -647,7 +639,7 @@ func (o *Options) NewControllerContext( cvoKubeClient := cb.KubeClientOrDie(o.Namespace, useProtobuf) o.PromQLTarget.KubeClient = cvoKubeClient - if err := addSchemes(); err != nil { + if err := internal.AddSchemes(); err != nil { return nil, err } rtClient := cb.RuntimeControllerClientOrDie("runtime-controller-client")