Internals Reference

Servers

class carbontube.servers.Pipeline(name, concurrency=10, backend_class=<class 'carbontube.storage.inmemory.EphemeralStorageBackend'>)

Pipeline server class

A pipeline must be defined only after you already at least one Phase.

handle_finished_job(job)

called when a job just finished processing.

When overriding this method make sure to call super() first

initialize()

Initializes the backend.

Subclasses can overload this in order to define their own backends.

on_finished(event)

called when a job just finished processing. You can override this at will

on_started(event)

called when a job just started processing.

This method is ok to be overriden by subclasses in order to take action appropriate action.

Clients

class carbontube.clients.PipelineClient(address, hwm=10)

Pipeline client

Has the ability to push jobs to a pipeline server

connect()

connects to the server

enqueue_job(data)

pushes a job to the pipeline.

Note that the data must be a dictionary with the following

keys:

• name - the pipeline name
• instructions - a dictionary with instructions for the first phase to execute

Parameters:data – the dictionary with the formatted payload. Returns:the payload sent to the server, which contains the job id

EXAMPLE:

>>> from carbontube.clients import PipelineClient

>>> properly_formatted = {
...     "name": "example1",
...     "instructions": {
...          "size": 100",
...     },
... }
>>> client = PipelineClient('tcp://127.0.0.1:5050')
>>> client.connect()
>>> ok, payload_sent = client.enqueue_job(properly_formatted)

Storage Backends

class carbontube.storage.BaseStorageBackend(name, *args, **kw)

base class for storage backends

connect()

this method is called by the pipeline once it started to listen on zmq sockets, so this is also an appropriate time to implement your own connection to a database in a backend subclass pass

consume_job_of_type(job_type)

dequeues a job for the given type. must return None when no job is ready.

Make sure to requeue this job in case it could not be fed into an immediate worker.

enqueue_job(job)

adds the job to its appropriate queue name

get_next_available_worker_for_type(job_type)

randomly picks a workers that is currently available

initialize()

backend-specific constructor. This method must be overriden by subclasses in order to setup database connections and such

register_worker(worker)

register the worker as available. must return a boolean. True if the worker was successfully registered, False otherwise

unregister_worker(worker)

unregisters the worker completely, removing it from the roster

class carbontube.storage.EphemeralStorageBackend(name, *args, **kw)

in-memory storage backend. It dies with the process and has no option for persistence whatsoever. Used only for testing purposes.

connect()

this method is called by the pipeline once it started to listen on zmq sockets, so this is also an appropriate time to implement your own connection to a database in a backend subclass pass

consume_job_of_type(job_type)

dequeues a job for the given type. must return None when no job is ready.

Make sure to requeue this job in case it could not be fed into an immediate worker.

enqueue_job(job)

adds the job to its appropriate queue name

get_next_available_worker_for_type(job_type)

randomly picks a workers that is currently available

initialize()

backend-specific constructor. This method must be overriden by subclasses in order to setup database connections and such

register_worker(worker)

register the worker as available. must return a boolean. True if the worker was successfully registered, False otherwise

unregister_worker(worker)

unregisters the worker completely, removing it from the roster

class carbontube.storage.RedisStorageBackend(name, *args, **kw)

Redis Storage Backend

connect()

this method is called by the pipeline once it started to listen on zmq sockets, so this is also an appropriate time to implement your own connection to a database in a backend subclass pass

enqueue_job(job, state)

adds the job to its appropriate queue name

get_next_available_worker_for_type(job_type)

randomly picks a workers that is currently available

initialize(redis_uri='redis://', worker_availability_timeout=300)

backend-specific constructor. This method must be overriden by subclasses in order to setup database connections and such

register_worker(worker)

register the worker as available. must return a boolean. True if the worker was successfully registered, False otherwise

unregister_worker(worker)

unregisters the worker completely, removing it from the roster

class carbontube.storage.RedisJobStorage(name, *args, **kw)

connect()

this method is called by the pipeline once it started to listen on zmq sockets, so this is also an appropriate time to implement your own connection to a database in a backend subclass pass

initialize(redis_uri='redis://', worker_availability_timeout=300)

backend-specific constructor. This method must be overriden by subclasses in order to setup database connections and such

Utilities

class carbontube.util.CompressedPickle(*args, **kw)

Serializes to and from zlib compressed pickle

pack(item)

Must receive a python object and return a safe primitive (dict, list, int, string, etc).

unpack(item)

must receive a string and return a python object

carbontube.util.parse_port(address)

parses the port from a zmq tcp address

Parameters:address – the string of address Returns:an int or None

carbontube.util.read_internal_file(path)

reads an internal file, mostly used for loading lua scripts

carbontube.util.sanitize_name(name)

ensures that a job type or pipeline name are safe for storage and handling.

Parameters:name – the string Returns:a safe string