[functions] Allow functions to pass runtime specific options

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java

@@ -300,6 +300,8 @@ abstract class FunctionDetailsCommand extends BaseCommand {
         @Parameter(names = "--max-message-retries", description = "How many times should we try to process a message before giving up")
         protected Integer maxMessageRetries;
         @Parameter(names = "--dead-letter-topic", description = "The topic where messages that are not processed successfully are sent to")
+        protected String customRuntimeOptions;
+        @Parameter(names = "--custom-runtime-options", description = "A string that encodes options to customize the runtime, see docs for configured runtime for details")
         protected String deadLetterTopic;
         protected FunctionConfig functionConfig;
         protected String userCodeFile;
@@ -437,6 +439,10 @@ void processArguments() throws Exception {
                 functionConfig.setTimeoutMs(timeoutMs);
             }
 
+            if (customRuntimeOptions != null) {
+                functionConfig.setCustomRuntimeOptions(customRuntimeOptions);
+            }
+
             // window configs
             WindowConfig windowConfig = functionConfig.getWindowConfig();
             if (null != windowLengthCount) {

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java

@@ -314,6 +314,8 @@ abstract class SinkDetailsCommand extends BaseCommand {
         protected Boolean autoAck;
         @Parameter(names = "--timeout-ms", description = "The message timeout in milliseconds")
         protected Long timeoutMs;
+        @Parameter(names = "--custom-runtime-options", description = "A string that encodes options to customize the runtime, see docs for configured runtime for details")
+        protected String customRuntimeOptions;
 
         protected SinkConfig sinkConfig;
 
@@ -446,6 +448,10 @@ void processArguments() throws Exception {
                 sinkConfig.setConfigs(parseConfigs(sinkConfigString));
             }
 
+            if (customRuntimeOptions != null) {
+                sinkConfig.setCustomRuntimeOptions(customRuntimeOptions);
+            }
+
             // check if configs are valid
             validateSinkConfigs(sinkConfig);
         }

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java

@@ -299,6 +299,8 @@ abstract class SourceDetailsCommand extends BaseCommand {
         protected String DEPRECATED_sourceConfigString;
         @Parameter(names = "--source-config", description = "Source config key/values")
         protected String sourceConfigString;
+        @Parameter(names = "--custom-runtime-options", description = "A string that encodes options to customize the runtime, see docs for configured runtime for details")
+        protected String customRuntimeOptions;
 
         protected SourceConfig sourceConfig;
 
@@ -392,6 +394,9 @@ void processArguments() throws Exception {
                 sourceConfig.setConfigs(parseConfigs(sourceConfigString));
             }
 
+            if (customRuntimeOptions != null) {
+                sourceConfig.setCustomRuntimeOptions(customRuntimeOptions);
+            }
             // check if source configs are valid
             validateSourceConfigs(sourceConfig);
         }

pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java

@@ -116,4 +116,8 @@ public enum Runtime {
     private String go;
     // Whether the subscriptions the functions created/used should be deleted when the functions is deleted
     private Boolean cleanupSubscription;
+    // This is an arbitrary string that can be interpreted by the function runtime
+    // to change behavior at runtime. Currently, this primarily used by the KubernetesManifestCustomizer
+    // interface
+    private String customRuntimeOptions;
 }

pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java

@@ -83,4 +83,8 @@ public class SinkConfig {
 
     // Any flags that you want to pass to the runtime.
     private String runtimeFlags;
+    // This is an arbitrary string that can be interpreted by the function runtime
+    // to change behavior at runtime. Currently, this primarily used by the KubernetesManifestCustomizer
+    // interface
+    private String customRuntimeOptions;
 }

pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java

@@ -67,4 +67,8 @@ public class SourceConfig {
     private String archive;
     // Any flags that you want to pass to the runtime.
     private String runtimeFlags;
+    // This is an arbitrary string that can be interpreted by the function runtime
+    // to change behavior at runtime. Currently, this primarily used by the KubernetesManifestCustomizer
+    // interface
+    private String customRuntimeOptions;
 }

pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java

@@ -346,7 +346,7 @@ private void startProcessMode(org.apache.pulsar.functions.proto.Function.Functio
                 null, /* python instance file */
                 null, /* log directory */
                 null, /* extra dependencies dir */
-                new DefaultSecretsProviderConfigurator(), false, Optional.empty())) {
+                new DefaultSecretsProviderConfigurator(), false, Optional.empty(), Optional.empty())) {
 
             for (int i = 0; i < parallelism; ++i) {
                 InstanceConfig instanceConfig = new InstanceConfig();

pulsar-functions/proto/src/main/proto/Function.proto

@@ -80,6 +80,7 @@ message FunctionDetails {
     RetryDetails retryDetails = 15;
     string runtimeFlags = 17;
     ComponentType componentType = 18;
+    string customRuntimeOptions = 19;
 }
 
 message ConsumerSpec {

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/ClearTextFunctionTokenAuthProvider.java

@@ -21,6 +21,7 @@
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.client.impl.auth.AuthenticationToken;
 import org.apache.pulsar.functions.instance.AuthenticationConfig;
+import org.apache.pulsar.functions.proto.Function;
 
 import java.util.Optional;
 
@@ -40,7 +41,7 @@ public void configureAuthenticationConfig(AuthenticationConfig authConfig, Optio
     }
 
     @Override
-    public Optional<FunctionAuthData> cacheAuthData(String tenant, String namespace, String name, AuthenticationDataSource authenticationDataSource) throws Exception {
+    public Optional<FunctionAuthData> cacheAuthData(Function.FunctionDetails funcDetails, AuthenticationDataSource authenticationDataSource) throws Exception {
         String token = null;
         try {
             token = getToken(authenticationDataSource);
@@ -55,13 +56,13 @@ public Optional<FunctionAuthData> cacheAuthData(String tenant, String namespace,
     }
 
     @Override
-    public Optional<FunctionAuthData> updateAuthData(String tenant, String namespace, String name,
+    public Optional<FunctionAuthData> updateAuthData(Function.FunctionDetails funcDetails,
                                                      Optional<FunctionAuthData> existingFunctionAuthData, AuthenticationDataSource authenticationDataSource) throws Exception {
-        return cacheAuthData(tenant, namespace, name, authenticationDataSource);
+        return cacheAuthData(funcDetails, authenticationDataSource);
     }
 
     @Override
-    public void cleanUpAuthData(String tenant, String namespace, String name, Optional<FunctionAuthData> functionAuthData) throws Exception {
+    public void cleanUpAuthData(Function.FunctionDetails funcDetails, Optional<FunctionAuthData> functionAuthData) throws Exception {
         //no-op
     }
 }

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/FunctionAuthProvider.java

@@ -20,6 +20,7 @@
 
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.functions.instance.AuthenticationConfig;
+import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.utils.Reflections;
 
 import java.util.Optional;
@@ -39,29 +40,22 @@ public interface FunctionAuthProvider {
 
     /**
      * Cache auth data in as part of function metadata for function that runtime may need to configure authentication
-     * @param tenant tenant that the function is running under
-     * @param namespace namespace that is the function is running under
-     * @param name name of the function
+     * @param funcDetails the function details
      * @param authenticationDataSource auth data
      * @return
      * @throws Exception
      */
-    Optional<FunctionAuthData> cacheAuthData(String tenant, String namespace, String name, AuthenticationDataSource authenticationDataSource) throws Exception;
-
-
-
-    Optional<FunctionAuthData> updateAuthData(String tenant, String namespace, String name, Optional<FunctionAuthData> existingFunctionAuthData, AuthenticationDataSource authenticationDataSource) throws Exception;
+    Optional<FunctionAuthData> cacheAuthData(Function.FunctionDetails funcDetails, AuthenticationDataSource authenticationDataSource) throws Exception;
 
+    Optional<FunctionAuthData> updateAuthData(Function.FunctionDetails funcDetails, Optional<FunctionAuthData> existingFunctionAuthData, AuthenticationDataSource authenticationDataSource) throws Exception;
 
     /**
      * Clean up operation for auth when function is terminated
-     * @param tenant tenant that the function is running under
-     * @param namespace namespace that is the function is running under
-     * @param name name of the function
+     * @param funcDetails the function details
      * @param functionAuthData function auth data
      * @throws Exception
      */
-    void cleanUpAuthData(String tenant, String namespace, String name, Optional<FunctionAuthData> functionAuthData) throws Exception;
+    void cleanUpAuthData(Function.FunctionDetails funcDetails, Optional<FunctionAuthData> functionAuthData) throws Exception;
 
     static FunctionAuthProvider getAuthProvider(String className) {
         return Reflections.createInstance(className, FunctionAuthProvider.class, Thread.currentThread().getContextClassLoader());

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/KubernetesFunctionAuthProvider.java

@@ -21,6 +21,8 @@
 import io.kubernetes.client.apis.CoreV1Api;
 import io.kubernetes.client.models.V1ServiceAccount;
 import io.kubernetes.client.models.V1StatefulSet;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.utils.Reflections;
 
 import java.util.Optional;
@@ -30,7 +32,21 @@
  */
 public interface KubernetesFunctionAuthProvider extends FunctionAuthProvider {
 
-    void initialize(CoreV1Api coreClient, String kubeNamespace, byte[] caBytes);
+    void initialize(CoreV1Api coreClient);
+
+    default void initialize(CoreV1Api coreClient, byte[] caBytes, java.util.function.Function<Function.FunctionDetails, String> namespaceCustomizerFunc) {
+        setCaBytes(caBytes);
+        setNamespaceProviderFunc(namespaceCustomizerFunc);
+        initialize(coreClient);
+    }
+
+    default void setCaBytes(byte[] caBytes) {
+
+    }
+
+    default void setNamespaceProviderFunc(java.util.function.Function<Function.FunctionDetails, String> funcDetails) {
+
+    }
 
     /**
      * Configure function statefulset spec based on function auth data

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/auth/KubernetesSecretsTokenAuthProvider.java

@@ -32,9 +32,9 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
-import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.auth.AuthenticationToken;
 import org.apache.pulsar.functions.instance.AuthenticationConfig;
+import org.apache.pulsar.functions.proto.Function;
 import org.apache.pulsar.functions.utils.Actions;
 import org.apache.pulsar.functions.utils.FunctionCommon;
 
@@ -62,16 +62,28 @@ public class KubernetesSecretsTokenAuthProvider implements KubernetesFunctionAut
 
 
     private CoreV1Api coreClient;
-    private String kubeNamespace;
     private byte[] caBytes;
+    private java.util.function.Function<Function.FunctionDetails, String> getNamespaceFromDetails;
 
     @Override
-    public void initialize(CoreV1Api coreClient, String kubeNamespace, byte[] caBytes) {
+    public void initialize(CoreV1Api coreClient) {
         this.coreClient = coreClient;
-        this.kubeNamespace = kubeNamespace;
+    }
+
+    @Override
+    public void setCaBytes(byte[] caBytes) {
         this.caBytes = caBytes;
     }
 
+    @Override
+    public void setNamespaceProviderFunc(java.util.function.Function<Function.FunctionDetails, String> getNamespaceFromDetails) {
+        this.getNamespaceFromDetails = getNamespaceFromDetails;
+    }
+
+    private String getKubeNamespace(Function.FunctionDetails funcDetails) {
+        return getNamespaceFromDetails.apply(funcDetails);
+    }
+
     @Override
     public void configureAuthDataStatefulSet(V1StatefulSet statefulSet, Optional<FunctionAuthData> functionAuthData) {
         if (!functionAuthData.isPresent()) {
@@ -115,13 +127,16 @@ public void configureAuthenticationConfig(AuthenticationConfig authConfig, Optio
 
 
     @Override
-    public Optional<FunctionAuthData> cacheAuthData(String tenant, String namespace, String name,
+    public Optional<FunctionAuthData> cacheAuthData(Function.FunctionDetails funcDetails,
                                                     AuthenticationDataSource authenticationDataSource) {
         String id = null;
+        String tenant = funcDetails.getTenant();
+        String namespace = funcDetails.getNamespace();
+        String name = funcDetails.getName();
         try {
             String token = getToken(authenticationDataSource);
             if (token != null) {
-                id = createSecret(token, tenant, namespace, name);
+                id = createSecret(token, funcDetails);
             }
         } catch (Exception e) {
             log.warn("Failed to get token for function {}", FunctionCommon.getFullyQualifiedName(tenant, namespace, name), e);
@@ -135,12 +150,12 @@ public Optional<FunctionAuthData> cacheAuthData(String tenant, String namespace,
     }
 
     @Override
-    public void cleanUpAuthData(String tenant, String namespace, String name, Optional<FunctionAuthData> functionAuthData) throws Exception {
+    public void cleanUpAuthData(Function.FunctionDetails funcDetails, Optional<FunctionAuthData> functionAuthData) throws Exception {
         if (!functionAuthData.isPresent()) {
             return;
         }
 
-        String fqfn = FunctionCommon.getFullyQualifiedName(tenant, namespace, name);
+        String fqfn = FunctionCommon.getFullyQualifiedName(funcDetails.getTenant(), funcDetails.getNamespace(), funcDetails.getName());
 
         String secretId = new String(functionAuthData.get().getData());
         // Make sure secretName is empty.  Defensive programing
@@ -150,6 +165,7 @@ public void cleanUpAuthData(String tenant, String namespace, String name, Option
         }
 
         String secretName = getSecretName(secretId);
+        String kubeNamespace = getKubeNamespace(funcDetails);
 
         Actions.Action deleteSecrets = Actions.Action.builder()
                 .actionName(String.format("Deleting secrets for function %s", fqfn))
@@ -233,7 +249,7 @@ public void cleanUpAuthData(String tenant, String namespace, String name, Option
     }
 
     @Override
-    public Optional<FunctionAuthData> updateAuthData(String tenant, String namespace, String name,
+    public Optional<FunctionAuthData> updateAuthData(Function.FunctionDetails funcDetails,
                                                      Optional<FunctionAuthData> existingFunctionAuthData,
                                                      AuthenticationDataSource authenticationDataSource) throws Exception {
 
@@ -248,15 +264,15 @@ public Optional<FunctionAuthData> updateAuthData(String tenant, String namespace
         try {
             token = getToken(authenticationDataSource);
         } catch (AuthenticationException e) {
-             // No token is passed so delete the token. Might be trying to switch over to using anonymous user
+            // No token is passed so delete the token. Might be trying to switch over to using anonymous user
             cleanUpAuthData(
-                    tenant, namespace, name,
+                    funcDetails,
                     existingFunctionAuthData);
             return Optional.empty();
         }
 
         if (token != null) {
-            upsertSecret(token, tenant, namespace, name, getSecretName(secretId));
+            upsertSecret(token, funcDetails, getSecretName(secretId));
             return Optional.of(FunctionAuthData.builder().data(secretId.getBytes()).build());
         }
 
@@ -273,8 +289,12 @@ Map<String, byte[]> buildSecretMap(String token) {
         return valueMap;
     }
 
-    private void upsertSecret(String token, String tenant, String namespace, String name, String secretName) throws InterruptedException {
+    private void upsertSecret(String token, Function.FunctionDetails funcDetails, String secretName) throws InterruptedException {
+        String tenant = funcDetails.getTenant();
+        String namespace = funcDetails.getNamespace();
+        String name = funcDetails.getName();
 
+        String kubeNamespace = getKubeNamespace(funcDetails);
         Actions.Action createAuthSecret = Actions.Action.builder()
                 .actionName(String.format("Upsert authentication secret for function %s/%s/%s", tenant, namespace, name))
                 .numRetries(NUM_RETRIES)
@@ -325,7 +345,11 @@ private void upsertSecret(String token, String tenant, String namespace, String
         }
     }
 
-    private String createSecret(String token, String tenant, String namespace, String name) throws ApiException, InterruptedException {
+    private String createSecret(String token, Function.FunctionDetails funcDetails) throws ApiException, InterruptedException {
+        String kubeNamespace = getKubeNamespace(funcDetails);
+        String tenant = funcDetails.getTenant();
+        String namespace = funcDetails.getNamespace();
+        String name = funcDetails.getName();
 
         StringBuilder sb = new StringBuilder();
         Actions.Action createAuthSecret = Actions.Action.builder()

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeCustomizer.java

@@ -0,0 +1,31 @@
+/**
+ * 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.runtime;
+
+import org.apache.pulsar.functions.utils.Reflections;
+
+import java.util.Map;
+
+public interface RuntimeCustomizer {
+    void initialize(final Map<String, Object> runtimeCustomizerConfig);
+
+    static RuntimeCustomizer getRuntimeCustomizer(String className) {
+        return Reflections.createInstance(className, RuntimeCustomizer.class, Thread.currentThread().getContextClassLoader());
+    }
+}