Allow user to specify Node Tolerations for cluster pods

This adds support to the operator for a user to be able to specify a set
of Node Tolerations to be applied to the pods in a cluster.

Author: accorvin

.travis/.travis.test-common.sh

@@ -330,6 +330,20 @@ testOverwriteLimits() {
   os::cmd::expect_success_and_text '${BIN} delete ${KIND} my-spark-cluster-overwritelim' '"my-spark-cluster-overwritelim" deleted'
 }
 
+testNodeTolerations() {
+  info
+  sleep 2
+  os::cmd::expect_success_and_text "${BIN} create -f $DIR/../examples/test/${CM}cluster-node-tolerations.yaml" '"?spark-cluster-with-tolerations"? created' && \
+  os::cmd::try_until_success "${BIN} get pod -l radanalytics.io/deployment=spark-cluster-with-tolerations-w -o=jsonpath='{.items[0].spec.tolerations}'" && \
+  local tolerations=`${BIN} get pod -l radanalytics.io/deployment=spark-cluster-with-tolerations-w -o='jsonpath="{.items[0].spec.tolerations}"' | sed 's/"//g'` && \
+  os::cmd::expect_success_and_text 'echo $tolerations' 'tolerationSeconds:60' && \
+  os::cmd::expect_success_and_text 'echo $tolerations' 'value:foo_value' && \
+  os::cmd::expect_success_and_text 'echo $tolerations' 'effect:NoExecute' && \
+  os::cmd::expect_success_and_text 'echo $tolerations' 'key:foo_key' && \
+  os::cmd::expect_success_and_text 'echo $tolerations' 'operator:Equal' && \
+  os::cmd::expect_success_and_text '${BIN} delete ${KIND} spark-cluster-with-tolerations' '"spark-cluster-with-tolerations" deleted'
+}
+
 testApp() {
   info
   [ "$CRD" = "0" ] && FOO="test/cm/" || FOO=""

.travis/.travis.test-oc-and-k8s.sh

@@ -50,6 +50,9 @@ run_tests() {
   sleep 5
 
   run_limit_request_tests || errorLogs
+  sleep 5
+
+  testNodeTolerations || errorLogs
 
   sleep 10
   testApp || appErrorLogs

README.md

@@ -76,6 +76,14 @@ You can see these being used in the *examples/test* directory.
 
 * *cpuLimit* and *memoryLimit* set limit values and take precedence over values taken from *cpu* and *memory* respectively
 
+# Node Tolerations for SparkCluster pods
+
+The operator supports specifying [Kubernetes node tolerations](https://kubernetes.io/docs/concepts/configuration/taint-and-toleration)
+which will be applied to all master and worker pods in a Spark cluster.
+You can see examples of this in use in the *examples/test* directory.
+
+* *nodeTolerations* specifies a list of Node Tolerations definitions that should
+  be applied to all master and worker nodes.
 
 ## Spark Applications
 

examples/test/cluster-node-tolerations.yaml

@@ -0,0 +1,15 @@
+apiVersion: radanalytics.io/v1
+kind: SparkCluster
+metadata:
+  name: spark-cluster-with-tolerations
+spec:
+  nodeTolerations:
+    - key: foo_key
+      operator: Equal
+      value: foo_value
+      effect: NoExecute
+      tolerationSeconds: 60
+  worker:
+    instances: "1"
+  master:
+    instances: "1"

examples/test/cm/cluster-node-tolerations.yaml

@@ -0,0 +1,18 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: spark-cluster-with-tolerations
+  labels:
+    radanalytics.io/kind: SparkCluster
+data:
+  config: |-
+    nodeTolerations:
+      - key: foo_key
+        operator: Equal
+        value: foo_value
+        effect: NoExecute
+        tolerationSeconds: 60
+    worker:
+      instances: "1"
+    master:
+      instances: "1"

src/main/java/io/radanalytics/operator/cluster/KubernetesSparkClusterDeployer.java

@@ -166,6 +166,9 @@ private ReplicationController getRCforMasterOrWorker(boolean isMaster, SparkClus
         podLabels.put(prefix + LabelsHelper.OPERATOR_POD_TYPE_LABEL, isMaster ? OPERATOR_TYPE_MASTER_LABEL : OPERATOR_TYPE_WORKER_LABEL);
         addLabels(podLabels, cluster, isMaster);
 
+        // pod tolerations
+        List<Toleration> tolerations = getTolerations(cluster, isMaster);
+
         PodTemplateSpecFluent.SpecNested<ReplicationControllerSpecFluent.TemplateNested<ReplicationControllerFluent.SpecNested<ReplicationControllerBuilder>>> rcBuilder = new ReplicationControllerBuilder().withNewMetadata()
                 .withName(podName).withLabels(labels)
                 .endMetadata()
@@ -178,7 +181,7 @@ private ReplicationController getRCforMasterOrWorker(boolean isMaster, SparkClus
                 )
                 .withSelector(selector)
                 .withNewTemplate().withNewMetadata().withLabels(podLabels).endMetadata()
-                .withNewSpec().withContainers(containerBuilder.build());
+                .withNewSpec().withTolerations(tolerations).withContainers(containerBuilder.build());
 
         ReplicationController rc = rcBuilder.endSpec().endTemplate().endSpec().build();
 
@@ -261,6 +264,16 @@ private void getLimitRequestValues(Boolean isMaster, Master m, Worker w, Map<Str
         Optional.ofNullable(isMaster ? m.getCpuRequest() : w.getCpuRequest()).ifPresent(cpuval -> requests.put("cpu", new Quantity(cpuval)));       
     }
 
+    private List<Toleration> getTolerations(SparkCluster cluster, boolean isMaster) {
+        List<Toleration> tolerations = new ArrayList<Toleration>();
+        List<NodeToleration> nodeTolerations = cluster.getNodeTolerations();
+        nodeTolerations.forEach(t -> {
+            tolerations.add(new Toleration(t.getEffect(), t.getKey(), t.getOperator(), (long) t.getTolerationSeconds(), t.getValue()));
+        });
+        return tolerations;
+
+    }
+
     private ContainerBuilder augmentContainerBuilder(SparkCluster cluster, ContainerBuilder builder, boolean isMaster) {
         Master m = null;
         Worker w = null;

src/main/resources/schema/sparkCluster.json

@@ -98,6 +98,36 @@
         }
       }
     },
+    "nodeTolerations": {
+      "type": "array",
+      "items": {
+        "type": "object",
+        "properties": {
+          "key": {
+            "type": "string"
+          },
+          "operator": {
+            "type": "string"
+          },
+          "value": {
+            "type": "string"
+          },
+          "effect": {
+            "type": "string"
+          },
+          "tolerationSeconds": {
+            "type": "integer",
+            "default": null
+          }
+        },
+        "required": [
+          "key",
+          "operator",
+          "value",
+          "effect"
+        ]
+      }
+    },
     "mavenDependencies": {
       "type": "array",
       "items": {