Repository: apache/camel-k
Pull request #2182 (2021-03-29 08:30:01)

feat(operator): NodeSelector install options
<!-- Description -->

With this PR we give the user the possibility to chose a `Node Selector` when installing the operator, both for regular and `olm` modes.



<!--
Enter your extended release note in the below block. If the PR requires
additional action from users switching to the new release, include the string
"action required". If no release note is required, write "NONE". 

You can (optionally) mark this PR with labels "kind/bug" or "kind/feature" to make sure
the text is added to the right section of the release notes. 
-->

**Release Note**
```release-note
feat(operator): NodeSelector install options
```


squakez: @astefanutti as you were involved in a similar PR for tolerations, would you mind having a look at this? Thanks!
astefanutti: @squakez could you please rebase?

Patch:
diff --git a/pkg/cmd/install.go b/pkg/cmd/install.go
index 0adbac4ce..1692c15e9 100644
--- a/pkg/cmd/install.go
+++ b/pkg/cmd/install.go
@@ -128,6 +128,7 @@ func newCmdInstall(rootCmdOptions *RootCmdOptions) (*cobra.Command, *installCmdO
 
 	// Pod settings
 	cmd.Flags().StringArrayP("toleration", "", nil, "Add a Toleration to the operator Pod")
+	cmd.Flags().StringArrayP("node-selector", "", nil, "Add a NodeSelector to the operator Pod")
 
 	// save
 	cmd.Flags().Bool("save", false, "Save the install parameters into the default kamel configuration file (kamel-config.yaml)")
@@ -175,6 +176,7 @@ type installCmdOptions struct {
 	Properties              []string `mapstructure:"properties"`
 	TraitProfile            string   `mapstructure:"trait-profile"`
 	Tolerations             []string `mapstructure:"tolerations"`
+	NodeSelectors           []string `mapstructure:"node-selectors"`
 	HTTPProxySecret         string   `mapstructure:"http-proxy-secret"`
 
 	registry         v1.IntegrationPlatformRegistrySpec
@@ -221,7 +223,7 @@ func (o *installCmdOptions) install(cobraCmd *cobra.Command, _ []string) error {
 		if installViaOLM {
 			fmt.Fprintln(cobraCmd.OutOrStdout(), "OLM is available in the cluster")
 			var installed bool
-			if installed, err = olm.Install(o.Context, olmClient, o.Namespace, o.Global, o.olmOptions, collection, o.Tolerations); err != nil {
+			if installed, err = olm.Install(o.Context, olmClient, o.Namespace, o.Global, o.olmOptions, collection, o.Tolerations, o.NodeSelectors); err != nil {
 				return err
 			}
 			if !installed {
@@ -272,7 +274,8 @@ func (o *installCmdOptions) install(cobraCmd *cobra.Command, _ []string) error {
 					Enabled: o.Monitoring,
 					Port:    o.MonitoringPort,
 				},
-				Tolerations: o.Tolerations,
+				Tolerations:   o.Tolerations,
+				NodeSelectors: o.NodeSelectors,
 			}
 			err = install.OperatorOrCollect(o.Context, c, cfg, collection, o.Force)
 			if err != nil {
diff --git a/pkg/install/operator.go b/pkg/install/operator.go
index 1fc6cf7c9..50d4e5b35 100644
--- a/pkg/install/operator.go
+++ b/pkg/install/operator.go
@@ -55,6 +55,7 @@ type OperatorConfiguration struct {
 	Health                OperatorHealthConfiguration
 	Monitoring            OperatorMonitoringConfiguration
 	Tolerations           []string
+	NodeSelectors         []string
 }
 
 // OperatorHealthConfiguration --
@@ -99,6 +100,18 @@ func OperatorOrCollect(ctx context.Context, c client.Client, cfg OperatorConfigu
 			}
 		}
 
+		if cfg.NodeSelectors != nil {
+			if d, ok := o.(*appsv1.Deployment); ok {
+				if d.Labels["camel.apache.org/component"] == "operator" {
+					nodeSelector, err := kubernetes.NewNodeSelectors(cfg.NodeSelectors)
+					if err != nil {
+						fmt.Println("Warning: could not parse the configured node selectors!")
+					}
+					d.Spec.Template.Spec.NodeSelector = nodeSelector
+				}
+			}
+		}
+
 		if d, ok := o.(*appsv1.Deployment); ok {
 			if d.Labels["camel.apache.org/component"] == "operator" {
 				// Metrics endpoint port
diff --git a/pkg/trait/container.go b/pkg/trait/container.go
index 8f3a7d1ac..238a8274a 100644
--- a/pkg/trait/container.go
+++ b/pkg/trait/container.go
@@ -186,8 +186,8 @@ func (t *containerTrait) configureContainer(e *Environment) error {
 	}
 
 	// combine Environment of integration with platform, kit, integration
-	for key, value := range e.collectConfigurationPairs("env") {
-		envvar.SetVal(&container.Env, key, value)
+	for _, env := range e.collectConfigurationPairs("env") {
+		envvar.SetVal(&container.Env, env.Name, env.Value)
 	}
 
 	envvar.SetVal(&container.Env, "CAMEL_K_DIGEST", e.Integration.Status.Digest)
diff --git a/pkg/trait/cron.go b/pkg/trait/cron.go
index 478fe122c..e87612884 100644
--- a/pkg/trait/cron.go
+++ b/pkg/trait/cron.go
@@ -266,7 +266,7 @@ func (t *cronTrait) getCronJobFor(e *Environment) *v1beta1.CronJob {
 
 	// Copy annotations from the integration resource
 	if e.Integration.Annotations != nil {
-		for k, v := range FilterTransferableAnnotations(e.Integration.Annotations) {
+		for k, v := range filterTransferableAnnotations(e.Integration.Annotations) {
 			annotations[k] = v
 		}
 	}
diff --git a/pkg/trait/deployment.go b/pkg/trait/deployment.go
index 8fd05ade7..11d24c6f8 100644
--- a/pkg/trait/deployment.go
+++ b/pkg/trait/deployment.go
@@ -60,9 +60,7 @@ func (t *deploymentTrait) Configure(e *Environment) (bool, error) {
 		return condition != nil && condition.Status == corev1.ConditionTrue, nil
 	}
 
-	//
 	// Don't deploy when a different strategy is needed (e.g. Knative, Cron)
-	//
 	strategy, err := e.DetermineControllerStrategy()
 	if err != nil {
 		e.Integration.Status.SetErrorCondition(
@@ -141,7 +139,7 @@ func (t *deploymentTrait) getDeploymentFor(e *Environment) *appsv1.Deployment {
 	// create a copy to avoid sharing the underlying annotation map
 	annotations := make(map[string]string)
 	if e.Integration.Annotations != nil {
-		for k, v := range FilterTransferableAnnotations(e.Integration.Annotations) {
+		for k, v := range filterTransferableAnnotations(e.Integration.Annotations) {
 			annotations[k] = v
 		}
 	}
diff --git a/pkg/trait/knative_service.go b/pkg/trait/knative_service.go
index a6f5eda24..a1e8ac19a 100644
--- a/pkg/trait/knative_service.go
+++ b/pkg/trait/knative_service.go
@@ -269,7 +269,7 @@ func (t *knativeServiceTrait) getServiceFor(e *Environment) *serving.Service {
 
 	// Copy annotations from the integration resource
 	if e.Integration.Annotations != nil {
-		for k, v := range FilterTransferableAnnotations(e.Integration.Annotations) {
+		for k, v := range filterTransferableAnnotations(e.Integration.Annotations) {
 			annotations[k] = v
 		}
 	}
diff --git a/pkg/trait/trait.go b/pkg/trait/trait.go
index 28c388ec5..35ea851eb 100644
--- a/pkg/trait/trait.go
+++ b/pkg/trait/trait.go
@@ -20,17 +20,17 @@ package trait
 import (
 	"context"
 
+	"github.com/pkg/errors"
+
 	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 
 	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/client"
 	"github.com/apache/camel-k/pkg/platform"
 	"github.com/apache/camel-k/pkg/util/kubernetes"
-	"github.com/pkg/errors"
-	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 )
 
-// Apply --
 func Apply(ctx context.Context, c client.Client, integration *v1.Integration, kit *v1.IntegrationKit) (*Environment, error) {
 	environment, err := newEnvironment(ctx, c, integration, kit)
 	if err != nil {
@@ -77,7 +77,7 @@ func newEnvironment(ctx context.Context, c client.Client, integration *v1.Integr
 	}
 
 	if kit == nil {
-		kit, err = GetIntegrationKit(ctx, c, integration)
+		kit, err = getIntegrationKit(ctx, c, integration)
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/trait/trait_types.go b/pkg/trait/trait_types.go
index a34c71882..b3cd99131 100644
--- a/pkg/trait/trait_types.go
+++ b/pkg/trait/trait_types.go
@@ -402,8 +402,8 @@ func (e *Environment) computeConfigMaps() []ctrl.Object {
 	// properties have the priority
 	userProperties := ""
 
-	for key, val := range e.collectConfigurationPairs("property") {
-		userProperties += fmt.Sprintf("%s=%s\n", key, val)
+	for _, prop := range e.collectConfigurationPairs("property") {
+		userProperties += fmt.Sprintf("%s=%s\n", prop.Name, prop.Value)
 	}
 
 	if userProperties != "" {
@@ -783,8 +783,12 @@ func (e *Environment) collectConfigurationValues(configurationType string) []str
 	return collectConfigurationValues(configurationType, e.Platform, e.IntegrationKit, e.Integration)
 }
 
-func (e *Environment) collectConfigurationPairs(configurationType string) map[string]string {
-	return CollectConfigurationPairs(configurationType, e.Platform, e.IntegrationKit, e.Integration)
+type variable struct {
+	Name, Value string
+}
+
+func (e *Environment) collectConfigurationPairs(configurationType string) []variable {
+	return collectConfigurationPairs(configurationType, e.Platform, e.IntegrationKit, e.Integration)
 }
 
 func (e *Environment) getIntegrationContainer() *corev1.Container {
diff --git a/pkg/trait/util.go b/pkg/trait/util.go
index debba1f75..695b7ee41 100644
--- a/pkg/trait/util.go
+++ b/pkg/trait/util.go
@@ -28,7 +28,7 @@ import (
 	user "github.com/mitchellh/go-homedir"
 	"github.com/scylladb/go-set/strset"
 
-	k8sclient "sigs.k8s.io/controller-runtime/pkg/client"
+	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
 
 	v1 "github.com/apache/camel-k/pkg/apis/camel/v1"
 	"github.com/apache/camel-k/pkg/client"
@@ -39,17 +39,13 @@ import (
 
 var exactVersionRegexp = regexp.MustCompile(`^(\d+)\.(\d+)\.([\w-.]+)$`)
 
-// GetIntegrationKit retrieves the kit set on the integration
-func GetIntegrationKit(ctx context.Context, c client.Client, integration *v1.Integration) (*v1.IntegrationKit, error) {
+// getIntegrationKit retrieves the kit set on the integration
+func getIntegrationKit(ctx context.Context, c client.Client, integration *v1.Integration) (*v1.IntegrationKit, error) {
 	if integration.Status.IntegrationKit == nil {
 		return nil, nil
 	}
 	kit := v1.NewIntegrationKit(integration.Status.IntegrationKit.Namespace, integration.Status.IntegrationKit.Name)
-	key := k8sclient.ObjectKey{
-		Namespace: integration.Status.IntegrationKit.Namespace,
-		Name:      integration.Status.IntegrationKit.Name,
-	}
-	err := c.Get(ctx, key, &kit)
+	err := c.Get(ctx, ctrl.ObjectKeyFromObject(&kit), &kit)
 	return &kit, err
 }
 
@@ -80,8 +76,8 @@ func collectConfigurationValues(configurationType string, configurable ...v1.Con
 	return s
 }
 
-func CollectConfigurationPairs(configurationType string, configurable ...v1.Configurable) map[string]string {
-	result := make(map[string]string)
+func collectConfigurationPairs(configurationType string, configurable ...v1.Configurable) []variable {
+	result := make([]variable, 0)
 
 	for _, c := range configurable {
 		c := c
@@ -98,14 +94,28 @@ func CollectConfigurationPairs(configurationType string, configurable ...v1.Conf
 		for _, entry := range entries {
 			if entry.Type == configurationType {
 				pair := strings.SplitN(entry.Value, "=", 2)
+				var k, v string
+				if len(pair) >= 1 {
+					k = strings.TrimSpace(pair[0])
+				}
 				if len(pair) == 2 {
-					k := strings.TrimSpace(pair[0])
-					v := strings.TrimSpace(pair[1])
+					v = strings.TrimSpace(pair[1])
+				}
+				if k == "" {
+					continue
+				}
 
-					if len(k) > 0 && len(v) > 0 {
-						result[k] = v
+				ok := false
+				for i, variable := range result {
+					if variable.Name == k {
+						result[i].Value = v
+						ok = true
+						break
 					}
 				}
+				if !ok {
+					result = append(result, variable{Name: k, Value: v})
+				}
 			}
 		}
 	}
@@ -129,8 +139,8 @@ func keyValuePairArrayAsStringMap(pairs []string) (map[string]string, error) {
 	return m, nil
 }
 
-// FilterTransferableAnnotations returns a map containing annotations that are meaningful for being transferred to child resources.
-func FilterTransferableAnnotations(annotations map[string]string) map[string]string {
+// filterTransferableAnnotations returns a map containing annotations that are meaningful for being transferred to child resources.
+func filterTransferableAnnotations(annotations map[string]string) map[string]string {
 	res := make(map[string]string)
 	for k, v := range annotations {
 		if strings.HasPrefix(k, "kubectl.kubernetes.io") {
diff --git a/pkg/util/kubernetes/factory.go b/pkg/util/kubernetes/factory.go
index a85c4723c..956627c1e 100644
--- a/pkg/util/kubernetes/factory.go
+++ b/pkg/util/kubernetes/factory.go
@@ -26,8 +26,9 @@ import (
 )
 
 var validTaintRegexp = regexp.MustCompile(`^([\w\/_\-\.]+)(=)?([\w_\-\.]+)?:(NoSchedule|NoExecute|PreferNoSchedule):?(\d*)?$`)
+var validNodeSelectorRegexp = regexp.MustCompile(`^([\w\/_\-\.]+)=([\w_\-\.]+)$`)
 
-// GetTolerations build an array of Tolerations from an array of string
+// NewTolerations build an array of Tolerations from an array of string
 func NewTolerations(taints []string) ([]corev1.Toleration, error) {
 	tolerations := make([]corev1.Toleration, 0)
 	for _, t := range taints {
@@ -60,3 +61,17 @@ func NewTolerations(taints []string) ([]corev1.Toleration, error) {
 
 	return tolerations, nil
 }
+
+// NewNodeSelectors build a map of NodeSelectors from an array of string
+func NewNodeSelectors(nsArray []string) (map[string]string, error) {
+	nodeSelectors := make(map[string]string)
+	for _, ns := range nsArray {
+		if !validNodeSelectorRegexp.MatchString(ns) {
+			return nil, fmt.Errorf("could not match node selector %v", ns)
+		}
+		// Parse the regexp groups
+		groups := validNodeSelectorRegexp.FindStringSubmatch(ns)
+		nodeSelectors[groups[1]] = groups[2]
+	}
+	return nodeSelectors, nil
+}
diff --git a/pkg/util/olm/operator.go b/pkg/util/olm/operator.go
index 896b24194..bb9ce2af8 100644
--- a/pkg/util/olm/operator.go
+++ b/pkg/util/olm/operator.go
@@ -138,7 +138,7 @@ func HasPermissionToInstall(ctx context.Context, client client.Client, namespace
 }
 
 // Install creates a subscription for the OLM package
-func Install(ctx context.Context, client client.Client, namespace string, global bool, options Options, collection *kubernetes.Collection, tolerations []string) (bool, error) {
+func Install(ctx context.Context, client client.Client, namespace string, global bool, options Options, collection *kubernetes.Collection, tolerations []string, nodeSelectors []string) (bool, error) {
 	options = fillDefaults(options)
 	if installed, err := IsOperatorInstalled(ctx, client, namespace, global, options); err != nil {
 		return false, err
@@ -171,6 +171,10 @@ func Install(ctx context.Context, client client.Client, namespace string, global
 	if err != nil {
 		return false, errors.Wrap(err, fmt.Sprintf("could not set tolerations"))
 	}
+	err = maybeSetNodeSelectors(&sub, nodeSelectors)
+	if err != nil {
+		return false, errors.Wrap(err, fmt.Sprintf("could not set node selectors"))
+	}
 
 	if collection != nil {
 		collection.Add(&sub)
@@ -216,6 +220,17 @@ func maybeSetTolerations(sub *operatorsv1alpha1.Subscription, tolArray []string)
 	return nil
 }
 
+func maybeSetNodeSelectors(sub *operatorsv1alpha1.Subscription, nsArray []string) error {
+	if nsArray != nil {
+		nodeSelectors, err := kubernetes.NewNodeSelectors(nsArray)
+		if err != nil {
+			return err
+		}
+		sub.Spec.Config.NodeSelector = nodeSelectors
+	}
+	return nil
+}
+
 // Uninstall removes CSV and subscription from the namespace
 func Uninstall(ctx context.Context, client client.Client, namespace string, global bool, options Options) error {
 	sub, err := findSubscription(ctx, client, namespace, global, options)


Test patch:
diff --git a/pkg/trait/util_test.go b/pkg/trait/util_test.go
index 9e9e35099..87d70cf59 100644
--- a/pkg/trait/util_test.go
+++ b/pkg/trait/util_test.go
@@ -94,8 +94,10 @@ func TestCollectConfigurationPairs(t *testing.T) {
 	e.Platform.ResyncStatusFullConfig()
 
 	pairs := e.collectConfigurationPairs("property")
-	assert.Equal(t, "integration", pairs["p1"])
-	assert.Equal(t, "kit", pairs["p2"])
-	assert.Equal(t, "platform", pairs["p3"])
-	assert.Equal(t, "integration", pairs["p4"])
+	assert.Equal(t, pairs, []variable{
+		{Name: "p1", Value: "integration"},
+		{Name: "p2", Value: "kit"},
+		{Name: "p3", Value: "platform"},
+		{Name: "p4", Value: "integration"},
+	})
 }
diff --git a/pkg/util/kubernetes/factory_test.go b/pkg/util/kubernetes/factory_test.go
new file mode 100644
index 000000000..1bda9f343
--- /dev/null
+++ b/pkg/util/kubernetes/factory_test.go
@@ -0,0 +1,136 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 kubernetes
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+)
+
+func TestValidTolerations(t *testing.T) {
+	validTolerations := [][]string{
+		{"key=value:NoSchedule"},
+		{"key=value:NoExecute"},
+		{"key=value:PreferNoSchedule"},
+		{"key=value:NoSchedule:120"},
+		{"key=value:NoExecute:120"},
+		{"key=value:PreferNoSchedule:120"},
+		{"existKey:NoSchedule"},
+		{"existKey:NoExecute"},
+		{"existKey:PreferNoSchedule"},
+		{"existKey:NoSchedule:120"},
+		{"existKey:NoExecute:120"},
+		{"existKey:PreferNoSchedule:120"},
+	}
+	for _, vd := range validTolerations {
+		_, err := NewTolerations(vd)
+		assert.Nil(t, err)
+	}
+}
+
+func TestInvalidTolerations(t *testing.T) {
+	validTolerations := [][]string{
+		{"key-NoSchedule"},
+		{"key=value:Something"},
+		{"key@wrong=value:PreferNoSchedule"},
+		{"key=value%wrong:NoSchedule:120"},
+		{"existKey"},
+		{"existKey:"},
+		{"existKey:Something"},
+		{"existKey:PreferNoSchedule:something"},
+	}
+	for _, vd := range validTolerations {
+		_, err := NewTolerations(vd)
+		assert.NotNil(t, err)
+	}
+}
+
+func TestValueTolerations(t *testing.T) {
+	tolerations := []string{
+		"key=value:NoSchedule",
+		"key=value:NoExecute:120",
+		"existKey:PreferNoSchedule",
+		"existKey:NoSchedule:120",
+	}
+	toleration, err := NewTolerations(tolerations)
+	assert.Nil(t, err)
+	assert.Equal(t, 4, len(toleration))
+
+	assert.Equal(t, "key", toleration[0].Key)
+	assert.Equal(t, v1.TolerationOpEqual, toleration[0].Operator)
+	assert.Equal(t, "value", toleration[0].Value)
+	assert.Equal(t, v1.TaintEffectNoSchedule, toleration[0].Effect)
+
+	assert.Equal(t, "key", toleration[1].Key)
+	assert.Equal(t, v1.TolerationOpEqual, toleration[1].Operator)
+	assert.Equal(t, "value", toleration[1].Value)
+	assert.Equal(t, v1.TaintEffectNoExecute, toleration[1].Effect)
+	assert.Equal(t, int64(120), *toleration[1].TolerationSeconds)
+
+	assert.Equal(t, "existKey", toleration[2].Key)
+	assert.Equal(t, v1.TolerationOpExists, toleration[2].Operator)
+	assert.Equal(t, "", toleration[2].Value)
+	assert.Equal(t, v1.TaintEffectPreferNoSchedule, toleration[2].Effect)
+
+	assert.Equal(t, "existKey", toleration[3].Key)
+	assert.Equal(t, v1.TolerationOpExists, toleration[3].Operator)
+	assert.Equal(t, "", toleration[3].Value)
+	assert.Equal(t, v1.TaintEffectNoSchedule, toleration[3].Effect)
+	assert.Equal(t, int64(120), *toleration[3].TolerationSeconds)
+}
+
+func TestValidNodeSelectors(t *testing.T) {
+	validNodeSelectors := [][]string{
+		{"key1=value"},
+		{"kubernetes.io/hostname=worker0"},
+		{"disktype=ssd"},
+		{"key=path-to-value"},
+		{"keyNum=123"},
+	}
+	for _, vds := range validNodeSelectors {
+		_, err := NewNodeSelectors(vds)
+		assert.Nil(t, err)
+	}
+}
+
+func TestInvalidNodeSelectors(t *testing.T) {
+	validNodeSelectors := [][]string{
+		{"key1"},
+		{"kubernetes.io@hostname=worker0"},
+		{"key=path/to/value"},
+	}
+	for _, vds := range validNodeSelectors {
+		_, err := NewNodeSelectors(vds)
+		assert.NotNil(t, err)
+	}
+}
+
+func TestValueNodeSelectors(t *testing.T) {
+	nodeSelectorsArray := []string{
+		"key=value",
+		"kubernetes.io/hostname=worker0",
+	}
+	nodeSelectors, err := NewNodeSelectors(nodeSelectorsArray)
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(nodeSelectors))
+
+	assert.Equal(t, "value", nodeSelectors["key"])
+	assert.Equal(t, "worker0", nodeSelectors["kubernetes.io/hostname"])
+}