You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/11/26 23:38:47 UTC
[GitHub] srkukarni closed pull request #2826: Secrets Interface
srkukarni closed pull request #2826: Secrets Interface
URL: https://github.com/apache/pulsar/pull/2826
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-cpp/python/pulsar/functions/context.py b/pulsar-client-cpp/python/pulsar/functions/context.py
index 47e86f978f..6575f7abf8 100644
--- a/pulsar-client-cpp/python/pulsar/functions/context.py
+++ b/pulsar-client-cpp/python/pulsar/functions/context.py
@@ -98,6 +98,11 @@ def get_user_config_map(self):
"""Returns the entire user-defined config as a dict (the dict will be empty if no user-defined config is supplied)"""
pass
+ @abstractmethod
+ def get_secret(self, secret_name):
+ """Returns the secret value associated with the name. None if nothing was found"""
+ pass
+
@abstractmethod
def record_metric(self, metric_name, metric_value):
"""Records the metric_value. metric_value has to satisfy isinstance(metric_value, numbers.Number)"""
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index 7564d32aa8..6ab50cf96e 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -73,6 +73,7 @@
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.runtime.ProcessRuntimeFactory;
import org.apache.pulsar.functions.runtime.RuntimeSpawner;
+import org.apache.pulsar.functions.secretsproviderconfigurator.DefaultSecretsProviderConfigurator;
import org.apache.pulsar.functions.utils.*;
import org.apache.pulsar.functions.windowing.WindowUtils;
@@ -258,6 +259,8 @@ void processArguments() throws Exception {
protected String DEPRECATED_userConfigString;
@Parameter(names = "--user-config", description = "User-defined config key/values")
protected String userConfigString;
+ @Parameter(names = "--secrets-config", description = "Secrets config key/values")
+ protected String secretsConfig;
@Parameter(names = "--retainOrdering", description = "Function consumes and processes messages in order", hidden = true)
protected Boolean DEPRECATED_retainOrdering;
@Parameter(names = "--retain-ordering", description = "Function consumes and processes messages in order")
@@ -406,6 +409,11 @@ void processArguments() throws Exception {
Map<String, Object> userConfigMap = new Gson().fromJson(userConfigString, type);
functionConfig.setUserConfig(userConfigMap);
}
+ if (null != secretsConfig) {
+ Type type = new TypeToken<Map<String, Object>>() {}.getType();
+ Map<String, Object> secretsMap = new Gson().fromJson(secretsConfig, type);
+ functionConfig.setSecrets(secretsMap);
+ }
if (functionConfig.getUserConfig() == null) {
functionConfig.setUserConfig(new HashMap<>());
}
@@ -1027,7 +1035,7 @@ protected static void startLocalRun(org.apache.pulsar.functions.proto.Function.F
}
try (ProcessRuntimeFactory containerFactory = new ProcessRuntimeFactory(serviceUrl, stateStorageServiceUrl, authConfig, null, null,
- null)) {
+ null, new DefaultSecretsProviderConfigurator())) {
List<RuntimeSpawner> spawners = new LinkedList<>();
for (int i = 0; i < parallelism; ++i) {
InstanceConfig instanceConfig = new InstanceConfig();
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
index a3bfd88025..5f7f8d2828 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
@@ -292,6 +292,8 @@ void runCmd() throws Exception {
protected String DEPRECATED_sinkConfigString;
@Parameter(names = "--sink-config", description = "User defined configs key/values")
protected String sinkConfigString;
+ @Parameter(names = "--secrets-config", description = "Secrets config key/values")
+ protected String secretsConfig;
@Parameter(names = "--auto-ack", description = "Whether or not the framework will automatically acknowleges messages", arity = 1)
protected boolean autoAck = true;
@Parameter(names = "--timeout-ms", description = "The message timeout in milliseconds")
@@ -406,6 +408,12 @@ void processArguments() throws Exception {
sinkConfig.setConfigs(parseConfigs(sinkConfigString));
}
+ if (null != secretsConfig) {
+ Type type = new TypeToken<Map<String, Object>>() {}.getType();
+ Map<String, Object> secretsMap = new Gson().fromJson(secretsConfig, type);
+ sinkConfig.setSecrets(secretsMap);
+ }
+
sinkConfig.setAutoAck(autoAck);
if (timeoutMs != null) {
sinkConfig.setTimeoutMs(timeoutMs);
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
index baeb5e0b09..c4d7f53cc7 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
@@ -274,6 +274,8 @@ void runCmd() throws Exception {
@Parameter(names = "--source-config-file", description = "The path to a YAML config file specifying the "
+ "source's configuration")
protected String sourceConfigFile;
+ @Parameter(names = "--secrets-config", description = "Secrets config key/values")
+ protected String secretsConfig;
@Parameter(names = "--cpu", description = "The CPU (in cores) that needs to be allocated per source instance (applicable only to Docker runtime)")
protected Double cpu;
@Parameter(names = "--ram", description = "The RAM (in bytes) that need to be allocated per source instance (applicable only to the process and Docker runtimes)")
@@ -371,6 +373,12 @@ void processArguments() throws Exception {
sourceConfig.setConfigs(parseConfigs(sourceConfigString));
}
+ if (null != secretsConfig) {
+ Type type = new TypeToken<Map<String, Object>>() {}.getType();
+ Map<String, Object> secretsMap = new Gson().fromJson(secretsConfig, type);
+ sourceConfig.setSecrets(secretsMap);
+ }
+
inferMissingArguments(sourceConfig);
// check if source configs are valid
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
index ea5866ac20..2163df1db9 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
@@ -79,6 +79,11 @@
private ProcessingGuarantees processingGuarantees;
private boolean retainOrdering;
private Map<String, Object> userConfig;
+ // This is a map of secretName(aka how the secret is going to be
+ // accessed in the function via context) to an object that
+ // encapsulates how the secret is fetched by the underlying
+ // secrets provider
+ private Map<String, Object> secrets;
private Runtime runtime;
private boolean autoAck;
private int maxMessageRetries = -1;
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java
index c4bf1ca4f4..355c6962e5 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java
@@ -55,6 +55,11 @@
private Map<String, ConsumerConfig> inputSpecs = new TreeMap<>();
private Map<String, Object> configs;
+ // This is a map of secretName(aka how the secret is going to be
+ // accessed in the function via context) to an object that
+ // encapsulates how the secret is fetched by the underlying
+ // secrets provider
+ private Map<String, Object> secrets;
private int parallelism = 1;
private FunctionConfig.ProcessingGuarantees processingGuarantees;
private boolean retainOrdering;
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java
index 57c0f79561..9dbe97c3cd 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java
@@ -46,6 +46,11 @@
private String schemaType;
private Map<String, Object> configs;
+ // This is a map of secretName(aka how the secret is going to be
+ // accessed in the function via context) to an object that
+ // encapsulates how the secret is fetched by the underlying
+ // secrets provider
+ private Map<String, Object> secrets;
private int parallelism = 1;
private FunctionConfig.ProcessingGuarantees processingGuarantees;
private Resources resources;
diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java
index c66ea6efa7..e9b79c3699 100644
--- a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java
+++ b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java
@@ -159,6 +159,13 @@
*/
Object getUserConfigValueOrDefault(String key, Object defaultValue);
+ /**
+ * Get the secret associated with this key
+ * @param secretName The name of the secret
+ * @return The secret if anything was found or null
+ */
+ String getSecret(String secretName);
+
/**
* Record a user defined metric
* @param metricName The name of the metric
diff --git a/pulsar-functions/instance/pom.xml b/pulsar-functions/instance/pom.xml
index e206b111cb..c2bac05503 100644
--- a/pulsar-functions/instance/pom.xml
+++ b/pulsar-functions/instance/pom.xml
@@ -71,6 +71,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions-secretsprovider</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.bookkeeper</groupId>
<artifactId>stream-storage-java-client</artifactId>
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
index 4d474333b9..406fe13db6 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
@@ -49,6 +49,7 @@
import org.apache.pulsar.functions.instance.state.StateContextImpl;
import org.apache.pulsar.functions.proto.Function.SinkSpec;
import org.apache.pulsar.functions.proto.InstanceCommunication.MetricsData;
+import org.apache.pulsar.functions.secretsprovider.SecretsProvider;
import org.apache.pulsar.functions.source.TopicSchema;
import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.core.SourceContext;
@@ -101,12 +102,16 @@ public void update(double value) {
private final TopicSchema topicSchema;
+ private final SecretsProvider secretsProvider;
+ private final Map<String, Object> secretsMap;
+
@Getter
@Setter
private StateContextImpl stateContext;
private Map<String, Object> userConfigs;
- public ContextImpl(InstanceConfig config, Logger logger, PulsarClient client, List<String> inputTopics) {
+ public ContextImpl(InstanceConfig config, Logger logger, PulsarClient client, List<String> inputTopics,
+ SecretsProvider secretsProvider) {
this.config = config;
this.logger = logger;
this.currentAccumulatedMetrics = new ConcurrentHashMap<>();
@@ -125,6 +130,14 @@ public ContextImpl(InstanceConfig config, Logger logger, PulsarClient client, Li
new TypeToken<Map<String, Object>>() {
}.getType());
}
+ this.secretsProvider = secretsProvider;
+ if (!StringUtils.isEmpty(config.getFunctionDetails().getSecretsMap())) {
+ secretsMap = new Gson().fromJson(config.getFunctionDetails().getSecretsMap(),
+ new TypeToken<Map<String, Object>>() {
+ }.getType());
+ } else {
+ secretsMap = new HashMap<>();
+ }
}
public void setCurrentMessageContext(Record<?> record) {
@@ -212,6 +225,15 @@ public Object getUserConfigValueOrDefault(String key, Object defaultValue) {
return userConfigs;
}
+ @Override
+ public String getSecret(String secretName) {
+ if (secretsMap.containsKey(secretName)) {
+ return secretsProvider.provideSecret(secretName, secretsMap.get(secretName));
+ } else {
+ return null;
+ }
+ }
+
private void ensureStateEnabled() {
checkState(null != stateContext, "State is not enabled.");
}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index ace5efc57b..f626c1066d 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -49,6 +49,7 @@
import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.MetricsData.Builder;
+import org.apache.pulsar.functions.secretsprovider.SecretsProvider;
import org.apache.pulsar.functions.sink.PulsarSink;
import org.apache.pulsar.functions.sink.PulsarSinkConfig;
import org.apache.pulsar.functions.sink.PulsarSinkDisable;
@@ -110,6 +111,8 @@
private Source source;
private Sink sink;
+ private final SecretsProvider secretsProvider;
+
public static final String METRICS_TOTAL_PROCESSED = "__total_processed__";
public static final String METRICS_TOTAL_SUCCESS = "__total_successfully_processed__";
public static final String METRICS_TOTAL_SYS_EXCEPTION = "__total_system_exceptions__";
@@ -122,13 +125,15 @@ public JavaInstanceRunnable(InstanceConfig instanceConfig,
FunctionCacheManager fnCache,
String jarFile,
PulsarClient pulsarClient,
- String stateStorageServiceUrl) {
+ String stateStorageServiceUrl,
+ SecretsProvider secretsProvider) {
this.instanceConfig = instanceConfig;
this.fnCache = fnCache;
this.jarFile = jarFile;
this.client = (PulsarClientImpl) pulsarClient;
this.stateStorageServiceUrl = stateStorageServiceUrl;
this.stats = new FunctionStats();
+ this.secretsProvider = secretsProvider;
}
/**
@@ -173,7 +178,7 @@ ContextImpl setupContext() {
}
Logger instanceLog = LoggerFactory.getLogger(
"function-" + instanceConfig.getFunctionDetails().getName());
- return new ContextImpl(instanceConfig, instanceLog, client, inputTopics);
+ return new ContextImpl(instanceConfig, instanceLog, client, inputTopics, secretsProvider);
}
/**
diff --git a/pulsar-functions/instance/src/main/python/Function_pb2.py b/pulsar-functions/instance/src/main/python/Function_pb2.py
index b09e1052fe..d6cc8af085 100644
--- a/pulsar-functions/instance/src/main/python/Function_pb2.py
+++ b/pulsar-functions/instance/src/main/python/Function_pb2.py
@@ -17,6 +17,7 @@
# under the License.
#
+# Generated by the protocol buffer compiler. DO NOT EDIT!
# Generated by the protocol buffer compiler. DO NOT EDIT!
# source: Function.proto
@@ -39,7 +40,7 @@
name='Function.proto',
package='proto',
syntax='proto3',
- serialized_pb=_b('\n\x0e\x46unction.proto\x12\x05proto\"3\n\tResources\x12\x0b\n\x03\x63pu\x18\x01 \x01(\x01\x12\x0b\n\x03ram\x18\x02 \x01(\x03\x12\x0c\n\x04\x64isk\x18\x03 \x01(\x03\"B\n\x0cRetryDetails\x12\x19\n\x11maxMessageRetries\x18\x01 \x01(\x05\x12\x17\n\x0f\x64\x65\x61\x64LetterTopic\x18\x02 \x01(\t\"\xd4\x03\n\x0f\x46unctionDetails\x12\x0e\n\x06tenant\x18\x01 \x01(\t\x12\x11\n\tnamespace\x18\x02 \x01(\t\x12\x0c\n\x04name\x18\x03 \x01(\t\x12\x11\n\tclassName\x18\x04 \x01(\t\x12\x10\n\x08logTopic\x18\x05 \x01(\t\x12\x39\n\x14processingGuarantees\x18\x06 \x01(\x0e\x32\x1b.proto.ProcessingGuarantees\x12\x12\n\nuserConfig\x18\x07 \x01(\t\x12/\n\x07runtime\x18\x08 \x01(\x0e\x32\x1e.proto.FunctionDetails.Runtime\x12\x0f\n\x07\x61utoAck\x18\t \x01(\x08\x12\x13\n\x0bparallelism\x18\n \x01(\x05\x12!\n\x06source\x18\x0b \x01(\x0b\x32\x11.proto.SourceSpec\x12\x1d\n\x04sink\x18\x0c \x01(\x0b\x32\x0f.proto.SinkSpec\x12#\n\tresources\x18\r \x01(\x0b\x32\x10.proto.Resources\x12\x12\n\npackageUrl\x18\x0e \x01(\t\x12)\n\x0cretryDetails\x18\x0f \x01(\x0b\x32\x13.proto.RetryDetails\"\x1f\n\x07Runtime\x12\x08\n\x04JAVA\x10\x00\x12\n\n\x06PYTHON\x10\x01\"R\n\x0c\x43onsumerSpec\x12\x12\n\nschemaType\x18\x01 \x01(\t\x12\x16\n\x0eserdeClassName\x18\x02 \x01(\t\x12\x16\n\x0eisRegexPattern\x18\x03 \x01(\x08\"\xe4\x03\n\nSourceSpec\x12\x11\n\tclassName\x18\x01 \x01(\t\x12\x0f\n\x07\x63onfigs\x18\x02 \x01(\t\x12\x15\n\rtypeClassName\x18\x05 \x01(\t\x12\x31\n\x10subscriptionType\x18\x03 \x01(\x0e\x32\x17.proto.SubscriptionType\x12Q\n\x16topicsToSerDeClassName\x18\x04 \x03(\x0b\x32-.proto.SourceSpec.TopicsToSerDeClassNameEntryB\x02\x18\x01\x12\x35\n\ninputSpecs\x18\n \x03(\x0b\x32!.proto.SourceSpec.InputSpecsEntry\x12\x11\n\ttimeoutMs\x18\x06 \x01(\x04\x12\x19\n\rtopicsPattern\x18\x07 \x01(\tB\x02\x18\x01\x12\x0f\n\x07\x62uiltin\x18\x08 \x01(\t\x12\x18\n\x10subscriptionName\x18\t \x01(\t\x1a=\n\x1bTopicsToSerDeClassNameEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x46\n\x0fInputSpecsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\"\n\x05value\x18\x02 \x01(\x0b\x32\x13.proto.ConsumerSpec:\x02\x38\x01\"\x91\x01\n\x08SinkSpec\x12\x11\n\tclassName\x18\x01 \x01(\t\x12\x0f\n\x07\x63onfigs\x18\x02 \x01(\t\x12\x15\n\rtypeClassName\x18\x05 \x01(\t\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x16\n\x0eserDeClassName\x18\x04 \x01(\t\x12\x0f\n\x07\x62uiltin\x18\x06 \x01(\t\x12\x12\n\nschemaType\x18\x07 \x01(\t\"H\n\x17PackageLocationMetaData\x12\x13\n\x0bpackagePath\x18\x01 \x01(\t\x12\x18\n\x10originalFileName\x18\x02 \x01(\t\"\xa1\x01\n\x10\x46unctionMetaData\x12/\n\x0f\x66unctionDetails\x18\x01 \x01(\x0b\x32\x16.proto.FunctionDetails\x12\x37\n\x0fpackageLocation\x18\x02 \x01(\x0b\x32\x1e.proto.PackageLocationMetaData\x12\x0f\n\x07version\x18\x03 \x01(\x04\x12\x12\n\ncreateTime\x18\x04 \x01(\x04\"Q\n\x08Instance\x12\x31\n\x10\x66unctionMetaData\x18\x01 \x01(\x0b\x32\x17.proto.FunctionMetaData\x12\x12\n\ninstanceId\x18\x02 \x01(\x05\"A\n\nAssignment\x12!\n\x08instance\x18\x01 \x01(\x0b\x32\x0f.proto.Instance\x12\x10\n\x08workerId\x18\x02 \x01(\t*O\n\x14ProcessingGuarantees\x12\x10\n\x0c\x41TLEAST_ONCE\x10\x00\x12\x0f\n\x0b\x41TMOST_ONCE\x10\x01\x12\x14\n\x10\x45\x46\x46\x45\x43TIVELY_ONCE\x10\x02*,\n\x10SubscriptionType\x12\n\n\x06SHARED\x10\x00\x12\x0c\n\x08\x46\x41ILOVER\x10\x01\x42-\n!org.apache.pulsar.functions.protoB\x08\x46unctionb\x06proto3')
+ serialized_pb=_b('\n\x0e\x46unction.proto\x12\x05proto\"3\n\tResources\x12\x0b\n\x03\x63pu\x18\x01 \x01(\x01\x12\x0b\n\x03ram\x18\x02 \x01(\x03\x12\x0c\n\x04\x64isk\x18\x03 \x01(\x03\"B\n\x0cRetryDetails\x12\x19\n\x11maxMessageRetries\x18\x01 \x01(\x05\x12\x17\n\x0f\x64\x65\x61\x64LetterTopic\x18\x02 \x01(\t\"\xe8\x03\n\x0f\x46unctionDetails\x12\x0e\n\x06tenant\x18\x01 \x01(\t\x12\x11\n\tnamespace\x18\x02 \x01(\t\x12\x0c\n\x04name\x18\x03 \x01(\t\x12\x11\n\tclassName\x18\x04 \x01(\t\x12\x10\n\x08logTopic\x18\x05 \x01(\t\x12\x39\n\x14processingGuarantees\x18\x06 \x01(\x0e\x32\x1b.proto.ProcessingGuarantees\x12\x12\n\nuserConfig\x18\x07 \x01(\t\x12\x12\n\nsecretsMap\x18\x10 \x01(\t\x12/\n\x07runtime\x18\x08 \x01(\x0e\x32\x1e.proto.FunctionDetails.Runtime\x12\x0f\n\x07\x61utoAck\x18\t \x01(\x08\x12\x13\n\x0bparallelism\x18\n \x01(\x05\x12!\n\x06source\x18\x0b \x01(\x0b\x32\x11.proto.SourceSpec\x12\x1d\n\x04sink\x18\x0c \x01(\x0b\x32\x0f.proto.SinkSpec\x12#\n\tresources\x18\r \x01(\x0b\x32\x10.proto.Resources\x12\x12\n\npackageUrl\x18\x0e \x01(\t\x12)\n\x0cretryDetails\x18\x0f \x01(\x0b\x32\x13.proto.RetryDetails\"\x1f\n\x07Runtime\x12\x08\n\x04JAVA\x10\x00\x12\n\n\x06PYTHON\x10\x01\"R\n\x0c\x43onsumerSpec\x12\x12\n\nschemaType\x18\x01 \x01(\t\x12\x16\n\x0eserdeClassName\x18\x02 \x01(\t\x12\x16\n\x0eisRegexPattern\x18\x03 \x01(\x08\"\xe4\x03\n\nSourceSpec\x12\x11\n\tclassName\x18\x01 \x01(\t\x12\x0f\n\x07\x63onfigs\x18\x02 \x01(\t\x12\x15\n\rtypeClassName\x18\x05 \x01(\t\x12\x31\n\x10subscriptionType\x18\x03 \x01(\x0e\x32\x17.proto.SubscriptionType\x12Q\n\x16topicsToSerDeClassName\x18\x04 \x03(\x0b\x32-.proto.SourceSpec.TopicsToSerDeClassNameEntryB\x02\x18\x01\x12\x35\n\ninputSpecs\x18\n \x03(\x0b\x32!.proto.SourceSpec.InputSpecsEntry\x12\x11\n\ttimeoutMs\x18\x06 \x01(\x04\x12\x19\n\rtopicsPattern\x18\x07 \x01(\tB\x02\x18\x01\x12\x0f\n\x07\x62uiltin\x18\x08 \x01(\t\x12\x18\n\x10subscriptionName\x18\t \x01(\t\x1a=\n\x1bTopicsToSerDeClassNameEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x46\n\x0fInputSpecsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\"\n\x05value\x18\x02 \x01(\x0b\x32\x13.proto.ConsumerSpec:\x02\x38\x01\"\x91\x01\n\x08SinkSpec\x12\x11\n\tclassName\x18\x01 \x01(\t\x12\x0f\n\x07\x63onfigs\x18\x02 \x01(\t\x12\x15\n\rtypeClassName\x18\x05 \x01(\t\x12\r\n\x05topic\x18\x03 \x01(\t\x12\x16\n\x0eserDeClassName\x18\x04 \x01(\t\x12\x0f\n\x07\x62uiltin\x18\x06 \x01(\t\x12\x12\n\nschemaType\x18\x07 \x01(\t\"H\n\x17PackageLocationMetaData\x12\x13\n\x0bpackagePath\x18\x01 \x01(\t\x12\x18\n\x10originalFileName\x18\x02 \x01(\t\"\xa1\x01\n\x10\x46unctionMetaData\x12/\n\x0f\x66unctionDetails\x18\x01 \x01(\x0b\x32\x16.proto.FunctionDetails\x12\x37\n\x0fpackageLocation\x18\x02 \x01(\x0b\x32\x1e.proto.PackageLocationMetaData\x12\x0f\n\x07version\x18\x03 \x01(\x04\x12\x12\n\ncreateTime\x18\x04 \x01(\x04\"Q\n\x08Instance\x12\x31\n\x10\x66unctionMetaData\x18\x01 \x01(\x0b\x32\x17.proto.FunctionMetaData\x12\x12\n\ninstanceId\x18\x02 \x01(\x05\"A\n\nAssignment\x12!\n\x08instance\x18\x01 \x01(\x0b\x32\x0f.proto.Instance\x12\x10\n\x08workerId\x18\x02 \x01(\t*O\n\x14ProcessingGuarantees\x12\x10\n\x0c\x41TLEAST_ONCE\x10\x00\x12\x0f\n\x0b\x41TMOST_ONCE\x10\x01\x12\x14\n\x10\x45\x46\x46\x45\x43TIVELY_ONCE\x10\x02*,\n\x10SubscriptionType\x12\n\n\x06SHARED\x10\x00\x12\x0c\n\x08\x46\x41ILOVER\x10\x01\x42-\n!org.apache.pulsar.functions.protoB\x08\x46unctionb\x06proto3')
)
_PROCESSINGGUARANTEES = _descriptor.EnumDescriptor(
@@ -63,8 +64,8 @@
],
containing_type=None,
options=None,
- serialized_start=1724,
- serialized_end=1803,
+ serialized_start=1744,
+ serialized_end=1823,
)
_sym_db.RegisterEnumDescriptor(_PROCESSINGGUARANTEES)
@@ -86,8 +87,8 @@
],
containing_type=None,
options=None,
- serialized_start=1805,
- serialized_end=1849,
+ serialized_start=1825,
+ serialized_end=1869,
)
_sym_db.RegisterEnumDescriptor(_SUBSCRIPTIONTYPE)
@@ -116,8 +117,8 @@
],
containing_type=None,
options=None,
- serialized_start=584,
- serialized_end=615,
+ serialized_start=604,
+ serialized_end=635,
)
_sym_db.RegisterEnumDescriptor(_FUNCTIONDETAILS_RUNTIME)
@@ -262,56 +263,63 @@
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='runtime', full_name='proto.FunctionDetails.runtime', index=7,
+ name='secretsMap', full_name='proto.FunctionDetails.secretsMap', index=7,
+ number=16, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='runtime', full_name='proto.FunctionDetails.runtime', index=8,
number=8, type=14, cpp_type=8, label=1,
has_default_value=False, default_value=0,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='autoAck', full_name='proto.FunctionDetails.autoAck', index=8,
+ name='autoAck', full_name='proto.FunctionDetails.autoAck', index=9,
number=9, type=8, cpp_type=7, label=1,
has_default_value=False, default_value=False,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='parallelism', full_name='proto.FunctionDetails.parallelism', index=9,
+ name='parallelism', full_name='proto.FunctionDetails.parallelism', index=10,
number=10, type=5, cpp_type=1, label=1,
has_default_value=False, default_value=0,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='source', full_name='proto.FunctionDetails.source', index=10,
+ name='source', full_name='proto.FunctionDetails.source', index=11,
number=11, type=11, cpp_type=10, label=1,
has_default_value=False, default_value=None,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='sink', full_name='proto.FunctionDetails.sink', index=11,
+ name='sink', full_name='proto.FunctionDetails.sink', index=12,
number=12, type=11, cpp_type=10, label=1,
has_default_value=False, default_value=None,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='resources', full_name='proto.FunctionDetails.resources', index=12,
+ name='resources', full_name='proto.FunctionDetails.resources', index=13,
number=13, type=11, cpp_type=10, label=1,
has_default_value=False, default_value=None,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='packageUrl', full_name='proto.FunctionDetails.packageUrl', index=13,
+ name='packageUrl', full_name='proto.FunctionDetails.packageUrl', index=14,
number=14, type=9, cpp_type=9, label=1,
has_default_value=False, default_value=_b("").decode('utf-8'),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
- name='retryDetails', full_name='proto.FunctionDetails.retryDetails', index=14,
+ name='retryDetails', full_name='proto.FunctionDetails.retryDetails', index=15,
number=15, type=11, cpp_type=10, label=1,
has_default_value=False, default_value=None,
message_type=None, enum_type=None, containing_type=None,
@@ -331,7 +339,7 @@
oneofs=[
],
serialized_start=147,
- serialized_end=615,
+ serialized_end=635,
)
@@ -375,8 +383,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=617,
- serialized_end=699,
+ serialized_start=637,
+ serialized_end=719,
)
@@ -413,8 +421,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=1053,
- serialized_end=1114,
+ serialized_start=1073,
+ serialized_end=1134,
)
_SOURCESPEC_INPUTSPECSENTRY = _descriptor.Descriptor(
@@ -450,8 +458,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=1116,
- serialized_end=1186,
+ serialized_start=1136,
+ serialized_end=1206,
)
_SOURCESPEC = _descriptor.Descriptor(
@@ -543,8 +551,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=702,
- serialized_end=1186,
+ serialized_start=722,
+ serialized_end=1206,
)
@@ -616,8 +624,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=1189,
- serialized_end=1334,
+ serialized_start=1209,
+ serialized_end=1354,
)
@@ -654,8 +662,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=1336,
- serialized_end=1408,
+ serialized_start=1356,
+ serialized_end=1428,
)
@@ -706,8 +714,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=1411,
- serialized_end=1572,
+ serialized_start=1431,
+ serialized_end=1592,
)
@@ -744,8 +752,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=1574,
- serialized_end=1655,
+ serialized_start=1594,
+ serialized_end=1675,
)
@@ -782,8 +790,8 @@
extension_ranges=[],
oneofs=[
],
- serialized_start=1657,
- serialized_end=1722,
+ serialized_start=1677,
+ serialized_end=1742,
)
_FUNCTIONDETAILS.fields_by_name['processingGuarantees'].enum_type = _PROCESSINGGUARANTEES
diff --git a/pulsar-functions/instance/src/main/python/contextimpl.py b/pulsar-functions/instance/src/main/python/contextimpl.py
index 24246caed7..83a63aa98a 100644
--- a/pulsar-functions/instance/src/main/python/contextimpl.py
+++ b/pulsar-functions/instance/src/main/python/contextimpl.py
@@ -48,12 +48,13 @@ def record(self, value):
self.min = value
class ContextImpl(pulsar.Context):
- def __init__(self, instance_config, logger, pulsar_client, user_code, consumers):
+ def __init__(self, instance_config, logger, pulsar_client, user_code, consumers, secrets_provider):
self.instance_config = instance_config
self.log = logger
self.pulsar_client = pulsar_client
self.user_code_dir = os.path.dirname(user_code)
self.consumers = consumers
+ self.secrets_provider = secrets_provider
self.current_accumulated_metrics = {}
self.accumulated_metrics = {}
self.publish_producers = {}
@@ -64,6 +65,9 @@ def __init__(self, instance_config, logger, pulsar_client, user_code, consumers)
self.user_config = json.loads(instance_config.function_details.userConfig) \
if instance_config.function_details.userConfig \
else []
+ self.secrets_map = json.loads(instance_config.function_details.secretsMap) \
+ if instance_config.function_details.secretsMap \
+ else {}
# Called on a per message basis to set the context for the current message
def set_current_message_context(self, msgid, topic):
@@ -107,6 +111,11 @@ def get_user_config_value(self, key):
def get_user_config_map(self):
return self.user_config
+ def get_secret(self, secret_key):
+ if not secret_key in self.secrets_map:
+ return None
+ return self.secrets_provider.provide_secret(secret_key, self.secrets_map[secret_key])
+
def record_metric(self, metric_name, metric_value):
if not metric_name in self.current_accumulated_metrics:
self.current_accumulated_metrics[metric_name] = AccumulatedMetricDatum()
diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py
index 03dafaef2f..67d6b11963 100644
--- a/pulsar-functions/instance/src/main/python/python_instance.py
+++ b/pulsar-functions/instance/src/main/python/python_instance.py
@@ -132,7 +132,7 @@ def update(self, object):
class PythonInstance(object):
- def __init__(self, instance_id, function_id, function_version, function_details, max_buffered_tuples, expected_healthcheck_interval, user_code, pulsar_client):
+ def __init__(self, instance_id, function_id, function_version, function_details, max_buffered_tuples, expected_healthcheck_interval, user_code, pulsar_client, secrets_provider):
self.instance_config = InstanceConfig(instance_id, function_id, function_version, function_details, max_buffered_tuples)
self.user_code = user_code
self.queue = queue.Queue(max_buffered_tuples)
@@ -157,6 +157,7 @@ def __init__(self, instance_id, function_id, function_version, function_details,
self.last_health_check_ts = time.time()
self.timeout_ms = function_details.source.timeoutMs if function_details.source.timeoutMs > 0 else None
self.expected_healthcheck_interval = expected_healthcheck_interval
+ self.secrets_provider = secrets_provider
def health_check(self):
self.last_health_check_ts = time.time()
@@ -226,7 +227,7 @@ def run(self):
except:
self.function_purefunction = function_kclass
- self.contextimpl = contextimpl.ContextImpl(self.instance_config, Log, self.pulsar_client, self.user_code, self.consumers)
+ self.contextimpl = contextimpl.ContextImpl(self.instance_config, Log, self.pulsar_client, self.user_code, self.consumers, self.secrets_provider)
# Now launch a thread that does execution
self.exeuction_thread = threading.Thread(target=self.actual_execution)
self.exeuction_thread.start()
diff --git a/pulsar-functions/instance/src/main/python/python_instance_main.py b/pulsar-functions/instance/src/main/python/python_instance_main.py
index 2acfd6e662..4601190bc5 100644
--- a/pulsar-functions/instance/src/main/python/python_instance_main.py
+++ b/pulsar-functions/instance/src/main/python/python_instance_main.py
@@ -29,6 +29,8 @@
import signal
import time
import zipfile
+import json
+import inspect
import pulsar
@@ -72,11 +74,12 @@ def main():
parser.add_argument('--logging_file', required=True, help='Log file name')
parser.add_argument('--logging_config_file', required=True, help='Config file for logging')
parser.add_argument('--expected_healthcheck_interval', required=True, help='Expected time in seconds between health checks', type=int)
+ parser.add_argument('--secrets_provider', required=True, help='The classname of the secrets provider')
+ parser.add_argument('--secrets_provider_config', required=False, help='The config that needs to be passed to secrets provider')
parser.add_argument('--install_usercode_dependencies', required=False, help='For packaged python like wheel files, do we need to install all dependencies', type=bool)
parser.add_argument('--dependency_repository', required=False, help='For packaged python like wheel files, which repository to pull the dependencies from')
parser.add_argument('--extra_dependency_repository', required=False, help='For packaged python like wheel files, any extra repository to pull the dependencies from')
-
args = parser.parse_args()
function_details = Function_pb2.FunctionDetails()
args.function_details = str(args.function_details)
@@ -120,11 +123,19 @@ def main():
if args.tls_trust_cert_path:
tls_trust_cert_path = args.tls_trust_cert_path
pulsar_client = pulsar.Client(args.pulsar_serviceurl, authentication, 30, 1, 1, 50000, None, use_tls, tls_trust_cert_path, tls_allow_insecure_connection)
+
+ secrets_provider = util.import_class(os.path.dirname(inspect.getfile(inspect.currentframe())), str(args.secrets_provider))
+ secrets_provider = secrets_provider()
+ secrets_provider_config = None
+ if args.secrets_provider_config is not None:
+ secrets_provider_config = json.loads(str(args.secrets_provider_config))
+ secrets_provider.init(secrets_provider_config)
+
pyinstance = python_instance.PythonInstance(str(args.instance_id), str(args.function_id),
str(args.function_version), function_details,
int(args.max_buffered_tuples),
int(args.expected_healthcheck_interval),
- str(args.py), pulsar_client)
+ str(args.py), pulsar_client, secrets_provider)
pyinstance.run()
server_instance = server.serve(args.port, pyinstance)
diff --git a/pulsar-functions/instance/src/main/python/secretsprovider.py b/pulsar-functions/instance/src/main/python/secretsprovider.py
new file mode 100644
index 0000000000..db8e68c094
--- /dev/null
+++ b/pulsar-functions/instance/src/main/python/secretsprovider.py
@@ -0,0 +1,61 @@
+#!/usr/bin/env python
+#
+# 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.
+#
+
+# -*- encoding: utf-8 -*-
+
+"""secretsprovider.py: Interfaces and definitions for Secret Providers
+"""
+from abc import abstractmethod
+import os
+
+class SecretsProvider:
+ """Interface for providing secrets information runtime"""
+ @abstractmethod
+ def init(self, config):
+ """Do any kind of initialization"""
+ pass
+
+ @abstractmethod
+ def provide_secret(self, secret_name, path_to_secret):
+ """Fetches the secret located at the path"""
+ pass
+
+
+"""A simple implementation that represents storing secrets in clear text """
+class ClearTextSecretsProvider(SecretsProvider):
+ def __init__(self):
+ pass
+
+ def init(self, config):
+ pass
+
+ def provide_secret(self, secret_name, path_to_secret):
+ return path_to_secret
+
+"""Implementation that fetches secrets from environment variables"""
+class EnvironmentBasedSecretsProvider(SecretsProvider):
+ def __init__(self):
+ pass
+
+ def init(self, config):
+ pass
+
+ def provide_secret(self, secret_name, path_to_secret):
+ return os.environ.get(secret_name)
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java
index f5108fc6c1..e3e32fda3e 100644
--- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java
@@ -38,6 +38,7 @@
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.functions.instance.state.StateContextImpl;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.secretsprovider.EnvironmentBasedSecretsProvider;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Matchers;
@@ -73,7 +74,8 @@ public void setup() {
config,
logger,
client,
- new ArrayList<>()
+ new ArrayList<>(),
+ new EnvironmentBasedSecretsProvider()
);
}
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
index 12d4f19a04..80b3b1da5a 100644
--- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceRunnableTest.java
@@ -56,7 +56,7 @@ private static InstanceConfig createInstanceConfig(boolean addCustom, String out
private JavaInstanceRunnable createRunnable(boolean addCustom, String outputSerde) throws Exception {
InstanceConfig config = createInstanceConfig(addCustom, outputSerde);
JavaInstanceRunnable javaInstanceRunnable = new JavaInstanceRunnable(
- config, null, null, null, null);
+ config, null, null, null, null, null);
return javaInstanceRunnable;
}
diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/SecretsFunction.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/SecretsFunction.java
new file mode 100644
index 0000000000..5cd54240d0
--- /dev/null
+++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/SecretsFunction.java
@@ -0,0 +1,34 @@
+/**
+ * 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.api.examples;
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+
+/**
+ * The classic Exclamation Function that appends an exclamation at the end
+ * of the input
+ */
+public class SecretsFunction implements Function<String, String> {
+ @Override
+ public String process(String input, Context context) {
+ String secret = context.getSecret("username");
+ return String.format("%s!", secret);
+ }
+}
diff --git a/pulsar-functions/pom.xml b/pulsar-functions/pom.xml
index f54effdb13..78383bb440 100644
--- a/pulsar-functions/pom.xml
+++ b/pulsar-functions/pom.xml
@@ -43,6 +43,7 @@
<module>runtime-shaded</module>
<module>runtime-all</module>
<module>worker</module>
+ <module>secretsprovider</module>
</modules>
<dependencyManagement>
diff --git a/pulsar-functions/proto/src/main/proto/Function.proto b/pulsar-functions/proto/src/main/proto/Function.proto
index 482d901536..8d93764854 100644
--- a/pulsar-functions/proto/src/main/proto/Function.proto
+++ b/pulsar-functions/proto/src/main/proto/Function.proto
@@ -57,6 +57,7 @@ message FunctionDetails {
string logTopic = 5;
ProcessingGuarantees processingGuarantees = 6;
string userConfig = 7;
+ string secretsMap = 16;
Runtime runtime = 8;
bool autoAck = 9;
int32 parallelism = 10;
diff --git a/pulsar-functions/python-examples/secrets_function.py b/pulsar-functions/python-examples/secrets_function.py
new file mode 100755
index 0000000000..df4132317d
--- /dev/null
+++ b/pulsar-functions/python-examples/secrets_function.py
@@ -0,0 +1,31 @@
+#!/usr/bin/env python
+#
+# 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.
+#
+
+
+from pulsar import Function
+
+# Example function that uses secrets
+class SecretsFunction(Function):
+ def __init__(self):
+ pass
+
+ def process(self, input, context):
+ return "%s!" % context.get_secret("username")
+ return
diff --git a/pulsar-functions/runtime/pom.xml b/pulsar-functions/runtime/pom.xml
index 843f5e3826..2aff44d6bd 100644
--- a/pulsar-functions/runtime/pom.xml
+++ b/pulsar-functions/runtime/pom.xml
@@ -45,6 +45,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions-secretsprovider</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-all</artifactId>
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
index 1551f6f345..608fa17b79 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
@@ -22,18 +22,23 @@
import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.converters.StringConverter;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
import com.google.protobuf.Empty;
import com.google.protobuf.util.JsonFormat;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceControlGrpc;
+import java.lang.reflect.Type;
+import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
@@ -95,6 +100,12 @@
@Parameter(names = "--expected_healthcheck_interval", description = "Expected interval in seconds between healtchecks", required = true)
protected int expectedHealthCheckInterval;
+ @Parameter(names = "--secrets_provider", description = "The classname of the secrets provider", required = true)
+ protected String secretsProviderClassName;
+
+ @Parameter(names = "--secrets_provider_config", description = "The config that needs to be passed to secrets provider", required = false)
+ protected String secretsProviderConfig;
+
private Server server;
private RuntimeSpawner runtimeSpawner;
private ThreadRuntimeFactory containerFactory;
@@ -122,13 +133,20 @@ public void start() throws Exception {
instanceConfig.setFunctionDetails(functionDetails);
instanceConfig.setPort(port);
+ Map<String, String> secretsProviderConfigMap = null;
+ if (!StringUtils.isEmpty(secretsProviderConfig)) {
+ Type type = new TypeToken<Map<String, String>>() {}.getType();
+ secretsProviderConfigMap = new Gson().fromJson(secretsProviderConfig, type);
+ }
+
containerFactory = new ThreadRuntimeFactory("LocalRunnerThreadGroup", pulsarServiceUrl,
stateStorageServiceUrl,
AuthenticationConfig.builder().clientAuthenticationPlugin(clientAuthenticationPlugin)
.clientAuthenticationParameters(clientAuthenticationParameters).useTls(isTrue(useTls))
.tlsAllowInsecureConnection(isTrue(tlsAllowInsecureConnection))
.tlsHostnameVerificationEnable(isTrue(tlsHostNameVerificationEnabled))
- .tlsTrustCertsFilePath(tlsTrustCertFilePath).build());
+ .tlsTrustCertsFilePath(tlsTrustCertFilePath).build(),
+ secretsProviderClassName, secretsProviderConfigMap);
runtimeSpawner = new RuntimeSpawner(
instanceConfig,
jarFile,
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
index 80eb8400af..2c2c7f5ebc 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
@@ -22,6 +22,8 @@
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
import com.google.protobuf.Empty;
import com.google.protobuf.util.JsonFormat;
import com.squareup.okhttp.Response;
@@ -33,6 +35,7 @@
import io.kubernetes.client.models.*;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.StringUtils;
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.metrics.PrometheusMetricsServer;
@@ -40,7 +43,10 @@
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.proto.InstanceControlGrpc;
+import org.apache.pulsar.functions.secretsprovider.EnvironmentBasedSecretsProvider;
+import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
+import java.lang.reflect.Type;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -97,6 +103,7 @@
private final String userCodePkgUrl;
private final String originalCodeFileName;
private final String pulsarAdminUrl;
+ private final SecretsProviderConfigurator secretsProviderConfigurator;
private boolean running;
@@ -119,6 +126,7 @@
String pulsarAdminUrl,
String stateStorageServiceUrl,
AuthenticationConfig authConfig,
+ SecretsProviderConfigurator secretsProviderConfigurator,
Integer expectedMetricsInterval) throws Exception {
this.appsClient = appsClient;
this.coreClient = coreClient;
@@ -130,7 +138,13 @@
this.userCodePkgUrl = userCodePkgUrl;
this.originalCodeFileName = pulsarRootDir + "/" + originalCodeFileName;
this.pulsarAdminUrl = pulsarAdminUrl;
+ this.secretsProviderConfigurator = secretsProviderConfigurator;
String logConfigFile = null;
+ String secretsProviderClassName = secretsProviderConfigurator.getSecretsProviderClassName(instanceConfig.getFunctionDetails());
+ String secretsProviderConfig = null;
+ if (secretsProviderConfigurator.getSecretsProviderConfig(instanceConfig.getFunctionDetails()) != null) {
+ secretsProviderConfig = new Gson().toJson(secretsProviderConfigurator.getSecretsProviderConfig(instanceConfig.getFunctionDetails()));
+ }
switch (instanceConfig.getFunctionDetails().getRuntime()) {
case JAVA:
logConfigFile = "kubernetes_instance_log4j2.yml";
@@ -141,7 +155,7 @@
}
this.processArgs = RuntimeUtils.composeArgs(instanceConfig, instanceFile, logDirectory, this.originalCodeFileName, pulsarServiceUrl, stateStorageServiceUrl,
authConfig, "$" + ENV_SHARD_ID, GRPC_PORT, -1l, logConfigFile,
- installUserCodeDependencies, pythonDependencyRepository, pythonExtraDependencyRepository);
+ secretsProviderClassName, secretsProviderConfig, installUserCodeDependencies, pythonDependencyRepository, pythonExtraDependencyRepository);
this.prometheusMetricsServerArgs = composePrometheusMetricsServerArgs(prometheusMetricsServerJarFile, expectedMetricsInterval);
running = false;
doChecks(instanceConfig.getFunctionDetails());
@@ -156,6 +170,10 @@ public void start() throws Exception {
try {
submitStatefulSet();
} catch (Exception e) {
+ log.error("Could not submit statefulset for {}/{}/{}, deleting service as well",
+ instanceConfig.getFunctionDetails().getTenant(),
+ instanceConfig.getFunctionDetails().getNamespace(),
+ instanceConfig.getFunctionDetails().getName(), e);
deleteService();
}
running = true;
@@ -536,8 +554,10 @@ private V1Container getFunctionContainer(List<String> instanceCommand, Function.
.valueFrom(new V1EnvVarSource()
.fieldRef(new V1ObjectFieldSelector()
.fieldPath("metadata.name")));
- container.setEnv(Arrays.asList(envVarPodName));
+ container.addEnvItem(envVarPodName);
+ // Configure secrets
+ secretsProviderConfigurator.configureKubernetesRuntimeSecretsProvider(container, instanceConfig.getFunctionDetails());
// set container resources
final V1ResourceRequirements resourceRequirements = new V1ResourceRequirements();
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
index 1a180aee0c..6ed3a21221 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
@@ -29,6 +29,7 @@
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
import java.util.Map;
@@ -56,6 +57,7 @@
private final String javaInstanceJarFile;
private final String pythonInstanceFile;
private final String prometheusMetricsServerJarFile;
+ private final SecretsProviderConfigurator secretsProviderConfigurator;
private final String logDirectory = "logs/functions";
private final Integer expectedMetricsInterval;
private AppsV1Api appsClient;
@@ -75,7 +77,8 @@ public KubernetesRuntimeFactory(String k8Uri,
String pulsarAdminUri,
String stateStorageServiceUri,
AuthenticationConfig authConfig,
- Integer expectedMetricsInterval) {
+ Integer expectedMetricsInterval,
+ SecretsProviderConfigurator secretsProviderConfigurator) {
this.k8Uri = k8Uri;
if (!isEmpty(jobNamespace)) {
this.jobNamespace = jobNamespace;
@@ -105,6 +108,7 @@ public KubernetesRuntimeFactory(String k8Uri,
this.pythonInstanceFile = this.pulsarRootDir + "/instances/python-instance/python_instance_main.py";
this.prometheusMetricsServerJarFile = this.pulsarRootDir + "/instances/PrometheusMetricsServer.jar";
this.expectedMetricsInterval = expectedMetricsInterval == null ? -1 : expectedMetricsInterval;
+ this.secretsProviderConfigurator = secretsProviderConfigurator;
}
@Override
@@ -148,6 +152,7 @@ public KubernetesRuntime createContainer(InstanceConfig instanceConfig, String c
pulsarAdminUri,
stateStorageServiceUri,
authConfig,
+ secretsProviderConfigurator,
expectedMetricsInterval);
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
index 2fa7f82238..76db513715 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
@@ -22,6 +22,7 @@
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.gson.Gson;
import com.google.protobuf.Empty;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
@@ -29,9 +30,12 @@
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
+import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.proto.InstanceControlGrpc;
+import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
+import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
import java.io.InputStream;
import java.util.List;
@@ -60,6 +64,7 @@
private ScheduledExecutorService timer;
private InstanceConfig instanceConfig;
private final Long expectedHealthCheckInterval;
+ private final SecretsProviderConfigurator secretsProviderConfigurator;
private static final long GRPC_TIMEOUT_SECS = 5;
ProcessRuntime(InstanceConfig instanceConfig,
@@ -69,11 +74,18 @@
String pulsarServiceUrl,
String stateStorageServiceUrl,
AuthenticationConfig authConfig,
+ SecretsProviderConfigurator secretsProviderConfigurator,
Long expectedHealthCheckInterval) throws Exception {
this.instanceConfig = instanceConfig;
this.instancePort = instanceConfig.getPort();
this.expectedHealthCheckInterval = expectedHealthCheckInterval;
+ this.secretsProviderConfigurator = secretsProviderConfigurator;
String logConfigFile = null;
+ String secretsProviderClassName = secretsProviderConfigurator.getSecretsProviderClassName(instanceConfig.getFunctionDetails());
+ String secretsProviderConfig = null;
+ if (secretsProviderConfigurator.getSecretsProviderConfig(instanceConfig.getFunctionDetails()) != null) {
+ secretsProviderConfig = new Gson().toJson(secretsProviderConfigurator.getSecretsProviderConfig(instanceConfig.getFunctionDetails()));
+ }
switch (instanceConfig.getFunctionDetails().getRuntime()) {
case JAVA:
logConfigFile = "java_instance_log4j2.yml";
@@ -84,7 +96,7 @@
}
this.processArgs = RuntimeUtils.composeArgs(instanceConfig, instanceFile, logDirectory, codeFile, pulsarServiceUrl, stateStorageServiceUrl,
authConfig, instanceConfig.getInstanceName(), instanceConfig.getPort(), expectedHealthCheckInterval,
- logConfigFile, false, null, null);
+ logConfigFile, secretsProviderClassName, secretsProviderConfig, false, null, null);
}
/**
@@ -258,6 +270,7 @@ private void startProcess() {
deathException = null;
try {
ProcessBuilder processBuilder = new ProcessBuilder(processArgs).inheritIO();
+ secretsProviderConfigurator.configureProcessRuntimeSecretsProvider(processBuilder, instanceConfig.getFunctionDetails());
log.info("ProcessBuilder starting the process with args {}", String.join(" ", processBuilder.command()));
process = processBuilder.start();
} catch (Exception ex) {
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntimeFactory.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntimeFactory.java
index 78b069cb69..41a28fbd1d 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntimeFactory.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntimeFactory.java
@@ -24,6 +24,7 @@
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
+import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
import org.apache.pulsar.functions.utils.functioncache.FunctionCacheEntry;
import java.nio.file.Paths;
@@ -37,6 +38,7 @@
private final String pulsarServiceUrl;
private final String stateStorageServiceUrl;
private AuthenticationConfig authConfig;
+ private SecretsProviderConfigurator secretsProviderConfigurator;
private String javaInstanceJarFile;
private String pythonInstanceFile;
private String logDirectory;
@@ -47,10 +49,12 @@ public ProcessRuntimeFactory(String pulsarServiceUrl,
AuthenticationConfig authConfig,
String javaInstanceJarFile,
String pythonInstanceFile,
- String logDirectory) {
+ String logDirectory,
+ SecretsProviderConfigurator secretsProviderConfigurator) {
this.pulsarServiceUrl = pulsarServiceUrl;
this.stateStorageServiceUrl = stateStorageServiceUrl;
this.authConfig = authConfig;
+ this.secretsProviderConfigurator = secretsProviderConfigurator;
this.javaInstanceJarFile = javaInstanceJarFile;
this.pythonInstanceFile = pythonInstanceFile;
this.logDirectory = logDirectory;
@@ -113,6 +117,7 @@ public ProcessRuntime createContainer(InstanceConfig instanceConfig, String code
pulsarServiceUrl,
stateStorageServiceUrl,
authConfig,
+ secretsProviderConfigurator,
expectedHealthCheckInterval);
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
index 26475369dd..f12222ef45 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
@@ -21,6 +21,7 @@
import com.google.protobuf.util.JsonFormat;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
@@ -49,6 +50,8 @@
Integer grpcPort,
Long expectedHealthCheckInterval,
String logConfigFile,
+ String secretsProviderClassName,
+ String secretsProviderConfig,
Boolean installUserCodeDepdendencies,
String pythonDependencyRepository,
String pythonExtraDependencyRepository) throws Exception {
@@ -150,6 +153,13 @@
}
args.add("--expected_healthcheck_interval");
args.add(String.valueOf(expectedHealthCheckInterval));
+
+ args.add("--secrets_provider");
+ args.add(secretsProviderClassName);
+ if (!StringUtils.isEmpty(secretsProviderConfig)) {
+ args.add("--secrets_provider_config");
+ args.add(secretsProviderConfig);
+ }
return args;
}
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntime.java
index 05cb87ff80..c49565d8a2 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntime.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntime.java
@@ -28,6 +28,7 @@
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
+import org.apache.pulsar.functions.secretsprovider.SecretsProvider;
import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManager;
import org.apache.pulsar.functions.instance.JavaInstanceRunnable;
import org.apache.pulsar.functions.utils.FunctionDetailsUtils;
@@ -51,7 +52,8 @@
ThreadGroup threadGroup,
String jarFile,
PulsarClient pulsarClient,
- String stateStorageServiceUrl) {
+ String stateStorageServiceUrl,
+ SecretsProvider secretsProvider) {
this.instanceConfig = instanceConfig;
if (instanceConfig.getFunctionDetails().getRuntime() != Function.FunctionDetails.Runtime.JAVA) {
throw new RuntimeException("Thread Container only supports Java Runtime");
@@ -61,7 +63,8 @@
fnCache,
jarFile,
pulsarClient,
- stateStorageServiceUrl);
+ stateStorageServiceUrl,
+ secretsProvider);
this.threadGroup = threadGroup;
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntimeFactory.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntimeFactory.java
index dfbbb64efa..83e4ea6264 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntimeFactory.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ThreadRuntimeFactory.java
@@ -21,19 +21,28 @@
import com.google.common.annotations.VisibleForTesting;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
import lombok.extern.slf4j.Slf4j;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
+import org.apache.pulsar.functions.secretsprovider.SecretsProvider;
+import org.apache.pulsar.functions.utils.Reflections;
import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManager;
import org.apache.pulsar.functions.utils.functioncache.FunctionCacheManagerImpl;
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+
/**
* Thread based function container factory implementation.
*/
@@ -44,15 +53,25 @@
private final FunctionCacheManager fnCache;
private final PulsarClient pulsarClient;
private final String storageServiceUrl;
+ private final SecretsProvider secretsProvider;
private volatile boolean closed;
public ThreadRuntimeFactory(String threadGroupName, String pulsarServiceUrl, String storageServiceUrl,
- AuthenticationConfig authConfig) throws Exception {
- this(threadGroupName, createPulsarClient(pulsarServiceUrl, authConfig), storageServiceUrl);
+ AuthenticationConfig authConfig, String secretsProviderClassName,
+ Map<String, String> secretsProviderConfig) throws Exception {
+ this(threadGroupName, createPulsarClient(pulsarServiceUrl, authConfig), storageServiceUrl, secretsProviderClassName, secretsProviderConfig);
}
@VisibleForTesting
- public ThreadRuntimeFactory(String threadGroupName, PulsarClient pulsarClient, String storageServiceUrl) {
+ public ThreadRuntimeFactory(String threadGroupName, PulsarClient pulsarClient, String storageServiceUrl,
+ String secretsProviderClassName, Map<String, String> secretsProviderConfig) {
+ try {
+ secretsProvider = (SecretsProvider) Reflections.createInstance(secretsProviderClassName, ClassLoader.getSystemClassLoader());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ secretsProvider.init(secretsProviderConfig);
+
this.fnCache = new FunctionCacheManagerImpl();
this.threadGroup = new ThreadGroup(threadGroupName);
this.pulsarClient = pulsarClient;
@@ -90,7 +109,8 @@ public ThreadRuntime createContainer(InstanceConfig instanceConfig, String jarFi
threadGroup,
jarFile,
pulsarClient,
- storageServiceUrl);
+ storageServiceUrl,
+ secretsProvider);
}
@Override
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
index e82c75febf..ef03c828d6 100644
--- a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
@@ -24,6 +24,7 @@
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.ConsumerSpec;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.secretsproviderconfigurator.DefaultSecretsProviderConfigurator;
import org.apache.pulsar.functions.utils.FunctionDetailsUtils;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.Test;
@@ -72,7 +73,9 @@ public KubernetesRuntimeTest() throws Exception {
this.stateStorageServiceUrl = "bk://localhost:4181";
this.logDirectory = "logs/functions";
this.factory = spy(new KubernetesRuntimeFactory(null, null, null, pulsarRootDir,
- false, true, "myrepo", "anotherrepo", null, pulsarServiceUrl, pulsarAdminUrl, stateStorageServiceUrl, null, null));
+ false, true, "myrepo", "anotherrepo",
+ null, pulsarServiceUrl, pulsarAdminUrl, stateStorageServiceUrl, null, null,
+ new DefaultSecretsProviderConfigurator()));
doNothing().when(this.factory).setupClient();
}
@@ -121,7 +124,7 @@ public void testJavaConstructor() throws Exception {
KubernetesRuntime container = factory.createContainer(config, userJarFile, userJarFile, 30l);
List<String> args = container.getProcessArgs();
- assertEquals(args.size(), 28);
+ assertEquals(args.size(), 30);
String expectedArgs = "java -cp " + javaInstanceJarFile
+ " -Dpulsar.functions.java.instance.jar=" + javaInstanceJarFile
+ " -Dlog4j.configurationFile=kubernetes_instance_log4j2.yml "
@@ -135,7 +138,8 @@ public void testJavaConstructor() throws Exception {
+ "' --pulsar_serviceurl " + pulsarServiceUrl
+ " --max_buffered_tuples 1024 --port " + args.get(23)
+ " --state_storage_serviceurl " + stateStorageServiceUrl
- + " --expected_healthcheck_interval -1";
+ + " --expected_healthcheck_interval -1"
+ + " --secrets_provider org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider";
assertEquals(String.join(" ", args), expectedArgs);
}
@@ -145,7 +149,7 @@ public void testPythonConstructor() throws Exception {
KubernetesRuntime container = factory.createContainer(config, userJarFile, userJarFile, 30l);
List<String> args = container.getProcessArgs();
- assertEquals(args.size(), 32);
+ assertEquals(args.size(), 34);
String expectedArgs = "python " + pythonInstanceFile
+ " --py " + pulsarRootDir + "/" + userJarFile
+ " --logging_directory " + logDirectory
@@ -160,7 +164,8 @@ public void testPythonConstructor() throws Exception {
+ " --function_details '" + JsonFormat.printer().omittingInsignificantWhitespace().print(config.getFunctionDetails())
+ "' --pulsar_serviceurl " + pulsarServiceUrl
+ " --max_buffered_tuples 1024 --port " + args.get(29)
- + " --expected_healthcheck_interval -1";
+ + " --expected_healthcheck_interval -1"
+ + " --secrets_provider secretsprovider.ClearTextSecretsProvider";
assertEquals(String.join(" ", args), expectedArgs);
}
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
index fe492b859a..5f4aeb5144 100644
--- a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
@@ -31,6 +31,7 @@
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.ConsumerSpec;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
+import org.apache.pulsar.functions.secretsproviderconfigurator.DefaultSecretsProviderConfigurator;
import org.apache.pulsar.functions.utils.FunctionDetailsUtils;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.Test;
@@ -67,7 +68,8 @@ public ProcessRuntimeTest() {
this.stateStorageServiceUrl = "bk://localhost:4181";
this.logDirectory = "Users/user/logs";
this.factory = new ProcessRuntimeFactory(
- pulsarServiceUrl, stateStorageServiceUrl, null, javaInstanceJarFile, pythonInstanceFile, logDirectory);
+ pulsarServiceUrl, stateStorageServiceUrl, null, javaInstanceJarFile, pythonInstanceFile, logDirectory,
+ new DefaultSecretsProviderConfigurator());
}
@AfterMethod
@@ -115,7 +117,7 @@ public void testJavaConstructor() throws Exception {
ProcessRuntime container = factory.createContainer(config, userJarFile, null, 30l);
List<String> args = container.getProcessArgs();
- assertEquals(args.size(), 28);
+ assertEquals(args.size(), 30);
String expectedArgs = "java -cp " + javaInstanceJarFile
+ " -Dpulsar.functions.java.instance.jar=" + javaInstanceJarFile
+ " -Dlog4j.configurationFile=java_instance_log4j2.yml "
@@ -129,7 +131,8 @@ public void testJavaConstructor() throws Exception {
+ "' --pulsar_serviceurl " + pulsarServiceUrl
+ " --max_buffered_tuples 1024 --port " + args.get(23)
+ " --state_storage_serviceurl " + stateStorageServiceUrl
- + " --expected_healthcheck_interval 30";
+ + " --expected_healthcheck_interval 30"
+ + " --secrets_provider org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider";
assertEquals(String.join(" ", args), expectedArgs);
}
@@ -139,7 +142,7 @@ public void testPythonConstructor() throws Exception {
ProcessRuntime container = factory.createContainer(config, userJarFile, null, 30l);
List<String> args = container.getProcessArgs();
- assertEquals(args.size(), 26);
+ assertEquals(args.size(), 28);
String expectedArgs = "python " + pythonInstanceFile
+ " --py " + userJarFile + " --logging_directory "
+ logDirectory + "/functions" + " --logging_file " + config.getFunctionDetails().getName()
@@ -149,7 +152,8 @@ public void testPythonConstructor() throws Exception {
+ " --function_details '" + JsonFormat.printer().omittingInsignificantWhitespace().print(config.getFunctionDetails())
+ "' --pulsar_serviceurl " + pulsarServiceUrl
+ " --max_buffered_tuples 1024 --port " + args.get(23)
- + " --expected_healthcheck_interval 30";
+ + " --expected_healthcheck_interval 30"
+ + " --secrets_provider secretsprovider.ClearTextSecretsProvider";
assertEquals(String.join(" ", args), expectedArgs);
}
diff --git a/pulsar-functions/secretsprovider/pom.xml b/pulsar-functions/secretsprovider/pom.xml
new file mode 100644
index 0000000000..ca98b7f11b
--- /dev/null
+++ b/pulsar-functions/secretsprovider/pom.xml
@@ -0,0 +1,54 @@
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions</artifactId>
+ <version>2.3.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>pulsar-functions-secretsprovider</artifactId>
+ <name>Pulsar Functions :: Secrets Provider</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>io.kubernetes</groupId>
+ <artifactId>client-java</artifactId>
+ <version>2.0.0</version>
+ <scope>compile</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions-proto</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProvider.java b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProvider.java
new file mode 100644
index 0000000000..acb7c6897c
--- /dev/null
+++ b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProvider.java
@@ -0,0 +1,45 @@
+/**
+ * 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.secretsprovider;
+
+import java.util.Map;
+
+/**
+ * Context provides contextual information to the executing function.
+ * Features like which message id we are handling, whats the topic name of the
+ * message, what are our operating constraints, etc can be accessed by the
+ * executing function
+ */
+public class ClearTextSecretsProvider implements SecretsProvider {
+ /**
+ * Initialize the SecretsProvider
+ * @return
+ */
+ @Override
+ public void init(Map<String, String> config) { }
+
+ /**
+ * Fetches a secret
+ * @return The actual secret
+ */
+ @Override
+ public String provideSecret(String secretName, Object pathToSecret) {
+ return pathToSecret.toString();
+ }
+}
\ No newline at end of file
diff --git a/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProvider.java b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProvider.java
new file mode 100644
index 0000000000..e109c5974b
--- /dev/null
+++ b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProvider.java
@@ -0,0 +1,45 @@
+/**
+ * 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.secretsprovider;
+
+import java.util.Map;
+
+/**
+ * Context provides contextual information to the executing function.
+ * Features like which message id we are handling, whats the topic name of the
+ * message, what are our operating constraints, etc can be accessed by the
+ * executing function
+ */
+public class EnvironmentBasedSecretsProvider implements SecretsProvider {
+ /**
+ * Initialize the SecretsProvider
+ * @return
+ */
+ @Override
+ public void init(Map<String, String> config) { }
+
+ /**
+ * Fetches a secret
+ * @return The actual secret
+ */
+ @Override
+ public String provideSecret(String secretName, Object pathToSecret) {
+ return System.getenv(secretName);
+ }
+}
\ No newline at end of file
diff --git a/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/SecretsProvider.java b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/SecretsProvider.java
new file mode 100644
index 0000000000..8894f8764f
--- /dev/null
+++ b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsprovider/SecretsProvider.java
@@ -0,0 +1,41 @@
+/**
+ * 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.secretsprovider;
+
+import java.util.Map;
+
+/**
+ * Context provides contextual information to the executing function.
+ * Features like which message id we are handling, whats the topic name of the
+ * message, what are our operating constraints, etc can be accessed by the
+ * executing function
+ */
+public interface SecretsProvider {
+ /**
+ * Initialize the SecretsProvider
+ * @return
+ */
+ void init(Map<String, String> config);
+
+ /**
+ * Fetches a secret
+ * @return The actual secret
+ */
+ String provideSecret(String secretName, Object pathToSecret);
+}
\ No newline at end of file
diff --git a/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/DefaultSecretsProviderConfigurator.java b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/DefaultSecretsProviderConfigurator.java
new file mode 100644
index 0000000000..65069f809e
--- /dev/null
+++ b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/DefaultSecretsProviderConfigurator.java
@@ -0,0 +1,65 @@
+/**
+ * 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.secretsproviderconfigurator;
+
+import io.kubernetes.client.models.V1Container;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
+
+import java.util.Map;
+
+/**
+ * Context provides contextual information to the executing function.
+ * Features like which message id we are handling, whats the topic name of the
+ * message, what are our operating constraints, etc can be accessed by the
+ * executing function
+ */
+public class DefaultSecretsProviderConfigurator implements SecretsProviderConfigurator {
+ @Override
+ public void init(Map<String, String> config) {
+
+ }
+
+ @Override
+ public String getSecretsProviderClassName(Function.FunctionDetails functionDetails) {
+ switch (functionDetails.getRuntime()) {
+ case JAVA:
+ return ClearTextSecretsProvider.class.getName();
+ case PYTHON:
+ return "secretsprovider.ClearTextSecretsProvider";
+ default:
+ throw new RuntimeException("Unknwon runtime " + functionDetails.getRuntime());
+ }
+ }
+
+ @Override
+ public Map<String, String> getSecretsProviderConfig(Function.FunctionDetails functionDetails) {
+ return null;
+ }
+
+ @Override
+ public void configureKubernetesRuntimeSecretsProvider(V1Container container, Function.FunctionDetails functionDetails) {
+ // noop
+ }
+
+ @Override
+ public void configureProcessRuntimeSecretsProvider(ProcessBuilder processBuilder, Function.FunctionDetails functionDetails) {
+ // noop
+ }
+}
\ No newline at end of file
diff --git a/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfigurator.java b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfigurator.java
new file mode 100644
index 0000000000..c5cac1aa19
--- /dev/null
+++ b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfigurator.java
@@ -0,0 +1,87 @@
+/**
+ * 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.secretsproviderconfigurator;
+
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import io.kubernetes.client.models.V1Container;
+import io.kubernetes.client.models.V1EnvVar;
+import io.kubernetes.client.models.V1EnvVarSource;
+import io.kubernetes.client.models.V1SecretKeySelector;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
+import org.apache.pulsar.functions.secretsprovider.EnvironmentBasedSecretsProvider;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+/**
+ * Context provides contextual information to the executing function.
+ * Features like which message id we are handling, whats the topic name of the
+ * message, what are our operating constraints, etc can be accessed by the
+ * executing function
+ */
+public class KubernetesSecretsProviderConfigurator implements SecretsProviderConfigurator {
+ @Override
+ public void init(Map<String, String> config) {
+
+ }
+
+ @Override
+ public String getSecretsProviderClassName(Function.FunctionDetails functionDetails) {
+ switch (functionDetails.getRuntime()) {
+ case JAVA:
+ return EnvironmentBasedSecretsProvider.class.getName();
+ case PYTHON:
+ return "secretsprovider.EnvironmentBasedSecretsProvider";
+ default:
+ throw new RuntimeException("Unknown function runtime " + functionDetails.getRuntime());
+ }
+ }
+
+ @Override
+ public Map<String, String> getSecretsProviderConfig(Function.FunctionDetails functionDetails) {
+ return null;
+ }
+
+ @Override
+ public void configureKubernetesRuntimeSecretsProvider(V1Container container, Function.FunctionDetails functionDetails) {
+ if (!StringUtils.isEmpty(functionDetails.getSecretsMap())) {
+ Type type = new TypeToken<Map<String, Object>>() {
+ }.getType();
+ Map<String, Object> secretsMap = new Gson().fromJson(functionDetails.getSecretsMap(), type);
+ for (Map.Entry<String, Object> entry : secretsMap.entrySet()) {
+ final V1EnvVar secretEnv = new V1EnvVar();
+ Map<String, String> kv = (Map<String, String>) entry.getValue();
+ secretEnv.name(entry.getKey())
+ .valueFrom(new V1EnvVarSource()
+ .secretKeyRef(new V1SecretKeySelector()
+ .name(kv.entrySet().iterator().next().getKey())
+ .key(kv.entrySet().iterator().next().getValue())));
+ container.addEnvItem(secretEnv);
+ }
+ }
+ }
+
+ @Override
+ public void configureProcessRuntimeSecretsProvider(ProcessBuilder processBuilder, Function.FunctionDetails functionDetails) {
+ // noop
+ }
+}
\ No newline at end of file
diff --git a/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/SecretsProviderConfigurator.java b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/SecretsProviderConfigurator.java
new file mode 100644
index 0000000000..b07e85aad0
--- /dev/null
+++ b/pulsar-functions/secretsprovider/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/SecretsProviderConfigurator.java
@@ -0,0 +1,59 @@
+/**
+ * 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.secretsproviderconfigurator;
+
+import io.kubernetes.client.models.V1Container;
+import org.apache.pulsar.functions.proto.Function;
+
+import java.util.Map;
+
+/**
+ * Context provides contextual information to the executing function.
+ * Features like which message id we are handling, whats the topic name of the
+ * message, what are our operating constraints, etc can be accessed by the
+ * executing function
+ */
+public interface SecretsProviderConfigurator {
+ /**
+ * Initialize the SecretsProviderConfigurator
+ * @return
+ */
+ void init(Map<String, String> config);
+
+ /**
+ * Return the Secrets Provider Classname
+ */
+ String getSecretsProviderClassName(Function.FunctionDetails functionDetails);
+
+ /**
+ * Return the secrets provider config
+ */
+ Map<String, String> getSecretsProviderConfig(Function.FunctionDetails functionDetails);
+
+ /**
+ * Attaches any secrets specific stuff to the k8 container for kubernetes runtime
+ */
+ void configureKubernetesRuntimeSecretsProvider(V1Container container, Function.FunctionDetails functionDetails);
+
+ /**
+ * Attaches any secrets specific stuff to the k8 container for kubernetes runtime
+ */
+ void configureProcessRuntimeSecretsProvider(ProcessBuilder processBuilder, Function.FunctionDetails functionDetails);
+
+}
\ No newline at end of file
diff --git a/pulsar-functions/submit-secret-function.sh b/pulsar-functions/submit-secret-function.sh
new file mode 100755
index 0000000000..fcd48bcf3b
--- /dev/null
+++ b/pulsar-functions/submit-secret-function.sh
@@ -0,0 +1,28 @@
+#!/bin/bash
+#
+# 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.
+#
+
+
+bin/pulsar-admin functions create \
+ --output output_topic \
+ --inputs input_topics \
+ --jar `pwd`/java-examples/target/pulsar-functions-api-examples.jar \
+ --classname org.apache.pulsar.functions.api.examples.SecretsFunction \
+ --secrets-config '{"username":"mysecret"}'
+
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
index 0586a53a2d..a7e45ccc42 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
@@ -188,6 +188,10 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ClassLoader
functionDetailsBuilder.setUserConfig(new Gson().toJson(configs));
}
+ if (functionConfig.getSecrets() != null && !functionConfig.getSecrets().isEmpty()) {
+ functionDetailsBuilder.setSecretsMap(new Gson().toJson(functionConfig.getSecrets()));
+ }
+
functionDetailsBuilder.setAutoAck(functionConfig.isAutoAck());
functionDetailsBuilder.setParallelism(functionConfig.getParallelism());
if (functionConfig.getResources() != null) {
@@ -275,6 +279,12 @@ public static FunctionConfig convertFromDetails(FunctionDetails functionDetails)
}
functionConfig.setUserConfig(userConfig);
+ if (!isEmpty(functionDetails.getSecretsMap())) {
+ Type type = new TypeToken<Map<String, Object>>() {}.getType();
+ Map<String, Object> secretsMap = new Gson().fromJson(functionDetails.getSecretsMap(), type);
+ functionConfig.setSecrets(secretsMap);
+ }
+
if (functionDetails.hasResources()) {
Resources resources = new Resources();
resources.setCpu(functionDetails.getResources().getCpu());
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
index cea9dc7b1b..4834db0ae3 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
@@ -165,6 +165,9 @@ public static FunctionDetails convert(SinkConfig sinkConfig, NarClassLoader clas
if (sinkConfig.getConfigs() != null) {
sinkSpecBuilder.setConfigs(new Gson().toJson(sinkConfig.getConfigs()));
}
+ if (sinkConfig.getSecrets() != null && !sinkConfig.getSecrets().isEmpty()) {
+ functionDetailsBuilder.setSecretsMap(new Gson().toJson(sinkConfig.getSecrets()));
+ }
if (typeArg != null) {
sinkSpecBuilder.setTypeClassName(typeArg);
}
@@ -228,6 +231,11 @@ public static SinkConfig convertFromDetails(FunctionDetails functionDetails) {
Type type = new TypeToken<Map<String, String>>() {}.getType();
sinkConfig.setConfigs(new Gson().fromJson(functionDetails.getSink().getConfigs(), type));
}
+ if (!isEmpty(functionDetails.getSecretsMap())) {
+ Type type = new TypeToken<Map<String, Object>>() {}.getType();
+ Map<String, Object> secretsMap = new Gson().fromJson(functionDetails.getSecretsMap(), type);
+ sinkConfig.setSecrets(secretsMap);
+ }
if (functionDetails.hasResources()) {
Resources resources = new Resources();
resources.setCpu(functionDetails.getResources().getCpu());
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
index 8956ff4ec7..8d7c1d017c 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
@@ -98,6 +98,10 @@ public static FunctionDetails convert(SourceConfig sourceConfig, NarClassLoader
sourceSpecBuilder.setConfigs(new Gson().toJson(sourceConfig.getConfigs()));
}
+ if (sourceConfig.getSecrets() != null && !sourceConfig.getSecrets().isEmpty()) {
+ functionDetailsBuilder.setSecretsMap(new Gson().toJson(sourceConfig.getSecrets()));
+ }
+
if (typeArg != null) {
sourceSpecBuilder.setTypeClassName(typeArg);
}
@@ -156,6 +160,11 @@ public static SourceConfig convertFromDetails(FunctionDetails functionDetails) {
Type type = new TypeToken<Map<String, String>>() {}.getType();
sourceConfig.setConfigs(new Gson().fromJson(sourceSpec.getConfigs(), type));
}
+ if (!isEmpty(functionDetails.getSecretsMap())) {
+ Type type = new TypeToken<Map<String, Object>>() {}.getType();
+ Map<String, Object> secretsMap = new Gson().fromJson(functionDetails.getSecretsMap(), type);
+ sourceConfig.setSecrets(secretsMap);
+ }
Function.SinkSpec sinkSpec = functionDetails.getSink();
sourceConfig.setTopicName(sinkSpec.getTopic());
if (!StringUtils.isEmpty(sinkSpec.getSchemaType())) {
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 f7dd4bc07e..08ca2fd81f 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
@@ -54,6 +54,10 @@
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.functions.runtime.Runtime;
+import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
+import org.apache.pulsar.functions.secretsproviderconfigurator.DefaultSecretsProviderConfigurator;
+import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
+import org.apache.pulsar.functions.utils.Reflections;
/**
* This class managers all aspects of functions assignments and running of function assignments for this worker
@@ -104,6 +108,14 @@ public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerSer
this.workerService = workerService;
this.functionAdmin = workerService.getFunctionAdmin();
+ SecretsProviderConfigurator secretsProviderConfigurator;
+ if (!StringUtils.isEmpty(workerConfig.getSecretsProviderConfiguratorClassName())) {
+ secretsProviderConfigurator = (SecretsProviderConfigurator) Reflections.createInstance(workerConfig.getSecretsProviderConfiguratorClassName(), ClassLoader.getSystemClassLoader());
+ } else {
+ secretsProviderConfigurator = new DefaultSecretsProviderConfigurator();
+ }
+ secretsProviderConfigurator.init(workerConfig.getSecretsProviderConfiguratorConfig());
+
AuthenticationConfig authConfig = AuthenticationConfig.builder()
.clientAuthenticationPlugin(workerConfig.getClientAuthenticationPlugin())
.clientAuthenticationParameters(workerConfig.getClientAuthenticationParameters())
@@ -116,7 +128,8 @@ public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerSer
workerConfig.getThreadContainerFactory().getThreadGroupName(),
workerConfig.getPulsarServiceUrl(),
workerConfig.getStateStorageServiceUrl(),
- authConfig);
+ authConfig,
+ ClearTextSecretsProvider.class.getName(), null);
} else if (workerConfig.getProcessContainerFactory() != null) {
this.runtimeFactory = new ProcessRuntimeFactory(
workerConfig.getPulsarServiceUrl(),
@@ -124,7 +137,8 @@ public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerSer
authConfig,
workerConfig.getProcessContainerFactory().getJavaInstanceJarLocation(),
workerConfig.getProcessContainerFactory().getPythonInstanceLocation(),
- workerConfig.getProcessContainerFactory().getLogDirectory());
+ workerConfig.getProcessContainerFactory().getLogDirectory(),
+ secretsProviderConfigurator);
} else if (workerConfig.getKubernetesContainerFactory() != null){
this.runtimeFactory = new KubernetesRuntimeFactory(
workerConfig.getKubernetesContainerFactory().getK8Uri(),
@@ -140,7 +154,8 @@ public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerSer
StringUtils.isEmpty(workerConfig.getKubernetesContainerFactory().getPulsarAdminUrl()) ? workerConfig.getPulsarWebServiceUrl() : workerConfig.getKubernetesContainerFactory().getPulsarAdminUrl(),
workerConfig.getStateStorageServiceUrl(),
authConfig,
- workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval() == null ? -1 : workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval());
+ workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval() == null ? -1 : workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval(),
+ secretsProviderConfigurator);
} else {
throw new RuntimeException("Either Thread, Process or Kubernetes Container Factory need to be set");
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index 2b3e816a79..6ca56979b5 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -147,6 +147,9 @@
}
private KubernetesContainerFactory kubernetesContainerFactory;
+ private String secretsProviderConfiguratorClassName;
+ private Map<String, String> secretsProviderConfiguratorConfig;
+
public String getFunctionMetadataTopic() {
return String.format("persistent://%s/%s", pulsarFunctionsNamespace, functionMetadataTopicName);
}
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 087238c6af..89e3b48620 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
@@ -243,8 +243,8 @@ public Response registerFunction(final String tenant, final String namespace, fi
Utils.uploadFileToBookkeeper(packageLocationMetaDataBuilder.getPackagePath(), file, worker().getDlogNamespace());
} else {
packageLocationMetaDataBuilder.setPackagePath(createPackagePath(tenant, namespace, componentName,
- fileDetail.getName()));
- packageLocationMetaDataBuilder.setOriginalFileName(fileDetail.getName());
+ fileDetail.getFileName()));
+ packageLocationMetaDataBuilder.setOriginalFileName(fileDetail.getFileName());
log.info("Uploading {} package to {}", componentType, packageLocationMetaDataBuilder.getPackagePath());
Utils.uploadFileToBookkeeper(packageLocationMetaDataBuilder.getPackagePath(), uploadedInputStreamAsFile, worker().getDlogNamespace());
}
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
index 0b8d177c70..b184a61d7b 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
@@ -32,6 +32,7 @@
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.Assignment;
import org.apache.pulsar.functions.runtime.ThreadRuntimeFactory;
+import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
import org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler;
import org.mockito.Mockito;
import org.mockito.invocation.Invocation;
@@ -140,7 +141,7 @@ public void testSchedule() throws Exception {
functionMetaDataList.add(function1);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
// set assignments
@@ -186,7 +187,7 @@ public void testNothingNewToSchedule() throws Exception {
functionMetaDataList.add(function1);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
// set assignments
@@ -233,7 +234,7 @@ public void testAddingFunctions() throws Exception {
functionMetaDataList.add(function2);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
// set assignments
@@ -293,7 +294,7 @@ public void testDeletingFunctions() throws Exception {
functionMetaDataList.add(function1);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
// set assignments
@@ -358,7 +359,7 @@ public void testScalingUp() throws Exception {
functionMetaDataList.add(function2);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
// set assignments
@@ -469,7 +470,7 @@ public void testScalingDown() throws Exception {
functionMetaDataList.add(function2);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
// set assignments
@@ -596,7 +597,7 @@ public void testHeartbeatFunction() throws Exception {
functionMetaDataList.add(function2);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
Map<String, Map<String, Function.Assignment>> currentAssignments = new HashMap<>();
@@ -650,7 +651,7 @@ public void testUpdate() throws Exception {
functionMetaDataList.add(function2);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
// set assignments
@@ -783,7 +784,7 @@ public void testAssignmentWorkerDoesNotExist() throws InterruptedException, NoSu
functionMetaDataList.add(function2);
doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
- ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy");
+ ThreadRuntimeFactory factory = new ThreadRuntimeFactory("dummy", null, "dummy", ClearTextSecretsProvider.class.getName(), null);
doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
// set assignments
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services