Add TableTriggers (#138)

* Add TableTriggers and controller

* Add TABLE_TRIGGERS metadata table

* Add table trigger unit tests

* Add table trigger metadata integration test

* Relax pipeline table integration tests

* Add sample cron trigger

* Bug fixes h/t srnand

Author: ryannedolan

.github/workflows/integration-tests.yml

@@ -55,6 +55,11 @@ jobs:
         kubectl get flinkdeployments
         kubectl get subscriptions
         kubectl get pipelines
+        kubectl get tabletemplates
+        kubectl get jobtemplates
+        kubectl get tabletriggers
+        kubectl get cronjobs
+        kubectl get jobs
     - name: Capture Flink Job Logs
       if: always()
       run: |

Makefile

@@ -40,9 +40,13 @@ quickstart: build deploy
 
 deploy-demo: deploy
 	kubectl apply -f ./deploy/samples/demodb.yaml
+	kubectl apply -f ./deploy/samples/tabletriggers.yaml
+	kubectl apply -f ./deploy/samples/crontrigger.yaml
 
 undeploy-demo: undeploy
 	kubectl delete -f ./deploy/samples/demodb.yaml || echo "skipping"
+	kubectl delete -f ./deploy/samples/tabletriggers.yaml || echo "skipping"
+	kubectl delete -f ./deploy/samples/crontrigger.yaml || echo "skipping"
 
 deploy-flink: deploy
 	kubectl create namespace flink || echo "skipping"
@@ -65,9 +69,10 @@ undeploy-flink:
 deploy-kafka: deploy deploy-flink
 	kubectl create namespace kafka || echo "skipping"
 	kubectl apply -f "https://strimzi.io/install/latest?namespace=kafka" -n kafka
+	sleep 10 # avoid kubectl race condition
+	kubectl wait --for=condition=Established=True crds/kafkas.kafka.strimzi.io
 	kubectl apply -f ./deploy/samples/kafkadb.yaml
 	kubectl apply -f ./deploy/dev/kafka.yaml
-	kubectl wait --for=condition=Established=True crds/kafkas.kafka.strimzi.io
 	kubectl wait kafka.kafka.strimzi.io/one --for=condition=Ready --timeout=10m -n kafka
 	kubectl wait kafkatopic.kafka.strimzi.io/kafka-database-existing-topic-1 --for=condition=Ready --timeout=10m
 	kubectl wait kafkatopic.kafka.strimzi.io/kafka-database-existing-topic-2 --for=condition=Ready --timeout=10m

deploy/rbac.yaml

@@ -5,12 +5,15 @@ metadata:
   name: hoptimator-operator
 rules:
 - apiGroups: ["hoptimator.linkedin.com"]
-  resources: ["acls", "databases", "engines", "jobtemplates", "kafkatopics", "pipelines", "sqljobs", "subscriptions", "tabletemplates", "views"]
+  resources: ["acls", "databases", "engines", "jobtemplates", "kafkatopics", "pipelines", "sqljobs", "subscriptions", "tabletemplates", "tabletriggers", "views"]
   verbs: ["get", "watch", "list", "update", "create"]
 - apiGroups: ["hoptimator.linkedin.com"]
-  resources: ["acls/status", "kafkatopics/status", "pipelines/status", "sqljobs/status", "subscriptions/status"]
-  verbs: ["get", "patch"]
+  resources: ["acls/status", "kafkatopics/status", "pipelines/status", "sqljobs/status", "subscriptions/status", "tabletriggers/status"]
+  verbs: ["get", "patch", "update", "create"]
 - apiGroups: ["flink.apache.org"]
   resources: ["flinkdeployments", "flinksessionjobs"]
   verbs: ["get", "update", "create"]
+- apiGroups: ["batch"]
+  resources: ["jobs"]
+  verbs: ["list", "create", "delete"]
 

deploy/samples/crontrigger.yaml

@@ -0,0 +1,24 @@
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: cron-trigger
+spec:
+  schedule: "* * * * *"
+  jobTemplate:
+    spec:
+      ttlSecondsAfterFinished: 10
+      template:
+        spec:
+          serviceAccountName: hoptimator-operator
+          containers:
+          - name: trigger
+            image: alpine/k8s:1.33.0
+            imagePullPolicy: IfNotPresent
+            command:
+            - /bin/bash
+            - -c
+            - >-
+              kubectl patch tabletrigger test-table-trigger -p
+              "{\"status\":{\"timestamp\": \"`date +%FT%TZ`\"}}"
+              --subresource=status --type=merge
+          restartPolicy: OnFailure

deploy/samples/tabletriggers.yaml

@@ -0,0 +1,24 @@
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: TableTrigger
+metadata:
+  name: test-table-trigger
+spec:
+  schema: KAFKA
+  table: existing-topic-1
+  yaml: |
+    apiVersion: batch/v1
+    kind: Job
+    metadata:
+      name: test-table-trigger-job
+    spec:
+      template:
+        spec:
+          containers:
+          - name: hello
+            image: alpine/k8s:1.33.0
+            command: ["bash",  "-c", "echo test-table-trigger fired at `date`"]
+          restartPolicy: Never
+      backoffLimit: 4
+      ttlSecondsAfterFinished: 90
+
+

generate-models.sh

@@ -17,5 +17,6 @@ docker run \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/sqljobs.crd.yaml" \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/subscriptions.crd.yaml" \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/tabletemplates.crd.yaml" \
+  -u "$(pwd)/hoptimator-k8s/src/main/resources/tabletriggers.crd.yaml" \
   -u "$(pwd)/hoptimator-k8s/src/main/resources/views.crd.yaml" \
   && echo "done."

hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/QuidemTestBase.java

@@ -35,7 +35,7 @@
 public abstract class QuidemTestBase {
 
   protected void run(String resourceName) throws IOException, URISyntaxException {
-    run(Objects.requireNonNull(Thread.currentThread().getContextClassLoader().getResource(resourceName)).toURI(), "");
+    run(resourceName, "");
   }
 
   protected void run(String resourceName, String jdbcProperties) throws IOException, URISyntaxException {
@@ -62,7 +62,12 @@ protected void run(URI resource, String jdbcProperties) throws IOException {
     for (String line : output) {
       System.out.println(line);
     }
-    Assertions.assertIterableEquals(input, output);
+    for (int i = 0; i < input.size(); i++) {
+      String line = output.get(i);
+      String expected = input.get(i);
+      Assertions.assertEquals(expected, line,
+        "Context:\n" + String.join("\n", output.subList(i, Math.min(i + 10, output.size() - 1))));
+    }
   }
 
   private static final class CustomCommandHandler implements CommandHandler {

hoptimator-k8s/build.gradle

@@ -1,5 +1,6 @@
 plugins {
   id 'java'
+  id 'java-test-fixtures'
   id 'maven-publish'
 }
 
@@ -10,6 +11,10 @@ dependencies {
   implementation libs.calcite.server
   implementation libs.kubernetes.client
 
+  testFixturesImplementation libs.kubernetes.client
+  testFixturesImplementation project(':hoptimator-api')
+  testFixturesImplementation project(':hoptimator-util')
+
   // These are included in case the respective Databases are installed
   testRuntimeOnly project(':hoptimator-demodb')
   testRuntimeOnly project(':hoptimator-kafka')

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

@@ -14,8 +14,10 @@
 import io.kubernetes.client.common.KubernetesListObject;
 import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.openapi.models.V1OwnerReference;
+import io.kubernetes.client.util.Yaml;
 import io.kubernetes.client.util.generic.GenericKubernetesApi;
 import io.kubernetes.client.util.generic.KubernetesApiResponse;
+import io.kubernetes.client.util.generic.options.DeleteOptions;
 import io.kubernetes.client.util.generic.options.ListOptions;
 
 import com.linkedin.hoptimator.util.Api;
@@ -115,10 +117,20 @@ public void delete(T obj) throws SQLException {
     if (obj.getMetadata().getNamespace() == null && !endpoint.clusterScoped()) {
       obj.getMetadata().namespace(context.namespace());
     }
+    delete(obj.getMetadata().getNamespace(), obj.getMetadata().getName());
+  }
+
+  public void delete(String namespace, String name) throws SQLException {
+    DeleteOptions options = new DeleteOptions();
+    options.setPropagationPolicy("Background");   // ensure deletes cascade
     KubernetesApiResponse<T> resp =
-        context.generic(endpoint).delete(obj.getMetadata().getNamespace(), obj.getMetadata().getName());
-    K8sUtils.checkResponse("Error deleting " + obj.getMetadata().getName(), resp);
-    log.info("Deleted K8s obj: {}:{}", obj.getKind(), obj.getMetadata().getName());
+        context.generic(endpoint).delete(namespace, name, options);
+    K8sUtils.checkResponse("Error deleting " + name, resp);
+    log.info("Deleted K8s obj: {}:{}", endpoint.kind(), name);
+  }
+
+  public void delete(String name) throws SQLException {
+    delete(context.namespace(), name);
   }
 
   @Override
@@ -170,7 +182,7 @@ public void updateStatus(T obj, Object status) throws SQLException {
       obj.getMetadata().namespace(context.namespace());
     }
     KubernetesApiResponse<T> resp = context.generic(endpoint).updateStatus(obj, x -> status);
-    K8sUtils.checkResponse("Error updating status of " + obj.getMetadata().getName(), resp);
+    K8sUtils.checkResponse(() -> "Error updating status of " + Yaml.dump(obj), resp);
     log.info("Updated K8s obj status: {}:{}", obj.getKind(), obj.getMetadata().getName());
   }
 }

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

@@ -2,6 +2,8 @@
 
 import io.kubernetes.client.openapi.models.V1ConfigMap;
 import io.kubernetes.client.openapi.models.V1ConfigMapList;
+import io.kubernetes.client.openapi.models.V1Job;
+import io.kubernetes.client.openapi.models.V1JobList;
 import io.kubernetes.client.openapi.models.V1Namespace;
 import io.kubernetes.client.openapi.models.V1NamespaceList;
 import io.kubernetes.client.openapi.models.V1Secret;
@@ -17,6 +19,8 @@
 import com.linkedin.hoptimator.k8s.models.V1alpha1PipelineList;
 import com.linkedin.hoptimator.k8s.models.V1alpha1TableTemplate;
 import com.linkedin.hoptimator.k8s.models.V1alpha1TableTemplateList;
+import com.linkedin.hoptimator.k8s.models.V1alpha1TableTrigger;
+import com.linkedin.hoptimator.k8s.models.V1alpha1TableTriggerList;
 import com.linkedin.hoptimator.k8s.models.V1alpha1View;
 import com.linkedin.hoptimator.k8s.models.V1alpha1ViewList;
 
@@ -31,6 +35,9 @@ public final class K8sApiEndpoints {
   public static final K8sApiEndpoint<V1ConfigMap, V1ConfigMapList> CONFIG_MAPS =
       new K8sApiEndpoint<>("ConfigMap", "", "v1", "configmaps", false,
           V1ConfigMap.class, V1ConfigMapList.class);
+  public static final K8sApiEndpoint<V1Job, V1JobList> JOBS =
+      new K8sApiEndpoint<>("Job", "batch", "v1", "jobs", false,
+          V1Job.class, V1JobList.class);
 
   // Hoptimator custom resources
   public static final K8sApiEndpoint<V1alpha1Database, V1alpha1DatabaseList> DATABASES =
@@ -51,6 +58,9 @@ public final class K8sApiEndpoints {
   public static final K8sApiEndpoint<V1alpha1JobTemplate, V1alpha1JobTemplateList> JOB_TEMPLATES =
       new K8sApiEndpoint<>("JobTemplate", "hoptimator.linkedin.com", "v1alpha1", "jobtemplates", false,
           V1alpha1JobTemplate.class, V1alpha1JobTemplateList.class);
+  public static final K8sApiEndpoint<V1alpha1TableTrigger, V1alpha1TableTriggerList> TABLE_TRIGGERS =
+      new K8sApiEndpoint<>("TableTrigger", "hoptimator.linkedin.com", "v1alpha1", "tabletriggers", false,
+          V1alpha1TableTrigger.class, V1alpha1TableTriggerList.class);
 
   private K8sApiEndpoints() {
   }