Skip to content

Commit c818b28

Browse files
hongzhi-gaoluoluoyuyu
hongzhi-gao
and
luoluoyuyu
authored
Feature/iotconsensus snap multi dir (#16811)
* move FolderManager to node-common module * move FolderManager to node-common module * Multi-directory snapshot support for IoTConsensus receiver * Revert "Multi-directory snapshot support for IoTConsensus receiver" This reverts commit 1fef8e4. * IoTConsensus receiver snapshot multi dirs * fix imports * Add keep same disk when loading snapshot * Use commons i18n for disk messages * Fix multi directory strategy package prefix * Fix IoTConsensus snapshot fragment placement * Use local data dirs for IoTConsensus snapshot receive * Honor folder availability for same disk selection --------- Co-authored-by: luoluoyuyu <zhenyu@apache.org>
1 parent b46c58b commit c818b28

42 files changed

Lines changed: 478 additions & 174 deletions

File tree

Some content is hidden

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

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,13 +22,15 @@
2222
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
2323
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
2424

25+
import java.util.List;
2526
import java.util.Optional;
2627

2728
public class ConsensusConfig {
2829

2930
private final TEndPoint thisNodeEndPoint;
3031
private final int thisNodeId;
3132
private final String storageDir;
33+
private final List<String> recvSnapshotDirs;
3234
private final TConsensusGroupType consensusGroupType;
3335
private final RatisConfig ratisConfig;
3436
private final IoTConsensusConfig iotConsensusConfig;
@@ -38,13 +40,15 @@ private ConsensusConfig(
3840
TEndPoint thisNode,
3941
int thisNodeId,
4042
String storageDir,
43+
List<String> recvSnapshotDirs,
4144
TConsensusGroupType consensusGroupType,
4245
RatisConfig ratisConfig,
4346
IoTConsensusConfig iotConsensusConfig,
4447
IoTConsensusV2Config iotConsensusV2Config) {
4548
this.thisNodeEndPoint = thisNode;
4649
this.thisNodeId = thisNodeId;
4750
this.storageDir = storageDir;
51+
this.recvSnapshotDirs = recvSnapshotDirs;
4852
this.consensusGroupType = consensusGroupType;
4953
this.ratisConfig = ratisConfig;
5054
this.iotConsensusConfig = iotConsensusConfig;
@@ -63,6 +67,10 @@ public String getStorageDir() {
6367
return storageDir;
6468
}
6569

70+
public List<String> getRecvSnapshotDirs() {
71+
return recvSnapshotDirs;
72+
}
73+
6674
public TConsensusGroupType getConsensusGroupType() {
6775
return consensusGroupType;
6876
}
@@ -88,6 +96,7 @@ public static class Builder {
8896
private TEndPoint thisNode;
8997
private int thisNodeId;
9098
private String storageDir;
99+
private List<String> recvSnapshotDirs;
91100
private TConsensusGroupType consensusGroupType;
92101
private RatisConfig ratisConfig;
93102
private IoTConsensusConfig iotConsensusConfig;
@@ -98,6 +107,7 @@ public ConsensusConfig build() {
98107
thisNode,
99108
thisNodeId,
100109
storageDir,
110+
recvSnapshotDirs,
101111
consensusGroupType,
102112
Optional.ofNullable(ratisConfig).orElseGet(() -> RatisConfig.newBuilder().build()),
103113
Optional.ofNullable(iotConsensusConfig)
@@ -121,6 +131,11 @@ public Builder setStorageDir(String storageDir) {
121131
return this;
122132
}
123133

134+
public Builder setRecvSnapshotDirs(List<String> recvSnapshotDirs) {
135+
this.recvSnapshotDirs = recvSnapshotDirs;
136+
return this;
137+
}
138+
124139
public Builder setConsensusGroupType(TConsensusGroupType groupType) {
125140
this.consensusGroupType = groupType;
126141
return this;

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java‎

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.apache.iotdb.commons.concurrent.ThreadName;
2727
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
2828
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
29+
import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException;
2930
import org.apache.iotdb.commons.exception.StartupException;
3031
import org.apache.iotdb.commons.request.IConsensusRequest;
3132
import org.apache.iotdb.commons.service.RegisterManager;
@@ -94,6 +95,7 @@ public class IoTConsensus implements IConsensus {
9495
private final TEndPoint thisNode;
9596
private final int thisNodeId;
9697
private final File storageDir;
98+
private final List<String> recvSnapshotDirs;
9799
private final IStateMachine.Registry registry;
98100
private final Map<ConsensusGroupId, IoTConsensusServerImpl> stateMachineMap =
99101
new ConcurrentHashMap<>();
@@ -110,6 +112,7 @@ public IoTConsensus(ConsensusConfig config, Registry registry) {
110112
this.thisNode = config.getThisNodeEndPoint();
111113
this.thisNodeId = config.getThisNodeId();
112114
this.storageDir = new File(config.getStorageDir());
115+
this.recvSnapshotDirs = config.getRecvSnapshotDirs();
113116
this.config = config.getIotConsensusConfig();
114117
this.registry = registry;
115118
this.service = new IoTConsensusRPCService(thisNode, config.getIotConsensusConfig());
@@ -177,6 +180,7 @@ private void initAndRecover() throws IOException {
177180
IoTConsensusServerImpl consensus =
178181
new IoTConsensusServerImpl(
179182
path.toString(),
183+
recvSnapshotDirs,
180184
new Peer(consensusGroupId, thisNodeId, thisNode),
181185
new TreeSet<>(),
182186
registry.apply(consensusGroupId),
@@ -186,6 +190,8 @@ private void initAndRecover() throws IOException {
186190
config);
187191
stateMachineMap.put(consensusGroupId, consensus);
188192
}
193+
} catch (DiskSpaceInsufficientException e) {
194+
throw new IOException(e);
189195
}
190196
}
191197
if (correctPeerListBeforeStart != null) {
@@ -283,16 +289,22 @@ public void createLocalPeer(ConsensusGroupId groupId, List<Peer> peers)
283289
return null;
284290
}
285291

286-
IoTConsensusServerImpl impl =
287-
new IoTConsensusServerImpl(
288-
path,
289-
new Peer(groupId, thisNodeId, thisNode),
290-
new TreeSet<>(peers),
291-
registry.apply(groupId),
292-
backgroundTaskService,
293-
clientManager,
294-
syncClientManager,
295-
config);
292+
IoTConsensusServerImpl impl = null;
293+
try {
294+
impl =
295+
new IoTConsensusServerImpl(
296+
path,
297+
recvSnapshotDirs,
298+
new Peer(groupId, thisNodeId, thisNode),
299+
new TreeSet<>(peers),
300+
registry.apply(groupId),
301+
backgroundTaskService,
302+
clientManager,
303+
syncClientManager,
304+
config);
305+
} catch (DiskSpaceInsufficientException e) {
306+
throw new RuntimeException(e);
307+
}
296308
impl.start();
297309
return impl;
298310
}))

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java‎

Lines changed: 75 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,9 @@
2626
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
2727
import org.apache.iotdb.commons.consensus.index.ComparableConsensusRequest;
2828
import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex;
29+
import org.apache.iotdb.commons.disk.FolderManager;
30+
import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
31+
import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException;
2932
import org.apache.iotdb.commons.request.IConsensusRequest;
3033
import org.apache.iotdb.commons.service.metric.MetricService;
3134
import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
@@ -115,6 +118,7 @@ public class IoTConsensusServerImpl {
115118
private final Lock stateMachineLock = new ReentrantLock();
116119
private final Condition stateMachineCondition = stateMachineLock.newCondition();
117120
private final String storageDir;
121+
private FolderManager recvFolderManager = null;
118122
private final TreeSet<Peer> configuration;
119123
private final AtomicLong searchIndex;
120124
private final LogDispatcher logDispatcher;
@@ -132,15 +136,29 @@ public class IoTConsensusServerImpl {
132136

133137
public IoTConsensusServerImpl(
134138
String storageDir,
139+
List<String> recvSnapshotDirs,
135140
Peer thisNode,
136141
TreeSet<Peer> configuration,
137142
IStateMachine stateMachine,
138143
ScheduledExecutorService backgroundTaskService,
139144
IClientManager<TEndPoint, AsyncIoTConsensusServiceClient> clientManager,
140145
IClientManager<TEndPoint, SyncIoTConsensusServiceClient> syncClientManager,
141-
IoTConsensusConfig config) {
146+
IoTConsensusConfig config)
147+
throws DiskSpaceInsufficientException {
142148
this.active = true;
143149
this.storageDir = storageDir;
150+
List<String> snapshotDirs = new ArrayList<>();
151+
if (recvSnapshotDirs != null) {
152+
for (String dir : recvSnapshotDirs) {
153+
snapshotDirs.add(dir + File.separator + SNAPSHOT_DIR_NAME);
154+
}
155+
} else {
156+
snapshotDirs.add(storageDir);
157+
}
158+
159+
this.recvFolderManager =
160+
new FolderManager(
161+
snapshotDirs, DirectoryStrategyType.MIN_FOLDER_OCCUPIED_SPACE_FIRST_STRATEGY);
144162
this.thisNode = thisNode;
145163
this.stateMachine = stateMachine;
146164
this.cacheQueueMap = new ConcurrentHashMap<>();
@@ -360,18 +378,45 @@ public void receiveSnapshotFragment(
360378
throws ConsensusGroupModifyPeerException {
361379
try {
362380
String targetFilePath = calculateSnapshotPath(snapshotId, originalFilePath);
363-
File targetFile = getSnapshotPath(targetFilePath);
364-
Path parentDir = Paths.get(targetFile.getParent());
365-
if (!Files.exists(parentDir)) {
366-
Files.createDirectories(parentDir);
367-
}
368-
try (FileOutputStream fos = new FileOutputStream(targetFile.getAbsolutePath(), true);
369-
FileChannel channel = fos.getChannel()) {
370-
channel.write(fileChunk.slice(), fileOffset);
381+
File existingFile = getExistingSnapshotFile(targetFilePath);
382+
if (existingFile != null) {
383+
writeSnapshotFragment(existingFile, fileChunk, fileOffset);
384+
return;
371385
}
386+
387+
recvFolderManager.getNextWithRetry(
388+
folder -> {
389+
writeSnapshotFragment(getSnapshotPath(folder, targetFilePath), fileChunk, fileOffset);
390+
return null;
391+
});
372392
} catch (IOException e) {
373393
throw new ConsensusGroupModifyPeerException(
374394
String.format(IoTConsensusMessages.ERROR_RECEIVING_SNAPSHOT, snapshotId), e);
395+
} catch (DiskSpaceInsufficientException e) {
396+
throw new ConsensusGroupModifyPeerException(
397+
String.format(IoTConsensusMessages.ERROR_RECEIVING_SNAPSHOT, snapshotId), e);
398+
}
399+
}
400+
401+
private File getExistingSnapshotFile(String targetFilePath) {
402+
for (String folder : recvFolderManager.getFolders()) {
403+
File targetFile = getSnapshotPath(folder, targetFilePath);
404+
if (targetFile.exists()) {
405+
return targetFile;
406+
}
407+
}
408+
return null;
409+
}
410+
411+
private void writeSnapshotFragment(File targetFile, ByteBuffer fileChunk, long fileOffset)
412+
throws IOException {
413+
Path parentDir = Paths.get(targetFile.getParent());
414+
if (!Files.exists(parentDir)) {
415+
Files.createDirectories(parentDir);
416+
}
417+
try (FileOutputStream fos = new FileOutputStream(targetFile.getAbsolutePath(), true);
418+
FileChannel channel = fos.getChannel()) {
419+
channel.write(fileChunk.slice(), fileOffset);
375420
}
376421
}
377422

@@ -404,12 +449,17 @@ private void clearOldSnapshot() {
404449

405450
public void loadSnapshot(String snapshotId) {
406451
// TODO: (xingtanzjr) throw exception if the snapshot load failed
407-
stateMachine.loadSnapshot(getSnapshotPath(snapshotId));
452+
recvFolderManager
453+
.getFolders()
454+
.forEach(
455+
dir -> {
456+
stateMachine.loadSnapshot(getSnapshotPath(dir, snapshotId));
457+
});
408458
}
409459

410-
private File getSnapshotPath(String snapshotRelativePath) {
411-
File storageDirFile = new File(storageDir);
412-
File snapshotDir = new File(storageDir, snapshotRelativePath);
460+
private File getSnapshotPath(String curStorageDir, String snapshotRelativePath) {
461+
File storageDirFile = new File(curStorageDir);
462+
File snapshotDir = new File(curStorageDir, snapshotRelativePath);
413463
try {
414464
if (!snapshotDir
415465
.getCanonicalFile()
@@ -829,15 +879,19 @@ public void cleanupRemoteSnapshot(Peer targetPeer) throws ConsensusGroupModifyPe
829879
}
830880

831881
public void cleanupSnapshot(String snapshotId) throws ConsensusGroupModifyPeerException {
832-
File snapshotDir = getSnapshotPath(snapshotId);
833-
if (snapshotDir.exists()) {
834-
try {
835-
FileUtils.deleteDirectory(snapshotDir);
836-
} catch (IOException e) {
837-
throw new ConsensusGroupModifyPeerException(e);
882+
List<String> allDirs = new ArrayList<>(Collections.singletonList(storageDir));
883+
allDirs.addAll(recvFolderManager.getFolders());
884+
for (String dir : allDirs) {
885+
File snapshotDir = getSnapshotPath(dir, snapshotId);
886+
if (snapshotDir.exists()) {
887+
try {
888+
FileUtils.deleteDirectory(snapshotDir);
889+
} catch (IOException e) {
890+
throw new ConsensusGroupModifyPeerException(e);
891+
}
892+
} else {
893+
logger.info(IoTConsensusMessages.FILE_NOT_EXIST, snapshotDir);
838894
}
839-
} else {
840-
logger.info(IoTConsensusMessages.FILE_NOT_EXIST, snapshotDir);
841895
}
842896
}
843897

‎iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,21 @@ public class ReplicateTest {
7171
new File("target" + File.separator + "2"),
7272
new File("target" + File.separator + "3"));
7373

74+
private final List<List<String>> peersRecvSnapshotDirs =
75+
Arrays.asList(
76+
Arrays.asList(
77+
"target" + File.separator + "1-1",
78+
"target" + File.separator + "1-2",
79+
"target" + File.separator + "1-3"),
80+
Arrays.asList(
81+
"target" + File.separator + "2-1",
82+
"target" + File.separator + "2-2",
83+
"target" + File.separator + "2-3"),
84+
Arrays.asList(
85+
"target" + File.separator + "3-1",
86+
"target" + File.separator + "3-2",
87+
"target" + File.separator + "3-3"));
88+
7489
private final ConsensusGroup group = new ConsensusGroup(gid, peers);
7590
private final List<IoTConsensus> servers = new ArrayList<>();
7691
private final List<TestStateMachine> stateMachines = new ArrayList<>();
@@ -81,6 +96,7 @@ public void setUp() throws Exception {
8196
file.mkdirs();
8297
stateMachines.add(new TestStateMachine());
8398
}
99+
peersRecvSnapshotDirs.forEach(innerList -> innerList.forEach(dir -> new File(dir).mkdirs()));
84100
initServer();
85101
}
86102

@@ -90,6 +106,16 @@ public void tearDown() throws Exception {
90106
for (File file : peersStorage) {
91107
FileUtils.deleteFully(file);
92108
}
109+
peersRecvSnapshotDirs.forEach(
110+
innerList ->
111+
innerList.forEach(
112+
dir -> {
113+
try {
114+
FileUtils.deleteFully(new File(dir));
115+
} catch (IOException e) {
116+
throw new RuntimeException(e);
117+
}
118+
}));
93119
}
94120

95121
private void initServer() throws IOException {
@@ -105,6 +131,7 @@ private void initServer() throws IOException {
105131
.setThisNodeId(peers.get(i).getNodeId())
106132
.setThisNode(peers.get(i).getEndpoint())
107133
.setStorageDir(peersStorage.get(i).getAbsolutePath())
134+
.setRecvSnapshotDirs(peersRecvSnapshotDirs.get(i))
108135
.setConsensusGroupType(TConsensusGroupType.DataRegion)
109136
.build(),
110137
groupId -> stateMachines.get(finalI))

‎iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StabilityTest.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
import java.io.File;
4646
import java.io.IOException;
4747
import java.util.ArrayList;
48+
import java.util.Arrays;
4849
import java.util.Collections;
4950
import java.util.HashMap;
5051
import java.util.List;
@@ -60,6 +61,12 @@ public class StabilityTest {
6061

6162
private final File storageDir = new File("target" + java.io.File.separator + "stability");
6263

64+
private final List<String> recvSnapshotDirs =
65+
Arrays.asList(
66+
"target" + File.separator + "1-1",
67+
"target" + File.separator + "1-2",
68+
"target" + File.separator + "1-3");
69+
6370
private IoTConsensus consensusImpl;
6471

6572
private final int basePort = 6667;
@@ -73,6 +80,7 @@ public void constructConsensus() throws IOException {
7380
.setThisNodeId(1)
7481
.setThisNode(new TEndPoint("0.0.0.0", basePort))
7582
.setStorageDir(storageDir.getAbsolutePath())
83+
.setRecvSnapshotDirs(recvSnapshotDirs)
7684
.setConsensusGroupType(TConsensusGroupType.DataRegion)
7785
.build(),
7886
gid -> new TestStateMachine())

0 commit comments

Comments
 (0)