Scaling Your Application

Let’s see what it would take to scale our HTTP server to have five instances. Ready?

curl -X PUT -H 'Content-Type: application/json' 
    marathon.example.com:8080/v2/apps --data '{"instances": 5}'

Of course, no one would ever actually want to have five instances of such a useless application, so why don’t we scale it down?

curl -X PUT -H 'Content-Type: application/json' 
    marathon.example.com:8080/v2/apps --data '{"instances": 1}'

Luckily, scaling is extremely easy!

Using Placement Constraints

Marathon supports constraints on where the application is launched. These constraints can be driven either by the hostname of the slave or any slave attribute. Constraints are provided in an array to the application; each constraint is itself an array of two or three elements, depending on whether there’s an argument. Let’s take a look at what constraints are available and how to use them:

GROUP_BY

This operator ensures that instances of the application are evenly spread across nodes with a particular attribute. You can use this to spread the application equitably across hosts (Example 3-3) or racks (assuming that the rack is encoded by the rack attribute on the slave; see Example 3-4).

Example 3-3. Spread evenly by host

{
    // ... snip ...
    "constraints": [["hostname", "GROUP_BY"]]
}

Example 3-4. Spread evenly by rack

{
    // ... snip ...
    "constraints": [["rack", "GROUP_BY"]]
}

UNIQUE

This operator ensures that every instance of the application has a different value for the UNIQUE constraint. It’s similar to GROUP_BY, except if there are not enough different values for the attribute, the application will fail to fully deploy, rather than running multiple instances on some slaves. Usually, GROUP_BY is the best choice for availability, since it’s generally preferable to have multiple tasks running on a single slave rather than fewer than the desired number. Example 3-5 shows how you could use UNIQUE to ensure that there’s no more than one instance of some application per slave in the cluster.

Example 3-5. Run at most one instance per host

{
    // ... snip ...
    "constraints": [["hostname", "UNIQUE"]]
}

CLUSTER

This operator allows you to run an application only on slaves with a specific value for an attribute. If certain slaves have special hardware configurations, or if you want to restrict placement to a specific rack, this operator can be useful; however, LIKE is more powerful and often a better fit. For example, suppose that we’ve got a mixture of ARM and x86 processors in our data center, and we’ve made a cpu_arch attribute for our slaves to help us distinguish their architectures. Example 3-6 shows how to ensure that our application only runs on x86 processors.

Example 3-6. Only run on x86

{
    // ... snip ...
    "constraints": [["cpu_arch", "CLUSTER", "x86"]] 
}

Note that the CLUSTER operator takes an argument.

LIKE

This operator ensures that the application runs only on slaves that have a certain attribute, and where the value of that attribute matches the provided regular expression. This can be used like CLUSTER, but it has more flexibility, since many values can be matched. For example, suppose that we’re running on Amazon, and we set the instance_type attribute on all of our slaves. Example 3-7 shows how we could restrict our application to run only on the biggest C4 compute-optimized machines.

Example 3-7. Only run on c4.4xlarge and c4.8xlarge machines

{
    // ... snip ...
    "constraints": [["cpu_arch", "LIKE", "c4.[48]xlarge"]] 
}

LIKE’s argument is a regular expression.

UNLIKE

UNLIKE is the complement of LIKE: it allows us to avoid running on certain slaves. Maybe you don’t want some daemons (such as ones processing user billing information) to run on machines that are in the DMZ, which is an unsecured network area. By setting the attribute dmz to true on machines in the DMZ, we can have our application avoid those machines, as in Example 3-8.

Example 3-8. Don’t run in the DMZ

{
    // ... snip ...
    "constraints": [["dmz", "UNLIKE", "true"]] 
}

UNLIKE’s argument can be a regular expression.

Placement constraints can also be combined. For example, we could ensure that our application isn’t running in the DMZ and is evenly distributed amongst the remaining machines as shown in Example 3-9.

{
    // ... snip ...
    "constraints": [["dmz", "UNLIKE", "true"],
                    ["hostname", "GROUP_BY"]]
}

