Feature/63 implement deprovisioning (#98)

* added deprovisioning for S3 and Azure Blob

* fixed overwriting destination containers

* do not allow to transition from ERROR to DEPROVISION_REQ

* deactivated terraform-plan due to problems with the Azure AD SP

Author: paullatzelsperger

.github/workflows/terraform.yaml

@@ -86,12 +86,13 @@ jobs:
         id: validate
         run: terraform -chdir=scripts validate -no-color
 
-      - name: Terraform Plan
-        id: plan
-        if: github.event_name == 'pull_request'
-        run: terraform -chdir=scripts plan -no-color -var "resourcesuffix=dev" -var "backend_account_key=${{ secrets.TF_BACKEND_KEY }}"
+      #     deactivated until the GithubActions-User has correct credentials
+#      - name: Terraform Plan
+#        id: plan
+#        if: github.event_name == 'pull_request'
+#        run: terraform -chdir=scripts plan -no-color -var "resourcesuffix=dev" -var "backend_account_key=${{ secrets.TF_BACKEND_KEY }}"
+
 
-#     deactivated until the GithubActions-User has correct credentials
 
 #      - name: Taint the connector instance
 #        if: github.ref == 'refs/heads/main' && github.event_name == 'push'

distributions/demo-e2e/dagx-config-docker.properties

@@ -7,7 +7,7 @@ dagx.vault.clientid=<YOUR_SERVICEPRINCIPAL_CLIENTID>
 dagx.vault.tenantid=<YOUR_SUBSCRIPTION_TENANTID>
 # that must be the same certificate that was used during "terraform apply" for the primary service principal!
 dagx.vault.certificate=./cert.pfx
-dagx.vault.name=dagx-demo-vault
+dagx.vault.name=dagx-dev-vault
 dagx.atlas.url=https://dev-dagx-atlas.westeurope.cloudapp.azure.com
 dagx.nifi.url=http://dev-dagx-nifi.westeurope.azurecontainer.io:8080/
 dagx.nifi.flow.url=http://dev-dagx-nifi.westeurope.azurecontainer.io:8888/

extensions/transfer/transfer-core/src/main/java/com/microsoft/dagx/transfer/core/CoreTransferExtension.java

@@ -11,6 +11,7 @@
 import com.microsoft.dagx.spi.system.ServiceExtension;
 import com.microsoft.dagx.spi.system.ServiceExtensionContext;
 import com.microsoft.dagx.spi.transfer.TransferProcessManager;
+import com.microsoft.dagx.spi.transfer.TransferProcessObservable;
 import com.microsoft.dagx.spi.transfer.TransferWaitStrategy;
 import com.microsoft.dagx.spi.transfer.flow.DataFlowManager;
 import com.microsoft.dagx.spi.transfer.provision.ProvisionManager;
@@ -84,6 +85,7 @@ public void initialize(ServiceExtensionContext context) {
                 .build();
 
         context.registerService(TransferProcessManager.class, processManager);
+        context.registerService(TransferProcessObservable.class, processManager);
 
         monitor.info("Initialized Core Transfer extension");
     }

extensions/transfer/transfer-core/src/main/java/com/microsoft/dagx/transfer/core/provision/ProvisionContextImpl.java

@@ -21,14 +21,17 @@
 public class ProvisionContextImpl implements ProvisionContext {
     private final Consumer<ProvisionedResource> resourceCallback;
     private final BiConsumer<ProvisionedDataDestinationResource, SecretToken> destinationCallback;
+    private final BiConsumer<ProvisionedDataDestinationResource, Throwable> deprovisionCallback;
     private final TransferProcessStore processStore;
 
     public ProvisionContextImpl(TransferProcessStore processStore,
                                 Consumer<ProvisionedResource> resourceCallback,
-                                BiConsumer<ProvisionedDataDestinationResource, SecretToken> destinationCallback) {
+                                BiConsumer<ProvisionedDataDestinationResource, SecretToken> destinationCallback,
+                                BiConsumer<ProvisionedDataDestinationResource, Throwable> deprovisionCallback) {
         this.processStore = processStore;
         this.resourceCallback = resourceCallback;
         this.destinationCallback = destinationCallback;
+        this.deprovisionCallback = deprovisionCallback;
     }
 
     @Override
@@ -41,6 +44,11 @@ public void callback(ProvisionedDataDestinationResource resource, SecretToken se
         destinationCallback.accept(resource, secretToken);
     }
 
+    @Override
+    public void deprovisioned(ProvisionedDataDestinationResource resource, Throwable error) {
+        deprovisionCallback.accept(resource, error);
+    }
+
     @Override
     public void create(String processId, String resourceDefinitionId, Object data) {
         processStore.createData(processId, resourceDefinitionId, data);

extensions/transfer/transfer-core/src/main/java/com/microsoft/dagx/transfer/core/provision/ProvisionManagerImpl.java

@@ -10,6 +10,7 @@
 import com.microsoft.dagx.spi.security.Vault;
 import com.microsoft.dagx.spi.transfer.provision.ProvisionManager;
 import com.microsoft.dagx.spi.transfer.provision.Provisioner;
+import com.microsoft.dagx.spi.transfer.response.ResponseStatus;
 import com.microsoft.dagx.spi.transfer.store.TransferProcessStore;
 import com.microsoft.dagx.spi.types.TypeManager;
 import com.microsoft.dagx.spi.types.domain.transfer.*;
@@ -38,7 +39,7 @@ public ProvisionManagerImpl(Vault vault, TypeManager typeManager, Monitor monito
 
     public void start(TransferProcessStore processStore) {
         this.processStore = processStore;
-        var context = new ProvisionContextImpl(this.processStore, this::onResource, this::onDestinationResource);
+        var context = new ProvisionContextImpl(this.processStore, this::onResource, this::onDestinationResource, this::onDeprovisionComplete);
         provisioners.forEach(provisioner -> provisioner.initialize(context));
     }
 
@@ -64,7 +65,29 @@ public void provision(TransferProcess process) {
     public void deprovision(TransferProcess process) {
         for (ProvisionedResource definition : process.getProvisionedResourceSet().getResources()) {
             Provisioner<?, ProvisionedResource> chosenProvisioner = getProvisioner(definition);
-            chosenProvisioner.deprovision(definition);
+            final ResponseStatus status = chosenProvisioner.deprovision(definition);
+            if (status != ResponseStatus.OK) {
+                process.transitionError("Error during deprovisioning");
+                processStore.update(process);
+            }
+        }
+    }
+
+    void onDeprovisionComplete(ProvisionedDataDestinationResource resource, Throwable deprovisionError) {
+        if (deprovisionError != null) {
+            monitor.severe("Deprovisioning error: ", deprovisionError);
+        } else {
+            monitor.info("Deprovisioning successfully completed.");
+
+            final TransferProcess transferProcess = processStore.find(resource.getTransferProcessId());
+            if (transferProcess != null) {
+                transferProcess.transitionDeprovisioned();
+                processStore.update(transferProcess);
+                monitor.debug("Process " + transferProcess.getId() + " is now " + TransferProcessStates.from(transferProcess.getState()));
+            } else {
+                monitor.severe("ProvisionManager: no TransferProcess found for deprovisioned resource");
+            }
+
         }
     }
 

extensions/transfer/transfer-core/src/main/java/com/microsoft/dagx/transfer/core/transfer/TransferProcessManagerImpl.java

@@ -7,35 +7,32 @@
 
 import com.microsoft.dagx.spi.message.RemoteMessageDispatcherRegistry;
 import com.microsoft.dagx.spi.monitor.Monitor;
-import com.microsoft.dagx.spi.transfer.TransferInitiateResponse;
-import com.microsoft.dagx.spi.transfer.TransferProcessManager;
-import com.microsoft.dagx.spi.transfer.TransferWaitStrategy;
+import com.microsoft.dagx.spi.transfer.*;
 import com.microsoft.dagx.spi.transfer.flow.DataFlowManager;
 import com.microsoft.dagx.spi.transfer.provision.ProvisionManager;
 import com.microsoft.dagx.spi.transfer.provision.ResourceManifestGenerator;
 import com.microsoft.dagx.spi.transfer.response.ResponseStatus;
 import com.microsoft.dagx.spi.transfer.store.TransferProcessStore;
 import com.microsoft.dagx.spi.types.domain.transfer.*;
 
-import java.util.List;
-import java.util.Objects;
+import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 import static com.microsoft.dagx.spi.types.domain.transfer.TransferProcess.Type.CLIENT;
 import static com.microsoft.dagx.spi.types.domain.transfer.TransferProcess.Type.PROVIDER;
-import static com.microsoft.dagx.spi.types.domain.transfer.TransferProcessStates.INITIAL;
-import static com.microsoft.dagx.spi.types.domain.transfer.TransferProcessStates.PROVISIONED;
+import static com.microsoft.dagx.spi.types.domain.transfer.TransferProcessStates.*;
 import static java.lang.String.format;
 import static java.util.UUID.randomUUID;
 
 /**
  *
  */
-public class TransferProcessManagerImpl implements TransferProcessManager {
+public class TransferProcessManagerImpl implements TransferProcessManager, TransferProcessObservable {
     private final AtomicBoolean active = new AtomicBoolean();
+    private final Map<String, List<TransferProcessListener>> listenerMap;
     private int batchSize = 5;
     private TransferWaitStrategy waitStrategy = () -> 5000L;  // default wait five seconds
     private ResourceManifestGenerator manifestGenerator;
@@ -48,6 +45,11 @@ public class TransferProcessManagerImpl implements TransferProcessManager {
     private StatusCheckerRegistry statusCheckerRegistry;
 
     private TransferProcessManagerImpl() {
+        listenerMap = new HashMap<>();
+    }
+
+    public Map<String, List<TransferProcessListener>> getListeners() {
+        return listenerMap;
     }
 
     public void start(TransferProcessStore processStore) {
@@ -74,6 +76,28 @@ public TransferInitiateResponse initiateProviderRequest(DataRequest dataRequest)
         return initiateRequest(PROVIDER, dataRequest);
     }
 
+    @Override
+    public void registerListener(String processId, TransferProcessListener listener) {
+        if (listenerMap.containsKey(processId)) {
+            final List<TransferProcessListener> list = listenerMap.get(processId);
+            if (!list.contains(listener)) {
+                list.add(listener);
+            }
+        } else {
+            final ArrayList<TransferProcessListener> list = new ArrayList<>();
+            list.add(listener);
+            listenerMap.put(processId, list);
+        }
+    }
+
+    @Override
+    public void unregister(TransferProcessListener listener) {
+        // unregister from all processes
+        listenerMap.forEach((key, value) -> value.remove(listener));
+        // clear the registration if no more listeners
+        listenerMap.entrySet().removeIf(e -> e.getValue().isEmpty());
+    }
+
     private TransferInitiateResponse initiateRequest(TransferProcess.Type type, DataRequest dataRequest) {
         // make the request idempotent: if the process exists, return
         var processId = transferProcessStore.processIdForTransferId(dataRequest.getId());
@@ -99,7 +123,9 @@ private void run() {
 
                 int finished = checkCompleted();
 
-                if (provisioning + provisioned + sent + finished == 0) {
+                int deprovisioning = checkDeprovisioningRequested();
+
+                if (provisioning + provisioned + sent + finished + deprovisioning == 0) {
                     Thread.sleep(waitStrategy.waitForMillis());
                 }
                 waitStrategy.success();
@@ -122,6 +148,27 @@ private void run() {
         }
     }
 
+    /**
+     * Transitions all processes that are in state DEPROVISIONING_REQ and deprovisions their associated
+     * resources. Then they are moved to DEPROVISIONING
+     *
+     * @return the number of transfer processes in DEPROVISIONING_REQ
+     */
+    private int checkDeprovisioningRequested() {
+        List<TransferProcess> processesDeprovisioning = transferProcessStore.nextForState(DEPROVISIONING_REQ.code(), batchSize);
+
+        for (var process : processesDeprovisioning) {
+            process.transitionDeprovisioning();
+            transferProcessStore.update(process);
+            monitor.debug("Process " + process.getId() + " is now " + TransferProcessStates.from(process.getState()));
+            provisionManager.deprovision(process);
+            publishDeprovisioned(process);
+        }
+
+        return processesDeprovisioning.size();
+    }
+
+
     /**
      * Transition all processes, who have provisioned resources, into the IN_PROCRESS or STREAMING status, depending on
      * whether they're finite or not.
@@ -160,30 +207,34 @@ private int checkCompleted() {
 
         for (var process : processesInProgress.stream().filter(p -> p.getType() == CLIENT).collect(Collectors.toList())) {
 
-            //only check resources for which a checker was registered.
-            // todo: maybe error out processes with uncheckable resources??
             List<ProvisionedResource> resources = process.getProvisionedResourceSet().getResources().stream().filter(this::hasChecker).collect(Collectors.toList());
 
-            //todo: comment this in if we want to error out uncheckable resources
-//            var resourcesWithNoChecker = resources.stream().filter(resource -> statusCheckerRegistry.resolve(resource) == null).collect(Collectors.toList());
-//            if (!resourcesWithNoChecker.isEmpty()) {
-//                final String violatingResourceClasses = resourcesWithNoChecker.stream().map(r -> r.getClass().getName()).collect(Collectors.joining(", "));
-//                monitor.severe("There is no StatusChecker for resource type " + violatingResourceClasses);
-//                process.transitionError("No StatusChecker found for a provisioned resource of type(s) " + violatingResourceClasses + ". The violating resource is part of transfer process " + process.getId());
-//                transferProcessStore.update(process);
-//                continue;
-//            }
-
             // update the process once ALL resources are completed
             if (resources.stream().allMatch(this::isComplete)) {
                 process.transitionCompleted();
                 monitor.debug("Process " + process.getId() + " is now " + TransferProcessStates.COMPLETED);
+                publishCompleted(process);
+
             }
             transferProcessStore.update(process);
         }
         return processesInProgress.size();
     }
 
+    private void publishCompleted(TransferProcess process) {
+        final List<TransferProcessListener> transferProcessListeners = listenerMap.get(process.getId());
+        if (transferProcessListeners != null) {
+            transferProcessListeners.forEach(l -> l.completed(process));
+        }
+    }
+
+    private void publishDeprovisioned(TransferProcess process) {
+        final List<TransferProcessListener> transferProcessListeners = listenerMap.get(process.getId());
+        if (transferProcessListeners != null) {
+            transferProcessListeners.forEach(l -> l.deprovisioned(process));
+        }
+    }
+
     private boolean hasChecker(ProvisionedResource provisionedResource) {
         return provisionedResource instanceof ProvisionedDataDestinationResource && statusCheckerRegistry.resolve((ProvisionedDataDestinationResource) provisionedResource) != null;
     }