support: nodeselector in apiserversource

pkg/apis/feature/features.go

@@ -102,6 +102,25 @@ func (e Flags) String() string {
 	return fmt.Sprintf("%+v", map[string]Flag(e))
 }
 
+func (e Flags) NodeSelector() map[string]string {
+	// Check if NodeSelector is not nil
+	if e == nil {
+		return map[string]string{}
+	}
+
+	nodeSelectorMap := make(map[string]string)
+
+	for k, v := range e {
+		if strings.Contains(k, NodeSelectorLabel) {
+			key := strings.TrimPrefix(k, NodeSelectorLabel)
+			value := strings.TrimSpace(string(v))
+			nodeSelectorMap[key] = value
+		}
+	}
+
+	return nodeSelectorMap
+}
+
 // NewFlagsConfigFromMap creates a Flags from the supplied Map
 func NewFlagsConfigFromMap(data map[string]string) (Flags, error) {
 	flags := newDefaults()
@@ -122,6 +141,8 @@ func NewFlagsConfigFromMap(data map[string]string) (Flags, error) {
 			flags[sanitizedKey] = Permissive
 		} else if k == TransportEncryption && strings.EqualFold(v, string(Strict)) {
 			flags[sanitizedKey] = Strict
+		} else if strings.Contains(k, NodeSelectorLabel) {
+			flags[sanitizedKey] = Flag(v)
 		} else {
 			return flags, fmt.Errorf("cannot parse the feature flag '%s' = '%s'", k, v)
 		}

pkg/apis/feature/features_test.go

@@ -56,10 +56,13 @@ func TestGetFlags(t *testing.T) {
 	require.True(t, flags.IsAllowed("my-enabled-flag"))
 	require.True(t, flags.IsAllowed("my-allowed-flag"))
 	require.False(t, flags.IsAllowed("non-disabled-flag"))
+
+	nodeSelector := flags.NodeSelector()
+	expectedNodeSelector := map[string]string{"testkey": "testvalue", "testkey1": "testvalue1", "testkey2": "testvalue2"}
+	require.Equal(t, expectedNodeSelector, nodeSelector)
 }
 
 func TestShouldNotOverrideDefaults(t *testing.T) {
-
 	f, err := NewFlagsConfigFromMap(map[string]string{})
 	require.Nil(t, err)
 	require.NotNil(t, f)

pkg/apis/feature/flag_names.go

@@ -25,4 +25,5 @@ const (
 	TransportEncryption = "transport-encryption"
 	EvenTypeAutoCreate  = "eventtype-auto-create"
 	OIDCAuthentication  = "authentication-oidc"
+	NodeSelectorLabel   = "apiserversources.nodeselector."
 )

pkg/apis/feature/testdata/config-features.yaml

@@ -25,3 +25,7 @@ data:
     my-enabled-flag: "enabled"
     my-disabled-flag: "disabled"
     my-allowed-flag: "allowed"
+    apiserversources.nodeselector.testkey: testvalue
+    apiserversources.nodeselector.testkey1: testvalue1
+    apiserversources.nodeselector.testkey2: testvalue2
+

pkg/reconciler/apiserversource/apiserversource.go

@@ -102,7 +102,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1.ApiServerSour
 		// no Namespace defined in dest.Ref, we will use the Namespace of the source
 		// as the Namespace of dest.Ref.
 		if dest.Ref.Namespace == "" {
-			//TODO how does this work with deprecated fields
+			// TODO how does this work with deprecated fields
 			dest.Ref.Namespace = source.GetNamespace()
 		}
 	}
@@ -118,7 +118,6 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1.ApiServerSour
 	if featureFlags.IsOIDCAuthentication() {
 		// Create the role
 		err := r.createOIDCRole(ctx, source)
-
 		if err != nil {
 			logging.FromContext(ctx).Errorw("Failed when creating the OIDC Role for ApiServerSource", zap.Error(err))
 			return err
@@ -225,6 +224,8 @@ func (r *Reconciler) createReceiveAdapter(ctx context.Context, src *v1.ApiServer
 	// 	return nil, err
 	// }
 
+	featureFlags := feature.FromContext(ctx)
+
 	adapterArgs := resources.ReceiveAdapterArgs{
 		Image:         r.receiveAdapterImage,
 		Source:        src,
@@ -235,6 +236,7 @@ func (r *Reconciler) createReceiveAdapter(ctx context.Context, src *v1.ApiServer
 		Configs:       r.configs,
 		Namespaces:    namespaces,
 		AllNamespaces: allNamespaces,
+		NodeSelector:  featureFlags.NodeSelector(),
 	}
 
 	expected, err := resources.MakeReceiveAdapter(&adapterArgs)
@@ -357,7 +359,6 @@ func (r *Reconciler) runAccessCheck(ctx context.Context, src *v1.ApiServerSource
 
 	src.Status.MarkNoSufficientPermissions(lastReason, "User %s cannot %s", user, missing)
 	return fmt.Errorf("insufficient permissions: User %s cannot %s", user, missing)
-
 }
 
 func (r *Reconciler) createCloudEventAttributes(src *v1.ApiServerSource) ([]duckv1.CloudEventAttributes, error) {
@@ -385,7 +386,6 @@ func (r *Reconciler) createOIDCRole(ctx context.Context, source *v1.ApiServerSou
 	roleName := resources.GetOIDCTokenRoleName(source.Name)
 
 	expected, err := resources.MakeOIDCRole(source)
-
 	if err != nil {
 		return fmt.Errorf("Cannot create OIDC role for ApiServerSource %s/%s: %w", source.GetName(), source.GetNamespace(), err)
 	}
@@ -417,7 +417,6 @@ func (r *Reconciler) createOIDCRole(ctx context.Context, source *v1.ApiServerSou
 	}
 
 	return nil
-
 }
 
 // createOIDCRoleBinding:  this function will call resources package to get the rolebinding object