Merge pull request istio-ecosystem#122 from services-mesh/concurrent_…

…map_fix

fixed concurrent map issue

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
@@ -34,7 +36,7 @@ type ServiceController struct {
 type serviceCache struct {
 	//map of dependencies key=identity value array of onboarded identities
 	cache map[string]*ServiceClusterEntry
-	mutex *sync.Mutex
+	mutex *sync.RWMutex
 }
 
 func (s *serviceCache) Put(service *k8sV1.Service) {
@@ -69,6 +71,8 @@ func (s *serviceCache) getKey(service *k8sV1.Service) string {
 }
 
 func (s *serviceCache) Get(key string) *ServiceClusterEntry {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
 	return s.cache[key]
 }
 
@@ -117,7 +121,7 @@ func NewServiceController(stopCh <-chan struct{}, handler ServiceHandler, config
 
 	podCache := serviceCache{}
 	podCache.cache = make(map[string]*ServiceClusterEntry)
-	podCache.mutex = &sync.Mutex{}
+	podCache.mutex = &sync.RWMutex{}
 
 	serviceController.Cache = &podCache
 	var err error
@@ -139,7 +143,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) {
@@ -35,7 +40,7 @@ func TestNewServiceController(t *testing.T) {
 func TestServiceCache_Put(t *testing.T) {
 	serviceCache := serviceCache{}
 	serviceCache.cache = make(map[string]*ServiceClusterEntry)
-	serviceCache.mutex = &sync.Mutex{}
+	serviceCache.mutex = &sync.RWMutex{}
 
 	//test service cache empty with admiral ignore should skip to save in cache
 
@@ -88,7 +93,7 @@ func TestServiceCache_Put(t *testing.T) {
 func TestServiceCache_GetLoadBalancer(t *testing.T) {
 	sc := serviceCache{}
 	sc.cache = make(map[string]*ServiceClusterEntry)
-	sc.mutex = &sync.Mutex{}
+	sc.mutex = &sync.RWMutex{}
 
 	service := &v1.Service{}
 	service.Name = "test-service"
@@ -237,3 +242,49 @@ func TestServiceCache_GetLoadBalancer(t *testing.T) {
 		})
 	}
 }
+
+func TestConcurrentGetAndPut(t *testing.T) {
+	serviceCache := serviceCache{}
+	serviceCache.cache = make(map[string]*ServiceClusterEntry)
+	serviceCache.mutex = &sync.RWMutex{}
+
+	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()
+
+}