The Executor

We’ve already learned what a scheduler is: it’s the component that interacts with the Mesos masters and the framework’s clients, manages the running tasks, and handles failovers. But what is an executor? An executor has three responsibilities:

• Executing tasks as requested by the scheduler

• Keeping the scheduler informed of the status of those tasks

• Handling other requests from the scheduler

Rewriting the CommandExecutor

Now that we’ve seen some examples of when we’d want to write an executor, let’s take a look at how to actually do so. We’re going to start by writing something very simple: an executor that is compatible with the built-in CommandExecutor.

We’ll start by looking at the skeleton of MyExecutor, our CommandExecutor clone. The imports and class declaration are presented in Example 5-1.

package com.example;
import org.apache.mesos.*;
import org.apache.mesos.Protos.*;
import java.util.*;
import java.io.File;
import org.json.*; 
import java.lang.ProcessBuilder.Redirect; 

public class MyExecutor implements Executor, Runnable { 
    // We'll discuss the inner bits as we go
}

We’ll use JSON to encode the task we want to run. This JSON will be read from the TaskInfo so that we know what to invoke.

The ProcessBuilder is the most convenient way to start a process in Java; we’ll use it to launch our application, and we’ll redirect the application’s stdout and stderr to separate log files from our executor.

Since there’s no callback API for finding out when a process finishes, we’ll start a thread to wait for the process—that’s why we implement Runnable.

The main method is shown in Example 5-2.

public static void main(String ... args) throws Exception {
    Executor executor = new MyExecutor();
    ExecutorDriver driver = new MesosExecutorDriver(executor);
    driver.run();
}

Although it’s tempting to try to configure some things via the command line passed in to main, you’ll be thankful that all executor configuration comes from the Mesos APIs—this will simplify development in the long run by ensuring all configuration comes from a single source. When the scheduler has total control over the executors, this simplifies development: global (scheduler) changes and local (executor) changes are now managed from a single codebase, the scheduler’s. This makes developing new features easier, and it simplifies operations, since the administrators need only to modify and reload the scheduler to make framework-wide changes.

As with the scheduler, in the main of our executor we create a corresponding driver to communicate with Mesos. Unfortunately, this driver turns out to be a major annoyance. When the framework calls MesosExecutorDriver.start(), it needs to have been started by a slave. The slave injects an environment variable, MESOS_SLAVE_PID, which is used to allow the executor to connect to the slave that started it and begin participating in the Mesos protocols. As a result, if you want to test your executor, you need to implement a mock ExecutorDriver. If you forget to do this, you’ll see a weird stack trace like this, regardless of what language you’re developing your executor in:

F0605 21:32:01.538770 18480 os.hpp:173] Expecting 'MESOS_SLAVE_PID' in
environment variables
*** Check failure stack trace: ***
    @     0x7fc535912dfd  google::LogMessage::Fail()
    @     0x7fc535914c3d  google::LogMessage::SendToLog()
    @     0x7fc5359129ec  google::LogMessage::Flush()
    @     0x7fc535915539  google::LogMessageFatal::~LogMessageFatal()
    @     0x7fc5352c9ff0  os::getenv()
    @     0x7fc53534139b  mesos::MesosExecutorDriver::start()
    @     0x7fc5359068ee  Java_org_apache_mesos_MesosExecutorDriver_start
    @     0x7fc5390127f8  (unknown)

We’ll block the main thread until our driver completes—when the task is finished, we’ll call driver.stop() elsewhere to ensure the executor exits correctly. Since executors decouple the notion of tasks from the containers and processes that run them, it’s easy to forget that sending a TASK_FINISHED doesn’t terminate the executor. You should always have executors shut themselves down once they’ve finished their work, or have the scheduler explicitly manage their lifetime by killing their canary tasks (see “Canary tasks”).

Just like for schedulers, many of the executor callbacks aren’t important to handle. Example 5-3 lists the callbacks we can ignore for now.

    public void frameworkMessage(ExecutorDriver driver, byte[] data) { } 
    public void registered(ExecutorDriver driver, ExecutorInfo executorInfo, 
                           FrameworkInfo frameworkInfo, SlaveInfo slaveInfo) {
        System.out.println("registered executor " + executorInfo);
    }
    public void disconnected(ExecutorDriver driver) { } 
    public void shutdown(ExecutorDriver driver) { } 
    public void reregistered(ExecutorDriver driver, SlaveInfo slaveInfo) { } 
    public void error(ExecutorDriver driver, java.lang.String message) { } 

