Skip to content

Commit 64c4f42

Browse files
luoluoyuyuluoluoyuyu
committed
Optimize pipe logging with shared PipePeriodicalLogReducer
Move PipePeriodicalLogReducer to node-commons and route ConfigNode and DataNode pipe logs through PipeLogger to rate-limit repetitive messages.
1 parent a437cd4 commit 64c4f42

19 files changed

Lines changed: 248 additions & 164 deletions

File tree

‎iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,8 @@ public final class ManagerMessages {
6767
"Detected completion of pipe {}, static meta: {}, remove it.";
6868
public static final String DETECT_PIPERUNTIMECRITICALEXCEPTION_FROM_AGENT_STOP_PIPE =
6969
"Detect PipeRuntimeCriticalException {} from agent, stop pipe {}.";
70+
public static final String DETECT_PIPERUNTIMESINKCRITICALEXCEPTION_FROM_AGENT_STOP_PIPE =
71+
"Detect PipeRuntimeSinkCriticalException {} from agent, stop pipe {}.";
7072
public static final String ENABLE_SEPARATION_OF_POWERS_IS_NOT_SUPPORTED =
7173
"Enable separation of powers is not supported";
7274
public static final String ENDEXECUTECQ_TIME_RANGE_IS_CURRENT_TIME_IS =

‎iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,8 @@ public final class ManagerMessages {
6767
"Detected completion of pipe {}, static meta: {}, remove it.";
6868
public static final String DETECT_PIPERUNTIMECRITICALEXCEPTION_FROM_AGENT_STOP_PIPE =
6969
"Detect PipeRuntimeCriticalException {} from agent, stop pipe {}.";
70+
public static final String DETECT_PIPERUNTIMESINKCRITICALEXCEPTION_FROM_AGENT_STOP_PIPE =
71+
"检测到 agent 上报 PipeRuntimeSinkCriticalException {}，停止 pipe {}。";
7072
public static final String ENABLE_SEPARATION_OF_POWERS_IS_NOT_SUPPORTED = "不支持启用权力分离";
7173
public static final String ENDEXECUTECQ_TIME_RANGE_IS_CURRENT_TIME_IS =
7274
"[EndExecuteCQ] {}, time range is [{}, {}), current time is {}";

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@
2626
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
2727
import org.apache.iotdb.commons.pipe.config.PipeConfig;
2828
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
29+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
30+
import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
2931
import org.apache.iotdb.commons.service.IService;
3032
import org.apache.iotdb.commons.service.ServiceType;
3133
import org.apache.iotdb.confignode.i18n.ManagerMessages;
@@ -56,6 +58,7 @@ public class PipeConfigNodeRuntimeAgent implements IService {
5658
@Override
5759
public synchronized void start() {
5860
PipeConfig.getInstance().printAllConfigs();
61+
PipeLogger.setLogger(PipePeriodicalLogReducer::log);
5962

6063
// PipeTasks will not be started here and will be started by "HandleLeaderChange"
6164
// procedure when the consensus layer notify leader ready
@@ -142,19 +145,21 @@ public void report(final EnrichedEvent event, final PipeRuntimeException pipeRun
142145
if (event.getPipeTaskMeta() != null) {
143146
report(event.getPipeTaskMeta(), pipeRuntimeException);
144147
} else {
145-
LOGGER.warn(
146-
ManagerMessages.ATTEMPT_TO_REPORT_PIPE_EXCEPTION_TO_A_NULL_PIPETASKMETA,
147-
pipeRuntimeException);
148+
PipeLogger.log(
149+
LOGGER::warn,
150+
pipeRuntimeException,
151+
ManagerMessages.ATTEMPT_TO_REPORT_PIPE_EXCEPTION_TO_A_NULL_PIPETASKMETA);
148152
}
149153
}
150154

151155
private void report(
152156
final PipeTaskMeta pipeTaskMeta, final PipeRuntimeException pipeRuntimeException) {
153-
LOGGER.warn(
157+
PipeLogger.log(
158+
LOGGER::warn,
159+
pipeRuntimeException,
154160
ManagerMessages.REPORT_PIPERUNTIMEEXCEPTION_TO_LOCAL_PIPETASKMETA_EXCEPTION_MESSAGE,
155161
pipeTaskMeta,
156-
pipeRuntimeException.getMessage(),
157-
pipeRuntimeException);
162+
pipeRuntimeException.getMessage());
158163

159164
pipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
160165

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java‎

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
3333
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
3434
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
35+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
3536
import org.apache.iotdb.confignode.i18n.ManagerMessages;
3637
import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
3738
import org.apache.iotdb.confignode.manager.pipe.metric.sink.PipeConfigRegionSinkMetrics;
@@ -105,10 +106,11 @@ private void initSource(final Map<String, String> sourceAttributes) throws Excep
105106
try {
106107
source.close();
107108
} catch (Exception closeException) {
108-
LOGGER.warn(
109+
PipeLogger.log(
110+
LOGGER::warn,
111+
closeException,
109112
ManagerMessages.FAILED_TO_CLOSE_EXTRACTOR_AFTER_FAILED_TO_INITIALIZE_EXTRACTOR
110-
+ "Ignore this exception.",
111-
closeException);
113+
+ "Ignore this exception.");
112114
}
113115
throw e;
114116
}
@@ -154,9 +156,11 @@ private void initSink(final Map<String, String> sinkAttributes) throws Exception
154156
try {
155157
outputPipeSink.close();
156158
} catch (final Exception closeException) {
157-
LOGGER.warn(
159+
PipeLogger.log(
160+
LOGGER::warn,
161+
closeException,
158162
ManagerMessages.FAILED_TO_CLOSE_SINK_AFTER_FAILED_TO_INITIALIZE_IT_IGNORE,
159-
closeException);
163+
closeException.getMessage());
160164
}
161165
throw e;
162166
}
@@ -208,19 +212,19 @@ public void close() {
208212
try {
209213
source.close();
210214
} catch (final Exception e) {
211-
LOGGER.info(ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPEEXTRACTOR, e);
215+
PipeLogger.log(LOGGER::info, e, ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPEEXTRACTOR);
212216
}
213217

214218
try {
215219
processor.close();
216220
} catch (final Exception e) {
217-
LOGGER.info(ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPEPROCESSOR, e);
221+
PipeLogger.log(LOGGER::info, e, ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPEPROCESSOR);
218222
}
219223

220224
try {
221225
outputPipeSink.close();
222226
} catch (final Exception e) {
223-
LOGGER.info(ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPECONNECTOR, e);
227+
PipeLogger.log(LOGGER::info, e, ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPECONNECTOR);
224228
} finally {
225229
// Should be after connector.close()
226230
super.close();

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.commons.concurrent.ThreadName;
2525
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
2626
import org.apache.iotdb.commons.pipe.config.PipeConfig;
27+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
2728
import org.apache.iotdb.confignode.i18n.ManagerMessages;
2829
import org.apache.iotdb.confignode.manager.ConfigManager;
2930
import org.apache.iotdb.confignode.manager.ProcedureManager;
@@ -95,7 +96,8 @@ private synchronized void sync() {
9596
isLastPipeSyncSuccessful = false;
9697

9798
if (configManager.getPipeManager().getPipeTaskCoordinator().isLocked()) {
98-
LOGGER.warn(
99+
PipeLogger.log(
100+
LOGGER::warn,
99101
ManagerMessages.PIPETASKCOORDINATORLOCK_IS_HELD_BY_ANOTHER_THREAD_SKIP_THIS_ROUND_OF_2);
100102
return;
101103
}
@@ -109,7 +111,9 @@ private synchronized void sync() {
109111
== PipeConfig.getInstance().getPipeMetaSyncerAutoRestartPipeCheckIntervalRound()) {
110112
somePipesNeedRestarting = autoRestartWithLock();
111113
if (somePipesNeedRestarting) {
112-
LOGGER.info(ManagerMessages.SOME_PIPES_NEED_RESTARTING_WILL_RESTART_THEM_AFTER_THIS_SYNC);
114+
PipeLogger.log(
115+
LOGGER::info,
116+
ManagerMessages.SOME_PIPES_NEED_RESTARTING_WILL_RESTART_THEM_AFTER_THIS_SYNC);
113117
}
114118
pipeAutoRestartRoundCounter.set(0);
115119
}
@@ -130,19 +134,22 @@ private synchronized void sync() {
130134
if (handleMetaChangeStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
131135
successfulSync = true;
132136
} else {
133-
LOGGER.warn(
137+
PipeLogger.log(
138+
LOGGER::warn,
134139
ManagerMessages.FAILED_TO_HANDLE_PIPE_META_CHANGE_RESULT_STATUS,
135140
handleMetaChangeStatus);
136141
}
137142
}
138143

139144
if (successfulSync) {
140-
LOGGER.info(
145+
PipeLogger.log(
146+
LOGGER::info,
141147
ManagerMessages.AFTER_THIS_SUCCESSFUL_SYNC_IF_PIPETASKINFO_IS_EMPTY_DURING_THIS);
142148
isLastPipeSyncSuccessful = true;
143149
}
144150
} else {
145-
LOGGER.warn(ManagerMessages.FAILED_TO_SYNC_PIPE_META_RESULT_STATUS, metaSyncStatus);
151+
PipeLogger.log(
152+
LOGGER::warn, ManagerMessages.FAILED_TO_SYNC_PIPE_META_RESULT_STATUS, metaSyncStatus);
146153
}
147154
}
148155

@@ -158,7 +165,8 @@ private boolean autoRestartWithLock() {
158165
final AtomicReference<PipeTaskInfo> pipeTaskInfo =
159166
configManager.getPipeManager().getPipeTaskCoordinator().tryLock();
160167
if (pipeTaskInfo == null) {
161-
LOGGER.warn(ManagerMessages.FAILED_TO_ACQUIRE_PIPE_LOCK_FOR_AUTO_RESTART_PIPE_TASK);
168+
PipeLogger.log(
169+
LOGGER::warn, ManagerMessages.FAILED_TO_ACQUIRE_PIPE_LOCK_FOR_AUTO_RESTART_PIPE_TASK);
162170
return false;
163171
}
164172
try {
@@ -176,15 +184,17 @@ private void checkAndRepairConsensusPipes() {
176184
.getRegionMaintainHandler()
177185
.checkAndRepairConsensusPipes();
178186
} catch (Exception e) {
179-
LOGGER.warn(ManagerMessages.FAILED_TO_CHECK_AND_REPAIR_CONSENSUS_PIPES, e);
187+
PipeLogger.log(LOGGER::warn, e, ManagerMessages.FAILED_TO_CHECK_AND_REPAIR_CONSENSUS_PIPES);
180188
}
181189
}
182190

183191
private boolean handleSuccessfulRestartWithLock() {
184192
final AtomicReference<PipeTaskInfo> pipeTaskInfo =
185193
configManager.getPipeManager().getPipeTaskCoordinator().tryLock();
186194
if (pipeTaskInfo == null) {
187-
LOGGER.warn(ManagerMessages.FAILED_TO_ACQUIRE_PIPE_LOCK_FOR_HANDLING_SUCCESSFUL_RESTART);
195+
PipeLogger.log(
196+
LOGGER::warn,
197+
ManagerMessages.FAILED_TO_ACQUIRE_PIPE_LOCK_FOR_HANDLING_SUCCESSFUL_RESTART);
188198
return false;
189199
}
190200
try {

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java‎

Lines changed: 25 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
3131
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInCoordinator;
3232
import org.apache.iotdb.commons.pipe.config.PipeConfig;
33+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
3334
import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
3435
import org.apache.iotdb.confignode.i18n.ManagerMessages;
3536
import org.apache.iotdb.confignode.manager.ConfigManager;
@@ -93,7 +94,8 @@ synchronized void parseHeartbeat(final int nodeId, final PipeHeartbeat pipeHeart
9394
final AtomicReference<PipeTaskInfo> pipeTaskInfo =
9495
configManager.getPipeManager().getPipeTaskCoordinator().tryLock();
9596
if (pipeTaskInfo == null) {
96-
LOGGER.warn(
97+
PipeLogger.log(
98+
LOGGER::warn,
9799
ManagerMessages.FAILED_TO_ACQUIRE_LOCK_WHEN_PARSEHEARTBEAT_FROM_NODE_ID,
98100
nodeId);
99101
return;
@@ -127,8 +129,10 @@ private int getExpectedHeartbeatNodeCount() {
127129
configManager.getNodeManager().getRegisteredDataNodeCount()
128130
+ (PipeConfig.getInstance().isSeperatedPipeHeartbeatEnabled() ? 1 : 0);
129131
if (expectedNodeCount <= 0) {
130-
LOGGER.warn(
131-
ManagerMessages.EXPECTED_PIPE_HEARTBEAT_NODE_COUNT_IS_FALLBACK_TO_1, expectedNodeCount);
132+
PipeLogger.log(
133+
LOGGER::warn,
134+
ManagerMessages.EXPECTED_PIPE_HEARTBEAT_NODE_COUNT_IS_FALLBACK_TO_1,
135+
expectedNodeCount);
132136
return 1;
133137
}
134138
return expectedNodeCount;
@@ -142,7 +146,8 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
142146
final PipeStaticMeta staticMeta = pipeMetaFromCoordinator.getStaticMeta();
143147
final PipeMeta pipeMetaFromAgent = pipeHeartbeat.getPipeMeta(staticMeta);
144148
if (pipeMetaFromAgent == null) {
145-
LOGGER.info(
149+
PipeLogger.log(
150+
LOGGER::info,
146151
ManagerMessages.PIPERUNTIMECOORDINATOR_MEETS_ERROR_IN_UPDATING_PIPEMETAKEEPER
147152
+ "pipeMetaFromAgent is null, pipeMetaFromCoordinator: {}",
148153
pipeMetaFromCoordinator);
@@ -157,7 +162,8 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
157162
if (Boolean.TRUE.equals(isPipeCompletedFromAgent)) {
158163

159164
temporaryMeta.markDataNodeCompleted(nodeId);
160-
LOGGER.info(
165+
PipeLogger.log(
166+
LOGGER::info,
161167
"Detected historical pipe completion report from DataNode {} for pipe {}. remainingEventCount: {}, remainingTime: {}, completedDataNodes: {}",
162168
nodeId,
163169
staticMeta.getPipeName(),
@@ -169,14 +175,16 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
169175
configManager.getNodeManager().getRegisteredDataNodeLocations().keySet();
170176
uncompletedDataNodeIds.removeAll(temporaryMeta.getCompletedDataNodeIds());
171177
if (uncompletedDataNodeIds.isEmpty()) {
172-
LOGGER.info(
178+
PipeLogger.log(
179+
LOGGER::info,
173180
"All DataNodes reported historical pipe {} completed. globalRemainingEventCount: {}, globalRemainingTime: {}, staticMeta: {}",
174181
staticMeta.getPipeName(),
175182
temporaryMeta.getGlobalRemainingEvents(),
176183
temporaryMeta.getGlobalRemainingTime(),
177184
staticMeta);
178185
pipeTaskInfo.get().removePipeMeta(staticMeta.getPipeName());
179-
LOGGER.info(
186+
PipeLogger.log(
187+
LOGGER::info,
180188
ManagerMessages.DETECTED_COMPLETION_OF_PIPE_STATIC_META_REMOVE_IT,
181189
staticMeta.getPipeName(),
182190
staticMeta);
@@ -267,7 +275,9 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
267275
needWriteConsensusOnConfigNodes.set(true);
268276
needPushPipeMetaToDataNodes.set(false);
269277

270-
LOGGER.warn(
278+
PipeLogger.log(
279+
LOGGER::warn,
280+
exception,
271281
ManagerMessages.DETECT_PIPERUNTIMECRITICALEXCEPTION_FROM_AGENT_STOP_PIPE,
272282
exception,
273283
pipeName);
@@ -296,11 +306,13 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
296306
needWriteConsensusOnConfigNodes.set(true);
297307
needPushPipeMetaToDataNodes.set(false);
298308

299-
LOGGER.warn(
300-
String.format(
301-
"Detect PipeRuntimeConnectorCriticalException %s "
302-
+ "from agent, stop pipe %s.",
303-
exception, pipeName));
309+
PipeLogger.log(
310+
LOGGER::warn,
311+
exception,
312+
ManagerMessages
313+
.DETECT_PIPERUNTIMESINKCRITICALEXCEPTION_FROM_AGENT_STOP_PIPE,
314+
exception,
315+
pipeName);
304316
});
305317
}
306318
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.iotdb.commons.concurrent.ThreadName;
2626
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
2727
import org.apache.iotdb.commons.pipe.config.PipeConfig;
28+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
2829
import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
2930
import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager;
3031
import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext;
@@ -85,7 +86,8 @@ private synchronized void heartbeat() {
8586
}
8687

8788
if (configManager.getPipeManager().getPipeTaskCoordinator().isLocked()) {
88-
LOGGER.warn(
89+
PipeLogger.log(
90+
LOGGER::warn,
8991
ManagerMessages.PIPETASKCOORDINATORLOCK_IS_HELD_BY_ANOTHER_THREAD_SKIP_THIS_ROUND_OF);
9092
return;
9193
}
@@ -130,7 +132,8 @@ private synchronized void heartbeat() {
130132
configNodeResp.getPipeRemainingEventCountList(),
131133
configNodeResp.getPipeRemainingTimeList()));
132134
} catch (final Exception e) {
133-
LOGGER.warn(ManagerMessages.FAILED_TO_COLLECT_PIPE_META_LIST_FROM_CONFIG_NODE_TASK, e);
135+
PipeLogger.log(
136+
LOGGER::warn, e, ManagerMessages.FAILED_TO_COLLECT_PIPE_META_LIST_FROM_CONFIG_NODE_TASK);
134137
}
135138
}
136139

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2323
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
2424
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
25+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
2526
import org.apache.iotdb.confignode.consensus.request.read.pipe.task.ShowPipePlanV2;
2627
import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
2728
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
@@ -215,7 +216,10 @@ public TShowPipeResp showPipes(final TShowPipeReq req) {
215216
.filter(req.whereClause, req.pipeName, req.isTableModel, req.userName)
216217
.convertToTShowPipeResp();
217218
} catch (final ConsensusException e) {
218-
LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_READ_API_EXECUTING_THE_CONSENSUS_LAYER_DUE, e);
219+
PipeLogger.log(
220+
LOGGER::warn,
221+
e,
222+
ConfigNodeMessages.FAILED_IN_THE_READ_API_EXECUTING_THE_CONSENSUS_LAYER_DUE);
219223
final TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
220224
res.setMessage(e.getMessage());
221225
return new PipeTableResp(res, Collections.emptyList()).convertToTShowPipeResp();
@@ -227,7 +231,7 @@ public TGetAllPipeInfoResp getAllPipeInfo() {
227231
return ((PipeTableResp) configManager.getConsensusManager().read(new ShowPipePlanV2()))
228232
.convertToTGetAllPipeInfoResp();
229233
} catch (IOException | ConsensusException e) {
230-
LOGGER.warn(ManagerMessages.FAILED_TO_GET_ALL_PIPE_INFO, e);
234+
PipeLogger.log(LOGGER::warn, e, ManagerMessages.FAILED_TO_GET_ALL_PIPE_INFO);
231235
return new TGetAllPipeInfoResp(
232236
new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode()).setMessage(e.getMessage()),
233237
Collections.emptyList());

0 commit comments

Comments
 (0)