Zlog: A shared-log on RADOS

A Distributed Shared Log on RADOS

So far we have seen how the CORFU system functions to achieve high-throughput appends to a globally ordered shared-log. Now we are going to discuss how the basic design of CORFU can be mapped onto Ceph. Specifically we are going to discuss the mapping of CORFU onto RADOS, a scalable, fault-tolerant object-based storage system that powers Ceph. Next I’ll give a brief overview of the salient features of RADOS that are important for this post, but a full discussion of the architecture of Ceph is beyond the scope of this post. There are plenty of resources for learning about Ceph, and I invite anyone not familiar with the technology to check them out!

Overview. The RADOS system is a distributed, scalable, and fault-tolerant object-based storage system that presents a cluster of unreliable nodes as a highly available object store. Each node is called as Object Storage Device (OSD) and is typically an off-the-shelf node with CPU, memory, and local disk. The RADOS system logically groups objects into placement groups and each placement group is assigned to one or more OSDs across which objects are stored redundantly (e.g. replication or erasure coding). Clients interact with RADOS through a library with a rich interface that completely hides the details of the underlying distributed system. A salient feature of RADOS that we make use of in this project is support for the injection of user-defined code for the creation of domain-specific transactional interfaces to object data that run within the cluster of storage devices.

Next I’ll discuss the layout of a log within RADOS, the custom storage interfaces designed for the CORFU protocol, and then present the client libraries and show examples of how easy it is to use the zlog implementation.

Logical Log Layout

In order to store a log in RADOS a choice must be made about how to assign log positions to addressable storage. For instance, each log position could be written to a dedicated object, or a single log could be striped across a fixed number of objects. Other strategies exist and each have different performance and implementation trade-offs, but that is a topic for a later post. However, for reasons related to bounding the slow tail finding mechanism, we’ve chosen a basic, initial implementation that stripes a log across a fixed (but adjustable across epochs) number of objects.

A log is defined by a stripe width parameter that specifies how many objects the log should be striped across (a good choice is the number of placement groups in the target storage pool). Each log position is mapped round-robin onto this set of objects by the formula object-id = log-position % stripe-width. One might be tempted to think this would cause scalability issues as an object grows in size, however, RADOS contains a wide variety of object interfaces that allow us to access at a fine granularity both bulk and structured data associated with an object, efficiently supporting access to individual log positions even as the amount of data associated with an object grows.

This mapping of log positions onto objects within RADOS is a direct analogy to the mapping of log positions in CORFU onto a set of SSDs organized into a redundancy group. Thus, the enforcement of the storage interface in CORFU must be achieved at the object level in RADOS. To accomplish this we use the object class feature of RADOS to install a custom interface that implements the interface expected by the CORFU protocol.

Zlog Custom Object Interface

The following code snippet shows the custom storage interface to objects in RADOS used to store log data. The ability to create custom object interfaces is a major feature of RADOS and you can learn more about it here.

First we have the set of possible return codes defined by our custom interface. Note that error handling is two-level. The outer layer is composed of RADOS specific errors (e.g. network errors or failures) and the inner layer is for our protocol. Thus, these codes are only valid when the outer layer reports success.

