ceph/src/rgw/rgw_orphan.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 <string>
   5
   6
   7 #include "common/config.h"
   8 #include "common/Formatter.h"
   9 #include "common/errno.h"
  10
  11 #include "rgw_op.h"
  12 #include "rgw_multi.h"
  13 #include "rgw_orphan.h"
  14 #include "rgw_zone.h"
  15 #include "rgw_bucket.h"
  16 #include "rgw_sal_rados.h"
  17
  18 #include "services/svc_zone.h"
  19 #include "services/svc_sys_obj.h"
  20
  21 #define dout_subsys ceph_subsys_rgw
  22
  23 #define DEFAULT_NUM_SHARDS 64
  24
  25 static string obj_fingerprint(const string& oid, const char *force_ns = NULL)
  26 {
  27   ssize_t pos = oid.find('_');
  28   if (pos < 0) {
  29     cerr << "ERROR: object does not have a bucket marker: " << oid << std::endl;
  30   }
  31
  32   string obj_marker = oid.substr(0, pos);
  33
  34   rgw_obj_key key;
  35
  36   rgw_obj_key::parse_raw_oid(oid.substr(pos + 1), &key);
  37
  38   if (key.ns.empty()) {
  39     return oid;
  40   }
  41
  42   string s = oid;
  43
  44   if (force_ns) {
  45     rgw_bucket b;
  46     rgw_obj new_obj(b, key);
  47     s = obj_marker + "_" + new_obj.get_oid();
  48   }
  49
  50   /* cut out suffix */
  51   size_t i = s.size() - 1;
  52   for (; i >= s.size() - 10; --i) {
  53     char c = s[i];
  54     if (!isdigit(c) && c != '.' && c != '_') {
  55       break;
  56     }
  57   }
  58
  59   return s.substr(0, i + 1);
  60 }
  61
  62 int RGWOrphanStore::read_job(const string& job_name, RGWOrphanSearchState & state)
  63 {
  64   set<string> keys;
  65   map<string, bufferlist> vals;
  66   keys.insert(job_name);
  67   int r = ioctx.omap_get_vals_by_keys(oid, keys, &vals);
  68   if (r < 0) {
  69     return r;
  70   }
  71
  72   map<string, bufferlist>::iterator iter = vals.find(job_name);
  73   if (iter == vals.end()) {
  74     return -ENOENT;
  75   }
  76
  77   try {
  78     bufferlist& bl = iter->second;
  79     decode(state, bl);
  80   } catch (buffer::error& err) {
  81     lderr(store->ctx()) << "ERROR: could not decode buffer" << dendl;
  82     return -EIO;
  83   }
  84
  85   return 0;
  86 }
  87
  88 int RGWOrphanStore::write_job(const string& job_name, const RGWOrphanSearchState& state)
  89 {
  90   map<string, bufferlist> vals;
  91   bufferlist bl;
  92   encode(state, bl);
  93   vals[job_name] = bl;
  94   int r = ioctx.omap_set(oid, vals);
  95   if (r < 0) {
  96     return r;
  97   }
  98
  99   return 0;
 100 }
 101
 102 int RGWOrphanStore::remove_job(const string& job_name)
 103 {
 104   set<string> keys;
 105   keys.insert(job_name);
 106
 107   int r = ioctx.omap_rm_keys(oid, keys);
 108   if (r < 0) {
 109     return r;
 110   }
 111
 112   return 0;
 113 }
 114
 115 int RGWOrphanStore::list_jobs(map <string,RGWOrphanSearchState>& job_list)
 116 {
 117   map <string,bufferlist> vals;
 118   int MAX_READ=1024;
 119   string marker="";
 120   int r = 0;
 121
 122   // loop through all the omap vals from index object, storing them to job_list,
 123   // read in batches of 1024, we update the marker every iteration and exit the
 124   // loop when we find that total size read out is less than batch size
 125   do {
 126     r = ioctx.omap_get_vals(oid, marker, MAX_READ, &vals);
 127     if (r < 0) {
 128       return r;
 129     }
 130     r = vals.size();
 131
 132     for (const auto &it : vals) {
 133       marker=it.first;
 134       RGWOrphanSearchState state;
 135       try {
 136         bufferlist bl = it.second;
 137         decode(state, bl);
 138       } catch (buffer::error& err) {
 139         lderr(store->ctx()) << "ERROR: could not decode buffer" << dendl;
 140         return -EIO;
 141       }
 142       job_list[it.first] = state;
 143     }
 144   } while (r == MAX_READ);
 145
 146   return 0;
 147 }
 148
 149 int RGWOrphanStore::init(const DoutPrefixProvider *dpp)
 150 {
 151   const rgw_pool& log_pool = store->svc()->zone->get_zone_params().log_pool;
 152   int r = rgw_init_ioctx(dpp, store->getRados()->get_rados_handle(), log_pool, ioctx);
 153   if (r < 0) {
 154     cerr << "ERROR: failed to open log pool (" << log_pool << " ret=" << r << std::endl;
 155     return r;
 156   }
 157
 158   return 0;
 159 }
 160
 161 int RGWOrphanStore::store_entries(const DoutPrefixProvider *dpp, const string& oid, const map<string, bufferlist>& entries)
 162 {
 163   librados::ObjectWriteOperation op;
 164   op.omap_set(entries);
 165   cout << "storing " << entries.size() << " entries at " << oid << std::endl;
 166   ldpp_dout(dpp, 20) << "storing " << entries.size() << " entries at " << oid << ": " << dendl;
 167   for (map<string, bufferlist>::const_iterator iter = entries.begin(); iter != entries.end(); ++iter) {
 168     ldpp_dout(dpp, 20) << " > " << iter->first << dendl;
 169   }
 170   int ret = rgw_rados_operate(dpp, ioctx, oid, &op, null_yield);
 171   if (ret < 0) {
 172     ldpp_dout(dpp, -1) << "ERROR: " << __func__ << "(" << oid << ") returned ret=" << ret << dendl;
 173   }
 174
 175   return 0;
 176 }
 177
 178 int RGWOrphanStore::read_entries(const string& oid, const string& marker, map<string, bufferlist> *entries, bool *truncated)
 179 {
 180 #define MAX_OMAP_GET 100
 181   int ret = ioctx.omap_get_vals(oid, marker, MAX_OMAP_GET, entries);
 182   if (ret < 0 && ret != -ENOENT) {
 183     cerr << "ERROR: " << __func__ << "(" << oid << ") returned ret=" << cpp_strerror(-ret) << std::endl;
 184   }
 185
 186   *truncated = (entries->size() == MAX_OMAP_GET);
 187
 188   return 0;
 189 }
 190
 191 int RGWOrphanSearch::init(const DoutPrefixProvider *dpp, const string& job_name, RGWOrphanSearchInfo *info, bool _detailed_mode)
 192 {
 193   int r = orphan_store.init(dpp);
 194   if (r < 0) {
 195     return r;
 196   }
 197
 198   constexpr int64_t MAX_LIST_OBJS_ENTRIES=100;
 199
 200   max_list_bucket_entries = std::max(store->ctx()->_conf->rgw_list_bucket_min_readahead,
 201                                      MAX_LIST_OBJS_ENTRIES);
 202
 203   detailed_mode = _detailed_mode;
 204   RGWOrphanSearchState state;
 205   r = orphan_store.read_job(job_name, state);
 206   if (r < 0 && r != -ENOENT) {
 207     ldpp_dout(dpp, -1) << "ERROR: failed to read state ret=" << r << dendl;
 208     return r;
 209   }
 210
 211   if (r == 0) {
 212     search_info = state.info;
 213     search_stage = state.stage;
 214   } else if (info) { /* r == -ENOENT, initiate a new job if info was provided */
 215     search_info = *info;
 216     search_info.job_name = job_name;
 217     search_info.num_shards = (info->num_shards ? info->num_shards : DEFAULT_NUM_SHARDS);
 218     search_info.start_time = ceph_clock_now();
 219     search_stage = RGWOrphanSearchStage(ORPHAN_SEARCH_STAGE_INIT);
 220
 221     r = save_state();
 222     if (r < 0) {
 223       ldpp_dout(dpp, -1) << "ERROR: failed to write state ret=" << r << dendl;
 224       return r;
 225     }
 226   } else {
 227       ldpp_dout(dpp, -1) << "ERROR: job not found" << dendl;
 228       return r;
 229   }
 230
 231   index_objs_prefix = RGW_ORPHAN_INDEX_PREFIX + string(".");
 232   index_objs_prefix += job_name;
 233
 234   for (int i = 0; i < search_info.num_shards; i++) {
 235     char buf[128];
 236
 237     snprintf(buf, sizeof(buf), "%s.rados.%d", index_objs_prefix.c_str(), i);
 238     all_objs_index[i] = buf;
 239
 240     snprintf(buf, sizeof(buf), "%s.buckets.%d", index_objs_prefix.c_str(), i);
 241     buckets_instance_index[i] = buf;
 242
 243     snprintf(buf, sizeof(buf), "%s.linked.%d", index_objs_prefix.c_str(), i);
 244     linked_objs_index[i] = buf;
 245   }
 246   return 0;
 247 }
 248
 249 int RGWOrphanSearch::log_oids(const DoutPrefixProvider *dpp, map<int, string>& log_shards, map<int, list<string> >& oids)
 250 {
 251   map<int, list<string> >::iterator miter = oids.begin();
 252
 253   list<log_iter_info> liters; /* a list of iterator pairs for begin and end */
 254
 255   for (; miter != oids.end(); ++miter) {
 256     log_iter_info info;
 257     info.oid = log_shards[miter->first];
 258     info.cur = miter->second.begin();
 259     info.end = miter->second.end();
 260     liters.push_back(info);
 261   }
 262
 263   list<log_iter_info>::iterator list_iter;
 264   while (!liters.empty()) {
 265      list_iter = liters.begin();
 266
 267      while (list_iter != liters.end()) {
 268        log_iter_info& cur_info = *list_iter;
 269
 270        list<string>::iterator& cur = cur_info.cur;
 271        list<string>::iterator& end = cur_info.end;
 272
 273        map<string, bufferlist> entries;
 274 #define MAX_OMAP_SET_ENTRIES 100
 275        for (int j = 0; cur != end && j != MAX_OMAP_SET_ENTRIES; ++cur, ++j) {
 276          ldpp_dout(dpp, 20) << "adding obj: " << *cur << dendl;
 277          entries[*cur] = bufferlist();
 278        }
 279
 280        int ret = orphan_store.store_entries(dpp, cur_info.oid, entries);
 281        if (ret < 0) {
 282          return ret;
 283        }
 284        list<log_iter_info>::iterator tmp = list_iter;
 285        ++list_iter;
 286        if (cur == end) {
 287          liters.erase(tmp);
 288        }
 289      }
 290   }
 291   return 0;
 292 }
 293
 294 int RGWOrphanSearch::build_all_oids_index(const DoutPrefixProvider *dpp)
 295 {
 296   librados::IoCtx ioctx;
 297
 298   int ret = rgw_init_ioctx(dpp, store->getRados()->get_rados_handle(), search_info.pool, ioctx);
 299   if (ret < 0) {
 300     ldpp_dout(dpp, -1) << __func__ << ": rgw_init_ioctx() returned ret=" << ret << dendl;
 301     return ret;
 302   }
 303
 304   ioctx.set_namespace(librados::all_nspaces);
 305   librados::NObjectIterator i = ioctx.nobjects_begin();
 306   librados::NObjectIterator i_end = ioctx.nobjects_end();
 307
 308   map<int, list<string> > oids;
 309
 310   int count = 0;
 311   uint64_t total = 0;
 312
 313   cout << "logging all objects in the pool" << std::endl;
 314
 315   for (; i != i_end; ++i) {
 316     string nspace = i->get_nspace();
 317     string oid = i->get_oid();
 318     string locator = i->get_locator();
 319
 320     ssize_t pos = oid.find('_');
 321     if (pos < 0) {
 322       cout << "unidentified oid: " << oid << ", skipping" << std::endl;
 323       /* what is this object, oids should be in the format of <bucket marker>_<obj>,
 324        * skip this entry
 325        */
 326       continue;
 327     }
 328     string stripped_oid = oid.substr(pos + 1);
 329     rgw_obj_key key;
 330     if (!rgw_obj_key::parse_raw_oid(stripped_oid, &key)) {
 331       cout << "cannot parse oid: " << oid << ", skipping" << std::endl;
 332       continue;
 333     }
 334
 335     if (key.ns.empty()) {
 336       /* skipping head objects, we don't want to remove these as they are mutable and
 337        * cleaning them up is racy (can race with object removal and a later recreation)
 338        */
 339       cout << "skipping head object: oid=" << oid << std::endl;
 340       continue;
 341     }
 342
 343     string oid_fp = obj_fingerprint(oid);
 344
 345     ldout(store->ctx(), 20) << "oid_fp=" << oid_fp << dendl;
 346
 347     int shard = orphan_shard(oid_fp);
 348     oids[shard].push_back(oid);
 349
 350 #define COUNT_BEFORE_FLUSH 1000
 351     ++total;
 352     if (++count >= COUNT_BEFORE_FLUSH) {
 353       ldout(store->ctx(), 1) << "iterated through " << total << " objects" << dendl;
 354       ret = log_oids(dpp, all_objs_index, oids);
 355       if (ret < 0) {
 356         cerr << __func__ << ": ERROR: log_oids() returned ret=" << ret << std::endl;
 357         return ret;
 358       }
 359       count = 0;
 360       oids.clear();
 361     }
 362   }
 363   ret = log_oids(dpp, all_objs_index, oids);
 364   if (ret < 0) {
 365     cerr << __func__ << ": ERROR: log_oids() returned ret=" << ret << std::endl;
 366     return ret;
 367   }
 368
 369   return 0;
 370 }
 371
 372 int RGWOrphanSearch::build_buckets_instance_index(const DoutPrefixProvider *dpp)
 373 {
 374   void *handle;
 375   int max = 1000;
 376   string section = "bucket.instance";
 377   int ret = store->ctl()->meta.mgr->list_keys_init(dpp, section, &handle);
 378   if (ret < 0) {
 379     ldpp_dout(dpp, -1) << "ERROR: can't get key: " << cpp_strerror(-ret) << dendl;
 380     return ret;
 381   }
 382
 383   map<int, list<string> > instances;
 384
 385   bool truncated;
 386
 387   RGWObjectCtx obj_ctx(store);
 388
 389   int count = 0;
 390   uint64_t total = 0;
 391
 392   do {
 393     list<string> keys;
 394     ret = store->ctl()->meta.mgr->list_keys_next(handle, max, keys, &truncated);
 395     if (ret < 0) {
 396       ldpp_dout(dpp, -1) << "ERROR: lists_keys_next(): " << cpp_strerror(-ret) << dendl;
 397       return ret;
 398     }
 399
 400     for (list<string>::iterator iter = keys.begin(); iter != keys.end(); ++iter) {
 401       ++total;
 402       ldpp_dout(dpp, 10) << "bucket_instance=" << *iter << " total=" << total << dendl;
 403       int shard = orphan_shard(*iter);
 404       instances[shard].push_back(*iter);
 405
 406       if (++count >= COUNT_BEFORE_FLUSH) {
 407         ret = log_oids(dpp, buckets_instance_index, instances);
 408         if (ret < 0) {
 409           ldpp_dout(dpp, -1) << __func__ << ": ERROR: log_oids() returned ret=" << ret << dendl;
 410           return ret;
 411         }
 412         count = 0;
 413         instances.clear();
 414       }
 415     }
 416
 417   } while (truncated);
 418
 419   ret = log_oids(dpp, buckets_instance_index, instances);
 420   if (ret < 0) {
 421     ldpp_dout(dpp, -1) << __func__ << ": ERROR: log_oids() returned ret=" << ret << dendl;
 422     return ret;
 423   }
 424   store->ctl()->meta.mgr->list_keys_complete(handle);
 425
 426   return 0;
 427 }
 428
 429 int RGWOrphanSearch::handle_stat_result(const DoutPrefixProvider *dpp, map<int, list<string> >& oids, RGWRados::Object::Stat::Result& result)
 430 {
 431   set<string> obj_oids;
 432   rgw_bucket& bucket = result.obj.bucket;
 433   if (!result.manifest) { /* a very very old object, or part of a multipart upload during upload */
 434     const string loc = bucket.bucket_id + "_" + result.obj.get_oid();
 435     obj_oids.insert(obj_fingerprint(loc));
 436
 437     /*
 438      * multipart parts don't have manifest on them, it's in the meta object. Instead of reading the
 439      * meta object, just add a "shadow" object to the mix
 440      */
 441     obj_oids.insert(obj_fingerprint(loc, "shadow"));
 442   } else {
 443     RGWObjManifest& manifest = *result.manifest;
 444
 445     if (!detailed_mode &&
 446         manifest.get_obj_size() <= manifest.get_head_size()) {
 447       ldpp_dout(dpp, 5) << "skipping object as it fits in a head" << dendl;
 448       return 0;
 449     }
 450
 451     RGWObjManifest::obj_iterator miter;
 452     for (miter = manifest.obj_begin(dpp); miter != manifest.obj_end(dpp); ++miter) {
 453       const rgw_raw_obj& loc = miter.get_location().get_raw_obj(store);
 454       string s = loc.oid;
 455       obj_oids.insert(obj_fingerprint(s));
 456     }
 457   }
 458
 459   for (set<string>::iterator iter = obj_oids.begin(); iter != obj_oids.end(); ++iter) {
 460     ldpp_dout(dpp, 20) << __func__ << ": oid for obj=" << result.obj << ": " << *iter << dendl;
 461
 462     int shard = orphan_shard(*iter);
 463     oids[shard].push_back(*iter);
 464   }
 465
 466   return 0;
 467 }
 468
 469 int RGWOrphanSearch::pop_and_handle_stat_op(const DoutPrefixProvider *dpp, map<int, list<string> >& oids, std::deque<RGWRados::Object::Stat>& ops)
 470 {
 471   RGWRados::Object::Stat& front_op = ops.front();
 472
 473   int ret = front_op.wait();
 474   if (ret < 0) {
 475     if (ret != -ENOENT) {
 476       ldpp_dout(dpp, -1) << "ERROR: stat_async() returned error: " << cpp_strerror(-ret) << dendl;
 477     }
 478     goto done;
 479   }
 480   ret = handle_stat_result(dpp, oids, front_op.result);
 481   if (ret < 0) {
 482     ldpp_dout(dpp, -1) << "ERROR: handle_stat_response() returned error: " << cpp_strerror(-ret) << dendl;
 483   }
 484 done:
 485   ops.pop_front();
 486   return ret;
 487 }
 488
 489 int RGWOrphanSearch::build_linked_oids_for_bucket(const DoutPrefixProvider *dpp, const string& bucket_instance_id, map<int, list<string> >& oids)
 490 {
 491   RGWObjectCtx obj_ctx(store);
 492   auto sysobj_ctx = store->svc()->sysobj->init_obj_ctx();
 493
 494   rgw_bucket orphan_bucket;
 495   int shard_id;
 496   int ret = rgw_bucket_parse_bucket_key(store->ctx(), bucket_instance_id,
 497                                         &orphan_bucket, &shard_id);
 498   if (ret < 0) {
 499     ldpp_dout(dpp, 0) << __func__ << " failed to parse bucket instance: "
 500                  << bucket_instance_id << " skipping" << dendl;
 501     return ret;
 502   }
 503
 504   RGWBucketInfo cur_bucket_info;
 505   ret = store->getRados()->get_bucket_info(store->svc(), orphan_bucket.tenant,
 506                                orphan_bucket.name, cur_bucket_info, nullptr, null_yield, dpp);
 507   if (ret < 0) {
 508     if (ret == -ENOENT) {
 509       /* probably raced with bucket removal */
 510       return 0;
 511     }
 512     ldpp_dout(dpp, -1) << __func__ << ": ERROR: RGWRados::get_bucket_instance_info() returned ret=" << ret << dendl;
 513     return ret;
 514   }
 515
 516   if (cur_bucket_info.bucket.bucket_id != orphan_bucket.bucket_id) {
 517     ldpp_dout(dpp, 0) << __func__ << ": Skipping stale bucket instance: "
 518                            << orphan_bucket.name << ": "
 519                            << orphan_bucket.bucket_id << dendl;
 520     return 0;
 521   }
 522
 523   if (cur_bucket_info.reshard_status == cls_rgw_reshard_status::IN_PROGRESS) {
 524     ldpp_dout(dpp, 0) << __func__ << ": reshard in progress. Skipping "
 525                            << orphan_bucket.name << ": "
 526                            << orphan_bucket.bucket_id << dendl;
 527     return 0;
 528   }
 529
 530   RGWBucketInfo bucket_info;
 531   ret = store->getRados()->get_bucket_instance_info(sysobj_ctx, bucket_instance_id, bucket_info, nullptr, nullptr, null_yield, dpp);
 532   if (ret < 0) {
 533     if (ret == -ENOENT) {
 534       /* probably raced with bucket removal */
 535       return 0;
 536     }
 537     ldpp_dout(dpp, -1) << __func__ << ": ERROR: RGWRados::get_bucket_instance_info() returned ret=" << ret << dendl;
 538     return ret;
 539   }
 540
 541   ldpp_dout(dpp, 10) << "building linked oids for bucket instance: " << bucket_instance_id << dendl;
 542   RGWRados::Bucket target(store->getRados(), bucket_info);
 543   RGWRados::Bucket::List list_op(&target);
 544
 545   string marker;
 546   list_op.params.marker = rgw_obj_key(marker);
 547   list_op.params.list_versions = true;
 548   list_op.params.enforce_ns = false;
 549
 550   bool truncated;
 551
 552   deque<RGWRados::Object::Stat> stat_ops;
 553
 554   do {
 555     vector<rgw_bucket_dir_entry> result;
 556
 557     ret = list_op.list_objects(dpp, max_list_bucket_entries,
 558                                &result, nullptr, &truncated, null_yield);
 559     if (ret < 0) {
 560       cerr << "ERROR: store->list_objects(): " << cpp_strerror(-ret) << std::endl;
 561       return ret;
 562     }
 563
 564     for (vector<rgw_bucket_dir_entry>::iterator iter = result.begin(); iter != result.end(); ++iter) {
 565       rgw_bucket_dir_entry& entry = *iter;
 566       if (entry.key.instance.empty()) {
 567         ldpp_dout(dpp, 20) << "obj entry: " << entry.key.name << dendl;
 568       } else {
 569         ldpp_dout(dpp, 20) << "obj entry: " << entry.key.name << " [" << entry.key.instance << "]" << dendl;
 570       }
 571
 572       ldpp_dout(dpp, 20) << __func__ << ": entry.key.name=" << entry.key.name << " entry.key.instance=" << entry.key.instance << dendl;
 573
 574       if (!detailed_mode &&
 575           entry.meta.accounted_size <= (uint64_t)store->ctx()->_conf->rgw_max_chunk_size) {
 576         ldpp_dout(dpp, 5) << __func__ << "skipping stat as the object " << entry.key.name
 577                               << "fits in a head" << dendl;
 578         continue;
 579       }
 580
 581       rgw_obj obj(bucket_info.bucket, entry.key);
 582
 583       RGWRados::Object op_target(store->getRados(), bucket_info, obj_ctx, obj);
 584
 585       stat_ops.push_back(RGWRados::Object::Stat(&op_target));
 586       RGWRados::Object::Stat& op = stat_ops.back();
 587
 588
 589       ret = op.stat_async(dpp);
 590       if (ret < 0) {
 591         ldpp_dout(dpp, -1) << "ERROR: stat_async() returned error: " << cpp_strerror(-ret) << dendl;
 592         return ret;
 593       }
 594       if (stat_ops.size() >= max_concurrent_ios) {
 595         ret = pop_and_handle_stat_op(dpp, oids, stat_ops);
 596         if (ret < 0) {
 597           if (ret != -ENOENT) {
 598             ldpp_dout(dpp, -1) << "ERROR: stat_async() returned error: " << cpp_strerror(-ret) << dendl;
 599           }
 600         }
 601       }
 602       if (oids.size() >= COUNT_BEFORE_FLUSH) {
 603         ret = log_oids(dpp, linked_objs_index, oids);
 604         if (ret < 0) {
 605           cerr << __func__ << ": ERROR: log_oids() returned ret=" << ret << std::endl;
 606           return ret;
 607         }
 608         oids.clear();
 609       }
 610     }
 611   } while (truncated);
 612
 613   while (!stat_ops.empty()) {
 614     ret = pop_and_handle_stat_op(dpp, oids, stat_ops);
 615     if (ret < 0) {
 616       if (ret != -ENOENT) {
 617         ldpp_dout(dpp, -1) << "ERROR: stat_async() returned error: " << cpp_strerror(-ret) << dendl;
 618       }
 619     }
 620   }
 621
 622   return 0;
 623 }
 624
 625 int RGWOrphanSearch::build_linked_oids_index(const DoutPrefixProvider *dpp)
 626 {
 627   map<int, list<string> > oids;
 628   map<int, string>::iterator iter = buckets_instance_index.find(search_stage.shard);
 629   for (; iter != buckets_instance_index.end(); ++iter) {
 630     ldpp_dout(dpp, 0) << "building linked oids index: " << iter->first << "/" << buckets_instance_index.size() << dendl;
 631     bool truncated;
 632
 633     string oid = iter->second;
 634
 635     do {
 636       map<string, bufferlist> entries;
 637       int ret = orphan_store.read_entries(oid, search_stage.marker, &entries, &truncated);
 638       if (ret == -ENOENT) {
 639         truncated = false;
 640         ret = 0;
 641       }
 642
 643       if (ret < 0) {
 644         ldpp_dout(dpp, -1) << __func__ << ": ERROR: read_entries() oid=" << oid << " returned ret=" << ret << dendl;
 645         return ret;
 646       }
 647
 648       if (entries.empty()) {
 649         break;
 650       }
 651
 652       for (map<string, bufferlist>::iterator eiter = entries.begin(); eiter != entries.end(); ++eiter) {
 653         ldpp_dout(dpp, 20) << " indexed entry: " << eiter->first << dendl;
 654         ret = build_linked_oids_for_bucket(dpp, eiter->first, oids);
 655         if (ret < 0) {
 656           ldpp_dout(dpp, -1) << __func__ << ": ERROR: build_linked_oids_for_bucket() indexed entry=" << eiter->first
 657                               << " returned ret=" << ret << dendl;
 658           return ret;
 659         }
 660       }
 661
 662       search_stage.shard = iter->first;
 663       search_stage.marker = entries.rbegin()->first; /* last entry */
 664     } while (truncated);
 665
 666     search_stage.marker.clear();
 667   }
 668
 669   int ret = log_oids(dpp, linked_objs_index, oids);
 670   if (ret < 0) {
 671     cerr << __func__ << ": ERROR: log_oids() returned ret=" << ret << std::endl;
 672     return ret;
 673   }
 674
 675   ret = save_state();
 676   if (ret < 0) {
 677     cerr << __func__ << ": ERROR: failed to write state ret=" << ret << std::endl;
 678     return ret;
 679   }
 680
 681   return 0;
 682 }
 683
 684 class OMAPReader {
 685   librados::IoCtx ioctx;
 686   string oid;
 687
 688   map<string, bufferlist> entries;
 689   map<string, bufferlist>::iterator iter;
 690   string marker;
 691   bool truncated;
 692
 693 public:
 694   OMAPReader(librados::IoCtx& _ioctx, const string& _oid) : ioctx(_ioctx), oid(_oid), truncated(true) {
 695     iter = entries.end();
 696   }
 697
 698   int get_next(string *key, bufferlist *pbl, bool *done);
 699 };
 700
 701 int OMAPReader::get_next(string *key, bufferlist *pbl, bool *done)
 702 {
 703   if (iter != entries.end()) {
 704     *key = iter->first;
 705     if (pbl) {
 706       *pbl = iter->second;
 707     }
 708     ++iter;
 709     *done = false;
 710     marker = *key;
 711     return 0;
 712   }
 713
 714   if (!truncated) {
 715     *done = true;
 716     return 0;
 717   }
 718
 719 #define MAX_OMAP_GET_ENTRIES 100
 720   int ret = ioctx.omap_get_vals(oid, marker, MAX_OMAP_GET_ENTRIES, &entries);
 721   if (ret < 0) {
 722     if (ret == -ENOENT) {
 723       *done = true;
 724       return 0;
 725     }
 726     return ret;
 727   }
 728
 729   truncated = (entries.size() == MAX_OMAP_GET_ENTRIES);
 730   iter = entries.begin();
 731   return get_next(key, pbl, done);
 732 }
 733
 734 int RGWOrphanSearch::compare_oid_indexes(const DoutPrefixProvider *dpp)
 735 {
 736   ceph_assert(linked_objs_index.size() == all_objs_index.size());
 737
 738   librados::IoCtx& ioctx = orphan_store.get_ioctx();
 739
 740   librados::IoCtx data_ioctx;
 741
 742   int ret = rgw_init_ioctx(dpp, store->getRados()->get_rados_handle(), search_info.pool, data_ioctx);
 743   if (ret < 0) {
 744     ldpp_dout(dpp, -1) << __func__ << ": rgw_init_ioctx() returned ret=" << ret << dendl;
 745     return ret;
 746   }
 747
 748   uint64_t time_threshold = search_info.start_time.sec() - stale_secs;
 749
 750   map<int, string>::iterator liter = linked_objs_index.begin();
 751   map<int, string>::iterator aiter = all_objs_index.begin();
 752
 753   for (; liter != linked_objs_index.end(); ++liter, ++aiter) {
 754     OMAPReader linked_entries(ioctx, liter->second);
 755     OMAPReader all_entries(ioctx, aiter->second);
 756
 757     bool done;
 758
 759     string cur_linked;
 760     bool linked_done = false;
 761
 762
 763     do {
 764       string key;
 765       int r = all_entries.get_next(&key, NULL, &done);
 766       if (r < 0) {
 767         return r;
 768       }
 769       if (done) {
 770         break;
 771       }
 772
 773       string key_fp = obj_fingerprint(key);
 774
 775       while (cur_linked < key_fp && !linked_done) {
 776         r = linked_entries.get_next(&cur_linked, NULL, &linked_done);
 777         if (r < 0) {
 778           return r;
 779         }
 780       }
 781
 782       if (cur_linked == key_fp) {
 783         ldpp_dout(dpp, 20) << "linked: " << key << dendl;
 784         continue;
 785       }
 786
 787       time_t mtime;
 788       r = data_ioctx.stat(key, NULL, &mtime);
 789       if (r < 0) {
 790         if (r != -ENOENT) {
 791           ldpp_dout(dpp, -1) << "ERROR: ioctx.stat(" << key << ") returned ret=" << r << dendl;
 792         }
 793         continue;
 794       }
 795       if (stale_secs && (uint64_t)mtime >= time_threshold) {
 796         ldpp_dout(dpp, 20) << "skipping: " << key << " (mtime=" << mtime << " threshold=" << time_threshold << ")" << dendl;
 797         continue;
 798       }
 799       ldpp_dout(dpp, 20) << "leaked: " << key << dendl;
 800       cout << "leaked: " << key << std::endl;
 801     } while (!done);
 802   }
 803
 804   return 0;
 805 }
 806
 807 int RGWOrphanSearch::run(const DoutPrefixProvider *dpp)
 808 {
 809   int r;
 810
 811   switch (search_stage.stage) {
 812
 813     case ORPHAN_SEARCH_STAGE_INIT:
 814       ldpp_dout(dpp, 0) << __func__ << "(): initializing state" << dendl;
 815       search_stage = RGWOrphanSearchStage(ORPHAN_SEARCH_STAGE_LSPOOL);
 816       r = save_state();
 817       if (r < 0) {
 818         ldpp_dout(dpp, -1) << __func__ << ": ERROR: failed to save state, ret=" << r << dendl;
 819         return r;
 820       }
 821       // fall through
 822     case ORPHAN_SEARCH_STAGE_LSPOOL:
 823       ldpp_dout(dpp, 0) << __func__ << "(): building index of all objects in pool" << dendl;
 824       r = build_all_oids_index(dpp);
 825       if (r < 0) {
 826         ldpp_dout(dpp, -1) << __func__ << ": ERROR: build_all_objs_index returned ret=" << r << dendl;
 827         return r;
 828       }
 829
 830       search_stage = RGWOrphanSearchStage(ORPHAN_SEARCH_STAGE_LSBUCKETS);
 831       r = save_state();
 832       if (r < 0) {
 833         ldpp_dout(dpp, -1) << __func__ << ": ERROR: failed to save state, ret=" << r << dendl;
 834         return r;
 835       }
 836       // fall through
 837
 838     case ORPHAN_SEARCH_STAGE_LSBUCKETS:
 839       ldpp_dout(dpp, 0) << __func__ << "(): building index of all bucket indexes" << dendl;
 840       r = build_buckets_instance_index(dpp);
 841       if (r < 0) {
 842         ldpp_dout(dpp, -1) << __func__ << ": ERROR: build_all_objs_index returned ret=" << r << dendl;
 843         return r;
 844       }
 845
 846       search_stage = RGWOrphanSearchStage(ORPHAN_SEARCH_STAGE_ITERATE_BI);
 847       r = save_state();
 848       if (r < 0) {
 849         ldpp_dout(dpp, -1) << __func__ << ": ERROR: failed to save state, ret=" << r << dendl;
 850         return r;
 851       }
 852       // fall through
 853
 854
 855     case ORPHAN_SEARCH_STAGE_ITERATE_BI:
 856       ldpp_dout(dpp, 0) << __func__ << "(): building index of all linked objects" << dendl;
 857       r = build_linked_oids_index(dpp);
 858       if (r < 0) {
 859         ldpp_dout(dpp, -1) << __func__ << ": ERROR: build_all_objs_index returned ret=" << r << dendl;
 860         return r;
 861       }
 862
 863       search_stage = RGWOrphanSearchStage(ORPHAN_SEARCH_STAGE_COMPARE);
 864       r = save_state();
 865       if (r < 0) {
 866         ldpp_dout(dpp, -1) << __func__ << ": ERROR: failed to save state, ret=" << r << dendl;
 867         return r;
 868       }
 869       // fall through
 870
 871     case ORPHAN_SEARCH_STAGE_COMPARE:
 872       r = compare_oid_indexes(dpp);
 873       if (r < 0) {
 874         ldpp_dout(dpp, -1) << __func__ << ": ERROR: build_all_objs_index returned ret=" << r << dendl;
 875         return r;
 876       }
 877
 878       break;
 879
 880     default:
 881       ceph_abort();
 882   };
 883
 884   return 0;
 885 }
 886
 887
 888 int RGWOrphanSearch::remove_index(map<int, string>& index)
 889 {
 890   librados::IoCtx& ioctx = orphan_store.get_ioctx();
 891
 892   for (map<int, string>::iterator iter = index.begin(); iter != index.end(); ++iter) {
 893     int r = ioctx.remove(iter->second);
 894     if (r < 0) {
 895       if (r != -ENOENT) {
 896         ldout(store->ctx(), 0) << "ERROR: couldn't remove " << iter->second << ": ret=" << r << dendl;
 897       }
 898     }
 899   }
 900   return 0;
 901 }
 902
 903 int RGWOrphanSearch::finish()
 904 {
 905   int r = remove_index(all_objs_index);
 906   if (r < 0) {
 907     ldout(store->ctx(), 0) << "ERROR: remove_index(" << all_objs_index << ") returned ret=" << r << dendl;
 908   }
 909   r = remove_index(buckets_instance_index);
 910   if (r < 0) {
 911     ldout(store->ctx(), 0) << "ERROR: remove_index(" << buckets_instance_index << ") returned ret=" << r << dendl;
 912   }
 913   r = remove_index(linked_objs_index);
 914   if (r < 0) {
 915     ldout(store->ctx(), 0) << "ERROR: remove_index(" << linked_objs_index << ") returned ret=" << r << dendl;
 916   }
 917
 918   r = orphan_store.remove_job(search_info.job_name);
 919   if (r < 0) {
 920     ldout(store->ctx(), 0) << "ERROR: could not remove job name (" << search_info.job_name << ") ret=" << r << dendl;
 921   }
 922
 923   return r;
 924 }
 925
 926
 927 int RGWRadosList::handle_stat_result(const DoutPrefixProvider *dpp,
 928                                      RGWRados::Object::Stat::Result& result,
 929                                      std::string& bucket_name,
 930                                      rgw_obj_key& obj_key,
 931                                      std::set<string>& obj_oids)
 932 {
 933   obj_oids.clear();
 934
 935   rgw_bucket& bucket = result.obj.bucket;
 936
 937   ldpp_dout(dpp, 20) << "RGWRadosList::" << __func__ <<
 938     " bucket=" << bucket <<
 939     ", has_manifest=" << result.manifest.has_value() <<
 940     dendl;
 941
 942   // iterator to store result of dlo/slo attribute find
 943   decltype(result.attrs)::iterator attr_it = result.attrs.end();
 944   const std::string oid = bucket.marker + "_" + result.obj.get_oid();
 945   ldpp_dout(dpp, 20) << "radoslist processing object=\"" <<
 946       oid << "\"" << dendl;
 947   if (visited_oids.find(oid) != visited_oids.end()) {
 948     // apparently we hit a loop; don't continue with this oid
 949     ldpp_dout(dpp, 15) <<
 950       "radoslist stopped loop at already visited object=\"" <<
 951       oid << "\"" << dendl;
 952     return 0;
 953   }
 954
 955   bucket_name = bucket.name;
 956   obj_key = result.obj.key;
 957
 958   if (!result.manifest) {
 959     /* a very very old object, or part of a multipart upload during upload */
 960     obj_oids.insert(oid);
 961
 962     /*
 963      * multipart parts don't have manifest on them, it's in the meta
 964      * object; we'll process them in
 965      * RGWRadosList::do_incomplete_multipart
 966      */
 967   } else if ((attr_it = result.attrs.find(RGW_ATTR_USER_MANIFEST)) !=
 968              result.attrs.end()) {
 969     // *** handle DLO object ***
 970
 971     obj_oids.insert(oid);
 972     visited_oids.insert(oid); // prevent dlo loops
 973     ldpp_dout(dpp, 15) << "radoslist added to visited list DLO=\"" <<
 974       oid << "\"" << dendl;
 975
 976     char* prefix_path_c = attr_it->second.c_str();
 977     const std::string& prefix_path = prefix_path_c;
 978
 979     const size_t sep_pos = prefix_path.find('/');
 980     if (string::npos == sep_pos) {
 981       return -EINVAL;
 982     }
 983
 984     const std::string bucket_name = prefix_path.substr(0, sep_pos);
 985     const std::string prefix = prefix_path.substr(sep_pos + 1);
 986
 987     add_bucket_prefix(bucket_name, prefix);
 988     ldpp_dout(dpp, 25) << "radoslist DLO oid=\"" << oid <<
 989       "\" added bucket=\"" << bucket_name << "\" prefix=\"" <<
 990       prefix << "\" to process list" << dendl;
 991   } else if ((attr_it = result.attrs.find(RGW_ATTR_SLO_MANIFEST)) !=
 992              result.attrs.end()) {
 993     // *** handle SLO object ***
 994
 995     obj_oids.insert(oid);
 996     visited_oids.insert(oid); // prevent slo loops
 997     ldpp_dout(dpp, 15) << "radoslist added to visited list SLO=\"" <<
 998       oid << "\"" << dendl;
 999
1000     RGWSLOInfo slo_info;
1001     bufferlist::const_iterator bliter = attr_it->second.begin();
1002     try {
1003       ::decode(slo_info, bliter);
1004     } catch (buffer::error& err) {
1005       ldpp_dout(dpp, 0) <<
1006         "ERROR: failed to decode slo manifest for " << oid << dendl;
1007       return -EIO;
1008     }
1009
1010     for (const auto& iter : slo_info.entries) {
1011       const string& path_str = iter.path;
1012
1013       const size_t sep_pos = path_str.find('/', 1 /* skip initial slash */);
1014       if (string::npos == sep_pos) {
1015         return -EINVAL;
1016       }
1017
1018       std::string bucket_name;
1019       std::string obj_name;
1020
1021       bucket_name = url_decode(path_str.substr(1, sep_pos - 1));
1022       obj_name = url_decode(path_str.substr(sep_pos + 1));
1023
1024       const rgw_obj_key obj_key(obj_name);
1025       add_bucket_filter(bucket_name, obj_key);
1026       ldpp_dout(dpp, 25) << "radoslist SLO oid=\"" << oid <<
1027         "\" added bucket=\"" << bucket_name << "\" obj_key=\"" <<
1028         obj_key << "\" to process list" << dendl;
1029     }
1030   } else {
1031     RGWObjManifest& manifest = *result.manifest;
1032
1033     // in multipart, the head object contains no data and just has the
1034     // manifest AND empty objects have no manifest, but they're
1035     // realized as empty rados objects
1036     if (0 == manifest.get_max_head_size() ||
1037         manifest.obj_begin(dpp) == manifest.obj_end(dpp)) {
1038       obj_oids.insert(oid);
1039       // first_insert = true;
1040     }
1041
1042     RGWObjManifest::obj_iterator miter;
1043     for (miter = manifest.obj_begin(dpp); miter != manifest.obj_end(dpp); ++miter) {
1044       const rgw_raw_obj& loc =
1045         miter.get_location().get_raw_obj(store);
1046       string s = loc.oid;
1047       obj_oids.insert(s);
1048     }
1049   }
1050
1051   return 0;
1052 } // RGWRadosList::handle_stat_result
1053
1054 int RGWRadosList::pop_and_handle_stat_op(
1055   const DoutPrefixProvider *dpp,
1056   RGWObjectCtx& obj_ctx,
1057   std::deque<RGWRados::Object::Stat>& ops)
1058 {
1059   std::string bucket_name;
1060   rgw_obj_key obj_key;
1061   std::set<std::string> obj_oids;
1062   RGWRados::Object::Stat& front_op = ops.front();
1063
1064   int ret = front_op.wait();
1065   if (ret < 0) {
1066     if (ret != -ENOENT) {
1067       ldpp_dout(dpp, -1) << "ERROR: stat_async() returned error: " <<
1068         cpp_strerror(-ret) << dendl;
1069     }
1070     goto done;
1071   }
1072
1073   ret = handle_stat_result(dpp, front_op.result, bucket_name, obj_key, obj_oids);
1074   if (ret < 0) {
1075     ldpp_dout(dpp, -1) << "ERROR: handle_stat_result() returned error: " <<
1076       cpp_strerror(-ret) << dendl;
1077   }
1078
1079   // output results
1080   for (const auto& o : obj_oids) {
1081     if (include_rgw_obj_name) {
1082       std::cout << o <<
1083         field_separator << bucket_name <<
1084         field_separator << obj_key <<
1085         std::endl;
1086     } else {
1087       std::cout << o << std::endl;
1088     }
1089   }
1090
1091 done:
1092
1093   // invalidate object context for this object to avoid memory leak
1094   // (see pr https://github.com/ceph/ceph/pull/30174)
1095   obj_ctx.invalidate(front_op.result.obj);
1096
1097   ops.pop_front();
1098   return ret;
1099 }
1100
1101
1102 #if 0 // code that may be the basis for expansion
1103 int RGWRadosList::build_buckets_instance_index()
1104 {
1105   void *handle;
1106   int max = 1000;
1107   string section = "bucket.instance";
1108   int ret = store->meta_mgr->list_keys_init(section, &handle);
1109   if (ret < 0) {
1110     lderr(store->ctx()) << "ERROR: can't get key: " << cpp_strerror(-ret) << dendl;
1111     return ret;
1112   }
1113
1114   map<int, list<string> > instances;
1115
1116   bool truncated;
1117
1118   RGWObjectCtx obj_ctx(store);
1119
1120   int count = 0;
1121   uint64_t total = 0;
1122
1123   do {
1124     list<string> keys;
1125     ret = store->meta_mgr->list_keys_next(handle, max, keys, &truncated);
1126     if (ret < 0) {
1127       lderr(store->ctx()) << "ERROR: lists_keys_next(): " << cpp_strerror(-ret) << dendl;
1128       return ret;
1129     }
1130
1131     for (list<string>::iterator iter = keys.begin(); iter != keys.end(); ++iter) {
1132       ++total;
1133       ldout(store->ctx(), 10) << "bucket_instance=" << *iter << " total=" << total << dendl;
1134       int shard = orphan_shard(*iter);
1135       instances[shard].push_back(*iter);
1136
1137       if (++count >= COUNT_BEFORE_FLUSH) {
1138         ret = log_oids(buckets_instance_index, instances);
1139         if (ret < 0) {
1140           lderr(store->ctx()) << __func__ << ": ERROR: log_oids() returned ret=" << ret << dendl;
1141           return ret;
1142         }
1143         count = 0;
1144         instances.clear();
1145       }
1146     }
1147   } while (truncated);
1148
1149   ret = log_oids(buckets_instance_index, instances);
1150   if (ret < 0) {
1151     lderr(store->ctx()) << __func__ << ": ERROR: log_oids() returned ret=" << ret << dendl;
1152     return ret;
1153   }
1154   store->meta_mgr->list_keys_complete(handle);
1155
1156   return 0;
1157 }
1158 #endif
1159
1160
1161 int RGWRadosList::process_bucket(
1162   const DoutPrefixProvider *dpp,
1163   const std::string& bucket_instance_id,
1164   const std::string& prefix,
1165   const std::set<rgw_obj_key>& entries_filter)
1166 {
1167   ldpp_dout(dpp, 10) << "RGWRadosList::" << __func__ <<
1168     " bucket_instance_id=" << bucket_instance_id <<
1169     ", prefix=" << prefix <<
1170     ", entries_filter.size=" << entries_filter.size() << dendl;
1171
1172   RGWBucketInfo bucket_info;
1173   RGWSysObjectCtx sys_obj_ctx = store->svc()->sysobj->init_obj_ctx();
1174   int ret = store->getRados()->get_bucket_instance_info(sys_obj_ctx,
1175                                                         bucket_instance_id,
1176                                                         bucket_info,
1177                                                         nullptr,
1178                                                         nullptr,
1179                                                         null_yield,
1180                                                         dpp);
1181   if (ret < 0) {
1182     if (ret == -ENOENT) {
1183       // probably raced with bucket removal
1184       return 0;
1185     }
1186     ldpp_dout(dpp, -1) << __func__ <<
1187       ": ERROR: RGWRados::get_bucket_instance_info() returned ret=" <<
1188       ret << dendl;
1189     return ret;
1190   }
1191
1192   RGWRados::Bucket target(store->getRados(), bucket_info);
1193   RGWRados::Bucket::List list_op(&target);
1194
1195   std::string marker;
1196   list_op.params.marker = rgw_obj_key(marker);
1197   list_op.params.list_versions = true;
1198   list_op.params.enforce_ns = false;
1199   list_op.params.allow_unordered = false;
1200   list_op.params.prefix = prefix;
1201
1202   bool truncated;
1203
1204   std::deque<RGWRados::Object::Stat> stat_ops;
1205   std::string prev_versioned_key_name = "";
1206
1207   RGWObjectCtx obj_ctx(store);
1208
1209   do {
1210     std::vector<rgw_bucket_dir_entry> result;
1211
1212     constexpr int64_t LIST_OBJS_MAX_ENTRIES = 100;
1213     ret = list_op.list_objects(dpp, LIST_OBJS_MAX_ENTRIES, &result,
1214                                NULL, &truncated, null_yield);
1215     if (ret == -ENOENT) {
1216       // race with bucket delete?
1217       ret = 0;
1218       break;
1219     } else if (ret < 0) {
1220       std::cerr << "ERROR: store->list_objects(): " << cpp_strerror(-ret) <<
1221         std::endl;
1222       return ret;
1223     }
1224
1225     for (std::vector<rgw_bucket_dir_entry>::iterator iter = result.begin();
1226          iter != result.end();
1227          ++iter) {
1228       rgw_bucket_dir_entry& entry = *iter;
1229
1230       if (entry.key.instance.empty()) {
1231         ldpp_dout(dpp, 20) << "obj entry: " << entry.key.name << dendl;
1232       } else {
1233         ldpp_dout(dpp, 20) << "obj entry: " << entry.key.name <<
1234           " [" << entry.key.instance << "]" << dendl;
1235       }
1236
1237       ldpp_dout(dpp, 20) << __func__ << ": entry.key.name=" <<
1238         entry.key.name << " entry.key.instance=" << entry.key.instance <<
1239         dendl;
1240
1241       // ignore entries that are not in the filter if there is a filter
1242       if (!entries_filter.empty() &&
1243           entries_filter.find(entry.key) == entries_filter.cend()) {
1244         continue;
1245       }
1246
1247       // we need to do this in two cases below, so use a lambda
1248       auto do_stat_key =
1249         [&](const rgw_obj_key& key) -> int {
1250           int ret;
1251
1252           rgw_obj obj(bucket_info.bucket, key);
1253
1254           RGWRados::Object op_target(store->getRados(), bucket_info,
1255                                      obj_ctx, obj);
1256
1257           stat_ops.push_back(RGWRados::Object::Stat(&op_target));
1258           RGWRados::Object::Stat& op = stat_ops.back();
1259
1260           ret = op.stat_async(dpp);
1261           if (ret < 0) {
1262             ldpp_dout(dpp, -1) << "ERROR: stat_async() returned error: " <<
1263               cpp_strerror(-ret) << dendl;
1264             return ret;
1265           }
1266
1267           if (stat_ops.size() >= max_concurrent_ios) {
1268             ret = pop_and_handle_stat_op(dpp, obj_ctx, stat_ops);
1269             if (ret < 0) {
1270               if (ret != -ENOENT) {
1271                 ldpp_dout(dpp, -1) <<
1272                   "ERROR: pop_and_handle_stat_op() returned error: " <<
1273                   cpp_strerror(-ret) << dendl;
1274               }
1275
1276               // clear error, so we'll continue processing directory
1277               ret = 0;
1278             }
1279           }
1280
1281           return ret;
1282         }; // do_stat_key lambda
1283
1284       // for versioned objects, make sure the head object is handled
1285       // as well by ignoring the instance identifier
1286       if (!entry.key.instance.empty() &&
1287           entry.key.name != prev_versioned_key_name) {
1288         // don't do the same key twice; even though out bucket index
1289         // listing allows unordered, since all versions of an object
1290         // use the same bucket index key, they'll all end up together
1291         // and sorted
1292         prev_versioned_key_name = entry.key.name;
1293
1294         rgw_obj_key uninstanced(entry.key.name);
1295
1296         ret = do_stat_key(uninstanced);
1297         if (ret < 0) {
1298           return ret;
1299         }
1300       }
1301
1302       ret = do_stat_key(entry.key);
1303       if (ret < 0) {
1304         return ret;
1305       }
1306     } // for iter loop
1307   } while (truncated);
1308
1309   while (!stat_ops.empty()) {
1310     ret = pop_and_handle_stat_op(dpp, obj_ctx, stat_ops);
1311     if (ret < 0) {
1312       if (ret != -ENOENT) {
1313         ldpp_dout(dpp, -1) << "ERROR: stat_async() returned error: " <<
1314           cpp_strerror(-ret) << dendl;
1315       }
1316     }
1317   }
1318
1319   return 0;
1320 }
1321
1322
1323 int RGWRadosList::run(const DoutPrefixProvider *dpp)
1324 {
1325   int ret;
1326   void* handle = nullptr;
1327
1328   ret = store->ctl()->meta.mgr->list_keys_init(dpp, "bucket", &handle);
1329   if (ret < 0) {
1330     ldpp_dout(dpp, -1) << "RGWRadosList::" << __func__ <<
1331       " ERROR: list_keys_init returned " <<
1332       cpp_strerror(-ret) << dendl;
1333     return ret;
1334   }
1335
1336   const int max_keys = 1000;
1337   bool truncated = true;
1338
1339   do {
1340     std::list<std::string> buckets;
1341     ret = store->ctl()->meta.mgr->list_keys_next(handle, max_keys,
1342                                                  buckets, &truncated);
1343
1344     for (std::string& bucket_id : buckets) {
1345       ret = run(dpp, bucket_id);
1346       if (ret == -ENOENT) {
1347         continue;
1348       } else if (ret < 0) {
1349         return ret;
1350       }
1351     }
1352   } while (truncated);
1353
1354   return 0;
1355 } // RGWRadosList::run()
1356
1357
1358 int RGWRadosList::run(const DoutPrefixProvider *dpp, const std::string& start_bucket_name)
1359 {
1360   RGWSysObjectCtx sys_obj_ctx = store->svc()->sysobj->init_obj_ctx();
1361   RGWObjectCtx obj_ctx(store);
1362   RGWBucketInfo bucket_info;
1363   int ret;
1364
1365   add_bucket_entire(start_bucket_name);
1366
1367   while (! bucket_process_map.empty()) {
1368     // pop item from map and capture its key data
1369     auto front = bucket_process_map.begin();
1370     std::string bucket_name = front->first;
1371     process_t process;
1372     std::swap(process, front->second);
1373     bucket_process_map.erase(front);
1374
1375     RGWBucketInfo bucket_info;
1376     ret = store->getRados()->get_bucket_info(store->svc(),
1377                                              tenant_name,
1378                                              bucket_name,
1379                                              bucket_info,
1380                                              nullptr,
1381                                              null_yield,
1382                                              dpp);
1383     if (ret == -ENOENT) {
1384       std::cerr << "WARNING: bucket " << bucket_name <<
1385         " does not exist; could it have been deleted very recently?" <<
1386         std::endl;
1387       continue;
1388     } else if (ret < 0) {
1389       std::cerr << "ERROR: could not get info for bucket " << bucket_name <<
1390         " -- " << cpp_strerror(-ret) << std::endl;
1391       return ret;
1392     }
1393
1394     const std::string bucket_id = bucket_info.bucket.get_key();
1395
1396     static const std::set<rgw_obj_key> empty_filter;
1397     static const std::string empty_prefix;
1398
1399     auto do_process_bucket =
1400       [dpp, &bucket_id, this]
1401       (const std::string& prefix,
1402        const std::set<rgw_obj_key>& entries_filter) -> int {
1403         int ret = process_bucket(dpp, bucket_id, prefix, entries_filter);
1404         if (ret == -ENOENT) {
1405           // bucket deletion race?
1406           return 0;
1407         } if (ret < 0) {
1408           ldpp_dout(dpp, -1) << "RGWRadosList::" << __func__ <<
1409             ": ERROR: process_bucket(); bucket_id=" <<
1410             bucket_id << " returned ret=" << ret << dendl;
1411         }
1412
1413         return ret;
1414       };
1415
1416     // either process the whole bucket *or* process the filters and/or
1417     // the prefixes
1418     if (process.entire_container) {
1419       ret = do_process_bucket(empty_prefix, empty_filter);
1420       if (ret < 0) {
1421         return ret;
1422       }
1423     } else {
1424       if (! process.filter_keys.empty()) {
1425         ret = do_process_bucket(empty_prefix, process.filter_keys);
1426         if (ret < 0) {
1427           return ret;
1428         }
1429       }
1430       for (const auto& p : process.prefixes) {
1431         ret = do_process_bucket(p, empty_filter);
1432         if (ret < 0) {
1433           return ret;
1434         }
1435       }
1436     }
1437   } // while (! bucket_process_map.empty())
1438
1439   if (include_rgw_obj_name) {
1440     goto done;
1441   }
1442
1443   // now handle incomplete multipart uploads by going back to the
1444   // initial bucket
1445
1446   ret = store->getRados()->get_bucket_info(store->svc(),
1447                                            tenant_name,
1448                                            start_bucket_name,
1449                                            bucket_info,
1450                                            nullptr,
1451                                            null_yield,
1452                                            dpp);
1453   if (ret == -ENOENT) {
1454     // bucket deletion race?
1455     return 0;
1456   } else if (ret < 0) {
1457     ldpp_dout(dpp, -1) << "RGWRadosList::" << __func__ <<
1458       ": ERROR: get_bucket_info returned ret=" << ret << dendl;
1459     return ret;
1460   }
1461
1462   ret = do_incomplete_multipart(dpp, store, bucket_info);
1463   if (ret < 0) {
1464     ldpp_dout(dpp, -1) << "RGWRadosList::" << __func__ <<
1465       ": ERROR: do_incomplete_multipart returned ret=" << ret << dendl;
1466     return ret;
1467   }
1468
1469 done:
1470
1471   return 0;
1472 } // RGWRadosList::run(string)
1473
1474
1475 int RGWRadosList::do_incomplete_multipart(
1476   const DoutPrefixProvider *dpp,
1477   rgw::sal::RGWRadosStore* store,
1478   RGWBucketInfo& bucket_info)
1479 {
1480   constexpr int max_uploads = 1000;
1481   constexpr int max_parts = 1000;
1482   static const std::string mp_ns = RGW_OBJ_NS_MULTIPART;
1483   static MultipartMetaFilter mp_filter;
1484
1485   int ret;
1486
1487   RGWRados::Bucket target(store->getRados(), bucket_info);
1488   RGWRados::Bucket::List list_op(&target);
1489   list_op.params.ns = mp_ns;
1490   list_op.params.filter = &mp_filter;
1491   // use empty string for initial list_op.params.marker
1492   // use empty strings for list_op.params.{prefix,delim}
1493
1494   bool is_listing_truncated;
1495
1496   do {
1497     std::vector<rgw_bucket_dir_entry> objs;
1498     std::map<string, bool> common_prefixes;
1499     ret = list_op.list_objects(dpp, max_uploads, &objs, &common_prefixes,
1500                                &is_listing_truncated, null_yield);
1501     if (ret == -ENOENT) {
1502       // could bucket have been removed while this is running?
1503       ldpp_dout(dpp, 5) << "RGWRadosList::" << __func__ <<
1504         ": WARNING: call to list_objects of multipart namespace got ENOENT; "
1505         "assuming bucket removal race" << dendl;
1506       break;
1507     } else if (ret < 0) {
1508       ldpp_dout(dpp, -1) << "RGWRadosList::" << __func__ <<
1509         ": ERROR: list_objects op returned ret=" << ret << dendl;
1510       return ret;
1511     }
1512
1513     if (!objs.empty()) {
1514       std::vector<RGWMultipartUploadEntry> uploads;
1515       RGWMultipartUploadEntry entry;
1516       for (const rgw_bucket_dir_entry& obj : objs) {
1517         const rgw_obj_key& key = obj.key;
1518         if (!entry.mp.from_meta(key.name)) {
1519           // we only want the meta objects, so skip all the components
1520           continue;
1521         }
1522         entry.obj = obj;
1523         uploads.push_back(entry);
1524         ldpp_dout(dpp, 20) << "RGWRadosList::" << __func__ <<
1525           " processing incomplete multipart entry " <<
1526           entry << dendl;
1527       }
1528
1529       // now process the uploads vector
1530       for (const auto& upload : uploads) {
1531         const RGWMPObj& mp = upload.mp;
1532         int parts_marker = 0;
1533         bool is_parts_truncated = false;
1534
1535
1536         do { // while (is_parts_truncated);
1537           std::map<uint32_t, RGWUploadPartInfo> parts;
1538           ret = list_multipart_parts(dpp, store, bucket_info, store->ctx(),
1539                                      mp.get_upload_id(), mp.get_meta(),
1540                                      max_parts, parts_marker,
1541                                      parts, &parts_marker,
1542                                      &is_parts_truncated);
1543           if (ret == -ENOENT) {
1544             ldpp_dout(dpp, 5) << "RGWRadosList::" << __func__ <<
1545               ": WARNING: list_multipart_parts returned ret=-ENOENT "
1546               "for " << mp.get_upload_id() << ", moving on" << dendl;
1547             break;
1548           } else if (ret < 0) {
1549             ldpp_dout(dpp, -1) << "RGWRadosList::" << __func__ <<
1550               ": ERROR: list_multipart_parts returned ret=" << ret <<
1551               dendl;
1552             return ret;
1553           }
1554
1555           for (auto& p : parts) {
1556             RGWObjManifest& manifest = p.second.manifest;
1557             for (auto obj_it = manifest.obj_begin(dpp);
1558                  obj_it != manifest.obj_end(dpp);
1559                  ++obj_it) {
1560               const rgw_raw_obj& loc =
1561                 obj_it.get_location().get_raw_obj(store);
1562               std::cout << loc.oid << std::endl;
1563             } // for (auto obj_it
1564           } // for (auto& p
1565         } while (is_parts_truncated);
1566       } // for (const auto& upload
1567     } // if objs not empty
1568   } while (is_listing_truncated);
1569
1570   return 0;
1571 } // RGWRadosList::do_incomplete_multipart