Write to intermediate file before moving to state file (#755)

* write to temp file and move to state file

* fixed memleak and other issues

* call windows replace function with MOVEFILE_WRITE_THROUGH flag

* moved few functions to platform package

* moved test files to correct dir

* addressed comments

Author: tamilmani1989

cni/ipam/ipam.go

@@ -294,9 +294,6 @@ func (plugin *ipamPlugin) Delete(args *cniSkel.CmdArgs) error {
 		return err
 	}
 
-	// Release the pool.
-	plugin.am.ReleasePool(nwCfg.Ipam.AddrSpace, nwCfg.Ipam.Subnet)
-
 	return nil
 }
 

cni/network/invoker_azure.go

@@ -58,7 +58,7 @@ func (invoker *AzureIPAMInvoker) Add(nwCfg *cni.NetworkConfig, subnetPrefix *net
 	}()
 
 	if nwCfg.IPV6Mode != "" {
-		nwCfg6 := nwCfg
+		nwCfg6 := *nwCfg
 		nwCfg6.Ipam.Environment = common.OptEnvironmentIPv6NodeIpam
 		nwCfg6.Ipam.Type = ipamV6
 
@@ -67,7 +67,7 @@ func (invoker *AzureIPAMInvoker) Add(nwCfg *cni.NetworkConfig, subnetPrefix *net
 			nwCfg6.Ipam.Subnet = invoker.nwInfo.Subnets[1].Prefix.String()
 		}
 
-		resultV6, err = invoker.plugin.DelegateAdd(ipamV6, nwCfg6)
+		resultV6, err = invoker.plugin.DelegateAdd(nwCfg6.Ipam.Type, &nwCfg6)
 		if err != nil {
 			err = invoker.plugin.Errorf("Failed to allocate v6 pool: %v", err)
 		}
@@ -83,18 +83,18 @@ func (invoker *AzureIPAMInvoker) Delete(address *net.IPNet, nwCfg *cni.NetworkCo
 
 	if nwCfg == nil {
 		return invoker.plugin.Errorf("nil nwCfg passed to CNI ADD, stack: %+v", string(debug.Stack()))
-	} else if address == nil {
+	}
+
+	if len(invoker.nwInfo.Subnets) > 0 {
+		nwCfg.Ipam.Subnet = invoker.nwInfo.Subnets[0].Prefix.String()
+	}
+
+	if address == nil {
 		if err := invoker.plugin.DelegateDel(nwCfg.Ipam.Type, nwCfg); err != nil {
-			return invoker.plugin.Errorf("Network not found, attempted to release address with error:  %v", err)
+			return invoker.plugin.Errorf("Attempted to release address with error:  %v", err)
 		}
 	} else if len(address.IP.To4()) == 4 {
-
-		// cleanup pool
-		if options[optReleasePool] == optValPool {
-			nwCfg.Ipam.Address = ""
-		}
-
-		nwCfg.Ipam.Subnet = invoker.nwInfo.Subnets[0].Prefix.String()
+		nwCfg.Ipam.Address = address.IP.String()
 		log.Printf("Releasing ipv4 address :%s pool: %s",
 			nwCfg.Ipam.Address, nwCfg.Ipam.Subnet)
 		if err := invoker.plugin.DelegateDel(nwCfg.Ipam.Type, nwCfg); err != nil {
@@ -105,6 +105,7 @@ func (invoker *AzureIPAMInvoker) Delete(address *net.IPNet, nwCfg *cni.NetworkCo
 		nwCfgIpv6 := *nwCfg
 		nwCfgIpv6.Ipam.Environment = common.OptEnvironmentIPv6NodeIpam
 		nwCfgIpv6.Ipam.Type = ipamV6
+		nwCfgIpv6.Ipam.Address = address.IP.String()
 		if len(invoker.nwInfo.Subnets) > 1 {
 			nwCfgIpv6.Ipam.Subnet = invoker.nwInfo.Subnets[1].Prefix.String()
 		}

cni/network/network.go

@@ -36,8 +36,6 @@ const (
 	// Supported IP version. Currently support only IPv4
 	ipVersion      = "4"
 	ipamV6         = "azure-vnet-ipamv6"
-	optReleasePool = "DeleteOnErr"
-	optValPool     = "pool"
 )
 
 // CNI Operation Types
@@ -438,20 +436,17 @@ func (plugin *netPlugin) Add(args *cniSkel.CmdArgs) error {
 		log.Printf("[cni-net] Creating network %v.", networkId)
 
 		if !nwCfg.MultiTenancy {
-
-			options[optReleasePool] = optValPool
 			result, resultV6, err = plugin.ipamInvoker.Add(nwCfg, &subnetPrefix, options)
 			if err != nil {
 				return err
 			}
 
 			defer func() {
 				if err != nil {
-					options[optReleasePool] = optValPool
 					if result != nil && len(result.IPs) > 0 {
 						plugin.ipamInvoker.Delete(&result.IPs[0].Address, nwCfg, options)
 					}
-					if resultV6 != nil && len(result.IPs) > 0 {
+					if resultV6 != nil && len(resultV6.IPs) > 0 {
 						plugin.ipamInvoker.Delete(&resultV6.IPs[0].Address, nwCfg, options)
 					}
 				}
@@ -460,6 +455,7 @@ func (plugin *netPlugin) Add(args *cniSkel.CmdArgs) error {
 
 		gateway := result.IPs[0].Gateway
 		subnetPrefix.IP = subnetPrefix.IP.Mask(subnetPrefix.Mask)
+		nwCfg.Ipam.Subnet = subnetPrefix.String()
 		// Find the master interface.
 		masterIfName := plugin.findMasterInterface(nwCfg, &subnetPrefix)
 		if masterIfName == "" {
@@ -541,7 +537,6 @@ func (plugin *netPlugin) Add(args *cniSkel.CmdArgs) error {
 		if !nwCfg.MultiTenancy {
 			// Network already exists.
 			log.Printf("[cni-net] Found network %v with subnet %v.", networkId, nwInfo.Subnets[0].Prefix.String())
-			nwInfo.Options[optReleasePool] = ""
 			result, resultV6, err = plugin.ipamInvoker.Add(nwCfg, &subnetPrefix, nwInfo.Options)
 			if err != nil {
 				return err
@@ -551,8 +546,12 @@ func (plugin *netPlugin) Add(args *cniSkel.CmdArgs) error {
 
 			defer func() {
 				if err != nil {
-					nwInfo.Options[optReleasePool] = ""
-					plugin.ipamInvoker.Delete(&result.IPs[0].Address, nwCfg, nwInfo.Options)
+					if result != nil && len(result.IPs) > 0 {
+						plugin.ipamInvoker.Delete(&result.IPs[0].Address, nwCfg, nwInfo.Options)
+					}
+					if resultV6 != nil && len(resultV6.IPs) > 0 {
+						plugin.ipamInvoker.Delete(&resultV6.IPs[0].Address, nwCfg, nwInfo.Options)
+					}
 				}
 			}()
 		}
@@ -847,6 +846,7 @@ func (plugin *netPlugin) Delete(args *cniSkel.CmdArgs) error {
 		if !nwCfg.MultiTenancy {
 			// attempt to release address associated with this Endpoint id
 			// This is to ensure clean up is done even in failure cases
+			log.Printf("release ip ep not found")
 			if err = plugin.ipamInvoker.Delete(nil, nwCfg, nwInfo.Options); err != nil {
 				log.Printf("Endpoint not found, attempted to release address with error: %v", err)
 			}
@@ -878,8 +878,7 @@ func (plugin *netPlugin) Delete(args *cniSkel.CmdArgs) error {
 	if !nwCfg.MultiTenancy {
 		// Call into IPAM plugin to release the endpoint's addresses.
 		for _, address := range epInfo.IPAddresses {
-			nwCfg.Ipam.Address = address.IP.String()
-			nwInfo.Options[optReleasePool] = ""
+			log.Printf("release ip:%s", address.IP.String())
 			err = plugin.ipamInvoker.Delete(&address, nwCfg, nwInfo.Options)
 			if err != nil {
 				err = plugin.Errorf("Failed to release address %v with error: %v", address, err)

cnm/plugin/main.go

@@ -152,7 +152,7 @@ func main() {
 		return
 	}
 
-	err = common.CreateDirectory(storeFileLocation)
+	err = platform.CreateDirectory(storeFileLocation)
 	if err != nil {
 		log.Errorf("Failed to create File Store directory %s, due to Error:%v", storeFileLocation, err.Error())
 		return

cns/service/main.go

@@ -439,7 +439,7 @@ func main() {
 	// Log platform information.
 	logger.Printf("Running on %v", platform.GetOSInfo())
 
-	err = acn.CreateDirectory(storeFileLocation)
+	err = platform.CreateDirectory(storeFileLocation)
 	if err != nil {
 		logger.Errorf("Failed to create File Store directory %s, due to Error:%v", storeFileLocation, err.Error())
 		return

common/utils.go

@@ -4,12 +4,10 @@
 package common
 
 import (
-	"bufio"
 	"encoding/binary"
 	"encoding/json"
 	"encoding/xml"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -123,35 +121,6 @@ func LogNetworkInterfaces() {
 	}
 }
 
-func CheckIfFileExists(filepath string) (bool, error) {
-	_, err := os.Stat(filepath)
-	if err == nil {
-		return true, nil
-	}
-
-	if os.IsNotExist(err) {
-		return false, nil
-	}
-
-	return true, err
-}
-
-func CreateDirectory(dirPath string) error {
-	var err error
-
-	if dirPath == "" {
-		log.Printf("dirPath is empty, nothing to create.")
-		return nil
-	}
-
-	isExist, _ := CheckIfFileExists(dirPath)
-	if !isExist {
-		err = os.Mkdir(dirPath, os.ModePerm)
-	}
-
-	return err
-}
-
 func IpToInt(ip net.IP) uint32 {
 	if len(ip) == 16 {
 		return binary.BigEndian.Uint32(ip[12:16])
@@ -200,38 +169,6 @@ func StartProcess(path string, args []string) error {
 	return err
 }
 
-// ReadFileByLines reads file line by line and return array of lines.
-func ReadFileByLines(filename string) ([]string, error) {
-	var (
-		lineStrArr []string
-	)
-
-	f, err := os.Open(filename)
-	if err != nil {
-		return nil, fmt.Errorf("Error opening %s file error %v", filename, err)
-	}
-
-	defer f.Close()
-
-	r := bufio.NewReader(f)
-
-	for {
-		lineStr, err := r.ReadString('\n')
-		if err != nil {
-			if err != io.EOF {
-				return nil, fmt.Errorf("Error reading %s file error %v", filename, err)
-			}
-
-			lineStrArr = append(lineStrArr, lineStr)
-			break
-		}
-
-		lineStrArr = append(lineStrArr, lineStr)
-	}
-
-	return lineStrArr, nil
-}
-
 // GetHostMetadata - retrieve VM metadata from wireserver
 func GetHostMetadata(fileName string) (Metadata, error) {
 	content, err := ioutil.ReadFile(fileName)

common/utils_test.go

@@ -187,7 +187,9 @@ func (am *addressManager) StartSource(options map[string]interface{}) error {
 	var isLoaded bool
 	environment, _ := options[common.OptEnvironment].(string)
 
-	if am.AddrSpaces != nil && len(am.AddrSpaces) > 0 {
+	if am.AddrSpaces != nil && len(am.AddrSpaces) > 0 &&
+		am.AddrSpaces[LocalDefaultAddressSpaceId] != nil &&
+		len(am.AddrSpaces[LocalDefaultAddressSpaceId].Pools) > 0 {
 		isLoaded = true
 	}