ceph/src/librbd/api/Group.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 "common/Cond.h"
   5 #include "common/errno.h"
   6
   7 #include "librbd/ExclusiveLock.h"
   8 #include "librbd/api/Group.h"
   9 #include "librbd/ImageCtx.h"
  10 #include "librbd/ImageState.h"
  11 #include "librbd/ImageWatcher.h"
  12 #include "librbd/Operations.h"
  13 #include "librbd/Utils.h"
  14 #include "librbd/internal.h"
  15 #include "librbd/io/AioCompletion.h"
  16
  17 #define dout_subsys ceph_subsys_rbd
  18 #undef dout_prefix
  19 #define dout_prefix *_dout << "librbd::api::Group: " << __func__ << ": "
  20
  21 using std::map;
  22 using std::pair;
  23 using std::set;
  24 using std::string;
  25 using std::vector;
  26 // list binds to list() here, so std::list is explicitly used below
  27
  28 using ceph::bufferlist;
  29 using librados::snap_t;
  30 using librados::IoCtx;
  31 using librados::Rados;
  32
  33
  34 namespace librbd {
  35 namespace api {
  36
  37 namespace {
  38
  39 template <typename I>
  40 snap_t get_group_snap_id(I* ictx,
  41                          const cls::rbd::SnapshotNamespace& in_snap_namespace) {
  42   ceph_assert(ceph_mutex_is_locked(ictx->image_lock));
  43   auto it = ictx->snap_ids.lower_bound({cls::rbd::GroupSnapshotNamespace{},
  44                                         ""});
  45   for (; it != ictx->snap_ids.end(); ++it) {
  46     if (it->first.first == in_snap_namespace) {
  47       return it->second;
  48     } else if (!std::holds_alternative<cls::rbd::GroupSnapshotNamespace>(
  49                  it->first.first)) {
  50       break;
  51     }
  52   }
  53   return CEPH_NOSNAP;
  54 }
  55
  56 string generate_uuid(librados::IoCtx& io_ctx)
  57 {
  58   Rados rados(io_ctx);
  59   uint64_t bid = rados.get_instance_id();
  60
  61   uint32_t extra = rand() % 0xFFFFFFFF;
  62   std::ostringstream bid_ss;
  63   bid_ss << std::hex << bid << std::hex << extra;
  64   return bid_ss.str();
  65 }
  66
  67 int group_snap_list(librados::IoCtx& group_ioctx, const char *group_name,
  68                     std::vector<cls::rbd::GroupSnapshot> *cls_snaps)
  69 {
  70   CephContext *cct = (CephContext *)group_ioctx.cct();
  71
  72   string group_id;
  73   vector<string> ind_snap_names;
  74
  75   int r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY,
  76                                  group_name, &group_id);
  77   if (r < 0) {
  78     lderr(cct) << "error reading group id object: "
  79                << cpp_strerror(r)
  80                << dendl;
  81     return r;
  82   }
  83   string group_header_oid = util::group_header_name(group_id);
  84
  85   const int max_read = 1024;
  86   cls::rbd::GroupSnapshot snap_last;
  87
  88   for (;;) {
  89     vector<cls::rbd::GroupSnapshot> snaps_page;
  90
  91     r = cls_client::group_snap_list(&group_ioctx, group_header_oid,
  92                                     snap_last, max_read, &snaps_page);
  93
  94     if (r < 0) {
  95       lderr(cct) << "error reading snap list from group: "
  96         << cpp_strerror(-r) << dendl;
  97       return r;
  98     }
  99     cls_snaps->insert(cls_snaps->end(), snaps_page.begin(), snaps_page.end());
 100     if (snaps_page.size() < max_read) {
 101       break;
 102     }
 103     snap_last = *snaps_page.rbegin();
 104   }
 105
 106   return 0;
 107 }
 108
 109 std::string calc_ind_image_snap_name(uint64_t pool_id,
 110                                      const std::string &group_id,
 111                                      const std::string &snap_id)
 112 {
 113   std::stringstream ind_snap_name_stream;
 114   ind_snap_name_stream << ".group." << std::hex << pool_id << "_"
 115                        << group_id << "_" << snap_id;
 116   return ind_snap_name_stream.str();
 117 }
 118
 119 int group_image_list(librados::IoCtx& group_ioctx, const char *group_name,
 120                      std::vector<cls::rbd::GroupImageStatus> *image_ids)
 121 {
 122   CephContext *cct = (CephContext *)group_ioctx.cct();
 123
 124   string group_id;
 125
 126   int r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY,
 127                                  group_name, &group_id);
 128   if (r < 0) {
 129     lderr(cct) << "error reading group id object: "
 130                << cpp_strerror(r)
 131                << dendl;
 132     return r;
 133   }
 134   string group_header_oid = util::group_header_name(group_id);
 135
 136   ldout(cct, 20) << "listing images in group name "
 137                  << group_name << " group id " << group_header_oid << dendl;
 138   image_ids->clear();
 139
 140   const int max_read = 1024;
 141   cls::rbd::GroupImageSpec start_last;
 142   do {
 143     std::vector<cls::rbd::GroupImageStatus> image_ids_page;
 144
 145     r = cls_client::group_image_list(&group_ioctx, group_header_oid,
 146                                      start_last, max_read, &image_ids_page);
 147
 148     if (r < 0) {
 149       lderr(cct) << "error reading image list from group: "
 150         << cpp_strerror(-r) << dendl;
 151       return r;
 152     }
 153     image_ids->insert(image_ids->end(),
 154                      image_ids_page.begin(), image_ids_page.end());
 155
 156     if (image_ids_page.size() > 0)
 157       start_last = image_ids_page.rbegin()->spec;
 158
 159     r = image_ids_page.size();
 160   } while (r == max_read);
 161
 162   return 0;
 163 }
 164
 165 int group_image_remove(librados::IoCtx& group_ioctx, string group_id,
 166                        librados::IoCtx& image_ioctx, string image_id)
 167 {
 168   CephContext *cct = (CephContext *)group_ioctx.cct();
 169
 170   string group_header_oid = util::group_header_name(group_id);
 171
 172   string image_header_oid = util::header_name(image_id);
 173
 174   ldout(cct, 20) << "removing image " << image_id
 175                  << " image id " << image_header_oid << dendl;
 176
 177   cls::rbd::GroupSpec group_spec(group_id, group_ioctx.get_id());
 178
 179   cls::rbd::GroupImageStatus incomplete_st(image_id, image_ioctx.get_id(),
 180                                 cls::rbd::GROUP_IMAGE_LINK_STATE_INCOMPLETE);
 181
 182   cls::rbd::GroupImageSpec spec(image_id, image_ioctx.get_id());
 183
 184   int r = cls_client::group_image_set(&group_ioctx, group_header_oid,
 185                                       incomplete_st);
 186
 187   if (r < 0) {
 188     lderr(cct) << "couldn't put image into removing state: "
 189                << cpp_strerror(-r) << dendl;
 190     return r;
 191   }
 192
 193   r = cls_client::image_group_remove(&image_ioctx, image_header_oid,
 194                                      group_spec);
 195   if ((r < 0) && (r != -ENOENT)) {
 196     lderr(cct) << "couldn't remove group reference from image"
 197                << cpp_strerror(-r) << dendl;
 198     return r;
 199   } else if (r >= 0) {
 200     ImageWatcher<>::notify_header_update(image_ioctx, image_header_oid);
 201   }
 202
 203   r = cls_client::group_image_remove(&group_ioctx, group_header_oid, spec);
 204   if (r < 0) {
 205     lderr(cct) << "couldn't remove image from group"
 206                << cpp_strerror(-r) << dendl;
 207     return r;
 208   }
 209
 210   return 0;
 211 }
 212
 213 int group_snap_remove_by_record(librados::IoCtx& group_ioctx,
 214                                 const cls::rbd::GroupSnapshot& group_snap,
 215                                 const std::string& group_id,
 216                                 const std::string& group_header_oid) {
 217
 218   CephContext *cct = (CephContext *)group_ioctx.cct();
 219   std::vector<C_SaferCond*> on_finishes;
 220   int r, ret_code;
 221
 222   std::vector<librbd::ImageCtx*> ictxs;
 223
 224   cls::rbd::GroupSnapshotNamespace ne{group_ioctx.get_id(), group_id,
 225                                       group_snap.id};
 226
 227   ldout(cct, 20) << "Removing snapshots" << dendl;
 228   int snap_count = group_snap.snaps.size();
 229
 230   for (int i = 0; i < snap_count; ++i) {
 231     librbd::IoCtx image_io_ctx;
 232     r = util::create_ioctx(group_ioctx, "image", group_snap.snaps[i].pool, {},
 233                            &image_io_ctx);
 234     if (r < 0) {
 235       return r;
 236     }
 237
 238     librbd::ImageCtx* image_ctx = new ImageCtx("", group_snap.snaps[i].image_id,
 239                                                nullptr, image_io_ctx, false);
 240
 241     C_SaferCond* on_finish = new C_SaferCond;
 242
 243     image_ctx->state->open(0, on_finish);
 244
 245     ictxs.push_back(image_ctx);
 246     on_finishes.push_back(on_finish);
 247   }
 248
 249   ret_code = 0;
 250   for (int i = 0; i < snap_count; ++i) {
 251     r = on_finishes[i]->wait();
 252     delete on_finishes[i];
 253     if (r < 0) {
 254       ictxs[i] = nullptr;
 255       ret_code = r;
 256     }
 257   }
 258   if (ret_code != 0) {
 259     goto finish;
 260   }
 261
 262   ldout(cct, 20) << "Opened participating images. " <<
 263                     "Deleting snapshots themselves." << dendl;
 264
 265   for (int i = 0; i < snap_count; ++i) {
 266     ImageCtx *ictx = ictxs[i];
 267     on_finishes[i] = new C_SaferCond;
 268
 269     std::string snap_name;
 270     ictx->image_lock.lock_shared();
 271     snap_t snap_id = get_group_snap_id(ictx, ne);
 272     r = ictx->get_snap_name(snap_id, &snap_name);
 273     ictx->image_lock.unlock_shared();
 274
 275     if (r >= 0) {
 276       ldout(cct, 20) << "removing individual snapshot from image " << ictx->name
 277                      << dendl;
 278       ictx->operations->snap_remove(ne, snap_name, on_finishes[i]);
 279     } else {
 280       // We are ok to ignore missing image snapshots. The snapshot could have
 281       // been inconsistent in the first place.
 282       on_finishes[i]->complete(0);
 283     }
 284   }
 285
 286   for (int i = 0; i < snap_count; ++i) {
 287     r = on_finishes[i]->wait();
 288     delete on_finishes[i];
 289     if (r < 0 && r != -ENOENT) {
 290       // if previous attempts to remove this snapshot failed then the image's
 291       // snapshot may not exist
 292       lderr(cct) << "Failed deleting image snapshot. Ret code: " << r << dendl;
 293       ret_code = r;
 294     }
 295   }
 296
 297   if (ret_code != 0) {
 298     goto finish;
 299   }
 300
 301   ldout(cct, 20) << "Removed images snapshots removing snapshot record."
 302                  << dendl;
 303
 304   r = cls_client::group_snap_remove(&group_ioctx, group_header_oid,
 305       group_snap.id);
 306   if (r < 0) {
 307     ret_code = r;
 308     goto finish;
 309   }
 310
 311 finish:
 312   for (int i = 0; i < snap_count; ++i) {
 313     if (ictxs[i] != nullptr) {
 314       ictxs[i]->state->close();
 315     }
 316   }
 317   return ret_code;
 318 }
 319
 320 int group_snap_rollback_by_record(librados::IoCtx& group_ioctx,
 321                                   const cls::rbd::GroupSnapshot& group_snap,
 322                                   const std::string& group_id,
 323                                   const std::string& group_header_oid,
 324                                   ProgressContext& pctx) {
 325   CephContext *cct = (CephContext *)group_ioctx.cct();
 326   std::vector<C_SaferCond*> on_finishes;
 327   int r, ret_code;
 328
 329   std::vector<librbd::ImageCtx*> ictxs;
 330
 331   cls::rbd::GroupSnapshotNamespace ne{group_ioctx.get_id(), group_id,
 332                                       group_snap.id};
 333
 334   ldout(cct, 20) << "Rolling back snapshots" << dendl;
 335   int snap_count = group_snap.snaps.size();
 336
 337   for (int i = 0; i < snap_count; ++i) {
 338     librados::IoCtx image_io_ctx;
 339     r = util::create_ioctx(group_ioctx, "image", group_snap.snaps[i].pool, {},
 340                            &image_io_ctx);
 341     if (r < 0) {
 342       return r;
 343     }
 344
 345     librbd::ImageCtx* image_ctx = new ImageCtx("", group_snap.snaps[i].image_id,
 346                                                nullptr, image_io_ctx, false);
 347
 348     C_SaferCond* on_finish = new C_SaferCond;
 349
 350     image_ctx->state->open(0, on_finish);
 351
 352     ictxs.push_back(image_ctx);
 353     on_finishes.push_back(on_finish);
 354   }
 355
 356   ret_code = 0;
 357   for (int i = 0; i < snap_count; ++i) {
 358     r = on_finishes[i]->wait();
 359     delete on_finishes[i];
 360     if (r < 0) {
 361       ictxs[i] = nullptr;
 362       ret_code = r;
 363     }
 364   }
 365   if (ret_code != 0) {
 366     goto finish;
 367   }
 368
 369   ldout(cct, 20) << "Requesting exclusive locks for images" << dendl;
 370   for (auto ictx: ictxs) {
 371     std::shared_lock owner_lock{ictx->owner_lock};
 372     if (ictx->exclusive_lock != nullptr) {
 373       ictx->exclusive_lock->block_requests(-EBUSY);
 374     }
 375   }
 376   for (int i = 0; i < snap_count; ++i) {
 377     ImageCtx *ictx = ictxs[i];
 378     std::shared_lock owner_lock{ictx->owner_lock};
 379
 380     on_finishes[i] = new C_SaferCond;
 381     if (ictx->exclusive_lock != nullptr) {
 382       ictx->exclusive_lock->acquire_lock(on_finishes[i]);
 383     }
 384   }
 385
 386   ret_code = 0;
 387   for (int i = 0; i < snap_count; ++i) {
 388     r = 0;
 389     ImageCtx *ictx = ictxs[i];
 390     if (ictx->exclusive_lock != nullptr) {
 391       r = on_finishes[i]->wait();
 392     }
 393     delete on_finishes[i];
 394     if (r < 0) {
 395       ret_code = r;
 396     }
 397   }
 398   if (ret_code != 0) {
 399     goto finish;
 400   }
 401
 402   for (int i = 0; i < snap_count; ++i) {
 403     ImageCtx *ictx = ictxs[i];
 404     on_finishes[i] = new C_SaferCond;
 405
 406     std::shared_lock owner_locker{ictx->owner_lock};
 407     std::string snap_name;
 408     ictx->image_lock.lock_shared();
 409     snap_t snap_id = get_group_snap_id(ictx, ne);
 410     r = ictx->get_snap_name(snap_id, &snap_name);
 411     ictx->image_lock.unlock_shared();
 412
 413     if (r >= 0) {
 414       ldout(cct, 20) << "rolling back to individual snapshot for image " << ictx->name
 415                      << dendl;
 416       ictx->operations->execute_snap_rollback(ne, snap_name, pctx, on_finishes[i]);
 417     } else {
 418       on_finishes[i]->complete(r);
 419     }
 420   }
 421
 422   for (int i = 0; i < snap_count; ++i) {
 423     r = on_finishes[i]->wait();
 424     delete on_finishes[i];
 425     if (r < 0 && r != -ENOENT) {
 426       lderr(cct) << "Failed rolling back group to snapshot. Ret code: " << r << dendl;
 427       ret_code = r;
 428     }
 429   }
 430
 431 finish:
 432   for (int i = 0; i < snap_count; ++i) {
 433     if (ictxs[i] != nullptr) {
 434       ictxs[i]->state->close();
 435     }
 436   }
 437   return ret_code;
 438 }
 439
 440 template <typename I>
 441 void notify_unquiesce(std::vector<I*> &ictxs,
 442                       const std::vector<uint64_t> &requests) {
 443   if (requests.empty()) {
 444     return;
 445   }
 446
 447   ceph_assert(requests.size() == ictxs.size());
 448   int image_count = ictxs.size();
 449   std::vector<C_SaferCond> on_finishes(image_count);
 450
 451   for (int i = 0; i < image_count; ++i) {
 452     ImageCtx *ictx = ictxs[i];
 453
 454     ictx->image_watcher->notify_unquiesce(requests[i], &on_finishes[i]);
 455   }
 456
 457   for (int i = 0; i < image_count; ++i) {
 458     on_finishes[i].wait();
 459   }
 460 }
 461
 462 template <typename I>
 463 int notify_quiesce(std::vector<I*> &ictxs, ProgressContext &prog_ctx,
 464                    std::vector<uint64_t> *requests) {
 465   int image_count = ictxs.size();
 466   std::vector<C_SaferCond> on_finishes(image_count);
 467
 468   requests->resize(image_count);
 469   for (int i = 0; i < image_count; ++i) {
 470     auto ictx = ictxs[i];
 471
 472     ictx->image_watcher->notify_quiesce(&(*requests)[i], prog_ctx,
 473                                         &on_finishes[i]);
 474   }
 475
 476   int ret_code = 0;
 477   for (int i = 0; i < image_count; ++i) {
 478     int r = on_finishes[i].wait();
 479     if (r < 0) {
 480       ret_code = r;
 481     }
 482   }
 483
 484   if (ret_code != 0) {
 485     notify_unquiesce(ictxs, *requests);
 486   }
 487
 488   return ret_code;
 489 }
 490
 491 } // anonymous namespace
 492
 493 template <typename I>
 494 int Group<I>::image_remove_by_id(librados::IoCtx& group_ioctx,
 495                                  const char *group_name,
 496                                  librados::IoCtx& image_ioctx,
 497                                  const char *image_id)
 498 {
 499   CephContext *cct = (CephContext *)group_ioctx.cct();
 500   ldout(cct, 20) << "io_ctx=" << &group_ioctx
 501     << " group name " << group_name << " image "
 502     << &image_ioctx << " id " << image_id << dendl;
 503
 504   string group_id;
 505
 506   int r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY, group_name,
 507       &group_id);
 508   if (r < 0) {
 509     lderr(cct) << "error reading group id object: "
 510       << cpp_strerror(r)
 511       << dendl;
 512     return r;
 513   }
 514
 515   ldout(cct, 20) << "removing image from group name " << group_name
 516                   << " group id " << group_id << dendl;
 517
 518   return group_image_remove(group_ioctx, group_id, image_ioctx, string(image_id));
 519 }
 520
 521 template <typename I>
 522 int Group<I>::create(librados::IoCtx& io_ctx, const char *group_name)
 523 {
 524   CephContext *cct = (CephContext *)io_ctx.cct();
 525
 526   string id = generate_uuid(io_ctx);
 527
 528   ldout(cct, 2) << "adding group to directory..." << dendl;
 529
 530   int r = cls_client::group_dir_add(&io_ctx, RBD_GROUP_DIRECTORY, group_name,
 531                                     id);
 532   if (r < 0) {
 533     lderr(cct) << "error adding group to directory: "
 534                << cpp_strerror(r)
 535                << dendl;
 536     return r;
 537   }
 538   string header_oid = util::group_header_name(id);
 539
 540   r = io_ctx.create(header_oid, true);
 541   if (r < 0) {
 542     lderr(cct) << "error creating group header: " << cpp_strerror(r) << dendl;
 543     goto err_remove_from_dir;
 544   }
 545
 546   return 0;
 547
 548 err_remove_from_dir:
 549   int remove_r = cls_client::group_dir_remove(&io_ctx, RBD_GROUP_DIRECTORY,
 550                                               group_name, id);
 551   if (remove_r < 0) {
 552     lderr(cct) << "error cleaning up group from rbd_directory "
 553                << "object after creation failed: " << cpp_strerror(remove_r)
 554                << dendl;
 555   }
 556
 557   return r;
 558 }
 559
 560 template <typename I>
 561 int Group<I>::remove(librados::IoCtx& io_ctx, const char *group_name)
 562 {
 563   CephContext *cct((CephContext *)io_ctx.cct());
 564   ldout(cct, 20) << "group_remove " << &io_ctx << " " << group_name << dendl;
 565
 566   std::string group_id;
 567   int r = cls_client::dir_get_id(&io_ctx, RBD_GROUP_DIRECTORY,
 568                                  std::string(group_name), &group_id);
 569   if (r < 0 && r != -ENOENT) {
 570     lderr(cct) << "error getting id of group" << dendl;
 571     return r;
 572   }
 573   string group_header_oid = util::group_header_name(group_id);
 574
 575   std::vector<cls::rbd::GroupSnapshot> snaps;
 576   r = group_snap_list(io_ctx, group_name, &snaps);
 577   if (r < 0 && r != -ENOENT) {
 578     lderr(cct) << "error listing group snapshots" << dendl;
 579     return r;
 580   }
 581
 582   for (auto &snap : snaps) {
 583     r = group_snap_remove_by_record(io_ctx, snap, group_id, group_header_oid);
 584     if (r < 0) {
 585       return r;
 586     }
 587   }
 588
 589   std::vector<cls::rbd::GroupImageStatus> images;
 590   r = group_image_list(io_ctx, group_name, &images);
 591   if (r < 0 && r != -ENOENT) {
 592     lderr(cct) << "error listing group images" << dendl;
 593     return r;
 594   }
 595
 596   for (auto image : images) {
 597     IoCtx image_ioctx;
 598     r = util::create_ioctx(io_ctx, "image", image.spec.pool_id, {},
 599                            &image_ioctx);
 600     if (r < 0) {
 601       return r;
 602     }
 603
 604     r = group_image_remove(io_ctx, group_id, image_ioctx, image.spec.image_id);
 605     if (r < 0 && r != -ENOENT) {
 606       lderr(cct) << "error removing image from a group" << dendl;
 607       return r;
 608     }
 609   }
 610
 611   string header_oid = util::group_header_name(group_id);
 612
 613   r = io_ctx.remove(header_oid);
 614   if (r < 0 && r != -ENOENT) {
 615     lderr(cct) << "error removing header: " << cpp_strerror(-r) << dendl;
 616     return r;
 617   }
 618
 619   r = cls_client::group_dir_remove(&io_ctx, RBD_GROUP_DIRECTORY,
 620                                        group_name, group_id);
 621   if (r < 0 && r != -ENOENT) {
 622     lderr(cct) << "error removing group from directory" << dendl;
 623     return r;
 624   }
 625
 626   return 0;
 627 }
 628
 629 template <typename I>
 630 int Group<I>::list(IoCtx& io_ctx, vector<string> *names)
 631 {
 632   CephContext *cct = (CephContext *)io_ctx.cct();
 633   ldout(cct, 20) << "io_ctx=" << &io_ctx << dendl;
 634
 635   int max_read = 1024;
 636   string last_read = "";
 637   int r;
 638   do {
 639     map<string, string> groups;
 640     r = cls_client::group_dir_list(&io_ctx, RBD_GROUP_DIRECTORY, last_read,
 641                                    max_read, &groups);
 642     if (r < 0) {
 643       if (r != -ENOENT) {
 644         lderr(cct) << "error listing group in directory: "
 645                    << cpp_strerror(r) << dendl;
 646       } else {
 647         r = 0;
 648       }
 649       return r;
 650     }
 651     for (pair<string, string> group : groups) {
 652       names->push_back(group.first);
 653     }
 654     if (!groups.empty()) {
 655       last_read = groups.rbegin()->first;
 656     }
 657     r = groups.size();
 658   } while (r == max_read);
 659
 660   return 0;
 661 }
 662
 663 template <typename I>
 664 int Group<I>::image_add(librados::IoCtx& group_ioctx, const char *group_name,
 665                         librados::IoCtx& image_ioctx, const char *image_name)
 666 {
 667   CephContext *cct = (CephContext *)group_ioctx.cct();
 668   ldout(cct, 20) << "io_ctx=" << &group_ioctx
 669                  << " group name " << group_name << " image "
 670                  << &image_ioctx << " name " << image_name << dendl;
 671
 672   if (group_ioctx.get_namespace() != image_ioctx.get_namespace()) {
 673     lderr(cct) << "group and image cannot be in different namespaces" << dendl;
 674     return -EINVAL;
 675   }
 676
 677   string group_id;
 678
 679   int r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY, group_name,
 680                                  &group_id);
 681   if (r < 0) {
 682     lderr(cct) << "error reading group id object: "
 683                << cpp_strerror(r)
 684                << dendl;
 685     return r;
 686   }
 687   string group_header_oid = util::group_header_name(group_id);
 688
 689
 690   ldout(cct, 20) << "adding image to group name " << group_name
 691                  << " group id " << group_header_oid << dendl;
 692
 693   string image_id;
 694
 695   r = cls_client::dir_get_id(&image_ioctx, RBD_DIRECTORY, image_name,
 696                              &image_id);
 697   if (r < 0) {
 698     lderr(cct) << "error reading image id object: "
 699                << cpp_strerror(-r) << dendl;
 700     return r;
 701   }
 702
 703   string image_header_oid = util::header_name(image_id);
 704
 705   ldout(cct, 20) << "adding image " << image_name
 706                  << " image id " << image_header_oid << dendl;
 707
 708   cls::rbd::GroupImageStatus incomplete_st(
 709     image_id, image_ioctx.get_id(),
 710     cls::rbd::GROUP_IMAGE_LINK_STATE_INCOMPLETE);
 711   cls::rbd::GroupImageStatus attached_st(
 712     image_id, image_ioctx.get_id(), cls::rbd::GROUP_IMAGE_LINK_STATE_ATTACHED);
 713
 714   r = cls_client::group_image_set(&group_ioctx, group_header_oid,
 715                                   incomplete_st);
 716
 717   cls::rbd::GroupSpec group_spec(group_id, group_ioctx.get_id());
 718
 719   if (r < 0) {
 720     lderr(cct) << "error adding image reference to group: "
 721                << cpp_strerror(-r) << dendl;
 722     return r;
 723   }
 724
 725   r = cls_client::image_group_add(&image_ioctx, image_header_oid, group_spec);
 726   if (r < 0) {
 727     lderr(cct) << "error adding group reference to image: "
 728                << cpp_strerror(-r) << dendl;
 729     cls::rbd::GroupImageSpec spec(image_id, image_ioctx.get_id());
 730     cls_client::group_image_remove(&group_ioctx, group_header_oid, spec);
 731     // Ignore errors in the clean up procedure.
 732     return r;
 733   }
 734   ImageWatcher<>::notify_header_update(image_ioctx, image_header_oid);
 735
 736   r = cls_client::group_image_set(&group_ioctx, group_header_oid,
 737                                   attached_st);
 738
 739   return r;
 740 }
 741
 742 template <typename I>
 743 int Group<I>::image_remove(librados::IoCtx& group_ioctx, const char *group_name,
 744                            librados::IoCtx& image_ioctx, const char *image_name)
 745 {
 746   CephContext *cct = (CephContext *)group_ioctx.cct();
 747   ldout(cct, 20) << "io_ctx=" << &group_ioctx
 748                 << " group name " << group_name << " image "
 749                 << &image_ioctx << " name " << image_name << dendl;
 750
 751   if (group_ioctx.get_namespace() != image_ioctx.get_namespace()) {
 752     lderr(cct) << "group and image cannot be in different namespaces" << dendl;
 753     return -EINVAL;
 754   }
 755
 756   string group_id;
 757
 758   int r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY, group_name,
 759       &group_id);
 760   if (r < 0) {
 761     lderr(cct) << "error reading group id object: "
 762       << cpp_strerror(r)
 763       << dendl;
 764     return r;
 765   }
 766
 767   ldout(cct, 20) << "removing image from group name " << group_name
 768     << " group id " << group_id << dendl;
 769
 770   string image_id;
 771   r = cls_client::dir_get_id(&image_ioctx, RBD_DIRECTORY, image_name,
 772       &image_id);
 773   if (r < 0) {
 774     lderr(cct) << "error reading image id object: "
 775       << cpp_strerror(-r) << dendl;
 776     return r;
 777   }
 778
 779   r = group_image_remove(group_ioctx, group_id, image_ioctx, image_id);
 780
 781   return r;
 782 }
 783
 784 template <typename I>
 785 int Group<I>::image_list(librados::IoCtx& group_ioctx,
 786                          const char *group_name,
 787                          std::vector<group_image_info_t>* images)
 788 {
 789   CephContext *cct = (CephContext *)group_ioctx.cct();
 790   ldout(cct, 20) << "io_ctx=" << &group_ioctx
 791                  << " group name " << group_name << dendl;
 792
 793   std::vector<cls::rbd::GroupImageStatus> image_ids;
 794
 795   group_image_list(group_ioctx, group_name, &image_ids);
 796
 797   for (auto image_id : image_ids) {
 798     IoCtx ioctx;
 799     int r = util::create_ioctx(group_ioctx, "image", image_id.spec.pool_id, {},
 800                                &ioctx);
 801     if (r < 0) {
 802       return r;
 803     }
 804
 805     std::string image_name;
 806     r = cls_client::dir_get_name(&ioctx, RBD_DIRECTORY,
 807                                  image_id.spec.image_id, &image_name);
 808     if (r < 0) {
 809       return r;
 810     }
 811
 812     images->push_back(
 813         group_image_info_t {
 814            image_name,
 815            ioctx.get_id(),
 816            static_cast<group_image_state_t>(image_id.state)});
 817   }
 818
 819   return 0;
 820 }
 821
 822 template <typename I>
 823 int Group<I>::rename(librados::IoCtx& io_ctx, const char *src_name,
 824                      const char *dest_name)
 825 {
 826   CephContext *cct((CephContext *)io_ctx.cct());
 827   ldout(cct, 20) << "group_rename " << &io_ctx << " " << src_name
 828                  << " -> " << dest_name << dendl;
 829
 830   std::string group_id;
 831   int r = cls_client::dir_get_id(&io_ctx, RBD_GROUP_DIRECTORY,
 832                                  std::string(src_name), &group_id);
 833   if (r < 0) {
 834     if (r != -ENOENT)
 835       lderr(cct) << "error getting id of group" << dendl;
 836     return r;
 837   }
 838
 839   r = cls_client::group_dir_rename(&io_ctx, RBD_GROUP_DIRECTORY,
 840                                    src_name, dest_name, group_id);
 841   if (r < 0 && r != -ENOENT) {
 842     lderr(cct) << "error renaming group from directory" << dendl;
 843     return r;
 844   }
 845
 846   return 0;
 847 }
 848
 849
 850 template <typename I>
 851 int Group<I>::image_get_group(I *ictx, group_info_t *group_info)
 852 {
 853   int r = ictx->state->refresh_if_required();
 854   if (r < 0)
 855     return r;
 856
 857   if (RBD_GROUP_INVALID_POOL != ictx->group_spec.pool_id) {
 858     IoCtx ioctx;
 859     r = util::create_ioctx(ictx->md_ctx, "group", ictx->group_spec.pool_id, {},
 860                            &ioctx);
 861     if (r < 0) {
 862       return r;
 863     }
 864
 865     std::string group_name;
 866     r = cls_client::dir_get_name(&ioctx, RBD_GROUP_DIRECTORY,
 867                                  ictx->group_spec.group_id, &group_name);
 868     if (r < 0)
 869       return r;
 870     group_info->pool = ioctx.get_id();
 871     group_info->name = group_name;
 872   } else {
 873     group_info->pool = RBD_GROUP_INVALID_POOL;
 874     group_info->name = "";
 875   }
 876
 877   return 0;
 878 }
 879
 880 template <typename I>
 881 int Group<I>::snap_create(librados::IoCtx& group_ioctx,
 882                           const char *group_name, const char *snap_name,
 883                           uint32_t flags) {
 884   CephContext *cct = (CephContext *)group_ioctx.cct();
 885
 886   string group_id;
 887   cls::rbd::GroupSnapshot group_snap;
 888   vector<cls::rbd::ImageSnapshotSpec> image_snaps;
 889   std::string ind_snap_name;
 890
 891   std::vector<librbd::ImageCtx*> ictxs;
 892   std::vector<C_SaferCond*> on_finishes;
 893   std::vector<uint64_t> quiesce_requests;
 894   NoOpProgressContext prog_ctx;
 895   uint64_t internal_flags = 0;
 896
 897   int r = util::snap_create_flags_api_to_internal(cct, flags, &internal_flags);
 898   if (r < 0) {
 899     return r;
 900   }
 901   internal_flags &= ~(SNAP_CREATE_FLAG_SKIP_NOTIFY_QUIESCE |
 902                       SNAP_CREATE_FLAG_IGNORE_NOTIFY_QUIESCE_ERROR);
 903
 904   r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY, group_name,
 905                              &group_id);
 906   if (r < 0) {
 907     lderr(cct) << "error reading group id object: "
 908                << cpp_strerror(r)
 909                << dendl;
 910     return r;
 911   }
 912
 913   std::vector<cls::rbd::GroupImageStatus> images;
 914   r = group_image_list(group_ioctx, group_name, &images);
 915   if (r < 0) {
 916     return r;
 917   }
 918   int image_count = images.size();
 919
 920   ldout(cct, 20) << "Found " << image_count << " images in group" << dendl;
 921
 922   image_snaps = vector<cls::rbd::ImageSnapshotSpec>(image_count,
 923       cls::rbd::ImageSnapshotSpec());
 924
 925   for (int i = 0; i < image_count; ++i) {
 926     image_snaps[i].pool = images[i].spec.pool_id;
 927     image_snaps[i].image_id = images[i].spec.image_id;
 928   }
 929
 930   string group_header_oid = util::group_header_name(group_id);
 931
 932   group_snap.id = generate_uuid(group_ioctx);
 933   group_snap.name = string(snap_name);
 934   group_snap.state = cls::rbd::GROUP_SNAPSHOT_STATE_INCOMPLETE;
 935   group_snap.snaps = image_snaps;
 936
 937   cls::rbd::GroupSnapshotNamespace ne{group_ioctx.get_id(), group_id,
 938                                       group_snap.id};
 939
 940   r = cls_client::group_snap_set(&group_ioctx, group_header_oid, group_snap);
 941   if (r == -EEXIST) {
 942     lderr(cct) << "snapshot with this name already exists: "
 943                << cpp_strerror(r)
 944                << dendl;
 945   }
 946   int ret_code = 0;
 947   if (r < 0) {
 948     ret_code = r;
 949     goto finish;
 950   }
 951
 952   for (auto image: images) {
 953     librbd::IoCtx image_io_ctx;
 954     r = util::create_ioctx(group_ioctx, "image", image.spec.pool_id, {},
 955                            &image_io_ctx);
 956     if (r < 0) {
 957       ret_code = r;
 958       goto finish;
 959     }
 960
 961     ldout(cct, 20) << "Opening image with id " << image.spec.image_id << dendl;
 962
 963     librbd::ImageCtx* image_ctx = new ImageCtx("", image.spec.image_id.c_str(),
 964                                                nullptr, image_io_ctx, false);
 965
 966     C_SaferCond* on_finish = new C_SaferCond;
 967
 968     image_ctx->state->open(0, on_finish);
 969
 970     ictxs.push_back(image_ctx);
 971     on_finishes.push_back(on_finish);
 972   }
 973   ldout(cct, 20) << "Issued open request waiting for the completion" << dendl;
 974   ret_code = 0;
 975   for (int i = 0; i < image_count; ++i) {
 976
 977     ldout(cct, 20) << "Waiting for completion on on_finish: " <<
 978       on_finishes[i] << dendl;
 979
 980     r = on_finishes[i]->wait();
 981     delete on_finishes[i];
 982     if (r < 0) {
 983       ictxs[i] = nullptr;
 984       ret_code = r;
 985     }
 986   }
 987   if (ret_code != 0) {
 988     goto remove_record;
 989   }
 990
 991   if ((flags & RBD_SNAP_CREATE_SKIP_QUIESCE) == 0) {
 992     ldout(cct, 20) << "Sending quiesce notification" << dendl;
 993     ret_code = notify_quiesce(ictxs, prog_ctx, &quiesce_requests);
 994     if (ret_code != 0 && (flags & RBD_SNAP_CREATE_IGNORE_QUIESCE_ERROR) == 0) {
 995       goto remove_record;
 996     }
 997   }
 998
 999   ldout(cct, 20) << "Requesting exclusive locks for images" << dendl;