Note that below there are some RADOS specific items: the ObjectWriteOperation can be thought of as just encoding the name of the target object for the particular operation, the bufferlist is a fancy object for holding binary data, and the data retrieved through cls_zlog_read is accessed through the RADOS-specific ObjectReadOperation object.

  /* Operation error codes. */
                      enum {
                        CLS_ZLOG_OK            = 0x00,
                        CLS_ZLOG_STALE_EPOCH   = 0x01,
                        CLS_ZLOG_READ_ONLY     = 0x02,
                        CLS_ZLOG_NOT_WRITTEN   = 0x03,
                        CLS_ZLOG_INVALIDATED   = 0x04,
                        CLS_ZLOG_INVALID_EPOCH = 0x05,
                      };
                    
                      /*
                       * Install a new epoch in an object.
                       */
                      void cls_zlog_seal(librados::ObjectWriteOperation& op, uint64_t epoch);
                    
                      /*
                       * Mark a log position as junk in an object.
                       */
                      void cls_zlog_fill(librados::ObjectWriteOperation& op, uint64_t epoch,
                        uint64_t position);
                    
                      /*
                       * Write to a log position in an object.
                       */
                      void cls_zlog_write(librados::ObjectWriteOperation& op, uint64_t epoch,
                        uint64_t position, ceph::bufferlist& data);
                    
                      /*
                       * Read the contents of a log position in an object.
                       */
                      void cls_zlog_read(librados::ObjectReadOperation& op, uint64_t epoch,
                        uint65_t position);
                    
                      /*
                       * Find the maximum log position written to an object.
                       */
                      void cls_zlog_max_position(librados::ObjectReadOperation& op, uint64_t epoch,
                        uint64_t *pposition, int *pret);
                    

The only substantial departure from the CORFU storage interface is that unlike in the CORFU protocol, finding the maximum position is a separate operation from seal which arises from certain limitations on what RADOS storage interfaces can do. However, cls_zlog_max_position does enforce the last sealed epoch, so when a new sequencer comes online and executes the slow tail finding procedure it should first seal each object and then use the newly installed epoch when querying for the maximum written position.

We now have enough detail on how the system is structured to describe the slow tail finding procedure. First we need to seal each object that a log is striped across, and then we’ll query for the maximum position written. In the following code snippet, the for loop and slot_to_oid function implement log striping, and for each object we construct and execute a seal operation directed at the target object with the specified epoch.

int Log::Seal(uint64_t epoch)
                    {
                      for (int i = 0; i < stripe_size_; i++) {
                        std::string oid = slot_to_oid(i);
                    
                        librados::ObjectWriteOperation op;
                        cls_zlog_seal(op, epoch);
                        int ret = ioctx_->operate(oid, &op);
                    
                        if (ret != zlog::CLS_ZLOG_OK) {
                          std::cerr << "failed to seal object" << std::endl;
                          return ret;
                        }
                      }
                    
                      return 0;
                    }
                    

Once we’ve sealed each object that a log is striped across we use the newly installed epoch to query each object again for the maximum position that has been written. Note that since no existing client has knowledge of this new epoch value and each object rejects requests tagged with old epochs, we can be certain that once complete we have a guarantee on the maximum log position written across all objects.

Like before we loop over each object that the log is striped across. For each object we construct and execute cls_zlog_max_position request. In the error free case we simply record the maximum position seen and return this value to the caller.

int Log::FindMaxPosition(uint64_t epoch, uint64_t *pposition)
                    {
                      uint64_t max_position = 0;
                    
                      for (int i = 0; i < stripe_size_; i++) {
                        std::string oid = slot_to_oid(i);
                    
                        librados::bufferlist bl;
                        librados::ObjectReadOperation op;
                        int op_ret;
                        uint64_t this_pos;
                    
                        cls_zlog_max_position(op, epoch, &this_pos, &op_ret);
                    
                        int ret = ioctx_->operate(oid, &op, &bl);
                        if (ret < 0) {
                          if (ret == -ENOENT) // no writes yet
                            continue;
                          std::cerr << "failed to find max pos " << ret << std::endl;
                          return ret;
                        }
                    
                        if (op_ret == zlog::CLS_ZLOG_OK)
                          max_position = std::max(max_position, this_pos);
                      }
                    
                      *pposition =  max_position;
                    
                      return 0;
                    }
                    

And that’s it. A sequencer installs a new epoch and executes the slow tail finding procedure for each new log, and anytime a sequencer is restarted. The resulting position of the slow tail finding mechanism is what the sequencer uses to initialize its in-memory cache of a log’s tail. Next we’ll discuss the design of the sequencer used in our implementation.

Zlog Sequencer

