Skip to content

Commit c5b39fd

Browse files
dnhatndnhatn
committed
Limit number of retaining translog files for peer recovery (#47414)
Today we control the extra translog (when soft-deletes is disabled) for peer recoveries by size and age. If users manually (force) flush many times within a short period, we can keep many small (or empty) translog files as neither the size or age condition is reached. We can protect the cluster from running out of the file descriptors in such a situation by limiting the number of retaining translog files.
1 parent 08915b6 commit c5b39fd

File tree

11 files changed

+138
-25
lines changed

11 files changed

+138
-25
lines changed

‎server/src/main/java/org/elasticsearch/index/IndexSettings.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -269,6 +269,14 @@ public final class IndexSettings {
269269
Setting.byteSizeSetting("index.translog.retention.size", settings -> INDEX_SOFT_DELETES_SETTING.get(settings) ? "-1" : "512MB",
270270
Property.Dynamic, Property.IndexScope);
271271

272+
/**
273+
* Controls the number of translog files that are no longer needed for persistence reasons will be kept around before being deleted.
274+
* This is a safeguard making sure that the translog deletion policy won't keep too many translog files especially when they're small.
275+
* This setting is intentionally not registered, it is only used in tests
276+
**/
277+
public static final Setting<Integer> INDEX_TRANSLOG_RETENTION_TOTAL_FILES_SETTING =
278+
Setting.intSetting("index.translog.retention.total_files", 100, 0, Setting.Property.IndexScope);
279+
272280
/**
273281
* Controls the maximum length of time since a retention lease is created or renewed before it is considered expired.
274282
*/
@@ -781,6 +789,14 @@ public TimeValue getTranslogRetentionAge() {
781789
return translogRetentionAge;
782790
}
783791

792+
/**
793+
* Returns the maximum number of translog files that that no longer required for persistence should be kept for peer recovery
794+
* when soft-deletes is disabled.
795+
*/
796+
public int getTranslogRetentionTotalFiles() {
797+
return INDEX_TRANSLOG_RETENTION_TOTAL_FILES_SETTING.get(getSettings());
798+
}
799+
784800
/**
785801
* Returns the generation threshold size. As sequence numbers can cause multiple generations to
786802
* be preserved for rollback purposes, we want to keep the size of individual generations from

‎server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -193,7 +193,8 @@ public InternalEngine(EngineConfig engineConfig) {
193193
}
194194
final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(
195195
engineConfig.getIndexSettings().getTranslogRetentionSize().getBytes(),
196-
engineConfig.getIndexSettings().getTranslogRetentionAge().getMillis()
196+
engineConfig.getIndexSettings().getTranslogRetentionAge().getMillis(),
197+
engineConfig.getIndexSettings().getTranslogRetentionTotalFiles()
197198
);
198199
store.incRef();
199200
IndexWriter writer = null;

‎server/src/main/java/org/elasticsearch/index/engine/NoOpEngine.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -151,7 +151,7 @@ public void trimUnreferencedTranslogFiles() {
151151

152152
if (minTranslogGeneration < lastCommitGeneration) {
153153
// a translog deletion policy that retains nothing but the last translog generation from safe commit
154-
final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(-1, -1);
154+
final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(-1, -1, 0);
155155
translogDeletionPolicy.setTranslogGenerationOfLastCommit(lastCommitGeneration);
156156
translogDeletionPolicy.setMinTranslogGenerationForRecovery(lastCommitGeneration);
157157

‎server/src/main/java/org/elasticsearch/index/engine/ReadOnlyEngine.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -227,7 +227,8 @@ private static TranslogStats translogStats(final EngineConfig config, final Segm
227227
final TranslogConfig translogConfig = config.getTranslogConfig();
228228
final TranslogDeletionPolicy translogDeletionPolicy = new TranslogDeletionPolicy(
229229
config.getIndexSettings().getTranslogRetentionSize().getBytes(),
230-
config.getIndexSettings().getTranslogRetentionAge().getMillis()
230+
config.getIndexSettings().getTranslogRetentionAge().getMillis(),
231+
config.getIndexSettings().getTranslogRetentionTotalFiles()
231232
);
232233
translogDeletionPolicy.setTranslogGenerationOfLastCommit(translogGenOfLastCommit);
233234

‎server/src/main/java/org/elasticsearch/index/translog/TranslogDeletionPolicy.java‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,9 +63,12 @@ public void assertNoOpenTranslogRefs() {
6363

6464
private long retentionAgeInMillis;
6565

66-
public TranslogDeletionPolicy(long retentionSizeInBytes, long retentionAgeInMillis) {
66+
private int retentionTotalFiles;
67+
68+
public TranslogDeletionPolicy(long retentionSizeInBytes, long retentionAgeInMillis, int retentionTotalFiles) {
6769
this.retentionSizeInBytes = retentionSizeInBytes;
6870
this.retentionAgeInMillis = retentionAgeInMillis;
71+
this.retentionTotalFiles = retentionTotalFiles;
6972
if (Assertions.ENABLED) {
7073
openTranslogRef = new ConcurrentHashMap<>();
7174
} else {
@@ -100,6 +103,10 @@ public synchronized void setRetentionAgeInMillis(long ageInMillis) {
100103
retentionAgeInMillis = ageInMillis;
101104
}
102105

106+
synchronized void setRetentionTotalFiles(int retentionTotalFiles) {
107+
this.retentionTotalFiles = retentionTotalFiles;
108+
}
109+
103110
/**
104111
* acquires the basis generation for a new snapshot. Any translog generation above, and including, the returned generation
105112
* will not be deleted until the returned {@link Releasable} is closed.
@@ -164,7 +171,8 @@ synchronized long minTranslogGenRequired(List<TranslogReader> readers, TranslogW
164171
} else {
165172
minByAgeAndSize = Math.max(minByAge, minBySize);
166173
}
167-
return Math.min(minByAgeAndSize, Math.min(minByLocks, minTranslogGenerationForRecovery));
174+
long minByNumFiles = getMinTranslogGenByTotalFiles(readers, writer, retentionTotalFiles);
175+
return Math.min(Math.max(minByAgeAndSize, minByNumFiles), Math.min(minByLocks, minTranslogGenerationForRecovery));
168176
}
169177

170178
static long getMinTranslogGenBySize(List<TranslogReader> readers, TranslogWriter writer, long retentionSizeInBytes) {
@@ -196,6 +204,16 @@ static long getMinTranslogGenByAge(List<TranslogReader> readers, TranslogWriter
196204
}
197205
}
198206

207+
static long getMinTranslogGenByTotalFiles(List<TranslogReader> readers, TranslogWriter writer, final int maxTotalFiles) {
208+
long minGen = writer.generation;
209+
int totalFiles = 1; // for the current writer
210+
for (int i = readers.size() - 1; i >= 0 && totalFiles < maxTotalFiles; i--) {
211+
totalFiles++;
212+
minGen = readers.get(i).generation;
213+
}
214+
return minGen;
215+
}
216+
199217
protected long currentTime() {
200218
return System.currentTimeMillis();
201219
}

‎server/src/main/java/org/elasticsearch/index/translog/TruncateTranslogAction.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -178,7 +178,8 @@ private boolean isTranslogClean(ShardPath shardPath, String translogUUID) throws
178178
indexSettings, BigArrays.NON_RECYCLING_INSTANCE);
179179
long primaryTerm = indexSettings.getIndexMetaData().primaryTerm(shardPath.getShardId().id());
180180
// We open translog to check for corruption, do not clean anything.
181-
final TranslogDeletionPolicy retainAllTranslogPolicy = new TranslogDeletionPolicy(Long.MAX_VALUE, Long.MAX_VALUE) {
181+
final TranslogDeletionPolicy retainAllTranslogPolicy = new TranslogDeletionPolicy(
182+
Long.MAX_VALUE, Long.MAX_VALUE, Integer.MAX_VALUE) {
182183
@Override
183184
long minTranslogGenRequired(List<TranslogReader> readers, TranslogWriter writer) {
184185
long minGen = writer.generation;

‎server/src/test/java/org/elasticsearch/index/shard/IndexShardIT.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,7 @@
9797
import java.util.Arrays;
9898
import java.util.Collection;
9999
import java.util.Collections;
100+
import java.util.Comparator;
100101
import java.util.List;
101102
import java.util.Locale;
102103
import java.util.concurrent.BrokenBarrierException;
@@ -136,6 +137,7 @@
136137
import static org.hamcrest.Matchers.greaterThan;
137138
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
138139
import static org.hamcrest.Matchers.instanceOf;
140+
import static org.hamcrest.Matchers.lessThanOrEqualTo;
139141
import static org.hamcrest.Matchers.notNullValue;
140142

141143
public class IndexShardIT extends ESSingleNodeTestCase {
@@ -991,6 +993,43 @@ public void testNoOpEngineFactoryTakesPrecedence() {
991993
}
992994
}
993995

996+
public void testLimitNumberOfRetainingTranslogFiles() throws Exception {
997+
String indexName = "test";
998+
int translogRetentionTotalFiles = randomIntBetween(0, 50);
999+
Settings.Builder settings = Settings.builder()
1000+
.put(SETTING_NUMBER_OF_SHARDS, 1)
1001+
.put(SETTING_NUMBER_OF_REPLICAS, 0)
1002+
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false)
1003+
.put(IndexSettings.INDEX_TRANSLOG_RETENTION_TOTAL_FILES_SETTING.getKey(), translogRetentionTotalFiles);
1004+
if (randomBoolean()) {
1005+
settings.put(IndexSettings.INDEX_TRANSLOG_RETENTION_SIZE_SETTING.getKey(), new ByteSizeValue(between(1, 1024 * 1024)));
1006+
}
1007+
if (randomBoolean()) {
1008+
settings.put(IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING.getKey(), TimeValue.timeValueMillis(between(1, 10_000)));
1009+
}
1010+
IndexService indexService = createIndex(indexName, settings.build());
1011+
IndexShard shard = indexService.getShard(0);
1012+
shard.rollTranslogGeneration();
1013+
CheckedRunnable<IOException> checkTranslog = () -> {
1014+
try (Stream<Path> files = Files.list(getTranslog(shard).location()).sorted(Comparator.reverseOrder())) {
1015+
long totalFiles = files.filter(f -> f.getFileName().toString().endsWith(Translog.TRANSLOG_FILE_SUFFIX)).count();
1016+
assertThat(totalFiles, either(lessThanOrEqualTo((long) translogRetentionTotalFiles)).or(equalTo(1L)));
1017+
}
1018+
};
1019+
for (int i = 0; i < 100; i++) {
1020+
client().prepareIndex(indexName, "_doc", Integer.toString(i)).setSource("{}", XContentType.JSON).get();
1021+
if (randomInt(100) < 10) {
1022+
client().admin().indices().prepareFlush(indexName).setWaitIfOngoing(true).get();
1023+
checkTranslog.run();
1024+
}
1025+
if (randomInt(100) < 10) {
1026+
shard.rollTranslogGeneration();
1027+
}
1028+
}
1029+
client().admin().indices().prepareFlush(indexName).get();
1030+
checkTranslog.run();
1031+
}
1032+
9941033
/**
9951034
* Asserts that there are no files in the specified path
9961035
*/

‎server/src/test/java/org/elasticsearch/index/translog/TranslogDeletionPolicyTests.java‎

Lines changed: 49 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ public void testNoRetention() throws IOException {
4747
List<BaseTranslogReader> allGens = new ArrayList<>(readersAndWriter.v1());
4848
allGens.add(readersAndWriter.v2());
4949
try {
50-
TranslogDeletionPolicy deletionPolicy = new MockDeletionPolicy(now, 0, 0);
50+
TranslogDeletionPolicy deletionPolicy = new MockDeletionPolicy(now, 0, 0, 0);
5151
assertMinGenRequired(deletionPolicy, readersAndWriter, 1L);
5252
final int committedReader = randomIntBetween(0, allGens.size() - 1);
5353
final long committedGen = allGens.get(committedReader).generation;
@@ -98,6 +98,25 @@ public void testAgeRetention() throws IOException {
9898
}
9999
}
100100

101+
public void testTotalFilesRetention() throws Exception {
102+
Tuple<List<TranslogReader>, TranslogWriter> readersAndWriter = createReadersAndWriter(System.currentTimeMillis());
103+
List<BaseTranslogReader> allGens = new ArrayList<>(readersAndWriter.v1());
104+
allGens.add(readersAndWriter.v2());
105+
try {
106+
assertThat(TranslogDeletionPolicy.getMinTranslogGenByTotalFiles(readersAndWriter.v1(), readersAndWriter.v2(),
107+
randomIntBetween(Integer.MIN_VALUE, 1)), equalTo(readersAndWriter.v2().generation));
108+
assertThat(TranslogDeletionPolicy.getMinTranslogGenByTotalFiles(readersAndWriter.v1(), readersAndWriter.v2(),
109+
randomIntBetween(allGens.size(), Integer.MAX_VALUE)), equalTo(allGens.get(0).generation));
110+
int numFiles = randomIntBetween(1, allGens.size());
111+
long selectedGeneration = allGens.get(allGens.size() - numFiles).generation;
112+
assertThat(TranslogDeletionPolicy.getMinTranslogGenByTotalFiles(readersAndWriter.v1(), readersAndWriter.v2(), numFiles),
113+
equalTo(selectedGeneration));
114+
} finally {
115+
IOUtils.close(readersAndWriter.v1());
116+
IOUtils.close(readersAndWriter.v2());
117+
}
118+
}
119+
101120
/**
102121
* Tests that age trumps size but recovery trumps both.
103122
*/
@@ -107,7 +126,7 @@ public void testRetentionHierarchy() throws IOException {
107126
List<BaseTranslogReader> allGens = new ArrayList<>(readersAndWriter.v1());
108127
allGens.add(readersAndWriter.v2());
109128
try {
110-
TranslogDeletionPolicy deletionPolicy = new MockDeletionPolicy(now, Long.MAX_VALUE, Long.MAX_VALUE);
129+
TranslogDeletionPolicy deletionPolicy = new MockDeletionPolicy(now, Long.MAX_VALUE, Long.MAX_VALUE, Integer.MAX_VALUE);
111130
deletionPolicy.setTranslogGenerationOfLastCommit(Long.MAX_VALUE);
112131
deletionPolicy.setMinTranslogGenerationForRecovery(Long.MAX_VALUE);
113132
int selectedReader = randomIntBetween(0, allGens.size() - 1);
@@ -116,33 +135,41 @@ public void testRetentionHierarchy() throws IOException {
116135
selectedReader = randomIntBetween(0, allGens.size() - 1);
117136
final long selectedGenerationBySize = allGens.get(selectedReader).generation;
118137
long size = allGens.stream().skip(selectedReader).map(BaseTranslogReader::sizeInBytes).reduce(Long::sum).get();
138+
selectedReader = randomIntBetween(0, allGens.size() - 1);
139+
final long selectedGenerationByTotalFiles = allGens.get(selectedReader).generation;
119140
deletionPolicy.setRetentionAgeInMillis(maxAge);
120141
deletionPolicy.setRetentionSizeInBytes(size);
121-
assertMinGenRequired(deletionPolicy, readersAndWriter, Math.max(selectedGenerationByAge, selectedGenerationBySize));
142+
final int totalFiles = allGens.size() - selectedReader;
143+
deletionPolicy.setRetentionTotalFiles(totalFiles);
144+
assertMinGenRequired(deletionPolicy, readersAndWriter,
145+
max3(selectedGenerationByAge, selectedGenerationBySize, selectedGenerationByTotalFiles));
122146
// make a new policy as committed gen can't go backwards (for now)
123-
deletionPolicy = new MockDeletionPolicy(now, size, maxAge);
147+
deletionPolicy = new MockDeletionPolicy(now, size, maxAge, totalFiles);
124148
long committedGen = randomFrom(allGens).generation;
125149
deletionPolicy.setTranslogGenerationOfLastCommit(randomLongBetween(committedGen, Long.MAX_VALUE));
126150
deletionPolicy.setMinTranslogGenerationForRecovery(committedGen);
127-
assertMinGenRequired(deletionPolicy, readersAndWriter,
128-
Math.min(committedGen, Math.max(selectedGenerationByAge, selectedGenerationBySize)));
151+
assertMinGenRequired(deletionPolicy, readersAndWriter, Math.min(committedGen,
152+
max3(selectedGenerationByAge, selectedGenerationBySize, selectedGenerationByTotalFiles)));
129153
long viewGen = randomFrom(allGens).generation;
130154
try (Releasable ignored = deletionPolicy.acquireTranslogGen(viewGen)) {
131155
assertMinGenRequired(deletionPolicy, readersAndWriter,
132-
Math.min(
133-
Math.min(committedGen, viewGen),
134-
Math.max(selectedGenerationByAge, selectedGenerationBySize)));
156+
min3(committedGen, viewGen, max3(selectedGenerationByAge, selectedGenerationBySize, selectedGenerationByTotalFiles)));
135157
// disable age
136158
deletionPolicy.setRetentionAgeInMillis(-1);
137-
assertMinGenRequired(deletionPolicy, readersAndWriter, Math.min(Math.min(committedGen, viewGen), selectedGenerationBySize));
159+
assertMinGenRequired(deletionPolicy, readersAndWriter,
160+
min3(committedGen, viewGen, Math.max(selectedGenerationBySize, selectedGenerationByTotalFiles)));
138161
// disable size
139162
deletionPolicy.setRetentionAgeInMillis(maxAge);
140163
deletionPolicy.setRetentionSizeInBytes(-1);
141-
assertMinGenRequired(deletionPolicy, readersAndWriter, Math.min(Math.min(committedGen, viewGen), selectedGenerationByAge));
142-
// disable both
164+
assertMinGenRequired(deletionPolicy, readersAndWriter,
165+
min3(committedGen, viewGen, Math.max(selectedGenerationByAge, selectedGenerationByTotalFiles)));
166+
// disable age and zie
143167
deletionPolicy.setRetentionAgeInMillis(-1);
144168
deletionPolicy.setRetentionSizeInBytes(-1);
145169
assertMinGenRequired(deletionPolicy, readersAndWriter, Math.min(committedGen, viewGen));
170+
// disable total files
171+
deletionPolicy.setRetentionTotalFiles(0);
172+
assertMinGenRequired(deletionPolicy, readersAndWriter, Math.min(committedGen, viewGen));
146173
}
147174
} finally {
148175
IOUtils.close(readersAndWriter.v1());
@@ -191,8 +218,8 @@ private static class MockDeletionPolicy extends TranslogDeletionPolicy {
191218

192219
long now;
193220

194-
MockDeletionPolicy(long now, long retentionSizeInBytes, long maxRetentionAgeInMillis) {
195-
super(retentionSizeInBytes, maxRetentionAgeInMillis);
221+
MockDeletionPolicy(long now, long retentionSizeInBytes, long maxRetentionAgeInMillis, int maxRetentionTotalFiles) {
222+
super(retentionSizeInBytes, maxRetentionAgeInMillis, maxRetentionTotalFiles);
196223
this.now = now;
197224
}
198225

@@ -201,4 +228,12 @@ protected long currentTime() {
201228
return now;
202229
}
203230
}
231+
232+
private static long max3(long x1, long x2, long x3) {
233+
return Math.max(Math.max(x1, x2), x3);
234+
}
235+
236+
private static long min3(long x1, long x2, long x3) {
237+
return Math.min(Math.min(x1, x2), x3);
238+
}
204239
}

‎server/src/test/java/org/elasticsearch/index/translog/TranslogTests.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2265,7 +2265,7 @@ public void testRecoveryFromAFutureGenerationCleansUp() throws IOException {
22652265
// engine blows up, after committing the above generation
22662266
translog.close();
22672267
TranslogConfig config = translog.getConfig();
2268-
final TranslogDeletionPolicy deletionPolicy = new TranslogDeletionPolicy(-1, -1);
2268+
final TranslogDeletionPolicy deletionPolicy = new TranslogDeletionPolicy(-1, -1, 0);
22692269
deletionPolicy.setTranslogGenerationOfLastCommit(randomLongBetween(comittedGeneration, Long.MAX_VALUE));
22702270
deletionPolicy.setMinTranslogGenerationForRecovery(comittedGeneration);
22712271
translog = new Translog(config, translog.getTranslogUUID(), deletionPolicy,
@@ -2324,7 +2324,7 @@ public void testRecoveryFromFailureOnTrimming() throws IOException {
23242324
// expected...
23252325
}
23262326
}
2327-
final TranslogDeletionPolicy deletionPolicy = new TranslogDeletionPolicy(-1, -1);
2327+
final TranslogDeletionPolicy deletionPolicy = new TranslogDeletionPolicy(-1, -1, 0);
23282328
deletionPolicy.setTranslogGenerationOfLastCommit(randomLongBetween(comittedGeneration, Long.MAX_VALUE));
23292329
deletionPolicy.setMinTranslogGenerationForRecovery(comittedGeneration);
23302330
try (Translog translog = new Translog(config, translogUUID, deletionPolicy,

‎test/framework/src/main/java/org/elasticsearch/index/translog/TranslogDeletionPolicies.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,13 +27,14 @@ public class TranslogDeletionPolicies {
2727
public static TranslogDeletionPolicy createTranslogDeletionPolicy() {
2828
return new TranslogDeletionPolicy(
2929
IndexSettings.INDEX_TRANSLOG_RETENTION_SIZE_SETTING.getDefault(Settings.EMPTY).getBytes(),
30-
IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING.getDefault(Settings.EMPTY).getMillis()
30+
IndexSettings.INDEX_TRANSLOG_RETENTION_AGE_SETTING.getDefault(Settings.EMPTY).getMillis(),
31+
IndexSettings.INDEX_TRANSLOG_RETENTION_TOTAL_FILES_SETTING.getDefault(Settings.EMPTY)
3132
);
3233
}
3334

3435
public static TranslogDeletionPolicy createTranslogDeletionPolicy(IndexSettings indexSettings) {
3536
return new TranslogDeletionPolicy(indexSettings.getTranslogRetentionSize().getBytes(),
36-
indexSettings.getTranslogRetentionAge().getMillis());
37+
indexSettings.getTranslogRetentionAge().getMillis(), indexSettings.getTranslogRetentionTotalFiles());
3738
}
3839

3940
}

0 commit comments

Comments
 (0)