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 #include "common/bit_vector.hpp"
  11
  12 #include <errno.h>
  13
  14 namespace librbd {
  15 namespace cls_client {
  16
  17 void create_image(librados::ObjectWriteOperation *op, uint64_t size,
  18                   uint8_t order, uint64_t features,
  19                   const std::string &object_prefix, int64_t data_pool_id)
  20 {
  21   bufferlist bl;
  22   encode(size, bl);
  23   encode(order, bl);
  24   encode(features, bl);
  25   encode(object_prefix, bl);
  26   encode(data_pool_id, bl);
  27
  28   op->exec("rbd", "create", bl);
  29 }
  30
  31 int create_image(librados::IoCtx *ioctx, const std::string &oid,
  32                  uint64_t size, uint8_t order, uint64_t features,
  33                  const std::string &object_prefix, int64_t data_pool_id)
  34 {
  35   librados::ObjectWriteOperation op;
  36   create_image(&op, size, order, features, object_prefix, data_pool_id);
  37
  38   return ioctx->operate(oid, &op);
  39 }
  40
  41 void get_features_start(librados::ObjectReadOperation *op, bool read_only)
  42 {
  43   bufferlist bl;
  44   encode(static_cast<uint64_t>(CEPH_NOSNAP), bl);
  45   encode(read_only, bl);
  46   op->exec("rbd", "get_features", bl);
  47 }
  48
  49 int get_features_finish(bufferlist::const_iterator *it, uint64_t *features,
  50                         uint64_t *incompatible_features)
  51 {
  52   try {
  53     decode(*features, *it);
  54     decode(*incompatible_features, *it);
  55   } catch (const buffer::error &err) {
  56     return -EBADMSG;
  57   }
  58
  59   return 0;
  60 }
  61
  62 int get_features(librados::IoCtx *ioctx, const std::string &oid,
  63                  bool read_only, uint64_t *features,
  64                  uint64_t *incompatible_features)
  65 {
  66   librados::ObjectReadOperation op;
  67   get_features_start(&op, read_only);
  68
  69   bufferlist out_bl;
  70   int r = ioctx->operate(oid, &op, &out_bl);
  71   if (r < 0) {
  72     return r;
  73   }
  74
  75   auto it = out_bl.cbegin();
  76   return get_features_finish(&it, features, incompatible_features);
  77 }
  78
  79 void set_features(librados::ObjectWriteOperation *op, uint64_t features,
  80                   uint64_t mask)
  81 {
  82   bufferlist bl;
  83   encode(features, bl);
  84   encode(mask, bl);
  85
  86   op->exec("rbd", "set_features", bl);
  87 }
  88
  89 int set_features(librados::IoCtx *ioctx, const std::string &oid,
  90                   uint64_t features, uint64_t mask)
  91 {
  92   librados::ObjectWriteOperation op;
  93   set_features(&op, features, mask);
  94
  95   return ioctx->operate(oid, &op);
  96 }
  97
  98 void get_object_prefix_start(librados::ObjectReadOperation *op)
  99 {
 100   bufferlist bl;
 101   op->exec("rbd", "get_object_prefix", bl);
 102 }
 103
 104 int get_object_prefix_finish(bufferlist::const_iterator *it,
 105                              std::string *object_prefix)
 106 {
 107   try {
 108     decode(*object_prefix, *it);
 109   } catch (const buffer::error &err) {
 110     return -EBADMSG;
 111   }
 112   return 0;
 113 }
 114
 115 int get_object_prefix(librados::IoCtx *ioctx, const std::string &oid,
 116                       std::string *object_prefix)
 117 {
 118   librados::ObjectReadOperation op;
 119   get_object_prefix_start(&op);
 120
 121   bufferlist out_bl;
 122   int r = ioctx->operate(oid, &op, &out_bl);
 123   if (r < 0) {
 124     return r;
 125   }
 126
 127   auto it = out_bl.cbegin();
 128   return get_object_prefix_finish(&it, object_prefix);
 129 }
 130
 131 void get_data_pool_start(librados::ObjectReadOperation *op) {
 132   bufferlist bl;
 133   op->exec("rbd", "get_data_pool", bl);
 134 }
 135
 136 int get_data_pool_finish(bufferlist::const_iterator *it, int64_t *data_pool_id) {
 137   try {
 138     decode(*data_pool_id, *it);
 139   } catch (const buffer::error &err) {
 140     return -EBADMSG;
 141   }
 142   return 0;
 143 }
 144
 145 int get_data_pool(librados::IoCtx *ioctx, const std::string &oid,
 146                   int64_t *data_pool_id) {
 147   librados::ObjectReadOperation op;
 148   get_data_pool_start(&op);
 149
 150   bufferlist out_bl;
 151   int r = ioctx->operate(oid, &op, &out_bl);
 152   if (r < 0) {
 153     return r;
 154   }
 155
 156   auto it = out_bl.cbegin();
 157   return get_data_pool_finish(&it, data_pool_id);
 158 }
 159
 160 void get_size_start(librados::ObjectReadOperation *op, snapid_t snap_id)
 161 {
 162   bufferlist bl;
 163   encode(snap_id, bl);
 164   op->exec("rbd", "get_size", bl);
 165 }
 166
 167 int get_size_finish(bufferlist::const_iterator *it, uint64_t *size,
 168                     uint8_t *order)
 169 {
 170   try {
 171     decode(*order, *it);
 172     decode(*size, *it);
 173   } catch (const buffer::error &err) {
 174     return -EBADMSG;
 175   }
 176   return 0;
 177 }
 178
 179 int get_size(librados::IoCtx *ioctx, const std::string &oid,
 180              snapid_t snap_id, uint64_t *size, uint8_t *order)
 181 {
 182   librados::ObjectReadOperation op;
 183   get_size_start(&op, snap_id);
 184
 185   bufferlist out_bl;
 186   int r = ioctx->operate(oid, &op, &out_bl);
 187   if (r < 0) {
 188     return r;
 189   }
 190
 191   auto it = out_bl.cbegin();
 192   return get_size_finish(&it, size, order);
 193 }
 194
 195 int set_size(librados::IoCtx *ioctx, const std::string &oid,
 196              uint64_t size)
 197 {
 198   librados::ObjectWriteOperation op;
 199   set_size(&op, size);
 200   return ioctx->operate(oid, &op);
 201 }
 202
 203 void set_size(librados::ObjectWriteOperation *op, uint64_t size)
 204 {
 205   bufferlist bl;
 206   encode(size, bl);
 207   op->exec("rbd", "set_size", bl);
 208 }
 209
 210 void get_flags_start(librados::ObjectReadOperation *op, snapid_t snap_id) {
 211   bufferlist in_bl;
 212   encode(static_cast<snapid_t>(snap_id), in_bl);
 213   op->exec("rbd", "get_flags", in_bl);
 214 }
 215
 216 int get_flags_finish(bufferlist::const_iterator *it, uint64_t *flags) {
 217   try {
 218     decode(*flags, *it);
 219   } catch (const buffer::error &err) {
 220     return -EBADMSG;
 221   }
 222   return 0;
 223 }
 224
 225 int get_flags(librados::IoCtx *ioctx, const std::string &oid,
 226               snapid_t snap_id, uint64_t *flags)
 227 {
 228   librados::ObjectReadOperation op;
 229   get_flags_start(&op, snap_id);
 230
 231   bufferlist out_bl;
 232   int r = ioctx->operate(oid, &op, &out_bl);
 233   if (r < 0) {
 234     return r;
 235   }
 236
 237   auto it = out_bl.cbegin();
 238   return get_flags_finish(&it, flags);
 239 }
 240
 241 void set_flags(librados::ObjectWriteOperation *op, snapid_t snap_id,
 242                uint64_t flags, uint64_t mask)
 243 {
 244   bufferlist inbl;
 245   encode(flags, inbl);
 246   encode(mask, inbl);
 247   encode(snap_id, inbl);
 248   op->exec("rbd", "set_flags", inbl);
 249 }
 250
 251 void op_features_get_start(librados::ObjectReadOperation *op)
 252 {
 253   bufferlist in_bl;
 254   op->exec("rbd", "op_features_get", in_bl);
 255 }
 256
 257 int op_features_get_finish(bufferlist::const_iterator *it, uint64_t *op_features)
 258 {
 259   try {
 260     decode(*op_features, *it);
 261   } catch (const buffer::error &err) {
 262     return -EBADMSG;
 263   }
 264   return 0;
 265 }
 266
 267 int op_features_get(librados::IoCtx *ioctx, const std::string &oid,
 268                     uint64_t *op_features)
 269 {
 270   librados::ObjectReadOperation op;
 271   op_features_get_start(&op);
 272
 273   bufferlist out_bl;
 274   int r = ioctx->operate(oid, &op, &out_bl);
 275   if (r < 0) {
 276     return r;
 277   }
 278
 279   auto it = out_bl.cbegin();
 280   return op_features_get_finish(&it, op_features);
 281 }
 282
 283 void op_features_set(librados::ObjectWriteOperation *op,
 284                      uint64_t op_features, uint64_t mask)
 285 {
 286   bufferlist inbl;
 287   encode(op_features, inbl);
 288   encode(mask, inbl);
 289   op->exec("rbd", "op_features_set", inbl);
 290 }
 291
 292 int op_features_set(librados::IoCtx *ioctx, const std::string &oid,
 293                     uint64_t op_features, uint64_t mask)
 294 {
 295   librados::ObjectWriteOperation op;
 296   op_features_set(&op, op_features, mask);
 297
 298   return ioctx->operate(oid, &op);
 299 }
 300
 301 void get_parent_start(librados::ObjectReadOperation *op, snapid_t snap_id)
 302 {
 303   bufferlist bl;
 304   encode(snap_id, bl);
 305   op->exec("rbd", "get_parent", bl);
 306 }
 307
 308 int get_parent_finish(bufferlist::const_iterator *it,
 309                       cls::rbd::ParentImageSpec *pspec,
 310                       uint64_t *parent_overlap)
 311 {
 312   *pspec = {};
 313   try {
 314     decode(pspec->pool_id, *it);
 315     decode(pspec->image_id, *it);
 316     decode(pspec->snap_id, *it);
 317     decode(*parent_overlap, *it);
 318   } catch (const buffer::error &) {
 319     return -EBADMSG;
 320   }
 321   return 0;
 322 }
 323
 324 int get_parent(librados::IoCtx *ioctx, const std::string &oid,
 325                snapid_t snap_id, cls::rbd::ParentImageSpec *pspec,
 326                uint64_t *parent_overlap)
 327 {
 328   librados::ObjectReadOperation op;
 329   get_parent_start(&op, snap_id);
 330
 331   bufferlist out_bl;
 332   int r = ioctx->operate(oid, &op, &out_bl);
 333   if (r < 0) {
 334     return r;
 335   }
 336
 337   auto it = out_bl.cbegin();
 338   return get_parent_finish(&it, pspec, parent_overlap);
 339 }
 340
 341 int set_parent(librados::IoCtx *ioctx, const std::string &oid,
 342                const cls::rbd::ParentImageSpec &pspec, uint64_t parent_overlap)
 343 {
 344   librados::ObjectWriteOperation op;
 345   set_parent(&op, pspec, parent_overlap);
 346   return ioctx->operate(oid, &op);
 347 }
 348
 349 void set_parent(librados::ObjectWriteOperation *op,
 350                 const cls::rbd::ParentImageSpec &pspec,
 351                 uint64_t parent_overlap) {
 352   assert(pspec.pool_namespace.empty());
 353
 354   bufferlist in_bl;
 355   encode(pspec.pool_id, in_bl);
 356   encode(pspec.image_id, in_bl);
 357   encode(pspec.snap_id, in_bl);
 358   encode(parent_overlap, in_bl);
 359
 360   op->exec("rbd", "set_parent", in_bl);
 361 }
 362
 363 int remove_parent(librados::IoCtx *ioctx, const std::string &oid)
 364 {
 365   librados::ObjectWriteOperation op;
 366   remove_parent(&op);
 367   return ioctx->operate(oid, &op);
 368 }
 369
 370 void remove_parent(librados::ObjectWriteOperation *op)
 371 {
 372   bufferlist inbl;
 373   op->exec("rbd", "remove_parent", inbl);
 374 }
 375
 376 void parent_get_start(librados::ObjectReadOperation* op) {
 377   bufferlist in_bl;
 378   op->exec("rbd", "parent_get", in_bl);
 379 }
 380
 381 int parent_get_finish(bufferlist::const_iterator* it,
 382                       cls::rbd::ParentImageSpec* parent_image_spec) {
 383   try {
 384     decode(*parent_image_spec, *it);
 385   } catch (const buffer::error &) {
 386     return -EBADMSG;
 387   }
 388   return 0;
 389 }
 390
 391 int parent_get(librados::IoCtx* ioctx, const std::string &oid,
 392                cls::rbd::ParentImageSpec* parent_image_spec) {
 393   librados::ObjectReadOperation op;
 394   parent_get_start(&op);
 395
 396   bufferlist out_bl;
 397   int r = ioctx->operate(oid, &op, &out_bl);
 398   if (r < 0) {
 399     return r;
 400   }
 401
 402   auto it = out_bl.cbegin();
 403   r = parent_get_finish(&it, parent_image_spec);
 404   if (r < 0) {
 405     return r;
 406   }
 407   return 0;
 408 }
 409
 410 void parent_overlap_get_start(librados::ObjectReadOperation* op,
 411                               snapid_t snap_id) {
 412   bufferlist in_bl;
 413   encode(snap_id, in_bl);
 414   op->exec("rbd", "parent_overlap_get", in_bl);
 415 }
 416
 417 int parent_overlap_get_finish(bufferlist::const_iterator* it,
 418                               std::optional<uint64_t>* parent_overlap) {
 419   try {
 420     decode(*parent_overlap, *it);
 421   } catch (const buffer::error &) {
 422     return -EBADMSG;
 423   }
 424   return 0;
 425 }
 426
 427 int parent_overlap_get(librados::IoCtx* ioctx, const std::string &oid,
 428                        snapid_t snap_id,
 429                        std::optional<uint64_t>* parent_overlap) {
 430   librados::ObjectReadOperation op;
 431   parent_overlap_get_start(&op, snap_id);
 432
 433   bufferlist out_bl;
 434   int r = ioctx->operate(oid, &op, &out_bl);
 435   if (r < 0) {
 436     return r;
 437   }
 438
 439   auto it = out_bl.cbegin();
 440   r = parent_overlap_get_finish(&it, parent_overlap);
 441   if (r < 0) {
 442     return r;
 443   }
 444   return 0;
 445 }
 446
 447 void parent_attach(librados::ObjectWriteOperation* op,
 448                    const cls::rbd::ParentImageSpec& parent_image_spec,
 449                    uint64_t parent_overlap, bool reattach) {
 450   bufferlist in_bl;
 451   encode(parent_image_spec, in_bl);
 452   encode(parent_overlap, in_bl);
 453   encode(reattach, in_bl);
 454   op->exec("rbd", "parent_attach", in_bl);
 455 }
 456
 457 int parent_attach(librados::IoCtx *ioctx, const std::string &oid,
 458                   const cls::rbd::ParentImageSpec& parent_image_spec,
 459                   uint64_t parent_overlap, bool reattach) {
 460   librados::ObjectWriteOperation op;
 461   parent_attach(&op, parent_image_spec, parent_overlap, reattach);
 462   return ioctx->operate(oid, &op);
 463 }
 464
 465 void parent_detach(librados::ObjectWriteOperation* op) {
 466   bufferlist in_bl;
 467   op->exec("rbd", "parent_detach", in_bl);
 468 }
 469
 470 int parent_detach(librados::IoCtx *ioctx, const std::string &oid) {
 471   librados::ObjectWriteOperation op;
 472   parent_detach(&op);
 473   return ioctx->operate(oid, &op);
 474 }
 475
 476 int add_child(librados::IoCtx *ioctx, const std::string &oid,
 477               const cls::rbd::ParentImageSpec &pspec,
 478               const std::string &c_imageid)
 479 {
 480   librados::ObjectWriteOperation op;
 481   add_child(&op, pspec, c_imageid);
 482   return ioctx->operate(oid, &op);
 483 }
 484
 485 void add_child(librados::ObjectWriteOperation *op,
 486                const cls::rbd::ParentImageSpec& pspec,
 487                const std::string &c_imageid)
 488 {
 489   assert(pspec.pool_namespace.empty());
 490
 491   bufferlist in;
 492   encode(pspec.pool_id, in);
 493   encode(pspec.image_id, in);
 494   encode(pspec.snap_id, in);
 495   encode(c_imageid, in);
 496
 497   op->exec("rbd", "add_child", in);
 498 }
 499
 500 void remove_child(librados::ObjectWriteOperation *op,
 501                   const cls::rbd::ParentImageSpec &pspec,
 502                   const std::string &c_imageid)
 503 {
 504   assert(pspec.pool_namespace.empty());
 505
 506   bufferlist in;
 507   encode(pspec.pool_id, in);
 508   encode(pspec.image_id, in);
 509   encode(pspec.snap_id, in);
 510   encode(c_imageid, in);
 511   op->exec("rbd", "remove_child", in);
 512 }
 513
 514 int remove_child(librados::IoCtx *ioctx, const std::string &oid,
 515                  const cls::rbd::ParentImageSpec &pspec,
 516                  const std::string &c_imageid)
 517 {
 518   librados::ObjectWriteOperation op;
 519   remove_child(&op, pspec, c_imageid);
 520   return ioctx->operate(oid, &op);
 521 }
 522
 523 void get_children_start(librados::ObjectReadOperation *op,
 524                         const cls::rbd::ParentImageSpec &pspec) {
 525   bufferlist in_bl;
 526   encode(pspec.pool_id, in_bl);
 527   encode(pspec.image_id, in_bl);
 528   encode(pspec.snap_id, in_bl);
 529   op->exec("rbd", "get_children", in_bl);
 530 }
 531
 532 int get_children_finish(bufferlist::const_iterator *it,
 533                         std::set<std::string>* children) {
 534   try {
 535     decode(*children, *it);
 536   } catch (const buffer::error &err) {
 537     return -EBADMSG;
 538   }
 539   return 0;
 540 }
 541
 542 int get_children(librados::IoCtx *ioctx, const std::string &oid,
 543                  const cls::rbd::ParentImageSpec &pspec, set<string>& children)
 544 {
 545   librados::ObjectReadOperation op;
 546   get_children_start(&op, pspec);
 547
 548   bufferlist out_bl;
 549   int r = ioctx->operate(oid, &op, &out_bl);
 550   if (r < 0) {
 551     return r;
 552   }
 553
 554   auto it = out_bl.cbegin();
 555   return get_children_finish(&it, &children);
 556 }
 557
 558 void snapshot_get_start(librados::ObjectReadOperation *op, snapid_t snap_id)
 559 {
 560   bufferlist bl;
 561   encode(snap_id, bl);
 562   op->exec("rbd", "snapshot_get", bl);
 563 }
 564
 565 int snapshot_get_finish(bufferlist::const_iterator* it,
 566                         cls::rbd::SnapshotInfo* snap_info)
 567 {
 568   try {
 569     decode(*snap_info, *it);
 570   } catch (const buffer::error &err) {
 571     return -EBADMSG;
 572   }
 573   return 0;
 574 }
 575
 576 int snapshot_get(librados::IoCtx *ioctx, const std::string &oid,
 577                  snapid_t snap_id, cls::rbd::SnapshotInfo* snap_info)
 578 {
 579   librados::ObjectReadOperation op;
 580   snapshot_get_start(&op, snap_id);
 581
 582   bufferlist out_bl;
 583   int r = ioctx->operate(oid, &op, &out_bl);
 584   if (r < 0) {
 585     return r;
 586   }
 587
 588   auto it = out_bl.cbegin();
 589   return snapshot_get_finish(&it, snap_info);
 590 }
 591
 592 void snapshot_add(librados::ObjectWriteOperation *op, snapid_t snap_id,
 593                   const std::string &snap_name,
 594                   const cls::rbd::SnapshotNamespace &snap_namespace)
 595 {
 596   bufferlist bl;
 597   encode(snap_name, bl);
 598   encode(snap_id, bl);
 599   encode(snap_namespace, bl);
 600   op->exec("rbd", "snapshot_add", bl);
 601 }
 602
 603 void snapshot_remove(librados::ObjectWriteOperation *op, snapid_t snap_id)
 604 {
 605   bufferlist bl;
 606   encode(snap_id, bl);
 607   op->exec("rbd", "snapshot_remove", bl);
 608 }
 609
 610 void snapshot_rename(librados::ObjectWriteOperation *op,
 611                      snapid_t src_snap_id,
 612                      const std::string &dst_name)
 613 {
 614   bufferlist bl;
 615   encode(src_snap_id, bl);
 616   encode(dst_name, bl);
 617   op->exec("rbd", "snapshot_rename", bl);
 618 }
 619
 620 void snapshot_trash_add(librados::ObjectWriteOperation *op,
 621                         snapid_t snap_id)
 622 {
 623   bufferlist bl;
 624   encode(snap_id, bl);
 625   op->exec("rbd", "snapshot_trash_add", bl);
 626 }
 627
 628 void get_snapcontext_start(librados::ObjectReadOperation *op)
 629 {
 630   bufferlist bl;
 631   op->exec("rbd", "get_snapcontext", bl);
 632 }
 633
 634 int get_snapcontext_finish(bufferlist::const_iterator *it,
 635                            ::SnapContext *snapc)
 636 {
 637   try {
 638     decode(*snapc, *it);
 639   } catch (const buffer::error &err) {
 640     return -EBADMSG;
 641   }
 642   if (!snapc->is_valid()) {
 643     return -EBADMSG;
 644   }
 645   return 0;
 646 }
 647
 648 int get_snapcontext(librados::IoCtx *ioctx, const std::string &oid,
 649                     ::SnapContext *snapc)
 650 {
 651   librados::ObjectReadOperation op;
 652   get_snapcontext_start(&op);
 653
 654   bufferlist out_bl;
 655   int r = ioctx->operate(oid, &op, &out_bl);
 656   if (r < 0) {
 657     return r;
 658   }
 659
 660   auto bl_it = out_bl.cbegin();
 661   return get_snapcontext_finish(&bl_it, snapc);
 662 }
 663
 664 void get_snapshot_name_start(librados::ObjectReadOperation *op,
 665                              snapid_t snap_id)
 666 {
 667   bufferlist bl;
 668   encode(snap_id, bl);
 669   op->exec("rbd", "get_snapshot_name", bl);
 670 }
 671
 672 int get_snapshot_name_finish(bufferlist::const_iterator *it,
 673                              std::string *name)
 674 {
 675   try {
 676     decode(*name, *it);
 677   } catch (const buffer::error &err) {
 678     return -EBADMSG;
 679   }
 680   return 0;
 681 }
 682
 683 int get_snapshot_name(librados::IoCtx *ioctx, const std::string &oid,
 684                       snapid_t snap_id, std::string *name)
 685 {
 686   librados::ObjectReadOperation op;
 687   get_snapshot_name_start(&op, snap_id);
 688
 689   bufferlist out_bl;
 690   int r = ioctx->operate(oid, &op, &out_bl);
 691   if (r < 0) {
 692     return r;
 693   }
 694
 695   auto it = out_bl.cbegin();
 696   return get_snapshot_name_finish(&it, name);
 697 }
 698
 699 void get_snapshot_timestamp_start(librados::ObjectReadOperation *op,
 700                                   snapid_t snap_id)
 701 {
 702   bufferlist bl;
 703   encode(snap_id, bl);
 704   op->exec("rbd", "get_snapshot_timestamp", bl);
 705 }
 706
 707 int get_snapshot_timestamp_finish(bufferlist::const_iterator *it,
 708                                   utime_t *timestamp)
 709 {
 710   try {
 711     decode(*timestamp, *it);
 712   } catch (const buffer::error &err) {
 713     return -EBADMSG;
 714   }
 715   return 0;
 716 }
 717
 718 int get_snapshot_timestamp(librados::IoCtx *ioctx, const std::string &oid,
 719                            snapid_t snap_id, utime_t *timestamp)
 720 {
 721   librados::ObjectReadOperation op;
 722   get_snapshot_timestamp_start(&op, snap_id);
 723
 724   bufferlist out_bl;
 725   int r = ioctx->operate(oid, &op, &out_bl);
 726   if (r < 0) {
 727     return r;
 728   }
 729
 730   auto it = out_bl.cbegin();
 731   return get_snapshot_timestamp_finish(&it, timestamp);
 732 }
 733
 734 void old_snapshot_add(librados::ObjectWriteOperation *op,
 735                       snapid_t snap_id, const std::string &snap_name)
 736 {
 737   bufferlist bl;
 738   encode(snap_name, bl);
 739   encode(snap_id, bl);
 740   op->exec("rbd", "snap_add", bl);
 741 }
 742
 743 void old_snapshot_remove(librados::ObjectWriteOperation *op,
 744                          const std::string &snap_name)
 745 {
 746   bufferlist bl;
 747   encode(snap_name, bl);
 748   op->exec("rbd", "snap_remove", bl);
 749 }
 750
 751 void old_snapshot_rename(librados::ObjectWriteOperation *op,
 752                          snapid_t src_snap_id, const std::string &dst_name)
 753 {
 754   bufferlist bl;
 755   encode(src_snap_id, bl);
 756   encode(dst_name, bl);
 757   op->exec("rbd", "snap_rename", bl);
 758 }
 759
 760 void old_snapshot_list_start(librados::ObjectReadOperation *op) {
 761   bufferlist in_bl;
 762   op->exec("rbd", "snap_list", in_bl);
 763 }
 764
 765 int old_snapshot_list_finish(bufferlist::const_iterator *it,
 766                              std::vector<string> *names,
 767                              std::vector<uint64_t> *sizes,
 768                              ::SnapContext *snapc) {
 769   try {
 770     uint32_t num_snaps;
 771     decode(snapc->seq, *it);
 772     decode(num_snaps, *it);
 773
 774     names->resize(num_snaps);
 775     sizes->resize(num_snaps);
 776     snapc->snaps.resize(num_snaps);
 777     for (uint32_t i = 0; i < num_snaps; ++i) {
 778       decode(snapc->snaps[i], *it);
 779       decode((*sizes)[i], *it);
 780       decode((*names)[i], *it);
 781     }
 782   } catch (const buffer::error &err) {
 783     return -EBADMSG;
 784   }
 785   return 0;
 786 }
 787
 788 int old_snapshot_list(librados::IoCtx *ioctx, const std::string &oid,
 789                       std::vector<string> *names,
 790                       std::vector<uint64_t> *sizes,
 791                       ::SnapContext *snapc)
 792 {
 793   librados::ObjectReadOperation op;
 794   old_snapshot_list_start(&op);
 795
 796   bufferlist out_bl;
 797   int r = ioctx->operate(oid, &op, &out_bl);
 798   if (r < 0) {
 799     return r;
 800   }
 801
 802   auto it = out_bl.cbegin();
 803   return old_snapshot_list_finish(&it, names, sizes, snapc);
 804 }
 805
 806 void get_all_features_start(librados::ObjectReadOperation *op) {
 807   bufferlist in;
 808   op->exec("rbd", "get_all_features", in);
 809 }
 810
 811 int get_all_features_finish(bufferlist::const_iterator *it,
 812                             uint64_t *all_features) {
 813   try {
 814     decode(*all_features, *it);
 815   } catch (const buffer::error &err) {
 816     return -EBADMSG;
 817   }
 818   return 0;
 819 }
 820
 821 int get_all_features(librados::IoCtx *ioctx, const std::string &oid,
 822                      uint64_t *all_features) {
 823   librados::ObjectReadOperation op;
 824   get_all_features_start(&op);
 825
 826   bufferlist out_bl;
 827   int r = ioctx->operate(oid, &op, &out_bl);
 828   if (r < 0) {
 829     return r;
 830   }
 831
 832   auto it = out_bl.cbegin();
 833   return get_all_features_finish(&it, all_features);
 834 }
 835
 836 int copyup(librados::IoCtx *ioctx, const std::string &oid,
 837            bufferlist data) {
 838   bufferlist out;
 839   return ioctx->exec(oid, "rbd", "copyup", data, out);
 840 }
 841
 842 void get_protection_status_start(librados::ObjectReadOperation *op,
 843                                  snapid_t snap_id)
 844 {
 845   bufferlist bl;
 846   encode(snap_id, bl);
 847   op->exec("rbd", "get_protection_status", bl);
 848 }
 849
 850 int get_protection_status_finish(bufferlist::const_iterator *it,
 851                                  uint8_t *protection_status)
 852 {
 853   try {
 854     decode(*protection_status, *it);
 855   } catch (const buffer::error &) {
 856     return -EBADMSG;
 857   }
 858   return 0;
 859 }
 860
 861 int get_protection_status(librados::IoCtx *ioctx, const std::string &oid,
 862                           snapid_t snap_id, uint8_t *protection_status)
 863 {
 864   librados::ObjectReadOperation op;
 865   get_protection_status_start(&op, snap_id);
 866
 867   bufferlist out_bl;
 868   int r = ioctx->operate(oid, &op, &out_bl);
 869   if (r < 0) {
 870     return r;
 871   }
 872
 873   auto it = out_bl.cbegin();
 874   return get_protection_status_finish(&it, protection_status);
 875 }
 876
 877 int set_protection_status(librados::IoCtx *ioctx, const std::string &oid,
 878                           snapid_t snap_id, uint8_t protection_status)
 879 {
 880   // TODO remove
 881   librados::ObjectWriteOperation op;
 882   set_protection_status(&op, snap_id, protection_status);
 883   return ioctx->operate(oid, &op);
 884 }
 885
 886 void set_protection_status(librados::ObjectWriteOperation *op,
 887                            snapid_t snap_id, uint8_t protection_status)
 888 {
 889   bufferlist in;
 890   encode(snap_id, in);
 891   encode(protection_status, in);
 892   op->exec("rbd", "set_protection_status", in);
 893 }
 894
 895 int snapshot_get_limit(librados::IoCtx *ioctx, const std::string &oid,
 896                        uint64_t *limit)
 897 {
 898   bufferlist in, out;
 899   int r =  ioctx->exec(oid, "rbd", "snapshot_get_limit", in, out);
 900
 901   if (r < 0) {
 902     return r;
 903   }
 904
 905   try {
 906     auto iter = out.cbegin();
 907     decode(*limit, iter);
 908   } catch (const buffer::error &err) {
 909     return -EBADMSG;
 910   }
 911
 912   return 0;
 913 }
 914
 915 void snapshot_set_limit(librados::ObjectWriteOperation *op, uint64_t limit)
 916 {
 917   bufferlist in;
 918   encode(limit, in);
 919   op->exec("rbd", "snapshot_set_limit", in);
 920 }
 921
 922 void get_stripe_unit_count_start(librados::ObjectReadOperation *op) {
 923   bufferlist empty_bl;
 924   op->exec("rbd", "get_stripe_unit_count", empty_bl);
 925 }
 926
 927 int get_stripe_unit_count_finish(bufferlist::const_iterator *it,
 928                                  uint64_t *stripe_unit,
 929                                  uint64_t *stripe_count) {
 930   ceph_assert(stripe_unit);
 931   ceph_assert(stripe_count);
 932
 933   try {
 934     decode(*stripe_unit, *it);
 935     decode(*stripe_count, *it);
 936   } catch (const buffer::error &err) {
 937     return -EBADMSG;
 938   }
 939   return 0;
 940 }
 941
 942 int get_stripe_unit_count(librados::IoCtx *ioctx, const std::string &oid,
 943                           uint64_t *stripe_unit, uint64_t *stripe_count)
 944 {
 945   librados::ObjectReadOperation op;
 946   get_stripe_unit_count_start(&op);
 947
 948   bufferlist out_bl;
 949   int r = ioctx->operate(oid, &op, &out_bl);
 950   if (r < 0) {
 951     return r;
 952   }
 953
 954   auto it = out_bl.cbegin();
 955   return get_stripe_unit_count_finish(&it, stripe_unit, stripe_count);
 956 }
 957
 958 void set_stripe_unit_count(librados::ObjectWriteOperation *op,
 959                            uint64_t stripe_unit, uint64_t stripe_count)
 960 {
 961   bufferlist bl;
 962   encode(stripe_unit, bl);
 963   encode(stripe_count, bl);
 964
 965   op->exec("rbd", "set_stripe_unit_count", bl);
 966 }
 967
 968 int set_stripe_unit_count(librados::IoCtx *ioctx, const std::string &oid,
 969                           uint64_t stripe_unit, uint64_t stripe_count)
 970 {
 971   librados::ObjectWriteOperation op;
 972   set_stripe_unit_count(&op, stripe_unit, stripe_count);
 973
 974   return ioctx->operate(oid, &op);
 975 }
 976
 977 void get_create_timestamp_start(librados::ObjectReadOperation *op) {
 978   bufferlist empty_bl;
 979   op->exec("rbd", "get_create_timestamp", empty_bl);
 980 }
 981
 982 int get_create_timestamp_finish(bufferlist::const_iterator *it,
 983                                 utime_t *timestamp) {
 984   ceph_assert(timestamp);
 985
 986   try {
 987     decode(*timestamp, *it);
 988   } catch (const buffer::error &err) {
 989     return -EBADMSG;
 990   }
 991   return 0;
 992 }
 993
 994 int get_create_timestamp(librados::IoCtx *ioctx, const std::string &oid,
 995                          utime_t *timestamp)
 996 {
 997   librados::ObjectReadOperation op;
 998   get_create_timestamp_start(&op);
 999
1000   bufferlist out_bl;
1001   int r = ioctx->operate(oid, &op, &out_bl);
1002   if (r < 0) {
1003     return r;
1004   }
1005
1006   auto it = out_bl.cbegin();
1007   return get_create_timestamp_finish(&it, timestamp);
1008 }
1009
1010 void get_access_timestamp_start(librados::ObjectReadOperation *op) {
1011   bufferlist empty_bl;
1012   op->exec("rbd", "get_access_timestamp", empty_bl);
1013 }
1014
1015 int get_access_timestamp_finish(bufferlist::const_iterator *it,
1016                                 utime_t *timestamp) {
1017   ceph_assert(timestamp);
1018
1019   try {
1020     decode(*timestamp, *it);
1021   } catch (const buffer::error &err) {
1022     return -EBADMSG;
1023   }
1024   return 0;
1025 }
1026
1027 int get_access_timestamp(librados::IoCtx *ioctx, const std::string &oid,
1028                          utime_t *timestamp)
1029 {
1030   librados::ObjectReadOperation op;
1031   get_access_timestamp_start(&op);
1032
1033   bufferlist out_bl;
1034   int r = ioctx->operate(oid, &op, &out_bl);
1035   if (r < 0) {
1036     return r;
1037   }
1038
1039   auto it = out_bl.cbegin();
1040   return get_access_timestamp_finish(&it, timestamp);
1041 }
1042
1043 void set_access_timestamp(librados::ObjectWriteOperation *op)
1044 {
1045     bufferlist empty_bl;
1046     op->exec("rbd","set_access_timestamp",empty_bl);
1047 }
1048
1049 int set_access_timestamp(librados::IoCtx *ioctx, const std::string &oid)
1050 {
1051     librados::ObjectWriteOperation op;
1052     set_access_timestamp(&op);
1053     return ioctx->operate(oid, &op);
1054 }
1055
1056 void get_modify_timestamp_start(librados::ObjectReadOperation *op) {
1057   bufferlist empty_bl;
1058   op->exec("rbd", "get_modify_timestamp", empty_bl);
1059 }
1060
1061 int get_modify_timestamp_finish(bufferlist::const_iterator *it,
1062                                   utime_t *timestamp) {
1063   ceph_assert(timestamp);
1064
1065   try {
1066     decode(*timestamp, *it);
1067   } catch (const buffer::error &err) {
1068     return -EBADMSG;
1069   }
1070   return 0;
1071 }
1072
1073 int get_modify_timestamp(librados::IoCtx *ioctx, const std::string &oid,
1074                            utime_t *timestamp)
1075 {
1076   librados::ObjectReadOperation op;
1077   get_modify_timestamp_start(&op);
1078
1079   bufferlist out_bl;
1080   int r = ioctx->operate(oid, &op, &out_bl);
1081   if (r < 0) {
1082     return r;
1083   }
1084
1085   auto it = out_bl.cbegin();
1086   return get_modify_timestamp_finish(&it, timestamp);
1087 }
1088
1089 void set_modify_timestamp(librados::ObjectWriteOperation *op)
1090 {
1091     bufferlist empty_bl;
1092     op->exec("rbd","set_modify_timestamp",empty_bl);
1093 }
1094
1095 int set_modify_timestamp(librados::IoCtx *ioctx, const std::string &oid)
1096 {
1097     librados::ObjectWriteOperation op;
1098     set_modify_timestamp(&op);
1099     return ioctx->operate(oid, &op);
1100 }
1101
1102
1103 /************************ rbd_id object methods ************************/
1104
1105 void get_id_start(librados::ObjectReadOperation *op) {
1106   bufferlist empty_bl;
1107   op->exec("rbd", "get_id", empty_bl);
1108 }
1109
1110 int get_id_finish(bufferlist::const_iterator *it, std::string *id) {
1111   try {
1112     decode(*id, *it);
1113   } catch (const buffer::error &err) {
1114     return -EBADMSG;
1115   }
1116   return 0;
1117 }
1118
1119 int get_id(librados::IoCtx *ioctx, const std::string &oid, std::string *id)
1120 {
1121   librados::ObjectReadOperation op;
1122   get_id_start(&op);
1123
1124   bufferlist out_bl;
1125   int r = ioctx->operate(oid, &op, &out_bl);
1126   if (r < 0) {
1127     return r;
1128   }
1129
1130   auto it = out_bl.cbegin();
1131   return get_id_finish(&it, id);
1132 }
1133
1134 void set_id(librados::ObjectWriteOperation *op, const std::string &id)
1135 {
1136   bufferlist bl;
1137   encode(id, bl);
1138   op->exec("rbd", "set_id", bl);
1139 }
1140
1141 int set_id(librados::IoCtx *ioctx, const std::string &oid, const std::string &id)
1142 {
1143   librados::ObjectWriteOperation op;
1144   set_id(&op, id);
1145
1146   return ioctx->operate(oid, &op);
1147 }
1148
1149 /******************** rbd_directory object methods ********************/
1150
1151 void dir_get_id_start(librados::ObjectReadOperation *op,
1152                       const std::string &image_name) {
1153   bufferlist bl;
1154   encode(image_name, bl);
1155
1156   op->exec("rbd", "dir_get_id", bl);
1157 }
1158
1159 int dir_get_id_finish(bufferlist::const_iterator *iter, std::string *image_id) {
1160   try {
1161     decode(*image_id, *iter);
1162   } catch (const buffer::error &err) {
1163     return -EBADMSG;
1164   }
1165
1166   return 0;
1167 }
1168
1169 int dir_get_id(librados::IoCtx *ioctx, const std::string &oid,
1170                const std::string &name, std::string *id) {
1171   librados::ObjectReadOperation op;
1172   dir_get_id_start(&op, name);
1173
1174   bufferlist out_bl;
1175   int r = ioctx->operate(oid, &op, &out_bl);
1176   if (r < 0) {
1177     return r;
1178   }
1179
1180   auto iter = out_bl.cbegin();
1181   return dir_get_id_finish(&iter, id);
1182 }
1183
1184 void dir_get_name_start(librados::ObjectReadOperation *op,
1185                         const std::string &id) {
1186   bufferlist in_bl;
1187   encode(id, in_bl);
1188   op->exec("rbd", "dir_get_name", in_bl);
1189 }
1190
1191 int dir_get_name_finish(bufferlist::const_iterator *it, std::string *name) {
1192   try {
1193     decode(*name, *it);
1194   } catch (const buffer::error &err) {
1195     return -EBADMSG;
1196   }
1197   return 0;
1198 }
1199
1200 int dir_get_name(librados::IoCtx *ioctx, const std::string &oid,
1201                  const std::string &id, std::string *name) {
1202   librados::ObjectReadOperation op;
1203   dir_get_name_start(&op, id);
1204
1205   bufferlist out_bl;
1206   int r = ioctx->operate(oid, &op, &out_bl);
1207   if (r < 0) {
1208     return r;
1209   }
1210
1211   auto it = out_bl.cbegin();
1212   return dir_get_name_finish(&it, name);
1213 }
1214
1215 void dir_list_start(librados::ObjectReadOperation *op,
1216                     const std::string &start, uint64_t max_return)
1217 {
1218   bufferlist in_bl;
1219   encode(start, in_bl);
1220   encode(max_return, in_bl);
1221
1222   op->exec("rbd", "dir_list", in_bl);
1223 }
1224
1225 int dir_list_finish(bufferlist::const_iterator *it, map<string, string> *images)
1226 {
1227   try {
1228     decode(*images, *it);
1229   } catch (const buffer::error &err) {
1230     return -EBADMSG;
1231   }
1232   return 0;
1233 }
1234
1235 int dir_list(librados::IoCtx *ioctx, const std::string &oid,
1236              const std::string &start, uint64_t max_return,
1237              map<string, string> *images)
1238 {
1239   librados::ObjectReadOperation op;
1240   dir_list_start(&op, start, max_return);
1241
1242   bufferlist out_bl;
1243   int r = ioctx->operate(oid, &op, &out_bl);
1244   if (r < 0) {
1245     return r;
1246   }
1247
1248   auto iter = out_bl.cbegin();
1249   return dir_list_finish(&iter, images);
1250 }
1251
1252 void dir_add_image(librados::ObjectWriteOperation *op,
1253                    const std::string &name, const std::string &id)
1254 {
1255   bufferlist bl;
1256   encode(name, bl);
1257   encode(id, bl);
1258   op->exec("rbd", "dir_add_image", bl);
1259 }
1260
1261 int dir_add_image(librados::IoCtx *ioctx, const std::string &oid,
1262                   const std::string &name, const std::string &id)
1263 {
1264   librados::ObjectWriteOperation op;
1265   dir_add_image(&op, name, id);
1266
1267   return ioctx->operate(oid, &op);
1268 }
1269
1270 int dir_remove_image(librados::IoCtx *ioctx, const std::string &oid,
1271                      const std::string &name, const std::string &id)
1272 {
1273   librados::ObjectWriteOperation op;
1274   dir_remove_image(&op, name, id);
1275
1276   return ioctx->operate(oid, &op);
1277 }
1278
1279 void dir_state_assert(librados::ObjectOperation *op,
1280                       cls::rbd::DirectoryState directory_state)
1281 {
1282   bufferlist bl;
1283   encode(directory_state, bl);
1284   op->exec("rbd", "dir_state_assert", bl);
1285 }
1286
1287 int dir_state_assert(librados::IoCtx *ioctx, const std::string &oid,
1288                      cls::rbd::DirectoryState directory_state)
1289 {
1290   librados::ObjectWriteOperation op;
1291   dir_state_assert(&op, directory_state);
1292
1293   return ioctx->operate(oid, &op);
1294 }
1295
1296 void dir_state_set(librados::ObjectWriteOperation *op,
1297                    cls::rbd::DirectoryState directory_state)
1298 {
1299   bufferlist bl;
1300   encode(directory_state, bl);
1301   op->exec("rbd", "dir_state_set", bl);
1302 }
1303
1304 int dir_state_set(librados::IoCtx *ioctx, const std::string &oid,
1305                   cls::rbd::DirectoryState directory_state)
1306 {
1307   librados::ObjectWriteOperation op;
1308   dir_state_set(&op, directory_state);
1309
1310   return ioctx->operate(oid, &op);
1311 }
1312
1313 void dir_remove_image(librados::ObjectWriteOperation *op,
1314                       const std::string &name, const std::string &id)
1315 {
1316   bufferlist bl;
1317   encode(name, bl);
1318   encode(id, bl);
1319
1320   op->exec("rbd", "dir_remove_image", bl);
1321 }
1322
1323 void dir_rename_image(librados::ObjectWriteOperation *op,
1324                       const std::string &src, const std::string &dest,
1325                       const std::string &id)
1326 {
1327   bufferlist in;
1328   encode(src, in);
1329   encode(dest, in);
1330   encode(id, in);
1331   op->exec("rbd", "dir_rename_image", in);
1332 }
1333
1334 void object_map_load_start(librados::ObjectReadOperation *op) {
1335   bufferlist in_bl;
1336   op->exec("rbd", "object_map_load", in_bl);
1337 }
1338
1339 int object_map_load_finish(bufferlist::const_iterator *it,
1340                            ceph::BitVector<2> *object_map) {
1341   try {
1342     decode(*object_map, *it);
1343   } catch (const buffer::error &err) {
1344     return -EBADMSG;
1345   }
1346   return 0;
1347 }
1348
1349 int object_map_load(librados::IoCtx *ioctx, const std::string &oid,
1350                     ceph::BitVector<2> *object_map)
1351 {
1352   librados::ObjectReadOperation op;
1353   object_map_load_start(&op);
1354
1355   bufferlist out_bl;
1356   int r = ioctx->operate(oid, &op, &out_bl);
1357   if (r < 0) {
1358     return r;
1359   }
1360
1361   auto it = out_bl.cbegin();
1362   return object_map_load_finish(&it, object_map);
1363 }
1364
1365 void object_map_save(librados::ObjectWriteOperation *rados_op,
1366                      const ceph::BitVector<2> &object_map)
1367 {
1368   ceph::BitVector<2> object_map_copy(object_map);
1369   object_map_copy.set_crc_enabled(false);
1370
1371   bufferlist in;
1372   encode(object_map_copy, in);
1373   rados_op->exec("rbd", "object_map_save", in);
1374 }
1375
1376 void object_map_resize(librados::ObjectWriteOperation *rados_op,
1377                        uint64_t object_count, uint8_t default_state)
1378 {
1379   bufferlist in;
1380   encode(object_count, in);
1381   encode(default_state, in);
1382   rados_op->exec("rbd", "object_map_resize", in);
1383 }
1384
1385 void object_map_update(librados::ObjectWriteOperation *rados_op,
1386                        uint64_t start_object_no, uint64_t end_object_no,
1387                        uint8_t new_object_state,
1388                        const boost::optional<uint8_t> &current_object_state)
1389 {
1390   bufferlist in;
1391   encode(start_object_no, in);
1392   encode(end_object_no, in);
1393   encode(new_object_state, in);
1394   encode(current_object_state, in);
1395   rados_op->exec("rbd", "object_map_update", in);
1396 }
1397
1398 void object_map_snap_add(librados::ObjectWriteOperation *rados_op)
1399 {
1400   bufferlist in;
1401   rados_op->exec("rbd", "object_map_snap_add", in);
1402 }
1403
1404 void object_map_snap_remove(librados::ObjectWriteOperation *rados_op,
1405                             const ceph::BitVector<2> &object_map)
1406 {
1407   ceph::BitVector<2> object_map_copy(object_map);
1408   object_map_copy.set_crc_enabled(false);
1409
1410   bufferlist in;
1411   encode(object_map_copy, in);
1412   rados_op->exec("rbd", "object_map_snap_remove", in);
1413 }
1414
1415 void metadata_set(librados::ObjectWriteOperation *op,
1416                   const map<string, bufferlist> &data)
1417 {
1418   bufferlist bl;
1419   encode(data, bl);
1420
1421   op->exec("rbd", "metadata_set", bl);
1422 }
1423
1424 int metadata_set(librados::IoCtx *ioctx, const std::string &oid,
1425                  const map<string, bufferlist> &data)
1426 {
1427   librados::ObjectWriteOperation op;
1428   metadata_set(&op, data);
1429
1430   return ioctx->operate(oid, &op);
1431 }
1432
1433 void metadata_remove(librados::ObjectWriteOperation *op,
1434                      const std::string &key)
1435 {
1436   bufferlist bl;
1437   encode(key, bl);
1438
1439   op->exec("rbd", "metadata_remove", bl);
1440 }
1441
1442 int metadata_remove(librados::IoCtx *ioctx, const std::string &oid,
1443                     const std::string &key)
1444 {
1445   librados::ObjectWriteOperation op;
1446   metadata_remove(&op, key);
1447
1448   return ioctx->operate(oid, &op);
1449 }
1450
1451 int metadata_list(librados::IoCtx *ioctx, const std::string &oid,
1452                   const std::string &start, uint64_t max_return,
1453                   map<string, bufferlist> *pairs)
1454 {
1455   librados::ObjectReadOperation op;
1456   metadata_list_start(&op, start, max_return);
1457
1458   bufferlist out_bl;
1459   int r = ioctx->operate(oid, &op, &out_bl);
1460   if (r < 0) {
1461     return r;
1462   }
1463
1464   auto it = out_bl.cbegin();
1465   return metadata_list_finish(&it, pairs);
1466 }
1467
1468 void metadata_list_start(librados::ObjectReadOperation *op,
1469                          const std::string &start, uint64_t max_return)
1470 {
1471   bufferlist in_bl;
1472   encode(start, in_bl);
1473   encode(max_return, in_bl);
1474   op->exec("rbd", "metadata_list", in_bl);
1475 }
1476
1477 int metadata_list_finish(bufferlist::const_iterator *it,
1478                          std::map<std::string, bufferlist> *pairs)
1479 {
1480   ceph_assert(pairs);
1481   try {
1482     decode(*pairs, *it);
1483   } catch (const buffer::error &err) {
1484     return -EBADMSG;
1485   }
1486   return 0;
1487 }
1488
1489 int metadata_get(librados::IoCtx *ioctx, const std::string &oid,
1490                  const std::string &key, string *s)
1491 {
1492   ceph_assert(s);
1493   bufferlist in, out;
1494   encode(key, in);
1495   int r = ioctx->exec(oid, "rbd", "metadata_get", in, out);
1496   if (r < 0)
1497     return r;
1498
1499   auto iter = out.cbegin();
1500   try {
1501     decode(*s, iter);
1502   } catch (const buffer::error &err) {
1503     return -EBADMSG;
1504   }
1505
1506   return 0;
1507 }
1508
1509 void child_attach(librados::ObjectWriteOperation *op, snapid_t snap_id,
1510                   const cls::rbd::ChildImageSpec& child_image)
1511 {
1512   bufferlist bl;
1513   encode(snap_id, bl);
1514   encode(child_image, bl);
1515   op->exec("rbd", "child_attach", bl);
1516 }
1517
1518 int child_attach(librados::IoCtx *ioctx, const std::string &oid,
1519                  snapid_t snap_id,
1520                  const cls::rbd::ChildImageSpec& child_image)
1521 {
1522   librados::ObjectWriteOperation op;
1523   child_attach(&op, snap_id, child_image);
1524
1525   int r = ioctx->operate(oid, &op);
1526   if (r < 0) {
1527     return r;
1528   }
1529   return 0;
1530 }
1531
1532 void child_detach(librados::ObjectWriteOperation *op, snapid_t snap_id,
1533                   const cls::rbd::ChildImageSpec& child_image)
1534 {
1535   bufferlist bl;
1536   encode(snap_id, bl);
1537   encode(child_image, bl);
1538   op->exec("rbd", "child_detach", bl);
1539 }
1540
1541 int child_detach(librados::IoCtx *ioctx, const std::string &oid,
1542                  snapid_t snap_id,
1543                  const cls::rbd::ChildImageSpec& child_image)
1544 {
1545   librados::ObjectWriteOperation op;
1546   child_detach(&op, snap_id, child_image);
1547
1548   int r = ioctx->operate(oid, &op);
1549   if (r < 0) {
1550     return r;
1551   }
1552   return 0;
1553 }
1554
1555 void children_list_start(librados::ObjectReadOperation *op,
1556                          snapid_t snap_id)
1557 {
1558   bufferlist bl;
1559   encode(snap_id, bl);
1560   op->exec("rbd", "children_list", bl);
1561 }
1562
1563 int children_list_finish(bufferlist::const_iterator *it,
1564                          cls::rbd::ChildImageSpecs *child_images)
1565 {
1566   child_images->clear();
1567   try {
1568     decode(*child_images, *it);
1569   } catch (const buffer::error &err) {
1570     return -EBADMSG;
1571   }
1572   return 0;
1573 }
1574
1575 int children_list(librados::IoCtx *ioctx, const std::string &oid,
1576                   snapid_t snap_id,
1577                   cls::rbd::ChildImageSpecs *child_images)
1578 {
1579   librados::ObjectReadOperation op;
1580   children_list_start(&op, snap_id);
1581
1582   bufferlist out_bl;
1583   int r = ioctx->operate(oid, &op, &out_bl);
1584   if (r < 0) {
1585     return r;
1586   }
1587
1588   auto it = out_bl.cbegin();
1589   r = children_list_finish(&it, child_images);
1590   if (r < 0) {
1591     return r;
1592   }
1593   return 0;
1594 }
1595
1596 int migration_set(librados::IoCtx *ioctx, const std::string &oid,
1597                   const cls::rbd::MigrationSpec &migration_spec) {
1598   librados::ObjectWriteOperation op;
1599   migration_set(&op, migration_spec);
1600   return ioctx->operate(oid, &op);
1601 }
1602
1603 void migration_set(librados::ObjectWriteOperation *op,
1604                    const cls::rbd::MigrationSpec &migration_spec) {
1605   bufferlist bl;
1606   encode(migration_spec, bl);
1607   op->exec("rbd", "migration_set", bl);
1608 }
1609
1610 int migration_set_state(librados::IoCtx *ioctx, const std::string &oid,
1611                         cls::rbd::MigrationState state,
1612                         const std::string &description) {
1613   librados::ObjectWriteOperation op;
1614   migration_set_state(&op, state, description);
1615   return ioctx->operate(oid, &op);
1616 }
1617
1618 void migration_set_state(librados::ObjectWriteOperation *op,
1619                          cls::rbd::MigrationState state,
1620                          const std::string &description) {
1621   bufferlist bl;
1622   encode(state, bl);
1623   encode(description, bl);
1624   op->exec("rbd", "migration_set_state", bl);
1625 }
1626
1627 void migration_get_start(librados::ObjectReadOperation *op) {
1628   bufferlist bl;
1629   op->exec("rbd", "migration_get", bl);
1630 }
1631
1632 int migration_get_finish(bufferlist::const_iterator *it,
1633                          cls::rbd::MigrationSpec *migration_spec) {
1634   try {
1635     decode(*migration_spec, *it);
1636   } catch (const buffer::error &err) {
1637     return -EBADMSG;
1638   }
1639   return 0;
1640 }
1641
1642 int migration_get(librados::IoCtx *ioctx, const std::string &oid,
1643                   cls::rbd::MigrationSpec *migration_spec) {
1644   librados::ObjectReadOperation op;
1645   migration_get_start(&op);
1646
1647   bufferlist out_bl;
1648   int r = ioctx->operate(oid, &op, &out_bl);
1649   if (r < 0) {
1650     return r;
1651   }
1652
1653   auto iter = out_bl.cbegin();
1654   r = migration_get_finish(&iter, migration_spec);
1655   if (r < 0) {
1656     return r;
1657   }
1658   return 0;
1659 }
1660
1661 int migration_remove(librados::IoCtx *ioctx, const std::string &oid) {
1662   librados::ObjectWriteOperation op;
1663   migration_remove(&op);
1664   return ioctx->operate(oid, &op);
1665 }
1666
1667 void migration_remove(librados::ObjectWriteOperation *op) {
1668   bufferlist bl;
1669   op->exec("rbd", "migration_remove", bl);
1670 }
1671
1672 int assert_snapc_seq(librados::IoCtx *ioctx, const std::string &oid,
1673                      uint64_t snapc_seq,
1674                      cls::rbd::AssertSnapcSeqState state) {
1675   librados::ObjectWriteOperation op;
1676   assert_snapc_seq(&op, snapc_seq, state);
1677   return ioctx->operate(oid, &op);
1678 }
1679
1680 void assert_snapc_seq(librados::ObjectWriteOperation *op,
1681                       uint64_t snapc_seq,
1682                       cls::rbd::AssertSnapcSeqState state) {
1683   bufferlist bl;
1684   encode(snapc_seq, bl);
1685   encode(state, bl);
1686   op->exec("rbd", "assert_snapc_seq", bl);
1687 }
1688
1689 void mirror_uuid_get_start(librados::ObjectReadOperation *op) {
1690   bufferlist bl;
1691   op->exec("rbd", "mirror_uuid_get", bl);
1692 }
1693
1694 int mirror_uuid_get_finish(bufferlist::const_iterator *it,
1695                            std::string *uuid) {
1696   try {
1697     decode(*uuid, *it);
1698   } catch (const buffer::error &err) {
1699     return -EBADMSG;
1700   }
1701   return 0;
1702 }
1703
1704 int mirror_uuid_get(librados::IoCtx *ioctx, std::string *uuid) {
1705   librados::ObjectReadOperation op;
1706   mirror_uuid_get_start(&op);
1707
1708   bufferlist out_bl;
1709   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1710   if (r < 0) {
1711     return r;
1712   }
1713
1714   auto it = out_bl.cbegin();
1715   r = mirror_uuid_get_finish(&it, uuid);
1716   if (r < 0) {
1717     return r;
1718   }
1719   return 0;
1720 }
1721
1722 int mirror_uuid_set(librados::IoCtx *ioctx, const std::string &uuid) {
1723   bufferlist in_bl;
1724   encode(uuid, in_bl);
1725
1726   bufferlist out_bl;
1727   int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_uuid_set", in_bl,
1728                       out_bl);
1729   if (r < 0) {
1730     return r;
1731   }
1732   return 0;
1733 }
1734
1735 void mirror_mode_get_start(librados::ObjectReadOperation *op) {
1736   bufferlist bl;
1737   op->exec("rbd", "mirror_mode_get", bl);
1738 }
1739
1740 int mirror_mode_get_finish(bufferlist::const_iterator *it,
1741                            cls::rbd::MirrorMode *mirror_mode) {
1742   try {
1743     uint32_t mirror_mode_decode;
1744     decode(mirror_mode_decode, *it);
1745     *mirror_mode = static_cast<cls::rbd::MirrorMode>(mirror_mode_decode);
1746   } catch (const buffer::error &err) {
1747     return -EBADMSG;
1748   }
1749
1750   return 0;
1751 }
1752
1753 int mirror_mode_get(librados::IoCtx *ioctx,
1754                     cls::rbd::MirrorMode *mirror_mode) {
1755   librados::ObjectReadOperation op;
1756   mirror_mode_get_start(&op);
1757
1758   bufferlist out_bl;
1759   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1760   if (r == -ENOENT) {
1761     *mirror_mode = cls::rbd::MIRROR_MODE_DISABLED;
1762     return 0;
1763   } else if (r < 0) {
1764     return r;
1765   }
1766
1767   auto it = out_bl.cbegin();
1768   r = mirror_mode_get_finish(&it, mirror_mode);
1769   if (r < 0) {
1770     return r;
1771   }
1772   return 0;
1773 }
1774
1775 int mirror_mode_set(librados::IoCtx *ioctx,
1776                     cls::rbd::MirrorMode mirror_mode) {
1777   bufferlist in_bl;
1778   encode(static_cast<uint32_t>(mirror_mode), in_bl);
1779
1780   bufferlist out_bl;
1781   int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_mode_set", in_bl,
1782                       out_bl);
1783   if (r < 0) {
1784     return r;
1785   }
1786   return 0;
1787 }
1788
1789 int mirror_peer_list(librados::IoCtx *ioctx,
1790                      std::vector<cls::rbd::MirrorPeer> *peers) {
1791   bufferlist in_bl;
1792   bufferlist out_bl;
1793   int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_list", in_bl,
1794                       out_bl);
1795   if (r < 0) {
1796     return r;
1797   }
1798
1799   peers->clear();
1800   try {
1801     auto bl_it = out_bl.cbegin();
1802     decode(*peers, bl_it);
1803   } catch (const buffer::error &err) {
1804     return -EBADMSG;
1805   }
1806   return 0;
1807 }
1808
1809 int mirror_peer_add(librados::IoCtx *ioctx, const std::string &uuid,
1810                     const std::string &cluster_name,
1811                     const std::string &client_name, int64_t pool_id) {
1812   cls::rbd::MirrorPeer peer(uuid, cluster_name, client_name, pool_id);
1813   bufferlist in_bl;
1814   encode(peer, in_bl);
1815
1816   bufferlist out_bl;
1817   int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_add", in_bl,
1818                       out_bl);
1819   if (r < 0) {
1820     return r;
1821   }
1822   return 0;
1823 }
1824
1825 int mirror_peer_remove(librados::IoCtx *ioctx,
1826                        const std::string &uuid) {
1827   bufferlist in_bl;
1828   encode(uuid, in_bl);
1829
1830   bufferlist out_bl;
1831   int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_remove", in_bl,
1832                       out_bl);
1833   if (r < 0) {
1834     return r;
1835   }
1836   return 0;
1837 }
1838
1839 int mirror_peer_set_client(librados::IoCtx *ioctx,
1840                            const std::string &uuid,
1841                            const std::string &client_name) {
1842   bufferlist in_bl;
1843   encode(uuid, in_bl);
1844   encode(client_name, in_bl);
1845
1846   bufferlist out_bl;
1847   int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_set_client",
1848                       in_bl, out_bl);
1849   if (r < 0) {
1850     return r;
1851   }
1852   return 0;
1853 }
1854
1855 int mirror_peer_set_cluster(librados::IoCtx *ioctx,
1856                             const std::string &uuid,
1857                             const std::string &cluster_name) {
1858   bufferlist in_bl;
1859   encode(uuid, in_bl);
1860   encode(cluster_name, in_bl);
1861
1862   bufferlist out_bl;
1863   int r = ioctx->exec(RBD_MIRRORING, "rbd", "mirror_peer_set_cluster",
1864                       in_bl, out_bl);
1865   if (r < 0) {
1866     return r;
1867   }
1868   return 0;
1869 }
1870
1871 void mirror_image_list_start(librados::ObjectReadOperation *op,
1872                              const std::string &start, uint64_t max_return)
1873 {
1874   bufferlist in_bl;
1875   encode(start, in_bl);
1876   encode(max_return, in_bl);
1877   op->exec("rbd", "mirror_image_list", in_bl);
1878 }
1879
1880 int mirror_image_list_finish(bufferlist::const_iterator *it,
1881                              std::map<string, string> *mirror_image_ids)
1882 {
1883   try {
1884     decode(*mirror_image_ids, *it);
1885   } catch (const buffer::error &err) {
1886     return -EBADMSG;
1887   }
1888   return 0;
1889 }
1890
1891 int mirror_image_list(librados::IoCtx *ioctx,
1892                       const std::string &start, uint64_t max_return,
1893                       std::map<std::string, std::string> *mirror_image_ids) {
1894   librados::ObjectReadOperation op;
1895   mirror_image_list_start(&op, start, max_return);
1896
1897   bufferlist out_bl;
1898   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1899   if (r < 0) {
1900     return r;
1901   }
1902
1903   auto bl_it = out_bl.cbegin();
1904   return mirror_image_list_finish(&bl_it, mirror_image_ids);
1905 }
1906
1907 void mirror_image_get_image_id_start(librados::ObjectReadOperation *op,
1908                                      const std::string &global_image_id) {
1909   bufferlist in_bl;
1910   encode(global_image_id, in_bl);
1911   op->exec( "rbd", "mirror_image_get_image_id", in_bl);
1912 }
1913
1914 int mirror_image_get_image_id_finish(bufferlist::const_iterator *it,
1915                                      std::string *image_id) {
1916   try {
1917     decode(*image_id, *it);
1918   } catch (const buffer::error &err) {
1919     return -EBADMSG;
1920   }
1921   return 0;
1922 }
1923
1924 int mirror_image_get_image_id(librados::IoCtx *ioctx,
1925                               const std::string &global_image_id,
1926                               std::string *image_id) {
1927   librados::ObjectReadOperation op;
1928   mirror_image_get_image_id_start(&op, global_image_id);
1929
1930   bufferlist out_bl;
1931   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1932   if (r < 0) {
1933     return r;
1934   }
1935
1936   auto it = out_bl.cbegin();
1937   return mirror_image_get_image_id_finish(&it, image_id);
1938 }
1939
1940 int mirror_image_get(librados::IoCtx *ioctx, const std::string &image_id,
1941                      cls::rbd::MirrorImage *mirror_image) {
1942   librados::ObjectReadOperation op;
1943   mirror_image_get_start(&op, image_id);
1944
1945   bufferlist out_bl;
1946   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
1947   if (r < 0) {
1948     return r;
1949   }
1950
1951   auto iter = out_bl.cbegin();
1952   r = mirror_image_get_finish(&iter, mirror_image);
1953   if (r < 0) {
1954     return r;
1955   }
1956   return 0;
1957 }
1958
1959 void mirror_image_get_start(librados::ObjectReadOperation *op,
1960                             const std::string &image_id) {
1961   bufferlist in_bl;
1962   encode(image_id, in_bl);
1963
1964   op->exec("rbd", "mirror_image_get", in_bl);
1965 }
1966
1967 int mirror_image_get_finish(bufferlist::const_iterator *iter,
1968                             cls::rbd::MirrorImage *mirror_image) {
1969   try {
1970     decode(*mirror_image, *iter);
1971   } catch (const buffer::error &err) {
1972     return -EBADMSG;
1973   }
1974   return 0;
1975 }
1976
1977 void mirror_image_set(librados::ObjectWriteOperation *op,
1978                       const std::string &image_id,
1979                       const cls::rbd::MirrorImage &mirror_image) {
1980   bufferlist bl;
1981   encode(image_id, bl);
1982   encode(mirror_image, bl);
1983
1984   op->exec("rbd", "mirror_image_set", bl);
1985 }
1986
1987 int mirror_image_set(librados::IoCtx *ioctx, const std::string &image_id,
1988                      const cls::rbd::MirrorImage &mirror_image) {
1989   librados::ObjectWriteOperation op;
1990   mirror_image_set(&op, image_id, mirror_image);
1991
1992   int r = ioctx->operate(RBD_MIRRORING, &op);
1993   if (r < 0) {
1994     return r;
1995   }
1996   return 0;
1997 }
1998
1999 void mirror_image_remove(librados::ObjectWriteOperation *op,
2000                          const std::string &image_id) {
2001   bufferlist bl;
2002   encode(image_id, bl);
2003
2004   op->exec("rbd", "mirror_image_remove", bl);
2005 }
2006
2007 int mirror_image_remove(librados::IoCtx *ioctx, const std::string &image_id) {
2008   librados::ObjectWriteOperation op;
2009   mirror_image_remove(&op, image_id);
2010
2011   int r = ioctx->operate(RBD_MIRRORING, &op);
2012   if (r < 0) {
2013     return r;
2014   }
2015   return 0;
2016 }
2017
2018 int mirror_image_status_set(librados::IoCtx *ioctx,
2019                             const std::string &global_image_id,
2020                             const cls::rbd::MirrorImageStatus &status) {
2021   librados::ObjectWriteOperation op;
2022   mirror_image_status_set(&op, global_image_id, status);
2023   return ioctx->operate(RBD_MIRRORING, &op);
2024 }
2025
2026 void mirror_image_status_set(librados::ObjectWriteOperation *op,
2027                              const std::string &global_image_id,
2028                              const cls::rbd::MirrorImageStatus &status) {
2029   bufferlist bl;
2030   encode(global_image_id, bl);
2031   encode(status, bl);
2032   op->exec("rbd", "mirror_image_status_set", bl);
2033 }
2034
2035 int mirror_image_status_remove(librados::IoCtx *ioctx,
2036                                const std::string &global_image_id) {
2037   librados::ObjectWriteOperation op;
2038   mirror_image_status_remove(&op, global_image_id);
2039   return ioctx->operate(RBD_MIRRORING, &op);
2040 }
2041
2042 void mirror_image_status_remove(librados::ObjectWriteOperation *op,
2043                                 const std::string &global_image_id) {
2044   bufferlist bl;
2045   encode(global_image_id, bl);
2046   op->exec("rbd", "mirror_image_status_remove", bl);
2047 }
2048
2049 int mirror_image_status_get(librados::IoCtx *ioctx,
2050                             const std::string &global_image_id,
2051                             cls::rbd::MirrorImageStatus *status) {
2052   librados::ObjectReadOperation op;
2053   mirror_image_status_get_start(&op, global_image_id);
2054
2055   bufferlist out_bl;
2056   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
2057   if (r < 0) {
2058     return r;
2059   }
2060
2061   auto iter = out_bl.cbegin();
2062   r = mirror_image_status_get_finish(&iter, status);
2063   if (r < 0) {
2064     return r;
2065   }
2066   return 0;
2067 }
2068
2069 void mirror_image_status_get_start(librados::ObjectReadOperation *op,
2070                                    const std::string &global_image_id) {
2071   bufferlist bl;
2072   encode(global_image_id, bl);
2073   op->exec("rbd", "mirror_image_status_get", bl);
2074 }
2075
2076 int mirror_image_status_get_finish(bufferlist::const_iterator *iter,
2077                                    cls::rbd::MirrorImageStatus *status) {
2078   try {
2079     decode(*status, *iter);
2080   } catch (const buffer::error &err) {
2081     return -EBADMSG;
2082   }
2083   return 0;
2084 }
2085
2086 int mirror_image_status_list(librados::IoCtx *ioctx,
2087                              const std::string &start, uint64_t max_return,
2088                              std::map<std::string, cls::rbd::MirrorImage> *images,
2089                              std::map<std::string, cls::rbd::MirrorImageStatus> *statuses) {
2090   librados::ObjectReadOperation op;
2091   mirror_image_status_list_start(&op, start, max_return);
2092
2093   bufferlist out_bl;
2094   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
2095   if (r < 0) {
2096     return r;
2097   }
2098
2099   auto iter = out_bl.cbegin();
2100   r = mirror_image_status_list_finish(&iter, images, statuses);
2101   if (r < 0) {
2102     return r;
2103   }
2104   return 0;
2105 }
2106
2107 void mirror_image_status_list_start(librados::ObjectReadOperation *op,
2108                                     const std::string &start,
2109                                     uint64_t max_return) {
2110   bufferlist bl;
2111   encode(start, bl);
2112   encode(max_return, bl);
2113   op->exec("rbd", "mirror_image_status_list", bl);
2114 }
2115
2116 int mirror_image_status_list_finish(bufferlist::const_iterator *iter,
2117                                     std::map<std::string, cls::rbd::MirrorImage> *images,
2118                                     std::map<std::string, cls::rbd::MirrorImageStatus> *statuses) {
2119   images->clear();
2120   statuses->clear();
2121   try {
2122     decode(*images, *iter);
2123     decode(*statuses, *iter);
2124   } catch (const buffer::error &err) {
2125     return -EBADMSG;
2126   }
2127   return 0;
2128 }
2129
2130 int mirror_image_status_get_summary(librados::IoCtx *ioctx,
2131                                     std::map<cls::rbd::MirrorImageStatusState, int> *states) {
2132   librados::ObjectReadOperation op;
2133   mirror_image_status_get_summary_start(&op);
2134
2135   bufferlist out_bl;
2136   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
2137   if (r < 0) {
2138     return r;
2139   }
2140
2141   auto iter = out_bl.cbegin();
2142   r = mirror_image_status_get_summary_finish(&iter, states);
2143   if (r < 0) {
2144     return r;
2145   }
2146   return 0;
2147 }
2148
2149 void mirror_image_status_get_summary_start(librados::ObjectReadOperation *op) {
2150   bufferlist bl;
2151   op->exec("rbd", "mirror_image_status_get_summary", bl);
2152 }
2153
2154 int mirror_image_status_get_summary_finish(bufferlist::const_iterator *iter,
2155                                            std::map<cls::rbd::MirrorImageStatusState, int> *states) {
2156   try {
2157     decode(*states, *iter);
2158   } catch (const buffer::error &err) {
2159     return -EBADMSG;
2160   }
2161   return 0;
2162 }
2163
2164 int mirror_image_status_remove_down(librados::IoCtx *ioctx) {
2165   librados::ObjectWriteOperation op;
2166   mirror_image_status_remove_down(&op);
2167   return ioctx->operate(RBD_MIRRORING, &op);
2168 }
2169
2170 void mirror_image_status_remove_down(librados::ObjectWriteOperation *op) {
2171   bufferlist bl;
2172   op->exec("rbd", "mirror_image_status_remove_down", bl);
2173 }
2174
2175 int mirror_image_instance_get(librados::IoCtx *ioctx,
2176                               const std::string &global_image_id,
2177                               entity_inst_t *instance) {
2178   librados::ObjectReadOperation op;
2179   mirror_image_instance_get_start(&op, global_image_id);
2180
2181   bufferlist out_bl;
2182   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
2183   if (r < 0) {
2184     return r;
2185   }
2186
2187   auto iter = out_bl.cbegin();
2188   r = mirror_image_instance_get_finish(&iter, instance);
2189   if (r < 0) {
2190     return r;
2191   }
2192   return 0;
2193 }
2194
2195 void mirror_image_instance_get_start(librados::ObjectReadOperation *op,
2196                                      const std::string &global_image_id) {
2197   bufferlist bl;
2198   encode(global_image_id, bl);
2199   op->exec("rbd", "mirror_image_instance_get", bl);
2200 }
2201
2202 int mirror_image_instance_get_finish(bufferlist::const_iterator *iter,
2203                                      entity_inst_t *instance) {
2204   try {
2205     decode(*instance, *iter);
2206   } catch (const buffer::error &err) {
2207     return -EBADMSG;
2208   }
2209   return 0;
2210 }
2211
2212 int mirror_image_instance_list(
2213     librados::IoCtx *ioctx, const std::string &start, uint64_t max_return,
2214     std::map<std::string, entity_inst_t> *instances) {
2215   librados::ObjectReadOperation op;
2216   mirror_image_instance_list_start(&op, start, max_return);
2217
2218   bufferlist out_bl;
2219   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
2220   if (r < 0) {
2221     return r;
2222   }
2223
2224   auto iter = out_bl.cbegin();
2225   r = mirror_image_instance_list_finish(&iter, instances);
2226   if (r < 0) {
2227     return r;
2228   }
2229   return 0;
2230 }
2231
2232 void mirror_image_instance_list_start(librados::ObjectReadOperation *op,
2233                                       const std::string &start,
2234                                       uint64_t max_return) {
2235   bufferlist bl;
2236   encode(start, bl);
2237   encode(max_return, bl);
2238   op->exec("rbd", "mirror_image_instance_list", bl);
2239 }
2240
2241 int mirror_image_instance_list_finish(
2242     bufferlist::const_iterator *iter,
2243     std::map<std::string, entity_inst_t> *instances) {
2244   instances->clear();
2245   try {
2246     decode(*instances, *iter);
2247   } catch (const buffer::error &err) {
2248     return -EBADMSG;
2249   }
2250   return 0;
2251 }
2252
2253 void mirror_instances_list_start(librados::ObjectReadOperation *op) {
2254   bufferlist bl;
2255   op->exec("rbd", "mirror_instances_list", bl);
2256 }
2257
2258 int mirror_instances_list_finish(bufferlist::const_iterator *iter,
2259                                  std::vector<std::string> *instance_ids) {
2260   instance_ids->clear();
2261   try {
2262     decode(*instance_ids, *iter);
2263   } catch (const buffer::error &err) {
2264     return -EBADMSG;
2265   }
2266   return 0;
2267 }
2268
2269 int mirror_instances_list(librados::IoCtx *ioctx,
2270                           std::vector<std::string> *instance_ids) {
2271   librados::ObjectReadOperation op;
2272   mirror_instances_list_start(&op);
2273
2274   bufferlist out_bl;
2275   int r = ioctx->operate(RBD_MIRROR_LEADER, &op, &out_bl);
2276   if (r < 0) {
2277     return r;
2278   }
2279
2280   auto iter = out_bl.cbegin();
2281   r = mirror_instances_list_finish(&iter, instance_ids);
2282   if (r < 0) {
2283     return r;
2284   }
2285   return 0;
2286 }
2287
2288 void mirror_instances_add(librados::ObjectWriteOperation *op,
2289                           const std::string &instance_id) {
2290   bufferlist bl;
2291   encode(instance_id, bl);
2292   op->exec("rbd", "mirror_instances_add", bl);
2293 }
2294
2295 int mirror_instances_add(librados::IoCtx *ioctx,
2296                          const std::string &instance_id) {
2297   librados::ObjectWriteOperation op;
2298   mirror_instances_add(&op, instance_id);
2299   return ioctx->operate(RBD_MIRROR_LEADER, &op);
2300 }
2301
2302 void mirror_instances_remove(librados::ObjectWriteOperation *op,
2303                              const std::string &instance_id) {
2304   bufferlist bl;
2305   encode(instance_id, bl);
2306   op->exec("rbd", "mirror_instances_remove", bl);
2307 }
2308
2309 int mirror_instances_remove(librados::IoCtx *ioctx,
2310                             const std::string &instance_id) {
2311   librados::ObjectWriteOperation op;
2312   mirror_instances_remove(&op, instance_id);
2313   return ioctx->operate(RBD_MIRROR_LEADER, &op);
2314 }
2315
2316 void mirror_image_map_list_start(librados::ObjectReadOperation *op,
2317                                  const std::string &start_after,
2318                                  uint64_t max_read) {
2319   bufferlist bl;
2320   encode(start_after, bl);
2321   encode(max_read, bl);
2322
2323   op->exec("rbd", "mirror_image_map_list", bl);
2324 }
2325
2326 int mirror_image_map_list_finish(bufferlist::const_iterator *iter,
2327                                  std::map<std::string, cls::rbd::MirrorImageMap> *image_mapping) {
2328   try {
2329     decode(*image_mapping, *iter);
2330   } catch (const buffer::error &err) {
2331     return -EBADMSG;
2332   }
2333   return 0;
2334 }
2335
2336 int mirror_image_map_list(
2337     librados::IoCtx *ioctx, const std::string &start_after,
2338     uint64_t max_read,
2339     std::map<std::string, cls::rbd::MirrorImageMap> *image_mapping) {
2340   librados::ObjectReadOperation op;
2341   mirror_image_map_list_start(&op, start_after, max_read);
2342
2343   bufferlist out_bl;
2344   int r = ioctx->operate(RBD_MIRRORING, &op, &out_bl);
2345   if (r < 0) {
2346     return r;
2347   }
2348
2349   auto iter = out_bl.cbegin();
2350   return mirror_image_map_list_finish(&iter, image_mapping);
2351 }
2352
2353 void mirror_image_map_update(librados::ObjectWriteOperation *op,
2354                              const std::string &global_image_id,
2355                              const cls::rbd::MirrorImageMap &image_map) {
2356   bufferlist bl;
2357   encode(global_image_id, bl);
2358   encode(image_map, bl);
2359
2360   op->exec("rbd", "mirror_image_map_update", bl);
2361 }
2362
2363 void mirror_image_map_remove(librados::ObjectWriteOperation *op,
2364                              const std::string &global_image_id) {
2365   bufferlist bl;
2366   encode(global_image_id, bl);
2367
2368   op->exec("rbd", "mirror_image_map_remove", bl);
2369 }
2370
2371 // Groups functions
2372 int group_dir_list(librados::IoCtx *ioctx, const std::string &oid,
2373                    const std::string &start, uint64_t max_return,
2374                    map<string, string> *cgs)
2375 {
2376   bufferlist in, out;
2377   encode(start, in);
2378   encode(max_return, in);
2379   int r = ioctx->exec(oid, "rbd", "group_dir_list", in, out);
2380   if (r < 0)
2381     return r;
2382
2383   auto iter = out.cbegin();
2384   try {
2385     decode(*cgs, iter);
2386   } catch (const buffer::error &err) {
2387     return -EBADMSG;
2388   }
2389
2390   return 0;
2391 }
2392
2393 int group_dir_add(librados::IoCtx *ioctx, const std::string &oid,
2394                   const std::string &name, const std::string &id)
2395 {
2396   bufferlist in, out;
2397   encode(name, in);
2398   encode(id, in);
2399   return ioctx->exec(oid, "rbd", "group_dir_add", in, out);
2400 }
2401
2402 int group_dir_rename(librados::IoCtx *ioctx, const std::string &oid,
2403                      const std::string &src, const std::string &dest,
2404                      const std::string &id)
2405 {
2406   bufferlist in, out;
2407   encode(src, in);
2408   encode(dest, in);
2409   encode(id, in);
2410   return ioctx->exec(oid, "rbd", "group_dir_rename", in, out);
2411 }
2412
2413 int group_dir_remove(librados::IoCtx *ioctx, const std::string &oid,
2414                      const std::string &name, const std::string &id)
2415 {
2416   bufferlist in, out;
2417   encode(name, in);
2418   encode(id, in);
2419   return ioctx->exec(oid, "rbd", "group_dir_remove", in, out);
2420 }
2421
2422 int group_image_remove(librados::IoCtx *ioctx, const std::string &oid,
2423                        const cls::rbd::GroupImageSpec &spec)
2424 {
2425   bufferlist bl, bl2;
2426   encode(spec, bl);
2427
2428   return ioctx->exec(oid, "rbd", "group_image_remove", bl, bl2);
2429 }
2430
2431 int group_image_list(librados::IoCtx *ioctx,
2432                      const std::string &oid,
2433                      const cls::rbd::GroupImageSpec &start,
2434                      uint64_t max_return,
2435                      std::vector<cls::rbd::GroupImageStatus> *images)
2436 {
2437   bufferlist bl, bl2;
2438   encode(start, bl);
2439   encode(max_return, bl);
2440
2441   int r = ioctx->exec(oid, "rbd", "group_image_list", bl, bl2);
2442   if (r < 0)
2443     return r;
2444
2445   auto iter = bl2.cbegin();
2446   try {
2447     decode(*images, iter);
2448   } catch (const buffer::error &err) {
2449     return -EBADMSG;
2450   }
2451
2452   return 0;
2453 }
2454
2455 int group_image_set(librados::IoCtx *ioctx, const std::string &oid,
2456                     const cls::rbd::GroupImageStatus &st)
2457 {
2458   bufferlist bl, bl2;
2459   encode(st, bl);
2460
2461   return ioctx->exec(oid, "rbd", "group_image_set", bl, bl2);
2462 }
2463
2464 int image_group_add(librados::IoCtx *ioctx, const std::string &oid,
2465                     const cls::rbd::GroupSpec &group_spec)
2466 {
2467   bufferlist bl, bl2;
2468   encode(group_spec, bl);
2469
2470   return ioctx->exec(oid, "rbd", "image_group_add", bl, bl2);
2471 }
2472
2473 int image_group_remove(librados::IoCtx *ioctx, const std::string &oid,
2474                        const cls::rbd::GroupSpec &group_spec)
2475 {
2476   bufferlist bl, bl2;
2477   encode(group_spec, bl);
2478
2479   return ioctx->exec(oid, "rbd", "image_group_remove", bl, bl2);
2480 }
2481
2482 void image_group_get_start(librados::ObjectReadOperation *op)
2483 {
2484   bufferlist in_bl;
2485   op->exec("rbd", "image_group_get", in_bl);
2486 }
2487
2488 int image_group_get_finish(bufferlist::const_iterator *iter,
2489                            cls::rbd::GroupSpec *group_spec)
2490 {
2491   try {
2492     decode(*group_spec, *iter);
2493   } catch (const buffer::error &err) {
2494     return -EBADMSG;
2495   }
2496   return 0;
2497 }
2498
2499 int image_group_get(librados::IoCtx *ioctx, const std::string &oid,
2500                     cls::rbd::GroupSpec *group_spec)
2501 {
2502   librados::ObjectReadOperation op;
2503   image_group_get_start(&op);
2504
2505   bufferlist out_bl;
2506   int r = ioctx->operate(oid, &op, &out_bl);
2507   if (r < 0) {
2508     return r;
2509   }
2510
2511   auto iter = out_bl.cbegin();
2512   return image_group_get_finish(&iter, group_spec);
2513 }
2514
2515 int group_snap_set(librados::IoCtx *ioctx, const std::string &oid,
2516                    const cls::rbd::GroupSnapshot &snapshot)
2517 {
2518   using ceph::encode;
2519   bufferlist inbl, outbl;
2520   encode(snapshot, inbl);
2521   int r = ioctx->exec(oid, "rbd", "group_snap_set", inbl, outbl);
2522   return r;
2523 }
2524
2525 int group_snap_remove(librados::IoCtx *ioctx, const std::string &oid,
2526                       const std::string &snap_id)
2527 {
2528   using ceph::encode;
2529   bufferlist inbl, outbl;
2530   encode(snap_id, inbl);
2531   return ioctx->exec(oid, "rbd", "group_snap_remove", inbl, outbl);
2532 }
2533
2534 int group_snap_get_by_id(librados::IoCtx *ioctx, const std::string &oid,
2535                          const std::string &snap_id,
2536                          cls::rbd::GroupSnapshot *snapshot)
2537 {
2538   using ceph::encode;
2539   using ceph::decode;
2540   bufferlist inbl, outbl;
2541
2542   encode(snap_id, inbl);
2543   int r = ioctx->exec(oid, "rbd", "group_snap_get_by_id", inbl, outbl);
2544   if (r < 0) {
2545     return r;
2546   }
2547
2548   auto iter = outbl.cbegin();
2549   try {
2550     decode(*snapshot, iter);
2551   } catch (const buffer::error &err) {
2552     return -EBADMSG;
2553   }
2554
2555   return 0;
2556 }
2557 int group_snap_list(librados::IoCtx *ioctx, const std::string &oid,
2558                     const cls::rbd::GroupSnapshot &start,
2559                     uint64_t max_return,
2560                     std::vector<cls::rbd::GroupSnapshot> *snapshots)
2561 {
2562   using ceph::encode;
2563   using ceph::decode;
2564   bufferlist inbl, outbl;
2565   encode(start, inbl);
2566   encode(max_return, inbl);
2567
2568   int r = ioctx->exec(oid, "rbd", "group_snap_list", inbl, outbl);
2569   if (r < 0) {
2570     return r;
2571   }
2572   auto iter = outbl.cbegin();
2573   try {
2574     decode(*snapshots, iter);
2575   } catch (const buffer::error &err) {
2576     return -EBADMSG;
2577   }
2578
2579   return 0;
2580 }
2581
2582 // rbd_trash functions
2583 void trash_add(librados::ObjectWriteOperation *op,
2584                const std::string &id,
2585                const cls::rbd::TrashImageSpec &trash_spec)
2586 {
2587   bufferlist bl;
2588   encode(id, bl);
2589   encode(trash_spec, bl);
2590   op->exec("rbd", "trash_add", bl);
2591 }
2592
2593 int trash_add(librados::IoCtx *ioctx, const std::string &id,
2594               const cls::rbd::TrashImageSpec &trash_spec)
2595 {
2596   librados::ObjectWriteOperation op;
2597   trash_add(&op, id, trash_spec);
2598
2599   return ioctx->operate(RBD_TRASH, &op);
2600 }
2601
2602 void trash_remove(librados::ObjectWriteOperation *op,
2603                   const std::string &id)
2604 {
2605   bufferlist bl;
2606   encode(id, bl);
2607   op->exec("rbd", "trash_remove", bl);
2608 }
2609
2610 int trash_remove(librados::IoCtx *ioctx, const std::string &id)
2611 {
2612   librados::ObjectWriteOperation op;
2613   trash_remove(&op, id);
2614
2615   return ioctx->operate(RBD_TRASH, &op);
2616 }
2617
2618 void trash_list_start(librados::ObjectReadOperation *op,
2619                       const std::string &start, uint64_t max_return)
2620 {
2621   bufferlist bl;
2622   encode(start, bl);
2623   encode(max_return, bl);
2624   op->exec("rbd", "trash_list", bl);
2625 }
2626
2627 int trash_list_finish(bufferlist::const_iterator *it,
2628                       map<string, cls::rbd::TrashImageSpec> *entries)
2629 {
2630   ceph_assert(entries);
2631
2632   try {
2633     decode(*entries, *it);
2634   } catch (const buffer::error &err) {
2635     return -EBADMSG;
2636   }
2637
2638   return 0;
2639 }
2640
2641 int trash_list(librados::IoCtx *ioctx,
2642                const std::string &start, uint64_t max_return,
2643                map<string, cls::rbd::TrashImageSpec> *entries)
2644 {
2645   librados::ObjectReadOperation op;
2646   trash_list_start(&op, start, max_return);
2647
2648   bufferlist out_bl;
2649   int r = ioctx->operate(RBD_TRASH, &op, &out_bl);
2650   if (r < 0) {
2651     return r;
2652   }
2653
2654   auto iter = out_bl.cbegin();
2655   return trash_list_finish(&iter, entries);
2656 }
2657
2658 void trash_get_start(librados::ObjectReadOperation *op,
2659                      const std::string &id)
2660 {
2661   bufferlist bl;
2662   encode(id, bl);
2663   op->exec("rbd", "trash_get", bl);
2664 }
2665
2666 int trash_get_finish(bufferlist::const_iterator *it,
2667                      cls::rbd::TrashImageSpec *trash_spec) {
2668   ceph_assert(trash_spec);
2669   try {
2670     decode(*trash_spec, *it);
2671   } catch (const buffer::error &err) {
2672     return -EBADMSG;
2673   }
2674
2675   return 0;
2676 }
2677
2678 int trash_get(librados::IoCtx *ioctx, const std::string &id,
2679               cls::rbd::TrashImageSpec *trash_spec)
2680 {
2681   librados::ObjectReadOperation op;
2682   trash_get_start(&op, id);
2683
2684   bufferlist out_bl;
2685   int r = ioctx->operate(RBD_TRASH, &op, &out_bl);
2686   if (r < 0) {
2687     return r;
2688   }
2689
2690   auto it = out_bl.cbegin();
2691   return trash_get_finish(&it, trash_spec);
2692 }
2693
2694 void trash_state_set(librados::ObjectWriteOperation *op,
2695                      const std::string &id,
2696                      const cls::rbd::TrashImageState &trash_state,
2697                      const cls::rbd::TrashImageState &expect_state)
2698 {
2699   bufferlist bl;
2700   encode(id, bl);
2701   encode(trash_state, bl);
2702   encode(expect_state, bl);
2703   op->exec("rbd", "trash_state_set", bl);
2704 }
2705
2706 int trash_state_set(librados::IoCtx *ioctx, const std::string &id,
2707                     const cls::rbd::TrashImageState &trash_state,
2708                     const cls::rbd::TrashImageState &expect_state)
2709 {
2710   librados::ObjectWriteOperation op;
2711   trash_state_set(&op, id, trash_state, expect_state);
2712
2713   return ioctx->operate(RBD_TRASH, &op);
2714 }
2715
2716 void namespace_add(librados::ObjectWriteOperation *op,
2717                    const std::string &name)
2718 {
2719   bufferlist bl;
2720   encode(name, bl);
2721   op->exec("rbd", "namespace_add", bl);
2722 }
2723
2724 int namespace_add(librados::IoCtx *ioctx, const std::string &name)
2725 {
2726   librados::ObjectWriteOperation op;
2727   namespace_add(&op, name);
2728
2729   return ioctx->operate(RBD_NAMESPACE, &op);
2730 }
2731
2732 void namespace_remove(librados::ObjectWriteOperation *op,
2733                       const std::string &name)
2734 {
2735   bufferlist bl;
2736   encode(name, bl);
2737   op->exec("rbd", "namespace_remove", bl);
2738 }
2739
2740 int namespace_remove(librados::IoCtx *ioctx, const std::string &name)
2741 {
2742   librados::ObjectWriteOperation op;
2743   namespace_remove(&op, name);
2744
2745   return ioctx->operate(RBD_NAMESPACE, &op);
2746 }
2747
2748 void namespace_list_start(librados::ObjectReadOperation *op,
2749                           const std::string &start, uint64_t max_return)
2750 {
2751   bufferlist bl;
2752   encode(start, bl);
2753   encode(max_return, bl);
2754   op->exec("rbd", "namespace_list", bl);
2755 }
2756
2757 int namespace_list_finish(bufferlist::const_iterator *it,
2758                           std::list<std::string> *entries)
2759 {
2760   ceph_assert(entries);
2761
2762   try {
2763     decode(*entries, *it);
2764   } catch (const buffer::error &err) {
2765     return -EBADMSG;
2766   }
2767
2768   return 0;
2769 }
2770
2771 int namespace_list(librados::IoCtx *ioctx,
2772                    const std::string &start, uint64_t max_return,
2773                    std::list<std::string> *entries)
2774 {
2775   librados::ObjectReadOperation op;
2776   namespace_list_start(&op, start, max_return);
2777
2778   bufferlist out_bl;
2779   int r = ioctx->operate(RBD_NAMESPACE, &op, &out_bl);
2780   if (r < 0) {
2781     return r;
2782   }
2783
2784   auto iter = out_bl.cbegin();
2785   return namespace_list_finish(&iter, entries);
2786 }
2787
2788 void sparsify(librados::ObjectWriteOperation *op, size_t sparse_size,
2789               bool remove_empty)
2790 {
2791   bufferlist bl;
2792   encode(sparse_size, bl);
2793   encode(remove_empty, bl);
2794   op->exec("rbd", "sparsify", bl);
2795 }
2796
2797 int sparsify(librados::IoCtx *ioctx, const std::string &oid, size_t sparse_size,
2798              bool remove_empty)
2799 {
2800   librados::ObjectWriteOperation op;
2801   sparsify(&op, sparse_size, remove_empty);
2802
2803   return ioctx->operate(oid, &op);
2804 }
2805
2806 } // namespace cls_client
2807 } // namespace librbd