Potential resolution of the file-based transfer

distributions/file/build.gradle.kts

@@ -0,0 +1,134 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ * All rights reserved.
+ */
+
+import com.bmuschko.gradle.docker.tasks.container.DockerCreateContainer
+import com.bmuschko.gradle.docker.tasks.container.DockerStartContainer
+import com.bmuschko.gradle.docker.tasks.image.DockerBuildImage
+import com.bmuschko.gradle.docker.tasks.image.DockerPushImage
+import com.bmuschko.gradle.docker.tasks.image.Dockerfile
+import java.io.FileInputStream
+import java.util.*
+
+plugins {
+ `java-library`
+ id("application")
+ id("com.github.johnrengelman.shadow") version "7.0.0"
+ id("com.bmuschko.docker-remote-api") version "6.7.0"
+}
+
+dependencies {
+ implementation(project(":runtime"))
+ implementation(project(":extensions:protocol:web"))
+ implementation(project(":extensions:control-http"))
+
+ implementation(project(":extensions:metadata:metadata-memory"))
+ implementation(project(":extensions:transfer:transfer-core"))
+ implementation(project(":extensions:transfer:transfer-store-memory"))
+ implementation(project(":extensions:demo:demo-file"))
+
+ implementation(project(":extensions:policy:policy-registry-memory"))
+ implementation(project(":extensions:iam:iam-mock"))
+ implementation(project(":extensions:ids:ids-policy-mock"))
+ implementation(project(":extensions:ids"))
+
+ implementation(project(":extensions:configuration:configuration-fs"))
+
+ testImplementation("org.junit.jupiter:junit-jupiter-api:5.5.2")
+ testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.5.2")
+
+}
+
+val filename = "github.properties"
+var email = ""
+var user = "microsoft"
+var pwd = ""
+var url = ""
+var imageName = ""
+
+// initializes variables
+tasks.register("initializer") {
+ val initializer by tasks
+ val configFile = project.file(filename)
+ if (!configFile.exists()) {
+ println("WARNING: No $filename file was found, default will be used. Publishing won't be available!")
+ } else {
+ val fis = FileInputStream(configFile)
+ val prop = Properties()
+ prop.load(fis)
+ email = prop.getProperty("email")
+ user = prop.getProperty("user")
+ pwd = prop.getProperty("password")
+ url = prop.getProperty("url")
+ }
+ imageName = "$user/dagx-file-demo:latest"
+
+ if (url != "") {
+ imageName = "$url/$imageName"
+ }
+
+ println("Will use the following docker config:")
+ println(" - URL: $url")
+ println(" - User: $user")
+ println(" - Email: $email")
+ println(" - Image: $imageName")
+
+}
+
+// generate docker file
+val createDockerfile by tasks.creating(Dockerfile::class) {
+ dependsOn("initializer")
+ from("openjdk:11-jre-slim")
+ runCommand("mkdir /app")
+ copyFile("./build/libs/dagx-file-demo.jar", "/app/dagx-file-demo.jar")
+
+ exposePort(8181)
+
+ entryPoint("java", "-Djava.security.egd=file:/dev/./urandom", "-jar", "/app/dagx-file-demo.jar")
+}
+
+// build the image
+val buildDemo by tasks.creating(DockerBuildImage::class) {
+ dependsOn("shadowJar", createDockerfile)
+ dockerFile.set(project.file("${buildDir}/docker/Dockerfile"))
+ inputDir.set(project.file("."))
+ images.add(imageName)
+}
+
+// create demo container
+val createDemoContainer by tasks.creating(DockerCreateContainer::class) {
+ dependsOn(buildDemo)
+ targetImageId(buildDemo.imageId)
+ hostConfig.portBindings.set(listOf("8181:8181"))
+ hostConfig.autoRemove.set(true)
+ containerName.set("dagx-file-demo")
+}
+
+// start runtime demo in docker
+val startDemo by tasks.creating(DockerStartContainer::class) {
+ dependsOn(createDemoContainer)
+ targetContainerId(createDemoContainer.containerId)
+}
+
+//publish to github
+val publishDemo by tasks.creating(DockerPushImage::class) {
+ dependsOn(buildDemo)
+
+ registryCredentials.email.set(email)
+ registryCredentials.username.set(user)
+ registryCredentials.password.set(pwd)
+ registryCredentials.url.set(imageName)
+ images.add(imageName)
+}
+
+application {
+ @Suppress("DEPRECATION")
+ mainClassName = "com.microsoft.dagx.runtime.DagxRuntime"
+}
+
+tasks.withType<com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar> {
+ exclude("**/pom.properties", "**/pom.xm")
+ mergeServiceFiles()
+ archiveFileName.set("dagx-file-demo.jar")
+}

