Repository: apache/incubator-heron (language: python)

Issue:
Add support for Pod Templates which are read from a ConfigMap. The ConfigMap name can be passed in as a config-property
This feature request is to enhance the Kubernetes Scheduler to add support for Pod Templates (similar to [Spark's Kubernetes feature](https://spark.apache.org/docs/latest/running-on-kubernetes.html#pod-template))

By providing a Pod Template as a ConfigMap, the Heron Kubernetes Scheduler can retrieve the template and provide it as a base starting point for Pod definition.


Discussion / hints:
This is the Spark Scala step that is used to create the Pod spec based on a ConfigMap PodTemplate.

https://github.com/apache/spark/blob/de59e01aa4853ef951da080c0d1908d53d133ebe/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/PodTemplateConfigMapStep.scala
Hi @nicknezis, I have started poking around in the code base for `Heron` again and I am trying to wrap my head around what needs to happen here. Again, I do not have any experience in the codebase.

> By providing a Pod Template as a ConfigMap, the Heron Kubernetes Scheduler can retrieve the template and provide it as a base starting point for Pod definition.

This appears to be setting up the `Config` on the Scheduler:
https://github.com/apache/incubator-heron/blob/4d9a7de106d6f1f6de6a96f996735c5fc636011b/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesScheduler.java#L57-L83

The following code block appears to be creating the YAML config based on the required record structures:
https://github.com/apache/spark/blob/de59e01aa4853ef951da080c0d1908d53d133ebe/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/PodTemplateConfigMapStep.scala#L39-L66

I am trying to figure out what my starting point for this should be on the `Heron` side, any ideas? What functionality is available as far as building the YAML record structures (`Config.Builder`)?

The following utility is preparing the `ConfigMap` and storing the YAML `<Key, Value>` pairs from the `topologyConfig`:
https://github.com/apache/incubator-heron/blob/4d9a7de106d6f1f6de6a96f996735c5fc636011b/heron/api/src/java/org/apache/heron/api/utils/TopologyUtils.java#L295-L323

A quick review of the code in the `heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/` directory, and some other related code:

* `KubernetesConstants`: namespaced constants for configurations.
* `KubernetesContext`: child class of `Context` with Kubernetes specific keys used to lookup values (no "setters", only "getters") stored in the supplied `Config` object.
* `KubernetesController`: the basic interface with Kubernetes to recover info about the topology configs as well as submit/kill/restart them.
* `KubernetesLauncher`: used to submit a topology to the Kubernetes cluster.
* `KubernetesScheduler`: the interface to manage the cluster, get info, kill, restart, resize, and update the cluster size based on the topology. It contains a `KubernetesController` object which facilitates the comms with the cluster.
* `KubernetesUtils`: various logging utilities.
* `V1Controller`: child class of the `KubernetesController` which contains the actual logic to interface with the Kubernetes cluster. The `PackingPlan` for the instances is central to operations with configurations stored in a `ContainerPlan` object.
* `Volumes`: utilities used to configure the clusters storage volumes.
* `PackingPlan`: `heron/spi/src/java/org/apache/heron/spi/packing/` has the classes for `PackingPlan`, `InstancePlan` and `ContainerPlan`.

I have very quickly compiled a basic `Pod Config`:
```YAML
# POD CONFIG WITH TEMPLATE:
apiVersion: v1
kind: Pod
metadata:
  name: heron-node
spec:
  replicas: 2
  selector:
    matchLabels:
      app: heron-node
  template:
    metadata:
      labels:
        app: heron-node
    spec:
      containers:
      - name: heron-node
        image: centos:8
        command: ['sh', '-c', 'echo "Heron Kubernetes node." && sleep 3600']
        resources:
          requests:
            memory: "1Gb"
            cpu: "250m"
          limits:
            memory: "5Gb"
            cpu: "500m"
        securityContext:
          allowPrivilegeEscalation: false
          capabilities:
            add: ["NET_ADMIN", "SYS_TIME"]
      restartPolicy: OnFailure
```

The code [snippet](https://github.com/apache/spark/blob/de59e01aa4853ef951da080c0d1908d53d133ebe/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/PodTemplateConfigMapStep.scala#L39-L54) from Spark is appending the following to the pod config, which creates a `Volume` mount pointing to the pod config template:

```YAML
  volumes:
    - name: pod-template-name  # from <POD_TEMPLATE_VOLUME>.
      configMap:
        name: configmap-name  # from <configmapName>.
        items:
        - key: pod-template-key  # from <POD_TEMPLATE_KEY>.
          path: executor-pod-spec-template-file-name # from <EXECUTOR_POD_SPEC_TEMPLATE_FILE_NAME>.
```

This code [snippet](https://github.com/apache/spark/blob/de59e01aa4853ef951da080c0d1908d53d133ebe/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/PodTemplateConfigMapStep.scala#L56-L61) is adding the `Volume` mount pointing to a container config template:

```YAML
  volumes:
    - name: pod-template-name  # from <POD_TEMPLATE_VOLUME>.
      path: executor-pod-spec-template-mount-path # from <EXECUTOR_POD_SPEC_TEMPLATE_MOUNTPATH>.
```

The constants (caps naming convention) are defined [here](https://github.com/apache/spark/blob/de59e01aa4853ef951da080c0d1908d53d133ebe/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala#L81-L85). If we are to follow what Spark is doing, we will need to declare these constants as conventions and document them. The most prudent place for these constants is `KubernetesConstants` as they will not be required outside. The `configmapName` is being acquired as [such](https://github.com/apache/spark/blob/de59e01aa4853ef951da080c0d1908d53d133ebe/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/PodTemplateConfigMapStep.scala#L37).

The remaining two routines in the Spark code appear to be extracting information from the configs - they are not modifying them.

The question now is what functionality exists within Heron to modify the `Config` object to append the YAML elements to the `Config` as well as where (object) the actual pod/container configs are stored within the `Config`. Another question is if there is support for both containers and pods configs, and what the structure within the `Config` object is. Once we have an understanding of these things I can hammer out a solution.

The following are client API components to interface with Kubernetes:
* `V1PodSpec`: [io.kubernetes.client.openapi.models.V1PodSpec](https://javadoc.io/static/io.kubernetes/client-java-api/13.0.0/io/kubernetes/client/openapi/models/V1PodSpec.html) configuration for the pods. Includes methods to `setVolumes` as well as `setSecurityContext`. This will be set in a `Pod Template`.
* `V1PodTemplateSpec`: [io.kubernetes.client.openapi.models.V1PodTemplateSpec
](https://javadoc.io/static/io.kubernetes/client-java-api/13.0.0/io/kubernetes/client/openapi/models/V1PodTemplateSpec.html) is the pod template config. This needs to be put into a `Stateful Set`.
* `V1StatefulSetSpec`: [io.kubernetes.client.openapi.models.V1StatefulSetSpec](https://javadoc.io/static/io.kubernetes/client-java-api/11.0.0/io/kubernetes/client/openapi/models/V1StatefulSetSpec.html) sateful set config.
* `V1StatefulSetSpecBuilder`: [io.kubernetes.client.openapi.models.V1StatefulSetSpecBuilder](https://javadoc.io/static/io.kubernetes/client-java-api/11.0.0/io/kubernetes/client/openapi/models/V1StatefulSetSpecBuilder.html) used to build/generate the `Stateful Set` configs.

I have never used Heron and am learning about the workflow and setup as I go, so please bear with me. I am also still getting familiar with the vast codebase.

Hi @nicknezis, I have a plan put together 🤔  and will start to hammer things out. I shall generate a WIP PR into the Heron repo once I have some of the more substantial components put together.

Fix patch:
diff --git a/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesUtils.java b/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesUtils.java
index 2363d7abdab..709662a9f30 100644
--- a/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesUtils.java
+++ b/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesUtils.java
@@ -90,6 +90,11 @@ static String Megabytes(ByteAmount amount) {
     return String.format("%sMi", Long.toString(amount.asMegabytes()));
   }
 
+  static double roundDecimal(double value, int places) {
+    double scale = Math.pow(10, places);
+    return Math.round(value * scale) / scale;
+  }
+
   static class V1ControllerUtils<T> {
     private static final Logger LOG = Logger.getLogger(V1Controller.class.getName());
 
diff --git a/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/V1Controller.java b/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/V1Controller.java
index 3bde340b096..669b1280e49 100644
--- a/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/V1Controller.java
+++ b/heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/V1Controller.java
@@ -99,6 +99,11 @@ public class V1Controller extends KubernetesController {
   private final AppsV1Api appsClient;
   private final CoreV1Api coreClient;
 
+  /**
+   * Configures the Kubernetes API Application and Core communications clients.
+   * @param configuration <code>topology</code> configurations.
+   * @param runtimeConfiguration Kubernetes runtime configurations.
+   */
   V1Controller(Config configuration, Config runtimeConfiguration) {
     super(configuration, runtimeConfiguration);
 
@@ -119,6 +124,11 @@ public class V1Controller extends KubernetesController {
     }
   }
 
+  /**
+   * Configures all components required by a <code>topology</code> and submits it to the Kubernetes scheduler.
+   * @param packingPlan Used to configure the StatefulSets <code>Resource</code>s and replica count.
+   * @return Success indicator.
+   */
   @Override
   boolean submit(PackingPlan packingPlan) {
     final String topologyName = getTopologyName();
@@ -160,6 +170,15 @@ boolean submit(PackingPlan packingPlan) {
     return true;
   }
 
+  /**
+   * Shuts down a <code>topology</code> by deleting all associated resources.
+   * <ul>
+   *   <li><code>Persistent Volume Claims</code> added by the <code>topology</code>.</li>
+   *   <li><code>StatefulSet</code> for both the <code>Executors</code> and <code>Manager</code>.</li>
+   *   <li>Headless <code>Service</code> which facilitates communication between all Pods.</li>
+   * </ul>
+   * @return Success indicator.
+   */
   @Override
   boolean killTopology() {
     removePersistentVolumeClaims();
@@ -186,6 +205,11 @@ boolean restart(int shardId) {
     return true;
   }
 
+  /**
+   * Adds a specified number of Pods to a <code>topology</code>'s <code>Executors</code>.
+   * @param containersToAdd Set of containers to be added.
+   * @return The passed in <code>Packing Plan</code>.
+   */
   @Override
   public Set<PackingPlan.ContainerPlan>
       addContainers(Set<PackingPlan.ContainerPlan> containersToAdd) {
@@ -210,6 +234,10 @@ boolean restart(int shardId) {
     return containersToAdd;
   }
 
+  /**
+   * Removes a specified number of Pods from a <code>topology</code>'s <code>Executors</code>.
+   * @param containersToRemove Set of containers to be removed.
+   */
   @Override
   public void removeContainers(Set<PackingPlan.ContainerPlan> containersToRemove) {
     final V1StatefulSet statefulSet;
@@ -232,6 +260,12 @@ public void removeContainers(Set<PackingPlan.ContainerPlan> containersToRemove)
     }
   }
 
+  /**
+   * Performs an in-place update of the replica count for a <code>topology</code>. This allows the
+   * <code>topology</code> Pod count to be scaled up or down.
+   * @param replicas The new number of Pod replicas required.
+   * @throws ApiException in the event there is a failure patching the StatefulSet.
+   */
   private void patchStatefulSetReplicas(int replicas) throws ApiException {
     final String body =
             String.format(KubernetesConstants.JSON_PATCH_STATEFUL_SET_REPLICAS_FORMAT,
@@ -253,11 +287,20 @@ private void patchStatefulSetReplicas(int replicas) throws ApiException {
         appsClient.getApiClient());
   }
 
+  /**
+   * Retrieves the <code>Executors</code> StatefulSet configurations for the Kubernetes cluster.
+   * @return <code>Executors</code> StatefulSet configurations.
+   * @throws ApiException in the event there is a failure retrieving the StatefulSet.
+   */
   V1StatefulSet getStatefulSet() throws ApiException {
     return appsClient.readNamespacedStatefulSet(getStatefulSetName(true), getNamespace(),
         null, null, null);
   }
 
+  /**
+   * Deletes the headless <code>Service</code> for a <code>topology</code>'s <code>Executors</code>
+   * and <code>Manager</code> using the <code>topology</code>'s name.
+   */
   void deleteService() {
     try (Response response = coreClient.deleteNamespacedServiceCall(getTopologyName(),
           getNamespace(), null, null, 0, null,
@@ -293,6 +336,10 @@ void deleteService() {
             + "] in namespace [" + getNamespace() + "] is deleted.");
   }
 
+  /**
+   * Deletes the StatefulSets for a <code>topology</code>'s <code>Executors</code> and <code>Manager</code>
+   * using <code>Label</code>s.
+   */
   void deleteStatefulSets() {
     try (Response response = appsClient.deleteCollectionNamespacedStatefulSetCall(getNamespace(),
         null, null, null, null, null, createTopologySelectorLabels(), null, null, null, null, null,
@@ -383,6 +430,10 @@ protected static String setShardIdEnvironmentVariableCommand(boolean isExecutor)
     return String.format(pattern, ENV_SHARD_ID, ENV_SHARD_ID);
   }
 
+  /**
+   * Creates a headless <code>Service</code> to facilitate communication between Pods in a <code>topology</code>.
+   * @return A fully configured <code>Service</code> to be used by a <code>topology</code>.
+   */
   private V1Service createTopologyService() {
     final String topologyName = getTopologyName();
 
@@ -483,14 +534,26 @@ private V1StatefulSet createStatefulSet(Resource containerResource, int numberOf
     return statefulSet;
   }
 
+  /**
+   * Extracts general Pod <code>Annotation</code>s from configurations.
+   * @return Key-value pairs of general <code>Annotation</code>s to be added to the Pod.
+   */
   private Map<String, String> getPodAnnotations() {
     return KubernetesContext.getPodAnnotations(getConfiguration());
   }
 
+  /**
+   * Extracts <code>Service Annotations</code> for configurations.
+   * @return Key-value pairs of service <code>Annotation</code>s to be added to the Pod.
+   */
   private Map<String, String> getServiceAnnotations() {
     return KubernetesContext.getServiceAnnotations(getConfiguration());
   }
 
+  /**
+   * Generates <code>Label</code>s to indicate Prometheus scraping and the exposed port.
+   * @return Key-value pairs of Prometheus <code>Annotation</code>s to be added to the Pod.
+   */
   private Map<String, String> getPrometheusAnnotations() {
     final Map<String, String> annotations = new HashMap<>();
     annotations.put(KubernetesConstants.ANNOTATION_PROMETHEUS_SCRAPE, "true");
@@ -500,6 +563,11 @@ private Map<String, String> getPrometheusAnnotations() {
     return annotations;
   }
 
+  /**
+   * Generates the <code>heron</code> and <code>topology</code> name <code>Match Label</code>s.
+   * @param topologyName Name of the <code>topology</code>.
+   * @return Key-value pairs of <code>Match Label</code>s to be added to the Pod.
+   */
   private Map<String, String> getPodMatchLabels(String topologyName) {
     final Map<String, String> labels = new HashMap<>();
     labels.put(KubernetesConstants.LABEL_APP, KubernetesConstants.LABEL_APP_VALUE);
@@ -507,6 +575,12 @@ private Map<String, String> getPodMatchLabels(String topologyName) {
     return labels;
   }
 
+  /**
+   * Extracts <code>Label</code>s from configurations, generates the <code>heron</code> and
+   * <code>topology</code> name <code>Label</code>s.
+   * @param topologyName Name of the <code>topology</code>.
+   * @return Key-value pairs of <code>Label</code>s to be added to the Pod.
+   */
   private Map<String, String> getPodLabels(String topologyName) {
     final Map<String, String> labels = new HashMap<>();
     labels.put(KubernetesConstants.LABEL_APP, KubernetesConstants.LABEL_APP_VALUE);
@@ -515,6 +589,10 @@ private Map<String, String> getPodLabels(String topologyName) {
     return labels;
   }
 
+  /**
+   * Extracts <code>Selector Labels</code> for<code>Service</code>s from configurations.
+   * @return Key-value pairs of <code>Service Labels</code> to be added to the Pod.
+   */
   private Map<String, String> getServiceLabels() {
     return KubernetesContext.getServiceLabels(getConfiguration());
   }
@@ -639,6 +717,10 @@ protected void addVolumesIfPresent(final V1PodSpec spec) {
     }
   }
 
+  /**
+   * Adds <code>Volume Mounts</code> for <code>Secrets</code> to a pod.
+   * @param podSpec <code>Pod Spec</code> to add secrets to.
+   */
   private void mountSecretsAsVolumes(V1PodSpec podSpec) {
     final Config config = getConfiguration();
     final Map<String, String> secrets = KubernetesContext.getPodSecretsToMount(config);
@@ -705,7 +787,7 @@ private void configureHeronContainer(Resource resource, int numberOfInstances,
    * @param container The <code>container</code> to be configured.
    * @param configuration The <code>Config</code> object to check if a resource request needs to be set.
    * @param resource User defined resources limits from input.
-   * @param isExecutor
+   * @param isExecutor Flag to indicate configuration for an <code>executor</code> or <code>manager</code>.
    */
   @VisibleForTesting
   protected void configureContainerResources(final V1Container container,
@@ -730,7 +812,7 @@ protected void configureContainerResources(final V1Container container,
     // with precedence [1] CLI, [2] Config.
     final Map<String, Quantity> limits = resourceRequirements.getLimits();
     final Quantity limitCPU = limitsCLI.getOrDefault(KubernetesConstants.CPU,
-        Quantity.fromString(Double.toString(roundDecimal(resource.getCpu(), 3))));
+        Quantity.fromString(Double.toString(KubernetesUtils.roundDecimal(resource.getCpu(), 3))));
     final Quantity limitMEMORY = limitsCLI.getOrDefault(KubernetesConstants.MEMORY,
         Quantity.fromString(KubernetesUtils.Megabytes(resource.getRam())));
 
@@ -907,6 +989,10 @@ protected void mountVolumeIfPresent(final V1Container container) {
     }
   }
 
+  /**
+   * Adds <code>Secret Key</code> references to a <code>container</code>.
+   * @param container <code>container</code> to be configured.
+   */
   private void setSecretKeyRefs(V1Container container) {
     final Config config = getConfiguration();
     final Map<String, String> podSecretKeyRefs = KubernetesContext.getPodSecretKeyRefs(config);
@@ -930,11 +1016,6 @@ private void setSecretKeyRefs(V1Container container) {
     }
   }
 
-  public static double roundDecimal(double value, int places) {
-    double scale = Math.pow(10, places);
-    return Math.round(value * scale) / scale;
-  }
-
   /**
    * Initiates the process of locating and loading <code>Pod Template</code> from a <code>ConfigMap</code>.
    * The loaded text is then parsed into a usable <code>Pod Template</code>.


Test patch:
diff --git a/heron/schedulers/tests/java/org/apache/heron/scheduler/kubernetes/V1ControllerTest.java b/heron/schedulers/tests/java/org/apache/heron/scheduler/kubernetes/V1ControllerTest.java
index 6436520ef4d..a4975fd9430 100644
--- a/heron/schedulers/tests/java/org/apache/heron/scheduler/kubernetes/V1ControllerTest.java
+++ b/heron/schedulers/tests/java/org/apache/heron/scheduler/kubernetes/V1ControllerTest.java
@@ -566,11 +566,11 @@ public void testConfigureContainerResources() {
     final Quantity defaultRAM = Quantity.fromString(
         KubernetesUtils.Megabytes(resourceDefault.getRam()));
     final Quantity defaultCPU = Quantity.fromString(
-        Double.toString(V1Controller.roundDecimal(resourceDefault.getCpu(), 3)));
+        Double.toString(KubernetesUtils.roundDecimal(resourceDefault.getCpu(), 3)));
     final Quantity customRAM = Quantity.fromString(
         KubernetesUtils.Megabytes(resourceCustom.getRam()));
     final Quantity customCPU = Quantity.fromString(
-        Double.toString(V1Controller.roundDecimal(resourceCustom.getCpu(), 3)));
+        Double.toString(KubernetesUtils.roundDecimal(resourceCustom.getCpu(), 3)));
     final Quantity customDisk = Quantity.fromString(
         KubernetesUtils.Megabytes(resourceCustom.getDisk()));