Combining constraints will allow you to precisely control how your applications are placed across the cluster.

Running Dockerized Applications

Marathon has first-class support for Docker containers. If your app is already Dockerized, then it’s very easy to run it on Marathon.

In order to get started, you’ll first need to make sure that your Mesos cluster has Docker support enabled. To do this, make sure that you’ve enabled the Docker containerizer and increased the executor timeout (see “Using Docker” for details).

At this point, we can simply add the Docker configuration to the JSON that we use to configure our application. Let’s take a look at the options we have when using Docker, as illustrated in Example 3-10. Note that this JSON configuration is partial; it must also include additional options such as those in Example 3-1.

{
    // ... snip ...
    "container": {
        "type": "DOCKER", 
        "docker": {
            "image": "group/image", 
            "network": "HOST" 
        }
    }
}

Since Mesos will continue to add new container types, such as Rocket or KVM containers, much of the container configuration isn’t Docker-specific. Here, we must specify that we’ll provide Docker-specific configuration, because this is a DOCKER container.

This is the most important part: where you specify your container image.²

We’re going to run Docker in host networking mode; this means that Docker is using the slave’s networking resources directly.

Once the Docker container launches, it’ll also have access to the Mesos sandbox directory, which will be available in the environment variable $MESOS_SANDBOX.

{
    "container": {
        "type": "DOCKER",
        "docker": {
            "image": "group/image",
            "network": "HOST"
        },
        "volumes": [ 
            { 
                "containerPath": "/var/hostlib",
                "hostPath": "/usr/lib",
                "mode": "RO" 
            },
            { 
                "containerPath": "/var/scratch",
                "hostPath": "/mount/ssd",
                "mode": "RW"
            }
        ]
    }
}

Health Checks

Marathon keeps track of whether it believes an application is healthy or not. This information is exported via the REST API, making it easy to use Marathon to centralize management of its application health checks. When you query the REST API for information about a particular application, all of its tasks will include their latest health check results. By default, Marathon assumes that a task is healthy if it is in the RUNNING state. Of course, just because an application has started executing doesn’t mean that it’s actually healthy. To improve the usefulness of the application’s health data, you can specify three additional types of health checks that the tasks should undergo: command-based, HTTP, and TCP.

Example 3-12 shows the common fields of the JSON health check specification.

{
    "gracePeriodSeconds": 300, 
    "intervalSeconds": 60, 
    "maxConsecutiveFailures": 3, 
    "timeoutSeconds": 30 
}

The grace period is the amount of time that a task is given to start itself up. Health checks will not begin to run against a particular task until this many seconds have elapsed. This parameter is optional, with a default value of a 5-minute grace period.

The interval is the amount of time between runs of this health check. This parameter is optional, with a default value of 1 minute between each check.

The maximum number of consecutive failures determines how many failed health checks in a row should cause Marathon to forcibly kill the task, so that it can be restarted elsewhere. If this is set to 0, then this health check failing will never result in the task getting killed. This parameter is optional, with tasks by default being killed after three failures of this health check.

Sometimes, a health check just hangs. Rather than simply ignoring the check, Marathon will treat the slow or hung health check as a failure if it doesn’t complete in this many seconds. This parameter is optional, and by default health checks have 20 seconds to run.

We’ll look at the two of the three types of health checks here: HTTP and TCP. Command health checks are still relatively new, and due to some of their limitations, I recommend waiting a bit longer for them to stabilize. Bear in mind, though, that HTTP and TCP health checks here are currently performed by the Marathon master, which means that they’re not scalable to thousands of tasks.³

{
    "protocol": "HTTP", 
    "path": "/healthcheck", 
    "portIndex": 0 
}

{
    "protocol": "TCP", 
    "portIndex": 0 
}

Application Versioning and Rolling Upgrades