Framework messages are a simple (but not guaranteed, as we’ll see in the upcoming sidebar) way to communicate between the scheduler and its executors by sending arbitrary data serialized as bytes. Note that framework messages aren’t routed to particular tasks.

This is invoked the first time that an executor connects to the slave. The most common thing to do is get data from the ExecutorInfo, since that can carry executor configuration information in its data field, which contains arbitrary bytes.

This is invoked when the slave disconnects from the executor, which typically indicates a slave restart. Rarely should an executor need to do anything special here.

This callback informs the executor to gracefully shut down. It is called when a slave restart fails to complete within the grace period, or when the executor’s framework completes. The executor will be forcibly killed if shutdown doesn’t complete within 5 seconds (the default, configurable on the slave command line with --executor_shutdown_grace_period).

This callback is invoked after a successful slave restart; it contains the new slave’s information.

This callback is invoked after a fatal error occurs. When this callback is invoked, the driver is no longer running.

As you can see, the executor can actually be a full participant in the slave checkpointing and recovery system; however, you can ignore this for almost all executors. We’ll make use of some of these callbacks later, but for now, let’s continue by looking at how to actually launch a task.

We’ll start by looking at what data we’ll need to access throughout the executor’s implementation. These are global variables that are defined in the MyExecutor class (see Example 5-1):

Process proc

We’ll need to store a reference to the process we launched so that we can wait for it to complete.

TaskID taskId

We also need to communicate the task’s status to Mesos: the driver API to communicate with our scheduler requires the TaskID as an argument.

ExecutorDriver driver

Of course, we’ll want to be able to use the driver, so we’ll store a reference to it as well.

Well-behaved executors should keep Mesos up-to-date with the status of their tasks. Since sending status updates has some boilerplate, we’ll write a convenience method, shown in Example 5-4.

private void statusUpdate(TaskState state) {
    TaskStatus status = TaskStatus.newBuilder()
        .setTaskId(taskId)
        .setState(state)
        .build()
    driver.sendStatusUpdate(status);
}

Here, we simply construct a bare-bones task status and send it via the driver. Sometimes, you’ll want to communicate additional information. Like most Mesos protobufs, TaskStatus supports arbitrary data, as well as a human-readable message.

In order to keep this example simple, our executor will only support running a single task, after which it will exit. We’ll enforce this invariant with a helper function, as seen in Example 5-5.

private boolean ensureOneLaunch(ExecutorDriver driver, TaskID id) { 
    if (this.taskId != null) { 
        TaskStatus status = TaskStatus.newBuilder()
            .setTaskId(id) 
            .setState(TaskState.TASK_ERROR)
            .setMessage("this executor only can run a single task") 
            .build();
        driver.sendStatusUpdate(status);
        return false;
    } else {
        return true;
    }
}

We’ll return true if this is the first task, and false if it should be ignored.

The code that launches our task will set the value of this.taskId. Thus, we can assume that if this.taskId was already set, we’ve already launched a task, so this one is invalid.

Note that we use the TaskID of the newly requested task, not the running task.

We’ll take advantage of the human-readable message in the status update to help the user diagnose why the task launch resulted in an error.

Example 5-6 illustrates how we actually launch a process. We’ll take the configuration JSON object as an input, and we’ll return the newly started process. To help separate our logs, we’ll redirect the task process’s stdout and stderr to different files than the executor’s stdout and stderr. For now, the task JSON will have only a single key: "cmd", which is the command line to run the task. We’ll invoke the task with a shell for simplicity.

private Process startProcess(JSONObject cfg) throws Exception{
    List<String> cmdArgs = Arrays.asList("bash", "-c", cfg.getString("cmd")); 
    ProcessBuilder pb = new ProcessBuilder(cmdArgs); 
    File stdoutFile = new File(System.getenv("MESOS_DIRECTORY"), "child_stdout"); 
    File stderrFile = new File(System.getenv("MESOS_DIRECTORY"), "child_stderr"); 
    pb.redirectOutput(Redirect.to(stdoutFile)); 
    pb.redirectError(Redirect.to(stderrFile)); 
    return pb.start();
}

