fix: repair windows cni lock issue (Azure#1712)

* Moving the lock from InitializeKeyValueStore() function to restore/save functions to improve cni performance on windows.

* fix: use defer function to unlock statefile.

* fix: fixing the IPAM lock and defer func

* fix: Optimizing cni file lock by moving SetSdnRemoteArpMacAddress() on startup for CRD and MultitenantCRD mode.

* adding store lock on telemetry service start to avoid race condition on windows.

Author: behzad-mir

cni/network/plugin/main.go

@@ -225,9 +225,10 @@ func rootExecute() error {
 		// Start telemetry process if not already started. This should be done inside lock, otherwise multiple process
 		// end up creating/killing telemetry process results in undesired state.
 		tb = telemetry.NewTelemetryBuffer()
-		tb.ConnectToTelemetryService(telemetryNumRetries, telemetryWaitTimeInMilliseconds)
+		if err = tb.ConnectCNIToTelemetryService(telemetryNumRetries, telemetryWaitTimeInMilliseconds, netPlugin.Plugin); err != nil {
+			log.Errorf("connection to telemetry service failed.")
+		}
 		defer tb.Close()
-
 		netPlugin.SetCNIReport(cniReport, tb)
 
 		t := time.Now()

cni/plugin.go

@@ -174,12 +174,15 @@ func (plugin *Plugin) InitializeKeyValueStore(config *common.PluginConfig) error
 		}
 	}
 
-	// Acquire store lock.
-	if err := plugin.Store.Lock(store.DefaultLockTimeout); err != nil {
-		log.Printf("[cni] Failed to lock store: %v.", err)
-		return err
+	// Acquiring store lock at this stage only for Linux. For optimization purpuses on Windows, the store lock will take place at later stage.
+	// TODO: The Linux store locking should be removed from here as well after some more validation.
+	if runtime.GOOS != "windows" {
+		// Acquire store lock.
+		if err := plugin.Store.Lock(store.DefaultLockTimeout); err != nil {
+			log.Printf("[cni] Failed to lock store: %v.", err)
+			return errors.Wrap(err, "error Acquiring store lock")
+		}
 	}
-
 	config.Store = plugin.Store
 
 	return nil
@@ -198,3 +201,25 @@ func (plugin *Plugin) UninitializeKeyValueStore() error {
 
 	return nil
 }
+
+// Lock key-value store. This function is being used for locking access to TelemetryService start for Windows Runtime.
+func (plugin *Plugin) LockKeyValueStore() error {
+	// Acquire store lock.
+	if err := plugin.Store.Lock(store.DefaultLockTimeout); err != nil {
+		log.Printf("[cni] Failed to lock store: %v.", err)
+		return errors.Wrap(err, "error Acquiring store lock")
+	}
+	return nil
+}
+
+// Unlock key-value store
+func (plugin *Plugin) UnLockKeyValueStore() error {
+	if plugin.Store != nil {
+		err := plugin.Store.Unlock()
+		if err != nil {
+			log.Printf("[cni] Failed to unlock store: %v.", err)
+			return errors.Wrap(err, "error unlock store lock")
+		}
+	}
+	return nil
+}

cns/restserver/api.go

@@ -20,7 +20,6 @@ import (
 	"github.com/Azure/azure-container-networking/cns/types"
 	"github.com/Azure/azure-container-networking/cns/wireserver"
 	"github.com/Azure/azure-container-networking/nmagent"
-	"github.com/Azure/azure-container-networking/platform"
 	"github.com/pkg/errors"
 )
 
@@ -885,14 +884,6 @@ func (service *HTTPRestService) getNetworkContainerByOrchestratorContext(w http.
 		return
 	}
 
-	// getNetworkContainerByOrchestratorContext gets called for multitenancy and
-	// setting the SDNRemoteArpMacAddress regKey is essential for the multitenancy
-	// to work correctly in case of windows platform. Return if there is an error
-	if err = platform.SetSdnRemoteArpMacAddress(); err != nil {
-		logger.Printf("[Azure CNS] SetSdnRemoteArpMacAddress failed with error: %s", err.Error())
-		return
-	}
-
 	getNetworkContainerResponse := service.getNetworkContainerResponse(req)
 	returnCode := getNetworkContainerResponse.Response.ReturnCode
 	err = service.Listener.Encode(w, &getNetworkContainerResponse)

cns/service/main.go

@@ -660,6 +660,13 @@ func main() {
 		}
 	}
 
