You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by si...@apache.org on 2018/04/20 23:54:39 UTC
[incubator-pulsar] branch master updated: Introduced
Upload/Download interface for uploading and downloading function code.
(#1624)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 4f73c54 Introduced Upload/Download interface for uploading and downloading function code. (#1624)
4f73c54 is described below
commit 4f73c545344e4b089ecf11291ad6f1d1767f9181
Author: Sanjeev Kulkarni <sa...@gmail.com>
AuthorDate: Fri Apr 20 16:54:37 2018 -0700
Introduced Upload/Download interface for uploading and downloading function code. (#1624)
* Moved some util functionality from worker to util
* Reverted K8 changes
* Removed k8
* Moved tests
* Change path
* Added upload and download server implementations
* Added upload and download client requests
* Dont need to depend on util
* Missed declarations
* Fix bug
* Fix bug
* Revert "Moved some util functionality from worker to util"
This reverts commit f2347cb05878adf31c9538be489ef174205be761.
* moved path
* Revert "Moved tests"
This reverts commit 7dc4b506018df67773f433aa09f24610dc1d648a.
* Fixed the paths
* Fixed Exception to be precise
---
.../pulsar/broker/admin/impl/FunctionsBase.java | 22 ++++++
.../org/apache/pulsar/client/admin/Functions.java | 34 +++++++++
.../client/admin/internal/FunctionsImpl.java | 36 ++++++++++
.../org/apache/pulsar/admin/cli/CmdFunctions.java | 60 ++++++++++++++++
.../org/apache/pulsar/functions/worker/Utils.java | 2 +-
.../functions/worker/rest/api/FunctionsImpl.java | 82 +++++++++++++++++++++-
.../worker/rest/api/v2/FunctionApiV2Resource.java | 23 ++++++
7 files changed, 257 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
index b53c1b1..5e0747b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
@@ -28,6 +28,7 @@ import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
+import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.pulsar.broker.admin.AdminResource;
@@ -153,4 +154,25 @@ public class FunctionsBase extends AdminResource implements Supplier<WorkerServi
}
+ @POST
+ @Path("/{tenant}/{namespace}/{functionName}/upload")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response uploadFunction(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("functionName") String functionName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail) {
+ return functions.uploadFunction(
+ tenant, namespace, functionName, uploadedInputStream, fileDetail);
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{functionName}/download")
+ public Response downloadFunction(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("functionName") String functionName,
+ final @QueryParam("filename") String fileName) {
+ return functions.downloadFunction(tenant, namespace, functionName, fileName);
+ }
+
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
index 4a3dbe5..795683d 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Functions.java
@@ -159,5 +159,39 @@ public interface Functions {
*/
String triggerFunction(String tenant, String namespace, String function, String triggerValue, String triggerFile) throws PulsarAdminException;
+ /**
+ * Upload Function Code.
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param function
+ * Function name
+ * @param fileName
+ * Code where file is located
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void uploadFunction(String tenant, String namespace, String function, String fileName) throws PulsarAdminException;
+
+ /**
+ * Download Function Code.
+ *
+ * @param tenant
+ * Tenant name
+ * @param namespace
+ * Namespace name
+ * @param function
+ * Function name
+ * @param fileName
+ * Code where function should be downloaded
+ *
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ void downloadFunction(String tenant, String namespace, String function, String fileName) throws PulsarAdminException;
+
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
index eb793cc..f831b9e 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/FunctionsImpl.java
@@ -41,6 +41,10 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import java.io.File;
import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
import java.util.List;
@Slf4j
@@ -163,6 +167,38 @@ public class FunctionsImpl extends BaseResource implements Functions {
throw getApiException(e);
}
}
+
+ @Override
+ public void uploadFunction(String tenant, String namespace, String functionName, String fileName) throws PulsarAdminException {
+ try {
+ final FormDataMultiPart mp = new FormDataMultiPart();
+
+ mp.bodyPart(new FileDataBodyPart("data", new File(fileName), MediaType.APPLICATION_OCTET_STREAM_TYPE));
+
+ request(functions.path(tenant).path(namespace).path(functionName).path("upload"))
+ .post(Entity.entity(mp, MediaType.MULTIPART_FORM_DATA), ErrorData.class);
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
+
+ @Override
+ public void downloadFunction(String tenant, String namespace, String function, String path) throws PulsarAdminException {
+ try {
+ Path pathToFile = Paths.get(path);
+ InputStream response = request(functions.path(tenant).path(namespace).path(function).path("download")
+ .queryParam("filename", pathToFile.getFileName().toString())).get(InputStream.class);
+ if (response != null) {
+ File targetFile = new File(path);
+ java.nio.file.Files.copy(
+ response,
+ targetFile.toPath(),
+ StandardCopyOption.REPLACE_EXISTING);
+ }
+ } catch (Exception e) {
+ throw getApiException(e);
+ }
+ }
public static void mergeJson(String json, Builder builder) throws IOException {
JsonFormat.parser().merge(json, builder);
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index 7e0b320..339b947 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -84,6 +84,8 @@ public class CmdFunctions extends CmdBase {
private final ListFunctions lister;
private final StateGetter stateGetter;
private final TriggerFunction triggerer;
+ private final UploadFunction uploader;
+ private final DownloadFunction downloader;
/**
* Base command
@@ -691,6 +693,50 @@ public class CmdFunctions extends CmdBase {
}
}
+ @Parameters(commandDescription = "Upload Pulsar Function code to Pulsar")
+ class UploadFunction extends FunctionCommand {
+ @Parameter(
+ names = "--jar",
+ description = "Path to the jar file for the function (if the function is written in Java)",
+ listConverter = StringConverter.class)
+ protected String jarFile;
+ @Parameter(
+ names = "--py",
+ description = "Path to the main Python file for the function (if the function is written in Python)",
+ listConverter = StringConverter.class)
+ protected String pyFile;
+
+ @Override
+ void runCmd() throws Exception {
+ if (jarFile == null && pyFile == null) {
+ throw new RuntimeException("Either a jar File or a python file needs to be specified");
+ }
+ String userCodeFile;
+ if (jarFile != null) {
+ userCodeFile = jarFile;
+ } else {
+ userCodeFile = pyFile;
+ }
+ admin.functions().uploadFunction(tenant, namespace, functionName, userCodeFile);
+ print("Uploaded successfully");
+ }
+ }
+
+ @Parameters(commandDescription = "Download Pulsar Function code from Pulsar")
+ class DownloadFunction extends FunctionCommand {
+ @Parameter(
+ names = "--downloadPath",
+ description = "Path where the file needs to be downloaded)",
+ listConverter = StringConverter.class, required = true)
+ protected String downloadPath;
+
+ @Override
+ void runCmd() throws Exception {
+ admin.functions().downloadFunction(tenant, namespace, functionName, downloadPath);
+ print("Downloaded successfully");
+ }
+ }
+
public CmdFunctions(PulsarAdmin admin) throws PulsarClientException {
super("functions", admin);
localRunner = new LocalRunner();
@@ -702,6 +748,8 @@ public class CmdFunctions extends CmdBase {
lister = new ListFunctions();
stateGetter = new StateGetter();
triggerer = new TriggerFunction();
+ uploader = new UploadFunction();
+ downloader = new DownloadFunction();
jcommander.addCommand("localrun", getLocalRunner());
jcommander.addCommand("create", getCreater());
jcommander.addCommand("delete", getDeleter());
@@ -711,6 +759,8 @@ public class CmdFunctions extends CmdBase {
jcommander.addCommand("list", getLister());
jcommander.addCommand("querystate", getStateGetter());
jcommander.addCommand("trigger", getTriggerer());
+ jcommander.addCommand("upload", getUploader());
+ jcommander.addCommand("download", getDownloader());
}
@VisibleForTesting
@@ -756,6 +806,16 @@ public class CmdFunctions extends CmdBase {
return triggerer;
}
+ @VisibleForTesting
+ UploadFunction getUploader() {
+ return uploader;
+ }
+
+ @VisibleForTesting
+ DownloadFunction getDownloader() {
+ return downloader;
+ }
+
private static FunctionConfig loadConfig(File file) throws IOException {
ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
index fba562f..5a9f339 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
@@ -127,7 +127,7 @@ public final class Utils {
public static void downloadFromBookkeeper(Namespace namespace,
OutputStream outputStream,
- String packagePath) throws Exception {
+ String packagePath) throws IOException {
DistributedLogManager dlm = namespace.openLog(packagePath);
InputStream in = new DLInputStream(dlm);
int read = 0;
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index 1df2871..b870619 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -23,7 +23,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
import com.google.gson.Gson;
import java.io.IOException;
import java.io.InputStream;
-import java.lang.reflect.Array;
+import java.io.OutputStream;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -36,9 +36,11 @@ import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
+import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
+import javax.ws.rs.core.StreamingOutput;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.Message;
@@ -526,6 +528,84 @@ public class FunctionsImpl {
}
}
+ @POST
+ @Path("/{tenant}/{namespace}/{functionName}/upload")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response uploadFunction(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("functionName") String functionName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail) {
+ // validate parameters
+ try {
+ if (tenant == null) {
+ throw new IllegalArgumentException("Tenant is not provided");
+ }
+ if (namespace == null) {
+ throw new IllegalArgumentException("Namespace is not provided");
+ }
+ if (functionName == null) {
+ throw new IllegalArgumentException("Function Name is not provided");
+ }
+ if (uploadedInputStream == null || fileDetail == null) {
+ throw new IllegalArgumentException("Function Package is not provided");
+ }
+ } catch (IllegalArgumentException e) {
+ log.error("Invalid upload function request @ /{}/{}/{}",
+ tenant, namespace, functionName, e);
+ return Response.status(Status.BAD_REQUEST)
+ .type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData(e.getMessage())).build();
+ }
+
+ String packageLocation = String.format(
+ "%s/%s/%s/%s",
+ tenant,
+ namespace,
+ functionName,
+ fileDetail.getFileName());
+
+ // Upload to bookkeeper
+ try {
+ log.info("Uploading function package to {}", packageLocation);
+
+ Utils.uploadToBookeeper(
+ worker().getDlogNamespace(),
+ uploadedInputStream,
+ packageLocation);
+ } catch (IOException e) {
+ log.error("Error uploading file {}", packageLocation, e);
+ return Response.serverError()
+ .type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData(e.getMessage()))
+ .build();
+ }
+
+ return Response.status(Status.OK).build();
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{functionName}/download")
+ public Response downloadFunction(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("functionName") String functionName,
+ final @QueryParam("filename") String fileName) {
+ String packageLocation = String.format(
+ "%s/%s/%s/%s",
+ tenant,
+ namespace,
+ functionName,
+ fileName);
+ return Response.status(Status.OK).entity(
+ new StreamingOutput() {
+ @Override
+ public void write(final OutputStream output) throws IOException {
+ Utils.downloadFromBookkeeper(worker().getDlogNamespace(),
+ output, packageLocation);
+ }
+ }).build();
+ }
+
private void validateListFunctionRequestParams(String tenant, String namespace) throws IllegalArgumentException {
if (tenant == null) {
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
index d2347ff..a2e7c0a 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
@@ -30,6 +30,7 @@ import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
+import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import java.io.IOException;
@@ -139,4 +140,26 @@ public class FunctionApiV2Resource extends FunctionApiResource {
final @FormDataParam("dataStream") InputStream uploadedInputStream) {
return functions.triggerFunction(tenant, namespace, functionName, input, uploadedInputStream);
}
+
+ @POST
+ @Path("/{tenant}/{namespace}/{functionName}/upload")
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ public Response uploadFunction(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("functionName") String functionName,
+ final @FormDataParam("data") InputStream uploadedInputStream,
+ final @FormDataParam("data") FormDataContentDisposition fileDetail) {
+ return functions.uploadFunction(
+ tenant, namespace, functionName, uploadedInputStream, fileDetail);
+ }
+
+ @GET
+ @Path("/{tenant}/{namespace}/{functionName}/download")
+ public Response downloadFunction(final @PathParam("tenant") String tenant,
+ final @PathParam("namespace") String namespace,
+ final @PathParam("functionName") String functionName,
+ final @QueryParam("filename") String fileName) {
+ return functions.downloadFunction(tenant, namespace, functionName, fileName);
+ }
+
}
--
To stop receiving notification emails like this one, please contact
sijie@apache.org.