refactor: convert provision-http extension to data-plane (#5156)

Author: ndr-brt

core/common/junit/src/main/java/org/eclipse/edc/junit/extensions/EmbeddedRuntime.java

@@ -124,6 +124,13 @@ public void shutdown() {
         return new TestServiceExtensionContext(monitor, config, serviceMocks);
     }
 
+    /**
+     * Utility method that doesn't require the addShutdownHook parameter
+     */
+    public void boot() {
+        boot(false);
+    }
+
     @Override
     protected @NotNull Monitor createMonitor() {
         // disable logs when "quiet" log level is set

core/common/lib/state-machine-lib/src/main/java/org/eclipse/edc/statemachine/AbstractStateEntityManager.java

@@ -82,9 +82,11 @@ private EntityRetryProcessConfiguration defaultEntityRetryProcessConfiguration()
 
     protected void update(E entity) {
         store.save(entity);
-        monitor.debug(() -> "[%s] %s %s is now in state %s"
+        var error = entity.getErrorDetail() == null ? "" : ". errorDetail: " + entity.getErrorDetail();
+
+        monitor.debug(() -> "[%s] %s %s is now in state %s%s"
                 .formatted(this.getClass().getSimpleName(), entity.getClass().getSimpleName(),
-                        entity.getId(), entity.stateAsString()));
+                        entity.getId(), entity.stateAsString(), error));
     }
 
     protected void breakLease(E entity) {

core/control-plane/control-plane-transfer-manager/src/main/java/org/eclipse/edc/connector/controlplane/transfer/process/TransferProcessManagerImpl.java

@@ -224,6 +224,8 @@ private boolean processInitial(TransferProcess process) {
                     update(process);
                     return true;
                 }
+            } else {
+                monitor.warning("control-plane provisioning has been deprecated, please convert your provision extensions to the data-plane model and deploy them there.");
             }
 
             process.transitionProvisioning(manifest);
@@ -247,6 +249,9 @@ private boolean processInitial(TransferProcess process) {
             }
 
             var manifest = manifestGenerator.generateProviderResourceManifest(process, dataAddress, policy);
+            if (!manifest.empty()) {
+                monitor.warning("control-plane provisioning has been deprecated, please convert your provision extensions to the data-plane model and deploy them there.");
+            }
             process.transitionProvisioning(manifest);
         }
 

core/data-plane/data-plane-core/src/main/java/org/eclipse/edc/connector/dataplane/framework/manager/DataPlaneManagerImpl.java

@@ -33,6 +33,7 @@
 import org.eclipse.edc.spi.response.StatusResult;
 import org.eclipse.edc.spi.result.Result;
 import org.eclipse.edc.spi.result.ServiceFailure;
+import org.eclipse.edc.spi.result.ServiceResult;
 import org.eclipse.edc.spi.result.StoreFailure;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.spi.types.domain.transfer.DataFlowProvisionMessage;
@@ -100,10 +101,16 @@ public Result<Void> validate(DataFlowStartMessage dataRequest) {
             return success();
         } else {
             var transferService = transferServiceRegistry.resolveTransferService(dataRequest);
-            return transferService != null ?
-                    transferService.validate(dataRequest) :
-                    Result.failure(format("Cannot find a transfer Service that can handle %s source and %s destination",
-                            dataRequest.getSourceDataAddress().getType(), dataRequest.getDestinationDataAddress().getType()));
+            if (transferService != null) {
+                return transferService.validate(dataRequest);
+            }
+
+            if (resourceDefinitionGeneratorManager.sourceTypes().contains(dataRequest.getSourceDataAddress().getType())) {
+                return Result.success();
+            }
+
+            return Result.failure(format("Cannot find a transfer Service that can handle %s source and %s destination",
+                    dataRequest.getSourceDataAddress().getType(), dataRequest.getDestinationDataAddress().getType()));
         }
     }
 
@@ -198,7 +205,11 @@ public StatusResult<Void> terminate(String dataFlowId, @Nullable String reason)
                     }
                     return stop(dataFlow, reason)
                             .onSuccess(flow -> {
-                                flow.transitToTerminated(reason);
+                                if (flow.resourcesToBeDeprovisioned().isEmpty()) {
+                                    flow.transitToTerminated(reason);
+                                } else {
+                                    flow.transitionToDeprovisioning();
+                                }
                                 update(dataFlow);
                             })
                             .mapEmpty();
@@ -223,9 +234,9 @@ public StatusResult<Void> restartFlows() {
     }
 
     @Override
-    public StatusResult<Void> resourceProvisioned(ProvisionedResource provisionedResource) {
+    public ServiceResult<Void> resourceProvisioned(ProvisionedResource provisionedResource) {
         return store.findByIdAndLease(provisionedResource.getFlowId())
-                .flatMap(StatusResult::from)
+                .flatMap(ServiceResult::from)
                 .onSuccess(flow -> {
                     flow.resourceProvisioned(List.of(provisionedResource));
                     update(flow);
@@ -234,9 +245,9 @@ public StatusResult<Void> resourceProvisioned(ProvisionedResource provisionedRes
     }
 
     @Override
-    public StatusResult<Void> resourceDeprovisioned(DeprovisionedResource deprovisionedResource) {
+    public ServiceResult<Void> resourceDeprovisioned(DeprovisionedResource deprovisionedResource) {
         return store.findByIdAndLease(deprovisionedResource.getFlowId())
-                .flatMap(StatusResult::from)
+                .flatMap(ServiceResult::from)
                 .onSuccess(flow -> {
                     flow.resourceDeprovisioned(List.of(deprovisionedResource));
                     update(flow);
@@ -376,17 +387,20 @@ private boolean processProvisioning(DataFlow dataFlow) {
 
                     if (provisionedResources.size() != results.size()) {
                         var failureDetail = results.stream().filter(StatusResult::failed).map(StatusResult::getFailureDetail).collect(joining(","));
+                        monitor.warning("Failed to provision flow " + flow.getId() + ": " + failureDetail);
                         flow.setErrorDetail(failureDetail);
                         flow.transitionToProvisioning();
                     }
 
                     update(flow);
                 })
                 .onFailure((flow, t) -> {
+                    monitor.warning("Failed to provision flow " + flow.getId(), t);
                     flow.transitionToProvisioning();
                     update(flow);
                 })
                 .onFinalFailure((flow, e) -> {
+                    monitor.severe("Cannot provision flow " + flow.getId(), e);
                     flow.transitToFailed("Cannot provision: " + e.getMessage());
                     update(flow);
                 })

core/data-plane/data-plane-core/src/main/java/org/eclipse/edc/connector/dataplane/framework/provision/ResourceDefinitionGeneratorManagerImpl.java

@@ -53,12 +53,17 @@ public List<ProvisionResource> generateConsumerResourceDefinition(DataFlow dataF
     @Override
     public List<ProvisionResource> generateProviderResourceDefinition(DataFlow dataFlow) {
         return providerGenerators.stream()
-                .filter(g -> g.supportedType().equals(dataFlow.getDestination().getType()))
+                .filter(g -> g.supportedType().equals(dataFlow.getSource().getType()))
                 .map(g -> g.generate(dataFlow))
                 .filter(Objects::nonNull)
                 .toList();
     }
 
+    @Override
+    public Set<String> sourceTypes() {
+        return providerGenerators.stream().map(ResourceDefinitionGenerator::supportedType).collect(toSet());
+    }
+
     @Override
     public Set<String> destinationTypes() {
         return consumerGenerators.stream().map(ResourceDefinitionGenerator::supportedType).collect(toSet());