added event sourcing and forwarding to Azure Event Grid (#106)

* added event sourcing and forwarding to Azure Event Grid

* supply default values for topic and topic endpoint

* added the topic props to the dockerfile and terraform

* added DTO for events

Co-authored-by: Paul Latzelsperger <[email protected]>

Author: paullatzelsperger

core/src/main/java/com/microsoft/dagx/system/DefaultServiceExtensionContext.java

@@ -98,6 +98,15 @@ public <T> T getService(Class<T> type) {
         return service;
     }
 
+    @Override
+    @SuppressWarnings("unchecked")
+    public <T> T getService(Class<T> type, boolean isOptional) {
+        if (!isOptional) {
+            return getService(type);
+        }
+        return (T) services.get(type);
+    }
+
     @Override
     public <T> void registerService(Class<T> type, T service) {
         services.put(type, service);

distributions/demo-e2e/Dockerfile

@@ -17,4 +17,6 @@ ENTRYPOINT java \
     -Ddagx.nifi.flow.url=${NIFI_FLOW_URL} \
     -Ddagx.cosmos.account.name=${COSMOS_ACCOUNT} \
     -Ddagx.cosmos.database.name=${COSMOS_DB} \
+    -Ddagx.events.topic.name=${TOPIC_NAME} \
+    -Ddagx.events.topic.endpoint=${TOPIC_ENDPOINT} \
     -Djava.security.edg=file:/dev/.urandom -jar dagx-demo-e2e.jar

distributions/demo-e2e/build.gradle.kts

@@ -22,9 +22,10 @@ dependencies {
     implementation(project(":extensions:transfer:transfer-provision-azure"))
     implementation(project(":extensions:transfer:transfer-nifi"))
 
+    implementation(project(":extensions:events:events-azure"))
+
     implementation(project(":extensions:ids"))
 
-    // todo: replace with atlas - but we need this for the time being to provide catalog entries
     implementation(project(":extensions:catalog:catalog-atlas"))
     implementation(project(":extensions:dataseed"))
 

extensions/catalog/catalog-atlas/src/main/java/com/microsoft/dagx/catalog/atlas/metadata/AtlasExtension.java

@@ -6,6 +6,7 @@
 package com.microsoft.dagx.catalog.atlas.metadata;
 
 import com.microsoft.dagx.schema.SchemaRegistry;
+import com.microsoft.dagx.spi.metadata.MetadataObservable;
 import com.microsoft.dagx.spi.metadata.MetadataStore;
 import com.microsoft.dagx.spi.security.Vault;
 import com.microsoft.dagx.spi.system.ServiceExtension;
@@ -24,7 +25,7 @@ public class AtlasExtension implements ServiceExtension {
 
     @Override
     public Set<String> provides() {
-        return Set.of(ATLAS_FEATURE);
+        return Set.of(ATLAS_FEATURE, "dagx:metadata-store-observable");
     }
 
     @Override
@@ -47,7 +48,10 @@ public void initialize(ServiceExtensionContext context) {
         var pwd = vault.resolveSecret(SECRET_ATLAS_PWD);
         var api = new AtlasApiImpl(atlasUrl, user, pwd, context.getService(OkHttpClient.class), context.getTypeManager());
         context.registerService(AtlasApi.class, api);
-        context.registerService(MetadataStore.class, new AtlasMetadataStore(api, context.getMonitor(), context.getService(SchemaRegistry.class)));
+        final AtlasMetadataStore store = new AtlasMetadataStore(api, context.getMonitor(), context.getService(SchemaRegistry.class));
+        context.registerService(MetadataStore.class, store);
+        context.registerService(MetadataObservable.class, store);
+
         context.getMonitor().info("Initialized Atlas API extension.");
 
         context.getTypeManager().registerTypes(AtlasDataCatalogEntry.class);

extensions/catalog/catalog-atlas/src/main/java/com/microsoft/dagx/catalog/atlas/metadata/AtlasMetadataStore.java

@@ -14,6 +14,8 @@
 import com.microsoft.dagx.schema.DataSchema;
 import com.microsoft.dagx.schema.SchemaRegistry;
 import com.microsoft.dagx.spi.DagxException;
+import com.microsoft.dagx.spi.metadata.MetadataListener;
+import com.microsoft.dagx.spi.metadata.MetadataObservable;
 import com.microsoft.dagx.spi.metadata.MetadataStore;
 import com.microsoft.dagx.spi.monitor.Monitor;
 import com.microsoft.dagx.spi.types.domain.metadata.DataEntry;
@@ -30,7 +32,7 @@
 import static java.util.stream.Collectors.toSet;
 
 @SuppressWarnings("unchecked")
-public class AtlasMetadataStore implements MetadataStore {
+public class AtlasMetadataStore extends MetadataObservable implements MetadataStore {
     private static final String ATLAS_PROPERTY_KEYNAME = "keyName";
     private static final String ATLAS_PROPERTY_TYPE = "type";
     private final AtlasApi atlasApi;
@@ -46,7 +48,7 @@ public AtlasMetadataStore(AtlasApi atlasApi, Monitor monitor, SchemaRegistry sch
 
     @Override
     public @Nullable DataEntry findForId(String id) {
-
+        getListeners().forEach(MetadataListener::searchInitiated);
         var properties = atlasApi.getEntityById(id);
 
         if (properties == null) {
@@ -101,11 +103,13 @@ private String getPolicyIdForEntity(AtlasEntity entry) {
 
     @Override
     public void save(DataEntry entry) {
+        getListeners().forEach(MetadataListener::metadataItemAdded);
         monitor.severe("Save not yet implemented");
     }
 
     @Override
     public @NotNull Collection<DataEntry> queryAll(Collection<Policy> policies) {
+        getListeners().forEach(MetadataListener::querySubmitted);
         if (policies.isEmpty()) {
             return Collections.emptyList();
         }

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

@@ -0,0 +1,18 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ * All rights reserved.
+ */
+
+plugins {
+    `java-library`
+}
+
+val eventGridSdkVersion: String by project
+
+dependencies {
+    api(project(":spi"))
+    implementation(project(":extensions:schema"))
+    implementation("com.azure:azure-messaging-eventgrid:${eventGridSdkVersion}")
+}
+
+

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

@@ -0,0 +1,93 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ *  All rights reserved.
+ *
+ */
+package com.microsoft.dagx.events.azure;
+
+import com.azure.core.credential.AzureKeyCredential;
+import com.azure.messaging.eventgrid.EventGridPublisherClientBuilder;
+import com.microsoft.dagx.spi.metadata.MetadataObservable;
+import com.microsoft.dagx.spi.monitor.Monitor;
+import com.microsoft.dagx.spi.security.Vault;
+import com.microsoft.dagx.spi.system.ServiceExtension;
+import com.microsoft.dagx.spi.system.ServiceExtensionContext;
+import com.microsoft.dagx.spi.transfer.TransferProcessObservable;
+
+import java.util.Objects;
+import java.util.Set;
+
+public class AzureEventExtension implements ServiceExtension {
+
+    private static final String DEFAULT_TOPIC_NAME = "connector-events";
+    private static final String DEFAULT_ENDPOINT_NAME_TEMPLATE = "https://%s.westeurope-1.eventgrid.azure.net/api/events";
+    private static final String TOPIC_NAME_SETTING = "dagx.events.topic.name";
+    private static final String TOPIC_ENDPOINT_SETTING = "dagx.events.topic.endpoint";
+    private Monitor monitor;
+
+    @Override
+    public Set<String> requires() {
+        return Set.of("dagx:transfer-process-observable");
+    }
+
+    @Override
+    public void initialize(ServiceExtensionContext context) {
+        monitor = context.getMonitor();
+
+        monitor.info("AzureEventsExtension: create event grid appender");
+        registerListeners(context);
+
+        monitor.info("Initialized Azure Events Extension");
+    }
+
+
+    @Override
+    public void start() {
+        monitor.info("Started Azure Events Extension");
+    }
+
+    @Override
+    public void shutdown() {
+        monitor.info("Shutdown Azure Events Extension");
+    }
+
+    private void registerListeners(ServiceExtensionContext context) {
+
+        var vault = context.getService(Vault.class);
+
+        var topicName = getTopic(context);
+        var endpoint = getEndpoint(context, topicName);
+        monitor.info("AzureEventExtension: will use topic endpoint " + endpoint);
+
+        var publisherClient = new EventGridPublisherClientBuilder()
+                .credential(new AzureKeyCredential(Objects.requireNonNull(vault.resolveSecret(topicName), "Did not find secret in vault: " + endpoint)))
+                .endpoint(endpoint)
+                .buildEventGridEventPublisherAsyncClient();
+
+        final AzureEventGridPublisher publisher = new AzureEventGridPublisher(monitor, publisherClient);
+
+        var processObservable = context.getService(TransferProcessObservable.class, true);
+        if (processObservable != null) {
+            processObservable.registerListener(publisher);
+        }
+
+        var metadataObservable = context.getService(MetadataObservable.class, true);
+        if (metadataObservable != null) {
+            metadataObservable.registerListener(publisher);
+        }
+
+
+    }
+
+    private String getTopic(ServiceExtensionContext context) {
+        return context.getSetting(TOPIC_NAME_SETTING, DEFAULT_TOPIC_NAME);
+    }
+
+    private String getEndpoint(ServiceExtensionContext context, String topicName) {
+        var ep = context.getSetting(TOPIC_ENDPOINT_SETTING, null);
+        if (ep == null) {
+            ep = String.format(DEFAULT_ENDPOINT_NAME_TEMPLATE, topicName);
+        }
+        return ep;
+    }
+}

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

@@ -0,0 +1,120 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ *  All rights reserved.
+ *
+ */
+
+package com.microsoft.dagx.events.azure;
+
+import com.azure.core.util.BinaryData;
+import com.azure.messaging.eventgrid.EventGridEvent;
+import com.azure.messaging.eventgrid.EventGridPublisherAsyncClient;
+import com.microsoft.dagx.spi.metadata.MetadataListener;
+import com.microsoft.dagx.spi.monitor.Monitor;
+import com.microsoft.dagx.spi.transfer.TransferProcessListener;
+import com.microsoft.dagx.spi.types.domain.transfer.TransferProcess;
+import com.microsoft.dagx.spi.types.domain.transfer.TransferProcessStates;
+import org.jetbrains.annotations.NotNull;
+import reactor.core.publisher.BaseSubscriber;
+import reactor.core.publisher.Mono;
+
+class AzureEventGridPublisher implements TransferProcessListener, MetadataListener {
+
+    private final Monitor monitor;
+    private final EventGridPublisherAsyncClient<EventGridEvent> client;
+    private final String eventTypeTransferprocess = "dagx/transfer/transferprocess";
+    private final String eventTypeMetadata = "dagx/metadata/store";
+
+    public AzureEventGridPublisher(Monitor monitor, EventGridPublisherAsyncClient<EventGridEvent> client) {
+        this.monitor = monitor;
+        this.client = client;
+    }
+
+    @Override
+    public void created(TransferProcess process) {
+        var dto = createDto(process);
+        if (process.getType() == TransferProcess.Type.CLIENT) {
+            sendEvent("createdClient", eventTypeTransferprocess, dto).subscribe(new LoggingSubscriber<>("Transfer process created"));
+        } else {
+            sendEvent("createdProvider", eventTypeTransferprocess, dto).subscribe(new LoggingSubscriber<>("Transfer process created"));
+        }
+    }
+
+    @Override
+    public void completed(TransferProcess process) {
+        sendEvent("completed", eventTypeTransferprocess, createDto(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"));
+
+    }
+
+    @Override
+    public void ended(TransferProcess process) {
+        sendEvent("ended", eventTypeTransferprocess, createDto(process)).subscribe(new LoggingSubscriber<>("Transfer process ended"));
+
+    }
+
+    @Override
+    public void error(TransferProcess process) {
+        sendEvent("error", eventTypeTransferprocess, createDto(process)).subscribe(new LoggingSubscriber<>("Transfer process errored!"));
+
+    }
+
+    @Override
+    public void querySubmitted() {
+        sendEvent("querySubmitted", eventTypeMetadata, null).subscribe(new LoggingSubscriber<>("query submitted"));
+    }
+
+    @Override
+    public void searchInitiated() {
+        sendEvent("searchInitiated", eventTypeMetadata, null).subscribe(new LoggingSubscriber<>("search initiated"));
+    }
+
+    @Override
+    public void metadataItemAdded() {
+        sendEvent("itemAdded", eventTypeMetadata, null).subscribe(new LoggingSubscriber<>("AzureEventGrid: metadata item added"));
+    }
+
+    @Override
+    public void metadataItemUpdated() {
+        sendEvent("itemUpdated", eventTypeMetadata, null).subscribe(new LoggingSubscriber<>("metadata item updated"));
+    }
+
+    private Mono<Void> sendEvent(String what, String where, Object payload) {
+        final BinaryData data = BinaryData.fromObject(payload);
+        var evt = new EventGridEvent(what, where, data, "0.1");
+        return client.sendEvent(evt);
+    }
+
+    @NotNull
+    private TransferProcessDto createDto(TransferProcess process) {
+        return TransferProcessDto.Builder.newInstance()
+                .state(TransferProcessStates.from(process.getState()))
+                .requestId(process.getDataRequest().getId())
+                .type(process.getType())
+                .build();
+    }
+
+    private class LoggingSubscriber<T> extends BaseSubscriber<T> {
+
+        private final String message;
+
+        LoggingSubscriber(String message) {
+            this.message = message;
+        }
+
+        @Override
+        protected void hookOnComplete() {
+            monitor.info("AzureEventGrid: " + message);
+        }
+
+        @Override
+        protected void hookOnError(@NotNull Throwable throwable) {
+            monitor.severe("Error during event publishing", throwable);
+        }
+    }
+}