making services multi thread safe

Author: abhi

netmaster/daemon.go

@@ -441,5 +441,5 @@ func (d *daemon) services(id string) ([]core.State, error) {
 		return []core.State{core.State(svc)}, nil
 	}
 
-	return nil, core.Errorf("Unexpected code path")
+	return nil, err
 }

netmaster/master/api.go

@@ -363,6 +363,7 @@ func ServiceProviderUpdateHandler(w http.ResponseWriter, r *http.Request, vars m
 			provider.Labels[k] = v
 			epCfg.Labels[k] = v
 		}
+
 		//maintain the containerId in endpointstat for recovery
 		epCfg.ContainerID = svcProvUpdReq.ContainerID
 
@@ -379,7 +380,9 @@ func ServiceProviderUpdateHandler(w http.ResponseWriter, r *http.Request, vars m
 		}
 
 		//update provider db
+		mastercfg.SvcMutex.Lock()
 		mastercfg.ProviderDb[providerDbID] = provider
+
 		for serviceID, service := range mastercfg.ServiceLBDb {
 			count := 0
 			if service.Tenant == svcProvUpdReq.Tenant {
@@ -395,19 +398,18 @@ func ServiceProviderUpdateHandler(w http.ResponseWriter, r *http.Request, vars m
 							append(mastercfg.ProviderDb[providerDbID].Services, serviceID)
 							//Update ServiceDB
 						mastercfg.ServiceLBDb[serviceID].Providers[providerID] = provider
-						stateDriver, err := utils.GetStateDriver()
-						if err != nil {
-							return nil, err
-						}
+
 						serviceLbState := &mastercfg.CfgServiceLBState{}
 						serviceLbState.StateDriver = stateDriver
 						err = serviceLbState.Read(serviceID)
 						if err != nil {
+							mastercfg.SvcMutex.Unlock()
 							return nil, err
 						}
 						serviceLbState.Providers[providerID] = provider
 						serviceLbState.Write()
 						SvcProviderUpdate(serviceID, false)
+						break
 					}
 				}
 			}