+	// Setting the remote ARP MAC address to 12-34-56-78-9a-bc on windows for external traffic
+	err = platform.SetSdnRemoteArpMacAddress()
+	if err != nil {
+		logger.Errorf("Failed to set remote ARP MAC address: %v", err)
+		return
+	}
+
 	// Initialze state in if CNS is running in CRD mode
 	// State must be initialized before we start HTTPRestService
 	if config.ChannelMode == cns.CRD {
@@ -697,12 +704,6 @@ func main() {
 			return
 		}
 
-		// Setting the remote ARP MAC address to 12-34-56-78-9a-bc on windows for external traffic
-		err = platform.SetSdnRemoteArpMacAddress()
-		if err != nil {
-			logger.Errorf("Failed to set remote ARP MAC address: %v", err)
-			return
-		}
 	}
 
 	// Initialize multi-tenant controller if the CNS is running in MultiTenantCRD mode.
@@ -714,12 +715,6 @@ func main() {
 			return
 		}
 
-		// Setting the remote ARP MAC address to 12-34-56-78-9a-bc on windows for external traffic
-		err = platform.SetSdnRemoteArpMacAddress()
-		if err != nil {
-			logger.Errorf("Failed to set remote ARP MAC address: %v", err)
-			return
-		}
 	}
 
 	logger.Printf("[Azure CNS] Start HTTP listener")

ipam/manager.go

