Skip to content

Commit

Permalink
[Remote Store] Optimize segments metadata upload
Browse files Browse the repository at this point in the history 
Signed-off-by: Ashish Singh <ssashish@amazon.com>
  • Loading branch information
@ashking94
ashking94 committed Jun 3, 2023
1 parent 5d5e8ad commit e2f69d7
Show file tree
Hide file tree
Showing 2 changed files with 54 additions and 60 deletions.
4 changes: 4 additions & 0 deletions server/src/main/java/org/opensearch/index/remote/RemoteRefreshSegmentTracker.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -346,6 +346,10 @@ public void addToLatestUploadedFiles(String file) {
computeBytesLag();
}

public Set<String> getLatestUploadedFiles() {
return new HashSet<>(latestUploadedFiles);
}

public void setLatestUploadedFiles(Set<String> files) {
this.latestUploadedFiles.clear();
this.latestUploadedFiles.addAll(files);
Expand Down
110 changes: 50 additions & 60 deletions server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.IndexOutput;
import org.opensearch.action.bulk.BackoffPolicy;
import org.opensearch.common.CheckedFunction;
import org.opensearch.common.concurrent.GatedCloseable;
import org.opensearch.common.unit.TimeValue;
import org.opensearch.common.util.concurrent.ConcurrentCollections;
Expand Down Expand Up @@ -152,11 +151,13 @@ public void onFailure(String file) {
// Track upload failure
segmentTracker.addUploadBytesFailed(latestFileNameSizeOnLocalMap.get(file));
}
}, remoteDirectory, storeDirectory, this::getChecksumOfLocalFile);
}, remoteDirectory, storeDirectory);
}

@Override
public void beforeRefresh() throws IOException {}
public void beforeRefresh() throws IOException {

}

/**
* Upload new segment files created as part of the last refresh to the remote segment store.
Expand Down Expand Up @@ -230,23 +231,25 @@ private synchronized void syncSegments(boolean isRetry) {
// Create a map of file name to size and update the refresh segment tracker
updateLocalSizeMapAndTracker(localSegmentsPostRefresh);

// Start the segments files upload
boolean newSegmentsUploadStatus = uploadNewSegments(localSegmentsPostRefresh);
if (newSegmentsUploadStatus) {
segmentInfoSnapshotFilename = uploadSegmentInfosSnapshot(latestSegmentInfos.get(), segmentInfos);
localSegmentsPostRefresh.add(segmentInfoSnapshotFilename);

// Start metadata file upload
remoteDirectory.uploadMetadata(
localSegmentsPostRefresh,
storeDirectory,
indexShard.getOperationPrimaryTerm(),
segmentInfos.getGeneration()
);
clearStaleFilesFromLocalSegmentChecksumMap(localSegmentsPostRefresh);
onSuccessfulSegmentsSync(refreshTimeMs, refreshSeqNo);
List<String> segmentFilesToUpload = localSegmentsPostRefresh.stream()
.filter(f -> skipUpload(f) == false)
.collect(Collectors.toList());

if (segmentFilesToUpload.isEmpty() || uploadNewSegments(segmentFilesToUpload)) {
if (segmentTracker.getLatestUploadedFiles().equals(localSegmentsPostRefresh) == false) {
segmentInfoSnapshotFilename = uploadSegmentInfosSnapshot(latestSegmentInfos.get(), segmentInfos);
localSegmentsPostRefresh.add(segmentInfoSnapshotFilename);
// Start metadata file upload
remoteDirectory.uploadMetadata(
localSegmentsPostRefresh,
storeDirectory,
indexShard.getOperationPrimaryTerm(),
segmentInfos.getGeneration()
);
clearStaleFilesFromLocalSegmentChecksumMap(localSegmentsPostRefresh);
onSuccessfulSegmentsSync(refreshTimeMs, refreshSeqNo, checkpoint);
}
indexShard.getEngine().translogManager().setMinSeqNoToKeep(lastRefreshedCheckpoint + 1);
checkpointPublisher.publish(indexShard, checkpoint);
// At this point since we have uploaded new segments, segment infos and segment metadata file,
// along with marking minSeqNoToKeep, upload has succeeded completely.
shouldRetry = false;
Expand Down Expand Up @@ -283,11 +286,7 @@ private synchronized void syncSegments(boolean isRetry) {
* @param localSegmentsPostRefresh list of segment files present post refresh
*/
private void clearStaleFilesFromLocalSegmentChecksumMap(Collection<String> localSegmentsPostRefresh) {
localSegmentChecksumMap.keySet()
.stream()
.filter(file -> !localSegmentsPostRefresh.contains(file))
.collect(Collectors.toSet())
.forEach(localSegmentChecksumMap::remove);
localSegmentChecksumMap.entrySet().removeIf(entry -> localSegmentsPostRefresh.contains(entry.getKey()) == false);
}

