text: // Licensed to the Apache Software Foundation (ASF) under one
  // or more contributor license agreements.  See the NOTICE file
  // distributed with this work for additional information
  // regarding copyright ownership.  The ASF licenses this file
  // to you under the Apache License, Version 2.0 (the
  // "License"); you may not use this file except in compliance
  // with the License.  You may obtain a copy of the License at
  //
  //     http://www.apache.org/licenses/LICENSE-2.0
  //
  // Unless required by applicable law or agreed to in writing, software
  // distributed under the License is distributed on an "AS IS" BASIS,
  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  // See the License for the specific language governing permissions and
  // limitations under the License.
  
  #include <unordered_map>
  
  #include <process/async.hpp>
  #include <process/check.hpp>
  #include <process/collect.hpp>
  #include <process/dispatch.hpp>
  #include <process/owned.hpp>
  
  #include <stout/net.hpp>
  #include <stout/path.hpp>
  #ifdef __WINDOWS__
  #include <stout/windows.hpp>
  #endif // __WINDOWS__
  
  #include <stout/os/find.hpp>
  #include <stout/os/killtree.hpp>
  #include <stout/os/read.hpp>
  
  #include "hdfs/hdfs.hpp"
  
  #include "slave/slave.hpp"
  
  #include "slave/containerizer/fetcher.hpp"
  
  using std::list;
  using std::map;
  using std::shared_ptr;
  using std::string;
  using std::transform;
  using std::vector;
  
  using mesos::fetcher::FetcherInfo;
  
  using process::Future;
  using process::Owned;
  
  namespace mesos {
  namespace internal {
  namespace slave {
  
  static const string FILE_URI_PREFIX = "file://";
  static const string FILE_URI_LOCALHOST = "file://localhost";
  
  static const string CACHE_FILE_NAME_PREFIX = "c";
  
  
  Fetcher::Fetcher() : process(new FetcherProcess())
  {
    spawn(process.get());
  }
  
  
  Fetcher::Fetcher(const process::Owned<FetcherProcess>& process)
    : process(process)
  {
    spawn(process.get());
  }
  
  
  Fetcher::~Fetcher()
  {
    terminate(process.get());
    process::wait(process.get());
  }
  
  
  Try<Nothing> Fetcher::recover(const SlaveID& slaveId, const Flags& flags)
  {
    // Good enough for now, simple, least-effort recovery.
    VLOG(1) << "Clearing fetcher cache";
  
    string cacheDirectory = paths::getSlavePath(flags.fetcher_cache_dir, slaveId);
    Result<string> path = os::realpath(cacheDirectory);
    if (path.isError()) {
      LOG(ERROR) << "Malformed fetcher cache directory path '" << cacheDirectory
                 << "', error: " + path.error();
  
      return Error(path.error());
    }
  
    if (path.isSome() && os::exists(path.get())) {
      Try<Nothing> rmdir = os::rmdir(path.get(), true);
      if (rmdir.isError()) {
        LOG(ERROR) << "Could not delete fetcher cache directory '"
                   << cacheDirectory << "', error: " + rmdir.error();
  
        return rmdir;
      }
    }
  
    return Nothing();
  }
  
  
  Try<string> Fetcher::basename(const string& uri)
  {
    // TODO(bernd-mesos): full URI parsing, then move this to stout.
    // There is a bug (or is it a feature?) in the original fetcher
    // code without caching that remains in effect here. URIs are
    // treated like file paths, looking for occurrences of "/",
    // but ignoring other separators that can show up
    // (e.g. "?", "=" in HTTP URLs).
  
    if (uri.find_first_of('\\') != string::npos ||
        uri.find_first_of('\'') != string::npos ||
        uri.find_first_of('\0') != string::npos) {
        return Error("Illegal characters in URI");
    }
  
    size_t index = uri.find("://");
    if (index != string::npos && 1 < index) {
      // URI starts with protocol specifier, e.g., http://, https://,
      // ftp://, ftps://, hdfs://, hftp://, s3://, s3n://.
  
      string path = uri.substr(index + 3);
      if (!strings::contains(path, "/") || path.size() <= path.find("/") + 1) {
        return Error("Malformed URI (missing path): " + uri);
      }
  
      return path.substr(path.find_last_of("/") + 1);
    }
    return Path(uri).basename();
  }
  
  
  Try<Nothing> Fetcher::validateUri(const string& uri)
  {
    Try<string> result = basename(uri);
    if (result.isError()) {
      return Error(result.error());
    }
  
    return Nothing();
  }
  
  
  Try<Nothing> Fetcher::validateOutputFile(const string& path)
  {
    Try<string> result = Path(path).basename();
    if (result.isError()) {
      return Error(result.error());
    }
  
    if (path.size() == 0) {
      return Error("URI output file path is empty");
    }
  
    // TODO(mrbrowning): Check that the filename's directory component is
    // actually a subdirectory of the sandbox, not just relative to it.
    if (strings::startsWith(path, '/')) {
      return Error("URI output file must be within the sandbox directory");
    }
  
    return Nothing();
  }
  
  
  static Try<Nothing> validateUris(const CommandInfo& commandInfo)
  {
    foreach (const CommandInfo::URI& uri, commandInfo.uris()) {
      Try<Nothing> uriValidation = Fetcher::validateUri(uri.value());
      if (uriValidation.isError()) {
        return Error(uriValidation.error());
      }
  
      if (uri.has_output_file()) {
        Try<Nothing> outputFileValidation =
          Fetcher::validateOutputFile(uri.output_file());
        if (outputFileValidation.isError()) {
          return Error(outputFileValidation.error());
        }
      }
    }
  
    return Nothing();
  }
  
  
  Result<string> Fetcher::uriToLocalPath(
      const string& uri,
      const Option<string>& frameworksHome)
  {
    if (!strings::startsWith(uri, FILE_URI_PREFIX) &&
        strings::contains(uri, "://")) {
      return None();
    }
  
    string path = uri;
    bool fileUri = false;
  
    if (strings::startsWith(path, FILE_URI_LOCALHOST)) {
      path = path.substr(FILE_URI_LOCALHOST.size());
      fileUri = true;
    } else if (strings::startsWith(path, FILE_URI_PREFIX)) {
      path = path.substr(FILE_URI_PREFIX.size());
      fileUri = true;
    }
  
  #ifndef __WINDOWS__
    const bool isRelativePath = !strings::startsWith(path, "/");
  
    if (isRelativePath) {
      if (fileUri) {
        return Error("File URI only supports absolute paths");
      }
  
      if (frameworksHome.isNone() || frameworksHome.get().empty()) {
        return Error("A relative path was passed for the resource but the "
                     "Mesos framework home was not specified. "
                     "Please either provide this config option "
                     "or avoid using a relative path");
      } else {
        path = path::join(frameworksHome.get(), path);
        LOG(INFO) << "Prepended Mesos frameworks home to relative path, "
                  << "making it: '" << path << "'";
      }
    }
  #endif // __WINDOWS__
  
    return path;
  }
  
  
  bool Fetcher::isNetUri(const string& uri)
  {
    return strings::startsWith(uri, "http://")  ||
           strings::startsWith(uri, "https://") ||
           strings::startsWith(uri, "ftp://")   ||
           strings::startsWith(uri, "ftps://");
  }
  
  
  Future<Nothing> Fetcher::fetch(
      const ContainerID& containerId,
      const CommandInfo& commandInfo,
      const string& sandboxDirectory,
      const Option<string>& user,
      const SlaveID& slaveId,
      const Flags& flags)
  {
    return dispatch(process.get(),
                    &FetcherProcess::fetch,
                    containerId,
                    commandInfo,
                    sandboxDirectory,
                    user,
                    slaveId,
                    flags);
  }
  
  
  void Fetcher::kill(const ContainerID& containerId)
  {
    dispatch(process.get(), &FetcherProcess::kill, containerId);
  }
  
  
  FetcherProcess::~FetcherProcess()
  {
    foreach (const ContainerID& containerId, subprocessPids.keys()) {
      kill(containerId);
    }
  }
  
  
  // Find out how large a potential download from the given URI is.
  static Try<Bytes> fetchSize(
      const string& uri,
      const Option<string>& frameworksHome)
  {
    VLOG(1) << "Fetching size for URI: " << uri;
  
    Result<string> path = Fetcher::uriToLocalPath(uri, frameworksHome);
    if (path.isError()) {
      return Error(path.error());
    }
    if (path.isSome()) {
      Try<Bytes> size = os::stat::size(path.get(), os::stat::FOLLOW_SYMLINK);
      if (size.isError()) {
        return Error("Could not determine file size for: '" + path.get() +
                       "', error: " + size.error());
      }
      return size.get();
    }
  
    if (Fetcher::isNetUri(uri)) {
      Try<Bytes> size = net::contentLength(uri);
      if (size.isError()) {
        return Error(size.error());
      }
      if (size.get() == 0) {
        return Error("URI reported content-length 0: " + uri);
      }
  
      return size.get();
    }
  
    // TODO(hausdorff): (MESOS-5460) Explore adding support for fetching from
    // HDFS.
  #ifndef __WINDOWS__
    Try<Owned<HDFS>> hdfs = HDFS::create();
    if (hdfs.isError()) {
      return Error("Failed to create HDFS client: " + hdfs.error());
    }
  
    Future<Bytes> size = hdfs.get()->du(uri);
    size.await();
  
    if (!size.isReady()) {
      return Error("Hadoop client could not determine size: " +
                   (size.isFailed() ? size.failure() : "discarded"));
    }
  
    return size.get();
  #else
    return Error("Windows currently does not support fetching files from HDFS");
  #endif // __WINDOWS__
  }
  
  
  Future<Nothing> FetcherProcess::fetch(
      const ContainerID& containerId,
      const CommandInfo& commandInfo,
      const string& sandboxDirectory,
      const Option<string>& user,
      const SlaveID& slaveId,
      const Flags& flags)
  {
    VLOG(1) << "Starting to fetch URIs for container: " << containerId
            << ", directory: " << sandboxDirectory;
  
    // TODO(bernd-mesos): This will disappear once we inject flags at
    // Fetcher/FetcherProcess creation time. For now we trust this is
    // always the exact same value.
    cache.setSpace(flags.fetcher_cache_size);
  
    Try<Nothing> validated = validateUris(commandInfo);
    if (validated.isError()) {
      return Failure("Could not fetch: " + validated.error());
    }
  
    Option<string> commandUser = user;
    if (commandInfo.has_user()) {
      commandUser = commandInfo.user();
    }
  
    string cacheDirectory = paths::getSlavePath(flags.fetcher_cache_dir, slaveId);
    if (commandUser.isSome()) {
      // Segregating per-user cache directories.
      cacheDirectory = path::join(cacheDirectory, commandUser.get());
    }
  
    // For each URI we determine if we should use the cache and if so we
    // try and either get the cache entry or create a cache entry. If
    // we're getting the cache entry then we might need to wait for that
    // cache entry to be downloaded. If we're creating a new cache entry
    // then we need to properly reserve the cache space (and perform any
    // evictions). Thus, there are three possibilities for each URI:
    //
    //   (1) We are not using the cache.
    //   (2) We are using the cache but need to wait for an entry to be
    //       downloaded.
    //   (3) We are using the cache and need to create a new entry.
    //
    // We capture whether or not we're using the cache using an Option
    // as a value in a map, i.e., if we are not trying to use the cache
    // as in (1) above then the Option is None otherwise as in (2) and
    // (3) the Option is Some. And to capture the asynchronous nature of
    // both (2) and (3) that Option holds a Future to the actual cache
    // entry.
    hashmap<CommandInfo::URI, Option<Future<shared_ptr<Cache::Entry>>>> entries;
  
    foreach (const CommandInfo::URI& uri, commandInfo.uris()) {
      if (!uri.cache()) {
        entries[uri] = None();
        continue;
      }
  
      // Check if this is already in the cache (but not necessarily
      // downloaded).
      const Option<shared_ptr<Cache::Entry>> entry =
        cache.get(commandUser, uri.value());
  
      if (entry.isSome()) {
        entry.get()->reference();
  
        // Wait for the URI to be downloaded into the cache (or fail)
        entries[uri] = entry.get()->completion()
          .then(defer(self(), [=]() {
            return Future<shared_ptr<Cache::Entry>>(entry.get());
          }));
      } else {
        shared_ptr<Cache::Entry> newEntry =
          cache.create(cacheDirectory, commandUser, uri);
  
        newEntry->reference();
  
        entries[uri] =
          async([=]() {
            return fetchSize(uri.value(), flags.frameworks_home);
          })
          .then(defer(self(), [=](const Try<Bytes>& requestedSpace) {
            return reserveCacheSpace(requestedSpace, newEntry);
          }));
      }
    }
  
    // NOTE: We explicitly call the continuation '_fetch' even though it
    // looks like we could easily inline it here because we want to be
    // able to mock the function for testing! Don't remove this!
    return _fetch(entries,
                  containerId,
                  sandboxDirectory,
                  cacheDirectory,
                  commandUser,
                  flags);
  }
  
  
  Future<Nothing> FetcherProcess::_fetch(
      const hashmap<CommandInfo::URI, Option<Future<shared_ptr<Cache::Entry>>>>&
        entries,
      const ContainerID& containerId,
      const string& sandboxDirectory,
      const string& cacheDirectory,
      const Option<string>& user,
      const Flags& flags)
  {
    // Get out all of the futures we need to wait for so we can wait on
    // them together via 'await'.
    list<Future<shared_ptr<Cache::Entry>>> futures;
  
    foreachvalue (const Option<Future<shared_ptr<Cache::Entry>>>& entry,
                  entries) {
      if (entry.isSome()) {
        futures.push_back(entry.get());
      }
    }
  
    return await(futures)
      .then(defer(self(), [=]() {
        // For each URI, if there is a potential cache entry and waiting
        // for its associated future was successful, extract the entry
        // from the future and store it in 'result'. Otherwise we assume
        // we are not using or cannot use the cache for this URI.
        hashmap<CommandInfo::URI, Option<shared_ptr<Cache::Entry>>> result;
  
        foreachpair (const CommandInfo::URI& uri,
                     const Option<Future<shared_ptr<Cache::Entry>>>& entry,
                     entries) {
          if (entry.isSome()) {
            if (entry.get().isReady()) {
              result[uri] = entry.get().get();
            } else {
              LOG(WARNING)
                << "Reverting to fetching directly into the sandbox for '"
                << uri.value()
                << "', due to failure to fetch through the cache, "
                << "with error: " << entry.get().failure();
  
              result[uri] = None();
            }
          } else {
            // No entry means bypassing the cache.
            result[uri] = None();
          }
        }
  
        // NOTE: While we could inline '__fetch' we've explicitly kept
        // it as a separate function to minimize complexity. Like with
        // '_fetch', this also enables this phase of the fetcher cache
        // to easily be mocked for testing!
        return __fetch(result,
                       containerId,
                       sandboxDirectory,
                       cacheDirectory,
                       user,
                       flags);
      }));
  }
  
  
  Future<Nothing> FetcherProcess::__fetch(
      const hashmap<CommandInfo::URI, Option<shared_ptr<Cache::Entry>>>& entries,
      const ContainerID& containerId,
      const string& sandboxDirectory,
      const string& cacheDirectory,
      const Option<string>& user,
      const Flags& flags)
  {
    // Now construct the FetcherInfo based on which URIs we're using
    // the cache for and which ones we are bypassing the cache.
    FetcherInfo info;
  
    foreachpair (const CommandInfo::URI& uri,
                 const Option<shared_ptr<Cache::Entry>>& entry,
                 entries) {
      FetcherInfo::Item* item = info.add_items();
  
      item->mutable_uri()->CopyFrom(uri);
  
      if (entry.isSome()) {
        if (entry.get()->completion().isPending()) {
          // Since the entry is not yet "complete", i.e.,
          // 'completion().isPending()', it must be the case that we created
          // the entry in FetcherProcess::fetch(). Otherwise the entry should
          // have been in the cache already and we would have waited for its
          // completion in FetcherProcess::fetch().
          item->set_action(FetcherInfo::Item::DOWNLOAD_AND_CACHE);
          item->set_cache_filename(entry.get()->filename);
        } else {
          CHECK_READY(entry.get()->completion());
          item->set_action(FetcherInfo::Item::RETRIEVE_FROM_CACHE);
          item->set_cache_filename(entry.get()->filename);
        }
      } else {
        item->set_action(FetcherInfo::Item::BYPASS_CACHE);
      }
    }
  
    info.set_sandbox_directory(sandboxDirectory);
    info.set_cache_directory(cacheDirectory);
  
    if (user.isSome()) {
      info.set_user(user.get());
    }
  
    if (!flags.frameworks_home.empty()) {
      info.set_frameworks_home(flags.frameworks_home);
    }
  
    return run(containerId, sandboxDirectory, user, info, flags)
      .repair(defer(self(), [=](const Future<Nothing>& future) {
        LOG(ERROR) << "Failed to run mesos-fetcher: " << future.failure();
  
        foreachvalue (const Option<shared_ptr<Cache::Entry>>& entry, entries) {
          if (entry.isSome()) {
            entry.get()->unreference();
  
            if (entry.get()->completion().isPending()) {
              // Unsuccessfully (or partially) downloaded! Remove from the cache.
              entry.get()->fail();
              cache.remove(entry.get()); // Might delete partial download.
            }
          }
        }
  
        return future; // Always propagate the failure!
      })
      // Call to `operator` here forces the conversion on MSVC. This is implicit
      // on clang an gcc.
      .operator std::function<process::Future<Nothing>(
          const process::Future<Nothing> &)>())
      .then(defer(self(), [=]() {
        foreachvalue (const Option<shared_ptr<Cache::Entry>>& entry, entries) {
          if (entry.isSome()) {
            entry.get()->unreference();
  
            if (entry.get()->completion().isPending()) {
              // Successfully downloaded and cached!
  
              Try<Nothing> adjust = cache.adjust(entry.get());
              if (adjust.isSome()) {
                entry.get()->complete();
              } else {
                LOG(WARNING) << "Failed to adjust the cache size for entry '"
                             << entry.get()->key << "' with error: "
                             << adjust.error();
  
                // Successfully fetched, but not reusable from the
                // cache, because we are deleting the entry now.
                entry.get()->fail();
                cache.remove(entry.get());
              }
            }
          }
        }
  
        return Nothing();
      }));
  }
  
  
  static off_t delta(
      const Bytes& actualSize,
      const shared_ptr<FetcherProcess::Cache::Entry>& entry)
  {
    if (actualSize < entry->size) {
      Bytes delta = entry->size - actualSize;
      LOG(WARNING) << "URI download result for '" << entry->key
                   << "' is smaller than expected by " << stringify(delta)
                   << " at: " << entry->path();
  
      return -off_t(delta.bytes());
    } else if (actualSize > entry->size) {
      Bytes delta = actualSize - entry->size;
      LOG(WARNING) << "URI download result for '" << entry->key
                   << "' is larger than expected by " << stringify(delta)
                   << " at: " << entry->path();
  
      return off_t(delta.bytes());
    }
  
    return 0;
  }
  
  
  // For testing only.
  // TODO(bernd-mesos): After refactoring slave/containerizer,fetcher so
  // that flags and slave ID get injected, replace this with two functions
  // one of which returns a list of cache file paths, the other the number
  // of entries in the cache table.
  Try<list<Path>> FetcherProcess::cacheFiles(
      const SlaveID& slaveId,
      const Flags& flags)
  {
    list<Path> result;
  
    const string cacheDirectory =
      slave::paths::getSlavePath(flags.fetcher_cache_dir, slaveId);
  
    if (!os::exists(cacheDirectory)) {
      return result;
    }
  
    const Try<list<string>> find =
      os::find(cacheDirectory, CACHE_FILE_NAME_PREFIX);
  
    if (find.isError()) {
      return Error("Could not access cache directory '" +
                   cacheDirectory + "' with error: " + find.error());
    }
  
    transform(find.get().begin(),
              find.get().end(),
              std::back_inserter(result),
              [](const string& path) { return Path(path); });
  
    return result;
  }
  
  
  // For testing only.
  size_t FetcherProcess::cacheSize()
  {
    return cache.size();
  }
  
  
  Bytes FetcherProcess::availableCacheSpace()
  {
    return cache.availableSpace();
  }
  
  
  Future<shared_ptr<FetcherProcess::Cache::Entry>>
  FetcherProcess::reserveCacheSpace(
      const Try<Bytes>& requestedSpace,
      const shared_ptr<FetcherProcess::Cache::Entry>& entry)
  {
    if (requestedSpace.isError()) {
      // Let anyone waiting on this future know that we've
      // failed to download and they should bypass the cache
      // (any new requests will try again).
      entry->fail();
      cache.remove(entry);
  
      return Failure("Could not determine size of cache file for '" +
                     entry->key + "' with error: " +
                     requestedSpace.error());
    }
  
    Try<Nothing> reservation = cache.reserve(requestedSpace.get());
  
    if (reservation.isError()) {
      // Let anyone waiting on this future know that we've
      // failed to download and they should bypass the cache
      // (any new requests will try again).
      entry->fail();
      cache.remove(entry);
  
      return Failure("Failed to reserve space in the cache: " +
                     reservation.error());
    }
  
    VLOG(1) << "Claiming fetcher cache space for: " << entry->key;
  
    cache.claimSpace(requestedSpace.get());
  
    // NOTE: We must set the entry size only when we are also claiming
    // the space! Other functions rely on this dependency (see
    // Cache::remove()).
    entry->size = requestedSpace.get();
  
    return entry;
  }
  
  
  Future<Nothing> FetcherProcess::run(
      const ContainerID& containerId,
      const string& sandboxDirectory,
      const Option<string>& user,
      const FetcherInfo& info,
      const Flags& flags)
  {
    // Before we fetch let's make sure we create 'stdout' and 'stderr'
    // files into which we can redirect the output of the mesos-fetcher
    // (and later redirect the child's stdout/stderr).
  
    // TODO(tillt): Considering updating fetcher::run to take paths
    // instead of file descriptors and then use Subprocess::PATH()
    // instead of Subprocess::FD(). The reason this can't easily be done
    // today is because we not only need to open the files but also
    // chown them.
    const string stdoutPath = path::join(info.sandbox_directory(), "stdout");
  
    Try<int> out = os::open(
        stdoutPath,
        O_WRONLY | O_CREAT | O_TRUNC | O_NONBLOCK | O_CLOEXEC,
        S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
  
    if (out.isError()) {
      return Failure("Failed to create 'stdout' file: " + out.error());
    }
  
    string stderrPath = path::join(info.sandbox_directory(), "stderr");
    Try<int> err = os::open(
        stderrPath,
        O_WRONLY | O_CREAT | O_TRUNC | O_NONBLOCK | O_CLOEXEC,
        S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
  
    if (err.isError()) {
      os::close(out.get());
      return Failure("Failed to create 'stderr' file: " + err.error());
    }
  
  // NOTE: `os::chown` is not supported on Windows. The flag that gets passed in
  // here is conditionally compiled out on Windows.
  #ifndef __WINDOWS__
    if (user.isSome()) {
      // TODO(megha.sharma): Fetcher should not create seperate stdout/stderr
      // files but rather use FDs prepared by the container logger.
      // See MESOS-6271 for more details.
      Try<Nothing> chownOut = os::chown(
          user.get(),
          stdoutPath,
          false);
  
      if (chownOut.isError()) {
        os::close(out.get());
        os::close(err.get());
        return Failure(
            "Failed to chown '" +
            stdoutPath +
            "' to user '" + user.get() + "' : " +
            chownOut.error());
      }
  
      Try<Nothing> chownErr = os::chown(
          user.get(),
          stderrPath,
          false);
  
      if (chownErr.isError()) {
        os::close(out.get());
        os::close(err.get());
        return Failure(
            "Failed to chown '" +
            stderrPath +
            "' to user '" + user.get() + "' : " +
            chownErr.error());
      }
    }
  #endif // __WINDOWS__
  
    // Return early if there are no URIs to fetch.
    if (info.items_size() == 0) {
        os::close(out.get());
        os::close(err.get());
        return Nothing();
    }
  
    string fetcherPath = path::join(flags.launcher_dir, "mesos-fetcher");
    Result<string> realpath = os::realpath(fetcherPath);
  
    if (!realpath.isSome()) {
      LOG(ERROR) << "Failed to determine the canonical path "
                 << "for the mesos-fetcher '"
                 << fetcherPath
                 << "': "
                 << (realpath.isError() ? realpath.error()
                                        : "No such file or directory");
  
      os::close(out.get());
      os::close(err.get());
  
      return Failure("Could not fetch URIs: failed to find mesos-fetcher");
    }
  
    // Now the actual mesos-fetcher command.
    string command = realpath.get();
  
    // We pass arguments to the fetcher program by means of an
    // environment variable.
    map<string, string> environment = os::environment();
  
    // The libprocess port is explicitly removed because this will conflict
    // with the already-running agent.
    environment.erase("LIBPROCESS_PORT");
    environment.erase("LIBPROCESS_ADVERTISE_PORT");
  
    environment["MESOS_FETCHER_INFO"] = stringify(JSON::protobuf(info));
  
    if (!flags.hadoop_home.empty()) {
      environment["HADOOP_HOME"] = flags.hadoop_home;
    }
  
    VLOG(1) << "Fetching URIs using command '" << command << "'";
  
    Try<Subprocess> fetcherSubprocess = subprocess(
        command,
        Subprocess::PIPE(),
        Subprocess::FD(out.get(), Subprocess::IO::OWNED),
        Subprocess::FD(err.get(), Subprocess::IO::OWNED),
        environment);
  
    if (fetcherSubprocess.isError()) {
      return Failure("Failed to execute mesos-fetcher: " +
                     fetcherSubprocess.error());
    }
  
    // Remember this PID in case we need to kill the subprocess. See
    // FetcherProcess::kill(). This value gets removed after we wait on
    // the subprocess.
    subprocessPids[containerId] = fetcherSubprocess.get().pid();
  
    return fetcherSubprocess.get().status()
      .then(defer(self(), [=](const Option<int>& status) -> Future<Nothing> {
        if (status.isNone()) {
          return Failure("No status available from mesos-fetcher");
        }
  
        if (status.get() != 0) {
          return Failure("Failed to fetch all URIs for container '" +
                         stringify(containerId) +
                         "' with exit status: " +
                         stringify(status.get()));
        }
  
        return Nothing();
      }))
      .onFailed(defer(self(), [=](const string&) {
        // To aid debugging what went wrong when attempting to fetch, grab the
        // fetcher's local log output from the sandbox and log it here.
        Try<string> text = os::read(stderrPath);
        if (text.isSome()) {
          LOG(WARNING) << "Begin fetcher log (stderr in sandbox) for container "
                       << containerId << " from running command: " << command
                       << "\n" << text.get() << "\n"
                       << "End fetcher log for container " << containerId;
        } else {
          LOG(ERROR) << "Fetcher log (stderr in sandbox) for container "
                     << containerId << " not readable: " << text.error();
        }
      }))
      .onAny(defer(self(), [=](const Future<Nothing>&) {
        // Clear the subprocess PID remembered from running mesos-fetcher.
        subprocessPids.erase(containerId);
      }));
  }
  
  
  void FetcherProcess::kill(const ContainerID& containerId)
  {
    if (subprocessPids.contains(containerId)) {
      VLOG(1) << "Killing the fetcher for container '" << containerId << "'";
      // Best effort kill the entire fetcher tree.
      os::killtree(subprocessPids.get(containerId).get(), SIGKILL);
  
      subprocessPids.erase(containerId);
    }
  }
  
  
  string FetcherProcess::Cache::nextFilename(const CommandInfo::URI& uri)
  {
    // Different URIs may have the same base name, so we need to
    // segregate the download results. This can be done by separate
    // directories or by different file names. We opt for the latter
    // since there may be tighter limits on how many sub-directories a
    // file system can bear than on how many files can be in a directory.
  
    // We put a fixed prefix upfront before the serial number so we can
    // later easily find cache files with os::find() to support testing.
  
    // Why we keep the file extension here: When fetching from cache, if
    // extraction is enabled, the extraction algorithm can look at the
    // extension of the cache file the same way as it would at a
    // download of the original URI, and external commands performing
    // the extraction do not get confused by their source file
    // missing an expected form of extension. This is included in the
    // following.
  
    // Just for human operators who want to take a look at the cache
    // and relate cache files to URIs, we also add some of the URI's
    // basename, but not too much so we do not exceed file name size
    // limits.
  
    Try<string> base = Fetcher::basename(uri.value());
    CHECK_SOME(base);
  
    string s = base.get();
    if (s.size() > 20) {
      // Grab only a prefix and a suffix, but for sure including the
      // file extension.
      s = s.substr(0, 10) + "_" + s.substr(s.size() - 10, string::npos);
    }
  
    ++filenameSerial;
  
    return CACHE_FILE_NAME_PREFIX + stringify(filenameSerial) + "-" + s;
  }
  
  
  static string cacheKey(const Option<string>& user, const string& uri)
  {
    return user.isNone() ? uri : user.get() + "@" + uri;
  }
  
  
  shared_ptr<FetcherProcess::Cache::Entry> FetcherProcess::Cache::create(
      const string& cacheDirectory,
      const Option<string>& user,
      const CommandInfo::URI& uri)
  {
    const string key = cacheKey(user, uri.value());
    const string filename = nextFilename(uri);
  
    auto entry = shared_ptr<Cache::Entry>(
        new Cache::Entry(key, cacheDirectory, filename));
  
    table.put(key, entry);
    lruSortedEntries.push_back(entry);
  
    VLOG(1) << "Created cache entry '" << key << "' with file: " << filename;
  
    return entry;
  }
  
  
  Option<shared_ptr<FetcherProcess::Cache::Entry>>
  FetcherProcess::Cache::get(
      const Option<string>& user,
      const string& uri)
  {
    const string key = cacheKey(user, uri);
  
    Option<shared_ptr<Entry>> entry = table.get(key);
    if (entry.isSome()) {
      // Refresh the cache entry by moving it to the back of lruSortedEntries.
      lruSortedEntries.remove(entry.get());
      lruSortedEntries.push_back(entry.get());
    }
  
    return entry;
  }
  
  
  bool FetcherProcess::Cache::contains(
      const Option<string>& user,
      const string& uri)
  {
    const string key = cacheKey(user, uri);
    return table.get(key).isSome();
  }
  
  
  bool FetcherProcess::Cache::contains(const shared_ptr<Cache::Entry>& entry)
  {
    Option<shared_ptr<Cache::Entry>> found = table.get(entry->key);
    if (found.isNone()) {
      return false;
    }
  
    return found == entry;
  }
  
  
  // We are removing an entry if:
  //
  //   (1) We failed to determine its prospective cache file size.
  //   (2) We failed to download it when invoking the mesos-fetcher.
  //   (3) We're evicting it to make room for another entry.
  //
  // In (1) and (2) the contract is that we'll have failed the entry's
  // future before we call remove, so the entry's future should no
  // longer be pending.
  //
  // In (3) it should be the case that the future is no longer pending,
  // because we shouldn't be able to evict something if we're
  // currently downloading it, because it should have a non-zero
  // reference count and therefore the future must either be ready or
  // failed in which case this is just case (1) above.
  //
  // NOTE: It is not necessarily the case that this cache entry has
  // zero references because there might be some waiters on the
  // downloading of this entry which haven't been able to run and find
  // out that the downloading failed.
  //
  // We want to attempt to delete the file regardless of if it being
  // downloaded since it might have been downloaded partially! Deleting
  // this file should not be racing with any other downloading or
  // deleting because all calls into the cache are serialized by the
  // FetcherProcess and since this entry is already in the cache there
  // should not be any other conflicting entries or files representing
  // this entry. Furthermore every cache file has a unique name. Thus
  // no new download conflicts with the manipulation of any pre-existing
  // cache content.
  Try<Nothing> FetcherProcess::Cache::remove(
      const shared_ptr<Cache::Entry>& entry)
  {
    VLOG(1) << "Removing cache entry '" << entry->key
            << "' with filename: " << entry->filename;
  
    CHECK(!entry->completion().isPending());
  
    CHECK(contains(entry));
  
    table.erase(entry->key);
    lruSortedEntries.remove(entry);
  
    // We may or may not have started downloading. The download may or may
    // not have been partial. In any case, clean up whatever is there.
    if (os::exists(entry->path().string())) {
      Try<Nothing> rm = os::rm(entry->path().string());
      if (rm.isError()) {
        return Error("Could not delete fetcher cache file '" +
                     entry->path().string() + "' with error: " + rm.error() +
                     " for entry '" + entry->key +
                     "', leaking cache space: " + stringify(entry->size));
      }
    }
  
    // NOTE: There is an assumption that if and only if 'entry->size > 0'
    // then we've claimed cache space for this entry! This currently only
    // gets set in reserveCacheSpace().
    if (entry->size > 0) {
      releaseSpace(entry->size);
  
      entry->size = 0;
    }
  
    return Nothing();
  }
  
  
  // Select LRU cache entries for cache eviction.
  Try<list<shared_ptr<FetcherProcess::Cache::Entry>>>
  FetcherProcess::Cache::selectVictims(const Bytes& requiredSpace)
  {
    list<shared_ptr<FetcherProcess::Cache::Entry>> victims;
  
    Bytes space = 0;
  
    foreach (const shared_ptr<Cache::Entry>& entry, lruSortedEntries) {
      if (!entry->isReferenced()) {
        victims.push_back(entry);
  
        space += entry->size;
        if (space >= requiredSpace) {
          return victims;
        }
      }
    }
  
    return Error("Could not find enough cache files to evict");
  }
  
  
  Try<Nothing> FetcherProcess::Cache::reserve(
      const Bytes& requestedSpace)
  {
    if (availableSpace() < requestedSpace) {
      Bytes missingSpace = requestedSpace - availableSpace();
  
      VLOG(1) << "Freeing up fetcher cache space for: " << missingSpace;
  
      const Try<list<shared_ptr<Cache::Entry>>> victims =
        selectVictims(missingSpace);
  
      if (victims.isError()) {
        return Error("Could not free up enough fetcher cache space");
      }
  
      foreach (const shared_ptr<Cache::Entry>& entry, victims.get()) {
        Try<Nothing> removal = remove(entry);
        if (removal.isError()) {
          return Error(removal.error());
        }
      }
    }
  
    return Nothing();
  }
  
  
  Try<Nothing> FetcherProcess::Cache::adjust(
      const shared_ptr<FetcherProcess::Cache::Entry>& entry)
  {
    CHECK(contains(entry));
  
    Try<Bytes> size = os::stat::size(
        entry.get()->path().string(),
        os::stat::DO_NOT_FOLLOW_SYMLINK);
  
    if (size.isSome()) {
      off_t d = delta(size.get(), entry);
      if (d <= 0) {
        entry->size = size.get();
  
        releaseSpace(Bytes(d));
      } else {
        return Error("More cache size now necessary, not adjusting " +
                     entry->key);
      }
    } else {
      // This should never be caused by Mesos itself, but cannot be excluded.
      return Error("Fetcher cache file for '" + entry->key +
                   "' disappeared from: " + entry->path().string());
    }
  
    return Nothing();
  }
  
  
  size_t FetcherProcess::Cache::size()
  {
    return table.size();
  }
  
  
  void FetcherProcess::Cache::setSpace(const Bytes& bytes)
  {
    if (space > 0) {
      // Dynamic cache size changes not supported.
      CHECK_EQ(space, bytes);
    } else {
      space = bytes;
    }
  }
  
  
  void FetcherProcess::Cache::claimSpace(const Bytes& bytes)
  {
    tally += bytes;
  
    if (tally > space) {
      // Used cache volume space exceeds the maximum amount set by
      // flags.fetcher_cache_size. This may be tolerated temporarily,
      // if there is sufficient physical space available. But it can
      // otherwise cause unspecified system behavior at any moment.
      LOG(WARNING) << "Fetcher cache space overflow - space used: " << tally
                   << ", exceeds total fetcher cache space: " << space;
    }
  
    VLOG(1) << "Claimed cache space: " << bytes << ", now using: " << tally;
  }
  
  
  void FetcherProcess::Cache::releaseSpace(const Bytes& bytes)
  {
    CHECK(bytes <= tally) << "Attempt to release more cache space than in use - "
                          << " requested: " << bytes << ", in use: " << tally;
  
  
    tally -= bytes;
  
    VLOG(1) << "Released cache space: " << bytes << ", now using: " << tally;
  }
  
  
  Bytes FetcherProcess::Cache::availableSpace()
  {
    if (tally > space) {
      LOG(WARNING) << "Fetcher cache space overflow - space used: " << tally
                   << ", exceeds total fetcher cache space: " << space;
      return 0;
    }
  
    return space - tally;
  }
  
  
  void FetcherProcess::Cache::Entry::complete()
  {
    CHECK_PENDING(promise.future());
  
    promise.set(Nothing());
  }
  
  
  Future<Nothing> FetcherProcess::Cache::Entry::completion()
  {
    return promise.future();
  }
  
  
  void FetcherProcess::Cache::Entry::fail()
  {
    CHECK_PENDING(promise.future());
  
    promise.fail("Could not download to fetcher cache: " + key);
  }
  
  
  void FetcherProcess::Cache::Entry::reference()
  {
    referenceCount++;
  }
  
  
  void FetcherProcess::Cache::Entry::unreference()
  {
    CHECK(referenceCount > 0);
  
    referenceCount--;
  }
  
  
  bool FetcherProcess::Cache::Entry::isReferenced()
  {
    return referenceCount > 0;
  }
  
  } // namespace slave {
  } // namespace internal {
  } // namespace mesos {
  