Good afternoon, dear readers!

In previous articles, I outlined in general terms about our open fault-tolerant Elliptics repository, as well as a little bit of detail . Today I will visually tell you and show you how to use Elliptics on the example of creating your own fault-tolerant HabraMusic.


')
HabraMusic is your personal music repository with regional support, data replication, minimal disk and network load, and a simple HTTP API that can be used in any of your applications or on a personal website.

Under the cut - step by step details.

Deploying Elliptics

First we need to put Elliptics. To do this, we collect packages in our repository for such common server systems as:

• Debian Wheezy;
• Ubuntu Precise 12.04;
• Ubuntu Trusty 14.04;
• RedHat 6 / CentOS 6.

For other distributions, you can build the system yourself from the sources that are located on our GitHub account .

For further work it is necessary to deliver the packages:

$ sudo apt-get install elliptics elliptics-client rift 

or

 $ sudo yum install elliptics elliptics-client rift 

If you want to try in the API for C ++, then you should also install the package elliptics-dev or elliptics-client-devel , depending on the distribution.

Great, packages are delivered. Now it's time to launch your cluster. In this tutorial, we will run all the servers on the same machine. In a real situation, they should be spread to different machines and data centers. On GitHub all configuration files which will be necessary for our tutorial are located.

First node-1.json look at the example of the node-1.json configuration file:

 { "loggers": { "type": "/srv/elliptics/1/log.txt", "level": 4 }, "options": { "join": true, "flags": 4, "remote": [ "autodiscovery:224.0.0.5:1025:2" ], "address": [ "localhost:1025:2" ], "wait_timeout": 60, "check_timeout": 60, "io_thread_num": 16, "nonblocking_io_thread_num": 16, "net_thread_num": 4, "daemon": true, "auth_cookie": "<-Don't forget to change it->", "cache": { "size": 1073741824 }, "indexes_shard_count": 16, "monitor_port": 20000 }, "backends": [ { "type": "blob", "group": 1, "history": "/srv/elliptics/1/history/", "data": "/srv/elliptics/1/blob/data", "sync": "-1", "blob_flags": "129", "blob_size": "10G", "records_in_blob": "1000000" } ] } 

To run Elliptics with this configuration file, you need to create the following directories:

• /srv/elliptics/1/ - here logs with the Debug level will be written to the log.txt file;
• /srv/elliptics/1/blob/ - blobs with data will be located here;
• /srv/elliptics/1/history/ - there will be a file with the beginning of the intervals from the DHT-ring.

Details on the remaining specified and unspecified fields can be found in our documentation .

To start, just run:

 $ dnet_ioserv -c node-1.json 

But we want to build a failover cluster. For this we need to raise at least one more car in the second group. In her config you will need to specify a different address, as well as other remote addresses. As a result, while the node-1.json, node-2.json configuration files are used to store data in 2 copies.

Create directories needed for the first two nodes, and run each of them:

 $ for id in `seq 1 2`; do mkdir /srv/elliptics/$id/{history,blob} -p; done $ for id in `seq 1 2`; do dnet_ioserv -c node-$id.json; done 

Meet C ++ / Python API

And now, when the servers are running, let's start our acquaintance with low-level C ++ and Python API. To do this, I will show every action in two languages ​​at once with explanations.

Connect to servers

First you need to create a client node to connect to Elliptics servers.

C ++:

 #include <elliptics/session.hpp> using namespace ioremap::elliptics; file_logger logger(“/dev/stderr”, DNET_LOG_ERROR); node n(file_logger); n.add_remote("localhost", 1025); n.add_remote("localhost", 1026); session sess(n); sess.set_groups({ 1, 2, 3 }); 

Python:

 import elliptics log = elliptics.Logger(“/dev/stderr”, elliptics.log_level.debug) node = elliptics.Node(log) node.add_remote(“localhost”, 1025) node.add_remote(“localhost”, 1026) sess = elliptics.Session(node) sess.group = [ 1, 2, 3 ] 

We write data in Elliptics

Write some data in Elliptics:

C ++:

 auto async_result = sess.write_data(std::string("test_key"), "some data", offset); 

Python:

 async_result = sess.write_data(“test_key”, “some_data”, offset) 

