Skip to content

Commit 6d17e91

Browse files
spinscalespinscale
committed
Watcher: Fix check for currently executed watches (#31137)
The ack watch action has a check for currently executed watches, to make sure that currently running watches cannot be acknowledged. This check only checked on the coordinating node for watches being executed, but should have checked the whole cluster using a WatcherStatsRequest, which is being switched to in this commit.
1 parent e80686c commit 6d17e91

File tree

2 files changed

+87
-62
lines changed

2 files changed

+87
-62
lines changed

‎x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/ack/TransportAckWatchAction.java‎

Lines changed: 57 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -29,13 +29,13 @@
2929
import org.elasticsearch.threadpool.ThreadPool;
3030
import org.elasticsearch.transport.TransportService;
3131
import org.elasticsearch.xpack.core.watcher.actions.ActionWrapper;
32-
import org.elasticsearch.xpack.core.watcher.execution.WatchExecutionSnapshot;
3332
import org.elasticsearch.xpack.core.watcher.transport.actions.ack.AckWatchAction;
3433
import org.elasticsearch.xpack.core.watcher.transport.actions.ack.AckWatchRequest;
3534
import org.elasticsearch.xpack.core.watcher.transport.actions.ack.AckWatchResponse;
35+
import org.elasticsearch.xpack.core.watcher.transport.actions.stats.WatcherStatsAction;
36+
import org.elasticsearch.xpack.core.watcher.transport.actions.stats.WatcherStatsRequest;
3637
import org.elasticsearch.xpack.core.watcher.watch.Watch;
3738
import org.elasticsearch.xpack.core.watcher.watch.WatchField;
38-
import org.elasticsearch.xpack.watcher.execution.ExecutionService;
3939
import org.elasticsearch.xpack.watcher.transport.actions.WatcherTransportAction;
4040
import org.elasticsearch.xpack.watcher.watch.WatchParser;
4141
import org.joda.time.DateTime;
@@ -53,85 +53,88 @@ public class TransportAckWatchAction extends WatcherTransportAction<AckWatchRequ
5353

5454
private final Clock clock;
5555
private final WatchParser parser;
56-
private ExecutionService executionService;
5756
private final Client client;
5857

5958
@Inject
6059
public TransportAckWatchAction(Settings settings, TransportService transportService, ThreadPool threadPool, ActionFilters actionFilters,
6160
IndexNameExpressionResolver indexNameExpressionResolver, Clock clock, XPackLicenseState licenseState,
62-
WatchParser parser, ExecutionService executionService, Client client, ClusterService clusterService) {
61+
WatchParser parser, Client client, ClusterService clusterService) {
6362
super(settings, AckWatchAction.NAME, transportService, threadPool, actionFilters, indexNameExpressionResolver,
6463
licenseState, clusterService, AckWatchRequest::new, AckWatchResponse::new);
6564
this.clock = clock;
6665
this.parser = parser;
67-
this.executionService = executionService;
6866
this.client = client;
6967
}
7068

7169
@Override
7270
protected void masterOperation(AckWatchRequest request, ClusterState state,
7371
ActionListener<AckWatchResponse> listener) throws Exception {
74-
// if the watch to be acked is running currently, reject this request
75-
List<WatchExecutionSnapshot> snapshots = executionService.currentExecutions();
76-
boolean isWatchRunning = snapshots.stream().anyMatch(s -> s.watchId().equals(request.getWatchId()));
77-
if (isWatchRunning) {
78-
listener.onFailure(new ElasticsearchStatusException("watch[{}] is running currently, cannot ack until finished",
72+
WatcherStatsRequest watcherStatsRequest = new WatcherStatsRequest();
73+
watcherStatsRequest.includeCurrentWatches(true);
74+
75+
executeAsyncWithOrigin(client, WATCHER_ORIGIN, WatcherStatsAction.INSTANCE, watcherStatsRequest, ActionListener.wrap(response -> {
76+
boolean isWatchRunning = response.getNodes().stream()
77+
.anyMatch(node -> node.getSnapshots().stream().anyMatch(snapshot -> snapshot.watchId().equals(request.getWatchId())));
78+
if (isWatchRunning) {
79+
listener.onFailure(new ElasticsearchStatusException("watch[{}] is running currently, cannot ack until finished",
7980
RestStatus.CONFLICT, request.getWatchId()));
80-
return;
81-
}
82-
83-
GetRequest getRequest = new GetRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId())
84-
.preference(Preference.LOCAL.type()).realtime(true);
85-
86-
executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, getRequest,
87-
ActionListener.<GetResponse>wrap((response) -> {
88-
if (response.isExists() == false) {
89-
listener.onFailure(new ResourceNotFoundException("Watch with id [{}] does not exist", request.getWatchId()));
90-
} else {
91-
DateTime now = new DateTime(clock.millis(), UTC);
92-
Watch watch = parser.parseWithSecrets(request.getWatchId(), true, response.getSourceAsBytesRef(),
81+
} else {
82+
GetRequest getRequest = new GetRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId())
83+
.preference(Preference.LOCAL.type()).realtime(true);
84+
85+
executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, getRequest,
86+
ActionListener.<GetResponse>wrap(getResponse -> {
87+
if (getResponse.isExists() == false) {
88+
listener.onFailure(new ResourceNotFoundException("Watch with id [{}] does not exist", request.getWatchId()));
89+
} else {
90+
DateTime now = new DateTime(clock.millis(), UTC);
91+
Watch watch = parser.parseWithSecrets(request.getWatchId(), true, getResponse.getSourceAsBytesRef(),
9392
now, XContentType.JSON);
94-
watch.version(response.getVersion());
95-
watch.status().version(response.getVersion());
96-
String[] actionIds = request.getActionIds();
97-
if (actionIds == null || actionIds.length == 0) {
98-
actionIds = new String[]{WatchField.ALL_ACTIONS_ID};
99-
}
93+
watch.version(getResponse.getVersion());
94+
watch.status().version(getResponse.getVersion());
95+
String[] actionIds = request.getActionIds();
96+
if (actionIds == null || actionIds.length == 0) {
97+
actionIds = new String[]{WatchField.ALL_ACTIONS_ID};
98+
}
10099

101-
// exit early in case nothing changes
102-
boolean isChanged = watch.ack(now, actionIds);
103-
if (isChanged == false) {
104-
listener.onResponse(new AckWatchResponse(watch.status()));
105-
return;
106-
}
100+
// exit early in case nothing changes
101+
boolean isChanged = watch.ack(now, actionIds);
102+
if (isChanged == false) {
103+
listener.onResponse(new AckWatchResponse(watch.status()));
104+
return;
105+
}
107106

108-
UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId());
109-
// this may reject this action, but prevents concurrent updates from a watch execution
110-
updateRequest.version(response.getVersion());
111-
updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
112-
XContentBuilder builder = jsonBuilder();
113-
builder.startObject()
107+
UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId());
108+
// this may reject this action, but prevents concurrent updates from a watch execution
109+
updateRequest.version(getResponse.getVersion());
110+
updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
111+
XContentBuilder builder = jsonBuilder();
112+
builder.startObject()
114113
.startObject(WatchField.STATUS.getPreferredName())
115114
.startObject("actions");
116115

117-
List<String> actionIdsAsList = Arrays.asList(actionIds);
118-
boolean updateAll = actionIdsAsList.contains("_all");
119-
for (ActionWrapper actionWrapper : watch.actions()) {
120-
if (updateAll || actionIdsAsList.contains(actionWrapper.id())) {
121-
builder.startObject(actionWrapper.id())
116+
List<String> actionIdsAsList = Arrays.asList(actionIds);
117+
boolean updateAll = actionIdsAsList.contains("_all");
118+
for (ActionWrapper actionWrapper : watch.actions()) {
119+
if (updateAll || actionIdsAsList.contains(actionWrapper.id())) {
120+
builder.startObject(actionWrapper.id())
122121
.field("ack", watch.status().actionStatus(actionWrapper.id()).ackStatus(), ToXContent.EMPTY_PARAMS)
123122
.endObject();
123+
}
124124
}
125-
}
126125

127-
builder.endObject().endObject().endObject();
128-
updateRequest.doc(builder);
126+
builder.endObject().endObject().endObject();
127+
updateRequest.doc(builder);
129128

130-
executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, updateRequest,
129+
executeAsyncWithOrigin(client.threadPool().getThreadContext(), WATCHER_ORIGIN, updateRequest,
131130
ActionListener.<UpdateResponse>wrap(
132-
(updateResponse) -> listener.onResponse(new AckWatchResponse(watch.status())),
133-
listener::onFailure), client::update);
134-
}
135-
}, listener::onFailure), client::get);
131+
(updateResponse) -> listener.onResponse(new AckWatchResponse(watch.status())),
132+
listener::onFailure), client::update);
133+
}
134+
}, listener::onFailure), client::get);
135+
136+
}
137+
138+
}, listener::onFailure));
136139
}
137140
}

