Fun with librados (python)

I have recently started to test the python bindings for librados and I’d like to share what I have learned so far.

To begin, you will need a ceph cluster, an easy way to get one going is to use ansible-ceph. I will not get into much detail about ansible-ceph in this post, but it is very straight forward to get it running, just follow the directions on the github README page. For the code on this post, there’s no need to add RGW or MDS nodes. I changed the default box from Ubuntu to CentOS 7.1. When SSHing to node mon0, the rados python module is already installed, so you should be able to run the code examples from there.

Before running I/O operations, let’s cover some cluster operations:

import rados

cluster = rados.Rados(conffile='/etc/ceph/ceph.conf')
cluster.connect()
cluster.create_pool('test')
print cluster.list_pools()
cluster.shutdown()

For any client application, lines 3 and 4 are always required to make a connection to the ceph cluster. When finished, make sure to call shutdown(). Lines 5 and 6 are examples of cluster level APIs that are available (see the librados API for a complete list).

Write and read whole objects

To perform I/O the librados API provides a few different options, let’s try to cover them in detail here. First you can write “full” objects, meaning, if you have the complete data, you can make one call to write the whole object at once:

import rados

cluster = rados.Rados(conffile='/etc/ceph/ceph.conf')
cluster.connect()
ioctx = cluster.open_ioctx('test')

# write full object at once
ioctx.write_full('obj', 'Hello World!')
obj = ioctx.read('obj')
assert(obj == 'Hello World!')

Note in line 5, there’s a call to open an input/output context. This call will open a connection to a specific pool. Once you have the I/O context, you can start writing and reading objects to/from that pool.

To write a full object, call write_full(). The first argument is the key or object name (e.g., 'obj'), and the second the data itself. To read the object, just call read() and provide the name of the object you want to read. If you want to read chunks of the object, you can provide length and offset arguments.

A new call to write_full() will completely overwrite the object:

ioctx.write_full('obj', 'Oi Mundo!')
obj = ioctx.read('obj')
assert(obj == 'Oi Mundo!')

Stream Data

If your application needs to stream data to the ceph cluster, you can use the write() method to write chunks of data. See below:

# stream data
ioctx.write('stream_obj', 'Data is ', 0)
ioctx.write('stream_obj', 'being streamed', 8)
ioctx.write('stream_obj', ' in', 22)

stream_obj = ioctx.read('stream_obj')
assert(stream_obj == 'Data is being streamed in')

But, now you need to be careful when trying to overwrite an object. A new call to write() will only overwrite the specific length at the specific offset provided:

# overwrite object
ioctx.write('stream_obj', 'New object', 0)
stream_obj = ioctx.read('stream_obj')
assert(stream_obj == 'New objecting streamed in')

A better way to overwrite an object is to truncate it first, which will resize the object to the new object length:

# better way to overwrite object
ioctx.trunc('stream_obj', 10)
ioctx.write('stream_obj', 'New', 0)
ioctx.write('stream_obj', ' object', 3)
stream_obj = ioctx.read('stream_obj')
assert(stream_obj == 'New object')

Async I/O

Librados also provides async I/O APIs. Async APIs are a bit more difficult to use and they also change the way you are writing an application, but they can be an effective way of improving performance of your applications. To use the async write and read functions, you need to provide call back functions. These functions are called when the operation is complete.

# async io
def w_oncomplete(completion):
    print('completed writing data to memory')

def w_onsafe(completion):
    print('data is safely stored on disk: %s' % completion.is_safe())

completion = ioctx.aio_write_full('aio_obj', "Let's test async IO",
                                  w_oncomplete, w_onsafe)

completion.wait_for_safe_and_cb()

print('all done writing object')

def r_oncomplete(completion, data_read):
    print(data_read)
    assert(data_read == "Let's test async IO")

completion = ioctx.aio_read('aio_obj', 8192, 0, r_oncomplete)

completion.wait_for_complete_and_cb()

Ok, so there’s a lot going on here, let’s break it down line by line:
Lines 2 and 6 are two different call back functions provided to aio_write_full. w_oncomplete is called when the data is written to memory on the OSDs, and w_onsafe is called when the data is safely stored on disk. The interesting part is that a Completion object is passed to those two functions; this object contains a few public methods allowing you to check the status of the operation. Depending on your needs you can provide just one these functions. It is actually not required to pass either of these two functions to aio_write_full(), but then your application will never be notified of when the operation has completed.

On line 12, there’s a call to wait_for_safe_and_cb(). This method waits for the async operation to complete, thus suspending this task and allowing others to proceed.

Line 17 defines a callback for the aio_read(). The difference between aio_read() and read() is that the object data is returned in the callback function. Notice on line 21 that aio_read() does not currently provide default values for length and offset like read() does, so for now, you must provide a value for those parameters.

Write and Read Object Metadata

Librados also has support for writing and reading object metadata:

# set metadata
ioctx.set_xattr('obj', 'author', 'Thiago')
print(ioctx.get_xattr('obj', 'author'))

Delete objects

To delete an object, simply call remove_object() or in case you have an instance of the object itself (see listing objects below), call the remove() method.

# remove object
ioctx.remove_object('hw')

Iterate through list of objects

To iterate through a list of all objects in a pool, call list_objects() and use a for loop. The Object class provides a number of APIs allowing you to perform specific operations directly on a object (i.e., write, read, remove, etc).

# listing objects in pool
object_iterator = ioctx.list_objects()
for o in object_iterator:
    print("Object: '%s', contents: %s" %
          (o.key, o.read()))

Close connections

To finish, don’t forget to close the connections:

ioctx.close()
cluster.shutdown()

What’s next?

Next, I’d like to eventually run some benchmark tests to compare ‘sync’ vs ‘async’ I/O with this API. If anybody has any insights to share please make sure to leave a comment!