Add pluggable controllers via ControllerProvider (#18)

ryannedolan · web-flow · commit 42b784630448 · 2023-05-30T11:29:27.000-07:00
* Publish to LinkedInJFrog * Publish hoptimator-cli and hoptimator-operator * Add pluggable controllers via ControllerProvider * Add user-defined properties to operator * Fix logging dependencies * Make Operator properties optional h/t @ehoner * Drop extraneous log4j.properties h/t @ehoner
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
@@ -25,5 +25,6 @@ junit = "junit:junit:4.12"
 kafkaClients = "org.apache.kafka:kafka-clients:2.7.1"
 kubernetesClient = "io.kubernetes:client-java:16.0.2"
 kubernetesExtendedClient = "io.kubernetes:client-java-extended:16.0.2"
-slf4jLog4j = "org.slf4j:slf4j-log4j12:1.7.36"
+slf4jSimple = "org.slf4j:slf4j-simple:1.7.30"
+slf4jApi = "org.slf4j:slf4j-api:1.7.30"
 sqlline = "sqlline:sqlline:1.12.0"
diff --git a/hoptimator-cli/build.gradle b/hoptimator-cli/build.gradle
@@ -18,10 +18,12 @@ dependencies {
   integration project(':hoptimator-kafka-adapter')
   integration project(':hoptimator-mysql-adapter')
   integration libs.flinkCsv
+  integration libs.flinkConnectorKafka
+  integration libs.slf4jSimple
 
   implementation libs.avro 
   implementation libs.sqlline
-  implementation libs.slf4jLog4j
+  implementation libs.slf4jApi
   implementation libs.flinkClients
   implementation libs.flinkTableRuntime
   implementation libs.flinkTablePlanner
diff --git a/hoptimator-flink-adapter/build.gradle b/hoptimator-flink-adapter/build.gradle
@@ -0,0 +1,22 @@
+plugins {
+  id 'java-library'
+  id 'idea'
+}
+
+dependencies {
+  implementation project(':hoptimator-catalog')
+  implementation project(':hoptimator-models')
+  implementation project(':hoptimator-operator')
+  implementation libs.kubernetesClient
+  implementation libs.kubernetesExtendedClient
+
+  testImplementation libs.junit
+  testImplementation libs.assertj
+}
+
+idea {
+  module {
+    downloadJavadoc = true
+    downloadSources = true
+  }
+}
diff --git a/hoptimator-flink-adapter/src/main/java/com/linkedin/hoptimator/operator/flink/FlinkControllerProvider.java b/hoptimator-flink-adapter/src/main/java/com/linkedin/hoptimator/operator/flink/FlinkControllerProvider.java
@@ -0,0 +1,32 @@
+package com.linkedin.hoptimator.operator.flink;
+
+import com.linkedin.hoptimator.operator.ControllerProvider;
+import com.linkedin.hoptimator.operator.Operator;
+import com.linkedin.hoptimator.models.V1alpha1SqlJob;
+
+import io.kubernetes.client.extended.controller.Controller;
+import io.kubernetes.client.extended.controller.builder.ControllerBuilder;
+import io.kubernetes.client.extended.controller.reconciler.Reconciler;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/** Provides a Controller plugin for FlinkDeployments. */
+public class FlinkControllerProvider implements ControllerProvider {
+
+  @Override
+  public Collection<Controller> controllers(Operator operator) {
+    operator.registerApi("FlinkDeployment", "flinkdeployment", "flinkdeployments",
+      "flink.apache.org", "v1beta1");
+
+    Reconciler reconciler = new FlinkSqlJobReconciler(operator);
+    Controller controller = ControllerBuilder.defaultBuilder(operator.informerFactory())
+      .withReconciler(reconciler)
+      .withName("flink-deployment-controller")
+      .withWorkerCount(1)
+      .watch(x -> ControllerBuilder.controllerWatchBuilder(V1alpha1SqlJob.class, x).build())
+      .build();
+
+    return Collections.singleton(controller);
+  }
+}
diff --git a/hoptimator-flink-adapter/src/main/java/com/linkedin/hoptimator/operator/flink/FlinkDeployment.java b/hoptimator-flink-adapter/src/main/java/com/linkedin/hoptimator/operator/flink/FlinkDeployment.java
diff --git a/hoptimator-flink-adapter/src/main/java/com/linkedin/hoptimator/operator/flink/FlinkSqlJobReconciler.java b/hoptimator-flink-adapter/src/main/java/com/linkedin/hoptimator/operator/flink/FlinkSqlJobReconciler.java
@@ -6,8 +6,6 @@
 import com.linkedin.hoptimator.operator.RequestEnvironment;
 import com.linkedin.hoptimator.models.V1alpha1SqlJob;
 
-import io.kubernetes.client.extended.controller.Controller;
-import io.kubernetes.client.extended.controller.builder.ControllerBuilder;
 import io.kubernetes.client.extended.controller.reconciler.Reconciler;
 import io.kubernetes.client.extended.controller.reconciler.Request;
 import io.kubernetes.client.extended.controller.reconciler.Result;
@@ -72,20 +70,6 @@ public Result reconcile(Request request) {
     return new Result(false);
   }
 
-  public static Controller controller(Operator operator) {
-    Reconciler reconciler = new FlinkSqlJobReconciler(operator);
-    return ControllerBuilder.defaultBuilder(operator.informerFactory())
-      .withReconciler(reconciler)
-      .withName("flink-deployment-controller")
-      .withWorkerCount(1)
-      //.withReadyFunc(resourceInformer::hasSynced) // optional, only starts controller when the
-      // cache has synced up
-      //.withWorkQueue(resourceWorkQueue)
-      //.watch()
-      .watch(x -> ControllerBuilder.controllerWatchBuilder(V1alpha1SqlJob.class, x).build())
-      .build();
-  }
-
   private static <T> List<T> list(List<T> maybeNull) {
     if (maybeNull == null) {
       return Collections.emptyList();
diff --git a/hoptimator-flink-adapter/src/main/resources/META-INF/services/com.linkedin.hoptimator.operator.ControllerProvider b/hoptimator-flink-adapter/src/main/resources/META-INF/services/com.linkedin.hoptimator.operator.ControllerProvider
@@ -0,0 +1 @@
+com.linkedin.hoptimator.operator.flink.FlinkControllerProvider
diff --git a/hoptimator-kafka-adapter/build.gradle b/hoptimator-kafka-adapter/build.gradle
@@ -5,9 +5,12 @@ plugins {
 
 dependencies {
   implementation project(':hoptimator-catalog')
+  implementation project(':hoptimator-models')
+  implementation project(':hoptimator-operator')
   implementation libs.calciteCore
   implementation libs.kafkaClients
-  implementation libs.flinkConnectorKafka
+  implementation libs.kubernetesClient
+  implementation libs.kubernetesExtendedClient
 
   testImplementation libs.junit
   testImplementation libs.assertj
diff --git a/hoptimator-kafka-adapter/src/main/java/com/linkedin/hoptimator/operator/kafka/KafkaControllerProvider.java b/hoptimator-kafka-adapter/src/main/java/com/linkedin/hoptimator/operator/kafka/KafkaControllerProvider.java
@@ -0,0 +1,33 @@
+package com.linkedin.hoptimator.operator.kafka;
+
+import com.linkedin.hoptimator.operator.ControllerProvider;
+import com.linkedin.hoptimator.operator.Operator;
+import com.linkedin.hoptimator.models.V1alpha1KafkaTopic;
+import com.linkedin.hoptimator.models.V1alpha1KafkaTopicList;
+
+import io.kubernetes.client.extended.controller.Controller;
+import io.kubernetes.client.extended.controller.builder.ControllerBuilder;
+import io.kubernetes.client.extended.controller.reconciler.Reconciler;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/** Provides a Controller plugin for KafkaTopics. */
+public class KafkaControllerProvider implements ControllerProvider {
+
+  @Override
+  public Collection<Controller> controllers(Operator operator) {
+    operator.registerApi("KafkaTopic", "kafkatopic", "kafkatopics", "hoptimator.linkedin.com",
+      "v1alpha1", V1alpha1KafkaTopic.class, V1alpha1KafkaTopicList.class);
+
+    Reconciler reconciler = new KafkaTopicReconciler(operator);
+    Controller controller = ControllerBuilder.defaultBuilder(operator.informerFactory())
+      .withReconciler(reconciler)
+      .withName("kafka-topic-controller")
+      .withWorkerCount(1)
+      .watch(x -> ControllerBuilder.controllerWatchBuilder(V1alpha1KafkaTopic.class, x).build())
+      .build();
+
+    return Collections.singleton(controller);
+  }
+}
diff --git a/hoptimator-kafka-adapter/src/main/java/com/linkedin/hoptimator/operator/kafka/KafkaTopicReconciler.java b/hoptimator-kafka-adapter/src/main/java/com/linkedin/hoptimator/operator/kafka/KafkaTopicReconciler.java
@@ -4,8 +4,6 @@
 import com.linkedin.hoptimator.operator.ConfigAssembler;
 import com.linkedin.hoptimator.models.V1alpha1KafkaTopic;
 
-import io.kubernetes.client.extended.controller.Controller;
-import io.kubernetes.client.extended.controller.builder.ControllerBuilder;
 import io.kubernetes.client.extended.controller.reconciler.Reconciler;
 import io.kubernetes.client.extended.controller.reconciler.Request;
 import io.kubernetes.client.extended.controller.reconciler.Result;
@@ -98,20 +96,6 @@ public Result reconcile(Request request) {
     return new Result(false);
   }
 
-  public static Controller controller(Operator operator) {
-    Reconciler reconciler = new KafkaTopicReconciler(operator);
-    return ControllerBuilder.defaultBuilder(operator.informerFactory())
-      .withReconciler(reconciler)
-      .withName("kafka-topic-controller")
-      .withWorkerCount(1)
-      //.withReadyFunc(resourceInformer::hasSynced) // optional, only starts controller when the
-      // cache has synced up
-      //.withWorkQueue(resourceWorkQueue)
-      //.watch()
-      .watch(x -> ControllerBuilder.controllerWatchBuilder(V1alpha1KafkaTopic.class, x).build())
-      .build();
-  }
-
   private static <T> List<T> list(List<T> maybeNull) {
     if (maybeNull == null) {
       return Collections.emptyList();
diff --git a/hoptimator-kafka-adapter/src/main/resources/META-INF/services/com.linkedin.hoptimator.operator.ControllerProvider b/hoptimator-kafka-adapter/src/main/resources/META-INF/services/com.linkedin.hoptimator.operator.ControllerProvider
@@ -0,0 +1 @@
+com.linkedin.hoptimator.operator.kafka.KafkaControllerProvider
diff --git a/hoptimator-models/generate-models.sh b/hoptimator-models/generate-models.sh
@@ -9,7 +9,7 @@ docker run \
   -ti \
   --network host \
   ghcr.io/kubernetes-client/java/crd-model-gen:v1.0.6 \
-  /generate.sh -o "$(pwd)/models" -n "" -p "com.linkedin.hoptimator" \
+  /generate.sh -o "$(pwd)/hoptimator-models" -n "" -p "com.linkedin.hoptimator" \
   -u "$(pwd)/deploy/kafkatopics.crd.yaml" \
   -u "$(pwd)/deploy/subscriptions.crd.yaml" \
   -u "$(pwd)/deploy/sqljobs.crd.yaml" \
diff --git a/hoptimator-operator/build.gradle b/hoptimator-operator/build.gradle
@@ -18,12 +18,13 @@ dependencies {
   // include adapters in integration tests 
   integration project(':hoptimator-kafka-adapter')
   integration project(':hoptimator-mysql-adapter')
+  integration project(':hoptimator-flink-adapter')
+  integration libs.slf4jSimple
 
   implementation libs.calciteCore
-  implementation libs.kafkaClients
   implementation libs.kubernetesClient
   implementation libs.kubernetesExtendedClient
-  implementation libs.slf4jLog4j
+  implementation libs.slf4jApi
 
   testImplementation libs.junit
   testImplementation libs.assertj
diff --git a/hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/ControllerProvider.java b/hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/ControllerProvider.java
@@ -0,0 +1,11 @@
+package com.linkedin.hoptimator.operator;
+
+import io.kubernetes.client.extended.controller.Controller;
+
+import java.util.Collection;
+
+/** Service Provider Interface to enable dynamically loading Controllers. */
+public interface ControllerProvider {
+
+  Collection<Controller> controllers(Operator operator);
+}
diff --git a/hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/ControllerService.java b/hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/ControllerService.java
@@ -0,0 +1,27 @@
+package com.linkedin.hoptimator.operator;
+
+import org.apache.calcite.plan.RelOptRule;
+
+import io.kubernetes.client.extended.controller.Controller;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.ServiceLoader;
+import java.util.stream.Collectors;
+
+/** Loads controller plugins via the ControllerProvider Service Provider Interface. */
+public final class ControllerService {
+
+  public static Collection<ControllerProvider> providers() {
+    ServiceLoader<ControllerProvider> loader = ServiceLoader.load(ControllerProvider.class);
+    List<ControllerProvider> providers = new ArrayList<>();
+    loader.iterator().forEachRemaining(x -> providers.add(x));
+    return providers;
+  }
+
+  public static Collection<Controller> controllers(Operator operator) {
+    return providers().stream().flatMap(x -> x.controllers(operator).stream())
+      .collect(Collectors.toList());
+  }
+}
diff --git a/hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/HoptimatorOperatorApp.java b/hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/HoptimatorOperatorApp.java
@@ -6,29 +6,46 @@
 import io.kubernetes.client.extended.controller.Controller;
 import io.kubernetes.client.extended.controller.ControllerManager;
 
-import com.linkedin.hoptimator.models.V1alpha1KafkaTopic;
-import com.linkedin.hoptimator.models.V1alpha1KafkaTopicList;
 import com.linkedin.hoptimator.models.V1alpha1Subscription;
 import com.linkedin.hoptimator.models.V1alpha1SubscriptionList;
 import com.linkedin.hoptimator.models.V1alpha1SqlJob;
 import com.linkedin.hoptimator.models.V1alpha1SqlJobList;
-import com.linkedin.hoptimator.operator.flink.FlinkSqlJobReconciler;
-import com.linkedin.hoptimator.operator.kafka.KafkaTopicReconciler;
 import com.linkedin.hoptimator.operator.subscription.SubscriptionReconciler;
 import com.linkedin.hoptimator.planner.HoptimatorPlanner;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
-import java.util.concurrent.TimeUnit;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 
 public class HoptimatorOperatorApp {
+  private static final Logger log = LoggerFactory.getLogger(HoptimatorOperatorApp.class);
+
+  final String modelPath;
+  final String namespace;
+  final Properties properties;
+
+  /** 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;
+  }
 
   public static void main(String[] args) throws Exception {
     if (args.length != 1) {
       throw new IllegalArgumentException("Missing model file argument.");
     }
-    HoptimatorPlanner.Factory plannerFactory = HoptimatorPlanner.Factory.fromModelFile(args[0],
-        new Properties());
+    new HoptimatorOperatorApp(args[0], "default", new Properties()).run();
+  }
+
+  public void run() throws Exception {
+    HoptimatorPlanner.Factory plannerFactory = HoptimatorPlanner.Factory.fromModelFile(modelPath,
+        properties);
 
     // ensure model file works, and that static classes are initialized in the main thread
     HoptimatorPlanner planner = plannerFactory.makePlanner();
@@ -37,25 +54,23 @@ public static void main(String[] args) throws Exception {
     apiClient.setHttpClient(apiClient.getHttpClient().newBuilder()
       .readTimeout(0, TimeUnit.SECONDS).build());
     SharedInformerFactory informerFactory = new SharedInformerFactory(apiClient);
-    Operator operator = new Operator("default", apiClient, informerFactory);
+    Operator operator = new Operator(namespace, apiClient, informerFactory, properties);
+    // TODO replace hard-coded "default" namespace with command-line argument
 
     operator.registerApi("SqlJob", "sqljob", "sqljobs",
       "hoptimator.linkedin.com", "v1alpha1", V1alpha1SqlJob.class, V1alpha1SqlJobList.class);
  
-    operator.registerApi("FlinkDeployment", "flinkdeployment", "flinkdeployments",
-      "flink.apache.org", "v1beta1");
-    
-    operator.registerApi("KafkaTopic", "kafkatopic", "kafkatopics", "hoptimator.linkedin.com",
-      "v1alpha1", V1alpha1KafkaTopic.class, V1alpha1KafkaTopicList.class);
-
     operator.registerApi("Subscription", "subscription", "subscriptions", "hoptimator.linkedin.com",
       "v1alpha1", V1alpha1Subscription.class, V1alpha1SubscriptionList.class);
 
+    List<Controller> controllers = new ArrayList<>();
+    controllers.addAll(ControllerService.controllers(operator));
+    controllers.add(SubscriptionReconciler.controller(operator, plannerFactory));
+
     ControllerManager controllerManager = new ControllerManager(operator.informerFactory(),
-      SubscriptionReconciler.controller(operator, plannerFactory),
-      KafkaTopicReconciler.controller(operator),
-      FlinkSqlJobReconciler.controller(operator));
+      controllers.toArray(new Controller[0]));
   
+    log.info("Starting operator with {} controllers.", controllers.size());
     controllerManager.run();  
   }
 }
diff --git a/hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/Operator.java b/hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/Operator.java
@@ -21,27 +21,44 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/** Single handle to all the clients required by all the controllers. */
+/** Single handle to all the clients and configs required by all the controllers. */
 public class Operator {
   private final static Logger log = LoggerFactory.getLogger(Operator.class);
  
   private final String namespace; 
   private final ApiClient apiClient;
   private final SharedInformerFactory informerFactory;
   private final Map<String, ApiInfo<?, ?>> apiInfo = new HashMap<>();
+  private final Properties properties;
 
-  public Operator(String namespace, ApiClient apiClient, SharedInformerFactory informerFactory) {
+  public Operator(String namespace, ApiClient apiClient, SharedInformerFactory informerFactory,
+      Properties properties) {
     this.namespace = namespace;
     this.apiClient = apiClient;
     this.informerFactory = informerFactory;
+    this.properties = properties;
+  }
+
+  public Operator(String namespace, ApiClient apiClient, SharedInformerFactory informerFactory) {
+    this(namespace, apiClient, informerFactory, new Properties());
+  }
+
+  public Operator(String namespace, ApiClient apiClient, Properties properties) {
+    this(namespace, apiClient, new SharedInformerFactory(apiClient), properties);
   }
 
   public Operator(String namespace, ApiClient apiClient) {
-    this(namespace, apiClient, new SharedInformerFactory(apiClient));
+    this(namespace, apiClient, new SharedInformerFactory(apiClient), new Properties());
+  }
+
+  /** Arbitrary global properties, which a controller plugin may use. */
+  public Properties properties() {
+    return properties;
   }
   
   public <T extends KubernetesObject, L extends KubernetesListObject> void registerApi(String kind,
diff --git a/hoptimator-operator/src/main/resources/log4j.properties b/hoptimator-operator/src/main/resources/log4j.properties
diff --git a/settings.gradle b/settings.gradle

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+com.linkedin.hoptimator.operator.flink.FlinkControllerProvider`