the connectorID is now either configured via a config entry, or assigned randomly

Author: paullatzelsperger

common/src/main/java/com/microsoft/dagx/common/settings/SettingsHelper.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ *  All rights reserved.
+ *
+ */
+
+package com.microsoft.dagx.common.settings;
+
+import com.microsoft.dagx.common.string.StringUtils;
+import com.microsoft.dagx.spi.system.ServiceExtensionContext;
+
+import java.util.UUID;
+
+public class SettingsHelper {
+    private static String connectorId;
+
+    /**
+     * Fetches the unique ID of the connector. If the {@code dagx.ids.connector.name} config value has been set, that value
+     * is returned, else a random name is chosen.
+     * The connector id is non-transient, that means two subsequent calls will produce the same result.
+     */
+    public static synchronized String getConnectorId(ServiceExtensionContext context) {
+        if (StringUtils.isNullOrEmpty(connectorId)) {
+            connectorId = context.getSetting("dagx.ids.connector.name", "dagx-connector-" + UUID.randomUUID());
+        }
+        return connectorId;
+    }
+}

extensions/events/events-azure/build.gradle.kts

@@ -12,6 +12,7 @@ val eventGridSdkVersion: String by project
 dependencies {
     api(project(":spi"))
     implementation(project(":extensions:schema"))
+    implementation(project(":common"))
     implementation("com.azure:azure-messaging-eventgrid:${eventGridSdkVersion}")
 }
 

extensions/events/events-azure/src/main/java/com/microsoft/dagx/events/azure/AzureEventExtension.java

@@ -17,6 +17,8 @@
 import java.util.Objects;
 import java.util.Set;
 
