ceph/src/librbd/api/Trash.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/Trash.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/ExclusiveLock.h"
  11 #include "librbd/ImageCtx.h"
  12 #include "librbd/ImageState.h"
  13 #include "librbd/internal.h"
  14 #include "librbd/Operations.h"
  15 #include "librbd/TrashWatcher.h"
  16 #include "librbd/Utils.h"
  17 #include "librbd/api/DiffIterate.h"
  18 #include "librbd/exclusive_lock/Policy.h"
  19 #include "librbd/image/RemoveRequest.h"
  20 #include "librbd/mirror/DisableRequest.h"
  21 #include "librbd/mirror/EnableRequest.h"
  22 #include "librbd/trash/MoveRequest.h"
  23 #include "librbd/trash/RemoveRequest.h"
  24 #include <json_spirit/json_spirit.h>
  25 #include "librbd/journal/DisabledPolicy.h"
  26 #include "librbd/image/ListWatchersRequest.h"
  27 #include <experimental/map>
  28
  29 #define dout_subsys ceph_subsys_rbd
  30 #undef dout_prefix
  31 #define dout_prefix *_dout << "librbd::api::Trash: " << __func__ << ": "
  32
  33 namespace librbd {
  34 namespace api {
  35
  36 template <typename I>
  37 const typename Trash<I>::TrashImageSources Trash<I>::RESTORE_SOURCE_WHITELIST {
  38     cls::rbd::TRASH_IMAGE_SOURCE_USER,
  39     cls::rbd::TRASH_IMAGE_SOURCE_MIRRORING,
  40     cls::rbd::TRASH_IMAGE_SOURCE_USER_PARENT
  41   };
  42
  43 namespace {
  44
  45 template <typename I>
  46 int disable_mirroring(I *ictx) {
  47   ldout(ictx->cct, 10) << dendl;
  48
  49   C_SaferCond ctx;
  50   auto req = mirror::DisableRequest<I>::create(ictx, false, true, &ctx);
  51   req->send();
  52   int r = ctx.wait();
  53   if (r < 0) {
  54     lderr(ictx->cct) << "failed to disable mirroring: " << cpp_strerror(r)
  55                      << dendl;
  56     return r;
  57   }
  58
  59   return 0;
  60 }
  61
  62 template <typename I>
  63 int enable_mirroring(IoCtx &io_ctx, const std::string &image_id) {
  64   auto cct = reinterpret_cast<CephContext*>(io_ctx.cct());
  65
  66   uint64_t features;
  67   uint64_t incompatible_features;
  68   int r = cls_client::get_features(&io_ctx, util::header_name(image_id), true,
  69                                    &features, &incompatible_features);
  70   if (r < 0) {
  71     lderr(cct) << "failed to retrieve features: " << cpp_strerror(r) << dendl;
  72     return r;
  73   }
  74
  75   if ((features & RBD_FEATURE_JOURNALING) == 0) {
  76     return 0;
  77   }
  78
  79   cls::rbd::MirrorMode mirror_mode;
  80   r = cls_client::mirror_mode_get(&io_ctx, &mirror_mode);
  81   if (r < 0 && r != -ENOENT) {
  82     lderr(cct) << "failed to retrieve mirror mode: " << cpp_strerror(r)
  83                << dendl;
  84     return r;
  85   }
  86
  87   if (mirror_mode != cls::rbd::MIRROR_MODE_POOL) {
  88     ldout(cct, 10) << "not pool mirroring mode" << dendl;
  89     return 0;
  90   }
  91
  92   ldout(cct, 10) << dendl;
  93
  94   ThreadPool *thread_pool;
  95   ContextWQ *op_work_queue;
  96   ImageCtx::get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
  97   C_SaferCond ctx;
  98   auto req = mirror::EnableRequest<I>::create(
  99     io_ctx, image_id, cls::rbd::MIRROR_IMAGE_MODE_JOURNAL, "", false,
 100     op_work_queue, &ctx);
 101   req->send();
 102   r = ctx.wait();
 103   if (r < 0) {
 104     lderr(cct) << "failed to enable mirroring: " << cpp_strerror(r)
 105                << dendl;
 106     return r;
 107   }
 108
 109   return 0;
 110 }
 111
 112 int list_trash_image_specs(
 113     librados::IoCtx &io_ctx,
 114     std::map<std::string, cls::rbd::TrashImageSpec>* trash_image_specs,
 115     bool exclude_user_remove_source) {
 116   CephContext *cct((CephContext *)io_ctx.cct());
 117   ldout(cct, 20) << "list_trash_image_specs " << &io_ctx << dendl;
 118
 119   bool more_entries;
 120   uint32_t max_read = 1024;
 121   std::string last_read;
 122   do {
 123     std::map<string, cls::rbd::TrashImageSpec> trash_entries;
 124     int r = cls_client::trash_list(&io_ctx, last_read, max_read,
 125                                    &trash_entries);
 126     if (r < 0 && r != -ENOENT) {
 127       lderr(cct) << "error listing rbd trash entries: " << cpp_strerror(r)
 128                  << dendl;
 129       return r;
 130     } else if (r == -ENOENT) {
 131       break;
 132     }
 133
 134     if (trash_entries.empty()) {
 135       break;
 136     }
 137
 138     for (const auto &entry : trash_entries) {
 139       if (exclude_user_remove_source &&
 140           entry.second.source == cls::rbd::TRASH_IMAGE_SOURCE_REMOVING) {
 141         continue;
 142       }
 143
 144       trash_image_specs->insert({entry.first, entry.second});
 145     }
 146
 147     last_read = trash_entries.rbegin()->first;
 148     more_entries = (trash_entries.size() >= max_read);
 149   } while (more_entries);
 150
 151   return 0;
 152 }
 153
 154 } // anonymous namespace
 155
 156 template <typename I>
 157 int Trash<I>::move(librados::IoCtx &io_ctx, rbd_trash_image_source_t source,
 158                    const std::string &image_name, const std::string &image_id,
 159                    uint64_t delay) {
 160   ceph_assert(!image_name.empty() && !image_id.empty());
 161   CephContext *cct((CephContext *)io_ctx.cct());
 162   ldout(cct, 20) << &io_ctx << " name=" << image_name << ", id=" << image_id
 163                  << dendl;
 164
 165   auto ictx = new I("", image_id, nullptr, io_ctx, false);
 166   int r = ictx->state->open(OPEN_FLAG_SKIP_OPEN_PARENT);
 167
 168   if (r < 0 && r != -ENOENT) {
 169     lderr(cct) << "failed to open image: " << cpp_strerror(r) << dendl;
 170     return r;
 171   }
 172
 173   if (r == 0) {
 174     cls::rbd::MirrorImage mirror_image;
 175     int mirror_r = cls_client::mirror_image_get(&ictx->md_ctx, ictx->id,
 176                                                 &mirror_image);
 177     if (mirror_r == -ENOENT) {
 178       ldout(ictx->cct, 10) << "mirroring is not enabled for this image"
 179                            << dendl;
 180     } else if (mirror_r < 0) {
 181       lderr(ictx->cct) << "failed to retrieve mirror image: "
 182                        << cpp_strerror(mirror_r) << dendl;
 183       return mirror_r;
 184     } else if (mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
 185       // a remote rbd-mirror might own the exclusive-lock on this image
 186       // and therefore we need to disable mirroring so that it closes the image
 187       r = disable_mirroring<I>(ictx);
 188       if (r < 0) {
 189         ictx->state->close();
 190         return r;
 191       }
 192     }
 193
 194     if (ictx->test_features(RBD_FEATURE_JOURNALING)) {
 195       std::unique_lock image_locker{ictx->image_lock};
 196       ictx->set_journal_policy(new journal::DisabledPolicy());
 197     }
 198
 199     ictx->owner_lock.lock_shared();
 200     if (ictx->exclusive_lock != nullptr) {
 201       ictx->exclusive_lock->block_requests(0);
 202
 203       r = ictx->operations->prepare_image_update(
 204         exclusive_lock::OPERATION_REQUEST_TYPE_GENERAL, true);
 205       if (r < 0) {
 206         lderr(cct) << "cannot obtain exclusive lock - not removing" << dendl;
 207         ictx->owner_lock.unlock_shared();
 208         ictx->state->close();
 209         return -EBUSY;
 210       }
 211     }
 212     ictx->owner_lock.unlock_shared();
 213
 214     ictx->image_lock.lock_shared();
 215     if (!ictx->migration_info.empty()) {
 216       lderr(cct) << "cannot move migrating image to trash" << dendl;
 217       ictx->image_lock.unlock_shared();
 218       ictx->state->close();
 219       return -EBUSY;
 220     }
 221     ictx->image_lock.unlock_shared();
 222
 223     if (mirror_r >= 0 &&
 224         mirror_image.mode != cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
 225       r = disable_mirroring<I>(ictx);
 226       if (r < 0) {
 227         ictx->state->close();
 228         return r;
 229       }
 230     }
 231
 232     ictx->state->close();
 233   }
 234
 235   utime_t delete_time{ceph_clock_now()};
 236   utime_t deferment_end_time{delete_time};
 237   deferment_end_time += delay;
 238   cls::rbd::TrashImageSpec trash_image_spec{
 239     static_cast<cls::rbd::TrashImageSource>(source), image_name,
 240     delete_time, deferment_end_time};
 241
 242   trash_image_spec.state = cls::rbd::TRASH_IMAGE_STATE_MOVING;
 243   C_SaferCond ctx;
 244   auto req = trash::MoveRequest<I>::create(io_ctx, image_id, trash_image_spec,
 245                                            &ctx);
 246   req->send();
 247
 248   r = ctx.wait();
 249   trash_image_spec.state = cls::rbd::TRASH_IMAGE_STATE_NORMAL;
 250   int ret = cls_client::trash_state_set(&io_ctx, image_id,
 251                                         trash_image_spec.state,
 252                                         cls::rbd::TRASH_IMAGE_STATE_MOVING);
 253   if (ret < 0 && ret != -EOPNOTSUPP) {
 254     lderr(cct) << "error setting trash image state: "
 255                << cpp_strerror(ret) << dendl;
 256     return ret;
 257   }
 258   if (r < 0) {
 259     return r;
 260   }
 261
 262   C_SaferCond notify_ctx;
 263   TrashWatcher<I>::notify_image_added(io_ctx, image_id, trash_image_spec,
 264                                       &notify_ctx);
 265   r = notify_ctx.wait();
 266   if (r < 0) {
 267     lderr(cct) << "failed to send update notification: " << cpp_strerror(r)
 268                << dendl;
 269   }
 270
 271   return 0;
 272 }
 273
 274 template <typename I>
 275 int Trash<I>::move(librados::IoCtx &io_ctx, rbd_trash_image_source_t source,
 276                    const std::string &image_name, uint64_t delay) {
 277   CephContext *cct((CephContext *)io_ctx.cct());
 278   ldout(cct, 20) << &io_ctx << " name=" << image_name << dendl;
 279
 280   // try to get image id from the directory
 281   std::string image_id;
 282   int r = cls_client::dir_get_id(&io_ctx, RBD_DIRECTORY, image_name,
 283                                  &image_id);
 284   if (r == -ENOENT) {
 285     r = io_ctx.stat(util::old_header_name(image_name), nullptr, nullptr);
 286     if (r == 0) {
 287       // cannot move V1 image to trash
 288       ldout(cct, 10) << "cannot move v1 image to trash" << dendl;
 289       return -EOPNOTSUPP;
 290     }
 291
 292     // search for an interrupted trash move request
 293     std::map<std::string, cls::rbd::TrashImageSpec> trash_image_specs;
 294     int r = list_trash_image_specs(io_ctx, &trash_image_specs, true);
 295     if (r < 0) {
 296       return r;
 297     }
 298
 299     std::experimental::erase_if(
 300       trash_image_specs, [image_name](const auto& pair) {
 301         const auto& spec = pair.second;
 302         return (spec.source != cls::rbd::TRASH_IMAGE_SOURCE_USER ||
 303                 spec.state != cls::rbd::TRASH_IMAGE_STATE_MOVING ||
 304                 spec.name != image_name);
 305       });
 306     if (trash_image_specs.empty()) {
 307       return -ENOENT;
 308     }
 309
 310     image_id = trash_image_specs.begin()->first;
 311     ldout(cct, 15) << "derived image id " << image_id << " from existing "
 312                    << "trash entry" << dendl;
 313   } else if (r < 0) {
 314     lderr(cct) << "failed to retrieve image id: " << cpp_strerror(r) << dendl;
 315     return r;
 316   }
 317
 318   if (image_name.empty() || image_id.empty()) {
 319     lderr(cct) << "invalid image name/id" << dendl;
 320     return -EINVAL;
 321   }
 322
 323   return Trash<I>::move(io_ctx, source, image_name, image_id, delay);
 324 }
 325
 326 template <typename I>
 327 int Trash<I>::get(IoCtx &io_ctx, const std::string &id,
 328               trash_image_info_t *info) {
 329   CephContext *cct((CephContext *)io_ctx.cct());
 330   ldout(cct, 20) << __func__ << " " << &io_ctx << dendl;
 331
 332   cls::rbd::TrashImageSpec spec;
 333   int r = cls_client::trash_get(&io_ctx, id, &spec);
 334   if (r == -ENOENT) {
 335     return r;
 336   } else if (r < 0) {
 337     lderr(cct) << "error retrieving trash entry: " << cpp_strerror(r)
 338                << dendl;
 339     return r;
 340   }
 341
 342   rbd_trash_image_source_t source = static_cast<rbd_trash_image_source_t>(
 343     spec.source);
 344   *info = trash_image_info_t{id, spec.name, source, spec.deletion_time.sec(),
 345                              spec.deferment_end_time.sec()};
 346   return 0;
 347 }
 348
 349 template <typename I>
 350 int Trash<I>::list(IoCtx &io_ctx, vector<trash_image_info_t> &entries,
 351                    bool exclude_user_remove_source) {
 352   CephContext *cct((CephContext *)io_ctx.cct());
 353   ldout(cct, 20) << __func__ << " " << &io_ctx << dendl;
 354
 355   std::map<std::string, cls::rbd::TrashImageSpec> trash_image_specs;
 356   int r = list_trash_image_specs(io_ctx, &trash_image_specs,
 357                                  exclude_user_remove_source);
 358   if (r < 0) {
 359     return r;
 360   }
 361
 362   entries.reserve(trash_image_specs.size());
 363   for (const auto& [image_id, spec] : trash_image_specs) {
 364     rbd_trash_image_source_t source =
 365         static_cast<rbd_trash_image_source_t>(spec.source);
 366     entries.push_back({image_id, spec.name, source,
 367                        spec.deletion_time.sec(),
 368                        spec.deferment_end_time.sec()});
 369   }
 370
 371   return 0;
 372 }
 373
 374 template <typename I>
 375 int Trash<I>::purge(IoCtx& io_ctx, time_t expire_ts,
 376                     float threshold, ProgressContext& pctx) {
 377   auto *cct((CephContext *) io_ctx.cct());
 378   ldout(cct, 20) << &io_ctx << dendl;
 379
 380   std::vector<librbd::trash_image_info_t> trash_entries;
 381   int r = librbd::api::Trash<I>::list(io_ctx, trash_entries, true);
 382   if (r < 0) {
 383     return r;
 384   }
 385
 386   trash_entries.erase(
 387       std::remove_if(trash_entries.begin(), trash_entries.end(),
 388                      [](librbd::trash_image_info_t info) {
 389                        return info.source != RBD_TRASH_IMAGE_SOURCE_USER &&
 390                          info.source != RBD_TRASH_IMAGE_SOURCE_USER_PARENT;
 391                      }),
 392       trash_entries.end());
 393
 394   std::set<std::string> to_be_removed;
 395   if (threshold != -1) {
 396     if (threshold < 0 || threshold > 1) {
 397       lderr(cct) << "argument 'threshold' is out of valid range"
 398                  << dendl;
 399       return -EINVAL;
 400     }
 401
 402     librados::bufferlist inbl;
 403     librados::bufferlist outbl;
 404     std::string pool_name = io_ctx.get_pool_name();
 405
 406     librados::Rados rados(io_ctx);
 407     rados.mon_command(R"({"prefix": "df", "format": "json"})", inbl,
 408                       &outbl, nullptr);
 409
 410     json_spirit::mValue json;
 411     if (!json_spirit::read(outbl.to_str(), json)) {
 412       lderr(cct) << "ceph df json output could not be parsed"
 413                  << dendl;
 414       return -EBADMSG;
 415     }
 416
 417     json_spirit::mArray arr = json.get_obj()["pools"].get_array();
 418
 419     double pool_percent_used = 0;
 420     uint64_t pool_total_bytes = 0;
 421
 422     std::map<std::string, std::vector<std::string>> datapools;
 423
 424     std::sort(trash_entries.begin(), trash_entries.end(),
 425         [](librbd::trash_image_info_t a, librbd::trash_image_info_t b) {
 426           return a.deferment_end_time < b.deferment_end_time;
 427         }
 428     );
 429
 430     for (const auto &entry : trash_entries) {
 431       int64_t data_pool_id = -1;
 432       r = cls_client::get_data_pool(&io_ctx, util::header_name(entry.id),
 433                                     &data_pool_id);
 434       if (r < 0 && r != -ENOENT && r != -EOPNOTSUPP) {
 435         lderr(cct) << "failed to query data pool: " << cpp_strerror(r) << dendl;
 436         return r;
 437       } else if (data_pool_id == -1) {
 438         data_pool_id = io_ctx.get_id();
 439       }
 440
 441       if (data_pool_id != io_ctx.get_id()) {
 442         librados::IoCtx data_io_ctx;
 443         r = util::create_ioctx(io_ctx, "image", data_pool_id,
 444                                {}, &data_io_ctx);
 445         if (r < 0) {
 446           lderr(cct) << "error accessing data pool" << dendl;
 447           continue;
 448         }
 449         auto data_pool = data_io_ctx.get_pool_name();
 450         datapools[data_pool].push_back(entry.id);
 451       } else {
 452         datapools[pool_name].push_back(entry.id);
 453       }
 454     }
 455
 456     uint64_t bytes_to_free = 0;
 457
 458     for (uint8_t i = 0; i < arr.size(); ++i) {
 459       json_spirit::mObject obj = arr[i].get_obj();
 460       std::string name = obj.find("name")->second.get_str();
 461       auto img = datapools.find(name);
 462       if (img != datapools.end()) {
 463         json_spirit::mObject stats = arr[i].get_obj()["stats"].get_obj();
 464         pool_percent_used = stats["percent_used"].get_real();
 465         if (pool_percent_used <= threshold) continue;
 466
 467         bytes_to_free = 0;
 468
 469         pool_total_bytes = stats["max_avail"].get_uint64() +
 470                            stats["bytes_used"].get_uint64();
 471
 472         auto bytes_threshold = (uint64_t) (pool_total_bytes *
 473                                           (pool_percent_used - threshold));
 474
 475         for (const auto &it : img->second) {
 476           auto ictx = new I("", it, nullptr, io_ctx, false);
 477           r = ictx->state->open(OPEN_FLAG_SKIP_OPEN_PARENT);
 478           if (r == -ENOENT) {
 479             continue;
 480           } else if (r < 0) {
 481             lderr(cct) << "failed to open image " << it << ": "
 482                        << cpp_strerror(r) << dendl;
 483           }
 484
 485           r = librbd::api::DiffIterate<I>::diff_iterate(
 486             ictx, cls::rbd::UserSnapshotNamespace(), nullptr, 0, ictx->size,
 487             false, true,
 488             [](uint64_t offset, size_t len, int exists, void *arg) {
 489                 auto *to_free = reinterpret_cast<uint64_t *>(arg);
 490                 if (exists)
 491                   (*to_free) += len;
 492                 return 0;
 493             }, &bytes_to_free);
 494
 495           ictx->state->close();
 496           if (r < 0) {
 497             lderr(cct) << "failed to calculate disk usage for image " << it
 498                        << ": " << cpp_strerror(r) << dendl;
 499             continue;
 500           }
 501
 502           to_be_removed.insert(it);
 503           if (bytes_to_free >= bytes_threshold) {
 504             break;
 505           }
 506         }
 507       }
 508     }
 509
 510     if (bytes_to_free == 0) {
 511       ldout(cct, 10) << "pool usage is lower than or equal to "
 512                      << (threshold * 100)
 513                      << "%" << dendl;
 514       return 0;
 515     }
 516   }
 517
 518   if (expire_ts == 0) {
 519     struct timespec now;
 520     clock_gettime(CLOCK_REALTIME, &now);
 521     expire_ts = now.tv_sec;
 522   }
 523
 524   for (const auto &entry : trash_entries) {
 525     if (expire_ts >= entry.deferment_end_time) {
 526       to_be_removed.insert(entry.id);
 527     }
 528   }
 529
 530   NoOpProgressContext remove_pctx;
 531   uint64_t list_size = to_be_removed.size(), i = 0;
 532   for (const auto &entry_id : to_be_removed) {
 533     r = librbd::api::Trash<I>::remove(io_ctx, entry_id, true, remove_pctx);
 534     if (r < 0) {
 535       if (r == -ENOTEMPTY) {
 536         ldout(cct, 5) << "image has snapshots - these must be deleted "
 537                       << "with 'rbd snap purge' before the image can be "
 538                       << "removed." << dendl;
 539       } else if (r == -EBUSY) {
 540         ldout(cct, 5) << "error: image still has watchers" << std::endl
 541                       << "This means the image is still open or the client "
 542                       << "using it crashed. Try again after closing/unmapping "
 543                       << "it or waiting 30s for the crashed client to timeout."
 544                       << dendl;
 545       } else if (r == -EUCLEAN) {
 546         ldout(cct, 5) << "Image is not in the expected state. Ensure moving "
 547                       << "the image to the trash completed successfully."
 548                       << dendl;
 549       } else if (r == -EMLINK) {
 550         ldout(cct, 5) << "Remove the image from the group and try again."
 551                       << dendl;
 552       } else {
 553         lderr(cct) << "remove error: " << cpp_strerror(r) << dendl;
 554       }
 555       return r;
 556     }
 557     pctx.update_progress(++i, list_size);
 558   }
 559
 560   return 0;
 561 }
 562
 563 template <typename I>
 564 int Trash<I>::remove(IoCtx &io_ctx, const std::string &image_id, bool force,
 565                      ProgressContext& prog_ctx) {
 566   CephContext *cct((CephContext *)io_ctx.cct());
 567   ldout(cct, 20) << "trash_remove " << &io_ctx << " " << image_id
 568                  << " " << force << dendl;
 569
 570   cls::rbd::TrashImageSpec trash_spec;
 571   int r = cls_client::trash_get(&io_ctx, image_id, &trash_spec);
 572   if (r < 0) {
 573     lderr(cct) << "error getting image id " << image_id
 574                << " info from trash: " << cpp_strerror(r) << dendl;
 575     return r;
 576   }
 577
 578   utime_t now = ceph_clock_now();
 579   if (now < trash_spec.deferment_end_time && !force) {
 580     lderr(cct) << "error: deferment time has not expired." << dendl;
 581     return -EPERM;
 582   }
 583   if (trash_spec.state == cls::rbd::TRASH_IMAGE_STATE_MOVING) {
 584     lderr(cct) << "error: image is pending moving to the trash."
 585                << dendl;
 586     return -EUCLEAN;
 587   } else if (trash_spec.state != cls::rbd::TRASH_IMAGE_STATE_NORMAL &&
 588              trash_spec.state != cls::rbd::TRASH_IMAGE_STATE_REMOVING) {
 589     lderr(cct) << "error: image is pending restoration." << dendl;
 590     return -EBUSY;
 591   }
 592
 593   ThreadPool *thread_pool;
 594   ContextWQ *op_work_queue;
 595   ImageCtx::get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
 596
 597   C_SaferCond cond;
 598   auto req = librbd::trash::RemoveRequest<I>::create(
 599       io_ctx, image_id, op_work_queue, force, prog_ctx, &cond);
 600   req->send();
 601
 602   r = cond.wait();
 603   if (r < 0) {
 604     return r;
 605   }
 606
 607   C_SaferCond notify_ctx;
 608   TrashWatcher<I>::notify_image_removed(io_ctx, image_id, &notify_ctx);
 609   r = notify_ctx.wait();
 610   if (r < 0) {
 611     lderr(cct) << "failed to send update notification: " << cpp_strerror(r)
 612                << dendl;
 613   }
 614
 615   return 0;
 616 }
 617
 618 template <typename I>
 619 int Trash<I>::restore(librados::IoCtx &io_ctx,
 620                       const TrashImageSources& trash_image_sources,
 621                       const std::string &image_id,
 622                       const std::string &image_new_name) {
 623   CephContext *cct((CephContext *)io_ctx.cct());
 624   ldout(cct, 20) << "trash_restore " << &io_ctx << " " << image_id << " "
 625                  << image_new_name << dendl;
 626
 627   cls::rbd::TrashImageSpec trash_spec;
 628   int r = cls_client::trash_get(&io_ctx, image_id, &trash_spec);
 629   if (r < 0) {
 630     lderr(cct) << "error getting image id " << image_id
 631                << " info from trash: " << cpp_strerror(r) << dendl;
 632     return r;
 633   }
 634
 635   if (trash_image_sources.count(trash_spec.source) == 0) {
 636     lderr(cct) << "Current trash source '" << trash_spec.source << "' "
 637                << "does not match expected: "
 638                << trash_image_sources << dendl;
 639     return -EINVAL;
 640   }
 641
 642   std::string image_name = image_new_name;
 643   if (trash_spec.state != cls::rbd::TRASH_IMAGE_STATE_NORMAL &&
 644       trash_spec.state != cls::rbd::TRASH_IMAGE_STATE_RESTORING) {
 645     lderr(cct) << "error restoring image id " << image_id
 646                << ", which is pending deletion" << dendl;
 647     return -EBUSY;
 648   }
 649   r = cls_client::trash_state_set(&io_ctx, image_id,
 650                                   cls::rbd::TRASH_IMAGE_STATE_RESTORING,
 651                                   cls::rbd::TRASH_IMAGE_STATE_NORMAL);
 652   if (r < 0 && r != -EOPNOTSUPP) {
 653     lderr(cct) << "error setting trash image state: "
 654                << cpp_strerror(r) << dendl;
 655     return r;
 656   }
 657
 658   if (image_name.empty()) {
 659     // if user didn't specify a new name, let's try using the old name
 660     image_name = trash_spec.name;
 661     ldout(cct, 20) << "restoring image id " << image_id << " with name "
 662                    << image_name << dendl;
 663   }
 664
 665   // check if no image exists with the same name
 666   bool create_id_obj = true;
 667   std::string existing_id;
 668   r = cls_client::get_id(&io_ctx, util::id_obj_name(image_name), &existing_id);
 669   if (r < 0 && r != -ENOENT) {
 670     lderr(cct) << "error checking if image " << image_name << " exists: "
 671                << cpp_strerror(r) << dendl;
 672     int ret = cls_client::trash_state_set(&io_ctx, image_id,
 673                                           cls::rbd::TRASH_IMAGE_STATE_NORMAL,
 674                                           cls::rbd::TRASH_IMAGE_STATE_RESTORING);
 675     if (ret < 0 && ret != -EOPNOTSUPP) {
 676       lderr(cct) << "error setting trash image state: "
 677                  << cpp_strerror(ret) << dendl;
 678     }
 679     return r;
 680   } else if (r != -ENOENT){
 681     // checking if we are recovering from an incomplete restore
 682     if (existing_id != image_id) {
 683       ldout(cct, 2) << "an image with the same name already exists" << dendl;
 684       int r2 = cls_client::trash_state_set(&io_ctx, image_id,
 685                                            cls::rbd::TRASH_IMAGE_STATE_NORMAL,
 686                                            cls::rbd::TRASH_IMAGE_STATE_RESTORING);
 687       if (r2 < 0 && r2 != -EOPNOTSUPP) {
 688       lderr(cct) << "error setting trash image state: "
 689                  << cpp_strerror(r2) << dendl;
 690      }
 691       return -EEXIST;
 692     }
 693     create_id_obj = false;
 694   }
 695
 696   if (create_id_obj) {
 697     ldout(cct, 2) << "adding id object" << dendl;
 698     librados::ObjectWriteOperation op;
 699     op.create(true);
 700     cls_client::set_id(&op, image_id);
 701     r = io_ctx.operate(util::id_obj_name(image_name), &op);
 702     if (r < 0) {
 703       lderr(cct) << "error adding id object for image " << image_name
 704                  << ": " << cpp_strerror(r) << dendl;
 705       return r;
 706     }
 707   }
 708
 709   ldout(cct, 2) << "adding rbd image to v2 directory..." << dendl;
 710   r = cls_client::dir_add_image(&io_ctx, RBD_DIRECTORY, image_name,
 711                                 image_id);
 712   if (r < 0 && r != -EEXIST) {
 713     lderr(cct) << "error adding image to v2 directory: "
 714                << cpp_strerror(r) << dendl;
 715     return r;
 716   }
 717
 718   r = enable_mirroring<I>(io_ctx, image_id);
 719   if (r < 0) {
 720     // not fatal -- ignore
 721   }
 722
 723   ldout(cct, 2) << "removing image from trash..." << dendl;
 724   r = cls_client::trash_remove(&io_ctx, image_id);
 725   if (r < 0 && r != -ENOENT) {
 726     lderr(cct) << "error removing image id " << image_id << " from trash: "
 727                << cpp_strerror(r) << dendl;
 728     return r;
 729   }
 730
 731   C_SaferCond notify_ctx;
 732   TrashWatcher<I>::notify_image_removed(io_ctx, image_id, &notify_ctx);
 733   r = notify_ctx.wait();
 734   if (r < 0) {
 735     lderr(cct) << "failed to send update notification: " << cpp_strerror(r)
 736                << dendl;
 737   }
 738
 739   return 0;
 740 }
 741
 742 } // namespace api
 743 } // namespace librbd
 744
 745 template class librbd::api::Trash<librbd::ImageCtx>;