@@ -416,13 +418,16 @@ func ServiceProviderUpdateHandler(w http.ResponseWriter, r *http.Request, vars m
 	} else if svcProvUpdReq.Event == "die" {
 		//Received a container die event. If it was a service provider -
 		//clear the provider db and the service db and change the etcd state
+
 		providerDbID := svcProvUpdReq.ContainerID
 		if providerDbID == "" {
 			return nil, fmt.Errorf("Invalid containerID in SvcProvUpdateRequest:(nil)")
 		}
 
+		mastercfg.SvcMutex.Lock()
 		provider := mastercfg.ProviderDb[providerDbID]
 		if provider == nil {
+			mastercfg.SvcMutex.Unlock()
 			// It is not a provider . Ignore event
 			return nil, nil
 		}
@@ -431,6 +436,7 @@ func ServiceProviderUpdateHandler(w http.ResponseWriter, r *http.Request, vars m
 			service := mastercfg.ServiceLBDb[serviceID]
 			providerID := getProviderID(provider)
 			if providerID == "" {
+				mastercfg.SvcMutex.Unlock()
 				return nil, fmt.Errorf("Invalid ProviderID from providerInfo:{%v}", provider)
 			}
 			if service.Providers[providerID] != nil {
@@ -440,16 +446,19 @@ func ServiceProviderUpdateHandler(w http.ResponseWriter, r *http.Request, vars m
 				serviceLbState.StateDriver = stateDriver
 				err = serviceLbState.Read(serviceID)
 				if err != nil {
+					mastercfg.SvcMutex.Unlock()
 					return nil, err
 				}
 				delete(serviceLbState.Providers, providerID)
 				serviceLbState.Write()
 				delete(mastercfg.ProviderDb, providerDbID)
 				SvcProviderUpdate(serviceID, false)
+
 			}
 		}
 
 	}
+	mastercfg.SvcMutex.Unlock()
 	srvUpdResp := &SvcProvUpdateResponse{
 		IPAddress: svcProvUpdReq.IPAddress,
 	}

netmaster/master/servicelb.go

@@ -30,9 +30,16 @@ func CreateServiceLB(stateDriver core.StateDriver, serviceLbCfg *intent.ConfigSe
 
 	var providersPresent bool
 	serviceIP := serviceLbCfg.IPAddress
+
 	log.Infof("Recevied Create Service Load Balancer config {%v}", serviceLbCfg)
+
+	//Check if service already exists.
 	svcID := getServiceID(serviceLbCfg.ServiceName, serviceLbCfg.Tenant)
+
+	mastercfg.SvcMutex.RLock()
 	oldServiceInfo := mastercfg.ServiceLBDb[svcID]
+	mastercfg.SvcMutex.RUnlock()
+
 	if oldServiceInfo != nil {
 		//ServiceInfo Exists
 		if reflect.DeepEqual(oldServiceInfo.Ports, serviceLbCfg.Ports) &&
@@ -43,6 +50,8 @@ func CreateServiceLB(stateDriver core.StateDriver, serviceLbCfg *intent.ConfigSe
 		serviceIP = oldServiceInfo.IPAddress
 		DeleteServiceLB(stateDriver, oldServiceInfo.ServiceName, oldServiceInfo.Tenant)
 	}
+
+	//New Service
 	serviceLbState := &mastercfg.CfgServiceLBState{}
 	serviceLbState.ServiceName = serviceLbCfg.ServiceName
 	serviceLbState.Tenant = serviceLbCfg.Tenant
@@ -73,10 +82,11 @@ func CreateServiceLB(stateDriver core.StateDriver, serviceLbCfg *intent.ConfigSe
 		return err
 	}
 	serviceLbState.IPAddress = addr
-
+	mastercfg.SvcMutex.Lock()
 	err = serviceLbState.Write()
 
 	if err != nil {
+		mastercfg.SvcMutex.Unlock()
 		return err
 	}
 
@@ -112,10 +122,14 @@ func CreateServiceLB(stateDriver core.StateDriver, serviceLbCfg *intent.ConfigSe
 		}
 	}
 
+	//Write into cluster store
 	err = serviceLbState.Write()
+
 	if err != nil {
+		mastercfg.SvcMutex.Unlock()
 		return err
 	}
+	mastercfg.SvcMutex.Unlock()
 
 	if providersPresent {
 		err = SvcProviderUpdate(serviceID, false)
@@ -130,16 +144,21 @@ func CreateServiceLB(stateDriver core.StateDriver, serviceLbCfg *intent.ConfigSe
 
 //DeleteServiceLB deletes from etcd state
 func DeleteServiceLB(stateDriver core.StateDriver, serviceName string, tenantName string) error {
-	log.Infof("Receiver Delete Service Load Balancer %s on %s", serviceName, tenantName)
+
+	log.Infof("Received Delete Service Load Balancer %s on %s", serviceName, tenantName)
 	serviceLBState := &mastercfg.CfgServiceLBState{}
 	serviceLBState.StateDriver = stateDriver
 	serviceLBState.ID = getServiceID(serviceName, tenantName)
 
+	mastercfg.SvcMutex.RLock()
 	err := serviceLBState.Read(serviceLBState.ID)
 	if err != nil {
+		mastercfg.SvcMutex.RUnlock()
 		log.Errorf("Error reading service lb config for service %s in tenant %s", serviceName, tenantName)
 		return err
 	}
+	mastercfg.SvcMutex.RUnlock()
+
 	// find the network from network id
 	nwCfg := &mastercfg.CfgNetworkState{}
 	nwCfg.StateDriver = stateDriver
@@ -156,6 +175,7 @@ func DeleteServiceLB(stateDriver core.StateDriver, serviceName string, tenantNam
 
 	serviceID := getServiceID(serviceLBState.ServiceName, serviceLBState.Tenant)
 
+	mastercfg.SvcMutex.Lock()
 	//Remove the service ID from the provider cache
 	for _, providerInfo := range mastercfg.ServiceLBDb[serviceID].Providers {
 		containerID := providerInfo.ContainerID
@@ -174,9 +194,12 @@ func DeleteServiceLB(stateDriver core.StateDriver, serviceName string, tenantNam
 
 	err = serviceLBState.Clear()
 	if err != nil {
+		mastercfg.SvcMutex.Unlock()
+
 		log.Errorf("Error deleing service lb config for service %s in tenant %s", serviceName, tenantName)
 		return err
 	}
+	mastercfg.SvcMutex.Unlock()
 
 	return nil
 
@@ -199,7 +222,9 @@ func RestoreServiceProviderLBDb() {
 	}
 	svcLBState.StateDriver = stateDriver
 	svcLBCfgs, err := svcLBState.ReadAll()
+
 	if err == nil {
+		mastercfg.SvcMutex.Lock()
 		for _, svcLBCfg := range svcLBCfgs {
 			svcLB := svcLBCfg.(*mastercfg.CfgServiceLBState)
 			//mastercfg.ServiceLBDb = make(map[string]*mastercfg.ServiceLBInfo)
@@ -225,9 +250,11 @@ func RestoreServiceProviderLBDb() {
 				mastercfg.ProviderDb[providerDBId] = providerInfo
 			}
 		}
+		mastercfg.SvcMutex.Unlock()
 	} else {
 		log.Errorf("Error reading service load balancer state from cluster store")
 	}
+
 	//Recover from endpoint state as well .
 	epCfgState := mastercfg.CfgEndpointState{}
 	epCfgState.StateDriver = stateDriver
@@ -248,7 +275,9 @@ func RestoreServiceProviderLBDb() {
 				for k, v := range ep.Labels {
 					providerInfo.Labels[k] = v
 				}
+				mastercfg.SvcMutex.Lock()
 				mastercfg.ProviderDb[providerDBId] = providerInfo
+				mastercfg.SvcMutex.Unlock()
 			}
 		}
 	} else {

netmaster/mastercfg/providerState.go

@@ -45,18 +45,21 @@ type Provider struct {
 	Network     string
 	Services    []string
 	Container   string //container endpoint id
+
 }
 
 // Write the state
 func (s *SvcProvider) Write() error {
 	key := fmt.Sprintf(svcProviderPath, s.ID)
-	return s.StateDriver.WriteState(key, s, json.Marshal)
+	err := s.StateDriver.WriteState(key, s, json.Marshal)
+	return err
 }
 
 // Read the state in for a given ID.
 func (s *SvcProvider) Read(id string) error {
 	key := fmt.Sprintf(svcProviderPath, id)
-	return s.StateDriver.ReadState(key, s, json.Unmarshal)
+	err := s.StateDriver.ReadState(key, s, json.Unmarshal)
+	return err
 }
 
 // ReadAll reads all the state for master bgp configurations and returns it.
@@ -67,7 +70,8 @@ func (s *SvcProvider) ReadAll() ([]core.State, error) {
 // Clear removes the configuration from the state store.
 func (s *SvcProvider) Clear() error {
 	key := fmt.Sprintf(svcProviderPath, s.ID)
-	return s.StateDriver.ClearState(key)
+	err := s.StateDriver.ClearState(key)
+	return err
 }
 
 // WatchAll state transitions and send them through the channel.

netmaster/mastercfg/servicelbState.go

@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/contiv/netplugin/core"
+	"sync"
 )
 
 const (
@@ -40,6 +41,9 @@ type ServiceLBInfo struct {
 //ServiceLBDb is map of all services
 var ServiceLBDb = make(map[string]*ServiceLBInfo) //DB for all services keyed by servicename.tenant
 
+//SvcDbMutex is mutex for service transaction
+var SvcMutex sync.RWMutex
+
 // CfgServiceLBState is the service object configuration
 type CfgServiceLBState struct {
 	core.CommonState
@@ -55,13 +59,15 @@ type CfgServiceLBState struct {
 // Write the state
 func (s *CfgServiceLBState) Write() error {
 	key := fmt.Sprintf(serviceLBConfigPath, s.ID)
-	return s.StateDriver.WriteState(key, s, json.Marshal)
+	err := s.StateDriver.WriteState(key, s, json.Marshal)
+	return err
 }
 
 // Read the state in for a given ID.
 func (s *CfgServiceLBState) Read(id string) error {
 	key := fmt.Sprintf(serviceLBConfigPath, id)
-	return s.StateDriver.ReadState(key, s, json.Unmarshal)
+	err := s.StateDriver.ReadState(key, s, json.Unmarshal)
+	return err
 }
 
 // ReadAll reads all the state for master bgp configurations and returns it.
@@ -72,7 +78,8 @@ func (s *CfgServiceLBState) ReadAll() ([]core.State, error) {
 // Clear removes the configuration from the state store.
 func (s *CfgServiceLBState) Clear() error {
 	key := fmt.Sprintf(serviceLBConfigPath, s.ID)
-	return s.StateDriver.ClearState(key)
+	err := s.StateDriver.ClearState(key)
+	return err
 }
 
 // WatchAll state transitions and send them through the channel.