private void beforeSegmentsSync(boolean isRetry) {
Expand All @@ -298,7 +297,8 @@ private void beforeSegmentsSync(boolean isRetry) {
segmentTracker.incrementTotalUploadsStarted();
}

private void onSuccessfulSegmentsSync(long refreshTimeMs, long refreshSeqNo) {
private void onSuccessfulSegmentsSync(long refreshTimeMs, long refreshSeqNo, ReplicationCheckpoint checkpoint) {
checkpointPublisher.publish(indexShard, checkpoint);
// Update latest uploaded segment files name in segment tracker
segmentTracker.setLatestUploadedFiles(latestFileNameSizeOnLocalMap.keySet());
// Update the remote refresh time and refresh seq no
Expand Down Expand Up @@ -365,6 +365,25 @@ String uploadSegmentInfosSnapshot(String latestSegmentsNFilename, SegmentInfos s
return segmentInfoSnapshotFilename;
}

/**
* Whether to upload a file or not depending on whether file is in excluded list or has been already uploaded.
*
* @param file that needs to be uploaded.
* @return true if the upload has to be skipped for the file.
*/
private boolean skipUpload(String file) {
try {
// Exclude files that are already uploaded and the exclude files to come up with the list of files to be uploaded.
return EXCLUDE_FILES.contains(file) || remoteDirectory.containsFile(file, getChecksumOfLocalFile(file));
} catch (IOException e) {
logger.error(
"Exception while reading checksum of local segment file: {}, ignoring the exception and re-uploading the file",
file
);
}
return false;
}

private boolean uploadNewSegments(Collection<String> localSegmentsPostRefresh) throws IOException {
AtomicBoolean uploadSuccess = new AtomicBoolean(true);
localSegmentsPostRefresh.forEach(file -> {
Expand Down Expand Up @@ -446,11 +465,12 @@ private void updateFinalUploadStatusInSegmentTracker(boolean uploadStatus, long
if (uploadStatus) {
long bytesUploaded = segmentTracker.getUploadBytesSucceeded() - bytesBeforeUpload;
long timeTakenInMS = (System.nanoTime() - startTimeInNS) / 1_000_000L;

segmentTracker.incrementTotalUploadsSucceeded();
segmentTracker.addUploadBytes(bytesUploaded);
segmentTracker.addUploadBytesPerSec((bytesUploaded * 1_000L) / timeTakenInMS);
segmentTracker.addUploadTimeMs(timeTakenInMS);
if (bytesUploaded > 0) {
segmentTracker.addUploadBytes(bytesUploaded);
segmentTracker.addUploadBytesPerSec((bytesUploaded * 1_000L) / timeTakenInMS);
segmentTracker.addUploadTimeMs(timeTakenInMS);
}
} else {
segmentTracker.incrementTotalUploadsFailed();
}
Expand All @@ -470,18 +490,10 @@ private static class FileUploader {

private final Directory storeDirectory;

private final CheckedFunction<String, String, IOException> checksumProvider;

public FileUploader(
UploadTracker uploadTracker,
RemoteSegmentStoreDirectory remoteDirectory,
Directory storeDirectory,
CheckedFunction<String, String, IOException> checksumProvider
) {
public FileUploader(UploadTracker uploadTracker, RemoteSegmentStoreDirectory remoteDirectory, Directory storeDirectory) {
this.uploadTracker = uploadTracker;
this.remoteDirectory = remoteDirectory;
this.storeDirectory = storeDirectory;
this.checksumProvider = checksumProvider;
}

/**
Expand All @@ -492,9 +504,6 @@ public FileUploader(
* @throws IOException is thrown if the upload fails.
*/
private void uploadFile(String file) throws IOException {
if (skipUpload(file)) {
return;
}
uploadTracker.beforeUpload(file);
boolean success = false;
try {
Expand All @@ -508,25 +517,6 @@ private void uploadFile(String file) throws IOException {
}
}

/**
* Whether to upload a file or not depending on whether file is in excluded list or has been already uploaded.
*
* @param file that needs to be uploaded.
* @return true if the upload has to be skipped for the file.
*/
private boolean skipUpload(String file) {
try {
// Exclude files that are already uploaded and the exclude files to come up with the list of files to be uploaded.
return EXCLUDE_FILES.contains(file) || remoteDirectory.containsFile(file, checksumProvider.apply(file));
} catch (IOException e) {
logger.error(
"Exception while reading checksum of local segment file: {}, ignoring the exception and re-uploading the file",
file
);
}
return false;
}

/**
* This method does the actual upload.
*
Expand Down

0 comments on commit e2f69d7

Please sign in to comment.