Skip to content

Commit 2926476

Browse files
wushengyeyouyawushengyeyouya
committed
Merge remote-tracking branch 'origin/dev-0.2.3-log-collector' into dev-0.2.3-log-collector
2 parents c445fc1 + 154abdc commit 2926476

File tree

65 files changed

+1075
-473
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

65 files changed

+1075
-473
lines changed

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@
5353
<jdk.compile.version>1.8</jdk.compile.version>
5454
<maven.version>3.3.3</maven.version>
5555
<gson.version>2.8.5</gson.version>
56-
<fasterxml.jackson.version>2.11.3</fasterxml.jackson.version>
56+
<fasterxml.jackson.version>2.13.2</fasterxml.jackson.version>
5757
<math3.version>3.1.1</math3.version>
5858
<httpclient.version>4.5.4</httpclient.version>
5959
<httpmime.version>4.5.4</httpmime.version>

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/pom.xml renamed to streamis-jobmanager/streamis-job-log/job-log-collector/flink-streamis-log-collector/pom.xml

File renamed without changes.

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/main/assembly/package.xml renamed to streamis-jobmanager/streamis-job-log/job-log-collector/flink-streamis-log-collector/src/main/assembly/package.xml

File renamed without changes.

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

Lines changed: 16 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package com.webank.wedatasphere.streamis.jobmanager.log.collector.flink;
22

33
import com.webank.wedatasphere.streamis.jobmanager.log.collector.config.StreamisLogAppenderConfig;
4+
import com.webank.wedatasphere.streamis.jobmanager.log.collector.log4j2.StreamisLog4j2AppenderConfig;
45
import com.webank.wedatasphere.streamis.jobmanager.plugin.StreamisConfigAutowired;
56
import org.apache.commons.lang3.StringUtils;
67
import org.apache.flink.configuration.Configuration;
@@ -11,6 +12,7 @@
1112
import org.apache.logging.log4j.core.Filter;
1213
import org.apache.logging.log4j.core.filter.LevelMatchFilter;
1314
import org.apache.logging.log4j.core.filter.RegexFilter;
15+
import org.apache.logging.log4j.core.filter.ThresholdFilter;
1416