At this point, we’re finally ready to handle launching a task. To do so, we must implement the launchTask callback of the Executor interface (see Example 5-7).

public void launchTask(ExecutorDriver driver, TaskInfo task) {
    synchronized (this) { 
        try {
            if (!ensureOneLaunch(driver, task.getTaskId())) { 
                return;
            }

            this.taskId = task.getTaskId(); 
            this.driver = driver; 

            statusUpdate(TaskState.TASK_STARTING); 

            byte[] taskData = task.getData().toByteArray(); 
            JSONObject cfg = new JSONObject(new String(taskData, "UTF-8"));
            this.proc = startProcess(cfg);  

            statusUpdate(TaskState.TASK_RUNNING); 

            Thread t = new Thread(this); 
            t.setDaemon(true);
            t.start();
        } catch (Exception e) {
            e.printStackTrace(); 
        }
    }
}

At this point, we nearly know the entirety of how our executor goes about managing the lifecycle of its task. Example 5-8 illustrates how we track the completion of the task.

public void run() {
    int exitCode;
    try {
        exitCode = proc.waitFor(); 
    } catch (Exception e) {
        exitCode = -99; 
    }
    synchronized (this) { 
        if (proc == null) { 
            return;
        }
        proc = null; 
        if (exitCode == 0) { 
            statusUpdate(TaskState.TASK_FINISHED);
        } else { 
            driver.sendStatusUpdate(TaskStatus.newBuilder()
                    .setTaskId(taskId)
                    .setState(TaskState.TASK_FAILED)
                    .setMessage("Process exited with code " + exitCode)
                    .build());
        }
    }
    driver.stop(); 
}

waitFor() will block until the process terminates, and it returns the process’s exit code.

If waitFor() throws an exception for some reason, we’ll assume something went terribly wrong and use a distinctive exit code to signal that.

As mentioned before, the mutex on this protects access to the global variables.

If we reach this point and proc is null, this is a special signal that means that process was killed by killTask (which we haven’t seen yet). In that case, our work’s already done.

By setting proc to null, we set that special signal from above.

When the process exits with a successful status, we report that the task finished (remember, FINISHED is a successful status in Mesos).

Otherwise, we know that the task failed, and so we send a FAILED status.

We stop() the executor driver at this point, so that our main function can return and the executor will gracefully shut down.

We’ve seen how our new executor handles accepting a task, starting a process, and sending the appropriate status updates when that process finishes. All we have left is to handle requests from the scheduler to kill the task early. Example 5-9 shows how to handle a killTask message.

public void killTask(ExecutorDriver driver, TaskID taskId) {
    synchronized (this) { 
        if (proc != null 
                && taskId.equals(this.taskId)) { 
            proc.destroy(); 
            statusUpdate(TaskState.TASK_KILLED); 
            proc = null;
        }
        driver.stop();
    }
}

We have now seen the entirety of how to implement a basic executor with functionality similar to the CommandExecutor. Of course, you might be wondering how to actually use this executor from your scheduler. To integrate this executor with our example scheduler from Chapter 4, we’ll simply need to update the makeTask method of the Job object, as shown in Example 5-10.

public TaskInfo makeTask(SlaveID targetSlave, FrameworkID fid) {
    TaskID id = TaskID.newBuilder()
        .setValue(this.id)
        .build();
    ExecutorID eid = ExecutorID.newBuilder() 
        .setValue(this.id)
        .build();
    CommandInfo ci = CommandInfo.newBuilder()
        .setValue("java -jar /path/to/custom/executor.jar") 
        .build();
    ExecutorInfo executor = ExecutorInfo.newBuilder() 
        .setExecutorId(eid)
        .setFrameworkId(fid)
        .setCommand(ci)
        .build();
    JSONObject cfg = new JSONObject();
    try {
        cfg.put("cmd", this.command);
        return TaskInfo.newBuilder()
            // ... Elided unchanged code ... 
            .setExecutor(executor) 
            .setData(ByteString.copyFrom(cfg.toString().getBytes("UTF-8"))) 
            .build();
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException();
    }
}

