[Heron-3723] Add support for Persistent Volumes for stateful storage

deploy/kubernetes/general/apiserver.yaml

@@ -92,6 +92,7 @@ spec:
               -D heron.statefulstorage.classname=org.apache.heron.statefulstorage.dlog.DlogStorage
               -D heron.statefulstorage.dlog.namespace.uri=distributedlog://zookeeper:2181/heron
               -D heron.kubernetes.pod.template.configmap.disabled=false
+              -D heron.kubernetes.persistent.volume.claims.cli.disabled=false
 
 ---
 apiVersion: v1

deploy/kubernetes/helm/templates/tools.yaml

@@ -159,6 +159,7 @@ spec:
               {{- end }}
               -D heron.kubernetes.resource.request.mode={{ .Values.topologyResourceRequestMode }}
               -D heron.kubernetes.pod.template.configmap.disabled={{ .Values.disablePodTemplates }}
+              -D heron.kubernetes.persistent.volume.claims.cli.disabled={{ .Values.disablePersistentVolumeMountsCLI }}
           envFrom:
             - configMapRef:
                 name: {{ .Release.Name }}-tools-config
@@ -321,3 +322,13 @@ rules:
   verbs:
   - get
   - list
+- apiGroups:
+  - ""
+  resources:
+  - persistentvolumeclaims
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - deletecollection

deploy/kubernetes/helm/values.yaml.template

@@ -77,6 +77,9 @@ packing: RoundRobin # ResourceCompliantRR, FirstFitDecreasing
 # Support for ConfigMap mounted PodTemplates
 disablePodTemplates: false
 
+# Support for Dynamic Persistent Volume Mounts from CLI input
+disablePersistentVolumeMountsCLI: false
+
 # Number of replicas for storage bookies, memory and storage requirements
 bookieReplicas: 3
 bookieCpuMin: 100m

deploy/kubernetes/minikube/apiserver.yaml

@@ -83,6 +83,7 @@ spec:
               -D heron.statefulstorage.classname=org.apache.heron.statefulstorage.dlog.DlogStorage
               -D heron.statefulstorage.dlog.namespace.uri=distributedlog://zookeeper:2181/heronbkdl
               -D heron.kubernetes.pod.template.configmap.disabled=false
+              -D heron.kubernetes.persistent.volume.claims.cli.disabled=false
 
 ---
 apiVersion: v1

heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesConstants.java

