[Heron-3724] Separate the Manager and Executors.

deploy/kubernetes/general/apiserver.yaml

@@ -91,7 +91,7 @@ spec:
               -D heron.uploader.dlog.topologies.namespace.uri=distributedlog://zookeeper:2181/heron
               -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.pod.template.disabled=false
               -D heron.kubernetes.persistent.volume.claims.cli.disabled=false
 
 ---

deploy/kubernetes/helm/templates/tools.yaml

@@ -158,7 +158,7 @@ spec:
               -D heron.class.repacking.algorithm=org.apache.heron.packing.binpacking.FirstFitDecreasingPacking
               {{- end }}
               -D heron.kubernetes.resource.request.mode={{ .Values.topologyResourceRequestMode }}
-              -D heron.kubernetes.pod.template.configmap.disabled={{ .Values.disablePodTemplates }}
+              -D heron.kubernetes.pod.template.disabled={{ .Values.disablePodTemplates }}
               -D heron.kubernetes.persistent.volume.claims.cli.disabled={{ .Values.disablePersistentVolumeMountsCLI }}
           envFrom:
             - configMapRef:
@@ -297,6 +297,7 @@ rules:
   - patch
   - update
   - watch
+  - deletecollection
 - apiGroups:
   - ""
   resources:

deploy/kubernetes/minikube/apiserver.yaml

@@ -82,7 +82,7 @@ spec:
               -D heron.uploader.dlog.topologies.namespace.uri=distributedlog://zookeeper:2181/heronbkdl
               -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.pod.template.disabled=false
               -D heron.kubernetes.persistent.volume.claims.cli.disabled=false
 
 ---

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

@@ -36,6 +36,7 @@ private KubernetesConstants() {
   public static final String CPU = "cpu";
 
   public static final String EXECUTOR_NAME = "executor";
+  public static final String MANAGER_NAME = "manager";
 
   // container env constants
   public static final String ENV_HOST = "HOST";
@@ -89,6 +90,9 @@ private KubernetesConstants() {
   public static final String JOB_LINK =
       "/api/v1/namespaces/kube-system/services/kubernetes-dashboard/proxy/#/pod";
 
+  protected static final String JSON_PATCH_STATEFUL_SET_REPLICAS_FORMAT =
+      "[{\"op\":\"replace\",\"path\":\"/spec/replicas\",\"value\":%d}]";
+
   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);

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

@@ -86,11 +86,11 @@ public enum KubernetesResourceRequestMode {
   public static final String KUBERNETES_VOLUME_AWS_EBS_FS_TYPE =
       "heron.kubernetes.volume.awsElasticBlockStore.fsType";
 
-  // pod template configmap
-  public static final String KUBERNETES_POD_TEMPLATE_CONFIGMAP_NAME =
-      "heron.kubernetes.pod.template.configmap.name";
-  public static final String KUBERNETES_POD_TEMPLATE_CONFIGMAP_DISABLED =
-      "heron.kubernetes.pod.template.configmap.disabled";
+  // Pod Template ConfigMap: heron.kubernetes.[executor | manager].pod.template
+  public static final String KUBERNETES_POD_TEMPLATE_LOCATION =
+      "heron.kubernetes.%s.pod.template";
+  public static final String KUBERNETES_POD_TEMPLATE_DISABLED =
+      "heron.kubernetes.pod.template.disabled";
 
   // container mount volume mount keys
   public static final String KUBERNETES_CONTAINER_VOLUME_MOUNT_NAME =
@@ -116,9 +116,15 @@ public enum KubernetesResourceRequestMode {
   // 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
+  // heron.kubernetes.[executor | manager].volumes.persistentVolumeClaim.VOLUME_NAME.OPTION=VALUE
   public static final String KUBERNETES_VOLUME_CLAIM_PREFIX =
-      "heron.kubernetes.volumes.persistentVolumeClaim.";
+      "heron.kubernetes.%s.volumes.persistentVolumeClaim.";
+  // heron.kubernetes.[executor | manager].limits.OPTION=VALUE
+  public static final String KUBERNETES_RESOURCE_LIMITS_PREFIX =
+      "heron.kubernetes.%s.limits.";
+  // heron.kubernetes.[executor | manager].requests.OPTION=VALUE
+  public static final String KUBERNETES_RESOURCE_REQUESTS_PREFIX =
+      "heron.kubernetes.%s.requests.";
 
   private KubernetesContext() {
   }
@@ -189,12 +195,14 @@ static String getContainerVolumeMountPath(Config config) {
     return config.getStringValue(KUBERNETES_CONTAINER_VOLUME_MOUNT_PATH);
   }
 
-  public static String getPodTemplateConfigMapName(Config config) {
-    return config.getStringValue(KUBERNETES_POD_TEMPLATE_CONFIGMAP_NAME);
+  public static String getPodTemplateConfigMapName(Config config, boolean isExecutor) {
+    final String key = String.format(KUBERNETES_POD_TEMPLATE_LOCATION,
+        isExecutor ? KubernetesConstants.EXECUTOR_NAME : KubernetesConstants.MANAGER_NAME);
+    return config.getStringValue(key);
   }
 
-  public static boolean getPodTemplateConfigMapDisabled(Config config) {
-    final String disabled = config.getStringValue(KUBERNETES_POD_TEMPLATE_CONFIGMAP_DISABLED);
+  public static boolean getPodTemplateDisabled(Config config) {
+    final String disabled = config.getStringValue(KUBERNETES_POD_TEMPLATE_DISABLED);
     return "true".equalsIgnoreCase(disabled);
   }
 
@@ -222,6 +230,18 @@ public static Map<String, String> getPodSecretKeyRefs(Config config) {
     return getConfigItemsByPrefix(config, KUBERNETES_POD_SECRET_KEY_REF_PREFIX);
   }
 
+  public static Map<String, String> getResourceLimits(Config config, boolean isExecutor) {
+    final String key = String.format(KUBERNETES_RESOURCE_LIMITS_PREFIX,
+        isExecutor ? KubernetesConstants.EXECUTOR_NAME : KubernetesConstants.MANAGER_NAME);
+    return getConfigItemsByPrefix(config, key);
+  }
+
+  public static Map<String, String> getResourceRequests(Config config, boolean isExecutor) {
+    final String key = String.format(KUBERNETES_RESOURCE_REQUESTS_PREFIX,
+        isExecutor ? KubernetesConstants.EXECUTOR_NAME : KubernetesConstants.MANAGER_NAME);
+    return getConfigItemsByPrefix(config, key);
+  }
+
   public static boolean getPersistentVolumeClaimDisabled(Config config) {
     final String disabled = config.getStringValue(KUBERNETES_PERSISTENT_VOLUME_CLAIMS_CLI_DISABLED);
     return "true".equalsIgnoreCase(disabled);
@@ -231,15 +251,18 @@ public static boolean getPersistentVolumeClaimDisabled(Config config) {
    * 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>.
+   * @param isExecutor Flag used to collect CLI commands for the <code>executor</code> and <code>manager</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) {
+      getVolumeClaimTemplates(Config config, boolean isExecutor) {
     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 String prefixKey = String.format(KUBERNETES_VOLUME_CLAIM_PREFIX,
+        isExecutor ? KubernetesConstants.EXECUTOR_NAME : KubernetesConstants.MANAGER_NAME);
+    final Set<String> completeConfigParam = getConfigKeys(config, prefixKey);
+    final int prefixLength = prefixKey.length();
     final int volumeNameIdx = 0;
     final int optionIdx = 1;
     final Matcher matcher = KubernetesConstants.VALID_LOWERCASE_RFC_1123_REGEX.matcher("");