We need to explicitly assign our executor an ID. The CommandExecutor uses the same ID for the task and executor, so we’ll do that too.

You’ll need to build the custom executor’s .jar and distribute it to your slaves. In the next section, we’ll look at some approaches to this.

The ExecutorInfo requires several pieces of information, including the FrameworkID. The scheduler was trivially modified to pass it to makeTask.

The unchanged configuration to the TaskInfo is not repeated here (refer back to Example 4-3). We did, however, drop the .setCommand(...), since now we’re setting the executor.

TaskInfo must have exactly one of the command or executor set, but never both—that’s an error.

Finally, we include the JSON configuration of the task.

That’s all there is to it! Now we’ve enhanced our scheduler from Chapter 4 to support our new, custom executor.

Bootstrapping Executor Installation

One challenge in creating your own executor is deploying the binaries everywhere. We’ll look at four solutions to this problem, and weigh their pros and cons:

Use HDFS, S3, or another non-POSIX store

Any Mesos executor can list a set of URIs that will be downloaded and optionally unzipped into the local working directory before the executor is launched. Mesos has built-in support for HTTP, FTP, and HDFS stores to download files (see “Configuring the process’s environment”). As a result, you can easily host your applications on a local HDFS cluster, Amazon S3, or another data store to avoid the need to deploy any new technologies, as the other strategies might require. The challenge is that this data store must be sufficiently scalable to be able to deliver the container to every concurrently launching task. A common reason that a newly developed framework falls over during scalability testing is that it launches hundreds of executors simultaneously; these executors all simultaneously request their binaries from a single server, which kills the server. This is commonly known as the thundering herd problem. Make sure that you sufficiently replicate the server that offers the binaries, or use a system with proven scalability, like S3.

The “next-level” version of this approach is to actually host the binaries in the scheduler process itself, by having the scheduler run an embedded HTTP server. Of course, this will exacerbate the thundering herd problem. The solution is to have the scheduler stagger the launch times of tasks, to ensure that the server is never overwhelmed. This is done by waiting for a random duration before submitting the launch to Mesos.

Use a configuration management system

Many Mesos clusters will be deployed using a configuration management system like Chef, Puppet, or Ansible. If this applies to your cluster, then you can also use that system to push your executor binaries out to every slave in your Mesos cluster. The benefit to this is that you can ensure that the binaries are always immediately available to launch tasks—you don’t need to wait for them to download or run the risk of an untimely network failure that causes your tasks to become LOST. The downside, however, is that deploying and upgrading to a new version of the executor is more difficult. Of course, you’ll want to version every deployed executor (to ensure that multiple versions can be running concurrently, during upgrades); nevertheless, it can be extremely tedious during development to wait for the executor to be pushed out to all slaves.

Use a shared POSIX filesystem

Often, a compute cluster already has a shared POSIX-compatible filesystem. Some popular choices here are NFS and GlusterFS. The shared filesystem approach is similar to the configuration management approach: in both cases, your executor exists at a known path. The benefit of having that path be on a shared filesystem is that you don’t need to wait for the configuration management system to actually push the binaries out to every slave; instead, you can immediately launch the executor as soon as it has been copied onto the shared filesystem. The downside is that it can be difficult to scale the shared filesystem, and if it fails, your entire cluster goes down.

Use Docker

Mesos has first-class support for Docker containers (see “Using Docker”), which are themselves a way to ship around complete binary images. Once you Dockerize your executor, you can start that container on any slave in the cluster. The challenge here is the Docker repository: you must find a Docker repository (be it Docker Hub or a self-hosted one) that can host your Docker containers with sufficient scalability and security. The most common pitfall with this approach is the time to download a Docker container: when a Docker container takes 10 minutes to download, the executor launch might time out. This has the symptom of unpredictable, transient LOST tasks.

