|
105 | 105 | import java.nio.file.NoSuchFileException; |
106 | 106 | import java.nio.file.Path; |
107 | 107 | import java.util.ArrayList; |
| 108 | +import java.util.Collection; |
108 | 109 | import java.util.Collections; |
109 | 110 | import java.util.HashMap; |
110 | 111 | import java.util.Iterator; |
|
122 | 123 | import static java.util.Collections.emptyMap; |
123 | 124 | import static java.util.Collections.unmodifiableMap; |
124 | 125 | import static org.opensearch.index.seqno.SequenceNumbers.LOCAL_CHECKPOINT_KEY; |
| 126 | +import static org.opensearch.index.store.Store.MetadataSnapshot.loadMetadata; |
125 | 127 |
|
126 | 128 | /** |
127 | 129 | * A Store provides plain access to files written by an opensearch index shard. Each shard |
@@ -334,6 +336,51 @@ public MetadataSnapshot getMetadata(SegmentInfos segmentInfos) throws IOExceptio |
334 | 336 | return new MetadataSnapshot(segmentInfos, directory, logger); |
335 | 337 | } |
336 | 338 |
|
| 339 | + /** |
| 340 | + * Segment Replication method - Fetch a map of StoreFileMetadata for segments, ignoring Segment_N files. |
| 341 | + * @param segmentInfos {@link SegmentInfos} from which to compute metadata. |
| 342 | + * @return {@link Map} map file name to {@link StoreFileMetadata}. |
| 343 | + */ |
| 344 | + public Map<String, StoreFileMetadata> getSegmentMetadataMap(SegmentInfos segmentInfos) throws IOException { |
| 345 | + assert indexSettings.isSegRepEnabled(); |
| 346 | + return loadMetadata(segmentInfos, directory, logger, true).fileMetadata; |
| 347 | + } |
| 348 | + |
| 349 | + /** |
| 350 | + * Segment Replication method |
| 351 | + * Returns a diff between the Maps of StoreFileMetadata that can be used for getting list of files to copy over to a replica for segment replication. The returned diff will hold a list of files that are: |
| 352 | + * <ul> |
| 353 | + * <li>identical: they exist in both maps and they can be considered the same ie. they don't need to be recovered</li> |
| 354 | + * <li>different: they exist in both maps but their they are not identical</li> |
| 355 | + * <li>missing: files that exist in the source but not in the target</li> |
| 356 | + * </ul> |
| 357 | + */ |
| 358 | + public static RecoveryDiff segmentReplicationDiff(Map<String, StoreFileMetadata> source, Map<String, StoreFileMetadata> target) { |
| 359 | + final List<StoreFileMetadata> identical = new ArrayList<>(); |
| 360 | + final List<StoreFileMetadata> different = new ArrayList<>(); |
| 361 | + final List<StoreFileMetadata> missing = new ArrayList<>(); |
| 362 | + for (StoreFileMetadata value : source.values()) { |
| 363 | + if (value.name().startsWith(IndexFileNames.SEGMENTS)) { |
| 364 | + continue; |
| 365 | + } |
| 366 | + if (target.containsKey(value.name()) == false) { |
| 367 | + missing.add(value); |
| 368 | + } else { |
| 369 | + final StoreFileMetadata fileMetadata = target.get(value.name()); |
| 370 | + if (fileMetadata.isSame(value)) { |
| 371 | + identical.add(value); |
| 372 | + } else { |
| 373 | + different.add(value); |
| 374 | + } |
| 375 | + } |
| 376 | + } |
| 377 | + return new RecoveryDiff( |
| 378 | + Collections.unmodifiableList(identical), |
| 379 | + Collections.unmodifiableList(different), |
| 380 | + Collections.unmodifiableList(missing) |
| 381 | + ); |
| 382 | + } |
| 383 | + |
337 | 384 | /** |
338 | 385 | * Renames all the given files from the key of the map to the |
339 | 386 | * value of the map. All successfully renamed files are removed from the map in-place. |
@@ -709,31 +756,34 @@ public void cleanupAndVerify(String reason, MetadataSnapshot sourceMetadata) thr |
709 | 756 | } |
710 | 757 |
|
711 | 758 | /** |
712 | | - * This method deletes every file in this store that is not contained in either the remote or local metadata snapshots. |
| 759 | + * Segment Replication method - |
| 760 | + * This method deletes every file in this store that is not referenced by the passed in SegmentInfos or |
| 761 | + * part of the latest on-disk commit point. |
713 | 762 | * This method is used for segment replication when the in memory SegmentInfos can be ahead of the on disk segment file. |
714 | 763 | * In this case files from both snapshots must be preserved. Verification has been done that all files are present on disk. |
715 | 764 | * @param reason the reason for this cleanup operation logged for each deleted file |
716 | | - * @param localSnapshot The local snapshot from in memory SegmentInfos. |
| 765 | + * @param infos {@link SegmentInfos} Files from this infos will be preserved on disk if present. |
717 | 766 | * @throws IllegalStateException if the latest snapshot in this store differs from the given one after the cleanup. |
718 | 767 | */ |
719 | | - public void cleanupAndPreserveLatestCommitPoint(String reason, MetadataSnapshot localSnapshot) throws IOException { |
| 768 | + public void cleanupAndPreserveLatestCommitPoint(String reason, SegmentInfos infos) throws IOException { |
| 769 | + assert indexSettings.isSegRepEnabled(); |
720 | 770 | // fetch a snapshot from the latest on disk Segments_N file. This can be behind |
721 | 771 | // the passed in local in memory snapshot, so we want to ensure files it references are not removed. |
722 | 772 | metadataLock.writeLock().lock(); |
723 | 773 | try (Lock writeLock = directory.obtainLock(IndexWriter.WRITE_LOCK_NAME)) { |
724 | | - cleanupFiles(reason, localSnapshot, getMetadata(readLastCommittedSegmentsInfo())); |
| 774 | + cleanupFiles(reason, getMetadata(readLastCommittedSegmentsInfo()), infos.files(true)); |
725 | 775 | } finally { |
726 | 776 | metadataLock.writeLock().unlock(); |
727 | 777 | } |
728 | 778 | } |
729 | 779 |
|
730 | | - private void cleanupFiles(String reason, MetadataSnapshot localSnapshot, @Nullable MetadataSnapshot additionalSnapshot) |
| 780 | + private void cleanupFiles(String reason, MetadataSnapshot localSnapshot, @Nullable Collection<String> additionalFiles) |
731 | 781 | throws IOException { |
732 | 782 | assert metadataLock.isWriteLockedByCurrentThread(); |
733 | 783 | for (String existingFile : directory.listAll()) { |
734 | 784 | if (Store.isAutogenerated(existingFile) |
735 | 785 | || localSnapshot.contains(existingFile) |
736 | | - || (additionalSnapshot != null && additionalSnapshot.contains(existingFile))) { |
| 786 | + || (additionalFiles != null && additionalFiles.contains(existingFile))) { |
737 | 787 | // don't delete snapshot file, or the checksums file (note, this is extra protection since the Store won't delete |
738 | 788 | // checksum) |
739 | 789 | continue; |
@@ -825,17 +875,9 @@ public void commitSegmentInfos(SegmentInfos latestSegmentInfos, long maxSeqNo, l |
825 | 875 | userData.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(maxSeqNo)); |
826 | 876 | latestSegmentInfos.setUserData(userData, true); |
827 | 877 | latestSegmentInfos.commit(directory()); |
828 | | - |
829 | | - // similar to TrimUnsafeCommits, create a commit with an appending IW, this will delete old commits and ensure all files |
830 | | - // associated with the SegmentInfos.commit are fsynced. |
831 | | - final List<IndexCommit> existingCommits = DirectoryReader.listCommits(directory); |
832 | | - assert existingCommits.isEmpty() == false : "Expected at least one commit but none found"; |
833 | | - final IndexCommit lastIndexCommit = existingCommits.get(existingCommits.size() - 1); |
834 | | - assert latestSegmentInfos.getSegmentsFileName().equals(lastIndexCommit.getSegmentsFileName()); |
835 | | - try (IndexWriter writer = newAppendingIndexWriter(directory, lastIndexCommit)) { |
836 | | - writer.setLiveCommitData(lastIndexCommit.getUserData().entrySet()); |
837 | | - writer.commit(); |
838 | | - } |
| 878 | + directory.sync(latestSegmentInfos.files(true)); |
| 879 | + directory.syncMetaData(); |
| 880 | + cleanupAndPreserveLatestCommitPoint("After commit", latestSegmentInfos); |
839 | 881 | } finally { |
840 | 882 | metadataLock.writeLock().unlock(); |
841 | 883 | } |
@@ -1033,6 +1075,11 @@ static LoadedMetadata loadMetadata(IndexCommit commit, Directory directory, Logg |
1033 | 1075 | } |
1034 | 1076 |
|
1035 | 1077 | static LoadedMetadata loadMetadata(SegmentInfos segmentInfos, Directory directory, Logger logger) throws IOException { |
| 1078 | + return loadMetadata(segmentInfos, directory, logger, false); |
| 1079 | + } |
| 1080 | + |
| 1081 | + static LoadedMetadata loadMetadata(SegmentInfos segmentInfos, Directory directory, Logger logger, boolean ignoreSegmentsFile) |
| 1082 | + throws IOException { |
1036 | 1083 | long numDocs = Lucene.getNumDocs(segmentInfos); |
1037 | 1084 | Map<String, String> commitUserDataBuilder = new HashMap<>(); |
1038 | 1085 | commitUserDataBuilder.putAll(segmentInfos.getUserData()); |
@@ -1067,8 +1114,10 @@ static LoadedMetadata loadMetadata(SegmentInfos segmentInfos, Directory director |
1067 | 1114 | if (maxVersion == null) { |
1068 | 1115 | maxVersion = org.opensearch.Version.CURRENT.minimumIndexCompatibilityVersion().luceneVersion; |
1069 | 1116 | } |
1070 | | - final String segmentsFile = segmentInfos.getSegmentsFileName(); |
1071 | | - checksumFromLuceneFile(directory, segmentsFile, builder, logger, maxVersion, true); |
| 1117 | + if (ignoreSegmentsFile == false) { |
| 1118 | + final String segmentsFile = segmentInfos.getSegmentsFileName(); |
| 1119 | + checksumFromLuceneFile(directory, segmentsFile, builder, logger, maxVersion, true); |
| 1120 | + } |
1072 | 1121 | return new LoadedMetadata(unmodifiableMap(builder), unmodifiableMap(commitUserDataBuilder), numDocs); |
1073 | 1122 | } |
1074 | 1123 |
|
@@ -1148,7 +1197,6 @@ public Map<String, StoreFileMetadata> asMap() { |
1148 | 1197 | * Helper method used to group store files according to segment and commit. |
1149 | 1198 | * |
1150 | 1199 | * @see MetadataSnapshot#recoveryDiff(MetadataSnapshot) |
1151 | | - * @see MetadataSnapshot#segmentReplicationDiff(MetadataSnapshot) |
1152 | 1200 | */ |
1153 | 1201 | private Iterable<List<StoreFileMetadata>> getGroupedFilesIterable() { |
1154 | 1202 | final Map<String, List<StoreFileMetadata>> perSegment = new HashMap<>(); |
@@ -1241,51 +1289,6 @@ public RecoveryDiff recoveryDiff(MetadataSnapshot recoveryTargetSnapshot) { |
1241 | 1289 | return recoveryDiff; |
1242 | 1290 | } |
1243 | 1291 |
|
1244 | | - /** |
1245 | | - * Segment Replication method |
1246 | | - * Returns a diff between the two snapshots that can be used for getting list of files to copy over to a replica for segment replication. The given snapshot is treated as the |
1247 | | - * target and this snapshot as the source. The returned diff will hold a list of files that are: |
1248 | | - * <ul> |
1249 | | - * <li>identical: they exist in both snapshots and they can be considered the same ie. they don't need to be recovered</li> |
1250 | | - * <li>different: they exist in both snapshots but their they are not identical</li> |
1251 | | - * <li>missing: files that exist in the source but not in the target</li> |
1252 | | - * </ul> |
1253 | | - */ |
1254 | | - public RecoveryDiff segmentReplicationDiff(MetadataSnapshot recoveryTargetSnapshot) { |
1255 | | - final List<StoreFileMetadata> identical = new ArrayList<>(); |
1256 | | - final List<StoreFileMetadata> different = new ArrayList<>(); |
1257 | | - final List<StoreFileMetadata> missing = new ArrayList<>(); |
1258 | | - final ArrayList<StoreFileMetadata> identicalFiles = new ArrayList<>(); |
1259 | | - for (List<StoreFileMetadata> segmentFiles : getGroupedFilesIterable()) { |
1260 | | - identicalFiles.clear(); |
1261 | | - boolean consistent = true; |
1262 | | - for (StoreFileMetadata meta : segmentFiles) { |
1263 | | - StoreFileMetadata storeFileMetadata = recoveryTargetSnapshot.get(meta.name()); |
1264 | | - if (storeFileMetadata == null) { |
1265 | | - // Do not consider missing files as inconsistent in SegRep as replicas may lag while primary updates |
1266 | | - // documents and generate new files specific to a segment |
1267 | | - missing.add(meta); |
1268 | | - } else if (storeFileMetadata.isSame(meta) == false) { |
1269 | | - consistent = false; |
1270 | | - different.add(meta); |
1271 | | - } else { |
1272 | | - identicalFiles.add(meta); |
1273 | | - } |
1274 | | - } |
1275 | | - if (consistent) { |
1276 | | - identical.addAll(identicalFiles); |
1277 | | - } else { |
1278 | | - different.addAll(identicalFiles); |
1279 | | - } |
1280 | | - } |
1281 | | - RecoveryDiff recoveryDiff = new RecoveryDiff( |
1282 | | - Collections.unmodifiableList(identical), |
1283 | | - Collections.unmodifiableList(different), |
1284 | | - Collections.unmodifiableList(missing) |
1285 | | - ); |
1286 | | - return recoveryDiff; |
1287 | | - } |
1288 | | - |
1289 | 1292 | /** |
1290 | 1293 | * Returns the number of files in this snapshot |
1291 | 1294 | */ |
|
0 commit comments