| 553 | } |
| 554 | |
| 555 | void __launchGroup( |
| 556 | const TaskGroupInfo& taskGroup, |
| 557 | const vector<ContainerID>& containerIds, |
| 558 | const Connection& connection, |
| 559 | const Future<vector<Response>>& responses) |
| 560 | { |
| 561 | if (shuttingDown) { |
| 562 | LOG(WARNING) << "Ignoring the launch group operation as the " |
| 563 | << "executor is shutting down"; |
| 564 | return; |
| 565 | } |
| 566 | |
| 567 | // This could happen if the agent process failed while the child |
| 568 | // containers were being launched. Shutdown the executor if this |
| 569 | // happens. |
| 570 | if (!responses.isReady()) { |
| 571 | LOG(ERROR) << "Unable to receive a response from the agent for " |
| 572 | << "the LAUNCH_CONTAINER call: " |
| 573 | << (responses.isFailed() ? responses.failure() : "discarded"); |
| 574 | _shutdown(); |
| 575 | return; |
| 576 | } |
| 577 | |
| 578 | CHECK_EQ(containerIds.size(), (size_t) taskGroup.tasks().size()); |
| 579 | CHECK_EQ(containerIds.size(), responses->size()); |
| 580 | |
| 581 | int index = 0; |
| 582 | auto responseIterator = responses->begin(); |
| 583 | foreach (const ContainerID& containerId, containerIds) { |
| 584 | const TaskInfo& task = taskGroup.tasks().Get(index++); |
| 585 | const TaskID& taskId = task.task_id(); |
| 586 | const Response& response = *(responseIterator++); |
| 587 | |
| 588 | CHECK(containers.contains(taskId)); |
| 589 | Container* container = containers.at(taskId).get(); |
| 590 | |
| 591 | // Check if we received a 200 OK response for the |
| 592 | // `LAUNCH_CONTAINER` call. Skip the rest of the container |
| 593 | // initialization if this is not the case. |
| 594 | if (response.code != process::http::Status::OK) { |
| 595 | LOG(ERROR) << "Received '" << response.status << "' (" << response.body |
| 596 | << ") while launching child container " << containerId |
| 597 | << " of task '" << taskId << "'"; |
| 598 | container->launchError = response.body; |
| 599 | continue; |
| 600 | } |
| 601 | |
| 602 | container->launched = true; |
| 603 | |
| 604 | const checks::runtime::Nested nestedRuntime{ |
| 605 | containerId, agent, authorizationHeader}; |
| 606 | |
| 607 | if (task.has_check()) { |
| 608 | Try<Owned<checks::Checker>> checker = |
| 609 | checks::Checker::create( |
| 610 | task.check(), |
| 611 | launcherDirectory, |
| 612 | defer(self(), &Self::taskCheckUpdated, taskId, lambda::_1), |