You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by rg...@apache.org on 2024/03/04 16:26:09 UTC
(pulsar) branch branch-3.1 updated: [improve][fn] Add configuration for connector & functions package url sources (#22184)
This is an automated email from the ASF dual-hosted git repository.
rgao pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new b183492bb03 [improve][fn] Add configuration for connector & functions package url sources (#22184)
b183492bb03 is described below
commit b183492bb03c023983ea7a5354c53c6c97bf62fd
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Mar 4 18:12:08 2024 +0200
[improve][fn] Add configuration for connector & functions package url sources (#22184)
(cherry picked from commit 207335a449f2bc9cdf6782c67f93f8c2fb267271)
---
conf/functions_worker.yml | 8 ++
.../worker/PulsarFunctionE2ESecurityTest.java | 6 ++
.../worker/PulsarFunctionPublishTest.java | 4 +
.../functions/worker/PulsarFunctionTlsTest.java | 9 +-
.../apache/pulsar/io/AbstractPulsarE2ETest.java | 6 ++
.../pulsar/functions/worker/WorkerConfig.java | 24 +++++
.../pulsar/functions/worker/FunctionActioner.java | 16 +++-
.../functions/worker/FunctionRuntimeManager.java | 3 +-
.../functions/worker/PackageUrlValidator.java | 101 +++++++++++++++++++++
.../functions/worker/PulsarWorkerService.java | 4 +-
.../functions/worker/rest/api/ComponentImpl.java | 32 +++++--
.../functions/worker/rest/api/FunctionsImpl.java | 3 +-
.../functions/worker/rest/api/SinksImpl.java | 9 +-
.../functions/worker/rest/api/SourcesImpl.java | 3 +-
.../functions/worker/FunctionActionerTest.java | 17 +++-
.../worker/FunctionRuntimeManagerTest.java | 2 +-
.../api/v3/AbstractFunctionApiResourceTest.java | 7 ++
.../rest/api/v3/AbstractFunctionsResourceTest.java | 18 +++-
.../worker/rest/api/v3/SinkApiV3ResourceTest.java | 56 ++++++------
.../rest/api/v3/SourceApiV3ResourceTest.java | 9 ++
20 files changed, 282 insertions(+), 55 deletions(-)
diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml
index 8c625369719..3871c74a887 100644
--- a/conf/functions_worker.yml
+++ b/conf/functions_worker.yml
@@ -408,7 +408,15 @@ saslJaasServerRoleTokenSignerSecretPath:
########################
connectorsDirectory: ./connectors
+# Whether to enable referencing connectors directory files by file url in connector (sink/source) creation
+enableReferencingConnectorDirectoryFiles: true
+# Regex patterns for enabling creation of connectors by referencing packages in matching http/https urls
+additionalEnabledConnectorUrlPatterns: []
functionsDirectory: ./functions
+# Whether to enable referencing functions directory files by file url in functions creation
+enableReferencingFunctionsDirectoryFiles: true
+# Regex patterns for enabling creation of functions by referencing packages in matching http/https urls
+additionalEnabledFunctionsUrlPatterns: []
# Enables extended validation for connector config with fine-grain annotation based validation
# during submission. Classloading with either enableClassloadingOfExternalFiles or
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
index 107aedd0766..cbf2f28b0b5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
@@ -34,6 +34,7 @@ import java.lang.reflect.Method;
import java.net.URL;
import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
@@ -267,6 +268,11 @@ public class PulsarFunctionE2ESecurityTest {
workerConfig.setAuthorizationEnabled(config.isAuthorizationEnabled());
workerConfig.setAuthorizationProvider(config.getAuthorizationProvider());
+ List<String> urlPatterns =
+ List.of(getPulsarApiExamplesJar().getParentFile().toURI() + ".*", "http://127\\.0\\.0\\.1:.*");
+ workerConfig.setAdditionalEnabledConnectorUrlPatterns(urlPatterns);
+ workerConfig.setAdditionalEnabledFunctionsUrlPatterns(urlPatterns);
+
PulsarWorkerService workerService = new PulsarWorkerService();
return workerService;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
index 7bcf1dec871..72e6a3766ae 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
@@ -268,6 +268,10 @@ public class PulsarFunctionPublishTest {
workerConfig.setAuthenticationEnabled(true);
workerConfig.setAuthorizationEnabled(true);
+ List<String> urlPatterns = List.of(getPulsarApiExamplesJar().getParentFile().toURI() + ".*");
+ workerConfig.setAdditionalEnabledConnectorUrlPatterns(urlPatterns);
+ workerConfig.setAdditionalEnabledFunctionsUrlPatterns(urlPatterns);
+
PulsarWorkerService workerService = new PulsarWorkerService();
return workerService;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
index 9882b15450e..3508cf0bfc7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
@@ -20,8 +20,8 @@ package org.apache.pulsar.functions.worker;
import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
@@ -32,6 +32,7 @@ import java.io.UncheckedIOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -153,6 +154,12 @@ public class PulsarFunctionTlsTest {
workerConfig.setUseTls(true);
workerConfig.setTlsEnableHostnameVerification(true);
workerConfig.setTlsAllowInsecureConnection(false);
+ File packagePath = new File(
+ PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath()).getParentFile();
+ List<String> urlPatterns =
+ List.of(packagePath.toURI() + ".*");
+ workerConfig.setAdditionalEnabledConnectorUrlPatterns(urlPatterns);
+ workerConfig.setAdditionalEnabledFunctionsUrlPatterns(urlPatterns);
fnWorkerServices[i] = WorkerServiceLoader.load(workerConfig);
configurations[i] = config;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java
index f968315a712..3c0dd0822b7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java
@@ -35,6 +35,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -306,6 +307,11 @@ public abstract class AbstractPulsarE2ETest {
workerConfig.setAuthenticationEnabled(true);
workerConfig.setAuthorizationEnabled(true);
+ List<String> urlPatterns =
+ List.of(getPulsarApiExamplesJar().getParentFile().toURI() + ".*", "http://127\\.0\\.0\\.1:.*");
+ workerConfig.setAdditionalEnabledConnectorUrlPatterns(urlPatterns);
+ workerConfig.setAdditionalEnabledFunctionsUrlPatterns(urlPatterns);
+
PulsarWorkerService workerService = new PulsarWorkerService();
return workerService;
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index 2d9698103fa..ec0e620d0ae 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -264,6 +264,18 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
doc = "The directory where nar packages are extractors"
)
private String narExtractionDirectory = NarClassLoader.DEFAULT_NAR_EXTRACTION_DIR;
+ @FieldContext(
+ category = CATEGORY_CONNECTORS,
+ doc = "Whether to enable referencing connectors directory files by file url in connector (sink/source) "
+ + "creation. Default is true."
+ )
+ private Boolean enableReferencingConnectorDirectoryFiles = true;
+ @FieldContext(
+ category = CATEGORY_FUNCTIONS,
+ doc = "Regex patterns for enabling creation of connectors by referencing packages in matching http/https "
+ + "urls."
+ )
+ private List<String> additionalEnabledConnectorUrlPatterns = new ArrayList<>();
@FieldContext(
category = CATEGORY_CONNECTORS,
doc = "Enables extended validation for connector config with fine-grain annotation based validation "
@@ -282,6 +294,18 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
doc = "The path to the location to locate builtin functions"
)
private String functionsDirectory = "./functions";
+ @FieldContext(
+ category = CATEGORY_FUNCTIONS,
+ doc = "Whether to enable referencing functions directory files by file url in functions creation. "
+ + "Default is true."
+ )
+ private Boolean enableReferencingFunctionsDirectoryFiles = true;
+ @FieldContext(
+ category = CATEGORY_FUNCTIONS,
+ doc = "Regex patterns for enabling creation of functions by referencing packages in matching http/https "
+ + "urls."
+ )
+ private List<String> additionalEnabledFunctionsUrlPatterns = new ArrayList<>();
@FieldContext(
category = CATEGORY_FUNC_METADATA_MNG,
doc = "The Pulsar topic used for storing function metadata"
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
index 250a7cc4c7b..389051fce42 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
@@ -83,18 +83,21 @@ public class FunctionActioner {
private final ConnectorsManager connectorsManager;
private final FunctionsManager functionsManager;
private final PulsarAdmin pulsarAdmin;
+ private final PackageUrlValidator packageUrlValidator;
public FunctionActioner(WorkerConfig workerConfig,
RuntimeFactory runtimeFactory,
Namespace dlogNamespace,
ConnectorsManager connectorsManager,
- FunctionsManager functionsManager, PulsarAdmin pulsarAdmin) {
+ FunctionsManager functionsManager, PulsarAdmin pulsarAdmin,
+ PackageUrlValidator packageUrlValidator) {
this.workerConfig = workerConfig;
this.runtimeFactory = runtimeFactory;
this.dlogNamespace = dlogNamespace;
this.connectorsManager = connectorsManager;
this.functionsManager = functionsManager;
this.pulsarAdmin = pulsarAdmin;
+ this.packageUrlValidator = packageUrlValidator;
}
@@ -152,6 +155,9 @@ public class FunctionActioner {
boolean isPkgUrlProvided = isFunctionPackageUrlSupported(packagePath);
String packageFile;
if (isPkgUrlProvided && packagePath.startsWith(FILE)) {
+ if (!packageUrlValidator.isValidPackageUrl(componentType, packagePath)) {
+ throw new IllegalArgumentException("Package URL " + packagePath + " is not valid");
+ }
URL url = new URL(packagePath);
File pkgFile = new File(url.toURI());
packageFile = pkgFile.getAbsolutePath();
@@ -168,7 +174,7 @@ public class FunctionActioner {
pkgDir,
new File(getDownloadFileName(functionMetaData.getFunctionDetails(),
pkgLocation)).getName());
- downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId, pkgLocation);
+ downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId, pkgLocation, componentType);
packageFile = pkgFile.getAbsolutePath();
}
return packageFile;
@@ -227,7 +233,8 @@ public class FunctionActioner {
}
private void downloadFile(File pkgFile, boolean isPkgUrlProvided, FunctionMetaData functionMetaData,
- int instanceId, Function.PackageLocationMetaData pkgLocation)
+ int instanceId, Function.PackageLocationMetaData pkgLocation,
+ FunctionDetails.ComponentType componentType)
throws IOException, PulsarAdminException {
FunctionDetails details = functionMetaData.getFunctionDetails();
@@ -252,6 +259,9 @@ public class FunctionActioner {
downloadFromHttp ? pkgLocationPath : pkgLocation);
if (downloadFromHttp) {
+ if (!packageUrlValidator.isValidPackageUrl(componentType, pkgLocationPath)) {
+ throw new IllegalArgumentException("Package URL " + pkgLocationPath + " is not valid");
+ }
FunctionCommon.downloadFromHttpUrl(pkgLocationPath, tempPkgFile);
} else if (downloadFromPackageManagementService) {
getPulsarAdmin().packages().download(pkgLocationPath, tempPkgFile.getPath());
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
index 8e6725e93af..b6e2bbb1ca0 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
@@ -220,7 +220,8 @@ public class FunctionRuntimeManager implements AutoCloseable {
functionAuthProvider, runtimeCustomizer);
this.functionActioner = new FunctionActioner(this.workerConfig, runtimeFactory,
- dlogNamespace, connectorsManager, functionsManager, workerService.getBrokerAdmin());
+ dlogNamespace, connectorsManager, functionsManager, workerService.getBrokerAdmin(),
+ workerService.getPackageUrlValidator());
this.membershipManager = membershipManager;
this.functionMetaDataManager = functionMetaDataManager;
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PackageUrlValidator.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PackageUrlValidator.java
new file mode 100644
index 00000000000..2a8fe8dddb1
--- /dev/null
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PackageUrlValidator.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.worker;
+
+import java.net.URI;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.pulsar.functions.proto.Function;
+
+/**
+ * Validates package URLs for functions and connectors.
+ * Validates that the package URL is either a file in the connectors or functions directory
+ * when referencing connector or function files is enabled, or matches one of the additional url patterns.
+ */
+public class PackageUrlValidator {
+ private final Path connectionsDirectory;
+ private final Path functionsDirectory;
+ private final List<Pattern> additionalConnectionsPatterns;
+ private final List<Pattern> additionalFunctionsPatterns;
+
+ public PackageUrlValidator(WorkerConfig workerConfig) {
+ this.connectionsDirectory = resolveDirectory(workerConfig.getEnableReferencingConnectorDirectoryFiles(),
+ workerConfig.getConnectorsDirectory());
+ this.functionsDirectory = resolveDirectory(workerConfig.getEnableReferencingFunctionsDirectoryFiles(),
+ workerConfig.getFunctionsDirectory());
+ this.additionalConnectionsPatterns =
+ compilePatterns(workerConfig.getAdditionalEnabledConnectorUrlPatterns());
+ this.additionalFunctionsPatterns =
+ compilePatterns(workerConfig.getAdditionalEnabledFunctionsUrlPatterns());
+ }
+
+ private static Path resolveDirectory(Boolean enabled, String directory) {
+ return enabled != null && enabled
+ ? Path.of(directory).normalize().toAbsolutePath() : null;
+ }
+
+ private static List<Pattern> compilePatterns(List<String> additionalPatterns) {
+ return additionalPatterns != null ? additionalPatterns.stream().map(Pattern::compile).collect(
+ Collectors.toList()) : Collections.emptyList();
+ }
+
+ boolean isValidFunctionsPackageUrl(URI functionPkgUrl) {
+ return doesMatch(functionPkgUrl, functionsDirectory, additionalFunctionsPatterns);
+ }
+
+ boolean isValidConnectionsPackageUrl(URI functionPkgUrl) {
+ return doesMatch(functionPkgUrl, connectionsDirectory, additionalConnectionsPatterns);
+ }
+
+ private boolean doesMatch(URI functionPkgUrl, Path directory, List<Pattern> patterns) {
+ if (directory != null && "file".equals(functionPkgUrl.getScheme())) {
+ Path filePath = Path.of(functionPkgUrl.getPath()).normalize().toAbsolutePath();
+ if (filePath.startsWith(directory)) {
+ return true;
+ }
+ }
+ String functionPkgUrlString = functionPkgUrl.normalize().toString();
+ for (Pattern pattern : patterns) {
+ if (pattern.matcher(functionPkgUrlString).matches()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ public boolean isValidPackageUrl(Function.FunctionDetails.ComponentType componentType, String functionPkgUrl) {
+ URI uri = URI.create(functionPkgUrl);
+ if (componentType == null) {
+ // if component type is not specified, we need to check both functions and connections
+ return isValidFunctionsPackageUrl(uri) || isValidConnectionsPackageUrl(uri);
+ }
+ switch (componentType) {
+ case FUNCTION:
+ return isValidFunctionsPackageUrl(uri);
+ case SINK:
+ case SOURCE:
+ return isValidConnectionsPackageUrl(uri);
+ default:
+ throw new IllegalArgumentException("Unknown component type: " + componentType);
+ }
+ }
+}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java
index f9f2738828b..84b943e5671 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java
@@ -119,7 +119,8 @@ public class PulsarWorkerService implements WorkerService {
private Sinks<PulsarWorkerService> sinks;
private Sources<PulsarWorkerService> sources;
private Workers<PulsarWorkerService> workers;
-
+ @Getter
+ private PackageUrlValidator packageUrlValidator;
private final PulsarClientCreator clientCreator;
public PulsarWorkerService() {
@@ -196,6 +197,7 @@ public class PulsarWorkerService implements WorkerService {
this.sinks = new SinksImpl(() -> PulsarWorkerService.this);
this.sources = new SourcesImpl(() -> PulsarWorkerService.this);
this.workers = new WorkerImpl(() -> PulsarWorkerService.this);
+ this.packageUrlValidator = new PackageUrlValidator(workerConfig);
}
@Override
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
index 5e105f7057e..6d07e587091 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java
@@ -1376,11 +1376,17 @@ public abstract class ComponentImpl implements Component<PulsarWorkerService> {
private StreamingOutput getStreamingOutput(String pkgPath, FunctionDetails.ComponentType componentType) {
return output -> {
if (pkgPath.startsWith(Utils.HTTP)) {
+ if (!worker().getPackageUrlValidator().isValidPackageUrl(componentType, pkgPath)) {
+ throw new IllegalArgumentException("Invalid package url: " + pkgPath);
+ }
URL url = URI.create(pkgPath).toURL();
try (InputStream inputStream = url.openStream()) {
IOUtils.copy(inputStream, output);
}
} else if (pkgPath.startsWith(Utils.FILE)) {
+ if (!worker().getPackageUrlValidator().isValidPackageUrl(componentType, pkgPath)) {
+ throw new IllegalArgumentException("Invalid package url: " + pkgPath);
+ }
URI url = URI.create(pkgPath);
File file = new File(url.getPath());
Files.copy(file.toPath(), output);
@@ -1804,12 +1810,17 @@ public abstract class ComponentImpl implements Component<PulsarWorkerService> {
return file;
}
- protected File getPackageFile(String functionPkgUrl, String existingPackagePath, InputStream uploadedInputStream)
+ protected File getPackageFile(FunctionDetails.ComponentType componentType, String functionPkgUrl,
+ String existingPackagePath, InputStream uploadedInputStream)
throws IOException, PulsarAdminException {
File componentPackageFile = null;
if (isNotBlank(functionPkgUrl)) {
- componentPackageFile = getPackageFile(functionPkgUrl);
+ componentPackageFile = getPackageFile(componentType, functionPkgUrl);
} else if (existingPackagePath.startsWith(Utils.FILE) || existingPackagePath.startsWith(Utils.HTTP)) {
+ if (!worker().getPackageUrlValidator().isValidPackageUrl(componentType, functionPkgUrl)) {
+ throw new IllegalArgumentException("Function Package url is not valid."
+ + "supported url (http/https/file)");
+ }
try {
componentPackageFile = FunctionCommon.extractFileFromPkgURL(existingPackagePath);
} catch (Exception e) {
@@ -1818,7 +1829,7 @@ public abstract class ComponentImpl implements Component<PulsarWorkerService> {
ComponentTypeUtils.toString(componentType), functionPkgUrl));
}
} else if (Utils.hasPackageTypePrefix(existingPackagePath)) {
- componentPackageFile = getPackageFile(existingPackagePath);
+ componentPackageFile = getPackageFile(componentType, existingPackagePath);
} else if (uploadedInputStream != null) {
componentPackageFile = WorkerUtils.dumpToTmpFile(uploadedInputStream);
} else if (!existingPackagePath.startsWith(Utils.BUILTIN)) {
@@ -1836,15 +1847,16 @@ public abstract class ComponentImpl implements Component<PulsarWorkerService> {
return componentPackageFile;
}
- protected File downloadPackageFile(String packageName) throws IOException, PulsarAdminException {
- return downloadPackageFile(worker(), packageName);
- }
-
- protected File getPackageFile(String functionPkgUrl) throws IOException, PulsarAdminException {
+ protected File getPackageFile(FunctionDetails.ComponentType componentType, String functionPkgUrl)
+ throws IOException, PulsarAdminException {
if (Utils.hasPackageTypePrefix(functionPkgUrl)) {
- return downloadPackageFile(functionPkgUrl);
+ if (!worker().getWorkerConfig().isFunctionsWorkerEnablePackageManagement()) {
+ throw new IllegalStateException("Function Package management service is disabled. "
+ + "Please enable it to use " + functionPkgUrl);
+ }
+ return downloadPackageFile(worker(), functionPkgUrl);
} else {
- if (!Utils.isFunctionPackageUrlSupported(functionPkgUrl)) {
+ if (!worker().getPackageUrlValidator().isValidPackageUrl(componentType, functionPkgUrl)) {
throw new IllegalArgumentException("Function Package url is not valid."
+ "supported url (http/https/file)");
}
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 6b81d2c4918..a075d3e18a0 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
@@ -144,7 +144,7 @@ public class FunctionsImpl extends ComponentImpl implements Functions<PulsarWork
// validate parameters
try {
if (isNotBlank(functionPkgUrl)) {
- componentPackageFile = getPackageFile(functionPkgUrl);
+ componentPackageFile = getPackageFile(componentType, functionPkgUrl);
functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
functionConfig, componentPackageFile);
} else {
@@ -305,6 +305,7 @@ public class FunctionsImpl extends ComponentImpl implements Functions<PulsarWork
// validate parameters
try {
componentPackageFile = getPackageFile(
+ componentType,
functionPkgUrl,
existingComponent.getPackageLocation().getPackagePath(),
uploadedInputStream);
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
index 51d1333a79c..6b8b41e5a8e 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
@@ -143,7 +143,7 @@ public class SinksImpl extends ComponentImpl implements Sinks<PulsarWorkerServic
// validate parameters
try {
if (isNotBlank(sinkPkgUrl)) {
- componentPackageFile = getPackageFile(sinkPkgUrl);
+ componentPackageFile = getPackageFile(componentType, sinkPkgUrl);
functionDetails = validateUpdateRequestParams(tenant, namespace, sinkName,
sinkConfig, componentPackageFile);
} else {
@@ -310,6 +310,7 @@ public class SinksImpl extends ComponentImpl implements Sinks<PulsarWorkerServic
// validate parameters
try {
componentPackageFile = getPackageFile(
+ componentType,
sinkPkgUrl,
existingComponent.getPackageLocation().getPackagePath(),
uploadedInputStream);
@@ -421,7 +422,8 @@ public class SinksImpl extends ComponentImpl implements Sinks<PulsarWorkerServic
try {
String builtin = functionDetails.getBuiltin();
if (isBlank(builtin)) {
- functionPackageFile = getPackageFile(transformFunction);
+ functionPackageFile =
+ getPackageFile(Function.FunctionDetails.ComponentType.FUNCTION, transformFunction);
}
Function.PackageLocationMetaData.Builder functionPackageLocation =
getFunctionPackageLocation(functionMetaDataBuilder.build(),
@@ -744,7 +746,8 @@ public class SinksImpl extends ComponentImpl implements Sinks<PulsarWorkerServic
transformFunctionPackage =
getBuiltinFunctionPackage(sinkConfig.getTransformFunction());
if (transformFunctionPackage == null) {
- File functionPackageFile = getPackageFile(sinkConfig.getTransformFunction());
+ File functionPackageFile = getPackageFile(Function.FunctionDetails.ComponentType.FUNCTION,
+ sinkConfig.getTransformFunction());
transformFunctionPackage =
new FunctionFilePackage(functionPackageFile, workerConfig.getNarExtractionDirectory(),
workerConfig.getEnableClassloadingOfExternalFiles(), ConnectorDefinition.class);
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
index dea69698dd2..51913061469 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
@@ -143,7 +143,7 @@ public class SourcesImpl extends ComponentImpl implements Sources<PulsarWorkerSe
// validate parameters
try {
if (isPkgUrlProvided) {
- componentPackageFile = getPackageFile(sourcePkgUrl);
+ componentPackageFile = getPackageFile(componentType, sourcePkgUrl);
functionDetails = validateUpdateRequestParams(tenant, namespace, sourceName,
sourceConfig, componentPackageFile);
} else {
@@ -304,6 +304,7 @@ public class SourcesImpl extends ComponentImpl implements Sources<PulsarWorkerSe
// validate parameters
try {
componentPackageFile = getPackageFile(
+ componentType,
sourcePkgUrl,
existingComponent.getPackageLocation().getPackagePath(),
uploadedInputStream);
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionActionerTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionActionerTest.java
index 4e4c3d2f234..ac5ca617ea4 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionActionerTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionActionerTest.java
@@ -30,6 +30,7 @@ import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.AssertJUnit.fail;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.apache.distributedlog.api.namespace.Namespace;
@@ -77,7 +78,8 @@ public class FunctionActionerTest {
@SuppressWarnings("resource")
FunctionActioner actioner = new FunctionActioner(workerConfig, factory, dlogNamespace,
- new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), mock(PulsarAdmin.class));
+ new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), mock(PulsarAdmin.class),
+ mock(PackageUrlValidator.class));
Function.FunctionMetaData function1 = Function.FunctionMetaData.newBuilder()
.setFunctionDetails(Function.FunctionDetails.newBuilder().setTenant("test-tenant")
.setNamespace("test-namespace").setName("func-1"))
@@ -109,6 +111,8 @@ public class FunctionActionerTest {
workerConfig.setPulsarServiceUrl("pulsar://localhost:6650");
workerConfig.setStateStorageServiceUrl("foo");
workerConfig.setFunctionAssignmentTopicName("assignments");
+ workerConfig.setAdditionalEnabledFunctionsUrlPatterns(List.of("file:///user/.*", "http://invalid/.*"));
+ workerConfig.setAdditionalEnabledConnectorUrlPatterns(List.of("file:///user/.*", "http://invalid/.*"));
String downloadDir = this.getClass().getProtectionDomain().getCodeSource().getLocation().getPath();
workerConfig.setDownloadDirectory(downloadDir);
@@ -122,11 +126,12 @@ public class FunctionActionerTest {
@SuppressWarnings("resource")
FunctionActioner actioner = new FunctionActioner(workerConfig, factory, dlogNamespace,
- new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), mock(PulsarAdmin.class));
+ new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), mock(PulsarAdmin.class),
+ new PackageUrlValidator(workerConfig));
// (1) test with file url. functionActioner should be able to consider file-url and it should be able to call
// RuntimeSpawner
- String pkgPathLocation = FILE + ":/user/my-file.jar";
+ String pkgPathLocation = FILE + ":///user/my-file.jar";
startFunction(actioner, pkgPathLocation, pkgPathLocation);
verify(runtime, times(1)).start();
@@ -194,7 +199,8 @@ public class FunctionActionerTest {
@SuppressWarnings("resource")
FunctionActioner actioner = new FunctionActioner(workerConfig, factory, dlogNamespace,
- new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), mock(PulsarAdmin.class));
+ new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), mock(PulsarAdmin.class),
+ mock(PackageUrlValidator.class));
String pkgPathLocation = "http://invalid/my-file.jar";
@@ -257,7 +263,8 @@ public class FunctionActionerTest {
@SuppressWarnings("resource")
FunctionActioner actioner = new FunctionActioner(workerConfig, factory, dlogNamespace,
- new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), pulsarAdmin);
+ new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), pulsarAdmin,
+ mock(PackageUrlValidator.class));
// (1) test with file url. functionActioner should be able to consider file-url and it should be able to call
// RuntimeSpawner
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java
index bc56b1766d3..7b57a1c8965 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionRuntimeManagerTest.java
@@ -720,7 +720,7 @@ public class FunctionRuntimeManagerTest {
FunctionActioner functionActioner = spy(new FunctionActioner(
workerConfig,
- kubernetesRuntimeFactory, null, null, null, null));
+ kubernetesRuntimeFactory, null, null, null, null, workerService.getPackageUrlValidator()));
try (final MockedStatic<RuntimeFactory> runtimeFactoryMockedStatic = Mockito
.mockStatic(RuntimeFactory.class);) {
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionApiResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionApiResourceTest.java
index 5845ff3afd9..388331ce6f2 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionApiResourceTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionApiResourceTest.java
@@ -29,6 +29,7 @@ import com.google.common.collect.Lists;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
+import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
@@ -59,6 +60,12 @@ import org.testng.Assert;
import org.testng.annotations.Test;
public abstract class AbstractFunctionApiResourceTest extends AbstractFunctionsResourceTest {
+ @Override
+ protected void customizeWorkerConfig(WorkerConfig workerConfig, Method method) {
+ if (method.getName().contains("Upload")) {
+ workerConfig.setFunctionsWorkerEnablePackageManagement(false);
+ }
+ }
@Test
public void testListFunctionsSuccess() {
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionsResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionsResourceTest.java
index 4cc4ed0b098..51ca4c83f9e 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionsResourceTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/AbstractFunctionsResourceTest.java
@@ -28,6 +28,7 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
+import java.lang.reflect.Method;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -63,6 +64,7 @@ import org.apache.pulsar.functions.worker.FunctionMetaDataManager;
import org.apache.pulsar.functions.worker.FunctionRuntimeManager;
import org.apache.pulsar.functions.worker.FunctionsManager;
import org.apache.pulsar.functions.worker.LeaderService;
+import org.apache.pulsar.functions.worker.PackageUrlValidator;
import org.apache.pulsar.functions.worker.PulsarWorkerService;
import org.apache.pulsar.functions.worker.WorkerConfig;
import org.apache.pulsar.functions.worker.WorkerUtils;
@@ -141,7 +143,7 @@ public abstract class AbstractFunctionsResourceTest {
}
@BeforeMethod
- public final void setup() throws Exception {
+ public final void setup(Method method) throws Exception {
this.mockedManager = mock(FunctionMetaDataManager.class);
this.mockedFunctionRunTimeManager = mock(FunctionRuntimeManager.class);
this.mockedRuntimeFactory = mock(RuntimeFactory.class);
@@ -181,21 +183,33 @@ public abstract class AbstractFunctionsResourceTest {
}).when(mockedPackages).download(any(), any());
// worker config
+ List<String> urlPatterns =
+ List.of("http://localhost.*", "file:.*", "https://repo1.maven.org/maven2/org/apache/pulsar/.*");
WorkerConfig workerConfig = new WorkerConfig()
.setWorkerId("test")
.setWorkerPort(8080)
.setFunctionMetadataTopicName("pulsar/functions")
.setNumFunctionPackageReplicas(3)
- .setPulsarServiceUrl("pulsar://localhost:6650/");
+ .setPulsarServiceUrl("pulsar://localhost:6650/")
+ .setAdditionalEnabledFunctionsUrlPatterns(urlPatterns)
+ .setAdditionalEnabledConnectorUrlPatterns(urlPatterns)
+ .setFunctionsWorkerEnablePackageManagement(true);
+ customizeWorkerConfig(workerConfig, method);
tempDirectory = PulsarFunctionTestTemporaryDirectory.create(getClass().getSimpleName());
tempDirectory.useTemporaryDirectoriesForWorkerConfig(workerConfig);
when(mockedWorkerService.getWorkerConfig()).thenReturn(workerConfig);
when(mockedWorkerService.getFunctionsManager()).thenReturn(functionsManager);
when(mockedWorkerService.getConnectorsManager()).thenReturn(connectorsManager);
+ PackageUrlValidator packageUrlValidator = new PackageUrlValidator(workerConfig);
+ when(mockedWorkerService.getPackageUrlValidator()).thenReturn(packageUrlValidator);
doSetup();
}
+ protected void customizeWorkerConfig(WorkerConfig workerConfig, Method method) {
+
+ }
+
protected File getDefaultNarFile() {
return getPulsarIOTwitterNar();
}
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java
index b9833380d70..c6c6303e480 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java
@@ -36,6 +36,7 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
@@ -58,6 +59,7 @@ import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.Function.FunctionMetaData;
import org.apache.pulsar.functions.utils.SinkConfigUtils;
+import org.apache.pulsar.functions.worker.WorkerConfig;
import org.apache.pulsar.functions.worker.WorkerUtils;
import org.apache.pulsar.functions.worker.rest.api.SinksImpl;
import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
@@ -80,6 +82,12 @@ public class SinkApiV3ResourceTest extends AbstractFunctionsResourceTest {
this.resource = spy(new SinksImpl(() -> mockedWorkerService));
}
+ @Override
+ protected void customizeWorkerConfig(WorkerConfig workerConfig, Method method) {
+ if (method.getName().contains("Upload") || method.getName().contains("BKPackage")) {
+ workerConfig.setFunctionsWorkerEnablePackageManagement(false);
+ }
+ }
@Override
protected Function.FunctionDetails.ComponentType getComponentType() {
return Function.FunctionDetails.ComponentType.SINK;
@@ -1486,17 +1494,15 @@ public class SinkApiV3ResourceTest extends AbstractFunctionsResourceTest {
SinkConfig sinkConfig = createDefaultSinkConfig();
sinkConfig.setArchive("builtin://cassandra");
- try (FileInputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) {
- resource.registerSink(
- tenant,
- namespace,
- sink,
- inputStream,
- mockedFormData,
- null,
- sinkConfig,
- null);
- }
+ resource.registerSink(
+ tenant,
+ namespace,
+ sink,
+ null,
+ mockedFormData,
+ null,
+ sinkConfig,
+ null);
}
/*
@@ -1526,21 +1532,19 @@ public class SinkApiV3ResourceTest extends AbstractFunctionsResourceTest {
SinkConfig sinkConfig = createDefaultSinkConfig();
sinkConfig.setArchive("builtin://cassandra");
- try (FileInputStream inputStream = new FileInputStream(getPulsarIOCassandraNar())) {
- try {
- resource.registerSink(
- tenant,
- namespace,
- sink,
- inputStream,
- mockedFormData,
- null,
- sinkConfig,
- null);
- Assert.fail();
- } catch (RuntimeException e) {
- Assert.assertEquals(e.getMessage(), injectedErrMsg);
- }
+ try {
+ resource.registerSink(
+ tenant,
+ namespace,
+ sink,
+ null,
+ mockedFormData,
+ null,
+ sinkConfig,
+ null);
+ Assert.fail();
+ } catch (RuntimeException e) {
+ Assert.assertEquals(e.getMessage(), injectedErrMsg);
}
}
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java
index c7e69484d30..f02acbd3663 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java
@@ -34,6 +34,7 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.lang.reflect.Method;
import java.util.LinkedList;
import java.util.List;
import javax.ws.rs.core.Response;
@@ -57,6 +58,7 @@ import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.source.TopicSchema;
import org.apache.pulsar.functions.utils.SourceConfigUtils;
import org.apache.pulsar.functions.utils.io.ConnectorUtils;
+import org.apache.pulsar.functions.worker.WorkerConfig;
import org.apache.pulsar.functions.worker.WorkerUtils;
import org.apache.pulsar.functions.worker.rest.api.SourcesImpl;
import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
@@ -80,6 +82,13 @@ public class SourceApiV3ResourceTest extends AbstractFunctionsResourceTest {
this.resource = spy(new SourcesImpl(() -> mockedWorkerService));
}
+ @Override
+ protected void customizeWorkerConfig(WorkerConfig workerConfig, Method method) {
+ if (method.getName().endsWith("UploadFailure") || method.getName().contains("BKPackage")) {
+ workerConfig.setFunctionsWorkerEnablePackageManagement(false);
+ }
+ }
+
@Override
protected FunctionDetails.ComponentType getComponentType() {
return FunctionDetails.ComponentType.SOURCE;