Mesos源码分析(13): MesosContainerier运行一个Task

2022-10-24 08:16:47

MesosContainerizer的实现在文件src/slave/containerizer/mesos/containerizer.cpp中

 

  1. Future<bool> MesosContainerizer::launch(
  2.     const ContainerID& containerId,
  3.     const TaskInfo& taskInfo,
  4.     const ExecutorInfo& executorInfo,
  5.     const
    string& directory,
  6.     const Option<string>& user,
  7.     const SlaveID& slaveId,
  8.     const PID<Slave>& slavePid,
  9.     bool checkpoint)
  10. {
  11.   return dispatch(process.get(),
  12.                   &MesosContainerizerProcess::launch,
  13.                   containerId,
  14.                   taskInfo,
  15.                   executorInfo,
  16.                   directory,
  17.                   user,
  18.                   slaveId,
  19.                   slavePid,
  20.                   checkpoint);
  21. }

 

转而调用MesosContainerizerProcess::launch，只有executorInfo.has_container()是Mesos的时候，才使用MesosContainerizer.

  1. // Launching an executor involves the following steps:
  2. // 1. Call prepare on each isolator.
  3. // 2. Fork the executor. The forked child is blocked from exec'ing until it has
  4. // been isolated.
  5. // 3. Isolate the executor. Call isolate with the pid for each isolator.
  6. // 4. Fetch the executor.
  7. // 5. Exec the executor. The forked child is signalled to continue. It will
  8. // first execute any preparation commands from isolators and then exec the
  9. // executor.
  10. Future<bool> MesosContainerizerProcess::launch(
  11.     const ContainerID& containerId,
  12.     const Option<TaskInfo>& taskInfo,
  13.     const ExecutorInfo& _executorInfo,
  14.     const
    string& directory,
  15.     const Option<string>& user,
  16.     const SlaveID& slaveId,
  17.     const PID<Slave>& slavePid,
  18.     bool checkpoint)
  19. {
  20.   if (containers_.contains(containerId)) {
  21.     return Failure("Container already started");
  22.   }
  23.  
  24.   if (taskInfo.isSome() &&
  25.       taskInfo.get().has_container() &&
  26.       taskInfo.get().container().type() != ContainerInfo::MESOS) {
  27.     return
    false;
  28.   }
  29.  
  30.   // NOTE: We make a copy of the executor info because we may mutate
  31.   // it with default container info.
  32.   ExecutorInfo executorInfo = _executorInfo;
  33.  
  34.   if (executorInfo.has_container() &&
  35.       executorInfo.container().type() != ContainerInfo::MESOS) {
  36.     return
    false;
  37.   }
  38.  
  39.   // Add the default container info to the executor info.
  40.   // TODO(jieyu): Rename the flag to be default_mesos_container_info.
  41.   if (!executorInfo.has_container() &&
  42.       flags.default_container_info.isSome()) {
  43.     executorInfo.mutable_container()->CopyFrom(
  44.         flags.default_container_info.get());
  45.   }
  46.  
  47.   LOG(INFO) << "Starting container '" << containerId
  48.             << "' for executor '" << executorInfo.executor_id()
  49.             << "' of framework '" << executorInfo.framework_id() << "'";
  50.  
  51.   Container* container = new Container();
  52.   container->directory = directory;
  53.   container->state = PROVISIONING;
  54.   container->resources = executorInfo.resources();
  55.  
  56.   // We need to set the `launchInfos` to be a ready future initially
  57.   // before we starting calling isolator->prepare() because otherwise,
  58.   // the destroy will wait forever trying to wait for this future to
  59.   // be ready , which it never will. See MESOS-4878.
  60.   container->launchInfos = list<Option<ContainerLaunchInfo>>();
  61.  
  62.   containers_.put(containerId, Owned<Container>(container));
  63.  
  64.   if (!executorInfo.has_container()) {
  65.     return prepare(containerId, taskInfo, executorInfo, directory, user, None())
  66.       .then(defer(self(),
  67.                   &Self::__launch,
  68.                   containerId,
  69.                   executorInfo,
  70.                   directory,
  71.                   user,
  72.                   slaveId,
  73.                   slavePid,
  74.                   checkpoint,
  75.                   lambda::_1));
  76.   }
  77.  
  78.   // Provision the root filesystem if needed.
  79.   CHECK_EQ(executorInfo.container().type(), ContainerInfo::MESOS);
  80.  
  81.   if (!executorInfo.container().mesos().has_image()) {
  82.     return _launch(containerId,
  83.                    taskInfo,
  84.                    executorInfo,
  85.                    directory,
  86.                    user,
  87.                    slaveId,
  88.                    slavePid,
  89.                    checkpoint,
  90.                    None());
  91.   }
  92.  
  93.   const Image& image = executorInfo.container().mesos().image();
  94.  
  95.   Future<ProvisionInfo> future =
  96.     provisioner->provision(containerId, image);
  97.  
  98.   container->provisionInfos.push_back(future);
  99.  
  100.   return future
  101.     .then(defer(PID<MesosContainerizerProcess>(this),
  102.                 &MesosContainerizerProcess::_launch,
  103.                 containerId,
  104.                 taskInfo,
  105.                 executorInfo,
  106.                 directory,
  107.                 user,
  108.                 slaveId,
  109.                 slavePid,
  110.                 checkpoint,
  111.                 lambda::_1));
  112. }

 

