Implement CNI update for Azure CNI (#265)

* Implement CNI Update for Azure CNI (#21)

Author: sharmasushant

cni/cni.go

@@ -9,10 +9,11 @@ import (
 
 const (
 	// CNI commands.
-	Cmd    = "CNI_COMMAND"
-	CmdAdd = "ADD"
-	CmdGet = "GET"
-	CmdDel = "DEL"
+	Cmd       = "CNI_COMMAND"
+	CmdAdd    = "ADD"
+	CmdGet    = "GET"
+	CmdDel    = "DEL"
+	CmdUpdate = "UPDATE"
 
 	// CNI errors.
 	ErrRuntime = 100
@@ -29,4 +30,5 @@ type PluginApi interface {
 	Add(args *cniSkel.CmdArgs) error
 	Get(args *cniSkel.CmdArgs) error
 	Delete(args *cniSkel.CmdArgs) error
+	Update(args *cniSkel.CmdArgs) error
 }

cni/ipam/ipam.go

@@ -290,3 +290,8 @@ func (plugin *ipamPlugin) Delete(args *cniSkel.CmdArgs) error {
 
 	return nil
 }
+
+// Update handles CNI update command.
+func (plugin *ipamPlugin) Update(args *cniSkel.CmdArgs) error {
+	return nil
+}

cni/netconfig.go

@@ -48,6 +48,7 @@ type NetworkConfig struct {
 	MultiTenancy               bool     `json:"multiTenancy,omitempty"`
 	EnableSnatOnHost           bool     `json:"enableSnatOnHost,omitempty"`
 	EnableExactMatchForPodName bool     `json:"enableExactMatchForPodName,omitempty"`
+	CNSUrl                     string   `json:"cnsurl,omitempty"`
 	Ipam                       struct {
 		Type          string `json:"type"`
 		Environment   string `json:"environment,omitempty"`

cni/network/mutlitenancy.go

@@ -205,6 +205,7 @@ func checkIfSubnetOverlaps(enableInfraVnet bool, nwCfg *cni.NetworkConfig, cnsNe
 	return false
 }
 
+// GetMultiTenancyCNIResult retrieves network goal state of a container from CNS
 func GetMultiTenancyCNIResult(
 	enableInfraVnet bool,
 	nwCfg *cni.NetworkConfig,
@@ -214,7 +215,7 @@ func GetMultiTenancyCNIResult(
 	ifName string) (*cniTypesCurr.Result, *cns.GetNetworkContainerResponse, net.IPNet, *cniTypesCurr.Result, error) {
 
 	if nwCfg.MultiTenancy {
-		result, cnsNetworkConfig, subnetPrefix, err := getContainerNetworkConfiguration(nwCfg, "", k8sPodName, k8sNamespace, ifName)
+		result, cnsNetworkConfig, subnetPrefix, err := getContainerNetworkConfiguration(nwCfg, nwCfg.CNSUrl, k8sPodName, k8sNamespace, ifName)
 		if err != nil {
 			log.Printf("GetContainerNetworkConfiguration failed for podname %v namespace %v with error %v", k8sPodName, k8sNamespace, err)
 			return nil, nil, net.IPNet{}, nil, err

cni/network/network.go

@@ -4,11 +4,13 @@
 package network
 
 import (
+	"encoding/json"
 	"fmt"
 	"net"
 
 	"github.com/Azure/azure-container-networking/cni"
 	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/cnsclient"
 	"github.com/Azure/azure-container-networking/common"
 	"github.com/Azure/azure-container-networking/log"
 	"github.com/Azure/azure-container-networking/network"
@@ -397,17 +399,20 @@ func (plugin *netPlugin) Add(args *cniSkel.CmdArgs) error {
 	}
 
 	epInfo = &network.EndpointInfo{
-		Id:               endpointId,
-		ContainerID:      args.ContainerID,
-		NetNsPath:        args.Netns,
-		IfName:           args.IfName,
-		EnableSnatOnHost: nwCfg.EnableSnatOnHost,
-		EnableInfraVnet:  enableInfraVnet,
-		Data:             make(map[string]interface{}),
-		DNS:              epDNSInfo,
-		Policies:         policies,
-	}
-	
+		Id:                 endpointId,
+		ContainerID:        args.ContainerID,
+		NetNsPath:          args.Netns,
+		IfName:             args.IfName,
+		Data:               make(map[string]interface{}),
+		DNS:                epDNSInfo,
+		Policies:           policies,
+		EnableSnatOnHost:   nwCfg.EnableSnatOnHost,
+		EnableMultiTenancy: nwCfg.MultiTenancy,
+		EnableInfraVnet:    enableInfraVnet,
+		PODName:            k8sPodName,
+		PODNameSpace:       k8sNamespace,
+	}
+
 	epPolicies := getPoliciesFromRuntimeCfg(nwCfg)
 	for _, epPolicy := range epPolicies {
 		epInfo.Policies = append(epInfo.Policies, epPolicy)
@@ -599,3 +604,159 @@ func (plugin *netPlugin) Delete(args *cniSkel.CmdArgs) error {
 
 	return nil
 }
+
+// Update handles CNI update commands.
+// Update is only supported for multitenancy and to update routes.
+func (plugin *netPlugin) Update(args *cniSkel.CmdArgs) error {
+	var (
+		result         *cniTypesCurr.Result
+		err            error
+		nwCfg          *cni.NetworkConfig
+		existingEpInfo *network.EndpointInfo
+	)
+
+	log.Printf("[cni-net] Processing UPDATE command with args {Netns:%v Args:%v Path:%v}.",
+		args.Netns, args.Args, args.Path)
+
+	// Parse network configuration from stdin.
+	nwCfg, err = cni.ParseNetworkConfig(args.StdinData)
+	if err != nil {
+		err = plugin.Errorf("Failed to parse network configuration: %v.", err)
+		return err
+	}
+
+	log.Printf("[cni-net] Read network configuration %+v.", nwCfg)
+
+	defer func() {
+		if result == nil {
+			result = &cniTypesCurr.Result{}
+		}
+
+		// Convert result to the requested CNI version.
+		res, vererr := result.GetAsVersion(nwCfg.CNIVersion)
+		if vererr != nil {
+			log.Printf("GetAsVersion failed with error %v", vererr)
+			plugin.Error(vererr)
+		}
+
+		if err == nil && res != nil {
+			// Output the result to stdout.
+			res.Print()
+		}
+
+		log.Printf("[cni-net] UPDATE command completed with result:%+v err:%v.", result, err)
+	}()
+
+	// Parse Pod arguments.
+	podCfg, err := cni.ParseCniArgs(args.Args)
+	if err != nil {
+		log.Printf("Error while parsing CNI Args during UPDATE %v", err)
+		return err
+	}
+
+	k8sNamespace := string(podCfg.K8S_POD_NAMESPACE)
+	if len(k8sNamespace) == 0 {
+		errMsg := "Required parameter Pod Namespace not specified in CNI Args during UPDATE"
+		log.Printf(errMsg)
+		return plugin.Errorf(errMsg)
+	}
+
+	k8sPodName := string(podCfg.K8S_POD_NAME)
+	if len(k8sPodName) == 0 {
+		errMsg := "Required parameter Pod Name not specified in CNI Args during UPDATE"
+		log.Printf(errMsg)
+		return plugin.Errorf(errMsg)
+	}
+
+	// Initialize values from network config.
+	networkID := nwCfg.Name
+
+	// Query the network.
+	_, err = plugin.nm.GetNetworkInfo(networkID)
+	if err != nil {
+		errMsg := fmt.Sprintf("Failed to query network during CNI UPDATE: %v", err)
+		log.Printf(errMsg)
+		return plugin.Errorf(errMsg)
+	}
+
+	// Query the existing endpoint since this is an update.
+	// Right now, we do not support updating pods that have multiple endpoints.
+	existingEpInfo, err = plugin.nm.GetEndpointInfoBasedOnPODDetails(networkID, k8sPodName, k8sNamespace)
+	if err != nil {
+		plugin.Errorf("Failed to retrieve target endpoint for CNI UPDATE [name=%v, namespace=%v]: %v", k8sPodName, k8sNamespace, err)
+		return err
+	} else {
+		log.Printf("Retrieved existing endpoint from state that may get update: %+v", existingEpInfo)
+	}
+
+	// now query CNS to get the target routes that should be there in the networknamespace (as a result of update)
+	log.Printf("Going to collect target routes for [name=%v, namespace=%v] from CNS.", k8sPodName, k8sNamespace)
+	cnsClient, err := cnsclient.NewCnsClient(nwCfg.CNSUrl)
+	if err != nil {
+		log.Printf("Initializing CNS client error in CNI Update%v", err)
+		log.Printf(err.Error())
+		return plugin.Errorf(err.Error())
+	}
+
+	// create struct with info for target POD
+	podInfo := cns.KubernetesPodInfo{PodName: k8sPodName, PodNamespace: k8sNamespace}
+	orchestratorContext, err := json.Marshal(podInfo)
+	if err != nil {
+		log.Printf("Marshalling KubernetesPodInfo failed with %v", err)
+		return plugin.Errorf(err.Error())
+	}
+
+	targetNetworkConfig, err := cnsClient.GetNetworkConfiguration(orchestratorContext)
+	if err != nil {
+		log.Printf("GetNetworkConfiguration failed with %v", err)
+		return plugin.Errorf(err.Error())
+	}
+
+	log.Printf("Network config received from cns for [name=%v, namespace=%v] is as follows -> %+v", k8sPodName, k8sNamespace, targetNetworkConfig)
+	targetEpInfo := &network.EndpointInfo{}
+
+	// get the target routes that should replace existingEpInfo.Routes inside the network namespace
+	log.Printf("Going to collect target routes for [name=%v, namespace=%v] from targetNetworkConfig.", k8sPodName, k8sNamespace)
+	if targetNetworkConfig.Routes != nil && len(targetNetworkConfig.Routes) > 0 {
+		for _, route := range targetNetworkConfig.Routes {
+			log.Printf("Adding route from routes to targetEpInfo %+v", route)
+			_, dstIPNet, _ := net.ParseCIDR(route.IPAddress)
+			gwIP := net.ParseIP(route.GatewayIPAddress)
+			targetEpInfo.Routes = append(targetEpInfo.Routes, network.RouteInfo{Dst: *dstIPNet, Gw: gwIP, DevName: existingEpInfo.IfName})
+			log.Printf("Successfully added route from routes to targetEpInfo %+v", route)
+		}
+	}
+
+	log.Printf("Going to collect target routes based on Cnetaddressspace for [name=%v, namespace=%v] from targetNetworkConfig.", k8sPodName, k8sNamespace)
+	ipconfig := targetNetworkConfig.IPConfiguration
+	for _, ipRouteSubnet := range targetNetworkConfig.CnetAddressSpace {
+		log.Printf("Adding route from cnetAddressspace to targetEpInfo %+v", ipRouteSubnet)
+		dstIPNet := net.IPNet{IP: net.ParseIP(ipRouteSubnet.IPAddress), Mask: net.CIDRMask(int(ipRouteSubnet.PrefixLength), 32)}
+		gwIP := net.ParseIP(ipconfig.GatewayIPAddress)
+		route := network.RouteInfo{Dst: dstIPNet, Gw: gwIP, DevName: existingEpInfo.IfName}
+		targetEpInfo.Routes = append(targetEpInfo.Routes, route)
+		log.Printf("Successfully added route from cnetAddressspace to targetEpInfo %+v", ipRouteSubnet)
+	}
+
+	log.Printf("Finished collecting new routes in targetEpInfo as follows: %+v", targetEpInfo.Routes)
+	log.Printf("Now saving existing infravnetaddress space if needed.")
+	for _, ns := range nwCfg.PodNamespaceForDualNetwork {
+		if k8sNamespace == ns {
+			targetEpInfo.EnableInfraVnet = true
+			targetEpInfo.InfraVnetAddressSpace = nwCfg.InfraVnetAddressSpace
+			log.Printf("Saving infravnet address space %s for [%s-%s]",
+				targetEpInfo.InfraVnetAddressSpace, existingEpInfo.PODNameSpace, existingEpInfo.PODName)
+			break
+		}
+	}
+
+	// Update the endpoint.
+	log.Printf("Now updating existing endpoint %v with targetNetworkConfig %+v.", existingEpInfo.Id, targetNetworkConfig)
+	err = plugin.nm.UpdateEndpoint(networkID, existingEpInfo, targetEpInfo)
+	if err != nil {
+		err = plugin.Errorf("Failed to update endpoint: %v", err)
+		return err
+	}
+
+	return nil
+}

cni/network/plugin/main.go

@@ -4,14 +4,19 @@
 package main
 
 import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
 	"os"
 	"reflect"
 
 	"github.com/Azure/azure-container-networking/cni"
 	"github.com/Azure/azure-container-networking/cni/network"
 	"github.com/Azure/azure-container-networking/common"
+	acn "github.com/Azure/azure-container-networking/common"
 	"github.com/Azure/azure-container-networking/log"
 	"github.com/Azure/azure-container-networking/telemetry"
+	"github.com/containernetworking/cni/pkg/skel"
 )
 
 const (
@@ -23,6 +28,22 @@ const (
 // Version is populated by make during build.
 var version string
 
+// Command line arguments for CNI plugin.
+var args = acn.ArgumentList{
+	{
+		Name:         acn.OptVersion,
+		Shorthand:    acn.OptVersionAlias,
+		Description:  "Print version information",
+		Type:         "bool",
+		DefaultValue: false,
+	},
+}
+
+// Prints version information.
+func printVersion() {
+	fmt.Printf("Azure CNI Version %v\n", version)
+}
+
 // If report write succeeded, mark the report flag state to false.
 func markSendReport(reportManager *telemetry.ReportManager) {
 	if err := reportManager.SetReportState(telemetry.CNITelemetryFile); err != nil {
@@ -48,8 +69,82 @@ func reportPluginError(reportManager *telemetry.ReportManager, err error) {
 	}
 }
 
+func validateConfig(jsonBytes []byte) error {
+	var conf struct {
+		Name string `json:"name"`
+	}
+	if err := json.Unmarshal(jsonBytes, &conf); err != nil {
+		return fmt.Errorf("error reading network config: %s", err)
+	}
+	if conf.Name == "" {
+		return fmt.Errorf("missing network name")
+	}
+	return nil
+}
+
+func getCmdArgsFromEnv() (string, *skel.CmdArgs, error) {
+	log.Printf("Going to read from stdin")
+	stdinData, err := ioutil.ReadAll(os.Stdin)
+	if err != nil {
+		return "", nil, fmt.Errorf("error reading from stdin: %v", err)
+	}
+
+	cmdArgs := &skel.CmdArgs{
+		ContainerID: os.Getenv("CNI_CONTAINERID"),
+		Netns:       os.Getenv("CNI_NETNS"),
+		IfName:      os.Getenv("CNI_IFNAME"),
+		Args:        os.Getenv("CNI_ARGS"),
+		Path:        os.Getenv("CNI_PATH"),
+		StdinData:   stdinData,
+	}
+
+	cmd := os.Getenv("CNI_COMMAND")
+	return cmd, cmdArgs, nil
+}
+
+func handleIfCniUpdate(update func(*skel.CmdArgs) error) (bool, error) {
+	isupdate := true
+
+	if os.Getenv("CNI_COMMAND") != cni.CmdUpdate {
+		return false, nil
+	}
+
+	log.Printf("CNI UPDATE received.")
+
+	_, cmdArgs, err := getCmdArgsFromEnv()
+	if err != nil {
+		log.Printf("Received error while retrieving cmds from environment: %+v", err)
+		return isupdate, err
+	}
+
+	log.Printf("Retrieved command args for update +%v", cmdArgs)
+	err = validateConfig(cmdArgs.StdinData)
+	if err != nil {
+		log.Printf("Failed to handle CNI UPDATE, err:%v.", err)
+		return isupdate, err
+	}
+
+	err = update(cmdArgs)
+	if err != nil {
+		log.Printf("Failed to handle CNI UPDATE, err:%v.", err)
+		return isupdate, err
+	}
+
+	return isupdate, nil
+}
+
 // Main is the entry point for CNI network plugin.
 func main() {
+
+	// Initialize and parse command line arguments.
+	acn.ParseArgs(&args, printVersion)
+	vers := acn.GetArg(acn.OptVersion).(bool)
+
+	if vers {
+		printVersion()
+		os.Exit(0)
+	}
+
 	var (
 		config common.PluginConfig
 		err    error
@@ -109,7 +204,10 @@ func main() {
 		panic("network plugin fatal error")
 	}
 
-	if err = netPlugin.Execute(cni.PluginApi(netPlugin)); err != nil {
+	handled, err := handleIfCniUpdate(netPlugin.Update)
+	if handled == true {
+		log.Printf("CNI UPDATE finished.")
+	} else if err = netPlugin.Execute(cni.PluginApi(netPlugin)); err != nil {
 		log.Printf("Failed to execute network plugin, err:%v.\n", err)
 		reportPluginError(reportManager, err)
 	}