Chapter 4. Understanding the Scheduler API

In this chapter, you will learn about the frameworks and how they interact with Mesos. To do this, we will develop a simple framework.

The following are the recipes covered in this chapter:

• Installing Protobuf
• Registering frameworks
• Handling events
• Declining offers
• Scheduling tasks
• Acknowledging task updates
• Killing tasks
• State persistence
• Reconciliation

Introduction

The following recipes will guide you through creating a simple Mesos framework. It's important that this framework is prepared to be as simple as possible, just to demonstrate how frameworks interact with Mesos. However, the price of simplicity is robustness. It's not designed to handle production traffic but could be used as a starting point for creating real-world solutions.

Our framework will be created in Go. Go is getting more and more popular in system infrastructure, mostly because it gives C/C++ system performance with syntax and memory management known from higher-level languages such as Python. Another point in its favor is a powerful standard library. Our framework will require only one external dependency, Protobuf: https://github.com/golang/protobuf.

In our implementation, we will use the Mesos HTTP API. It's worth mentioning that this API is by design not RESTful, but still pretty easy to use.

Before you start, install Go on your machine and create a new directory...

Installing Protobuf

In this recipe, we will be installing Protobuf to generate Golang structures used in communication with Mesos.

Getting ready

Install Golang, Protobuf, and the Golang Protobuf bindings:

sudo add-apt-repository ppa:ubuntu-lxc/lxd-stable
sudo apt-get update
sudo apt-get install golang git
export GOPATH=~/go
export PATH=$PATH:$GOPATH/bin
sudo apt-get install protobuf-compiler
go get -u github.com/golang/protobuf/{proto,protoc-gen-go}

How to do it...

1. Create a project directory and go there:

mkdir -p $GOPATH/src/simple-scheduler
cd $GOPATH/src/simple-scheduler

2. Download the Mesos protobuf message definitions:

wget https://raw.githubusercontent.com/apache/mesos/170ac/include/mesos/v1/mesos.proto
wget https://raw.githubusercontent.com/apache/mesos/170ac/include/mesos/v1/scheduler/scheduler.proto

3. Tweak them to our needs. Change the default package of generated bindings to one that matches Golang. We will keep the whole project in one main package and the generated bindings should fit in...

Registering frameworks

In this recipe, we will learn how frameworks register in Mesos to receive offers and state updates.

How to do it...

We will create the scheduler.go file and implement our framework inside of it.

Before we start, we need to define some globals and imports that we will need later:

import (
        "bufio"
        "bytes"
        "log"
        "net/http"
        "os"
        "strconv"
        "strings"
        "github.com/golang/protobuf/jsonpb"
)
// Url to Mesos master scheduler API
const schedulerApiUrl = "http://10.10.10.10:5050/api/v1/scheduler"
// Current framework configuration
var frameworkInfo FrameworkInfo
// Marshaler to serialize Protobuf Message to JSON
var marshaller = jsonpb.Marshaler{
        EnumsAsInts: false,
        Indent: " ",
        OrigName: true,
}

jsonpb.Marshaler is a part of the Golang Protobuf binding. It's responsible for converting structs into JSON. We will use it to serialize the messages we send to Mesos. It's important to use a proper Marshaler...

Handling events

In this recipe, we will learn how Mesos notifies frameworks about updates.

How to do it...

To handle events, we need to parse data that we obtain from Mesos. To do that, add the following code at the end of the loop in the subscribe() function:

var event Event
sonpb.UnmarshalString(data, &event)
log.Printf("Got: [%s]", event.String())

After parsing, we need to handle events that can vary in type. To do this, we will use the switch control statement on the event type:

switch *event.Type {
case Event_SUBSCRIBED:
        log.Print("Subscribed")
        frameworkInfo.Id = event.Subscribed.FrameworkId
        mesosStreamID = res.Header.Get("Mesos-Stream-Id")
case Event_HEARTBEAT:
        log.Print("PING")
}

Remember to declare the global mesosStreamID variable:

var mesosStreamID string

How it works...

Events come with persistent HTTP connection one-by-one. Protobuf binding comes with the method to parse the data into Golang structs. Then in our event loop, we can perform specific actions...

Declining offers

In this recipe, we will handle offers by declining them so they can be presented to other frameworks.

How to do it...

So far, our framework has been acquiring resources, although it's not using them. This is because it gets offers but does not respond to them. To make offers free, we need to explicitly refuse them.

To refuse Offers, we need to handle the Offers event. We can do this by adding the following code in the event type switch:

case Event_OFFERS:
         log.Printf("Handle offers returns: %v", handleOffers(event.Offers))