大家注意ExecutorInfo里面的ContainerInfo和TaskInfo里面的ContainerInfo不同。

如果大家看protocol buffer的定义文件include/mesos/mesos.proto里面，ExecutorInfo里面有一个ContainerInfo

  1. message ExecutorInfo {
  2.   required ExecutorID executor_id = 1;
  3.   optional FrameworkID framework_id = 8; // TODO(benh): Make this required.
  4.   required CommandInfo command = 7;
  5.   // Executor provided with a container will launch the container
  6.   // with the executor's CommandInfo and we expect the container to
  7.   // act as a Mesos executor.
  8.   optional ContainerInfo container = 11;
  9.   repeated Resource resources = 5;
  10.   optional string name = 9;
  11.  
  12.   // 'source' is an identifier style string used by frameworks to
  13.   // track the source of an executor. This is useful when it's
  14.   // possible for different executor ids to be related semantically.
  15.   //
  16.   // NOTE: 'source' is exposed alongside the resource usage of the
  17.   // executor via JSON on the slave. This allows users to import usage
  18.   // information into a time series database for monitoring.
  19.   optional string source = 10;
  20.  
  21.   optional bytes data = 4;
  22.  
  23.   // Service discovery information for the executor. It is not
  24.   // interpreted or acted upon by Mesos. It is up to a service
  25.   // discovery system to use this information as needed and to handle
  26.   // executors without service discovery information.
  27.   optional DiscoveryInfo discovery = 12;
  28. }

 

如果ExecutorInfo的ContainerInfo有值，则executor会启动在这个container里面。

那marathon里面的container info放在哪里呢？

TaskInfo里面也有一个ContainerInfo

  1. message TaskInfo {
  2.   required string name = 1;
  3.   required TaskID task_id = 2;
  4.   required SlaveID slave_id = 3;
  5.   repeated Resource resources = 4;
  6.   optional ExecutorInfo executor = 5;
  7.   optional CommandInfo command = 7;
  8.   // Task provided with a container will launch the container as part
  9.   // of this task paired with the task's CommandInfo.
  10.   optional ContainerInfo container = 9;
  11.   optional bytes data = 6;
  12.   // A health check for the task (currently in *alpha* and initial
  13.   // support will only be for TaskInfo's that have a CommandInfo).
  14.   optional HealthCheck health_check = 8;
  15.  
  16.   // Labels are free-form key value pairs which are exposed through
  17.   // master and slave endpoints. Labels will not be interpreted or
  18.   // acted upon by Mesos itself. As opposed to the data field, labels
  19.   // will be kept in memory on master and slave processes. Therefore,
  20.   // labels should be used to tag tasks with light-weight meta-data.
  21.   // Labels should not contain duplicate key-value pairs.
  22.   optional Labels labels = 10;
  23.  
  24.   // Service discovery information for the task. It is not interpreted
  25.   // or acted upon by Mesos. It is up to a service discovery system
  26.   // to use this information as needed and to handle tasks without
  27.   // service discovery information.
  28.   optional DiscoveryInfo discovery = 11;
  29. }

 

如果TaskInfo里面的ContainerInfo有值，才是真正的运行容器，容器里面运行任务。

 

