Support Environment chains (#42)

Author: ryannedolan

hoptimator-catalog/src/main/java/com/linkedin/hoptimator/catalog/Resource.java

@@ -152,7 +152,11 @@ public interface Environment {
     Environment EMPTY = new SimpleEnvironment();
     Environment PROCESS = new ProcessEnvironment();
 
-    String getOrDefault(String key, String defaultValue);
+    String getOrDefault(String key, Supplier<String> f);
+
+    default Environment orElse(Environment other) {
+      return (k, f) -> getOrDefault(k, () -> other.getOrDefault(k, f));
+    }
   }
 
   /** Basic Environment implementation */
@@ -162,6 +166,14 @@ public static class SimpleEnvironment implements Environment {
     public SimpleEnvironment() {
     }
 
+    public SimpleEnvironment(Properties properties) {
+      properties.forEach((k, v) -> vars.put(k.toString(), v.toString()));
+    }
+
+    public SimpleEnvironment(Map<String, String> vars) {
+      exportAll(vars);
+    }
+
     protected void export(String property, String value) {
       vars.put(property, value);
     }
@@ -180,20 +192,24 @@ public SimpleEnvironment with(String key, String value) {
     }
 
     @Override
-    public String getOrDefault(String key, String defaultValue) {
-      if (defaultValue == null && !vars.containsKey(key)) {
-        throw new IllegalArgumentException("No variable '" + key + "' found in the environment");
+    public String getOrDefault(String key, Supplier<String> f) {
+      if (!vars.containsKey(key)) {
+        if (f == null || f.get() == null) {
+          throw new IllegalArgumentException("No variable '" + key + "' found in the environment");
+        } else {
+          return f.get();
+        }
       }
-      return vars.getOrDefault(key, defaultValue);
+      return vars.get(key);
     }
   }
 
   /** Returns "{{key}}" for any key without a default */
   public static class DummyEnvironment implements Environment {
     @Override
-    public String getOrDefault(String key, String defaultValue) {
-      if (defaultValue != null) {
-        return defaultValue;
+    public String getOrDefault(String key, Supplier<String> f) {
+      if (f != null && f.get() != null) {
+        return f.get();
       } else {
         return "{{" + key + "}}";
       }
@@ -204,13 +220,13 @@ public String getOrDefault(String key, String defaultValue) {
   public static class ProcessEnvironment implements Environment {
 
     @Override
-    public String getOrDefault(String key, String defaultValue) {
+    public String getOrDefault(String key, Supplier<String> f) {
       String value = System.getenv(key);
       if (value == null) {
         value = System.getProperty(key);
       }
-      if (value == null) {
-        value = defaultValue;
+      if (value == null && f != null) {
+        value = f.get();
       }
       if (value == null) {
         throw new IllegalArgumentException("Missing system property `" + key + "`");
@@ -270,7 +286,7 @@ public String render(Resource resource) {
         }
         String key = m.group(2);
         String defaultValue = m.group(4);
-        String value = resource.getOrDefault(key, () -> env.getOrDefault(key, defaultValue));
+        String value = resource.getOrDefault(key, () -> env.getOrDefault(key, () -> defaultValue));
         if (value == null) {
           throw new IllegalArgumentException(template + " has no value for key " + key + ".");
         }

hoptimator-catalog/src/test/java/com/linkedin/hoptimator/catalog/ResourceTest.java

@@ -0,0 +1,28 @@
+package com.linkedin.hoptimator.catalog;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+
+public class ResourceTest {
+
+  @Test
+  public void handlesChainedEnvironments() {
+    Resource.Environment env = new Resource.SimpleEnvironment() {{
+      export("one", "1");
+      export("foo", "bar");
+    }}.orElse(new Resource.SimpleEnvironment() {{
+      export("two", "2");
+      export("foo", "car");
+    }}.orElse(new Resource.SimpleEnvironment() {{
+      export("three", "3");
+      export("foo", "dar");
+    }}));
+    assertEquals("1", env.getOrDefault("one", () -> "x"));
+    assertEquals("2", env.getOrDefault("two", () -> "x"));
+    assertEquals("3", env.getOrDefault("three", () -> "x"));
+    assertEquals("bar", env.getOrDefault("foo", () -> "x"));
+    assertEquals("x", env.getOrDefault("oof", () -> "x"));
+  }
+}

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/HoptimatorOperatorApp.java

@@ -6,6 +6,7 @@
 import io.kubernetes.client.extended.controller.Controller;
 import io.kubernetes.client.extended.controller.ControllerManager;
 
+import com.linkedin.hoptimator.catalog.Resource;
 import com.linkedin.hoptimator.models.V1alpha1Subscription;
 import com.linkedin.hoptimator.models.V1alpha1SubscriptionList;
 import com.linkedin.hoptimator.operator.subscription.SubscriptionReconciler;
@@ -34,12 +35,14 @@ public class HoptimatorOperatorApp {
   final String modelPath;
   final String namespace;
   final Properties properties;
+  final Resource.Environment environment;
 
   /** This constructor is likely to evolve and break. */
   public HoptimatorOperatorApp(String modelPath, String namespace, Properties properties) {
     this.modelPath = modelPath;
     this.namespace = namespace;
     this.properties = properties;
+    this.environment = new Resource.SimpleEnvironment(properties);
   }
 
   public static void main(String[] args) throws Exception {
@@ -91,7 +94,7 @@ public void run() throws Exception {
 
     List<Controller> controllers = new ArrayList<>();
     controllers.addAll(ControllerService.controllers(operator));
-    controllers.add(SubscriptionReconciler.controller(operator, plannerFactory));
+    controllers.add(SubscriptionReconciler.controller(operator, plannerFactory, environment));
 
     ControllerManager controllerManager = new ControllerManager(operator.informerFactory(),
       controllers.toArray(new Controller[0]));

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/subscription/SubscriptionEnvironment.java

@@ -26,27 +26,13 @@
  *     as a basis for deriving K8s object names, Kafka topic names, etc. The
  *     name is guaranteed to be a valid K8s object name, e.g. `my-subscription`.
  *  - `pipeline.avroSchema`, an Avro schema for the pipeline's output type.
- *
- * In addition, any "hints" in the Subscription object (`.spec.hints`) are
- * exported as-is. These can be used to provide optional properties to
- * templates. When using such hints in a template, ensure that you provide a
- * default value, e.g. `{{numPartitions:null}``, since they will usually be
- * missing.
  */
 public class SubscriptionEnvironment extends Resource.SimpleEnvironment {
 
-  public SubscriptionEnvironment(String namespace, String name, Pipeline pipeline,
-      Map<String, String> hints) {
-    if (hints != null) {
-      exportAll(hints);
-    }
+  public SubscriptionEnvironment(String namespace, String name, Pipeline pipeline) {
     export("pipeline.namespace", namespace);
     export("pipeline.name", name);
     export("pipeline.avroSchema", AvroConverter.avro("com.linkedin.hoptimator", "OutputRecord",
       pipeline.outputType()).toString(false));
   }
-
-  public SubscriptionEnvironment(String namespace, String name, Pipeline pipeline) {
-    this(namespace, name, pipeline, Collections.emptyMap());
-  }
 }

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/subscription/SubscriptionReconciler.java

@@ -43,10 +43,13 @@ public class SubscriptionReconciler implements Reconciler {
 
   private final Operator operator;
   private final HoptimatorPlanner.Factory plannerFactory;
+  private final Resource.Environment environment;
 
-  private SubscriptionReconciler(Operator operator, HoptimatorPlanner.Factory plannerFactory) {
+  private SubscriptionReconciler(Operator operator, HoptimatorPlanner.Factory plannerFactory,
+      Resource.Environment environment) {
     this.operator = operator;
     this.plannerFactory = plannerFactory;
+    this.environment = environment;
   }
 
   @Override
@@ -84,11 +87,13 @@ public Result reconcile(Request request) {
         log.info("Planning a new pipeline for {}/{} with SQL `{}`...", kind, name, object.getSpec().getSql());
 
         Pipeline pipeline = pipeline(object);
-        SubscriptionEnvironment env = new SubscriptionEnvironment(namespace, name, pipeline);
-        SubscriptionEnvironment sinkEnv = new SubscriptionEnvironment(namespace, name, pipeline,
-          object.getSpec().getHints());
-        Resource.TemplateFactory templateFactory = new Resource.SimpleTemplateFactory(env);
-        Resource.TemplateFactory sinkTemplateFactory = new Resource.SimpleTemplateFactory(sinkEnv);
+        Resource.Environment subEnv = new SubscriptionEnvironment(namespace, name, pipeline)
+          .orElse(environment);
+        Resource.TemplateFactory templateFactory = new Resource.SimpleTemplateFactory(subEnv);
+
+        // For sink resources, also expose hints.
+        Resource.TemplateFactory sinkTemplateFactory = new Resource.SimpleTemplateFactory(subEnv
+          .orElse(new Resource.SimpleEnvironment(object.getSpec().getHints())));
 
         // Render resources related to all source tables.
         List<String> upstreamResources = pipeline.upstreamResources().stream()
@@ -266,8 +271,8 @@ private static boolean isReady(DynamicKubernetesObject obj) {
     return true;
   }
 
-  public static Controller controller(Operator operator, HoptimatorPlanner.Factory plannerFactory) {
-    Reconciler reconciler = new SubscriptionReconciler(operator, plannerFactory);
+  public static Controller controller(Operator operator, HoptimatorPlanner.Factory plannerFactory, Resource.Environment environment) {
+    Reconciler reconciler = new SubscriptionReconciler(operator, plannerFactory, environment);
     return ControllerBuilder.defaultBuilder(operator.informerFactory())
       .withReconciler(reconciler)
       .withName("subscription-controller")