Use RemoteSegmentStoreDirectory instead of RemoteDirectory #4240
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -11,32 +11,54 @@ | |
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.apache.logging.log4j.message.ParameterizedMessage; | ||
| import org.apache.lucene.codecs.CodecUtil; | ||
| import org.apache.lucene.index.IndexFileNames; | ||
| import org.apache.lucene.index.SegmentInfos; | ||
| import org.apache.lucene.search.ReferenceManager; | ||
| import org.apache.lucene.store.Directory; | ||
| import org.apache.lucene.store.FilterDirectory; | ||
| import org.apache.lucene.store.IOContext; | ||
| import org.apache.lucene.store.IndexInput; | ||
| import org.opensearch.common.concurrent.GatedCloseable; | ||
| import org.opensearch.index.engine.EngineException; | ||
| import org.opensearch.index.store.RemoteSegmentStoreDirectory; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Collection; | ||
| import java.util.Comparator; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * RefreshListener implementation to upload newly created segment files to the remote store | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public final class RemoteStoreRefreshListener implements ReferenceManager.RefreshListener { | ||
| // Visible for testing | ||
| static final Set<String> EXCLUDE_FILES = Set.of("write.lock"); | ||
| // Visible for testing | ||
| static final int LAST_N_METADATA_FILES_TO_KEEP = 10; | ||
|
|
||
| private final IndexShard indexShard; | ||
| private final Directory storeDirectory; | ||
| private final RemoteSegmentStoreDirectory remoteDirectory; | ||
| private final Map<String, String> localSegmentChecksumMap; | ||
| private long primaryTerm; | ||
| private static final Logger logger = LogManager.getLogger(RemoteStoreRefreshListener.class); | ||
|
|
||
| public RemoteStoreRefreshListener(IndexShard indexShard) { | ||
| this.indexShard = indexShard; | ||
| this.storeDirectory = indexShard.store().directory(); | ||
| this.remoteDirectory = (RemoteSegmentStoreDirectory) ((FilterDirectory) ((FilterDirectory) indexShard.remoteStore().directory()) | ||
| .getDelegate()).getDelegate(); | ||
| this.primaryTerm = indexShard.getOperationPrimaryTerm(); | ||
| localSegmentChecksumMap = new HashMap<>(); | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -46,42 +68,112 @@ public void beforeRefresh() throws IOException { | |
|
|
||
| /** | ||
| * Upload new segment files created as part of the last refresh to the remote segment store. | ||
| * This method also uploads remote_segments_metadata file which contains metadata of each segment file uploaded. | ||
| * @param didRefresh true if the refresh opened a new reference | ||
| */ | ||
| @Override | ||
| public void afterRefresh(boolean didRefresh) { | ||
| synchronized (this) { | ||
| try { | ||
| if (indexShard.shardRouting.primary()) { | ||
|
There was a problem hiding this comment. What happens if index shard was demoted to replica from primary? Shouldn't There was a problem hiding this comment. Only primary should be uploading the segments. If replica is demoted, this check makes sure that we stop uploading segments from replica. Ideally, There was a problem hiding this comment. So in this case the usage of There was a problem hiding this comment. If the shard is demoted and promoted again, we need to make sure that we call There was a problem hiding this comment. Ok, I will try to go step by step by step:
Does it make sense? There was a problem hiding this comment. Yeah, this is definitely a bug. Thanks for catching it. I will fix it and will add tests around it as you suggested in another comment. There was a problem hiding this comment. @reta I changed the implementation and added tests. Please take a look again. |
||
| if (this.primaryTerm != indexShard.getOperationPrimaryTerm()) { | ||
| this.primaryTerm = indexShard.getOperationPrimaryTerm(); | ||
| this.remoteDirectory.init(); | ||
| } | ||
| try { | ||
| String lastCommittedLocalSegmentFileName = SegmentInfos.getLastCommitSegmentsFileName(storeDirectory); | ||
| if (!remoteDirectory.containsFile( | ||
| lastCommittedLocalSegmentFileName, | ||
| getChecksumOfLocalFile(lastCommittedLocalSegmentFileName) | ||
| )) { | ||
| deleteStaleCommits(); | ||
|
There was a problem hiding this comment. Would clean up stall refresh ? Can we do this in the background? There was a problem hiding this comment. Clean up does not happen at each refresh. It would only be called when refresh is called during flush (as new segments_N file gets created). We can definitely preform stale commit deletion in the background. I have created a tracking issue here: #4315 |
||
| } | ||
| try (GatedCloseable<SegmentInfos> segmentInfosGatedCloseable = indexShard.getSegmentInfosSnapshot()) { | ||
| SegmentInfos segmentInfos = segmentInfosGatedCloseable.get(); | ||
| Collection<String> refreshedLocalFiles = segmentInfos.files(true); | ||
|
|
||
| List<String> segmentInfosFiles = refreshedLocalFiles.stream() | ||
| .filter(file -> file.startsWith(IndexFileNames.SEGMENTS)) | ||
| .collect(Collectors.toList()); | ||
| Optional<String> latestSegmentInfos = segmentInfosFiles.stream() | ||
| .max(Comparator.comparingLong(IndexFileNames::parseGeneration)); | ||
|
|
||
| if (latestSegmentInfos.isPresent()) { | ||
| refreshedLocalFiles.addAll(SegmentInfos.readCommit(storeDirectory, latestSegmentInfos.get()).files(true)); | ||
| segmentInfosFiles.stream() | ||
| .filter(file -> !file.equals(latestSegmentInfos.get())) | ||
| .forEach(refreshedLocalFiles::remove); | ||
|
|
||
| boolean uploadStatus = uploadNewSegments(refreshedLocalFiles); | ||
| if (uploadStatus) { | ||
| remoteDirectory.uploadMetadata( | ||
| refreshedLocalFiles, | ||
| storeDirectory, | ||
| indexShard.getOperationPrimaryTerm(), | ||
| segmentInfos.getGeneration() | ||
| ); | ||
| localSegmentChecksumMap.keySet() | ||
| .stream() | ||
| .filter(file -> !refreshedLocalFiles.contains(file)) | ||
| .collect(Collectors.toSet()) | ||
| .forEach(localSegmentChecksumMap::remove); | ||
| } | ||
| } | ||
| } catch (EngineException e) { | ||
|
There was a problem hiding this comment. Since we swallow all exceptions, what happens if shard was promoted to primary but segments were not refreshed due to some failures? Should we go back to replica mode? Or should we schedule the retry later on? There was a problem hiding this comment. With the current implementation, we retry segment upload in the next refresh. If the upload fails, we don't want to demote primary as the failure could be due to issues with remote segment store itself. |
||
| logger.warn("Exception while reading SegmentInfosSnapshot", e); | ||
| } | ||
| } catch (IOException e) { | ||
| // We don't want to fail refresh if upload of new segments fails. The missed segments will be re-tried | ||
| // in the next refresh. This should not affect durability of the indexed data after remote trans-log integration. | ||
| logger.warn("Exception while uploading new segments to the remote segment store", e); | ||
| } | ||
| } | ||
| } catch (Throwable t) { | ||
| logger.error("Exception in RemoteStoreRefreshListener.afterRefresh()", t); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Visible for testing | ||
| boolean uploadNewSegments(Collection<String> localFiles) throws IOException { | ||
| AtomicBoolean uploadSuccess = new AtomicBoolean(true); | ||
| localFiles.stream().filter(file -> !EXCLUDE_FILES.contains(file)).filter(file -> { | ||
| try { | ||
| return !remoteDirectory.containsFile(file, getChecksumOfLocalFile(file)); | ||
| } catch (IOException e) { | ||
| logger.info( | ||
| "Exception while reading checksum of local segment file: {}, ignoring the exception and re-uploading the file", | ||
| file | ||
| ); | ||
| return true; | ||
| } | ||
| }).forEach(file -> { | ||
| try { | ||
| remoteDirectory.copyFrom(storeDirectory, file, file, IOContext.DEFAULT); | ||
| } catch (IOException e) { | ||
| uploadSuccess.set(false); | ||
| // ToDO: Handle transient and permanent un-availability of the remote store (GitHub #3397) | ||
| logger.warn(() -> new ParameterizedMessage("Exception while uploading file {} to the remote segment store", file), e); | ||
| } | ||
| }); | ||
| return uploadSuccess.get(); | ||
|
Comment on lines
+138
to
+159
There was a problem hiding this comment. the upload is happening serially, can we either bulk upload or do concurrent async uploads |
||
| } | ||
|
|
||
| private String getChecksumOfLocalFile(String file) throws IOException { | ||
| if (!localSegmentChecksumMap.containsKey(file)) { | ||
| try (IndexInput indexInput = storeDirectory.openInput(file, IOContext.DEFAULT)) { | ||
| String checksum = Long.toString(CodecUtil.retrieveChecksum(indexInput)); | ||
| localSegmentChecksumMap.put(file, checksum); | ||
| } | ||
| } | ||
| return localSegmentChecksumMap.get(file); | ||
| } | ||
|
|
||
| private void deleteStaleCommits() { | ||
| try { | ||
| remoteDirectory.deleteStaleSegments(LAST_N_METADATA_FILES_TO_KEEP); | ||
| } catch (IOException e) { | ||
| logger.info("Exception while deleting stale commits from remote segment store, will retry delete post next commit", e); | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -27,27 +27,37 @@ public class RemoteIndexInput extends IndexInput { | |
|
|
||
| private final InputStream inputStream; | ||
| private final long size; | ||
| private long filePointer; | ||
|
|
||
| public RemoteIndexInput(String name, InputStream inputStream, long size) { | ||
| super(name); | ||
| this.inputStream = inputStream; | ||
| this.size = size; | ||
| this.filePointer = 0; | ||
| } | ||
|
|
||
| @Override | ||
| public byte readByte() throws IOException { | ||
| byte[] buffer = new byte[1]; | ||
| int numberOfBytesRead = inputStream.read(buffer); | ||
| if (numberOfBytesRead != -1) { | ||
| filePointer += numberOfBytesRead; | ||
| } | ||
| return buffer[0]; | ||
| } | ||
|
|
||
| @Override | ||
| public void readBytes(byte[] b, int offset, int len) throws IOException { | ||
| int bytesRead = inputStream.read(b, offset, len); | ||
| if (bytesRead == len) { | ||
| filePointer += bytesRead; | ||
| } else { | ||
| while (bytesRead > 0 && bytesRead < len) { | ||
| filePointer += bytesRead; | ||
| len -= bytesRead; | ||
| offset += bytesRead; | ||
| bytesRead = inputStream.read(b, offset, len); | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -61,22 +71,25 @@ public long length() { | |
| return size; | ||
| } | ||
|
|
||
| /** | ||
| * Guaranteed to throw an exception and leave the RemoteIndexInput unmodified. | ||
| * This method is not implemented as it is not used for the file transfer to/from the remote store. | ||
| * | ||
| * @throws UnsupportedOperationException always | ||
| */ | ||
| @Override | ||
| public void seek(long pos) throws IOException { | ||
|
There was a problem hiding this comment. Just curious, why There was a problem hiding this comment. The implementation was incorrect. This was pointed by @andrross earlier. seek() should provide functionality to go to the exact point provided by the |
||
| throw new UnsupportedOperationException(); | ||
| } | ||
|
|
||
| /** | ||
| * Returns the current position in this file in terms of number of bytes read so far. | ||
| */ | ||
| @Override | ||
| public long getFilePointer() { | ||
| return filePointer; | ||
| } | ||
|
|
||
| /** | ||
| * Guaranteed to throw an exception and leave the RemoteIndexInput unmodified. | ||
| * This method is not implemented as it is not used for the file transfer to/from the remote store. | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
would have preferred individual dependencies as supplier rather than a full
IndexSharddependency, to protect against misuseThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah, I thought about that but there are different IndexShard methods (
getOperationPrimaryTerm,getSegmentInfosSnapshot) that are getting called in theafterRefresh()flow.Created #4316 to track this.