Restorable indexing tasks #1881

gianm · 2015-10-28T05:51:52Z

Some changes that make it possible to restart tasks on the same hardware.

This is done by killing and respawning the jvms rather than reconnecting to existing
jvms, for a couple reasons. One is that it lets you restore tasks after server reboots
too, and another is that it lets you upgrade all the software on a box at once by just
restarting everything.

The main changes are,

Add "canRestore" and "stopGracefully" methods to Tasks that say if a task can
stop gracefully, and actually do a graceful stop. RealtimeIndexTask is the only
one that currently implements this.
Add "stop" method to TaskRunners that attempts to do an orderly shutdown.
ThreadPoolTaskRunner- call stopGracefully on restorable tasks, wait for exit
ForkingTaskRunner- close output stream to restorable tasks, wait for exit
RemoteTaskRunner- do nothing special, we actually don't want to shutdown
Add "restore" method to TaskRunners that attempts to bootstrap tasks from last run.
Only ForkingTaskRunner does anything here. It maintains a "restore.json" file with
a list of restorable tasks.
Have the CliPeon's ExecutorLifecycle lock the task base directory to prevent a restored
task and a zombie old task from stomping on each other.

gianm · 2015-10-28T07:25:25Z

jdk8 DruidCoordinatorTest.testCoordinatorRun:349 expected:<1> but was:<0>

drcrallen · 2015-10-28T15:33:51Z

indexing-service/src/main/java/io/druid/indexing/common/task/RealtimeIndexTask.java

