| /* |
| * |
| * 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 |
| * |
| * http://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.bookkeeper.bookie; |
| |
| import static java.nio.charset.StandardCharsets.UTF_8; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.MoreObjects; |
| import com.google.common.collect.MapMaker; |
| import com.google.common.collect.Sets; |
| import io.netty.buffer.ByteBuf; |
| import io.netty.buffer.ByteBufAllocator; |
| import io.netty.buffer.PooledByteBufAllocator; |
| import io.netty.buffer.Unpooled; |
| import io.netty.util.concurrent.FastThreadLocal; |
| import java.io.BufferedReader; |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.FileNotFoundException; |
| import java.io.FilenameFilter; |
| import java.io.IOException; |
| import java.io.InputStreamReader; |
| import java.io.RandomAccessFile; |
| import java.nio.ByteBuffer; |
| import java.nio.channels.AsynchronousCloseException; |
| import java.nio.channels.FileChannel; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.SortedMap; |
| import java.util.TreeMap; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.CopyOnWriteArrayList; |
| import java.util.regex.Pattern; |
| import org.apache.bookkeeper.bookie.storage.CompactionEntryLog; |
| import org.apache.bookkeeper.bookie.storage.EntryLogScanner; |
| import org.apache.bookkeeper.bookie.storage.EntryLogger; |
| import org.apache.bookkeeper.conf.ServerConfiguration; |
| import org.apache.bookkeeper.stats.NullStatsLogger; |
| import org.apache.bookkeeper.stats.StatsLogger; |
| import org.apache.bookkeeper.util.DiskChecker; |
| import org.apache.bookkeeper.util.HardLink; |
| import org.apache.bookkeeper.util.IOUtils; |
| import org.apache.bookkeeper.util.collections.ConcurrentLongLongHashMap; |
| import org.apache.bookkeeper.util.collections.ConcurrentLongLongHashMap.BiConsumerLong; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * This class manages the writing of the bookkeeper entries. All the new |
| * entries are written to a common log. The LedgerCache will have pointers |
| * into files created by this class with offsets into the files to find |
| * the actual ledger entry. The entry log files created by this class are |
| * identified by a long. |
| */ |
| public class DefaultEntryLogger implements EntryLogger { |
| private static final Logger LOG = LoggerFactory.getLogger(DefaultEntryLogger.class); |
| |
| @VisibleForTesting |
| static final int UNINITIALIZED_LOG_ID = -0xDEAD; |
| |
| static class BufferedLogChannel extends BufferedChannel { |
| private final long logId; |
| private final EntryLogMetadata entryLogMetadata; |
| private final File logFile; |
| private long ledgerIdAssigned = UNASSIGNED_LEDGERID; |
| |
| public BufferedLogChannel(ByteBufAllocator allocator, FileChannel fc, int writeCapacity, int readCapacity, |
| long logId, File logFile, long unpersistedBytesBound) throws IOException { |
| super(allocator, fc, writeCapacity, readCapacity, unpersistedBytesBound); |
| this.logId = logId; |
| this.entryLogMetadata = new EntryLogMetadata(logId); |
| this.logFile = logFile; |
| } |
| public long getLogId() { |
| return logId; |
| } |
| |
| public File getLogFile() { |
| return logFile; |
| } |
| |
| public void registerWrittenEntry(long ledgerId, long entrySize) { |
| entryLogMetadata.addLedgerSize(ledgerId, entrySize); |
| } |
| |
| public ConcurrentLongLongHashMap getLedgersMap() { |
| return entryLogMetadata.getLedgersMap(); |
| } |
| |
| public Long getLedgerIdAssigned() { |
| return ledgerIdAssigned; |
| } |
| |
| public void setLedgerIdAssigned(Long ledgerId) { |
| this.ledgerIdAssigned = ledgerId; |
| } |
| |
| @Override |
| public String toString() { |
| return MoreObjects.toStringHelper(BufferedChannel.class) |
| .add("logId", logId) |
| .add("logFile", logFile) |
| .add("ledgerIdAssigned", ledgerIdAssigned) |
| .toString(); |
| } |
| |
| /** |
| * Append the ledger map at the end of the entry log. |
| * Updates the entry log file header with the offset and size of the map. |
| */ |
| void appendLedgersMap() throws IOException { |
| |
| long ledgerMapOffset = this.position(); |
| |
| ConcurrentLongLongHashMap ledgersMap = this.getLedgersMap(); |
| int numberOfLedgers = (int) ledgersMap.size(); |
| |
| // Write the ledgers map into several batches |
| |
| final int maxMapSize = LEDGERS_MAP_HEADER_SIZE + LEDGERS_MAP_ENTRY_SIZE * LEDGERS_MAP_MAX_BATCH_SIZE; |
| final ByteBuf serializedMap = ByteBufAllocator.DEFAULT.buffer(maxMapSize); |
| |
| try { |
| ledgersMap.forEach(new BiConsumerLong() { |
| int remainingLedgers = numberOfLedgers; |
| boolean startNewBatch = true; |
| int remainingInBatch = 0; |
| |
| @Override |
| public void accept(long ledgerId, long size) { |
| if (startNewBatch) { |
| int batchSize = Math.min(remainingLedgers, LEDGERS_MAP_MAX_BATCH_SIZE); |
| int ledgerMapSize = LEDGERS_MAP_HEADER_SIZE + LEDGERS_MAP_ENTRY_SIZE * batchSize; |
| |
| serializedMap.clear(); |
| serializedMap.writeInt(ledgerMapSize - 4); |
| serializedMap.writeLong(INVALID_LID); |
| serializedMap.writeLong(LEDGERS_MAP_ENTRY_ID); |
| serializedMap.writeInt(batchSize); |
| |
| startNewBatch = false; |
| remainingInBatch = batchSize; |
| } |
| // Dump the ledger in the current batch |
| serializedMap.writeLong(ledgerId); |
| serializedMap.writeLong(size); |
| --remainingLedgers; |
| |
| if (--remainingInBatch == 0) { |
| // Close current batch |
| try { |
| write(serializedMap); |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| |
| startNewBatch = true; |
| } |
| } |
| }); |
| } catch (RuntimeException e) { |
| if (e.getCause() instanceof IOException) { |
| throw (IOException) e.getCause(); |
| } else { |
| throw e; |
| } |
| } finally { |
| serializedMap.release(); |
| } |
| // Flush the ledger's map out before we write the header. |
| // Otherwise the header might point to something that is not fully |
| // written |
| super.flush(); |
| |
| // Update the headers with the map offset and count of ledgers |
| ByteBuffer mapInfo = ByteBuffer.allocate(8 + 4); |
| mapInfo.putLong(ledgerMapOffset); |
| mapInfo.putInt(numberOfLedgers); |
| mapInfo.flip(); |
| this.fileChannel.write(mapInfo, LEDGERS_MAP_OFFSET_POSITION); |
| } |
| } |
| |
| private final LedgerDirsManager ledgerDirsManager; |
| private final boolean entryLogPerLedgerEnabled; |
| |
| final RecentEntryLogsStatus recentlyCreatedEntryLogsStatus; |
| |
| /** |
| * locks for compaction log. |
| */ |
| private final Object compactionLogLock = new Object(); |
| |
| private volatile BufferedLogChannel compactionLogChannel; |
| |
| final EntryLoggerAllocator entryLoggerAllocator; |
| private final EntryLogManager entryLogManager; |
| |
| private final CopyOnWriteArrayList<EntryLogListener> listeners = new CopyOnWriteArrayList<EntryLogListener>(); |
| |
| private static final int HEADER_V0 = 0; // Old log file format (no ledgers map index) |
| private static final int HEADER_V1 = 1; // Introduced ledger map index |
| static final int HEADER_CURRENT_VERSION = HEADER_V1; |
| |
| private static class Header { |
| final int version; |
| final long ledgersMapOffset; |
| final int ledgersCount; |
| |
| Header(int version, long ledgersMapOffset, int ledgersCount) { |
| this.version = version; |
| this.ledgersMapOffset = ledgersMapOffset; |
| this.ledgersCount = ledgersCount; |
| } |
| } |
| |
| /** |
| * The 1K block at the head of the entry logger file |
| * that contains the fingerprint and meta-data. |
| * |
| * <pre> |
| * Header is composed of: |
| * Fingerprint: 4 bytes "BKLO" |
| * Log file HeaderVersion enum: 4 bytes |
| * Ledger map offset: 8 bytes |
| * Ledgers Count: 4 bytes |
| * </pre> |
| */ |
| static final int LOGFILE_HEADER_SIZE = 1024; |
| final ByteBuf logfileHeader = Unpooled.buffer(LOGFILE_HEADER_SIZE); |
| |
| static final int HEADER_VERSION_POSITION = 4; |
| static final int LEDGERS_MAP_OFFSET_POSITION = HEADER_VERSION_POSITION + 4; |
| |
| /** |
| * Ledgers map is composed of multiple parts that can be split into separated entries. Each of them is composed of: |
| * |
| * <pre> |
| * length: (4 bytes) [0-3] |
| * ledger id (-1): (8 bytes) [4 - 11] |
| * entry id: (8 bytes) [12-19] |
| * num ledgers stored in current metadata entry: (4 bytes) [20 - 23] |
| * ledger entries: sequence of (ledgerid, size) (8 + 8 bytes each) [24..] |
| * </pre> |
| */ |
| static final int LEDGERS_MAP_HEADER_SIZE = 4 + 8 + 8 + 4; |
| static final int LEDGERS_MAP_ENTRY_SIZE = 8 + 8; |
| |
| // Break the ledgers map into multiple batches, each of which can contain up to 10K ledgers |
| static final int LEDGERS_MAP_MAX_BATCH_SIZE = 10000; |
| |
| static final long INVALID_LID = -1L; |
| |
| // EntryId used to mark an entry (belonging to INVALID_ID) as a component of the serialized ledgers map |
| static final long LEDGERS_MAP_ENTRY_ID = -2L; |
| |
| static final int MIN_SANE_ENTRY_SIZE = 8 + 8; |
| static final long MB = 1024 * 1024; |
| |
| private final int maxSaneEntrySize; |
| |
| private final ByteBufAllocator allocator; |
| |
| final ServerConfiguration conf; |
| |
| /** |
| * Entry Log Listener. |
| */ |
| interface EntryLogListener { |
| /** |
| * Rotate a new entry log to write. |
| */ |
| void onRotateEntryLog(); |
| } |
| |
| public DefaultEntryLogger(ServerConfiguration conf) throws IOException { |
| this(conf, new LedgerDirsManager(conf, conf.getLedgerDirs(), |
| new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()))); |
| } |
| |
| /** |
| * Create an EntryLogger that stores it's log files in the given directories. |
| */ |
| public DefaultEntryLogger(ServerConfiguration conf, |
| LedgerDirsManager ledgerDirsManager) throws IOException { |
| this(conf, ledgerDirsManager, null, NullStatsLogger.INSTANCE, PooledByteBufAllocator.DEFAULT); |
| } |
| |
| public DefaultEntryLogger(ServerConfiguration conf, |
| LedgerDirsManager ledgerDirsManager, EntryLogListener listener, StatsLogger statsLogger, |
| ByteBufAllocator allocator) throws IOException { |
| //We reserve 500 bytes as overhead for the protocol. This is not 100% accurate |
| // but the protocol varies so an exact value is difficult to determine |
| this.maxSaneEntrySize = conf.getNettyMaxFrameSizeBytes() - 500; |
| this.allocator = allocator; |
| this.ledgerDirsManager = ledgerDirsManager; |
| this.conf = conf; |
| entryLogPerLedgerEnabled = conf.isEntryLogPerLedgerEnabled(); |
| if (listener != null) { |
| addListener(listener); |
| } |
| |
| // Initialize the entry log header buffer. This cannot be a static object |
| // since in our unit tests, we run multiple Bookies and thus EntryLoggers |
| // within the same JVM. All of these Bookie instances access this header |
| // so there can be race conditions when entry logs are rolled over and |
| // this header buffer is cleared before writing it into the new logChannel. |
| logfileHeader.writeBytes("BKLO".getBytes(UTF_8)); |
| logfileHeader.writeInt(HEADER_CURRENT_VERSION); |
| logfileHeader.writerIndex(LOGFILE_HEADER_SIZE); |
| |
| // Find the largest logId |
| long logId = INVALID_LID; |
| for (File dir : ledgerDirsManager.getAllLedgerDirs()) { |
| if (!dir.exists()) { |
| throw new FileNotFoundException( |
| "Entry log directory '" + dir + "' does not exist"); |
| } |
| long lastLogId = getLastLogId(dir); |
| if (lastLogId > logId) { |
| logId = lastLogId; |
| } |
| } |
| this.recentlyCreatedEntryLogsStatus = new RecentEntryLogsStatus(logId + 1); |
| this.entryLoggerAllocator = new EntryLoggerAllocator(conf, ledgerDirsManager, recentlyCreatedEntryLogsStatus, |
| logId, allocator); |
| if (entryLogPerLedgerEnabled) { |
| this.entryLogManager = new EntryLogManagerForEntryLogPerLedger(conf, ledgerDirsManager, |
| entryLoggerAllocator, listeners, recentlyCreatedEntryLogsStatus, statsLogger); |
| } else { |
| this.entryLogManager = new EntryLogManagerForSingleEntryLog(conf, ledgerDirsManager, entryLoggerAllocator, |
| listeners, recentlyCreatedEntryLogsStatus); |
| } |
| } |
| |
| EntryLogManager getEntryLogManager() { |
| return entryLogManager; |
| } |
| |
| void addListener(EntryLogListener listener) { |
| if (null != listener) { |
| listeners.add(listener); |
| } |
| } |
| |
| /** |
| * If the log id of current writable channel is the same as entryLogId and the position |
| * we want to read might end up reading from a position in the write buffer of the |
| * buffered channel, route this read to the current logChannel. Else, |
| * read from the BufferedReadChannel that is provided. |
| * @param entryLogId |
| * @param channel |
| * @param buff remaining() on this bytebuffer tells us the last position that we |
| * expect to read. |
| * @param pos The starting position from where we want to read. |
| * @return |
| */ |
| private int readFromLogChannel(long entryLogId, BufferedReadChannel channel, ByteBuf buff, long pos) |
| throws IOException { |
| BufferedLogChannel bc = entryLogManager.getCurrentLogIfPresent(entryLogId); |
| if (null != bc) { |
| synchronized (bc) { |
| if (pos + buff.writableBytes() >= bc.getFileChannelPosition()) { |
| return bc.read(buff, pos); |
| } |
| } |
| } |
| return channel.read(buff, pos); |
| } |
| |
| /** |
| * A thread-local variable that wraps a mapping of log ids to bufferedchannels |
| * These channels should be used only for reading. logChannel is the one |
| * that is used for writes. |
| */ |
| private final ThreadLocal<Map<Long, BufferedReadChannel>> logid2Channel = |
| new ThreadLocal<Map<Long, BufferedReadChannel>>() { |
| @Override |
| public Map<Long, BufferedReadChannel> initialValue() { |
| // Since this is thread local there only one modifier |
| // We dont really need the concurrency, but we need to use |
| // the weak values. Therefore using the concurrency level of 1 |
| return new MapMaker().concurrencyLevel(1) |
| .weakValues() |
| .makeMap(); |
| } |
| }; |
| |
| /** |
| * Each thread local buffered read channel can share the same file handle because reads are not relative |
| * and don't cause a change in the channel's position. We use this map to store the file channels. Each |
| * file channel is mapped to a log id which represents an open log file. |
| */ |
| private final ConcurrentMap<Long, FileChannel> logid2FileChannel = new ConcurrentHashMap<Long, FileChannel>(); |
| |
| /** |
| * Put the logId, bc pair in the map responsible for the current thread. |
| * @param logId |
| * @param bc |
| */ |
| public BufferedReadChannel putInReadChannels(long logId, BufferedReadChannel bc) { |
| Map<Long, BufferedReadChannel> threadMap = logid2Channel.get(); |
| return threadMap.put(logId, bc); |
| } |
| |
| /** |
| * Remove all entries for this log file in each thread's cache. |
| * @param logId |
| */ |
| public void removeFromChannelsAndClose(long logId) { |
| FileChannel fileChannel = logid2FileChannel.remove(logId); |
| if (null != fileChannel) { |
| try { |
| fileChannel.close(); |
| } catch (IOException e) { |
| LOG.warn("Exception while closing channel for log file:" + logId); |
| } |
| } |
| } |
| |
| public BufferedReadChannel getFromChannels(long logId) { |
| return logid2Channel.get().get(logId); |
| } |
| |
| @VisibleForTesting |
| long getLeastUnflushedLogId() { |
| return recentlyCreatedEntryLogsStatus.getLeastUnflushedLogId(); |
| } |
| |
| @Override |
| public Set<Long> getFlushedLogIds() { |
| Set<Long> logIds = new HashSet<>(); |
| synchronized (recentlyCreatedEntryLogsStatus) { |
| for (File dir : ledgerDirsManager.getAllLedgerDirs()) { |
| if (dir.exists() && dir.isDirectory()) { |
| File[] files = dir.listFiles(file -> file.getName().endsWith(".log")); |
| if (files != null && files.length > 0) { |
| for (File f : files) { |
| long logId = fileName2LogId(f.getName()); |
| if (recentlyCreatedEntryLogsStatus.isFlushedLogId(logId)) { |
| logIds.add(logId); |
| } |
| } |
| } |
| } |
| } |
| } |
| return logIds; |
| } |
| |
| long getPreviousAllocatedEntryLogId() { |
| return entryLoggerAllocator.getPreallocatedLogId(); |
| } |
| |
| /** |
| * Get the current log file for compaction. |
| */ |
| private File getCurCompactionLogFile() { |
| synchronized (compactionLogLock) { |
| if (compactionLogChannel == null) { |
| return null; |
| } |
| return compactionLogChannel.getLogFile(); |
| } |
| } |
| |
| void prepareSortedLedgerStorageCheckpoint(long numBytesFlushed) throws IOException { |
| entryLogManager.prepareSortedLedgerStorageCheckpoint(numBytesFlushed); |
| } |
| |
| void prepareEntryMemTableFlush() { |
| entryLogManager.prepareEntryMemTableFlush(); |
| } |
| |
| boolean commitEntryMemTableFlush() throws IOException { |
| return entryLogManager.commitEntryMemTableFlush(); |
| } |
| |
| /** |
| * get EntryLoggerAllocator, Just for tests. |
| */ |
| EntryLoggerAllocator getEntryLoggerAllocator() { |
| return entryLoggerAllocator; |
| } |
| |
| /** |
| * Remove entry log. |
| * |
| * @param entryLogId |
| * Entry Log File Id |
| */ |
| @Override |
| public boolean removeEntryLog(long entryLogId) { |
| removeFromChannelsAndClose(entryLogId); |
| File entryLogFile; |
| try { |
| entryLogFile = findFile(entryLogId); |
| } catch (FileNotFoundException e) { |
| LOG.error("Trying to delete an entryLog file that could not be found: " |
| + entryLogId + ".log"); |
| return false; |
| } |
| if (!entryLogFile.delete()) { |
| LOG.warn("Could not delete entry log file {}", entryLogFile); |
| } |
| return true; |
| } |
| |
| private long getLastLogId(File dir) { |
| long id = readLastLogId(dir); |
| // read success |
| if (id > 0) { |
| return id; |
| } |
| // read failed, scan the ledger directories to find biggest log id |
| File[] logFiles = dir.listFiles(file -> file.getName().endsWith(".log")); |
| List<Long> logs = new ArrayList<Long>(); |
| if (logFiles != null) { |
| for (File lf : logFiles) { |
| long logId = fileName2LogId(lf.getName()); |
| logs.add(logId); |
| } |
| } |
| // no log file found in this directory |
| if (0 == logs.size()) { |
| return INVALID_LID; |
| } |
| // order the collections |
| Collections.sort(logs); |
| return logs.get(logs.size() - 1); |
| } |
| |
| /** |
| * reads id from the "lastId" file in the given directory. |
| */ |
| private long readLastLogId(File f) { |
| FileInputStream fis; |
| try { |
| fis = new FileInputStream(new File(f, "lastId")); |
| } catch (FileNotFoundException e) { |
| return INVALID_LID; |
| } |
| try (BufferedReader br = new BufferedReader(new InputStreamReader(fis, UTF_8))) { |
| String lastIdString = br.readLine(); |
| return Long.parseLong(lastIdString, 16); |
| } catch (IOException | NumberFormatException e) { |
| return INVALID_LID; |
| } |
| } |
| |
| /** |
| * Flushes all rotated log channels. After log channels are flushed, |
| * move leastUnflushedLogId ptr to current logId. |
| */ |
| void checkpoint() throws IOException { |
| entryLogManager.checkpoint(); |
| } |
| |
| @Override |
| public void flush() throws IOException { |
| entryLogManager.flush(); |
| } |
| |
| long addEntry(long ledger, ByteBuffer entry) throws IOException { |
| return entryLogManager.addEntry(ledger, Unpooled.wrappedBuffer(entry), true); |
| } |
| |
| long addEntry(long ledger, ByteBuf entry, boolean rollLog) throws IOException { |
| return entryLogManager.addEntry(ledger, entry, rollLog); |
| } |
| |
| @Override |
| public long addEntry(long ledger, ByteBuf entry) throws IOException { |
| return entryLogManager.addEntry(ledger, entry, true); |
| } |
| |
| private final FastThreadLocal<ByteBuf> sizeBuffer = new FastThreadLocal<ByteBuf>() { |
| @Override |
| protected ByteBuf initialValue() throws Exception { |
| // Max usage is size (4 bytes) + ledgerId (8 bytes) + entryid (8 bytes) |
| return Unpooled.buffer(4 + 8 + 8); |
| } |
| }; |
| |
| private long addEntryForCompaction(long ledgerId, ByteBuf entry) throws IOException { |
| synchronized (compactionLogLock) { |
| int entrySize = entry.readableBytes() + 4; |
| if (compactionLogChannel == null) { |
| createNewCompactionLog(); |
| } |
| |
| ByteBuf sizeBuffer = this.sizeBuffer.get(); |
| sizeBuffer.clear(); |
| sizeBuffer.writeInt(entry.readableBytes()); |
| compactionLogChannel.write(sizeBuffer); |
| |
| long pos = compactionLogChannel.position(); |
| compactionLogChannel.write(entry); |
| compactionLogChannel.registerWrittenEntry(ledgerId, entrySize); |
| return (compactionLogChannel.getLogId() << 32L) | pos; |
| } |
| } |
| |
| private void flushCompactionLog() throws IOException { |
| synchronized (compactionLogLock) { |
| if (compactionLogChannel != null) { |
| compactionLogChannel.appendLedgersMap(); |
| compactionLogChannel.flushAndForceWrite(false); |
| LOG.info("Flushed compaction log file {} with logId {}.", |
| compactionLogChannel.getLogFile(), |
| compactionLogChannel.getLogId()); |
| // since this channel is only used for writing, after flushing the channel, |
| // we had to close the underlying file channel. Otherwise, we might end up |
| // leaking fds which cause the disk spaces could not be reclaimed. |
| compactionLogChannel.close(); |
| } else { |
| throw new IOException("Failed to flush compaction log which has already been removed."); |
| } |
| } |
| } |
| |
| private void createNewCompactionLog() throws IOException { |
| synchronized (compactionLogLock) { |
| if (compactionLogChannel == null) { |
| compactionLogChannel = entryLogManager.createNewLogForCompaction(); |
| } |
| } |
| } |
| |
| /** |
| * Remove the current compaction log, usually invoked when compaction failed and |
| * we need to do some clean up to remove the compaction log file. |
| */ |
| private void removeCurCompactionLog() { |
| synchronized (compactionLogLock) { |
| if (compactionLogChannel != null) { |
| if (!compactionLogChannel.getLogFile().delete()) { |
| LOG.warn("Could not delete compaction log file {}", compactionLogChannel.getLogFile()); |
| } |
| |
| try { |
| compactionLogChannel.close(); |
| } catch (IOException e) { |
| LOG.error("Failed to close file channel for compaction log {}", compactionLogChannel.getLogId(), |
| e); |
| } |
| compactionLogChannel = null; |
| } |
| } |
| } |
| |
| static long logIdForOffset(long offset) { |
| return offset >> 32L; |
| } |
| |
| |
| static long posForOffset(long location) { |
| return location & 0xffffffffL; |
| } |
| |
| |
| /** |
| * Exception type for representing lookup errors. Useful for disambiguating different error |
| * conditions for reporting purposes. |
| */ |
| static class EntryLookupException extends Exception { |
| EntryLookupException(String message) { |
| super(message); |
| } |
| |
| /** |
| * Represents case where log file is missing. |
| */ |
| static class MissingLogFileException extends EntryLookupException { |
| MissingLogFileException(long ledgerId, long entryId, long entryLogId, long pos) { |
| super(String.format("Missing entryLog %d for ledgerId %d, entry %d at offset %d", |
| entryLogId, |
| ledgerId, |
| entryId, |
| pos)); |
| } |
| } |
| |
| /** |
| * Represents case where entry log is present, but does not contain the specified entry. |
| */ |
| static class MissingEntryException extends EntryLookupException { |
| MissingEntryException(long ledgerId, long entryId, long entryLogId, long pos) { |
| super(String.format("pos %d (entry %d for ledgerId %d) past end of entryLog %d", |
| pos, |
| entryId, |
| ledgerId, |
| entryLogId)); |
| } |
| } |
| |
| /** |
| * Represents case where log is present, but encoded entry length header is invalid. |
| */ |
| static class InvalidEntryLengthException extends EntryLookupException { |
| InvalidEntryLengthException(long ledgerId, long entryId, long entryLogId, long pos) { |
| super(String.format("Invalid entry length at pos %d (entry %d for ledgerId %d) for entryLog %d", |
| pos, |
| entryId, |
| ledgerId, |
| entryLogId)); |
| } |
| } |
| |
| /** |
| * Represents case where the entry at pos is wrong. |
| */ |
| static class WrongEntryException extends EntryLookupException { |
| WrongEntryException(long foundEntryId, long foundLedgerId, long ledgerId, |
| long entryId, long entryLogId, long pos) { |
| super(String.format( |
| "Found entry %d, ledger %d at pos %d entryLog %d, should have found entry %d for ledgerId %d", |
| foundEntryId, |
| foundLedgerId, |
| pos, |
| entryLogId, |
| entryId, |
| ledgerId)); |
| } |
| } |
| } |
| |
| private BufferedReadChannel getFCForEntryInternal( |
| long ledgerId, long entryId, long entryLogId, long pos) |
| throws EntryLookupException, IOException { |
| try { |
| return getChannelForLogId(entryLogId); |
| } catch (FileNotFoundException e) { |
| throw new EntryLookupException.MissingLogFileException(ledgerId, entryId, entryLogId, pos); |
| } |
| } |
| |
| private ByteBuf readEntrySize(long ledgerId, long entryId, long entryLogId, long pos, BufferedReadChannel fc) |
| throws EntryLookupException, IOException { |
| ByteBuf sizeBuff = sizeBuffer.get(); |
| sizeBuff.clear(); |
| |
| long entrySizePos = pos - 4; // we want to get the entrySize as well as the ledgerId and entryId |
| |
| try { |
| if (readFromLogChannel(entryLogId, fc, sizeBuff, entrySizePos) != sizeBuff.capacity()) { |
| throw new EntryLookupException.MissingEntryException(ledgerId, entryId, entryLogId, entrySizePos); |
| } |
| } catch (BufferedChannelBase.BufferedChannelClosedException | AsynchronousCloseException e) { |
| throw new EntryLookupException.MissingLogFileException(ledgerId, entryId, entryLogId, entrySizePos); |
| } |
| return sizeBuff; |
| } |
| |
| void checkEntry(long ledgerId, long entryId, long location) throws EntryLookupException, IOException { |
| long entryLogId = logIdForOffset(location); |
| long pos = posForOffset(location); |
| BufferedReadChannel fc = getFCForEntryInternal(ledgerId, entryId, entryLogId, pos); |
| ByteBuf sizeBuf = readEntrySize(ledgerId, entryId, entryLogId, pos, fc); |
| validateEntry(ledgerId, entryId, entryLogId, pos, sizeBuf); |
| } |
| |
| private void validateEntry(long ledgerId, long entryId, long entryLogId, long pos, ByteBuf sizeBuff) |
| throws IOException, EntryLookupException { |
| int entrySize = sizeBuff.readInt(); |
| |
| // entrySize does not include the ledgerId |
| if (entrySize > maxSaneEntrySize) { |
| LOG.warn("Sanity check failed for entry size of " + entrySize + " at location " + pos + " in " |
| + entryLogId); |
| } |
| if (entrySize < MIN_SANE_ENTRY_SIZE) { |
| LOG.error("Read invalid entry length {}", entrySize); |
| throw new EntryLookupException.InvalidEntryLengthException(ledgerId, entryId, entryLogId, pos); |
| } |
| |
| long thisLedgerId = sizeBuff.getLong(4); |
| long thisEntryId = sizeBuff.getLong(12); |
| if (thisLedgerId != ledgerId || thisEntryId != entryId) { |
| throw new EntryLookupException.WrongEntryException( |
| thisEntryId, thisLedgerId, ledgerId, entryId, entryLogId, pos); |
| } |
| } |
| |
| @Override |
| public ByteBuf readEntry(long ledgerId, long entryId, long entryLocation) |
| throws IOException, Bookie.NoEntryException { |
| return internalReadEntry(ledgerId, entryId, entryLocation, true /* validateEntry */); |
| } |
| |
| @Override |
| public ByteBuf readEntry(long location) throws IOException, Bookie.NoEntryException { |
| return internalReadEntry(location, -1L, -1L, false /* validateEntry */); |
| } |
| |
| |
| private ByteBuf internalReadEntry(long ledgerId, long entryId, long location, boolean validateEntry) |
| throws IOException, Bookie.NoEntryException { |
| long entryLogId = logIdForOffset(location); |
| long pos = posForOffset(location); |
| |
| |
| BufferedReadChannel fc = null; |
| int entrySize = -1; |
| try { |
| fc = getFCForEntryInternal(ledgerId, entryId, entryLogId, pos); |
| |
| ByteBuf sizeBuff = readEntrySize(ledgerId, entryId, entryLogId, pos, fc); |
| entrySize = sizeBuff.getInt(0); |
| if (validateEntry) { |
| validateEntry(ledgerId, entryId, entryLogId, pos, sizeBuff); |
| } |
| } catch (EntryLookupException e) { |
| throw new IOException("Bad entry read from log file id: " + entryLogId, e); |
| } |
| |
| ByteBuf data = allocator.buffer(entrySize, entrySize); |
| int rc = readFromLogChannel(entryLogId, fc, data, pos); |
| if (rc != entrySize) { |
| data.release(); |
| throw new IOException("Bad entry read from log file id: " + entryLogId, |
| new EntryLookupException("Short read for " + ledgerId + "@" |
| + entryId + " in " + entryLogId + "@" |
| + pos + "(" + rc + "!=" + entrySize + ")")); |
| } |
| data.writerIndex(entrySize); |
| |
| return data; |
| } |
| |
| /** |
| * Read the header of an entry log. |
| */ |
| private Header getHeaderForLogId(long entryLogId) throws IOException { |
| BufferedReadChannel bc = getChannelForLogId(entryLogId); |
| |
| // Allocate buffer to read (version, ledgersMapOffset, ledgerCount) |
| ByteBuf headers = allocator.directBuffer(LOGFILE_HEADER_SIZE); |
| try { |
| bc.read(headers, 0); |
| |
| // Skip marker string "BKLO" |
| headers.readInt(); |
| |
| int headerVersion = headers.readInt(); |
| if (headerVersion < HEADER_V0 || headerVersion > HEADER_CURRENT_VERSION) { |
| LOG.info("Unknown entry log header version for log {}: {}", entryLogId, headerVersion); |
| } |
| |
| long ledgersMapOffset = headers.readLong(); |
| int ledgersCount = headers.readInt(); |
| return new Header(headerVersion, ledgersMapOffset, ledgersCount); |
| } finally { |
| headers.release(); |
| } |
| } |
| |
| private BufferedReadChannel getChannelForLogId(long entryLogId) throws IOException { |
| BufferedReadChannel fc = getFromChannels(entryLogId); |
| if (fc != null) { |
| return fc; |
| } |
| File file = findFile(entryLogId); |
| // get channel is used to open an existing entry log file |
| // it would be better to open using read mode |
| FileChannel newFc = new RandomAccessFile(file, "r").getChannel(); |
| FileChannel oldFc = logid2FileChannel.putIfAbsent(entryLogId, newFc); |
| if (null != oldFc) { |
| newFc.close(); |
| newFc = oldFc; |
| } |
| // We set the position of the write buffer of this buffered channel to Long.MAX_VALUE |
| // so that there are no overlaps with the write buffer while reading |
| fc = new BufferedReadChannel(newFc, conf.getReadBufferBytes()); |
| putInReadChannels(entryLogId, fc); |
| return fc; |
| } |
| |
| /** |
| * Whether the log file exists or not. |
| */ |
| @Override |
| public boolean logExists(long logId) { |
| for (File d : ledgerDirsManager.getAllLedgerDirs()) { |
| File f = new File(d, Long.toHexString(logId) + ".log"); |
| if (f.exists()) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Returns a set with the ids of all the entry log files. |
| * |
| * @throws IOException |
| */ |
| public Set<Long> getEntryLogsSet() throws IOException { |
| Set<Long> entryLogs = Sets.newTreeSet(); |
| |
| final FilenameFilter logFileFilter = new FilenameFilter() { |
| @Override |
| public boolean accept(File dir, String name) { |
| return name.endsWith(".log"); |
| } |
| }; |
| |
| for (File d : ledgerDirsManager.getAllLedgerDirs()) { |
| File[] files = d.listFiles(logFileFilter); |
| if (files == null) { |
| throw new IOException("Failed to get list of files in directory " + d); |
| } |
| |
| for (File f : files) { |
| Long entryLogId = Long.parseLong(f.getName().split(".log")[0], 16); |
| entryLogs.add(entryLogId); |
| } |
| } |
| return entryLogs; |
| } |
| |
| private File findFile(long logId) throws FileNotFoundException { |
| for (File d : ledgerDirsManager.getAllLedgerDirs()) { |
| File f = new File(d, Long.toHexString(logId) + ".log"); |
| if (f.exists()) { |
| return f; |
| } |
| } |
| throw new FileNotFoundException("No file for log " + Long.toHexString(logId)); |
| } |
| |
| /** |
| * Scan entry log. |
| * |
| * @param entryLogId Entry Log Id |
| * @param scanner Entry Log Scanner |
| * @throws IOException |
| */ |
| @Override |
| public void scanEntryLog(long entryLogId, EntryLogScanner scanner) throws IOException { |
| // Buffer where to read the entrySize (4 bytes) and the ledgerId (8 bytes) |
| ByteBuf headerBuffer = Unpooled.buffer(4 + 8); |
| BufferedReadChannel bc; |
| // Get the BufferedChannel for the current entry log file |
| try { |
| bc = getChannelForLogId(entryLogId); |
| } catch (IOException e) { |
| LOG.warn("Failed to get channel to scan entry log: " + entryLogId + ".log"); |
| throw e; |
| } |
| // Start the read position in the current entry log file to be after |
| // the header where all of the ledger entries are. |
| long pos = LOGFILE_HEADER_SIZE; |
| |
| // Start with a reasonably sized buffer size |
| ByteBuf data = allocator.directBuffer(1024 * 1024); |
| |
| try { |
| |
| // Read through the entry log file and extract the ledger ID's. |
| while (true) { |
| // Check if we've finished reading the entry log file. |
| if (pos >= bc.size()) { |
| break; |
| } |
| if (readFromLogChannel(entryLogId, bc, headerBuffer, pos) != headerBuffer.capacity()) { |
| LOG.warn("Short read for entry size from entrylog {}", entryLogId); |
| return; |
| } |
| long offset = pos; |
| |
| int entrySize = headerBuffer.readInt(); |
| if (entrySize <= 0) { // hitting padding |
| pos++; |
| headerBuffer.clear(); |
| continue; |
| } |
| long ledgerId = headerBuffer.readLong(); |
| headerBuffer.clear(); |
| |
| pos += 4; |
| if (ledgerId == INVALID_LID || !scanner.accept(ledgerId)) { |
| // skip this entry |
| pos += entrySize; |
| continue; |
| } |
| // read the entry |
| data.clear(); |
| data.capacity(entrySize); |
| int rc = readFromLogChannel(entryLogId, bc, data, pos); |
| if (rc != entrySize) { |
| LOG.warn("Short read for ledger entry from entryLog {}@{} ({} != {})", |
| entryLogId, pos, rc, entrySize); |
| return; |
| } |
| // process the entry |
| scanner.process(ledgerId, offset, data); |
| |
| // Advance position to the next entry |
| pos += entrySize; |
| } |
| } finally { |
| data.release(); |
| } |
| } |
| |
| public EntryLogMetadata getEntryLogMetadata(long entryLogId, AbstractLogCompactor.Throttler throttler) |
| throws IOException { |
| // First try to extract the EntryLogMetadata from the index, if there's no index then fallback to scanning the |
| // entry log |
| try { |
| return extractEntryLogMetadataFromIndex(entryLogId); |
| } catch (Exception e) { |
| LOG.info("Failed to get ledgers map index from: {}.log : {}", entryLogId, e.getMessage()); |
| |
| // Fall-back to scanning |
| return extractEntryLogMetadataByScanning(entryLogId, throttler); |
| } |
| } |
| |
| EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOException { |
| Header header = getHeaderForLogId(entryLogId); |
| |
| if (header.version < HEADER_V1) { |
| throw new IOException("Old log file header without ledgers map on entryLogId " + entryLogId); |
| } |
| |
| if (header.ledgersMapOffset == 0L) { |
| // The index was not stored in the log file (possibly because the bookie crashed before flushing it) |
| throw new IOException("No ledgers map index found on entryLogId " + entryLogId); |
| } |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Recovering ledgers maps for log {} at offset: {}", entryLogId, header.ledgersMapOffset); |
| } |
| |
| BufferedReadChannel bc = getChannelForLogId(entryLogId); |
| |
| // There can be multiple entries containing the various components of the serialized ledgers map |
| long offset = header.ledgersMapOffset; |
| EntryLogMetadata meta = new EntryLogMetadata(entryLogId); |
| |
| final int maxMapSize = LEDGERS_MAP_HEADER_SIZE + LEDGERS_MAP_ENTRY_SIZE * LEDGERS_MAP_MAX_BATCH_SIZE; |
| ByteBuf ledgersMap = allocator.directBuffer(maxMapSize); |
| |
| try { |
| while (offset < bc.size()) { |
| // Read ledgers map size |
| sizeBuffer.get().clear(); |
| bc.read(sizeBuffer.get(), offset); |
| |
| int ledgersMapSize = sizeBuffer.get().readInt(); |
| if (ledgersMapSize <= 0) { |
| break; |
| } |
| // Read the index into a buffer |
| ledgersMap.clear(); |
| bc.read(ledgersMap, offset + 4, ledgersMapSize); |
| |
| // Discard ledgerId and entryId |
| long lid = ledgersMap.readLong(); |
| if (lid != INVALID_LID) { |
| throw new IOException("Cannot deserialize ledgers map from ledger " + lid); |
| } |
| |
| long entryId = ledgersMap.readLong(); |
| if (entryId != LEDGERS_MAP_ENTRY_ID) { |
| throw new IOException("Cannot deserialize ledgers map from entryId " + entryId); |
| } |
| |
| // Read the number of ledgers in the current entry batch |
| int ledgersCount = ledgersMap.readInt(); |
| |
| // Extract all (ledger,size) tuples from buffer |
| for (int i = 0; i < ledgersCount; i++) { |
| long ledgerId = ledgersMap.readLong(); |
| long size = ledgersMap.readLong(); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Recovering ledgers maps for log {} -- Found ledger: {} with size: {}", |
| entryLogId, ledgerId, size); |
| } |
| meta.addLedgerSize(ledgerId, size); |
| } |
| if (ledgersMap.isReadable()) { |
| throw new IOException("Invalid entry size when reading ledgers map"); |
| } |
| |
| // Move to next entry, if any |
| offset += ledgersMapSize + 4; |
| } |
| } catch (IndexOutOfBoundsException e) { |
| throw new IOException(e); |
| } finally { |
| ledgersMap.release(); |
| } |
| |
| if (meta.getLedgersMap().size() != header.ledgersCount) { |
| throw new IOException("Not all ledgers were found in ledgers map index. expected: " + header.ledgersCount |
| + " -- found: " + meta.getLedgersMap().size() + " -- entryLogId: " + entryLogId); |
| } |
| |
| return meta; |
| } |
| |
| private EntryLogMetadata extractEntryLogMetadataByScanning(long entryLogId, |
| AbstractLogCompactor.Throttler throttler) |
| throws IOException { |
| final EntryLogMetadata meta = new EntryLogMetadata(entryLogId); |
| |
| // Read through the entry log file and extract the entry log meta |
| scanEntryLog(entryLogId, new EntryLogScanner() { |
| @Override |
| public void process(long ledgerId, long offset, ByteBuf entry) throws IOException { |
| if (throttler != null) { |
| throttler.acquire(entry.readableBytes()); |
| } |
| // add new entry size of a ledger to entry log meta |
| meta.addLedgerSize(ledgerId, entry.readableBytes() + 4); |
| } |
| |
| @Override |
| public boolean accept(long ledgerId) { |
| return ledgerId >= 0; |
| } |
| }); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Retrieved entry log meta data entryLogId: {}, meta: {}", entryLogId, meta); |
| } |
| return meta; |
| } |
| |
| /** |
| * Shutdown method to gracefully stop entry logger. |
| */ |
| @Override |
| public void close() { |
| // since logChannel is buffered channel, do flush when shutting down |
| LOG.info("Stopping EntryLogger"); |
| try { |
| flush(); |
| for (FileChannel fc : logid2FileChannel.values()) { |
| fc.close(); |
| } |
| // clear the mapping, so we don't need to go through the channels again in finally block in normal case. |
| logid2FileChannel.clear(); |
| entryLogManager.close(); |
| synchronized (compactionLogLock) { |
| if (compactionLogChannel != null) { |
| compactionLogChannel.close(); |
| compactionLogChannel = null; |
| } |
| } |
| } catch (IOException ie) { |
| // we have no idea how to avoid io exception during shutting down, so just ignore it |
| LOG.error("Error flush entry log during shutting down, which may cause entry log corrupted.", ie); |
| } finally { |
| for (FileChannel fc : logid2FileChannel.values()) { |
| IOUtils.close(LOG, fc); |
| } |
| |
| entryLogManager.forceClose(); |
| synchronized (compactionLogLock) { |
| IOUtils.close(LOG, compactionLogChannel); |
| } |
| } |
| // shutdown the pre-allocation thread |
| entryLoggerAllocator.stop(); |
| } |
| |
| protected LedgerDirsManager getLedgerDirsManager() { |
| return ledgerDirsManager; |
| } |
| |
| /** |
| * Convert log filename (hex format with suffix) to logId in long. |
| */ |
| static long fileName2LogId(String fileName) { |
| if (fileName != null && fileName.contains(".")) { |
| fileName = fileName.split("\\.")[0]; |
| } |
| try { |
| return Long.parseLong(fileName, 16); |
| } catch (Exception nfe) { |
| LOG.error("Invalid log file name {} found when trying to convert to logId.", fileName, nfe); |
| } |
| return INVALID_LID; |
| } |
| |
| /** |
| * Convert log Id to hex string. |
| */ |
| static String logId2HexString(long logId) { |
| return Long.toHexString(logId); |
| } |
| |
| /** |
| * Datastructure which maintains the status of logchannels. When a |
| * logChannel is created entry of < entryLogId, false > will be made to this |
| * sortedmap and when logChannel is rotated and flushed then the entry is |
| * updated to < entryLogId, true > and all the lowest entries with |
| * < entryLogId, true > status will be removed from the sortedmap. So that way |
| * we could get least unflushed LogId. |
| * |
| */ |
| static class RecentEntryLogsStatus { |
| private final SortedMap<Long, Boolean> entryLogsStatusMap; |
| private long leastUnflushedLogId; |
| |
| RecentEntryLogsStatus(long leastUnflushedLogId) { |
| entryLogsStatusMap = new TreeMap<>(); |
| this.leastUnflushedLogId = leastUnflushedLogId; |
| } |
| |
| synchronized void createdEntryLog(Long entryLogId) { |
| entryLogsStatusMap.put(entryLogId, false); |
| } |
| |
| synchronized void flushRotatedEntryLog(Long entryLogId) { |
| entryLogsStatusMap.replace(entryLogId, true); |
| while ((!entryLogsStatusMap.isEmpty()) && (entryLogsStatusMap.get(entryLogsStatusMap.firstKey()))) { |
| long leastFlushedLogId = entryLogsStatusMap.firstKey(); |
| entryLogsStatusMap.remove(leastFlushedLogId); |
| leastUnflushedLogId = leastFlushedLogId + 1; |
| } |
| } |
| |
| synchronized long getLeastUnflushedLogId() { |
| return leastUnflushedLogId; |
| } |
| |
| synchronized boolean isFlushedLogId(long entryLogId) { |
| return entryLogsStatusMap.getOrDefault(entryLogId, Boolean.FALSE) || entryLogId < leastUnflushedLogId; |
| } |
| } |
| |
| @Override |
| public CompactionEntryLog newCompactionLog(long logToCompact) throws IOException { |
| createNewCompactionLog(); |
| |
| File compactingLogFile = getCurCompactionLogFile(); |
| long compactionLogId = fileName2LogId(compactingLogFile.getName()); |
| File compactedLogFile = compactedLogFileFromCompacting(compactingLogFile, logToCompact); |
| File finalLogFile = new File(compactingLogFile.getParentFile(), |
| compactingLogFile.getName().substring(0, |
| compactingLogFile.getName().indexOf(".log") + 4)); |
| return new EntryLoggerCompactionEntryLog( |
| compactionLogId, logToCompact, compactingLogFile, compactedLogFile, finalLogFile); |
| |
| } |
| |
| private class EntryLoggerCompactionEntryLog implements CompactionEntryLog { |
| private final long compactionLogId; |
| private final long logIdToCompact; |
| private final File compactingLogFile; |
| private final File compactedLogFile; |
| private final File finalLogFile; |
| |
| EntryLoggerCompactionEntryLog(long compactionLogId, long logIdToCompact, |
| File compactingLogFile, |
| File compactedLogFile, |
| File finalLogFile) { |
| this.compactionLogId = compactionLogId; |
| this.logIdToCompact = logIdToCompact; |
| this.compactingLogFile = compactingLogFile; |
| this.compactedLogFile = compactedLogFile; |
| this.finalLogFile = finalLogFile; |
| } |
| |
| @Override |
| public long addEntry(long ledgerId, ByteBuf entry) throws IOException { |
| return addEntryForCompaction(ledgerId, entry); |
| } |
| @Override |
| public void scan(EntryLogScanner scanner) throws IOException { |
| scanEntryLog(compactionLogId, scanner); |
| } |
| @Override |
| public void flush() throws IOException { |
| flushCompactionLog(); |
| } |
| @Override |
| public void abort() { |
| removeCurCompactionLog(); |
| if (compactedLogFile.exists()) { |
| if (!compactedLogFile.delete()) { |
| LOG.warn("Could not delete file: {}", compactedLogFile); |
| } |
| } |
| } |
| |
| @Override |
| public void markCompacted() throws IOException { |
| if (compactingLogFile.exists()) { |
| if (!compactedLogFile.exists()) { |
| HardLink.createHardLink(compactingLogFile, compactedLogFile); |
| } |
| } else { |
| throw new IOException("Compaction log doesn't exist any more after flush: " + compactingLogFile); |
| } |
| removeCurCompactionLog(); |
| } |
| |
| @Override |
| public void makeAvailable() throws IOException { |
| if (!finalLogFile.exists()) { |
| HardLink.createHardLink(compactedLogFile, finalLogFile); |
| } |
| } |
| @Override |
| public void finalizeAndCleanup() { |
| if (compactedLogFile.exists()) { |
| if (!compactedLogFile.delete()) { |
| LOG.warn("Could not delete file: {}", compactedLogFile); |
| } |
| } |
| if (compactingLogFile.exists()) { |
| if (!compactingLogFile.delete()) { |
| LOG.warn("Could not delete file: {}", compactingLogFile); |
| } |
| } |
| } |
| |
| @Override |
| public long getDstLogId() { |
| return compactionLogId; |
| } |
| @Override |
| public long getSrcLogId() { |
| return logIdToCompact; |
| } |
| |
| @Override |
| public String toString() { |
| return MoreObjects.toStringHelper(this) |
| .add("logId", compactionLogId) |
| .add("compactedLogId", logIdToCompact) |
| .add("compactingLogFile", compactingLogFile) |
| .add("compactedLogFile", compactedLogFile) |
| .add("finalLogFile", finalLogFile) |
| .toString(); |
| } |
| } |
| |
| @Override |
| public Collection<CompactionEntryLog> incompleteCompactionLogs() { |
| List<File> ledgerDirs = ledgerDirsManager.getAllLedgerDirs(); |
| List<CompactionEntryLog> compactionLogs = new ArrayList<>(); |
| |
| for (File dir : ledgerDirs) { |
| File[] compactingPhaseFiles = dir.listFiles( |
| file -> file.getName().endsWith(TransactionalEntryLogCompactor.COMPACTING_SUFFIX)); |
| if (compactingPhaseFiles != null) { |
| for (File file : compactingPhaseFiles) { |
| if (file.delete()) { |
| LOG.info("Deleted failed compaction file {}", file); |
| } |
| } |
| } |
| File[] compactedPhaseFiles = dir.listFiles( |
| file -> file.getName().endsWith(TransactionalEntryLogCompactor.COMPACTED_SUFFIX)); |
| if (compactedPhaseFiles != null) { |
| for (File compactedFile : compactedPhaseFiles) { |
| LOG.info("Found compacted log file {} has partially flushed index, recovering index.", |
| compactedFile); |
| |
| File compactingLogFile = new File(compactedFile.getParentFile(), "doesntexist"); |
| long compactionLogId = -1L; |
| long compactedLogId = -1L; |
| String[] parts = compactedFile.getName().split(Pattern.quote(".")); |
| boolean valid = true; |
| if (parts.length != 4) { |
| valid = false; |
| } else { |
| try { |
| compactionLogId = Long.parseLong(parts[0], 16); |
| compactedLogId = Long.parseLong(parts[2], 16); |
| } catch (NumberFormatException nfe) { |
| valid = false; |
| } |
| } |
| |
| if (!valid) { |
| LOG.info("Invalid compacted file found ({}), deleting", compactedFile); |
| if (!compactedFile.delete()) { |
| LOG.warn("Couldn't delete invalid compacted file ({})", compactedFile); |
| } |
| continue; |
| } |
| File finalLogFile = new File(compactedFile.getParentFile(), compactionLogId + ".log"); |
| |
| compactionLogs.add( |
| new EntryLoggerCompactionEntryLog(compactionLogId, compactedLogId, |
| compactingLogFile, compactedFile, finalLogFile)); |
| } |
| } |
| } |
| return compactionLogs; |
| } |
| |
| private static File compactedLogFileFromCompacting(File compactionLogFile, long compactingLogId) { |
| File dir = compactionLogFile.getParentFile(); |
| String filename = compactionLogFile.getName(); |
| String newSuffix = ".log." + DefaultEntryLogger.logId2HexString(compactingLogId) |
| + TransactionalEntryLogCompactor.COMPACTED_SUFFIX; |
| String hardLinkFilename = filename.replace(TransactionalEntryLogCompactor.COMPACTING_SUFFIX, newSuffix); |
| return new File(dir, hardLinkFilename); |
| } |
| } |