最终会调用MesosContainerizerProcess::__launch

  1. Future<bool> MesosContainerizerProcess::__launch(
  2.     const ContainerID& containerId,
  3.     const ExecutorInfo& executorInfo,
  4.     const
    string& directory,
  5.     const Option<string>& user,
  6.     const SlaveID& slaveId,
  7.     const PID<Slave>& slavePid,
  8.     bool checkpoint,
  9.     const list<Option<ContainerLaunchInfo>>& launchInfos)
  10. {
  11. ……
  12.   // Prepare environment variables for the executor.
  13.   map<string, string> environment = executorEnvironment(
  14.       executorInfo,
  15.       directory,
  16.       slaveId,
  17.       slavePid,
  18.       checkpoint,
  19.       flags);
  20.  
  21.   // Determine the root filesystem for the container. Only one
  22.   // isolator should return the container root filesystem.
  23.   Option<string> rootfs;
  24.  
  25.   // Determine the executor launch command for the container.
  26.   // At most one command can be returned from docker runtime
  27.   // isolator if a docker image is specifed.
  28.   Option<CommandInfo> executorLaunchCommand;
  29.   Option<string> workingDirectory;
  30.  
  31.   foreach (const Option<ContainerLaunchInfo>& launchInfo, launchInfos) {
  32.     if (launchInfo.isSome() && launchInfo->has_rootfs()) {
  33.       if (rootfs.isSome()) {
  34.         return Failure("Only one isolator should return the container rootfs");
  35.       } else {
  36.         rootfs = launchInfo->rootfs();
  37.       }
  38.     }
  39.  
  40.     if (launchInfo.isSome() && launchInfo->has_environment()) {
  41.       foreach (const Environment::Variable& variable,
  42.                launchInfo->environment().variables()) {
  43.         const
    string& name = variable.name();
  44.         const
    string& value = variable.value();
  45.  
  46.         if (environment.count(name)) {
  47.           VLOG(1) << "Overwriting environment variable '"
  48.                   << name << "', original: '"
  49.                   << environment[name] << "', new: '"
  50.                   << value << "', for container "
  51.                   << containerId;
  52.         }
  53.  
  54.         environment[name] = value;
  55.       }
  56.     }
  57.  
  58.     if (launchInfo.isSome() && launchInfo->has_command()) {
  59.       if (executorLaunchCommand.isSome()) {
  60.         return Failure("At most one command can be returned from isolators");
  61.       } else {
  62.         executorLaunchCommand = launchInfo->command();
  63.       }
  64.     }
  65.  
  66.     if (launchInfo.isSome() && launchInfo->has_working_directory()) {
  67.       if (workingDirectory.isSome()) {
  68.         return Failure(
  69.             "At most one working directory can be returned from isolators");
  70.       } else {
  71.         workingDirectory = launchInfo->working_directory();
  72.       }
  73.     }
  74.   }
  75.  
  76.   // TODO(jieyu): Consider moving this to 'executorEnvironment' and
  77.   // consolidating with docker containerizer.
  78.   environment["MESOS_SANDBOX"] =
  79.     rootfs.isSome() ? flags.sandbox_directory : directory;
  80.  
  81.   // Include any enviroment variables from CommandInfo.
  82.   foreach (const Environment::Variable& variable,
  83.            executorInfo.command().environment().variables()) {
  84.     environment[variable.name()] = variable.value();
  85.   }
  86.  
  87.   JSON::Array commandArray;
  88.   int namespaces = 0;
  89.   foreach (const Option<ContainerLaunchInfo>& launchInfo, launchInfos) {
  90.     if (!launchInfo.isSome()) {
  91.       continue;
  92.     }
  93.  
  94.     // Populate the list of additional commands to be run inside the container
  95.     // context.
  96.     foreach (const CommandInfo& command, launchInfo->commands()) {
  97.       commandArray.values.emplace_back(JSON::protobuf(command));
  98.     }
  99.  
  100.     // Process additional environment variables returned by isolators.
  101.     if (launchInfo->has_environment()) {
  102.       foreach (const Environment::Variable& variable,
  103.           launchInfo->environment().variables()) {
  104.         environment[variable.name()] = variable.value();
  105.       }
  106.     }
  107.  
  108.     if (launchInfo->has_namespaces()) {
  109.       namespaces |= launchInfo->namespaces();
  110.     }
  111.   }
  112.  
  113.   // TODO(jieyu): Use JSON::Array once we have generic parse support.
  114.   JSON::Object commands;
  115.   commands.values["commands"] = commandArray;
  116.  
  117.   return logger->prepare(executorInfo, directory)
  118.     .then(defer(
  119.         self(),
  120.         [=](const ContainerLogger::SubprocessInfo& subprocessInfo)
  121.           -> Future<bool> {
  122.     // Use a pipe to block the child until it's been isolated.
  123.     int pipes[2];
  124.  
  125.     // We assume this should not fail under reasonable conditions so
  126.     // we use CHECK.
  127.     CHECK(pipe(pipes) == 0);
  128.  
  129.     // Prepare the flags to pass to the launch process.
  130.     MesosContainerizerLaunch::Flags launchFlags;
  131.  
  132.     launchFlags.command = executorLaunchCommand.isSome()
  133.       ? JSON::protobuf(executorLaunchCommand.get())
  134.       : JSON::protobuf(executorInfo.command());
  135.  
  136.     launchFlags.sandbox = rootfs.isSome()
  137.       ? flags.sandbox_directory
  138.       : directory;
  139.  
  140.     // NOTE: If the executor shares the host filesystem, we should not
  141.     // allow them to 'cd' into an arbitrary directory because that'll
  142.     // create security issues.
  143.     if (rootfs.isNone() && workingDirectory.isSome()) {
  144.       LOG(WARNING) << "Ignore working directory '" << workingDirectory.get()
  145.                    << "' specified in container launch info for container "
  146.                    << containerId << " since the executor is using the "
  147.                    << "host filesystem";
  148.     } else {
  149.       launchFlags.working_directory = workingDirectory;
  150.     }
  151.  
  152.     launchFlags.pipe_read = pipes[0];
  153.     launchFlags.pipe_write = pipes[1];
  154.     launchFlags.commands = commands;
  155.  
  156.     // Fork the child using launcher.
  157.     vector<string> argv(2);
  158.     argv[0] = MESOS_CONTAINERIZER;
  159.     argv[1] = MesosContainerizerLaunch::NAME;
  160.  
  161.     Try<pid_t> forked = launcher->fork(
  162.         containerId,
  163.         path::join(flags.launcher_dir, MESOS_CONTAINERIZER),
  164.         argv,
  165.         Subprocess::FD(STDIN_FILENO),
  166.         (local ? Subprocess::FD(STDOUT_FILENO)
  167.                : Subprocess::IO(subprocessInfo.out)),
  168.         (local ? Subprocess::FD(STDERR_FILENO)
  169.                : Subprocess::IO(subprocessInfo.err)),
  170.         launchFlags,
  171.         environment,
  172.         None(),
  173.         namespaces); // 'namespaces' will be ignored by PosixLauncher.
  174.  
  175.     if (forked.isError()) {
  176.       return Failure("Failed to fork executor: " + forked.error());
  177.     }
  178.     pid_t pid = forked.get();
  179.  
  180.     // Checkpoint the executor's pid if requested.
  181.     if (checkpoint) {
  182.       const
    string& path = slave::paths::getForkedPidPath(
  183.           slave::paths::getMetaRootDir(flags.work_dir),
  184.           slaveId,
  185.           executorInfo.framework_id(),
  186.           executorInfo.executor_id(),
  187.           containerId);
  188.  
  189.       LOG(INFO) << "Checkpointing executor's forked pid " << pid
  190.                 << " to '" << path << "'";
  191.  
  192.       Try<Nothing> checkpointed =
  193.         slave::state::checkpoint(path, stringify(pid));
  194.  
  195.       if (checkpointed.isError()) {
  196.         LOG(ERROR) << "Failed to checkpoint executor's forked pid to '"
  197.                    << path << "': " << checkpointed.error();
  198.  
  199.         return Failure("Could not checkpoint executor's pid");
  200.       }
  201.     }
  202.  
  203.     // Monitor the executor's pid. We keep the future because we'll
  204.     // refer to it again during container destroy.
  205.     Future<Option<int>> status = process::reap(pid);
  206.     status.onAny(defer(self(), &Self::reaped, containerId));
  207.     containers_[containerId]->status = status;
  208.  
  209.     return isolate(containerId, pid)
  210.       .then(defer(self(),
  211.                   &Self::fetch,
  212.                   containerId,
  213.                   executorInfo.command(),
  214.                   directory,
  215.                   user,
  216.                   slaveId))
  217.       .then(defer(self(), &Self::exec, containerId, pipes[1]))
  218.       .onAny(lambda::bind(&os::close, pipes[0]))
  219.       .onAny(lambda::bind(&os::close, pipes[1]));
  220.   }));
  221. }

 