+                  public void run()
+                  {
+                    persistLatch.countDown();
+                    committer.run();


Is the contract for persist that it will run in a non-daemon thread? also, if the committer errors out is there any special "oh crap" message that needs to fly?

I believe it runs in a daemon thread, but that's okay because there's a non-daemon thread (the shutdown hook) which is waiting for it to finish.

If the committer errors out then that would get logged by the plumber.

do we need to interchange persistLatch.countDown() and committer.run()? I believe persistLatch.await() should wait for task to really finish.

yes, this is a good point. I was thinking that the persistlatch needs to count down even if committer.run() throws an exception- but we can address this with a try/catch

nishantmonu51 · 2015-10-28T17:37:14Z

indexing-service/src/main/java/io/druid/indexing/common/task/RealtimeIndexTask.java

+            plumber.finishJob();
+          } else {
+            log.info("Persisting pending data without handoff, in preparation for restart.");
+            final Committer committer = committerSupplier.get();


here the firehose is already closed while doing shutdown.
Is it safe to call commit on a closed firehose ?

It seems this might not work for KafkaEightFirehoseFactory too since the underlying connector is already closed?

@nishantmonu51 that's a good point, this does not really work very well with the kafka firehose. But then again, RealtimeIndexTasks never did work well with kafka…

do you think it's worth it to rework things such that this works well for kafka and the event receiver? I think to do that we would want the behavior,

if you're using EventReceiverFirehose, stopGracefully causes the servlet to stop accepting new data, and the task will drain existing data, then stop.

if you're using the Kafka firehose, stopGracefully causes the task to simply stop reading data, then persist/commit, then stop.

This could probably be accomplished somehow…

I agree things dont work very well with kafka firehose at present also, till now we also didn't had a concept for graceful restart, If we are going to provide that functionality, I think we should also look into how we can make it working with our current firehoses.
Also, to others in community who might have written their own custom firehoses, a call to commit after a firehose has been shutdown may be unexpected and might result in weird errors.

Both the behaviours for EventReceiverFirehose and KafkaFirehose seem good and points to that we may need to add an API to the firehose where instead of completely shutting it down, we ask firehose to stop reading any further events, ingest all events which might be in some buffers, persist and call commit on firehose, shutdown the firehose and release any resources being held up by the firehose .

I also think some people do use kafka firehose (with either partition or replication) and it works in the specific cases, with this change that will break.

@nishantmonu51 @himanshug thinking of doing this by just having special behavior triggered by an instanceof check for EventReceiverFirehose. There doesn't seem to be a nicer way to do it with the current firehose interface. Basically- the ERF would get closed and drained, all other firehoses we would simply immediately stop reading. For those we would rely on commit() being an effective way to get back undrained data.

nishantmonu51 · 2015-10-29T09:22:35Z

After these changes, what is the behaviour of task logs ?
Do they get uploaded to s3 on each restart with diff file name or the new logs will override prev logs ?
or they get appended to the same file and finally when the task completes/fails they are uploaded to S3 ?

gianm · 2015-10-29T19:38:01Z

@nishantmonu51 they get uploaded to S3 on each restart, with new uploads overriding previous uploads. The FTR opens the log in "append" mode so each upload will contain logs from all previous runs of the same task.

gianm · 2015-10-29T19:38:39Z

@nishantmonu51 I do think that in the future we should make the uploading happen in chunks rather than all at once- but this would be a different PR

himanshug · 2015-10-29T19:50:17Z

indexing-service/src/main/java/io/druid/indexing/overlord/ForkingTaskRunner.java

+        taskRestoreInfo = jsonMapper.readValue(restoreFile, TaskRestoreInfo.class);
+      }
+      catch (Exception e) {
+        log.warn(e, "Failed to restore tasks from file[%s]. Skipping restore.", restoreFile);


this should be error as restoreFile existence should mean that there are valid tasks to restore and for whatever reason reading the file failed.

sounds good to me.

gianm · 2015-10-29T20:18:05Z

@himanshug @nishantmonu51 @drcrallen pushed a new commit, the main change is in firehose-closing behavior. also looking at adding some tests.

gianm · 2015-10-29T20:28:11Z

hmm, I think there's a further problem. If you don't close the kafka firehose, and no new data is forthcoming, I believe it will block forever on hasMore. I don't think there's anything we can do about that right now, but that points to potentially wanting to have a timeout on hasMore or a poll-style Firehose interface.

gianm · 2015-10-29T20:29:00Z

What that means is that if you do try to stopGracefully a realtime task reading from a dry kafka firehose, it will likely time out and be killed after the gracefulShutdownTimeout (default 5 minutes)

guobingkun · 2015-10-29T21:55:59Z

indexing-service/src/main/java/io/druid/indexing/common/task/RealtimeIndexTask.java

@@ -62,6 +63,8 @@
 import java.io.IOException;
 import java.util.Map;
 import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;


thanks, yes, removed.

himanshug · 2015-10-30T03:19:01Z

indexing-service/src/main/java/io/druid/indexing/worker/executor/ExecutorLifecycle.java

+            log.info("Acquired lock file[%s] in %,dms.", taskLockFile, System.currentTimeMillis() - startLocking);
+          }
+        } else {
+          throw new ISE("Already started!");


can this ever happen or just to catch if start() is called twice (or more)?

just to catch if start() is called more than once- which it should not be.

gianm · 2015-10-30T04:09:12Z

@himanshug @nishantmonu51 @drcrallen @guobingkun pushed updates for outstanding comments + also some unit tests for restoring realtime tasks

drcrallen · 2015-11-17T22:46:25Z

indexing-service/src/main/java/io/druid/indexing/overlord/ThreadPoolTaskRunner.java

+        log.info("Starting graceful shutdown of task[%s].", task.getId());
+
+        try {
+          task.stopGracefully();


Can we emit a metric about this?

It would be nice to have a way to keep track of how many graceful stops we do

…are. This is done by killing and respawning the jvms rather than reconnecting to existing jvms, for a couple reasons. One is that it lets you restore tasks after server reboots too, and another is that it lets you upgrade all the software on a box at once by just restarting everything. The main changes are, 1) Add "canRestore" and "stopGracefully" methods to Tasks that say if a task can stop gracefully, and actually do a graceful stop. RealtimeIndexTask is the only one that currently implements this. 2) Add "stop" method to TaskRunners that attempts to do an orderly shutdown. ThreadPoolTaskRunner- call stopGracefully on restorable tasks, wait for exit ForkingTaskRunner- close output stream to restorable tasks, wait for exit RemoteTaskRunner- do nothing special, we actually don't want to shutdown 3) Add "restore" method to TaskRunners that attempts to bootstrap tasks from last run. Only ForkingTaskRunner does anything here. It maintains a "restore.json" file with a list of restorable tasks. 4) Have the CliPeon's ExecutorLifecycle lock the task base directory to avoid a restored task and a zombie old task from stomping on each other.

gianm · 2015-11-23T19:22:37Z

@drcrallen @himanshug updated the branch with changes from code review

fjy · 2015-11-24T00:06:15Z

we've been using this internally for awhile, seems to work

himanshug · 2015-11-24T01:50:28Z

👍 with the changes too

Restorable indexing tasks

