Skip to content

Commit da7976b

Browse files
RocMarshalRocMarshal
committed
[FLINK-38229][runtime-web] Enhanced Job History Retention Policies for HistoryServer
1 parent 066a957 commit da7976b

File tree

7 files changed

+322
-20
lines changed

7 files changed

+322
-20
lines changed

‎docs/layouts/shortcodes/generated/history_server_configuration.html‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,12 @@
3232
<td>Integer</td>
3333
<td>The maximum number of jobs to retain in each archive directory defined by `historyserver.archive.fs.dir`. If set to `-1`(default), there is no limit to the number of archives. If set to `0` or less than `-1` HistoryServer will throw an <code class="highlighter-rouge">IllegalConfigurationException</code>. </td>
3434
</tr>
35+
<tr>
36+
<td><h5>historyserver.archive.retained-ttl</h5></td>
37+
<td style="word-wrap: break-word;">(none)</td>
38+
<td>Duration</td>
39+
<td>The time-to-live duration to retain the jobs archived in each archive directory defined by `historyserver.archive.fs.dir`. Note, when there are multiple history server instances with different configurations, they are working independently today and may have conflict configs. This is an existing problem. For the configuration, when there are multiple history servers instances. You can enable the configuration option like following: <ul><li>Using the same `<code class="highlighter-rouge">historyserver.archive.fs.dir</code>` directory as the refresh directories, you should enable and configure this feature in only one HistoryServer instance to avoid errors caused by multiple instances simultaneously cleaning up remote files,</li><li>Or you can keep the value of this configuration consistent across them.</li></ul></td>
40+
</tr>
3541
<tr>
3642
<td><h5>historyserver.log.jobmanager.url-pattern</h5></td>
3743
<td style="word-wrap: break-word;">(none)</td>

‎flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525

2626
import static org.apache.flink.configuration.ConfigOptions.key;
2727
import static org.apache.flink.configuration.description.TextElement.code;
28+
import static org.apache.flink.configuration.description.TextElement.text;
2829

2930
/** The set of configuration options relating to the HistoryServer. */
3031
@PublicEvolving
@@ -143,5 +144,29 @@ public class HistoryServerOptions {
143144
code("IllegalConfigurationException"))
144145
.build());
145146

147+
public static final ConfigOption<Duration> HISTORY_SERVER_RETAINED_TTL =
148+
key("historyserver.archive.retained-ttl")
149+
.durationType()
150+
.noDefaultValue()
151+
.withDescription(
152+
Description.builder()
153+
.text(
154+
String.format(
155+
"The time-to-live duration to retain the jobs archived in each archive directory defined by `%s`. ",
156+
HISTORY_SERVER_ARCHIVE_DIRS.key()))
157+
.text(
158+
"Note, when there are multiple history server instances with different configurations, "
159+
+ "they are working independently today and may have conflict configs. "
160+
+ "This is an existing problem. "
161+
+ "For the configuration, when there are multiple history servers instances. You can enable the configuration option like following: ")
162+
.list(
163+
text(
164+
"Using the same `%s` directory as the refresh directories, "
165+
+ "you should enable and configure this feature in only one HistoryServer instance to avoid errors caused by multiple instances simultaneously cleaning up remote files,",
166+
code(HISTORY_SERVER_ARCHIVE_DIRS.key())),
167+
text(
168+
"Or you can keep the value of this configuration consistent across them."))
169+
.build());
170+
146171
private HistoryServerOptions() {}
147172
}

‎flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java‎

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import org.apache.flink.configuration.Configuration;
2323
import org.apache.flink.configuration.GlobalConfiguration;
2424
import org.apache.flink.configuration.HistoryServerOptions;
25-
import org.apache.flink.configuration.IllegalConfigurationException;
2625
import org.apache.flink.core.fs.FileSystem;
2726
import org.apache.flink.core.fs.Path;
2827
import org.apache.flink.core.plugin.PluginUtils;
@@ -38,6 +37,7 @@
3837
import org.apache.flink.runtime.security.SecurityUtils;
3938
import org.apache.flink.runtime.util.EnvironmentInformation;
4039
import org.apache.flink.runtime.util.Runnables;
40+
import org.apache.flink.runtime.webmonitor.history.retaining.JobArchivedRetainedStrategyImpl;
4141
import org.apache.flink.runtime.webmonitor.utils.LogUrlUtil;
4242
import org.apache.flink.runtime.webmonitor.utils.WebFrontendBootstrap;
4343
import org.apache.flink.util.ExceptionUtils;
@@ -238,19 +238,13 @@ public HistoryServer(
238238

239239
refreshIntervalMillis =
240240
config.get(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_REFRESH_INTERVAL).toMillis();
241-
int maxHistorySize = config.get(HistoryServerOptions.HISTORY_SERVER_RETAINED_JOBS);
242-
if (maxHistorySize == 0 || maxHistorySize < -1) {
243-
throw new IllegalConfigurationException(
244-
"Cannot set %s to 0 or less than -1",
245-
HistoryServerOptions.HISTORY_SERVER_RETAINED_JOBS.key());
246-
}
247241
archiveFetcher =
248242
new HistoryServerArchiveFetcher(
249243
refreshDirs,
250244
webDir,
251245
jobArchiveEventListener,
252246
cleanupExpiredArchives,
253-
maxHistorySize);
247+
JobArchivedRetainedStrategyImpl.createFrom(config));
254248