1000
1001   for (auto ictx: ictxs) {
1002     std::shared_lock owner_lock{ictx->owner_lock};
1003     if (ictx->exclusive_lock != nullptr) {
1004       ictx->exclusive_lock->block_requests(-EBUSY);
1005     }
1006   }
1007   for (int i = 0; i < image_count; ++i) {
1008     ImageCtx *ictx = ictxs[i];
1009     std::shared_lock owner_lock{ictx->owner_lock};
1010
1011     on_finishes[i] = new C_SaferCond;
1012     if (ictx->exclusive_lock != nullptr) {
1013       ictx->exclusive_lock->acquire_lock(on_finishes[i]);
1014     }
1015   }
1016
1017   ret_code = 0;
1018   for (int i = 0; i < image_count; ++i) {
1019     r = 0;
1020     ImageCtx *ictx = ictxs[i];
1021     if (ictx->exclusive_lock != nullptr) {
1022       r = on_finishes[i]->wait();
1023     }
1024     delete on_finishes[i];
1025     if (r < 0) {
1026       ret_code = r;
1027     }
1028   }
1029   if (ret_code != 0) {
1030     notify_unquiesce(ictxs, quiesce_requests);
1031     goto remove_record;
1032   }
1033
1034   ind_snap_name = calc_ind_image_snap_name(group_ioctx.get_id(), group_id,
1035                                             group_snap.id);
1036
1037   for (int i = 0; i < image_count; ++i) {
1038     ImageCtx *ictx = ictxs[i];
1039
1040     C_SaferCond* on_finish = new C_SaferCond;
1041
1042     std::shared_lock owner_locker{ictx->owner_lock};
1043     ictx->operations->execute_snap_create(
1044         ne, ind_snap_name.c_str(), on_finish, 0,
1045         SNAP_CREATE_FLAG_SKIP_NOTIFY_QUIESCE, prog_ctx);
1046
1047     on_finishes[i] = on_finish;
1048   }
1049
1050   ret_code = 0;
1051   for (int i = 0; i < image_count; ++i) {
1052     r = on_finishes[i]->wait();
1053     delete on_finishes[i];
1054     if (r < 0) {
1055       ret_code = r;
1056     } else {
1057       ImageCtx *ictx = ictxs[i];
1058       ictx->image_lock.lock_shared();
1059       snap_t snap_id = get_group_snap_id(ictx, ne);
1060       ictx->image_lock.unlock_shared();
1061       if (snap_id == CEPH_NOSNAP) {
1062         ldout(cct, 20) << "Couldn't find created snapshot with namespace: "
1063                        << ne << dendl;
1064         ret_code = -ENOENT;
1065       } else {
1066         image_snaps[i].snap_id = snapid_t(snap_id);
1067         image_snaps[i].pool = ictx->md_ctx.get_id();
1068         image_snaps[i].image_id = ictx->id;
1069       }
1070     }
1071   }
1072   if (ret_code != 0) {
1073     goto remove_image_snaps;
1074   }
1075
1076   group_snap.snaps = image_snaps;
1077   group_snap.state = cls::rbd::GROUP_SNAPSHOT_STATE_COMPLETE;
1078
1079   r = cls_client::group_snap_set(&group_ioctx, group_header_oid, group_snap);
1080   if (r < 0) {
1081     ret_code = r;
1082     goto remove_image_snaps;
1083   }
1084
1085   ldout(cct, 20) << "Sending unquiesce notification" << dendl;
1086   notify_unquiesce(ictxs, quiesce_requests);
1087
1088   goto finish;
1089
1090 remove_image_snaps:
1091   notify_unquiesce(ictxs, quiesce_requests);
1092
1093   for (int i = 0; i < image_count; ++i) {
1094     ImageCtx *ictx = ictxs[i];
1095     ldout(cct, 20) << "Removing individual snapshot with name: " <<
1096       ind_snap_name << dendl;
1097
1098     on_finishes[i] = new C_SaferCond;
1099     std::string snap_name;
1100     ictx->image_lock.lock_shared();
1101     snap_t snap_id = get_group_snap_id(ictx, ne);
1102     r = ictx->get_snap_name(snap_id, &snap_name);
1103     ictx->image_lock.unlock_shared();
1104     if (r >= 0) {
1105       ictx->operations->snap_remove(ne, snap_name.c_str(), on_finishes[i]);
1106     } else {
1107       // Ignore missing image snapshots. The whole snapshot could have been
1108       // inconsistent.
1109       on_finishes[i]->complete(0);
1110     }
1111   }
1112
1113   for (int i = 0, n = on_finishes.size(); i < n; ++i) {
1114     r = on_finishes[i]->wait();
1115     delete on_finishes[i];
1116     if (r < 0 && r != -ENOENT) { // if previous attempts to remove this snapshot failed then the image's snapshot may not exist
1117       lderr(cct) << "Failed cleaning up image snapshot. Ret code: " << r << dendl;
1118       // just report error, but don't abort the process
1119     }
1120   }
1121
1122 remove_record:
1123   r = cls_client::group_snap_remove(&group_ioctx, group_header_oid,
1124       group_snap.id);
1125   if (r < 0) {
1126     lderr(cct) << "error while cleaning up group snapshot" << dendl;
1127     // we ignore return value in clean up
1128   }
1129
1130 finish:
1131   for (int i = 0, n = ictxs.size(); i < n; ++i) {
1132     if (ictxs[i] != nullptr) {
1133       ictxs[i]->state->close();
1134     }
1135   }
1136   return ret_code;
1137 }
1138
1139 template <typename I>
1140 int Group<I>::snap_remove(librados::IoCtx& group_ioctx, const char *group_name,
1141                           const char *snap_name)
1142 {
1143   CephContext *cct = (CephContext *)group_ioctx.cct();
1144
1145   string group_id;
1146   int r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY,
1147                                  group_name, &group_id);
1148   if (r < 0) {
1149     lderr(cct) << "error reading group id object: "
1150                << cpp_strerror(r)
1151                << dendl;
1152     return r;
1153   }
1154
1155   std::vector<cls::rbd::GroupSnapshot> snaps;
1156   r = group_snap_list(group_ioctx, group_name, &snaps);
1157   if (r < 0) {
1158     return r;
1159   }
1160
1161   cls::rbd::GroupSnapshot *group_snap = nullptr;
1162   for (auto &snap : snaps) {
1163     if (snap.name == string(snap_name)) {
1164       group_snap = &snap;
1165       break;
1166     }
1167   }
1168   if (group_snap == nullptr) {
1169     return -ENOENT;
1170   }
1171
1172   string group_header_oid = util::group_header_name(group_id);
1173   r = group_snap_remove_by_record(group_ioctx, *group_snap, group_id,
1174                                   group_header_oid);
1175   return r;
1176 }
1177
1178 template <typename I>
1179 int Group<I>::snap_rename(librados::IoCtx& group_ioctx, const char *group_name,
1180                           const char *old_snap_name,
1181                           const char *new_snap_name) {
1182   CephContext *cct = (CephContext *)group_ioctx.cct();
1183   if (0 == strcmp(old_snap_name, new_snap_name))
1184     return -EEXIST;
1185
1186   std::string group_id;
1187   int r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY,
1188                                  group_name, &group_id);
1189   if (r == -ENOENT) {
1190     return r;
1191   } else if (r < 0) {
1192     lderr(cct) << "error reading group id object: " << cpp_strerror(r) << dendl;
1193     return r;
1194   }
1195
1196   std::vector<cls::rbd::GroupSnapshot> group_snaps;
1197   r = group_snap_list(group_ioctx, group_name, &group_snaps);
1198   if (r < 0) {
1199     return r;
1200   }
1201
1202   cls::rbd::GroupSnapshot group_snap;
1203   for (auto &snap : group_snaps) {
1204     if (snap.name == old_snap_name) {
1205       group_snap = snap;
1206       break;
1207     }
1208   }
1209
1210   if (group_snap.id.empty()) {
1211     return -ENOENT;
1212   }
1213
1214   std::string group_header_oid = util::group_header_name(group_id);
1215   group_snap.name = new_snap_name;
1216   r = cls_client::group_snap_set(&group_ioctx, group_header_oid, group_snap);
1217   if (r < 0) {
1218     return r;
1219   }
1220
1221   return 0;
1222 }
1223
1224 template <typename I>
1225 int Group<I>::snap_list(librados::IoCtx& group_ioctx, const char *group_name,
1226                         std::vector<group_snap_info_t> *snaps)
1227 {
1228   std::vector<cls::rbd::GroupSnapshot> cls_snaps;
1229
1230   int r = group_snap_list(group_ioctx, group_name, &cls_snaps);
1231   if (r < 0) {
1232     return r;
1233   }
1234
1235   for (auto snap : cls_snaps) {
1236     snaps->push_back(
1237         group_snap_info_t {
1238            snap.name,
1239            static_cast<group_snap_state_t>(snap.state)});
1240
1241   }
1242   return 0;
1243 }
1244
1245 template <typename I>
1246 int Group<I>::snap_rollback(librados::IoCtx& group_ioctx,
1247                             const char *group_name, const char *snap_name,
1248                             ProgressContext& pctx)
1249 {
1250   CephContext *cct = (CephContext *)group_ioctx.cct();
1251
1252   string group_id;
1253   int r = cls_client::dir_get_id(&group_ioctx, RBD_GROUP_DIRECTORY,
1254                                  group_name, &group_id);
1255   if (r < 0) {
1256     lderr(cct) << "error reading group id object: "
1257                << cpp_strerror(r) << dendl;
1258     return r;
1259   }
1260
1261   std::vector<cls::rbd::GroupSnapshot> snaps;
1262   r = group_snap_list(group_ioctx, group_name, &snaps);
1263   if (r < 0) {
1264     return r;
1265   }
1266
1267   cls::rbd::GroupSnapshot *group_snap = nullptr;
1268   for (auto &snap : snaps) {
1269     if (snap.name == string(snap_name)) {
1270       group_snap = &snap;
1271       break;
1272     }
1273   }
1274   if (group_snap == nullptr) {
1275     return -ENOENT;
1276   }
1277
1278   string group_header_oid = util::group_header_name(group_id);
1279   r = group_snap_rollback_by_record(group_ioctx, *group_snap, group_id,
1280                                     group_header_oid, pctx);
1281   return r;
1282 }
1283
1284 } // namespace api
1285 } // namespace librbd
1286
1287 template class librbd::api::Group<librbd::ImageCtx>;