This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new e2014ff5329 Reduce repeated failure logs (#18175) (#18193)
e2014ff5329 is described below
commit e2014ff53294d719da3202cc2830b5ab9e6fbe8a
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 14 09:32:58 2026 +0800
Reduce repeated failure logs (#18175) (#18193)
* Reduce repeated failure logs
* Fix Chinese log message encoding
---
.../handlers/rpc/DataNodeTSStatusRPCHandler.java | 12 ++++-
.../manager/partition/PartitionManager.java | 8 ++-
.../agent/runtime/PipeConfigNodeRuntimeAgent.java | 10 ++--
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 4 +-
.../agent/runtime/PipeDataNodeRuntimeAgent.java | 10 ++--
.../protocol/thrift/IoTDBDataNodeReceiver.java | 4 +-
.../thrift/impl/DataNodeRegionManager.java | 7 ++-
.../schemaregion/SchemaRegionLoader.java | 8 ++-
.../storageengine/rescon/disk/FolderManager.java | 11 +++-
.../apache/iotdb/commons/conf/CommonConfig.java | 43 ++++++++++-----
.../LoggerPeriodicalLogReducer.java} | 63 ++++++++++++++--------
.../iotdb/commons/pipe/config/PipeConfig.java | 4 +-
.../iotdb/commons/pipe/config/PipeDescriptor.java | 16 +++---
.../commons/pipe/receiver/IoTDBFileReceiver.java | 24 ++++++---
.../commons/pipe/resource/log/PipeLogger.java | 12 +++--
.../LoggerPeriodicalLogReducerTest.java} | 21 ++++----
16 files changed, 173 insertions(+), 84 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java
index 7c93f363dd4..a44e3781c25 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.confignode.client.async.handlers.rpc;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -56,7 +57,7 @@ public class DataNodeTSStatusRPCHandler extends
DataNodeAsyncRequestRPCHandler<T
nodeLocationMap.remove(requestId);
LOGGER.info("Successfully {} on DataNode: {}", requestType,
formattedTargetLocation);
} else {
- LOGGER.error(
+ logFailure(
"Failed to {} on DataNode: {}, response: {}",
requestType,
formattedTargetLocation,
@@ -76,7 +77,7 @@ public class DataNodeTSStatusRPCHandler extends
DataNodeAsyncRequestRPCHandler<T
+ formattedTargetLocation
+ ", exception: "
+ e.getMessage();
- LOGGER.error(errorMsg);
+ logFailure(errorMsg);
responseMap.put(
requestId,
@@ -86,4 +87,11 @@ public class DataNodeTSStatusRPCHandler extends
DataNodeAsyncRequestRPCHandler<T
// Always CountDown
countDownLatch.countDown();
}
+
+ private void logFailure(final String format, final Object... args) {
+ if (!LoggerPeriodicalLogReducer.shouldLog(format, args)) {
+ return;
+ }
+ LOGGER.error(format, args);
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
index e186be44354..056b819d5bc 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.SchemaPartitionTable;
import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
@@ -280,7 +281,12 @@ public class PartitionManager {
return resp;
}
- LOGGER.error("Create SchemaPartition failed because: ", e);
+ if (LoggerPeriodicalLogReducer.shouldLog(
+ "Create SchemaPartition failed because: "
+ + e.getClass().getName()
+ + String.valueOf(e.getMessage()))) {
+ LOGGER.error("Create SchemaPartition failed because: ", e);
+ }
resp.setStatus(
new
TSStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode())
.setMessage(e.getMessage()));
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
index 900ccc8c10b..f5a864cd974 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
@@ -21,13 +21,13 @@ package
org.apache.iotdb.confignode.manager.pipe.agent.runtime;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor;
import
org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalPhantomReferenceCleaner;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
-import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
import org.apache.iotdb.commons.service.IService;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
@@ -58,7 +58,7 @@ public class PipeConfigNodeRuntimeAgent implements IService {
@Override
public synchronized void start() {
PipeConfig.getInstance().printAllConfigs();
- initPipePeriodicalLogReducer();
+ initLoggerPeriodicalLogReducer();
// PipeTasks will not be started here and will be started by
"HandleLeaderChange"
// procedure when the consensus layer notify leader ready
@@ -95,10 +95,10 @@ public class PipeConfigNodeRuntimeAgent implements IService
{
LOGGER.info("PipeRuntimeConfigNodeAgent stopped");
}
- private void initPipePeriodicalLogReducer() {
- PipePeriodicalLogReducer.setMemoryResizeFunction(
+ private void initLoggerPeriodicalLogReducer() {
+ LoggerPeriodicalLogReducer.setMemoryResizeFunction(
PipeConfigNodeResourceManager.memory()::resizeLogReducerMemory);
- PipeLogger.setLogger(PipePeriodicalLogReducer::log);
+ PipeLogger.setLogger(LoggerPeriodicalLogReducer::log);
}
public boolean isShutdown() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 792433829c6..b38a7c186ec 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -25,8 +25,8 @@ import org.apache.iotdb.commons.conf.ConfigurationFileUtils;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.conf.TrimProperties;
import org.apache.iotdb.commons.exception.BadNodeUrlException;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.commons.pipe.config.PipeDescriptor;
-import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
import org.apache.iotdb.commons.schema.SchemaConstant;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.utils.NodeUrlUtils;
@@ -2647,7 +2647,7 @@ public class IoTDBDescriptor {
private void loadPipeHotModifiedProp(TrimProperties properties) throws
IOException {
PipeDescriptor.loadPipeProps(commonDescriptor.getConfig(), properties,
true);
- PipePeriodicalLogReducer.update();
+ LoggerPeriodicalLogReducer.update();
}
@SuppressWarnings("squid:S3518") // "proportionSum" can't be zero
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
index f88e98d7165..235e7c1b0c9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java
@@ -25,6 +25,7 @@ import
org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor;
import
org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalPhantomReferenceCleaner;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
@@ -32,7 +33,6 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
-import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
import org.apache.iotdb.commons.service.IService;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.utils.TestOnly;
@@ -89,23 +89,23 @@ public class PipeDataNodeRuntimeAgent implements IService {
IoTDBPipePattern.setDevicePathGetter(CompactionPathUtils::getPath);
IoTDBPipePattern.setMeasurementPathGetter(CompactionPathUtils::getPath);
- initPipePeriodicalLogReducer();
+ initLoggerPeriodicalLogReducer();
}
- private void initPipePeriodicalLogReducer() {
+ private void initLoggerPeriodicalLogReducer() {
if (pipeLogReducerMemoryBlock == null) {
pipeLogReducerMemoryBlock =
PipeDataNodeResourceManager.memory()
.tryAllocate(PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes());
}
- PipePeriodicalLogReducer.setMemoryResizeFunction(
+ LoggerPeriodicalLogReducer.setMemoryResizeFunction(
targetSizeInBytes -> {
PipeDataNodeResourceManager.memory()
.resize(pipeLogReducerMemoryBlock, Math.max(0,
targetSizeInBytes), false);
return pipeLogReducerMemoryBlock.getMemoryUsageInBytes();
});
- PipeLogger.setLogger(PipePeriodicalLogReducer::log);
+ PipeLogger.setLogger(LoggerPeriodicalLogReducer::log);
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
index 945b4a55209..a1975f8bbe2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IllegalPathException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
@@ -31,7 +32,6 @@ import
org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
-import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
import
org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapPseudoTPipeTransferRequest;
import
org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferSliceReqHandler;
import
org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
@@ -929,7 +929,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
static boolean shouldLogStatementException(
final long receiverId, final Statement statement, final Exception e) {
// Use the reducer cache as a gate. The actual stack trace is logged only
when it passes.
- return PipePeriodicalLogReducer.log(
+ return LoggerPeriodicalLogReducer.log(
message -> {},
"Receiver id = %s, statement = %s, exception = %s, message = %s",
receiverId,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java
index 1b474ac7257..7d4631a0742 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.exception.ConsensusException;
@@ -126,7 +127,11 @@ public class DataNodeRegionManager {
tsStatus = new TSStatus(TSStatusCode.ILLEGAL_PATH.getStatusCode());
tsStatus.setMessage("Create Schema Region failed because storageGroup
path is illegal.");
} catch (MetadataException e2) {
- LOGGER.error("Create Schema Region {} failed because {}", storageGroup,
e2.getMessage());
+ if (LoggerPeriodicalLogReducer.shouldLog(
+ "Create Schema Region failed because of %s",
+ e2.getClass().getName() + String.valueOf(e2.getMessage()))) {
+ LOGGER.error("Create Schema Region {} failed because {}",
storageGroup, e2.getMessage());
+ }
tsStatus = new
TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode());
tsStatus.setMessage(
String.format("Create Schema Region failed because of %s",
e2.getMessage()));
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionLoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionLoader.java
index 5d47acfda5c..161f428d529 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionLoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionLoader.java
@@ -125,7 +125,13 @@ public class SchemaRegionLoader {
throws MetadataException {
try {
return currentConstructor.newInstance(schemaRegionParams);
- } catch (InstantiationException | IllegalAccessException |
InvocationTargetException e) {
+ } catch (InvocationTargetException e) {
+ if (e.getTargetException() instanceof MetadataException) {
+ throw (MetadataException) e.getTargetException();
+ }
+ logger.warn(e.getMessage(), e);
+ throw new MetadataException(e);
+ } catch (InstantiationException | IllegalAccessException e) {
logger.warn(e.getMessage(), e);
throw new MetadataException(e);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManager.java
index 0851a698aba..9315c63cd55 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManager.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.rescon.disk;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.db.exception.DiskSpaceInsufficientException;
import
org.apache.iotdb.db.storageengine.rescon.disk.strategy.DirectoryStrategy;
import
org.apache.iotdb.db.storageengine.rescon.disk.strategy.DirectoryStrategyType;
@@ -62,7 +63,10 @@ public class FolderManager {
try {
this.selectStrategy.setFolders(folders);
} catch (DiskSpaceInsufficientException e) {
- logger.error("All folders are full, change system mode to read-only.",
e);
+ if (LoggerPeriodicalLogReducer.shouldLog(
+ "All folders are full, change system mode to read-only.")) {
+ logger.error("All folders are full, change system mode to read-only.",
e);
+ }
CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly);
CommonDescriptor.getInstance().getConfig().setStatusReason(NodeStatus.DISK_FULL);
throw e;
@@ -73,7 +77,10 @@ public class FolderManager {
try {
return folders.get(selectStrategy.nextFolderIndex());
} catch (DiskSpaceInsufficientException e) {
- logger.error("All folders are full, change system mode to read-only.",
e);
+ if (LoggerPeriodicalLogReducer.shouldLog(
+ "All folders are full, change system mode to read-only.")) {
+ logger.error("All folders are full, change system mode to read-only.",
e);
+ }
CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly);
CommonDescriptor.getInstance().getConfig().setStatusReason(NodeStatus.DISK_FULL);
throw e;
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 37929ed1860..91fa09daac1 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -319,8 +319,8 @@ public class CommonConfig {
64 * 1024 * 1024,
(int) Math.min(Runtime.getRuntime().maxMemory() / 64,
Integer.MAX_VALUE));
private boolean pipeReceiverLoadConversionEnabled = false;
- private volatile long pipePeriodicalLogMinIntervalSeconds = 60;
- private volatile long pipeLoggerCacheMaxSizeInBytes = 16 * MB;
+ private volatile long loggerPeriodicalLogMinIntervalSeconds = 60;
+ private volatile long loggerCacheMaxSizeInBytes = 64 * MB;
private volatile double pipeMetaReportMaxLogNumPerRound = 0.1;
private volatile int pipeMetaReportMaxLogIntervalRounds = 360;
@@ -1690,29 +1690,46 @@ public class CommonConfig {
logger.info("pipeReceiverConversionEnabled is set to {}.",
pipeReceiverLoadConversionEnabled);
}
+ public long getLoggerPeriodicalLogMinIntervalSeconds() {
+ return loggerPeriodicalLogMinIntervalSeconds;
+ }
+
+ public void setLoggerPeriodicalLogMinIntervalSeconds(long
loggerPeriodicalLogMinIntervalSeconds) {
+ if (this.loggerPeriodicalLogMinIntervalSeconds ==
loggerPeriodicalLogMinIntervalSeconds) {
+ return;
+ }
+ this.loggerPeriodicalLogMinIntervalSeconds =
loggerPeriodicalLogMinIntervalSeconds;
+ logger.info(
+ "loggerPeriodicalLogMinIntervalSeconds is set to {}.",
+ loggerPeriodicalLogMinIntervalSeconds);
+ }
+
public long getPipePeriodicalLogMinIntervalSeconds() {
- return pipePeriodicalLogMinIntervalSeconds;
+ return getLoggerPeriodicalLogMinIntervalSeconds();
}
public void setPipePeriodicalLogMinIntervalSeconds(long
pipePeriodicalLogMinIntervalSeconds) {
- if (this.pipePeriodicalLogMinIntervalSeconds ==
pipePeriodicalLogMinIntervalSeconds) {
+
setLoggerPeriodicalLogMinIntervalSeconds(pipePeriodicalLogMinIntervalSeconds);
+ }
+
+ public long getLoggerCacheMaxSizeInBytes() {
+ return loggerCacheMaxSizeInBytes;
+ }
+
+ public void setLoggerCacheMaxSizeInBytes(long loggerCacheMaxSizeInBytes) {
+ if (this.loggerCacheMaxSizeInBytes == loggerCacheMaxSizeInBytes) {
return;
}
- this.pipePeriodicalLogMinIntervalSeconds =
pipePeriodicalLogMinIntervalSeconds;
- logger.info(
- "pipePeriodicalLogMinIntervalSeconds is set to {}.",
pipePeriodicalLogMinIntervalSeconds);
+ this.loggerCacheMaxSizeInBytes = loggerCacheMaxSizeInBytes;
+ logger.info("loggerCacheMaxSizeInBytes is set to {}.",
loggerCacheMaxSizeInBytes);
}
public long getPipeLoggerCacheMaxSizeInBytes() {
- return pipeLoggerCacheMaxSizeInBytes;
+ return getLoggerCacheMaxSizeInBytes();
}
public void setPipeLoggerCacheMaxSizeInBytes(long
pipeLoggerCacheMaxSizeInBytes) {
- if (this.pipeLoggerCacheMaxSizeInBytes == pipeLoggerCacheMaxSizeInBytes) {
- return;
- }
- this.pipeLoggerCacheMaxSizeInBytes = pipeLoggerCacheMaxSizeInBytes;
- logger.info("pipeLoggerCacheMaxSizeInBytes is set to {}.",
pipeLoggerCacheMaxSizeInBytes);
+ setLoggerCacheMaxSizeInBytes(pipeLoggerCacheMaxSizeInBytes);
}
public int getPipeReceiverReqDecompressedMaxLengthInBytes() {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducer.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducer.java
similarity index 56%
rename from
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducer.java
rename to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducer.java
index fa3f29ccbe3..524e76841b7 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducer.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducer.java
@@ -17,22 +17,23 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.resource.log;
+package org.apache.iotdb.commons.log;
-import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.tsfile.utils.RamUsageEstimator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.slf4j.helpers.MessageFormatter;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.LongUnaryOperator;
-public class PipePeriodicalLogReducer {
- private static final Logger LOGGER =
LoggerFactory.getLogger(PipePeriodicalLogReducer.class);
+public class LoggerPeriodicalLogReducer {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(LoggerPeriodicalLogReducer.class);
private static final LongUnaryOperator DEFAULT_MEMORY_RESIZE_FUNCTION =
sizeInBytes -> sizeInBytes;
@@ -41,10 +42,9 @@ public class PipePeriodicalLogReducer {
protected static final Cache<String, String> LOGGER_CACHE =
Caffeine.newBuilder()
- .expireAfterWrite(
-
PipeConfig.getInstance().getPipePeriodicalLogMinIntervalSeconds(),
TimeUnit.SECONDS)
- .weigher(PipePeriodicalLogReducer::estimateSize)
-
.maximumWeight(PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes())
+ .expireAfterWrite(getLogMinIntervalSeconds(), TimeUnit.SECONDS)
+ .weigher(LoggerPeriodicalLogReducer::estimateSize)
+ .maximumWeight(getLoggerCacheMaxSizeInBytes())
.build();
private static int estimateSize(final String key, final String value) {
@@ -54,27 +54,32 @@ public class PipePeriodicalLogReducer {
public static boolean log(
final Consumer<String> loggerFunction, final String rawMessage, final
Object... formatter) {
- final String loggerMessage = String.format(rawMessage, formatter);
- if (!LOGGER_CACHE.asMap().containsKey(loggerMessage)) {
- LOGGER_CACHE.put(loggerMessage, loggerMessage);
+ final String loggerMessage = formatMessage(rawMessage, formatter);
+ if (shouldLog(loggerMessage)) {
loggerFunction.accept(loggerMessage);
return true;
}
return false;
}
+ public static boolean shouldLog(final String loggerMessage) {
+ return LOGGER_CACHE.asMap().putIfAbsent(loggerMessage, loggerMessage) ==
null;
+ }
+
+ public static boolean shouldLog(final String rawMessage, final Object...
formatter) {
+ return shouldLog(formatMessage(rawMessage, formatter));
+ }
+
public static synchronized void setMemoryResizeFunction(
final LongUnaryOperator memoryResizeFunction) {
- PipePeriodicalLogReducer.memoryResizeFunction =
+ LoggerPeriodicalLogReducer.memoryResizeFunction =
memoryResizeFunction == null ? DEFAULT_MEMORY_RESIZE_FUNCTION :
memoryResizeFunction;
update();
}
public static synchronized void update() {
- final long maxWeight =
- memoryResizeFunction.applyAsLong(
- PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes());
- LOGGER.info("PipePeriodicalLogReducer is allocated to {} bytes.",
maxWeight);
+ final long maxWeight =
memoryResizeFunction.applyAsLong(getLoggerCacheMaxSizeInBytes());
+ LOGGER.info("LoggerPeriodicalLogReducer is allocated to {} bytes.",
maxWeight);
update(maxWeight);
}
@@ -82,15 +87,29 @@ public class PipePeriodicalLogReducer {
LOGGER_CACHE
.policy()
.expireAfterWrite()
- .ifPresent(
- time ->
- time.setExpiresAfter(
-
PipeConfig.getInstance().getPipePeriodicalLogMinIntervalSeconds(),
- TimeUnit.SECONDS));
+ .ifPresent(time -> time.setExpiresAfter(getLogMinIntervalSeconds(),
TimeUnit.SECONDS));
LOGGER_CACHE.policy().eviction().ifPresent(eviction ->
eviction.setMaximum(maxWeight));
}
- private PipePeriodicalLogReducer() {
+ public static String formatMessage(final String rawMessage, final Object...
formatter) {
+ if (formatter == null || formatter.length == 0) {
+ return rawMessage;
+ }
+ if (rawMessage.contains("{}")) {
+ return MessageFormatter.arrayFormat(rawMessage, formatter).getMessage();
+ }
+ return String.format(rawMessage, formatter);
+ }
+
+ private static long getLogMinIntervalSeconds() {
+ return
CommonDescriptor.getInstance().getConfig().getLoggerPeriodicalLogMinIntervalSeconds();
+ }
+
+ private static long getLoggerCacheMaxSizeInBytes() {
+ return
CommonDescriptor.getInstance().getConfig().getLoggerCacheMaxSizeInBytes();
+ }
+
+ private LoggerPeriodicalLogReducer() {
// static
}
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
index 0840bb48e59..16250dc1e8a 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -636,7 +636,7 @@ public class PipeConfig {
getPipeAirGapReceiverMaxPayloadSizeInBytes());
LOGGER.info("PipeReceiverLoadConversionEnabled: {}",
isPipeReceiverLoadConversionEnabled());
LOGGER.info(
- "PipePeriodicalLogMinIntervalSeconds: {}",
getPipePeriodicalLogMinIntervalSeconds());
+ "LoggerPeriodicalLogMinIntervalSeconds: {}",
getPipePeriodicalLogMinIntervalSeconds());
LOGGER.info(
"PipeRetryLocallyForParallelOrUserConflict: {}",
isPipeRetryLocallyForParallelOrUserConflict());
@@ -645,7 +645,7 @@ public class PipeConfig {
LOGGER.info("PipeMetaReportMaxLogIntervalRounds: {}",
getPipeMetaReportMaxLogIntervalRounds());
LOGGER.info("PipeTsFilePinMaxLogNumPerRound: {}",
getPipeTsFilePinMaxLogNumPerRound());
LOGGER.info("PipeTsFilePinMaxLogIntervalRounds: {}",
getPipeTsFilePinMaxLogIntervalRounds());
- LOGGER.info("PipeLoggerCacheMaxSizeInBytes: {}",
getPipeLoggerCacheMaxSizeInBytes());
+ LOGGER.info("LoggerCacheMaxSizeInBytes: {}",
getPipeLoggerCacheMaxSizeInBytes());
LOGGER.info("PipeMemoryManagementEnabled: {}",
getPipeMemoryManagementEnabled());
LOGGER.info("PipeMemoryAllocateMaxRetries: {}",
getPipeMemoryAllocateMaxRetries());
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
index ee035ba3eb1..2cf877db505 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
@@ -476,16 +476,20 @@ public class PipeDescriptor {
properties.getProperty(
"pipe_receiver_load_conversion_enabled",
String.valueOf(config.isPipeReceiverLoadConversionEnabled()))));
- config.setPipePeriodicalLogMinIntervalSeconds(
+ config.setLoggerPeriodicalLogMinIntervalSeconds(
Long.parseLong(
properties.getProperty(
- "pipe_periodical_log_min_interval_seconds",
-
String.valueOf(config.getPipePeriodicalLogMinIntervalSeconds()))));
- config.setPipeLoggerCacheMaxSizeInBytes(
+ "logger_periodical_log_min_interval_seconds",
+ properties.getProperty(
+ "pipe_periodical_log_min_interval_seconds",
+
String.valueOf(config.getLoggerPeriodicalLogMinIntervalSeconds())))));
+ config.setLoggerCacheMaxSizeInBytes(
Long.parseLong(
properties.getProperty(
- "pipe_logger_cache_max_size_in_bytes",
- String.valueOf(config.getPipeLoggerCacheMaxSizeInBytes()))));
+ "logger_cache_max_size_in_bytes",
+ properties.getProperty(
+ "pipe_logger_cache_max_size_in_bytes",
+ String.valueOf(config.getLoggerCacheMaxSizeInBytes())))));
config.setPipeMemoryAllocateMaxRetries(
Integer.parseInt(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
index b1e29c8563d..cb293de1444 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.audit.UserEntity;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.IllegalPathException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
@@ -179,17 +180,26 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
try {
receiverFileBaseDir = getReceiverFileBaseDir();
if (Objects.isNull(receiverFileBaseDir)) {
+ if (LoggerPeriodicalLogReducer.shouldLog(
+ "Receiver id = %s: Failed to init pipe receiver file folder
manager because all disks of folders are full.")) {
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = %s: Failed to init pipe receiver file folder
manager because all disks of folders are full.",
+ receiverId.get());
+ }
+ return new
TPipeTransferResp(StatusUtils.getStatus(TSStatusCode.DISK_SPACE_INSUFFICIENT));
+ }
+ } catch (Exception e) {
+ if (LoggerPeriodicalLogReducer.shouldLog(
+ "Receiver id = %s: Failed to create pipe receiver file folder
because all disks of folders are full."
+ + e.getClass().getName()
+ + e.getMessage())) {
PipeLogger.log(
LOGGER::warn,
- "Receiver id = %s: Failed to init pipe receiver file folder
manager because all disks of folders are full.",
+ e,
+ "Receiver id = %s: Failed to create pipe receiver file folder
because all disks of folders are full.",
receiverId.get());
- return new
TPipeTransferResp(StatusUtils.getStatus(TSStatusCode.DISK_SPACE_INSUFFICIENT));
}
- } catch (Exception e) {
- LOGGER.warn(
- "Receiver id = {}: Failed to create pipe receiver file folder
because all disks of folders are full.",
- receiverId.get(),
- e);
return new
TPipeTransferResp(StatusUtils.getStatus(TSStatusCode.DISK_SPACE_INSUFFICIENT));
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java
index 70b494da032..720999aae97 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.commons.pipe.resource.log;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
+
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.util.function.Consumer;
@@ -26,11 +28,12 @@ import java.util.function.Consumer;
public class PipeLogger {
private static PipePeriodicalLogger logger =
(loggerFunction, rawMessage, formatter) ->
- loggerFunction.accept(String.format(rawMessage, formatter));
+
loggerFunction.accept(LoggerPeriodicalLogReducer.formatMessage(rawMessage,
formatter));
public static void log(
final Consumer<String> loggerFunction, final String rawMessage, final
Object... formatter) {
- logger.log(loggerFunction, rawMessage, formatter);
+ logger.log(
+ loggerFunction, "%s",
LoggerPeriodicalLogReducer.formatMessage(rawMessage, formatter));
}
public static void log(
@@ -40,7 +43,10 @@ public class PipeLogger {
final Object... formatter) {
final ByteArrayOutputStream out = new ByteArrayOutputStream();
throwable.printStackTrace(new PrintStream(out));
- logger.log(loggerFunction, rawMessage + "\n" + out, formatter);
+ logger.log(
+ loggerFunction,
+ "%s",
+ LoggerPeriodicalLogReducer.formatMessage(rawMessage, formatter) + "\n"
+ out);
}
public static void setLogger(final PipePeriodicalLogger logger) {
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducerTest.java
similarity index 67%
rename from
iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducerTest.java
rename to
iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducerTest.java
index d390cbb9fdd..a0a6df14e9c 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducerTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducerTest.java
@@ -17,9 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.commons.pipe.resource.log;
+package org.apache.iotdb.commons.log;
-import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.junit.After;
import org.junit.Assert;
@@ -28,20 +28,20 @@ import org.junit.Test;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-public class PipePeriodicalLogReducerTest {
+public class LoggerPeriodicalLogReducerTest {
@After
public void tearDown() {
- PipePeriodicalLogReducer.setMemoryResizeFunction(null);
+ LoggerPeriodicalLogReducer.setMemoryResizeFunction(null);
}
@Test
public void testLogReducesDuplicateMessages() {
final AtomicInteger logCount = new AtomicInteger(0);
- final String message = "PipePeriodicalLogReducerTest-" + System.nanoTime();
+ final String message = "LoggerPeriodicalLogReducerTest-" +
System.nanoTime();
- Assert.assertTrue(PipePeriodicalLogReducer.log(log ->
logCount.incrementAndGet(), message));
- Assert.assertFalse(PipePeriodicalLogReducer.log(log ->
logCount.incrementAndGet(), message));
+ Assert.assertTrue(LoggerPeriodicalLogReducer.log(log ->
logCount.incrementAndGet(), message));
+ Assert.assertFalse(LoggerPeriodicalLogReducer.log(log ->
logCount.incrementAndGet(), message));
Assert.assertEquals(1, logCount.get());
}
@@ -50,16 +50,17 @@ public class PipePeriodicalLogReducerTest {
final AtomicLong requestedSizeInBytes = new AtomicLong(-1);
final long allocatedSizeInBytes = 1024;
- PipePeriodicalLogReducer.setMemoryResizeFunction(
+ LoggerPeriodicalLogReducer.setMemoryResizeFunction(
sizeInBytes -> {
requestedSizeInBytes.set(sizeInBytes);
return allocatedSizeInBytes;
});
Assert.assertEquals(
- PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes(),
requestedSizeInBytes.get());
+
CommonDescriptor.getInstance().getConfig().getLoggerCacheMaxSizeInBytes(),
+ requestedSizeInBytes.get());
Assert.assertEquals(
allocatedSizeInBytes,
-
PipePeriodicalLogReducer.LOGGER_CACHE.policy().eviction().get().getMaximum());
+
LoggerPeriodicalLogReducer.LOGGER_CACHE.policy().eviction().get().getMaximum());
}
}
