目前我们有两种路径使用Ceph的块存储：

- 利用QEMU/KVM通过librbd与 Ceph 块设备交互，主要为虚拟机提供块存储设备，如下图所示； - 利用kernel module与Host kernel交互，主要为物理机提供块设备支持。

Librbd 是Ceph提供的块存储接口的抽象，它提供C/C++、Python等多种接口。对于C++，最主要的两个类就是RBD 和 Image。 RBD 主要负责创建、删除、克隆映像等操作，而Image 类负责映像的读写等操作。

准备工作

对于任何客户端应用，都需要首先连接到一个运行良好的Ceph集群。

获取集群句柄

//声明Rados对象，并初始化librados::Rados rados;ret = rados.init("admin"); // just use the client.admin keyringif (ret < 0) { // let's handle any error that might have come back    std::cerr << "couldn't initialize rados! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "we just set up a rados cluster object" << std::endl;}//获取配置文件信息: /etc/ceph/ceph.conf// 1. 根据命令行参数/*ret = rados.conf_parse_argv(argc, argv);if (ret < 0) {    // This really can't happen, but we need to check to be a good citizen.    std::cerr << "failed to parse config options! error " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "we just parsed our config options" << std::endl;    // We also want to apply the config file if the user specified    // one, and conf_parse_argv won't do that for us.    for (int i = 0; i < argc; ++i) {
              if ((strcmp(argv[i], "-c") == 0) || (strcmp(argv[i], "--conf") == 0)) {            ret = rados.conf_read_file(argv[i+1]);            if (ret < 0) {                // This could fail if the config file is malformed, but it'd be hard.                std::cerr << "failed to parse config file " << argv[i+1]                        << "! error" << ret << std::endl;                ret = EXIT_FAILURE;                return EXIT_FAILURE;            }            break;        }    }}*/// 2. 程序里面指定ret = rados.conf_read_file("/etc/ceph/ceph.conf");if (ret < 0) {    // This could fail if the config file is malformed, but it'd be hard.    std::cerr << "failed to parse config file! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10 
    11 
    12 
    13 
    14 
    15 
    16 
    17 
    18 
    19 
    20 
    21 
    22 
    23 
    24 
    25 
    26 
    27 
    28 
    29 
    30 
    31 
    32 
    33 
    34 
    35 
    36 
    37 
    38 
    39 
    40 
    41 
    42 
    43 
    44 
    45 
    46 
    47

连接集群

ret = rados.connect();if (ret < 0) {    std::cerr << "couldn't connect to cluster! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "we just connected to the rados cluster" << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8

创建I/O上下文环境

如果没有存储池，需要先新建一个存储池。

新建存储池

const char *pool_name = "gnar";ret = rados.pool_create(pool_name);if (ret < 0) {    std::cerr << "couldn't create pool! error " << ret << std::endl;    ret = EXIT_FAILURE;    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;} else {    std::cout << "we just created a new pool named " << pool_name << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10

新建I/O上下文环境

librados::IoCtx io_ctx; //I/O上下文const char *pool_name = "gnar";ret = rados.ioctx_create(pool_name, io_ctx);if (ret < 0) {    std::cerr << "couldn't setup ioctx! err " << ret << std::endl;    ret = EXIT_FAILURE;    rados.shutdown(); //断开集群连接    return EXIT_FAILURE; } else {    std::cout << "we just created an ioctx for our pool" << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10 
    11

若干RBD映像API

声明RBD对象，创建rbd映像

librbd::RBD rbd;const char *image_name = "rumboo";uint64_t init_size = (uint64_t) 200 * 1024 * 1024; //映像初始化大小200MBuint64_t features = 1; //影响feature个数int order = 22; //默认值为22, 即4MB (1 << 22)ret = rbd.create2(io_ctx, image_name, init_size, features, &order);if (ret < 0) {    std::cerr << "couldn't create rbd image! err " << ret << std::endl;    ret = EXIT_FAILURE;    io_ctx.close(); //关闭I/O上下文    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;} else {    std::cout << "We just created an rbd image" << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10 
    11 
    12 
    13 
    14 
    15

打开rbd映像

librbd::RBD rbd;const char *image_name = "rumboo";librbd::Image image;ret = rbd.open(io_ctx, image, image_name);if (ret < 0) {    std::cerr << "couldn't open rbd image! err " << ret << std::endl;    ret = EXIT_FAILURE;    io_ctx.close(); //关闭I/O上下文    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;} else {    std::cout << "We just opened an rbd image" << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10 
    11 
    12 
    13

查看映像大小

uint64_t size = 0;ret = image.size(&size);if (ret < 0) {    std::cerr << "couldn't get image size! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "The size of the image is " << size << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9

调整映像大小

size = (uint64_t) 500 * 1024 * 1024; //调整映像大小500MBret = image.resize(size);if (ret < 0) {    std::cerr << "couldn't change the size of the image! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "We just change the size of the image" << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9

查看映像ID

std::string id;ret = image.get_id(&id);if (ret < 0) {    std::cerr << "couldn't get image ID! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "The ID of the image is " << id << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9

查看映像features

features = 0;ret = image.features(&features);if (ret < 0) {    std::cerr << "couldn't get image features! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "The features of the image are " << features << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9

查看映像状态信息

librbd::image_info_t info;ret = image.stat(info, sizeof(info));if (ret < 0) {    std::cerr << "couldn't get image stat_info! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "info.size is " << info.size << std::endl;    std::cout << "info.obj_size is " << info.obj_size << std::endl;    std::cout << "info.num_objs is " << info.num_objs << std::endl;    std::cout << "info.order is " << info.order << std::endl;    std::cout << "info.block_name_prefix is " << info.block_name_prefix << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10 
    11 
    12 
    13

查看存储池ID

std::cout << "data pool id is " << image.get_data_pool_id() << std::endl;   
   
    1

查看block_name_prefix

std::cout << "block name prefix is " << image.get_block_name_prefix() << std::endl;   
   
    1

查看flags

uint64_t flags = 0;ret = image.get_flags(&flags);if (ret < 0) {    std::cerr << "couldn't get image flags! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "image flags is " << flags << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9

查看条带化参数

std::cout << "image stripe unit is " << image.get_stripe_unit() << std::endl;std::cout << "image stripe count is " << image.get_stripe_count() << std::endl;   
   
    1 
    2

RBD映像数据读写

数据读写 – synchronous

uint64_t ofs_w = (uint64_t) 0; //读写偏移量uint64_t ofs_r = (uint64_t) 0;size_t len_w = 100; //读写长度size_t len_r = 100;ceph::bufferlist bl_w; //读写bufferlistceph::bufferlist bl_r;const char *fn_i = "input"; //读写文件名const char *fn_o = "output";std::string error;ret = bl_r.read_file(fn_i, &error);std::cout << "read file ret = " << ret << std::endl;if (ret < 0) {    std::cerr << "couldn't read file! err " << ret << std::endl;    ret = EXIT_FAILURE;    image.close(); //关闭rbd映像    io_ctx.close(); //关闭I/O上下文    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;} else {    std::cout << "We just read a file" << std::endl;}ssize_t ret_w = image.write2(ofs_w, len_w, bl_r, 0);ssize_t ret_r = image.read2(ofs_r, len_r, bl_w, 0);ret = bl_w.write_file(fn_o, 0644);std::cout << "write file ret = " << ret << std::endl;if (ret < 0) {    std::cerr << "couldn't write file! err " << ret << std::endl;    ret = EXIT_FAILURE;    image.close(); //关闭rbd映像    io_ctx.close(); //关闭I/O上下文    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;} else {    std::cout << "We just wrote a file" << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10 
    11 
    12 
    13 
    14 
    15 
    16 
    17 
    18 
    19 
    20 
    21 
    22 
    23 
    24 
    25 
    26 
    27 
    28 
    29 
    30 
    31 
    32 
    33 
    34 
    35 
    36 
    37

数据读写 – asynchronous

std::string data = "foo"; uint64_t ofs_aiow = (uint64_t) 100; //读写偏移量uint64_t ofs_aior = (uint64_t) 100;size_t len_aiow = 600; //读写长度size_t len_aior = 600;ceph::bufferlist bl_aiow; //读写bufferlistceph::bufferlist bl_aior;librbd::RBD::AioCompletion *write_completion = new librbd::RBD::AioCompletion(    NULL, (librbd::callback_t) simple_write_cb); //读写AioCompletionlibrbd::RBD::AioCompletion *read_completion = new librbd::RBD::AioCompletion(    NULL, (librbd::callback_t) simple_read_cb);for (int i = 0; i < 200; ++i) {    bl_aior.append(data);}std::cout << bl_aior.to_str() << std::endl;ret = image.aio_write2(ofs_aiow, len_aiow, bl_aior, write_completion, 0);if (ret < 0) {    std::cerr << "couldn't start write! error " << ret << std::endl;    ret = EXIT_FAILURE;    image.close(); //关闭rbd映像    io_ctx.close(); //关闭I/O上下文    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;}write_completion->wait_for_complete(); //等待写完成ret_w = write_completion->get_return_value();if (ret_w < 0) {    std::cerr << "couldn't write! error " << ret << std::endl;    ret_w = EXIT_FAILURE;    image.close(); //关闭rbd映像    io_ctx.close(); //关闭I/O上下文    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;} else {    std::cout << "we just write data successfully, return value is " << ret_w << std::endl;}ret = image.aio_read2(ofs_aior, len_aior, bl_aiow, read_completion, 0);if (ret < 0) {    std::cerr << "couldn't start read! error " << ret << std::endl;    ret = EXIT_FAILURE;    image.close(); //关闭rbd映像    io_ctx.close(); //关闭I/O上下文    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;}read_completion->wait_for_complete(); //等待读完成ret_r = read_completion->get_return_value();if (ret_r < 0) {    std::cerr << "couldn't read! error " << ret << std::endl;    ret_r = EXIT_FAILURE;    image.close(); //关闭rbd映像    io_ctx.close(); //关闭I/O上下文    rados.shutdown(); //断开集群连接    return EXIT_FAILURE;} else {    std::cout << "we just read data successfully, return value is " << ret_r << std::endl;}std::cout << bl_aiow.to_str() << std::endl;write_completion->release();read_completion->release();void simple_write_cb(librbd::completion_t cb, void *arg) {    std::cout << "write completion cb called!" << std::endl;}//简单的回调函数，用于librbd::RBD::AioCompletionvoid simple_read_cb(librbd::completion_t cb, void *arg) {    std::cout << "read completion cb called!" << std::endl;}   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10 
    11 
    12 
    13 
    14 
    15 
    16 
    17 
    18 
    19 
    20 
    21 
    22 
    23 
    24 
    25 
    26 
    27 
    28 
    29 
    30 
    31 
    32 
    33 
    34 
    35 
    36 
    37 
    38 
    39 
    40 
    41 
    42 
    43 
    44 
    45 
    46 
    47 
    48 
    49 
    50 
    51 
    52 
    53 
    54 
    55 
    56 
    57 
    58 
    59 
    60 
    61 
    62 
    63 
    64 
    65 
    66 
    67 
    68 
    69 
    70 
    71 
    72 
    73

收尾工作

在最后，一定不能忘记关闭rbd映像、I/O上下文，断开集群连接。

ret = image.close(); //关闭rbd映像if (ret < 0) {    std::cerr << "couldn't close rbd image! err " << ret << std::endl;    ret = EXIT_FAILURE;    return EXIT_FAILURE;} else {    std::cout << "we just closed an rbd image" << std::endl;}io_ctx.close(); //关闭I/O上下文rados.shutdown(); //断开集群连接return EXIT_SUCCESS;   
   
    1 
    2 
    3 
    4 
    5 
    6 
    7 
    8 
    9 
    10 
    11