ceph/src/librbd/api/Image.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 "librbd/api/Image.h"
   5 #include "include/rados/librados.hpp"
   6 #include "common/dout.h"
   7 #include "common/errno.h"
   8 #include "common/Cond.h"
   9 #include "cls/rbd/cls_rbd_client.h"
  10 #include "librbd/DeepCopyRequest.h"
  11 #include "librbd/ExclusiveLock.h"
  12 #include "librbd/ImageCtx.h"
  13 #include "librbd/ImageState.h"
  14 #include "librbd/internal.h"
  15 #include "librbd/Utils.h"
  16 #include "librbd/api/Config.h"
  17 #include "librbd/api/Trash.h"
  18 #include "librbd/image/CloneRequest.h"
  19 #include "librbd/image/RemoveRequest.h"
  20 #include "librbd/image/PreRemoveRequest.h"
  21 #include <boost/scope_exit.hpp>
  22
  23 #define dout_subsys ceph_subsys_rbd
  24 #undef dout_prefix
  25 #define dout_prefix *_dout << "librbd::api::Image: " << __func__ << ": "
  26
  27 namespace librbd {
  28 namespace api {
  29
  30 namespace {
  31
  32 bool compare_by_pool(const librbd::linked_image_spec_t& lhs,
  33                      const librbd::linked_image_spec_t& rhs)
  34 {
  35   if (lhs.pool_id != rhs.pool_id) {
  36     return lhs.pool_id < rhs.pool_id;
  37   } else if (lhs.pool_namespace != rhs.pool_namespace) {
  38     return lhs.pool_namespace < rhs.pool_namespace;
  39   }
  40   return false;
  41 }
  42
  43 bool compare(const librbd::linked_image_spec_t& lhs,
  44              const librbd::linked_image_spec_t& rhs)
  45 {
  46   if (lhs.pool_name != rhs.pool_name) {
  47     return lhs.pool_name < rhs.pool_name;
  48   } else if (lhs.pool_id != rhs.pool_id) {
  49     return lhs.pool_id < rhs.pool_id;
  50   } else if (lhs.pool_namespace != rhs.pool_namespace) {
  51     return lhs.pool_namespace < rhs.pool_namespace;
  52   } else if (lhs.image_name != rhs.image_name) {
  53     return lhs.image_name < rhs.image_name;
  54   } else if (lhs.image_id != rhs.image_id) {
  55     return lhs.image_id < rhs.image_id;
  56   }
  57   return false;
  58 }
  59
  60 template <typename I>
  61 int pre_remove_image(librados::IoCtx& io_ctx, const std::string& image_id) {
  62   I *image_ctx = I::create("", image_id, nullptr, io_ctx, false);
  63   int r = image_ctx->state->open(OPEN_FLAG_SKIP_OPEN_PARENT);
  64   if (r < 0) {
  65     return r;
  66   }
  67
  68   C_SaferCond ctx;
  69   auto req = image::PreRemoveRequest<I>::create(image_ctx, false, &ctx);
  70   req->send();
  71
  72   r = ctx.wait();
  73   image_ctx->state->close();
  74   return r;
  75 }
  76
  77 } // anonymous namespace
  78
  79 template <typename I>
  80 int Image<I>::get_op_features(I *ictx, uint64_t *op_features) {
  81   CephContext *cct = ictx->cct;
  82   ldout(cct, 20) << "image_ctx=" << ictx << dendl;
  83
  84   int r = ictx->state->refresh_if_required();
  85   if (r < 0) {
  86     return r;
  87   }
  88
  89   RWLock::RLocker snap_locker(ictx->snap_lock);
  90   *op_features = ictx->op_features;
  91   return 0;
  92 }
  93
  94 template <typename I>
  95 int Image<I>::list_images(librados::IoCtx& io_ctx,
  96                           std::vector<image_spec_t> *images) {
  97   CephContext *cct = (CephContext *)io_ctx.cct();
  98   ldout(cct, 20) << "list " << &io_ctx << dendl;
  99
 100   int r;
 101   images->clear();
 102
 103   if (io_ctx.get_namespace().empty()) {
 104     bufferlist bl;
 105     r = io_ctx.read(RBD_DIRECTORY, bl, 0, 0);
 106     if (r == -ENOENT) {
 107       return 0;
 108     } else if (r < 0) {
 109       lderr(cct) << "error listing v1 images: " << cpp_strerror(r) << dendl;
 110       return r;
 111     }
 112
 113     // V1 format images are in a tmap
 114     if (bl.length()) {
 115       auto p = bl.cbegin();
 116       bufferlist header;
 117       std::map<std::string, bufferlist> m;
 118       decode(header, p);
 119       decode(m, p);
 120       for (auto& it : m) {
 121         images->push_back({.id ="", .name = it.first});
 122       }
 123     }
 124   }
 125
 126   // V2 format images
 127   std::map<std::string, std::string> image_names_to_ids;
 128   r = list_images_v2(io_ctx, &image_names_to_ids);
 129   if (r < 0) {
 130     lderr(cct) << "error listing v2 images: " << cpp_strerror(r) << dendl;
 131     return r;
 132   }
 133
 134   for (const auto& img_pair : image_names_to_ids) {
 135     images->push_back({.id = img_pair.second,
 136                        .name = img_pair.first});
 137   }
 138
 139   // include V2 images in a partially removed state
 140   std::vector<librbd::trash_image_info_t> trash_images;
 141   r = Trash<I>::list(io_ctx, trash_images, false);
 142   if (r < 0 && r != -EOPNOTSUPP) {
 143     lderr(cct) << "error listing trash images: " << cpp_strerror(r) << dendl;
 144     return r;
 145   }
 146
 147   for (const auto& trash_image : trash_images) {
 148     if (trash_image.source == RBD_TRASH_IMAGE_SOURCE_REMOVING) {
 149       images->push_back({.id = trash_image.id,
 150                          .name = trash_image.name});
 151
 152     }
 153   }
 154
 155   return 0;
 156 }
 157
 158 template <typename I>
 159 int Image<I>::list_images_v2(librados::IoCtx& io_ctx, ImageNameToIds *images) {
 160   CephContext *cct = (CephContext *)io_ctx.cct();
 161   ldout(cct, 20) << "io_ctx=" << &io_ctx << dendl;
 162
 163   // new format images are accessed by class methods
 164   int r;
 165   int max_read = 1024;
 166   string last_read = "";
 167   do {
 168     map<string, string> images_page;
 169     r = cls_client::dir_list(&io_ctx, RBD_DIRECTORY, last_read, max_read,
 170                              &images_page);
 171     if (r < 0 && r != -ENOENT) {
 172       lderr(cct) << "error listing image in directory: "
 173                  << cpp_strerror(r) << dendl;
 174       return r;
 175     } else if (r == -ENOENT) {
 176       break;
 177     }
 178     for (map<string, string>::const_iterator it = images_page.begin();
 179          it != images_page.end(); ++it) {
 180       images->insert(*it);
 181     }
 182     if (!images_page.empty()) {
 183       last_read = images_page.rbegin()->first;
 184     }
 185     r = images_page.size();
 186   } while (r == max_read);
 187
 188   return 0;
 189 }
 190
 191 template <typename I>
 192 int Image<I>::get_parent(I *ictx,
 193                          librbd::linked_image_spec_t *parent_image,
 194                          librbd::snap_spec_t *parent_snap) {
 195   auto cct = ictx->cct;
 196   ldout(cct, 20) << "image_ctx=" << ictx << dendl;
 197
 198   int r = ictx->state->refresh_if_required();
 199   if (r < 0) {
 200     return r;
 201   }
 202
 203   RWLock::RLocker snap_locker(ictx->snap_lock);
 204   RWLock::RLocker parent_locker(ictx->parent_lock);
 205
 206   bool release_parent_locks = false;
 207   BOOST_SCOPE_EXIT_ALL(ictx, &release_parent_locks) {
 208     if (release_parent_locks) {
 209       ictx->parent->parent_lock.put_read();
 210       ictx->parent->snap_lock.put_read();
 211     }
 212   };
 213
 214   // if a migration is in-progress, the true parent is the parent
 215   // of the migration source image
 216   auto parent = ictx->parent;
 217   if (!ictx->migration_info.empty() && ictx->parent != nullptr) {
 218     release_parent_locks = true;
 219     ictx->parent->snap_lock.get_read();
 220     ictx->parent->parent_lock.get_read();
 221
 222     parent = ictx->parent->parent;
 223   }
 224
 225   if (parent == nullptr) {
 226     return -ENOENT;
 227   }
 228
 229   parent_image->pool_id = parent->md_ctx.get_id();
 230   parent_image->pool_name = parent->md_ctx.get_pool_name();
 231   parent_image->pool_namespace = parent->md_ctx.get_namespace();
 232
 233   RWLock::RLocker parent_snap_locker(parent->snap_lock);
 234   parent_snap->id = parent->snap_id;
 235   parent_snap->namespace_type = RBD_SNAP_NAMESPACE_TYPE_USER;
 236   if (parent->snap_id != CEPH_NOSNAP) {
 237     auto snap_info = parent->get_snap_info(parent->snap_id);
 238     if (snap_info == nullptr) {
 239       lderr(cct) << "error finding parent snap name: " << cpp_strerror(r)
 240                  << dendl;
 241       return -ENOENT;
 242     }
 243
 244     parent_snap->namespace_type = static_cast<snap_namespace_type_t>(
 245       cls::rbd::get_snap_namespace_type(snap_info->snap_namespace));
 246     parent_snap->name = snap_info->name;
 247   }
 248
 249   parent_image->image_id = parent->id;
 250   parent_image->image_name = parent->name;
 251   parent_image->trash = true;
 252
 253   librbd::trash_image_info_t trash_info;
 254   r = Trash<I>::get(parent->md_ctx, parent->id, &trash_info);
 255   if (r == -ENOENT || r == -EOPNOTSUPP) {
 256     parent_image->trash = false;
 257   } else if (r < 0) {
 258     lderr(cct) << "error looking up trash status: " << cpp_strerror(r)
 259                << dendl;
 260     return r;
 261   }
 262
 263   return 0;
 264 }
 265
 266 template <typename I>
 267 int Image<I>::list_children(I *ictx,
 268                             std::vector<librbd::linked_image_spec_t> *images) {
 269   images->clear();
 270   return list_descendants(ictx, 1, images);
 271 }
 272
 273 template <typename I>
 274 int Image<I>::list_children(I *ictx,
 275                             const cls::rbd::ParentImageSpec &parent_spec,
 276                             std::vector<librbd::linked_image_spec_t> *images) {
 277   images->clear();
 278   return list_descendants(ictx, parent_spec, 1, images);
 279 }
 280
 281 template <typename I>
 282 int Image<I>::list_descendants(
 283     librados::IoCtx& io_ctx, const std::string &image_id,
 284     const std::optional<size_t> &max_level,
 285     std::vector<librbd::linked_image_spec_t> *images) {
 286   ImageCtx *ictx = new librbd::ImageCtx("", image_id, nullptr,
 287                                         io_ctx, true);
 288   int r = ictx->state->open(OPEN_FLAG_SKIP_OPEN_PARENT);
 289   if (r < 0) {
 290     if (r == -ENOENT) {
 291       return 0;
 292     }
 293     lderr(ictx->cct) << "failed to open descendant " << image_id
 294                      << " from pool " << io_ctx.get_pool_name() << ":"
 295                      << cpp_strerror(r) << dendl;
 296     return r;
 297   }
 298
 299   r = list_descendants(ictx, max_level, images);
 300
 301   int r1 = ictx->state->close();
 302   if (r1 < 0) {
 303     lderr(ictx->cct) << "error when closing descendant " << image_id
 304                      << " from pool " << io_ctx.get_pool_name() << ":"
 305                      << cpp_strerror(r) << dendl;
 306   }
 307
 308   return r;
 309 }
 310
 311 template <typename I>
 312 int Image<I>::list_descendants(
 313     I *ictx, const std::optional<size_t> &max_level,
 314     std::vector<librbd::linked_image_spec_t> *images) {
 315   RWLock::RLocker l(ictx->snap_lock);
 316   std::vector<librados::snap_t> snap_ids;
 317   if (ictx->snap_id != CEPH_NOSNAP) {
 318     snap_ids.push_back(ictx->snap_id);
 319   } else {
 320     snap_ids = ictx->snaps;
 321   }
 322   for (auto snap_id : snap_ids) {
 323     cls::rbd::ParentImageSpec parent_spec{ictx->md_ctx.get_id(),
 324                                           ictx->md_ctx.get_namespace(),
 325                                           ictx->id, snap_id};
 326     int r = list_descendants(ictx, parent_spec, max_level, images);
 327     if (r < 0) {
 328       return r;
 329     }
 330   }
 331   return 0;
 332 }
 333
 334 template <typename I>
 335 int Image<I>::list_descendants(
 336     I *ictx, const cls::rbd::ParentImageSpec &parent_spec,
 337     const std::optional<size_t> &max_level,
 338     std::vector<librbd::linked_image_spec_t> *images) {
 339   auto child_max_level = max_level;
 340   if (child_max_level) {
 341     if (child_max_level == 0) {
 342       return 0;
 343     }
 344     (*child_max_level)--;
 345   }
 346   CephContext *cct = ictx->cct;
 347   ldout(cct, 20) << "ictx=" << ictx << dendl;
 348
 349   // no children for non-layered or old format image
 350   if (!ictx->test_features(RBD_FEATURE_LAYERING, ictx->snap_lock)) {
 351     return 0;
 352   }
 353
 354   librados::Rados rados(ictx->md_ctx);
 355
 356   // search all pools for clone v1 children dependent on this snapshot
 357   std::list<std::pair<int64_t, std::string> > pools;
 358   int r = rados.pool_list2(pools);
 359   if (r < 0) {
 360     lderr(cct) << "error listing pools: " << cpp_strerror(r) << dendl;
 361     return r;
 362   }
 363
 364   for (auto& it : pools) {
 365     int64_t base_tier;
 366     r = rados.pool_get_base_tier(it.first, &base_tier);
 367     if (r == -ENOENT) {
 368       ldout(cct, 1) << "pool " << it.second << " no longer exists" << dendl;
 369       continue;
 370     } else if (r < 0) {
 371       lderr(cct) << "error retrieving base tier for pool " << it.second
 372                  << dendl;
 373       return r;
 374     }
 375     if (it.first != base_tier) {
 376       // pool is a cache; skip it
 377       continue;
 378     }
 379
 380     IoCtx ioctx;
 381     r = util::create_ioctx(ictx->md_ctx, "child image", it.first, {}, &ioctx);
 382     if (r == -ENOENT) {
 383       continue;
 384     } else if (r < 0) {
 385       return r;
 386     }
 387
 388     std::set<std::string> image_ids;
 389     r = cls_client::get_children(&ioctx, RBD_CHILDREN, parent_spec,
 390                                  image_ids);
 391     if (r < 0 && r != -ENOENT) {
 392       lderr(cct) << "error reading list of children from pool " << it.second
 393                  << dendl;
 394       return r;
 395     }
 396
 397     for (auto& image_id : image_ids) {
 398       images->push_back({
 399         it.first, "", ictx->md_ctx.get_namespace(), image_id, "", false});
 400       r = list_descendants(ictx->md_ctx, image_id, child_max_level, images);
 401       if (r < 0) {
 402         return r;
 403       }
 404     }
 405   }
 406
 407   // retrieve clone v2 children attached to this snapshot
 408   IoCtx parent_io_ctx;
 409   r = util::create_ioctx(ictx->md_ctx, "parent image", parent_spec.pool_id,
 410                          parent_spec.pool_namespace, &parent_io_ctx);
 411   if (r < 0) {
 412     return r;
 413   }
 414
 415   cls::rbd::ChildImageSpecs child_images;
 416   r = cls_client::children_list(&parent_io_ctx,
 417                                 util::header_name(parent_spec.image_id),
 418                                 parent_spec.snap_id, &child_images);
 419   if (r < 0 && r != -ENOENT && r != -EOPNOTSUPP) {
 420     lderr(cct) << "error retrieving children: " << cpp_strerror(r) << dendl;
 421     return r;
 422   }
 423
 424   for (auto& child_image : child_images) {
 425     images->push_back({
 426       child_image.pool_id, "", child_image.pool_namespace,
 427       child_image.image_id, "", false});
 428     if (!child_max_level || *child_max_level > 0) {
 429       IoCtx ioctx;
 430       r = util::create_ioctx(ictx->md_ctx, "child image", child_image.pool_id,
 431                              child_image.pool_namespace, &ioctx);
 432       if (r == -ENOENT) {
 433         continue;
 434       } else if (r < 0) {
 435         return r;
 436       }
 437       r = list_descendants(ioctx, child_image.image_id, child_max_level,
 438                            images);
 439       if (r < 0) {
 440         return r;
 441       }
 442     }
 443   }
 444
 445   // batch lookups by pool + namespace
 446   std::sort(images->begin(), images->end(), compare_by_pool);
 447
 448   int64_t child_pool_id = -1;
 449   librados::IoCtx child_io_ctx;
 450   std::map<std::string, std::pair<std::string, bool>> child_image_id_to_info;
 451   for (auto& image : *images) {
 452     if (child_pool_id == -1 || child_pool_id != image.pool_id ||
 453         child_io_ctx.get_namespace() != image.pool_namespace) {
 454       r = util::create_ioctx(ictx->md_ctx, "child image", image.pool_id,
 455                              image.pool_namespace, &child_io_ctx);
 456       if (r < 0) {
 457         return r;
 458       }
 459       child_pool_id = image.pool_id;
 460
 461       child_image_id_to_info.clear();
 462
 463       std::map<std::string, std::string> image_names_to_ids;
 464       r = list_images_v2(child_io_ctx, &image_names_to_ids);
 465       if (r < 0) {
 466         lderr(cct) << "error listing v2 images: " << cpp_strerror(r) << dendl;
 467         return r;
 468       }
 469
 470       for (auto& [name, id] : image_names_to_ids) {
 471         child_image_id_to_info.insert({id, {name, false}});
 472       }
 473
 474       std::vector<librbd::trash_image_info_t> trash_images;
 475       r = Trash<I>::list(child_io_ctx, trash_images, false);
 476       if (r < 0 && r != -EOPNOTSUPP) {
 477         lderr(cct) << "error listing trash images: " << cpp_strerror(r)
 478                    << dendl;
 479         return r;
 480       }
 481
 482       for (auto& it : trash_images) {
 483         child_image_id_to_info.insert({
 484           it.id,
 485           {it.name,
 486            it.source == RBD_TRASH_IMAGE_SOURCE_REMOVING ? false : true}});
 487       }
 488     }
 489
 490     auto it = child_image_id_to_info.find(image.image_id);
 491     if (it == child_image_id_to_info.end()) {
 492           lderr(cct) << "error looking up name for image id "
 493                      << image.image_id << " in pool "
 494                      << child_io_ctx.get_pool_name()
 495                      << (image.pool_namespace.empty() ?
 496                           "" : "/" + image.pool_namespace) << dendl;
 497       return -ENOENT;
 498     }
 499
 500     image.pool_name = child_io_ctx.get_pool_name();
 501     image.image_name = it->second.first;
 502     image.trash = it->second.second;
 503   }
 504
 505   // final sort by pool + image names
 506   std::sort(images->begin(), images->end(), compare);
 507   return 0;
 508 }
 509
 510 template <typename I>
 511 int Image<I>::deep_copy(I *src, librados::IoCtx& dest_md_ctx,
 512                         const char *destname, ImageOptions& opts,
 513                         ProgressContext &prog_ctx) {
 514   CephContext *cct = (CephContext *)dest_md_ctx.cct();
 515   ldout(cct, 20) << src->name
 516                  << (src->snap_name.length() ? "@" + src->snap_name : "")
 517                  << " -> " << destname << " opts = " << opts << dendl;
 518
 519   uint64_t features;
 520   uint64_t src_size;
 521   {
 522     RWLock::RLocker snap_locker(src->snap_lock);
 523
 524     if (!src->migration_info.empty()) {
 525       lderr(cct) << "cannot deep copy migrating image" << dendl;
 526       return -EBUSY;
 527     }
 528
 529     features = (src->features & ~RBD_FEATURES_IMPLICIT_ENABLE);
 530     src_size = src->get_image_size(src->snap_id);
 531   }
 532   uint64_t format = 2;
 533   if (opts.get(RBD_IMAGE_OPTION_FORMAT, &format) != 0) {
 534     opts.set(RBD_IMAGE_OPTION_FORMAT, format);
 535   }
 536   if (format == 1) {
 537     lderr(cct) << "old format not supported for destination image" << dendl;
 538     return -EINVAL;
 539   }
 540   uint64_t stripe_unit = src->stripe_unit;
 541   if (opts.get(RBD_IMAGE_OPTION_STRIPE_UNIT, &stripe_unit) != 0) {
 542     opts.set(RBD_IMAGE_OPTION_STRIPE_UNIT, stripe_unit);
 543   }
 544   uint64_t stripe_count = src->stripe_count;
 545   if (opts.get(RBD_IMAGE_OPTION_STRIPE_COUNT, &stripe_count) != 0) {
 546     opts.set(RBD_IMAGE_OPTION_STRIPE_COUNT, stripe_count);
 547   }
 548   uint64_t order = src->order;
 549   if (opts.get(RBD_IMAGE_OPTION_ORDER, &order) != 0) {
 550     opts.set(RBD_IMAGE_OPTION_ORDER, order);
 551   }
 552   if (opts.get(RBD_IMAGE_OPTION_FEATURES, &features) != 0) {
 553     opts.set(RBD_IMAGE_OPTION_FEATURES, features);
 554   }
 555   if (features & ~RBD_FEATURES_ALL) {
 556     lderr(cct) << "librbd does not support requested features" << dendl;
 557     return -ENOSYS;
 558   }
 559
 560   uint64_t flatten = 0;
 561   if (opts.get(RBD_IMAGE_OPTION_FLATTEN, &flatten) == 0) {
 562     opts.unset(RBD_IMAGE_OPTION_FLATTEN);
 563   }
 564
 565   cls::rbd::ParentImageSpec parent_spec;
 566   if (flatten > 0) {
 567     parent_spec.pool_id = -1;
 568   } else {
 569     RWLock::RLocker snap_locker(src->snap_lock);
 570     RWLock::RLocker parent_locker(src->parent_lock);
 571
 572     // use oldest snapshot or HEAD for parent spec
 573     if (!src->snap_info.empty()) {
 574       parent_spec = src->snap_info.begin()->second.parent.spec;
 575     } else {
 576       parent_spec = src->parent_md.spec;
 577     }
 578   }
 579
 580   int r;
 581   if (parent_spec.pool_id == -1) {
 582     r = create(dest_md_ctx, destname, "", src_size, opts, "", "", false);
 583   } else {
 584     librados::IoCtx parent_io_ctx;
 585     r = util::create_ioctx(src->md_ctx, "parent image", parent_spec.pool_id,
 586                            parent_spec.pool_namespace, &parent_io_ctx);
 587     if (r < 0) {
 588       return r;
 589     }
 590
 591     ConfigProxy config{cct->_conf};
 592     api::Config<I>::apply_pool_overrides(dest_md_ctx, &config);
 593
 594     C_SaferCond ctx;
 595     std::string dest_id = util::generate_image_id(dest_md_ctx);
 596     auto *req = image::CloneRequest<I>::create(
 597       config, parent_io_ctx, parent_spec.image_id, "", parent_spec.snap_id,
 598       dest_md_ctx, destname, dest_id, opts, "", "", src->op_work_queue, &ctx);
 599     req->send();
 600     r = ctx.wait();
 601   }
 602   if (r < 0) {
 603     lderr(cct) << "header creation failed" << dendl;
 604     return r;
 605   }
 606   opts.set(RBD_IMAGE_OPTION_ORDER, static_cast<uint64_t>(order));
 607
 608   auto dest = new I(destname, "", nullptr, dest_md_ctx, false);
 609   r = dest->state->open(0);
 610   if (r < 0) {
 611     lderr(cct) << "failed to read newly created header" << dendl;
 612     return r;
 613   }
 614
 615   C_SaferCond lock_ctx;
 616   {
 617     RWLock::WLocker locker(dest->owner_lock);
 618
 619     if (dest->exclusive_lock == nullptr ||
 620         dest->exclusive_lock->is_lock_owner()) {
 621       lock_ctx.complete(0);
 622     } else {
 623       dest->exclusive_lock->acquire_lock(&lock_ctx);
 624     }
 625   }
 626
 627   r = lock_ctx.wait();
 628   if (r < 0) {
 629     lderr(cct) << "failed to request exclusive lock: " << cpp_strerror(r)
 630                << dendl;
 631     dest->state->close();
 632     return r;
 633   }
 634
 635   r = deep_copy(src, dest, flatten > 0, prog_ctx);
 636
 637   int close_r = dest->state->close();
 638   if (r == 0 && close_r < 0) {
 639     r = close_r;
 640   }
 641   return r;
 642 }
 643
 644 template <typename I>
 645 int Image<I>::deep_copy(I *src, I *dest, bool flatten,
 646                         ProgressContext &prog_ctx) {
 647   CephContext *cct = src->cct;
 648   librados::snap_t snap_id_start = 0;
 649   librados::snap_t snap_id_end;
 650   {
 651     RWLock::RLocker snap_locker(src->snap_lock);
 652     snap_id_end = src->snap_id;
 653   }
 654
 655   ThreadPool *thread_pool;
 656   ContextWQ *op_work_queue;
 657   ImageCtx::get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
 658
 659   C_SaferCond cond;
 660   SnapSeqs snap_seqs;
 661   auto req = DeepCopyRequest<I>::create(src, dest, snap_id_start, snap_id_end,
 662                                         flatten, boost::none, op_work_queue,
 663                                         &snap_seqs, &prog_ctx, &cond);
 664   req->send();
 665   int r = cond.wait();
 666   if (r < 0) {
 667     return r;
 668   }
 669
 670   return 0;
 671 }
 672
 673 template <typename I>
 674 int Image<I>::snap_set(I *ictx,
 675                        const cls::rbd::SnapshotNamespace &snap_namespace,
 676                        const char *snap_name) {
 677   ldout(ictx->cct, 20) << "snap_set " << ictx << " snap = "
 678                        << (snap_name ? snap_name : "NULL") << dendl;
 679
 680   // ignore return value, since we may be set to a non-existent
 681   // snapshot and the user is trying to fix that
 682   ictx->state->refresh_if_required();
 683
 684   uint64_t snap_id = CEPH_NOSNAP;
 685   std::string name(snap_name == nullptr ? "" : snap_name);
 686   if (!name.empty()) {
 687     RWLock::RLocker snap_locker(ictx->snap_lock);
 688     snap_id = ictx->get_snap_id(cls::rbd::UserSnapshotNamespace{},
 689                                 snap_name);
 690     if (snap_id == CEPH_NOSNAP) {
 691       return -ENOENT;
 692     }
 693   }
 694
 695   return snap_set(ictx, snap_id);
 696 }
 697
 698 template <typename I>
 699 int Image<I>::snap_set(I *ictx, uint64_t snap_id) {
 700   ldout(ictx->cct, 20) << "snap_set " << ictx << " "
 701                        << "snap_id=" << snap_id << dendl;
 702
 703   // ignore return value, since we may be set to a non-existent
 704   // snapshot and the user is trying to fix that
 705   ictx->state->refresh_if_required();
 706
 707   C_SaferCond ctx;
 708   ictx->state->snap_set(snap_id, &ctx);
 709   int r = ctx.wait();
 710   if (r < 0) {
 711     if (r != -ENOENT) {
 712       lderr(ictx->cct) << "failed to " << (snap_id == CEPH_NOSNAP ? "un" : "")
 713                        << "set snapshot: " << cpp_strerror(r) << dendl;
 714     }
 715     return r;
 716   }
 717
 718   return 0;
 719 }
 720
 721 template <typename I>
 722 int Image<I>::remove(IoCtx& io_ctx, const std::string &image_name,
 723                      ProgressContext& prog_ctx)
 724 {
 725   CephContext *cct((CephContext *)io_ctx.cct());
 726   ldout(cct, 20) << "name=" << image_name << dendl;
 727
 728   // look up the V2 image id based on the image name
 729   std::string image_id;
 730   int r = cls_client::dir_get_id(&io_ctx, RBD_DIRECTORY, image_name,
 731                                  &image_id);
 732   if (r == -ENOENT) {
 733     // check if it already exists in trash from an aborted trash remove attempt
 734     std::vector<trash_image_info_t> trash_entries;
 735     r = Trash<I>::list(io_ctx, trash_entries, false);
 736     if (r < 0) {
 737       return r;
 738     } else if (r >= 0) {
 739       for (auto& entry : trash_entries) {
 740         if (entry.name == image_name &&
 741             entry.source == RBD_TRASH_IMAGE_SOURCE_REMOVING) {
 742           return Trash<I>::remove(io_ctx, entry.id, true, prog_ctx);
 743         }
 744       }
 745     }
 746
 747     // fall-through if we failed to locate the image in the V2 directory and
 748     // trash
 749   } else if (r < 0) {
 750     lderr(cct) << "failed to retrieve image id: " << cpp_strerror(r) << dendl;
 751     return r;
 752   } else {
 753     // attempt to move the image to the trash (and optionally immediately
 754     // delete the image)
 755     ConfigProxy config(cct->_conf);
 756     Config<I>::apply_pool_overrides(io_ctx, &config);
 757
 758     rbd_trash_image_source_t trash_image_source =
 759       RBD_TRASH_IMAGE_SOURCE_REMOVING;
 760     uint64_t expire_seconds = 0;
 761     if (config.get_val<bool>("rbd_move_to_trash_on_remove")) {
 762       // keep the image in the trash upon remove requests
 763       trash_image_source = RBD_TRASH_IMAGE_SOURCE_USER;
 764       expire_seconds = config.get_val<uint64_t>(
 765         "rbd_move_to_trash_on_remove_expire_seconds");
 766     } else {
 767       // attempt to pre-validate the removal before moving to trash and
 768       // removing
 769       r = pre_remove_image<I>(io_ctx, image_id);
 770       if (r < 0 && r != -ENOENT) {
 771         return r;
 772       }
 773     }
 774
 775     r = Trash<I>::move(io_ctx, trash_image_source, image_name, image_id,
 776                        expire_seconds);
 777     if (r >= 0) {
 778       if (trash_image_source == RBD_TRASH_IMAGE_SOURCE_REMOVING) {
 779         // proceed with attempting to immediately remove the image
 780         r = Trash<I>::remove(io_ctx, image_id, true, prog_ctx);
 781
 782         if (r == -ENOTEMPTY || r == -EBUSY || r == -EMLINK) {
 783           // best-effort try to restore the image if the removal
 784           // failed for possible expected reasons
 785           Trash<I>::restore(io_ctx, trash_image_source, image_id, image_name);
 786         }
 787       }
 788       return r;
 789     } else if (r < 0 && r != -EOPNOTSUPP) {
 790       return r;
 791     }
 792
 793     // fall-through if trash isn't supported
 794   }
 795
 796   ThreadPool *thread_pool;
 797   ContextWQ *op_work_queue;
 798   ImageCtx::get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
 799
 800   // might be a V1 image format that cannot be moved to the trash
 801   // and would not have been listed in the V2 directory -- or the OSDs
 802   // are too old and don't support the trash feature
 803   C_SaferCond cond;
 804   auto req = librbd::image::RemoveRequest<I>::create(
 805     io_ctx, image_name, "", false, false, prog_ctx, op_work_queue, &cond);
 806   req->send();
 807
 808   return cond.wait();
 809 }
 810
 811 } // namespace api
 812 } // namespace librbd
 813
 814 template class librbd::api::Image<librbd::ImageCtx>;