Repository: virtual-kubelet/virtual-kubelet
Pull request #871 (2020-07-31 18:24:33)

Set Node Leader Owner Reference
This sets / updates the node lease owner reference to the current
node. Previously, we did not set this, which had the interesting
problem of leaking node leases on clusters with node churn.


Patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 386d3339..079f9877 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -77,70 +77,6 @@ git remote add upstream git@github.com:virtual-kubelet/virtual-kubelet.git
 ```
 3. Submit a pull request.
 
-
-## Submission and Review guidelines
-
-We welcome and appreciate everyone to submit and review changes. Here are some guidelines to follow for help ensure
-a successful contribution experience.
-
-Please note these are general guidelines, and while they are a good starting point, they are not specifically rules.
-If you have a question about something, feel free to ask:
-
-- [#virtual-kubelet](https://kubernetes.slack.com/archives/C8YU1QP8W) on Kubernetes Slack
-- [virtualkubelet-dev@lists.cncf.io](mailto:virtualkubelet-dev@lists.cncf.io)
-- GitHub Issues
-
-#### Don't make breaking API changes.
-
-Since Virtual Kubelet has reached 1.0 it is a major goal of the project to keep a stable API.
-Breaking changes must only be considered if a 2.0 release is on the table, which should only come with thoughtful
-consideration of the projects users as well as maintenance burden.
-
-Also note that behavior changes in the runtime can have cascading effects that cause unintended failures. Behavior
-changes should come well documented and with ample consideration for downstream effects. If possible, they should be
-opt-in.
-
-#### Public APIs
-
-Public API's should be extendable and flexible without requiring breaking changes.
-While we can always add a new function (`Foo2()`), a new type, etc, doing so makes it harder for people to update to
-the new behavior.
-
-Build API interfaces that do not need to be changed to adopt new or improved functionality. Opinions on how a particular
-thing should work should be encoded by the user rather than implicit in the runtime. Defaults are fine, but defaults
-should be overridable.
-
-The smaller the surface area of an API, the easier it is to do more interesting things with it.
-
-#### Building blocks
-
-Don't overload functionality. If something is complicated to setup we can provide helpers or wrappers to do that, but
-don't require users to do things a certain way because this tends to diminish the usefulness, especially as it relates
-to runtimes.
-
-We also do not want the maintenance burden of every users individual edge cases.
-
-#### Use context.Context
-
-Probably if it is a public/exported API, it should take a `context.Context`. Even if it doesn't need one today, it may
-need it tomorrow, and then we have a breaking API change.
-
-We use `context.Context` for storing loggers, tracing spans, and cancellation all across the project. Better safe
-than sorry: add a `context.Context`.
-
-#### Errors
-
-Callers can't handle errors if they don't know what the error is, so make sure they can figure that out.
-We use a package `errdefs` to define the types of errors we currently look out for. We do not typically look for
-concrete error types, so check out `errdefs` and see if there is already an error type in there for your needs, or even
-create a new one.
-
-#### Testing
-
-Ideally all behavior would be tested, in practice this is not the case. Unit tests are great, and fast. There is also
-an end-to-end test suite for testing the overall behavior of the system. Please add tests. This is also a great place
-to get started if you are new to the codebase.
-
 ## Code of conduct
 
 Virtual Kubelet follows the [CNCF Code of Conduct](https://github.com/cncf/foundation/blob/master/code-of-conduct.md).
diff --git a/node/node.go b/node/node.go
index 30d8aedb..6f6f363e 100644
--- a/node/node.go
+++ b/node/node.go
@@ -206,8 +206,7 @@ func (n *NodeController) Run(ctx context.Context) error {
 		return n.controlLoop(ctx)
 	}
 
-	n.lease = newLease(n.lease)
-	setLeaseAttrs(n.lease, n.n, n.pingInterval*5)
+	n.lease = newLease(ctx, n.lease, n.n, n.pingInterval)
 
 	l, err := ensureLease(ctx, n.leases, n.lease)
 	if err != nil {
@@ -347,7 +346,7 @@ func (n *NodeController) handlePing(ctx context.Context) (retErr error) {
 }
 
 func (n *NodeController) updateLease(ctx context.Context) error {
-	l, err := updateNodeLease(ctx, n.leases, newLease(n.lease))
+	l, err := updateNodeLease(ctx, n.leases, newLease(ctx, n.lease, n.n, n.pingInterval))
 	if err != nil {
 		return err
 	}
@@ -601,7 +600,8 @@ func updateNodeStatus(ctx context.Context, nodes v1.NodeInterface, nodeFromProvi
 	return updatedNode, nil
 }
 
-func newLease(base *coord.Lease) *coord.Lease {
+// This will return a new lease. It will either update base lease (and the set the renewal time appropriately), or create a brand new lease
+func newLease(ctx context.Context, base *coord.Lease, node *corev1.Node, leaseRenewalInterval time.Duration) *coord.Lease {
 	var lease *coord.Lease
 	if base == nil {
 		lease = &coord.Lease{}
@@ -610,21 +610,60 @@ func newLease(base *coord.Lease) *coord.Lease {
 	}
 
 	lease.Spec.RenewTime = &metav1.MicroTime{Time: time.Now()}
-	return lease
-}
 
-func setLeaseAttrs(l *coord.Lease, n *corev1.Node, dur time.Duration) {
-	if l.Name == "" {
-		l.Name = n.Name
+	if lease.Spec.LeaseDurationSeconds == nil {
+		// This is 25 due to historical reasons. It was supposed to be * 5, but...reasons
+		d := int32(leaseRenewalInterval.Seconds()) * 25
+		lease.Spec.LeaseDurationSeconds = &d
+	}
+
+	if lease.Name == "" {
+		lease.Name = node.Name
 	}
-	if l.Spec.HolderIdentity == nil {
-		l.Spec.HolderIdentity = &n.Name
+	if lease.Spec.HolderIdentity == nil {
+		// Let's do a copy here
+		name := node.Name
+		lease.Spec.HolderIdentity = &name
 	}
 
-	if l.Spec.LeaseDurationSeconds == nil {
-		d := int32(dur.Seconds()) * 5
-		l.Spec.LeaseDurationSeconds = &d
+	// Copied and pasted from: https://github.com/kubernetes/kubernetes/blob/442a69c3bdf6fe8e525b05887e57d89db1e2f3a5/pkg/kubelet/nodelease/controller.go#L213-L216
+	// Setting owner reference needs node's UID. Note that it is different from
+	// kubelet.nodeRef.UID. When lease is initially created, it is possible that
+	// the connection between master and node is not ready yet. So try to set
+	// owner reference every time when renewing the lease, until successful.
+	//
+	// We have a special case to deal with in the node may be deleted and
+	// come back with a different UID. In this case the lease object should
+	// be deleted due to a owner reference cascading deletion, and when we renew
+	// lease again updateNodeLease will call ensureLease, and establish a new
+	// lease with the right node ID
+	if l := len(lease.OwnerReferences); l == 0 {
+		lease.OwnerReferences = []metav1.OwnerReference{
+			{
+				APIVersion: corev1.SchemeGroupVersion.WithKind("Node").Version,
+				Kind:       corev1.SchemeGroupVersion.WithKind("Node").Kind,
+				Name:       node.Name,
+				UID:        node.UID,
+			},
+		}
+	} else if l > 0 {
+		var foundThisNode, foundNode bool
+		for _, ref := range lease.OwnerReferences {
+			if ref.APIVersion == corev1.SchemeGroupVersion.WithKind("Node").Version && ref.Kind == corev1.SchemeGroupVersion.WithKind("Node").Kind {
+				foundNode = true
+				if node.UID == ref.UID && node.Name == ref.Name {
+					foundThisNode = true
+				}
+			}
+		}
+		if !foundThisNode && !foundNode {
+			log.G(ctx).Warn("Found that lease had owner references, but no nodes in owner references")
+		} else if foundNode {
+			log.G(ctx).Warn("Found that lease had owner references, but nodes in owner references that is not this node")
+		}
 	}
+
+	return lease
 }
 
 func updateNodeStatusHeartbeat(n *corev1.Node) {
diff --git a/node/nodeutil/filter.go b/node/nodeutil/filter.go
deleted file mode 100644
index 026353f3..00000000
--- a/node/nodeutil/filter.go
+++ /dev/null
@@ -1,30 +0,0 @@
-package nodeutil
-
-import (
-	"context"
-
-	"github.com/virtual-kubelet/virtual-kubelet/node"
-	v1 "k8s.io/api/core/v1"
-)
-
-// FilterPodsForNodeName creates an event filter function that filters pod events such that pod.Sepc.NodeName matches the provided name
-// Use the return value of this as the PodEventFilterFunc in PodControllerConfig
-func FilterPodsForNodeName(name string) node.PodEventFilterFunc {
-	return func(_ context.Context, p *v1.Pod) bool {
-		return p.Spec.NodeName == name
-	}
-}
-
-// PodFilters turns a list of pod filters into a single filter.
-// When run, each item in the list is itterated in order until the first `true` result.
-// If nothing returns true, the filter is false.
-func PodFilters(filters ...node.PodEventFilterFunc) node.PodEventFilterFunc {
-	return func(ctx context.Context, p *v1.Pod) bool {
-		for _, f := range filters {
-			if f(ctx, p) {
-				return true
-			}
-		}
-		return false
-	}
-}
diff --git a/node/podcontroller.go b/node/podcontroller.go
index ca03c57c..fe490227 100644
--- a/node/podcontroller.go
+++ b/node/podcontroller.go
@@ -88,12 +88,6 @@ type PodNotifier interface {
 	NotifyPods(context.Context, func(*corev1.Pod))
 }
 
-// PodEventFilterFunc is used to filter pod events received from Kubernetes.
-//
-// Filters that return true means the event handler will be run
-// Filters that return false means the filter will *not* be run.
-type PodEventFilterFunc func(context.Context, *corev1.Pod) bool
-
 // PodController is the controller implementation for Pod resources.
 type PodController struct {
 	provider PodLifecycleHandler
@@ -121,8 +115,6 @@ type PodController struct {
 	// (derived from Kubernetes' cache library) -> a *knownPod struct.
 	knownPods sync.Map
 
-	podEventFilterFunc PodEventFilterFunc
-
 	// ready is a channel which will be closed once the pod controller is fully up and running.
 	// this channel will never be closed if there is an error on startup.
 	ready chan struct{}
@@ -156,8 +148,6 @@ type PodControllerConfig struct {
 
 	// PodInformer is used as a local cache for pods
 	// This should be configured to only look at pods scheduled to the node which the controller will be managing
-	// If the informer does not filter based on node, then you must provide a `PodEventFilterFunc` parameter so event handlers
-	//   can filter pods not assigned to this node.
 	PodInformer corev1informers.PodInformer
 
 	EventRecorder record.EventRecorder
@@ -166,7 +156,7 @@ type PodControllerConfig struct {
 
 	// Informers used for filling details for things like downward API in pod spec.
 	//
-	// We are using informers here instead of listers because we'll need the
+	// We are using informers here instead of listeners because we'll need the
 	// informer for certain features (like notifications for updated ConfigMaps)
 	ConfigMapInformer corev1informers.ConfigMapInformer
 	SecretInformer    corev1informers.SecretInformer
@@ -174,13 +164,6 @@ type PodControllerConfig struct {
 
 	// RateLimiter defines the rate limit of work queue
 	RateLimiter workqueue.RateLimiter
-
-	// Add custom filtering for pod informer event handlers
-	// Use this for cases where the pod informer handles more than pods assigned to this node
-	//
-	// For example, if the pod informer is not filtering based on pod.Spec.NodeName, you should
-	// set that filter here so the pod controller does not handle events for pods assigned to other nodes.
-	PodEventFilterFunc PodEventFilterFunc
 }
 
 // NewPodController creates a new pod controller with the provided config.
@@ -215,18 +198,17 @@ func NewPodController(cfg PodControllerConfig) (*PodController, error) {
 	}
 
 	pc := &PodController{
-		client:             cfg.PodClient,
-		podsInformer:       cfg.PodInformer,
-		podsLister:         cfg.PodInformer.Lister(),
-		provider:           cfg.Provider,
-		resourceManager:    rm,
-		ready:              make(chan struct{}),
-		done:               make(chan struct{}),
-		recorder:           cfg.EventRecorder,
-		k8sQ:               workqueue.NewNamedRateLimitingQueue(cfg.RateLimiter, "syncPodsFromKubernetes"),
-		deletionQ:          workqueue.NewNamedRateLimitingQueue(cfg.RateLimiter, "deletePodsFromKubernetes"),
-		podStatusQ:         workqueue.NewNamedRateLimitingQueue(cfg.RateLimiter, "syncPodStatusFromProvider"),
-		podEventFilterFunc: cfg.PodEventFilterFunc,
+		client:          cfg.PodClient,
+		podsInformer:    cfg.PodInformer,
+		podsLister:      cfg.PodInformer.Lister(),
+		provider:        cfg.Provider,
+		resourceManager: rm,
+		ready:           make(chan struct{}),
+		done:            make(chan struct{}),
+		recorder:        cfg.EventRecorder,
+		k8sQ:            workqueue.NewNamedRateLimitingQueue(cfg.RateLimiter, "syncPodsFromKubernetes"),
+		deletionQ:       workqueue.NewNamedRateLimitingQueue(cfg.RateLimiter, "deletePodsFromKubernetes"),
+		podStatusQ:      workqueue.NewNamedRateLimitingQueue(cfg.RateLimiter, "syncPodStatusFromProvider"),
 	}
 
 	return pc, nil
@@ -285,8 +267,7 @@ func (pc *PodController) Run(ctx context.Context, podSyncWorkers int) (retErr er
 	// Set up event handlers for when Pod resources change. Since the pod cache is in-sync, the informer will generate
 	// synthetic add events at this point. It again avoids the race condition of adding handlers while the cache is
 	// syncing.
-
-	var eventHandler cache.ResourceEventHandler = cache.ResourceEventHandlerFuncs{
+	pc.podsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(pod interface{}) {
 			if key, err := cache.MetaNamespaceKeyFunc(pod); err != nil {
 				log.G(ctx).Error(err)
@@ -319,22 +300,7 @@ func (pc *PodController) Run(ctx context.Context, podSyncWorkers int) (retErr er
 				pc.deletionQ.Forget(key)
 			}
 		},
-	}
-
-	if pc.podEventFilterFunc != nil {
-		eventHandler = cache.FilteringResourceEventHandler{
-			FilterFunc: func(obj interface{}) bool {
-				p, ok := obj.(*corev1.Pod)
-				if !ok {
-					return false
-				}
-				return pc.podEventFilterFunc(ctx, p)
-			},
-			Handler: eventHandler,
-		}
-	}
-
-	pc.podsInformer.Informer().AddEventHandler(eventHandler)
+	})
 
 	// Perform a reconciliation step that deletes any dangling pods from the provider.
 	// This happens only when the virtual-kubelet is starting, and operates on a "best-effort" basis.


Test patch:
diff --git a/node/node_test.go b/node/node_test.go
index 9fa80129..c5b5de5f 100644
--- a/node/node_test.go
+++ b/node/node_test.go
@@ -228,8 +228,7 @@ func TestEnsureLease(t *testing.T) {
 	n := testNode(t)
 	ctx := context.Background()
 
-	lease := newLease(nil)
-	setLeaseAttrs(lease, n, 1*time.Second)
+	lease := newLease(ctx, nil, n, 1*time.Second)
 
 	l1, err := ensureLease(ctx, c, lease.DeepCopy())
 	assert.NilError(t, err)
@@ -278,12 +277,13 @@ func TestUpdateNodeStatus(t *testing.T) {
 }
 
 func TestUpdateNodeLease(t *testing.T) {
+	ctx := context.Background()
+
 	leases := testclient.NewSimpleClientset().CoordinationV1beta1().Leases(corev1.NamespaceNodeLease)
-	lease := newLease(nil)
 	n := testNode(t)
-	setLeaseAttrs(lease, n, 0)
 
-	ctx := context.Background()
+	lease := newLease(ctx, nil, n, time.Duration(0))
+
 	l, err := updateNodeLease(ctx, leases, lease)
 	assert.NilError(t, err)
 	assert.Equal(t, l.Name, lease.Name)
@@ -304,6 +304,25 @@ func TestUpdateNodeLease(t *testing.T) {
 	assert.Assert(t, cmp.DeepEqual(compare.Spec.HolderIdentity, lease.Spec.HolderIdentity))
 }
 
+func TestFixNodeLeaseReferences(t *testing.T) {
+	ctx := context.Background()
+	n := testNode(t)
+
+	lease1 := newLease(ctx, nil, n, time.Second)
+	// Let's break owner references
+	lease1.OwnerReferences = nil
+	time.Sleep(2 * time.Nanosecond)
+	lease2 := newLease(ctx, lease1, n, time.Second)
+
+	// Make sure that newLease actually did its jobs
+	assert.Assert(t, lease2.Spec.RenewTime.Nanosecond() > lease1.Spec.RenewTime.Nanosecond())
+
+	// Let's check if owner references got set
+	assert.Assert(t, is.Len(lease2.OwnerReferences, 1))
+	assert.Assert(t, is.Equal(lease2.OwnerReferences[0].UID, n.UID))
+	assert.Assert(t, is.Equal(lease2.OwnerReferences[0].Name, n.Name))
+}
+
 // TestPingAfterStatusUpdate checks that Ping continues to be called with the specified interval
 // after a node status update occurs, when leases are disabled.
 //
diff --git a/node/nodeutil/filter_test.go b/node/nodeutil/filter_test.go
deleted file mode 100644
index 992de852..00000000
--- a/node/nodeutil/filter_test.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package nodeutil
-
-import (
-	"context"
-	"testing"
-
-	"gotest.tools/assert"
-	v1 "k8s.io/api/core/v1"
-)
-
-func TestFilterPodsForNodeName(t *testing.T) {
-	ctx := context.Background()
-	// Doesn't match pod with wrong name
-	assert.Check(t, !FilterPodsForNodeName(t.Name())(ctx, &v1.Pod{Spec: v1.PodSpec{NodeName: "NotOurNode"}}))
-	// Match pod with wrong name
-	assert.Check(t, FilterPodsForNodeName(t.Name())(ctx, &v1.Pod{Spec: v1.PodSpec{NodeName: t.Name()}}))
-}
diff --git a/node/podcontroller_test.go b/node/podcontroller_test.go
index f8e986ae..c87d8c67 100644
--- a/node/podcontroller_test.go
+++ b/node/podcontroller_test.go
@@ -90,58 +90,3 @@ func TestCompareLabels(t *testing.T) {
 	}
 	assert.Assert(t, !podsEffectivelyEqual(p1, p2))
 }
-
-// TestPodEventFilter ensure that pod filters are run for each event
-func TestPodEventFilter(t *testing.T) {
-	tc := newTestController()
-
-	wait := make(chan struct{}, 3)
-	tc.podEventFilterFunc = func(_ context.Context, pod *corev1.Pod) bool {
-		wait <- struct{}{}
-		return true
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	go tc.Run(ctx, 1)
-
-	ctxT, cancelT := context.WithTimeout(ctx, 30*time.Second)
-	defer cancelT()
-
-	select {
-	case <-ctxT.Done():
-		t.Fatal(ctxT.Err())
-	case <-tc.Done():
-		t.Fatal(tc.Err())
-	case <-tc.Ready():
-	}
-
-	pod := &corev1.Pod{}
-	pod.ObjectMeta.Namespace = "default"
-	pod.ObjectMeta.Name = "nginx"
-	pod.Spec = newPodSpec()
-
-	podC := tc.client.CoreV1().Pods(testNamespace)
-
-	_, err := podC.Create(ctx, pod, metav1.CreateOptions{})
-	assert.NilError(t, err)
-
-	pod.Annotations = map[string]string{"updated": "true"}
-	_, err = podC.Update(ctx, pod, metav1.UpdateOptions{})
-	assert.NilError(t, err)
-
-	err = podC.Delete(ctx, pod.Name, metav1.DeleteOptions{})
-	assert.NilError(t, err)
-
-	ctxT, cancelT = context.WithTimeout(ctx, 30*time.Second)
-	defer cancelT()
-	for i := 0; i < 3; i++ {
-		// check that the event filter fires
-		select {
-		case <-ctxT.Done():
-			t.Fatal(ctxT.Err())
-		case <-wait:
-		}
-	}
-}