<commit_before>/**
 * 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 <list>
#include <sstream>
#include <string>

#include <glog/logging.h>

#include <process/collect.hpp>

#include <stout/error.hpp>
#include <stout/foreach.hpp>
#include <stout/os.hpp>
#include <stout/path.hpp>
#include <stout/stringify.hpp>
#include <stout/strings.hpp>

#include "linux/fs.hpp"
#include "linux/ns.hpp"

#include "slave/paths.hpp"

#include "slave/containerizer/isolators/filesystem/linux.hpp"

using namespace process;

using std::list;
using std::ostringstream;
using std::string;

using mesos::slave::ContainerState;
using mesos::slave::ContainerLimitation;
using mesos::slave::ContainerPrepareInfo;
using mesos::slave::Isolator;

namespace mesos {
namespace internal {
namespace slave {

Try<Isolator*> LinuxFilesystemIsolatorProcess::create(
    const Flags& flags,
    const Owned<Provisioner>& provisioner)
{
  Result<string> user = os::user();
  if (!user.isSome()) {
    return Error("Failed to determine user: " +
                 (user.isError() ? user.error() : "username not found"));
  }

  if (user.get() != "root") {
    return Error("LinuxFilesystemIsolator requires root privileges");
  }

  Owned<MesosIsolatorProcess> process(
      new LinuxFilesystemIsolatorProcess(flags, provisioner));

  return new MesosIsolator(process);
}


LinuxFilesystemIsolatorProcess::LinuxFilesystemIsolatorProcess(
    const Flags& _flags,
    const Owned<Provisioner>& _provisioner)
  : flags(_flags),
    provisioner(_provisioner) {}


LinuxFilesystemIsolatorProcess::~LinuxFilesystemIsolatorProcess() {}


Future<Nothing> LinuxFilesystemIsolatorProcess::recover(
    const list<ContainerState>& states,
    const hashset<ContainerID>& orphans)
{
  // Read the mount table in the host mount namespace to recover paths
  // to containers' work directories if their root filesystems are
  // changed. Method 'cleanup()' relies on this information to clean
  // up mounts in the host mount namespace for each container.
  Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
  if (table.isError()) {
    return Failure("Failed to get mount table: " + table.error());
  }

  foreach (const ContainerState& state, states) {
    Owned<Info> info(new Info(state.directory()));

    foreach (const fs::MountInfoTable::Entry& entry, table.get().entries) {
      if (entry.root == info->directory) {
        info->sandbox = entry.target;
        break;
      }
    }

    infos.put(state.container_id(), info);
  }

  // Recover both known and unknown orphans by scanning the mount
  // table and finding those mounts whose roots are under slave's
  // sandbox root directory. Those mounts are container's work
  // directory mounts. Mounts from unknown orphans will be cleaned up
  // immediately. Mounts from known orphans will be cleaned up when
  // those known orphan containers are being destroyed by the slave.
  hashset<ContainerID> unknownOrphans;

  string sandboxRootDir = paths::getSandboxRootDir(flags.work_dir);

  foreach (const fs::MountInfoTable::Entry& entry, table.get().entries) {
    if (!strings::startsWith(entry.root, sandboxRootDir)) {
      continue;
    }

    // TODO(jieyu): Here, we retrieve the container ID by taking the
    // basename of 'entry.root'. This assumes that the slave's sandbox
    // root directory are organized according to the comments in the
    // beginning of slave/paths.hpp.
    ContainerID containerId;
    containerId.set_value(Path(entry.root).basename());

    if (infos.contains(containerId)) {
      continue;
    }

    Owned<Info> info(new Info(entry.root));

    if (entry.root != entry.target) {
      info->sandbox = entry.target;
    }

    infos.put(containerId, info);

    // Remember all the unknown orphan containers.
    if (!orphans.contains(containerId)) {
      unknownOrphans.insert(containerId);
    }
  }

  // Cleanup mounts from unknown orphans.
  list<Future<Nothing>> futures;
  foreach (const ContainerID& containerId, unknownOrphans) {
    futures.push_back(cleanup(containerId));
  }

  return collect(futures)
    .then(defer(PID<LinuxFilesystemIsolatorProcess>(this),
                &LinuxFilesystemIsolatorProcess::_recover,
                states,
                orphans));
}


Future<Nothing> LinuxFilesystemIsolatorProcess::_recover(
    const list<ContainerState>& states,
    const hashset<ContainerID>& orphans)
{
  return provisioner->recover(states, orphans)
    .then([]() -> Future<Nothing> { return Nothing(); });
}


Future<Option<ContainerPrepareInfo>> LinuxFilesystemIsolatorProcess::prepare(
    const ContainerID& containerId,
    const ExecutorInfo& executorInfo,
    const string& directory,
    const Option<string>& user)
{
  if (infos.contains(containerId)) {
    return Failure("Container has already been prepared");
  }

  infos.put(containerId, Owned<Info>(new Info(directory)));

  if (!executorInfo.has_container()) {
    return __prepare(containerId, executorInfo, directory, user, None());
  }

  // Provision the root filesystem if needed.
  CHECK_EQ(executorInfo.container().type(), ContainerInfo::MESOS);

  if (!executorInfo.container().mesos().has_image()) {
    return _prepare(containerId, executorInfo, directory, user, None());
  }

  const Image& image = executorInfo.container().mesos().image();

  return provisioner->provision(containerId, image)
    .then(defer(PID<LinuxFilesystemIsolatorProcess>(this),
                &LinuxFilesystemIsolatorProcess::_prepare,
                containerId,
                executorInfo,
                directory,
                user,
                lambda::_1));
}


Future<Option<ContainerPrepareInfo>> LinuxFilesystemIsolatorProcess::_prepare(
    const ContainerID& containerId,
    const ExecutorInfo& executorInfo,
    const string& directory,
    const Option<string>& user,
    const Option<string>& rootfs)
{
  CHECK(executorInfo.has_container());
  CHECK_EQ(executorInfo.container().type(), ContainerInfo::MESOS);

  // We will provision the images specified in ContainerInfo::volumes
  // as well. We will mutate ContainerInfo::volumes to include the
  // paths to the provisioned root filesystems (by setting the
  // 'host_path') if the volume specifies an image as the source.
  Owned<ExecutorInfo> _executorInfo(new ExecutorInfo(executorInfo));
  list<Future<Nothing>> futures;

  for (int i = 0; i < _executorInfo->container().volumes_size(); i++) {
    Volume* volume = _executorInfo->mutable_container()->mutable_volumes(i);

    if (!volume->has_image()) {
      continue;
    }

    const Image& image = volume->image();

    futures.push_back(
        provisioner->provision(containerId, image)
          .then([volume](const string& path) -> Future<Nothing> {
            volume->set_host_path(path);
            return Nothing();
          }));
  }

  return collect(futures)
    .then([=]() -> Future<Option<ContainerPrepareInfo>> {
      return __prepare(containerId, *_executorInfo, directory, user, rootfs);
    });
}


Future<Option<ContainerPrepareInfo>> LinuxFilesystemIsolatorProcess::__prepare(
    const ContainerID& containerId,
    const ExecutorInfo& executorInfo,
    const string& directory,
    const Option<string>& user,
    const Option<string>& rootfs)
{
  CHECK(infos.contains(containerId));

  const Owned<Info>& info = infos[containerId];

  ContainerPrepareInfo prepareInfo;
  prepareInfo.set_namespaces(CLONE_NEWNS);

  if (rootfs.isNone()) {
    // It the container does not change its root filesystem, we need
    // to do a self bind mount of the container's work directory and
    // mark it as a shared mount. This is necessary for any mounts
    // underneath it to be propagated into the container's mount
    // namespace. This is how we can update persistent volumes.
    LOG(INFO) << "Bind mounting work directory '" << directory
              << "' for container " << containerId;

    Try<Nothing> mount = fs::mount(
        directory,
        directory,
        None(),
        MS_BIND,
        NULL);

    if (mount.isError()) {
      return Failure(
          "Failed to self bind mount work directory '" +
          directory + "': " + mount.error());
    }

    mount = fs::mount(
        None(),
        directory,
        None(),
        MS_SHARED,
        NULL);

    if (mount.isError()) {
      return Failure(
          "Failed to mark work directory '" + directory +
          "' as a shared mount: " + mount.error());
    }
  } else {
    // If the container changes its root filesystem, we need to mount
    // the container's work directory into its root filesystem
    // (creating it if needed) so that the executor and the task can
    // access the work directory.
    //
    // NOTE: The mount of the work directory must be a shared mount in
    // the host filesystem so that any mounts underneath it will
    // propagate into the container's mount namespace. This is how we
    // can update persistent volumes for the container.

    // This is the mount point of the work directory in the root filesystem.
    const string sandbox = path::join(rootfs.get(), flags.sandbox_directory);

    // Save the path 'sandbox' which will be used in 'cleanup()'.
    info->sandbox = sandbox;

    if (!os::exists(sandbox)) {
      Try<Nothing> mkdir = os::mkdir(sandbox);
      if (mkdir.isError()) {
        return Failure(
            "Failed to create sandbox mount point at '" +
            sandbox + "': " + mkdir.error());
      }
    }

    LOG(INFO) << "Bind mounting work directory from '" << directory
              << "' to '" << sandbox << "' for container " << containerId;

    Try<Nothing> mount = fs::mount(
        directory,
        sandbox,
        None(),
        MS_BIND,
        NULL);

    if (mount.isError()) {
      return Failure(
          "Failed to mount work directory '" + directory +
          "' to '" + sandbox + "': " + mount.error());
    }

    mount = fs::mount(
        None(),
        sandbox,
        None(),
        MS_SHARED,
        NULL);

    if (mount.isError()) {<commit_msg>Fixed MESOS-3430 by making the sandbox/work directory mount as shared and slave.<commit_after>/**
 * 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 <list>
#include <sstream>
#include <string>

#include <glog/logging.h>

#include <process/collect.hpp>

#include <stout/error.hpp>
#include <stout/foreach.hpp>
#include <stout/os.hpp>
#include <stout/path.hpp>
#include <stout/stringify.hpp>
#include <stout/strings.hpp>

#include "linux/fs.hpp"
#include "linux/ns.hpp"

#include "slave/paths.hpp"

#include "slave/containerizer/isolators/filesystem/linux.hpp"

using namespace process;

using std::list;
using std::ostringstream;
using std::string;

using mesos::slave::ContainerState;
using mesos::slave::ContainerLimitation;
using mesos::slave::ContainerPrepareInfo;
using mesos::slave::Isolator;

namespace mesos {
namespace internal {
namespace slave {

Try<Isolator*> LinuxFilesystemIsolatorProcess::create(
    const Flags& flags,
    const Owned<Provisioner>& provisioner)
{
  Result<string> user = os::user();
  if (!user.isSome()) {
    return Error("Failed to determine user: " +
                 (user.isError() ? user.error() : "username not found"));
  }

  if (user.get() != "root") {
    return Error("LinuxFilesystemIsolator requires root privileges");
  }

  Owned<MesosIsolatorProcess> process(
      new LinuxFilesystemIsolatorProcess(flags, provisioner));

  return new MesosIsolator(process);
}


LinuxFilesystemIsolatorProcess::LinuxFilesystemIsolatorProcess(
    const Flags& _flags,
    const Owned<Provisioner>& _provisioner)
  : flags(_flags),
    provisioner(_provisioner) {}


LinuxFilesystemIsolatorProcess::~LinuxFilesystemIsolatorProcess() {}


Future<Nothing> LinuxFilesystemIsolatorProcess::recover(
    const list<ContainerState>& states,
    const hashset<ContainerID>& orphans)
{
  // Read the mount table in the host mount namespace to recover paths
  // to containers' work directories if their root filesystems are
  // changed. Method 'cleanup()' relies on this information to clean
  // up mounts in the host mount namespace for each container.
  Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
  if (table.isError()) {
    return Failure("Failed to get mount table: " + table.error());
  }

  foreach (const ContainerState& state, states) {
    Owned<Info> info(new Info(state.directory()));

    foreach (const fs::MountInfoTable::Entry& entry, table.get().entries) {
      if (entry.root == info->directory) {
        info->sandbox = entry.target;
        break;
      }
    }

    infos.put(state.container_id(), info);
  }

  // Recover both known and unknown orphans by scanning the mount
  // table and finding those mounts whose roots are under slave's
  // sandbox root directory. Those mounts are container's work
  // directory mounts. Mounts from unknown orphans will be cleaned up
  // immediately. Mounts from known orphans will be cleaned up when
  // those known orphan containers are being destroyed by the slave.
  hashset<ContainerID> unknownOrphans;

  string sandboxRootDir = paths::getSandboxRootDir(flags.work_dir);

  foreach (const fs::MountInfoTable::Entry& entry, table.get().entries) {
    if (!strings::startsWith(entry.root, sandboxRootDir)) {
      continue;
    }

    // TODO(jieyu): Here, we retrieve the container ID by taking the
    // basename of 'entry.root'. This assumes that the slave's sandbox
    // root directory are organized according to the comments in the
    // beginning of slave/paths.hpp.
    ContainerID containerId;
    containerId.set_value(Path(entry.root).basename());

    if (infos.contains(containerId)) {
      continue;
    }

    Owned<Info> info(new Info(entry.root));

    if (entry.root != entry.target) {
      info->sandbox = entry.target;
    }

    infos.put(containerId, info);

    // Remember all the unknown orphan containers.
    if (!orphans.contains(containerId)) {
      unknownOrphans.insert(containerId);
    }
  }

  // Cleanup mounts from unknown orphans.
  list<Future<Nothing>> futures;
  foreach (const ContainerID& containerId, unknownOrphans) {
    futures.push_back(cleanup(containerId));
  }

  return collect(futures)
    .then(defer(PID<LinuxFilesystemIsolatorProcess>(this),
                &LinuxFilesystemIsolatorProcess::_recover,
                states,
                orphans));
}


Future<Nothing> LinuxFilesystemIsolatorProcess::_recover(
    const list<ContainerState>& states,
    const hashset<ContainerID>& orphans)
{
  return provisioner->recover(states, orphans)
    .then([]() -> Future<Nothing> { return Nothing(); });
}


Future<Option<ContainerPrepareInfo>> LinuxFilesystemIsolatorProcess::prepare(
    const ContainerID& containerId,
    const ExecutorInfo& executorInfo,
    const string& directory,
    const Option<string>& user)
{
  if (infos.contains(containerId)) {
    return Failure("Container has already been prepared");
  }

  infos.put(containerId, Owned<Info>(new Info(directory)));

  if (!executorInfo.has_container()) {
    return __prepare(containerId, executorInfo, directory, user, None());
  }

  // Provision the root filesystem if needed.
  CHECK_EQ(executorInfo.container().type(), ContainerInfo::MESOS);

  if (!executorInfo.container().mesos().has_image()) {
    return _prepare(containerId, executorInfo, directory, user, None());
  }

  const Image& image = executorInfo.container().mesos().image();

  return provisioner->provision(containerId, image)
    .then(defer(PID<LinuxFilesystemIsolatorProcess>(this),
                &LinuxFilesystemIsolatorProcess::_prepare,
                containerId,
                executorInfo,
                directory,
                user,
                lambda::_1));
}


Future<Option<ContainerPrepareInfo>> LinuxFilesystemIsolatorProcess::_prepare(
    const ContainerID& containerId,
    const ExecutorInfo& executorInfo,
    const string& directory,
    const Option<string>& user,
    const Option<string>& rootfs)
{
  CHECK(executorInfo.has_container());
  CHECK_EQ(executorInfo.container().type(), ContainerInfo::MESOS);

  // We will provision the images specified in ContainerInfo::volumes
  // as well. We will mutate ContainerInfo::volumes to include the
  // paths to the provisioned root filesystems (by setting the
  // 'host_path') if the volume specifies an image as the source.
  Owned<ExecutorInfo> _executorInfo(new ExecutorInfo(executorInfo));
  list<Future<Nothing>> futures;

  for (int i = 0; i < _executorInfo->container().volumes_size(); i++) {
    Volume* volume = _executorInfo->mutable_container()->mutable_volumes(i);

    if (!volume->has_image()) {
      continue;
    }

    const Image& image = volume->image();

    futures.push_back(
        provisioner->provision(containerId, image)
          .then([volume](const string& path) -> Future<Nothing> {
            volume->set_host_path(path);
            return Nothing();
          }));
  }

  return collect(futures)
    .then([=]() -> Future<Option<ContainerPrepareInfo>> {
      return __prepare(containerId, *_executorInfo, directory, user, rootfs);
    });
}


Future<Option<ContainerPrepareInfo>> LinuxFilesystemIsolatorProcess::__prepare(
    const ContainerID& containerId,
    const ExecutorInfo& executorInfo,
    const string& directory,
    const Option<string>& user,
    const Option<string>& rootfs)
{
  CHECK(infos.contains(containerId));

  const Owned<Info>& info = infos[containerId];

  ContainerPrepareInfo prepareInfo;
  prepareInfo.set_namespaces(CLONE_NEWNS);

  if (rootfs.isNone()) {
    // It the container does not change its root filesystem, we need
    // to do a self bind mount of the container's work directory and
    // mark it as a shared mount. This is necessary for any mounts
    // underneath it to be propagated into the container's mount
    // namespace. This is how we can update persistent volumes.
    LOG(INFO) << "Bind mounting work directory '" << directory
              << "' for container " << containerId;

    Try<Nothing> mount = fs::mount(
        directory,
        directory,
        None(),
        MS_BIND,
        NULL);

    if (mount.isError()) {
      return Failure(
          "Failed to self bind mount work directory '" +
          directory + "': " + mount.error());
    }

    mount = fs::mount(
        None(),
        directory,
        None(),
        MS_SLAVE,
        NULL);

    if (mount.isError()) {
      return Failure(
          "Failed to mark work directory '" + directory +
          "' as a slave mount: " + mount.error());
    }

    mount = fs::mount(
        None(),
        directory,
        None(),
        MS_SHARED,
        NULL);

    if (mount.isError()) {
      return Failure(
          "Failed to mark work directory '" + directory +
          "' as a shared mount: " + mount.error());
    }
  } else {
    // If the container changes its root filesystem, we need to mount
    // the container's work directory into its root filesystem
    // (creating it if needed) so that the executor and the task can
    // access the work directory.
    //
    // NOTE: The mount of the work directory must be a shared mount in
    // the host filesystem so that any mounts underneath it will
    // propagate into the container's mount namespace. This is how we
    // can update persistent volumes for the container.

    // This is the mount point of the work directory in the root filesystem.
    const string sandbox = path::join(rootfs.get(), flags.sandbox_directory);

    // Save the path 'sandbox' which will be used in 'cleanup()'.
    info->sandbox = sandbox;

    if (!os::exists(sandbox)) {
      Try<Nothing> mkdir = os::mkdir(sandbox);
      if (mkdir.isError()) {
        return Failure(
            "Failed to create sandbox mount point at '" +
            sandbox + "': " + mkdir.error());
      }
    }

    LOG(INFO) << "Bind mounting work directory from '" << directory
              << "' to '" << sandbox << "' for container " << containerId;

    Try<Nothing> mount = fs::mount(
        directory,
        sandbox,
        None(),
        MS_BIND,
        NULL);

    if (mount.isError()) {
      return Failure(
          "Failed to mount work directory '" + directory +
          "' to '" + sandbox + "': " + mount.error());
    }

    mount = fs::mount(
        None(),
        sandbox,
        None(),
        MS_SLAVE,
        NULL);

    if (mount.isError()) {
      return Failure(
          "Failed to mark sandbox '" + sandbox +
          "' as a slave mount: " + mount.error());
    }

    mount = fs::mount(
        None(),
        sandbox,
        None(),
        MS_SHARED,
        NULL);

    if (mount.isError()) {
{'fraction_non_alphanumeric': 0.11346465191581219, 'fraction_numerical': 0.0007195538765965102, 'mean_word_length': 3.1988293051359515, 'pattern_counts': {'":': 0, '<': 151, '<?xml version=': 0, '>': 141, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '673212', 'n_tokens_mistral': 6466, 'n_tokens_neox': 6231, 'n_words': 1890}