Skip to content

Commit 4746880

Browse files
lukewhitinglukewhiting
authored
DLM Frozen Health Indicator - Health Cache Publishing (#158278)
* DLM Frozen Health Indicator - Health Cache Publishing * Update docs/changelog/158278.yaml * Simplify down the stats being reported
1 parent 4953a94 commit 4746880

18 files changed

Lines changed: 1098 additions & 40 deletions

‎docs/changelog/158278.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
area: Data streams
2+
issues: []
3+
pr: 158278
4+
summary: DLM Frozen Health Indicator - Health Cache Publishing
5+
type: enhancement

‎server/src/main/java/org/elasticsearch/health/node/DlmFrozenTransitionsHealthInfo.java‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the "Elastic License
4+
* 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
5+
* Public License v 1"; you may not use this file except in compliance with, at
6+
* your election, the "Elastic License 2.0", the "GNU Affero General Public
7+
* License v3.0 only", or the "Server Side Public License, v 1".
8+
*/
9+
10+
package org.elasticsearch.health.node;
11+
12+
import org.elasticsearch.cluster.metadata.ProjectId;
13+
import org.elasticsearch.common.io.stream.StreamInput;
14+
import org.elasticsearch.common.io.stream.StreamOutput;
15+
import org.elasticsearch.common.io.stream.Writeable;
16+
17+
import java.io.IOException;
18+
import java.util.Locale;
19+
import java.util.Map;
20+
import java.util.stream.Collectors;
21+
22+
/**
23+
* Represents the health of the DLM (data stream lifecycle) frozen-tier transition feature, as evaluated on the
24+
* elected master node.
25+
*
26+
* @param transitionsEnabled Whether the DLM frozen transition feature is enabled. When {@code false}, no new
27+
* transitions will be submitted, though in-flight transitions continue to completion.
28+
* @param serviceRunning Whether the DLM frozen transition service's periodic scheduler is running on the
29+
* current master. Detected via the scheduler's {@link java.util.concurrent.ScheduledFuture}:
30+
* {@code isDone()} becomes {@code true} if the task dies from an unhandled
31+
* {@link Error}, which {@code isShutdown()} on the executor cannot detect.
32+
* @param defaultRepositoryConfigured Whether a default snapshot repository ({@code repositories.default_repository}) is
33+
* configured. Without one, eligible indices cannot be marked for frozen conversion.
34+
* @param overdueIndices A capped sample of indices, keyed by project then index name, that are past their
35+
* {@code frozen_after} age by more than the configured stuck threshold and have not
36+
* completed their frozen-tier transition, together with their current transition
37+
* state. {@code totalOverdueIndicesCount} may exceed the number of entries here.
38+
* @param totalOverdueIndicesCount The total number of overdue indices found across all projects, regardless of
39+
* whether they fit in {@code overdueIndices}.
40+
* @param generatedAtMillis Epoch-millisecond timestamp at which the master built this snapshot. Used to
41+
* detect stale data (e.g. after a master failover before the new master has
42+
* published its first snapshot).
43+
* @param publishIntervalMillis The publisher's configured interval. The indicator treats the snapshot as stale
44+
* when {@code now - generatedAtMillis > STALE_AFTER_PUBLISH_INTERVALS * publishIntervalMillis}.
45+
*/
46+
public record DlmFrozenTransitionsHealthInfo(
47+
boolean transitionsEnabled,
48+
boolean serviceRunning,
49+
boolean defaultRepositoryConfigured,
50+
Map<ProjectId, Map<String, TransitionState>> overdueIndices,
51+
int totalOverdueIndicesCount,
52+
long generatedAtMillis,
53+
long publishIntervalMillis
54+
) implements Writeable {
55+
56+
public DlmFrozenTransitionsHealthInfo {
57+
overdueIndices = overdueIndices.entrySet()
58+
.stream()
59+
.collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> Map.copyOf(e.getValue())));
60+
}
61+
62+
public DlmFrozenTransitionsHealthInfo(StreamInput in) throws IOException {
63+
this(
64+
in.readBoolean(),
65+
in.readBoolean(),
66+
in.readBoolean(),
67+
in.readMap(ProjectId::readFrom, i -> i.readMap(v -> v.readEnum(TransitionState.class))),
68+
in.readVInt(),
69+
in.readVLong(),
70+
in.readVLong()
71+
);
72+
}
73+
74+
@Override
75+
public void writeTo(StreamOutput out) throws IOException {
76+
out.writeBoolean(transitionsEnabled);
77+
out.writeBoolean(serviceRunning);
78+
out.writeBoolean(defaultRepositoryConfigured);
79+
out.writeMap(overdueIndices, (o, id) -> id.writeTo(o), (o, m) -> o.writeMap(m, StreamOutput::writeEnum));
80+
out.writeVInt(totalOverdueIndicesCount);
81+
out.writeVLong(generatedAtMillis);
82+
out.writeVLong(publishIntervalMillis);
83+
}
84+
85+
/**
86+
* The transition state of an overdue index, as tracked by the transition executor on the current master node.
87+
* {@code UNMARKED} and {@code MARKED} are derived from durable cluster state; {@code QUEUED} and {@code RUNNING}
88+
* are best-effort and reset to {@code MARKED} across a master failover.
89+
*/
90+
public enum TransitionState {
91+
UNMARKED,
92+
MARKED,
93+
QUEUED,
94+
RUNNING;
95+
96+
@Override
97+
public String toString() {
98+
return name().toLowerCase(Locale.ROOT);
99+
}
100+
}
101+
}

