Expose configs via pipeline element table

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineElement.java

@@ -2,6 +2,7 @@
 
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -11,13 +12,15 @@
 
 /** Represents a pipeline element status and its associated pipelines. */
 public class K8sPipelineElement {
-  private String name;
+  private final String name;
   private final K8sPipelineElementStatus status;
+  private final Map<String, String> configs;
   private final Set<V1alpha1Pipeline> pipelines = new HashSet<>();
 
-  public K8sPipelineElement(V1alpha1Pipeline pipeline, K8sPipelineElementStatus status) {
+  public K8sPipelineElement(V1alpha1Pipeline pipeline, K8sPipelineElementStatus status, Map<String, String> configs) {
     this.name = status.getName();
     this.status = status;
+    this.configs = configs;
     this.pipelines.add(pipeline);
   }
 
@@ -29,6 +32,10 @@ public K8sPipelineElementStatus status() {
     return status;
   }
 
+  public Map<String, String> configs() {
+    return configs;
+  }
+
   public void addPipeline(V1alpha1Pipeline pipeline) {
     pipelines.add(pipeline);
   }

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineElementApi.java

@@ -1,6 +1,12 @@
 package com.linkedin.hoptimator.k8s;
 
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import io.kubernetes.client.util.generic.KubernetesApiResponse;
+import io.kubernetes.client.util.generic.dynamic.DynamicKubernetesObject;
+import io.kubernetes.client.util.generic.dynamic.Dynamics;
 import java.sql.SQLException;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
@@ -11,9 +17,15 @@
 import com.linkedin.hoptimator.k8s.status.K8sPipelineElementStatus;
 import com.linkedin.hoptimator.k8s.status.K8sPipelineElementStatusEstimator;
 import com.linkedin.hoptimator.util.Api;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 
 /** Provides all pipeline elements in a {@link com.linkedin.hoptimator.k8s.K8sContext} instance. */
 public class K8sPipelineElementApi implements Api<K8sPipelineElement> {
+  private final static Logger log = LoggerFactory.getLogger(K8sPipelineElementApi.class);
   private final K8sContext context;
 
   K8sPipelineElementApi(K8sContext context) {
@@ -27,18 +39,74 @@ private Collection<K8sPipelineElement> discoverAllElements(K8sContext context) t
     Map<String, K8sPipelineElement> elements = new HashMap<>();
     K8sPipelineElementStatusEstimator statusEstimator = new K8sPipelineElementStatusEstimator(context);
     for (V1alpha1Pipeline pipeline : pipelines) {
-      List<K8sPipelineElementStatus> elementStatuses = statusEstimator.estimateStatuses(pipeline);
-      for (K8sPipelineElementStatus elementStatus : elementStatuses) {
+      String namespace = Objects.requireNonNull(pipeline.getMetadata()).getNamespace();
+      List<String> elementYamls = getPipelineElements(pipeline);
+      for (String elementYaml : elementYamls) {
+        K8sPipelineElementStatus elementStatus = statusEstimator.estimateElementStatus(elementYaml, namespace);
+        Map<String, String> configurations = getElementConfiguration(elementYaml, namespace);
         String key = elementStatus.getName();
         if (!elements.containsKey(key)) {
-          elements.put(key, new K8sPipelineElement(pipeline, elementStatus));
+          elements.put(key, new K8sPipelineElement(pipeline, elementStatus, configurations));
         }
         elements.get(key).addPipeline(pipeline);
       }
     }
     return elements.values();
   }
 
+  /**
+   * Returns list of all elements specified in the given pipeline.
+   */
+  List<String> getPipelineElements(V1alpha1Pipeline pipeline) {
+    return Arrays.stream(Objects.requireNonNull(Objects.requireNonNull(pipeline.getSpec()).getYaml())
+            .split("\n---\n"))
+        .map(String::trim)
+        .filter(x -> !x.isEmpty())
+        .collect(Collectors.toList());
+  }
+
+  /**
+   * Returns spec.configs of the given element, if present.
+   */
+  Map<String, String> getElementConfiguration(String elementYaml, String pipelineNamespace) {
+    DynamicKubernetesObject obj = Dynamics.newFromYaml(elementYaml);
+    String name = obj.getMetadata().getName();
+    String namespace = obj.getMetadata().getNamespace() == null ? pipelineNamespace : obj.getMetadata().getNamespace();
+    String kind = obj.getKind();
+    String nameWithKind = String.format("%s/%s", kind, name);
+    try {
+      KubernetesApiResponse<DynamicKubernetesObject> existing =
+          context.dynamic(obj.getApiVersion(), K8sUtils.guessPlural(obj)).get(namespace, name);
+      String failureMessage =
+          String.format("Failed to fetch %s in namespace %s: %s.", nameWithKind, namespace, existing.toString());
+      existing.onFailure((code, status) -> log.warn(failureMessage));
+      if (!existing.isSuccess()) {
+        return new HashMap<>();
+      }
+      DynamicKubernetesObject element = existing.getObject();
+      if (element == null || element.getRaw() == null) {
+        return new HashMap<>();
+      }
+      if (!element.getRaw().has("spec") || !element.getRaw().getAsJsonObject("spec").has("configs")) {
+        return new HashMap<>();
+      }
+
+      // Extract configs as a Map
+      JsonObject configs = element.getRaw().getAsJsonObject("spec").getAsJsonObject("configs");
+      Map<String, String> configMap = new HashMap<>();
+      for (Map.Entry<String, JsonElement> entry : configs.entrySet()) {
+        configMap.put(entry.getKey(), entry.getValue().getAsString());
+      }
+      return configMap;
+    } catch (Exception e) {
+      String failureMessage =
+          String.format("Encountered exception while checking status of %s in namespace %s: %s", nameWithKind,
+              namespace, e);
+      log.error(failureMessage);
+      return new HashMap<>();
+    }
+  }
+
   /**
    * Lists all pipeline elements in the context.
    */

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineElementMapApi.java

@@ -28,7 +28,7 @@ public Collection<K8sPipelineElementMapEntry> list() throws SQLException {
   }
 
   private static Stream<K8sPipelineElementMapEntry> mapEntriesFromElement(K8sPipelineElement element) {
-    String elementName = element.status().getName();
+    String elementName = element.name();
     return element.pipelineNames()
         .stream()
         .map(pipelineName -> new K8sPipelineElementMapEntry(elementName, pipelineName));

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineElementTable.java

@@ -1,9 +1,9 @@
 package com.linkedin.hoptimator.k8s;
 
+import com.linkedin.hoptimator.util.RemoteTable;
 import org.apache.calcite.schema.Schema;
+import org.json.simple.JSONObject;
 
-import com.linkedin.hoptimator.k8s.status.K8sPipelineElementStatus;
-import com.linkedin.hoptimator.util.RemoteTable;
 
 public class K8sPipelineElementTable extends RemoteTable<K8sPipelineElement, K8sPipelineElementTable.Row> {
 
@@ -13,12 +13,14 @@ public static class Row {
     public boolean READY;
     public boolean FAILED;
     public String MESSAGE;
+    public String CONFIGS;
 
-    public Row(String name, boolean ready, boolean failed, String message) {
+    public Row(String name, boolean ready, boolean failed, String message, String configs) {
       this.NAME = name;
       this.READY = ready;
       this.FAILED = failed;
       this.MESSAGE = message;
+      this.CONFIGS = configs;
     }
 
     @Override
@@ -34,8 +36,11 @@ public K8sPipelineElementTable(K8sPipelineElementApi pipelineElementApi) {
 
   @Override
   public Row toRow(K8sPipelineElement k8sPipelineElement) {
-    K8sPipelineElementStatus status = k8sPipelineElement.status();
-    return new Row(status.getName(), status.isReady(), status.isFailed(), status.getMessage());
+    return new Row(k8sPipelineElement.name(),
+        k8sPipelineElement.status().isReady(),
+        k8sPipelineElement.status().isFailed(),
+        k8sPipelineElement.status().getMessage(),
+        JSONObject.toJSONString(k8sPipelineElement.configs()));
   }
 
   @Override

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/status/K8sPipelineElementStatusEstimator.java

@@ -47,7 +47,7 @@ public List<K8sPipelineElementStatus> estimateStatuses(V1alpha1Pipeline pipeline
   /**
    * Estimates status of an element. If we can not retrieve it from K8s, we assume that it's not ready and not failed yet.
    */
-  private K8sPipelineElementStatus estimateElementStatus(String elementYaml, String pipelineNamespace) {
+  public K8sPipelineElementStatus estimateElementStatus(String elementYaml, String pipelineNamespace) {
     DynamicKubernetesObject obj = Dynamics.newFromYaml(elementYaml);
     String name = obj.getMetadata().getName();
     String namespace = obj.getMetadata().getNamespace() == null ? pipelineNamespace : obj.getMetadata().getNamespace();