Add new strategy named ThresholdFilter

Author: Davidhua1996

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/flink/FlinkStreamisConfigAutowired.java

@@ -11,6 +11,7 @@
 import org.apache.logging.log4j.core.Filter;
 import org.apache.logging.log4j.core.filter.LevelMatchFilter;
 import org.apache.logging.log4j.core.filter.RegexFilter;
+import org.apache.logging.log4j.core.filter.ThresholdFilter;
 
 import java.util.Enumeration;
 import java.util.List;
@@ -53,7 +54,10 @@ public StreamisLogAppenderConfig logAppenderConfig(StreamisLogAppenderConfig.Bui
             if ("LevelMatch".equals(filterStrategy)){
                 builder.withFilter(LevelMatchFilter.newBuilder().setOnMatch(Filter.Result.ACCEPT).setOnMismatch(Filter.Result.DENY)
                         .setLevel(Level.getLevel(this.configuration.getString(LOG_FILTER_LEVEL_MATCH))).build());
-            } else if ("RegexMatch".equals(filterStrategy)){
+            }else if ("ThresholdFilter".equals(filterStrategy)){
+                builder.withFilter(ThresholdFilter.createFilter(Level
+                        .getLevel(this.configuration.getString(LOG_FILTER_THRESHOLD_MATCH)), Filter.Result.ACCEPT, Filter.Result.DENY));
+            }else if ("RegexMatch".equals(filterStrategy)){
                 builder.withFilter(RegexFilter.createFilter( this.configuration.getString(LOG_FILTER_REGEX),
                         null, true, Filter.Result.ACCEPT, Filter.Result.DENY));
             }

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/flink/FlinkStreamisConfigDefine.java

@@ -112,6 +112,11 @@ public class FlinkStreamisConfigDefine {
     public static final ConfigOption<String> LOG_FILTER_LEVEL_MATCH = ConfigOptions.key("stream.log.filter.level-match.level")
             .stringType().defaultValue("ERROR").withDescription("Level value of LevelMatch filter strategy");
 
+    /**
+     * Level value of ThresholdMatch filter strategy
+     */
+    public static final ConfigOption<String> LOG_FILTER_THRESHOLD_MATCH = ConfigOptions.key("stream.log.filter.threshold.level")
+            .stringType().defaultValue("ERROR").withDescription("Level value of ThresholdMatch filter strategy");
     /**
      * Regex value of RegexMatch filter strategy
      */