Merge pull request #87 from pperiyasamy/skip-unnecessary-node-event

OCPBUGS-1227: Skip unnecessary node events in config reconciler

.gitignore

@@ -1,6 +1,7 @@
 *.pyc
 build/
 website/public
+dev-env/unittest/bin
 .envrc
 
 # generated by helm chart-releaser

dev-env/unittest/setup-envtest.sh

@@ -0,0 +1,96 @@
+#!/usr/bin/env bash
+
+#  Copyright 2020 The Kubernetes Authors.
+#
+#  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.
+
+set -o errexit
+set -o pipefail
+
+# Turn colors in this script off by setting the NO_COLOR variable in your
+# environment to any value:
+#
+# $ NO_COLOR=1 test.sh
+NO_COLOR=${NO_COLOR:-""}
+if [ -z "$NO_COLOR" ]; then
+  header=$'\e[1;33m'
+  reset=$'\e[0m'
+else
+  header=''
+  reset=''
+fi
+
+function header_text {
+  echo "$header$*$reset"
+}
+
+function setup_envtest_env {
+  header_text "setting up env vars"
+
+  # Setup env vars
+  KUBEBUILDER_ASSETS=${KUBEBUILDER_ASSETS:-""}
+  if [[ -z "${KUBEBUILDER_ASSETS}" ]]; then
+    export KUBEBUILDER_ASSETS=$1/bin
+  fi
+}
+
+# fetch k8s API gen tools and make it available under envtest_root_dir/bin.
+#
+# Skip fetching and untaring the tools by setting the SKIP_FETCH_TOOLS variable
+# in your environment to any value:
+#
+# $ SKIP_FETCH_TOOLS=1 ./check-everything.sh
+#
+# If you skip fetching tools, this script will use the tools already on your
+# machine.
+function fetch_envtest_tools {
+  SKIP_FETCH_TOOLS=${SKIP_FETCH_TOOLS:-""}
+  if [ -n "$SKIP_FETCH_TOOLS" ]; then
+    return 0
+  fi
+
+  tmp_root=/tmp
+  envtest_root_dir=$tmp_root/envtest
+
+  k8s_version="${ENVTEST_K8S_VERSION:-1.19.2}"
+  goarch="$(go env GOARCH)"
+  goos="$(go env GOOS)"
+
+  if [[ "$goos" != "linux" && "$goos" != "darwin" ]]; then
+    echo "OS '$goos' not supported. Aborting." >&2
+    return 1
+  fi
+
+  local dest_dir="${1}"
+
+  # use the pre-existing version in the temporary folder if it matches our k8s version
+  if [[ -x "${dest_dir}/bin/kube-apiserver" ]]; then
+    version=$("${dest_dir}"/bin/kube-apiserver --version)
+    if [[ $version == *"${k8s_version}"* ]]; then
+      header_text "Using cached envtest tools from ${dest_dir}"
+      return 0
+    fi
+  fi
+
+  header_text "fetching envtest tools@${k8s_version} (into '${dest_dir}')"
+  envtest_tools_archive_name="kubebuilder-tools-$k8s_version-$goos-$goarch.tar.gz"
+  envtest_tools_download_url="https://storage.googleapis.com/kubebuilder-tools/$envtest_tools_archive_name"
+
+  envtest_tools_archive_path="$tmp_root/$envtest_tools_archive_name"
+  if [ ! -f $envtest_tools_archive_path ]; then
+    curl -sL ${envtest_tools_download_url} -o "$envtest_tools_archive_path"
+  fi
+
+  mkdir -p "${dest_dir}"
+  tar -C "${dest_dir}" --strip-components=1 -zvxf "$envtest_tools_archive_path"
+}

internal/k8s/controllers/config_controller.go

@@ -26,10 +26,13 @@ import (
 	metallbv1beta2 "go.universe.tf/metallb/api/v1beta2"
 	"go.universe.tf/metallb/internal/config"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
@@ -149,6 +152,23 @@ func (r *ConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 }
 
 func (r *ConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	p := predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			newNodeObj, ok := e.ObjectNew.(*corev1.Node)
+			if !ok {
+				return true
+			}
+			oldNodeObj, ok := e.ObjectOld.(*corev1.Node)
+			if !ok {
+				return true
+			}
+			// If there is no changes in node labels, ignore event.
+			if labels.Equals(labels.Set(oldNodeObj.Labels), labels.Set(newNodeObj.Labels)) {
+				return false
+			}
+			return true
+		},
+	}
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&metallbv1beta2.BGPPeer{}).
 		Watches(&source.Kind{Type: &metallbv1beta1.IPAddressPool{}}, &handler.EnqueueRequestForObject{}).
@@ -159,6 +179,7 @@ func (r *ConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Watches(&source.Kind{Type: &metallbv1beta1.AddressPool{}}, &handler.EnqueueRequestForObject{}).
 		Watches(&source.Kind{Type: &metallbv1beta1.Community{}}, &handler.EnqueueRequestForObject{}).
 		Watches(&source.Kind{Type: &corev1.Secret{}}, &handler.EnqueueRequestForObject{}).
+		WithEventFilter(p).
 		Complete(r)
 }
 

internal/k8s/controllers/config_controller_test.go

@@ -18,11 +18,15 @@ package controllers
 
 import (
 	"context"
+	"path/filepath"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	. "github.com/onsi/gomega"
 	v1beta1 "go.universe.tf/metallb/api/v1beta1"
 	v1beta2 "go.universe.tf/metallb/api/v1beta2"
 	"go.universe.tf/metallb/internal/config"
@@ -31,6 +35,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	k8sscheme "k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
@@ -137,6 +144,130 @@ func TestConfigController(t *testing.T) {
 	}
 }
 
