Fixing bug of not properly cleaning up after ALL interfaces when SOME of them failed during create.

The root cause is that in some cases we have returned earlier during interface creation, than starting the asynch thread.
But if some other threads have already progressed to the point of forking, they were basically prematurely reaped by the early
termination of the whole process.
Nasty.

Resolution is getting rid of ALL early return cases. In case of an early failure we simply push the result in to the syncher object, even
if the error happened in the main thread.
Only Syncher can return from the main thread, no sooner and no later than all results are available: be them pushed from the main thread, or from a fork.

Author: Levovar

pkg/cnidel/cnidel.go

@@ -63,13 +63,13 @@ func DelegateInterfaceSetup(netConf *datastructs.NetConf, danmClient danmclients
   }
   rawConfig, err := getCniPluginConfig(netConf, netInfo, ipamOptions, ep)
   if err != nil {
-    freeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
+    FreeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
     return nil, err
   }
   cniType := netInfo.Spec.NetworkType
   cniResult,err := execCniPlugin(cniType, CniAddOp, netInfo, rawConfig, ep)
   if err != nil {
-    freeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
+    FreeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
     return nil, errors.New("Error delegating ADD to CNI plugin:" + cniType + " because:" + err.Error())
   }
   if cniResult != nil {
@@ -187,19 +187,19 @@ func setEpIfaceAddress(cniResult *current.Result, epIface *danmtypes.DanmEpIface
 func DelegateInterfaceDelete(netConf *datastructs.NetConf, danmClient danmclientset.Interface, netInfo *danmtypes.DanmNet, ep *danmtypes.DanmEp) error {
   rawConfig, err := getCniPluginConfig(netConf, netInfo, datastructs.IpamConfig{Type: ipamType}, ep)
   if err != nil {
-    freeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
+    FreeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
     return err
   }
   cniType := netInfo.Spec.NetworkType
   _, err = execCniPlugin(cniType, CniDelOp, netInfo, rawConfig, ep)
   if err != nil {
-    freeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
+    FreeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
     return errors.New("Error delegating DEL to CNI plugin:" + cniType + " because:" + err.Error())
   }
-  return freeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
+  return FreeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
 }
 
-func freeDelegatedIps(danmClient danmclientset.Interface, netInfo *danmtypes.DanmNet, ip4, ip6 string) error {
+func FreeDelegatedIps(danmClient danmclientset.Interface, netInfo *danmtypes.DanmNet, ip4, ip6 string) error {
   err4 := freeDelegatedIp(danmClient, netInfo, ip4)
   err6 := freeDelegatedIp(danmClient, netInfo, ip6)
   if err4 != nil {

pkg/danmep/danmep.go

@@ -6,18 +6,14 @@ import (
   "log"
   "runtime"
   "strconv"
+  "time"
   "github.com/containernetworking/plugins/pkg/ns"
   "github.com/containernetworking/plugins/pkg/utils/sysctl"
   meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
   danmtypes "github.com/nokia/danm/crd/apis/danm/v1"
   danmclientset "github.com/nokia/danm/crd/client/clientset/versioned"
 )
 
-const (
-  defaultNamespace = "default"
-  dockerApiVersion = "1.22"
-)
-
 type sysctlFunction func(danmtypes.DanmEp) bool
 type sysctlObject struct {
   sysctlName  string
@@ -181,3 +177,22 @@ func isIPv6NotNeeded(ep danmtypes.DanmEp) bool {
   }
   return false
 }
+
+func PutDanmEp(danmClient danmclientset.Interface, ep danmtypes.DanmEp) error {
+  _, err := danmClient.DanmV1().DanmEps(ep.Namespace).Create(&ep)
+  if err != nil {
+    return errors.New("DanmEp object could not be PUT to K8s API server due to error:" + err.Error())
+  }
+  //We block the thread until DanmEp is really created in the API server, just in case
+  //We achieve this by not returning until Get for the same resource is successful
+  //Otherwise garbage collection could leak during CNI ADD if another thread finished unsuccessfully,
+  //simply because the DanmEp directing interface deletion does not yet exist
+  for i := 0; i < 100; i++ {
+    tempEp, err := danmClient.DanmV1().DanmEps(ep.Namespace).Get(ep.ObjectMeta.Name, meta_v1.GetOptions{})
+    if err == nil && tempEp.ObjectMeta.Name == ep.ObjectMeta.Name {
+      return nil
+    }
+    time.Sleep(10 * time.Millisecond)
+  }
+  return errors.New("DanmEp creation was supposedly successful, but the object hasn't really appeared within 1 sec")
+}

pkg/metacni/metacni.go

@@ -236,20 +236,22 @@ func setupNetworking(args *cniArgs) (*current.Result, error) {
     defParam := datastructs.Interface{SequenceId: 0, Ip: "dynamic",}
     err = createIface(args, danmClient, args.defaultNetwork, defParam, syncher, allocatedDevices)
     if err != nil {
-      return nil, err
+      syncher.PushResult(args.defaultNetwork.ObjectMeta.Name, err, nil)
     }
   }
   for nicID, nicParams := range args.interfaces {
     nicParams.SequenceId = nicID
     nicParams.DefaultIfaceName = defaultIfName
     netInfo, err := netcontrol.GetNetworkFromInterface(danmClient, nicParams, args.pod.ObjectMeta.Namespace)
     if err != nil {
-      return nil, errors.New("failed to get network object for Pod:" + args.pod.ObjectMeta.Name +
-                             "'s connection no.:" + strconv.Itoa(nicID) + " due to:" + err.Error())
+      syncher.PushResult("", errors.New("failed to get network object for Pod:" + args.pod.ObjectMeta.Name +
+                             "'s connection no.:" + strconv.Itoa(nicID) + " due to:" + err.Error()), nil)
+      continue
     }
     err = createIface(args, danmClient, netInfo, nicParams, syncher, allocatedDevices)
     if err != nil {
-      return nil, err
+      syncher.PushResult(netInfo.ObjectMeta.Name, err, nil)
+      continue
     }
   }
   err = syncher.GetAggregatedResult()
@@ -367,13 +369,15 @@ func createDelegatedInterface(syncher *syncher.Syncher, danmClient danmclientset
     syncher.PushResult(netInfo.ObjectMeta.Name, errors.New("CNI delegation failed due to error:" + err.Error()), nil)
     return
   }
-  _, err = danmClient.DanmV1().DanmEps(ep.Namespace).Create(&ep)
+  err = danmep.PutDanmEp(danmClient, ep)
   if err != nil {
+    cnidel.FreeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
     syncher.PushResult(netInfo.ObjectMeta.Name, errors.New("DanmEp object could not be PUT to K8s due to error:" + err.Error()), nil)
     return
   }
   err = danmep.SetDanmEpSysctls(ep)
   if err != nil {
+    cnidel.FreeDelegatedIps(danmClient, netInfo, ep.Spec.Iface.Address, ep.Spec.Iface.AddressIPv6)
     syncher.PushResult(netInfo.ObjectMeta.Name, errors.New("Sysctls could not be set due to error:" + err.Error()), nil)
     return
   }
@@ -400,7 +404,7 @@ func createDanmInterface(syncher *syncher.Syncher, danmClient danmclientset.Inte
     syncher.PushResult(netInfo.ObjectMeta.Name, errors.New("DanmEp object could not be created due to error:" + err.Error()), nil)
     return
   }
-  _, err = danmClient.DanmV1().DanmEps(ep.Namespace).Create(&ep)
+  err = danmep.PutDanmEp(danmClient, ep)
   if err != nil {
     ipam.GarbageCollectIps(danmClient, netInfo, ip4, ip6)
     syncher.PushResult(netInfo.ObjectMeta.Name, errors.New("EP could not be PUT into K8s due to error:" + err.Error()), nil)

pkg/syncher/syncher.go

@@ -28,13 +28,13 @@ func NewSyncher(numOfResults int) *Syncher {
 }
 
 func (synch *Syncher) PushResult(cniName string, opRes error, cniRes *current.Result) {
+  synch.mux.Lock()
+  defer synch.mux.Unlock()
   cniOpResult := cniOpResult {
     CniName: cniName,
     OpResult: opRes,
     CniResult: cniRes,
   }
-  synch.mux.Lock()
-  defer synch.mux.Unlock()
   synch.CniResults = append(synch.CniResults, cniOpResult)
 }