You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/03/04 21:34:08 UTC
(pulsar) branch branch-2.11 updated (8a96cc7e73c -> b107387a829)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 8a96cc7e73c [improve][broker] Add fine-grain authorization to retention admin API (#22163)
new fdf2be14682 Adjust license header format
new b107387a829 [improve][fn] Add configuration for connector & functions package url sources (#22184)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
conf/functions_worker.yml | 8 ++
.../broker/admin/TopicPoliciesAuthZTest.java | 2 +-
.../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 | 88 +++++++++++-------
.../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 | 89 +++---------------
.../functions/worker/rest/api/SinksImpl.java | 3 +-
.../functions/worker/rest/api/SourcesImpl.java | 3 +-
.../functions/worker/FunctionActionerTest.java | 73 +++++++--------
.../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 ++
21 files changed, 358 insertions(+), 189 deletions(-)
create mode 100644 pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PackageUrlValidator.java
(pulsar) 01/02: Adjust license header format
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fdf2be1468221907aa91245713fdca827775d67b
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Mon Mar 4 22:52:59 2024 +0200
Adjust license header format
---
.../java/org/apache/pulsar/broker/admin/TopicPoliciesAuthZTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesAuthZTest.java
index 3243c04baa2..b1a1f5bebc8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesAuthZTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesAuthZTest.java
@@ -1,4 +1,4 @@
-/*
+/**
* 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
(pulsar) 02/02: [improve][fn] Add configuration for connector & functions package url sources (#22184)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b107387a8296f4f2fca310953dacee722e485709
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 | 88 +++++++++++-------
.../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 | 89 +++---------------
.../functions/worker/rest/api/SinksImpl.java | 3 +-
.../functions/worker/rest/api/SourcesImpl.java | 3 +-
.../functions/worker/FunctionActionerTest.java | 73 +++++++--------
.../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, 357 insertions(+), 188 deletions(-)
diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml
index 08c782b8be5..1a0330eb6f9 100644
--- a/conf/functions_worker.yml
+++ b/conf/functions_worker.yml
@@ -401,7 +401,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 6a4430e2669..8cc767743d4 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
@@ -35,6 +35,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;
@@ -266,6 +267,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();
workerService.init(workerConfig, null, false);
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 33366fa888e..517a278a8bb 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
@@ -261,6 +261,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();
workerService.init(workerConfig, null, false);
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 c6370dec7b2..d42a6139525 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;
@@ -144,6 +145,12 @@ public class PulsarFunctionTlsTest {
workerConfig.setBrokerClientAuthenticationEnabled(true);
workerConfig.setTlsEnabled(true);
workerConfig.setUseTls(true);
+ 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 5009fca0195..2dc7cd89d42 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;
@@ -296,6 +297,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();
workerService.init(workerConfig, null, false);
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 5432d5c4610..a1600be8a0c 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
@@ -251,6 +251,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 "
@@ -269,6 +281,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 542eb30062a..a592a1726a7 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
@@ -30,9 +30,9 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.io.MoreFiles;
import com.google.common.io.RecursiveDeleteOption;
import java.io.File;
-import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.net.URISyntaxException;
import java.net.URL;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
@@ -82,18 +82,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;
}
@@ -108,30 +111,13 @@ public class FunctionActioner {
String packageFile;
- String pkgLocation = functionMetaData.getPackageLocation().getPackagePath();
- boolean isPkgUrlProvided = isFunctionPackageUrlSupported(pkgLocation);
+ Function.PackageLocationMetaData pkgLocation = functionMetaData.getPackageLocation();
if (runtimeFactory.externallyManaged()) {
- packageFile = pkgLocation;
+ packageFile = pkgLocation.getPackagePath();
} else {
- if (isPkgUrlProvided && pkgLocation.startsWith(FILE)) {
- URL url = new URL(pkgLocation);
- File pkgFile = new File(url.toURI());
- packageFile = pkgFile.getAbsolutePath();
- } else if (FunctionCommon.isFunctionCodeBuiltin(functionDetails)) {
- File pkgFile = getBuiltinArchive(FunctionDetails.newBuilder(functionMetaData.getFunctionDetails()));
- packageFile = pkgFile.getAbsolutePath();
- } else {
- File pkgDir = new File(workerConfig.getDownloadDirectory(),
- getDownloadPackagePath(functionMetaData, instanceId));
- pkgDir.mkdirs();
- File pkgFile = new File(
- pkgDir,
- new File(getDownloadFileName(functionMetaData.getFunctionDetails(),
- functionMetaData.getPackageLocation())).getName());
- downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId);
- packageFile = pkgFile.getAbsolutePath();
- }
+ packageFile = getPackageFile(functionMetaData, functionDetails, instanceId, pkgLocation,
+ InstanceUtils.calculateSubjectType(functionDetails));
}
// Setup for batch sources if necessary
@@ -150,6 +136,39 @@ public class FunctionActioner {
}
}
+ private String getPackageFile(FunctionMetaData functionMetaData, FunctionDetails functionDetails, int instanceId,
+ Function.PackageLocationMetaData pkgLocation,
+ FunctionDetails.ComponentType componentType)
+ throws URISyntaxException, IOException, ClassNotFoundException, PulsarAdminException {
+ String packagePath = pkgLocation.getPackagePath();
+ 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();
+ } else if (FunctionCommon.isFunctionCodeBuiltin(functionDetails, componentType)) {
+ File pkgFile = getBuiltinArchive(
+ componentType,
+ FunctionDetails.newBuilder(functionMetaData.getFunctionDetails()));
+ packageFile = pkgFile.getAbsolutePath();
+ } else {
+ File pkgDir = new File(workerConfig.getDownloadDirectory(),
+ getDownloadPackagePath(functionMetaData, instanceId));
+ pkgDir.mkdirs();
+ File pkgFile = new File(
+ pkgDir,
+ new File(getDownloadFileName(functionMetaData.getFunctionDetails(),
+ pkgLocation)).getName());
+ downloadFile(pkgFile, isPkgUrlProvided, functionMetaData, instanceId, pkgLocation, componentType);
+ packageFile = pkgFile.getAbsolutePath();
+ }
+ return packageFile;
+ }
+
RuntimeSpawner getRuntimeSpawner(Function.Instance instance, String packageFile) {
FunctionMetaData functionMetaData = instance.getFunctionMetaData();
int instanceId = instance.getInstanceId();
@@ -198,7 +217,9 @@ public class FunctionActioner {
}
private void downloadFile(File pkgFile, boolean isPkgUrlProvided, FunctionMetaData functionMetaData,
- int instanceId) throws FileNotFoundException, IOException, PulsarAdminException {
+ int instanceId, Function.PackageLocationMetaData pkgLocation,
+ FunctionDetails.ComponentType componentType)
+ throws IOException, PulsarAdminException {
FunctionDetails details = functionMetaData.getFunctionDetails();
File pkgDir = pkgFile.getParentFile();
@@ -214,14 +235,17 @@ public class FunctionActioner {
pkgDir,
pkgFile.getName() + "." + instanceId + "." + UUID.randomUUID());
} while (tempPkgFile.exists() || !tempPkgFile.createNewFile());
- String pkgLocationPath = functionMetaData.getPackageLocation().getPackagePath();
+ String pkgLocationPath = pkgLocation.getPackagePath();
boolean downloadFromHttp = isPkgUrlProvided && pkgLocationPath.startsWith(HTTP);
boolean downloadFromPackageManagementService = isPkgUrlProvided && hasPackageTypePrefix(pkgLocationPath);
log.info("{}/{}/{} Function package file {} will be downloaded from {}", tempPkgFile, details.getTenant(),
details.getNamespace(), details.getName(),
- downloadFromHttp ? pkgLocationPath : functionMetaData.getPackageLocation());
+ 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());
@@ -248,7 +272,7 @@ public class FunctionActioner {
} catch (FileAlreadyExistsException faee) {
// file already exists
log.warn("Function package has been downloaded from {} and saved at {}",
- functionMetaData.getPackageLocation(), pkgFile);
+ pkgLocation, pkgFile);
}
} finally {
tempPkgFile.delete();
@@ -486,8 +510,9 @@ public class FunctionActioner {
File.separatorChar);
}
- private File getBuiltinArchive(FunctionDetails.Builder functionDetails) throws IOException, ClassNotFoundException {
- if (functionDetails.hasSource()) {
+ private File getBuiltinArchive(FunctionDetails.ComponentType componentType, FunctionDetails.Builder functionDetails)
+ throws IOException, ClassNotFoundException {
+ if (componentType == FunctionDetails.ComponentType.SOURCE && functionDetails.hasSource()) {
SourceSpec sourceSpec = functionDetails.getSource();
if (!StringUtils.isEmpty(sourceSpec.getBuiltin())) {
Connector connector = connectorsManager.getConnector(sourceSpec.getBuiltin());
@@ -502,7 +527,7 @@ public class FunctionActioner {
}
}
- if (functionDetails.hasSink()) {
+ if (componentType == FunctionDetails.ComponentType.SINK && functionDetails.hasSink()) {
SinkSpec sinkSpec = functionDetails.getSink();
if (!StringUtils.isEmpty(sinkSpec.getBuiltin())) {
Connector connector = connectorsManager.getConnector(sinkSpec.getBuiltin());
@@ -518,7 +543,8 @@ public class FunctionActioner {
}
}
- if (!StringUtils.isEmpty(functionDetails.getBuiltin())) {
+ if (componentType == FunctionDetails.ComponentType.FUNCTION
+ && !StringUtils.isEmpty(functionDetails.getBuiltin())) {
return functionsManager.getFunctionArchive(functionDetails.getBuiltin()).toFile();
}
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 006437cee3b..3bcaa794931 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..742f1bdb4b6
--- /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 70237305a33..707e4358b5f 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() {
@@ -201,6 +202,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 7f7b3b78efb..46bdc19a1df 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
@@ -1356,11 +1356,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);
@@ -1781,12 +1787,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) {
@@ -1795,7 +1806,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)) {
@@ -1813,15 +1824,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 4f1b4477f1d..68d4cc7cb00 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
@@ -53,7 +53,6 @@ import org.apache.pulsar.functions.instance.InstanceUtils;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.utils.ComponentTypeUtils;
-import org.apache.pulsar.functions.utils.FunctionCommon;
import org.apache.pulsar.functions.utils.FunctionConfigUtils;
import org.apache.pulsar.functions.utils.FunctionFilePackage;
import org.apache.pulsar.functions.utils.FunctionMetaDataUtils;
@@ -139,29 +138,13 @@ public class FunctionsImpl extends ComponentImpl implements Functions<PulsarWork
}
Function.FunctionDetails functionDetails;
- boolean isPkgUrlProvided = isNotBlank(functionPkgUrl);
File componentPackageFile = null;
try {
// validate parameters
try {
- if (isPkgUrlProvided) {
- if (Utils.hasPackageTypePrefix(functionPkgUrl)) {
- componentPackageFile = downloadPackageFile(functionPkgUrl);
- } else {
- if (!Utils.isFunctionPackageUrlSupported(functionPkgUrl)) {
- throw new IllegalArgumentException("Function Package url is not valid."
- + "supported url (http/https/file)");
- }
- try {
-
- componentPackageFile = FunctionCommon.extractFileFromPkgURL(functionPkgUrl);
- } catch (Exception e) {
- throw new IllegalArgumentException(String.format("Encountered error \"%s\" "
- + "when getting %s package from %s", e.getMessage(),
- ComponentTypeUtils.toString(componentType), functionPkgUrl), e);
- }
- }
+ if (isNotBlank(functionPkgUrl)) {
+ componentPackageFile = getPackageFile(componentType, functionPkgUrl);
functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
functionConfig, componentPackageFile);
} else {
@@ -320,60 +303,18 @@ public class FunctionsImpl extends ComponentImpl implements Functions<PulsarWork
// validate parameters
try {
- if (isNotBlank(functionPkgUrl)) {
- if (Utils.hasPackageTypePrefix(functionPkgUrl)) {
- componentPackageFile = downloadPackageFile(functionPkgUrl);
- } else {
- try {
- componentPackageFile = FunctionCommon.extractFileFromPkgURL(functionPkgUrl);
- } catch (Exception e) {
- throw new IllegalArgumentException(String.format("Encountered error \"%s\" "
- + "when getting %s package from %s", e.getMessage(),
- ComponentTypeUtils.toString(componentType), functionPkgUrl));
- }
- }
- functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
- mergedConfig, componentPackageFile);
-
- } else if (existingComponent.getPackageLocation().getPackagePath().startsWith(Utils.FILE)
- || existingComponent.getPackageLocation().getPackagePath().startsWith(Utils.HTTP)) {
- try {
- componentPackageFile = FunctionCommon.extractFileFromPkgURL(
- existingComponent.getPackageLocation().getPackagePath());
- } catch (Exception e) {
- throw new IllegalArgumentException(String.format("Encountered error \"%s\" "
- + "when getting %s package from %s", e.getMessage(),
- ComponentTypeUtils.toString(componentType), functionPkgUrl));
- }
- functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
- mergedConfig, componentPackageFile);
- } else if (uploadedInputStream != null) {
-
- componentPackageFile = WorkerUtils.dumpToTmpFile(uploadedInputStream);
- functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
- mergedConfig, componentPackageFile);
-
- } else if (existingComponent.getPackageLocation().getPackagePath().startsWith(Utils.BUILTIN)) {
- functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
- mergedConfig, componentPackageFile);
- if (!isFunctionCodeBuiltin(functionDetails)
- && (componentPackageFile == null || fileDetail == null)) {
- throw new IllegalArgumentException(ComponentTypeUtils.toString(componentType)
- + " Package is not provided");
- }
- } else {
- componentPackageFile = FunctionCommon.createPkgTempFile();
- componentPackageFile.deleteOnExit();
- if (worker().getWorkerConfig().isFunctionsWorkerEnablePackageManagement()) {
- worker().getBrokerAdmin().packages().download(
- existingComponent.getPackageLocation().getPackagePath(),
- componentPackageFile.getAbsolutePath());
- } else {
- WorkerUtils.downloadFromBookkeeper(worker().getDlogNamespace(),
- componentPackageFile, existingComponent.getPackageLocation().getPackagePath());
- }
- functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
- mergedConfig, componentPackageFile);
+ componentPackageFile = getPackageFile(
+ componentType,
+ functionPkgUrl,
+ existingComponent.getPackageLocation().getPackagePath(),
+ uploadedInputStream);
+ functionDetails = validateUpdateRequestParams(tenant, namespace, functionName,
+ mergedConfig, componentPackageFile);
+ if (existingComponent.getPackageLocation().getPackagePath().startsWith(Utils.BUILTIN)
+ && !isFunctionCodeBuiltin(functionDetails)
+ && (componentPackageFile == null || fileDetail == null)) {
+ throw new IllegalArgumentException(ComponentTypeUtils.toString(componentType)
+ + " Package is not provided");
}
} catch (Exception e) {
log.error("Invalid update {} request @ /{}/{}/{}", ComponentTypeUtils.toString(componentType),
@@ -778,7 +719,7 @@ public class FunctionsImpl extends ComponentImpl implements Functions<PulsarWork
if (archive.startsWith(org.apache.pulsar.common.functions.Utils.BUILTIN)) {
archive = archive.replaceFirst("^builtin://", "");
- FunctionsManager functionsManager = this.worker().getFunctionsManager();
+ FunctionsManager functionsManager = worker().getFunctionsManager();
FunctionArchive function = functionsManager.getFunction(archive);
// check if builtin function exists
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 8cd020b6b33..69998ced75f 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 {
@@ -305,6 +305,7 @@ public class SinksImpl extends ComponentImpl implements Sinks<PulsarWorkerServic
// validate parameters
try {
componentPackageFile = getPackageFile(
+ componentType,
sinkPkgUrl,
existingComponent.getPackageLocation().getPackagePath(),
uploadedInputStream);
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 59bc5888a34..4acd673b584 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 9ccdc09bd7d..b6957081e48 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,8 +78,8 @@ public class FunctionActionerTest {
@SuppressWarnings("resource")
FunctionActioner actioner = new FunctionActioner(workerConfig, factory, dlogNamespace,
- new ConnectorsManager(workerConfig), new FunctionsManager(workerConfig), mock(PulsarAdmin.class));
- Runtime runtime = mock(Runtime.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"))
@@ -110,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);
@@ -123,44 +126,44 @@ 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";
- Function.FunctionMetaData function1 = Function.FunctionMetaData.newBuilder()
- .setFunctionDetails(Function.FunctionDetails.newBuilder().setTenant("test-tenant")
- .setNamespace("test-namespace").setName("func-1"))
- .setPackageLocation(PackageLocationMetaData.newBuilder().setPackagePath(pkgPathLocation).build())
- .build();
- Function.Instance instance = Function.Instance.newBuilder().setFunctionMetaData(function1).setInstanceId(0)
- .build();
- FunctionRuntimeInfo functionRuntimeInfo = mock(FunctionRuntimeInfo.class);
- doReturn(instance).when(functionRuntimeInfo).getFunctionInstance();
-
- actioner.startFunction(functionRuntimeInfo);
+ String pkgPathLocation = FILE + ":///user/my-file.jar";
+ startFunction(actioner, pkgPathLocation);
verify(runtime, times(1)).start();
// (2) test with http-url, downloading file from http should fail with UnknownHostException due to invalid url
- pkgPathLocation = "http://invalid/my-file.jar";
- function1 = Function.FunctionMetaData.newBuilder()
- .setFunctionDetails(Function.FunctionDetails.newBuilder().setTenant("test-tenant")
- .setNamespace("test-namespace").setName("func-1"))
- .setPackageLocation(PackageLocationMetaData.newBuilder().setPackagePath(pkgPathLocation).build())
- .build();
- instance = Function.Instance.newBuilder().setFunctionMetaData(function1).setInstanceId(0).build();
- functionRuntimeInfo = mock(FunctionRuntimeInfo.class);
- doReturn(instance).when(functionRuntimeInfo).getFunctionInstance();
- doThrow(new IllegalStateException("StartupException")).when(functionRuntimeInfo).setStartupException(any());
+ String invalidPkgPathLocation = "http://invalid/my-file.jar";
try {
- actioner.startFunction(functionRuntimeInfo);
+ startFunction(actioner, invalidPkgPathLocation);
fail();
} catch (IllegalStateException ex) {
assertEquals(ex.getMessage(), "StartupException");
}
}
+ private void startFunction(FunctionActioner actioner, String pkgPathLocation) {
+ PackageLocationMetaData packageLocation = PackageLocationMetaData.newBuilder()
+ .setPackagePath(pkgPathLocation)
+ .build();
+ Function.FunctionMetaData function = Function.FunctionMetaData.newBuilder()
+ .setFunctionDetails(Function.FunctionDetails.newBuilder().setTenant("test-tenant")
+ .setNamespace("test-namespace").setName("func-1"))
+ .setPackageLocation(packageLocation)
+ .build();
+ Function.Instance instance = Function.Instance.newBuilder().setFunctionMetaData(function).setInstanceId(0)
+ .build();
+ FunctionRuntimeInfo functionRuntimeInfo = mock(FunctionRuntimeInfo.class);
+ doReturn(instance).when(functionRuntimeInfo).getFunctionInstance();
+ doThrow(new IllegalStateException("StartupException")).when(functionRuntimeInfo).setStartupException(any());
+
+ actioner.startFunction(functionRuntimeInfo);
+ }
+
@Test
public void testFunctionAuthDisabled() throws Exception {
WorkerConfig workerConfig = new WorkerConfig();
@@ -185,7 +188,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";
@@ -248,22 +252,13 @@ 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
String pkgPathLocation = "function://public/default/test-function@latest";
- Function.FunctionMetaData function1 = Function.FunctionMetaData.newBuilder()
- .setFunctionDetails(Function.FunctionDetails.newBuilder().setTenant("test-tenant")
- .setNamespace("test-namespace").setName("func-1"))
- .setPackageLocation(PackageLocationMetaData.newBuilder().setPackagePath(pkgPathLocation).build())
- .build();
- Function.Instance instance = Function.Instance.newBuilder().setFunctionMetaData(function1).setInstanceId(0)
- .build();
- FunctionRuntimeInfo functionRuntimeInfo = mock(FunctionRuntimeInfo.class);
- doReturn(instance).when(functionRuntimeInfo).getFunctionInstance();
-
- actioner.startFunction(functionRuntimeInfo);
+ startFunction(actioner, pkgPathLocation);
verify(runtime, times(1)).start();
}
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 bfaa7431660..88e6fc0fbc5 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
@@ -721,7 +721,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 d6aa13dc2bf..454170df1e0 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 9da30128f1c..31fadea6a8d 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 a9927514efd..9f094409bbe 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;
@@ -57,6 +58,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;
@@ -79,6 +81,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;
@@ -1380,17 +1388,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);
}
/*
@@ -1420,21 +1426,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 3a76ca1a469..95304e1a58b 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
@@ -33,6 +33,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;
@@ -56,6 +57,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;
@@ -79,6 +81,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;