255249
this.shutdownHook =
256250
ShutdownHookUtil.addShutdownHook(

‎flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
3030
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
3131
import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
32+
import org.apache.flink.runtime.webmonitor.history.retaining.JobArchivesRetainedStrategy;
3233
import org.apache.flink.util.FileUtils;
3334
import org.apache.flink.util.jackson.JacksonMapperFactory;
3435

@@ -112,8 +113,7 @@ public ArchiveEventType getType() {
112113
private final List<HistoryServer.RefreshLocation> refreshDirs;
113114
private final Consumer<ArchiveEvent> jobArchiveEventListener;
114115
private final boolean processExpiredArchiveDeletion;
115-
private final boolean processBeyondLimitArchiveDeletion;
116-
private final int maxHistorySize;
116+
private final JobArchivesRetainedStrategy jobRetainedStrategy;
117117

118118
/** Cache of all available jobs identified by their id. */
119119
private final Map<Path, Set<String>> cachedArchivesPerRefreshDirectory;
@@ -127,13 +127,12 @@ public ArchiveEventType getType() {
127127
File webDir,
128128
Consumer<ArchiveEvent> jobArchiveEventListener,
129129
boolean cleanupExpiredArchives,
130-
int maxHistorySize)
130+
JobArchivesRetainedStrategy jobRetainedStrategy)
131131
throws IOException {
132132
this.refreshDirs = checkNotNull(refreshDirs);
133133
this.jobArchiveEventListener = jobArchiveEventListener;
134134
this.processExpiredArchiveDeletion = cleanupExpiredArchives;
135-
this.maxHistorySize = maxHistorySize;
136-
this.processBeyondLimitArchiveDeletion = this.maxHistorySize > 0;
135+
this.jobRetainedStrategy = checkNotNull(jobRetainedStrategy);
137136
this.cachedArchivesPerRefreshDirectory = new HashMap<>();
138137
for (HistoryServer.RefreshLocation refreshDir : refreshDirs) {
139138
cachedArchivesPerRefreshDirectory.put(refreshDir.getPath(), new HashSet<>());
@@ -159,7 +158,7 @@ void fetchArchives() {
159158
Map<Path, Set<String>> jobsToRemove = new HashMap<>();
160159
cachedArchivesPerRefreshDirectory.forEach(
161160
(path, archives) -> jobsToRemove.put(path, new HashSet<>(archives)));
162-
Map<Path, Set<Path>> archivesBeyondSizeLimit = new HashMap<>();
161+
Map<Path, Set<Path>> archivesBeyondRetainedLimit = new HashMap<>();
163162
for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
164163
Path refreshDir = refreshLocation.getPath();
165164
LOG.debug("Checking archive directory {}.", refreshDir);
@@ -176,7 +175,7 @@ void fetchArchives() {
176175
continue;
177176
}
178177

179-
int historySize = 0;
178+
int fileOrderedIndexOnModifiedTime = 0;
180179
for (FileStatus jobArchive : jobArchives) {
181180
Path jobArchivePath = jobArchive.getPath();
182181
String jobID = jobArchivePath.getName();
@@ -186,9 +185,10 @@ void fetchArchives() {
186185

187186
jobsToRemove.get(refreshDir).remove(jobID);
188187

189-
historySize++;
190-
if (historySize > maxHistorySize && processBeyondLimitArchiveDeletion) {
191-
archivesBeyondSizeLimit
188+
fileOrderedIndexOnModifiedTime++;
189+
if (!jobRetainedStrategy.shouldRetain(
190+
jobArchive, fileOrderedIndexOnModifiedTime)) {
191+
archivesBeyondRetainedLimit
192192
.computeIfAbsent(refreshDir, ignored -> new HashSet<>())
193193
.add(jobArchivePath);
194194
continue;
@@ -220,8 +220,8 @@ void fetchArchives() {
220220
&& processExpiredArchiveDeletion) {
221221
events.addAll(cleanupExpiredJobs(jobsToRemove));
222222
}
223-
if (!archivesBeyondSizeLimit.isEmpty() && processBeyondLimitArchiveDeletion) {
224-
events.addAll(cleanupJobsBeyondSizeLimit(archivesBeyondSizeLimit));
223+
if (!archivesBeyondRetainedLimit.isEmpty()) {
224+
events.addAll(cleanupJobsBeyondSizeLimit(archivesBeyondRetainedLimit));
225225
}
226226
if (!events.isEmpty()) {
227227
updateJobOverview(webOverviewDir, webDir);

‎flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/retaining/JobArchivedRetainedStrategyImpl.java‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.flink.runtime.webmonitor.history.retaining;
20+
21+
import org.apache.flink.configuration.IllegalConfigurationException;
22+
import org.apache.flink.configuration.ReadableConfig;
23+
import org.apache.flink.core.fs.FileStatus;
24+
25+
import javax.annotation.Nullable;
26+
27+
import java.time.Duration;
28+
import java.time.Instant;
29+
import java.util.Arrays;
30+
import java.util.Collections;
31+
import java.util.List;
32+
import java.util.Optional;
33+
34+
import static org.apache.flink.configuration.HistoryServerOptions.HISTORY_SERVER_RETAINED_JOBS;
35+
import static org.apache.flink.configuration.HistoryServerOptions.HISTORY_SERVER_RETAINED_TTL;
36+
37+
/** The retained strategy. */
38+
public class JobArchivedRetainedStrategyImpl implements JobArchivesRetainedStrategy {
39+
40+
public static JobArchivesRetainedStrategy createFrom(ReadableConfig config) {
41+
int maxHistorySizeByOldKey = config.get(HISTORY_SERVER_RETAINED_JOBS);
42+
if (maxHistorySizeByOldKey == 0 || maxHistorySizeByOldKey < -1) {
43+
throw new IllegalConfigurationException(
44+
"Cannot set %s to 0 or less than -1", HISTORY_SERVER_RETAINED_JOBS.key());
45+
}
46+
Optional<Duration> retainedTtlOpt = config.getOptional(HISTORY_SERVER_RETAINED_TTL);
47+
return new JobArchivedRetainedStrategyImpl(
48+
new QuantityBasedJobRetainedStrategy(maxHistorySizeByOldKey),
49+
new TimeToLiveBasedJobRetainedStrategy(retainedTtlOpt.orElse(null)));
50+
}
51+
52+
private final List<JobArchivesRetainedStrategy> strategies;
53+
54+
JobArchivedRetainedStrategyImpl(JobArchivesRetainedStrategy... strategies) {
55+
this.strategies =
56+
strategies == null || strategies.length == 0
57+
? Collections.emptyList()
58+
: Arrays.asList(strategies);
59+
}
60+
61+
@Override
62+
public boolean shouldRetain(FileStatus file, int fileOrderedIndex) {
63+
if (strategies.isEmpty()) {
64+
return true;
65+
}
66+
return strategies.stream().allMatch(s -> s.shouldRetain(file, fileOrderedIndex));
67+
}
68+
}
69+
70+
/** The time to live based retained strategy. */
71+
class TimeToLiveBasedJobRetainedStrategy implements JobArchivesRetainedStrategy {
72+
73+
@Nullable private final Duration ttlThreshold;
74+
75+
TimeToLiveBasedJobRetainedStrategy(Duration ttlThreshold) {
76+
this.ttlThreshold = ttlThreshold;
77+
}
78+
79+
@Override
80+
public boolean shouldRetain(FileStatus file, int fileOrderedIndex) {
81+
if (ttlThreshold == null || ttlThreshold.toMillis() <= 0L) {
82+
return true;
83+
}
84+
return Instant.now().toEpochMilli() - file.getModificationTime() < ttlThreshold.toMillis();
85+
}
86+
}
87+
88+
/** The job quantity based retained strategy. */
89+
class QuantityBasedJobRetainedStrategy implements JobArchivesRetainedStrategy {
90+
91+
private final int quantityThreshold;
92+
93+
QuantityBasedJobRetainedStrategy(int quantityThreshold) {
94+
this.quantityThreshold = quantityThreshold;
95+
}
96+
97+
@Override
98+
public boolean shouldRetain(FileStatus file, int fileOrderedIndex) {
99+
if (quantityThreshold <= 0) {
100+
return true;
101+
}
102+
return quantityThreshold >= fileOrderedIndex;
103+
}
104+
}

‎flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/retaining/JobArchivesRetainedStrategy.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.flink.runtime.webmonitor.history.retaining;
20+
21+
import org.apache.flink.core.fs.FileStatus;
22+
23+
/** To define the strategy interface to judge whether the file should be retained. */
24+
public interface JobArchivesRetainedStrategy {
25+
26+
/**
27+
* Judge whether the file should be retained.
28+
*
29+
* @param file the target file to judge.
30+
* @param fileOrderedIndex the specified order index position of the target file,
31+
* @return The result that indicates whether the file should be retained.
32+
*/
33+
boolean shouldRetain(FileStatus file, int fileOrderedIndex);
34+
}

0 commit comments

Comments
 (0)