import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
public Protos.Status run() {
  LOG.info("{} starting MesosExecutorDriver...", name);

  final MesosExecutorDriver driver = new MesosExecutorDriver(singularityExecutor);

  Runtime.getRuntime().addShutdownHook(new Thread("SingularityExecutorRunnerGracefulShutdown") {

    @Override
    public void run() {
      LOG.info("Executor is shutting down, ensuring shutdown via shutdown hook");
      monitor.shutdown(Optional.of((ExecutorDriver) driver));
    }

  });

  return driver.run();
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
@SuppressFBWarnings("DM_EXIT")
public void sendStatusUpdate(ExecutorDriver driver, Protos.TaskInfo taskInfo, Protos.TaskState taskState, String message, Logger logger) {
  logger.info("Sending status update \"{}\" ({})", message, taskState.name());

  message = message.substring(0, Math.min(configuration.getMaxTaskMessageLength(), message.length()));

  try {
    final Protos.TaskStatus.Builder builder = Protos.TaskStatus.newBuilder()
        .setTaskId(taskInfo.getTaskId())
        .setState(taskState)
        .setMessage(message);

    driver.sendStatusUpdate(builder.build());
  } catch (Throwable t) {
    try {
      logger.error("Exception while sending status updates, exiting", t);
    } finally {
      System.exit(4);
    }
  }
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
private Future startExitChecker(final Optional<ExecutorDriver> driver) {
  LOG.info("Starting an exit checker that will run in {}", JavaUtils.durationFromMillis(configuration.getIdleExecutorShutdownWaitMillis()));

  return exitChecker.schedule(new Runnable() {

    @Override
    public void run() {
      LOG.info("Exit checker running...");

      try {
        checkForExit(driver);
      } catch (Throwable t) {
        logAndExit(2, "While shutting down", t);
      }
    }
  }, configuration.getIdleExecutorShutdownWaitMillis(), TimeUnit.MILLISECONDS);
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
/**
 * Invoked when a task running within this executor has been killed
 * (via SchedulerDriver::killTask). Note that no status update will
 * be sent on behalf of the executor, the executor is responsible
 * for creating a new TaskStatus (i.e., with TASK_KILLED) and
 * invoking ExecutorDriver::sendStatusUpdate.
 */
@Override
public void killTask(ExecutorDriver executorDriver, Protos.TaskID taskID) {
  final String taskId = taskID.getValue();

  LOG.info("Asked to kill task {}", taskId);

  KillState killState = monitor.requestKill(taskId);

  switch (killState) {
    case DIDNT_EXIST:
    case INCONSISTENT_STATE:
      LOG.warn("Couldn't kill task {} due to killState {}", taskId, killState);
      break;
    case DESTROYING_PROCESS:
    case INTERRUPTING_PRE_PROCESS:
    case KILLING_PROCESS:
      LOG.info("Requested kill of task {} with killState {}", taskId, killState);
      break;
  }
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
/**
 * Invoked when a task has been launched on this executor (initiated
 * via {@link org.apache.mesos.SchedulerDriver#launchTasks}. Note that this task can be
 * realized with a thread, a process, or some simple computation,
 * however, no other callbacks will be invoked on this executor
 * until this callback has returned.
 *
 * @param driver The executor driver that launched the task.
 * @param task   Describes the task that was launched.
 * @see org.apache.mesos.ExecutorDriver
 * @see org.apache.mesos.Protos.TaskInfo
 */
@Override public void launchTask(ExecutorDriver driver, Protos.TaskInfo task) {
    LOGGER.info("Launching task in PinUserBoardExecutor...");
    Protos.TaskStatus taskStatus = Protos.TaskStatus.newBuilder().setTaskId(task.getTaskId())
            .setState(Protos.TaskState.TASK_RUNNING).build();
    driver.sendStatusUpdate(taskStatus);
    String url = task.getData().toStringUtf8();

    byte[] message = new byte[0];

    try {
        message = computePi().getBytes();
    } catch (IOException e) {
        LOGGER.error("Error computing Pi :" + e.getMessage());
    }

    LOGGER.info("Sending framework message and marking task finished." + getClass().getName());
    driver.sendFrameworkMessage(message);

    taskStatus = Protos.TaskStatus.newBuilder().setTaskId(task.getTaskId()).setState(Protos.TaskState.TASK_FINISHED)
            .build();
    driver.sendStatusUpdate(taskStatus);
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
@Override
public void killTask(final ExecutorDriver driver, final TaskID taskId) {
  LOG.info("Killing task : " + taskId.getValue());
  if (taskTracker != null) {
    LOG.info("Revoking task tracker map/reduce slots");
    revokeSlots();

    // Send the TASK_FINISHED status
    new Thread("TaskFinishedUpdate") {
      @Override
      public void run() {
        driver.sendStatusUpdate(TaskStatus.newBuilder()
          .setTaskId(taskId)
          .setState(TaskState.TASK_FINISHED)
          .build());
      }
    }.start();
  }
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
@Override
public ExecutorTask createTask(Protos.TaskInfo taskInfo, ExecutorDriver driver) throws ExecutorTaskException {

    CassandraTask cassandraTask = CassandraTask.parse(taskInfo);

    switch(cassandraTask.getType()) {
        case CASSANDRA_DAEMON:
            try {
                cassandra = CassandraDaemonProcess.create(scheduledExecutorService, taskInfo, driver);
                return cassandra;
            } catch (IOException e) {
                throw new ExecutorTaskException(e);
            }
        default:
            return launchClusterTask(cassandraTask);
    }
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
private CassandraDaemonProcess(
        ScheduledExecutorService scheduledExecutorService,
        CassandraDaemonTask cassandraTask,
        ExecutorDriver executorDriver,
        Protos.TaskInfo taskInfo,
        ProcessBuilder processBuilder,
        boolean exitOnTermination) throws IOException {
    super(executorDriver, taskInfo, processBuilder, exitOnTermination);
    this.task = cassandraTask;

    this.probe = new Probe(cassandraTask);
    this.mode = new AtomicReference<>(CassandraMode.STARTING);
    scheduledExecutorService.scheduleAtFixedRate(
            ModeReporter.create(task,
                    probe,
                    executorDriver,
                    open,
                    mode),
            1, 1, TimeUnit.SECONDS);
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
public static CheckHandler create(
        ExecutorDriver executorDriver,
        Protos.TaskInfo taskInfo,
        LaunchedTask launchedTask,
        Protos.HealthCheck healthCheck,
        ScheduledExecutorService scheduledExecutorService,
        CheckStats healthCheckStats,
        String checkType)
        throws CheckValidationException {
    return new CheckHandler(
            executorDriver,
            taskInfo,
            launchedTask,
            new ProcessRunner(),
            healthCheck,
            scheduledExecutorService,
            healthCheckStats,
            checkType);
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
/**
 * Allows providing a custom {@link ProcessRunner} for testing.
 */
@VisibleForTesting
CheckHandler(
        ExecutorDriver executorDriver,
        Protos.TaskInfo taskInfo,
        LaunchedTask launchedTask,
        ProcessRunner processRunner,
        Protos.HealthCheck healthCheck,
        ScheduledExecutorService scheduledExecutorService,
        CheckStats healthCheckStats,
        String checkType)
        throws CheckValidationException {
    validate(healthCheck);
    this.healthCheck = healthCheck;
    this.scheduledExecutorService = scheduledExecutorService;
    this.healthCheckRunner = new CheckRunner(
            executorDriver,
            taskInfo,
            launchedTask,
            processRunner,
            healthCheck,
            healthCheckStats,
            checkType);
    this.checkType = checkType;
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
private CheckRunner(
        ExecutorDriver executorDriver,
        Protos.TaskInfo taskInfo,
        LaunchedTask launchedTask,
        ProcessRunner processRunner,
        Protos.HealthCheck healthCheck,
        CheckStats healthCheckStats,
        String checkType) {
    this.executorDriver = executorDriver;
    this.taskInfo = taskInfo;
    this.launchedTask = launchedTask;
    this.processRunner = processRunner;
    this.healthCheck = healthCheck;
    this.healthCheckStats = healthCheckStats;
    this.checkType = checkType;
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
private void scheduleReadinessCheck(
        ExecutorDriver executorDriver,
        Protos.TaskInfo taskInfo,
        LaunchedTask launchedTask) {

    Optional<Protos.HealthCheck> readinessCheckOptional = Optional.empty();
    try {
        readinessCheckOptional = new ExecutorTaskLabelReader(taskInfo).getReadinessCheck();
    } catch (TaskException e) {
        LOGGER.error(String.format(
                "Failed to extract readiness check from task: %s", taskInfo.getTaskId().getValue()), e);
        return;
    }

    if (!readinessCheckOptional.isPresent()) {
        return;
    }

    scheduleCheck(executorDriver, taskInfo, readinessCheckOptional.get(), launchedTask, "Readiness");
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
@Override
public void registered(final ExecutorDriver executorDriver, final Protos.ExecutorInfo executorInfo, final Protos.FrameworkInfo frameworkInfo, final Protos.SlaveInfo slaveInfo) {
    if (!executorInfo.getData().isEmpty()) {
        Map<String, String> data = SerializationUtils.deserialize(executorInfo.getData().toByteArray());
        BasicDataSource dataSource = new BasicDataSource();
        dataSource.setDriverClassName(data.get("event_trace_rdb_driver"));
        dataSource.setUrl(data.get("event_trace_rdb_url"));
        dataSource.setPassword(data.get("event_trace_rdb_password"));
        dataSource.setUsername(data.get("event_trace_rdb_username"));
        jobEventBus = new JobEventBus(new JobEventRdbConfiguration(dataSource));
    }
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
@Override
public void frameworkMessage(final ExecutorDriver executorDriver, final byte[] bytes) {
    if (null != bytes && "STOP".equals(new String(bytes))) {
        log.error("call frameworkMessage executor stopped.");
        executorDriver.stop();
    }
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
@Override
public void registered(ExecutorDriver driver, Protos.ExecutorInfo executorInfo, Protos.FrameworkInfo frameworkInfo, Protos.SlaveInfo slaveInfo) {
    final String[] bootConfig = executorInfo.getData().toStringUtf8().split(";");
    this.dcosAuthToken = nullOnEmpty(bootConfig[0]);
    this.metricsContainersEnabled = Boolean.parseBoolean(bootConfig[1]);
    slaveId = slaveInfo.getId().getValue();
    System.out.println("HumioExecutor.registered");
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
protected void sendStatusUpdate(ExecutorDriver driver, Protos.TaskInfo task, Protos.TaskState state, boolean healthy) {
    driver.sendStatusUpdate(Protos.TaskStatus.newBuilder()
        .setExecutorId(task.getExecutor().getExecutorId())
        .setTaskId(task.getTaskId())
        .setHealthy(healthy)
        .setState(state)
        .build());
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
@Override
public void frameworkMessage(ExecutorDriver driver, byte[] data) {
    System.out.println("HumioExecutor.frameworkMessage");
    final List<TaskDetails> taskDetails = SerializationUtils.deserialize(data);

    updateElasticBeatConfig(HUMIO_FILEBEAT_YAML, taskDetails);
    updateElasticBeatConfig(HUMIO_METRICBEAT_YAML, taskDetails.stream().filter(taskDetail -> taskDetail.getContainerId() != null).collect(Collectors.toList()));
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
public TestExecutorTask(Protos.TaskInfo taskInfo, ExecutorDriver driver) {
    this.taskStatus = Protos.TaskStatus.newBuilder()
            .setTaskId(Protos.TaskID.newBuilder().setValue("test-task-id"))
            .setState(Protos.TaskState.TASK_FINISHED)
            .build();
    this.driver = driver;
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
private void checkForExit(final Optional<ExecutorDriver> driver) {
  try {
    exitLock.lockInterruptibly();
  } catch (InterruptedException e) {
    LOG.warn("Interrupted acquiring exit lock", e);
    return;
  }

  boolean shuttingDown = false;

  try {
    if (tasks.isEmpty()) {
      LOG.info("Shutting down executor due to no tasks being submitted within {}", JavaUtils.durationFromMillis(configuration.getIdleExecutorShutdownWaitMillis()));
      runState = RunState.SHUTDOWN;
      shuttingDown = true;
    }
  } finally {
    exitLock.unlock();
  }

  if (shuttingDown) {
    shutdown(driver);
  } else if (runState == RunState.SHUTDOWN) {
    LOG.info("Already shutting down...");
  } else {
    LOG.info("Tasks wasn't empty, exit checker doing nothing...");
  }
}
 

import org.apache.mesos.ExecutorDriver; //导入依赖的package包/类
private void checkIdleExecutorShutdown(ExecutorDriver driver) {
  exitLock.lock();

  try {
    clearExitCheckerUnsafe();

    if (tasks.isEmpty()) {
      exitCheckerFuture = Optional.of(startExitChecker(Optional.of(driver)));
    }
  } finally {
    exitLock.unlock();
  }
}