| 10954 | |
| 10955 | |
| 10956 | void Master::__removeSlave( |
| 10957 | Slave* slave, |
| 10958 | const string& message, |
| 10959 | const Option<TimeInfo>& unreachableTime) |
| 10960 | { |
| 10961 | // We want to remove the slave first, to avoid the allocator |
| 10962 | // re-allocating the recovered resources. |
| 10963 | allocator->removeSlave(slave->id); |
| 10964 | |
| 10965 | // Transition tasks to TASK_UNREACHABLE/TASK_GONE_BY_OPERATOR/TASK_LOST |
| 10966 | // and remove them. We only use TASK_UNREACHABLE/TASK_GONE_BY_OPERATOR if |
| 10967 | // the framework has opted in to the PARTITION_AWARE capability. |
| 10968 | foreachkey (const FrameworkID& frameworkId, utils::copy(slave->tasks)) { |
| 10969 | Framework* framework = getFramework(frameworkId); |
| 10970 | CHECK(framework != nullptr) |
| 10971 | << "Framework " << frameworkId << " not found while removing agent " |
| 10972 | << *slave << "; agent tasks: " << slave->tasks; |
| 10973 | |
| 10974 | TaskState newTaskState = TASK_UNREACHABLE; |
| 10975 | TaskStatus::Reason newTaskReason = TaskStatus::REASON_SLAVE_REMOVED; |
| 10976 | |
| 10977 | // Needed to convey task unreachability because we lose this |
| 10978 | // information from the task state if `TASK_LOST` is used. |
| 10979 | bool unreachable = true; |
| 10980 | |
| 10981 | if (!framework->capabilities.partitionAware) { |
| 10982 | newTaskState = TASK_LOST; |
| 10983 | } else if (unreachableTime.isNone()) { |
| 10984 | unreachable = false; |
| 10985 | newTaskState = TASK_GONE_BY_OPERATOR; |
| 10986 | newTaskReason = TaskStatus::REASON_SLAVE_REMOVED_BY_OPERATOR; |
| 10987 | } |
| 10988 | |
| 10989 | foreachvalue (Task* task, utils::copy(slave->tasks.at(frameworkId))) { |
| 10990 | const StatusUpdate& update = protobuf::createStatusUpdate( |
| 10991 | task->framework_id(), |
| 10992 | task->slave_id(), |
| 10993 | task->task_id(), |
| 10994 | newTaskState, |
| 10995 | TaskStatus::SOURCE_MASTER, |
| 10996 | None(), |
| 10997 | message, |
| 10998 | newTaskReason, |
| 10999 | (task->has_executor_id() ? |
| 11000 | Option<ExecutorID>(task->executor_id()) : None()), |
| 11001 | None(), |
| 11002 | None(), |
| 11003 | None(), |
| 11004 | None(), |
| 11005 | unreachableTime.isSome() ? unreachableTime : None()); |
| 11006 | |
| 11007 | updateTask(task, update); |
| 11008 | removeTask(task, unreachable); |
| 11009 | |
| 11010 | if (!framework->connected()) { |
| 11011 | LOG(WARNING) << "Dropping update " << update |
| 11012 | << " for disconnected " |
| 11013 | << " framework " << frameworkId; |
nothing calls this directly
no test coverage detected