‎server/src/main/java/org/elasticsearch/health/node/HealthInfo.java‎

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,34 +25,51 @@
2525
/**
2626
* This class wraps all the data returned by the health node.
2727
*
28-
* @param diskInfoByNode A Map of node id to DiskHealthInfo for that node
29-
* @param dslHealthInfo The data stream lifecycle health information
30-
* @param repositoriesInfoByNode A Map of node id to RepositoriesHealthInfo for that node
31-
* @param fileSettingsHealthInfo The file-based settings health information
28+
* @param diskInfoByNode A Map of node id to DiskHealthInfo for that node
29+
* @param dslHealthInfo The data stream lifecycle health information
30+
* @param repositoriesInfoByNode A Map of node id to RepositoriesHealthInfo for that node
31+
* @param fileSettingsHealthInfo The file-based settings health information
32+
* @param dlmFrozenTransitionsHealthInfo The DLM frozen-tier transition health information
3233
*/
3334
public record HealthInfo(
3435
Map<String, DiskHealthInfo> diskInfoByNode,
3536
@Nullable DataStreamLifecycleHealthInfo dslHealthInfo,
3637
Map<String, RepositoriesHealthInfo> repositoriesInfoByNode,
37-
FileSettingsHealthInfo fileSettingsHealthInfo
38+
FileSettingsHealthInfo fileSettingsHealthInfo,
39+
@Nullable DlmFrozenTransitionsHealthInfo dlmFrozenTransitionsHealthInfo
3840
) implements Writeable {
3941

4042
public static final HealthInfo EMPTY_HEALTH_INFO = new HealthInfo(Map.of(), NO_DSL_ERRORS, Map.of(), INDETERMINATE);
4143

4244
private static final TransportVersion FILE_SETTINGS_HEALTH_INFO = TransportVersion.fromName("file_settings_health_info");
45+
private static final TransportVersion DLM_FROZEN_TRANSITIONS_HEALTH_INFO = TransportVersion.fromName(
46+
"dlm_frozen_transitions_health_info"
47+
);
4348

4449
public HealthInfo {
4550
requireNonNull(fileSettingsHealthInfo);
4651
}
4752

53+
public HealthInfo(
54+
Map<String, DiskHealthInfo> diskInfoByNode,
55+
@Nullable DataStreamLifecycleHealthInfo dslHealthInfo,
56+
Map<String, RepositoriesHealthInfo> repositoriesInfoByNode,
57+
FileSettingsHealthInfo fileSettingsHealthInfo
58+
) {
59+
this(diskInfoByNode, dslHealthInfo, repositoriesInfoByNode, fileSettingsHealthInfo, null);
60+
}
61+
4862
public HealthInfo(StreamInput input) throws IOException {
4963
this(
5064
input.readMap(DiskHealthInfo::new),
5165
input.readOptionalWriteable(DataStreamLifecycleHealthInfo::new),
5266
input.readMap(RepositoriesHealthInfo::new),
5367
input.getTransportVersion().supports(FILE_SETTINGS_HEALTH_INFO)
5468
? input.readOptionalWriteable(FileSettingsHealthInfo::new)
55-
: INDETERMINATE
69+
: INDETERMINATE,
70+
input.getTransportVersion().supports(DLM_FROZEN_TRANSITIONS_HEALTH_INFO)
71+
? input.readOptionalWriteable(DlmFrozenTransitionsHealthInfo::new)
72+
: null
5673
);
5774
}
5875

@@ -64,5 +81,8 @@ public void writeTo(StreamOutput output) throws IOException {
6481
if (output.getTransportVersion().supports(FILE_SETTINGS_HEALTH_INFO)) {
6582
output.writeOptionalWriteable(fileSettingsHealthInfo);
6683
}
84+
if (output.getTransportVersion().supports(DLM_FROZEN_TRANSITIONS_HEALTH_INFO)) {
85+
output.writeOptionalWriteable(dlmFrozenTransitionsHealthInfo);
86+
}
6787
}
6888
}

