Skip to content

#111433 Watch Next Run Interval Resets On Shard Move or Node Restart #115102

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 6 commits into from
Oct 22, 2024
Merged

#111433 Watch Next Run Interval Resets On Shard Move or Node Restart #115102

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
bee4aa1
Switch Watcher scheduler to use last exec time
lukewhiting Oct 18, 2024
e3ffbeb
Add tests for last runtime calculation
lukewhiting Oct 18, 2024
c7ecfd8
Update docs/changelog/115102.yaml
lukewhiting Oct 18, 2024
6f37ce1
Add counter to watcher job executions to check no additional executio…
lukewhiting Oct 22, 2024
34a2203
Merge branch 'main' of github.com:elastic/elasticsearch into 111433-w…
lukewhiting Oct 22, 2024
6906a74
Merge branch 'main' of github.com:elastic/elasticsearch into 111433-w…
lukewhiting Oct 22, 2024
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
6 changes: 6 additions & 0 deletions docs/changelog/115102.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
pr: 115102
summary: Watch Next Run Interval Resets On Shard Move or Node Restart
area: Watcher
type: bug
issues:
- 111433
40 changes: 38 additions & 2 deletions .../org/elasticsearch/xpack/watcher/trigger/schedule/engine/TickerScheduleTriggerEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
import org.elasticsearch.xpack.core.watcher.support.WatcherDateTimeUtils;
import org.elasticsearch.xpack.core.watcher.trigger.TriggerEvent;
import org.elasticsearch.xpack.core.watcher.watch.Watch;
import org.elasticsearch.xpack.core.watcher.watch.WatchStatus;
import org.elasticsearch.xpack.watcher.trigger.schedule.IntervalSchedule;
import org.elasticsearch.xpack.watcher.trigger.schedule.Schedule;
import org.elasticsearch.xpack.watcher.trigger.schedule.ScheduleRegistry;
import org.elasticsearch.xpack.watcher.trigger.schedule.ScheduleTrigger;
Expand All @@ -32,6 +34,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
Expand Down Expand Up @@ -67,7 +70,11 @@ public synchronized void start(Collection<Watch> jobs) {
Map<String, ActiveSchedule> startingSchedules = Maps.newMapWithExpectedSize(jobs.size());
for (Watch job : jobs) {
if (job.trigger() instanceof ScheduleTrigger trigger) {
startingSchedules.put(job.id(), new ActiveSchedule(job.id(), trigger.getSchedule(), startTime));
if (trigger.getSchedule() instanceof IntervalSchedule) {
startingSchedules.put(job.id(), new ActiveSchedule(job.id(), trigger.getSchedule(), calculateLastStartTime(job)));
} else {
startingSchedules.put(job.id(), new ActiveSchedule(job.id(), trigger.getSchedule(), startTime));
}
}
}
// why are we calling putAll() here instead of assigning a brand
Expand Down Expand Up @@ -108,10 +115,39 @@ public void add(Watch watch) {
// watcher indexing listener
// this also means that updating an existing watch would not retrigger the schedule time, if it remains the same schedule
if (currentSchedule == null || currentSchedule.schedule.equals(trigger.getSchedule()) == false) {
schedules.put(watch.id(), new ActiveSchedule(watch.id(), trigger.getSchedule(), clock.millis()));
if (trigger.getSchedule() instanceof IntervalSchedule) {
schedules.put(watch.id(), new ActiveSchedule(watch.id(), trigger.getSchedule(), calculateLastStartTime(watch)));
} else {
schedules.put(watch.id(), new ActiveSchedule(watch.id(), trigger.getSchedule(), clock.millis()));
}

}
}

/**
* Attempts to calculate the epoch millis of the last time the watch was checked, If the watch has never been checked, the timestamp of
* the last state change is used. If the watch has never been checked and has never been in an active state, the current time is used.
* @param job the watch to calculate the last start time for
* @return the epoch millis of the last time the watch was checked or now
*/
private long calculateLastStartTime(Watch job) {
var lastChecked = Optional.ofNullable(job)
.map(Watch::status)
.map(WatchStatus::lastChecked)
.map(ZonedDateTime::toInstant)
.map(Instant::toEpochMilli);

return lastChecked.orElseGet(
() -> Optional.ofNullable(job)
.map(Watch::status)
.map(WatchStatus::state)
.map(WatchStatus.State::getTimestamp)
.map(ZonedDateTime::toInstant)
.map(Instant::toEpochMilli)
.orElse(clock.millis())
);
}

@Override
public boolean remove(String jobId) {
logger.debug("Removing watch [{}] from engine (engine is running: {})", jobId, isRunning.get());
Expand Down
239 changes: 239 additions & 0 deletions ...va/org/elasticsearch/xpack/watcher/trigger/schedule/engine/TickerScheduleEngineTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
import org.elasticsearch.xpack.core.watcher.trigger.TriggerEvent;
import org.elasticsearch.xpack.core.watcher.watch.ClockMock;
import org.elasticsearch.xpack.core.watcher.watch.Watch;
import org.elasticsearch.xpack.core.watcher.watch.WatchStatus;
import org.elasticsearch.xpack.watcher.condition.InternalAlwaysCondition;
import org.elasticsearch.xpack.watcher.input.none.ExecutableNoneInput;
import org.elasticsearch.xpack.watcher.trigger.schedule.Schedule;
Expand Down Expand Up @@ -283,6 +284,244 @@ public void testAddOnlyWithNewSchedule() {
assertThat(engine.getSchedules().get("_id"), not(is(activeSchedule)));
}

/**
* This test verifies that a watch with a valid lastCheckedTime executes before the interval time to ensure the job resumes waiting
* from the same point it left off before the reallocation / restart
*/
public void testWatchWithLastCheckedTimeExecutesBeforeInitialInterval() throws Exception {
final var firstLatch = new CountDownLatch(1);
final var secondLatch = new CountDownLatch(1);

Watch watch = new Watch(
"watch",
new ScheduleTrigger(interval("1s")),
new ExecutableNoneInput(),
InternalAlwaysCondition.INSTANCE,
null,
null,
Collections.emptyList(),
null,
new WatchStatus(-1L, null, null, clock.instant().minusMillis(500).atZone(ZoneOffset.UTC), null, null, null),
SequenceNumbers.UNASSIGNED_SEQ_NO,
SequenceNumbers.UNASSIGNED_PRIMARY_TERM
);

var watches = Collections.singletonList(watch);

var runCount = new AtomicInteger(0);

engine.register(events -> {
for (TriggerEvent ignored : events) {
if (runCount.get() == 0) {
logger.info("job first fire");
firstLatch.countDown();
} else {
logger.info("job second fire");
secondLatch.countDown();
}
runCount.incrementAndGet();
}
});

engine.start(watches);
advanceClockIfNeeded(clock.instant().plusMillis(510).atZone(ZoneOffset.UTC));
if (firstLatch.await(3, TimeUnit.SECONDS) == false) {
fail("waiting too long for all watches to be triggered");
}

advanceClockIfNeeded(clock.instant().plusMillis(1100).atZone(ZoneOffset.UTC));
if (secondLatch.await(3, TimeUnit.SECONDS) == false) {
fail("waiting too long for all watches to be triggered");
}

assertThat(runCount.get(), is(2));

engine.stop();
}

/**
* This test verifies that a watch without a lastCheckedTime but with a valid activationTime executes before the interval time to
* ensure the job resumes waiting from the same point it left off before the reallocation / restart
*/
public void testWatchWithNoLastCheckedTimeButHasActivationTimeExecutesBeforeInitialInterval() throws Exception {
final var firstLatch = new CountDownLatch(1);
final var secondLatch = new CountDownLatch(1);

Watch watch = new Watch(
"watch",
new ScheduleTrigger(interval("1s")),
new ExecutableNoneInput(),
InternalAlwaysCondition.INSTANCE,
null,
null,
Collections.emptyList(),
null,
new WatchStatus(
-1L,
new WatchStatus.State(true, clock.instant().minusMillis(500).atZone(ZoneOffset.UTC)),
null,
null,
null,
null,
null
),
SequenceNumbers.UNASSIGNED_SEQ_NO,
SequenceNumbers.UNASSIGNED_PRIMARY_TERM
);

var watches = Collections.singletonList(watch);

var runCount = new AtomicInteger(0);

engine.register(events -> {
for (TriggerEvent ignored : events) {
if (runCount.get() == 0) {
logger.info("job first fire");
firstLatch.countDown();
} else {
logger.info("job second fire");
secondLatch.countDown();
}
runCount.incrementAndGet();
}
});

engine.start(watches);
advanceClockIfNeeded(clock.instant().plusMillis(510).atZone(ZoneOffset.UTC));
if (firstLatch.await(3, TimeUnit.SECONDS) == false) {
fail("waiting too long for all watches to be triggered");
}

advanceClockIfNeeded(clock.instant().plusMillis(1100).atZone(ZoneOffset.UTC));
if (secondLatch.await(3, TimeUnit.SECONDS) == false) {
fail("waiting too long for all watches to be triggered");
}

assertThat(runCount.get(), is(2));

engine.stop();
}

/**
* This test verifies that a watch added after service start with a lastCheckedTime executes before the interval time to ensure the job
* resumes waiting from the same point it left off before the reallocation / restart
*/
public void testAddWithLastCheckedTimeExecutesBeforeInitialInterval() throws Exception {
final var firstLatch = new CountDownLatch(1);
final var secondLatch = new CountDownLatch(1);

Watch watch = new Watch(
"watch",
new ScheduleTrigger(interval("1s")),
new ExecutableNoneInput(),
InternalAlwaysCondition.INSTANCE,
null,
null,
Collections.emptyList(),
null,
new WatchStatus(-1L, null, null, clock.instant().minusMillis(500).atZone(ZoneOffset.UTC), null, null, null),
SequenceNumbers.UNASSIGNED_SEQ_NO,
SequenceNumbers.UNASSIGNED_PRIMARY_TERM
);

var runCount = new AtomicInteger(0);

engine.register(events -> {
for (TriggerEvent ignored : events) {
if (runCount.get() == 0) {
logger.info("job first fire");
firstLatch.countDown();
} else {
logger.info("job second fire");
secondLatch.countDown();
}
runCount.incrementAndGet();
}
});

engine.start(Collections.emptyList());
advanceClockIfNeeded(clock.instant().plusMillis(1100).atZone(ZoneOffset.UTC));
engine.add(watch);

advanceClockIfNeeded(clock.instant().plusMillis(510).atZone(ZoneOffset.UTC));
if (firstLatch.await(3, TimeUnit.SECONDS) == false) {
fail("waiting too long for all watches to be triggered");
}

advanceClockIfNeeded(clock.instant().plusMillis(1100).atZone(ZoneOffset.UTC));
if (secondLatch.await(3, TimeUnit.SECONDS) == false) {
fail("waiting too long for all watches to be triggered");
}

assertThat(runCount.get(), is(2));

engine.stop();
}

/**
* This test verifies that a watch added after service start without a lastCheckedTime but with a valid activationTime executes before
* the interval time to ensure the job resumes waiting from the same point it left off before the reallocation / restart
*/
public void testAddWithNoLastCheckedTimeButHasActivationTimeExecutesBeforeInitialInterval() throws Exception {
final var firstLatch = new CountDownLatch(1);
final var secondLatch = new CountDownLatch(1);

Watch watch = new Watch(
"watch",
new ScheduleTrigger(interval("1s")),
new ExecutableNoneInput(),
InternalAlwaysCondition.INSTANCE,
null,
null,
Collections.emptyList(),
null,
new WatchStatus(
-1L,
new WatchStatus.State(true, clock.instant().minusMillis(500).atZone(ZoneOffset.UTC)),
null,
null,
null,
null,
null
),
SequenceNumbers.UNASSIGNED_SEQ_NO,
SequenceNumbers.UNASSIGNED_PRIMARY_TERM
);

var runCount = new AtomicInteger(0);

engine.register(events -> {
for (TriggerEvent ignored : events) {
if (runCount.get() == 0) {
logger.info("job first fire");
firstLatch.countDown();
} else {
logger.info("job second fire");
secondLatch.countDown();
}
runCount.incrementAndGet();
}
});

engine.start(Collections.emptyList());
advanceClockIfNeeded(clock.instant().plusMillis(1100).atZone(ZoneOffset.UTC));
engine.add(watch);

advanceClockIfNeeded(clock.instant().plusMillis(510).atZone(ZoneOffset.UTC));
if (firstLatch.await(3, TimeUnit.SECONDS) == false) {
fail("waiting too long for all watches to be triggered");
}

advanceClockIfNeeded(clock.instant().plusMillis(1100).atZone(ZoneOffset.UTC));
Copy link
Member

Choose a reason for hiding this comment

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

It would be good to have short comments on each of these 4 tests describing what they're for. It took me a few minutes to figure out how they all differed. And in the case of this test, are you trying to show that it does not execute too many times if you advance the clock a good bit? Is it worth adding another latch or two to prove that the watch hasn't run when you expect it not to have run yet?

Copy link
Contributor Author

@lukewhiting lukewhiting Oct 22, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

So there's essentially 2 conditions tested across 4 tests here. Watches with a lastCheckedTime and those who don't have a lastCheckedTime but do have a lastActivationTime. For each of those, we test both the startup of the watcher service and adding those watches to an already running service.

For the startup tests, we you the latches to ensure it runs once before the interval (to verify it picked up the last run time) and for each add to running service test we start up watcher, tick the clock forward to show some time passing then add the watches and check they execute once before the interval time and once again after the interval time.

I have added some comments to better explain each test and good idea on adding a check to make sure they don't run more times than expected. I have added that as well.

if (secondLatch.await(3, TimeUnit.SECONDS) == false) {
fail("waiting too long for all watches to be triggered");
}

assertThat(runCount.get(), is(2));

engine.stop();
}

private Watch createWatch(String name, Schedule schedule) {
return new Watch(
name,
Expand Down