| 10491 | |
| 10492 | |
| 10493 | void Framework::recoverExecutor( |
| 10494 | const ExecutorState& state, |
| 10495 | bool recheckpointExecutor, |
| 10496 | const hashset<TaskID>& tasksToRecheckpoint) |
| 10497 | { |
| 10498 | LOG(INFO) << "Recovering executor '" << state.id |
| 10499 | << "' of framework " << id(); |
| 10500 | |
| 10501 | CHECK_NOTNULL(slave); |
| 10502 | |
| 10503 | if (state.runs.empty() || state.latest.isNone() || state.info.isNone()) { |
| 10504 | LOG(WARNING) << "Skipping recovery of executor '" << state.id |
| 10505 | << "' of framework " << id() |
| 10506 | << " because its latest run or executor info" |
| 10507 | << " cannot be recovered"; |
| 10508 | |
| 10509 | // GC the top level executor work directory. |
| 10510 | slave->garbageCollect(paths::getExecutorPath( |
| 10511 | slave->flags.work_dir, slave->info.id(), id(), state.id)); |
| 10512 | |
| 10513 | // GC the top level executor meta directory. |
| 10514 | slave->garbageCollect(paths::getExecutorPath( |
| 10515 | slave->metaDir, slave->info.id(), id(), state.id)); |
| 10516 | |
| 10517 | return; |
| 10518 | } |
| 10519 | |
| 10520 | // Verify that Resource.AllocationInfo is set, this should |
| 10521 | // be injected by the agent when recovering. |
| 10522 | foreach (const Resource& resource, state.info->resources()) { |
| 10523 | CHECK(resource.has_allocation_info()); |
| 10524 | } |
| 10525 | |
| 10526 | // We are only interested in the latest run of the executor! |
| 10527 | // So, we GC all the old runs. |
| 10528 | // NOTE: We don't schedule the top level executor work and meta |
| 10529 | // directories for GC here, because they will be scheduled when |
| 10530 | // the latest executor run terminates. |
| 10531 | const ContainerID& latest = state.latest.get(); |
| 10532 | foreachvalue (const RunState& run, state.runs) { |
| 10533 | CHECK_SOME(run.id); |
| 10534 | const ContainerID& runId = run.id.get(); |
| 10535 | if (latest != runId) { |
| 10536 | // GC the executor run's work directory. |
| 10537 | // TODO(vinod): Expose this directory to webui by recovering the |
| 10538 | // tasks and doing a 'files->attach()'. |
| 10539 | slave->garbageCollect(paths::getExecutorRunPath( |
| 10540 | slave->flags.work_dir, slave->info.id(), id(), state.id, runId)); |
| 10541 | |
| 10542 | // GC the executor run's meta directory. |
| 10543 | slave->garbageCollect(paths::getExecutorRunPath( |
| 10544 | slave->metaDir, slave->info.id(), id(), state.id, runId)); |
| 10545 | } |
| 10546 | } |
| 10547 | |
| 10548 | Option<RunState> run = state.runs.get(latest); |
| 10549 | CHECK_SOME(run) |
| 10550 | << "Cannot find latest run " << latest << " for executor " << state.id |
no test coverage detected