Skip to content

Commit 466938a

Browse files
Davidhua1996Davidhua1996
authored
Optimization of log collector module (#60)
* Add keyword filter strategy in log4j/log4j2. * Use the stream log event timestamp as the enqueue time. * Init the structure of supporting multiple disks/drivers. * Load balancer strategies for storing stream log. * Adjust the default value of configuration. * Add test unit for job log storage. * optimize the restful performance. * Add debug mode in log collector; Add the strategy to discard log cache. * Fix the problem in closing log4j bucket. * Optimize the strategies.
1 parent 64c9627 commit 466938a

File tree

47 files changed

+2103
-151
lines changed

Some content is hidden

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

47 files changed

+2103
-151
lines changed

‎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: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import com.webank.wedatasphere.streamis.jobmanager.log.collector.config.StreamisLogAppenderConfig;
44
import com.webank.wedatasphere.streamis.jobmanager.log.collector.log4j2.StreamisLog4j2AppenderConfig;
5+
import com.webank.wedatasphere.streamis.jobmanager.log.collector.log4j2.filters.KeywordThresholdFilter;
56
import com.webank.wedatasphere.streamis.jobmanager.plugin.StreamisConfigAutowired;
67
import org.apache.commons.lang3.StringUtils;
78
import org.apache.flink.configuration.Configuration;
@@ -56,20 +57,26 @@ public StreamisLogAppenderConfig logAppenderConfig(StreamisLogAppenderConfig.Bui
5657
if ("LevelMatch".equals(filterStrategy)) {
5758
((StreamisLog4j2AppenderConfig.Builder)builder).withFilter(LevelMatchFilter.newBuilder().setOnMatch(Filter.Result.ACCEPT).setOnMismatch(Filter.Result.DENY)
5859
.setLevel(Level.getLevel(this.configuration.getString(LOG_FILTER_LEVEL_MATCH))).build());
59-
} else if ("ThresholdFilter".equals(filterStrategy)) {
60+
} else if ("ThresholdMatch".equals(filterStrategy)) {
6061
((StreamisLog4j2AppenderConfig.Builder)builder).withFilter(ThresholdFilter.createFilter(Level
6162
.getLevel(this.configuration.getString(LOG_FILTER_THRESHOLD_MATCH)), Filter.Result.ACCEPT, Filter.Result.DENY));
6263
} else if ("RegexMatch".equals(filterStrategy)) {
6364
((StreamisLog4j2AppenderConfig.Builder)builder).withFilter(RegexFilter.createFilter(this.configuration.getString(LOG_FILTER_REGEX),
6465
null, true, Filter.Result.ACCEPT, Filter.Result.DENY));
66+
} else if ("Keyword".equals(filterStrategy)){
67+
((StreamisLog4j2AppenderConfig.Builder)builder).withFilter(
68+
new KeywordThresholdFilter(
69+
StringUtils.split(this.configuration.getString(LOG_FILTER_KEYWORDS), ","),
70+
StringUtils.split(this.configuration.getString(LOG_FILTER_KEYWORDS_EXCLUDE), ",")));
6571
}
6672
}
6773
}
6874
String hadoopUser = EnvironmentInformation.getHadoopUser();
6975
if (hadoopUser.equals("<no hadoop dependency found>") || hadoopUser.equals("<unknown>")){
7076
hadoopUser = System.getProperty("user.name");
7177
}
72-
return builder.setRpcConnTimeout(this.configuration.getInteger(LOG_RPC_CONN_TIMEOUT))
78+
return builder.setDebugMode(this.configuration.getBoolean(DEBUG_MODE))
79+
.setRpcConnTimeout(this.configuration.getInteger(LOG_RPC_CONN_TIMEOUT))
7380
.setRpcSocketTimeout(this.configuration.getInteger(LOG_RPC_SOCKET_TIMEOUT))
7481
.setRpcSendRetryCnt(this.configuration.getInteger(LOG_RPC_SEND_RETRY_COUNT))
7582
.setRpcServerRecoveryTimeInSec(this.configuration.getInteger(LOG_RPC_SERVER_RECOVERY_TIME))
@@ -81,10 +88,13 @@ public StreamisLogAppenderConfig logAppenderConfig(StreamisLogAppenderConfig.Bui
8188
hadoopUser))
8289
.setRpcCacheSize(this.configuration.getInteger(LOG_RPC_CACHE_SIZE))
8390
.setRpcCacheMaxConsumeThread(this.configuration.getInteger(LOG_PRC_CACHE_MAX_CONSUME_THREAD))
91+
.setDiscard(this.configuration.getBoolean(LOG_RPC_CACHE_DISCARD))
92+
.setDiscardWindow(this.configuration.getInteger(LOG_RPC_CACHE_DISCARD_WINDOW))
8493
.setRpcBufferSize(this.configuration.getInteger(LOG_RPC_BUFFER_SIZE))
8594
.setRpcBufferExpireTimeInSec(this.configuration.getInteger(LOG_RPC_BUFFER_EXPIRE_TIME)).build();
8695
}
8796

