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