+import static com.microsoft.dagx.common.settings.SettingsHelper.getConnectorId;
+
 public class AzureEventExtension implements ServiceExtension {
 
     private static final String DEFAULT_TOPIC_NAME = "connector-events";
@@ -64,7 +66,8 @@ private void registerListeners(ServiceExtensionContext context) {
                 .endpoint(endpoint)
                 .buildEventGridEventPublisherAsyncClient();
 
-        final AzureEventGridPublisher publisher = new AzureEventGridPublisher(monitor, publisherClient);
+
+        final AzureEventGridPublisher publisher = new AzureEventGridPublisher(getConnectorId(context), monitor, publisherClient);
 
         var processObservable = context.getService(TransferProcessObservable.class, true);
         if (processObservable != null) {

extensions/events/events-azure/src/main/java/com/microsoft/dagx/events/azure/AzureEventGridPublisher.java

@@ -24,15 +24,17 @@ class AzureEventGridPublisher implements TransferProcessListener, MetadataListen
     private final EventGridPublisherAsyncClient<EventGridEvent> client;
     private final String eventTypeTransferprocess = "dagx/transfer/transferprocess";
     private final String eventTypeMetadata = "dagx/metadata/store";
+    private final String connectorId;
 
-    public AzureEventGridPublisher(Monitor monitor, EventGridPublisherAsyncClient<EventGridEvent> client) {
+    public AzureEventGridPublisher(String connectorId, Monitor monitor, EventGridPublisherAsyncClient<EventGridEvent> client) {
+        this.connectorId = connectorId;
         this.monitor = monitor;
         this.client = client;
     }
 
     @Override
     public void created(TransferProcess process) {
-        var dto = createDto(process);
+        var dto = createTransferProcessDto(process);
         if (process.getType() == TransferProcess.Type.CLIENT) {
             sendEvent("createdClient", eventTypeTransferprocess, dto).subscribe(new LoggingSubscriber<>("Transfer process created"));
         } else {
@@ -42,46 +44,46 @@ public void created(TransferProcess process) {
 
     @Override
     public void completed(TransferProcess process) {
-        sendEvent("completed", eventTypeTransferprocess, createDto(process)).subscribe(new LoggingSubscriber<>("Transfer process completed"));
+        sendEvent("completed", eventTypeTransferprocess, createTransferProcessDto(process)).subscribe(new LoggingSubscriber<>("Transfer process completed"));
     }
 
 
     @Override
     public void deprovisioned(TransferProcess process) {
-        sendEvent("deprovisioned", eventTypeTransferprocess, createDto(process)).subscribe(new LoggingSubscriber<>("Transfer process resources deprovisioned"));
+        sendEvent("deprovisioned", eventTypeTransferprocess, createTransferProcessDto(process)).subscribe(new LoggingSubscriber<>("Transfer process resources deprovisioned"));
 
     }
 
     @Override
     public void ended(TransferProcess process) {
-        sendEvent("ended", eventTypeTransferprocess, createDto(process)).subscribe(new LoggingSubscriber<>("Transfer process ended"));
+        sendEvent("ended", eventTypeTransferprocess, createTransferProcessDto(process)).subscribe(new LoggingSubscriber<>("Transfer process ended"));
 
     }
 
     @Override
     public void error(TransferProcess process) {
-        sendEvent("error", eventTypeTransferprocess, createDto(process)).subscribe(new LoggingSubscriber<>("Transfer process errored!"));
+        sendEvent("error", eventTypeTransferprocess, createTransferProcessDto(process)).subscribe(new LoggingSubscriber<>("Transfer process errored!"));
 
     }
 
     @Override
     public void querySubmitted() {
-        sendEvent("querySubmitted", eventTypeMetadata, null).subscribe(new LoggingSubscriber<>("query submitted"));
+        sendEvent("queryReceived", eventTypeMetadata, new EventDto(connectorId)).subscribe(new LoggingSubscriber<>("query submitted"));
     }
 
     @Override
     public void searchInitiated() {
-        sendEvent("searchInitiated", eventTypeMetadata, null).subscribe(new LoggingSubscriber<>("search initiated"));
+        sendEvent("searchInitiated", eventTypeMetadata, new EventDto(connectorId)).subscribe(new LoggingSubscriber<>("search initiated"));
     }
 
     @Override
     public void metadataItemAdded() {
-        sendEvent("itemAdded", eventTypeMetadata, null).subscribe(new LoggingSubscriber<>("AzureEventGrid: metadata item added"));
+        sendEvent("itemAdded", eventTypeMetadata, new EventDto(connectorId)).subscribe(new LoggingSubscriber<>("AzureEventGrid: metadata item added"));
     }
 
     @Override
     public void metadataItemUpdated() {
-        sendEvent("itemUpdated", eventTypeMetadata, null).subscribe(new LoggingSubscriber<>("metadata item updated"));
+        sendEvent("itemUpdated", eventTypeMetadata, new EventDto(connectorId)).subscribe(new LoggingSubscriber<>("metadata item updated"));
     }
 
     private Mono<Void> sendEvent(String what, String where, Object payload) {
@@ -91,8 +93,9 @@ private Mono<Void> sendEvent(String what, String where, Object payload) {
     }
 
     @NotNull
-    private TransferProcessDto createDto(TransferProcess process) {
+    private TransferProcessDto createTransferProcessDto(TransferProcess process) {
         return TransferProcessDto.Builder.newInstance()
+                .connector(connectorId)
                 .state(TransferProcessStates.from(process.getState()))
                 .requestId(process.getDataRequest().getId())
                 .type(process.getType())

extensions/events/events-azure/src/main/java/com/microsoft/dagx/events/azure/EventDto.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ *  All rights reserved.
+ *
+ */
+
+package com.microsoft.dagx.events.azure;
+
+public class EventDto {
+    private final String connectorId;
+
+    protected EventDto(String connectorId) {
+        this.connectorId = connectorId;
+    }
+
+    public String getConnectorId() {
+        return connectorId;
+    }
+}

extensions/events/events-azure/src/main/java/com/microsoft/dagx/events/azure/TransferProcessDto.java

@@ -7,6 +7,7 @@
 package com.microsoft.dagx.events.azure;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonPOJOBuilder;
 import com.microsoft.dagx.spi.types.domain.transfer.TransferProcess;
@@ -19,14 +20,19 @@
  * reasons of security.
  */
 @JsonDeserialize(builder = TransferProcessDto.Builder.class)
-public class TransferProcessDto {
+public class TransferProcessDto extends EventDto {
+    @JsonProperty("requestId")
     private String requestId;
+    @JsonProperty("transferProcessType")
     private TransferProcess.Type type;
+    @JsonProperty("transferProcessState")
     private String state;
+    @JsonProperty("transferProcessStateCode")
     private int stateCode;
 
-    private TransferProcessDto() {
+    private TransferProcessDto(String connectorId) {
 
+        super(connectorId);
     }
 
     public String getRequestId() {
@@ -45,12 +51,14 @@ public int getStateCode() {
         return stateCode;
     }
 
+
     @JsonPOJOBuilder(withPrefix = "")
     public static final class Builder {
         private String requestId;
         private TransferProcess.Type type;
         private String state;
         private int stateCode;
+        private String connectorId;
 
         private Builder() {
         }
@@ -78,12 +86,17 @@ public Builder state(TransferProcessStates state) {
 
 
         public TransferProcessDto build() {
-            TransferProcessDto transferProcessDto = new TransferProcessDto();
+            TransferProcessDto transferProcessDto = new TransferProcessDto(connectorId);
             transferProcessDto.state = state;
             transferProcessDto.type = type;
             transferProcessDto.requestId = requestId;
             transferProcessDto.stateCode = stateCode;
             return transferProcessDto;
         }
+
+        public Builder connector(String connectorId) {
+            this.connectorId = connectorId;
+            return this;
+        }
     }
 }

extensions/ids/ids-core/build.gradle.kts

@@ -13,10 +13,11 @@ plugins {
 
 dependencies {
     api(project(":spi"))
+    api(project(":common"))
     api(project(":extensions:ids:ids-spi"))
 
     api("de.fraunhofer.iais.eis.ids.infomodel:java:${infoModelVersion}")
-    
+
     implementation("jakarta.ws.rs:jakarta.ws.rs-api:${rsApi}")
     implementation(project(":policy:policy-engine"))
 }

extensions/ids/ids-core/src/main/java/com/microsoft/dagx/ids/core/IdsCoreServiceExtension.java

@@ -25,7 +25,7 @@
 
 import java.util.Set;
 
-import static com.microsoft.dagx.ids.spi.IdsConfiguration.CONNECTOR_NAME;
+import static com.microsoft.dagx.common.settings.SettingsHelper.getConnectorId;
 
 /**
  * Implements the IDS Controller REST API.
@@ -51,7 +51,7 @@ public void initialize(ServiceExtensionContext context) {
         context.registerService(IdsDescriptorService.class, descriptorService);
 
         var identityService = context.getService(IdentityService.class);
-        var connectorName = context.getSetting(CONNECTOR_NAME, "connectorName");
+        var connectorName = getConnectorId(context);
         var dapsService = new DapsServiceImpl(connectorName, identityService);
         context.registerService(DapsService.class, dapsService);
 

extensions/transfer/transfer-store-cosmos/src/main/java/com/microsoft/dagx/transfer/store/cosmos/CosmosTransferProcessStoreExtension.java

@@ -23,7 +23,8 @@
 
 import java.util.ArrayList;
 import java.util.Set;
-import java.util.UUID;
+
+import static com.microsoft.dagx.common.settings.SettingsHelper.getConnectorId;
 
 /**
  * Provides an in-memory implementation of the {@link com.microsoft.dagx.spi.transfer.store.TransferProcessStore} for testing.
@@ -91,7 +92,7 @@ public void initialize(ServiceExtensionContext context) {
         var partitionKey = context.getSetting(COSMOS_PARTITION_KEY_SETTING, DEFAULT_PARTITION_KEY);
 
         // get unique connector name
-        var connectorId = context.getSetting("dagx.ids.connector.name", "dagx-connector-" + UUID.randomUUID());
+        var connectorId = getConnectorId(context);
 
         monitor.info("CosmosTransferProcessStore will use connector id '" + connectorId + "'");
         context.registerService(TransferProcessStore.class, new CosmosTransferProcessStore(container, context.getTypeManager(), partitionKey, connectorId));