最终运行的二进制文件为const char MESOS_CONTAINERIZER[] = "mesos-containerizer";

 

Mesos-containerizer是一个独立运行的二进制文件，它的main函数在src/slave/containerizer/mesos/main.c

  1. int main(int argc, char** argv)
  2. {
  3.   return Subcommand::dispatch(
  4.       None(),
  5.       argc,
  6.       argv,
  7.       new MesosContainerizerLaunch(),
  8.       new MesosContainerizerMount());
  9. }

 

Src/slave/containerizer/mesos/launch.cpp中MesosContainerizerLaunch::execute()函数最终调用

  1. if (command.get().shell()) {
  2.   // Execute the command using shell.
  3.   execlp("sh", "sh", "-c", command.get().value().c_str(), (char*) NULL);
  4. } else {
  5.   // Use os::execvpe to launch the command.
  6.   char** argv = new
    char*[command.get().arguments().size() + 1];
  7.   for (int i = 0; i < command.get().arguments().size(); i++) {
  8.     argv[i] = strdup(command.get().arguments(i).c_str());
  9.   }
  10.   argv[command.get().arguments().size()] = NULL;
  11.  
  12.   execvp(command.get().value().c_str(), argv);
  13. }

 

来运行executor的二进制文件。

 

如果是前面叙述的TestFramework，则运行的executor是TestExecutor，也就要求mesos-slave的相应目录下有这个二进制文件。

 

Mesos源码分析(13): MesosContainerier运行一个Task

上一篇：输入每个值连续出现几次的问题（其中包括while括号中出现任意输入问题）


下一篇：iOS11适配-Safe Area