Resque is a great library for creating background jobs. Usually these jobs are fire and forget tasks that may take considerable computation time, or may contain expensive operations. But what if you need to monitor these processes after they are queued? With a bit of looking under the hood, it is quite easy to query what these workers are up to.
Scheduled Tasks
Take this example class:
# Example Resque job class from https://github.com/defunkt/resque
class Archive
@queue = :file_serve
def self.perform(repo_id, branch = 'master')
repo = Repository.find(repo_id)
repo.create_archive(branch)
end
end
We can call Resque.enqueue(Archive, 1, "branch") and this will queue up this job to run inside of Redis. Resque uses Redis on the back end to store its queues, and they are implemented using Lists. The name for the list is determined by the @queue value defined in the Resque job class. Querying the Redis key of this name shows us the jobs in this queue.
The Resque gem conveniently provides access to the Redis CLI tools via the redis class method. This method returns a connected Redis session, and can be chained with methods the Redis gem supports. Since this is a List type, we will need to query this key using the Redis command lindex. The scheduled_jobs method below will return a Ruby Array of each element in the Redis List:
def scheduled_jobs
payloads = []
index = 0; while (payload = Resque.redis.lindex("queue:#{@queue}", index)) do
payloads << payload
end
payloads
end
MultiJson.decode can be called to convert the serialized payload back to a Ruby object. Once the payload is decoded, you can inspect the values of the payload object. An example decoded payload may look like the following:
{"class":"Archive","args":[1,"master"]}
The Hash key "args" contains the arguments as they were passed to Resque.enqueue. "class" can be used to determine the specific job class if the same queue is used for multiple jobs.
Currently Running Tasks
Resque’s workers look at the Redis List to determine what the next job is. When it finds a job to be done, it removes the element from the Redis List, and beings work. This means that once a job has begun, it can no longer be queried by looking at the Redis List elements. We need to look at the Resque workers.
The Resque gem provides a Resque::Worker namespace that contains the methods specific to workers. Specifically, Resque::Worker.working returns an Array of workers with currently running jobs. Each Worker instance in the Array responds to job, which contains useful information including the queue, and the payload for the job:
worker.job # => {"queue"=>"file_serve", "run_at"=>"2012/09/28 15:09:35 EDT", "payload"=>{"class"=>"Archive", "args"=>[1, 'master']}}
We can return any running job’s payloads using this method:
def currently_running
Resque::Worker.working.map(&:job).map(&:payload)
end
Summary
To get a comprehensive picture of jobs (both queued and running) we can combine these two different query methods:
def jobs
scheduled_jobs + currently_running
end
With knowledge of what jobs are queued, and what jobs are currently running, we can take specific actions in an application. Perhaps you want to disallow a user from scheduling multiple jobs of the same type? Or perhaps you want to enforce a simple dependency system in which certain jobs will only run after another job completes? An example may be postponing a Solr reindexing until after a job that extensively modifies your collection finishes. You could even enforce mutual exclusion of jobs, such as importing users into a system, and checking for duplicate users.
Moving jobs to be performed at a time in the future can be accomplished using the Resque Scheduler gem. An example of mutual exclusion may be checking for the other job inside the perform method of each Resque job class. If the other job is found to be scheduled or running, you can enqueue the job to run at a point in the future using the enqueue_in method.
Now that is better than firing and forgetting!
—Ben Simpson