ceph/src/librbd/image/CreateRequest.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/image/CreateRequest.h"
   5 #include "include/ceph_assert.h"
   6 #include "common/dout.h"
   7 #include "common/errno.h"
   8 #include "common/ceph_context.h"
   9 #include "cls/rbd/cls_rbd_client.h"
  10 #include "osdc/Striper.h"
  11 #include "librbd/Features.h"
  12 #include "librbd/Journal.h"
  13 #include "librbd/ObjectMap.h"
  14 #include "librbd/Utils.h"
  15 #include "librbd/image/ValidatePoolRequest.h"
  16 #include "librbd/journal/CreateRequest.h"
  17 #include "librbd/journal/RemoveRequest.h"
  18 #include "librbd/mirror/EnableRequest.h"
  19 #include "journal/Journaler.h"
  20
  21
  22 #define dout_subsys ceph_subsys_rbd
  23 #undef dout_prefix
  24 #define dout_prefix *_dout << "librbd::image::CreateRequest: " << __func__ \
  25                            << ": "
  26
  27 namespace librbd {
  28 namespace image {
  29
  30 using util::create_rados_callback;
  31 using util::create_context_callback;
  32
  33 namespace {
  34
  35 int validate_features(CephContext *cct, uint64_t features) {
  36   if (features & ~RBD_FEATURES_ALL) {
  37     lderr(cct) << "librbd does not support requested features." << dendl;
  38     return -ENOSYS;
  39   }
  40   if ((features & RBD_FEATURE_OPERATIONS) != 0) {
  41     lderr(cct) << "cannot use internally controlled features" << dendl;
  42     return -EINVAL;
  43   }
  44   if ((features & RBD_FEATURE_FAST_DIFF) != 0 &&
  45       (features & RBD_FEATURE_OBJECT_MAP) == 0) {
  46     lderr(cct) << "cannot use fast diff without object map" << dendl;
  47     return -EINVAL;
  48   }
  49   if ((features & RBD_FEATURE_OBJECT_MAP) != 0 &&
  50       (features & RBD_FEATURE_EXCLUSIVE_LOCK) == 0) {
  51     lderr(cct) << "cannot use object map without exclusive lock" << dendl;
  52     return -EINVAL;
  53   }
  54   if ((features & RBD_FEATURE_JOURNALING) != 0 &&
  55       (features & RBD_FEATURE_EXCLUSIVE_LOCK) == 0) {
  56     lderr(cct) << "cannot use journaling without exclusive lock" << dendl;
  57     return -EINVAL;
  58   }
  59
  60   return 0;
  61 }
  62
  63 int validate_striping(CephContext *cct, uint8_t order, uint64_t stripe_unit,
  64                       uint64_t stripe_count) {
  65   if ((stripe_unit && !stripe_count) ||
  66       (!stripe_unit && stripe_count)) {
  67     lderr(cct) << "must specify both (or neither) of stripe-unit and "
  68                << "stripe-count" << dendl;
  69     return -EINVAL;
  70   } else if (stripe_unit && ((1ull << order) % stripe_unit || stripe_unit > (1ull << order))) {
  71     lderr(cct) << "stripe unit is not a factor of the object size" << dendl;
  72     return -EINVAL;
  73   }
  74   return 0;
  75 }
  76
  77 bool validate_layout(CephContext *cct, uint64_t size, file_layout_t &layout) {
  78   if (!librbd::ObjectMap<>::is_compatible(layout, size)) {
  79     lderr(cct) << "image size not compatible with object map" << dendl;
  80     return false;
  81   }
  82
  83   return true;
  84 }
  85
  86 int get_image_option(const ImageOptions &image_options, int option,
  87                      uint8_t *value) {
  88   uint64_t large_value;
  89   int r = image_options.get(option, &large_value);
  90   if (r < 0) {
  91     return r;
  92   }
  93   *value = static_cast<uint8_t>(large_value);
  94   return 0;
  95 }
  96
  97 } // anonymous namespace
  98
  99 template<typename I>
 100 int CreateRequest<I>::validate_order(CephContext *cct, uint8_t order) {
 101   if (order > 25 || order < 12) {
 102     lderr(cct) << "order must be in the range [12, 25]" << dendl;
 103     return -EDOM;
 104   }
 105   return 0;
 106 }
 107
 108 #undef dout_prefix
 109 #define dout_prefix *_dout << "librbd::image::CreateRequest: " << this << " " \
 110                            << __func__ << ": "
 111
 112 template<typename I>
 113 CreateRequest<I>::CreateRequest(const ConfigProxy& config, IoCtx &ioctx,
 114                                 const std::string &image_name,
 115                                 const std::string &image_id, uint64_t size,
 116                                 const ImageOptions &image_options,
 117                                 bool skip_mirror_enable,
 118                                 cls::rbd::MirrorImageMode mirror_image_mode,
 119                                 const std::string &non_primary_global_image_id,
 120                                 const std::string &primary_mirror_uuid,
 121                                 ContextWQ *op_work_queue, Context *on_finish)
 122   : m_config(config), m_image_name(image_name), m_image_id(image_id),
 123     m_size(size), m_skip_mirror_enable(skip_mirror_enable),
 124     m_mirror_image_mode(mirror_image_mode),
 125     m_non_primary_global_image_id(non_primary_global_image_id),
 126     m_primary_mirror_uuid(primary_mirror_uuid),
 127     m_op_work_queue(op_work_queue), m_on_finish(on_finish) {
 128
 129   m_io_ctx.dup(ioctx);
 130   m_cct = reinterpret_cast<CephContext *>(m_io_ctx.cct());
 131
 132   m_id_obj = util::id_obj_name(m_image_name);
 133   m_header_obj = util::header_name(m_image_id);
 134   m_objmap_name = ObjectMap<>::object_map_name(m_image_id, CEPH_NOSNAP);
 135   m_force_non_primary = !non_primary_global_image_id.empty();
 136
 137   if (image_options.get(RBD_IMAGE_OPTION_FEATURES, &m_features) != 0) {
 138     m_features = librbd::rbd_features_from_string(
 139       m_config.get_val<std::string>("rbd_default_features"), nullptr);
 140     m_negotiate_features = true;
 141   }
 142
 143   uint64_t features_clear = 0;
 144   uint64_t features_set = 0;
 145   image_options.get(RBD_IMAGE_OPTION_FEATURES_CLEAR, &features_clear);
 146   image_options.get(RBD_IMAGE_OPTION_FEATURES_SET, &features_set);
 147
 148   uint64_t features_conflict = features_clear & features_set;
 149   features_clear &= ~features_conflict;
 150   features_set &= ~features_conflict;
 151   m_features |= features_set;
 152   m_features &= ~features_clear;
 153
 154   if ((m_features & RBD_FEATURE_OBJECT_MAP) == RBD_FEATURE_OBJECT_MAP) {
 155       m_features |= RBD_FEATURE_FAST_DIFF;
 156   }
 157
 158   if (image_options.get(RBD_IMAGE_OPTION_STRIPE_UNIT, &m_stripe_unit) != 0 ||
 159       m_stripe_unit == 0) {
 160     m_stripe_unit = m_config.get_val<Option::size_t>("rbd_default_stripe_unit");
 161   }
 162   if (image_options.get(RBD_IMAGE_OPTION_STRIPE_COUNT, &m_stripe_count) != 0 ||
 163       m_stripe_count == 0) {
 164     m_stripe_count = m_config.get_val<uint64_t>("rbd_default_stripe_count");
 165   }
 166   if (get_image_option(image_options, RBD_IMAGE_OPTION_ORDER, &m_order) != 0 ||
 167       m_order == 0) {
 168     m_order = config.get_val<uint64_t>("rbd_default_order");
 169   }
 170   if (get_image_option(image_options, RBD_IMAGE_OPTION_JOURNAL_ORDER,
 171       &m_journal_order) != 0) {
 172     m_journal_order = m_config.get_val<uint64_t>("rbd_journal_order");
 173   }
 174   if (get_image_option(image_options, RBD_IMAGE_OPTION_JOURNAL_SPLAY_WIDTH,
 175                        &m_journal_splay_width) != 0) {
 176     m_journal_splay_width = m_config.get_val<uint64_t>(
 177       "rbd_journal_splay_width");
 178   }
 179   if (image_options.get(RBD_IMAGE_OPTION_JOURNAL_POOL, &m_journal_pool) != 0) {
 180     m_journal_pool = m_config.get_val<std::string>("rbd_journal_pool");
 181   }
 182   if (image_options.get(RBD_IMAGE_OPTION_DATA_POOL, &m_data_pool) != 0) {
 183     m_data_pool = m_config.get_val<std::string>("rbd_default_data_pool");
 184   }
 185
 186   m_layout.object_size = 1ull << m_order;
 187   if (m_stripe_unit == 0 || m_stripe_count == 0) {
 188     m_layout.stripe_unit = m_layout.object_size;
 189     m_layout.stripe_count = 1;
 190   } else {
 191     m_layout.stripe_unit = m_stripe_unit;
 192     m_layout.stripe_count = m_stripe_count;
 193   }
 194
 195   if (!m_data_pool.empty() && m_data_pool != ioctx.get_pool_name()) {
 196     m_features |= RBD_FEATURE_DATA_POOL;
 197   } else {
 198     m_data_pool.clear();
 199     m_features &= ~RBD_FEATURE_DATA_POOL;
 200   }
 201
 202   if ((m_stripe_unit != 0 && m_stripe_unit != (1ULL << m_order)) ||
 203       (m_stripe_count != 0 && m_stripe_count != 1)) {
 204     m_features |= RBD_FEATURE_STRIPINGV2;
 205   } else {
 206     m_features &= ~RBD_FEATURE_STRIPINGV2;
 207   }
 208
 209   ldout(m_cct, 10) << "name=" << m_image_name << ", "
 210                    << "id=" << m_image_id << ", "
 211                    << "size=" << m_size << ", "
 212                    << "features=" << m_features << ", "
 213                    << "order=" << (uint64_t)m_order << ", "
 214                    << "stripe_unit=" << m_stripe_unit << ", "
 215                    << "stripe_count=" << m_stripe_count << ", "
 216                    << "journal_order=" << (uint64_t)m_journal_order << ", "
 217                    << "journal_splay_width="
 218                    << (uint64_t)m_journal_splay_width << ", "
 219                    << "journal_pool=" << m_journal_pool << ", "
 220                    << "data_pool=" << m_data_pool << dendl;
 221 }
 222
 223 template<typename I>
 224 void CreateRequest<I>::send() {
 225   ldout(m_cct, 20) << dendl;
 226
 227   int r = validate_features(m_cct, m_features);
 228   if (r < 0) {
 229     complete(r);
 230     return;
 231   }
 232
 233   r = validate_order(m_cct, m_order);
 234   if (r < 0) {
 235     complete(r);
 236     return;
 237   }
 238
 239   r = validate_striping(m_cct, m_order, m_stripe_unit, m_stripe_count);
 240   if (r < 0) {
 241     complete(r);
 242     return;
 243   }
 244
 245   if (((m_features & RBD_FEATURE_OBJECT_MAP) != 0) &&
 246       (!validate_layout(m_cct, m_size, m_layout))) {
 247     complete(-EINVAL);
 248     return;
 249   }
 250
 251   validate_data_pool();
 252 }
 253
 254 template <typename I>
 255 void CreateRequest<I>::validate_data_pool() {
 256   m_data_io_ctx = m_io_ctx;
 257   if ((m_features & RBD_FEATURE_DATA_POOL) != 0) {
 258     librados::Rados rados(m_io_ctx);
 259     int r = rados.ioctx_create(m_data_pool.c_str(), m_data_io_ctx);
 260     if (r < 0) {
 261       lderr(m_cct) << "data pool " << m_data_pool << " does not exist" << dendl;
 262       complete(r);
 263       return;
 264     }
 265     m_data_pool_id = m_data_io_ctx.get_id();
 266     m_data_io_ctx.set_namespace(m_io_ctx.get_namespace());
 267   }
 268
 269   if (!m_config.get_val<bool>("rbd_validate_pool")) {
 270     add_image_to_directory();
 271     return;
 272   }
 273
 274   ldout(m_cct, 15) << dendl;
 275
 276   auto ctx = create_context_callback<
 277     CreateRequest<I>, &CreateRequest<I>::handle_validate_data_pool>(this);
 278   auto req = ValidatePoolRequest<I>::create(m_data_io_ctx, m_op_work_queue,
 279                                             ctx);
 280   req->send();
 281 }
 282
 283 template <typename I>
 284 void CreateRequest<I>::handle_validate_data_pool(int r) {
 285   ldout(m_cct, 15) << "r=" << r << dendl;
 286
 287   if (r == -EINVAL) {
 288     lderr(m_cct) << "pool does not support RBD images" << dendl;
 289     complete(r);
 290     return;
 291   } else if (r < 0) {
 292     lderr(m_cct) << "failed to validate pool: " << cpp_strerror(r) << dendl;
 293     complete(r);
 294     return;
 295   }
 296
 297   add_image_to_directory();
 298 }
 299
 300 template<typename I>
 301 void CreateRequest<I>::add_image_to_directory() {
 302   ldout(m_cct, 15) << dendl;
 303
 304   librados::ObjectWriteOperation op;
 305   if (!m_io_ctx.get_namespace().empty()) {
 306     cls_client::dir_state_assert(&op, cls::rbd::DIRECTORY_STATE_READY);
 307   }
 308   cls_client::dir_add_image(&op, m_image_name, m_image_id);
 309
 310   using klass = CreateRequest<I>;
 311   librados::AioCompletion *comp =
 312     create_rados_callback<klass, &klass::handle_add_image_to_directory>(this);
 313   int r = m_io_ctx.aio_operate(RBD_DIRECTORY, comp, &op);
 314   ceph_assert(r == 0);
 315   comp->release();
 316 }
 317
 318 template<typename I>
 319 void CreateRequest<I>::handle_add_image_to_directory(int r) {
 320   ldout(m_cct, 15) << "r=" << r << dendl;
 321
 322   if (r == -EEXIST) {
 323     ldout(m_cct, 5) << "directory entry for image " << m_image_name
 324                     << " already exists" << dendl;
 325     complete(r);
 326     return;
 327   } else if (!m_io_ctx.get_namespace().empty() && r == -ENOENT) {
 328     ldout(m_cct, 5) << "namespace " << m_io_ctx.get_namespace()
 329                     << " does not exist" << dendl;
 330     complete(r);
 331     return;
 332   } else if (r < 0) {
 333     lderr(m_cct) << "error adding image to directory: " << cpp_strerror(r)
 334                  << dendl;
 335     complete(r);
 336     return;
 337   }
 338
 339   create_id_object();
 340 }
 341
 342 template<typename I>
 343 void CreateRequest<I>::create_id_object() {
 344   ldout(m_cct, 15) << dendl;
 345
 346   librados::ObjectWriteOperation op;
 347   op.create(true);
 348   cls_client::set_id(&op, m_image_id);
 349
 350   using klass = CreateRequest<I>;
 351   librados::AioCompletion *comp =
 352     create_rados_callback<klass, &klass::handle_create_id_object>(this);
 353   int r = m_io_ctx.aio_operate(m_id_obj, comp, &op);
 354   ceph_assert(r == 0);
 355   comp->release();
 356 }
 357
 358 template<typename I>
 359 void CreateRequest<I>::handle_create_id_object(int r) {
 360   ldout(m_cct, 15) << "r=" << r << dendl;
 361
 362   if (r == -EEXIST) {
 363     ldout(m_cct, 5) << "id object for  " << m_image_name << " already exists"
 364                     << dendl;
 365     m_r_saved = r;
 366     remove_from_dir();
 367     return;
 368   } else if (r < 0) {
 369     lderr(m_cct) << "error creating RBD id object: " << cpp_strerror(r)
 370                  << dendl;
 371     m_r_saved = r;
 372     remove_from_dir();
 373     return;
 374   }
 375
 376   negotiate_features();
 377 }
 378
 379 template<typename I>
 380 void CreateRequest<I>::negotiate_features() {
 381   if (!m_negotiate_features) {
 382     create_image();
 383     return;
 384   }
 385
 386   ldout(m_cct, 15) << dendl;
 387
 388   librados::ObjectReadOperation op;
 389   cls_client::get_all_features_start(&op);
 390
 391   using klass = CreateRequest<I>;
 392   librados::AioCompletion *comp =
 393     create_rados_callback<klass, &klass::handle_negotiate_features>(this);
 394
 395   m_outbl.clear();
 396   int r = m_io_ctx.aio_operate(RBD_DIRECTORY, comp, &op, &m_outbl);
 397   ceph_assert(r == 0);
 398   comp->release();
 399 }
 400
 401 template<typename I>
 402 void CreateRequest<I>::handle_negotiate_features(int r) {
 403   ldout(m_cct, 15) << "r=" << r << dendl;
 404
 405   uint64_t all_features;
 406   if (r >= 0) {
 407     auto it = m_outbl.cbegin();
 408     r = cls_client::get_all_features_finish(&it, &all_features);
 409   }
 410   if (r < 0) {
 411     ldout(m_cct, 10) << "error retrieving server supported features set: "
 412                      << cpp_strerror(r) << dendl;
 413   } else if ((m_features & all_features) != m_features) {
 414     m_features &= all_features;
 415     ldout(m_cct, 10) << "limiting default features set to server supported: "
 416                      << m_features << dendl;
 417   }
 418
 419   create_image();
 420 }
 421
 422 template<typename I>
 423 void CreateRequest<I>::create_image() {
 424   ldout(m_cct, 15) << dendl;
 425   ceph_assert(m_data_pool.empty() || m_data_pool_id != -1);
 426
 427   ostringstream oss;
 428   oss << RBD_DATA_PREFIX;
 429   if (m_data_pool_id != -1) {
 430     oss << stringify(m_io_ctx.get_id()) << ".";
 431   }
 432   oss << m_image_id;
 433   if (oss.str().length() > RBD_MAX_BLOCK_NAME_PREFIX_LENGTH) {
 434     lderr(m_cct) << "object prefix '" << oss.str() << "' too large" << dendl;
 435     m_r_saved = -EINVAL;
 436     remove_id_object();
 437     return;
 438   }
 439
 440   librados::ObjectWriteOperation op;
 441   op.create(true);
 442   cls_client::create_image(&op, m_size, m_order, m_features, oss.str(),
 443                            m_data_pool_id);
 444
 445   using klass = CreateRequest<I>;
 446   librados::AioCompletion *comp =
 447     create_rados_callback<klass, &klass::handle_create_image>(this);
 448   int r = m_io_ctx.aio_operate(m_header_obj, comp, &op);
 449   ceph_assert(r == 0);
 450   comp->release();
 451 }
 452
 453 template<typename I>
 454 void CreateRequest<I>::handle_create_image(int r) {
 455   ldout(m_cct, 15) << "r=" << r << dendl;
 456
 457   if (r == -EEXIST) {
 458     ldout(m_cct, 5) << "image id already in-use" << dendl;
 459     complete(-EBADF);
 460     return;
 461   } else if (r < 0) {
 462     lderr(m_cct) << "error writing header: " << cpp_strerror(r) << dendl;
 463     m_r_saved = r;
 464     remove_id_object();
 465     return;
 466   }
 467
 468   set_stripe_unit_count();
 469 }
 470
 471 template<typename I>
 472 void CreateRequest<I>::set_stripe_unit_count() {
 473   if ((!m_stripe_unit && !m_stripe_count) ||
 474       ((m_stripe_count == 1) && (m_stripe_unit == (1ull << m_order)))) {
 475     object_map_resize();
 476     return;
 477   }
 478
 479   ldout(m_cct, 15) << dendl;
 480
 481   librados::ObjectWriteOperation op;
 482   cls_client::set_stripe_unit_count(&op, m_stripe_unit, m_stripe_count);
 483
 484   using klass = CreateRequest<I>;
 485   librados::AioCompletion *comp =
 486     create_rados_callback<klass, &klass::handle_set_stripe_unit_count>(this);
 487   int r = m_io_ctx.aio_operate(m_header_obj, comp, &op);
 488   ceph_assert(r == 0);
 489   comp->release();
 490 }
 491
 492 template<typename I>
 493 void CreateRequest<I>::handle_set_stripe_unit_count(int r) {
 494   ldout(m_cct, 15) << "r=" << r << dendl;
 495
 496   if (r < 0) {
 497     lderr(m_cct) << "error setting stripe unit/count: "
 498                  << cpp_strerror(r) << dendl;
 499     m_r_saved = r;
 500     remove_header_object();
 501     return;
 502   }
 503
 504   object_map_resize();
 505 }
 506
 507 template<typename I>
 508 void CreateRequest<I>::object_map_resize() {
 509   if ((m_features & RBD_FEATURE_OBJECT_MAP) == 0) {
 510     fetch_mirror_mode();
 511     return;
 512   }
 513
 514   ldout(m_cct, 15) << dendl;
 515
 516   librados::ObjectWriteOperation op;
 517   cls_client::object_map_resize(&op, Striper::get_num_objects(m_layout, m_size),
 518                                 OBJECT_NONEXISTENT);
 519
 520   using klass = CreateRequest<I>;
 521   librados::AioCompletion *comp =
 522     create_rados_callback<klass, &klass::handle_object_map_resize>(this);
 523   int r = m_io_ctx.aio_operate(m_objmap_name, comp, &op);
 524   ceph_assert(r == 0);
 525   comp->release();
 526 }
 527
 528 template<typename I>
 529 void CreateRequest<I>::handle_object_map_resize(int r) {
 530   ldout(m_cct, 15) << "r=" << r << dendl;
 531
 532   if (r < 0) {
 533     lderr(m_cct) << "error creating initial object map: "
 534                  << cpp_strerror(r) << dendl;
 535
 536     m_r_saved = r;
 537     remove_header_object();
 538     return;
 539   }
 540
 541   fetch_mirror_mode();
 542 }
 543
 544 template<typename I>
 545 void CreateRequest<I>::fetch_mirror_mode() {
 546   if ((m_features & RBD_FEATURE_JOURNALING) == 0) {
 547     mirror_image_enable();
 548     return;
 549   }
 550
 551   ldout(m_cct, 15) << dendl;
 552
 553   librados::ObjectReadOperation op;
 554   cls_client::mirror_mode_get_start(&op);
 555
 556   using klass = CreateRequest<I>;
 557   librados::AioCompletion *comp =
 558     create_rados_callback<klass, &klass::handle_fetch_mirror_mode>(this);
 559   m_outbl.clear();
 560   int r = m_io_ctx.aio_operate(RBD_MIRRORING, comp, &op, &m_outbl);
 561   ceph_assert(r == 0);
 562   comp->release();
 563 }
 564
 565 template<typename I>
 566 void CreateRequest<I>::handle_fetch_mirror_mode(int r) {
 567   ldout(m_cct, 15) << "r=" << r << dendl;
 568
 569   if ((r < 0) && (r != -ENOENT)) {
 570     lderr(m_cct) << "failed to retrieve mirror mode: " << cpp_strerror(r)
 571                  << dendl;
 572
 573     m_r_saved = r;
 574     remove_object_map();
 575     return;
 576   }
 577
 578   m_mirror_mode = cls::rbd::MIRROR_MODE_DISABLED;
 579   if (r == 0) {
 580     auto it = m_outbl.cbegin();
 581     r = cls_client::mirror_mode_get_finish(&it, &m_mirror_mode);
 582     if (r < 0) {
 583       lderr(m_cct) << "Failed to retrieve mirror mode" << dendl;
 584
 585       m_r_saved = r;
 586       remove_object_map();
 587       return;
 588     }
 589   }
 590
 591   journal_create();
 592 }
 593
 594 template<typename I>
 595 void CreateRequest<I>::journal_create() {
 596   ldout(m_cct, 15) << dendl;
 597
 598   using klass = CreateRequest<I>;
 599   Context *ctx = create_context_callback<klass, &klass::handle_journal_create>(
 600     this);
 601
 602   // only link to remote primary mirror uuid if in journal-based
 603   // mirroring mode
 604   bool use_primary_mirror_uuid = (
 605     m_force_non_primary &&
 606     m_mirror_image_mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL);
 607
 608   librbd::journal::TagData tag_data;
 609   tag_data.mirror_uuid = (use_primary_mirror_uuid ? m_primary_mirror_uuid :
 610                           librbd::Journal<I>::LOCAL_MIRROR_UUID);
 611
 612   auto req = librbd::journal::CreateRequest<I>::create(
 613     m_io_ctx, m_image_id, m_journal_order, m_journal_splay_width,
 614     m_journal_pool, cls::journal::Tag::TAG_CLASS_NEW, tag_data,
 615     librbd::Journal<I>::IMAGE_CLIENT_ID, m_op_work_queue, ctx);
 616   req->send();
 617 }
 618
 619 template<typename I>
 620 void CreateRequest<I>::handle_journal_create(int r) {
 621   ldout(m_cct, 15) << "r=" << r << dendl;
 622
 623   if (r < 0) {
 624     lderr(m_cct) << "error creating journal: " << cpp_strerror(r)
 625                  << dendl;
 626
 627     m_r_saved = r;
 628     remove_object_map();
 629     return;
 630   }
 631
 632   mirror_image_enable();
 633 }
 634
 635 template<typename I>
 636 void CreateRequest<I>::mirror_image_enable() {
 637   if (((m_mirror_mode != cls::rbd::MIRROR_MODE_POOL) && !m_force_non_primary) ||
 638       m_skip_mirror_enable) {
 639     complete(0);
 640     return;
 641   }
 642
 643   ldout(m_cct, 15) << dendl;
 644   auto ctx = create_context_callback<
 645     CreateRequest<I>, &CreateRequest<I>::handle_mirror_image_enable>(this);
 646
 647   auto req = mirror::EnableRequest<I>::create(
 648     m_io_ctx, m_image_id, m_mirror_image_mode,
 649     m_non_primary_global_image_id, m_op_work_queue, ctx);
 650   req->send();
 651 }
 652
 653 template<typename I>
 654 void CreateRequest<I>::handle_mirror_image_enable(int r) {
 655   ldout(m_cct, 15) << "r=" << r << dendl;
 656
 657   if (r < 0) {
 658     lderr(m_cct) << "cannot enable mirroring: " << cpp_strerror(r)
 659                  << dendl;
 660
 661     m_r_saved = r;
 662     journal_remove();
 663     return;
 664   }
 665
 666   complete(0);
 667 }
 668
 669 template<typename I>
 670 void CreateRequest<I>::complete(int r) {
 671   ldout(m_cct, 10) << "r=" << r << dendl;
 672
 673   m_data_io_ctx.close();
 674   auto on_finish = m_on_finish;
 675   delete this;
 676   on_finish->complete(r);
 677 }
 678
 679 // cleanup
 680 template<typename I>
 681 void CreateRequest<I>::journal_remove() {
 682   if ((m_features & RBD_FEATURE_JOURNALING) == 0) {
 683     remove_object_map();
 684     return;
 685   }
 686
 687   ldout(m_cct, 15) << dendl;
 688
 689   using klass = CreateRequest<I>;
 690   Context *ctx = create_context_callback<klass, &klass::handle_journal_remove>(
 691     this);
 692
 693   librbd::journal::RemoveRequest<I> *req =
 694     librbd::journal::RemoveRequest<I>::create(
 695       m_io_ctx, m_image_id, librbd::Journal<I>::IMAGE_CLIENT_ID, m_op_work_queue,
 696       ctx);
 697   req->send();
 698 }
 699
 700 template<typename I>
 701 void CreateRequest<I>::handle_journal_remove(int r) {
 702   ldout(m_cct, 15) << "r=" << r << dendl;
 703
 704   if (r < 0) {
 705     lderr(m_cct) << "error cleaning up journal after creation failed: "
 706                  << cpp_strerror(r) << dendl;
 707   }
 708
 709   remove_object_map();
 710 }
 711
 712 template<typename I>
 713 void CreateRequest<I>::remove_object_map() {
 714   if ((m_features & RBD_FEATURE_OBJECT_MAP) == 0) {
 715     remove_header_object();
 716     return;
 717   }
 718
 719   ldout(m_cct, 15) << dendl;
 720
 721   using klass = CreateRequest<I>;
 722   librados::AioCompletion *comp =
 723     create_rados_callback<klass, &klass::handle_remove_object_map>(this);
 724   int r = m_io_ctx.aio_remove(m_objmap_name, comp);
 725   ceph_assert(r == 0);
 726   comp->release();
 727 }
 728
 729 template<typename I>
 730 void CreateRequest<I>::handle_remove_object_map(int r) {
 731   ldout(m_cct, 15) << "r=" << r << dendl;
 732
 733   if (r < 0) {
 734     lderr(m_cct) << "error cleaning up object map after creation failed: "
 735                  << cpp_strerror(r) << dendl;
 736   }
 737
 738   remove_header_object();
 739 }
 740
 741 template<typename I>
 742 void CreateRequest<I>::remove_header_object() {
 743   ldout(m_cct, 15) << dendl;
 744
 745   using klass = CreateRequest<I>;
 746   librados::AioCompletion *comp =
 747     create_rados_callback<klass, &klass::handle_remove_header_object>(this);
 748   int r = m_io_ctx.aio_remove(m_header_obj, comp);
 749   ceph_assert(r == 0);
 750   comp->release();
 751 }
 752
 753 template<typename I>
 754 void CreateRequest<I>::handle_remove_header_object(int r) {
 755   ldout(m_cct, 15) << "r=" << r << dendl;
 756
 757   if (r < 0) {
 758     lderr(m_cct) << "error cleaning up image header after creation failed: "
 759                  << cpp_strerror(r) << dendl;
 760   }
 761
 762   remove_id_object();
 763 }
 764
 765 template<typename I>
 766 void CreateRequest<I>::remove_id_object() {
 767   ldout(m_cct, 15) << dendl;
 768
 769   using klass = CreateRequest<I>;
 770   librados::AioCompletion *comp =
 771     create_rados_callback<klass, &klass::handle_remove_id_object>(this);
 772   int r = m_io_ctx.aio_remove(m_id_obj, comp);
 773   ceph_assert(r == 0);
 774   comp->release();
 775 }
 776
 777 template<typename I>
 778 void CreateRequest<I>::handle_remove_id_object(int r) {
 779   ldout(m_cct, 15) << "r=" << r << dendl;
 780
 781   if (r < 0) {
 782     lderr(m_cct) << "error cleaning up id object after creation failed: "
 783                  << cpp_strerror(r) << dendl;
 784   }
 785
 786   remove_from_dir();
 787 }
 788
 789 template<typename I>
 790 void CreateRequest<I>::remove_from_dir() {
 791   ldout(m_cct, 15) << dendl;
 792
 793   librados::ObjectWriteOperation op;
 794   cls_client::dir_remove_image(&op, m_image_name, m_image_id);
 795
 796   using klass = CreateRequest<I>;
 797   librados::AioCompletion *comp =
 798     create_rados_callback<klass, &klass::handle_remove_from_dir>(this);
 799   int r = m_io_ctx.aio_operate(RBD_DIRECTORY, comp, &op);
 800   ceph_assert(r == 0);
 801   comp->release();
 802 }
 803
 804 template<typename I>
 805 void CreateRequest<I>::handle_remove_from_dir(int r) {
 806   ldout(m_cct, 15) << "r=" << r << dendl;
 807
 808   if (r < 0) {
 809     lderr(m_cct) << "error cleaning up image from rbd_directory object "
 810                  << "after creation failed: " << cpp_strerror(r) << dendl;
 811   }
 812
 813   complete(m_r_saved);
 814 }
 815
 816 } //namespace image
 817 } //namespace librbd
 818
 819 template class librbd::image::CreateRequest<librbd::ImageCtx>;