‎x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/transport/actions/ack/TransportAckWatchActionTests.java‎

Lines changed: 30 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -6,12 +6,16 @@
66
package org.elasticsearch.xpack.watcher.transport.actions.ack;
77

88
import org.elasticsearch.ElasticsearchException;
9+
import org.elasticsearch.Version;
910
import org.elasticsearch.action.ActionListener;
1011
import org.elasticsearch.action.get.GetResponse;
1112
import org.elasticsearch.action.support.ActionFilters;
13+
import org.elasticsearch.action.support.ContextPreservingActionListener;
1214
import org.elasticsearch.action.support.PlainActionFuture;
1315
import org.elasticsearch.client.Client;
16+
import org.elasticsearch.cluster.ClusterName;
1417
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
18+
import org.elasticsearch.cluster.node.DiscoveryNode;
1519
import org.elasticsearch.cluster.service.ClusterService;
1620
import org.elasticsearch.common.bytes.BytesArray;
1721
import org.elasticsearch.common.settings.Settings;
@@ -22,11 +26,13 @@
2226
import org.elasticsearch.test.ESTestCase;
2327
import org.elasticsearch.threadpool.ThreadPool;
2428
import org.elasticsearch.transport.TransportService;
29+
import org.elasticsearch.xpack.core.watcher.WatcherMetaData;
2530
import org.elasticsearch.xpack.core.watcher.execution.WatchExecutionSnapshot;
2631
import org.elasticsearch.xpack.core.watcher.transport.actions.ack.AckWatchRequest;
2732
import org.elasticsearch.xpack.core.watcher.transport.actions.ack.AckWatchResponse;
33+
import org.elasticsearch.xpack.core.watcher.transport.actions.stats.WatcherStatsAction;
34+
import org.elasticsearch.xpack.core.watcher.transport.actions.stats.WatcherStatsResponse;
2835
import org.elasticsearch.xpack.core.watcher.watch.Watch;
29-
import org.elasticsearch.xpack.watcher.execution.ExecutionService;
3036
import org.elasticsearch.xpack.watcher.watch.WatchParser;
3137
import org.junit.Before;
3238

