e2e: nodeselector in apiserversauce

test/rekt/apiserversource_test.go

@@ -197,3 +197,18 @@ func TestApiserversourceSendEventWithJWTOIDC(t *testing.T) {
 
 	env.Test(ctx, t, apiserversourcefeatures.ApiserversourceSendEventWithJWT())
 }
+
+func TestApiServerSourceDeployment(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+		environment.WithPollTimings(5*time.Second, 2*time.Minute),
+	)
+
+	env.Test(ctx, t, apiserversourcefeatures.DeployAPIServerSourceWithNodeSelector())
+}

test/rekt/features/apiserversource/data_plane.go

@@ -28,10 +28,10 @@ import (
 
 	"knative.dev/eventing/pkg/eventingtls/eventingtlstesting"
 	"knative.dev/eventing/test/rekt/resources/addressable"
+	"knative.dev/eventing/test/rekt/resources/configmap"
 
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
 	"k8s.io/apimachinery/pkg/util/sets"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
@@ -279,7 +279,7 @@ func SendsEventsWithEventTypes() *feature.Feature {
 
 	f := new(feature.Feature)
 
-	//Install the broker
+	// Install the broker
 	brokerName := feature.MakeRandomK8sName("broker")
 	f.Setup("install broker", broker.Install(brokerName, broker.WithEnvConfig()...))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
@@ -822,6 +822,48 @@ func SendsEventsWithRetries() *feature.Feature {
 	return f
 }
 
+func DeployAPIServerSourceWithNodeSelector() *feature.Feature {
+	f := feature.NewFeature()
+
+	f.Setup("setup config-features", func(ctx context.Context, t feature.T) {
+		env := environment.FromContext(ctx)
+		ns := env.Namespace()
+		configmap.Install("config-features", ns)
+	})
+	f.Setup("setup node labels", apiserversource.SetupNodeLabels())
+
+	source := feature.MakeRandomK8sName("apiserversource")
+	sink := feature.MakeRandomK8sName("sink")
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
+
+	sacmName := feature.MakeRandomK8sName("apiserversource")
+	f.Setup("Create Service Account for ApiServerSource with RBAC for sources.knative.dev/v1 PingSources",
+		setupAccountAndRoleForPingSources(sacmName))
+
+	cfg := []manifest.CfgFn{
+		apiserversource.WithServiceAccountName(sacmName),
+		apiserversource.WithEventMode("Reference"),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
+		apiserversource.WithResources(v1.APIVersionKindSelector{
+			APIVersion: "sources.knative.dev/v1",
+			Kind:       "PingSource",
+		}),
+	}
+
+	f.Requirement("install ApiServerSource", apiserversource.Install(source, cfg...))
+	f.Requirement("ApiServerSource goes ready", apiserversource.IsReady(source))
+
+	f.Stable("ApiServerSource using nodeSelector").Must("must use it from config-features", apiserversource.VerifyNodeSelectorDeployment(source))
+
+	f.Teardown("reset resources", func(ctx context.Context, t feature.T) {
+		f.DeleteResources(ctx, t)
+		apiserversource.ResetNodeLabels(ctx, t)
+	})
+
+	return f
+}
+
 func SendsEventsWithBrokerAsSinkTLS() *feature.Feature {
 	src := feature.MakeRandomK8sName("apiserversource")
 	sacmName := feature.MakeRandomK8sName("apiserversource")
@@ -902,5 +944,4 @@ func SendsEventsWithBrokerAsSinkTLS() *feature.Feature {
 	)
 
 	return f
-
 }

test/rekt/resources/apiserversource/apiserversource.go

@@ -19,18 +19,23 @@ package apiserversource
 import (
 	"context"
 	"embed"
+	"fmt"
 	"strings"
 	"time"
 
+	appsv1 "k8s.io/api/apps/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 
 	v1 "knative.dev/eventing/pkg/apis/sources/v1"
+	"knative.dev/pkg/client/injection/kube/client"
 
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
 	"knative.dev/reconciler-test/pkg/manifest"
 )
 
@@ -166,3 +171,96 @@ func labelSelectorToStringMap(selector *metav1.LabelSelector) map[string]interfa
 
 	return r
 }
+
+func MatchLabels(labels1 map[string]string, labels2 map[string]string) bool {
+	for k, v := range labels1 {
+		if labels2[k] != v {
+			return false
+		}
+	}
+	return true
+}
+
+func TestLabels() map[string]string {
+	return map[string]string{
+		"testkey":  "testvalue",
+		"testkey1": "testvalue1",
+		"testkey2": "testvalue2",
+	}
+}
+
+func VerifyNodeSelectorDeployment(source string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		env := environment.FromContext(ctx)
+		ns := env.Namespace()
+
+		kubeClient := client.Get(ctx)
+
+		deps, err := kubeClient.AppsV1().Deployments(ns).List(ctx, metav1.ListOptions{})
+		if err != nil {
+			t.Fatalf("error getting deployment: %v", err)
+		}
+
+		var dep appsv1.Deployment
+
+		for _, d := range deps.Items {
+			if kmeta.ChildName(fmt.Sprintf("apiserversource-%s-", source), string(d.GetUID())) == d.Name {
+				dep = d
+			}
+		}
+
+		if !MatchLabels(dep.Spec.Template.Spec.NodeSelector, TestLabels()) {
+			t.Fatalf("NodeSelector labels do not match: %v", dep.Spec.Template.Spec.NodeSelector)
+		}
+	}
+}
+
+func SetupNodeLabels() feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		nodes, err := client.Get(ctx).CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+		if err != nil {
+			t.Fatalf("Could not list nodes: %v", err)
+		}
+
+		if len(nodes.Items) == 0 {
+			t.Fatal("No nodes found")
+		}
+
+		node := &nodes.Items[0]
+
+		node.Labels["testkey"] = "testvalue"
+		node.Labels["testkey1"] = "testvalue1"
+		node.Labels["testkey2"] = "testvalue2"
+
+		_, err = client.Get(ctx).CoreV1().Nodes().Update(ctx, node, metav1.UpdateOptions{})
+
+		if err != nil {
+			t.Fatalf("Could not update node: %v", err)
+		}
+	}
+}
+
+func ResetNodeLabels(ctx context.Context, t feature.T) {
+	nodes, err := client.Get(ctx).CoreV1().Nodes().List(ctx, metav1.ListOptions{
+		LabelSelector: "testkey=testvalue,testkey1=testvalue1,testkey2=testvalue2",
+	})
+	if err != nil {
+		t.Fatalf("Could not list nodes: %v", err)
+	}
+
+	if len(nodes.Items) == 0 {
+		t.Fatal("No nodes found")
+	}
+
+	node := &nodes.Items[0]
+
+	delete(node.Labels, "testkey")
+	delete(node.Labels, "testkey1")
+	delete(node.Labels, "testkey2")
+
+	_, err = client.Get(ctx).CoreV1().Nodes().Update(ctx, node, metav1.UpdateOptions{})
+
+	if err != nil {
+		t.Fatalf("Could not update node: %v", err)
+	}
+}

test/rekt/resources/configmap/config-features.yaml

@@ -0,0 +1,17 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+  labels:
+    knative.dev/config-propagation: original
+    knative.dev/config-category: eventing
+data:
+  _example: |
+    my-enabled-flag: "enabled"
+    my-disabled-flag: "disabled"
+    my-allowed-flag: "allowed"
+    apiserversources.nodeselector.testkey: testvalue
+    apiserversources.nodeselector.testkey1: testvalue1
+    apiserversources.nodeselector.testkey2: testvalue2
+

test/rekt/resources/configmap/configmap.go

@@ -0,0 +1,46 @@
+/*
+Copyright 2021 The Knative 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.
+*/
+
+package configmap
+
+import (
+	"context"
+	"embed"
+
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+// Install will create a configmap resource and will load the configmap to the context.
+func Install(name string, ns string, opts ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name":      name,
+		"namespace": ns,
+	}
+
+	for _, fn := range opts {
+		fn(cfg)
+	}
+
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
+		}
+	}
+}