Skip to content

Improve Watcher test framework resiliency #40658

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 18 commits into from
Apr 9, 2019
Merged

Improve Watcher test framework resiliency #40658

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
0cc878b
Improve Watcher test framework resiliency
gwbrown Mar 29, 2019
7a01c22
Replace locking with AtomicReference + track pause
gwbrown Apr 1, 2019
c81c129
Merge branch 'master' into watcher/test-resiliency
gwbrown Apr 1, 2019
908430f
Unmute additional test
gwbrown Apr 1, 2019
26a6b5e
Switch test to use TimeWarp as it's more reliable
gwbrown Apr 1, 2019
49f66e9
[?] Don't break if a watch is run on >1 scheduler
gwbrown Apr 1, 2019
d7374de
Re-mute test that isn't helped by these changes
gwbrown Apr 1, 2019
3b89d6e
Make sure the watch is picked up immediately
gwbrown Apr 1, 2019
44fce8b
Re-mute another test to which isn't helped
gwbrown Apr 2, 2019
a47e774
Merge branch 'master' into watcher/test-resiliency
gwbrown Apr 2, 2019
8091110
Adjust synchronization per review
gwbrown Apr 4, 2019
5ea0551
Merge branch 'master' into watcher/test-resiliency
gwbrown Apr 4, 2019
723df70
Unmute one more test
gwbrown Apr 4, 2019
77899f6
Condense stream calls
gwbrown Apr 4, 2019
d5dde77
[testing] Go back to no synchronization
gwbrown Apr 5, 2019
0fe11c4
Ensure new index is green in concrete index test
gwbrown Apr 5, 2019
4756aee
Revert "[testing] Go back to no synchronization"
gwbrown Apr 5, 2019
b3f0025
Merge branch 'master' into watcher/test-resiliency
gwbrown Apr 8, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 3 additions & 5 deletions ...ugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherConcreteIndexTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,11 +23,6 @@