extensions/demo/demo-file/build.gradle.kts

@@ -0,0 +1,16 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ * All rights reserved.
+ */
+
+val rsApi: String by project
+
+plugins {
+ `java-library`
+}
+
+dependencies {
+ api(project(":spi"))
+ implementation(project(":extensions:schema"))
+ implementation("jakarta.ws.rs:jakarta.ws.rs-api:${rsApi}")
+}

extensions/demo/demo-file/src/main/java/com/microsoft/dagx/demo/file/DemoFileApiController.java

@@ -0,0 +1,73 @@
+package com.microsoft.dagx.demo.file;
+
+import java.util.UUID;
+
+import com.microsoft.dagx.spi.monitor.Monitor;
+import com.microsoft.dagx.spi.transfer.TransferProcessManager;
+import com.microsoft.dagx.spi.types.domain.metadata.DataEntry;
+import com.microsoft.dagx.spi.types.domain.transfer.DataAddress;
+import com.microsoft.dagx.spi.types.domain.transfer.DataRequest;
+
+import com.microsoft.dagx.demo.file.schema.FileSchema;
+
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.UriInfo;
+
+@Consumes({MediaType.APPLICATION_JSON})
+@Produces({MediaType.APPLICATION_JSON})
+@Path("/files")
+public class DemoFileApiController {
+
+ private final TransferProcessManager processManager;
+ private final Monitor monitor;
+
+ public DemoFileApiController(TransferProcessManager processManager, Monitor monitor) {
+ this.processManager = processManager;
+ this.monitor = monitor;
+ }
+
+ // Request an artifact. E.g. to request test.txt and store it in /tmp directory run:
+ // http://localhost:8181/api/files/file1?file=test.txt&folder=%2ftmp&compress=false
+ @GET
+ @Path("{artifactId}")
+ public Response requestArtifact(@Context UriInfo uri,
+ @PathParam("artifactId") String artifactId,
+ @QueryParam("file") String fileName,
+ @QueryParam("compress") String compress,
+ @QueryParam("folder") String folderName) {
+
+ monitor.info("File '" + artifactId + "' request initialized");
+
+ var entry = DataEntry.Builder.newInstance().id(artifactId).build();
+ var connector = uri.getBaseUri().getHost() + ":" + uri.getBaseUri().getPort(); // connector should consume own request
+
+ var destination = DataAddress.Builder.newInstance()
+ .type(FileSchema.TYPE)
+ .property(FileSchema.TARGET_FILE_NAME, fileName)
+ .property(FileSchema.TARGET_DIRECTORY, folderName)
+ .property(FileSchema.IS_COMPRESSION_REQUESTED, compress)
+ .keyName("file")
+ .build();
+
+ var request = DataRequest.Builder.newInstance()
+ .id(UUID.randomUUID().toString())
+ .protocol("ids-rest")
+ .dataEntry(entry)
+ .dataDestination(destination)
+ .destinationType("dagx:file")
+ .connectorId(connector)
+ .connectorAddress("http://" + connector)
+ .build();
+
+ processManager.initiateClientRequest(request);
+ return Response.accepted().build();
+ }
+}

extensions/demo/demo-file/src/main/java/com/microsoft/dagx/demo/file/DemoFileExtension.java

