void Master::_accept(
????const FrameworkID& frameworkId,
????const SlaveID& slaveId,
????const Resources& offeredResources,
????const scheduler::Call::Accept& accept,
????const Future<list<Future<bool>>>& _authorizations)
{
??Framework* framework = getFramework(frameworkId);
……
?
??Slave* slave = slaves.registered.get(slaveId);
?
??if (slave == NULL || !slave->connected) {
????foreach (const Offer::Operation& operation, accept.operations()) {
??????if (operation.type() != Offer::Operation::LAUNCH) {
????????continue;
??????}
?
??????foreach (const TaskInfo& task, operation.launch().task_infos()) {
????????const TaskStatus::Reason reason =
????????????slave == NULL ? TaskStatus::REASON_SLAVE_REMOVED
??????????????????????????: TaskStatus::REASON_SLAVE_DISCONNECTED;
????????const StatusUpdate& update = protobuf::createStatusUpdate(
????????????framework->id(),
????????????task.slave_id(),
????????????task.task_id(),
????????????TASK_LOST,
????????????TaskStatus::SOURCE_MASTER,
????????????None(),
????????????slave == NULL ? "Slave removed" : "Slave disconnected",
????????????reason);
?
????????metrics->tasks_lost++;
?
????????metrics->incrementTasksStates(
????????????TASK_LOST,
????????????TaskStatus::SOURCE_MASTER,
????????????reason);
?
????????forward(update, UPID(), framework);
??????}
????}
?
????// Tell the allocator about the recovered resources.
????allocator->recoverResources(
????????frameworkId,
????????slaveId,
????????offeredResources,
????????None());
?
????return;
??}
?
??// Some offer operations update the offered resources. We keep
??// updated offered resources here. When a task is successfully
??// launched, we remove its resource from offered resources.
??Resources _offeredResources = offeredResources;
?
??// The order of `authorizations` must match the order of the operations in
??// `accept.operations()`, as they are iterated through simultaneously.
??CHECK_READY(_authorizations);
??list<Future<bool>> authorizations = _authorizations.get();
?
??foreach (const Offer::Operation& operation, accept.operations()) {
????switch (operation.type()) {
??????// The RESERVE operation allows a principal to reserve resources.
??????case Offer::Operation::RESERVE: {
????????Future<bool> authorization = authorizations.front();
????????authorizations.pop_front();
?
????????CHECK(!authorization.isDiscarded());
?
????????if (authorization.isFailed()) {
??????????// TODO(greggomann): We may want to retry this failed authorization
??????????// request rather than dropping it immediately.
??????????drop(framework,
???????????????operation,
???????????????"Authorization of principal ‘" + framework->info.principal() +
???????????????"‘ to reserve resources failed: " +
???????????????authorization.failure());
?
??????????continue;
????????} else
if (!authorization.get()) {
??????????drop(framework,
???????????????operation,
???????????????"Not authorized to reserve resources as ‘" +
?????????????????framework->info.principal() + "‘");
?
??????????continue;
????????}
?
????????Option<string> principal = framework->info.has_principal()
??????????? framework->info.principal()
??????????: Option<string>::none();
?
????????// Make sure this reserve operation is valid.
????????Option<Error> error = validation::operation::validate(
????????????operation.reserve(), principal);
?
????????if (error.isSome()) {
??????????drop(framework, operation, error.get().message);
??????????continue;
????????}
?
????????// Test the given operation on the included resources.
????????Try<Resources> resources = _offeredResources.apply(operation);
????????if (resources.isError()) {
??????????drop(framework, operation, resources.error());
??????????continue;
????????}
?
????????_offeredResources = resources.get();
?
????????LOG(INFO) << "Applying RESERVE operation for resources "
??????????????????<< operation.reserve().resources() << " from framework "
??????????????????<< *framework << " to slave " << *slave;
?
????????apply(framework, slave, operation);
????????break;
??????}
?
??????// The UNRESERVE operation allows a principal to unreserve resources.
??????case Offer::Operation::UNRESERVE: {
????????Future<bool> authorization = authorizations.front();
????????authorizations.pop_front();
?
????????CHECK(!authorization.isDiscarded());
?
????????if (authorization.isFailed()) {
??????????// TODO(greggomann): We may want to retry this failed authorization
??????????// request rather than dropping it immediately.
??????????drop(framework,
???????????????operation,
???????????????"Authorization of principal ‘" + framework->info.principal() +
???????????????"‘ to unreserve resources failed: " +
???????????????authorization.failure());
?
??????????continue;
????????} else
if (!authorization.get()) {
??????????drop(framework,
???????????????operation,
???????????????"Not authorized to unreserve resources as ‘" +
?????????????????framework->info.principal() + "‘");
?
??????????continue;
????????}
?
????????// Make sure this unreserve operation is valid.
????????Option<Error> error = validation::operation::validate(
????????????operation.unreserve());
?
????????if (error.isSome()) {
??????????drop(framework, operation, error.get().message);
??????????continue;
????????}
?
????????// Test the given operation on the included resources.
????????Try<Resources> resources = _offeredResources.apply(operation);
????????if (resources.isError()) {
??????????drop(framework, operation, resources.error());
??????????continue;
????????}
?
????????_offeredResources = resources.get();
?
????????LOG(INFO) << "Applying UNRESERVE operation for resources "
??????????????????<< operation.unreserve().resources() << " from framework "
??????????????????<< *framework << " to slave " << *slave;
?
????????apply(framework, slave, operation);
????????break;
??????}
?
??????case Offer::Operation::CREATE: {
????????Future<bool> authorization = authorizations.front();
????????authorizations.pop_front();
?
????????CHECK(!authorization.isDiscarded());
?
????????if (authorization.isFailed()) {
??????????// TODO(greggomann): We may want to retry this failed authorization
??????????// request rather than dropping it immediately.
??????????drop(framework,
???????????????operation,
???????????????"Authorization of principal ‘" + framework->info.principal() +
???????????????"‘ to create persistent volumes failed: " +
???????????????authorization.failure());
?
??????????continue;
????????} else
if (!authorization.get()) {
??????????drop(framework,
???????????????operation,
???????????????"Not authorized to create persistent volumes as ‘" +
?????????????????framework->info.principal() + "‘");
?
??????????continue;
????????}
?
????????// Make sure this create operation is valid.
????????Option<Error> error = validation::operation::validate(
????????????operation.create(), slave->checkpointedResources);
?
????????if (error.isSome()) {
??????????drop(framework, operation, error.get().message);
??????????continue;
????????}
?
????????Try<Resources> resources = _offeredResources.apply(operation);
????????if (resources.isError()) {
??????????drop(framework, operation, resources.error());
??????????continue;
????????}
?
????????_offeredResources = resources.get();
?
????????LOG(INFO) << "Applying CREATE operation for volumes "
??????????????????<< operation.create().volumes() << " from framework "
??????????????????<< *framework << " to slave " << *slave;
?
????????apply(framework, slave, operation);
????????break;
??????}
?
??????case Offer::Operation::DESTROY: {
????????Future<bool> authorization = authorizations.front();
????????authorizations.pop_front();
?
????????CHECK(!authorization.isDiscarded());
?
????????if (authorization.isFailed()) {
??????????// TODO(greggomann): We may want to retry this failed authorization
??????????// request rather than dropping it immediately.
??????????drop(framework,
???????????????operation,
???????????????"Authorization of principal ‘" + framework->info.principal() +
???????????????"‘ to destroy persistent volumes failed: " +
???????????????authorization.failure());
?
??????????continue;
????????} else
if (!authorization.get()) {
??????????drop(framework,
???????????????operation,
???????????????"Not authorized to destroy persistent volumes as ‘" +
?????????????????framework->info.principal() + "‘");
?
??????????continue;
????????}
?
????????// Make sure this destroy operation is valid.
????????Option<Error> error = validation::operation::validate(
????????????operation.destroy(), slave->checkpointedResources);
?
????????if (error.isSome()) {
??????????drop(framework, operation, error.get().message);
??????????continue;
????????}
?
????????Try<Resources> resources = _offeredResources.apply(operation);
????????if (resources.isError()) {
??????????drop(framework, operation, resources.error());
??????????continue;
????????}
?
????????_offeredResources = resources.get();
?
????????LOG(INFO) << "Applying DESTROY operation for volumes "
??????????????????<< operation.create().volumes() << " from framework "
??????????????????<< *framework << " to slave " << *slave;
?
????????apply(framework, slave, operation);
????????break;
??????}
?
??????case Offer::Operation::LAUNCH: {
????????foreach (const TaskInfo& task, operation.launch().task_infos()) {
??????????Future<bool> authorization = authorizations.front();
??????????authorizations.pop_front();
?
??????????// NOTE: The task will not be in ‘pendingTasks‘ if
??????????// ‘killTask()‘ for the task was called before we are here.
??????????// No need to launch the task if it‘s no longer pending.
??????????// However, we still need to check the authorization result
??????????// and do the validation so that we can send status update
??????????// in case the task has duplicated ID.
??????????bool pending = framework->pendingTasks.contains(task.task_id());
?
??????????// Remove from pending tasks.
??????????framework->pendingTasks.erase(task.task_id());
?
??????????CHECK(!authorization.isDiscarded());
?
??????????if (authorization.isFailed() || !authorization.get()) {
????????????string user = framework->info.user(); // Default user.
????????????if (task.has_command() && task.command().has_user()) {
??????????????user = task.command().user();
????????????} else
if (task.has_executor() &&
???????????????????????task.executor().command().has_user()) {
??????????????user = task.executor().command().user();
????????????}
?
????????????const StatusUpdate& update = protobuf::createStatusUpdate(
????????????????framework->id(),
????????????????task.slave_id(),
????????????????task.task_id(),
????????????????TASK_ERROR,
????????????????TaskStatus::SOURCE_MASTER,
????????????????None(),
????????????????authorization.isFailed() ?
????????????????????"Authorization failure: " + authorization.failure() :
????????????????????"Not authorized to launch as user ‘" + user + "‘",
????????????????TaskStatus::REASON_TASK_UNAUTHORIZED);
?
????????????metrics->tasks_error++;
?
????????????metrics->incrementTasksStates(
????????????????TASK_ERROR,
????????????????TaskStatus::SOURCE_MASTER,
????????????????TaskStatus::REASON_TASK_UNAUTHORIZED);
?
????????????forward(update, UPID(), framework);
?
????????????continue;
??????????}
?
??????????// Validate the task.
?
??????????// Make a copy of the original task so that we can
??????????// fill the missing `framework_id` in ExecutorInfo
??????????// if needed. This field was added to the API later
??????????// and thus was made optional.
??????????TaskInfo task_(task);
??????????if (task.has_executor() && !task.executor().has_framework_id()) {
????????????task_.mutable_executor()
????????????????->mutable_framework_id()->CopyFrom(framework->id());
??????????}
?
??????????const Option<Error>& validationError = validation::task::validate(
??????????????task_,
??????????????framework,
??????????????slave,
??????????????_offeredResources);
?
??????????if (validationError.isSome()) {
????????????const StatusUpdate& update = protobuf::createStatusUpdate(
????????????????framework->id(),
????????????????task_.slave_id(),
????????????????task_.task_id(),
????????????????TASK_ERROR,
????????????????TaskStatus::SOURCE_MASTER,
????????????????None(),
????????????????validationError.get().message,
????????????????TaskStatus::REASON_TASK_INVALID);
?
????????????metrics->tasks_error++;
?
????????????metrics->incrementTasksStates(
????????????????TASK_ERROR,
????????????????TaskStatus::SOURCE_MASTER,
????????????????TaskStatus::REASON_TASK_INVALID);
?
????????????forward(update, UPID(), framework);
?
????????????continue;
??????????}
?
??????????// Add task.
??????????if (pending) {
????????????_offeredResources -= addTask(task_, framework, slave);
?
????????????// TODO(bmahler): Consider updating this log message to
????????????// indicate when the executor is also being launched.
????????????LOG(INFO) << "Launching task " << task_.task_id()
??????????????????????<< " of framework " << *framework
??????????????????????<< " with resources " << task_.resources()
??????????????????????<< " on slave " << *slave;
?
????????????RunTaskMessage message;
????????????message.mutable_framework()->MergeFrom(framework->info);
?
????????????// TODO(anand): We set ‘pid‘ to UPID() for http frameworks
????????????// as ‘pid‘ was made optional in 0.24.0. In 0.25.0, we
????????????// no longer have to set pid here for http frameworks.
????????????message.set_pid(framework->pid.getOrElse(UPID()));
????????????message.mutable_task()->MergeFrom(task_);
?
????????????if (HookManager::hooksAvailable()) {
??????????????// Set labels retrieved from label-decorator hooks.
??????????????message.mutable_task()->mutable_labels()->CopyFrom(
??????????????????HookManager::masterLaunchTaskLabelDecorator(
??????????????????????task_,
??????????????????????framework->info,
??????????????????????slave->info));
????????????}
?
????????????send(slave->pid, message);
??????????}
????????}
????????break;
??????}
?
??????default:
????????LOG(ERROR) << "Unsupported offer operation " << operation.type();
????????break;
????}
??}
?
??if (!_offeredResources.empty()) {
????// Tell the allocator about the unused (e.g., refused) resources.
????allocator->recoverResources(
????????frameworkId,
????????slaveId,
????????_offeredResources,
????????accept.filters());
??}
}
