Skip to content

Commit aec44fe

Browse files
DaveCTurnerDaveCTurner
committed
Decouple DiskThresholdMonitor & ClusterInfoService (#44105)
Today the `ClusterInfoService` requires the `DiskThresholdMonitor` at construction time so that it can notify it when nodes report changes in their disk usage, but this is awkward to construct: the `DiskThresholdMonitor` requires a `RerouteService` which requires an `AllocationService` which comees from the `ClusterModule` which requires the `ClusterInfoService`. Today we break the cycle with a `LazilyInitializedRerouteService` which is itself a little ugly. This commit replaces this with a more traditional subject/observer relationship between the `ClusterInfoService` and the `DiskThresholdMonitor`.
1 parent e70cad4 commit aec44fe

File tree

8 files changed

+84
-128
lines changed

8 files changed

+84
-128
lines changed

‎server/src/main/java/org/elasticsearch/cluster/ClusterInfoService.java

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,23 @@
1919

2020
package org.elasticsearch.cluster;
2121

22+
import java.util.function.Consumer;
23+
2224
/**
23-
* Interface for a class used to gather information about a cluster at
24-
* regular intervals
25+
* Interface for a class used to gather information about a cluster periodically.
2526
*/
27+
@FunctionalInterface
2628
public interface ClusterInfoService {
2729

28-
/** The latest cluster information */
30+
/**
31+
* @return the latest cluster information
32+
*/
2933
ClusterInfo getClusterInfo();
34+
35+
/**
36+
* Add a listener for new cluster information
37+
*/
38+
default void addListener(Consumer<ClusterInfo> clusterInfoConsumer) {
39+
throw new UnsupportedOperationException();
40+
}
3041
}

‎server/src/main/java/org/elasticsearch/cluster/EmptyClusterInfoService.java

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,10 @@
1919

2020
package org.elasticsearch.cluster;
2121

22+
import java.util.function.Consumer;
23+
2224
/**
23-
* ClusterInfoService that provides empty maps for disk usage and shard sizes
25+
* {@link ClusterInfoService} that provides empty maps for disk usage and shard sizes
2426
*/
2527
public class EmptyClusterInfoService implements ClusterInfoService {
2628
public static final EmptyClusterInfoService INSTANCE = new EmptyClusterInfoService();
@@ -29,4 +31,9 @@ public class EmptyClusterInfoService implements ClusterInfoService {
2931
public ClusterInfo getClusterInfo() {
3032
return ClusterInfo.EMPTY;
3133
}
34+
35+
@Override
36+
public void addListener(Consumer<ClusterInfo> clusterInfoConsumer) {
37+
// never updated, so we can discard the listener
38+
}
3239
}

‎server/src/main/java/org/elasticsearch/cluster/InternalClusterInfoService.java

Lines changed: 20 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import org.apache.logging.log4j.LogManager;
2323
import org.apache.logging.log4j.Logger;
24+
import org.apache.logging.log4j.message.ParameterizedMessage;
2425
import org.elasticsearch.action.ActionListener;
2526
import org.elasticsearch.action.LatchedActionListener;
2627
import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
@@ -46,6 +47,8 @@
4647
import org.elasticsearch.threadpool.ThreadPool;
4748
import org.elasticsearch.transport.ReceiveTimeoutTransportException;
4849

50+
import java.util.ArrayList;
51+
import java.util.Collections;
4952
import java.util.List;
5053
import java.util.concurrent.CountDownLatch;
5154
import java.util.concurrent.TimeUnit;
@@ -85,10 +88,9 @@ public class InternalClusterInfoService implements ClusterInfoService, LocalNode
8588
private final ClusterService clusterService;
8689
private final ThreadPool threadPool;
8790
private final NodeClient client;
88-
private final Consumer<ClusterInfo> listener;
91+
private final List<Consumer<ClusterInfo>> listeners = Collections.synchronizedList(new ArrayList<>(1));
8992

90-
public InternalClusterInfoService(Settings settings, ClusterService clusterService, ThreadPool threadPool, NodeClient client,
91-
Consumer<ClusterInfo> listener) {
93+
public InternalClusterInfoService(Settings settings, ClusterService clusterService, ThreadPool threadPool, NodeClient client) {
9294
this.leastAvailableSpaceUsages = ImmutableOpenMap.of();
9395
this.mostAvailableSpaceUsages = ImmutableOpenMap.of();
9496
this.shardRoutingToDataPath = ImmutableOpenMap.of();
@@ -109,7 +111,6 @@ public InternalClusterInfoService(Settings settings, ClusterService clusterServi
109111
this.clusterService.addLocalNodeMasterListener(this);
110112
// Add to listen for state changes (when nodes are added)
111113
this.clusterService.addListener(this);
112-
this.listener = listener;
113114
}
114115

115116
private void setEnabled(boolean enabled) {
@@ -356,14 +357,25 @@ public void onFailure(Exception e) {
356357
Thread.currentThread().interrupt(); // restore interrupt status
357358
}
358359
ClusterInfo clusterInfo = getClusterInfo();
359-
try {
360-
listener.accept(clusterInfo);
361-
} catch (Exception e) {
362-
logger.info("Failed executing ClusterInfoService listener", e);
360+
boolean anyListeners = false;
361+
for (final Consumer<ClusterInfo> listener : listeners) {
362+
anyListeners = true;
363+
try {
364+
logger.trace("notifying [{}] of new cluster info", listener);
365+
listener.accept(clusterInfo);
366+
} catch (Exception e) {
367+
logger.info(new ParameterizedMessage("failed to notify [{}] of new cluster info", listener), e);
368+
}
363369
}
370+
assert anyListeners : "expected to notify at least one listener";
364371
return clusterInfo;
365372
}
366373

374+
@Override
375+
public void addListener(Consumer<ClusterInfo> clusterInfoConsumer) {
376+
listeners.add(clusterInfoConsumer);
377+
}
378+
367379
static void buildShardLevelInfo(Logger logger, ShardStats[] stats, ImmutableOpenMap.Builder<String, Long> newShardSizes,
368380
ImmutableOpenMap.Builder<ShardRouting, String> newShardRoutingToDataPath) {
369381
for (ShardStats s : stats) {

‎server/src/main/java/org/elasticsearch/cluster/routing/LazilyInitializedRerouteService.java

Lines changed: 0 additions & 42 deletions
This file was deleted.

‎server/src/main/java/org/elasticsearch/node/Node.java

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,8 @@
2626
import org.elasticsearch.Build;
2727
import org.elasticsearch.ElasticsearchException;
2828
import org.elasticsearch.ElasticsearchTimeoutException;
29-
import org.elasticsearch.action.ActionType;
3029
import org.elasticsearch.action.ActionModule;
30+
import org.elasticsearch.action.ActionType;
3131
import org.elasticsearch.action.admin.cluster.snapshots.status.TransportNodesSnapshotsStatus;
3232
import org.elasticsearch.action.search.SearchExecutionStatsCollector;
3333
import org.elasticsearch.action.search.SearchPhaseController;
@@ -38,7 +38,6 @@
3838
import org.elasticsearch.bootstrap.BootstrapContext;
3939
import org.elasticsearch.client.Client;
4040
import org.elasticsearch.client.node.NodeClient;
41-
import org.elasticsearch.cluster.ClusterInfo;
4241
import org.elasticsearch.cluster.ClusterInfoService;
4342
import org.elasticsearch.cluster.ClusterModule;
4443
import org.elasticsearch.cluster.ClusterName;
@@ -57,7 +56,6 @@
5756
import org.elasticsearch.cluster.node.DiscoveryNode;
5857
import org.elasticsearch.cluster.node.DiscoveryNodeRole;
5958
import org.elasticsearch.cluster.routing.BatchedRerouteService;
60-
import org.elasticsearch.cluster.routing.LazilyInitializedRerouteService;
6159
import org.elasticsearch.cluster.routing.RerouteService;
6260
import org.elasticsearch.cluster.routing.allocation.DiskThresholdMonitor;
6361
import org.elasticsearch.cluster.service.ClusterService;
@@ -180,7 +178,6 @@
180178
import java.util.Set;
181179
import java.util.concurrent.CountDownLatch;
182180
import java.util.concurrent.TimeUnit;
183-
import java.util.function.Consumer;
184181
import java.util.function.Function;
185182
import java.util.function.UnaryOperator;
186183
import java.util.stream.Collectors;
@@ -370,11 +367,7 @@ protected Node(
370367
.newHashPublisher());
371368
final IngestService ingestService = new IngestService(clusterService, threadPool, this.environment,
372369
scriptModule.getScriptService(), analysisModule.getAnalysisRegistry(), pluginsService.filterPlugins(IngestPlugin.class));
373-
final LazilyInitializedRerouteService lazilyInitializedRerouteService = new LazilyInitializedRerouteService();
374-
final DiskThresholdMonitor diskThresholdMonitor = new DiskThresholdMonitor(settings, clusterService::state,
375-
clusterService.getClusterSettings(), client, threadPool::relativeTimeInMillis, lazilyInitializedRerouteService);
376-
final ClusterInfoService clusterInfoService = newClusterInfoService(settings, clusterService, threadPool, client,
377-
diskThresholdMonitor::onNewInfo);
370+
final ClusterInfoService clusterInfoService = newClusterInfoService(settings, clusterService, threadPool, client);
378371
final UsageService usageService = new UsageService();
379372

380373
ModulesBuilder modules = new ModulesBuilder();
@@ -511,7 +504,10 @@ protected Node(
511504

512505
final RerouteService rerouteService
513506
= new BatchedRerouteService(clusterService, clusterModule.getAllocationService()::reroute);
514-
lazilyInitializedRerouteService.setRerouteService(rerouteService);
507+
final DiskThresholdMonitor diskThresholdMonitor = new DiskThresholdMonitor(settings, clusterService::state,
508+
clusterService.getClusterSettings(), client, threadPool::relativeTimeInMillis, rerouteService);
509+
clusterInfoService.addListener(diskThresholdMonitor::onNewInfo);
510+
515511
final DiscoveryModule discoveryModule = new DiscoveryModule(settings, threadPool, transportService, namedWriteableRegistry,
516512
networkService, clusterService.getMasterService(), clusterService.getClusterApplierService(),
517513
clusterService.getClusterSettings(), pluginsService.filterPlugins(DiscoveryPlugin.class),
@@ -1021,8 +1017,8 @@ private List<NetworkService.CustomNameResolver> getCustomNameResolvers(List<Disc
10211017

10221018
/** Constructs a ClusterInfoService which may be mocked for tests. */
10231019
protected ClusterInfoService newClusterInfoService(Settings settings, ClusterService clusterService,
1024-
ThreadPool threadPool, NodeClient client, Consumer<ClusterInfo> listeners) {
1025-
return new InternalClusterInfoService(settings, clusterService, threadPool, client, listeners);
1020+
ThreadPool threadPool, NodeClient client) {
1021+
return new InternalClusterInfoService(settings, clusterService, threadPool, client);
10261022
}
10271023

10281024
/** Constructs a {@link org.elasticsearch.http.HttpServerTransport} which may be mocked for tests. */

‎server/src/test/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDeciderTests.java

Lines changed: 25 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -100,12 +100,9 @@ public void testDiskThreshold() {
100100
new SameShardAllocationDecider(Settings.EMPTY, clusterSettings),
101101
makeDecider(diskSettings))));
102102

103-
ClusterInfoService cis = new ClusterInfoService() {
104-
@Override
105-
public ClusterInfo getClusterInfo() {
106-
logger.info("--> calling fake getClusterInfo");
107-
return clusterInfo;
108-
}
103+
ClusterInfoService cis = () -> {
104+
logger.info("--> calling fake getClusterInfo");
105+
return clusterInfo;
109106
};
110107
AllocationService strategy = new AllocationService(deciders,
111108
new TestGatewayAllocator(), new BalancedShardsAllocator(Settings.EMPTY), cis);
@@ -278,12 +275,9 @@ public void testDiskThresholdWithAbsoluteSizes() {
278275
new SameShardAllocationDecider(Settings.EMPTY, clusterSettings),
279276
makeDecider(diskSettings))));
280277

281-
ClusterInfoService cis = new ClusterInfoService() {
282-
@Override
283-
public ClusterInfo getClusterInfo() {
284-
logger.info("--> calling fake getClusterInfo");
285-
return clusterInfo;
286-
}
278+
ClusterInfoService cis = () -> {
279+
logger.info("--> calling fake getClusterInfo");
280+
return clusterInfo;
287281
};
288282

289283
AllocationService strategy = new AllocationService(deciders, new TestGatewayAllocator(),
@@ -328,12 +322,9 @@ public ClusterInfo getClusterInfo() {
328322
usagesBuilder.put(nodeWithoutPrimary, new DiskUsage(nodeWithoutPrimary, "", "/dev/null", 100, 35)); // 65% used
329323
usages = usagesBuilder.build();
330324
final ClusterInfo clusterInfo2 = new DevNullClusterInfo(usages, usages, shardSizes);
331-
cis = new ClusterInfoService() {
332-
@Override
333-
public ClusterInfo getClusterInfo() {
334-
logger.info("--> calling fake getClusterInfo");
335-
return clusterInfo2;
336-
}
325+
cis = () -> {
326+
logger.info("--> calling fake getClusterInfo");
327+
return clusterInfo2;
337328
};
338329
strategy = new AllocationService(deciders, new TestGatewayAllocator(),
339330
new BalancedShardsAllocator(Settings.EMPTY), cis);
@@ -516,12 +507,9 @@ Settings.EMPTY, new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLU
516507
),
517508
makeDecider(diskSettings))));
518509

519-
ClusterInfoService cis = new ClusterInfoService() {
520-
@Override
521-
public ClusterInfo getClusterInfo() {
522-
logger.info("--> calling fake getClusterInfo");
523-
return clusterInfo;
524-
}
510+
ClusterInfoService cis = () -> {
511+
logger.info("--> calling fake getClusterInfo");
512+
return clusterInfo;
525513
};
526514

527515
AllocationService strategy = new AllocationService(deciders, new TestGatewayAllocator(),
@@ -580,12 +568,9 @@ Settings.EMPTY, new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLU
580568
),
581569
makeDecider(diskSettings))));
582570

583-
ClusterInfoService cis = new ClusterInfoService() {
584-
@Override
585-
public ClusterInfo getClusterInfo() {
586-
logger.info("--> calling fake getClusterInfo");
587-
return clusterInfo;
588-
}
571+
ClusterInfoService cis = () -> {
572+
logger.info("--> calling fake getClusterInfo");
573+
return clusterInfo;
589574
};
590575

591576
AllocationService strategy = new AllocationService(deciders, new TestGatewayAllocator(),
@@ -677,12 +662,9 @@ public void testShardRelocationsTakenIntoAccount() {
677662
Settings.EMPTY, new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)
678663
), decider)));
679664

680-
ClusterInfoService cis = new ClusterInfoService() {
681-
@Override
682-
public ClusterInfo getClusterInfo() {
683-
logger.info("--> calling fake getClusterInfo");
684-
return clusterInfo;
685-
}
665+
ClusterInfoService cis = () -> {
666+
logger.info("--> calling fake getClusterInfo");
667+
return clusterInfo;
686668
};
687669

688670
AllocationService strategy = new AllocationService(deciders, new TestGatewayAllocator(),
@@ -856,12 +838,9 @@ public void testCanRemainWithShardRelocatingAway() {
856838
}
857839

858840
// Creating AllocationService instance and the services it depends on...
859-
ClusterInfoService cis = new ClusterInfoService() {
860-
@Override
861-
public ClusterInfo getClusterInfo() {
862-
logger.info("--> calling fake getClusterInfo");
863-
return clusterInfo;
864-
}
841+
ClusterInfoService cis = () -> {
842+
logger.info("--> calling fake getClusterInfo");
843+
return clusterInfo;
865844
};
866845
AllocationDeciders deciders = new AllocationDeciders(new HashSet<>(Arrays.asList(
867846
new SameShardAllocationDecider(
@@ -948,13 +927,10 @@ public void testForSingleDataNode() {
948927

949928
// Two shards should start happily
950929
assertThat(decision.type(), equalTo(Decision.Type.YES));
951-
assertThat(((Decision.Single) decision).getExplanation(), containsString("there is only a single data node present"));
952-
ClusterInfoService cis = new ClusterInfoService() {
953-
@Override
954-
public ClusterInfo getClusterInfo() {
955-
logger.info("--> calling fake getClusterInfo");
956-
return clusterInfo;
957-
}
930+
assertThat(decision.getExplanation(), containsString("there is only a single data node present"));
931+
ClusterInfoService cis = () -> {
932+
logger.info("--> calling fake getClusterInfo");
933+
return clusterInfo;
958934
};
959935

960936
AllocationDeciders deciders = new AllocationDeciders(new HashSet<>(Arrays.asList(

‎test/framework/src/main/java/org/elasticsearch/cluster/MockInternalClusterInfoService.java

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,6 @@
1818
*/
1919
package org.elasticsearch.cluster;
2020

21-
import java.util.Arrays;
22-
import java.util.Collections;
23-
import java.util.concurrent.CountDownLatch;
24-
import java.util.function.Consumer;
25-
2621
import org.elasticsearch.Version;
2722
import org.elasticsearch.action.ActionListener;
2823
import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
@@ -40,6 +35,10 @@
4035
import org.elasticsearch.test.ESTestCase;
4136
import org.elasticsearch.threadpool.ThreadPool;
4237

38+
import java.util.Arrays;
39+
import java.util.Collections;
40+
import java.util.concurrent.CountDownLatch;
41+
4342
import static java.util.Collections.emptyMap;
4443
import static java.util.Collections.emptySet;
4544

@@ -71,9 +70,8 @@ public static NodeStats makeStats(String nodeName, DiskUsage usage) {
7170
null, null, null, null);
7271
}
7372

74-
public MockInternalClusterInfoService(Settings settings, ClusterService clusterService, ThreadPool threadPool, NodeClient client,
75-
Consumer<ClusterInfo> listener) {
76-
super(settings, clusterService, threadPool, client, listener);
73+
public MockInternalClusterInfoService(Settings settings, ClusterService clusterService, ThreadPool threadPool, NodeClient client) {
74+
super(settings, clusterService, threadPool, client);
7775
this.clusterName = ClusterName.CLUSTER_NAME_SETTING.get(settings);
7876
stats[0] = makeStats("node_t1", new DiskUsage("node_t1", "n1", "/dev/null", 100, 100));
7977
stats[1] = makeStats("node_t2", new DiskUsage("node_t2", "n2", "/dev/null", 100, 100));

0 commit comments

Comments
 (0)