xvrl · 2015-11-24T06:50:21Z

would be great to add some integration tests for this feature.

gianm force-pushed the restartable-tasks branch from b21cc5b to a9d5f25 Compare October 28, 2015 06:18

gianm closed this Oct 28, 2015

gianm reopened this Oct 28, 2015

drcrallen reviewed Oct 28, 2015
View reviewed changes

gianm mentioned this pull request Oct 28, 2015

Epic: Realtime Ingestion Improvements #1642

Closed

nishantmonu51 reviewed Oct 28, 2015
View reviewed changes

himanshug reviewed Oct 29, 2015
View reviewed changes

gianm force-pushed the restartable-tasks branch from a9d5f25 to 0f6d71b Compare October 29, 2015 20:16

guobingkun reviewed Oct 29, 2015
View reviewed changes

fjy added this to the 0.9.0 milestone Oct 29, 2015

gianm mentioned this pull request Oct 30, 2015

Upload TaskLogs in chunks #1891

Closed

himanshug reviewed Oct 30, 2015
View reviewed changes

gianm force-pushed the restartable-tasks branch from 0f6d71b to f44ccb8 Compare October 30, 2015 03:54

gianm force-pushed the restartable-tasks branch from f44ccb8 to 9ce5222 Compare October 30, 2015 20:13

drcrallen reviewed Nov 17, 2015
View reviewed changes

EventReceiverFirehose: Drain buffer when closed, until empty.

3656909

gianm force-pushed the restartable-tasks branch from e899be4 to d1dfe8e Compare November 21, 2015 03:17

fjy closed this Nov 23, 2015

fjy reopened this Nov 23, 2015

gianm force-pushed the restartable-tasks branch from d1dfe8e to 501dcb4 Compare November 23, 2015 19:22

fjy closed this Nov 24, 2015

fjy reopened this Nov 24, 2015

fjy closed this Nov 24, 2015

fjy reopened this Nov 24, 2015

fjy added a commit that referenced this pull request Nov 24, 2015

Merge pull request #1881 from gianm/restartable-tasks

8e83d80

Restorable indexing tasks

fjy merged commit 8e83d80 into apache:master Nov 24, 2015

gianm modified the milestones: 0.8.3, 0.9.0 Dec 1, 2015

This was referenced Dec 1, 2015

0.8.3 backports #2022

Merged

druid-0.8.3 release notes #2044

Closed

gianm mentioned this pull request Feb 25, 2016

UnRegister port in ForkingTaskRunner #2543

Merged

gianm deleted the restartable-tasks branch February 25, 2016 16:50

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Restorable indexing tasks #1881

Restorable indexing tasks #1881

gianm commented Oct 28, 2015

gianm commented Oct 28, 2015

drcrallen Oct 28, 2015

gianm Oct 28, 2015

himanshug Oct 29, 2015

gianm Oct 29, 2015

nishantmonu51 Oct 28, 2015

nishantmonu51 Oct 28, 2015

gianm Oct 28, 2015

nishantmonu51 Oct 29, 2015

himanshug Oct 29, 2015

gianm Oct 29, 2015

nishantmonu51 commented Oct 29, 2015

gianm commented Oct 29, 2015

gianm commented Oct 29, 2015

himanshug Oct 29, 2015

gianm Oct 29, 2015

gianm commented Oct 29, 2015

gianm commented Oct 29, 2015

gianm commented Oct 29, 2015

guobingkun Oct 29, 2015

gianm Oct 30, 2015

himanshug Oct 30, 2015

gianm Oct 30, 2015

gianm commented Oct 30, 2015

drcrallen Nov 17, 2015

drcrallen Nov 17, 2015

gianm Nov 23, 2015

gianm commented Nov 23, 2015

fjy commented Nov 24, 2015

himanshug commented Nov 24, 2015

xvrl commented Nov 24, 2015

Restorable indexing tasks #1881

Restorable indexing tasks #1881

Conversation

gianm commented Oct 28, 2015

gianm commented Oct 28, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nishantmonu51 commented Oct 29, 2015

gianm commented Oct 29, 2015

gianm commented Oct 29, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gianm commented Oct 29, 2015

gianm commented Oct 29, 2015

gianm commented Oct 29, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gianm commented Oct 30, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gianm commented Nov 23, 2015

fjy commented Nov 24, 2015

himanshug commented Nov 24, 2015

xvrl commented Nov 24, 2015