ceph/src/rgw/rgw_reshard.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #include <limits>
   5 #include <sstream>
   6
   7 #include "rgw_rados.h"
   8 #include "rgw_zone.h"
   9 #include "rgw_bucket.h"
  10 #include "rgw_reshard.h"
  11 #include "rgw_sal.h"
  12 #include "cls/rgw/cls_rgw_client.h"
  13 #include "cls/lock/cls_lock_client.h"
  14 #include "common/errno.h"
  15 #include "common/ceph_json.h"
  16
  17 #include "common/dout.h"
  18
  19 #include "services/svc_zone.h"
  20 #include "services/svc_sys_obj.h"
  21 #include "services/svc_tier_rados.h"
  22
  23 #define dout_context g_ceph_context
  24 #define dout_subsys ceph_subsys_rgw
  25
  26 const string reshard_oid_prefix = "reshard.";
  27 const string reshard_lock_name = "reshard_process";
  28 const string bucket_instance_lock_name = "bucket_instance_lock";
  29
  30 /* All primes up to 2000 used to attempt to make dynamic sharding use
  31  * a prime numbers of shards. Note: this list also includes 1 for when
  32  * 1 shard is the most appropriate, even though 1 is not prime.
  33  */
  34 const std::initializer_list<uint16_t> RGWBucketReshard::reshard_primes = {
  35   1, 2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61,
  36   67, 71, 73, 79, 83, 89, 97, 101, 103, 107, 109, 113, 127, 131, 137,
  37   139, 149, 151, 157, 163, 167, 173, 179, 181, 191, 193, 197, 199, 211,
  38   223, 227, 229, 233, 239, 241, 251, 257, 263, 269, 271, 277, 281, 283,
  39   293, 307, 311, 313, 317, 331, 337, 347, 349, 353, 359, 367, 373, 379,
  40   383, 389, 397, 401, 409, 419, 421, 431, 433, 439, 443, 449, 457, 461,
  41   463, 467, 479, 487, 491, 499, 503, 509, 521, 523, 541, 547, 557, 563,
  42   569, 571, 577, 587, 593, 599, 601, 607, 613, 617, 619, 631, 641, 643,
  43   647, 653, 659, 661, 673, 677, 683, 691, 701, 709, 719, 727, 733, 739,
  44   743, 751, 757, 761, 769, 773, 787, 797, 809, 811, 821, 823, 827, 829,
  45   839, 853, 857, 859, 863, 877, 881, 883, 887, 907, 911, 919, 929, 937,
  46   941, 947, 953, 967, 971, 977, 983, 991, 997, 1009, 1013, 1019, 1021,
  47   1031, 1033, 1039, 1049, 1051, 1061, 1063, 1069, 1087, 1091, 1093,
  48   1097, 1103, 1109, 1117, 1123, 1129, 1151, 1153, 1163, 1171, 1181,
  49   1187, 1193, 1201, 1213, 1217, 1223, 1229, 1231, 1237, 1249, 1259,
  50   1277, 1279, 1283, 1289, 1291, 1297, 1301, 1303, 1307, 1319, 1321,
  51   1327, 1361, 1367, 1373, 1381, 1399, 1409, 1423, 1427, 1429, 1433,
  52   1439, 1447, 1451, 1453, 1459, 1471, 1481, 1483, 1487, 1489, 1493,
  53   1499, 1511, 1523, 1531, 1543, 1549, 1553, 1559, 1567, 1571, 1579,
  54   1583, 1597, 1601, 1607, 1609, 1613, 1619, 1621, 1627, 1637, 1657,
  55   1663, 1667, 1669, 1693, 1697, 1699, 1709, 1721, 1723, 1733, 1741,
  56   1747, 1753, 1759, 1777, 1783, 1787, 1789, 1801, 1811, 1823, 1831,
  57   1847, 1861, 1867, 1871, 1873, 1877, 1879, 1889, 1901, 1907, 1913,
  58   1931, 1933, 1949, 1951, 1973, 1979, 1987, 1993, 1997, 1999
  59 };
  60
  61 class BucketReshardShard {
  62   rgw::sal::RGWRadosStore *store;
  63   const RGWBucketInfo& bucket_info;
  64   int num_shard;
  65   RGWRados::BucketShard bs;
  66   vector<rgw_cls_bi_entry> entries;
  67   map<RGWObjCategory, rgw_bucket_category_stats> stats;
  68   deque<librados::AioCompletion *>& aio_completions;
  69   uint64_t max_aio_completions;
  70   uint64_t reshard_shard_batch_size;
  71
  72   int wait_next_completion() {
  73     librados::AioCompletion *c = aio_completions.front();
  74     aio_completions.pop_front();
  75
  76     c->wait_for_complete();
  77
  78     int ret = c->get_return_value();
  79     c->release();
  80
  81     if (ret < 0) {
  82       derr << "ERROR: reshard rados operation failed: " << cpp_strerror(-ret) << dendl;
  83       return ret;
  84     }
  85
  86     return 0;
  87   }
  88
  89   int get_completion(librados::AioCompletion **c) {
  90     if (aio_completions.size() >= max_aio_completions) {
  91       int ret = wait_next_completion();
  92       if (ret < 0) {
  93         return ret;
  94       }
  95     }
  96
  97     *c = librados::Rados::aio_create_completion(nullptr, nullptr);
  98     aio_completions.push_back(*c);
  99
 100     return 0;
 101   }
 102
 103 public:
 104   BucketReshardShard(rgw::sal::RGWRadosStore *_store, const RGWBucketInfo& _bucket_info,
 105                      int _num_shard,
 106                      deque<librados::AioCompletion *>& _completions) :
 107     store(_store), bucket_info(_bucket_info), bs(store->getRados()),
 108     aio_completions(_completions)
 109   {
 110     num_shard = (bucket_info.num_shards > 0 ? _num_shard : -1);
 111     bs.init(bucket_info.bucket, num_shard, nullptr /* no RGWBucketInfo */);
 112
 113     max_aio_completions =
 114       store->ctx()->_conf.get_val<uint64_t>("rgw_reshard_max_aio");
 115     reshard_shard_batch_size =
 116       store->ctx()->_conf.get_val<uint64_t>("rgw_reshard_batch_size");
 117   }
 118
 119   int get_num_shard() {
 120     return num_shard;
 121   }
 122
 123   int add_entry(rgw_cls_bi_entry& entry, bool account, RGWObjCategory category,
 124                 const rgw_bucket_category_stats& entry_stats) {
 125     entries.push_back(entry);
 126     if (account) {
 127       rgw_bucket_category_stats& target = stats[category];
 128       target.num_entries += entry_stats.num_entries;
 129       target.total_size += entry_stats.total_size;
 130       target.total_size_rounded += entry_stats.total_size_rounded;
 131       target.actual_size += entry_stats.actual_size;
 132     }
 133     if (entries.size() >= reshard_shard_batch_size) {
 134       int ret = flush();
 135       if (ret < 0) {
 136         return ret;
 137       }
 138     }
 139
 140     return 0;
 141   }
 142
 143   int flush() {
 144     if (entries.size() == 0) {
 145       return 0;
 146     }
 147
 148     librados::ObjectWriteOperation op;
 149     for (auto& entry : entries) {
 150       store->getRados()->bi_put(op, bs, entry);
 151     }
 152     cls_rgw_bucket_update_stats(op, false, stats);
 153
 154     librados::AioCompletion *c;
 155     int ret = get_completion(&c);
 156     if (ret < 0) {
 157       return ret;
 158     }
 159     ret = bs.bucket_obj.aio_operate(c, &op);
 160     if (ret < 0) {
 161       derr << "ERROR: failed to store entries in target bucket shard (bs=" << bs.bucket << "/" << bs.shard_id << ") error=" << cpp_strerror(-ret) << dendl;
 162       return ret;
 163     }
 164     entries.clear();
 165     stats.clear();
 166     return 0;
 167   }
 168
 169   int wait_all_aio() {
 170     int ret = 0;
 171     while (!aio_completions.empty()) {
 172       int r = wait_next_completion();
 173       if (r < 0) {
 174         ret = r;
 175       }
 176     }
 177     return ret;
 178   }
 179 }; // class BucketReshardShard
 180
 181
 182 class BucketReshardManager {
 183   rgw::sal::RGWRadosStore *store;
 184   const RGWBucketInfo& target_bucket_info;
 185   deque<librados::AioCompletion *> completions;
 186   int num_target_shards;
 187   vector<BucketReshardShard *> target_shards;
 188
 189 public:
 190   BucketReshardManager(rgw::sal::RGWRadosStore *_store,
 191                        const RGWBucketInfo& _target_bucket_info,
 192                        int _num_target_shards) :
 193     store(_store), target_bucket_info(_target_bucket_info),
 194     num_target_shards(_num_target_shards)
 195   {
 196     target_shards.resize(num_target_shards);
 197     for (int i = 0; i < num_target_shards; ++i) {
 198       target_shards[i] = new BucketReshardShard(store, target_bucket_info, i, completions);
 199     }
 200   }
 201
 202   ~BucketReshardManager() {
 203     for (auto& shard : target_shards) {
 204       int ret = shard->wait_all_aio();
 205       if (ret < 0) {
 206         ldout(store->ctx(), 20) << __func__ <<
 207           ": shard->wait_all_aio() returned ret=" << ret << dendl;
 208       }
 209     }
 210   }
 211
 212   int add_entry(int shard_index,
 213                 rgw_cls_bi_entry& entry, bool account, RGWObjCategory category,
 214                 const rgw_bucket_category_stats& entry_stats) {
 215     int ret = target_shards[shard_index]->add_entry(entry, account, category,
 216                                                     entry_stats);
 217     if (ret < 0) {
 218       derr << "ERROR: target_shards.add_entry(" << entry.idx <<
 219         ") returned error: " << cpp_strerror(-ret) << dendl;
 220       return ret;
 221     }
 222
 223     return 0;
 224   }
 225
 226   int finish() {
 227     int ret = 0;
 228     for (auto& shard : target_shards) {
 229       int r = shard->flush();
 230       if (r < 0) {
 231         derr << "ERROR: target_shards[" << shard->get_num_shard() << "].flush() returned error: " << cpp_strerror(-r) << dendl;
 232         ret = r;
 233       }
 234     }
 235     for (auto& shard : target_shards) {
 236       int r = shard->wait_all_aio();
 237       if (r < 0) {
 238         derr << "ERROR: target_shards[" << shard->get_num_shard() << "].wait_all_aio() returned error: " << cpp_strerror(-r) << dendl;
 239         ret = r;
 240       }
 241       delete shard;
 242     }
 243     target_shards.clear();
 244     return ret;
 245   }
 246 }; // class BucketReshardManager
 247
 248 RGWBucketReshard::RGWBucketReshard(rgw::sal::RGWRadosStore *_store,
 249                                    const RGWBucketInfo& _bucket_info,
 250                                    const map<string, bufferlist>& _bucket_attrs,
 251                                    RGWBucketReshardLock* _outer_reshard_lock) :
 252   store(_store), bucket_info(_bucket_info), bucket_attrs(_bucket_attrs),
 253   reshard_lock(store, bucket_info, true),
 254   outer_reshard_lock(_outer_reshard_lock)
 255 { }
 256
 257 int RGWBucketReshard::set_resharding_status(rgw::sal::RGWRadosStore* store,
 258                                             const RGWBucketInfo& bucket_info,
 259                                             const string& new_instance_id,
 260                                             int32_t num_shards,
 261                                             cls_rgw_reshard_status status)
 262 {
 263   if (new_instance_id.empty()) {
 264     ldout(store->ctx(), 0) << __func__ << " missing new bucket instance id" << dendl;
 265     return -EINVAL;
 266   }
 267
 268   cls_rgw_bucket_instance_entry instance_entry;
 269   instance_entry.set_status(new_instance_id, num_shards, status);
 270
 271   int ret = store->getRados()->bucket_set_reshard(bucket_info, instance_entry);
 272   if (ret < 0) {
 273     ldout(store->ctx(), 0) << "RGWReshard::" << __func__ << " ERROR: error setting bucket resharding flag on bucket index: "
 274                   << cpp_strerror(-ret) << dendl;
 275     return ret;
 276   }
 277   return 0;
 278 }
 279
 280 // reshard lock assumes lock is held
 281 int RGWBucketReshard::clear_resharding(rgw::sal::RGWRadosStore* store,
 282                                        const RGWBucketInfo& bucket_info)
 283 {
 284   int ret = clear_index_shard_reshard_status(store, bucket_info);
 285   if (ret < 0) {
 286     ldout(store->ctx(), 0) << "RGWBucketReshard::" << __func__ <<
 287       " ERROR: error clearing reshard status from index shard " <<
 288       cpp_strerror(-ret) << dendl;
 289     return ret;
 290   }
 291
 292   cls_rgw_bucket_instance_entry instance_entry;
 293   ret = store->getRados()->bucket_set_reshard(bucket_info, instance_entry);
 294   if (ret < 0) {
 295     ldout(store->ctx(), 0) << "RGWReshard::" << __func__ <<
 296       " ERROR: error setting bucket resharding flag on bucket index: " <<
 297       cpp_strerror(-ret) << dendl;
 298     return ret;
 299   }
 300
 301   return 0;
 302 }
 303
 304 int RGWBucketReshard::clear_index_shard_reshard_status(rgw::sal::RGWRadosStore* store,
 305                                                        const RGWBucketInfo& bucket_info)
 306 {
 307   uint32_t num_shards = bucket_info.num_shards;
 308
 309   if (num_shards < std::numeric_limits<uint32_t>::max()) {
 310     int ret = set_resharding_status(store, bucket_info,
 311                                     bucket_info.bucket.bucket_id,
 312                                     (num_shards < 1 ? 1 : num_shards),
 313                                     cls_rgw_reshard_status::NOT_RESHARDING);
 314     if (ret < 0) {
 315       ldout(store->ctx(), 0) << "RGWBucketReshard::" << __func__ <<
 316         " ERROR: error clearing reshard status from index shard " <<
 317         cpp_strerror(-ret) << dendl;
 318       return ret;
 319     }
 320   }
 321
 322   return 0;
 323 }
 324
 325 static int create_new_bucket_instance(rgw::sal::RGWRadosStore *store,
 326                                       int new_num_shards,
 327                                       const RGWBucketInfo& bucket_info,
 328                                       map<string, bufferlist>& attrs,
 329                                       RGWBucketInfo& new_bucket_info)
 330 {
 331   new_bucket_info = bucket_info;
 332
 333   store->getRados()->create_bucket_id(&new_bucket_info.bucket.bucket_id);
 334
 335   new_bucket_info.num_shards = new_num_shards;
 336   new_bucket_info.objv_tracker.clear();
 337
 338   new_bucket_info.new_bucket_instance_id.clear();
 339   new_bucket_info.reshard_status = cls_rgw_reshard_status::NOT_RESHARDING;
 340
 341   int ret = store->svc()->bi->init_index(new_bucket_info);
 342   if (ret < 0) {
 343     cerr << "ERROR: failed to init new bucket indexes: " << cpp_strerror(-ret) << std::endl;
 344     return ret;
 345   }
 346
 347   ret = store->getRados()->put_bucket_instance_info(new_bucket_info, true, real_time(), &attrs);
 348   if (ret < 0) {
 349     cerr << "ERROR: failed to store new bucket instance info: " << cpp_strerror(-ret) << std::endl;
 350     return ret;
 351   }
 352
 353   return 0;
 354 }
 355
 356 int RGWBucketReshard::create_new_bucket_instance(int new_num_shards,
 357                                                  RGWBucketInfo& new_bucket_info)
 358 {
 359   return ::create_new_bucket_instance(store, new_num_shards,
 360                                       bucket_info, bucket_attrs, new_bucket_info);
 361 }
 362
 363 int RGWBucketReshard::cancel()
 364 {
 365   int ret = reshard_lock.lock();
 366   if (ret < 0) {
 367     return ret;
 368   }
 369
 370   ret = clear_resharding();
 371
 372   reshard_lock.unlock();
 373   return ret;
 374 }
 375
 376 class BucketInfoReshardUpdate
 377 {
 378   rgw::sal::RGWRadosStore *store;
 379   RGWBucketInfo& bucket_info;
 380   std::map<string, bufferlist> bucket_attrs;
 381
 382   bool in_progress{false};
 383
 384   int set_status(cls_rgw_reshard_status s) {
 385     bucket_info.reshard_status = s;
 386     int ret = store->getRados()->put_bucket_instance_info(bucket_info, false, real_time(), &bucket_attrs);
 387     if (ret < 0) {
 388       ldout(store->ctx(), 0) << "ERROR: failed to write bucket info, ret=" << ret << dendl;
 389       return ret;
 390     }
 391     return 0;
 392   }
 393
 394 public:
 395   BucketInfoReshardUpdate(rgw::sal::RGWRadosStore *_store,
 396                           RGWBucketInfo& _bucket_info,
 397                           map<string, bufferlist>& _bucket_attrs,
 398                           const string& new_bucket_id) :
 399     store(_store),
 400     bucket_info(_bucket_info),
 401     bucket_attrs(_bucket_attrs)
 402   {
 403     bucket_info.new_bucket_instance_id = new_bucket_id;
 404   }
 405
 406   ~BucketInfoReshardUpdate() {
 407     if (in_progress) {
 408       // resharding must not have ended correctly, clean up
 409       int ret =
 410         RGWBucketReshard::clear_index_shard_reshard_status(store, bucket_info);
 411       if (ret < 0) {
 412         lderr(store->ctx()) << "Error: " << __func__ <<
 413           " clear_index_shard_status returned " << ret << dendl;
 414       }
 415       bucket_info.new_bucket_instance_id.clear();
 416
 417       // clears new_bucket_instance as well
 418       set_status(cls_rgw_reshard_status::NOT_RESHARDING);
 419     }
 420   }
 421
 422   int start() {
 423     int ret = set_status(cls_rgw_reshard_status::IN_PROGRESS);
 424     if (ret < 0) {
 425       return ret;
 426     }
 427     in_progress = true;
 428     return 0;
 429   }
 430
 431   int complete() {
 432     int ret = set_status(cls_rgw_reshard_status::DONE);
 433     if (ret < 0) {
 434       return ret;
 435     }
 436     in_progress = false;
 437     return 0;
 438   }
 439 };
 440
 441
 442 RGWBucketReshardLock::RGWBucketReshardLock(rgw::sal::RGWRadosStore* _store,
 443                                            const std::string& reshard_lock_oid,
 444                                            bool _ephemeral) :
 445   store(_store),
 446   lock_oid(reshard_lock_oid),
 447   ephemeral(_ephemeral),
 448   internal_lock(reshard_lock_name)
 449 {
 450   const int lock_dur_secs = store->ctx()->_conf.get_val<uint64_t>(
 451     "rgw_reshard_bucket_lock_duration");
 452   duration = std::chrono::seconds(lock_dur_secs);
 453
 454 #define COOKIE_LEN 16
 455   char cookie_buf[COOKIE_LEN + 1];
 456   gen_rand_alphanumeric(store->ctx(), cookie_buf, sizeof(cookie_buf) - 1);
 457   cookie_buf[COOKIE_LEN] = '\0';
 458
 459   internal_lock.set_cookie(cookie_buf);
 460   internal_lock.set_duration(duration);
 461 }
 462
 463 int RGWBucketReshardLock::lock() {
 464   internal_lock.set_must_renew(false);
 465   int ret;
 466   if (ephemeral) {
 467     ret = internal_lock.lock_exclusive_ephemeral(&store->getRados()->reshard_pool_ctx,
 468                                                  lock_oid);
 469   } else {
 470     ret = internal_lock.lock_exclusive(&store->getRados()->reshard_pool_ctx, lock_oid);
 471   }
 472   if (ret < 0) {
 473     ldout(store->ctx(), 0) << "RGWReshardLock::" << __func__ <<
 474       " failed to acquire lock on " << lock_oid << " ret=" << ret << dendl;
 475     return ret;
 476   }
 477   reset_time(Clock::now());
 478
 479   return 0;
 480 }
 481
 482 void RGWBucketReshardLock::unlock() {
 483   int ret = internal_lock.unlock(&store->getRados()->reshard_pool_ctx, lock_oid);
 484   if (ret < 0) {
 485     ldout(store->ctx(), 0) << "WARNING: RGWBucketReshardLock::" << __func__ <<
 486       " failed to drop lock on " << lock_oid << " ret=" << ret << dendl;
 487   }
 488 }
 489
 490 int RGWBucketReshardLock::renew(const Clock::time_point& now) {
 491   internal_lock.set_must_renew(true);
 492   int ret;
 493   if (ephemeral) {
 494     ret = internal_lock.lock_exclusive_ephemeral(&store->getRados()->reshard_pool_ctx,
 495                                                  lock_oid);
 496   } else {
 497     ret = internal_lock.lock_exclusive(&store->getRados()->reshard_pool_ctx, lock_oid);
 498   }
 499   if (ret < 0) { /* expired or already locked by another processor */
 500     std::stringstream error_s;
 501     if (-ENOENT == ret) {
 502       error_s << "ENOENT (lock expired or never initially locked)";
 503     } else {
 504       error_s << ret << " (" << cpp_strerror(-ret) << ")";
 505     }
 506     ldout(store->ctx(), 5) << __func__ << "(): failed to renew lock on " <<
 507       lock_oid << " with error " << error_s.str() << dendl;
 508     return ret;
 509   }
 510   internal_lock.set_must_renew(false);
 511
 512   reset_time(now);
 513   ldout(store->ctx(), 20) << __func__ << "(): successfully renewed lock on " <<
 514     lock_oid << dendl;
 515
 516   return 0;
 517 }
 518
 519
 520 int RGWBucketReshard::do_reshard(int num_shards,
 521                                  RGWBucketInfo& new_bucket_info,
 522                                  int max_entries,
 523                                  bool verbose,
 524                                  ostream *out,
 525                                  Formatter *formatter)
 526 {
 527   rgw_bucket& bucket = bucket_info.bucket;
 528
 529   int ret = 0;
 530
 531   if (out) {
 532     (*out) << "tenant: " << bucket_info.bucket.tenant << std::endl;
 533     (*out) << "bucket name: " << bucket_info.bucket.name << std::endl;
 534     (*out) << "old bucket instance id: " << bucket_info.bucket.bucket_id <<
 535       std::endl;
 536     (*out) << "new bucket instance id: " << new_bucket_info.bucket.bucket_id <<
 537       std::endl;
 538   }
 539
 540   /* update bucket info -- in progress*/
 541   list<rgw_cls_bi_entry> entries;
 542
 543   if (max_entries < 0) {
 544     ldout(store->ctx(), 0) << __func__ <<
 545       ": can't reshard, negative max_entries" << dendl;
 546     return -EINVAL;
 547   }
 548
 549   // NB: destructor cleans up sharding state if reshard does not
 550   // complete successfully
 551   BucketInfoReshardUpdate bucket_info_updater(store, bucket_info, bucket_attrs, new_bucket_info.bucket.bucket_id);
 552
 553   ret = bucket_info_updater.start();
 554   if (ret < 0) {
 555     ldout(store->ctx(), 0) << __func__ << ": failed to update bucket info ret=" << ret << dendl;
 556     return ret;
 557   }
 558
 559   int num_target_shards = (new_bucket_info.num_shards > 0 ? new_bucket_info.num_shards : 1);
 560
 561   BucketReshardManager target_shards_mgr(store, new_bucket_info, num_target_shards);
 562
 563   bool verbose_json_out = verbose && (formatter != nullptr) && (out != nullptr);
 564
 565   if (verbose_json_out) {
 566     formatter->open_array_section("entries");
 567   }
 568
 569   uint64_t total_entries = 0;
 570
 571   if (!verbose_json_out && out) {
 572     (*out) << "total entries:";
 573   }
 574
 575   const int num_source_shards =
 576     (bucket_info.num_shards > 0 ? bucket_info.num_shards : 1);
 577   string marker;
 578   for (int i = 0; i < num_source_shards; ++i) {
 579     bool is_truncated = true;
 580     marker.clear();
 581     while (is_truncated) {
 582       entries.clear();
 583       ret = store->getRados()->bi_list(bucket, i, string(), marker, max_entries, &entries, &is_truncated);
 584       if (ret < 0 && ret != -ENOENT) {
 585         derr << "ERROR: bi_list(): " << cpp_strerror(-ret) << dendl;
 586         return ret;
 587       }
 588
 589       for (auto iter = entries.begin(); iter != entries.end(); ++iter) {
 590         rgw_cls_bi_entry& entry = *iter;
 591         if (verbose_json_out) {
 592           formatter->open_object_section("entry");
 593
 594           encode_json("shard_id", i, formatter);
 595           encode_json("num_entry", total_entries, formatter);
 596           encode_json("entry", entry, formatter);
 597         }
 598         total_entries++;
 599
 600         marker = entry.idx;
 601
 602         int target_shard_id;
 603         cls_rgw_obj_key cls_key;
 604         RGWObjCategory category;
 605         rgw_bucket_category_stats stats;
 606         bool account = entry.get_info(&cls_key, &category, &stats);
 607         rgw_obj_key key(cls_key);
 608         rgw_obj obj(new_bucket_info.bucket, key);
 609         RGWMPObj mp;
 610         if (key.ns == RGW_OBJ_NS_MULTIPART && mp.from_meta(key.name)) {
 611           // place the multipart .meta object on the same shard as its head object
 612           obj.index_hash_source = mp.get_key();
 613         }
 614         int ret = store->getRados()->get_target_shard_id(new_bucket_info, obj.get_hash_object(), &target_shard_id);
 615         if (ret < 0) {
 616           lderr(store->ctx()) << "ERROR: get_target_shard_id() returned ret=" << ret << dendl;
 617           return ret;
 618         }
 619
 620         int shard_index = (target_shard_id > 0 ? target_shard_id : 0);
 621
 622         ret = target_shards_mgr.add_entry(shard_index, entry, account,
 623                                           category, stats);
 624         if (ret < 0) {
 625           return ret;
 626         }
 627
 628         Clock::time_point now = Clock::now();
 629         if (reshard_lock.should_renew(now)) {
 630           // assume outer locks have timespans at least the size of ours, so
 631           // can call inside conditional
 632           if (outer_reshard_lock) {
 633             ret = outer_reshard_lock->renew(now);
 634             if (ret < 0) {
 635               return ret;
 636             }
 637           }
 638           ret = reshard_lock.renew(now);
 639           if (ret < 0) {
 640             lderr(store->ctx()) << "Error renewing bucket lock: " << ret << dendl;
 641             return ret;
 642           }
 643         }
 644
 645         if (verbose_json_out) {
 646           formatter->close_section();
 647           formatter->flush(*out);
 648         } else if (out && !(total_entries % 1000)) {
 649           (*out) << " " << total_entries;
 650         }
 651       } // entries loop
 652     }
 653   }
 654
 655   if (verbose_json_out) {
 656     formatter->close_section();
 657     formatter->flush(*out);
 658   } else if (out) {
 659     (*out) << " " << total_entries << std::endl;
 660   }
 661
 662   ret = target_shards_mgr.finish();
 663   if (ret < 0) {
 664     lderr(store->ctx()) << "ERROR: failed to reshard" << dendl;
 665     return -EIO;
 666   }
 667
 668   ret = store->ctl()->bucket->link_bucket(new_bucket_info.owner, new_bucket_info.bucket, bucket_info.creation_time, null_yield);
 669   if (ret < 0) {
 670     lderr(store->ctx()) << "failed to link new bucket instance (bucket_id=" << new_bucket_info.bucket.bucket_id << ": " << cpp_strerror(-ret) << ")" << dendl;
 671     return ret;
 672   }
 673
 674   ret = bucket_info_updater.complete();
 675   if (ret < 0) {
 676     ldout(store->ctx(), 0) << __func__ << ": failed to update bucket info ret=" << ret << dendl;
 677     /* don't error out, reshard process succeeded */
 678   }
 679
 680   return 0;
 681   // NB: some error clean-up is done by ~BucketInfoReshardUpdate
 682 } // RGWBucketReshard::do_reshard
 683
 684 int RGWBucketReshard::get_status(list<cls_rgw_bucket_instance_entry> *status)
 685 {
 686   return store->svc()->bi_rados->get_reshard_status(bucket_info, status);
 687 }
 688
 689
 690 int RGWBucketReshard::execute(int num_shards, int max_op_entries,
 691                               bool verbose, ostream *out, Formatter *formatter,
 692                               RGWReshard* reshard_log)
 693 {
 694   int ret = reshard_lock.lock();
 695   if (ret < 0) {
 696     return ret;
 697   }
 698
 699   RGWBucketInfo new_bucket_info;
 700   ret = create_new_bucket_instance(num_shards, new_bucket_info);
 701   if (ret < 0) {
 702     // shard state is uncertain, but this will attempt to remove them anyway
 703     goto error_out;
 704   }
 705
 706   if (reshard_log) {
 707     ret = reshard_log->update(bucket_info, new_bucket_info);
 708     if (ret < 0) {
 709       goto error_out;
 710     }
 711   }
 712
 713   // set resharding status of current bucket_info & shards with
 714   // information about planned resharding
 715   ret = set_resharding_status(new_bucket_info.bucket.bucket_id,
 716                               num_shards, cls_rgw_reshard_status::IN_PROGRESS);
 717   if (ret < 0) {
 718     goto error_out;
 719   }
 720
 721   ret = do_reshard(num_shards,
 722                    new_bucket_info,
 723                    max_op_entries,
 724                    verbose, out, formatter);
 725   if (ret < 0) {
 726     goto error_out;
 727   }
 728
 729   // at this point we've done the main work; we'll make a best-effort
 730   // to clean-up but will not indicate any errors encountered
 731
 732   reshard_lock.unlock();
 733
 734   // resharding successful, so remove old bucket index shards; use
 735   // best effort and don't report out an error; the lock isn't needed
 736   // at this point since all we're using a best effor to to remove old
 737   // shard objects
 738   ret = store->svc()->bi->clean_index(bucket_info);
 739   if (ret < 0) {
 740     lderr(store->ctx()) << "Error: " << __func__ <<
 741       " failed to clean up old shards; " <<
 742       "RGWRados::clean_bucket_index returned " << ret << dendl;
 743   }
 744
 745   ret = store->ctl()->bucket->remove_bucket_instance_info(bucket_info.bucket,
 746                                                        bucket_info, null_yield);
 747   if (ret < 0) {
 748     lderr(store->ctx()) << "Error: " << __func__ <<
 749       " failed to clean old bucket info object \"" <<
 750       bucket_info.bucket.get_key() <<
 751       "\"created after successful resharding with error " << ret << dendl;
 752   }
 753
 754   ldout(store->ctx(), 1) << __func__ <<
 755     " INFO: reshard of bucket \"" << bucket_info.bucket.name << "\" from \"" <<
 756     bucket_info.bucket.get_key() << "\" to \"" <<
 757     new_bucket_info.bucket.get_key() << "\" completed successfully" << dendl;
 758
 759   return 0;
 760
 761 error_out:
 762
 763   reshard_lock.unlock();
 764
 765   // since the real problem is the issue that led to this error code
 766   // path, we won't touch ret and instead use another variable to
 767   // temporarily error codes
 768   int ret2 = store->svc()->bi->clean_index(new_bucket_info);
 769   if (ret2 < 0) {
 770     lderr(store->ctx()) << "Error: " << __func__ <<
 771       " failed to clean up shards from failed incomplete resharding; " <<
 772       "RGWRados::clean_bucket_index returned " << ret2 << dendl;
 773   }
 774
 775   ret2 = store->ctl()->bucket->remove_bucket_instance_info(new_bucket_info.bucket,
 776                                                         new_bucket_info,
 777                                                         null_yield);
 778   if (ret2 < 0) {
 779     lderr(store->ctx()) << "Error: " << __func__ <<
 780       " failed to clean bucket info object \"" <<
 781       new_bucket_info.bucket.get_key() <<
 782       "\"created during incomplete resharding with error " << ret2 << dendl;
 783   }
 784
 785   return ret;
 786 } // execute
 787
 788
 789 RGWReshard::RGWReshard(rgw::sal::RGWRadosStore* _store, bool _verbose, ostream *_out,
 790                        Formatter *_formatter) :
 791   store(_store), instance_lock(bucket_instance_lock_name),
 792   verbose(_verbose), out(_out), formatter(_formatter)
 793 {
 794   num_logshards = store->ctx()->_conf.get_val<uint64_t>("rgw_reshard_num_logs");
 795 }
 796
 797 string RGWReshard::get_logshard_key(const string& tenant,
 798                                     const string& bucket_name)
 799 {
 800   return tenant + ":" + bucket_name;
 801 }
 802
 803 #define MAX_RESHARD_LOGSHARDS_PRIME 7877
 804
 805 void RGWReshard::get_bucket_logshard_oid(const string& tenant, const string& bucket_name, string *oid)
 806 {
 807   string key = get_logshard_key(tenant, bucket_name);
 808
 809   uint32_t sid = ceph_str_hash_linux(key.c_str(), key.size());
 810   uint32_t sid2 = sid ^ ((sid & 0xFF) << 24);
 811   sid = sid2 % MAX_RESHARD_LOGSHARDS_PRIME % num_logshards;
 812
 813   get_logshard_oid(int(sid), oid);
 814 }
 815
 816 int RGWReshard::add(cls_rgw_reshard_entry& entry)
 817 {
 818   if (!store->svc()->zone->can_reshard()) {
 819     ldout(store->ctx(), 20) << __func__ << " Resharding is disabled"  << dendl;
 820     return 0;
 821   }
 822
 823   string logshard_oid;
 824
 825   get_bucket_logshard_oid(entry.tenant, entry.bucket_name, &logshard_oid);
 826
 827   librados::ObjectWriteOperation op;
 828   cls_rgw_reshard_add(op, entry);
 829
 830   int ret = rgw_rados_operate(store->getRados()->reshard_pool_ctx, logshard_oid, &op, null_yield);
 831   if (ret < 0) {
 832     lderr(store->ctx()) << "ERROR: failed to add entry to reshard log, oid=" << logshard_oid << " tenant=" << entry.tenant << " bucket=" << entry.bucket_name << dendl;
 833     return ret;
 834   }
 835   return 0;
 836 }
 837
 838 int RGWReshard::update(const RGWBucketInfo& bucket_info, const RGWBucketInfo& new_bucket_info)
 839 {
 840   cls_rgw_reshard_entry entry;
 841   entry.bucket_name = bucket_info.bucket.name;
 842   entry.bucket_id = bucket_info.bucket.bucket_id;
 843   entry.tenant = bucket_info.owner.tenant;
 844
 845   int ret = get(entry);
 846   if (ret < 0) {
 847     return ret;
 848   }
 849
 850   entry.new_instance_id = new_bucket_info.bucket.name + ":"  + new_bucket_info.bucket.bucket_id;
 851
 852   ret = add(entry);
 853   if (ret < 0) {
 854     ldout(store->ctx(), 0) << __func__ << ":Error in updating entry bucket " << entry.bucket_name << ": " <<
 855       cpp_strerror(-ret) << dendl;
 856   }
 857
 858   return ret;
 859 }
 860
 861
 862 int RGWReshard::list(int logshard_num, string& marker, uint32_t max, std::list<cls_rgw_reshard_entry>& entries, bool *is_truncated)
 863 {
 864   string logshard_oid;
 865
 866   get_logshard_oid(logshard_num, &logshard_oid);
 867
 868   int ret = cls_rgw_reshard_list(store->getRados()->reshard_pool_ctx, logshard_oid, marker, max, entries, is_truncated);
 869
 870   if (ret < 0) {
 871     if (ret == -ENOENT) {
 872       *is_truncated = false;
 873       ret = 0;
 874     } else {
 875       lderr(store->ctx()) << "ERROR: failed to list reshard log entries, oid=" << logshard_oid << dendl;
 876       if (ret == -EACCES) {
 877         lderr(store->ctx()) << "access denied to pool " << store->svc()->zone->get_zone_params().reshard_pool
 878                           << ". Fix the pool access permissions of your client" << dendl;
 879       }
 880     }
 881   }
 882
 883   return ret;
 884 }
 885
 886 int RGWReshard::get(cls_rgw_reshard_entry& entry)
 887 {
 888   string logshard_oid;
 889
 890   get_bucket_logshard_oid(entry.tenant, entry.bucket_name, &logshard_oid);
 891
 892   int ret = cls_rgw_reshard_get(store->getRados()->reshard_pool_ctx, logshard_oid, entry);
 893   if (ret < 0) {
 894     if (ret != -ENOENT) {
 895       lderr(store->ctx()) << "ERROR: failed to get entry from reshard log, oid=" << logshard_oid << " tenant=" << entry.tenant <<
 896         " bucket=" << entry.bucket_name << dendl;
 897     }
 898     return ret;
 899   }
 900
 901   return 0;
 902 }
 903
 904 int RGWReshard::remove(cls_rgw_reshard_entry& entry)
 905 {
 906   string logshard_oid;
 907
 908   get_bucket_logshard_oid(entry.tenant, entry.bucket_name, &logshard_oid);
 909
 910   librados::ObjectWriteOperation op;
 911   cls_rgw_reshard_remove(op, entry);
 912
 913   int ret = rgw_rados_operate(store->getRados()->reshard_pool_ctx, logshard_oid, &op, null_yield);
 914   if (ret < 0) {
 915     lderr(store->ctx()) << "ERROR: failed to remove entry from reshard log, oid=" << logshard_oid << " tenant=" << entry.tenant << " bucket=" << entry.bucket_name << dendl;
 916     return ret;
 917   }
 918
 919   return ret;
 920 }
 921
 922 int RGWReshard::clear_bucket_resharding(const string& bucket_instance_oid, cls_rgw_reshard_entry& entry)
 923 {
 924   int ret = cls_rgw_clear_bucket_resharding(store->getRados()->reshard_pool_ctx, bucket_instance_oid);
 925   if (ret < 0) {
 926     lderr(store->ctx()) << "ERROR: failed to clear bucket resharding, bucket_instance_oid=" << bucket_instance_oid << dendl;
 927     return ret;
 928   }
 929
 930   return 0;
 931 }
 932
 933 int RGWReshardWait::wait(optional_yield y)
 934 {
 935   std::unique_lock lock(mutex);
 936
 937   if (going_down) {
 938     return -ECANCELED;
 939   }
 940
 941 #ifdef HAVE_BOOST_CONTEXT
 942   if (y) {
 943     auto& context = y.get_io_context();
 944     auto& yield = y.get_yield_context();
 945
 946     Waiter waiter(context);
 947     waiters.push_back(waiter);
 948     lock.unlock();
 949
 950     waiter.timer.expires_after(duration);
 951
 952     boost::system::error_code ec;
 953     waiter.timer.async_wait(yield[ec]);
 954
 955     lock.lock();
 956     waiters.erase(waiters.iterator_to(waiter));
 957     return -ec.value();
 958   }
 959 #endif
 960
 961   cond.wait_for(lock, duration);
 962
 963   if (going_down) {
 964     return -ECANCELED;
 965   }
 966
 967   return 0;
 968 }
 969
 970 void RGWReshardWait::stop()
 971 {
 972   std::scoped_lock lock(mutex);
 973   going_down = true;
 974   cond.notify_all();
 975   for (auto& waiter : waiters) {
 976     // unblock any waiters with ECANCELED
 977     waiter.timer.cancel();
 978   }
 979 }
 980
 981 int RGWReshard::process_single_logshard(int logshard_num)
 982 {
 983   string marker;
 984   bool truncated = true;
 985
 986   CephContext *cct = store->ctx();
 987   constexpr uint32_t max_entries = 1000;
 988
 989   string logshard_oid;
 990   get_logshard_oid(logshard_num, &logshard_oid);
 991
 992   RGWBucketReshardLock logshard_lock(store, logshard_oid, false);
 993
 994   int ret = logshard_lock.lock();
 995   if (ret < 0) {
 996     ldout(store->ctx(), 5) << __func__ << "(): failed to acquire lock on " <<
 997       logshard_oid << ", ret = " << ret <<dendl;
 998     return ret;
 999   }
1000
1001   do {
1002     std::list<cls_rgw_reshard_entry> entries;
1003     ret = list(logshard_num, marker, max_entries, entries, &truncated);
1004     if (ret < 0) {
1005       ldout(cct, 10) << "cannot list all reshards in logshard oid=" <<
1006         logshard_oid << dendl;
1007       continue;
1008     }
1009
1010     for(auto& entry: entries) { // logshard entries
1011       if(entry.new_instance_id.empty()) {
1012
1013         ldout(store->ctx(), 20) << __func__ << " resharding " <<
1014           entry.bucket_name  << dendl;
1015
1016         rgw_bucket bucket;
1017         RGWBucketInfo bucket_info;
1018         map<string, bufferlist> attrs;
1019
1020         ret = store->getRados()->get_bucket_info(store->svc(),
1021                                                  entry.tenant, entry.bucket_name,
1022                                                  bucket_info, nullptr,
1023                                                  null_yield, &attrs);
1024         if (ret < 0) {
1025           ldout(cct, 0) <<  __func__ <<
1026             ": Error in get_bucket_info for bucket " << entry.bucket_name <<
1027             ": " << cpp_strerror(-ret) << dendl;
1028           if (ret != -ENOENT) {
1029             // any error other than ENOENT will abort
1030             return ret;
1031           }
1032
1033           // we've encountered a reshard queue entry for an apparently
1034           // non-existent bucket; let's try to recover by cleaning up
1035           ldout(cct, 0) <<  __func__ <<
1036             ": removing reshard queue entry for non-existent bucket " <<
1037             entry.bucket_name << dendl;
1038
1039           ret = remove(entry);
1040           if (ret < 0) {
1041             ldout(cct, 0) << __func__ <<
1042               ": Error removing non-existent bucket " <<
1043               entry.bucket_name << " from resharding queue: " <<
1044               cpp_strerror(-ret) << dendl;
1045             return ret;
1046           }
1047
1048           // we cleaned up, move on to the next entry
1049           goto finished_entry;
1050         }
1051
1052         RGWBucketReshard br(store, bucket_info, attrs, nullptr);
1053         ret = br.execute(entry.new_num_shards, max_entries, false, nullptr,
1054                          nullptr, this);
1055         if (ret < 0) {
1056           ldout(store->ctx(), 0) <<  __func__ <<
1057             ": Error during resharding bucket " << entry.bucket_name << ":" <<
1058             cpp_strerror(-ret)<< dendl;
1059           return ret;
1060         }
1061
1062         ldout(store->ctx(), 20) << __func__ <<
1063           " removing reshard queue entry for bucket " << entry.bucket_name <<
1064           dendl;
1065
1066         ret = remove(entry);
1067         if (ret < 0) {
1068           ldout(cct, 0) << __func__ << ": Error removing bucket " <<
1069             entry.bucket_name << " from resharding queue: " <<
1070             cpp_strerror(-ret) << dendl;
1071           return ret;
1072         }
1073       } // if new instance id is empty
1074
1075     finished_entry:
1076
1077       Clock::time_point now = Clock::now();
1078       if (logshard_lock.should_renew(now)) {
1079         ret = logshard_lock.renew(now);
1080         if (ret < 0) {
1081           return ret;
1082         }
1083       }
1084
1085       entry.get_key(&marker);
1086     } // entry for loop
1087   } while (truncated);
1088
1089   logshard_lock.unlock();
1090   return 0;
1091 }
1092
1093
1094 void  RGWReshard::get_logshard_oid(int shard_num, string *logshard)
1095 {
1096   char buf[32];
1097   snprintf(buf, sizeof(buf), "%010u", (unsigned)shard_num);
1098
1099   string objname(reshard_oid_prefix);
1100   *logshard =  objname + buf;
1101 }
1102
1103 int RGWReshard::process_all_logshards()
1104 {
1105   if (!store->svc()->zone->can_reshard()) {
1106     ldout(store->ctx(), 20) << __func__ << " Resharding is disabled"  << dendl;
1107     return 0;
1108   }
1109   int ret = 0;
1110
1111   for (int i = 0; i < num_logshards; i++) {
1112     string logshard;
1113     get_logshard_oid(i, &logshard);
1114
1115     ldout(store->ctx(), 20) << "processing logshard = " << logshard << dendl;
1116
1117     ret = process_single_logshard(i);
1118
1119     ldout(store->ctx(), 20) << "finish processing logshard = " << logshard << " , ret = " << ret << dendl;
1120   }
1121
1122   return 0;
1123 }
1124
1125 bool RGWReshard::going_down()
1126 {
1127   return down_flag;
1128 }
1129
1130 void RGWReshard::start_processor()
1131 {
1132   worker = new ReshardWorker(store->ctx(), this);
1133   worker->create("rgw_reshard");
1134 }
1135
1136 void RGWReshard::stop_processor()
1137 {
1138   down_flag = true;
1139   if (worker) {
1140     worker->stop();
1141     worker->join();
1142   }
1143   delete worker;
1144   worker = nullptr;
1145 }
1146
1147 void *RGWReshard::ReshardWorker::entry() {
1148   do {
1149     utime_t start = ceph_clock_now();
1150     reshard->process_all_logshards();
1151
1152     if (reshard->going_down())
1153       break;
1154
1155     utime_t end = ceph_clock_now();
1156     end -= start;
1157     int secs = cct->_conf.get_val<uint64_t>("rgw_reshard_thread_interval");
1158
1159     if (secs <= end.sec())
1160       continue; // next round
1161
1162     secs -= end.sec();
1163
1164     std::unique_lock locker{lock};
1165     cond.wait_for(locker, std::chrono::seconds(secs));
1166   } while (!reshard->going_down());
1167
1168   return NULL;
1169 }
1170
1171 void RGWReshard::ReshardWorker::stop()
1172 {
1173   std::lock_guard l{lock};
1174   cond.notify_all();
1175 }