SimpleCloud Part 2 – The Job Manager

In the previous installment I talked a little about the cloud, what Zend is doing in the cloud and what the example application for my ZPCAP webinar did.  One of the primary characteristics of scalability is the ability to process data as resources are available.  To do that I implemented the Zend Server Job Queue with an abstraction layer that I’ve written about three different versions for.  I think the fourth will be the charm :-).

The Zend Server Job Queue works by making an HTTP call to a server which will execute a PHP script.  That HTTP request is the “job” which is going to be executed.  The job is simply the Job Queue daemon pretending to be a browser.  While that works pretty well I prefer a mechanism that is more structured than simply running an arbitrary script.  Having small, defined, structured tasks allow you to spread those jobs over many servers quite easily.

So what I did was write a management system that is relatively simple which allows me to define those tasks and execute them on pretty much any server that is behind a load balancer.  And on the cloud, that load balancer can have a thousand machines behind it AND it can be reconfigured without changing your application.  One of the keys of elastic scalability is that you can throw an application “out there” and it will “work”.  That is why the Zend Server Job Queue is a good idea in the cloud.  Because it uses a protocol that requires one entry point to be defined and the rest is up to the infrastructure to work out.  (I personally am of the opinion that PHP developers are too dependent on config files).

There are two parts to this manager.  1) the queueing mechanism, 2) the executing mechanism.  Both are handled in the same class, named comzendjobqueueManager.  When a job is executed, it does not execute, it sends a request to the load balancer using a REST-like API.  The Job Queueing mechanism, by default, manages the queue on the local host.  I wanted the job server to manage its own queue.  This REST-like API will send the request to the load balancer, which sends it to a host.  In that REST-like call is contained the serialized object of the job that needs to be executed, along with an dependent data/references to data.  That host then queues the job on itself and then returns a serialized PHP object that provides the host name and the job number.  This result object can then be attached to a session so you can directly query the job queue server on subsequent requests.

The code for the manager is as follows.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

namespace com\zend\jobqueue;
 
class Manager
{
 
    const CONFIG_NAME    = 'JobQueueConfig';
 
    public function sendJobQueueRequest(JobAbstract $job)
    {
        $url = Zend_Registry::get(self::CONFIG_NAME)->queueurl . '?' . http_build_query(array('name' => base64_encode(get_class($job))));
        $http = new Zend_Http_Client($url);
        $http->setMethod('POST');
        $http->setRawData(base64_encode(serialize($job)));
        $body = $http->request()->getBody();
        $response = unserialize(base64_decode($body));
        if (!$response instanceof Response) {
            throw new Exception('Unable to get a properly formatted response from the server');
        }
        return $response;
    }
 
    public function getCompletedJob(Response $res)
    {
        $jq = new ZendJobQueue($res->getServerName());
        $job = $jq->getJobStatus($res->getJobNumber());
        $status = $job['status'];
        if ($status == ZendJobQueue::STATUS_OK) {
            $output = Zend_Http_Response::fromString($job['output']);
            $response = unserialize(base64_decode(trim($output->getBody())));
            return $response;
        }
    }
 
    public function executeJob()
    {
        $params = ZendJobQueue::getCurrentJobParams();
        if (isset($params['obj'])) {
            $obj = unserialize(base64_decode($params['obj']));
            if ($obj instanceof JobAbstract) {
                try {
                    $obj->run();
                    echo base64_encode(serialize($obj));
                    ZendJobQueue::setCurrentJobStatus(ZendJobQueue::OK);
                    exit;
                } catch (Exception $e) {
                    zend_monitor_set_aggregation_hint(get_class($obj) . ': ' . $e->getMessage());
                    zend_monitor_custom_event('Failed Job', $e->getMessage());
                    echo base64_encode(serialize($e));
                }
            }
        }
        ZendJobQueue::setCurrentJobStatus(ZendJobQueue::FAILED);
    }
 
    public function createJob($name)
    {
        $q = new ZendJobQueue();
        $qOptions = array('name' => base64_decode($name));
 
        $num = $q->createHttpJob(
            Zend_Registry::get(self::CONFIG_NAME)->executeurl,
            array(
                'obj' => file_get_contents('php://input')
            ),
            $qOptions
        );
 
        $response = new Response();
        $response->setJobNumber($num);
        $response->setServerName(php_uname('n'));
        echo base64_encode(serialize($response));
    }
}

Sequence of Events

sendJobQueueRequest() is the first to be called.  The job is passed via a parameter and is subsequently serialized.  A connection is made to the URL, which is stored in a Zend_Config object.  That URL can be a local host name or the load balancer’s host name.  Using this you can also set up different pools of servers quite easily simply by creating multiple load balancers and have each pool managed based off of its individual resource needs.

sendJobQueueRequest() called on the front end will cause createJob() to be called on the back end.  This queues the job locally by specifying a LOCAL URL that will be responsible for executing the job and creates a response object which contains the unique hostname of the machine and the unique job number on that machine.  It is serialized and echoed.  sendJobQueueRequest() then reads the response and unserializes it into a Response object which can be attached to a session.

This is the code on the backend URL that will be executed to queue the job.

1
2
3
4
5

use com\zend\jobqueue\Manager;
require_once '../bootstrap.php';
 
$q = new Manager();
$q->createJob($_GET['name']);

Don’t worry about the bootstrap.php yet.  It simply contains some configuration mechanisms and instantiates the SimpleCloud adapters.  We’ll cover that later.

This is the code for the response object (created in createJob()). The front end machine can call getCompletedJob() and pass the response object to check and see if the job is done.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

namespace com\zend\jobqueue;
 
class Response
{
 
    private $_jobNumber;
    private $_serverName;
 
    public function getJobNumber()
    {
        return $this->_jobNumber;
    }
 
    public function getServerName()
    {
        return $this->_serverName;
    }
 
    public function setJobNumber($num)
    {
        $this->_jobNumber = $num;
    }
 
    public function setServerName($name)
    {
        $this->_serverName = $name;
    }
 
}

At some point in the future, as resources are available, the URL, noted by Zend_Registry::get(self::CONFIG_NAME)->executeurl in createJob() will be executed.  The code of that URL is

1
2
3
4
5

use com\zend\jobqueue\Manager;
require_once '../bootstrap.php';
 
$q = new Manager();
$q->executeJob();

Pretty simple, eh?  That’s because most of the magic happens in the Manager class.  This is when executeJob() is called.  It takes that serialized object, unserializes it, and executes the run() method.  We will look at the difference between execute() and run() in a subsequent post.  If the job executes fine, the job is re-serialized and echoed.  If there is an exception thrown, THAT is serialized.

That’s the manager.  Next we will look at the abstract job class and after that we will get into the SimpleCloud components.