[improve][fn]: Added logLevel to change the log level of Function/Source/Sink Pod/Process to desired Level

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

@@ -88,6 +88,7 @@ public enum Runtime {
 
     private String outputSerdeClassName;
     private String logTopic;
+    private String logLevel;
     private ProcessingGuarantees processingGuarantees;
     // Do we want function instances to process data in the same order as in the input topics
     // This essentially means that every partition of input topic is consumed by only one instance

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

@@ -62,6 +62,8 @@ public class SinkConfig {
 
     private String deadLetterTopic;
 
+    private String logLevel;
+
     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

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

@@ -48,6 +48,8 @@ public class SourceConfig {
 
     private String schemaType;
 
+    private String logLevel;
+
     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

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

@@ -249,7 +249,9 @@ abstract class FunctionDetailsCommand extends BaseCommand {
         @Parameter(names = "--log-topic", description = "The topic to which the logs of a Pulsar Function are produced"
                 + " #Java, Python, Go")
         protected String logTopic;
-
+        @Parameter(names = "--logLevel", description = "Log level at which the logs of a Pulsar Function are produced"
+                + " #Java")
+        protected String logLevel;
         @Parameter(names = {"-st", "--schema-type"}, description = "The builtin schema type or "
                 + "custom schema class name to be used for messages output by the function #Java")
         protected String schemaType = "";
@@ -518,6 +520,9 @@ void processArguments() throws Exception {
             if (null != logTopic) {
                 functionConfig.setLogTopic(logTopic);
             }
+            if (null != logLevel) {
+                functionConfig.setLogLevel(logLevel);
+            }
             if (null != className) {
                 functionConfig.setClassName(className);
             }

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

@@ -410,7 +410,8 @@ abstract class SinkDetailsCommand extends BaseCommand {
         @Parameter(names = "--transform-function-config", description = "Configuration of the transform function "
                 + "applied before the Sink")
         protected String transformFunctionConfig;
-
+        @Parameter(names = "--logLevel", description = "Log level at which the logs of a Pulsar Sink are produced")
+        protected String logLevel;
         protected SinkConfig sinkConfig;
 
         private void mergeArgs() {
@@ -606,6 +607,10 @@ void processArguments() throws Exception {
                 sinkConfig.setTransformFunctionConfig(transformFunctionConfig);
             }
 
+            if (null != logLevel) {
+                sinkConfig.setLogLevel(logLevel);
+            }
+
             // check if configs are valid
             validateSinkConfigs(sinkConfig);
         }

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

@@ -365,6 +365,8 @@ abstract class SourceDetailsCommand extends BaseCommand {
         @Parameter(names = "--secrets", description = "The map of secretName to an object that encapsulates "
                 + "how the secret is fetched by the underlying secrets provider")
         protected String secretsString;
+        @Parameter(names = "--logLevel", description = "Log level at which the logs of a Pulsar Source are produced")
+        protected String logLevel;
 
         protected SourceConfig sourceConfig;
 
@@ -499,6 +501,10 @@ void processArguments() throws Exception {
                 sourceConfig.setSecrets(secretsMap);
             }
 
+            if (null != logLevel) {
+                sourceConfig.setLogLevel(logLevel);
+            }
+
             // check if source configs are valid
             validateSourceConfigs(sourceConfig);
         }

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

@@ -98,6 +98,7 @@ message FunctionDetails {
     bool retainOrdering = 21;
     bool retainKeyOrdering = 22;
     SubscriptionPosition subscriptionPosition = 23;
+    string logLevel = 24;
 }
 
 message ConsumerSpec {

pulsar-functions/runtime-all/src/main/resources/java_instance_log4j2.xml

@@ -120,7 +120,7 @@
             </AppenderRef>
         </Logger>
         <Root>
-            <level>info</level>
+            <level>${sys:pulsar.log.level}</level>
             <AppenderRef>
                 <ref>${sys:pulsar.log.appender}</ref>
                 <level>${sys:pulsar.log.level}</level>

pulsar-functions/runtime-all/src/main/resources/kubernetes_instance_log4j2.xml

@@ -50,7 +50,7 @@
             </AppenderRef>
         </Logger>
         <Root>
-            <level>info</level>
+            <level>${sys:pulsar.log.level}</level>
             <AppenderRef>
                 <ref>Console</ref>
                 <level>${sys:pulsar.log.level}</level>

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

@@ -78,6 +78,7 @@ public static List<String> composeCmd(InstanceConfig instanceConfig,
                                           Integer grpcPort,
                                           Long expectedHealthCheckInterval,
                                           String logConfigFile,
+                                          String logLevel,
                                           String secretsProviderClassName,
                                           String secretsProviderConfig,
                                           Boolean installUserCodeDependencies,
@@ -92,7 +93,7 @@ public static List<String> composeCmd(InstanceConfig instanceConfig,
         cmd.addAll(getCmd(instanceConfig, instanceFile, extraDependenciesDir, logDirectory,
                 originalCodeFileName, originalTransformFunctionFileName, pulsarServiceUrl, stateStorageServiceUrl,
                 authConfig, shardId, grpcPort, expectedHealthCheckInterval,
-                logConfigFile, secretsProviderClassName, secretsProviderConfig,
+                logConfigFile, logLevel, secretsProviderClassName, secretsProviderConfig,
                 installUserCodeDependencies, pythonDependencyRepository,
                 pythonExtraDependencyRepository, narExtractionDirectory,
                 functionInstanceClassPath, false, pulsarWebServiceUrl));
@@ -309,6 +310,7 @@ public static List<String> getCmd(InstanceConfig instanceConfig,
                                       Integer grpcPort,
                                       Long expectedHealthCheckInterval,
                                       String logConfigFile,
+                                      String logLevel,
                                       String secretsProviderClassName,
                                       String secretsProviderConfig,
                                       Boolean installUserCodeDependencies,
@@ -354,6 +356,7 @@ public static List<String> getCmd(InstanceConfig instanceConfig,
                 }
                 args.add(String.format("-D%s=%s", FUNCTIONS_INSTANCE_CLASSPATH, systemFunctionInstanceClasspath));
             }
+            args.add("-Dpulsar.log.level=" + logLevel);
             args.add("-Dlog4j.configurationFile=" + logConfigFile);
             args.add("-Dpulsar.function.log.dir=" + genFunctionLogFolder(logDirectory, instanceConfig));
             args.add("-Dpulsar.function.log.file=" + String.format(

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntime.java

@@ -74,6 +74,7 @@
 import lombok.extern.slf4j.Slf4j;
 import okhttp3.Response;
 import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.functions.auth.KubernetesFunctionAuthProvider;
 import org.apache.pulsar.functions.instance.AuthenticationConfig;
 import org.apache.pulsar.functions.instance.InstanceConfig;
@@ -236,6 +237,10 @@ public class KubernetesRuntime implements Runtime {
             case GO:
                 break;
         }
+        String logLevel = instanceConfig.getFunctionDetails().getLogLevel();
+        if (StringUtils.isBlank(logLevel)) {
+            logLevel = "info";
+        }
 
         this.authConfig = authConfig;
 
@@ -275,6 +280,7 @@ public class KubernetesRuntime implements Runtime {
                         grpcPort,
                         -1L,
                         logConfigFile,
+                        logLevel,
                         secretsProviderClassName,
                         secretsProviderConfig,
                         installUserCodeDependencies,

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/process/ProcessRuntime.java

@@ -122,6 +122,11 @@ class ProcessRuntime implements Runtime {
             case GO:
                 break;
         }
+        String logLevel = instanceConfig.getFunctionDetails().getLogLevel();
+        if (StringUtils.isBlank(logLevel)) {
+            logLevel = "info";
+        }
+
         this.extraDependenciesDir = extraDependenciesDir;
         this.narExtractionDirectory = narExtractionDirectory;
         this.processArgs = RuntimeUtils.composeCmd(
@@ -142,6 +147,7 @@ class ProcessRuntime implements Runtime {
             instanceConfig.getPort(),
             expectedHealthCheckInterval,
             logConfigFile,
+            logLevel,
             secretsProviderClassName,
             secretsProviderConfig,
             false,

pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/RuntimeUtilsTest.java

@@ -213,6 +213,7 @@ public void getAdditionalJavaRuntimeArguments(boolean k8sRuntime) throws Excepti
                 23,
                 1234L,
                 "logConfigFile",
+                "info",
                 "secretsProviderClassName",
                 "secretsProviderConfig",
                 false,

pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeTest.java

@@ -475,6 +475,7 @@ private void verifyJavaInstance(InstanceConfig config, String depsDir, boolean s
         String expectedArgs = "exec java -cp " + classpath
                 + extraDepsEnv
                 + " -Dpulsar.functions.instance.classpath=/pulsar/lib/*"
+                + " -Dpulsar.log.level=info"
                 + " -Dlog4j.configurationFile=kubernetes_instance_log4j2.xml "
                 + "-Dpulsar.function.log.dir=" + logDirectory + "/" + FunctionCommon.getFullyQualifiedName(config.getFunctionDetails())
                 + " -Dpulsar.function.log.file=" + config.getFunctionDetails().getName() + "-$SHARD_ID"

pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/process/ProcessRuntimeTest.java

@@ -219,7 +219,7 @@
 
         factory = createProcessRuntimeFactory(null);
 
         verifyJavaInstance(config);
     }
 
     @Test
@@ -324,6 +324,7 @@
         String expectedArgs = "java -cp " + classpath
                 + extraDepsEnv
                 + " -Dpulsar.functions.instance.classpath=/pulsar/lib/*"
+                + " -Dpulsar.log.level=info"
                 + " -Dlog4j.configurationFile=java_instance_log4j2.xml "
                 + "-Dpulsar.function.log.dir=" + logDirectory + "/functions/" + FunctionCommon.getFullyQualifiedName(config.getFunctionDetails())
                 + " -Dpulsar.function.log.file=" + config.getFunctionDetails().getName() + "-" + config.getInstanceId()

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java

@@ -32,6 +32,7 @@
 import com.google.gson.reflect.TypeToken;
 import java.io.File;
 import java.lang.reflect.Type;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -70,6 +71,7 @@ public static class ExtractedFunctionDetails {
 
     static final Integer MAX_PENDING_ASYNC_REQUESTS_DEFAULT = 1000;
     static final Boolean FORWARD_SOURCE_MESSAGE_PROPERTY_DEFAULT = Boolean.TRUE;
+    private static final List<String> VALID_LOG_LEVELS = Arrays.asList("INFO", "DEBUG", "WARN", "ERROR");
 
     private static final ObjectMapper OBJECT_MAPPER = ObjectMapperFactory.create();
 
@@ -272,6 +274,9 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ExtractedFu
         if (functionConfig.getLogTopic() != null) {
             functionDetailsBuilder.setLogTopic(functionConfig.getLogTopic());
         }
+        if (functionConfig.getLogLevel() != null) {
+            functionDetailsBuilder.setLogLevel(functionConfig.getLogLevel());
+        }
         if (functionConfig.getRuntime() != null) {
             functionDetailsBuilder.setRuntime(FunctionCommon.convertRuntime(functionConfig.getRuntime()));
         }
@@ -447,6 +452,9 @@ public static FunctionConfig convertFromDetails(FunctionDetails functionDetails)
         if (!isEmpty(functionDetails.getLogTopic())) {
             functionConfig.setLogTopic(functionDetails.getLogTopic());
         }
+        if (!isEmpty(functionDetails.getLogLevel())) {
+            functionConfig.setLogLevel(functionDetails.getLogLevel());
+        }
         if (functionDetails.getSink().getForwardSourceMessageProperty()) {
             functionConfig.setForwardSourceMessageProperty(functionDetails.getSink().getForwardSourceMessageProperty());
         }
@@ -806,6 +814,13 @@ public static void doCommonChecks(FunctionConfig functionConfig) {
             }
         }
 
+        if (!isEmpty(functionConfig.getLogLevel())) {
+            if (!VALID_LOG_LEVELS.contains(functionConfig.getLogLevel().toUpperCase())) {
+                throw new IllegalArgumentException(
+                        String.format("LogLevel %s is invalid", functionConfig.getLogLevel()));
+            }
+        }
+
         if (functionConfig.getParallelism() != null && functionConfig.getParallelism() <= 0) {
             throw new IllegalArgumentException("Function parallelism must be a positive number");
         }
@@ -1017,6 +1032,9 @@ public static FunctionConfig validateUpdate(FunctionConfig existingConfig, Funct
         if (!StringUtils.isEmpty(newConfig.getLogTopic())) {
             mergedConfig.setLogTopic(newConfig.getLogTopic());
         }
+        if (!StringUtils.isEmpty(newConfig.getLogLevel())) {
+            mergedConfig.setLogLevel(newConfig.getLogLevel());
+        }
         if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees()
                 .equals(existingConfig.getProcessingGuarantees())) {
             throw new IllegalArgumentException("Processing Guarantees cannot be altered");

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java

@@ -31,6 +31,7 @@
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -62,6 +63,8 @@
 @Slf4j
 public class SinkConfigUtils {
 
+    private static final List<String> VALID_LOG_LEVELS = Arrays.asList("INFO", "DEBUG", "WARN", "ERROR");
+
     @Getter
     @Setter
     @AllArgsConstructor
@@ -88,6 +91,9 @@ public static FunctionDetails convert(SinkConfig sinkConfig, ExtractedSinkDetail
             functionDetailsBuilder.setName(sinkConfig.getName());
         }
         functionDetailsBuilder.setRuntime(FunctionDetails.Runtime.JAVA);
+        if (sinkConfig.getLogLevel() != null) {
+            functionDetailsBuilder.setLogLevel(sinkConfig.getLogLevel());
+        }
         if (sinkConfig.getParallelism() != null) {
             functionDetailsBuilder.setParallelism(sinkConfig.getParallelism());
         } else {
@@ -396,6 +402,9 @@ public static SinkConfig convertFromDetails(FunctionDetails functionDetails) {
         if (!isEmpty(functionDetails.getUserConfig())) {
             sinkConfig.setTransformFunctionConfig(functionDetails.getUserConfig());
         }
+        if (!isEmpty(functionDetails.getLogLevel())) {
+            sinkConfig.setLogLevel(functionDetails.getLogLevel());
+        }
 
 
         return sinkConfig;
@@ -548,6 +557,13 @@ public static ExtractedSinkDetails validateAndExtractDetails(SinkConfig sinkConf
             throw new IllegalArgumentException("Only one of retain ordering or retain key ordering can be set");
         }
 
+        if (!isEmpty(sinkConfig.getLogLevel())) {
+            if (!VALID_LOG_LEVELS.contains(sinkConfig.getLogLevel().toUpperCase())) {
+                throw new IllegalArgumentException(
+                        String.format("LogLevel %s is invalid", sinkConfig.getLogLevel()));
+            }
+        }
+
         // validate user defined config if enabled and classloading is enabled
         if (validateConnectorConfig) {
             if (sinkFunction.isEnableClassloading()) {
@@ -712,6 +728,9 @@ public static SinkConfig validateUpdate(SinkConfig existingConfig, SinkConfig ne
         if (newConfig.getSourceSubscriptionPosition() != null) {
             mergedConfig.setSourceSubscriptionPosition(newConfig.getSourceSubscriptionPosition());
         }
+        if (!StringUtils.isEmpty(newConfig.getLogLevel())) {
+            mergedConfig.setLogLevel(newConfig.getLogLevel());
+        }
         return mergedConfig;
     }