Method TaskStatusUpdateStream

src/slave/task_status_update_manager.cpp:651–707

TaskStatusUpdateStream::TaskStatusUpdateStream(
    const TaskID& _taskId,
    const FrameworkID& _frameworkId,
    const SlaveID& _slaveId,
    const Flags& _flags,
    bool _checkpoint,
    const Option<ExecutorID>& executorId,
    const Option<ContainerID>& containerId)
    : checkpoint(_checkpoint),
      terminated(false),
      taskId(_taskId),
      frameworkId(_frameworkId),
      slaveId(_slaveId),
      flags(_flags),
      error(None())
{
  if (checkpoint) {
    CHECK_SOME(executorId);
    CHECK_SOME(containerId);

    path = paths::getTaskUpdatesPath(
        paths::getMetaRootDir(flags.work_dir),
        slaveId,
        frameworkId,
        executorId.get(),
        containerId.get(),
        taskId);

    // Create the base updates directory, if it doesn't exist.
    const string& dirName = Path(path.get()).dirname();
    Try<Nothing> directory = os::mkdir(dirName);
    if (directory.isError()) {
      error = "Failed to create '" + dirName + "': " + directory.error();
      return;
    }

    // Open the updates file.
    // NOTE: We don't use `O_SYNC` here because we only read this file
    // if the host did not crash. `os::write` success implies the kernel
    // will have flushed our data to the page cache. This is sufficient
    // for the recovery scenarios we use this data for.
    Try<int_fd> result = os::open(
        path.get(),
        O_CREAT | O_WRONLY | O_APPEND | O_CLOEXEC,
        S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);

    if (result.isError()) {
      error = "Failed to open '" + path.get() +
              "' for status updates: " + result.error();
      return;
    }

    // We keep the file open through the lifetime of the task, because it
    // makes it easy to append status update records to the file.
    fd = result.get();
  }
}