Modify deployers to keep state and undo changes on failure (#143)

* Add snapshot/restore behavior

* Address feedback & make provider more generic

* Add snapshot() to reset providers

* Refactor to restore under deployer directly

Author: jogrogan

hoptimator-api/src/main/java/com/linkedin/hoptimator/Deployer.java

@@ -15,4 +15,7 @@ public interface Deployer {
 
   /** Render a list of specs, usually YAML. */
   List<String> specify() throws SQLException;
+
+  /** Deployers are expected to track the state of changes made and revert them on demand. */
+  void restore();
 }

hoptimator-api/src/main/java/com/linkedin/hoptimator/DeployerProvider.java

@@ -8,4 +8,7 @@ public interface DeployerProvider {
 
   /** Find deployers capable of deploying the obj. */
   <T extends Deployable> Collection<Deployer> deployers(T obj, Properties connectionProperties);
+
+  /** A DeployerProvider with lower priority will execute first */
+  int priority();
 }

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java

@@ -20,6 +20,7 @@
 package com.linkedin.hoptimator.jdbc;
 
 import com.linkedin.hoptimator.Database;
+import com.linkedin.hoptimator.Deployer;
 import com.linkedin.hoptimator.MaterializedView;
 import com.linkedin.hoptimator.Pipeline;
 import com.linkedin.hoptimator.View;
@@ -30,6 +31,7 @@
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
@@ -133,15 +135,21 @@ public void execute(SqlCreateView create, CalcitePrepare.Context context) {
     RelProtoDataType protoType = RelDataTypeImpl.proto(analyzed.rowType);
     ViewTable viewTable = new ViewTable(Object.class, protoType, sql, schemaPath, viewPath);
     View view = new View(viewPath, sql);
+
+    Collection<Deployer> deployers = null;
     try {
+      deployers = DeploymentService.deployers(view, connectionProperties);
       ValidationService.validateOrThrow(viewTable);
       if (create.getReplace()) {
-        DeploymentService.update(view, connectionProperties);
+        DeploymentService.update(deployers);
       } else {
-        DeploymentService.create(view, connectionProperties);
+        DeploymentService.create(deployers);
       }
       schemaPlus.add(viewName, viewTable);
     } catch (Exception e) {
+      if (deployers != null) {
+        DeploymentService.restore(deployers);
+      }
       throw new DdlException(create, e.getMessage(), e);
     }
   }
@@ -185,6 +193,8 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
     final String sql = q.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
     final List<String> schemaPath = pair.left.path(null);
 
+    Collection<Deployer> deployers = null;
+
     String schemaName = schemaPlus.getName();
     String viewName = pair.right;
     List<String> viewPath = new ArrayList<>(schemaPath);
@@ -235,14 +245,18 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
       // TODO support CREATE ... WITH (options...)
       ValidationService.validateOrThrow(hook);
 
+      deployers = DeploymentService.deployers(hook, connectionProperties);
       if (create.getReplace()) {
-        DeploymentService.update(hook, connectionProperties);
+        DeploymentService.update(deployers);
       } else {
-        DeploymentService.create(hook, connectionProperties);
+        DeploymentService.create(deployers);
       }
 
       schemaPlus.add(viewName, materializedViewTable);
-    } catch (SQLException e) {
+    } catch (SQLException | RuntimeException e) {
+      if (deployers != null) {
+        DeploymentService.restore(deployers);
+      }
       throw new DdlException(create, e.getMessage(), e);
     }
   }
@@ -275,19 +289,25 @@ public void execute(SqlDropObject drop, CalcitePrepare.Context context) {
     List<String> viewPath = new ArrayList<>(schemaPath);
     viewPath.add(viewName);
 
+    Collection<Deployer> deployers = null;
     try {
       if (table instanceof MaterializedViewTable) {
         MaterializedViewTable materializedViewTable = (MaterializedViewTable) table;
         View view = new View(viewPath, materializedViewTable.viewSql());
-        DeploymentService.delete(view, connectionProperties);
+        deployers = DeploymentService.deployers(view, connectionProperties);
+        DeploymentService.delete(deployers);
       } else if (table instanceof ViewTable) {
         ViewTable viewTable = (ViewTable) table;
         View view = new View(viewPath, viewTable.getViewSql());
-        DeploymentService.delete(view, connectionProperties);
+        deployers = DeploymentService.deployers(view, connectionProperties);
+        DeploymentService.delete(deployers);
       } else {
         throw new DdlException(drop, viewName + " is not a view.");
       }
     } catch (Exception e) {
+      if (deployers != null) {
+        DeploymentService.restore(deployers);
+      }
       throw new DdlException(drop, e.getMessage(), e);
     }
     schemaPlus.removeTable(viewName);

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

@@ -16,42 +16,62 @@ public abstract class K8sDeployer<T extends KubernetesObject, U extends Kubernet
     implements Deployer {
 
   private final K8sApi<T, U> api;
+  private final K8sSnapshot snapshot;
 
   K8sDeployer(K8sContext context, K8sApiEndpoint<T, U> endpoint) {
     this.api = new K8sApi<>(context, endpoint);
+    this.snapshot = new K8sSnapshot(context);
   }
 
   @Override
   public void create() throws SQLException {
-    api.create(toK8sObject());
+    create(toK8sObject());
   }
 
   public V1OwnerReference createAndReference() throws SQLException {
     T obj = toK8sObject();
-    api.create(obj);
+    create(obj);
     return api.reference(obj);
   }
 
+  private void create(T obj) throws SQLException {
+    snapshot.store(obj);
+    api.create(obj);
+  }
+
   @Override
   public void delete() throws SQLException {
-    api.delete(toK8sObject());
+    T obj = toK8sObject();
+    snapshot.store(obj);
+    api.delete(obj);
   }
 
   @Override
   public void update() throws SQLException {
-    api.update(toK8sObject());
+    update(toK8sObject());
   }
 
   public V1OwnerReference updateAndReference() throws SQLException {
     T obj = toK8sObject();
-    api.update(obj);
+    update(obj);
     return api.reference(obj);
   }
 
+  private void update(T obj) throws SQLException {
+    snapshot.store(obj);
+    api.update(obj);
+  }
+
   @Override
   public List<String> specify() throws SQLException {
     return Collections.singletonList(Yaml.dump(toK8sObject()));
   }
 
+  @Override
+  public void restore() {
+    snapshot.restore();
+  }
+
+
   protected abstract T toK8sObject() throws SQLException;
 }

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

@@ -32,4 +32,9 @@ public <T extends Deployable> Collection<Deployer> deployers(T obj, Properties c
 
    return list;
   }
+
+  @Override
+  public int priority() {
+    return 1;
+  }
 }

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