@@ -4,13 +4,15 @@
 package ipam
 
 import (
+	"runtime"
 	"sync"
 	"time"
 
 	"github.com/Azure/azure-container-networking/common"
 	"github.com/Azure/azure-container-networking/log"
 	"github.com/Azure/azure-container-networking/platform"
 	"github.com/Azure/azure-container-networking/store"
+	"github.com/pkg/errors"
 )
 
 const (
@@ -100,6 +102,22 @@ func (am *addressManager) restore(rehydrateIpamInfoOnReboot bool) error {
 		return nil
 	}
 
+	// Acquiring store lock at this stage for optimization purpuses on Windows
+	if runtime.GOOS == "windows" {
+		// Acquire store lock.
+		if err := am.store.Lock(store.DefaultLockTimeout); err != nil {
+			log.Printf("[ipam] Failed to lock store: %v.", err)
+			return errors.Wrap(err, "error Acquiring store lock")
+		}
+		// Remove the lock on the key-value store
+		defer func() {
+			err := am.store.Unlock()
+			if err != nil {
+				log.Printf("[ipam] Failed to unlock store: %v.", err)
+			}
+		}()
+	}
+
 	// Read any persisted state.
 	err := am.store.Read(storeKey, am)
 	if err != nil {
@@ -169,6 +187,22 @@ func (am *addressManager) save() error {
 	// Update time stamp.
 	am.TimeStamp = time.Now()
 
+	// Acquiring store lock at this stage for optimization purpuses on Windows
+	if runtime.GOOS == "windows" {
+		// Acquire store lock.
+		if err := am.store.Lock(store.DefaultLockTimeout); err != nil {
+			log.Printf("[ipam] Failed to lock store: %v.", err)
+			return errors.Wrap(err, "error Acquiring store lock")
+		}
+		// Remove the lock on the key-value store
+		defer func() {
+			err := am.store.Unlock()
+			if err != nil {
+				log.Printf("[ipam] Failed to unlock store: %v.", err)
+			}
+		}()
+	}
+
 	log.Printf("[ipam] saving ipam state.\n")
 	err := am.store.Write(storeKey, am)
 	if err == nil {

network/manager.go

@@ -5,6 +5,7 @@ package network
 
 import (
 	"net"
+	"runtime"
 	"sync"
 	"time"
 
@@ -15,6 +16,7 @@ import (
 	"github.com/Azure/azure-container-networking/netlink"
 	"github.com/Azure/azure-container-networking/platform"
 	"github.com/Azure/azure-container-networking/store"
+	"github.com/pkg/errors"
 )
 
 const (
@@ -127,6 +129,22 @@ func (nm *networkManager) restore(isRehydrationRequired bool) error {
 	// After a reboot, all address resources are implicitly released.
 	// Ignore the persisted state if it is older than the last reboot time.
 
+	// Acquiring store lock at this stage for optimization purpuses on Windows
+	if runtime.GOOS == "windows" {
+		// Acquire store lock.
+		if err := nm.store.Lock(store.DefaultLockTimeout); err != nil {
+			log.Printf("[cni] Failed to lock store: %v.", err)
+			return errors.Wrap(err, "error Acquiring store lock")
+		}
+		// Remove the lock on the key-value store
+		defer func() {
+			err := nm.store.Unlock()
+			if err != nil {
+				log.Printf("[cni] Failed to unlock store: %v.", err)
+			}
+		}()
+	}
+
 	// Read any persisted state.
 	err := nm.store.Read(storeKey, nm)
 	if err != nil {
@@ -170,7 +188,6 @@ func (nm *networkManager) restore(isRehydrationRequired bool) error {
 					for extIfName := range nm.ExternalInterfaces {
 						delete(nm.ExternalInterfaces, extIfName)
 					}
-
 					return nil
 				}
 			}
@@ -225,6 +242,22 @@ func (nm *networkManager) save() error {
 	// Update time stamp.
 	nm.TimeStamp = time.Now()
 
+	// Acquiring store lock at this stage for optimization purpuses on Windows
+	if runtime.GOOS == "windows" {
+		// Acquire store lock.
+		if err := nm.store.Lock(store.DefaultLockTimeout); err != nil {
+			log.Printf("[cni] Failed to lock store: %v.", err)
+			return errors.Wrap(err, "error Acquiring store lock")
+		}
+		// Remove the lock on the key-value store
+		defer func() {
+			err := nm.store.Unlock()
+			if err != nil {
+				log.Printf("[cni] Failed to unlock store: %v.", err)
+			}
+		}()
+	}
+
 	err := nm.store.Write(storeKey, nm)
 	if err == nil {
 		log.Printf("[net] Save succeeded.\n")

telemetry/telemetrybuffer.go

@@ -11,13 +11,16 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"runtime"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/Azure/azure-container-networking/cni"
 	"github.com/Azure/azure-container-networking/common"
 	"github.com/Azure/azure-container-networking/log"
 	"github.com/Azure/azure-container-networking/platform"
+	"github.com/pkg/errors"
 )
 
 // TelemetryConfig - telemetry config read by telemetry service
@@ -319,6 +322,42 @@ func (tb *TelemetryBuffer) ConnectToTelemetryService(telemetryNumRetries, teleme
 	}
 }
 
+// ConnectToTelemetryService for CNI - Attempt to spawn telemetry process if it's not already running. This function will have store lock for CNI.
+// TODO: This function should eventually get removed when stateless CNI is developed.
+func (tb *TelemetryBuffer) ConnectCNIToTelemetryService(telemetryNumRetries, telemetryWaitTimeInMilliseconds int, netPlugin *cni.Plugin) error {
+	path, dir := getTelemetryServiceDirectory()
+	args := []string{"-d", dir}
+	for attempt := 0; attempt < 2; attempt++ {
+		if err := tb.Connect(); err != nil {
+			log.Logf("Connection to telemetry socket failed: %v", err)
+			if runtime.GOOS == "windows" {
+				if err = netPlugin.LockKeyValueStore(); err != nil {
+					log.Logf("lock acquire error: %v", err)
+					return errors.Wrap(err, "lock acquire error")
+				}
+			}
+			if err = tb.Cleanup(FdName); err != nil {
+				return errors.Wrap(err, "cleanup failed")
+			}
+			if err = StartTelemetryService(path, args); err != nil {
+				return errors.Wrap(err, "StartTelemetryService failed")
+			}
+			WaitForTelemetrySocket(telemetryNumRetries, time.Duration(telemetryWaitTimeInMilliseconds))
+			if runtime.GOOS == "windows" {
+				if err = netPlugin.UnLockKeyValueStore(); err != nil {
+					log.Logf("failed to relinquish lock error: %v", err)
+					return errors.Wrap(err, "failed to relinquish lock error")
+				}
+			}
+		} else {
+			tb.Connected = true
+			log.Logf("Connected to telemetry service")
+			return nil
+		}
+	}
+	return nil
+}
+
 func getTelemetryServiceDirectory() (path string, dir string) {
 	path = fmt.Sprintf("%v/%v", CniInstallDir, TelemetryServiceProcessName)
 	if exists, _ := platform.CheckIfFileExists(path); !exists {