97+
8898
/**
8999
* According to :
90100
* String launchCommand =

‎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: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import org.apache.flink.configuration.ConfigOption;
44
import org.apache.flink.configuration.ConfigOptions;
5+
import scala.Int;
56

67
import java.util.List;
78

@@ -86,8 +87,19 @@ public class FlinkStreamisConfigDefine {
8687
* Max cache consume threads in log RPC module
8788
*/
8889
public static final ConfigOption<Integer> LOG_PRC_CACHE_MAX_CONSUME_THREAD = ConfigOptions.key("stream.log.rpc.cache.max-consume-thread")
89-
.intType().defaultValue(10).withDescription("Max cache consume threads in log RPC module");
90+
.intType().defaultValue(2).withDescription("Max cache consume threads in log RPC module");
9091

92+
/**
93+
* If discard the useless log
94+
*/
95+
public static final ConfigOption<Boolean> LOG_RPC_CACHE_DISCARD = ConfigOptions.key("stream.log.rpc.cache.discard")
96+
.booleanType().defaultValue(true).withDescription("If discard the useless log");
97+
98+
/**
99+
* The window size of discarding
100+
*/
101+
public static final ConfigOption<Integer> LOG_RPC_CACHE_DISCARD_WINDOW = ConfigOptions.key("stream.log.rpc.cache.discard-window")
102+
.intType().defaultValue(2).withDescription("The window size of discarding");
91103
/**
92104
* Buffer size in log RPC module
93105
*/
@@ -104,7 +116,7 @@ public class FlinkStreamisConfigDefine {
104116
* Log filter strategy list
105117
*/
106118
public static final ConfigOption<List<String>> LOG_FILTER_STRATEGIES = ConfigOptions.key("stream.log.filter.strategies")
107-
.stringType().asList().defaultValues("LevelMatch").withDescription("Log filter strategy list");
119+
.stringType().asList().defaultValues("Keyword").withDescription("Log filter strategy list");
108120

109121
/**
110122
* Level value of LevelMatch filter strategy
@@ -122,4 +134,22 @@ public class FlinkStreamisConfigDefine {
122134
*/
123135
public static final ConfigOption<String> LOG_FILTER_REGEX = ConfigOptions.key("stream.log.filter.regex.value")
124136
.stringType().defaultValue(".*").withDescription("Regex value of RegexMatch filter strategy");
137+
138+
/**
139+
* Accept keywords of Keyword filter strategy
140+
*/
141+
public static final ConfigOption<String> LOG_FILTER_KEYWORDS = ConfigOptions.key("stream.log.filter.keywords")
142+
.stringType().defaultValue("ERROR").withDescription("Accept keywords of Keyword filter strategy");
143+
144+
/**
145+
* Exclude keywords of Keyword filter strategy
146+
*/
147+
public static final ConfigOption<String> LOG_FILTER_KEYWORDS_EXCLUDE = ConfigOptions.key("stream.log.filter.keywords.exclude")
148+
.stringType().defaultValue("").withDescription("Exclude keywords of Keyword filter strategy");
149+
150+
/**
151+
* Debug mode
152+
*/
153+
public static final ConfigOption<Boolean> DEBUG_MODE = ConfigOptions.key("stream.log.debug")
154+
.booleanType().defaultValue(false).withDescription("Debug mode");
125155
}

‎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: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,15 @@
33
import org.apache.flink.configuration.Configuration;
44
import org.apache.flink.configuration.GlobalConfiguration;
55
import org.junit.Test;
6+
import org.slf4j.Logger;
7+
import org.slf4j.LoggerFactory;
68

79
import java.util.Enumeration;
810
import java.util.Objects;
911
import java.util.Properties;
1012