What use would a distributed, scalable PaaS have if it couldn’t do rolling upgrades? An upgrade is any time that you modify the application, which is typically done by making a PUT to the /v2/apps/$appid route on Marathon. Every upgrade creates a new version, which is named with the timestamp of when the upgrade was made. When you query the REST API for information about the application, every task will say which version it is running. You can query which versions are available from the /v2/apps/$appid/versions route on Marathon, and you can find details of the versions from the /v2/apps/$appid/versions/$version route. Example 3-15 demonstrates how to configure the upgrade behavior of an application.

{
    "upgradeStrategy": {
        "minimumHealthCapacity": 0.5,
        "maximumOverCapacity": 0.1
    }
}

With Marathon, you can easily specify how much of the application you want to be available during an upgrade via the "minimumHealthCapacity" field of the application descriptor. The "minimumHealthCapacity" can be set to any value between 0 and 1.

If you set this to 0, when you deploy a new version Marathon will first kill all the old tasks, then start up new ones.

If you set this to 0.6, when you deploy a new version Marathon will first kill 40% of the old tasks, then start up 60% of the new tasks, then kill the rest of the old tasks and start the rest of the new tasks.

If you set this to 1, when you deploy a new version Marathon will first start up all the new tasks (at which point your application could use twice as many resources as during regular operations), and then shut down the old tasks.

The "maximumOverCapacity" setting provides an additional level of safety, so that an application doesn’t consume too many resources temporarily during an upgrade. The "maximumOverCapacity" can be set to any value between 0 and 1.

If you set this to 0, when you deploy a new version Marathon will not start more tasks than the target number of instances.

If you set this to 0.6, when you deploy a new version Marathon will never exceed a combined total of 160% of the target number of tasks.

If you set this to 1, when you deploy a new version Marathon will be allowed to start every new task before it kills any of the old tasks, if it so chooses.

The Event Bus

Often, when building applications with Marathon, you’ll want to receive notifications when various events happen, such as when an application gets a new deployment, or when it scales up or down. These notifications could then be used to reconfigure proxies and routers, log statistics, and generate reports. Marathon has a built-in feature for automatically POSTing all of its internal events to a provided URI. To use this feature, simply provide two additional command-line arguments to Marathon:

--event_subscriber

This argument should be passed the value http_callback in order to enable this subsystem. At this time, no other options are supported.

--http_endpoints

This argument is a comma-separated list of destination URIs to send the JSON-formatted events to. For example, a valid argument would be http://host1/foo,http://host2/bar.

Setting Up HAProxy with Marathon

What a great day! We’ve learned all about Marathon and decided that it would make an excellent platform for our applications. Now that our applications are running on Marathon, however, we’ve reached an impasse: how can the outside world actually connect to them? Some types of applications, such as Celery or Resque workers, need no additional configuration: they communicate via a shared database. Other types of applications, such as HTTP servers and Redis instances, need to be made easily discoverable. The most popular way to do this on Mesos clusters is to run proxies on static, non-Mesos managed hosts, and automatically update those proxies to point at running instances of the application. This way, the application is always available on a known host and port (i.e., the proxy), but we can still dynamically scale its capacity on the backend. Also, each proxy host can typically serve many applications, each on a different port, so that a small number of proxy hosts are sufficient for hundreds of backends. Typically, proxies are constrained either by the total bandwidth used by all active connections, or by the total CPU usage if they’re providing SSL termination.⁴

There are two proxies that are far and away the most popular choices for use in modern application stacks: HAProxy and Nginx. HAProxy is a proxy with very few features: it can proxy HTTP and TCP connections, perform basic health checks, and terminate SSL. It is built for stability and performance: there have been no reported crashes or deadlock bugs in HAProxy for 13 years. This is why HAProxy is popular amongst Mesos users today.

Nginx, by comparison, has no limits on its features. Besides acting as a proxy, it is able to run custom user code written in Lua, JavaScript, or JVM-based languages. This code can affect the proxy operations, or even serve responses directly. Many systems actually use both proxies: clients connect to HAProxy instances, which themselves forward the requests to Nginx instances. The Nginx instances then either respond directly, or proxy the requests to backend servers to be processed.