‎server/src/main/java/org/elasticsearch/health/node/HealthInfoCache.java‎

Lines changed: 38 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,8 @@ public class HealthInfoCache implements ClusterStateListener {
3636
private volatile ConcurrentHashMap<String, RepositoriesHealthInfo> repositoriesInfoByNode = new ConcurrentHashMap<>();
3737
private volatile FileSettingsHealthInfo fileSettingsHealthInfo = INDETERMINATE;
3838
@Nullable
39+
private volatile DlmFrozenTransitionsHealthInfo dlmFrozenTransitionsHealthInfo = null;
40+
@Nullable
3941
private volatile String masterNodeId = null;
4042

4143
private HealthInfoCache() {}
@@ -52,6 +54,17 @@ public void updateNodeHealth(
5254
@Nullable DataStreamLifecycleHealthInfo latestDslHealthInfo,
5355
@Nullable RepositoriesHealthInfo repositoriesHealthInfo,
5456
@Nullable FileSettingsHealthInfo fileSettingsHealthInfo
57+
) {
58+
updateNodeHealth(nodeId, diskHealthInfo, latestDslHealthInfo, repositoriesHealthInfo, fileSettingsHealthInfo, null);
59+
}
60+
61+
public void updateNodeHealth(
62+
String nodeId,
63+
@Nullable DiskHealthInfo diskHealthInfo,
64+
@Nullable DataStreamLifecycleHealthInfo latestDslHealthInfo,
65+
@Nullable RepositoriesHealthInfo repositoriesHealthInfo,
66+
@Nullable FileSettingsHealthInfo fileSettingsHealthInfo,
67+
@Nullable DlmFrozenTransitionsHealthInfo latestDlmFrozenTransitionsHealthInfo
5568
) {
5669
if (diskHealthInfo != null) {
5770
diskInfoByNode.put(nodeId, diskHealthInfo);
@@ -69,6 +82,9 @@ public void updateNodeHealth(
6982
this.fileSettingsHealthInfo = fileSettingsHealthInfo;
7083
}
7184
}
85+
if (latestDlmFrozenTransitionsHealthInfo != null) {
86+
this.dlmFrozenTransitionsHealthInfo = latestDlmFrozenTransitionsHealthInfo;
87+
}
7288
}
7389

7490
@Override
@@ -90,18 +106,24 @@ public void clusterChanged(ClusterChangedEvent event) {
90106
// Processing a delayed update after the cache has been emptied because
91107
// the node is not the health node anymore has small impact since it will
92108
// be reset in the next round again.
93-
} else if (diskInfoByNode.isEmpty() == false
109+
} else if (hasAnyHealthInfo()) {
110+
logger.debug("Node [{}][{}] is no longer the health node, emptying the cache.", localNode.getName(), localNode.getId());
111+
diskInfoByNode = new ConcurrentHashMap<>();
112+
dslHealthInfo = null;
113+
repositoriesInfoByNode = new ConcurrentHashMap<>();
114+
fileSettingsHealthInfo = INDETERMINATE;
115+
masterNodeId = null;
116+
dlmFrozenTransitionsHealthInfo = null;
117+
}
118+
}
119+
120+
private boolean hasAnyHealthInfo() {
121+
return diskInfoByNode.isEmpty() == false
94122
|| dslHealthInfo != null
95123
|| repositoriesInfoByNode.isEmpty() == false
96124
|| fileSettingsHealthInfo != INDETERMINATE
97-
|| masterNodeId != null) {
98-
logger.debug("Node [{}][{}] is no longer the health node, emptying the cache.", localNode.getName(), localNode.getId());
99-
diskInfoByNode = new ConcurrentHashMap<>();
100-
dslHealthInfo = null;
101-
repositoriesInfoByNode = new ConcurrentHashMap<>();
102-
fileSettingsHealthInfo = INDETERMINATE;
103-
masterNodeId = null;
104-
}
125+
|| masterNodeId != null
126+
|| dlmFrozenTransitionsHealthInfo != null;
105127
}
106128