Now that we’ve reviewed several options, how do you choose the right one? If you are already heavily using Docker, then I recommend continuing with Docker to host your Mesos executors. Otherwise, if you have a shared POSIX filesystem, that option offers the fastest and easiest development. If you would rather avoid the challenge of setting up a shared POSIX filesystem for this purpose, then serving the binaries from HDFS or S3 still offers rapid deployment, and nearly everyone has this capability. Using a configuration management system does centralize deployment, but it dramatically slows down your ability to redeploy a newer or older version of an executor, and so it is an option of last resort.

In any case, remember that your scheduler specifies exactly from where and how to fetch and launch the executor. Consequently, each version of a scheduler can choose to launch different forwards- and/or backwards-compatible versions of the executor. By ensuring that your scheduler always launches a compatible executor, you can generally avoid internal version mismatches in your frameworks.

Adding Heartbeats

One of the advantages of the Mesos architecture is that temporary communication disruptions don’t affect the operation of the running executors. We often leverage this functionality when designing for high availability, since each executor can continue to process its work or service client requests in spite of other failures. When building a pool of servers scheduler, for example (see “Pool of Servers Scheduler”), this is exactly the behavior we want. On the other hand, sometimes we want to ensure that every executor is running so that we can react quickly to failures—after all, there’s no way to distinguish a temporary and permanent network blip.²

We’ll ensure that every executor is functioning by requiring them to heartbeat every so often (the period will depend on how quickly we want to react to failures). To heartbeat, we’ll use framework messages, which have several interesting features and caveats:

• Framework messages are unreliable: there’s no guarantee they’re ever delivered.

• Framework messages are scalable: they usually use an optimized transmission mechanism directly from the executor to the scheduler, thus avoiding bottlenecking at the master.

• Framework messages can carry an arbitrary binary payload: you can serialize any data you’d like into them, such as progress updates.

• Framework messages are between executors and schedulers: it’s up to you to include the task ID in the binary payload if you want to send messages about a particular task on an executor.

First, let’s add heartbeats to our executor, as shown in Example 5-11.

public static void main(String ... args) {
    // ... snip ...
    Timer timer = new Timer(); 
    timer.schedule(new TimerTask() {
        public void run() {
            driver.sendFrameworkMessage(new byte[1]); 
        }
    }, 0, 5000); 
    // ... snip ...
}

We create a new Timer when our executor starts up to manage the heartbeats.

The task we create will simply send a framework message without any interesting data. In a production framework, you could include useful information here, such as when the message was sent (to track the message delivery delay) or what the completion percentage of each of the executor’s tasks is.

We start this task immediately, sending a heartbeat every 5 seconds. To increase scalability, heartbeats are often sent every 30 seconds or every few minutes.

Although it’s easy to send heartbeats on the executor, we also need to augment the scheduler. First, we enhance the Job to automatically fail itself when it doesn’t receive a heartbeat in time, as seen in Example 5-12.

public class Job {
    private Timer timer = new Timer(); 
    private TimerTask missedHeartbeatTask; 
    // ... snip ...

    public void heartbeat() { 
        if (missedHeartbeatTask != null) { 
            missedHeartbeatTask.cancel();
        }
        missedHeartbeatTask = new HeartbeatTask();
        timer.schedule(missedHeartbeatTask, 20000); 
    }

    public static Job fromJSON(JSONObject obj) {
        Job job = new Job();
        // ... snip ...
        job.heartbeat(); 
        return job;
    }

    public void succeed() {
        // ... snip ...
        missedHeartbeatTask.cancel(); 
    }

    public void fail() {
        // ... snip ...
        missedHeartbeatTask.cancel();
    }

    private class HeartbeatTask extends TimerTask { 
        public void run() {
            System.out.println("Heartbeat missed; failing");
            fail(); 
        }
    }
}

Each Job will now have a Timer to manage the expiration of its latest heartbeat.

We must store the TimerTask that will cause the current instance of the Job to fail, so that we can cancel it when we receive a heartbeat.

The heartbeat() method will get called every time we receive a heartbeat from the executor.

Every time we receive a heartbeat, we must cancel the current task, since if that task ran we’d fail the Job.

When we receive the heartbeat, we schedule the next failure due to lack of heartbeat in 20 seconds. Since we expect a heartbeat every 5 seconds, this should only run if we miss several heartbeats in a row.

When we first create the Job, we heartbeat it once to ensure that it fails if the executor never starts running.