@@ -36,14 +42,14 @@
3642

3743
import static org.hamcrest.Matchers.is;
3844
import static org.mockito.Matchers.anyObject;
45+
import static org.mockito.Matchers.eq;
3946
import static org.mockito.Mockito.doAnswer;
4047
import static org.mockito.Mockito.mock;
4148
import static org.mockito.Mockito.when;
4249

4350
public class TransportAckWatchActionTests extends ESTestCase {
4451

4552
private TransportAckWatchAction action;
46-
private ExecutionService executionService;
4753
private Client client;
4854

4955
@Before
@@ -53,13 +59,13 @@ public void setupAction() {
5359
ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
5460
when(threadPool.getThreadContext()).thenReturn(threadContext);
5561
WatchParser watchParser = mock(WatchParser.class);
56-
executionService = mock(ExecutionService.class);
5762
ClusterService clusterService = mock(ClusterService.class);
5863
client = mock(Client.class);
5964
when(client.threadPool()).thenReturn(threadPool);
6065
action = new TransportAckWatchAction(Settings.EMPTY, transportService, threadPool,
6166
new ActionFilters(Collections.emptySet()), new IndexNameExpressionResolver(Settings.EMPTY),
62-
Clock.systemUTC(), new XPackLicenseState(Settings.EMPTY), watchParser, executionService, client, clusterService);
67+
Clock.systemUTC(), new XPackLicenseState(Settings.EMPTY), watchParser, client, clusterService);
68+
when(client.threadPool()).thenReturn(threadPool);
6369
}
6470