107129
/**
@@ -110,6 +132,12 @@ public void clusterChanged(ClusterChangedEvent event) {
110132
*/
111133
public HealthInfo getHealthInfo() {
112134
// A shallow copy is enough because the inner data is immutable.
113-
return new HealthInfo(Map.copyOf(diskInfoByNode), dslHealthInfo, Map.copyOf(repositoriesInfoByNode), fileSettingsHealthInfo);
135+
return new HealthInfo(
136+
Map.copyOf(diskInfoByNode),
137+
dslHealthInfo,
138+
Map.copyOf(repositoriesInfoByNode),
139+
fileSettingsHealthInfo,
140+
dlmFrozenTransitionsHealthInfo
141+
);
114142
}
115143
}

‎server/src/main/java/org/elasticsearch/health/node/UpdateHealthInfoCacheAction.java‎

Lines changed: 46 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,9 @@ public class UpdateHealthInfoCacheAction extends ActionType<AcknowledgedResponse
4242
private static final Logger logger = LogManager.getLogger(UpdateHealthInfoCacheAction.class);
4343

4444
private static final TransportVersion FILE_SETTINGS_HEALTH_INFO = TransportVersion.fromName("file_settings_health_info");
45+
private static final TransportVersion DLM_FROZEN_TRANSITIONS_HEALTH_INFO = TransportVersion.fromName(
46+
"dlm_frozen_transitions_health_info"
47+
);
4548

