src/ceph/src/cls/rbd/cls_rbd_client.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "cls/rbd/cls_rbd_client.h"
   5 #include "cls/lock/cls_lock_client.h"
   6 #include "include/buffer.h"
   7 #include "include/encoding.h"
   8 #include "include/rbd_types.h"
   9 #include "include/rados/librados.hpp"
  10
  11 #include <errno.h>
  12
  13 namespace librbd {
  14   namespace cls_client {
  15
  16     void get_immutable_metadata_start(librados::ObjectReadOperation *op) {
  17       bufferlist bl, empty_bl;
  18       snapid_t snap = CEPH_NOSNAP;
  19       ::encode(snap, bl);
  20       op->exec("rbd", "get_size", bl);
  21       op->exec("rbd", "get_object_prefix", empty_bl);
  22     }
  23
  24     int get_immutable_metadata_finish(bufferlist::iterator *it,
  25                                       std::string *object_prefix,
  26                                       uint8_t *order) {
  27       try {
  28         uint64_t size;
  29         // get_size
  30         ::decode(*order, *it);
  31         ::decode(size, *it);
  32         // get_object_prefix
  33         ::decode(*object_prefix, *it);
  34       } catch (const buffer::error &err) {
  35         return -EBADMSG;
  36       }
  37       return 0;
  38
  39     }
  40
  41     int get_immutable_metadata(librados::IoCtx *ioctx, const std::string &oid,
  42                                std::string *object_prefix, uint8_t *order)
  43     {
  44       librados::ObjectReadOperation op;
  45       get_immutable_metadata_start(&op);
  46
  47       bufferlist out_bl;
  48       int r = ioctx->operate(oid, &op, &out_bl);
  49       if (r < 0) {
  50         return r;
  51       }
  52
  53       bufferlist::iterator it = out_bl.begin();
  54       return get_immutable_metadata_finish(&it, object_prefix, order);
  55     }
  56
  57     void get_mutable_metadata_start(librados::ObjectReadOperation *op,
  58                                     bool read_only) {
  59       snapid_t snap = CEPH_NOSNAP;
  60       bufferlist size_bl;
  61       ::encode(snap, size_bl);
  62       op->exec("rbd", "get_size", size_bl);
  63
  64       bufferlist features_bl;
  65       ::encode(snap, features_bl);
  66       ::encode(read_only, features_bl);
  67       op->exec("rbd", "get_features", features_bl);
  68
  69       bufferlist empty_bl;
  70       op->exec("rbd", "get_snapcontext", empty_bl);
  71
  72       bufferlist parent_bl;
  73       ::encode(snap, parent_bl);
  74       op->exec("rbd", "get_parent", parent_bl);
  75
  76       rados::cls::lock::get_lock_info_start(op, RBD_LOCK_NAME);
  77     }
  78
  79     int get_mutable_metadata_finish(bufferlist::iterator *it,
  80                                     uint64_t *size, uint64_t *features,
  81                                     uint64_t *incompatible_features,
  82                                     std::map<rados::cls::lock::locker_id_t,
  83                                              rados::cls::lock::locker_info_t> *lockers,
  84                                     bool *exclusive_lock, std::string *lock_tag,
  85                                     ::SnapContext *snapc, ParentInfo *parent) {
  86       assert(size);
  87       assert(features);
  88       assert(incompatible_features);
  89       assert(lockers);
  90       assert(exclusive_lock);
  91       assert(snapc);
  92       assert(parent);
  93
  94       try {
  95         uint8_t order;
  96         // get_size
  97         ::decode(order, *it);
  98         ::decode(*size, *it);
  99         // get_features
 100         ::decode(*features, *it);
 101         ::decode(*incompatible_features, *it);
 102         // get_snapcontext
 103         ::decode(*snapc, *it);
 104         // get_parent
 105         ::decode(parent->spec.pool_id, *it);
 106         ::decode(parent->spec.image_id, *it);
 107         ::decode(parent->spec.snap_id, *it);
 108         ::decode(parent->overlap, *it);
 109
 110         // get_lock_info
 111         ClsLockType lock_type = LOCK_NONE;
 112         int r = rados::cls::lock::get_lock_info_finish(it, lockers, &lock_type,
 113                                                        lock_tag);
 114         if (r == -EOPNOTSUPP) {
 115           r = 0;
 116         }
 117         if (r == 0) {
 118           *exclusive_lock = (lock_type == LOCK_EXCLUSIVE);
 119         }
 120       } catch (const buffer::error &err) {
 121         return -EBADMSG;
 122       }
 123       return 0;
 124     }
 125
 126     int get_mutable_metadata(librados::IoCtx *ioctx, const std::string &oid,
 127                              bool read_only, uint64_t *size, uint64_t *features,
 128                              uint64_t *incompatible_features,
 129                              map<rados::cls::lock::locker_id_t,
 130                                  rados::cls::lock::locker_info_t> *lockers,
 131                              bool *exclusive_lock,
 132                              string *lock_tag,
 133                              ::SnapContext *snapc,
 134                              ParentInfo *parent)
 135     {
 136       librados::ObjectReadOperation op;
 137       get_mutable_metadata_start(&op, read_only);
 138
 139       bufferlist out_bl;
 140       int r = ioctx->operate(oid, &op, &out_bl);
 141       if (r < 0) {
 142         return r;
 143       }
 144
 145       bufferlist::iterator it = out_bl.begin();
 146       return get_mutable_metadata_finish(&it, size, features,
 147                                          incompatible_features, lockers,
 148                                          exclusive_lock, lock_tag, snapc,
 149                                          parent);
 150     }
 151
 152     void create_image(librados::ObjectWriteOperation *op, uint64_t size,
 153                       uint8_t order, uint64_t features,
 154                       const std::string &object_prefix, int64_t data_pool_id)
 155     {
 156       bufferlist bl;
 157       ::encode(size, bl);
 158       ::encode(order, bl);
 159       ::encode(features, bl);
 160       ::encode(object_prefix, bl);
 161       ::encode(data_pool_id, bl);
 162
 163       op->exec("rbd", "create", bl);
 164     }
 165
 166     int create_image(librados::IoCtx *ioctx, const std::string &oid,
 167                      uint64_t size, uint8_t order, uint64_t features,
 168                      const std::string &object_prefix, int64_t data_pool_id)
 169     {
 170       librados::ObjectWriteOperation op;
 171       create_image(&op, size, order, features, object_prefix, data_pool_id);
 172
 173       return ioctx->operate(oid, &op);
 174     }
 175
 176     int get_features(librados::IoCtx *ioctx, const std::string &oid,
 177                      snapid_t snap_id, uint64_t *features)
 178     {
 179       bufferlist inbl, outbl;
 180       ::encode(snap_id, inbl);
 181
 182       int r = ioctx->exec(oid, "rbd", "get_features", inbl, outbl);
 183       if (r < 0)
 184         return r;
 185
 186       try {
 187         bufferlist::iterator iter = outbl.begin();
 188         ::decode(*features, iter);
 189       } catch (const buffer::error &err) {
 190         return -EBADMSG;
 191       }
 192
 193       return 0;
 194     }
 195
 196     void set_features(librados::ObjectWriteOperation *op, uint64_t features,
 197                       uint64_t mask)
 198     {
 199       bufferlist bl;
 200       ::encode(features, bl);
 201       ::encode(mask, bl);
 202
 203       op->exec("rbd", "set_features", bl);
 204     }
 205
 206     int set_features(librados::IoCtx *ioctx, const std::string &oid,
 207                       uint64_t features, uint64_t mask)
 208     {
 209       librados::ObjectWriteOperation op;
 210       set_features(&op, features, mask);
 211
 212       return ioctx->operate(oid, &op);
 213     }
 214
 215     int get_object_prefix(librados::IoCtx *ioctx, const std::string &oid,
 216                           std::string *object_prefix)
 217     {
 218       bufferlist inbl, outbl;
 219       int r = ioctx->exec(oid, "rbd", "get_object_prefix", inbl, outbl);
 220       if (r < 0)
 221         return r;
 222
 223       try {
 224         bufferlist::iterator iter = outbl.begin();
 225         ::decode(*object_prefix, iter);
 226       } catch (const buffer::error &err) {
 227         return -EBADMSG;
 228       }
 229
 230       return 0;
 231     }
 232
 233     void get_data_pool_start(librados::ObjectReadOperation *op) {
 234       bufferlist bl;
 235       op->exec("rbd", "get_data_pool", bl);
 236     }
 237
 238     int get_data_pool_finish(bufferlist::iterator *it, int64_t *data_pool_id) {
 239       try {
 240         ::decode(*data_pool_id, *it);
 241       } catch (const buffer::error &err) {
 242         return -EBADMSG;
 243       }
 244       return 0;
 245     }
 246
 247     int get_data_pool(librados::IoCtx *ioctx, const std::string &oid,
 248                       int64_t *data_pool_id) {
 249       librados::ObjectReadOperation op;
 250       get_data_pool_start(&op);
 251
 252       bufferlist out_bl;
 253       int r = ioctx->operate(oid, &op, &out_bl);
 254       if (r < 0) {
 255         return r;
 256       }
 257
 258       bufferlist::iterator it = out_bl.begin();
 259       return get_data_pool_finish(&it, data_pool_id);
 260     }
 261
 262     int get_size(librados::IoCtx *ioctx, const std::string &oid,
 263                  snapid_t snap_id, uint64_t *size, uint8_t *order)
 264     {
 265       bufferlist inbl, outbl;
 266       ::encode(snap_id, inbl);
 267
 268       int r = ioctx->exec(oid, "rbd", "get_size", inbl, outbl);
 269       if (r < 0)
 270         return r;
 271
 272       try {
 273         bufferlist::iterator iter = outbl.begin();
 274         ::decode(*order, iter);
 275         ::decode(*size, iter);
 276       } catch (const buffer::error &err) {
 277         return -EBADMSG;
 278       }
 279
 280       return 0;
 281     }
 282
 283     int set_size(librados::IoCtx *ioctx, const std::string &oid,
 284                  uint64_t size)
 285     {
 286       librados::ObjectWriteOperation op;
 287       set_size(&op, size);
 288       return ioctx->operate(oid, &op);
 289     }
 290
 291     void set_size(librados::ObjectWriteOperation *op, uint64_t size)
 292     {
 293       bufferlist bl;
 294       ::encode(size, bl);
 295       op->exec("rbd", "set_size", bl);
 296     }
 297
 298     int get_parent(librados::IoCtx *ioctx, const std::string &oid,
 299                    snapid_t snap_id, ParentSpec *pspec,
 300                    uint64_t *parent_overlap)
 301     {
 302       bufferlist inbl, outbl;
 303       ::encode(snap_id, inbl);
 304
 305       int r = ioctx->exec(oid, "rbd", "get_parent", inbl, outbl);
 306       if (r < 0)
 307         return r;
 308
 309       try {
 310         bufferlist::iterator iter = outbl.begin();
 311         ::decode(pspec->pool_id, iter);
 312         ::decode(pspec->image_id, iter);
 313         ::decode(pspec->snap_id, iter);
 314         ::decode(*parent_overlap, iter);
 315       } catch (const buffer::error &err) {
 316         return -EBADMSG;
 317       }
 318
 319       return 0;
 320     }
 321
 322     int set_parent(librados::IoCtx *ioctx, const std::string &oid,
 323                    const ParentSpec &pspec, uint64_t parent_overlap)
 324     {
 325       librados::ObjectWriteOperation op;
 326       set_parent(&op, pspec, parent_overlap);
 327       return ioctx->operate(oid, &op);
 328     }
 329
 330     void set_parent(librados::ObjectWriteOperation *op,
 331                     const ParentSpec &pspec, uint64_t parent_overlap) {
 332       bufferlist in_bl;
 333       ::encode(pspec.pool_id, in_bl);
 334       ::encode(pspec.image_id, in_bl);
 335       ::encode(pspec.snap_id, in_bl);
 336       ::encode(parent_overlap, in_bl);
 337
 338       op->exec("rbd", "set_parent", in_bl);
 339     }
 340
 341     void get_flags_start(librados::ObjectReadOperation *op,
 342                          const std::vector<snapid_t> &snap_ids) {
 343       bufferlist in_bl;
 344       ::encode(static_cast<snapid_t>(CEPH_NOSNAP), in_bl);
 345
 346       op->exec("rbd", "get_flags", in_bl);
 347       for (size_t i = 0; i < snap_ids.size(); ++i) {
 348         bufferlist snap_bl;
 349         ::encode(snap_ids[i], snap_bl);
 350         op->exec("rbd", "get_flags", snap_bl);
 351       }
 352
 353     }
 354
 355     int get_flags_finish(bufferlist::iterator *it, uint64_t *flags,
 356                          const std::vector<snapid_t> &snap_ids,
 357                          std::vector<uint64_t> *snap_flags) {
 358       snap_flags->resize(snap_ids.size());
 359       try {
 360         ::decode(*flags, *it);
 361         for (size_t i = 0; i < snap_flags->size(); ++i) {
 362           ::decode((*snap_flags)[i], *it);
 363         }
 364       } catch (const buffer::error &err) {
 365         return -EBADMSG;
 366       }
 367       return 0;
 368     }
 369
 370     int get_flags(librados::IoCtx *ioctx, const std::string &oid,
 371                   uint64_t *flags, const std::vector<snapid_t> &snap_ids,
 372                   vector<uint64_t> *snap_flags)
 373     {
 374       librados::ObjectReadOperation op;
 375       get_flags_start(&op, snap_ids);
 376
 377       bufferlist out_bl;
 378       int r = ioctx->operate(oid, &op, &out_bl);
 379       if (r < 0) {
 380         return r;
 381       }
 382
 383       bufferlist::iterator it = out_bl.begin();
 384       return get_flags_finish(&it, flags, snap_ids, snap_flags);
 385     }
 386
 387     void set_flags(librados::ObjectWriteOperation *op, snapid_t snap_id,
 388                    uint64_t flags, uint64_t mask)
 389     {
 390       bufferlist inbl;
 391       ::encode(flags, inbl);
 392       ::encode(mask, inbl);
 393       ::encode(snap_id, inbl);
 394       op->exec("rbd", "set_flags", inbl);
 395     }
 396
 397     int remove_parent(librados::IoCtx *ioctx, const std::string &oid)
 398     {
 399       librados::ObjectWriteOperation op;
 400       remove_parent(&op);
 401       return ioctx->operate(oid, &op);
 402     }
 403
 404     void remove_parent(librados::ObjectWriteOperation *op)
 405     {
 406       bufferlist inbl;
 407       op->exec("rbd", "remove_parent", inbl);
 408     }
 409
 410     int add_child(librados::IoCtx *ioctx, const std::string &oid,
 411                   const ParentSpec &pspec, const std::string &c_imageid)
 412     {
 413       librados::ObjectWriteOperation op;
 414       add_child(&op, pspec, c_imageid);
 415       return ioctx->operate(oid, &op);
 416     }
 417
 418     void add_child(librados::ObjectWriteOperation *op,
 419                   const ParentSpec pspec, const std::string &c_imageid)
 420     {
 421       bufferlist in;
 422       ::encode(pspec.pool_id, in);
 423       ::encode(pspec.image_id, in);
 424       ::encode(pspec.snap_id, in);
 425       ::encode(c_imageid, in);
 426
 427       op->exec("rbd", "add_child", in);
 428     }
 429
 430     void remove_child(librados::ObjectWriteOperation *op,
 431                       const ParentSpec &pspec, const std::string &c_imageid)
 432     {
 433       bufferlist in;
 434       ::encode(pspec.pool_id, in);
 435       ::encode(pspec.image_id, in);
 436       ::encode(pspec.snap_id, in);
 437       ::encode(c_imageid, in);
 438       op->exec("rbd", "remove_child", in);
 439     }
 440
 441     int remove_child(librados::IoCtx *ioctx, const std::string &oid,
 442                      const ParentSpec &pspec, const std::string &c_imageid)
 443     {
 444       librados::ObjectWriteOperation op;
 445       remove_child(&op, pspec, c_imageid);
 446       return ioctx->operate(oid, &op);
 447     }
 448
 449     void get_children_start(librados::ObjectReadOperation *op,
 450                             const ParentSpec &pspec) {
 451       bufferlist in_bl;
 452       ::encode(pspec.pool_id, in_bl);
 453       ::encode(pspec.image_id, in_bl);
 454       ::encode(pspec.snap_id, in_bl);
 455       op->exec("rbd", "get_children", in_bl);
 456     }
 457
 458     int get_children_finish(bufferlist::iterator *it,
 459                             std::set<std::string>* children) {
 460       try {
 461         ::decode(*children, *it);
 462       } catch (const buffer::error &err) {
 463         return -EBADMSG;
 464       }
 465       return 0;
 466     }
 467
 468     int get_children(librados::IoCtx *ioctx, const std::string &oid,
 469                      const ParentSpec &pspec, set<string>& children)
 470     {
 471       librados::ObjectReadOperation op;
 472       get_children_start(&op, pspec);
 473
 474       bufferlist out_bl;
 475       int r = ioctx->operate(oid, &op, &out_bl);
 476       if (r < 0) {
 477         return r;
 478       }
 479
 480       bufferlist::iterator it = out_bl.begin();
 481       return get_children_finish(&it, &children);
 482     }
 483
 484     void snapshot_add(librados::ObjectWriteOperation *op, snapid_t snap_id,
 485                       const std::string &snap_name, const cls::rbd::SnapshotNamespace &snap_namespace)
 486     {
 487       bufferlist bl;
 488       ::encode(snap_name, bl);
 489       ::encode(snap_id, bl);
 490       ::encode(cls::rbd::SnapshotNamespaceOnDisk(snap_namespace), bl);
 491       op->exec("rbd", "snapshot_add", bl);
 492     }
 493
 494     void snapshot_remove(librados::ObjectWriteOperation *op, snapid_t snap_id)
 495     {
 496       bufferlist bl;
 497       ::encode(snap_id, bl);
 498       op->exec("rbd", "snapshot_remove", bl);
 499     }
 500
 501     void snapshot_rename(librados::ObjectWriteOperation *op,
 502                          snapid_t src_snap_id,
 503                          const std::string &dst_name)
 504     {
 505       bufferlist bl;
 506       ::encode(src_snap_id, bl);
 507       ::encode(dst_name, bl);
 508       op->exec("rbd", "snapshot_rename", bl);
 509     }
 510
 511     int get_snapcontext(librados::IoCtx *ioctx, const std::string &oid,
 512                         ::SnapContext *snapc)
 513     {
 514       bufferlist inbl, outbl;
 515
 516       int r = ioctx->exec(oid, "rbd", "get_snapcontext", inbl, outbl);
 517       if (r < 0)
 518         return r;
 519
 520       try {
 521         bufferlist::iterator iter = outbl.begin();
 522         ::decode(*snapc, iter);
 523       } catch (const buffer::error &err) {
 524         return -EBADMSG;
 525       }
 526
 527       if (!snapc->is_valid())
 528         return -EBADMSG;
 529
 530       return 0;
 531     }
 532
 533     void snapshot_list_start(librados::ObjectReadOperation *op,
 534                              const std::vector<snapid_t> &ids) {
 535       for (auto snap_id : ids) {
 536         bufferlist bl1, bl2, bl3, bl4;
 537         ::encode(snap_id, bl1);
 538         op->exec("rbd", "get_snapshot_name", bl1);
 539         ::encode(snap_id, bl2);
 540         op->exec("rbd", "get_size", bl2);
 541         ::encode(snap_id, bl3);
 542         op->exec("rbd", "get_parent", bl3);
 543         ::encode(snap_id, bl4);
 544         op->exec("rbd", "get_protection_status", bl4);
 545       }
 546     }
 547
 548     int snapshot_list_finish(bufferlist::iterator *it,
 549                              const std::vector<snapid_t> &ids,
 550                              std::vector<string> *names,
 551                              std::vector<uint64_t> *sizes,
 552                              std::vector<ParentInfo> *parents,
 553                              std::vector<uint8_t> *protection_statuses)
 554     {
 555       names->resize(ids.size());
 556       sizes->resize(ids.size());
 557       parents->resize(ids.size());
 558       protection_statuses->resize(ids.size());
 559       try {
 560         for (size_t i = 0; i < names->size(); ++i) {
 561           uint8_t order;
 562           // get_snapshot_name
 563           ::decode((*names)[i], *it);
 564           // get_size
 565           ::decode(order, *it);
 566           ::decode((*sizes)[i], *it);
 567           // get_parent
 568           ::decode((*parents)[i].spec.pool_id, *it);
 569           ::decode((*parents)[i].spec.image_id, *it);
 570           ::decode((*parents)[i].spec.snap_id, *it);
 571           ::decode((*parents)[i].overlap, *it);
 572           // get_protection_status
 573           ::decode((*protection_statuses)[i], *it);
 574         }
 575       } catch (const buffer::error &err) {
 576         return -EBADMSG;
 577       }
 578       return 0;
 579     }
 580
 581     int snapshot_list(librados::IoCtx *ioctx, const std::string &oid,
 582                       const std::vector<snapid_t> &ids,
 583                       std::vector<string> *names,
 584                       std::vector<uint64_t> *sizes,
 585                       std::vector<ParentInfo> *parents,
 586                       std::vector<uint8_t> *protection_statuses)
 587     {
 588       librados::ObjectReadOperation op;
 589       snapshot_list_start(&op, ids);
 590
 591       bufferlist out_bl;
 592       int r = ioctx->operate(oid, &op, &out_bl);
 593       if (r < 0) {
 594         return r;
 595       }
 596
 597       bufferlist::iterator it = out_bl.begin();
 598       return snapshot_list_finish(&it, ids, names, sizes, parents,
 599                                   protection_statuses);
 600     }
 601
 602     void snapshot_timestamp_list_start(librados::ObjectReadOperation *op,
 603                                        const std::vector<snapid_t> &ids)
 604     {
 605       for (auto snap_id : ids) {
 606         bufferlist bl;
 607         ::encode(snap_id, bl);
 608         op->exec("rbd", "get_snapshot_timestamp", bl);
 609       }
 610     }
 611
 612     int snapshot_timestamp_list_finish(bufferlist::iterator *it,
 613                                        const std::vector<snapid_t> &ids,
 614                                        std::vector<utime_t> *timestamps)
 615     {
 616       timestamps->resize(ids.size());
 617       try {
 618         for (size_t i = 0; i < timestamps->size(); ++i) {
 619           utime_t t;
 620           ::decode(t, *it);
 621           (*timestamps)[i] = t;
 622         }
 623       } catch (const buffer::error &err) {
 624         return -EBADMSG;
 625       }
 626       return 0;
 627     }
 628
 629     int snapshot_timestamp_list(librados::IoCtx *ioctx, const std::string &oid,
 630                                 const std::vector<snapid_t> &ids,
 631                                 std::vector<utime_t> *timestamps)
 632     {
 633       librados::ObjectReadOperation op;
 634       snapshot_timestamp_list_start(&op, ids);
 635
 636       bufferlist out_bl;
 637       int r = ioctx->operate(oid, &op, &out_bl);
 638       if (r < 0) {
 639         return r;
 640       }
 641
 642       bufferlist::iterator it = out_bl.begin();
 643       return snapshot_timestamp_list_finish(&it, ids, timestamps);
 644     }
 645
 646     void snapshot_namespace_list_start(librados::ObjectReadOperation *op,
 647                                        const std::vector<snapid_t> &ids)
 648     {
 649       for (auto snap_id : ids) {
 650         bufferlist bl;
 651         ::encode(snap_id, bl);
 652         op->exec("rbd", "get_snapshot_namespace", bl);
 653       }
 654     }
 655
 656     int snapshot_namespace_list_finish(bufferlist::iterator *it,
 657                                        const std::vector<snapid_t> &ids,
 658                                        std::vector<cls::rbd::SnapshotNamespace> *namespaces)
 659     {
 660       namespaces->resize(ids.size());
 661       try {
 662         for (size_t i = 0; i < namespaces->size(); ++i) {
 663           cls::rbd::SnapshotNamespaceOnDisk e;
 664           ::decode(e, *it);
 665           (*namespaces)[i] = e.snapshot_namespace;
 666         }
 667       } catch (const buffer::error &err) {
 668         return -EBADMSG;
 669       }
 670       return 0;
 671     }
 672
 673     int snapshot_namespace_list(librados::IoCtx *ioctx, const std::string &oid,
 674                                 const std::vector<snapid_t> &ids,
 675                                 std::vector<cls::rbd::SnapshotNamespace> *namespaces)
 676     {
 677       librados::ObjectReadOperation op;
 678       snapshot_namespace_list_start(&op, ids);
 679
 680       bufferlist out_bl;
 681       int r = ioctx->operate(oid, &op, &out_bl);
 682       if (r < 0) {
 683         return r;
 684       }
 685
 686       bufferlist::iterator it = out_bl.begin();
 687       return snapshot_namespace_list_finish(&it, ids, namespaces);
 688     }
 689
 690     void old_snapshot_add(librados::ObjectWriteOperation *op,
 691                           snapid_t snap_id, const std::string &snap_name)
 692     {
 693       bufferlist bl;
 694       ::encode(snap_name, bl);
 695       ::encode(snap_id, bl);
 696       op->exec("rbd", "snap_add", bl);
 697     }
 698
 699     void old_snapshot_remove(librados::ObjectWriteOperation *op,
 700                              const std::string &snap_name)
 701     {
 702       bufferlist bl;
 703       ::encode(snap_name, bl);
 704       op->exec("rbd", "snap_remove", bl);
 705     }
 706
 707     void old_snapshot_rename(librados::ObjectWriteOperation *op,
 708                              snapid_t src_snap_id, const std::string &dst_name)
 709     {
 710       bufferlist bl;
 711       ::encode(src_snap_id, bl);
 712       ::encode(dst_name, bl);
 713       op->exec("rbd", "snap_rename", bl);
 714     }
 715
 716     void old_snapshot_list_start(librados::ObjectReadOperation *op) {
 717       bufferlist in_bl;
 718       op->exec("rbd", "snap_list", in_bl);
 719     }
 720
 721     int old_snapshot_list_finish(bufferlist::iterator *it,
 722                                  std::vector<string> *names,
 723                                  std::vector<uint64_t> *sizes,
 724                                  ::SnapContext *snapc) {
 725       try {
 726         uint32_t num_snaps;
 727         ::decode(snapc->seq, *it);
 728         ::decode(num_snaps, *it);
 729
 730         names->resize(num_snaps);
 731         sizes->resize(num_snaps);
 732         snapc->snaps.resize(num_snaps);
 733         for (uint32_t i = 0; i < num_snaps; ++i) {
 734           ::decode(snapc->snaps[i], *it);
 735           ::decode((*sizes)[i], *it);
 736           ::decode((*names)[i], *it);
 737         }
 738       } catch (const buffer::error &err) {
 739         return -EBADMSG;
 740       }
 741       return 0;
 742     }
 743
 744     int old_snapshot_list(librados::IoCtx *ioctx, const std::string &oid,
 745                           std::vector<string> *names,
 746                           std::vector<uint64_t> *sizes,
 747                           ::SnapContext *snapc)
 748     {
 749       librados::ObjectReadOperation op;
 750       old_snapshot_list_start(&op);
 751
 752       bufferlist out_bl;
 753       int r = ioctx->operate(oid, &op, &out_bl);
 754       if (r < 0) {
 755         return r;
 756       }
 757
 758       bufferlist::iterator it = out_bl.begin();
 759       return old_snapshot_list_finish(&it, names, sizes, snapc);
 760     }
 761
 762     void get_all_features_start(librados::ObjectReadOperation *op) {
 763       bufferlist in;
 764       op->exec("rbd", "get_all_features", in);
 765     }
 766
 767     int get_all_features_finish(bufferlist::iterator *it,
 768                                 uint64_t *all_features) {
 769       try {
 770         ::decode(*all_features, *it);
 771       } catch (const buffer::error &err) {
 772         return -EBADMSG;
 773       }
 774       return 0;
 775     }
 776
 777     int get_all_features(librados::IoCtx *ioctx, const std::string &oid,
 778                          uint64_t *all_features) {
 779       librados::ObjectReadOperation op;
 780       get_all_features_start(&op);
 781
 782       bufferlist out_bl;
 783       int r = ioctx->operate(oid, &op, &out_bl);
 784       if (r < 0) {
 785         return r;
 786       }
 787
 788       bufferlist::iterator it = out_bl.begin();
 789       return get_all_features_finish(&it, all_features);
 790     }
 791
 792     int copyup(librados::IoCtx *ioctx, const std::string &oid,
 793                bufferlist data) {
 794       bufferlist out;
 795       return ioctx->exec(oid, "rbd", "copyup", data, out);
 796     }
 797
 798     int get_protection_status(librados::IoCtx *ioctx, const std::string &oid,
 799                               snapid_t snap_id, uint8_t *protection_status)
 800     {
 801       bufferlist in, out;
 802       ::encode(snap_id.val, in);
 803
 804       int r = ioctx->exec(oid, "rbd", "get_protection_status", in, out);
 805       if (r < 0)
 806         return r;
 807
 808       try {
 809         bufferlist::iterator iter = out.begin();
 810         ::decode(*protection_status, iter);
 811       } catch (const buffer::error &err) {
 812         return -EBADMSG;
 813       }
 814
 815       return 0;
 816     }
 817
 818     int set_protection_status(librados::IoCtx *ioctx, const std::string &oid,
 819                               snapid_t snap_id, uint8_t protection_status)
 820     {
 821       // TODO remove
 822       librados::ObjectWriteOperation op;
 823       set_protection_status(&op, snap_id, protection_status);
 824       return ioctx->operate(oid, &op);
 825     }
 826
 827     void set_protection_status(librados::ObjectWriteOperation *op,
 828                                snapid_t snap_id, uint8_t protection_status)
 829     {
 830       bufferlist in;
 831       ::encode(snap_id, in);
 832       ::encode(protection_status, in);
 833       op->exec("rbd", "set_protection_status", in);
 834     }
 835
 836     int snapshot_get_limit(librados::IoCtx *ioctx, const std::string &oid,
 837                            uint64_t *limit)
 838     {
 839       bufferlist in, out;
 840       int r =  ioctx->exec(oid, "rbd", "snapshot_get_limit", in, out);
 841
 842       if (r < 0) {
 843         return r;
 844       }
 845
 846       try {
 847         bufferlist::iterator iter = out.begin();
 848         ::decode(*limit, iter);
 849       } catch (const buffer::error &err) {
 850         return -EBADMSG;
 851       }
 852
 853       return 0;
 854     }
 855
 856     void snapshot_set_limit(librados::ObjectWriteOperation *op, uint64_t limit)
 857     {
 858       bufferlist in;
 859       ::encode(limit, in);
 860       op->exec("rbd", "snapshot_set_limit", in);
 861     }
 862
 863     void get_stripe_unit_count_start(librados::ObjectReadOperation *op) {
 864       bufferlist empty_bl;
 865       op->exec("rbd", "get_stripe_unit_count", empty_bl);
 866     }
 867
 868     int get_stripe_unit_count_finish(bufferlist::iterator *it,
 869                                      uint64_t *stripe_unit,
 870                                      uint64_t *stripe_count) {
 871       assert(stripe_unit);
 872       assert(stripe_count);
 873
 874       try {
 875         ::decode(*stripe_unit, *it);
 876         ::decode(*stripe_count, *it);
 877       } catch (const buffer::error &err) {
 878         return -EBADMSG;
 879       }
 880       return 0;
 881     }
 882
 883     int get_stripe_unit_count(librados::IoCtx *ioctx, const std::string &oid,
 884                               uint64_t *stripe_unit, uint64_t *stripe_count)
 885     {
 886       librados::ObjectReadOperation op;
 887       get_stripe_unit_count_start(&op);
 888
 889       bufferlist out_bl;
 890       int r = ioctx->operate(oid, &op, &out_bl);
 891       if (r < 0) {
 892         return r;
 893       }
 894
 895       bufferlist::iterator it = out_bl.begin();
 896       return get_stripe_unit_count_finish(&it, stripe_unit, stripe_count);
 897     }
 898
 899     void set_stripe_unit_count(librados::ObjectWriteOperation *op,
 900                                uint64_t stripe_unit, uint64_t stripe_count)
 901     {
 902       bufferlist bl;
 903       ::encode(stripe_unit, bl);
 904       ::encode(stripe_count, bl);
 905
 906       op->exec("rbd", "set_stripe_unit_count", bl);
 907     }
 908
 909     int set_stripe_unit_count(librados::IoCtx *ioctx, const std::string &oid,
 910                               uint64_t stripe_unit, uint64_t stripe_count)
 911     {
 912       librados::ObjectWriteOperation op;
 913       set_stripe_unit_count(&op, stripe_unit, stripe_count);
 914
 915       return ioctx->operate(oid, &op);
 916     }
 917
 918     void get_create_timestamp_start(librados::ObjectReadOperation *op) {
 919       bufferlist empty_bl;
 920       op->exec("rbd", "get_create_timestamp", empty_bl);
 921     }
 922
 923     int get_create_timestamp_finish(bufferlist::iterator *it,
 924                                     utime_t *timestamp) {
 925       assert(timestamp);
 926
 927       try {
 928         ::decode(*timestamp, *it);
 929       } catch (const buffer::error &err) {
 930         return -EBADMSG;
 931       }
 932       return 0;
 933     }
 934
 935     int get_create_timestamp(librados::IoCtx *ioctx, const std::string &oid,
 936                              utime_t *timestamp)
 937     {
 938       librados::ObjectReadOperation op;
 939       get_create_timestamp_start(&op);
 940
 941       bufferlist out_bl;
 942       int r = ioctx->operate(oid, &op, &out_bl);
 943       if (r < 0) {
 944         return r;
 945       }
 946
 947       bufferlist::iterator it = out_bl.begin();
 948       return get_create_timestamp_finish(&it, timestamp);
 949     }
 950
 951     /************************ rbd_id object methods ************************/
 952
 953     void get_id_start(librados::ObjectReadOperation *op) {
 954       bufferlist empty_bl;
 955       op->exec("rbd", "get_id", empty_bl);
 956     }
 957
 958     int get_id_finish(bufferlist::iterator *it, std::string *id) {
 959       try {
 960         ::decode(*id, *it);
 961       } catch (const buffer::error &err) {
 962         return -EBADMSG;
 963       }
 964       return 0;
 965     }
 966
 967     int get_id(librados::IoCtx *ioctx, const std::string &oid, std::string *id)
 968     {
 969       librados::ObjectReadOperation op;
 970       get_id_start(&op);
 971
 972       bufferlist out_bl;
 973       int r = ioctx->operate(oid, &op, &out_bl);
 974       if (r < 0) {
 975         return r;
 976       }
 977
 978       bufferlist::iterator it = out_bl.begin();
 979       return get_id_finish(&it, id);
 980     }
 981
 982     void set_id(librados::ObjectWriteOperation *op, const std::string id)
 983     {
 984       bufferlist bl;
 985       ::encode(id, bl);
 986       op->exec("rbd", "set_id", bl);
 987     }
 988
 989     int set_id(librados::IoCtx *ioctx, const std::string &oid, std::string id)
 990     {
 991       librados::ObjectWriteOperation op;
 992       set_id(&op, id);
 993
 994       return ioctx->operate(oid, &op);
 995     }
 996
 997     /******************** rbd_directory object methods ********************/
 998
 999     void dir_get_id_start(librados::ObjectReadOperation *op,
1000                           const std::string &image_name) {
1001       bufferlist bl;
1002       ::encode(image_name, bl);
1003
1004       op->exec("rbd", "dir_get_id", bl);
1005     }
1006
1007     int dir_get_id_finish(bufferlist::iterator *iter, std::string *image_id) {
1008       try {
1009         ::decode(*image_id, *iter);
1010       } catch (const buffer::error &err) {
1011         return -EBADMSG;
1012       }
1013
1014       return 0;
1015     }
1016
1017     int dir_get_id(librados::IoCtx *ioctx, const std::string &oid,
1018                    const std::string &name, std::string *id) {
1019       librados::ObjectReadOperation op;
1020       dir_get_id_start(&op, name);
1021
1022       bufferlist out_bl;
1023       int r = ioctx->operate(oid, &op, &out_bl);
1024       if (r < 0) {
1025         return r;
1026       }
1027
1028       bufferlist::iterator iter = out_bl.begin();
1029       return dir_get_id_finish(&iter, id);
1030     }
1031
1032     void dir_get_name_start(librados::ObjectReadOperation *op,
1033                             const std::string &id) {
1034       bufferlist in_bl;
1035       ::encode(id, in_bl);
1036       op->exec("rbd", "dir_get_name", in_bl);
1037     }
1038
1039     int dir_get_name_finish(bufferlist::iterator *it, std::string *name) {
1040       try {
1041         ::decode(*name, *it);
1042       } catch (const buffer::error &err) {
1043         return -EBADMSG;
1044       }
1045       return 0;
1046     }
1047
1048     int dir_get_name(librados::IoCtx *ioctx, const std::string &oid,
1049                      const std::string &id, std::string *name) {
1050       librados::ObjectReadOperation op;
1051       dir_get_name_start(&op, id);
1052
1053       bufferlist out_bl;
1054       int r = ioctx->operate(oid, &op, &out_bl);
1055       if (r < 0) {
1056         return r;
1057       }
1058
1059       bufferlist::iterator it = out_bl.begin();
1060       return dir_get_name_finish(&it, name);
1061     }
1062
1063     void dir_list_start(librados::ObjectReadOperation *op,
1064                         const std::string &start, uint64_t max_return)
1065     {
1066       bufferlist in_bl;
1067       ::encode(start, in_bl);
1068       ::encode(max_return, in_bl);
1069
1070       op->exec("rbd", "dir_list", in_bl);
1071     }
1072
1073     int dir_list_finish(bufferlist::iterator *it, map<string, string> *images)
1074     {
1075       try {
1076         ::decode(*images, *it);
1077       } catch (const buffer::error &err) {
1078         return -EBADMSG;
1079       }
1080       return 0;
1081     }
1082
1083     int dir_list(librados::IoCtx *ioctx, const std::string &oid,
1084                  const std::string &start, uint64_t max_return,
1085                  map<string, string> *images)
1086     {
1087       librados::ObjectReadOperation op;
1088       dir_list_start(&op, start, max_return);
1089
1090       bufferlist out_bl;
1091       int r = ioctx->operate(oid, &op, &out_bl);
1092       if (r < 0) {
1093         return r;
1094       }
1095
1096       bufferlist::iterator iter = out_bl.begin();
1097       return dir_list_finish(&iter, images);
1098     }
1099
1100     void dir_add_image(librados::ObjectWriteOperation *op,
1101                        const std::string &name, const std::string &id)
1102     {
1103       bufferlist bl;
1104       ::encode(name, bl);
1105       ::encode(id, bl);
1106       op->exec("rbd", "dir_add_image", bl);
1107     }
1108
1109     int dir_add_image(librados::IoCtx *ioctx, const std::string &oid,
1110                       const std::string &name, const std::string &id)
1111     {
1112       librados::ObjectWriteOperation op;
1113       dir_add_image(&op, name, id);
1114
1115       return ioctx->operate(oid, &op);
1116     }
1117
1118     int dir_remove_image(librados::IoCtx *ioctx, const std::string &oid,
1119                          const std::string &name, const std::string &id)
1120     {
1121       librados::ObjectWriteOperation op;
1122       dir_remove_image(&op, name, id);
1123
1124       return ioctx->operate(oid, &op);
1125     }
1126
1127     void dir_remove_image(librados::ObjectWriteOperation *op,
1128                           const std::string &name, const std::string &id)
1129     {
1130       bufferlist bl;
1131       ::encode(name, bl);
1132       ::encode(id, bl);
1133
1134       op->exec("rbd", "dir_remove_image", bl);
1135     }
1136
1137     void dir_rename_image(librados::ObjectWriteOperation *op,
1138                          const std::string &src, const std::string &dest,
1139                          const std::string &id)
1140     {
1141       bufferlist in;
1142       ::encode(src, in);
1143       ::encode(dest, in);
1144       ::encode(id, in);
1145       op->exec("rbd", "dir_rename_image", in);
1146     }
1147
1148     void object_map_load_start(librados::ObjectReadOperation *op) {
1149       bufferlist in_bl;
1150       op->exec("rbd", "object_map_load", in_bl);
1151     }
1152
1153     int object_map_load_finish(bufferlist::iterator *it,
1154                                ceph::BitVector<2> *object_map) {
1155       try {
1156         ::decode(*object_map, *it);
1157       } catch (const buffer::error &err) {
1158         return -EBADMSG;
1159       }
1160       return 0;
1161     }
1162
1163     int object_map_load(librados::IoCtx *ioctx, const std::string &oid,
1164                         ceph::BitVector<2> *object_map)
1165     {
1166       librados::ObjectReadOperation op;
1167       object_map_load_start(&op);
1168
1169       bufferlist out_bl;
1170       int r = ioctx->operate(oid, &op, &out_bl);
1171       if (r < 0) {
1172         return r;
1173       }
1174
1175       bufferlist::iterator it = out_bl.begin();
1176       return object_map_load_finish(&it, object_map);
1177     }
1178
1179     void object_map_save(librados::ObjectWriteOperation *rados_op,
1180                          const ceph::BitVector<2> &object_map)
1181     {
1182       ceph::BitVector<2> object_map_copy(object_map);
1183       object_map_copy.set_crc_enabled(false);
1184
1185       bufferlist in;
1186       ::encode(object_map_copy, in);
1187       rados_op->exec("rbd", "object_map_save", in);
1188     }
1189
1190     void object_map_resize(librados::ObjectWriteOperation *rados_op,
1191                            uint64_t object_count, uint8_t default_state)
1192     {
1193       bufferlist in;
1194       ::encode(object_count, in);
1195       ::encode(default_state, in);
1196       rados_op->exec("rbd", "object_map_resize", in);
1197     }
1198
1199     void object_map_update(librados::ObjectWriteOperation *rados_op,
1200                            uint64_t start_object_no, uint64_t end_object_no,
1201                            uint8_t new_object_state,
1202                            const boost::optional<uint8_t> &current_object_state)
1203     {
1204       bufferlist in;
1205       ::encode(start_object_no, in);
1206       ::encode(end_object_no, in);
1207       ::encode(new_object_state, in);
1208       ::encode(current_object_state, in);
1209       rados_op->exec("rbd", "object_map_update", in);
1210     }
1211
1212     void object_map_snap_add(librados::ObjectWriteOperation *rados_op)
1213     {
1214       bufferlist in;
1215       rados_op->exec("rbd", "object_map_snap_add", in);
1216     }
1217
1218     void object_map_snap_remove(librados::ObjectWriteOperation *rados_op,
1219                                 const ceph::BitVector<2> &object_map)
1220     {
1221       ceph::BitVector<2> object_map_copy(object_map);
1222       object_map_copy.set_crc_enabled(false);
1223
1224       bufferlist in;
1225       ::encode(object_map_copy, in);
1226       rados_op->exec("rbd", "object_map_snap_remove", in);
1227     }
1228
1229     void metadata_set(librados::ObjectWriteOperation *op,
1230                      const map<string, bufferlist> &data)
1231     {
1232       bufferlist bl;
1233       ::encode(data, bl);
1234
1235       op->exec("rbd", "metadata_set", bl);
1236     }
1237
1238     int metadata_set(librados::IoCtx *ioctx, const std::string &oid,
1239                      const map<string, bufferlist> &data)
1240     {
1241       librados::ObjectWriteOperation op;
1242       metadata_set(&op, data);
1243
1244       return ioctx->operate(oid, &op);
1245     }
1246
1247     void metadata_remove(librados::ObjectWriteOperation *op,
1248                          const std::string &key)
1249     {
1250       bufferlist bl;
1251       ::encode(key, bl);
1252
1253       op->exec("rbd", "metadata_remove", bl);
1254     }
1255
1256     int metadata_remove(librados::IoCtx *ioctx, const std::string &oid,
1257                      const std::string &key)
1258     {
1259       librados::ObjectWriteOperation op;
1260       metadata_remove(&op, key);
1261
1262       return ioctx->operate(oid, &op);
1263     }
1264
1265     int metadata_list(librados::IoCtx *ioctx, const std::string &oid,
1266                       const std::string &start, uint64_t max_return,
1267                       map<string, bufferlist> *pairs)
1268     {
1269       librados::ObjectReadOperation op;
1270       metadata_list_start(&op, start, max_return);
1271
1272       bufferlist out_bl;
1273       int r = ioctx->operate(oid, &op, &out_bl);
1274       if (r < 0) {
1275         return r;
1276       }
1277
1278       bufferlist::iterator it = out_bl.begin();
1279       return metadata_list_finish(&it, pairs);
1280     }
1281
1282     void metadata_list_start(librados::ObjectReadOperation *op,
1283                              const std::string &start, uint64_t max_return)
1284     {
1285       bufferlist in_bl;
1286       ::encode(start, in_bl);
1287       ::encode(max_return, in_bl);
1288       op->exec("rbd", "metadata_list", in_bl);
1289     }
1290
1291     int metadata_list_finish(bufferlist::iterator *it,
1292                              std::map<std::string, bufferlist> *pairs)
1293     {
1294       assert(pairs);
1295       try {
1296         ::decode(*pairs, *it);
1297       } catch (const buffer::error &err) {
1298         return -EBADMSG;
1299       }
1300       return 0;
1301     }
1302
1303     int metadata_get(librados::IoCtx *ioctx, const std::string &oid,
1304                      const std::string &key, string *s)
1305     {
1306       assert(s);
1307       bufferlist in, out;
1308       ::encode(key, in);
1309       int r = ioctx->exec(oid, "rbd", "metadata_get", in, out);
1310       if (r < 0)
1311         return r;
1312
1313       bufferlist::iterator iter = out.begin();
1314       try {
1315         ::decode(*s, iter);
1316       } catch (const buffer::error &err) {
1317         return -EBADMSG;
1318       }
1319
1320       return 0;
1321     }
1322
1323     void mirror_uuid_get_start(librados::ObjectReadOperation *op) {
1324       bufferlist bl;
1325       op->exec("rbd", "mirror_uuid_get", bl);
1326     }
1327
1328     int mirror_uuid_get_finish(bufferlist::iterator *it,
1329                                std::string *uuid) {
1330       try {
1331         ::decode(*uuid, *it);
1332       } catch (const buffer::error &err) {
1333         return -EBADMSG;
1334       }
1335       return 0;
1336     }
1337
1338     int mirror_uuid_get(librados::IoCtx *ioctx, std::string *uuid) {
1339       librados::ObjectReadOperation op;
1340       mirror_uuid_get_start(&op);
1341
1342       bufferlist out_bl;
1343       int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1344       if (r < 0) {
1345         return r;
1346       }
1347
1348       bufferlist::iterator it = out_bl.begin();
1349       r = mirror_uuid_get_finish(&it, uuid);
1350       if (r < 0) {
1351         return r;
1352       }
1353       return 0;
1354     }
1355
1356     int mirror_uuid_set(librados::IoCtx *ioctx, const std::string &uuid) {
1357       bufferlist in_bl;
1358       ::encode(uuid, in_bl);
1359
1360       bufferlist out_bl;
1361       int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_uuid_set", in_bl,
1362                           out_bl);
1363       if (r < 0) {
1364         return r;
1365       }
1366       return 0;
1367     }
1368
1369     void mirror_mode_get_start(librados::ObjectReadOperation *op) {
1370       bufferlist bl;
1371       op->exec("rbd", "mirror_mode_get", bl);
1372     }
1373
1374     int mirror_mode_get_finish(bufferlist::iterator *it,
1375                                cls::rbd::MirrorMode *mirror_mode) {
1376       try {
1377         uint32_t mirror_mode_decode;
1378         ::decode(mirror_mode_decode, *it);
1379         *mirror_mode = static_cast<cls::rbd::MirrorMode>(mirror_mode_decode);
1380       } catch (const buffer::error &err) {
1381         return -EBADMSG;
1382       }
1383
1384       return 0;
1385     }
1386
1387     int mirror_mode_get(librados::IoCtx *ioctx,
1388                         cls::rbd::MirrorMode *mirror_mode) {
1389       librados::ObjectReadOperation op;
1390       mirror_mode_get_start(&op);
1391
1392       bufferlist out_bl;
1393       int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1394       if (r == -ENOENT) {
1395         *mirror_mode = cls::rbd::MIRROR_MODE_DISABLED;
1396         return 0;
1397       } else if (r < 0) {
1398         return r;
1399       }
1400
1401       bufferlist::iterator it = out_bl.begin();
1402       r = mirror_mode_get_finish(&it, mirror_mode);
1403       if (r < 0) {
1404         return r;
1405       }
1406       return 0;
1407     }
1408
1409     int mirror_mode_set(librados::IoCtx *ioctx,
1410                         cls::rbd::MirrorMode mirror_mode) {
1411       bufferlist in_bl;
1412       ::encode(static_cast<uint32_t>(mirror_mode), in_bl);
1413
1414       bufferlist out_bl;
1415       int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_mode_set", in_bl,
1416                           out_bl);
1417       if (r < 0) {
1418         return r;
1419       }
1420       return 0;
1421     }
1422
1423     int mirror_peer_list(librados::IoCtx *ioctx,
1424                          std::vector<cls::rbd::MirrorPeer> *peers) {
1425       bufferlist in_bl;
1426       bufferlist out_bl;
1427       int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_list", in_bl,
1428                           out_bl);
1429       if (r < 0) {
1430         return r;
1431       }
1432
1433       peers->clear();
1434       try {
1435         bufferlist::iterator bl_it = out_bl.begin();
1436         ::decode(*peers, bl_it);
1437       } catch (const buffer::error &err) {
1438         return -EBADMSG;
1439       }
1440       return 0;
1441     }
1442
1443     int mirror_peer_add(librados::IoCtx *ioctx, const std::string &uuid,
1444                         const std::string &cluster_name,
1445                         const std::string &client_name, int64_t pool_id) {
1446       cls::rbd::MirrorPeer peer(uuid, cluster_name, client_name, pool_id);
1447       bufferlist in_bl;
1448       ::encode(peer, in_bl);
1449
1450       bufferlist out_bl;
1451       int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_add", in_bl,
1452                           out_bl);
1453       if (r < 0) {
1454         return r;
1455       }
1456       return 0;
1457     }
1458
1459     int mirror_peer_remove(librados::IoCtx *ioctx,
1460                            const std::string &uuid) {
1461       bufferlist in_bl;
1462       ::encode(uuid, in_bl);
1463
1464       bufferlist out_bl;
1465       int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_remove", in_bl,
1466                           out_bl);
1467       if (r < 0) {
1468         return r;
1469       }
1470       return 0;
1471     }
1472
1473     int mirror_peer_set_client(librados::IoCtx *ioctx,
1474                                const std::string &uuid,
1475                                const std::string &client_name) {
1476       bufferlist in_bl;
1477       ::encode(uuid, in_bl);
1478       ::encode(client_name, in_bl);
1479
1480       bufferlist out_bl;
1481       int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_set_client",
1482                           in_bl, out_bl);
1483       if (r < 0) {
1484         return r;
1485       }
1486       return 0;
1487     }
1488
1489     int mirror_peer_set_cluster(librados::IoCtx *ioctx,
1490                                 const std::string &uuid,
1491                                 const std::string &cluster_name) {
1492       bufferlist in_bl;
1493       ::encode(uuid, in_bl);
1494       ::encode(cluster_name, in_bl);
1495
1496       bufferlist out_bl;
1497       int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_set_cluster",
1498                           in_bl, out_bl);
1499       if (r < 0) {
1500         return r;
1501       }
1502       return 0;
1503     }
1504
1505     void mirror_image_list_start(librados::ObjectReadOperation *op,
1506                                  const std::string &start, uint64_t max_return)
1507     {
1508       bufferlist in_bl;
1509       ::encode(start, in_bl);
1510       ::encode(max_return, in_bl);
1511       op->exec("rbd", "mirror_image_list", in_bl);
1512     }
1513
1514     int mirror_image_list_finish(bufferlist::iterator *it,
1515                                  std::map<string, string> *mirror_image_ids)
1516     {
1517       try {
1518         ::decode(*mirror_image_ids, *it);
1519       } catch (const buffer::error &err) {
1520         return -EBADMSG;
1521       }
1522       return 0;
1523     }
1524
1525     int mirror_image_list(librados::IoCtx *ioctx,
1526                           const std::string &start, uint64_t max_return,
1527                           std::map<std::string, std::string> *mirror_image_ids) {
1528       librados::ObjectReadOperation op;
1529       mirror_image_list_start(&op, start, max_return);
1530
1531       bufferlist out_bl;
1532       int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1533       if (r < 0) {
1534         return r;
1535       }
1536
1537       bufferlist::iterator bl_it = out_bl.begin();
1538       return mirror_image_list_finish(&bl_it, mirror_image_ids);
1539     }
1540
1541     void mirror_image_get_image_id_start(librados::ObjectReadOperation *op,
1542                                          const std::string &global_image_id) {
1543       bufferlist in_bl;
1544       ::encode(global_image_id, in_bl);
1545       op->exec( "rbd", "mirror_image_get_image_id", in_bl);
1546     }
1547
1548     int mirror_image_get_image_id_finish(bufferlist::iterator *it,
1549                                          std::string *image_id) {
1550       try {
1551         ::decode(*image_id, *it);
1552       } catch (const buffer::error &err) {
1553         return -EBADMSG;
1554       }
1555       return 0;
1556     }
1557
1558     int mirror_image_get_image_id(librados::IoCtx *ioctx,
1559                                   const std::string &global_image_id,
1560                                   std::string *image_id) {
1561       librados::ObjectReadOperation op;
1562       mirror_image_get_image_id_start(&op, global_image_id);
1563
1564       bufferlist out_bl;
1565       int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1566       if (r < 0) {
1567         return r;
1568       }
1569
1570       bufferlist::iterator it = out_bl.begin();
1571       return mirror_image_get_image_id_finish(&it, image_id);
1572     }
1573
1574     int mirror_image_get(librados::IoCtx *ioctx, const std::string &image_id,
1575                          cls::rbd::MirrorImage *mirror_image) {
1576       librados::ObjectReadOperation op;
1577       mirror_image_get_start(&op, image_id);
1578
1579       bufferlist out_bl;
1580       int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1581       if (r < 0) {
1582         return r;
1583       }
1584
1585       bufferlist::iterator iter = out_bl.begin();
1586       r = mirror_image_get_finish(&iter, mirror_image);
1587       if (r < 0) {
1588         return r;
1589       }
1590       return 0;
1591     }
1592
1593     void mirror_image_get_start(librados::ObjectReadOperation *op,
1594                                 const std::string &image_id) {
1595       bufferlist in_bl;
1596       ::encode(image_id, in_bl);
1597
1598       op->exec("rbd", "mirror_image_get", in_bl);
1599     }
1600
1601     int mirror_image_get_finish(bufferlist::iterator *iter,
1602                                 cls::rbd::MirrorImage *mirror_image) {
1603       try {
1604         ::decode(*mirror_image, *iter);
1605       } catch (const buffer::error &err) {
1606         return -EBADMSG;
1607       }
1608       return 0;
1609     }
1610
1611     void mirror_image_set(librados::ObjectWriteOperation *op,
1612                           const std::string &image_id,
1613                           const cls::rbd::MirrorImage &mirror_image) {
1614       bufferlist bl;
1615       ::encode(image_id, bl);
1616       ::encode(mirror_image, bl);
1617
1618       op->exec("rbd", "mirror_image_set", bl);
1619     }
1620
1621     int mirror_image_set(librados::IoCtx *ioctx, const std::string &image_id,
1622                          const cls::rbd::MirrorImage &mirror_image) {
1623       librados::ObjectWriteOperation op;
1624       mirror_image_set(&op, image_id, mirror_image);
1625
1626       int r = ioctx->operate(RBD_MIRRORING, &op);
1627       if (r < 0) {
1628         return r;
1629       }
1630       return 0;
1631     }
1632
1633     void mirror_image_remove(librados::ObjectWriteOperation *op,
1634                              const std::string &image_id) {
1635       bufferlist bl;
1636       ::encode(image_id, bl);
1637
1638       op->exec("rbd", "mirror_image_remove", bl);
1639     }
1640
1641     int mirror_image_remove(librados::IoCtx *ioctx, const std::string &image_id) {
1642       librados::ObjectWriteOperation op;
1643       mirror_image_remove(&op, image_id);
1644
1645       int r = ioctx->operate(RBD_MIRRORING, &op);
1646       if (r < 0) {
1647         return r;
1648       }
1649       return 0;
1650     }
1651
1652     int mirror_image_status_set(librados::IoCtx *ioctx,
1653                                 const std::string &global_image_id,
1654                                 const cls::rbd::MirrorImageStatus &status) {
1655       librados::ObjectWriteOperation op;
1656       mirror_image_status_set(&op, global_image_id, status);
1657       return ioctx->operate(RBD_MIRRORING, &op);
1658     }
1659
1660     void mirror_image_status_set(librados::ObjectWriteOperation *op,
1661                                  const std::string &global_image_id,
1662                                  const cls::rbd::MirrorImageStatus &status) {
1663       bufferlist bl;
1664       ::encode(global_image_id, bl);
1665       ::encode(status, bl);
1666       op->exec("rbd", "mirror_image_status_set", bl);
1667     }
1668
1669     int mirror_image_status_remove(librados::IoCtx *ioctx,
1670                                    const std::string &global_image_id) {
1671       librados::ObjectWriteOperation op;
1672       mirror_image_status_remove(&op, global_image_id);
1673       return ioctx->operate(RBD_MIRRORING, &op);
1674     }
1675
1676     void mirror_image_status_remove(librados::ObjectWriteOperation *op,
1677                                     const std::string &global_image_id) {
1678       bufferlist bl;
1679       ::encode(global_image_id, bl);
1680       op->exec("rbd", "mirror_image_status_remove", bl);
1681     }
1682
1683     int mirror_image_status_get(librados::IoCtx *ioctx,
1684                                 const std::string &global_image_id,
1685                                 cls::rbd::MirrorImageStatus *status) {
1686       librados::ObjectReadOperation op;
1687       mirror_image_status_get_start(&op, global_image_id);
1688
1689       bufferlist out_bl;
1690       int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1691       if (r < 0) {
1692         return r;
1693       }
1694
1695       bufferlist::iterator iter = out_bl.begin();
1696       r = mirror_image_status_get_finish(&iter, status);
1697       if (r < 0) {
1698         return r;
1699       }
1700       return 0;
1701     }
1702
1703     void mirror_image_status_get_start(librados::ObjectReadOperation *op,
1704                                        const std::string &global_image_id) {
1705       bufferlist bl;
1706       ::encode(global_image_id, bl);
1707       op->exec("rbd", "mirror_image_status_get", bl);
1708     }
1709
1710     int mirror_image_status_get_finish(bufferlist::iterator *iter,
1711                                        cls::rbd::MirrorImageStatus *status) {
1712       try {
1713         ::decode(*status, *iter);
1714       } catch (const buffer::error &err) {
1715         return -EBADMSG;
1716       }
1717       return 0;
1718     }
1719
1720     int mirror_image_status_list(librados::IoCtx *ioctx,
1721         const std::string &start, uint64_t max_return,
1722         std::map<std::string, cls::rbd::MirrorImage> *images,
1723         std::map<std::string, cls::rbd::MirrorImageStatus> *statuses) {
1724       librados::ObjectReadOperation op;
1725       mirror_image_status_list_start(&op, start, max_return);
1726
1727       bufferlist out_bl;
1728       int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1729       if (r < 0) {
1730         return r;
1731       }
1732
1733       bufferlist::iterator iter = out_bl.begin();
1734       r = mirror_image_status_list_finish(&iter, images, statuses);
1735       if (r < 0) {
1736         return r;
1737       }
1738       return 0;
1739     }
1740
1741     void mirror_image_status_list_start(librados::ObjectReadOperation *op,
1742                                         const std::string &start,
1743                                         uint64_t max_return) {
1744       bufferlist bl;
1745       ::encode(start, bl);
1746       ::encode(max_return, bl);
1747       op->exec("rbd", "mirror_image_status_list", bl);
1748     }
1749
1750     int mirror_image_status_list_finish(bufferlist::iterator *iter,
1751         std::map<std::string, cls::rbd::MirrorImage> *images,
1752         std::map<std::string, cls::rbd::MirrorImageStatus> *statuses) {
1753       images->clear();
1754       statuses->clear();
1755       try {
1756         ::decode(*images, *iter);
1757         ::decode(*statuses, *iter);
1758       } catch (const buffer::error &err) {
1759         return -EBADMSG;
1760       }
1761       return 0;
1762     }
1763
1764     int mirror_image_status_get_summary(librados::IoCtx *ioctx,
1765         std::map<cls::rbd::MirrorImageStatusState, int> *states) {
1766       librados::ObjectReadOperation op;
1767       mirror_image_status_get_summary_start(&op);
1768
1769       bufferlist out_bl;
1770       int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1771       if (r < 0) {
1772         return r;
1773       }
1774
1775       bufferlist::iterator iter = out_bl.begin();
1776       r = mirror_image_status_get_summary_finish(&iter, states);
1777       if (r < 0) {
1778         return r;
1779       }
1780       return 0;
1781     }
1782
1783     void mirror_image_status_get_summary_start(
1784       librados::ObjectReadOperation *op) {
1785       bufferlist bl;
1786       op->exec("rbd", "mirror_image_status_get_summary", bl);
1787     }
1788
1789     int mirror_image_status_get_summary_finish(bufferlist::iterator *iter,
1790         std::map<cls::rbd::MirrorImageStatusState, int> *states) {
1791       try {
1792         ::decode(*states, *iter);
1793       } catch (const buffer::error &err) {
1794         return -EBADMSG;
1795       }
1796       return 0;
1797     }
1798
1799     int mirror_image_status_remove_down(librados::IoCtx *ioctx) {
1800       librados::ObjectWriteOperation op;
1801       mirror_image_status_remove_down(&op);
1802       return ioctx->operate(RBD_MIRRORING, &op);
1803     }
1804
1805     void mirror_image_status_remove_down(librados::ObjectWriteOperation *op) {
1806       bufferlist bl;
1807       op->exec("rbd", "mirror_image_status_remove_down", bl);
1808     }
1809
1810     void mirror_instances_list_start(librados::ObjectReadOperation *op) {
1811       bufferlist bl;
1812       op->exec("rbd", "mirror_instances_list", bl);
1813     }
1814
1815     int mirror_instances_list_finish(bufferlist::iterator *iter,
1816                                      std::vector<std::string> *instance_ids) {
1817       instance_ids->clear();
1818       try {
1819         ::decode(*instance_ids, *iter);
1820       } catch (const buffer::error &err) {
1821         return -EBADMSG;
1822       }
1823       return 0;
1824     }
1825
1826     int mirror_instances_list(librados::IoCtx *ioctx,
1827                               std::vector<std::string> *instance_ids) {
1828       librados::ObjectReadOperation op;
1829       mirror_instances_list_start(&op);
1830
1831       bufferlist out_bl;
1832       int r = ioctx->operate(RBD_MIRROR_LEADER, &op, &out_bl);
1833       if (r < 0) {
1834         return r;
1835       }
1836
1837       bufferlist::iterator iter = out_bl.begin();
1838       r = mirror_instances_list_finish(&iter, instance_ids);
1839       if (r < 0) {
1840         return r;
1841       }
1842       return 0;
1843     }
1844
1845     void mirror_instances_add(librados::ObjectWriteOperation *op,
1846                               const std::string &instance_id) {
1847       bufferlist bl;
1848       ::encode(instance_id, bl);
1849       op->exec("rbd", "mirror_instances_add", bl);
1850     }
1851
1852     int mirror_instances_add(librados::IoCtx *ioctx,
1853                              const std::string &instance_id) {
1854       librados::ObjectWriteOperation op;
1855       mirror_instances_add(&op, instance_id);
1856       return ioctx->operate(RBD_MIRROR_LEADER, &op);
1857     }
1858
1859     void mirror_instances_remove(librados::ObjectWriteOperation *op,
1860                                  const std::string &instance_id) {
1861       bufferlist bl;
1862       ::encode(instance_id, bl);
1863       op->exec("rbd", "mirror_instances_remove", bl);
1864     }
1865
1866     int mirror_instances_remove(librados::IoCtx *ioctx,
1867                                 const std::string &instance_id) {
1868       librados::ObjectWriteOperation op;
1869       mirror_instances_remove(&op, instance_id);
1870       return ioctx->operate(RBD_MIRROR_LEADER, &op);
1871     }
1872
1873     // Consistency groups functions
1874     int group_create(librados::IoCtx *ioctx, const std::string &oid)
1875     {
1876       bufferlist bl, bl2;
1877
1878       return ioctx->exec(oid, "rbd", "group_create", bl, bl2);
1879     }
1880
1881     int group_dir_list(librados::IoCtx *ioctx, const std::string &oid,
1882                      const std::string &start, uint64_t max_return,
1883                      map<string, string> *cgs)
1884     {
1885       bufferlist in, out;
1886       ::encode(start, in);
1887       ::encode(max_return, in);
1888       int r = ioctx->exec(oid, "rbd", "group_dir_list", in, out);
1889       if (r < 0)
1890         return r;
1891
1892       bufferlist::iterator iter = out.begin();
1893       try {
1894         ::decode(*cgs, iter);
1895       } catch (const buffer::error &err) {
1896         return -EBADMSG;
1897       }
1898
1899       return 0;
1900     }
1901
1902     int group_dir_add(librados::IoCtx *ioctx, const std::string &oid,
1903                    const std::string &name, const std::string &id)
1904     {
1905       bufferlist in, out;
1906       ::encode(name, in);
1907       ::encode(id, in);
1908       return ioctx->exec(oid, "rbd", "group_dir_add", in, out);
1909     }
1910
1911     int group_dir_remove(librados::IoCtx *ioctx, const std::string &oid,
1912                       const std::string &name, const std::string &id)
1913     {
1914       bufferlist in, out;
1915       ::encode(name, in);
1916       ::encode(id, in);
1917       return ioctx->exec(oid, "rbd", "group_dir_remove", in, out);
1918     }
1919
1920     int group_image_remove(librados::IoCtx *ioctx, const std::string &oid,
1921                            const cls::rbd::GroupImageSpec &spec)
1922     {
1923       bufferlist bl, bl2;
1924       ::encode(spec, bl);
1925
1926       return ioctx->exec(oid, "rbd", "group_image_remove", bl, bl2);
1927     }
1928
1929     int group_image_list(librados::IoCtx *ioctx,
1930                          const std::string &oid,
1931                          const cls::rbd::GroupImageSpec &start,
1932                          uint64_t max_return,
1933                          std::vector<cls::rbd::GroupImageStatus> *images)
1934     {
1935       bufferlist bl, bl2;
1936       ::encode(start, bl);
1937       ::encode(max_return, bl);
1938
1939       int r = ioctx->exec(oid, "rbd", "group_image_list", bl, bl2);
1940       if (r < 0)
1941         return r;
1942
1943       bufferlist::iterator iter = bl2.begin();
1944       try {
1945         ::decode(*images, iter);
1946       } catch (const buffer::error &err) {
1947         return -EBADMSG;
1948       }
1949
1950       return 0;
1951     }
1952
1953     int group_image_set(librados::IoCtx *ioctx, const std::string &oid,
1954                         const cls::rbd::GroupImageStatus &st)
1955     {
1956       bufferlist bl, bl2;
1957       ::encode(st, bl);
1958
1959       return ioctx->exec(oid, "rbd", "group_image_set", bl, bl2);
1960     }
1961
1962     int image_add_group(librados::IoCtx *ioctx, const std::string &oid,
1963                         const cls::rbd::GroupSpec &group_spec)
1964     {
1965       bufferlist bl, bl2;
1966       ::encode(group_spec, bl);
1967
1968       return ioctx->exec(oid, "rbd", "image_add_group", bl, bl2);
1969     }
1970
1971     int image_remove_group(librados::IoCtx *ioctx, const std::string &oid,
1972                            const cls::rbd::GroupSpec &group_spec)
1973     {
1974       bufferlist bl, bl2;
1975       ::encode(group_spec, bl);
1976
1977       return ioctx->exec(oid, "rbd", "image_remove_group", bl, bl2);
1978     }
1979
1980     void image_get_group_start(librados::ObjectReadOperation *op)
1981     {
1982       bufferlist in_bl;
1983       op->exec("rbd", "image_get_group", in_bl);
1984     }
1985
1986     int image_get_group_finish(bufferlist::iterator *iter,
1987                                cls::rbd::GroupSpec *group_spec)
1988     {
1989       try {
1990         ::decode(*group_spec, *iter);
1991       } catch (const buffer::error &err) {
1992         return -EBADMSG;
1993       }
1994       return 0;
1995     }
1996
1997     int image_get_group(librados::IoCtx *ioctx, const std::string &oid,
1998                         cls::rbd::GroupSpec *group_spec)
1999     {
2000       librados::ObjectReadOperation op;
2001       image_get_group_start(&op);
2002
2003       bufferlist out_bl;
2004       int r = ioctx->operate(oid, &op, &out_bl);
2005       if (r < 0) {
2006         return r;
2007       }
2008
2009       bufferlist::iterator iter = out_bl.begin();
2010       return image_get_group_finish(&iter, group_spec);
2011     }
2012
2013     // rbd_trash functions
2014     void trash_add(librados::ObjectWriteOperation *op,
2015                    const std::string &id,
2016                    const cls::rbd::TrashImageSpec &trash_spec)
2017     {
2018       bufferlist bl;
2019       ::encode(id, bl);
2020       ::encode(trash_spec, bl);
2021       op->exec("rbd", "trash_add", bl);
2022     }
2023
2024     int trash_add(librados::IoCtx *ioctx, const std::string &id,
2025                   const cls::rbd::TrashImageSpec &trash_spec)
2026     {
2027       librados::ObjectWriteOperation op;
2028       trash_add(&op, id, trash_spec);
2029
2030       return ioctx->operate(RBD_TRASH, &op);
2031     }
2032
2033     void trash_remove(librados::ObjectWriteOperation *op,
2034                       const std::string &id)
2035     {
2036       bufferlist bl;
2037       ::encode(id, bl);
2038       op->exec("rbd", "trash_remove", bl);
2039     }
2040
2041     int trash_remove(librados::IoCtx *ioctx, const std::string &id)
2042     {
2043       librados::ObjectWriteOperation op;
2044       trash_remove(&op, id);
2045
2046       return ioctx->operate(RBD_TRASH, &op);
2047     }
2048
2049     void trash_list_start(librados::ObjectReadOperation *op,
2050                           const std::string &start, uint64_t max_return)
2051     {
2052       bufferlist bl;
2053       ::encode(start, bl);
2054       ::encode(max_return, bl);
2055       op->exec("rbd", "trash_list", bl);
2056     }
2057
2058     int trash_list_finish(bufferlist::iterator *it,
2059                           map<string, cls::rbd::TrashImageSpec> *entries)
2060     {
2061       assert(entries);
2062
2063       try {
2064         ::decode(*entries, *it);
2065       } catch (const buffer::error &err) {
2066         return -EBADMSG;
2067       }
2068
2069       return 0;
2070     }
2071
2072     int trash_list(librados::IoCtx *ioctx,
2073                    const std::string &start, uint64_t max_return,
2074                    map<string, cls::rbd::TrashImageSpec> *entries)
2075     {
2076       librados::ObjectReadOperation op;
2077       trash_list_start(&op, start, max_return);
2078
2079       bufferlist out_bl;
2080       int r = ioctx->operate(RBD_TRASH, &op, &out_bl);
2081       if (r < 0) {
2082         return r;
2083       }
2084
2085       bufferlist::iterator iter = out_bl.begin();
2086       return trash_list_finish(&iter, entries);
2087     }
2088
2089     void trash_get_start(librados::ObjectReadOperation *op,
2090                          const std::string &id)
2091     {
2092       bufferlist bl;
2093       ::encode(id, bl);
2094       op->exec("rbd", "trash_get", bl);
2095     }
2096
2097     int trash_get_finish(bufferlist::iterator *it,
2098                           cls::rbd::TrashImageSpec *trash_spec) {
2099       assert(trash_spec);
2100       try {
2101         ::decode(*trash_spec, *it);
2102       } catch (const buffer::error &err) {
2103         return -EBADMSG;
2104       }
2105
2106       return 0;
2107     }
2108
2109
2110     int trash_get(librados::IoCtx *ioctx, const std::string &id,
2111                   cls::rbd::TrashImageSpec *trash_spec)
2112     {
2113       librados::ObjectReadOperation op;
2114       trash_get_start(&op, id);
2115
2116       bufferlist out_bl;
2117       int r = ioctx->operate(RBD_TRASH, &op, &out_bl);
2118       if (r < 0) {
2119         return r;
2120       }
2121
2122       bufferlist::iterator it = out_bl.begin();
2123       return trash_get_finish(&it, trash_spec);
2124     }
2125
2126   } // namespace cls_client
2127 } // namespace librbd