Make cluster reconcilers able to sync using a shared mutex

pkg/config/types.go

@@ -232,9 +232,14 @@ func ForCluster(ctx context.Context, n uint16, cfg *Config, fn func(context.Cont
 	wg := &sync.WaitGroup{}
 	wg.Add(int(n))
 	foundErr := false
+
+	// mutex shared by cluster threads when they need to coordinate
+	// TODO: This is limited to only one lock operation, consider supporting more in the future
+	mux := &sync.Mutex{}
 	for i := ClusterNumber(1); i <= ClusterNumber(n); i++ {
 		go func(j ClusterNumber) {
 			clusterCtx := util.WithClusterNumber(ctx, uint16(j))
+			clusterCtx = util.WithMutex(clusterCtx, mux)
 			logger := util.Logger(clusterCtx)
 			logger.Tracef("ForCluster goroutine starting...")
 			clusterInfo := NewClusterInfo(cfg, j)

pkg/util/context.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"sync"
 
 	"github.com/sirupsen/logrus"
 )
@@ -81,6 +82,23 @@ func JoinPaths(ctx context.Context, filePaths ...string) string {
 	return filepath.Join(filePaths...)
 }
 
+var muxKey = muxKeyImpl{}
+
+type muxKeyImpl struct{}
+
+func WithMutex(ctx context.Context, mux *sync.Mutex) context.Context {
+	return context.WithValue(ctx, muxKey, mux)
+}
+
+func GetMutex(ctx context.Context) (*sync.Mutex, bool) {
+	mux, ok := ctx.Value(muxKey).(*sync.Mutex)
+	if !ok {
+		logrus.Debug("Didn't find mux from context, defaulting to nil")
+		return nil, false
+	}
+	return mux, ok
+}
+
 func Logger(ctx context.Context) *logrus.Entry {
 	logger := logrus.WithContext(ctx)
 	// If cluster number is set, add that logging field