4649
public static class Request extends HealthNodeRequest {
4750
private final String nodeId;
@@ -53,19 +56,23 @@ public static class Request extends HealthNodeRequest {
5356
private final RepositoriesHealthInfo repositoriesHealthInfo;
5457
@Nullable
5558
private final FileSettingsHealthInfo fileSettingsHealthInfo;
59+
@Nullable
60+
private final DlmFrozenTransitionsHealthInfo dlmFrozenTransitionsHealthInfo;
5661

5762
public Request(
5863
String nodeId,
5964
DiskHealthInfo diskHealthInfo,
6065
DataStreamLifecycleHealthInfo dslHealthInfo,
6166
RepositoriesHealthInfo repositoriesHealthInfo,
62-
@Nullable FileSettingsHealthInfo fileSettingsHealthInfo
67+
@Nullable FileSettingsHealthInfo fileSettingsHealthInfo,
68+
@Nullable DlmFrozenTransitionsHealthInfo dlmFrozenTransitionsHealthInfo
6369
) {
6470
this.nodeId = nodeId;
6571
this.diskHealthInfo = diskHealthInfo;
6672
this.dslHealthInfo = dslHealthInfo;
6773
this.repositoriesHealthInfo = repositoriesHealthInfo;
6874
this.fileSettingsHealthInfo = fileSettingsHealthInfo;
75+
this.dlmFrozenTransitionsHealthInfo = dlmFrozenTransitionsHealthInfo;
6976
}
7077

7178
public Request(String nodeId, DataStreamLifecycleHealthInfo dslHealthInfo) {
@@ -74,6 +81,7 @@ public Request(String nodeId, DataStreamLifecycleHealthInfo dslHealthInfo) {
7481
this.repositoriesHealthInfo = null;
7582
this.dslHealthInfo = dslHealthInfo;
7683
this.fileSettingsHealthInfo = null;
84+
this.dlmFrozenTransitionsHealthInfo = null;
7785
}
7886

7987
public Request(StreamInput in) throws IOException {
@@ -85,6 +93,9 @@ public Request(StreamInput in) throws IOException {
8593
this.fileSettingsHealthInfo = in.getTransportVersion().supports(FILE_SETTINGS_HEALTH_INFO)
8694
? in.readOptionalWriteable(FileSettingsHealthInfo::new)
8795
: null;
96+
this.dlmFrozenTransitionsHealthInfo = in.getTransportVersion().supports(DLM_FROZEN_TRANSITIONS_HEALTH_INFO)
97+
? in.readOptionalWriteable(DlmFrozenTransitionsHealthInfo::new)
98+
: null;
8899
}
89100

90101
public String getNodeId() {
@@ -108,6 +119,11 @@ public FileSettingsHealthInfo getFileSettingsHealthInfo() {
108119
return fileSettingsHealthInfo;
109120
}
110121

122+
@Nullable
123+
public DlmFrozenTransitionsHealthInfo getDlmFrozenTransitionsHealthInfo() {
124+
return dlmFrozenTransitionsHealthInfo;
125+
}
126+
111127
@Override
112128
public ActionRequestValidationException validate() {
113129
return null;
@@ -123,6 +139,9 @@ public void writeTo(StreamOutput out) throws IOException {
123139
if (out.getTransportVersion().supports(FILE_SETTINGS_HEALTH_INFO)) {
124140
out.writeOptionalWriteable(fileSettingsHealthInfo);
125141
}
142+
if (out.getTransportVersion().supports(DLM_FROZEN_TRANSITIONS_HEALTH_INFO)) {
143+
out.writeOptionalWriteable(dlmFrozenTransitionsHealthInfo);
144+
}
126145
}
127146

128147
@Override
@@ -150,12 +169,20 @@ public boolean equals(Object o) {
150169
&& Objects.equals(diskHealthInfo, request.diskHealthInfo)
151170
&& Objects.equals(dslHealthInfo, request.dslHealthInfo)
152171
&& Objects.equals(repositoriesHealthInfo, request.repositoriesHealthInfo)
153-
&& Objects.equals(fileSettingsHealthInfo, request.fileSettingsHealthInfo);
172+
&& Objects.equals(fileSettingsHealthInfo, request.fileSettingsHealthInfo)
173+
&& Objects.equals(dlmFrozenTransitionsHealthInfo, request.dlmFrozenTransitionsHealthInfo);
154174
}
155175

156176
@Override
157177
public int hashCode() {
158-
return Objects.hash(nodeId, diskHealthInfo, dslHealthInfo, repositoriesHealthInfo, fileSettingsHealthInfo);
178+
return Objects.hash(
179+
nodeId,
180+
diskHealthInfo,
181+
dslHealthInfo,
182+
repositoriesHealthInfo,
183+
fileSettingsHealthInfo,
184+
dlmFrozenTransitionsHealthInfo
185+
);
159186
}
160187

161188
public static class Builder {
@@ -164,6 +191,7 @@ public static class Builder {
164191
private RepositoriesHealthInfo repositoriesHealthInfo;
165192
private DataStreamLifecycleHealthInfo dslHealthInfo;
166193
private FileSettingsHealthInfo fileSettingsHealthInfo;
194+
private DlmFrozenTransitionsHealthInfo dlmFrozenTransitionsHealthInfo;
167195

168196
public Builder nodeId(String nodeId) {
169197
this.nodeId = nodeId;
@@ -190,8 +218,20 @@ public Builder fileSettingsHealthInfo(FileSettingsHealthInfo fileSettingsHealthI
190218
return this;
191219
}
192220

221+
public Builder dlmFrozenTransitionsHealthInfo(DlmFrozenTransitionsHealthInfo dlmFrozenTransitionsHealthInfo) {
222+
this.dlmFrozenTransitionsHealthInfo = dlmFrozenTransitionsHealthInfo;
223+
return this;
224+
}
225+
193226
public Request build() {
194-
return new Request(nodeId, diskHealthInfo, dslHealthInfo, repositoriesHealthInfo, fileSettingsHealthInfo);
227+
return new Request(
228+
nodeId,
229+
diskHealthInfo,
230+
dslHealthInfo,
231+
repositoriesHealthInfo,
232+
fileSettingsHealthInfo,
233+
dlmFrozenTransitionsHealthInfo
234+
);
195235
}
196236
}
197237
}
@@ -245,7 +285,8 @@ protected void healthOperation(
245285
request.getDiskHealthInfo(),
246286
request.getDslHealthInfo(),
247287
request.getRepositoriesHealthInfo(),
248-
request.getFileSettingsHealthInfo()
288+
request.getFileSettingsHealthInfo(),
289+
request.getDlmFrozenTransitionsHealthInfo()
249290
);
250291
listener.onResponse(AcknowledgedResponse.of(true));
251292
}

‎server/src/main/resources/transport/definitions/referable/dlm_frozen_transitions_health_info.csv‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
9527000

0 commit comments

Comments
 (0)