The above code will send a request to write data to the Elliptics server. However, after that you need to wait for the answers and process them. Many teams send multiple requests to Elliptics servers. For example, data recording sends requests simultaneously to each group on request. Each request receives a response, from which you can find out some information common to all commands (for example, the address of the server that processed the request, or the result code of the command execution), as well as information specific to this command (such as the path to the blob in which a recording was made, with coordinates inside it).

Handling server response

There are several options for processing the result of the command. They should be considered in more detail.

Synchronous request

C ++:

 try { async_result.wait(); } catch (error &e) { std::cerr << “Error while write execution: “ << e.message(); return; } auto sync_result = async_result.get(); for (write_result_entry entry : async_result.get()) { // Process entry std::cerr << "Received answer from: " << dnet_server_convert_dnet_addr(result.address()) << "status: " << result.status() << std::endl; } 

Python:

 try: async_result.wait() except Exception as e: print "Write failed: ", e for entry in async_result.get(): print("Reply from {}, status: {}".format(entry.address, entry.error.code)) 

You can get rid of exceptions by invoking the session method set_exceptions_policy, which is present in both C ++ and the Python API. In this case, you need to do a manual check for errors:

C ++:

 if (async_result.error()) { //    } 

Python:

 if (async_result.error) { //    } 

Iterated query

Sometimes the total amount of data from the server may be too large to be stored in memory at once. There may also be a requirement to process data as it arrives. In this case, you can use an iteration: the body of the loop will be executed for each response received from the server, as their performances. Between packets, the user's thread will sleep.

C ++:

 for (auto entry : async_result) { } 

Python:

 for entry in async_result: pass 

Asynchronous request

Also, as soon as the answer is completely received, you can execute your code in the Elliptics stream.

C ++:

 async_result.connect([] (const sync_write_result &result, const error_info &error) { if (error) { //   return; } for (auto entry : result) { //   } }); 

Python:

 def handler(results, error): if error.code != 0: //   return for entry in results: //   pass async_result.connect(handler) 

Asynchronous iterative query

This method is recommended to use if there is a lot of data from the server, you want to process them as they arrive and you do not want to block any of the threads to wait for the data. The first callback will be called for every response received from the server, and the last one will be called after the last response has been received. And the only argument will be information about a possible error, if the final response code is not zero.

C ++

 async_result.connect([] (const write_result_entry &entry) { //   }, [] (const error_info &error) { if (error) { //   } }); 

Python:

 def result_handler(result, error): //   pass def error_handler(results, error): if error.code != 0: //   pass async_result.connect(result_handler, error_handler) 

Reading data

To read the data, it is enough to execute the code below, and then process the response in the same way as in the case of write:

C ++:

 auto async_result = sess.read_data(std::string("test_key"), offset, size); 

Python:

 async_result = sess.read_data(“test_key”, offset, size) 

It should be remembered that for reading the server could not find the file the first time. In this case, the number of answers may be more than one, and all, except the last, will be with a negative error code.

Data deletion

Deleting data is also easy.

C ++:

 auto async_result = sess.remove_data(std::string("test_key")); 

Python:

 async_result = sess.remove_data(“test_key”) 

And much more

available in our documentation .

HTTP API

We recommend using Rift to work with the HTTP API. Detailed documentation is available here . Rift is based on our high-performance asynchronous HTTP server TheVoid , specially designed to work as an HTTP backend for Nginx.

Rift has support for two main modes of operation - with and without support for buckets. In this post, we will consider the baketa mode as more flexible and, of course, difficult. Buckets allow you to share files and rights of different users. For example, you can give one user the right to write to the bakt, and another will only be able to read from it.

As an example, consider the way to create your service for listening to music.

First we need to raise at least one more Elliptics server for storing such Rift meta-information as buckets.

 $ dnet_ioserv -c node-3.json 

Now you need to configure and raise Rift, its config is listed in the GitHub repository under the name rift.json , let's take a closer look:

 { "endpoints": [ "0.0.0.0:8080" ], "backlog": 128, "threads": 2, "buffer_size": 65536, "logger": { "file": "/srv/rift/log.txt", "level": 4 }, "daemon": { "fork": true, "uid": 1000 }, "monitor-port": 21000, "application": { "remotes": [ "localhost:1025:2", "localhost:1026:2", "localhost:1027:2" ], "groups": [], "metadata-groups": [ 3 ], "bucket": { "timeout": 60 }, "read-timeout": 10, "write-timeout": 16, "redirect-port": 8081, "path-prefix": "/srv/elliptics/" } } 

The configuration file can be divided into two sections. The first is responsible for configuring the HTTP server as a whole:

• “Endpoints” - a list of ports and unix-sockets that the server listens to incoming connections;
• “Logger” - logging system configuration;
• “Daemon” - description of server demonization.

The second section - “application” - is responsible for the configuration of the Rift:

• “Remotes” - list of Elliptics nodes;
• “Metadata-groups” - a list of groups in which all meta-information is stored, for example, bucket;
• “Redirect-port” - the port that will listen to Nginx on data machines;
• “Path-prefix” is the initial part of the path to blobs on data machines.

Details about all the parameters of the configuration file can be found here .

 $ rift_server -c rift.json 

All operations in Rift with activated bucks require authorization. The signature generation algorithm is described in the documentation. Here I will use an extremely simple http-client in Python, which will sign all requests properly. It lies in the archive and is called http_auth.py .

Create a bake, in which we will fill our music. And I would like to have the right to download music only the administrator of the resource, that is, we, and anyone could listen. Consider the baket configuration file:

 { "groups": [ 1, 2 ], "acl": [ { "user": "admin", "token": "favorite_music", "flags": 6 }, { "user": "*", "token": "very_secret_token", "flags": 1 } ] } 

The first field “groups” says that the data will be written in two copies, for copies in groups 1 and 2. More information about groups can be found in previous articles .
The second field “acl” contains access rights for different users, in this example, for each user 3 fields were specified:

• “User” is the public user id;
• “Token” is a secret token by which authorization takes place;
• “Flags” is a description of user rights, currently the following flags and their bit combinations are available:
  
  • 1 - no authorization required for this user;
  • 2 - this user has the right to write data to the bakt;
  • 4 - this user has the right to change the data of the bake itself (list of groups, acl, etc.).

Only “admin” possesses the rights to write new files and change the baket, its secret key is also indicated, with which the authorization will take place. But read access has any external user without the need for authorization. Now let's create a bake music in the all directory:

 $ ./http_auth.py http://localhost:8080/update-bucket/all/music --file update-bucket.json 200 

A directory is a special entity that allows, for example, finding all the buckets it contains. In the created bake we will fill in some music by the key example.mp3 :

 $ ./http_auth.py http://localhost:8080/upload/music/example.mp3?user=admin --file example.mp3 --token favorite_music 200 

After that, you can listen to music directly from any player:

 $ mplayer http://localhost:8080/get/music/example.mp3 

The /get handle supports the Last-Modified and Range headers for more efficient traffic management, but this is still not the most rational use of resources. It would be much more efficient to stream data using Nginx from the physical machines closest to the user on which this file is stored. This is achieved by supporting the regionality in Elliptics.

For this we need Nginx with our eblob module. It can be found on GitHub .

You can assemble it using any suitable Nginx (versions 1.7.x supported) are approximately as follows:

 $ git clone https://github.com/toshic/nginx_patches.git $ wget 'http://nginx.org/download/nginx-1.7.0.tar.gz' $ tar -xzvf nginx-1.7.0.tar.gz $ cd nginx-1.7.0 $ ./configure --prefix=/opt/nginx --add-module=../nginx_patches/nginx-eblob $ make $ sudo checkinstall # :) 

To activate an eblob module, simply add the “eblob;” block to the location section. Thus, the minimum configuration file looks like this:

 worker_processes 8; events { worker_connections 1024; } http { sendfile on; server { listen 8081; server_name localhost; location / { root /var/srw; eblob; } } } 

Of course, for the combat environment Nginx should not be configured so naively.

Great, now that nginx is configured and running, you can stream music directly from it:

 $ mplayer -q http://localhost:8080/redirect/music/example.mp3 

On this request, the server will generate a special type of url, upon receiving which Nginx will be able to send music directly from blobs, in which Elliptics stores its data.

Much more about the Rift can be read in our documentation . There is also a detailed description of all available methods.

Instead of conclusion

You can read about this and much more in our documentation on doc.reverbrain.com :

• Elliptics ;
• TheVoid ;
• Rift .

I remind you that all the projects provided in this article are open and located on GitHub: https://github.com/reverbrain .