Implement DataPlaneInstanceManager

eclipse-edc · Jun 3, 2024 · 3c901cc · 3c901cc
1 parent 53776d8
commit 3c901cc
Show file tree

Hide file tree

Showing 36 changed files with 648 additions and 244 deletions.
diff --git a/...st/java/org/eclipse/edc/connector/controlplane/services/asset/AssetEventDispatchTest.java b/...st/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"

diff --git a/...nnector/controlplane/services/contractdefinition/ContractDefinitionEventDispatchTest.java b/...nnector/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"

diff --git a/...ector/controlplane/services/contractnegotiation/ContractNegotiationEventDispatchTest.java b/...ector/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

diff --git a/...c/connector/controlplane/services/policydefinition/PolicyDefinitionEventDispatchTest.java b/...c/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")

diff --git a/.../java/org/eclipse/edc/connector/controlplane/services/secret/SecretEventDispatchTest.java b/.../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"

diff --git a/...edc/connector/controlplane/services/transferprocess/TransferProcessEventDispatchTest.java b/...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());

diff --git a/core/data-plane-selector/data-plane-selector-core/build.gradle.kts b/core/data-plane-selector/data-plane-selector-core/build.gradle.kts
@@ -27,7 +27,8 @@ dependencies {
 
     testImplementation(project(":core:common:junit"))
     testImplementation(testFixtures(project(":spi:data-plane-selector:data-plane-selector-spi")))
-    testImplementation(project(":core:common:junit"))
+
+    testImplementation(libs.awaitility)
 }
 
 
diff --git a/...rc/main/java/org/eclipse/edc/connector/dataplane/selector/DataPlaneSelectorExtension.java b/...rc/main/java/org/eclipse/edc/connector/dataplane/selector/DataPlaneSelectorExtension.java
@@ -14,36 +14,77 @@
 
 package org.eclipse.edc.connector.dataplane.selector;
 
-import org.eclipse.edc.connector.dataplane.selector.manager.DataPlaneInstanceManagerImpl;
+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.manager.DataPlaneInstanceManager;
+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 DataPlaneInstanceManager manager;
+    private DataPlaneSelectorManager manager;
+
+    @Override
+    public String name() {
+        return NAME;
+    }
 
     @Override
     public void initialize(ServiceExtensionContext context) {
-        manager = DataPlaneInstanceManagerImpl.Builder.newInstance()
+        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();
     }
 

diff --git a/...a/org/eclipse/edc/connector/dataplane/selector/DataPlaneSelectorManagerConfiguration.java b/...a/org/eclipse/edc/connector/dataplane/selector/DataPlaneSelectorManagerConfiguration.java
@@ -0,0 +1,26 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.dataplane.selector;
+
+import org.eclipse.edc.spi.retry.WaitStrategy;
+
+import java.time.Duration;
+
+public record DataPlaneSelectorManagerConfiguration(
+        WaitStrategy waitStrategy,
+        int batchSize,
+        Duration checkPeriod
+) {
+}
diff --git a/...va/org/eclipse/edc/connector/dataplane/selector/manager/DataPlaneInstanceManagerImpl.java b/...va/org/eclipse/edc/connector/dataplane/selector/manager/DataPlaneInstanceManagerImpl.java