When the task finishes, we cancel the pending failure task, since it’s no longer relevant.

A HeartbeatTask is a subclass of TimerTask. We only need to implement the run() method, which is only called if the task isn’t canceled before its time is up.

If our task does get to run, we just fail() the Job, which will cause the Job to get rescheduled later.

Now that our Job has been enhanced to track heartbeats, all we need to do is pass along framework messages from the scheduler to the proper Job. To do this, we’ll implement the frameworkMessage callback in the scheduler, as shown in Example 5-13.

public void frameworkMessage(SchedulerDriver driver, ExecutorID executorId,
                             SlaveID slaveId, byte[] data) {
    String id = executorId.getValue(); 
    synchronized (jobs) {
        for (Job j : jobs) { 
            if (j.getId().equals(id)) {
                j.heartbeat(); 
            }
        }
    }
}

For the example executor in this chapter (and for the CommandExecutor), we set the executor ID to be equal to the job ID and the task ID.

Since we don’t care about the data in the framework message, we simply scan through all the current jobs to find if one of them has the same ID as the executor that just sent the heartbeat.

Once we find a match, we call the heartbeat() method on that Job, so that the Example 5-12 code can do its thing.

This is all it takes to add basic heartbeating functionality to your framework. Of course, there are several enhancements that should be made in a production-quality system. Firstly, the heartbeat mechanism shown here creates one timer for every Job’s heartbeat; for a system with hundreds of Jobs, every Job can share the same Timer to ensure scalability since each Timer creates a new thread. Secondly, this heartbeat system doesn’t actually kill the non-heartbeating Jobs; it just treats them as dead. We should always kill the executors that aren’t heartbeating by using the killTask API, or else we could end up with orphaned or zombie executors, which will consume resources indefinitely on the cluster without doing any useful work. Finally, when we’re not sure if an executor was properly killed, we should be careful not to start an executor with a duplicate ID on a slave. We should actually track all the executor IDs we’ve used, and generate a new, unique ID for each executor. With the preceding code, you can encounter undefined behavior when you try to relaunch a Job on the same slave when the earlier executor is still running but believed dead due to lack of heartbeats.

Now, we’ve seen how to build a simple executor by leveraging the task-level and executor-level APIs. We’ve even added an additional functionality, heartbeats, that wasn’t possible with the built-in executor. In the next few sections, we’ll look at other advanced features that we can add to our executors.

Advanced Executor Features

As you’ve probably discovered by now, it’s difficult to write an executor. Unlike most software that we write, the executor is particularly challenging to test, since we need to build it, package it, and ship it across our cluster, at which point we must wait to see whether it correctly interacts with our scheduler.

Before we start looking at additional features we can add to our executor, keep in mind that an executor doesn’t have to make its task a subprocess. In fact, it’s quite common to have tasks run in the same process as the executor—this can lead to greater efficiency and a simpler design. The only reason we split our executor and task into different processes was simply so that we could imitate the behavior of the built-in CommandExecutor in Mesos.

Summary

Executors are the Mesos abstraction for a framework’s workers. They should be used when a framework requires richer interaction between its scheduler and workers. Building an executor is usually very challenging, because of the need to frequently deploy new builds to the cluster during development. By surmounting these challenges, you can build custom executors to enable frameworks to dynamically scale the containers in which their tasks are running, to efficiently share expensive resources, and to provide an easy API to send messages between the scheduler and workers.

In this chapter, after reviewing some of the reasons one might build an executor, we implemented an executor that was mostly compatible with the built-in Mesos CommandExecutor and which we integrated with the job scheduler from Chapter 4. Then, we added heartbeats, which allowed us to more rapidly detect certain types of partial failures on the cluster, resulting in a better experience for the humans that must wait for their jobs to complete.

Finally, we looked at some advanced uses of custom executors: how to choose between framework messages and status updates, considerations when integrating with centralized distributed logging systems, and how and why to develop executors that support multiple tasks.

Now that we’ve looked across the entire spectrum of building applications on Mesos—utilizing existing frameworks, building application-specific schedulers, creating rich executors—we’ll move on to more advanced topics, such as Mesos internals, Docker integration, and exotic, cutting-edge APIs.