There is a lot of flexibility in how the sequencer is implemented. All it needs to do is ensure that it can get a new epoch value, and that it seals all of the storage devices before starting to hand out log positions. Our current implementation is a standard Boost.ASIO asynchronous TCP server. It is capable of managing any number of logs stored in multiple RADOS pools. In the later section Epoch Management I discuss options for how the sequencer can acquire new epoch values. The code below shows the interface to the sequencer service, which contains only one important method to query (and optionally increment) the tail of a log.

class SeqrClient {
                     public:
                      SeqrClient(const char *host, const char *port) :
                        socket_(io_service_), host_(host), port_(port)
                      {}
                    
                      void Connect();
                    
                      /*
                       * Retrieve (and optionally increment) the tail position for a given log.
                       */
                      int CheckTail(uint64_t epoch, const std::string& pool,
                          const std::string& name, uint64_t *position, bool next);
                    
                     private:
                      /* ... boost stuff removed ... */
                    };
                    

Log clients don’t interact with the sequencer directly, but rather indirectly through a client library that implements the log abstraction. For instance, the sequencer is contacted automatically when a client wants to append to the log.

Zlog Client Library

We now have all the bits and pieces necessary to demonstrate an end-to-end example. The following code shows the components of the log abstraction from the client library.

class Log {
                     public:
                      /* ... cut ... */
                      int Append(ceph::bufferlist& bl, uint64_t *pposition);
                      int Read(uint64_t position, ceph::bufferlist& bl);
                      int Trim(uint64_t position);
                      int Fill(uint64_t position);
                      /* ... cut ... */
                    }
                    

The details of the log abstraction shown here were discussed in the previous section on the CORFU system, so we’ll jump right into how this stuff is implemented. The next code snippet is for the Append method that upon successful completion returns to the caller the log position where the entry has been stored.

The first step is to ask the sequencer what the current tail is. Specifically we ask it to give us the next position in the log by first atomically incrementing its in-memory counter using the CheckTail method of the sequencer client library.

int Log::Append(ceph::bufferlist& data, uint64_t *pposition)
                    {
                      for (;;) {
                        uint64_t position;
                        int ret = CheckTail(&position, true);
                        if (ret)
                          return ret;
                    

If successful, the sequencer will have given us the new tail position, so we should be able to write to this location without any hassle. To do this we construct a cls_zlog_write operation and execute it on the target object for this log position.

    librados::ObjectWriteOperation op;
                        zlog::cls_zlog_write(op, epoch_, position, data);
                    
                        std::string oid = position_to_oid(position);
                        ret = ioctx_->operate(oid, &op);
                        if (ret < 0) {
                          std::cerr << "append: failed ret " << ret << std::endl;
                          return ret;
                        }
                    
                        if (ret == zlog::CLS_ZLOG_OK) {
                          if (pposition)
                            *pposition = position;
                          return 0;
                        }
                    

If no RADOS-level errors occurred (e.g. network errors) then we examine the possible return values for our custom storage interface. First, if the operation was successful we return the position to the client and we are done. Otherwise there are two possible return values. First, we could been told by the storage device that our epoch value is too old (zlog::CLS_ZLOG_STALE_EPOCH). This can occur if the sequencer had to restart, or if the log was reconfigured (e.g. the stripe width was changed). The only other possible return value is that the log position has already been written (zlog::CLS_ZLOG_READ_ONLY). This can occur if it was filled with junk by another client to invalidate the position, or if following a sequencer restart another client beat us to the punch.

    if (ret == zlog::CLS_ZLOG_STALE_EPOCH) {
                          ret = RefreshProjection();
                          if (ret)
                            return ret;
                          continue;
                        }
                    
                        assert(ret == zlog::CLS_ZLOG_READ_ONLY);
                      }
                      assert(0);
                    }
                    

And that’s pretty much it. You can check out the rest of the implementation including the details of the custom storage interface on Github:

• https://github.com/cruzdb/zlog
• https://github.com/cruzdb/zlog/tree/master/src/libzlog/backend

We’ll finish up this post with a basic example of how to use the interface, discuss some future work we’ll be looking at, and address the issue of dealing with epoch values.