feat: add data-plane-instance heartbeat (#4230)

* define e2e tests

* Implement inmemory and sql stateful stores

* Implement DataPlaneInstanceManager

core/common/connector-core/build.gradle.kts

@@ -19,6 +19,7 @@ plugins {
 }
 
 dependencies {
+    api(project(":spi:common:auth-spi"))
     api(project(":spi:common:core-spi"))
     api(project(":spi:common:http-spi"))
     api(project(":spi:common:keys-spi"))

core/common/connector-core/src/main/java/org/eclipse/edc/connector/core/CoreDefaultServicesExtension.java

@@ -17,6 +17,7 @@
 import dev.failsafe.RetryPolicy;
 import okhttp3.EventListener;
 import okhttp3.OkHttpClient;
+import org.eclipse.edc.api.auth.spi.ControlClientAuthenticationProvider;
 import org.eclipse.edc.connector.core.agent.NoOpParticipantIdMapper;
 import org.eclipse.edc.connector.core.base.OkHttpClientConfiguration;
 import org.eclipse.edc.connector.core.base.OkHttpClientFactory;
@@ -37,6 +38,7 @@
 import org.eclipse.edc.transaction.spi.NoopTransactionContext;
 import org.eclipse.edc.transaction.spi.TransactionContext;
 
+import java.util.Collections;
 import java.util.concurrent.Executors;
 
 /**
@@ -125,6 +127,11 @@ public EdcHttpClient edcHttpClient(ServiceExtensionContext context) {
         );
     }
 
+    @Provider(isDefault = true)
+    public ControlClientAuthenticationProvider controlClientAuthenticationProvider() {
+        return Collections::emptyMap;
+    }
+
     @Provider
     public OkHttpClient okHttpClient(ServiceExtensionContext context) {
         var configuration = OkHttpClientConfiguration.Builder.newInstance()

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/asset/AssetEventDispatchTest.java

@@ -19,6 +19,7 @@
 import org.eclipse.edc.connector.controlplane.asset.spi.event.AssetDeleted;
 import org.eclipse.edc.connector.controlplane.asset.spi.event.AssetEvent;
 import org.eclipse.edc.connector.controlplane.services.spi.asset.AssetService;
+import org.eclipse.edc.connector.dataplane.selector.spi.client.DataPlaneClientFactory;
 import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
 import org.eclipse.edc.junit.extensions.EdcExtension;
 import org.eclipse.edc.spi.event.EventRouter;
@@ -44,13 +45,14 @@
 @ExtendWith(EdcExtension.class)
 public class AssetEventDispatchTest {
 
-    private final EventSubscriber eventSubscriber = mock(EventSubscriber.class);
+    private final EventSubscriber eventSubscriber = mock();
 
     @BeforeEach
     void setUp(EdcExtension extension) {
-        extension.registerServiceMock(ProtocolWebhook.class, mock(ProtocolWebhook.class));
-        extension.registerServiceMock(DataPlaneInstanceStore.class, mock(DataPlaneInstanceStore.class));
+        extension.registerServiceMock(ProtocolWebhook.class, mock());
+        extension.registerServiceMock(DataPlaneInstanceStore.class, mock());
         extension.registerServiceMock(IdentityService.class, mock());
+        extension.registerServiceMock(DataPlaneClientFactory.class, mock());
         extension.setConfiguration(Map.of(
                 "web.http.port", String.valueOf(getFreePort()),
                 "web.http.path", "/api"

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/contractdefinition/ContractDefinitionEventDispatchTest.java

@@ -19,6 +19,7 @@
 import org.eclipse.edc.connector.controlplane.contract.spi.event.contractdefinition.ContractDefinitionEvent;
 import org.eclipse.edc.connector.controlplane.contract.spi.types.offer.ContractDefinition;
 import org.eclipse.edc.connector.controlplane.services.spi.contractdefinition.ContractDefinitionService;
+import org.eclipse.edc.connector.dataplane.selector.spi.client.DataPlaneClientFactory;
 import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
 import org.eclipse.edc.junit.extensions.EdcExtension;
 import org.eclipse.edc.spi.event.EventRouter;
@@ -49,6 +50,7 @@ void setUp(EdcExtension extension) {
         extension.registerServiceMock(ProtocolWebhook.class, mock(ProtocolWebhook.class));
         extension.registerServiceMock(DataPlaneInstanceStore.class, mock(DataPlaneInstanceStore.class));
         extension.registerServiceMock(IdentityService.class, mock());
+        extension.registerServiceMock(DataPlaneClientFactory.class, mock());
         extension.setConfiguration(Map.of(
                 "web.http.port", String.valueOf(getFreePort()),
                 "web.http.path", "/api"

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/contractnegotiation/ContractNegotiationEventDispatchTest.java

@@ -28,6 +28,7 @@
 import org.eclipse.edc.connector.controlplane.policy.spi.PolicyDefinition;
 import org.eclipse.edc.connector.controlplane.policy.spi.store.PolicyDefinitionStore;
 import org.eclipse.edc.connector.controlplane.services.spi.contractnegotiation.ContractNegotiationProtocolService;
+import org.eclipse.edc.connector.dataplane.selector.spi.client.DataPlaneClientFactory;
 import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
 import org.eclipse.edc.junit.extensions.EdcExtension;
 import org.eclipse.edc.policy.model.Policy;
@@ -69,7 +70,7 @@
 class ContractNegotiationEventDispatchTest {
     private static final String CONSUMER = "consumer";
 
-    private final EventSubscriber eventSubscriber = mock(EventSubscriber.class);
+    private final EventSubscriber eventSubscriber = mock();
     private final IdentityService identityService = mock();
     private final ClaimToken token = ClaimToken.Builder.newInstance().claim(ParticipantAgentService.DEFAULT_IDENTITY_CLAIM_KEY, CONSUMER).build();
 
@@ -85,9 +86,10 @@ void setUp(EdcExtension extension) {
                 "edc.negotiation.provider.send.retry.limit", "0"
         ));
         extension.registerServiceMock(NegotiationWaitStrategy.class, () -> 1);
-        extension.registerServiceMock(ProtocolWebhook.class, mock(ProtocolWebhook.class));
-        extension.registerServiceMock(DataPlaneInstanceStore.class, mock(DataPlaneInstanceStore.class));
+        extension.registerServiceMock(ProtocolWebhook.class, mock());
+        extension.registerServiceMock(DataPlaneInstanceStore.class, mock());
         extension.registerServiceMock(IdentityService.class, identityService);
+        extension.registerServiceMock(DataPlaneClientFactory.class, mock());
     }
 
     @Test

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/policydefinition/PolicyDefinitionEventDispatchTest.java

@@ -20,6 +20,7 @@
 import org.eclipse.edc.connector.controlplane.policy.spi.event.PolicyDefinitionEvent;
 import org.eclipse.edc.connector.controlplane.policy.spi.event.PolicyDefinitionUpdated;
 import org.eclipse.edc.connector.controlplane.services.spi.policydefinition.PolicyDefinitionService;
+import org.eclipse.edc.connector.dataplane.selector.spi.client.DataPlaneClientFactory;
 import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
 import org.eclipse.edc.junit.extensions.EdcExtension;
 import org.eclipse.edc.policy.model.Policy;
@@ -47,9 +48,10 @@ public class PolicyDefinitionEventDispatchTest {
 
     @BeforeEach
     void setUp(EdcExtension extension) {
-        extension.registerServiceMock(ProtocolWebhook.class, mock(ProtocolWebhook.class));
-        extension.registerServiceMock(DataPlaneInstanceStore.class, mock(DataPlaneInstanceStore.class));
+        extension.registerServiceMock(ProtocolWebhook.class, mock());
+        extension.registerServiceMock(DataPlaneInstanceStore.class, mock());
         extension.registerServiceMock(IdentityService.class, mock());
+        extension.registerServiceMock(DataPlaneClientFactory.class, mock());
         extension.setConfiguration(Map.of(
                 "web.http.port", String.valueOf(getFreePort()),
                 "web.http.path", "/api")

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/secret/SecretEventDispatchTest.java

@@ -14,6 +14,7 @@
 
 package org.eclipse.edc.connector.controlplane.services.secret;
 
+import org.eclipse.edc.connector.dataplane.selector.spi.client.DataPlaneClientFactory;
 import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
 import org.eclipse.edc.connector.secret.spi.event.SecretCreated;
 import org.eclipse.edc.connector.secret.spi.event.SecretDeleted;
@@ -41,13 +42,14 @@
 @ExtendWith(EdcExtension.class)
 public class SecretEventDispatchTest {
 
-    private final EventSubscriber eventSubscriber = mock(EventSubscriber.class);
+    private final EventSubscriber eventSubscriber = mock();
 
     @BeforeEach
     void setUp(EdcExtension extension) {
-        extension.registerServiceMock(ProtocolWebhook.class, mock(ProtocolWebhook.class));
-        extension.registerServiceMock(DataPlaneInstanceStore.class, mock(DataPlaneInstanceStore.class));
+        extension.registerServiceMock(ProtocolWebhook.class, mock());
+        extension.registerServiceMock(DataPlaneInstanceStore.class, mock());
         extension.registerServiceMock(IdentityService.class, mock());
+        extension.registerServiceMock(DataPlaneClientFactory.class, mock());
         extension.setConfiguration(Map.of(
                 "web.http.port", String.valueOf(getFreePort()),
                 "web.http.path", "/api"

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/transferprocess/TransferProcessEventDispatchTest.java

@@ -35,6 +35,7 @@
 import org.eclipse.edc.connector.controlplane.transfer.spi.types.protocol.TransferProcessAck;
 import org.eclipse.edc.connector.controlplane.transfer.spi.types.protocol.TransferStartMessage;
 import org.eclipse.edc.connector.core.event.EventExecutorServiceContainer;
+import org.eclipse.edc.connector.dataplane.selector.spi.client.DataPlaneClientFactory;
 import org.eclipse.edc.junit.extensions.EdcExtension;
 import org.eclipse.edc.policy.model.Policy;
 import org.eclipse.edc.spi.EdcException;
@@ -86,7 +87,7 @@
 public class TransferProcessEventDispatchTest {
 
     public static final Duration TIMEOUT = Duration.ofSeconds(30);
-    private final EventSubscriber eventSubscriber = mock(EventSubscriber.class);
+    private final EventSubscriber eventSubscriber = mock();
     private final IdentityService identityService = mock();
 
     @NotNull
@@ -112,9 +113,10 @@ void setUp(EdcExtension extension) {
         extension.registerServiceMock(EventExecutorServiceContainer.class, new EventExecutorServiceContainer(newSingleThreadExecutor()));
         extension.registerServiceMock(IdentityService.class, identityService);
         extension.registerServiceMock(ProtocolWebhook.class, () -> "http://dummy");
-        extension.registerServiceMock(PolicyArchive.class, mock(PolicyArchive.class));
-        extension.registerServiceMock(ContractNegotiationStore.class, mock(ContractNegotiationStore.class));
-        extension.registerServiceMock(ParticipantAgentService.class, mock(ParticipantAgentService.class));
+        extension.registerServiceMock(PolicyArchive.class, mock());
+        extension.registerServiceMock(ContractNegotiationStore.class, mock());
+        extension.registerServiceMock(ParticipantAgentService.class, mock());
+        extension.registerServiceMock(DataPlaneClientFactory.class, mock());
         var dataAddressValidatorRegistry = mock(DataAddressValidatorRegistry.class);
         when(dataAddressValidatorRegistry.validateSource(any())).thenReturn(ValidationResult.success());
         when(dataAddressValidatorRegistry.validateDestination(any())).thenReturn(ValidationResult.success());

core/control-plane/control-plane-transfer/src/main/java/org/eclipse/edc/connector/controlplane/transfer/TransferProcessDefaultServicesExtension.java

@@ -46,8 +46,8 @@ public String name() {
     }
 
     @Provider
-    public DataFlowManager dataFlowManager() {
-        return new DataFlowManagerImpl();
+    public DataFlowManager dataFlowManager(ServiceExtensionContext context) {
+        return new DataFlowManagerImpl(context.getMonitor());
     }
 
     @Provider

core/control-plane/control-plane-transfer/src/main/java/org/eclipse/edc/connector/controlplane/transfer/flow/DataFlowManagerImpl.java

@@ -21,6 +21,7 @@
 import org.eclipse.edc.connector.controlplane.transfer.spi.types.DataFlowResponse;
 import org.eclipse.edc.connector.controlplane.transfer.spi.types.TransferProcess;
 import org.eclipse.edc.policy.model.Policy;
+import org.eclipse.edc.spi.monitor.Monitor;
 import org.eclipse.edc.spi.response.StatusResult;
 import org.jetbrains.annotations.NotNull;
 
@@ -41,6 +42,11 @@
 public class DataFlowManagerImpl implements DataFlowManager {
 
     private final List<PrioritizedDataFlowController> controllers = new ArrayList<>();
+    private final Monitor monitor;
+
+    public DataFlowManagerImpl(Monitor monitor) {
+        this.monitor = monitor;
+    }
 
     @Override
     public void register(DataFlowController controller) {
@@ -58,7 +64,9 @@ public void register(int priority, DataFlowController controller) {
         try {
             return chooseControllerAndApply(transferProcess, controller -> controller.start(transferProcess, policy));
         } catch (Exception e) {
-            return StatusResult.failure(FATAL_ERROR, runtimeException(transferProcess.getId(), e.getLocalizedMessage()));
+            var message = runtimeException(transferProcess.getId(), e.getMessage());
+            monitor.severe(message, e);
+            return StatusResult.failure(FATAL_ERROR, message);
         }
     }
 

core/control-plane/control-plane-transfer/src/test/java/org/eclipse/edc/connector/controlplane/transfer/flow/DataFlowManagerImplTest.java

@@ -38,7 +38,7 @@
 
 class DataFlowManagerImplTest {
 
-    private final DataFlowManagerImpl manager = new DataFlowManagerImpl();
+    private final DataFlowManagerImpl manager = new DataFlowManagerImpl(mock());
 
     @Nested
     class Initiate {

core/data-plane-selector/data-plane-selector-core/build.gradle.kts

@@ -18,12 +18,17 @@ plugins {
 
 dependencies {
     api(project(":spi:data-plane-selector:data-plane-selector-spi"))
-    implementation(project(":spi:common:transaction-spi"))
+    api(project(":spi:common:transaction-spi"))
+
+    implementation(project(":core:common:lib:query-lib"))
+    implementation(project(":core:common:lib:state-machine-lib"))
+    implementation(project(":core:common:lib:store-lib"))
     implementation(project(":core:common:lib:util-lib"))
 
     testImplementation(project(":core:common:junit"))
     testImplementation(testFixtures(project(":spi:data-plane-selector:data-plane-selector-spi")))
-    testImplementation(project(":core:common:junit"))
+
+    testImplementation(libs.awaitility)
 }
 
 

core/data-plane-selector/data-plane-selector-core/src/main/java/org/eclipse/edc/connector/dataplane/selector/DataPlaneSelectorDefaultServicesExtension.java

@@ -20,9 +20,13 @@
 import org.eclipse.edc.connector.dataplane.selector.store.InMemoryDataPlaneInstanceStore;
 import org.eclipse.edc.connector.dataplane.selector.strategy.DefaultSelectionStrategyRegistry;
 import org.eclipse.edc.runtime.metamodel.annotation.Extension;
+import org.eclipse.edc.runtime.metamodel.annotation.Inject;
 import org.eclipse.edc.runtime.metamodel.annotation.Provider;
+import org.eclipse.edc.spi.query.CriterionOperatorRegistry;
 import org.eclipse.edc.spi.system.ServiceExtension;
 
+import java.time.Clock;
+
 /**
  * Provides default service implementations for fallback
  * Omitted {@link Extension} since this module contains the extension {@link DataPlaneSelectorExtension}
@@ -31,14 +35,20 @@ public class DataPlaneSelectorDefaultServicesExtension implements ServiceExtensi
 
     public static final String NAME = "Data Plane Selector Default Services";
 
+    @Inject
+    private Clock clock;
+
+    @Inject
+    private CriterionOperatorRegistry criterionOperatorRegistry;
+
     @Override
     public String name() {
         return NAME;
     }
 
     @Provider(isDefault = true)
     public DataPlaneInstanceStore instanceStore() {
-        return new InMemoryDataPlaneInstanceStore();
+        return new InMemoryDataPlaneInstanceStore(clock, criterionOperatorRegistry);
     }
 
     @Provider(isDefault = true)

core/data-plane-selector/data-plane-selector-core/src/main/java/org/eclipse/edc/connector/dataplane/selector/DataPlaneSelectorExtension.java

@@ -14,27 +14,91 @@
 
 package org.eclipse.edc.connector.dataplane.selector;
 
+import org.eclipse.edc.connector.dataplane.selector.manager.DataPlaneSelectorManagerImpl;
 import org.eclipse.edc.connector.dataplane.selector.service.EmbeddedDataPlaneSelectorService;
 import org.eclipse.edc.connector.dataplane.selector.spi.DataPlaneSelectorService;
+import org.eclipse.edc.connector.dataplane.selector.spi.client.DataPlaneClientFactory;
+import org.eclipse.edc.connector.dataplane.selector.spi.manager.DataPlaneSelectorManager;
 import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
 import org.eclipse.edc.connector.dataplane.selector.spi.strategy.SelectionStrategyRegistry;
 import org.eclipse.edc.runtime.metamodel.annotation.Extension;
 import org.eclipse.edc.runtime.metamodel.annotation.Inject;
 import org.eclipse.edc.runtime.metamodel.annotation.Provider;
+import org.eclipse.edc.runtime.metamodel.annotation.Setting;
+import org.eclipse.edc.spi.retry.ExponentialWaitStrategy;
 import org.eclipse.edc.spi.system.ServiceExtension;
+import org.eclipse.edc.spi.system.ServiceExtensionContext;
 import org.eclipse.edc.transaction.spi.TransactionContext;
 
-@Extension(value = "DataPlane core selector")
+import java.time.Duration;
+
+import static org.eclipse.edc.connector.dataplane.selector.DataPlaneSelectorExtension.NAME;
+import static org.eclipse.edc.statemachine.AbstractStateEntityManager.DEFAULT_BATCH_SIZE;
+import static org.eclipse.edc.statemachine.AbstractStateEntityManager.DEFAULT_ITERATION_WAIT;
+
+@Extension(NAME)
 public class DataPlaneSelectorExtension implements ServiceExtension {
 
+    public static final String NAME = "Data Plane Selector core";
+
+    private static final int DEFAULT_CHECK_PERIOD = 60;
+
+    @Setting(value = "the iteration wait time in milliseconds in the data plane selector state machine.", defaultValue = DEFAULT_ITERATION_WAIT + "", type = "long")
+    private static final String DATA_PLANE_SELECTOR_STATE_MACHINE_ITERATION_WAIT_MILLIS = "edc.data.plane.selector.state-machine.iteration-wait-millis";
+
+    @Setting(value = "the batch size in the data plane selector state machine.", defaultValue = DEFAULT_BATCH_SIZE + "", type = "int")
+    private static final String DATA_PLANE_SELECTOR_STATE_MACHINE_BATCH_SIZE = "edc.data.plane.selector.state-machine.batch-size";
+
+    @Setting(value = "the check period for data plane availability, in seconds", defaultValue = DEFAULT_CHECK_PERIOD + "", type = "int")
+    private static final String DATA_PLANE_SELECTOR_CHECK_PERIOD = "edc.data.plane.selector.state-machine.check.period";
+
     @Inject
     private DataPlaneInstanceStore instanceStore;
-
     @Inject
     private TransactionContext transactionContext;
-
     @Inject
     private SelectionStrategyRegistry selectionStrategyRegistry;
+    @Inject
+    private DataPlaneClientFactory clientFactory;
+
+    private DataPlaneSelectorManager manager;
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @Override
+    public void initialize(ServiceExtensionContext context) {
+        var config = context.getConfig();
+        var iterationWait = config.getLong(DATA_PLANE_SELECTOR_STATE_MACHINE_ITERATION_WAIT_MILLIS, DEFAULT_ITERATION_WAIT);
+        var checkPeriod = config.getInteger(DATA_PLANE_SELECTOR_CHECK_PERIOD, DEFAULT_CHECK_PERIOD);
+
+        var configuration = new DataPlaneSelectorManagerConfiguration(
+                new ExponentialWaitStrategy(iterationWait),
+                config.getInteger(DATA_PLANE_SELECTOR_STATE_MACHINE_BATCH_SIZE, DEFAULT_BATCH_SIZE),
+                Duration.ofSeconds(checkPeriod)
+        );
+
+        manager = DataPlaneSelectorManagerImpl.Builder.newInstance()
+                .clientFactory(clientFactory)
+                .store(instanceStore)
+                .monitor(context.getMonitor())
+                .configuration(configuration)
+                .build();
+    }
+
+    @Override
+    public void start() {
+        manager.start();
+    }
+
+    @Override
+    public void shutdown() {
+        if (manager != null) {
+            manager.stop();
+        }
+    }
 
     @Provider
     public DataPlaneSelectorService dataPlaneSelectorService() {