| 528 | |
| 529 | |
| 530 | Future<Nothing> FetcherProcess::__fetch( |
| 531 | const hashmap<CommandInfo::URI, Option<shared_ptr<Cache::Entry>>>& entries, |
| 532 | const ContainerID& containerId, |
| 533 | const string& sandboxDirectory, |
| 534 | const string& cacheDirectory, |
| 535 | const Option<string>& user) |
| 536 | { |
| 537 | // Now construct the FetcherInfo based on which URIs we're using |
| 538 | // the cache for and which ones we are bypassing the cache. |
| 539 | FetcherInfo info; |
| 540 | |
| 541 | foreachpair (const CommandInfo::URI& uri, |
| 542 | const Option<shared_ptr<Cache::Entry>>& entry, |
| 543 | entries) { |
| 544 | FetcherInfo::Item* item = info.add_items(); |
| 545 | |
| 546 | item->mutable_uri()->CopyFrom(uri); |
| 547 | |
| 548 | if (entry.isSome()) { |
| 549 | if (entry.get()->completion().isPending()) { |
| 550 | // Since the entry is not yet "complete", i.e., |
| 551 | // 'completion().isPending()', it must be the case that we created |
| 552 | // the entry in FetcherProcess::fetch(). Otherwise the entry should |
| 553 | // have been in the cache already and we would have waited for its |
| 554 | // completion in FetcherProcess::fetch(). |
| 555 | item->set_action(FetcherInfo::Item::DOWNLOAD_AND_CACHE); |
| 556 | item->set_cache_filename(entry.get()->filename); |
| 557 | } else { |
| 558 | CHECK_READY(entry.get()->completion()); |
| 559 | item->set_action(FetcherInfo::Item::RETRIEVE_FROM_CACHE); |
| 560 | item->set_cache_filename(entry.get()->filename); |
| 561 | } |
| 562 | } else { |
| 563 | item->set_action(FetcherInfo::Item::BYPASS_CACHE); |
| 564 | } |
| 565 | } |
| 566 | |
| 567 | info.set_sandbox_directory(sandboxDirectory); |
| 568 | info.set_cache_directory(cacheDirectory); |
| 569 | |
| 570 | if (user.isSome()) { |
| 571 | info.set_user(user.get()); |
| 572 | } |
| 573 | |
| 574 | if (!flags.frameworks_home.empty()) { |
| 575 | info.set_frameworks_home(flags.frameworks_home); |
| 576 | } |
| 577 | |
| 578 | info.mutable_stall_timeout() |
| 579 | ->set_nanoseconds(flags.fetcher_stall_timeout.ns()); |
| 580 | |
| 581 | return run(containerId, sandboxDirectory, user, info) |
| 582 | .repair(defer(self(), [=](const Future<Nothing>& future) { |
| 583 | ++metrics.task_fetches_failed; |
| 584 | |
| 585 | LOG(ERROR) << "Failed to run mesos-fetcher: " << future.failure(); |
| 586 | |
| 587 | foreachvalue (const Option<shared_ptr<Cache::Entry>>& entry, entries) { |