[DCOS-46585] Fix supervised driver retry logic for outdated tasks

[samvantran]

What changes were proposed in this pull request?

Resolves DCOS-46585 && COPS-4325

Customers experienced --supervised Spark jobs were retried multiple times in scenarios where an agent would crash, come back, and re-register even when those jobs had already relaunched on a different agent. That is:

- supervised driver is running on agent1
- agent1 crashes
- driver is relaunched on another agent as `<task-id>-retry-1`
- agent1 comes back online and re-registers with scheduler
- spark relaunches the same job as `<task-id>-retry-2`
- now there are two jobs running simultaneously

This is because when an agent would come back and re-register it would send a status update TASK_FAILED for its old driver-task. Previous logic would indiscriminately remove the submissionId from Zookeeper's launchedDrivers node and add it to retryList node. Then, when a new offer came in, it would relaunch another -retry- task even though one was previously running.

For example logs, scroll to bottom

How was this patch tested?

• Added a unit test to simulate behavior described above
• Tested manually on a DC/OS cluster by

  - launching a --supervised spark job
  - dcos node ssh <to the agent with the running spark-driver>
  - systemctl stop dcos-mesos-slave
  - docker kill <driver-container-id>
  - [ wait until spark job is relaunched ]
  - systemctl start dcos-mesos-slave
  - [ observe spark driver is not relaunched as `-retry-2` ]
  

Log snippets included below. Notice the -retry-1 task is running when status update for the old task comes in afterward:

19/01/15 19:21:38 TRACE MesosClusterScheduler: Received offers from Mesos: 
... [offers] ...
19/01/15 19:21:39 TRACE MesosClusterScheduler: Using offer 5d421001-0630-4214-9ecb-d5838a2ec149-O2532 to launch driver driver-20190115192138-0001 with taskId: value: "driver-20190115192138-0001"
...
19/01/15 19:21:42 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001 state=TASK_STARTING message=''
19/01/15 19:21:43 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001 state=TASK_RUNNING message=''
...
19/01/15 19:29:12 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001 state=TASK_LOST message='health check timed out' reason=REASON_SLAVE_REMOVED
...
19/01/15 19:31:12 TRACE MesosClusterScheduler: Using offer 5d421001-0630-4214-9ecb-d5838a2ec149-O2681 to launch driver driver-20190115192138-0001 with taskId: value: "driver-20190115192138-0001-retry-1"
...
19/01/15 19:31:15 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001-retry-1 state=TASK_STARTING message=''
19/01/15 19:31:16 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001-retry-1 state=TASK_RUNNING message=''
...
19/01/15 19:33:45 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001 state=TASK_FAILED message='Unreachable agent re-reregistered'
...
19/01/15 19:33:45 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001 state=TASK_FAILED message='Abnormal executor termination: unknown container' reason=REASON_EXECUTOR_TERMINATED
19/01/15 19:33:45 ERROR MesosClusterScheduler: Unable to find driver with driver-20190115192138-0001 in status update
...
19/01/15 19:33:47 TRACE MesosClusterScheduler: Using offer 5d421001-0630-4214-9ecb-d5838a2ec149-O2729 to launch driver driver-20190115192138-0001 with taskId: value: "driver-20190115192138-0001-retry-2"
...
19/01/15 19:33:50 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001-retry-2 state=TASK_STARTING message=''
19/01/15 19:33:51 INFO MesosClusterScheduler: Received status update: taskId=driver-20190115192138-0001-retry-2 state=TASK_RUNNING message=''

[akirillov]

thanks, @samvantran. Provided fix relies on prior knowledge of task id format (its string representation and parsing) and on the ordering of task ids based on this representation. This heuristic doesn't look strong enough and the fix will stop working in case UUID semantics change in the future. I'd suggest relying on some state tracking data structure which will keep track of (re)launched drivers ordered by time/status. Or even better keep the lost driver ids in a set to perform taskIsOutdated check against it.

[akirillov]

this can be simplified to just getRetryCountFromTaskId(taskId) < getRetryCountFromTaskId(state.frameworkId) without if statement

[samvantran]

ah right. thanks!

[samvantran]

Thanks for the review @akirillov! Good call on using a set for tracking retried drivers. I've made changes based on your comments. Please take a 2nd look 🙏

[samvantran]

ah right. thanks!

[akirillov]

thanks, @samvantran. One thing is missing now - the cleanup logic, ids are only added to retriedDrivers set and will still live there until the Dispatcher is terminated thus potentially causing unnecessary memory pressure.

What if to use a reversed approach and keep track of running drivers and ignore any other ids coming with TASK_LOST/TASK_FAILED. I.e. when a driver needs to be relaunched - a prior id is removed from the list, and the new id is added. When the old id is coming in with failed task message - just ignore it if it's not in running.

[akirillov]

having state mutation logic in a boolean function which checks some condition looks a bit confusing

[samvantran]

Yeah my original concerns with this approach was that a long lived scheduler could 1) accumulate a lot of IDs and 2) degrade in lookup performance but realistically it wouldn't be that much. Imagine 1 million failed/lost jobs... that would catch cause alarm for any user and they'd like have bigger problems on their hands.

But to your point... I actually had this much longer response written up before I made the HashSet change so I might as well share it:

• re. tracking drivers, there is already the launchedDrivers map that does just that. It tracks status and last retryState (with number of retries, lastFailedStatus etc). When I was parsing retry counts, I was basing it on the fact that the map would have latest task state (which is updated here)
• there is even pendingRetryDrivers array that temporarily stores which drivers need to be relaunched and is purged when they are.
• When you say we remove "prior" IDs and ignore "old" IDs with status TASK_FAILED, how do we know they're old? If we follow my test example we get these events

  task001 -> TASK_RUNNING
  task001 -> TASK_LOST              // retriedDrivers = [task001]
  task001-retry1 -> TASK_RUNNING    // is this where we remove task001 ? 
  task001 -> TASK_FAILED 
  

• from the example above, how do we know task001-retry-1 is associated with task001 without parsing the taskId string?

I kind of am thinking we should go back to parsing retry count from taskId string because:

1. like you said, there would be added memory pressure over time with a set
2. no extra data structures needed
3. parsing retry count is O(1) operation
4. parsing taskId against the {RETRY_SEP} constant has been done for at least 2.5 years (see these methods). It still exists in Spark 3.0-SNAPSHOT so I don't suspect UUID semantics will change any time soon.

WDYT?

[akirillov]

thanks, @samvantran. LGTM!

One minor comment: def taskIsOutdated probably deserves a scaladoc

[samvantran]

Good call. I'll add docs. Thanks for the offline discussion. I think this is the right way to go!