Skip to content

Commit 67c105b

Browse files
sadanand48sadanand48
authored
HDDS-13906. Reduce Bootstrap Write lock time on OM during bootstrapping execution. (#9585)
1 parent 4f16a96 commit 67c105b

4 files changed

Lines changed: 401 additions & 111 deletions

File tree

‎hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMDbCheckpointServletInodeBasedXfer.java‎

Lines changed: 33 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -36,9 +36,8 @@
3636
import static org.junit.jupiter.api.Assertions.assertTrue;
3737
import static org.junit.jupiter.api.Assertions.fail;
3838
import static org.mockito.ArgumentMatchers.anyCollection;
39-
import static org.mockito.ArgumentMatchers.anyInt;
40-
import static org.mockito.ArgumentMatchers.anyMap;
4139
import static org.mockito.ArgumentMatchers.anySet;
40+
import static org.mockito.ArgumentMatchers.anyString;
4241
import static org.mockito.Mockito.any;
4342
import static org.mockito.Mockito.anyBoolean;
4443
import static org.mockito.Mockito.doAnswer;
@@ -84,16 +83,13 @@
8483
import javax.servlet.WriteListener;
8584
import javax.servlet.http.HttpServletRequest;
8685
import javax.servlet.http.HttpServletResponse;
87-
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
88-
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
8986
import org.apache.commons.lang3.RandomStringUtils;
9087
import org.apache.hadoop.fs.FileUtil;
9188
import org.apache.hadoop.hdds.client.ReplicationConfig;
9289
import org.apache.hadoop.hdds.client.ReplicationFactor;
9390
import org.apache.hadoop.hdds.client.ReplicationType;
9491
import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
9592
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
96-
import org.apache.hadoop.hdds.utils.Archiver;
9793
import org.apache.hadoop.hdds.utils.IOUtils;
9894
import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
9995
import org.apache.hadoop.hdds.utils.db.DBStore;
@@ -239,19 +235,19 @@ public void write(int b) throws IOException {
239235
doCallRealMethod().when(omDbCheckpointServletMock)
240236
.writeDbDataToStream(any(), any(), any(), any(), any());
241237
doCallRealMethod().when(omDbCheckpointServletMock)
242-
.writeDBToArchive(any(), any(), any(), any(), any(), any(), anyBoolean());
238+
.collectFilesFromDir(any(), any(), any(), anyBoolean(), any());
239+
doCallRealMethod().when(omDbCheckpointServletMock).collectDbDataToTransfer(any(), any(), any());
243240

244241
when(omDbCheckpointServletMock.getBootstrapStateLock())
245242
.thenReturn(lock);
246243
doCallRealMethod().when(omDbCheckpointServletMock).getCheckpoint(any(), anyBoolean());
247244
assertNull(doCallRealMethod().when(omDbCheckpointServletMock).getBootstrapTempData());
248245
doCallRealMethod().when(omDbCheckpointServletMock).
249246
processMetadataSnapshotRequest(any(), any(), anyBoolean(), anyBoolean());
250-
doCallRealMethod().when(omDbCheckpointServletMock).writeDbDataToStream(any(), any(), any(), any());
251247
doCallRealMethod().when(omDbCheckpointServletMock).getCompactionLogDir();
252248
doCallRealMethod().when(omDbCheckpointServletMock).getSstBackupDir();
253249
doCallRealMethod().when(omDbCheckpointServletMock)
254-
.transferSnapshotData(anySet(), any(), anyCollection(), anyCollection(), any(), any(), anyMap());
250+
.collectSnapshotData(anySet(), anyCollection(), anyCollection(), any(), any());
255251
doCallRealMethod().when(omDbCheckpointServletMock).createAndPrepareCheckpoint(anyBoolean());
256252
doCallRealMethod().when(omDbCheckpointServletMock).getSnapshotDirsFromDB(any(), any(), any());
257253
}
@@ -298,14 +294,14 @@ public void testWriteDBToArchiveClosesFilesListStream() throws Exception {
298294
// Do not use CALLS_REAL_METHODS for java.nio.file.Files: internal/private static
299295
// methods (eg Files.provider()) get intercepted too and Mockito will try to invoke
300296
// them reflectively, which fails on JDK9+ without --add-opens.
301-
try (MockedStatic<Files> files = mockStatic(Files.class);
302-
TarArchiveOutputStream tar = new TarArchiveOutputStream(new java.io.ByteArrayOutputStream())) {
297+
try (MockedStatic<Files> files = mockStatic(Files.class)) {
303298
files.when(() -> Files.exists(dbDir)).thenReturn(true);
304299
files.when(() -> Files.list(dbDir)).thenReturn(stream);
305300

306-
boolean result = servlet.writeDBToArchive(
307-
new HashSet<>(), dbDir, new AtomicLong(Long.MAX_VALUE),
308-
tar, folder, null, true);
301+
OMDBArchiver archiver = new OMDBArchiver();
302+
archiver.setTmpDir(folder);
303+
boolean result = servlet.collectFilesFromDir(new HashSet<>(), dbDir,
304+
new AtomicLong(Long.MAX_VALUE), true, archiver);
309305
assertTrue(result);
310306
}
311307

@@ -446,49 +442,38 @@ public void testWriteDBToArchive(boolean expectOnlySstFiles) throws Exception {
446442
// Create dummy files: one SST, one non-SST
447443
Path sstFile = dbDir.resolve("test.sst");
448444
Files.write(sstFile, "sst content".getBytes(StandardCharsets.UTF_8)); // Write some content to make it non-empty
449-
450445
Path nonSstFile = dbDir.resolve("test.log");
451446
Files.write(nonSstFile, "log content".getBytes(StandardCharsets.UTF_8));
452447
Set<String> sstFilesToExclude = new HashSet<>();
453448
AtomicLong maxTotalSstSize = new AtomicLong(1000000); // Sufficient size
454-
Map<String, String> hardLinkFileMap = new java.util.HashMap<>();
449+
OMDBArchiver omDbArchiver = new OMDBArchiver();
455450
Path tmpDir = folder.resolve("tmp");
456451
Files.createDirectories(tmpDir);
457-
TarArchiveOutputStream mockArchiveOutputStream = mock(TarArchiveOutputStream.class);
452+
omDbArchiver.setTmpDir(tmpDir);
453+
OMDBArchiver omDbArchiverSpy = spy(omDbArchiver);
458454
List<String> fileNames = new ArrayList<>();
459-
try (MockedStatic<Archiver> archiverMock = mockStatic(Archiver.class)) {
460-
archiverMock.when(() -> Archiver.linkAndIncludeFile(any(), any(), any(), any())).thenAnswer(invocation -> {
461-
// Get the actual mockArchiveOutputStream passed from writeDBToArchive
462-
TarArchiveOutputStream aos = invocation.getArgument(2);
463-
File sourceFile = invocation.getArgument(0);
464-
String fileId = invocation.getArgument(1);
465-
fileNames.add(sourceFile.getName());
466-
aos.putArchiveEntry(new TarArchiveEntry(sourceFile, fileId));
467-
aos.write(new byte[100], 0, 100); // Simulate writing
468-
aos.closeArchiveEntry();
469-
return 100L;
470-
});
471-
boolean success = omDbCheckpointServletMock.writeDBToArchive(
472-
sstFilesToExclude, dbDir, maxTotalSstSize, mockArchiveOutputStream,
473-
tmpDir, hardLinkFileMap, expectOnlySstFiles);
474-
assertTrue(success);
475-
verify(mockArchiveOutputStream, times(fileNames.size())).putArchiveEntry(any());
476-
verify(mockArchiveOutputStream, times(fileNames.size())).closeArchiveEntry();
477-
verify(mockArchiveOutputStream, times(fileNames.size())).write(any(byte[].class), anyInt(),
478-
anyInt()); // verify write was called once
479-
480-
boolean containsNonSstFile = false;
481-
for (String fileName : fileNames) {
482-
if (expectOnlySstFiles) {
483-
assertTrue(fileName.endsWith(".sst"), "File is not an SST File");
484-
} else {
485-
containsNonSstFile = true;
486-
}
455+
doAnswer((invocation) -> {
456+
File sourceFile = invocation.getArgument(0);
457+
fileNames.add(sourceFile.getName());
458+
omDbArchiver.recordFileEntry(sourceFile, invocation.getArgument(1));
459+
return null;
460+
}).when(omDbArchiverSpy).recordFileEntry(any(), anyString());
461+
boolean success =
462+
omDbCheckpointServletMock.collectFilesFromDir(sstFilesToExclude, dbDir, maxTotalSstSize, expectOnlySstFiles,
463+
omDbArchiverSpy);
464+
assertTrue(success);
465+
verify(omDbArchiverSpy, times(fileNames.size())).recordFileEntry(any(), anyString());
466+
boolean containsNonSstFile = false;
467+
for (String fileName : fileNames) {
468+
if (expectOnlySstFiles) {
469+
assertTrue(fileName.endsWith(".sst"), "File is not an SST File");
470+
} else {
471+
containsNonSstFile = true;
487472
}
473+
}
488474

489-
if (!expectOnlySstFiles) {
490-
assertTrue(containsNonSstFile, "SST File is not expected");
491-
}
475+
if (!expectOnlySstFiles) {
476+
assertTrue(containsNonSstFile, "SST File is not expected");
492477
}
493478
}
494479

@@ -905,6 +890,7 @@ private void setupClusterAndMocks(String volumeName, String bucketName,
905890
doCallRealMethod().when(omDbCheckpointServletMock).initialize(any(), any(),
906891
eq(false), any(), any(), eq(false));
907892
doCallRealMethod().when(omDbCheckpointServletMock).getSnapshotDirsFromDB(any(), any(), any());
893+
doCallRealMethod().when(omDbCheckpointServletMock).collectDbDataToTransfer(any(), any(), any());
908894
omDbCheckpointServletMock.initialize(spyDbStore, om.getMetrics().getDBCheckpointMetrics(),
909895
false,
910896
om.getOmAdminUsernames(), om.getOmAdminGroups(), false);

‎hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMDBArchiver.java‎

Lines changed: 165 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,165 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. 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+
package org.apache.hadoop.ozone.om;
19+
20+
import static org.apache.hadoop.hdds.utils.Archiver.includeFile;
21+
import static org.apache.hadoop.hdds.utils.Archiver.tar;
22+
import static org.apache.hadoop.hdds.utils.HddsServerUtil.includeRatisSnapshotCompleteFlag;
23+
import static org.apache.hadoop.ozone.om.OMDBCheckpointServletInodeBasedXfer.writeHardlinkFile;
24+
25+
import java.io.File;
26+
import java.io.IOException;
27+
import java.io.OutputStream;
28+
import java.nio.file.Files;
29+
import java.nio.file.Path;
30+
import java.util.HashMap;
31+
import java.util.Map;
32+
import org.apache.commons.compress.archivers.ArchiveOutputStream;
33+
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
34+
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
35+
import org.apache.hadoop.util.Time;
36+
import org.slf4j.Logger;
37+
import org.slf4j.LoggerFactory;
38+
39+
/**
40+
* Class for handling operations relevant to archiving the OM DB tarball.
41+
* Mainly maintains a map for recording the files collected from reading
42+
* the checkpoint and snapshot DB's. It temporarily creates hardlinks and stores
43+
* the link data in the map to release the bootstrap lock quickly
44+
* and do the actual write at the end outside the lock.
45+
*/
46+
public class OMDBArchiver {
47+
48+
private Path tmpDir;
49+
private Map<String, File> filesToWriteIntoTarball;
50+
private Map<String, String> hardLinkFileMap;
51+
private static final Logger LOG = LoggerFactory.getLogger(OMDBArchiver.class);
52+
private boolean completed;
53+
54+
public OMDBArchiver() {
55+
this.tmpDir = null;
56+
this.filesToWriteIntoTarball = new HashMap<>();
57+
this.hardLinkFileMap = null;
58+
this.completed = false;
59+
}
60+
61+
public void setTmpDir(Path tmpDir) {
62+
this.tmpDir = tmpDir;
63+
}
64+
65+
public Path getTmpDir() {
66+
return tmpDir;
67+
}
68+
69+
public Map<String, String> getHardLinkFileMap() {
70+
return hardLinkFileMap;
71+
}
72+
73+
public Map<String, File> getFilesToWriteIntoTarball() {
74+
return filesToWriteIntoTarball;
75+
}
76+
77+
public void setHardLinkFileMap(Map<String, String> hardLinkFileMap) {
78+
this.hardLinkFileMap = hardLinkFileMap;
79+
}
80+
81+
public boolean isCompleted() {
82+
return completed;
83+
}
84+
85+
public void setCompleted(boolean completed) {
86+
this.completed = completed;
87+
}
88+
89+
/**
90+
* Records the given file entry into the map after taking a hardlink.
91+
*
92+
* @param file the file to create a hardlink and record into the map
93+
* @param entryName name of the entry corresponding to file
94+
* @return the file size
95+
* @throws IOException in case of hardlink failure
96+
*/
97+
public long recordFileEntry(File file, String entryName) throws IOException {
98+
if (tmpDir == null) {
99+
throw new IllegalStateException(
100+
"Temporary directory not set. Call setTmpDir() before recordFileEntry().");
101+
}
102+
File link = tmpDir.resolve(entryName).toFile();
103+
long bytes = 0;
104+
try {
105+
Path linkPath = link.toPath();
106+
if (Files.exists(linkPath)) {
107+
// If the existing file is already a link to the same source, just reuse it.
108+
if (Files.isSameFile(linkPath, file.toPath())) {
109+
filesToWriteIntoTarball.put(entryName, link);
110+
return file.length();
111+
}
112+
// Otherwise, remove the stale link/entry so we can recreate it.
113+
Files.delete(linkPath);
114+
}
115+
Files.createLink(linkPath, file.toPath());
116+
filesToWriteIntoTarball.put(entryName, link);
117+
bytes = file.length();
118+
} catch (IOException ioe) {
119+
LOG.error("Couldn't create hardlink for file {} while including it in tarball.",
120+
file.getAbsolutePath(), ioe);
121+
throw ioe;
122+
}
123+
return bytes;
124+
}
125+
126+
/**
127+
* Writes all the files captured by the map into the archive and
128+
* also includes the hardlinkFile and the completion marker file.
129+
*
130+
* @param conf the configuration object to obtain metadata paths
131+
* @param outputStream the tarball archive output stream
132+
* @throws IOException in case of write failure to the archive
133+
*/
134+
public void writeToArchive(OzoneConfiguration conf, OutputStream outputStream)
135+
throws IOException {
136+
long bytesWritten = 0;
137+
long lastLoggedTime = Time.monotonicNow();
138+
long filesWritten = 0;
139+
try (ArchiveOutputStream<TarArchiveEntry> archiveOutput = tar(outputStream)) {
140+
for (Map.Entry<String, File> kv : filesToWriteIntoTarball.entrySet()) {
141+
String entryName = kv.getKey();
142+
File link = kv.getValue();
143+
try {
144+
bytesWritten += includeFile(link, entryName, archiveOutput);
145+
filesWritten++;
146+
if (Time.monotonicNow() - lastLoggedTime >= 30000) {
147+
LOG.info("Transferred {} KB, #files {} to checkpoint tarball stream...",
148+
bytesWritten / (1024), filesWritten);
149+
lastLoggedTime = Time.monotonicNow();
150+
}
151+
} catch (IOException ioe) {
152+
LOG.error("Failed to write file {} to checkpoint tarball archive.",
153+
link.getAbsolutePath(), ioe);
154+
throw ioe;
155+
} finally {
156+
Files.deleteIfExists(link.toPath());
157+
}
158+
}
159+
if (isCompleted()) {
160+
writeHardlinkFile(conf, hardLinkFileMap, archiveOutput);
161+
includeRatisSnapshotCompleteFlag(archiveOutput);
162+
}
163+
}
164+
}
165+
}

0 commit comments

Comments
 (0)