In this section, we’re only going to discuss HAProxy, but be aware that Nginx can be used in exactly the same way, if you require the additional functionality it offers.

Let’s take a look at the most popular configurations for setting up proxies with Mesos.

What Is Chronos?

Chronos is a Mesos framework that provides a highly available, distributed time-based job scheduler, like cron. With Chronos, you can launch jobs that are either command-line programs (optionally downloaded from URIs) or Docker containers. Like Marathon, Chronos has a web UI and an easily programmable REST API. Although the Chronos endpoints have a slightly different interface than Marathon’s, we won’t go over their details, as the online documentation is sufficient. Chronos has four key features for scheduling jobs:

Interval scheduling

Chronos uses the standard ISO 8601 notation to specify repeating intervals. Through the web UI or REST API, you can specify the frequency at which you want the job to rerun. This makes Chronos suitable for running quick jobs every few minutes, or a nightly data generation job.

Dependency tracking

In theory, Chronos supports dependent jobs. Dependent jobs run only if their prerequisites ran successfully. Unfortunately, Chronos’s implementation of dependent jobs misses the mark. In Chronos, a job’s dependencies are considered satisfied if the last invocation of that job was successful. A dependent job is run as soon as all of its dependencies have been run at least once. This means that unless every parent has the exact same interval, the dependent jobs can run at potentially unexpected times. In “Chronos Operational Concerns”, we’ll look at a popular mitigation strategy for this.

Deferred start

Chronos can also be used to run a job once or on a repeating interval at a point in the future. When you specify the ISO 8601 interval on which to repeat the job, you also specify when you’d like the first run of the job to start. This means that you can use Chronos to schedule jobs to run days, weeks, or months in the future, which can be useful for planning purposes.

Worker scalability

The key differentiator between Chronos and most other job schedulers is that Chronos runs its jobs inside of containers on Mesos. This means that jobs are isolated and workers are scalable, so no matter what load Chronos generates, you’re able to add new Mesos slaves to handle the additional work and ensure that the jobs receive their necessary resources.

Like all production-ready Mesos frameworks, Chronos has high-availability support with hot spares—most Chronos deployments will run at least two instances of the scheduler, to ensure that Chronos never has more than a couple seconds of downtime if a scheduler or its host fails. Chronos also comes with a synchronization tool, chronos-sync.rb. This tool synchronizes the Chronos state to disk in JSON format. Using chronos-sync.rb, you can version the jobs you run on your cluster by committing the file to version control. This is the most popular approach to ensure repeatability, since it allows you to keep your Chronos configuration in a repository.

The combination of its REST API and synchronize-to-disk tool make Chronos easy to integrate. Since its features are somewhat limited, it’s easy to use, but it’s not a complete solution for dependent workflow management.

Running Chronos on Marathon

One of the most common techniques seen in Mesos framework deployment is to run framework schedulers on Marathon. The reason for this is that those schedulers can rely on Marathon to ensure they’re always running somewhere. This removes a huge burden from the operator, who would otherwise need to ensure that every scheduler had a well-monitored, multimachine deployment.

In order to be deployable on Marathon, a scheduler must do some sort of leader election. As we’ll learn in Chapter 4, a scheduler connects to Mesos from a single host. This means that if we’d like to run our scheduler in two locations (e.g., to have a hot spare), we’ll need some sort of signaling to ensure that only one scheduler runs at a time. The most common way to solve this is to use ZooKeeper, since it has an easily integrated leader election component (see “Adding High Availability” for details). Marathon and Chronos both take this approach.

Let’s look at an example JSON expression for launching Chronos on Marathon (Example 3-16). In order to maximize the efficiency of our port allocations, we’ll allow Chronos to bind to any port. To connect to Chronos, we’ll either click through the Marathon UI or use one of the HAProxy service discovery mechanisms described earlier.

