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