* Note: If you use a store its reference count should be increased before using it by calling #incRef and a * corresponding #decRef must be called in a try/finally block to release the store again ie.: *
* store.incRef();
* try {
* // use the store...
*
* } finally {
* store.decRef();
* }
*
*/
public class Store extends AbstractIndexShardComponent implements Closeable, RefCounted {
/**
* Legacy index setting, kept for 7.x BWC compatibility. This setting has no effect in 8.x. Do not use.
* TODO: Remove in 9.0
*/
@Deprecated
public static final Settingnull
*
* @throws IOException if the index is corrupted or the segments file is not present
*/
private static SegmentInfos readSegmentsInfo(IndexCommit commit, Directory directory) throws IOException {
assert commit == null || commit.getDirectory() == directory;
try {
return commit == null ? Lucene.readSegmentInfos(directory) : Lucene.readSegmentInfos(commit);
} catch (EOFException eof) {
// TODO this should be caught by lucene - EOF is almost certainly an index corruption
throw new CorruptIndexException("Read past EOF while reading segment infos", "commit(" + commit + ")", eof);
} catch (IOException exception) {
throw exception; // IOExceptions like too many open files are not necessarily a corruption - just bubble it up
} catch (Exception ex) {
throw new CorruptIndexException("Hit unexpected exception while reading segment infos", "commit(" + commit + ")", ex);
}
}
final void ensureOpen() {
if (this.refCounter.refCount() <= 0) {
throw new AlreadyClosedException("store is already closed");
}
}
/**
* Returns a new MetadataSnapshot for the given commit. If the given commit is null
* the latest commit point is used.
*
* Note that this method requires the caller verify it has the right to access the store and
* no concurrent file changes are happening. If in doubt, you probably want to use one of the following:
*
* {@link #readMetadataSnapshot(Path, ShardId, NodeEnvironment.ShardLocker, Logger)} to read a meta data while locking
* {@link IndexShard#snapshotStoreMetadata()} to safely read from an existing shard
* {@link IndexShard#acquireLastIndexCommit(boolean)} to get an {@link IndexCommit} which is safe to use but has to be freed
* @param commit the index commit to read the snapshot from or null if the latest snapshot should be read from the
* directory
* @throws CorruptIndexException if the lucene index is corrupted. This can be caused by a checksum mismatch or an
* unexpected exception when opening the index reading the segments file.
* @throws IndexFormatTooOldException if the lucene index is too old to be opened.
* @throws IndexFormatTooNewException if the lucene index is too new to be opened.
* @throws FileNotFoundException if one or more files referenced by a commit are not present.
* @throws NoSuchFileException if one or more files referenced by a commit are not present.
* @throws IndexNotFoundException if the commit point can't be found in this store
*/
public MetadataSnapshot getMetadata(IndexCommit commit) throws IOException {
return getMetadata(commit, false);
}
/**
* Returns a new MetadataSnapshot for the given commit. If the given commit is null
* the latest commit point is used.
*
* Note that this method requires the caller verify it has the right to access the store and
* no concurrent file changes are happening. If in doubt, you probably want to use one of the following:
*
* {@link #readMetadataSnapshot(Path, ShardId, NodeEnvironment.ShardLocker, Logger)} to read a meta data while locking
* {@link IndexShard#snapshotStoreMetadata()} to safely read from an existing shard
* {@link IndexShard#acquireLastIndexCommit(boolean)} to get an {@link IndexCommit} which is safe to use but has to be freed
*
* @param commit the index commit to read the snapshot from or null if the latest snapshot should be read from the
* directory
* @param lockDirectory if true the index writer lock will be obtained before reading the snapshot. This should
* only be used if there is no started shard using this store.
* @throws CorruptIndexException if the lucene index is corrupted. This can be caused by a checksum mismatch or an
* unexpected exception when opening the index reading the segments file.
* @throws IndexFormatTooOldException if the lucene index is too old to be opened.
* @throws IndexFormatTooNewException if the lucene index is too new to be opened.
* @throws FileNotFoundException if one or more files referenced by a commit are not present.
* @throws NoSuchFileException if one or more files referenced by a commit are not present.
* @throws IndexNotFoundException if the commit point can't be found in this store
*/
public MetadataSnapshot getMetadata(IndexCommit commit, boolean lockDirectory) throws IOException {
ensureOpen();
failIfCorrupted();
assert lockDirectory ? commit == null : true : "IW lock should not be obtained if there is a commit point available";
// if we lock the directory we also acquire the write lock since that makes sure that nobody else tries to lock the IW
// on this store at the same time.
java.util.concurrent.locks.Lock lock = lockDirectory ? metadataLock.writeLock() : metadataLock.readLock();
lock.lock();
try (Closeable ignored = lockDirectory ? directory.obtainLock(IndexWriter.WRITE_LOCK_NAME) : () -> {}) {
return MetadataSnapshot.loadFromIndexCommit(commit, directory, logger);
} catch (CorruptIndexException | IndexFormatTooOldException | IndexFormatTooNewException ex) {
markStoreCorrupted(ex);
throw ex;
} finally {
lock.unlock();
}
}
/**
* Renames all the given files from the key of the map to the
* value of the map. All successfully renamed files are removed from the map in-place.
*/
public void renameTempFilesSafe(Map* Note: Close can safely be called multiple times. * * @throws AlreadyClosedException iff the reference counter can not be incremented. * @see #decRef * @see #tryIncRef() */ @Override public final void incRef() { refCounter.incRef(); } /** * Tries to increment the refCount of this Store instance. This method will return {@code true} iff the refCount was * incremented successfully otherwise {@code false}. RefCounts are used to determine when a * Store can be closed safely, i.e. as soon as there are no more references. Be sure to always call a * corresponding {@link #decRef}, in a finally clause; otherwise the store may never be closed. Note that * {@link #close} simply calls decRef(), which means that the Store will not really be closed until {@link * #decRef} has been called for all outstanding references. *
* Note: Close can safely be called multiple times. * * @see #decRef() * @see #incRef() */ @Override public final boolean tryIncRef() { return refCounter.tryIncRef(); } /** * Decreases the refCount of this Store instance. If the refCount drops to 0, then this * store is closed. * * @see #incRef */ @Override public final boolean decRef() { return refCounter.decRef(); } @Override public final boolean hasReferences() { return refCounter.hasReferences(); } @Override public void close() { if (isClosed.compareAndSet(false, true)) { // only do this once! decRef(); logger.debug("store reference count on close: {}", refCounter.refCount()); } } /** * @return true if the {@link Store#close()} method has been called. This indicates that the current * store is either closed or being closed waiting for all references to it to be released. * You might prefer to use {@link Store#ensureOpen()} instead. */ public boolean isClosing() { return isClosed.get(); } private void closeInternal() { // Leverage try-with-resources to close the shard lock for us try (Closeable c = shardLock) { try { directory.innerClose(); // this closes the distributorDirectory as well } finally { onClose.accept(shardLock); } } catch (IOException e) { assert false : e; logger.warn(() -> "exception on closing store for [" + shardId + "]", e); } } private static ByteSizeDirectory byteSizeDirectory(Directory directory, IndexSettings indexSettings, Logger logger) { if (directory instanceof ByteSizeDirectory byteSizeDirectory) { return byteSizeDirectory; } else { final TimeValue refreshInterval = indexSettings.getValue(INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING); logger.debug("store stats are refreshed with {} [{}]", INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING.getKey(), refreshInterval); return new ByteSizeCachingDirectory(directory, refreshInterval); } } /** * Reads a MetadataSnapshot from the given index locations or returns an empty snapshot if it can't be read. * * @throws IOException if the index we try to read is corrupted */ public static MetadataSnapshot readMetadataSnapshot( Path indexLocation, ShardId shardId, NodeEnvironment.ShardLocker shardLocker, Logger logger ) throws IOException { try ( ShardLock lock = shardLocker.lock(shardId, "read metadata snapshot", TimeUnit.SECONDS.toMillis(5)); Directory dir = new NIOFSDirectory(indexLocation) ) { failIfCorrupted(dir); return MetadataSnapshot.loadFromIndexCommit(null, dir, logger); } catch (IndexNotFoundException ex) { // that's fine - happens all the time no need to log } catch (CorruptIndexException ex) { logger.info(() -> format("%s: corrupted", shardId), ex); } catch (FileNotFoundException | NoSuchFileException ex) { logger.info("Failed to open / find files while reading metadata snapshot", ex); } catch (ShardLockObtainFailedException ex) { logger.info(() -> format("%s: failed to obtain shard lock", shardId), ex); } return MetadataSnapshot.EMPTY; } /** * Tries to open an index for the given location. This includes reading the * segment infos and possible corruption markers. If the index can not * be opened, an exception is thrown */ public static void tryOpenIndex(Path indexLocation, ShardId shardId, NodeEnvironment.ShardLocker shardLocker, Logger logger) throws IOException, ShardLockObtainFailedException { try ( ShardLock lock = shardLocker.lock(shardId, "open index", TimeUnit.SECONDS.toMillis(5)); Directory dir = new NIOFSDirectory(indexLocation) ) { failIfCorrupted(dir); SegmentInfos segInfo = Lucene.readSegmentInfos(dir); logger.trace("{} loaded segment info [{}]", shardId, segInfo); } } /** * The returned IndexOutput validates the files checksum. *
* Note: Checksums are calculated by default since version 4.8.0. This method only adds the
* verification against the checksum in the given metadata and does not add any significant overhead.
*/
public IndexOutput createVerifyingOutput(String fileName, final StoreFileMetadata metadata, final IOContext context)
throws IOException {
IndexOutput output = directory().createOutput(fileName, context);
boolean success = false;
try {
assert metadata.writtenBy() != null;
output = new VerifyingIndexOutput(metadata, output);
success = true;
} finally {
if (success == false) {
IOUtils.closeWhileHandlingException(output);
}
}
return output;
}
public static void verify(IndexOutput output) throws IOException {
if (output instanceof VerifyingIndexOutput verifyingIndexOutput) {
verifyingIndexOutput.verify();
}
}
public IndexInput openVerifyingInput(String filename, IOContext context, StoreFileMetadata metadata) throws IOException {
assert metadata.writtenBy() != null;
return new VerifyingIndexInput(directory().openInput(filename, context));
}
public static void verify(IndexInput input) throws IOException {
if (input instanceof VerifyingIndexInput) {
((VerifyingIndexInput) input).verify();
}
}
public boolean checkIntegrityNoException(StoreFileMetadata md) {
return checkIntegrityNoException(md, directory());
}
public static boolean checkIntegrityNoException(StoreFileMetadata md, Directory directory) {
try {
checkIntegrity(md, directory);
return true;
} catch (IOException e) {
return false;
}
}
public static void checkIntegrity(final StoreFileMetadata md, final Directory directory) throws IOException {
try (IndexInput input = directory.openInput(md.name(), IOContext.READONCE)) {
if (input.length() != md.length()) { // first check the length no matter how old this file is
throw new CorruptIndexException(
"expected length=" + md.length() + " != actual length: " + input.length() + " : file truncated?",
input
);
}
// throw exception if the file is corrupt
String checksum = Store.digestToString(CodecUtil.checksumEntireFile(input));
// throw exception if metadata is inconsistent
if (checksum.equals(md.checksum()) == false) {
throw new CorruptIndexException(
"inconsistent metadata: lucene checksum=" + checksum + ", metadata checksum=" + md.checksum(),
input
);
}
}
}
public boolean isMarkedCorrupted() throws IOException {
ensureOpen();
/* marking a store as corrupted is basically adding a _corrupted to all
* the files. This prevent
*/
final String[] files = directory().listAll();
for (String file : files) {
if (file.startsWith(CORRUPTED_MARKER_NAME_PREFIX)) {
return true;
}
}
return false;
}
/**
* Deletes all corruption markers from this store.
*/
public void removeCorruptionMarker() throws IOException {
ensureOpen();
final Directory directory = directory();
IOException firstException = null;
final String[] files = directory.listAll();
for (String file : files) {
if (file.startsWith(CORRUPTED_MARKER_NAME_PREFIX)) {
try {
directory.deleteFile(file);
} catch (IOException ex) {
if (firstException == null) {
firstException = ex;
} else {
firstException.addSuppressed(ex);
}
}
}
}
if (firstException != null) {
throw firstException;
}
}
public void failIfCorrupted() throws IOException {
ensureOpen();
failIfCorrupted(directory);
}
private static void failIfCorrupted(Directory directory) throws IOException {
final String[] files = directory.listAll();
List
* Note: This class will ignore the {@code segments.gen} file since it's optional and might
* change concurrently for safety reasons.
*
* @see StoreFileMetadata
*
* @param numDocs the number of documents in this store snapshot
*/
public record MetadataSnapshot(Map
* Individual files are compared by name, length, checksum and (if present) a UUID that was assigned when the file was originally
* written. The segment info ({@code *.si}) files and the segments file ({@code segments_N}) are also checked to be a byte-for-byte
* match.
*
* Files are collected together into a group for each segment plus one group of "per-commit" ({@code segments_N}) files. Each
* per-segment group is subdivided into a nongenerational group (most of them) and a generational group (e.g. {@code *.liv},
* {@code *.fnm}, {@code *.dvm}, {@code *.dvd} that have been updated by subsequent commits).
*
* For each segment, if any nongenerational files are different then the whole segment is considered to be different and will be
* recovered in full. If all the nongenerational files are the same but any generational files are different then all the
* generational files are considered to be different and will be recovered in full, but the nongenerational files are left alone.
* Finally, if any file is different then all the per-commit files are recovered too.
*/
public RecoveryDiff recoveryDiff(final MetadataSnapshot targetSnapshot) {
final List
* This class supports random access (it is possible to seek backward and forward) in order to accommodate retry
* mechanism that is used in some repository plugins (S3 for example). However, the checksum is only calculated on
* the first read. All consecutive reads of the same data are not used to calculate the checksum.
*/
public static class VerifyingIndexInput extends ChecksumIndexInput {
private final IndexInput input;
private final Checksum digest;
private final long checksumPosition;
private final byte[] checksum = new byte[8];
private long verifiedPosition = 0;
public VerifyingIndexInput(IndexInput input) {
this(input, new BufferedChecksum(new CRC32()));
}
VerifyingIndexInput(IndexInput input, Checksum digest) {
super("VerifyingIndexInput(" + input + ")");
this.input = input;
this.digest = digest;
checksumPosition = input.length() - 8;
}
@Override
public byte readByte() throws IOException {
long pos = input.getFilePointer();
final byte b = input.readByte();
pos++;
if (pos > verifiedPosition) {
if (pos <= checksumPosition) {
digest.update(b);
} else {
checksum[(int) (pos - checksumPosition - 1)] = b;
}
verifiedPosition = pos;
}
return b;
}
@Override
public void readBytes(byte[] b, int offset, int len) throws IOException {
long pos = input.getFilePointer();
input.readBytes(b, offset, len);
if (pos + len > verifiedPosition) {
// Conversion to int is safe here because (verifiedPosition - pos) can be at most len, which is integer
int alreadyVerified = (int) Math.max(0, verifiedPosition - pos);
if (pos < checksumPosition) {
if (pos + len < checksumPosition) {
digest.update(b, offset + alreadyVerified, len - alreadyVerified);
} else {
int checksumOffset = (int) (checksumPosition - pos);
if (checksumOffset - alreadyVerified > 0) {
digest.update(b, offset + alreadyVerified, checksumOffset - alreadyVerified);
}
System.arraycopy(b, offset + checksumOffset, checksum, 0, len - checksumOffset);
}
} else {
// Conversion to int is safe here because checksumPosition is (file length - 8) so
// (pos - checksumPosition) cannot be bigger than 8 unless we are reading after the end of file
assert pos - checksumPosition < 8;
System.arraycopy(b, offset, checksum, (int) (pos - checksumPosition), len);
}
verifiedPosition = pos + len;
}
}
@Override
public long getChecksum() {
return digest.getValue();
}
@Override
public void seek(long pos) throws IOException {
if (pos < verifiedPosition) {
// going within verified region - just seek there
input.seek(pos);
} else {
if (verifiedPosition > getFilePointer()) {
// portion of the skip region is verified and portion is not
// skipping the verified portion
input.seek(verifiedPosition);
// and checking unverified
super.seek(pos);
} else {
super.seek(pos);
}
}
}
@Override
public void close() throws IOException {
input.close();
}
@Override
public long getFilePointer() {
return input.getFilePointer();
}
@Override
public long length() {
return input.length();
}
@Override
public IndexInput clone() {
throw new UnsupportedOperationException();
}
@Override
public IndexInput slice(String sliceDescription, long offset, long length) throws IOException {
throw new UnsupportedOperationException();
}
public long getStoredChecksum() {
try {
return CodecUtil.readBELong(new ByteArrayDataInput(checksum));
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
public long verify() throws CorruptIndexException {
long storedChecksum = getStoredChecksum();
if (getChecksum() == storedChecksum) {
return storedChecksum;
}
throw new CorruptIndexException(
"verification failed : calculated="
+ Store.digestToString(getChecksum())
+ " stored="
+ Store.digestToString(storedChecksum),
this
);
}
}
public void deleteQuiet(String... files) {
ensureOpen();
StoreDirectory directory = this.directory;
for (String file : files) {
try {
directory.deleteFile("Store.deleteQuiet", file);
} catch (Exception ex) {
// ignore :(
}
}
}
/**
* Marks this store as corrupted. This method writes a {@code corrupted_${uuid}} file containing the given exception
* message. If a store contains a {@code corrupted_${uuid}} file {@link #isMarkedCorrupted()} will return
* 1. Replica can use unsafe commit in peer-recovery. This happens when a replica with a safe commit c1(max_seqno=1)
* and an unsafe commit c2(max_seqno=2) recovers from a primary with c1(max_seqno=1). If a new document(seqno=2)
* is added without flushing, the global checkpoint is advanced to 2; and the replica recovers again, it will use
* the unsafe commit c2(max_seqno=2 at most gcp=2) as the starting commit for sequenced-based recovery even the
* commit c2 contains a stale operation and the document(with seqno=2) will not be replicated to the replica.
*
* 2. Min translog gen for recovery can go backwards in peer-recovery. This happens when are replica with a safe commit
* c1(local_checkpoint=1, recovery_translog_gen=1) and an unsafe commit c2(local_checkpoint=2, recovery_translog_gen=2).
* The replica recovers from a primary, and keeps c2 as the last commit, then sets last_translog_gen to 2. Flushing a new
* commit on the replica will cause exception as the new last commit c3 will have recovery_translog_gen=1. The recovery
* translog generation of a commit is calculated based on the current local checkpoint. The local checkpoint of c3 is 1
* while the local checkpoint of c2 is 2.
*/
public void trimUnsafeCommits(final Path translogPath) throws IOException {
metadataLock.writeLock().lock();
try {
final List
*
* true.
*/
public void markStoreCorrupted(IOException exception) throws IOException {
ensureOpen();
if (isMarkedCorrupted() == false) {
final String corruptionMarkerName = CORRUPTED_MARKER_NAME_PREFIX + UUIDs.randomBase64UUID();
try (IndexOutput output = this.directory().createOutput(corruptionMarkerName, IOContext.DEFAULT)) {
CodecUtil.writeHeader(output, CODEC, CORRUPTED_MARKER_CODEC_VERSION);
BytesStreamOutput out = new BytesStreamOutput();
out.writeException(exception);
BytesReference bytes = out.bytes();
output.writeVInt(bytes.length());
BytesRef ref = bytes.toBytesRef();
output.writeBytes(ref.bytes, ref.offset, ref.length);
CodecUtil.writeFooter(output);
} catch (IOException | ImmutableDirectoryException ex) {
if (exception != null) {
ex.addSuppressed(exception);
}
if (ex instanceof ImmutableDirectoryException) {
logger.debug("Can't mark store with an immutable directory as corrupted", ex);
} else {
logger.warn("Can't mark store as corrupted", ex);
}
}
directory().sync(Collections.singleton(corruptionMarkerName));
}
}
/**
* A listener that is executed once the store is closed and all references to it are released
*/
public interface OnClose extends Consumer