{
    "cmd": "java -jar chronos.jar -Xmx350m -Xms350m ---port $PORT0", 
    "uris": [
        "http://fetchable/uri/with/chronos.jar" 
    ],
    "cpus": 1, 
    "mem": 384, 
    "instances": 2, 
    "id": "chronos",
    "ports": [0], 
    "constraints": [["hostname", "UNIQUE"]] 
}

The simplest way to verify that Chronos is running is to click on the assigned port number from the Marathon app view. Each task’s port assignment will be a link to that host and port combination, which will load the Chronos UI. The Chronos UI is somewhat buggy, so it’s recommended to use curl or chronos-sync.rb to configure your instance of Chronos.

Chronos Operational Concerns

Chronos is a useful part of the Mesos ecosystem; however, care must be taken not to overly rely on it. There are a few things that one must be aware of when running Chronos in production:

1. The dependent jobs feature isn’t useful—you must use a workflow manager.

2. Chronos is still a complex distributed system—it’s only necessary if you need jobs to run scalably in containers.

Almost no one uses the Chronos dependent jobs feature. Instead, they use Chronos to schedule invocations of popular data workflow tools. For simple workflows (such as running a few processes in order), Bash scripts are a popular option; however, beyond 20 lines, Bash scripts become unwieldy, and their syntax for complex loops and conditionals leaves much to be desired. make is another popular way to express dependencies, since it has superior syntax for workflows and supports basic parallelism. Luigi is a much richer and more powerful workflow manager. It supports checkpoints in HDFS and databases, which can improve the efficiency of retries of jobs that failed partway through, since it won’t need to redo earlier checkpointed work. Regardless of your particular needs and use case, you should use some other tool to order the jobs that Chronos will be scheduling.

If you’ve made it this far, you’re almost ready to use Chronos in production! The final consideration is whether you truly need the scalable isolation that Chronos gets from Mesos. Even though you’re running a Mesos cluster, you don’t need to use Chronos for your task management. There are many other job-scheduling servers that could easily be run on Marathon; only use Chronos if you need each job to run in its own container.

Chronos on Marathon: Summary

Chronos is a powerful tool for Mesos clusters that need reliable, scalable, interval-based job scheduling. It powers the data generation processes at many companies, including Airbnb. Just remember the several quirks, workarounds, and advice discussed here!

Chronos on Marathon is just one example of how Marathon can be used to provide high availability and ease of deployment to other Mesos schedulers. DCOS, a commercial Mesos product, takes this further and requires all frameworks on DCOS to be hosted by Marathon. Whether you’re deploying Chronos, Kafka, or Cotton, Marathon is an excellent choice for hosting other Mesos frameworks’ schedulers.

Singularity

Singularity is a Mesos framework developed by HubSpot. It offers the ability to deploy services just like Marathon (including Dockerized applications); beyond this, if a service begins to fail its health checks after a deployment, Singularity will automatically roll back the deployment. Singularity also supports repeating jobs, like Chronos, as well as one-shot jobs, which can be triggered via a REST API. This simplifies adding containerized, asynchronous background processing to any service or repeating job. For managing service discovery and proxies, Singularity is integrated with another tool called Baragon, which manages and integrates HAProxy, Nginx, and Amazon’s Elastic Load Balancer.

Aurora

Aurora is the Mesos framework that powers Twitter. Aurora has a highly advanced job description API written in Python. This API, called Thermos, allows users to specify how to build and install applications, as well as the workflow and sequence of processes to run within the Mesos executor. Aurora comes with a high-performance distributed service discovery API with bindings for C++, Java, and Python. It’s able to run repeating jobs with cron-style syntax. Also, just like Singularity, Aurora can automatically detect when a service’s health checks start failing and roll it back to the previous working version. Aurora’s most powerful features are related to its multitenancy controls: Aurora can allow for some tasks to use spare cluster capacity, and then preempt those tasks when higher-priority production tasks need to be scheduled. It also can enforce quotas between users. These features were born out of the needs of a company running huge numbers of applications written by hundreds of developers on massive Mesos clusters; if that sounds like the kind of organization you’re building a Mesos cluster for, Aurora’s worth a closer look.