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