+func TestNodeEvent(t *testing.T) {
+	g := NewGomegaWithT(t)
+	testEnv := &envtest.Environment{
+		CRDDirectoryPaths:     []string{filepath.Join("../../..", "config", "crd", "bases")},
+		ErrorIfCRDPathMissing: true,
+		Scheme:                scheme,
+	}
+	cfg, err := testEnv.Start()
+	g.Expect(err).To(BeNil())
+	defer func() {
+		err = testEnv.Stop()
+		g.Expect(err).To(BeNil())
+	}()
+	err = v1beta1.AddToScheme(k8sscheme.Scheme)
+	g.Expect(err).To(BeNil())
+	err = v1beta2.AddToScheme(k8sscheme.Scheme)
+	g.Expect(err).To(BeNil())
+	m, err := manager.New(cfg, manager.Options{})
+	g.Expect(err).To(BeNil())
+
+	var configUpdate int
+	var mutex sync.Mutex
+	mockHandler := func(l log.Logger, cfg *config.Config) SyncState {
+		mutex.Lock()
+		defer mutex.Unlock()
+		configUpdate++
+		return SyncStateSuccess
+	}
+	var forceReload int
+	mockForceReload := func() {
+		mutex.Lock()
+		defer mutex.Unlock()
+		forceReload++
+	}
+	r := &ConfigReconciler{
+		Client:         m.GetClient(),
+		Logger:         log.NewNopLogger(),
+		Scheme:         scheme,
+		Namespace:      testNamespace,
+		ValidateConfig: config.DontValidate,
+		Handler:        mockHandler,
+		ForceReload:    mockForceReload,
+	}
+	err = r.SetupWithManager(m)
+	g.Expect(err).To(BeNil())
+	ctx := context.Background()
+	go func() {
+		err = m.Start(ctx)
+		g.Expect(err).To(BeNil())
+	}()
+
+	// test new node event.
+	node := &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "test-node"},
+		Spec:       corev1.NodeSpec{},
+	}
+	node.Labels = make(map[string]string)
+	node.Labels["test"] = "e2e"
+	err = m.GetClient().Create(ctx, node)
+	g.Expect(err).To(BeNil())
+	g.Eventually(func() int {
+		mutex.Lock()
+		defer mutex.Unlock()
+		return configUpdate
+	}, 5*time.Second, 200*time.Millisecond).Should(Equal(1))
+	g.Eventually(func() int {
+		mutex.Lock()
+		defer mutex.Unlock()
+		return forceReload
+	}, 5*time.Second, 200*time.Millisecond).Should(Equal(0))
+
+	// test update node event with no changes into node label.
+	g.Eventually(func() error {
+		err = m.GetClient().Get(ctx, types.NamespacedName{Name: "test-node"}, node)
+		if err != nil {
+			return err
+		}
+		node.Labels = make(map[string]string)
+		node.Spec.PodCIDR = "192.168.10.0/24"
+		node.Labels["test"] = "e2e"
+		err = m.GetClient().Update(ctx, node)
+		if err != nil {
+			return err
+		}
+		return nil
+	}, 5*time.Second, 200*time.Millisecond).Should(BeNil())
+	g.Eventually(func() int {
+		mutex.Lock()
+		defer mutex.Unlock()
+		return configUpdate
+	}, 5*time.Second, 200*time.Millisecond).Should(Equal(1))
+	g.Eventually(func() int {
+		mutex.Lock()
+		defer mutex.Unlock()
+		return forceReload
+	}, 5*time.Second, 200*time.Millisecond).Should(Equal(0))
+
+	// test update node event with changes into node label.
+	g.Eventually(func() error {
+		err = m.GetClient().Get(ctx, types.NamespacedName{Name: "test-node"}, node)
+		if err != nil {
+			return err
+		}
+		node.Labels = make(map[string]string)
+		node.Labels["test"] = "e2e"
+		node.Labels["test"] = "update"
+		err = m.GetClient().Update(ctx, node)
+		if err != nil {
+			return err
+		}
+		return nil
+	}, 5*time.Second, 200*time.Millisecond).Should(BeNil())
+	g.Eventually(func() int {
+		mutex.Lock()
+		defer mutex.Unlock()
+		return configUpdate
+	}, 5*time.Second, 200*time.Millisecond).Should(Equal(2))
+	g.Eventually(func() int {
+		mutex.Lock()
+		defer mutex.Unlock()
+		return forceReload
+	}, 5*time.Second, 200*time.Millisecond).Should(Equal(0))
+}
+
 var (
 	testNamespace                  = "test-controller"
 	scheme                         = runtime.NewScheme()

tasks.py

@@ -666,9 +666,10 @@ def _replace(pattern):
 @task
 def test(ctx):
     """Run unit tests."""
-    run("go test -short ./...")
-    run("go test -short -race ./...")
-
+    envtest_asset_dir = os.getcwd() + "/dev-env/unittest"
+    run("source {}/setup-envtest.sh; fetch_envtest_tools {}".format(envtest_asset_dir, envtest_asset_dir), echo=True)
+    run("source {}/setup-envtest.sh; setup_envtest_env {}; go test -short ./...".format(envtest_asset_dir, envtest_asset_dir), echo=True)
+    run("source {}/setup-envtest.sh; setup_envtest_env {}; go test -short -race ./...".format(envtest_asset_dir, envtest_asset_dir), echo=True)
 
 @task
 def checkpatch(ctx):