Amazon Story

Decentralized Scaling and Monitoring

The second set of log data is sent to the server status domain (table). The data in this domain is crucial for the operation of the system, in particular the scaling algorithm and health monitoring of the servers.

#!/usr/bin/php
<?php
// bootstrap & initialization
require_once '../application/init.php';
// daemonize
Util_Daemonize::getInstance()->daemonize();
// instantiate EC2 object
$ec2 = new Util_AmazonEC2Instances($opts['aws']['access_key'],
                                   $opts['aws']['secret_access_key']);
// instantiate request queue object
$requestQueue = new Util_AmazonSqsQueue($opts['aws']['request_queue_name'],
                                        $opts['aws']['access_key'],
                                        $opts['aws']['secret_access_key']);
// get current timestamp
$lastServerStatusScrub = time();
// instantiate Amazon SimpleDB object
$sdb = new SimpleDB($opts['aws']['access_key'], $opts['aws']['secret_access_key']);
// instantiate Amazon scaler object
$scaler = new Util_AmazonInstancesScaler($requestQueue,
                                         $ec2,
                                         $sdb);
// loop indefinitely
while (true) {
    // let everybody else know we're still alive and kickin'
    $scaler->sendImAlive($opts['simpleDB']['mpt_server_status'], 'auto-scaler');
    // refresh instances data
    $ec2->init();
    // is the current (local) instance in charge?
    if ($scaler->amIinCharge()) {
        // init number of instances that need to be launched / terminated
        $instancesDelta = $scaler->getInstancesDelta();
        // apply min. & max. instances caps
        $actualInstancesDelta = $scaler->applyInstanceLimits($instancesDelta);
        // launch new instances
        if ($actualInstancesDelta > 0) {
            $result = $ec2->run(array('minCount'      => $actualInstancesDelta,
                                                              'maxCount'      => $actualInstancesDelta,
                                                              'instanceType'  => $opts['ec2']['instanceType'])
                                                              );
        // instances don't necessarily get shut down right away
        // since we're paying in full hour increments, there's no
        // need to terminate an instance until the hour is nearly up
        } elseif ($instancesDelta < 0) {
            // see whether any instances are close to having run a multiple of a full hour
            $agingInstances = $ec2->getAgingInstances(3600);
            // loop over aging instances
            for ($i = 0; ($i < -$instancesDelta) && isset($agingInstances[$i]); $i++) {
                // terminate instances
                $ec2->terminate($agingInstances[$i]['instanceId']);
            }
        }
        // now let's check on the health of all previously running servers
        foreach ($ec2->getInstances() as $instance) {
            // machine is not well ... not terminated ... and not just launched
            if (!$scaler->isInstanceHealthy($instance)
                && !in_array($instance['instanceId'], $terminatedInstances)
                && !$scaler->isInstanceNew($instance)
            ) {
                // reboot unhealthy instance
                $ec2->reboot($instance['instanceId']);
            }
        }
        // is it time to scrub the server status table?
        if (time() - $lastServerStatusScrub > $opts['simpleDB']['seconds_between_server_status_deletions']) {
            $delCount = $scaler->deleteAgedServerStatus($opts['simpleDB']['mpt_server_status'],
                                                                                                                $opts['simpleDB']['max_server_status_age']);
        }
    }
    // go to sleep for a bit
    sleep($opts['ec2']['autoscaling']['seconds_between_scaling']);
}
?>

AWS offers its own monitoring and scaling service, which supports metrics such as CPU, RAM, or disk usage. Unfortunately, the metric driving the scaling algorithm is the number of items in the request queue, which is not directly supported by AWS's Auto Scaling service. Besides, I liked the challenge of trying to come up with a decentralized approach to monitoring and scaling so as to avoid a single point of failure. After all, that is the same benefit that cloud-based providers highlight when proclaiming their offerings' high availability and scalability.

On startup, each server runs a process that is capable of using the centrally logged status data to scale the number of instances up or down. However, at any given time, only one machine's process is actually in charge and thus allowed to make such changes. Listing 4 contains a simplified version of the PHP script responsible for scaling on each server.

Line 5 of Listing 4 requires the file init.php, which bootstraps the application by initializing resources and loading configuration files. A call to Util_Daemonize on line 8 then turns the process into a daemon that can run in the background without timing out and without being tied to a parent process. Lines 11, 14, 20, and 23 then instantiate classes that wrap the Zend Framework classes interacting with AWS. This approach wraps most of the business logic while still relying on the framework to do the heavy lifting in terms of communicating with AWS.

Util_AmazonEC2Instances has methods to discover, launch, shutdown, and reboot Amazon virtual machine instances. Util_AmazonSqsQueue knows how to add and retrieve messages from Amazon's SQS. SimpleDB provides access to the NoSQL storage for logging and status.

Finally, Util_AmazonInstancesScaler contains the algorithm to determine how many instance are needed at any given time. Util_AmazonInstancesScaler also has methods to log status information to the NoSQL domain.

Line 26 starts the infinite loop that comprises the core of what this script does. To begin, the script sends a message to the SimpleDB mpt_server_status domain to let it know that the current process, auto-scaler, is currently alive and running (line 29). The script then tells the Util_AmazonEC2Instances object to retrieve from AWS the list of currently running instances. The init() method also identifies the current instance in the list of all running instances.

On line 35, the script decides whether the current instance of the script and server is the one in charge of scaling. If not, nothing remains to be done, and the script goes to sleep for a predefined period before repeating the cycle.

The criteria for determining which instance is in charge is fairly simple. The amIinCharge() method identifies the oldest instance that is considered healthy as the one in charge. Healthy in this context means that all processes on the instance have been reporting to the NoSQL domain that they are running without a problem.

Line 38 contains a call to method getInstancesDelta(), which encapsulates the scaling algorithm. It returns a signed integer that indicates a change to the total number of instances. A +5 means that five new instances need to be launched; whereas a -2 means that two currently running instances need to shut down.

The algorithm itself is pretty straightforward: Given the number of items in the request queue, how many instances must be added or removed to bring the average processing time below the maximum? The initial implementation assumes each video clip takes three minutes to process. For example, assume there are currently four instances and 35 requests in the queue. Furthermore, assume each request takes three minutes to process, and the longest you want a client to wait for the result is 10 minutes. Without changing the number of instances, the average wait time for the requests in the queue is [4 x (3 + 6 + 9 + 12 + 15 + 18 + 21 + 24) + (3 x 27)]/35 = 14.66 minutes. That is to say, the first four requests take three minutes each, the second set of four requests takes six minutes each, and so on. Add three more machines for a total of seven, and the average processing time drops to [7 x (3 + 6 + 9 + 12 + 15)]/35 = 9 minutes. Therefore, method getInstancesDelta() returns +3. You can refer to the box titled "Managing Instances" for more details on how Listing 4 manages the inventory of instances. Of course, there is great room for improvement. For starters, you should be able to use the activity log to get a better estimate of how long it takes to process each request. Another enhancement reserved for a subsequent release is to use cyclical patterns to anticipate changes in demand. For example, if we determine that traffic doubles around 8am when people get to work, we might want to launch additional instances shortly before 8am to anticipate that demand.

Summary

By building a system that wraps a single binary, this company was able to create a service that can handle tens of thousands of requests per day. What would have been very expensive to do on purchased hardware occurred fairly easily using cloud-based services.

Future plans for this project include support for additional types of processing; a different input/output format, such as JSON; and a desktop-based processing client.