This is an automated email from the ASF dual-hosted git repository.
keith-turner pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.1 by this push:
new 382010dac8 backport log recovery performance improvements (#6395)
382010dac8 is described below
commit 382010dac828c4d72ee9ad80b434eaddb62a2af1
Author: amcdonaldccri <[email protected]>
AuthorDate: Fri Nov 6 10:30:58 2026 -0500
backport log recovery performance improvements (#6395)
* Back porting improves tablet load times for tablet w/ walogs and no data
#4873
This commit makes two major changes. First it changed log recovery to use
block caches. Second it checks if a tablet has any data in walogs before
acquiring the recovery lock. These two changes together really speed up loading
tablets that have no data in walogs. These changes introduce an extra opening
of the walogs to see if the recovery lock needs to be acquired. Using the block
caches for this extra opening should avoid any extra cost. The block caches
also help in the case where [...]
In some simple test saw an 8x speedup in tablet load times.
Anytime a tablet has an unclean shutdown it will have the walogs of the
dead tserver assigned to it even if had no data in those walogs. These change
make loading tablets in that situation much faster.
* Back porting switches volumes prior to recovery check (#4889)
In #4873 a check was added to inspect walogs during tablet load to see
if they had any data for the tablet. This check happens prior to
volume replacement that also runs during tablet load. Therefore if
volume replacement is needed for the walogs then this check will fail
because it can not find the files and the tablet will fail to load.
To fix this problem modified the new check to switch volumes if needed
prior to running the check.
* Back porting Avoids listing the sorted logs dir multiple times during log
recovery. (#4874)
The log recovery code would list the sorted walog files multiple times
during recovery. These changes modify the code to only list the files
once. Also the listing is cached for a short period of time to improve
the case of multiple tablet referencing the same walogs. This along
with #4873 should result in much less traffic to the namenode when an
entire accumulo cluster shutsdown and needs to recover.
{"fundingSource": "41203", "team": "FED.ICGSA.OPS.MOE", "fshGit":
"dummy-lo", "fshDocker": "sha256:20cf0045"}
---------
Co-authored-by: andrew mcdonald <[email protected]>
---
.../org/apache/accumulo/server/fs/VolumeUtil.java | 4 +
.../apache/accumulo/tserver/AssignmentHandler.java | 6 +-
.../org/apache/accumulo/tserver/TabletServer.java | 62 ++++++----
.../accumulo/tserver/log/RecoveryLogsIterator.java | 92 +++++++--------
.../accumulo/tserver/log/ResolvedSortedLog.java | 130 +++++++++++++++++++++
.../accumulo/tserver/log/SortedLogRecovery.java | 65 +++++++----
.../accumulo/tserver/log/TabletServerLogger.java | 58 ++++++++-
.../apache/accumulo/tserver/logger/LogReader.java | 5 +-
.../tserver/log/RecoveryLogsIteratorTest.java | 25 ++--
.../tserver/log/SortedLogRecoveryTest.java | 26 ++++-
10 files changed, 357 insertions(+), 116 deletions(-)
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
index 35e1fbe1bc..7fd7b98bf1 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java
@@ -89,6 +89,10 @@ public class VolumeUtil {
return null;
}
+ public static LogEntry switchVolume(LogEntry le, List<Pair<Path,Path>>
replacements) {
+ return switchVolumes(le, replacements);
+ }
+
protected static LogEntry switchVolumes(LogEntry le, List<Pair<Path,Path>>
replacements) {
Path switchedPath = switchVolume(le.filename, FileType.WAL, replacements);
String switchedString;
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java
index e6154e1392..a4e66d32f0 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java
@@ -156,9 +156,7 @@ class AssignmentHandler implements Runnable {
Tablet tablet = null;
boolean successful = false;
- try {
- server.acquireRecoveryMemory(extent);
-
+ try (var recoveryMemory = server.acquireRecoveryMemory(tabletMetadata)) {
TabletResourceManager trm =
server.resourceManager.createTabletResourceManager(extent,
server.getTableConfiguration(extent));
TabletData data = new TabletData(tabletMetadata);
@@ -205,8 +203,6 @@ class AssignmentHandler implements Runnable {
TableId tableId = extent.tableId();
ProblemReports.getInstance(server.getContext()).report(new
ProblemReport(tableId, TABLET_LOAD,
extent.getUUID().toString(), server.getClientAddressString(), e));
- } finally {
- server.releaseRecoveryMemory(extent);
}
if (successful) {
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index ecb2687180..8f7a23582f 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -29,6 +29,7 @@ import static
org.apache.accumulo.core.util.threads.ThreadPools.watchCriticalSch
import static
org.apache.accumulo.core.util.threads.ThreadPools.watchNonCriticalScheduledTask;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.lang.management.ManagementFactory;
import java.net.UnknownHostException;
import java.security.SecureRandom;
@@ -89,6 +90,7 @@ import
org.apache.accumulo.core.master.thrift.TabletServerStatus;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
import org.apache.accumulo.core.metrics.MetricsInfo;
import org.apache.accumulo.core.process.thrift.ServerProcessService;
@@ -119,10 +121,9 @@ import
org.apache.accumulo.server.compaction.CompactionWatcher;
import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.fs.VolumeChooserEnvironmentImpl;
import org.apache.accumulo.server.fs.VolumeManager;
-import org.apache.accumulo.server.log.SortedLogState;
+import org.apache.accumulo.server.fs.VolumeUtil;
import org.apache.accumulo.server.log.WalStateManager;
import org.apache.accumulo.server.log.WalStateManager.WalMarkerException;
-import org.apache.accumulo.server.manager.recovery.RecoveryPath;
import org.apache.accumulo.server.rpc.ServerAddress;
import org.apache.accumulo.server.rpc.TServerUtils;
import org.apache.accumulo.server.rpc.ThriftProcessorTypes;
@@ -547,15 +548,44 @@ public class TabletServer extends AbstractServer
managerMessages.addLast(m);
}
- void acquireRecoveryMemory(KeyExtent extent) {
- if (!extent.isMeta()) {
+ private static final AutoCloseable NOOP_CLOSEABLE = () -> {};
+
+ AutoCloseable acquireRecoveryMemory(TabletMetadata tabletMetadata) {
+ if (tabletMetadata.getExtent().isMeta() || !needsRecovery(tabletMetadata))
{
+ return NOOP_CLOSEABLE;
+ } else {
recoveryLock.lock();
+ return () -> recoveryLock.unlock();
}
}
- void releaseRecoveryMemory(KeyExtent extent) {
- if (!extent.isMeta()) {
- recoveryLock.unlock();
+ public boolean needsRecovery(TabletMetadata tabletMetadata) {
+ var logEntries = tabletMetadata.getLogs();
+
+ if (logEntries.isEmpty()) {
+ return false;
+ }
+
+ // This method is called prior to volumes being switched for a tablet
during the load process,
+ // so switch volumes before calling needsRecovery()
+ var switchedLogEntries = new ArrayList<LogEntry>(logEntries.size());
+
+ for (LogEntry logEntry : logEntries) {
+ var switchedWalog = VolumeUtil.switchVolume(logEntry,
context.getVolumeReplacements());
+ LogEntry walog;
+ if (switchedWalog != null) {
+ log.debug("Volume switched for needsRecovery {} -> {}", logEntry,
switchedWalog);
+ walog = switchedWalog;
+ } else {
+ walog = logEntry;
+ }
+ switchedLogEntries.add(walog);
+ }
+
+ try {
+ return logger.needsRecovery(getContext(), tabletMetadata.getExtent(),
switchedLogEntries);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
}
}
@@ -1250,24 +1280,10 @@ public class TabletServer extends AbstractServer
public void recover(VolumeManager fs, KeyExtent extent, List<LogEntry>
logEntries,
Set<String> tabletFiles, MutationReceiver mutationReceiver) throws
IOException {
- List<Path> recoveryDirs = new ArrayList<>();
List<LogEntry> sorted = new ArrayList<>(logEntries);
sorted.sort((e1, e2) -> (int) (e1.timestamp - e2.timestamp));
- for (LogEntry entry : sorted) {
- Path recovery = null;
- Path finished = RecoveryPath.getRecoveryPath(new Path(entry.filename));
- finished = SortedLogState.getFinishedMarkerPath(finished);
- TabletServer.log.debug("Looking for " + finished);
- if (fs.exists(finished)) {
- recovery = finished.getParent();
- }
- if (recovery == null) {
- throw new IOException(
- "Unable to find recovery files for extent " + extent + " logEntry:
" + entry);
- }
- recoveryDirs.add(recovery);
- }
- logger.recover(getContext(), extent, recoveryDirs, tabletFiles,
mutationReceiver);
+
+ logger.recover(getContext(), extent, sorted, tabletFiles,
mutationReceiver);
}
public int createLogId() {
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
index b56612314a..af13b11ffb 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
@@ -22,11 +22,10 @@ import java.io.IOException;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
-import java.util.TreeMap;
+import java.util.SortedSet;
import org.apache.accumulo.core.crypto.CryptoEnvironmentImpl;
import org.apache.accumulo.core.data.Key;
@@ -34,21 +33,20 @@ import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.file.FileOperations;
import org.apache.accumulo.core.file.FileSKVIterator;
+import org.apache.accumulo.core.file.blockfile.impl.CacheProvider;
import org.apache.accumulo.core.iterators.IteratorAdapter;
import org.apache.accumulo.core.spi.crypto.CryptoEnvironment;
import org.apache.accumulo.core.spi.crypto.CryptoService;
import org.apache.accumulo.server.ServerContext;
-import org.apache.accumulo.server.fs.VolumeManager;
-import org.apache.accumulo.server.log.SortedLogState;
import org.apache.accumulo.tserver.logger.LogEvents;
import org.apache.accumulo.tserver.logger.LogFileKey;
import org.apache.accumulo.tserver.logger.LogFileValue;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.cache.Cache;
import com.google.common.collect.Iterators;
/**
@@ -63,11 +61,17 @@ public class RecoveryLogsIterator
private final Iterator<Entry<Key,Value>> iter;
private final CryptoEnvironment env = new
CryptoEnvironmentImpl(CryptoEnvironment.Scope.RECOVERY);
+ public RecoveryLogsIterator(ServerContext context, List<ResolvedSortedLog>
recoveryLogDirs,
+ LogFileKey start, LogFileKey end, boolean checkFirstKey) throws
IOException {
+ this(context, recoveryLogDirs, start, end, checkFirstKey, null, null);
+ }
+
/**
* Scans the files in each recoveryLogDir over the range [start,end].
*/
- public RecoveryLogsIterator(ServerContext context, List<Path>
recoveryLogDirs, LogFileKey start,
- LogFileKey end, boolean checkFirstKey) throws IOException {
+ public RecoveryLogsIterator(ServerContext context, List<ResolvedSortedLog>
recoveryLogDirs,
+ LogFileKey start, LogFileKey end, boolean checkFirstKey,
Cache<String,Long> fileLenCache,
+ CacheProvider cacheProvider) throws IOException {
List<Iterator<Entry<Key,Value>>> iterators = new
ArrayList<>(recoveryLogDirs.size());
fileIters = new ArrayList<>();
@@ -77,33 +81,31 @@ public class RecoveryLogsIterator
final CryptoService cryptoService =
context.getCryptoFactory().getService(env,
context.getConfiguration().getAllCryptoProperties());
- for (Path logDir : recoveryLogDirs) {
- LOG.debug("Opening recovery log dir {}", logDir.getName());
- TreeMap<Path,FileStatus> logFiles = getFiles(vm, logDir);
- var fs = vm.getFileSystemByPath(logDir);
+ for (ResolvedSortedLog logDir : recoveryLogDirs) {
+ LOG.debug("Opening recovery log dir {}", logDir.getDir().getName());
+ SortedSet<Path> logFiles = logDir.getChildren();
+ var fs = vm.getFileSystemByPath(logDir.getDir());
// only check the first key once to prevent extra iterator creation and
seeking
if (checkFirstKey && !logFiles.isEmpty()) {
- validateFirstKey(context, cryptoService, fs, logFiles, logDir);
+ validateFirstKey(context, cryptoService, fs, logFiles,
logDir.getDir(), fileLenCache,
+ cacheProvider);
}
- for (Entry<Path,FileStatus> entry : logFiles.entrySet()) {
- FileSKVIterator fileIter =
FileOperations.getInstance().newReaderBuilder()
- .forFile(entry.getKey().toString(), fs, fs.getConf(),
cryptoService, entry.getValue())
-
.withTableConfiguration(context.getConfiguration()).seekToBeginning().build();
+ for (Path log : logFiles) {
+ FileSKVIterator fileIter =
+ openLogFile(context, log, cryptoService, fs, fileLenCache,
cacheProvider);
if (range != null) {
fileIter.seek(range, Collections.emptySet(), false);
}
Iterator<Entry<Key,Value>> scanIter = new IteratorAdapter(fileIter);
if (scanIter.hasNext()) {
- LOG.debug("Write ahead log {} has data in range {} {}",
entry.getKey().getName(), start,
- end);
+ LOG.debug("Write ahead log {} has data in range {} {}",
log.getName(), start, end);
iterators.add(scanIter);
fileIters.add(fileIter);
} else {
- LOG.debug("Write ahead log {} has no data in range {} {}",
entry.getKey().getName(),
- start, end);
+ LOG.debug("Write ahead log {} has no data in range {} {}",
log.getName(), start, end);
fileIter.close();
}
}
@@ -135,46 +137,32 @@ public class RecoveryLogsIterator
}
}
- /**
- * Check for sorting signal files (finished/failed) and get the logs in the
provided directory.
- */
- private TreeMap<Path,FileStatus> getFiles(VolumeManager fs, Path directory)
throws IOException {
- boolean foundFinish = false;
- // Path::getName compares the last component of each Path value. In this
case, the last
- // component should
- // always have the format 'part-r-XXXXX.rf', where XXXXX are one-up values.
- TreeMap<Path,FileStatus> logFiles = new
TreeMap<>(Comparator.comparing(Path::getName));
- for (FileStatus child : fs.listStatus(directory)) {
- if (child.getPath().getName().startsWith("_")) {
- continue;
- }
- if (SortedLogState.isFinished(child.getPath().getName())) {
- foundFinish = true;
- continue;
- }
- if (SortedLogState.FAILED.getMarker().equals(child.getPath().getName()))
{
- continue;
- }
- FileSystem ns = fs.getFileSystemByPath(child.getPath());
- Path fullLogPath = ns.makeQualified(child.getPath());
- logFiles.put(fullLogPath, child);
+ FileSKVIterator openLogFile(ServerContext context, Path logFile,
CryptoService cs, FileSystem fs,
+ Cache<String,Long> fileLenCache, CacheProvider cacheProvider) throws
IOException {
+ var builder = FileOperations.getInstance().newReaderBuilder()
+ .forFile(logFile.toString(), fs, fs.getConf(), cs)
+ .withTableConfiguration(context.getConfiguration());
+
+ if (fileLenCache != null) {
+ builder = builder.withFileLenCache(fileLenCache);
}
- if (!foundFinish) {
- throw new IOException(
- "Sort '" + SortedLogState.FINISHED.getMarker() + "' flag not found
in " + directory);
+
+ if (cacheProvider != null) {
+ builder = builder.withCacheProvider(cacheProvider);
}
- return logFiles;
+
+ return builder.seekToBeginning().build();
}
/**
* Check that the first entry in the WAL is OPEN. Only need to do this once.
*/
private void validateFirstKey(ServerContext context, CryptoService cs,
FileSystem fs,
- TreeMap<Path,FileStatus> logFiles, Path fullLogPath) throws IOException {
- Entry<Path,FileStatus> first = logFiles.firstEntry();
- try (FileSKVIterator fileIter =
FileOperations.getInstance().newReaderBuilder()
- .forFile(first.getKey().toString(), fs, fs.getConf(), cs,
first.getValue())
-
.withTableConfiguration(context.getConfiguration()).seekToBeginning().build()) {
+ SortedSet<Path> logFiles, Path fullLogPath, Cache<String,Long>
fileLenCache,
+ CacheProvider cacheProvider) throws IOException {
+ Path first = logFiles.first();
+ try (FileSKVIterator fileIter =
+ openLogFile(context, first, cs, fs, fileLenCache, cacheProvider)) {
Iterator<Entry<Key,Value>> iterator = new IteratorAdapter(fileIter);
if (iterator.hasNext()) {
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java
new file mode 100644
index 0000000000..91041f9396
--- /dev/null
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.tserver.log;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
+import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.accumulo.server.log.SortedLogState;
+import org.apache.accumulo.server.manager.recovery.RecoveryPath;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * Write ahead logs have two paths in DFS. There is the path of the original
unsorted walog and the
+ * path of the sorted walog. The purpose of this class is to convert the
unsorted wal path to a
+ * sorted wal path and validate the sorted dir exists and is finished.
+ */
+public class ResolvedSortedLog {
+
+ private final SortedSet<Path> children;
+ private final LogEntry origin;
+ private final Path sortedLogDir;
+
+ private ResolvedSortedLog(LogEntry origin, Path sortedLogDir,
SortedSet<Path> children) {
+ this.origin = origin;
+ this.sortedLogDir = sortedLogDir;
+ this.children = Collections.unmodifiableSortedSet(children);
+ }
+
+ /**
+ * @return the unsorted walog path from which this was created.
+ */
+ public LogEntry getOrigin() {
+ return origin;
+ }
+
+ /**
+ * @return the path of the directory in which sorted logs are stored
+ */
+ public Path getDir() {
+ return sortedLogDir;
+ }
+
+ /**
+ * @return When an unsorted walog is sorted the sorted data is stored in one
or more rfiles, this
+ * returns the paths of those rfiles.
+ */
+ public SortedSet<Path> getChildren() {
+ return children;
+ }
+
+ @Override
+ public String toString() {
+ return sortedLogDir.toString();
+ }
+
+ /**
+ * For a given path of an unsorted walog check to see if the corresponding
sorted log dir exists
+ * and is finished. If it is return an immutable object containing
information about the sorted
+ * walogs.
+ */
+ public static ResolvedSortedLog resolve(LogEntry logEntry, VolumeManager fs)
throws IOException {
+
+ // convert the path of an unsorted log to the expected path for the
corresponding sorted log
+ // dir
+ Path sortedLogPath = RecoveryPath.getRecoveryPath(new
Path(logEntry.filename));
+
+ return resolve(logEntry, sortedLogPath, fs);
+ }
+
+ /**
+ * Resolve a sorted log directory path. This is useful for diagnostic tools
that operate directly
+ * on sorted recovery logs without going through the normal recovery flow
with LogEntry objects.
+ */
+ public static ResolvedSortedLog resolve(Path sortedLogDir, VolumeManager fs)
throws IOException {
+ return resolve(new LogEntry(null, 0, sortedLogDir.toString()),
sortedLogDir, fs);
+ }
+
+ private static ResolvedSortedLog resolve(LogEntry logEntry, Path
sortedLogPath, VolumeManager fs)
+ throws IOException {
+ boolean foundFinish = false;
+ // Path::getName compares the last component of each Path value. In this
case, the last
+ // component should
+ // always have the format 'part-r-XXXXX.rf', where XXXXX are one-up values.
+ SortedSet<Path> logFiles = new
TreeSet<>(Comparator.comparing(Path::getName));
+ for (FileStatus child : fs.listStatus(sortedLogPath)) {
+ if (child.getPath().getName().startsWith("_")) {
+ continue;
+ }
+ if (SortedLogState.isFinished(child.getPath().getName())) {
+ foundFinish = true;
+ continue;
+ }
+ if (SortedLogState.FAILED.getMarker().equals(child.getPath().getName()))
{
+ continue;
+ }
+ FileSystem ns = fs.getFileSystemByPath(child.getPath());
+ Path fullLogPath = ns.makeQualified(child.getPath());
+ logFiles.add(fullLogPath);
+ }
+ if (!foundFinish) {
+ throw new IOException("Sort '" + SortedLogState.FINISHED.getMarker() +
"' flag not found in "
+ + sortedLogPath + " for walog " + logEntry.filename);
+ }
+
+ return new ResolvedSortedLog(logEntry, sortedLogPath, logFiles);
+ }
+}
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java
index 40e8496ba7..71210de120 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java
@@ -42,6 +42,7 @@ import java.util.Set;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.file.blockfile.impl.CacheProvider;
import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.tserver.logger.LogEvents;
@@ -51,6 +52,7 @@ import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.cache.Cache;
import com.google.common.collect.Collections2;
import com.google.common.collect.Iterators;
import com.google.common.collect.PeekingIterator;
@@ -65,8 +67,23 @@ public class SortedLogRecovery {
private final ServerContext context;
- public SortedLogRecovery(ServerContext context) {
+ private final CacheProvider cacheProvider;
+
+ private final Cache<String,Long> fileLenCache;
+
+ public SortedLogRecovery(ServerContext context, Cache<String,Long>
fileLenCache,
+ CacheProvider cacheProvider) {
this.context = context;
+ this.cacheProvider = cacheProvider;
+ this.fileLenCache = fileLenCache;
+ }
+
+ public boolean needsRecovery(KeyExtent extent, List<ResolvedSortedLog>
recoveryDirs)
+ throws IOException {
+ Entry<Integer,List<ResolvedSortedLog>> maxEntry =
+ findLogsThatDefineTablet(extent, recoveryDirs);
+ int tabletId = maxEntry.getKey();
+ return tabletId != -1;
}
static LogFileKey maxKey(LogEvents event) {
@@ -100,11 +117,12 @@ public class SortedLogRecovery {
return key;
}
- private int findMaxTabletId(KeyExtent extent, List<Path> recoveryLogDirs)
throws IOException {
+ private int findMaxTabletId(KeyExtent extent, List<ResolvedSortedLog>
recoveryLogDirs)
+ throws IOException {
int tabletId = -1;
try (var rli = new RecoveryLogsIterator(context, recoveryLogDirs,
minKey(DEFINE_TABLET),
- maxKey(DEFINE_TABLET), true)) {
+ maxKey(DEFINE_TABLET), true, fileLenCache, cacheProvider)) {
KeyExtent alternative = extent;
if (extent.isRootTablet()) {
@@ -140,18 +158,18 @@ public class SortedLogRecovery {
* @return The maximum tablet ID observed AND the list of logs that
contained the maximum tablet
* ID.
*/
- private Entry<Integer,List<Path>> findLogsThatDefineTablet(KeyExtent extent,
- List<Path> recoveryDirs) throws IOException {
- Map<Integer,List<Path>> logsThatDefineTablet = new HashMap<>();
+ private Entry<Integer,List<ResolvedSortedLog>>
findLogsThatDefineTablet(KeyExtent extent,
+ List<ResolvedSortedLog> recoveryDirs) throws IOException {
+ Map<Integer,List<ResolvedSortedLog>> logsThatDefineTablet = new
HashMap<>();
- for (Path walDir : recoveryDirs) {
+ for (ResolvedSortedLog walDir : recoveryDirs) {
int tabletId = findMaxTabletId(extent,
Collections.singletonList(walDir));
if (tabletId == -1) {
- log.debug("Did not find tablet {} in recovery log {}", extent,
walDir.getName());
+ log.debug("Did not find tablet {} in recovery log {}", extent,
walDir.getDir().getName());
} else {
logsThatDefineTablet.computeIfAbsent(tabletId, k -> new
ArrayList<>()).add(walDir);
log.debug("Found tablet {} with id {} in recovery log {}", extent,
tabletId,
- walDir.getName());
+ walDir.getDir().getName());
}
}
@@ -196,8 +214,8 @@ public class SortedLogRecovery {
}
- private long findRecoverySeq(List<Path> recoveryLogs, Set<String>
tabletFiles, int tabletId)
- throws IOException {
+ private long findRecoverySeq(List<ResolvedSortedLog> recoveryLogs,
Set<String> tabletFiles,
+ int tabletId) throws IOException {
HashSet<String> suffixes = new HashSet<>();
for (String path : tabletFiles) {
suffixes.add(getPathSuffix(path));
@@ -207,8 +225,9 @@ public class SortedLogRecovery {
long lastFinish = 0;
long recoverySeq = 0;
- try (RecoveryLogsIterator rli = new RecoveryLogsIterator(context,
recoveryLogs,
- minKey(COMPACTION_START, tabletId), maxKey(COMPACTION_START,
tabletId), false)) {
+ try (RecoveryLogsIterator rli =
+ new RecoveryLogsIterator(context, recoveryLogs,
minKey(COMPACTION_START, tabletId),
+ maxKey(COMPACTION_START, tabletId), false, fileLenCache,
cacheProvider)) {
DeduplicatingIterator ddi = new DeduplicatingIterator(rli);
@@ -258,14 +277,15 @@ public class SortedLogRecovery {
return recoverySeq;
}
- private void playbackMutations(List<Path> recoveryLogs, MutationReceiver mr,
int tabletId,
- long recoverySeq) throws IOException {
+ private void playbackMutations(List<ResolvedSortedLog> recoveryLogs,
MutationReceiver mr,
+ int tabletId, long recoverySeq) throws IOException {
LogFileKey start = minKey(MUTATION, tabletId);
start.setSeq(recoverySeq);
LogFileKey end = maxKey(MUTATION, tabletId);
- try (var rli = new RecoveryLogsIterator(context, recoveryLogs, start, end,
false)) {
+ try (var rli = new RecoveryLogsIterator(context, recoveryLogs, start, end,
false, fileLenCache,
+ cacheProvider)) {
while (rli.hasNext()) {
Entry<LogFileKey,LogFileValue> entry = rli.next();
LogFileKey logFileKey = entry.getKey();
@@ -286,20 +306,21 @@ public class SortedLogRecovery {
}
}
- Collection<String> asNames(List<Path> recoveryLogs) {
- return Collections2.transform(recoveryLogs, Path::getName);
+ Collection<String> asNames(List<ResolvedSortedLog> recoveryLogs) {
+ return Collections2.transform(recoveryLogs, rl -> rl.getDir().getName());
}
- public void recover(KeyExtent extent, List<Path> recoveryDirs, Set<String>
tabletFiles,
- MutationReceiver mr) throws IOException {
+ public void recover(KeyExtent extent, List<ResolvedSortedLog> recoveryDirs,
+ Set<String> tabletFiles, MutationReceiver mr) throws IOException {
- Entry<Integer,List<Path>> maxEntry = findLogsThatDefineTablet(extent,
recoveryDirs);
+ Entry<Integer,List<ResolvedSortedLog>> maxEntry =
+ findLogsThatDefineTablet(extent, recoveryDirs);
// A tablet may leave a tserver and then come back, in which case it would
have a different and
// higher tablet id. Only want to consider events in the log related to
the last time the tablet
// was loaded.
int tabletId = maxEntry.getKey();
- List<Path> logsThatDefineTablet = maxEntry.getValue();
+ List<ResolvedSortedLog> logsThatDefineTablet = maxEntry.getValue();
if (tabletId == -1) {
log.info("Tablet {} is not defined in recovery logs {} ", extent,
asNames(recoveryDirs));
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
index caa6ace170..1a9968bcb9 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java
@@ -23,6 +23,7 @@ import static
org.apache.accumulo.core.util.threads.ThreadPoolNames.TSERVER_WAL_
import java.io.IOException;
import java.nio.channels.ClosedChannelException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -40,7 +41,12 @@ import org.apache.accumulo.core.client.Durability;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.fate.zookeeper.ServiceLock;
+import org.apache.accumulo.core.file.blockfile.impl.BasicCacheProvider;
+import org.apache.accumulo.core.file.blockfile.impl.CacheProvider;
+import org.apache.accumulo.core.logging.LoggingBlockCache;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
+import org.apache.accumulo.core.spi.cache.CacheType;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
import org.apache.accumulo.core.util.Halt;
import org.apache.accumulo.core.util.Retry;
import org.apache.accumulo.core.util.Retry.RetryFactory;
@@ -51,6 +57,7 @@ import
org.apache.accumulo.server.replication.proto.Replication.Status;
import org.apache.accumulo.server.util.ReplicationTableUtil;
import org.apache.accumulo.tserver.TabletMutations;
import org.apache.accumulo.tserver.TabletServer;
+import org.apache.accumulo.tserver.TabletServerResourceManager;
import org.apache.accumulo.tserver.log.DfsLogger.LoggerOperation;
import org.apache.accumulo.tserver.log.DfsLogger.ServerResources;
import org.apache.accumulo.tserver.tablet.CommitSession;
@@ -58,6 +65,9 @@ import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+
/**
* Central logging facility for the TServerInfo.
*
@@ -76,6 +86,9 @@ public class TabletServerLogger {
private final TabletServer tserver;
+ // Cache for resolved sorted logs to avoid repeated expensive I/O operations
+ private final Cache<LogEntry,ResolvedSortedLog> sortedLogCache;
+
// The current logger
private DfsLogger currentLog = null;
private final SynchronousQueue<Object> nextLog = new SynchronousQueue<>();
@@ -157,6 +170,8 @@ public class TabletServerLogger {
this.createRetry = null;
this.writeRetryFactory = writeRetryFactory;
this.maxAge = maxAge;
+ this.sortedLogCache =
+ CacheBuilder.newBuilder().expireAfterWrite(1,
TimeUnit.MINUTES).maximumSize(10_000).build();
}
private DfsLogger initializeLoggers(final AtomicInteger logIdOut) throws
IOException {
@@ -558,11 +573,48 @@ public class TabletServerLogger {
return seq;
}
- public void recover(ServerContext context, KeyExtent extent, List<Path>
recoveryDirs,
+ private List<ResolvedSortedLog> resolve(Collection<LogEntry> walogs) throws
IOException {
+ List<ResolvedSortedLog> sortedLogs = new ArrayList<>(walogs.size());
+ VolumeManager fs = tserver.getContext().getVolumeManager();
+ for (var logEntry : walogs) {
+ try {
+ ResolvedSortedLog resolvedLog =
+ sortedLogCache.get(logEntry, () ->
ResolvedSortedLog.resolve(logEntry, fs));
+ sortedLogs.add(resolvedLog);
+ } catch (Exception e) {
+ throw new IOException("Failed to resolve sorted log for " +
logEntry.filename, e);
+ }
+ }
+ return sortedLogs;
+ }
+
+ private CacheProvider createCacheProvider(TabletServerResourceManager
resourceMgr) {
+ return new BasicCacheProvider(
+ LoggingBlockCache.wrap(CacheType.INDEX, resourceMgr.getIndexCache()),
+ LoggingBlockCache.wrap(CacheType.DATA, resourceMgr.getDataCache()));
+ }
+
+ public boolean needsRecovery(ServerContext context, KeyExtent extent,
Collection<LogEntry> walogs)
+ throws IOException {
+ try {
+ var resourceMgr = tserver.getResourceManager();
+ var cacheProvider = createCacheProvider(resourceMgr);
+ SortedLogRecovery recovery =
+ new SortedLogRecovery(context, resourceMgr.getFileLenCache(),
cacheProvider);
+ return recovery.needsRecovery(extent, resolve(walogs));
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ public void recover(ServerContext context, KeyExtent extent,
Collection<LogEntry> walogs,
Set<String> tabletFiles, MutationReceiver mr) throws IOException {
try {
- SortedLogRecovery recovery = new SortedLogRecovery(context);
- recovery.recover(extent, recoveryDirs, tabletFiles, mr);
+ var resourceMgr = tserver.getResourceManager();
+ var cacheProvider = createCacheProvider(resourceMgr);
+ SortedLogRecovery recovery =
+ new SortedLogRecovery(context, resourceMgr.getFileLenCache(),
cacheProvider);
+ recovery.recover(extent, resolve(walogs), tabletFiles, mr);
} catch (Exception e) {
throw new IOException(e);
}
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
index 704a57804c..efbaa0c4c7 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
@@ -48,6 +48,7 @@ import org.apache.accumulo.start.spi.KeywordExecutable;
import org.apache.accumulo.tserver.log.DfsLogger;
import org.apache.accumulo.tserver.log.DfsLogger.LogHeaderIncompleteException;
import org.apache.accumulo.tserver.log.RecoveryLogsIterator;
+import org.apache.accumulo.tserver.log.ResolvedSortedLog;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
@@ -175,8 +176,8 @@ public class LogReader implements KeywordExecutable {
} else {
// read the log entries in a sorted RFile. This has to be a
directory that contains the
// finished file.
- try (var rli = new RecoveryLogsIterator(context,
Collections.singletonList(path), null,
- null, false)) {
+ try (var rli = new RecoveryLogsIterator(context,
+ Collections.singletonList(ResolvedSortedLog.resolve(path, fs)),
null, null, false)) {
while (rli.hasNext()) {
Entry<LogFileKey,LogFileValue> entry = rli.next();
printLogEvent(entry.getKey(), entry.getValue(), row, rowMatcher,
ke, tabletIds,
diff --git
a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java
b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java
index 75d933db37..13595ad16b 100644
---
a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java
+++
b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java
@@ -132,7 +132,8 @@ public class RecoveryLogsIteratorTest extends WithTestNames
{
createRecoveryDir(logs, dirs, true);
- try (RecoveryLogsIterator rli = new RecoveryLogsIterator(context, dirs,
null, null, false)) {
+ try (RecoveryLogsIterator rli =
+ new RecoveryLogsIterator(context, pathsToResolvedLogs(dirs), null,
null, false)) {
while (rli.hasNext()) {
Entry<LogFileKey,LogFileValue> entry = rli.next();
assertEquals(1, entry.getKey().getTabletId(), "TabletId does not
match");
@@ -159,7 +160,7 @@ public class RecoveryLogsIteratorTest extends WithTestNames
{
createRecoveryDir(logs, dirs, false);
assertThrows(IOException.class,
- () -> new RecoveryLogsIterator(context, dirs, null, null, false),
+ () -> new RecoveryLogsIterator(context, pathsToResolvedLogs(dirs),
null, null, false),
"Finish marker should not be found");
}
@@ -168,9 +169,10 @@ public class RecoveryLogsIteratorTest extends
WithTestNames {
String destPath = workDir + "/test.rf";
fs.create(new Path(destPath));
- assertThrows(
- IOException.class, () -> new RecoveryLogsIterator(context,
- Collections.singletonList(new Path(destPath)), null, null, false),
+ assertThrows(IOException.class,
+ () -> new RecoveryLogsIterator(context,
+ Collections.singletonList(ResolvedSortedLog.resolve(new
Path(destPath), fs)), null,
+ null, false),
"Finish marker should not be found for a single file.");
}
@@ -192,7 +194,7 @@ public class RecoveryLogsIteratorTest extends WithTestNames
{
createRecoveryDir(logs, dirs, true);
assertThrows(IllegalStateException.class,
- () -> new RecoveryLogsIterator(context, dirs, null, null, true),
+ () -> new RecoveryLogsIterator(context, pathsToResolvedLogs(dirs),
null, null, true),
"First log entry is not OPEN so exception should be thrown.");
}
@@ -219,7 +221,8 @@ public class RecoveryLogsIteratorTest extends WithTestNames
{
createRecoveryDir(logs, dirs, true);
- try (RecoveryLogsIterator rli = new RecoveryLogsIterator(context, dirs,
null, null, true)) {
+ try (RecoveryLogsIterator rli =
+ new RecoveryLogsIterator(context, pathsToResolvedLogs(dirs), null,
null, true)) {
while (rli.hasNext()) {
Entry<LogFileKey,LogFileValue> entry = rli.next();
assertNotNull(entry.getKey());
@@ -248,4 +251,12 @@ public class RecoveryLogsIteratorTest extends
WithTestNames {
dirs.add(new Path(destPath));
}
}
+
+ private List<ResolvedSortedLog> pathsToResolvedLogs(List<Path> dirs) throws
IOException {
+ List<ResolvedSortedLog> resolvedLogs = new ArrayList<>();
+ for (Path dir : dirs) {
+ resolvedLogs.add(ResolvedSortedLog.resolve(dir, fs));
+ }
+ return resolvedLogs;
+ }
}
diff --git
a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java
b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java
index 1e4a79c71c..801644d67c 100644
---
a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java
+++
b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java
@@ -51,10 +51,15 @@ import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.dataImpl.KeyExtent;
+import
org.apache.accumulo.core.file.blockfile.cache.impl.BlockCacheConfiguration;
+import org.apache.accumulo.core.file.blockfile.cache.tinylfu.TinyLfuBlockCache;
+import org.apache.accumulo.core.file.blockfile.impl.BasicCacheProvider;
+import org.apache.accumulo.core.file.blockfile.impl.CacheProvider;
import org.apache.accumulo.core.file.rfile.bcfile.Compression;
import org.apache.accumulo.core.file.rfile.bcfile.CompressionAlgorithm;
import org.apache.accumulo.core.file.rfile.bcfile.Utils;
import org.apache.accumulo.core.file.streams.SeekableDataInputStream;
+import org.apache.accumulo.core.spi.cache.CacheType;
import org.apache.accumulo.core.spi.crypto.CryptoServiceFactory;
import org.apache.accumulo.core.spi.crypto.GenericCryptoServiceFactory;
import org.apache.accumulo.core.util.Pair;
@@ -75,6 +80,9 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "paths not
set by user input")
@@ -165,6 +173,15 @@ public class SortedLogRecoveryTest extends WithTestNames {
private List<Mutation> recover(Map<String,KeyValue[]> logs, Set<String>
files, KeyExtent extent,
int bufferSize) throws IOException {
+ CacheProvider cacheProvider = new BasicCacheProvider(
+ new TinyLfuBlockCache(
+
BlockCacheConfiguration.forTabletServer(DefaultConfiguration.getInstance()),
+ CacheType.INDEX),
+ new TinyLfuBlockCache(
+
BlockCacheConfiguration.forTabletServer(DefaultConfiguration.getInstance()),
+ CacheType.DATA));
+ Cache<String,Long> fileLenCache = CacheBuilder.newBuilder().build();
+
final String workdir = new File(tempDir, testName()).getAbsolutePath();
try (var fs = VolumeManagerImpl.getLocalForTesting(workdir)) {
CryptoServiceFactory cryptoFactory = new GenericCryptoServiceFactory();
@@ -198,9 +215,14 @@ public class SortedLogRecoveryTest extends WithTestNames {
dirs.add(new Path(destPath));
}
// Recover
- SortedLogRecovery recovery = new SortedLogRecovery(context);
+ SortedLogRecovery recovery = new SortedLogRecovery(context,
fileLenCache, cacheProvider);
CaptureMutations capture = new CaptureMutations();
- recovery.recover(extent, dirs, files, capture);
+ // Convert Path objects to ResolvedSortedLog objects for recovery
+ List<ResolvedSortedLog> resolvedLogs = new ArrayList<>();
+ for (Path dir : dirs) {
+ resolvedLogs.add(ResolvedSortedLog.resolve(dir, fs));
+ }
+ recovery.recover(extent, resolvedLogs, files, capture);
verify(context);
return capture.result;
}