6571
public void testWatchNotFound() throws Exception {
@@ -71,6 +77,13 @@ public void testWatchNotFound() throws Exception {
7177
return null;
7278
}).when(client).get(anyObject(), anyObject());
7379

80+
doAnswer(invocation -> {
81+
ContextPreservingActionListener listener = (ContextPreservingActionListener) invocation.getArguments()[2];
82+
listener.onResponse(new WatcherStatsResponse(new ClusterName("clusterName"), new WatcherMetaData(false),
83+
Collections.emptyList(), Collections.emptyList()));
84+
return null;
85+
}).when(client).execute(eq(WatcherStatsAction.INSTANCE), anyObject(), anyObject());
86+
7487
AckWatchRequest ackWatchRequest = new AckWatchRequest(watchId);
7588
PlainActionFuture<AckWatchResponse> listener = PlainActionFuture.newFuture();
7689
action.masterOperation(ackWatchRequest, null, listener);
@@ -82,9 +95,18 @@ public void testWatchNotFound() throws Exception {
8295

8396
public void testThatWatchCannotBeAckedWhileRunning() throws Exception {
8497
String watchId = "my_watch_id";
85-
WatchExecutionSnapshot snapshot = mock(WatchExecutionSnapshot.class);
86-
when(snapshot.watchId()).thenReturn(watchId);
87-
when(executionService.currentExecutions()).thenReturn(Collections.singletonList(snapshot));
98+
99+
doAnswer(invocation -> {
100+
ContextPreservingActionListener listener = (ContextPreservingActionListener) invocation.getArguments()[2];
101+
DiscoveryNode discoveryNode = new DiscoveryNode("node_2", buildNewFakeTransportAddress(), Version.CURRENT);
102+
WatcherStatsResponse.Node node = new WatcherStatsResponse.Node(discoveryNode);
103+
WatchExecutionSnapshot snapshot = mock(WatchExecutionSnapshot.class);
104+
when(snapshot.watchId()).thenReturn(watchId);
105+
node.setSnapshots(Collections.singletonList(snapshot));
106+
listener.onResponse(new WatcherStatsResponse(new ClusterName("clusterName"),
107+
new WatcherMetaData(false), Collections.singletonList(node), Collections.emptyList()));
108+
return null;
109+
}).when(client).execute(eq(WatcherStatsAction.INSTANCE), anyObject(), anyObject());
88110

89111
AckWatchRequest ackWatchRequest = new AckWatchRequest(watchId);
90112
PlainActionFuture<AckWatchResponse> listener = PlainActionFuture.newFuture();
@@ -95,4 +117,4 @@ public void testThatWatchCannotBeAckedWhileRunning() throws Exception {
95117
assertThat(e.getMessage(), is("watch[my_watch_id] is running currently, cannot ack until finished"));
96118
assertThat(e.status(), is(RestStatus.CONFLICT));
97119
}
98-
}
120+
}

0 commit comments

Comments
 (0)