add log level

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -20,6 +20,7 @@
 package com.dtstack.flink.sql;
 
 
+
 import com.dtstack.flink.sql.exec.ExecuteProcessHelper;
 import com.dtstack.flink.sql.exec.ParamsInfo;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -18,6 +18,8 @@
 
 package com.dtstack.flink.sql.exec;
 
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.LoggerContext;
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.config.CalciteConfig;
 import com.dtstack.flink.sql.enums.ClusterMode;
@@ -108,6 +110,7 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
         String remoteSqlPluginPath = options.getRemoteSqlPluginPath();
         String pluginLoadMode = options.getPluginLoadMode();
         String deployMode = options.getMode();
+        String logLevel = options.getLogLevel();
 
         Preconditions.checkArgument(checkRemoteSqlPluginPath(remoteSqlPluginPath, deployMode, pluginLoadMode),
                 "Non-local mode or shipfile deployment mode, remoteSqlPluginPath is required");
@@ -125,6 +128,7 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
                 .setDeployMode(deployMode)
                 .setConfProp(confProperties)
                 .setJarUrlList(jarURList)
+                .setLogLevel(logLevel)
                 .build();
 
     }
@@ -150,6 +154,8 @@ public static StreamExecutionEnvironment getStreamExecution(ParamsInfo paramsInf
         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
         StreamQueryConfig streamQueryConfig = StreamEnvConfigManager.getStreamQueryConfig(tableEnv, paramsInfo.getConfProp());
 
+        setLogLevel(paramsInfo.getLogLevel());
+
         SqlParser.setLocalSqlPluginRoot(paramsInfo.getLocalSqlPluginPath());
         SqlTree sqlTree = SqlParser.parseSql(paramsInfo.getSql());
 
@@ -341,4 +347,12 @@ public static StreamExecutionEnvironment getStreamExeEnv(Properties confProperti
         StreamEnvConfigManager.streamExecutionEnvironmentConfig(env, confProperties);
         return env;
     }
+
+    private static void setLogLevel(String level){
+        LoggerContext loggerContext= (LoggerContext) LoggerFactory.getILoggerFactory();
+        //设置全局日志级别
+        ch.qos.logback.classic.Logger logger = loggerContext.getLogger("root");
+        logger.setLevel(Level.toLevel(level, Level.INFO));
+    }
+
 }

core/src/main/java/com/dtstack/flink/sql/exec/ParamsInfo.java

@@ -38,17 +38,19 @@ public class ParamsInfo {
     private String remoteSqlPluginPath;
     private String pluginLoadMode;
     private String deployMode;
+    private String logLevel;
     private Properties confProp;
 
     public ParamsInfo(String sql, String name, List<URL> jarUrlList, String localSqlPluginPath,
-                      String remoteSqlPluginPath, String pluginLoadMode, String deployMode, Properties confProp) {
+                      String remoteSqlPluginPath, String pluginLoadMode, String deployMode, String logLevel, Properties confProp) {
         this.sql = sql;
         this.name = name;
         this.jarUrlList = jarUrlList;
         this.localSqlPluginPath = localSqlPluginPath;
         this.remoteSqlPluginPath = remoteSqlPluginPath;
         this.pluginLoadMode = pluginLoadMode;
         this.deployMode = deployMode;
+        this.logLevel = logLevel;
         this.confProp = confProp;
     }
 
@@ -84,6 +86,10 @@ public Properties getConfProp() {
         return confProp;
     }
 
+    public String getLogLevel() {
+        return logLevel;
+    }
+
     @Override
     public String toString() {
         return "ParamsInfo{" +
@@ -94,6 +100,7 @@ public String toString() {
                 ", remoteSqlPluginPath='" + remoteSqlPluginPath + '\'' +
                 ", pluginLoadMode='" + pluginLoadMode + '\'' +
                 ", deployMode='" + deployMode + '\'' +
+                ", logLevel='" + logLevel + '\'' +
                 ", confProp=" + confProp +
                 '}';
     }
@@ -114,6 +121,7 @@ public static class Builder {
         private String remoteSqlPluginPath;
         private String pluginLoadMode;
         private String deployMode;
+        private String logLevel;
         private Properties confProp;
 
 
@@ -152,14 +160,19 @@ public ParamsInfo.Builder setDeployMode(String deployMode) {
             return this;
         }
 
+        public ParamsInfo.Builder setLogLevel(String logLevel) {
+            this.logLevel = logLevel;
+            return this;
+        }
+
         public ParamsInfo.Builder setConfProp(Properties confProp) {
             this.confProp = confProp;
             return this;
         }
 
         public ParamsInfo build() {
             return new ParamsInfo(sql, name, jarUrlList, localSqlPluginPath,
-                    remoteSqlPluginPath, pluginLoadMode, deployMode, confProp);
+                    remoteSqlPluginPath, pluginLoadMode, deployMode, logLevel, confProp);
         }
     }
 }

core/src/main/java/com/dtstack/flink/sql/option/Options.java

@@ -69,7 +69,7 @@ public class Options {
     @OptionRequired(description = "plugin load mode, by classpath or shipfile")
     private String pluginLoadMode = EPluginLoadMode.CLASSPATH.name();
 
-    private String logLevel = "logLevel";
+    private String logLevel;
 
     public String getMode() {
         return mode;