Merge pull request #5 from kaloom/retry-2

Add longer retry decupled from the event thread.

Author: benoitTK

controller/controller.go

@@ -1,4 +1,5 @@
 /*
+/*
 Copyright 2017-2019 Kaloom Inc.
 
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -59,6 +60,7 @@ type Controller struct {
 	kubeClient *kubernetes.Clientset
 	runtime    Runtime
 	cniPlugin  *cni.NetworkPlugin
+	eventQueue *EventQueue
 }
 
 // Run starts a Pod resource controller
@@ -111,6 +113,7 @@ func NewController(kubeClient *kubernetes.Clientset, endpoint, cniBinPath, cniCo
 		kubeClient: kubeClient,
 		runtime:    runTime,
 		cniPlugin:  cniPlugin,
+		eventQueue: newQueue(),
 	}
 	return c, nil
 }

controller/eventqueue.go

@@ -0,0 +1,95 @@
+/*
+Copyright 2017-2021 Kaloom Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"container/list"
+	"fmt"
+	"github.com/golang/glog"
+	"sync"
+)
+
+type OpType int
+
+const (
+	Add OpType = iota
+	Delete
+)
+
+// Event struct
+type Event struct {
+	opType OpType
+	data   interface{}
+}
+
+// EventQueue is a FIFO type queue
+type EventQueue struct {
+	q    *list.List
+	m    map[string]*list.Element // ref in the eventQueue list
+	lock sync.Mutex
+	cond *sync.Cond
+}
+
+// newQueue will create a new FIFO queue
+func newQueue() *EventQueue {
+	eq := &EventQueue{m: make(map[string]*list.Element), q: list.New()}
+	eq.q.Init()
+	eq.cond = sync.NewCond(&eq.lock)
+	return eq
+}
+
+// getKey will create a unique key identifying the event
+func (ev *Event) getKey() string {
+	return fmt.Sprintf("%+v", ev.data)
+}
+
+// Enqueue will push the new event in the FIFO queue
+// If a similar event already exists with a opposite operation type, both event will be discarded.
+func (eq *EventQueue) Enqueue(event *Event) {
+	eq.cond.L.Lock()
+	defer eq.cond.L.Unlock()
+
+	key := event.getKey()
+	glog.V(5).Infof("Enqueuing using key:%s", key)
+	if e, ok := eq.m[key]; ok {
+		ev := e.Value.(Event)
+		if ev.opType != event.opType {
+			glog.Infof("Cancelling events from queue - events cancels each others:", ev)
+			eq.q.Remove(e)
+			delete(eq.m, key)
+			return
+		}
+	}
+
+	e := eq.q.PushBack(*event)
+	glog.V(5).Infof("Enqueue new event:", event)
+	eq.m[key] = e
+	eq.cond.Signal()
+}
+
+// Dequeue will remove the first element from the queue and return it for processing.
+// The caller MUST use the mutex provided by the EventQueue struct
+func (eq *EventQueue) Dequeue() *Event {
+	e := eq.q.Front() // First element
+	if e == nil {
+		return nil
+	}
+	ev := e.Value.(Event)
+	eq.q.Remove(e)
+	delete(eq.m, ev.getKey())
+	return &ev
+}

controller/pods.go

@@ -47,10 +47,51 @@ type cniPodNetworkProperty struct {
 type cniPodNetworks []cniPodNetwork
 
 const (
-	maxRetries = 5
-	retryDelay = 1 * time.Second
+	maxRetries = 60
+	retryDelay = 10 * time.Second
 )
 
+// Process will take a element from the FIFO queue and attempt to process it (either add or remove network)
+func (c *Controller) Process(e *Event) {
+	var err error
+	switch e.opType {
+	case Add:
+		for i := 0; i < maxRetries; i++ {
+			err = c.cniPlugin.AddNetwork(e.data.(*cni.Parameters))
+			if err == nil {
+				if i > 0 {
+					glog.Infof("Succeeded adding network %+v after %d attempt", e.data, i)
+				} else {
+					glog.V(5).Infof("Succeeded adding network %+v on first attempt", e.data)
+				}
+				return
+			}
+			glog.Warningf("Failed adding network %+v... retrying %d/%d. err:%v", e.data, i+1, maxRetries, err)
+			time.Sleep(retryDelay)
+		}
+		glog.Errorf("Failed adding network %+v after %d attempt. err:%v", e.data, maxRetries, err)
+
+	case Delete:
+		for i := 0; i < maxRetries; i++ {
+			err = c.cniPlugin.DeleteNetwork(e.data.(*cni.Parameters))
+			if err == nil {
+				if i > 0 {
+					glog.Infof("Succeeded deleting network %+v after %d attempt", e.data, i)
+				} else {
+					glog.V(5).Infof("Succeeded deleting network %+v on first attempt", e.data)
+				}
+				return
+			}
+			glog.Warningf("Failed deleting network %+v... retrying %d/%d. err:%v", e.data, i+1, maxRetries, err)
+			time.Sleep(retryDelay)
+		}
+		glog.Errorf("Failed deleting network %+v after %d attempt. err:%v", e.data, maxRetries, err)
+
+	default:
+		glog.Errorf("processing invalid operation type value in event %+v", e)
+	}
+}
+
 func getNetworkSet(networks string) (gset.GSet, error) {
 	nets := cniPodNetworks{}
 	netSetBuilder := gset.NewBuilder()
@@ -122,19 +163,8 @@ func (c *Controller) addNetwork(podObj *apiv1.Pod, networkName string, np cniPod
 		return err
 	}
 
-	for i := 0; i < maxRetries; i++ {
-		err = c.cniPlugin.AddNetwork(cniParams)
-		if err == nil {
-			if i > 0 {
-				glog.V(4).Infof("Succeeded adding network %s on Pod %s after %d attempt", networkName, podObj.ObjectMeta.Name, i)
-			}
-			break
-		}
-		glog.Warningf("Failed adding network %s on Pod %s... retrying %d/%d. err:%v", networkName, podObj.ObjectMeta.Name, i+1, maxRetries, err)
-		time.Sleep(retryDelay)
-	}
-
-	return err
+	c.eventQueue.Enqueue(&Event{opType: Add, data: cniParams})
+	return nil
 }
 
 func (c *Controller) delNetwork(podObj *apiv1.Pod, networkName string, np cniPodNetworkProperty) error {
@@ -147,19 +177,8 @@ func (c *Controller) delNetwork(podObj *apiv1.Pod, networkName string, np cniPod
 		return err
 	}
 
-	for i := 0; i < maxRetries; i++ {
-		err = c.cniPlugin.DeleteNetwork(cniParams)
-		if err == nil {
-			if i > 0 {
-				glog.V(4).Infof("Succeeded deleting network %s on Pod %s after %d attempt", networkName, podObj.ObjectMeta.Name, i)
-			}
-			break
-		}
-		glog.Warningf("Failed deleting network %s on Pod %s... retrying %d/%d. err:%v", networkName, podObj.ObjectMeta.Name, i+1, maxRetries, err)
-		time.Sleep(retryDelay)
-	}
-
-	return err
+	c.eventQueue.Enqueue(&Event{opType: Delete, data: cniParams})
+	return nil
 }
 
 func (c *Controller) podUpdated(oldObj, newObj interface{}) {
@@ -207,6 +226,7 @@ func (c *Controller) podUpdated(oldObj, newObj interface{}) {
 			np := cniPodNetworkProperty{}
 			for _, n := range nets {
 				np.IfMAC = n.IfMAC
+				np.IsPrimary = n.IsPrimary
 				err := c.delNetwork(newPod, n.NetworkName, np)
 				if err != nil {
 					glog.Errorf("Failed to delete network %s on pod %s", n.NetworkName, podName)
@@ -223,15 +243,38 @@ func (c *Controller) podUpdated(oldObj, newObj interface{}) {
 		np := cniPodNetworkProperty{}
 		for _, n := range nets {
 			np.IfMAC = n.IfMAC
+			np.IsPrimary = n.IsPrimary
 			err := c.addNetwork(newPod, n.NetworkName, np)
 			if err != nil {
-				glog.V(4).Infof("Failed to add network %s on pod %s", n.NetworkName, podName)
+				glog.Errorf("Failed to add network %s on pod %s", n.NetworkName, podName)
 			}
 		}
 	}
 }
 
+func (c *Controller) eventQueueWorker() {
+	for {
+		c.eventQueue.cond.L.Lock()
+
+		for c.eventQueue.q.Len() == 0 {
+			c.eventQueue.cond.Wait()
+		}
+
+		ev := c.eventQueue.Dequeue()
+		c.eventQueue.cond.L.Unlock()
+
+		if ev != nil {
+			glog.V(5).Infof("Processing event:", ev)
+			c.Process(ev)
+		}
+	}
+}
+
 func (c *Controller) watchPods(ctx context.Context, nodeName string) (cache.Controller, error) {
+
+	// Initialize the worker queue
+	go c.eventQueueWorker()
+
 	// Currently there is no field selector for a Pod annotation
 	// https://github.com/kubernetes/kubernetes/blob/master/pkg/registry/core/pod/strategy.go
 	fs := fields.Set{