chore(k8s) Use a single scheme for all of k8s (#2796) (#2812)

We were modifying the same scheme in many places when configuring the
controllers. This is unecessary and is likely to introduce bugs.
We now use the same scheme for everything and modify it at initialization

Signed-off-by: Charly Molter <charly.molter@konghq.com>
(cherry picked from commit fdc5f1a)

Co-authored-by: Charly Molter <charly.molter@konghq.com>

.golangci.yml

@@ -34,6 +34,8 @@ linters-settings:
       alias: mesh_proto
     - pkg: github.com/kumahq/kuma/pkg/util/proto
       alias: util_proto
+    - pkg: github.com/kumahq/kuma/pkg/plugins/bootstrap/k8s
+      alias: bootstrap_k8s
   gomodguard:
     blocked:
       modules:

pkg/plugins/bootstrap/k8s/plugin.go

@@ -38,7 +38,10 @@ func init() {
 }
 
 func (p *plugin) BeforeBootstrap(b *core_runtime.Builder, _ core_plugins.PluginConfig) error {
-	scheme := kube_runtime.NewScheme()
+	scheme, err := NewScheme()
+	if err != nil {
+		return err
+	}
 	config := kube_ctrl.GetConfigOrDie()
 	mgr, err := kube_ctrl.NewManager(
 		config,
@@ -91,10 +94,6 @@ func secretClient(systemNamespace string, config *rest.Config, scheme *kube_runt
 	if err != nil {
 		return nil, err
 	}
-	// Add kube core scheme first, otherwise cache won't start
-	if err := kube_core.AddToScheme(scheme); err != nil {
-		return nil, errors.Wrapf(err, "could not add %q to scheme", kube_core.SchemeGroupVersion)
-	}
 
 	// We are listing secrets by our custom "type", therefore we need to add index by this field into cache
 	err = kubeCache.IndexField(context.Background(), &kube_core.Secret{}, "type", func(object kube_runtime.Object) []string {

pkg/plugins/bootstrap/k8s/scheme.go

@@ -0,0 +1,33 @@
+package k8s
+
+import (
+	"github.com/pkg/errors"
+	appsv1 "k8s.io/api/apps/v1"
+	kube_core "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	kube_runtime "k8s.io/apimachinery/pkg/runtime"
+
+	mesh_k8s "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/api/v1alpha1"
+	k8scnicncfio "github.com/kumahq/kuma/pkg/plugins/runtime/k8s/apis/k8s.cni.cncf.io"
+)
+
+// NewScheme creates a new scheme with all the necessary schemas added already (kuma CRD, builtin resources, cni CRDs...).
+func NewScheme() (*kube_runtime.Scheme, error) {
+	s := kube_runtime.NewScheme()
+	if err := kube_core.AddToScheme(s); err != nil {
+		return nil, errors.Wrapf(err, "could not add %q to scheme", kube_core.SchemeGroupVersion)
+	}
+	if err := mesh_k8s.AddToScheme(s); err != nil {
+		return nil, errors.Wrapf(err, "could not add %q to scheme", mesh_k8s.GroupVersion)
+	}
+	if err := k8scnicncfio.AddToScheme(s); err != nil {
+		return nil, errors.Wrapf(err, "could not add %q to scheme", k8scnicncfio.GroupVersion)
+	}
+	if err := apiextensionsv1.AddToScheme(s); err != nil {
+		return nil, errors.Wrapf(err, "could not add %q to scheme", apiextensionsv1.SchemeGroupVersion)
+	}
+	if err := appsv1.AddToScheme(s); err != nil {
+		return nil, errors.Wrapf(err, "could not add %q to scheme", apiextensionsv1.SchemeGroupVersion)
+	}
+	return s, nil
+}

pkg/plugins/config/k8s/k8s_suite_test.go

@@ -6,17 +6,17 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	kube_core "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
+	"github.com/kumahq/kuma/pkg/plugins/bootstrap/k8s"
 	"github.com/kumahq/kuma/pkg/test"
 )
 
 var k8sClient client.Client
 var testEnv *envtest.Environment
-var k8sClientScheme = runtime.NewScheme()
+var k8sClientScheme *runtime.Scheme
 
 func TestKubernetes(t *testing.T) {
 	test.RunSpecs(t, "Kubernetes Config Suite")
@@ -30,7 +30,7 @@ var _ = BeforeSuite(test.Within(time.Minute, func() {
 	Expect(err).ToNot(HaveOccurred())
 	Expect(cfg).ToNot(BeNil())
 
-	err = kube_core.AddToScheme(k8sClientScheme)
+	k8sClientScheme, err = k8s.NewScheme()
 	Expect(err).NotTo(HaveOccurred())
 
 	// +kubebuilder:scaffold:scheme

pkg/plugins/config/k8s/plugin.go

@@ -2,7 +2,6 @@ package k8s
 
 import (
 	"github.com/pkg/errors"
-	kube_core "k8s.io/api/core/v1"
 
 	core_plugins "github.com/kumahq/kuma/pkg/core/plugins"
 	core_store "github.com/kumahq/kuma/pkg/core/resources/store"
@@ -22,9 +21,6 @@ func (p *plugin) NewConfigStore(pc core_plugins.PluginContext, _ core_plugins.Pl
 	if !ok {
 		return nil, errors.Errorf("k8s controller runtime Manager hasn't been configured")
 	}
-	if err := kube_core.AddToScheme(mgr.GetScheme()); err != nil {
-		return nil, errors.Wrapf(err, "could not add %q to scheme", kube_core.SchemeGroupVersion)
-	}
 	converter, ok := k8s_extensions.FromResourceConverterContext(pc.Extensions())
 	if !ok {
 		return nil, errors.Errorf("k8s resource converter hasn't been configured")

pkg/plugins/resources/k8s/k8s_suite_test.go

@@ -30,7 +30,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
-	mesh_k8s "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/api/v1alpha1"
+	"github.com/kumahq/kuma/pkg/plugins/bootstrap/k8s"
 	k8s_registry "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/pkg/registry"
 
 	// +kubebuilder:scaffold:imports
@@ -41,7 +41,7 @@ import (
 
 var k8sClient client.Client
 var testEnv *envtest.Environment
-var k8sClientScheme = runtime.NewScheme()
+var k8sClientScheme *runtime.Scheme
 
 func TestKubernetes(t *testing.T) {
 	test.RunSpecs(t, "Kubernetes Resources Suite")
@@ -60,14 +60,10 @@ var _ = BeforeSuite(test.Within(time.Minute, func() {
 	Expect(err).ToNot(HaveOccurred())
 	Expect(cfg).ToNot(BeNil())
 
-	err = sample_v1alpha1.AddToScheme(k8sClientScheme)
-	Expect(err).NotTo(HaveOccurred())
-
-	err = mesh_k8s.AddToScheme(k8sClientScheme)
-	Expect(err).NotTo(HaveOccurred())
+	k8sClientScheme, err = k8s.NewScheme()
+	Expect(err).ToNot(HaveOccurred())
 
-	err = kube_core.AddToScheme(k8sClientScheme)
-	Expect(err).NotTo(HaveOccurred())
+	Expect(sample_v1alpha1.AddToScheme(k8sClientScheme)).To(Succeed())
 
 	// +kubebuilder:scaffold:scheme
 

pkg/plugins/runtime/k8s/controllers/configmap_controller.go

@@ -60,11 +60,6 @@ func (r *ConfigMapReconciler) Reconcile(req kube_ctrl.Request) (kube_ctrl.Result
 }
 
 func (r *ConfigMapReconciler) SetupWithManager(mgr kube_ctrl.Manager) error {
-	for _, addToScheme := range []func(*kube_runtime.Scheme) error{kube_core.AddToScheme, mesh_k8s.AddToScheme} {
-		if err := addToScheme(mgr.GetScheme()); err != nil {
-			return err
-		}
-	}
 	return kube_ctrl.NewControllerManagedBy(mgr).
 		For(&kube_core.ConfigMap{}).
 		Watches(&kube_source.Kind{Type: &kube_core.Service{}}, &kube_handler.EnqueueRequestsFromMapFunc{

pkg/plugins/runtime/k8s/controllers/mesh_controller.go

@@ -4,8 +4,6 @@ import (
 	"context"
 
 	"github.com/go-logr/logr"
-	"github.com/pkg/errors"
-	kube_core "k8s.io/api/core/v1"
 	kube_apierrs "k8s.io/apimachinery/pkg/api/errors"
 	kube_runtime "k8s.io/apimachinery/pkg/runtime"
 	kube_ctrl "sigs.k8s.io/controller-runtime"
@@ -69,12 +67,6 @@ func (r *MeshReconciler) Reconcile(req kube_ctrl.Request) (kube_ctrl.Result, err
 }
 
 func (r *MeshReconciler) SetupWithManager(mgr kube_ctrl.Manager) error {
-	if err := kube_core.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", kube_core.SchemeGroupVersion)
-	}
-	if err := mesh_k8s.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", mesh_k8s.GroupVersion)
-	}
 	return kube_ctrl.NewControllerManagedBy(mgr).
 		For(&mesh_k8s.Mesh{}).
 		Complete(r)

pkg/plugins/runtime/k8s/controllers/mesh_defaults_controller.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/pkg/errors"
-	kube_core "k8s.io/api/core/v1"
 	kube_apierrs "k8s.io/apimachinery/pkg/api/errors"
 	kube_ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
@@ -58,12 +57,6 @@ func (r *MeshDefaultsReconciler) Reconcile(req kube_ctrl.Request) (kube_ctrl.Res
 }
 
 func (r *MeshDefaultsReconciler) SetupWithManager(mgr kube_ctrl.Manager) error {
-	if err := kube_core.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", kube_core.SchemeGroupVersion)
-	}
-	if err := mesh_k8s.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", mesh_k8s.GroupVersion)
-	}
 	return kube_ctrl.NewControllerManagedBy(mgr).
 		For(&mesh_k8s.Mesh{}, builder.WithPredicates(onlyCreate)).
 		Complete(r)

pkg/plugins/runtime/k8s/controllers/namespace_controller.go

@@ -17,7 +17,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
-	k8scnicncfio "github.com/kumahq/kuma/pkg/plugins/runtime/k8s/apis/k8s.cni.cncf.io"
 	network_v1 "github.com/kumahq/kuma/pkg/plugins/runtime/k8s/apis/k8s.cni.cncf.io/v1"
 	"github.com/kumahq/kuma/pkg/plugins/runtime/k8s/metadata"
 )
@@ -123,15 +122,6 @@ func (r *NamespaceReconciler) deleteNetworkAttachmentDefinition(log logr.Logger,
 }
 
 func (r *NamespaceReconciler) SetupWithManager(mgr kube_ctrl.Manager) error {
-	if err := kube_core.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", kube_core.SchemeGroupVersion)
-	}
-	if err := k8scnicncfio.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", k8scnicncfio.GroupVersion)
-	}
-	if err := apiextensionsv1.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", apiextensionsv1.SchemeGroupVersion)
-	}
 	return kube_ctrl.NewControllerManagedBy(mgr).
 		For(&kube_core.Namespace{}, builder.WithPredicates(namespaceEvents)).
 		Complete(r)

pkg/plugins/runtime/k8s/controllers/pod_controller.go

@@ -266,11 +266,6 @@ func (r *PodReconciler) createOrUpdateIngress(pod *kube_core.Pod, services []*ku
 }
 
 func (r *PodReconciler) SetupWithManager(mgr kube_ctrl.Manager) error {
-	for _, addToScheme := range []func(*kube_runtime.Scheme) error{kube_core.AddToScheme, mesh_k8s.AddToScheme} {
-		if err := addToScheme(mgr.GetScheme()); err != nil {
-			return err
-		}
-	}
 	return kube_ctrl.NewControllerManagedBy(mgr).
 		For(&kube_core.Pod{}).
 		// on Service update reconcile affected Pods (all Pods in the same namespace)

pkg/plugins/runtime/k8s/controllers/pod_status_controller.go

@@ -114,12 +114,6 @@ func (r *PodStatusReconciler) Reconcile(req kube_ctrl.Request) (kube_ctrl.Result
 }
 
 func (r *PodStatusReconciler) SetupWithManager(mgr kube_ctrl.Manager) error {
-	for _, addToScheme := range []func(*kube_runtime.Scheme) error{kube_core.AddToScheme, mesh_k8s.AddToScheme} {
-		if err := addToScheme(mgr.GetScheme()); err != nil {
-			return err
-		}
-	}
-
 	return kube_ctrl.NewControllerManagedBy(mgr).
 		For(&kube_core.Pod{}, builder.WithPredicates(podStatusEvents)).
 		Complete(r)

pkg/plugins/runtime/k8s/controllers/service_controller.go

@@ -7,7 +7,6 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 	kube_core "k8s.io/api/core/v1"
-	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	kube_apierrs "k8s.io/apimachinery/pkg/api/errors"
 	kube_types "k8s.io/apimachinery/pkg/types"
 	kube_ctrl "sigs.k8s.io/controller-runtime"
@@ -16,7 +15,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
-	k8scnicncfio "github.com/kumahq/kuma/pkg/plugins/runtime/k8s/apis/k8s.cni.cncf.io"
 	"github.com/kumahq/kuma/pkg/plugins/runtime/k8s/metadata"
 )
 
@@ -71,15 +69,6 @@ func (r *ServiceReconciler) Reconcile(req kube_ctrl.Request) (kube_ctrl.Result,
 }
 
 func (r *ServiceReconciler) SetupWithManager(mgr kube_ctrl.Manager) error {
-	if err := kube_core.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", kube_core.SchemeGroupVersion)
-	}
-	if err := k8scnicncfio.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", k8scnicncfio.GroupVersion)
-	}
-	if err := apiextensionsv1.AddToScheme(mgr.GetScheme()); err != nil {
-		return errors.Wrapf(err, "could not add %q to scheme", apiextensionsv1.SchemeGroupVersion)
-	}
 	return kube_ctrl.NewControllerManagedBy(mgr).
 		For(&kube_core.Service{}, builder.WithPredicates(serviceEvents)).
 		Complete(r)

pkg/plugins/runtime/k8s/controllers/suite_test.go

@@ -7,20 +7,17 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	kube_core "k8s.io/api/core/v1"
-	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
-	meshv1alpha1 "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/api/v1alpha1"
-	k8scnicncfio "github.com/kumahq/kuma/pkg/plugins/runtime/k8s/apis/k8s.cni.cncf.io"
+	"github.com/kumahq/kuma/pkg/plugins/bootstrap/k8s"
 	"github.com/kumahq/kuma/pkg/test"
 )
 
 var k8sClient client.Client
 var testEnv *envtest.Environment
-var k8sClientScheme = runtime.NewScheme()
+var k8sClientScheme *runtime.Scheme
 
 func TestAPIs(t *testing.T) {
 	test.RunSpecs(t, "Namespace Controller Suite")
@@ -36,13 +33,7 @@ var _ = BeforeSuite(test.Within(time.Minute, func() {
 	Expect(err).ToNot(HaveOccurred())
 	Expect(cfg).ToNot(BeNil())
 
-	err = meshv1alpha1.AddToScheme(k8sClientScheme)
-	Expect(err).NotTo(HaveOccurred())
-	err = kube_core.AddToScheme(k8sClientScheme)
-	Expect(err).NotTo(HaveOccurred())
-	err = k8scnicncfio.AddToScheme(k8sClientScheme)
-	Expect(err).NotTo(HaveOccurred())
-	err = apiextensionsv1.AddToScheme(k8sClientScheme)
+	k8sClientScheme, err = k8s.NewScheme()
 	Expect(err).NotTo(HaveOccurred())
 
 	// +kubebuilder:scaffold:scheme

pkg/plugins/runtime/k8s/webhooks/injector/injector_suite_test.go

@@ -23,18 +23,17 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	kube_core "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
-	mesh_k8s "github.com/kumahq/kuma/pkg/plugins/resources/k8s/native/api/v1alpha1"
+	"github.com/kumahq/kuma/pkg/plugins/bootstrap/k8s"
 	"github.com/kumahq/kuma/pkg/test"
 )
 
 var k8sClient client.Client
 var testEnv *envtest.Environment
-var k8sClientScheme = runtime.NewScheme()
+var k8sClientScheme *runtime.Scheme
 
 func TestInjector(t *testing.T) {
 	test.RunSpecs(t, "Kubernetes Resources Suite")
@@ -52,10 +51,7 @@ var _ = BeforeSuite(test.Within(time.Minute, func() {
 	Expect(err).ToNot(HaveOccurred())
 	Expect(cfg).ToNot(BeNil())
 
-	err = mesh_k8s.AddToScheme(k8sClientScheme)
-	Expect(err).NotTo(HaveOccurred())
-
-	err = kube_core.AddToScheme(k8sClientScheme)
+	k8sClientScheme, err = k8s.NewScheme()
 	Expect(err).NotTo(HaveOccurred())
 
 	// +kubebuilder:scaffold:scheme

pkg/plugins/runtime/k8s/webhooks/service_validator_test.go

@@ -8,10 +8,9 @@ import (
 	. "github.com/onsi/ginkgo/extensions/table"
 	. "github.com/onsi/gomega"
 	admissionv1beta1 "k8s.io/api/admission/v1beta1"
-	kube_core "k8s.io/api/core/v1"
-	kube_runtime "k8s.io/apimachinery/pkg/runtime"
 	kube_admission "sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
+	"github.com/kumahq/kuma/pkg/plugins/bootstrap/k8s"
 	. "github.com/kumahq/kuma/pkg/plugins/runtime/k8s/webhooks"
 )
 
@@ -20,11 +19,10 @@ var _ = Describe("ServiceValidator", func() {
 	var decoder *kube_admission.Decoder
 
 	BeforeEach(func() {
-		scheme := kube_runtime.NewScheme()
+		scheme, err := k8s.NewScheme()
 		// expect
-		Expect(kube_core.AddToScheme(scheme)).To(Succeed())
+		Expect(err).ToNot(HaveOccurred())
 
-		var err error
 		// when
 		decoder, err = kube_admission.NewDecoder(scheme)
 		// then