@@ -52,6 +52,7 @@ private KubernetesConstants() {
   public static final String LABEL_APP = "app";
   public static final String LABEL_APP_VALUE = "heron";
   public static final String LABEL_TOPOLOGY = "topology";
+  public static final String LABEL_ON_DEMAND = "onDemand";
 
   // prometheus annotation keys
   public static final String ANNOTATION_PROMETHEUS_SCRAPE = "prometheus.io/scrape";
@@ -88,11 +89,13 @@ private KubernetesConstants() {
   public static final String JOB_LINK =
       "/api/v1/namespaces/kube-system/services/kubernetes-dashboard/proxy/#/pod";
 
-
   public static final Pattern VALID_POD_NAME_REGEX =
       Pattern.compile("[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*",
           Pattern.CASE_INSENSITIVE);
 
+  public static final Pattern VALID_LOWERCASE_RFC_1123_REGEX =
+      Pattern.compile("[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*");
+
   public static final List<String> VALID_IMAGE_PULL_POLICIES = Collections.unmodifiableList(
       Arrays.asList(
           "IfNotPresent",
@@ -107,4 +110,14 @@ private KubernetesConstants() {
           "node.kubernetes.io/unreachable"
       )
   );
+
+  enum VolumeClaimTemplateConfigKeys {
+    claimName,
+    storageClassName,
+    sizeLimit,
+    accessModes,
+    volumeMode,
+    path,               // Added to container.
+    subPath,            // Added to container.
+  }
 }

heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesContext.java

@@ -23,7 +23,11 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.regex.Matcher;
 
+import org.apache.heron.scheduler.TopologySubmissionException;
 import org.apache.heron.spi.common.Config;
 import org.apache.heron.spi.common.Context;
 
@@ -109,6 +113,13 @@ public enum KubernetesResourceRequestMode {
   public static final String KUBERNETES_POD_SECRET_KEY_REF_PREFIX =
       "heron.kubernetes.pod.secretKeyRef.";
 
+  // Persistent Volume Claims
+  public static final String KUBERNETES_PERSISTENT_VOLUME_CLAIMS_CLI_DISABLED =
+      "heron.kubernetes.persistent.volume.claims.cli.disabled";
+  // heron.kubernetes.volumes.persistentVolumeClaim.VOLUME_NAME.OPTION=OPTION_VALUE
+  public static final String KUBERNETES_VOLUME_CLAIM_PREFIX =
+      "heron.kubernetes.volumes.persistentVolumeClaim.";
+
   private KubernetesContext() {
   }
 
@@ -211,6 +222,80 @@ public static Map<String, String> getPodSecretKeyRefs(Config config) {
     return getConfigItemsByPrefix(config, KUBERNETES_POD_SECRET_KEY_REF_PREFIX);
   }
 
+  public static boolean getPersistentVolumeClaimDisabled(Config config) {
+    final String disabled = config.getStringValue(KUBERNETES_PERSISTENT_VOLUME_CLAIMS_CLI_DISABLED);
+    return "true".equalsIgnoreCase(disabled);
+  }
+
+  /**
+   * Collects parameters form the <code>CLI</code> and generates a mapping between <code>Volumes</code>
+   * and their configuration <code>key-value</code> pairs.
+   * @param config Contains the configuration options collected from the <code>CLI</code>.
+   * @return A mapping between <code>Volumes</code> and their configuration <code>key-value</code> pairs.
+   * Will return an empty list if there are no Volume Claim Templates to be generated.
+   */
+  public static Map<String, Map<KubernetesConstants.VolumeClaimTemplateConfigKeys, String>>
+      getVolumeClaimTemplates(Config config) {
+    final Logger LOG = Logger.getLogger(V1Controller.class.getName());
+
+    final Set<String> completeConfigParam = getConfigKeys(config, KUBERNETES_VOLUME_CLAIM_PREFIX);
+    final int prefixLength = KUBERNETES_VOLUME_CLAIM_PREFIX.length();
+    final int volumeNameIdx = 0;
+    final int optionIdx = 1;
+    final Matcher matcher = KubernetesConstants.VALID_LOWERCASE_RFC_1123_REGEX.matcher("");
+
+    final Map<String, Map<KubernetesConstants.VolumeClaimTemplateConfigKeys, String>> volumes
+        = new HashMap<>();
+
+    try {
+      for (String param : completeConfigParam) {
+        final String[] tokens = param.substring(prefixLength).split("\\.");
+        final String volumeName = tokens[volumeNameIdx];
+        final KubernetesConstants.VolumeClaimTemplateConfigKeys key =
+            KubernetesConstants.VolumeClaimTemplateConfigKeys.valueOf(tokens[optionIdx]);
+        final String value = config.getStringValue(param);
+
+        Map<KubernetesConstants.VolumeClaimTemplateConfigKeys, String> volume =
+            volumes.get(volumeName);
+        if (volume == null) {
+          // Validate new Volume Names.
+          if (!matcher.reset(volumeName).matches()) {
+            throw new TopologySubmissionException(
+                String.format("Volume name `%s` does not match lowercase RFC-1123 pattern",
+                    volumeName));
+          }
+          volume = new HashMap<>();
+          volumes.put(volumeName, volume);
+        }
+
+        /* Validate Claim and Storage Class names.
+          [1] `claimNameNotOnDemand`: checks for a `claimName` which is not `OnDemand`.
+          [2] `storageClassName`: Check if it is the provided `option`.
+          Conditions [1] OR [2] are True, then...
+          [3] Check for a valid lowercase RFC-1123 pattern.
+         */
+        boolean claimNameNotOnDemand =
+            KubernetesConstants.VolumeClaimTemplateConfigKeys.claimName.equals(key)
+                && !KubernetesConstants.LABEL_ON_DEMAND.equalsIgnoreCase(value);
+        if ((claimNameNotOnDemand // [1]
+            ||
+            KubernetesConstants.VolumeClaimTemplateConfigKeys.storageClassName.equals(key)) // [2]
+            && !matcher.reset(value).matches()) { // [3]
+          throw new TopologySubmissionException(
+              String.format("Option `%s` value `%s` does not match lowercase RFC-1123 pattern",
+                  key, value));
+        }
+
+        volume.put(key, value);
+      }
+    } catch (IndexOutOfBoundsException | IllegalArgumentException e) {
+      final String message = "Invalid Persistent Volume Claim CLI parameter provided";
+      LOG.log(Level.CONFIG, message);
+      throw new TopologySubmissionException(message);
+    }
+    return volumes;
+  }
+
   static Set<String> getConfigKeys(Config config, String keyPrefix) {
     Set<String> annotations = new HashSet<>();
     for (String s : config.getKeySet()) {

heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesUtils.java

@@ -122,4 +122,27 @@ protected List<T> mergeListsDedupe(List<T> primaryList, List<T> secondaryList,
       }
     }
   }
+
+  /**
+   * Generic testing class for test runners in Kubernetes Scheduler.
+   * @param <T1> Test input object type.
+   * @param <T2> Expected test object type.
+   */
+  static class TestTuple<T1, T2> {
+    public final String description;
+    public final T1 input;
+    public final T2 expected;
+
+    /**
+     * Configure the test object.
+     * @param description Description of the test to be run.
+     * @param input Input test case.
+     * @param expected Expected output form test.
+     */
+    TestTuple(String description, T1 input, T2 expected) {
+      this.description = description;
+      this.expected = expected;
+      this.input = input;
+    }
+  }
 }