public class WatcherConcreteIndexTests extends AbstractWatcherIntegrationTestCase {

@Override
protected boolean timeWarped() {
return false;
}

public void testCanUseAnyConcreteIndexName() throws Exception {
String newWatcherIndexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
String watchResultsIndex = randomAlphaOfLength(11).toLowerCase(Locale.ROOT);
Expand All @@ -45,6 +40,9 @@ public void testCanUseAnyConcreteIndexName() throws Exception {
.get();

assertTrue(putWatchResponse.isCreated());
refresh();

timeWarp().trigger("mywatch");

assertBusy(() -> {
SearchResponse searchResult = client().prepareSearch(watchResultsIndex).setTrackTotalHits(true).get();
Expand Down
3 changes: 1 addition & 2 deletions ...st/java/org/elasticsearch/xpack/watcher/actions/webhook/WebhookHttpsIntegrationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,10 @@
import org.elasticsearch.xpack.core.watcher.history.WatchRecord;
import org.elasticsearch.xpack.core.watcher.support.xcontent.XContentSource;
import org.elasticsearch.xpack.watcher.actions.ActionBuilders;
import org.elasticsearch.xpack.watcher.common.http.BasicAuth;
import org.elasticsearch.xpack.watcher.common.http.HttpMethod;
import org.elasticsearch.xpack.watcher.common.http.HttpRequestTemplate;
import org.elasticsearch.xpack.watcher.common.http.Scheme;
import org.elasticsearch.xpack.watcher.common.http.BasicAuth;
import org.elasticsearch.xpack.watcher.common.text.TextTemplate;
import org.elasticsearch.xpack.watcher.condition.InternalAlwaysCondition;
import org.elasticsearch.xpack.watcher.test.AbstractWatcherIntegrationTestCase;
Expand Down Expand Up @@ -67,7 +67,6 @@ public void stopWebservice() throws Exception {
webServer.close();
}

@AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/35503")
public void testHttps() throws Exception {
webServer.enqueue(new MockResponse().setResponseCode(200).setBody("body"));
HttpRequestTemplate.Builder builder = HttpRequestTemplate.builder("localhost", webServer.getPort())
Expand Down
4 changes: 1 addition & 3 deletions ...rc/test/java/org/elasticsearch/xpack/watcher/actions/webhook/WebhookIntegrationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
*/
package org.elasticsearch.xpack.watcher.actions.webhook;

import org.apache.lucene.util.LuceneTestCase;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.transport.TransportAddress;
Expand All @@ -18,9 +17,9 @@
import org.elasticsearch.xpack.core.watcher.history.WatchRecord;
import org.elasticsearch.xpack.core.watcher.support.xcontent.XContentSource;
import org.elasticsearch.xpack.watcher.actions.ActionBuilders;
import org.elasticsearch.xpack.watcher.common.http.BasicAuth;
import org.elasticsearch.xpack.watcher.common.http.HttpMethod;
import org.elasticsearch.xpack.watcher.common.http.HttpRequestTemplate;
import org.elasticsearch.xpack.watcher.common.http.BasicAuth;
import org.elasticsearch.xpack.watcher.common.text.TextTemplate;
import org.elasticsearch.xpack.watcher.condition.InternalAlwaysCondition;
import org.elasticsearch.xpack.watcher.test.AbstractWatcherIntegrationTestCase;
Expand All @@ -44,7 +43,6 @@
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;

@LuceneTestCase.AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/35503")
public class WebhookIntegrationTests extends AbstractWatcherIntegrationTestCase {

private MockWebServer webServer = new MockWebServer();
Expand Down
14 changes: 11 additions & 3 deletions ...rc/test/java/org/elasticsearch/xpack/watcher/test/AbstractWatcherIntegrationTestCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
*/
package org.elasticsearch.xpack.watcher.test;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
Expand Down Expand Up @@ -580,6 +582,7 @@ public EmailSent send(Email email, Authentication auth, Profile profile, String
}

protected static class TimeWarp {
private static final Logger logger = LogManager.getLogger(TimeWarp.class);

private final List<ScheduleTriggerEngineMock> schedulers;
private final ClockMock clock;
Expand All @@ -598,9 +601,14 @@ public ClockMock clock() {
}

public void trigger(String watchId, int times, TimeValue timeValue) {
boolean isTriggered = schedulers.stream().anyMatch(scheduler -> scheduler.trigger(watchId, times, timeValue));
String msg = String.format(Locale.ROOT, "could not find watch [%s] to trigger", watchId);
assertThat(msg, isTriggered, is(true));
long triggeredCount = schedulers.stream().map(scheduler -> scheduler.trigger(watchId, times, timeValue))
.filter(b -> b)
.count();
String msg = String.format(Locale.ROOT, "watch was triggered on [%d] schedulers, expected [1]", triggeredCount);
if (triggeredCount > 1) {
logger.warn(msg);
}
assertThat(msg, triggeredCount, greaterThanOrEqualTo(1L));
}
}

Expand Down
2 changes: 0 additions & 2 deletions ...her/src/test/java/org/elasticsearch/xpack/watcher/test/integration/BasicWatcherTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
*/
package org.elasticsearch.xpack.watcher.test.integration;

import org.apache.lucene.util.LuceneTestCase;
import org.elasticsearch.ElasticsearchParseException;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
Expand Down Expand Up @@ -63,7 +62,6 @@

@TestLogging("org.elasticsearch.xpack.watcher:DEBUG," +
"org.elasticsearch.xpack.watcher.WatcherIndexingListener:TRACE")
@LuceneTestCase.AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/35503")
public class BasicWatcherTests extends AbstractWatcherIntegrationTestCase {

public void testIndexWatch() throws Exception {
Expand Down
1 change: 0 additions & 1 deletion ...st/java/org/elasticsearch/xpack/watcher/test/integration/HttpSecretsIntegrationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,6 @@ protected Settings nodeSettings(int nodeOrdinal) {
return super.nodeSettings(nodeOrdinal);
}

@AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/40587")
public void testHttpInput() throws Exception {
WatcherClient watcherClient = watcherClient();
watcherClient.preparePutWatch("_id")
Expand Down
1 change: 0 additions & 1 deletion ...watcher/src/test/java/org/elasticsearch/xpack/watcher/test/integration/WatchAckTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,6 @@ public void testAckSingleAction() throws Exception {
assertThat(throttledCount, greaterThan(0L));
}

@AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/35506")
public void testAckAllActions() throws Exception {
PutWatchResponse putWatchResponse = watcherClient().preparePutWatch()
.setId("_id")
Expand Down
28 changes: 18 additions & 10 deletions ...cher/src/test/java/org/elasticsearch/xpack/watcher/trigger/ScheduleTriggerEngineMock.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,10 @@
import java.time.ZonedDateTime;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
* A mock scheduler to help with unit testing. Provide {@link ScheduleTriggerEngineMock#trigger} method to manually trigger
Expand All @@ -31,7 +33,8 @@
public class ScheduleTriggerEngineMock extends ScheduleTriggerEngine {
private static final Logger logger = LogManager.getLogger(ScheduleTriggerEngineMock.class);

private final ConcurrentMap<String, Watch> watches = new ConcurrentHashMap<>();
private final AtomicReference<Map<String, Watch>> watches = new AtomicReference<>(new ConcurrentHashMap<>());
Copy link
Contributor

@jakelandis jakelandis Apr 3, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Curious why using an AtomicReference here ?

Copy link
Contributor

@jakelandis jakelandis Apr 3, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nvmd.. I see now that you are swapping out the instances below.

New question, why atomic swaps vs. shared lock (for this test code) ?

Copy link
Contributor Author

@gwbrown gwbrown Apr 3, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

An earlier version of this used a lock and it made the tests very slow (this version takes about ~2-3m to run the whole suite, I killed the version with locks at ~10m), I'm not entirely sure why.

private final AtomicBoolean paused = new AtomicBoolean(false);

public ScheduleTriggerEngineMock(ScheduleRegistry scheduleRegistry, Clock clock) {
super(scheduleRegistry, clock);
Expand All @@ -50,37 +53,42 @@ public ScheduleTriggerEvent parseTriggerEvent(TriggerService service, String wat

@Override
public void start(Collection<Watch> jobs) {
Copy link
Contributor

@jakelandis jakelandis Apr 3, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this method by itself is not thread safe. I would advise to synchronize this method, or use a shared lock between this start/stop/add/remove, which negates the need for the atomic reference, or throw an exception if attempted to start/stop twice.

Copy link
Contributor Author

@gwbrown gwbrown Apr 3, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point, I'll take another look at this. There's a few cases where this could potentially have issues now that I look at it again.

Copy link
Member

@martijnvg martijnvg Apr 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This method is invoked from TriggerService#start(...) and that method is synchronized, so I don't think we need to synchronize this method.

Copy link
Contributor Author

@gwbrown gwbrown Apr 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've added synchronized to the methods which modify watches to ensure this is thread safe. How does this look @jakelandis?

jobs.forEach(this::add);
Map<String, Watch> newWatches = new ConcurrentHashMap<>();
jobs.forEach((watch) -> newWatches.put(watch.id(), watch));
watches.set(newWatches);
paused.set(false);
}

@Override
public void stop() {
watches.clear();
watches.set(new ConcurrentHashMap<>());
}

@Override
public void add(Watch watch) {
logger.debug("adding watch [{}]", watch.id());
watches.put(watch.id(), watch);
watches.get().put(watch.id(), watch);
}

@Override
public void pauseExecution() {
// No action is needed because this engine does not trigger watches on a schedule (instead
// they must be triggered manually).
paused.set(true);
}

@Override
public boolean remove(String jobId) {
return watches.remove(jobId) != null;
return watches.get().remove(jobId) != null;
}

public boolean trigger(String jobName) {
return trigger(jobName, 1, null);
}

public boolean trigger(String jobName, int times, TimeValue interval) {
if (watches.containsKey(jobName) == false) {
if (paused.get()) {
return false;
}
if (watches.get().containsKey(jobName) == false) {
return false;
}

Expand All @@ -89,7 +97,7 @@ public boolean trigger(String jobName, int times, TimeValue interval) {
logger.debug("firing watch [{}] at [{}]", jobName, now);
ScheduleTriggerEvent event = new ScheduleTriggerEvent(jobName, now, now);
consumers.forEach(consumer -> consumer.accept(Collections.singletonList(event)));
if (interval != null) {
if (interval != null) {
if (clock instanceof ClockMock) {
((ClockMock) clock).fastForward(interval);
} else {
Expand Down