@@ -20,39 +20,71 @@ class K8sMaterializedViewDeployer implements Deployer {
   private final MaterializedView view;
   private final K8sContext context;
   private final Properties connectionProperties;
+  private final K8sViewDeployer viewDeployer;
+  private final List<Deployer> deployers;
+
+  private final Object crudLock = new Object();
 
   K8sMaterializedViewDeployer(MaterializedView view, K8sContext context, Properties connectionProperties) {
     this.view = view;
     this.context = context;
     this.connectionProperties = connectionProperties;
+    this.viewDeployer = new K8sViewDeployer(view, true, context);
+    this.deployers = new ArrayList<>();
   }
 
   @Override
   public void create() throws SQLException {
-    String name = name();
-    List<String> pipelineSpecs = pipelineSpecs();
-    V1OwnerReference viewRef = new K8sViewDeployer(view, true, context).createAndReference();
-    K8sContext viewContext = context.withOwner(viewRef);
-    V1OwnerReference pipelineRef = new K8sPipelineDeployer(name, pipelineSpecs, sql(), viewContext).createAndReference();
-    K8sContext pipelineContext = viewContext.withLabel("pipeline", name).withOwner(pipelineRef);
-    new K8sYamlDeployerImpl(pipelineContext, pipelineSpecs).update();  // update, cuz some elements may already exist
+    synchronized (crudLock) {
+      String name = name();
+      List<String> pipelineSpecs = pipelineSpecs();
+      V1OwnerReference viewRef = viewDeployer.createAndReference();
+      K8sContext viewContext = context.withOwner(viewRef);
+      K8sPipelineDeployer pipelineDeployer = new K8sPipelineDeployer(name, pipelineSpecs, sql(), viewContext);
+      deployers.add(pipelineDeployer);
+      V1OwnerReference pipelineRef = pipelineDeployer.createAndReference();
+      K8sContext pipelineContext = viewContext.withLabel("pipeline", name).withOwner(pipelineRef);
+      K8sYamlDeployerImpl yamlDeployer = new K8sYamlDeployerImpl(pipelineContext, pipelineSpecs);
+      deployers.add(yamlDeployer);
+      yamlDeployer.update();  // update, cuz some elements may already exist
+    }
   }
 
   @Override
   public void update() throws SQLException {
-    String name = name();
-    List<String> pipelineSpecs = pipelineSpecs();
-    V1OwnerReference viewRef = new K8sViewDeployer(view, true, context).updateAndReference();
-    K8sContext viewContext = context.withOwner(viewRef);
-    V1OwnerReference pipelineRef = new K8sPipelineDeployer(name, pipelineSpecs, sql(), viewContext).updateAndReference();
-    K8sContext pipelineContext = viewContext.withLabel("pipeline", name).withOwner(pipelineRef);
-    new K8sYamlDeployerImpl(pipelineContext, pipelineSpecs).update();
+    synchronized (crudLock) {
+      String name = name();
+      List<String> pipelineSpecs = pipelineSpecs();
+      V1OwnerReference viewRef = viewDeployer.updateAndReference();
+      K8sContext viewContext = context.withOwner(viewRef);
+      K8sPipelineDeployer pipelineDeployer = new K8sPipelineDeployer(name, pipelineSpecs, sql(), viewContext);
+      deployers.add(pipelineDeployer);
+      V1OwnerReference pipelineRef = pipelineDeployer.updateAndReference();
+      K8sContext pipelineContext = viewContext.withLabel("pipeline", name).withOwner(pipelineRef);
+      K8sYamlDeployerImpl yamlDeployer = new K8sYamlDeployerImpl(pipelineContext, pipelineSpecs);
+      deployers.add(yamlDeployer);
+      yamlDeployer.update();
+    }
   }
 
   @Override
   public void delete() throws SQLException {
-    // Delete will cascade to any owned objects, including the Pipeline object.
-    new K8sViewDeployer(view, true, context).delete();
+    synchronized (crudLock) {
+      // Delete will cascade to any owned objects, including the Pipeline object.
+      viewDeployer.delete();
+    }
+  }
+
+  @Override
+  public void restore() {
+    synchronized (crudLock) {
+      // There may be a scenario where a view creates multiple pipelines.
+      // Restoring the pipelines in reverse order ensures the original state is preserved.
+      for (int i = deployers.size() - 1; i >= 0; i--) {
+        deployers.get(i).restore();
+      }
+      viewDeployer.restore();
+    }
   }
 
   List<String> pipelineSpecs() throws SQLException {