1517
import java.util.Enumeration;
1618
import java.util.List;
@@ -48,19 +50,24 @@ public StreamisLogAppenderConfig logAppenderConfig(StreamisLogAppenderConfig.Bui
4850
gateway += this.configuration.getString(LOG_COLLECT_PATH, "/");
4951
builder.setRpcAddress(gateway);
5052
}
51-
List<String> filterStrategies = this.configuration.get(LOG_FILTER_STRATEGIES);
52-
for(String filterStrategy : filterStrategies){
53-
if ("LevelMatch".equals(filterStrategy)){
54-
builder.withFilter(LevelMatchFilter.newBuilder().setOnMatch(Filter.Result.ACCEPT).setOnMismatch(Filter.Result.DENY)
55-
.setLevel(Level.getLevel(this.configuration.getString(LOG_FILTER_LEVEL_MATCH))).build());
56-
} else if ("RegexMatch".equals(filterStrategy)){
57-
builder.withFilter(RegexFilter.createFilter( this.configuration.getString(LOG_FILTER_REGEX),
58-
null, true, Filter.Result.ACCEPT, Filter.Result.DENY));
53+
if (builder instanceof StreamisLog4j2AppenderConfig.Builder) {
54+
List<String> filterStrategies = this.configuration.get(LOG_FILTER_STRATEGIES);
55+
for (String filterStrategy : filterStrategies) {
56+
if ("LevelMatch".equals(filterStrategy)) {
57+
((StreamisLog4j2AppenderConfig.Builder)builder).withFilter(LevelMatchFilter.newBuilder().setOnMatch(Filter.Result.ACCEPT).setOnMismatch(Filter.Result.DENY)
58+
.setLevel(Level.getLevel(this.configuration.getString(LOG_FILTER_LEVEL_MATCH))).build());
59+
} else if ("ThresholdFilter".equals(filterStrategy)) {
60+
((StreamisLog4j2AppenderConfig.Builder)builder).withFilter(ThresholdFilter.createFilter(Level
61+
.getLevel(this.configuration.getString(LOG_FILTER_THRESHOLD_MATCH)), Filter.Result.ACCEPT, Filter.Result.DENY));
62+
} else if ("RegexMatch".equals(filterStrategy)) {
63+
((StreamisLog4j2AppenderConfig.Builder)builder).withFilter(RegexFilter.createFilter(this.configuration.getString(LOG_FILTER_REGEX),
64+
null, true, Filter.Result.ACCEPT, Filter.Result.DENY));
65+
}
5966
}
6067
}
6168
String hadoopUser = EnvironmentInformation.getHadoopUser();
6269
if (hadoopUser.equals("<no hadoop dependency found>") || hadoopUser.equals("<unknown>")){
63-
hadoopUser = "";
70+
hadoopUser = System.getProperty("user.name");
6471
}
6572
return builder.setRpcConnTimeout(this.configuration.getInteger(LOG_RPC_CONN_TIMEOUT))
6673
.setRpcSocketTimeout(this.configuration.getInteger(LOG_RPC_SOCKET_TIMEOUT))

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,11 @@ public class FlinkStreamisConfigDefine {
112112
public static final ConfigOption<String> LOG_FILTER_LEVEL_MATCH = ConfigOptions.key("stream.log.filter.level-match.level")
113113
.stringType().defaultValue("ERROR").withDescription("Level value of LevelMatch filter strategy");
114114

115+
/**
116+
* Level value of ThresholdMatch filter strategy
117+
*/
118+
public static final ConfigOption<String> LOG_FILTER_THRESHOLD_MATCH = ConfigOptions.key("stream.log.filter.threshold.level")
119+
.stringType().defaultValue("ERROR").withDescription("Level value of ThresholdMatch filter strategy");
115120
/**
116121
* Regex value of RegexMatch filter strategy
117122
*/

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/main/resources/META-INF/services/com.webank.wedatasphere.streamis.jobmanager.plugin.StreamisConfigAutowired renamed to streamis-jobmanager/streamis-job-log/job-log-collector/flink-streamis-log-collector/src/main/resources/META-INF/services/com.webank.wedatasphere.streamis.jobmanager.plugin.StreamisConfigAutowired

File renamed without changes.

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/test/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/flink/FlinkConfigurationLoadTest.java renamed to streamis-jobmanager/streamis-job-log/job-log-collector/flink-streamis-log-collector/src/test/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/flink/FlinkConfigurationLoadTest.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@
44
import org.apache.flink.configuration.GlobalConfiguration;
55
import org.junit.Test;
66

7-
import java.io.File;
87
import java.util.Enumeration;
98
import java.util.Objects;
109
import java.util.Properties;

streamis-jobmanager/streamis-job-log/flink-streamis-log-collector/src/test/resources/flink-conf.yaml renamed to streamis-jobmanager/streamis-job-log/job-log-collector/flink-streamis-log-collector/src/test/resources/flink-conf.yaml

File renamed without changes.

‎streamis-jobmanager/streamis-job-log/job-log-collector/streamis-job-log-collector-core/pom.xml‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<parent>
6+
<artifactId>streamis-job-log</artifactId>
7+
<groupId>com.webank.wedatasphere.streamis</groupId>
8+
<version>0.2.0</version>
9+
</parent>
10+
<modelVersion>4.0.0</modelVersion>
11+
12+
<artifactId>streamis-job-log-collector-core</artifactId>
13+
14+
<properties>
15+
<maven.compiler.source>8</maven.compiler.source>
16+
<maven.compiler.target>8</maven.compiler.target>
17+
<httpclient.version>4.5.13</httpclient.version>
18+
<httpmine.version>4.5.4</httpmine.version>
19+
</properties>
20+
21+
<dependencies>
22+
<dependency>
23+
<groupId>com.webank.wedatasphere.streamis</groupId>
24+
<artifactId>streamis-job-log-common</artifactId>
25+
<version>${streamis.version}</version>
26+
</dependency>
27+
<!--http client module-->
28+
<dependency>
29+
<groupId>org.apache.httpcomponents</groupId>
30+
<artifactId>httpclient</artifactId>
31+
<version>${httpclient.version}</version>
32+
</dependency>
33+
<dependency>
34+
<groupId>org.apache.httpcomponents</groupId>
35+
<artifactId>httpmime</artifactId>
36+
<version>${httpmine.version}</version>
37+
</dependency>
38+
</dependencies>
39+
</project>

streamis-jobmanager/streamis-job-log/streamis-job-log-collector/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/ExceptionListener.java renamed to streamis-jobmanager/streamis-job-log/job-log-collector/streamis-job-log-collector-core/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/ExceptionListener.java

File renamed without changes.

0 commit comments

Comments
 (0)