Feature/100 implement cosmos based transferprocessstore (#102)

* added cosmos-based transfer process store

* added barebones cosmos-db-based transferprocess store

* fixed GH actions config error

* fixed missing line breaks

* fixed serialization problems with TransferProcesses

* made DataEntry non-generic

* fixed i-tests

* added more tests for the CosmosDbStore

* re-enabled terraform plan

* re-enabled terraform plan

* really reenabled...

* disabled wronly enabled test

* replaced nextForState with a transfaction-safe storedProcedure

* removed explicit lock/unlock feature

* only run test in CI

* fixed terraform format, removed secret

* same connector should be able to renew the lease

* deploy the connector instance only after the storage account is ready

* code cosmetics

* try to run terraform after docker images are built

* removed branch restriction

* disabled azure vault tests

* removed commented code

* reenabled terraform

* re-enabled taint and apply steps on pushes on main

* added lease/break-lease behaviour to all modifying calls

* code cleanup. added a retry policy that retries after request throttling

* added unique connectorId to the lease mechanism

* fixed CI switch

* PR updates

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

Author: paullatzelsperger

.github/workflows/terraform.yaml

@@ -44,6 +44,10 @@
 
 name: 'Terraform'
 
+# on:
+#   workflow_run:
+#     workflows: [ "Build Docker Images" ]
+#     types: [ completed ]
 on:
   push:
     branches:
@@ -54,7 +58,8 @@ jobs:
   terraform:
     name: 'Terraform'
     runs-on: ubuntu-latest
-    environment: production
+    environment: dev
+
     env:
       ARM_CLIENT_ID: ${{ secrets.AZURE_AD_CIENT_ID }}
       ARM_CLIENT_SECRET: ${{ secrets.AZURE_AD_CLIENT_SECRET }}
@@ -86,19 +91,18 @@ jobs:
         id: validate
         run: terraform -chdir=scripts validate -no-color
 
-      #     deactivated until the GithubActions-User has correct credentials
-#      - name: Terraform Plan
-#        id: plan
-#        if: github.event_name == 'pull_request'
-#        run: terraform -chdir=scripts plan -no-color -var "resourcesuffix=dev" -var "backend_account_key=${{ secrets.TF_BACKEND_KEY }}"
+      - name: Terraform Plan
+        id: plan
+        run: terraform -chdir=scripts plan -no-color -var "resourcesuffix=dev" -var "backend_account_key=${{ secrets.TF_BACKEND_KEY }}"
 
 
+      - name: Taint the connector instance
+        id: taint
+        if: github.ref == 'refs/heads/main' && github.event_name == 'push'
+        run: terraform -chdir=scripts taint azurerm_container_group.connector-instance
+        continue-on-error: true
 
-#      - name: Taint the connector instance
-#        if: github.ref == 'refs/heads/main' && github.event_name == 'push'
-#        run: terraform -chdir=scripts taint azurerm_container_group.connector-instance
-#        continue-on-error: true
-#
-#      - name: Terraform Apply
-#        if: github.ref == 'refs/heads/main' && github.event_name == 'push'
-#        run: terraform -chdir=scripts apply -var "resourcesuffix=dev" -var "backend_account_key=${{ secrets.TF_BACKEND_KEY }}" -auto-approve
+      - name: Terraform Apply
+        id: apply
+        if: github.ref == 'refs/heads/main' && github.event_name == 'push'
+        run: terraform -chdir=scripts apply -var "resourcesuffix=dev" -var "backend_account_key=${{ secrets.TF_BACKEND_KEY }}" -auto-approve

.github/workflows/verify.yaml

@@ -22,7 +22,7 @@ jobs:
           NIFI_API_AUTH: ${{ secrets.NIFI_API_AUTH }}
           S3_SECRET_ACCESS_KEY: ${{ secrets.S3_SECRET_ACCESS_KEY }}
           S3_ACCESS_KEY_ID: ${{ secrets.S3_ACCESS_KEY_ID }}
-        run: ./gradlew clean check -x :extensions:catalog:catalog-atlas:check -x :extensions:security:security-azure:check  -x :extensions:transfer:transfer-nifi:check -x :external:nifi:processors:check
+        run: ./gradlew clean check -x :extensions:catalog:catalog-atlas:check -x :extensions:security:security-azure:check  -x :extensions:transfer:transfer-nifi:check -x :external:nifi:processors:check -x extensions:transfer:transfer-store-cosmos:check
 
       - name: Publish Unit Test Results
         id: publish-results
@@ -89,7 +89,14 @@ jobs:
         run: ./gradlew external:nifi:processors:check
 
       - name: Test Azure Vault Integration
+        if: ${{ false }} # disabled for now, because it's VERY slow
         id: azure-vault-tests
         env:
           AZ_STORAGE_SAS: ${{ secrets.AZ_STORAGE_SAS }}
-        run: ./gradlew extensions:security:security-azure:check
+        run: ./gradlew extensions:security:security-azure:check
+
+      - name: Test Cosmos-based TransferProcessStore
+        id: cosmos-transferprocessstore-test
+        env:
+          COSMOS_KEY: ${{ secrets.COSMOS_DB_MASTERKEY }}
+        run: ./gradlew extensions:transfer:transfer-store-cosmos:check

.gitignore

@@ -48,3 +48,5 @@ distributions/azure/azure.properties
 **/secrets
 kubeconfig
 **/dagx-config.properties
+
+scripts/terraform.tfvars

distributions/demo-e2e/Dockerfile

@@ -15,4 +15,6 @@ ENTRYPOINT java \
     -Ddagx.atlas.url=${ATLAS_URL} \
     -Ddagx.nifi.url=${NIFI_URL} \
     -Ddagx.nifi.flow.url=${NIFI_FLOW_URL} \
+    -Ddagx.cosmos.account.name=${COSMOS_ACCOUNT} \
+    -Ddagx.cosmos.database.name=${COSMOS_DB} \
     -Djava.security.edg=file:/dev/.urandom -jar dagx-demo-e2e.jar

distributions/demo-e2e/build.gradle.kts

@@ -17,7 +17,7 @@ dependencies {
     implementation(project(":extensions:control-http"))
 
     implementation(project(":extensions:transfer:transfer-core"))
-    implementation(project(":extensions:transfer:transfer-store-memory"))
+    implementation(project(":extensions:transfer:transfer-store-cosmos"))
     implementation(project(":extensions:transfer:transfer-provision-aws"))
     implementation(project(":extensions:transfer:transfer-provision-azure"))
     implementation(project(":extensions:transfer:transfer-nifi"))

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

@@ -5,14 +5,18 @@
 
 package com.microsoft.dagx.catalog.atlas.metadata;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.microsoft.dagx.spi.types.domain.metadata.DataCatalogEntry;
 import com.microsoft.dagx.spi.types.domain.transfer.DataAddress;
 
+@JsonTypeName("dagx:atlascatalogentry")
 public class AtlasDataCatalogEntry implements DataCatalogEntry {
 
+    @JsonProperty
     private final DataAddress address;
 
-    public AtlasDataCatalogEntry(DataAddress address) {
+    public AtlasDataCatalogEntry(@JsonProperty("address") DataAddress address) {
 
         this.address = address;
     }

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

@@ -49,6 +49,8 @@ public void initialize(ServiceExtensionContext context) {
         context.registerService(AtlasApi.class, api);
         context.registerService(MetadataStore.class, new AtlasMetadataStore(api, context.getMonitor(), context.getService(SchemaRegistry.class)));
         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

@@ -45,7 +45,7 @@ public AtlasMetadataStore(AtlasApi atlasApi, Monitor monitor, SchemaRegistry sch
     }
 
     @Override
-    public @Nullable DataEntry<?> findForId(String id) {
+    public @Nullable DataEntry findForId(String id) {
 
         var properties = atlasApi.getEntityById(id);
 
@@ -100,12 +100,12 @@ private String getPolicyIdForEntity(AtlasEntity entry) {
     }
 
     @Override
-    public void save(DataEntry<?> entry) {
+    public void save(DataEntry entry) {
         monitor.severe("Save not yet implemented");
     }
 
     @Override
-    public @NotNull Collection<DataEntry<?>> queryAll(Collection<Policy> policies) {
+    public @NotNull Collection<DataEntry> queryAll(Collection<Policy> policies) {
         if (policies.isEmpty()) {
             return Collections.emptyList();
         }

extensions/catalog/catalog-atlas/src/test/java/com/microsoft/dagx/catalog/atlas/metadata/AtlasMetadataStoreTest.java

@@ -174,7 +174,7 @@ void queryAll() {
         replay(atlasApiMock);
         replay(monitorMock);
 
-        final Collection<DataEntry<?>> entries = atlasMetadataStore.queryAll(Collections.singleton(policy));
+        final Collection<DataEntry> entries = atlasMetadataStore.queryAll(Collections.singleton(policy));
         assertThat(entries).isNotNull().isNotEmpty().doesNotContainNull();
 
         assertThat(entries).allSatisfy(this::assertAzureEntry);
@@ -232,7 +232,7 @@ private AtlasEntity createS3Entity(String name) {
                 "region", "neverland"));
     }
 
-    private void assertAzureEntry(DataEntry<?> entry) {
+    private void assertAzureEntry(DataEntry entry) {
         assertThat(entry.getCatalogEntry().getAddress().getProperties()).isNotNull()
                 .hasFieldOrPropertyWithValue("keyName", KEY_NAME)
                 .hasFieldOrPropertyWithValue("type", AzureBlobStoreSchema.TYPE)

extensions/demo/demo-nifi/src/main/java/com/microsoft/dagx/demo/nifi/NifiDemoServiceExtension.java

@@ -12,7 +12,6 @@
 import com.microsoft.dagx.spi.policy.PolicyRegistry;
 import com.microsoft.dagx.spi.system.ServiceExtension;
 import com.microsoft.dagx.spi.system.ServiceExtensionContext;
-import com.microsoft.dagx.spi.types.domain.metadata.DataCatalogEntry;
 import com.microsoft.dagx.spi.types.domain.metadata.DataEntry;
 import com.microsoft.dagx.spi.types.domain.metadata.GenericDataCatalogEntry;
 
@@ -60,10 +59,10 @@ private void saveDataEntries() {
                 .property("type", AzureBlobStoreSchema.TYPE)
                 .build();
 
-        DataEntry<DataCatalogEntry> entry1 = DataEntry.Builder.newInstance().id("test123").policyId(USE_EU_POLICY).catalogEntry(sourceFileCatalog).build();
+        DataEntry entry1 = DataEntry.Builder.newInstance().id("test123").policyId(USE_EU_POLICY).catalogEntry(sourceFileCatalog).build();
         metadataStore.save(entry1);
 
-        DataEntry<DataCatalogEntry> entry2 = DataEntry.Builder.newInstance().id("test456").policyId(USE_US_OR_EU_POLICY).catalogEntry(sourceFileCatalog).build();
+        DataEntry entry2 = DataEntry.Builder.newInstance().id("test456").policyId(USE_US_OR_EU_POLICY).catalogEntry(sourceFileCatalog).build();
         metadataStore.save(entry2);
     }