1113
public class FlinkConfigurationLoadTest {
14+
private static final Logger LOG = LoggerFactory.getLogger(FlinkConfigurationLoadTest.class);
1215
@Test
1316
public void loadConfiguration() {
1417
String configDir = Objects.requireNonNull(FlinkConfigurationLoadTest.class.getResource("/")).getFile();
@@ -21,4 +24,5 @@ public void loadConfiguration() {
2124
}
2225
GlobalConfiguration.loadConfiguration(configDir, dynamicConfiguration);
2326
}
27+
2428
}

‎streamis-jobmanager/streamis-job-log/job-log-collector/flink-streamis-log-collector/src/test/resources/META-INF/services/com.webank.wedatasphere.streamis.jobmanager.plugin.StreamisConfigAutowired‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
com.webank.wedatasphere.streamis.jobmanager.log.collector.flink.FlinkStreamisConfigAutowired

‎streamis-jobmanager/streamis-job-log/job-log-collector/flink-streamis-log-collector/src/test/resources/log4j2.xml‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
~ Copyright 2019 WeBank
4+
~
5+
~ Licensed under the Apache License, Version 2.0 (the "License");
6+
~ you may not use this file except in compliance with the License.
7+
~ You may obtain a copy of the License at
8+
~
9+
~ http://www.apache.org/licenses/LICENSE-2.0
10+
~
11+
~ Unless required by applicable law or agreed to in writing, software
12+
~ distributed under the License is distributed on an "AS IS" BASIS,
13+
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
~ See the License for the specific language governing permissions and
15+
~ limitations under the License.
16+
~
17+
-->
18+
19+
<configuration status="error" monitorInterval="30">
20+
<appenders>
21+
<StreamRpcLog name="StreamRpcLog" appName="stream_application">
22+
<PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n"/>
23+
<RpcLogSender sendRetryCnt="3" connectionTimeout="3000"
24+
socketTimeout="20000" serverRecoveryTimeInSec="5" maxDelayTimeInSec="60">
25+
<AuthConfig tokenCodeKey="" tokenCode="" tokenUser="" tokenUserKey=""/>
26+
<SendLogCache size="300" maxConsumeThread="10"/>
27+
<SendBuffer size="50" expireTimeInSec="2"/>
28+
</RpcLogSender>
29+
</StreamRpcLog>`
30+
</appenders>
31+
<loggers>
32+
<root level="INFO" additivity="true">
33+
<!-- <appender-ref ref="Console"/>-->
34+
<appender-ref ref="StreamRpcLog"/>
35+
</root>
36+
</loggers>
37+
</configuration>

‎streamis-jobmanager/streamis-job-log/job-log-collector/streamis-job-log-collector-core/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/config/RpcLogSenderConfig.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,10 @@ public class RpcLogSenderConfig {
3838
*/
3939
private int maxDelayTimeInSec = 60;
4040

41+
/**
42+
* If open debug mode
43+
*/
44+
private boolean debugMode = false;
4145
/**
4246
* Auth config
4347
*/
@@ -149,6 +153,14 @@ public void setServerRecoveryTimeInSec(int serverRecoveryTimeInSec) {
149153
this.serverRecoveryTimeInSec = serverRecoveryTimeInSec;
150154
}
151155

156+
public boolean isDebugMode() {
157+
return debugMode;
158+
}
159+
160+
public void setDebugMode(boolean debugMode) {
161+
this.debugMode = debugMode;
162+
}
163+
152164
@Override
153165
public String toString() {
154166
return "RpcLogSenderConfig{" +
@@ -161,6 +173,8 @@ public String toString() {
161173
", authConfig=" + authConfig +
162174
", cacheConfig=" + cacheConfig +
163175
", bufferConfig=" + bufferConfig +
176+
", debug=" + debugMode +
164177
'}';
165178
}
179+
166180
}

‎streamis-jobmanager/streamis-job-log/job-log-collector/streamis-job-log-collector-core/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/config/SendLogCacheConfig.java‎

Lines changed: 32 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,17 @@ public class SendLogCacheConfig {
1313
/**
1414
* Max number of consuming thread
1515
*/
16-
private int maxConsumeThread = 10;
16+
private int maxConsumeThread = 2;
17+
18+
/**
19+
* The switch to discard log
20+
*/
21+
private boolean discard = true;
22+
23+
/**
24+
* Discard window in second
25+
*/
26+
private int discardWindow = 2;
1727

1828
public SendLogCacheConfig(){
1929

@@ -40,11 +50,32 @@ public void setMaxConsumeThread(int maxConsumeThread) {
4050
this.maxConsumeThread = maxConsumeThread;
4151
}
4252

53+
public boolean isDiscard() {
54+
return discard;
55+
}
56+
57+
public void setDiscard(boolean discard) {
58+
this.discard = discard;
59+
}
60+
61+
public int getDiscardWindow() {
62+
return discardWindow;
63+
}
64+
65+
public void setDiscardWindow(int discardWindow) {
66+
this.discardWindow = discardWindow;
67+
}
68+
4369
@Override
4470
public String toString() {
4571
return "SendLogCacheConfig{" +
4672
"size=" + size +
4773
", maxConsumeThread=" + maxConsumeThread +
74+
", discard=" + discard +
75+
", discardWindow=" + discardWindow +
4876
'}';
4977
}
78+
79+
80+
5081
}

‎streamis-jobmanager/streamis-job-log/job-log-collector/streamis-job-log-collector-core/src/main/java/com/webank/wedatasphere/streamis/jobmanager/log/collector/config/StreamisLogAppenderConfig.java‎

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

3+
import com.webank.wedatasphere.streamis.jobmanager.log.collector.message.filters.LogMessageFilter;
4+
35
import java.util.ArrayList;
6+
import java.util.List;
47
import java.util.Objects;
58
import java.util.Optional;
69

@@ -14,9 +17,15 @@ public class StreamisLogAppenderConfig {
1417

1518
protected final RpcLogSenderConfig senderConfig;
1619

17-
protected StreamisLogAppenderConfig(String applicationName, RpcLogSenderConfig rpcLogSenderConfig){
20+
/**
21+
* Message filters
22+
*/
23+
protected final List<LogMessageFilter> messageFilters;
24+
protected StreamisLogAppenderConfig(String applicationName, RpcLogSenderConfig rpcLogSenderConfig,
25+
List<LogMessageFilter> messageFilters){
1826
this.applicationName = applicationName;
1927
this.senderConfig = null != rpcLogSenderConfig? rpcLogSenderConfig : new RpcLogSenderConfig();
28+
this.messageFilters = messageFilters;
2029
}
2130

2231
public static class Builder{
@@ -30,6 +39,11 @@ public static class Builder{
3039
*/
3140
protected final RpcLogSenderConfig rpcLogSenderConfig;
3241

42+
/**
43+
* Message filters
44+
*/
45+
protected final List<LogMessageFilter> messageFilters = new ArrayList<>();
46+
3347
public Builder(String applicationName,
3448
RpcLogSenderConfig rpcLogSenderConfig){
3549
this.applicationName = applicationName;
@@ -189,8 +203,47 @@ public StreamisLogAppenderConfig.Builder setRpcBufferExpireTimeInSec(int expireT
189203
return this;
190204
}
191205

206+
/**
207+
* Add log message filter
208+
* @param messageFilter message filter
209+
* @return builder
210+
*/
211+
public StreamisLogAppenderConfig.Builder withMessageFilter(LogMessageFilter messageFilter){
212+
this.messageFilters.add(messageFilter);
213+
return this;
214+
}
215+
216+
/**
217+
* Set to discard the useless log
218+
* @param discard discard
219+
* @return builder
220+
*/
221+
public StreamisLogAppenderConfig.Builder setDiscard(boolean discard){
222+
this.rpcLogSenderConfig.getCacheConfig().setDiscard(discard);
223+
return this;
224+
}
225+
226+
/**
227+
* Set the window size of discarding
228+
* @param windowSize
229+
* @return
230+
*/
231+
public StreamisLogAppenderConfig.Builder setDiscardWindow(int windowSize){
232+
this.rpcLogSenderConfig.getCacheConfig().setDiscardWindow(windowSize);
233+
return this;
234+
}
235+
/**
236+
* Switch to debug
237+
* @param debugMode debug mode
238+
* @return builder
239+
*/
240+
public StreamisLogAppenderConfig.Builder setDebugMode(boolean debugMode){
241+
this.rpcLogSenderConfig.setDebugMode(debugMode);
242+
return this;
243+
}
244+
192245
public StreamisLogAppenderConfig build(){
193-
return new StreamisLogAppenderConfig(applicationName, rpcLogSenderConfig);
246+
return new StreamisLogAppenderConfig(applicationName, rpcLogSenderConfig, messageFilters);
194247
}
195248
}
196249
public String getApplicationName() {
@@ -202,4 +255,7 @@ public RpcLogSenderConfig getSenderConfig() {
202255
return senderConfig;
203256
}
204257

258+
public List<LogMessageFilter> getMessageFilters() {
259+
return messageFilters;
260+
}
205261
}

0 commit comments

Comments
 (0)