@@ -0,0 +1,79 @@
+package com.microsoft.dagx.demo.file;
+
+import com.microsoft.dagx.demo.file.transfer.FileFlowController;
+import com.microsoft.dagx.demo.file.transfer.provision.FolderResourceDefinitionGenerator;
+import com.microsoft.dagx.demo.file.transfer.provision.RandomFileArtifactResourceDefinitionGenerator;
+import com.microsoft.dagx.demo.file.transfer.provision.artifact.RandomFileArtifactProvisioner;
+import com.microsoft.dagx.demo.file.transfer.provision.folder.FolderProvisioner;
+import com.microsoft.dagx.spi.metadata.MetadataStore;
+import com.microsoft.dagx.spi.monitor.Monitor;
+import com.microsoft.dagx.spi.policy.PolicyRegistry;
+import com.microsoft.dagx.spi.protocol.web.WebService;
+import com.microsoft.dagx.spi.system.ServiceExtension;
+import com.microsoft.dagx.spi.system.ServiceExtensionContext;
+import com.microsoft.dagx.spi.transfer.TransferProcessManager;
+import com.microsoft.dagx.spi.transfer.flow.DataFlowManager;
+import com.microsoft.dagx.spi.transfer.provision.ProvisionManager;
+import com.microsoft.dagx.spi.transfer.provision.ResourceManifestGenerator;
+import com.microsoft.dagx.spi.transfer.store.TransferProcessStore;
+
+import java.util.Set;
+
+public class DemoFileExtension implements ServiceExtension {
+ private Monitor monitor;
+
+ @Override
+ public void initialize(ServiceExtensionContext context) {
+ monitor = context.getMonitor();
+
+ var processManager = context.getService(TransferProcessManager.class);
+ var processStore = context.getService(TransferProcessStore.class);
+ var dataFlowManager = context.getService(DataFlowManager.class);
+ var manifestGenerator = context.getService(ResourceManifestGenerator.class);
+ var provisionManager = context.getService(ProvisionManager.class);
+
+ // Register API
+ var webService = context.getService(WebService.class);
+ webService.registerController(new DemoFileApiController(processManager, monitor));
+
+ // Register Provisioner
+ var folderDefGenerator = new FolderResourceDefinitionGenerator();
+ var fileArtifactDefGenerator = new RandomFileArtifactResourceDefinitionGenerator();
+ manifestGenerator.registerClientGenerator(folderDefGenerator);
+ manifestGenerator.registerProviderGenerator(fileArtifactDefGenerator);
+
+ provisionManager.register(new RandomFileArtifactProvisioner(monitor));
+ provisionManager.register(new FolderProvisioner(monitor));
+
+ // Register Data Flow
+ var flowController = new FileFlowController(processStore, monitor);
+ dataFlowManager.register(flowController);
+
+ // Generate Demo Data
+ generateDemoData(context);
+
+ monitor.info("Initialized Demo File extension");
+ }
+
+ @Override
+ public void start() {
+ monitor.info("Started Demo File extension");
+ }
+
+ @Override
+ public void shutdown() {
+ monitor.info("Shutdown Demo File extension");
+ }
+
+ @Override
+ public Set<String> requires() {
+ return Set.of("policy-registry");
+ }
+
+ private void generateDemoData(ServiceExtensionContext context) {
+ var metadataStore = context.getService(MetadataStore.class);
+ var policyRegistry = context.getService(PolicyRegistry.class);
+ var dataGenerator = new DemoMetaDataGenerator(metadataStore, policyRegistry);
+ dataGenerator.generate("file", 10); // generates metadata "file1", "file2", "file3", ..
+ }
+}

extensions/demo/demo-file/src/main/java/com/microsoft/dagx/demo/file/DemoMetaDataGenerator.java

@@ -0,0 +1,31 @@
+package com.microsoft.dagx.demo.file;
+
+import com.microsoft.dagx.policy.model.Policy;
+import com.microsoft.dagx.spi.metadata.MetadataStore;
+import com.microsoft.dagx.spi.policy.PolicyRegistry;
+import com.microsoft.dagx.spi.types.domain.metadata.DataEntry;
+import com.microsoft.dagx.spi.types.domain.metadata.GenericDataCatalog;
+
+import java.util.UUID;
+
+public class DemoMetaDataGenerator {
+ private static final String EmptyPolicyId = "EmptyPolicy";
+ private final MetadataStore metadataStore;
+
+ public DemoMetaDataGenerator(MetadataStore metadataStore, PolicyRegistry policyRegistry) {
+ this.metadataStore = metadataStore;
+ var emptyPolicy = Policy.Builder.newInstance().id(EmptyPolicyId).build();
+ policyRegistry.registerPolicy(emptyPolicy);
+ }
+
+ public void generate(String prefix, int count) {
+ var catalog = GenericDataCatalog.Builder.newInstance()
+ .property("demoGroup", UUID.randomUUID().toString())
+ .build();
+
+ for (var i = 1; i <= count; i++) {
+ var entry = DataEntry.Builder.newInstance().id(prefix + i).policyId(EmptyPolicyId).catalog(catalog).build();
+ metadataStore.save(entry);
+ }
+ }
+}

extensions/demo/demo-file/src/main/java/com/microsoft/dagx/demo/file/schema/FileSchema.java

@@ -0,0 +1,25 @@
+package com.microsoft.dagx.demo.file.schema;
+
+import com.microsoft.dagx.schema.DataSchema;
+import com.microsoft.dagx.schema.SchemaAttribute;
+
+public class FileSchema extends DataSchema {
+
+ public static final String TYPE = "FileSchema";
+ public static String TARGET_FILE_NAME = "targetFileName";
+ public static String TARGET_DIRECTORY = "targetFolderName";
+ public static String IS_COMPRESSION_REQUESTED = "isCompressionRequested";
+
+ @Override
+ protected void addAttributes() {
+ attributes.add(new SchemaAttribute(TARGET_FILE_NAME, true));
+ attributes.add(new SchemaAttribute(TARGET_DIRECTORY, true));
+ attributes.add(new SchemaAttribute(IS_COMPRESSION_REQUESTED, false));
+ }
+
+ @Override
+ public String getName() {
+ return TYPE;
+ }
+
+}