fix: data races

.github/workflows/tests.yaml

@@ -45,5 +45,6 @@ jobs:
           export GOOS=linux
 
           go test \
+            --race \
             -tags test \
             ./cmd/... ./pkg/... ./test/utils

pkg/hook/controller/kubernetes_bindings_controller.go

@@ -56,7 +56,7 @@ type kubernetesBindingsController struct {
 }
 
 // kubernetesHooksController should implement the KubernetesHooksController
-var _ KubernetesBindingsController = &kubernetesBindingsController{}
+var _ KubernetesBindingsController = (*kubernetesBindingsController)(nil)
 
 // NewKubernetesBindingsController returns an implementation of KubernetesBindingsController
 var NewKubernetesBindingsController = func(logger *log.Logger) *kubernetesBindingsController {

pkg/hook/controller/schedule_bindings_controller.go

@@ -8,7 +8,7 @@ import (
 	schedulemanager "github.com/flant/shell-operator/pkg/schedule_manager"
 )
 
-// A link between a hook and a kube monitor
+// ScheduleBindingToCrontabLink a link between a hook and a kube monitor
 type ScheduleBindingToCrontabLink struct {
 	BindingName string
 	Crontab     string
@@ -29,7 +29,7 @@ type ScheduleBindingsController interface {
 	HandleEvent(crontab string) []BindingExecutionInfo
 }
 
-// scheduleHooksController is a main implementation of KubernetesHooksController
+// scheduleBindingsController is a main implementation of KubernetesHooksController
 type scheduleBindingsController struct {
 	// dependencies
 	scheduleManager schedulemanager.ScheduleManager
@@ -43,7 +43,7 @@ type scheduleBindingsController struct {
 }
 
 // kubernetesHooksController should implement the KubernetesHooksController
-var _ ScheduleBindingsController = &scheduleBindingsController{}
+var _ ScheduleBindingsController = (*scheduleBindingsController)(nil)
 
 // NewScheduleBindingsController returns an implementation of ScheduleBindingsController
 var NewScheduleBindingsController = func() *scheduleBindingsController {

pkg/kube_events_manager/factory.go

@@ -49,6 +49,12 @@ func NewFactoryStore() *FactoryStore {
 	}
 }
 
+func (c *FactoryStore) Reset() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.data = make(map[FactoryIndex]Factory)
+}
+
 func (c *FactoryStore) add(index FactoryIndex, f dynamicinformer.DynamicSharedInformerFactory) {
 	ctx, cancel := context.WithCancel(context.Background())
 	c.data[index] = Factory{

pkg/kube_events_manager/monitor.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log/slog"
 	"sort"
+	"sync"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
 
@@ -35,14 +36,14 @@ type monitor struct {
 	// Namespace informer to get new namespaces
 	NamespaceInformer *namespaceInformer
 	// map of dynamically starting informers
-	VaryingInformers map[string][]*resourceInformer
+	VaryingInformers varyingInformers
 
 	eventCb       func(kemtypes.KubeEvent)
 	eventsEnabled bool
 	// Index of namespaces statically defined in monitor configuration
-	staticNamespaces map[string]bool
+	staticNamespaces sync.Map
 
-	cancelForNs map[string]context.CancelFunc
+	cancelForNs cancelForNs
 
 	ctx           context.Context
 	cancel        context.CancelFunc
@@ -51,6 +52,85 @@ type monitor struct {
 	logger *log.Logger
 }
 
+type varyingInformers struct {
+	value sync.Map
+}
+
+func (v *varyingInformers) Store(key string, value []*resourceInformer) {
+	v.value.Store(key, value)
+}
+
+func (v *varyingInformers) Load(key string) ([]*resourceInformer, bool) {
+	value, ok := v.value.Load(key)
+	if !ok {
+		return nil, false
+	}
+	if value, ok := value.([]*resourceInformer); ok {
+		return value, true
+	}
+	return nil, false
+}
+
+func (v *varyingInformers) Range(f func(key string, value []*resourceInformer) bool) {
+	v.value.Range(func(key, value any) bool {
+		if key, ok := key.(string); ok {
+			if value, ok := value.([]*resourceInformer); ok {
+				return f(key, value)
+			}
+		}
+		return true
+	})
+}
+
+func (v *varyingInformers) RangeValue(f func(value []*resourceInformer)) {
+	v.value.Range(func(_, value any) bool {
+		if value, ok := value.([]*resourceInformer); ok {
+			f(value)
+		}
+		return true
+	})
+}
+
+func (v *varyingInformers) Delete(key string) {
+	v.value.Delete(key)
+}
+
+type cancelForNs struct {
+	value sync.Map
+}
+
+func (c *cancelForNs) Store(key string, value context.CancelFunc) {
+	c.value.Store(key, value)
+}
+
+func (c *cancelForNs) Load(key string) (context.CancelFunc, bool) {
+	value, ok := c.value.Load(key)
+	if !ok {
+		return nil, false
+	}
+	if value, ok := value.(context.CancelFunc); ok {
+		return value, true
+	}
+	return nil, false
+}
+
+func (c *cancelForNs) Range(f func(key string, value context.CancelFunc) bool) {
+	c.value.Range(func(key, value any) bool {
+		if key, ok := key.(string); ok {
+			if value, ok := value.(context.CancelFunc); ok {
+				return f(key, value)
+			}
+		}
+		return true
+	})
+}
+
+func (c *cancelForNs) Delete(key string) {
+	c.value.Delete(key)
+}
+
+var _ Monitor = (*monitor)(nil)
+
 func NewMonitor(ctx context.Context, client *klient.Client, mstor metric.Storage, config *MonitorConfig, eventCb func(kemtypes.KubeEvent), logger *log.Logger) *monitor {
 	cctx, cancel := context.WithCancel(ctx)
 
@@ -62,9 +142,9 @@ func NewMonitor(ctx context.Context, client *klient.Client, mstor metric.Storage
 		Config:            config,
 		eventCb:           eventCb,
 		ResourceInformers: make([]*resourceInformer, 0),
-		VaryingInformers:  make(map[string][]*resourceInformer),
-		cancelForNs:       make(map[string]context.CancelFunc),
-		staticNamespaces:  make(map[string]bool),
+		VaryingInformers:  varyingInformers{value: sync.Map{}},
+		cancelForNs:       cancelForNs{value: sync.Map{}},
+		staticNamespaces:  sync.Map{},
 		logger:            logger,
 	}
 }
@@ -98,7 +178,7 @@ func (m *monitor) CreateInformers() error {
 		// This list of informers is static.
 		for _, nsName := range nsNames {
 			if nsName != "" {
-				m.staticNamespaces[nsName] = true
+				m.staticNamespaces.Store(nsName, true)
 			}
 			informers, err := m.CreateInformersForNamespace(nsName)
 			if err != nil {
@@ -115,29 +195,28 @@ func (m *monitor) CreateInformers() error {
 			func(nsName string) {
 				// Added/Modified event: check, create and run informers for Ns
 				// ignore event if namespace is already has static ResourceInformers
-				if _, ok := m.staticNamespaces[nsName]; ok {
+				if _, ok := m.staticNamespaces.Load(nsName); ok {
 					return
 				}
 				// ignore already started informers
-				_, ok := m.VaryingInformers[nsName]
-				if ok {
+				if _, ok := m.VaryingInformers.Load(nsName); ok {
 					return
 				}
 
 				logEntry.Info("got ns, create dynamic ResourceInformers", slog.String("name", nsName))
 
-				var err error
-				m.VaryingInformers[nsName], err = m.CreateInformersForNamespace(nsName)
+				varyingInformers, err := m.CreateInformersForNamespace(nsName)
 				if err != nil {
 					logEntry.Error("create ResourceInformers for ns",
 						slog.String("name", nsName),
 						log.Err(err))
 				}
+				m.VaryingInformers.Store(nsName, varyingInformers)
 
-				var ctx context.Context
-				ctx, m.cancelForNs[nsName] = context.WithCancel(m.ctx)
+				ctx, cancelForNs := context.WithCancel(m.ctx)
+				m.cancelForNs.Store(nsName, cancelForNs)
 
-				for _, informer := range m.VaryingInformers[nsName] {
+				for _, informer := range varyingInformers {
 					informer.withContext(ctx)
 					if m.eventsEnabled {
 						informer.enableKubeEventCb()
@@ -150,22 +229,23 @@ func (m *monitor) CreateInformers() error {
 				logEntry.Info("deleted ns, stop dynamic ResourceInformers", slog.String("name", nsName))
 
 				// ignore statically specified namespaces
-				if _, ok := m.staticNamespaces[nsName]; ok {
+				if _, ok := m.staticNamespaces.Load(nsName); ok {
 					return
 				}
 
 				// ignore already stopped informers
-				_, ok := m.cancelForNs[nsName]
-				if !ok {
+				if _, ok := m.cancelForNs.Load(nsName); !ok {
 					return
 				}
 
-				m.cancelForNs[nsName]()
+				if fn, ok := m.cancelForNs.Load(nsName); ok {
+					fn()
+				}
 
 				// TODO wait
 
-				delete(m.VaryingInformers, nsName)
-				delete(m.cancelForNs, nsName)
+				m.VaryingInformers.Delete(nsName)
+				m.cancelForNs.Delete(nsName)
 			},
 		)
 		if err != nil {
@@ -175,17 +255,17 @@ func (m *monitor) CreateInformers() error {
 			logEntry.Info("got ns, create dynamic ResourceInformers", slog.String("name", nsName))
 
 			// ignore event if namespace is already has static ResourceInformers
-			if _, ok := m.staticNamespaces[nsName]; ok {
+			if _, ok := m.staticNamespaces.Load(nsName); ok {
 				continue
 			}
 
-			var err error
-			m.VaryingInformers[nsName], err = m.CreateInformersForNamespace(nsName)
+			varyingInformers, err := m.CreateInformersForNamespace(nsName)
 			if err != nil {
 				logEntry.Error("create ResourceInformers for ns",
 					slog.String("name", nsName),
 					log.Err(err))
 			}
+			m.VaryingInformers.Store(nsName, varyingInformers)
 		}
 	}
 
@@ -200,11 +280,11 @@ func (m *monitor) Snapshot() []kemtypes.ObjectAndFilterResult {
 		objects = append(objects, informer.getCachedObjects()...)
 	}
 
-	for nsName := range m.VaryingInformers {
-		for _, informer := range m.VaryingInformers[nsName] {
+	m.VaryingInformers.RangeValue(func(value []*resourceInformer) {
+		for _, informer := range value {
 			objects = append(objects, informer.getCachedObjects()...)
 		}
-	}
+	})
 
 	// Sort objects by namespace and name
 	sort.Sort(kemtypes.ByNamespaceAndName(objects))
@@ -218,11 +298,12 @@ func (m *monitor) EnableKubeEventCb() {
 	for _, informer := range m.ResourceInformers {
 		informer.enableKubeEventCb()
 	}
-	for nsName := range m.VaryingInformers {
-		for _, informer := range m.VaryingInformers[nsName] {
+	// Execute eventCb for events accumulated during "Synchronization" phase.
+	m.VaryingInformers.RangeValue(func(value []*resourceInformer) {
+		for _, informer := range value {
 			informer.enableKubeEventCb()
 		}
-	}
+	})
 	// Enable events for future VaryingInformers.
 	m.eventsEnabled = true
 }
@@ -269,14 +350,15 @@ func (m *monitor) Start(parentCtx context.Context) {
 		informer.start()
 	}
 
-	for nsName := range m.VaryingInformers {
-		var ctx context.Context
-		ctx, m.cancelForNs[nsName] = context.WithCancel(m.ctx)
-		for _, informer := range m.VaryingInformers[nsName] {
+	m.VaryingInformers.Range(func(nsName string, value []*resourceInformer) bool {
+		ctx, cancelForNs := context.WithCancel(m.ctx)
+		m.cancelForNs.Store(nsName, cancelForNs)
+		for _, informer := range value {
 			informer.withContext(ctx)
 			informer.start()
 		}
-	}
+		return true
+	})
 
 	if m.NamespaceInformer != nil {
 		m.NamespaceInformer.withContext(m.ctx)
@@ -299,11 +381,11 @@ func (m *monitor) PauseHandleEvents() {
 		informer.pauseHandleEvents()
 	}
 
-	for _, informers := range m.VaryingInformers {
-		for _, informer := range informers {
+	m.VaryingInformers.RangeValue(func(value []*resourceInformer) {
+		for _, informer := range value {
 			informer.pauseHandleEvents()
 		}
-	}
+	})
 
 	if m.NamespaceInformer != nil {
 		m.NamespaceInformer.pauseHandleEvents()
@@ -319,12 +401,12 @@ func (m *monitor) SnapshotOperations() (*CachedObjectsInfo /*total*/, *CachedObj
 		last.add(informer.getCachedObjectsInfoIncrement())
 	}
 
-	for nsName := range m.VaryingInformers {
-		for _, informer := range m.VaryingInformers[nsName] {
+	m.VaryingInformers.RangeValue(func(value []*resourceInformer) {
+		for _, informer := range value {
 			total.add(informer.getCachedObjectsInfo())
 			last.add(informer.getCachedObjectsInfoIncrement())
 		}
-	}
+	})
 
 	return total, last
 }