Skip to content

Commit

Permalink
Watcher: Mark watcher as started only after loading watches (#30403)
Browse files Browse the repository at this point in the history 
Starting watcher should wait for the watcher to be started before
marking the status as started, which is now done via a callback.

Also, reloading watcher could set the execution service to paused. This could
lead to watches not being executed, when run in tests. This fix does not
change the paused flag in the execution service, just clears out the
current queue and executions.

Closes #30381
  • Loading branch information
@spinscale
spinscale committed May 7, 2018
1 parent 35c5eab commit 38bf2bd
Show file tree
Hide file tree
Showing 6 changed files with 80 additions and 23 deletions.
11 changes: 5 additions & 6 deletions ...plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/WatcherLifeCycleService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -111,8 +111,7 @@ public void clusterChanged(ClusterChangedEvent event) {
// if this is not a data node, we need to start it ourselves possibly
if (event.state().nodes().getLocalNode().isDataNode() == false &&
isWatcherStoppedManually == false && this.state.get() == WatcherState.STOPPED) {
watcherService.start(event.state());
this.state.set(WatcherState.STARTED);
watcherService.start(event.state(), () -> this.state.set(WatcherState.STARTED));
return;
}

Expand Down Expand Up @@ -160,8 +159,8 @@ public void clusterChanged(ClusterChangedEvent event) {
if (state.get() == WatcherState.STARTED) {
watcherService.reload(event.state(), "new local watcher shard allocation ids");
} else if (state.get() == WatcherState.STOPPED) {
watcherService.start(event.state());
this.state.set(WatcherState.STARTED);
this.state.set(WatcherState.STARTING);
watcherService.start(event.state(), () -> this.state.set(WatcherState.STARTED));
}
} else {
clearAllocationIds();
Expand All @@ -172,8 +171,8 @@ public void clusterChanged(ClusterChangedEvent event) {
WatcherState watcherState = this.state.get();
if (event.localNodeMaster()) {
if (watcherState != WatcherState.STARTED && watcherState != WatcherState.STARTING) {
watcherService.start(event.state());
this.state.set(WatcherState.STARTED);
this.state.set(WatcherState.STARTING);
watcherService.start(event.state(), () -> this.state.set(WatcherState.STARTED));
}
} else {
if (watcherState == WatcherState.STARTED || watcherState == WatcherState.STARTING) {
Expand Down
29 changes: 24 additions & 5 deletions x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/WatcherService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -183,23 +183,40 @@ void reload(ClusterState state, String reason) {
// by checking the cluster state version before and after loading the watches we can potentially just exit without applying the
// changes
processedClusterStateVersion.set(state.getVersion());
pauseExecution(reason);
triggerService.pauseExecution();
int cancelledTaskCount = executionService.clearExecutionsAndQueue();
logger.info("reloading watcher, reason [{}], cancelled [{}] queued tasks", reason, cancelledTaskCount);

executor.execute(wrapWatcherService(() -> reloadInner(state, reason, false),
e -> logger.error("error reloading watcher", e)));
}

public void start(ClusterState state) {
/**
* start the watcher service, load watches in the background
*
* @param state the current cluster state
* @param postWatchesLoadedCallback the callback to be triggered, when watches where loaded successfully
*/
public void start(ClusterState state, Runnable postWatchesLoadedCallback) {
executionService.unPause();
processedClusterStateVersion.set(state.getVersion());
executor.execute(wrapWatcherService(() -> reloadInner(state, "starting", true),
executor.execute(wrapWatcherService(() -> {
if (reloadInner(state, "starting", true)) {
postWatchesLoadedCallback.run();
}
},
e -> logger.error("error starting watcher", e)));
}

/**
* reload the watches and start scheduling them
* reload watches and start scheduling them
*
* @param state the current cluster state
* @param reason the reason for reloading, will be logged
* @param loadTriggeredWatches should triggered watches be loaded in this run, not needed for reloading, only for starting
* @return true if no other loading of a newer cluster state happened in parallel, false otherwise
*/
private synchronized void reloadInner(ClusterState state, String reason, boolean loadTriggeredWatches) {
private synchronized boolean reloadInner(ClusterState state, String reason, boolean loadTriggeredWatches) {
// exit early if another thread has come in between
if (processedClusterStateVersion.get() != state.getVersion()) {
logger.debug("watch service has not been reloaded for state [{}], another reload for state [{}] in progress",
Expand All @@ -221,9 +238,11 @@ private synchronized void reloadInner(ClusterState state, String reason, boolean
executionService.executeTriggeredWatches(triggeredWatches);
}
logger.debug("watch service has been reloaded, reason [{}]", reason);
return true;
} else {
logger.debug("watch service has not been reloaded for state [{}], another reload for state [{}] in progress",
state.getVersion(), processedClusterStateVersion.get());
return false;
}
}

Expand Down
16 changes: 15 additions & 1 deletion ...gin/watcher/src/main/java/org/elasticsearch/xpack/watcher/execution/ExecutionService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -121,11 +121,25 @@ public void unPause() {
}

/**
* Pause the execution of the watcher executor
* Pause the execution of the watcher executor, and empty the state.
* Pausing means, that no new watch executions will be done unless this pausing is explicitely unset.
* This is important when watcher is stopped, so that scheduled watches do not accidentally get executed.
* This should not be used when we need to reload watcher based on some cluster state changes, then just calling
* {@link #clearExecutionsAndQueue()} is the way to go
*
* @return the number of tasks that have been removed
*/
public int pause() {
paused.set(true);
return clearExecutionsAndQueue();
}

/**
* Empty the currently queued tasks and wait for current executions to finish.
*
* @return the number of tasks that have been removed
*/
public int clearExecutionsAndQueue() {
int cancelledTaskCount = executor.queue().drainTo(new ArrayList<>());
this.clearExecutions();
return cancelledTaskCount;
Expand Down
6 changes: 3 additions & 3 deletions ...n/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherLifeCycleServiceTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@ public void testManualStartStop() {
reset(watcherService);
when(watcherService.validate(clusterState)).thenReturn(true);
lifeCycleService.clusterChanged(new ClusterChangedEvent("any", clusterState, stoppedClusterState));
verify(watcherService, times(1)).start(eq(clusterState));
verify(watcherService, times(1)).start(eq(clusterState), anyObject());

// no change, keep going
reset(watcherService);
Expand Down Expand Up @@ -457,7 +457,7 @@ public void testWatcherStartsOnlyOnMasterWhenOldNodesAreInCluster() throws Excep

// now start again
lifeCycleService.clusterChanged(new ClusterChangedEvent("any", state, watcherStoppedState));
verify(watcherService).start(any(ClusterState.class));
verify(watcherService).start(any(ClusterState.class), anyObject());
}

public void testDistributedWatchExecutionDisabledWith5xNodesInCluster() throws Exception {
Expand Down Expand Up @@ -509,7 +509,7 @@ public void testWatcherServiceDoesNotStartIfIndexTemplatesAreMissing() throws Ex
when(watcherService.validate(eq(state))).thenReturn(true);

lifeCycleService.clusterChanged(new ClusterChangedEvent("any", state, state));
verify(watcherService, times(0)).start(any(ClusterState.class));
verify(watcherService, times(0)).start(any(ClusterState.class), anyObject());
}

public void testWatcherStopsWhenMasterNodeIsMissing() {
Expand Down
24 changes: 23 additions & 1 deletion x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherServiceTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

Expand Down Expand Up @@ -199,7 +200,7 @@ void stopExecutor() {
when(client.clearScroll(any(ClearScrollRequest.class))).thenReturn(clearScrollFuture);
clearScrollFuture.onResponse(new ClearScrollResponse(true, 1));

service.start(clusterState);
service.start(clusterState, () -> {});

ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
verify(triggerService).start(captor.capture());
Expand Down Expand Up @@ -238,6 +239,27 @@ void stopExecutor() {
verify(triggerEngine).pauseExecution();
}

// if we have to reload the watcher service, the execution service should not be paused, as this might
// result in missing executions
public void testReloadingWatcherDoesNotPauseExecutionService() {
ExecutionService executionService = mock(ExecutionService.class);
TriggerService triggerService = mock(TriggerService.class);
WatcherService service = new WatcherService(Settings.EMPTY, triggerService, mock(TriggeredWatchStore.class),
executionService, mock(WatchParser.class), mock(Client.class), executorService) {
@Override
void stopExecutor() {
}
};

ClusterState.Builder csBuilder = new ClusterState.Builder(new ClusterName("_name"));
csBuilder.metaData(MetaData.builder());

service.reload(csBuilder.build(), "whatever");
verify(executionService).clearExecutionsAndQueue();
verify(executionService, never()).pause();
verify(triggerService).pauseExecution();
}

private static DiscoveryNode newNode() {
return new DiscoveryNode("node", ESTestCase.buildNewFakeTransportAddress(), Collections.emptyMap(),
new HashSet<>(asList(DiscoveryNode.Role.values())), Version.CURRENT);
Expand Down
17 changes: 10 additions & 7 deletions .../java/org/elasticsearch/xpack/watcher/test/integration/ExecutionVarsIntegrationTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.util.Map;
import java.util.function.Function;

import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.elasticsearch.xpack.watcher.actions.ActionBuilders.loggingAction;
import static org.elasticsearch.xpack.watcher.client.WatchSourceBuilders.watchBuilder;
import static org.elasticsearch.xpack.watcher.input.InputBuilders.simpleInput;
Expand All @@ -35,6 +36,8 @@

public class ExecutionVarsIntegrationTests extends AbstractWatcherIntegrationTestCase {

private String watchId = randomAlphaOfLength(20);

@Override
protected List<Class<? extends Plugin>> pluginTypes() {
List<Class<? extends Plugin>> types = super.pluginTypes();
Expand Down Expand Up @@ -106,7 +109,7 @@ protected Map<String, Function<Map<String, Object>, Object>> pluginScripts() {
public void testVars() throws Exception {
WatcherClient watcherClient = watcherClient();

PutWatchResponse putWatchResponse = watcherClient.preparePutWatch("_id").setSource(watchBuilder()
PutWatchResponse putWatchResponse = watcherClient.preparePutWatch(watchId).setSource(watchBuilder()
.trigger(schedule(cron("0/1 * * * * ?")))
.input(simpleInput("value", 5))
.condition(new ScriptCondition(
Expand All @@ -125,7 +128,7 @@ public void testVars() throws Exception {

assertThat(putWatchResponse.isCreated(), is(true));

timeWarp().trigger("_id");
timeWarp().trigger(watchId);

flush();
refresh();
Expand All @@ -134,11 +137,11 @@ public void testVars() throws Exception {
// defaults to match all;
});

assertThat(searchResponse.getHits().getTotalHits(), is(1L));
assertHitCount(searchResponse, 1L);

Map<String, Object> source = searchResponse.getHits().getAt(0).getSourceAsMap();

assertValue(source, "watch_id", is("_id"));
assertValue(source, "watch_id", is(watchId));
assertValue(source, "state", is("executed"));

// we don't store the computed vars in history
Expand Down Expand Up @@ -170,7 +173,7 @@ public void testVars() throws Exception {
public void testVarsManual() throws Exception {
WatcherClient watcherClient = watcherClient();

PutWatchResponse putWatchResponse = watcherClient.preparePutWatch("_id").setSource(watchBuilder()
PutWatchResponse putWatchResponse = watcherClient.preparePutWatch(watchId).setSource(watchBuilder()
.trigger(schedule(cron("0/1 * * * * ? 2020")))
.input(simpleInput("value", 5))
.condition(new ScriptCondition(
Expand All @@ -192,13 +195,13 @@ public void testVarsManual() throws Exception {
boolean debug = randomBoolean();

ExecuteWatchResponse executeWatchResponse = watcherClient
.prepareExecuteWatch("_id")
.prepareExecuteWatch(watchId)
.setDebug(debug)
.get();
assertThat(executeWatchResponse.getRecordId(), notNullValue());
XContentSource source = executeWatchResponse.getRecordSource();

assertValue(source, "watch_id", is("_id"));
assertValue(source, "watch_id", is(watchId));
assertValue(source, "state", is("executed"));

if (debug) {
Expand Down

0 comments on commit 38bf2bd

Please sign in to comment.