kubernetes executor: support node selector and tolerations

charts/libs/runner/Chart.yaml

@@ -15,7 +15,7 @@ type: library
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.1.0
+version: 0.1.1
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to

charts/libs/runner/README.md

@@ -1,6 +1,6 @@
 # runner
 
-![Version: 0.1.0](https://img.shields.io/badge/Version-0.1.0-informational?style=flat-square) ![Type: library](https://img.shields.io/badge/Type-library-informational?style=flat-square) ![AppVersion: 1.16.0](https://img.shields.io/badge/AppVersion-1.16.0-informational?style=flat-square)
+![Version: 0.1.1](https://img.shields.io/badge/Version-0.1.1-informational?style=flat-square) ![Type: library](https://img.shields.io/badge/Type-library-informational?style=flat-square) ![AppVersion: 1.16.0](https://img.shields.io/badge/AppVersion-1.16.0-informational?style=flat-square)
 
 A Helm chart library for otf runner resources and configuration.
 
@@ -18,8 +18,10 @@ A Helm chart library for otf runner resources and configuration.
 | kubernetesLabels | list | `[]` | Set additional labels on kubernetes jobs. Name and value are separated by an equals sign, e.g. `foo=bar`. |
 | kubernetesLimitCPU | string | `nil` | Set a CPU limit for kubernetes jobs. |
 | kubernetesLimitMemory | string | `nil` | Set a memory limit for kubernetes jobs. |
+| kubernetesNodeSelector | list | `[]` | Constrain kubernetes jobs to nodes with these labels. Name and value are separated by an equals sign, e.g. `foo=bar`. |
 | kubernetesRequestCPU | string | `nil` | Set the requested CPU resources for kubernetes jobs. |
 | kubernetesRequestMemory | string | `nil` | Set the requested memory resources for kubernetes jobs. |
 | kubernetesTTLAfterFinish | string | `nil` | Delete finished kubernetes jobs after this duration. |
+| kubernetesTolerations | list | `[]` | Add tolerations to kubernetes jobs, in the format `key[=value]:effect` (operator is `Exists` when no value is given, otherwise `Equal`), e.g. `dedicated=terraform:NoSchedule`. |
 | pluginCache | bool | `nil` | Enable shared provider plugin cache for terraform providers. Note this is only concurrency safe in opentofu 1.10.0 and greater. |
 

charts/libs/runner/templates/_envs.yaml

@@ -16,6 +16,14 @@
 - name: OTF_KUBERNETES_LABELS
   value: {{ . | join "," }}
 {{- end }}
+{{- with .kubernetesNodeSelector }}
+- name: OTF_KUBERNETES_NODE_SELECTOR
+  value: {{ . | join "," }}
+{{- end }}
+{{- with .kubernetesTolerations }}
+- name: OTF_KUBERNETES_TOLERATIONS
+  value: {{ . | join "," }}
+{{- end }}
 {{- with .kubernetesRequestCPU }}
 - name: OTF_KUBERNETES_REQUEST_CPU
   value: {{ . }}

charts/libs/runner/values.yaml

@@ -12,6 +12,10 @@ kubernetesLimitCPU: ~
 kubernetesLimitMemory: ~
 # -- Set additional labels on kubernetes jobs. Name and value are separated by an equals sign, e.g. `foo=bar`.
 kubernetesLabels: []
+# -- Constrain kubernetes jobs to nodes with these labels. Name and value are separated by an equals sign, e.g. `foo=bar`.
+kubernetesNodeSelector: []
+# -- Add tolerations to kubernetes jobs, in the format `key[=value]:effect` (operator is `Exists` when no value is given, otherwise `Equal`), e.g. `dedicated=terraform:NoSchedule`.
+kubernetesTolerations: []
 # -- Set the kubernetes job image. Defaults to `leg100/otf-job` tagged with the current OTF version. See [docs](https://docs.otf.ninja/config/flags/#-kubernetes-job-image).
 kubernetesJobImage: ""
 # -- Delete finished kubernetes jobs after this duration.

charts/otf-agent/Chart.yaml

@@ -2,7 +2,7 @@ apiVersion: v2
 name: otf-agent
 description: Helm chart for the OTF agent
 type: application
-version: 0.1.33
+version: 0.1.34
 appVersion: 0.6.3
 maintainers:
   - name: Louis Garman

charts/otf-agent/README.md

@@ -1,6 +1,6 @@
 # Helm Chart for `otf-agent`
 
-![Version: 0.1.33](https://img.shields.io/badge/Version-0.1.33-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.6.3](https://img.shields.io/badge/AppVersion-0.6.3-informational?style=flat-square)
+![Version: 0.1.34](https://img.shields.io/badge/Version-0.1.34-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.6.3](https://img.shields.io/badge/AppVersion-0.6.3-informational?style=flat-square)
 
 Installs the [otf agent](https://docs.otf.ninja/runners/).
 
@@ -93,8 +93,10 @@ address=10.244.0.18 agent.pool_id=apool-5b90443ed82ef769
 | runner.kubernetesLabels | list | `[]` | Set additional labels on kubernetes jobs. Name and value are separated by an equals sign, e.g. `foo=bar`. |
 | runner.kubernetesLimitCPU | string | `nil` | Set a CPU limit for kubernetes jobs. |
 | runner.kubernetesLimitMemory | string | `nil` | Set a memory limit for kubernetes jobs. |
+| runner.kubernetesNodeSelector | list | `[]` | Constrain kubernetes jobs to nodes with these labels. Name and value are separated by an equals sign, e.g. `foo=bar`. |
 | runner.kubernetesRequestCPU | string | `nil` | Set the requested CPU resources for kubernetes jobs. |
 | runner.kubernetesRequestMemory | string | `nil` | Set the requested memory resources for kubernetes jobs. |
 | runner.kubernetesTTLAfterFinish | string | `nil` | Delete finished kubernetes jobs after this duration. |
+| runner.kubernetesTolerations | list | `[]` | Add tolerations to kubernetes jobs, in the format `key[=value]:effect` (operator is `Exists` when no value is given, otherwise `Equal`), e.g. `dedicated=terraform:NoSchedule`. |
 | runner.pluginCache | bool | `nil` | Enable shared provider plugin cache for terraform providers. Note this is only concurrency safe in opentofu 1.10.0 and greater. |
 

charts/otfd/Chart.yaml

@@ -13,7 +13,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.4.2
+version: 0.4.3
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.

charts/otfd/README.md

@@ -1,6 +1,6 @@
 # Helm chart for `otfd`
 
-![Version: 0.4.2](https://img.shields.io/badge/Version-0.4.2-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.6.3](https://img.shields.io/badge/AppVersion-0.6.3-informational?style=flat-square)
+![Version: 0.4.3](https://img.shields.io/badge/Version-0.4.3-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 0.6.3](https://img.shields.io/badge/AppVersion-0.6.3-informational?style=flat-square)
 
 Installs the [otf](https://github.com/leg100/otf) daemon.
 
@@ -115,8 +115,10 @@ Note: you should only use this for testing purposes.
 | runner.kubernetesLabels | list | `[]` | Set additional labels on kubernetes jobs. Name and value are separated by an equals sign, e.g. `foo=bar`. |
 | runner.kubernetesLimitCPU | string | `nil` | Set a CPU limit for kubernetes jobs. |
 | runner.kubernetesLimitMemory | string | `nil` | Set a memory limit for kubernetes jobs. |
+| runner.kubernetesNodeSelector | list | `[]` | Constrain kubernetes jobs to nodes with these labels. Name and value are separated by an equals sign, e.g. `foo=bar`. |
 | runner.kubernetesRequestCPU | string | `nil` | Set the requested CPU resources for kubernetes jobs. |
 | runner.kubernetesRequestMemory | string | `nil` | Set the requested memory resources for kubernetes jobs. |
 | runner.kubernetesTTLAfterFinish | string | `nil` | Delete finished kubernetes jobs after this duration. |
+| runner.kubernetesTolerations | list | `[]` | Add tolerations to kubernetes jobs, in the format `key[=value]:effect` (operator is `Exists` when no value is given, otherwise `Equal`), e.g. `dedicated=terraform:NoSchedule`. |
 | runner.pluginCache | bool | `nil` | Enable shared provider plugin cache for terraform providers. Note this is only concurrency safe in opentofu 1.10.0 and greater. |
 

docs/docs/config/flags.md

@@ -170,6 +170,20 @@ Set a memory limit for kubernetes jobs.
 
 Set additional labels on kubernetes jobs. Name and value are separated by an equals sign, e.g. `foo=bar`.
 
+## `--kubernetes-node-selector`
+
+* System: `otfd`, `otf-agent`
+* Default: `""`
+
+Constrain kubernetes jobs to nodes with these labels. Name and value are separated by an equals sign, e.g. `foo=bar`. Specify multiple selectors separated by a comma. Combine with `--kubernetes-tolerations` to schedule jobs onto a dedicated, tainted node pool.
+
+## `--kubernetes-tolerations`
+
+* System: `otfd`, `otf-agent`
+* Default: `""`
+
+Add tolerations to kubernetes jobs, in the format `key[=value]:effect`. The operator is `Exists` when no value is given, otherwise `Equal`; omit the effect to tolerate all effects for the key. `effect`, when given, must be one of `NoSchedule`, `PreferNoSchedule` or `NoExecute`, e.g. `dedicated=terraform:NoSchedule`. Specify multiple tolerations separated by a comma.
+
 ## `--kubernetes-ttl-after-finish`
 
 * System: `otfd`, `otf-agent`

docs/docs/executors.md

@@ -26,6 +26,8 @@ This executor is only functional when `otfd` or `otf-agent` is deployed via the
 There are a number of flags that customise the jobs:
 
 * [`--kubernetes-job-image`](config/flags.md#-kubernetes-job-image)
+* [`--kubernetes-node-selector`](config/flags.md#-kubernetes-node-selector)
+* [`--kubernetes-tolerations`](config/flags.md#-kubernetes-tolerations)
 * [`--kubernetes-request-cpu`](config/flags.md#-kubernetes-request-cpu)
 * [`--kubernetes-request-memory`](config/flags.md#-kubernetes-request-memory)
 * [`--kubernetes-ttl-after-finish`](config/flags.md#-kubernetes-ttl-after-finish)

internal/runner/executor_kube.go

@@ -72,7 +72,6 @@ var (
 	defaultKubeConfig kubeConfig
 )
 
-
 type kubeConfig struct {
 	Namespace      string
 	Image          string
@@ -87,6 +86,8 @@ type kubeConfig struct {
 	limitCPU      *k8sresource.Quantity
 	limitMemory   *k8sresource.Quantity
 	labels        map[string]string
+	nodeSelector  map[string]string
+	tolerations   []corev1.Toleration
 
 	flags kubeConfigFlags
 }
@@ -99,6 +100,8 @@ type kubeConfigFlags struct {
 	RequestMemory string
 	LimitCPU      string
 	LimitMemory   string
+	NodeSelector  []string
+	Tolerations   []string
 }
 
 func registerKubeFlags(flags *pflag.FlagSet, cfg *kubeConfig) {
@@ -109,6 +112,34 @@ func registerKubeFlags(flags *pflag.FlagSet, cfg *kubeConfig) {
 	flags.StringVar(&cfg.flags.LimitCPU, "kubernetes-limit-cpu", cfg.flags.LimitCPU, "CPU limit for kubernetes job.")
 	flags.StringVar(&cfg.flags.LimitMemory, "kubernetes-limit-memory", cfg.flags.LimitMemory, "Memory limit for kubernetes job.")
 	flags.StringSliceVar(&cfg.flags.Labels, "kubernetes-labels", cfg.flags.Labels, "Set additional labels on kubernetes jobs. Name and value are separated by an equals sign, e.g. `foo=bar`.")
+	flags.StringSliceVar(&cfg.flags.NodeSelector, "kubernetes-node-selector", cfg.flags.NodeSelector, "Constrain kubernetes jobs to nodes with these labels. Name and value are separated by an equals sign, e.g. `foo=bar`.")
+	flags.StringSliceVar(&cfg.flags.Tolerations, "kubernetes-tolerations", cfg.flags.Tolerations, "Add tolerations to kubernetes jobs, in the format `key[=value]:effect` (operator is `Exists` when no value is given, otherwise `Equal`), e.g. `dedicated=terraform:NoSchedule`. Omit the effect to tolerate all effects for the key.")
+}
+
+// parseToleration parses a toleration from the format key[=value]:effect,
+// mirroring the syntax used by `kubectl taint`. The operator is Exists when no
+// value is supplied, otherwise Equal. An empty effect tolerates all effects.
+func parseToleration(s string) (corev1.Toleration, error) {
+	keyValue, effect, _ := strings.Cut(s, ":")
+	switch corev1.TaintEffect(effect) {
+	case "", corev1.TaintEffectNoSchedule, corev1.TaintEffectPreferNoSchedule, corev1.TaintEffectNoExecute:
+	default:
+		return corev1.Toleration{}, fmt.Errorf("invalid toleration effect %q: must be one of NoSchedule, PreferNoSchedule, NoExecute", effect)
+	}
+
+	toleration := corev1.Toleration{Effect: corev1.TaintEffect(effect)}
+	if key, value, ok := strings.Cut(keyValue, "="); ok {
+		toleration.Key = key
+		toleration.Value = value
+		toleration.Operator = corev1.TolerationOpEqual
+	} else {
+		toleration.Key = keyValue
+		toleration.Operator = corev1.TolerationOpExists
+	}
+	if toleration.Key == "" {
+		return corev1.Toleration{}, fmt.Errorf("invalid toleration %q: key must not be empty", s)
+	}
+	return toleration, nil
 }
 
 type kubeExecutor struct {
@@ -177,6 +208,23 @@ func newKubeExecutor(
 		executor.Config.labels[k] = v
 	}
 
+	executor.Config.nodeSelector = make(map[string]string)
+	for _, selector := range kubeConfig.flags.NodeSelector {
+		k, v, ok := strings.Cut(selector, "=")
+		if !ok {
+			return nil, fmt.Errorf("invalid node selector: must be in format name=value")
+		}
+		executor.Config.nodeSelector[k] = v
+	}
+
+	for _, t := range kubeConfig.flags.Tolerations {
+		toleration, err := parseToleration(t)
+		if err != nil {
+			return nil, err
+		}
+		executor.Config.tolerations = append(executor.Config.tolerations, toleration)
+	}
+
 	// assume running in-cluster; otherwise use config path
 	config, err := rest.InClusterConfig()
 	if errors.Is(err, rest.ErrNotInCluster) {
@@ -265,6 +313,8 @@ func (s *kubeExecutor) SpawnOperation(ctx context.Context, _ *errgroup.Group, jo
 				Spec: corev1.PodSpec{
 					ServiceAccountName: s.Config.ServiceAccount,
 					RestartPolicy:      corev1.RestartPolicyNever,
+					NodeSelector:       s.Config.nodeSelector,
+					Tolerations:        s.Config.tolerations,
 					Resources: &corev1.ResourceRequirements{
 						Requests: corev1.ResourceList{
 							corev1.ResourceCPU:    s.Config.requestCPU,

internal/runner/executor_kube_test.go

@@ -74,13 +74,81 @@ func TestNewKubeExecutor(t *testing.T) {
 		)
 		assert.Error(t, err)
 	})
+
+	t.Run("with node selector", func(t *testing.T) {
+		cfg := defaultKubeConfig
+		cfg.flags.NodeSelector = []string{"foo=bar", "coo=boo"}
+
+		executor, err := newKubeExecutor(
+			logr.Discard(),
+			defaultOperationConfig(),
+			cfg,
+		)
+		require.NoError(t, err)
+		assert.Equal(t, map[string]string{"foo": "bar", "coo": "boo"}, executor.Config.nodeSelector)
+	})
+
+	t.Run("with invalid node selector", func(t *testing.T) {
+		cfg := defaultKubeConfig
+		cfg.flags.NodeSelector = []string{"foobar"}
+
+		_, err := newKubeExecutor(
+			logr.Discard(),
+			defaultOperationConfig(),
+			cfg,
+		)
+		assert.Error(t, err)
+	})
+
+	t.Run("with tolerations", func(t *testing.T) {
+		cfg := defaultKubeConfig
+		cfg.flags.Tolerations = []string{"dedicated=terraform:NoSchedule", "spot:NoExecute", "gpu"}
+
+		executor, err := newKubeExecutor(
+			logr.Discard(),
+			defaultOperationConfig(),
+			cfg,
+		)
+		require.NoError(t, err)
+		assert.Equal(t, []corev1.Toleration{
+			{Key: "dedicated", Operator: corev1.TolerationOpEqual, Value: "terraform", Effect: corev1.TaintEffectNoSchedule},
+			{Key: "spot", Operator: corev1.TolerationOpExists, Effect: corev1.TaintEffectNoExecute},
+			{Key: "gpu", Operator: corev1.TolerationOpExists},
+		}, executor.Config.tolerations)
+	})
+
+	t.Run("with invalid toleration effect", func(t *testing.T) {
+		cfg := defaultKubeConfig
+		cfg.flags.Tolerations = []string{"dedicated=terraform:Bogus"}
+
+		_, err := newKubeExecutor(
+			logr.Discard(),
+			defaultOperationConfig(),
+			cfg,
+		)
+		assert.Error(t, err)
+	})
+
+	t.Run("with empty toleration key", func(t *testing.T) {
+		cfg := defaultKubeConfig
+		cfg.flags.Tolerations = []string{":NoSchedule"}
+
+		_, err := newKubeExecutor(
+			logr.Discard(),
+			defaultOperationConfig(),
+			cfg,
+		)
+		assert.Error(t, err)
+	})
 }
 
 func TestKubeExecutor_SpawnOperation(t *testing.T) {
 	cfg := defaultKubeConfig
 	cfg.flags.Labels = []string{"foo=bar"}
 	cfg.flags.LimitCPU = "3000m"
 	cfg.flags.LimitMemory = "512Mi"
+	cfg.flags.NodeSelector = []string{"dedicated=terraform"}
+	cfg.flags.Tolerations = []string{"dedicated=terraform:NoSchedule"}
 
 	executor, err := newKubeExecutor(
 		logr.Discard(),
@@ -123,6 +191,12 @@ func TestKubeExecutor_SpawnOperation(t *testing.T) {
 	assert.Equal(t, wantLabels, jobsClient.job.Labels)
 	assert.Equal(t, wantLabels, secretsClient.secret.Labels)
 	assert.Equal(t, map[string]string{"jobToken": "token"}, secretsClient.secret.StringData)
+
+	podSpec := jobsClient.job.Spec.Template.Spec
+	assert.Equal(t, map[string]string{"dedicated": "terraform"}, podSpec.NodeSelector)
+	assert.Equal(t, []corev1.Toleration{
+		{Key: "dedicated", Operator: corev1.TolerationOpEqual, Value: "terraform", Effect: corev1.TaintEffectNoSchedule},
+	}, podSpec.Tolerations)
 }
 
 type fakeSecretsClient struct {