Of course, we need to implement the handleOffers() function. But before that, we should prepare the call() function, which will be useful for implementation. This function will be used to send messages to Mesos:

func call(message *Call) error {
        message.FrameworkId = frameworkInfo.Id
        body, _ := marshaller.MarshalToString(message)
        req, _ := http.NewRequest("POST", schedulerApiUrl, bytes.NewBuffer([]byte(body)))
        req.Header...

Scheduling tasks

In this recipe, we will ask Mesos to run a task.

How to do it...

First of all, we need to know what task we want to schedule. To allow communication with the framework, we will prepare the HTTP API.

To handle HTTP requests, we need to implement the HTTP handler and bind it to some ports.

Let's declare the following variables in the main() function:

listen := ":9090"
webuiURL := fmt.Sprintf("http://%s%s", hostname, listen)

Let's also declare one global variable:

var taskID uint64

Tasks can be launched only when offers are available. To communicate with the offer handler, we will use a channel. If you are not familiar with Golang channels, they're similar to a message queue or pipe. Basically, you write at one end and read at another. Let's declare it globally:

var commandChan = make(chan string, 100)

Extend the FrameworkInfo definition with the just-created web UI URL:

frameworkInfo = FrameworkInfo{
        User: &user,
        Name: &name,
        Hostname: &hostname,
...

Acknowledging task updates

In this recipe, we will learn how Mesos communicates the task's state changes and how to handle these changes.

How to do it...

Right now, we should be able to spawn new tasks on a Mesos cluster but when we deploy tasks with an invalid command, such as false:

curl -X POST "http://localhost:9090/?cmd=false"

This task immediately ends with an exit code equal to 1. In the Mesos UI, we can see these tasks have failed but it's not presented in completed tasks. This is because each task state update needs to be acknowledged by the framework.

In the subscribe event switch, add the following case:

case Event_UPDATE:
     log.Printf("Handle update returns: %v", handleUpdate(event.Update))

Then implement the handleUpdate function:

func handleUpdate(update *Event_Update) error {
      return call(&Call{
             Type: Call_ACKNOWLEDGE.Enum(),
             Acknowledge: &Call_Acknowledge{
                   AgentId: update.Status.AgentId,
                   TaskId: update...

Killing tasks

In this recipe, we will ask Mesos to shut down a task.

How to do it...

Killing tasks is similar to launching tasks with one difference: tasks can be killed at any time.

Let's add support for the delete method in the web function:

case "DELETE":
     id := r.Form["id"][0]
     err := kill(id)
     if err != nil {
             fmt.Fprint(w, err)
     } else {
             fmt.Print(w, "KILLED")
     }

We need to implement the kill function. It'll be similar to acknowledge because it will only prepare and send a message to Mesos. To kill tasks, we need to know the task's ID. We will get it from the user and agent ID on which the task is launched. We need to modify our framework to keep track of the launched tasks. We will keep that in the global map, where task ID will be the key and task last update will be the value:

var tasksState = make(map[string]*TaskStatus)

This map will be updated every time we get an update. So in the hadleUpdate() function, we need to add the following:

tasksState...

State persistence

In this recipe, we will learn how to persist framework state between restarts.

How to do it...

Every time we restart our framework, it starts from scratch, losing all information about the tasks it's scheduled. After the framework goes down, Mesos kills all its tasks. This behavior is not acceptable when we want to upgrade the framework without restarting its tasks. To change this, we must do two things: tell Mesos to keep tasks after framework communication fails, and keep framework state between restarts.

In the main() function, declare the variable holding the framework failover seconds and checkpointing flag:

failoverTimeout := float64(3600)
checkpoint := true

We will use them in the framework info declaration.

Then we need to store the framework info and task state. The framework info is changed only after the subscribe method. After subscription, it's worthwhile saving, since it contains a framework that is required to reconnect.

Declare a global variable with a path to...

Reconciliation

In this recipe, we will learn how to reconcile the state between the framework and Mesos.

How to do it...

Although our tasks persist after the framework restarts, they cannot be managed because the framework does not know about them. To make it aware of tasks it had scheduled previously, we need to keep them between restarts. We will do it in the same way as framework info. We will store the task state after every update and load it after registration.

Declare the global variable with the path to task state file:

var stateFile = fmt.Sprintf("%s/%s", os.TempDir(), "state.json")

In the handle update function, just after updating the task state, save it to the file:

stateJSON, _ := json.Marshal(tasksState)
ioutil.WriteFile(stateFile, stateJSON, 0644)

We will load the previous state in the dedicated function and that should be called after the frameworkId is stored in the framework info:

func reconcile() {
      oldState, err := ioutil.ReadFile(stateFile)
      if err == nil {
      ...