concurrent map issue (#225)

admiral/pkg/controller/admiral/service.go

@@ -2,17 +2,19 @@ package admiral
 
 import (
 	"fmt"
+	"time"
+
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/common"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/rest"
-	"time"
+
+	"sync"
 
 	k8sV1 "k8s.io/api/core/v1"
 	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
-	"sync"
 )
 
 // Handler interface contains the methods that are required
@@ -69,6 +71,8 @@ func (s *serviceCache) getKey(service *k8sV1.Service) string {
 }
 
 func (s *serviceCache) Get(key string) *ServiceClusterEntry {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
 	return s.cache[key]
 }
 
@@ -146,7 +150,7 @@ func NewServiceController(stopCh <-chan struct{}, handler ServiceHandler, config
 		&k8sV1.Service{}, resyncPeriod, cache.Indexers{},
 	)
 
-	NewController("service-ctrl-" + config.Host , stopCh, &serviceController, serviceController.informer)
+	NewController("service-ctrl-"+config.Host, stopCh, &serviceController, serviceController.informer)
 
 	return &serviceController, nil
 }

admiral/pkg/controller/admiral/service_test.go

@@ -1,15 +1,20 @@
 package admiral
 
 import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
 	"github.com/google/go-cmp/cmp"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/controller/common"
 	"github.com/istio-ecosystem/admiral/admiral/pkg/test"
-	"k8s.io/api/core/v1"
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/client-go/tools/clientcmd"
-	"sync"
-	"testing"
-	"time"
 )
 
 func TestNewServiceController(t *testing.T) {
@@ -248,3 +253,49 @@ func TestServiceCache_GetLoadBalancer(t *testing.T) {
 		})
 	}
 }
+
+func TestConcurrentGetAndPut(t *testing.T) {
+	serviceCache := serviceCache{}
+	serviceCache.cache = make(map[string]*ServiceClusterEntry)
+	serviceCache.mutex = &sync.Mutex{}
+
+	serviceCache.Put(&v1.Service{
+		ObjectMeta: metaV1.ObjectMeta{Name: "testname", Namespace: "testns"},
+	})
+
+	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(3*time.Second))
+	defer cancel()
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	// Producer go routine
+	go func(ctx context.Context) {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				serviceCache.Put(&v1.Service{
+					ObjectMeta: metaV1.ObjectMeta{Name: "testname", Namespace: string(uuid.NewUUID())},
+				})
+			}
+		}
+	}(ctx)
+
+	// Consumer go routine
+	go func(ctx context.Context) {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				assert.NotNil(t, serviceCache.Get("testns"))
+			}
+		}
+	}(ctx)
+
+	wg.Wait()
+
+}