ceph/src/rgw/rgw_bucket.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 <cerrno>
   5 #include <map>
   6 #include <sstream>
   7 #include <string>
   8 #include <string_view>
   9
  10 #include <boost/format.hpp>
  11
  12 #include "common/errno.h"
  13 #include "common/ceph_json.h"
  14 #include "include/scope_guard.h"
  15
  16 #include "rgw_datalog.h"
  17 #include "rgw_rados.h"
  18 #include "rgw_zone.h"
  19 #include "rgw_acl.h"
  20 #include "rgw_acl_s3.h"
  21 #include "rgw_tag_s3.h"
  22
  23 #include "include/types.h"
  24 #include "rgw_bucket.h"
  25 #include "rgw_user.h"
  26 #include "rgw_string.h"
  27 #include "rgw_multi.h"
  28 #include "rgw_op.h"
  29 #include "rgw_bucket_sync.h"
  30
  31 #include "services/svc_zone.h"
  32 #include "services/svc_sys_obj.h"
  33 #include "services/svc_bucket.h"
  34 #include "services/svc_bucket_sync.h"
  35 #include "services/svc_meta.h"
  36 #include "services/svc_meta_be_sobj.h"
  37 #include "services/svc_user.h"
  38 #include "services/svc_cls.h"
  39 #include "services/svc_bilog_rados.h"
  40
  41 #include "include/rados/librados.hpp"
  42 // until everything is moved from rgw_common
  43 #include "rgw_common.h"
  44 #include "rgw_reshard.h"
  45 #include "rgw_lc.h"
  46 #include "rgw_bucket_layout.h"
  47
  48 // stolen from src/cls/version/cls_version.cc
  49 #define VERSION_ATTR "ceph.objclass.version"
  50
  51 #include "cls/user/cls_user_types.h"
  52
  53 #include "rgw_sal_rados.h"
  54
  55 #define dout_context g_ceph_context
  56 #define dout_subsys ceph_subsys_rgw
  57
  58 #define BUCKET_TAG_TIMEOUT 30
  59
  60 using namespace std;
  61
  62 // default number of entries to list with each bucket listing call
  63 // (use marker to bridge between calls)
  64 static constexpr size_t listing_max_entries = 1000;
  65
  66 void init_bucket(rgw_bucket *b, const char *t, const char *n, const char *dp, const char *ip, const char *m, const char *id)
  67 {
  68   b->tenant = t;
  69   b->name = n;
  70   b->marker = m;
  71   b->bucket_id = id;
  72   b->explicit_placement.data_pool = rgw_pool(dp);
  73   b->explicit_placement.index_pool = rgw_pool(ip);
  74 }
  75
  76 /*
  77  * The tenant_name is always returned on purpose. May be empty, of course.
  78  */
  79 static void parse_bucket(const string& bucket,
  80                          string *tenant_name,
  81                          string *bucket_name,
  82                          string *bucket_instance = nullptr /* optional */)
  83 {
  84   /*
  85    * expected format: [tenant/]bucket:bucket_instance
  86    */
  87   int pos = bucket.find('/');
  88   if (pos >= 0) {
  89     *tenant_name = bucket.substr(0, pos);
  90   } else {
  91     tenant_name->clear();
  92   }
  93   string bn = bucket.substr(pos + 1);
  94   pos = bn.find (':');
  95   if (pos < 0) {
  96     *bucket_name = std::move(bn);
  97     return;
  98   }
  99   *bucket_name = bn.substr(0, pos);
 100   if (bucket_instance) {
 101     *bucket_instance = bn.substr(pos + 1);
 102   }
 103
 104   /*
 105    * deal with the possible tenant:bucket:bucket_instance case
 106    */
 107   if (tenant_name->empty()) {
 108     pos = bucket_instance->find(':');
 109     if (pos >= 0) {
 110       *tenant_name = *bucket_name;
 111       *bucket_name = bucket_instance->substr(0, pos);
 112       *bucket_instance = bucket_instance->substr(pos + 1);
 113     }
 114   }
 115 }
 116
 117 /*
 118  * Note that this is not a reversal of parse_bucket(). That one deals
 119  * with the syntax we need in metadata and such. This one deals with
 120  * the representation in RADOS pools. We chose '/' because it's not
 121  * acceptable in bucket names and thus qualified buckets cannot conflict
 122  * with the legacy or S3 buckets.
 123  */
 124 std::string rgw_make_bucket_entry_name(const std::string& tenant_name,
 125                                        const std::string& bucket_name) {
 126   std::string bucket_entry;
 127
 128   if (bucket_name.empty()) {
 129     bucket_entry.clear();
 130   } else if (tenant_name.empty()) {
 131     bucket_entry = bucket_name;
 132   } else {
 133     bucket_entry = tenant_name + "/" + bucket_name;
 134   }
 135
 136   return bucket_entry;
 137 }
 138
 139 /*
 140  * Tenants are separated from buckets in URLs by a colon in S3.
 141  * This function is not to be used on Swift URLs, not even for COPY arguments.
 142  */
 143 void rgw_parse_url_bucket(const string &bucket, const string& auth_tenant,
 144                           string &tenant_name, string &bucket_name) {
 145
 146   int pos = bucket.find(':');
 147   if (pos >= 0) {
 148     /*
 149      * N.B.: We allow ":bucket" syntax with explicit empty tenant in order
 150      * to refer to the legacy tenant, in case users in new named tenants
 151      * want to access old global buckets.
 152      */
 153     tenant_name = bucket.substr(0, pos);
 154     bucket_name = bucket.substr(pos + 1);
 155   } else {
 156     tenant_name = auth_tenant;
 157     bucket_name = bucket;
 158   }
 159 }
 160
 161 int rgw_bucket_parse_bucket_instance(const string& bucket_instance, string *bucket_name, string *bucket_id, int *shard_id)
 162 {
 163   auto pos = bucket_instance.rfind(':');
 164   if (pos == string::npos) {
 165     return -EINVAL;
 166   }
 167
 168   string first = bucket_instance.substr(0, pos);
 169   string second = bucket_instance.substr(pos + 1);
 170
 171   pos = first.find(':');
 172
 173   if (pos == string::npos) {
 174     *shard_id = -1;
 175     *bucket_name = first;
 176     *bucket_id = second;
 177     return 0;
 178   }
 179
 180   *bucket_name = first.substr(0, pos);
 181   *bucket_id = first.substr(pos + 1);
 182
 183   string err;
 184   *shard_id = strict_strtol(second.c_str(), 10, &err);
 185   if (!err.empty()) {
 186     return -EINVAL;
 187   }
 188
 189   return 0;
 190 }
 191
 192 // parse key in format: [tenant/]name:instance[:shard_id]
 193 int rgw_bucket_parse_bucket_key(CephContext *cct, const string& key,
 194                                 rgw_bucket *bucket, int *shard_id)
 195 {
 196   std::string_view name{key};
 197   std::string_view instance;
 198
 199   // split tenant/name
 200   auto pos = name.find('/');
 201   if (pos != string::npos) {
 202     auto tenant = name.substr(0, pos);
 203     bucket->tenant.assign(tenant.begin(), tenant.end());
 204     name = name.substr(pos + 1);
 205   } else {
 206     bucket->tenant.clear();
 207   }
 208
 209   // split name:instance
 210   pos = name.find(':');
 211   if (pos != string::npos) {
 212     instance = name.substr(pos + 1);
 213     name = name.substr(0, pos);
 214   }
 215   bucket->name.assign(name.begin(), name.end());
 216
 217   // split instance:shard
 218   pos = instance.find(':');
 219   if (pos == string::npos) {
 220     bucket->bucket_id.assign(instance.begin(), instance.end());
 221     if (shard_id) {
 222       *shard_id = -1;
 223     }
 224     return 0;
 225   }
 226
 227   // parse shard id
 228   auto shard = instance.substr(pos + 1);
 229   string err;
 230   auto id = strict_strtol(shard.data(), 10, &err);
 231   if (!err.empty()) {
 232     if (cct) {
 233       ldout(cct, 0) << "ERROR: failed to parse bucket shard '"
 234           << instance.data() << "': " << err << dendl;
 235     }
 236     return -EINVAL;
 237   }
 238
 239   if (shard_id) {
 240     *shard_id = id;
 241   }
 242   instance = instance.substr(0, pos);
 243   bucket->bucket_id.assign(instance.begin(), instance.end());
 244   return 0;
 245 }
 246
 247 static void dump_mulipart_index_results(list<rgw_obj_index_key>& objs_to_unlink,
 248         Formatter *f)
 249 {
 250   for (const auto& o : objs_to_unlink) {
 251     f->dump_string("object",  o.name);
 252   }
 253 }
 254
 255 void check_bad_user_bucket_mapping(rgw::sal::Store* store, rgw::sal::User* user,
 256                                    bool fix,
 257                                    optional_yield y,
 258                                    const DoutPrefixProvider *dpp)
 259 {
 260   rgw::sal::BucketList user_buckets;
 261   string marker;
 262
 263   CephContext *cct = store->ctx();
 264
 265   size_t max_entries = cct->_conf->rgw_list_buckets_max_chunk;
 266
 267   do {
 268     int ret = user->list_buckets(dpp, marker, string(), max_entries, false, user_buckets, y);
 269     if (ret < 0) {
 270       ldout(store->ctx(), 0) << "failed to read user buckets: "
 271                              << cpp_strerror(-ret) << dendl;
 272       return;
 273     }
 274
 275     map<string, std::unique_ptr<rgw::sal::Bucket>>& buckets = user_buckets.get_buckets();
 276     for (auto i = buckets.begin();
 277          i != buckets.end();
 278          ++i) {
 279       marker = i->first;
 280
 281       auto& bucket = i->second;
 282
 283       std::unique_ptr<rgw::sal::Bucket> actual_bucket;
 284       int r = store->get_bucket(dpp, user, user->get_tenant(), bucket->get_name(), &actual_bucket, null_yield);
 285       if (r < 0) {
 286         ldout(store->ctx(), 0) << "could not get bucket info for bucket=" << bucket << dendl;
 287         continue;
 288       }
 289
 290       if (actual_bucket->get_name().compare(bucket->get_name()) != 0 ||
 291           actual_bucket->get_tenant().compare(bucket->get_tenant()) != 0 ||
 292           actual_bucket->get_marker().compare(bucket->get_marker()) != 0 ||
 293           actual_bucket->get_bucket_id().compare(bucket->get_bucket_id()) != 0) {
 294         cout << "bucket info mismatch: expected " << actual_bucket << " got " << bucket << std::endl;
 295         if (fix) {
 296           cout << "fixing" << std::endl;
 297           r = actual_bucket->chown(dpp, user, nullptr, null_yield);
 298           if (r < 0) {
 299             cerr << "failed to fix bucket: " << cpp_strerror(-r) << std::endl;
 300           }
 301         }
 302       }
 303     }
 304   } while (user_buckets.is_truncated());
 305 }
 306
 307 // returns true if entry is in the empty namespace. note: function
 308 // type conforms to type RGWBucketListNameFilter
 309 bool rgw_bucket_object_check_filter(const std::string& oid)
 310 {
 311   const static std::string empty_ns;
 312   rgw_obj_key key; // thrown away but needed for parsing
 313   return rgw_obj_key::oid_to_key_in_ns(oid, &key, empty_ns);
 314 }
 315
 316 int rgw_remove_object(const DoutPrefixProvider *dpp, rgw::sal::Store* store, rgw::sal::Bucket* bucket, rgw_obj_key& key)
 317 {
 318   RGWObjectCtx rctx(store);
 319
 320   if (key.instance.empty()) {
 321     key.instance = "null";
 322   }
 323
 324   std::unique_ptr<rgw::sal::Object> object = bucket->get_object(key);
 325
 326   return object->delete_object(dpp, &rctx, null_yield);
 327 }
 328
 329 static void set_err_msg(std::string *sink, std::string msg)
 330 {
 331   if (sink && !msg.empty())
 332     *sink = msg;
 333 }
 334
 335 int RGWBucket::init(rgw::sal::Store* _store, RGWBucketAdminOpState& op_state,
 336                     optional_yield y, const DoutPrefixProvider *dpp, std::string *err_msg)
 337 {
 338   if (!_store) {
 339     set_err_msg(err_msg, "no storage!");
 340     return -EINVAL;
 341   }
 342
 343   store = _store;
 344
 345   std::string bucket_name = op_state.get_bucket_name();
 346
 347   if (bucket_name.empty() && op_state.get_user_id().empty())
 348     return -EINVAL;
 349
 350   user = store->get_user(op_state.get_user_id());
 351   std::string tenant = user->get_tenant();
 352
 353   // split possible tenant/name
 354   auto pos = bucket_name.find('/');
 355   if (pos != string::npos) {
 356     tenant = bucket_name.substr(0, pos);
 357     bucket_name = bucket_name.substr(pos + 1);
 358   }
 359
 360   int r = store->get_bucket(dpp, user.get(), tenant, bucket_name, &bucket, y);
 361   if (r < 0) {
 362       set_err_msg(err_msg, "failed to fetch bucket info for bucket=" + bucket_name);
 363       return r;
 364   }
 365
 366   op_state.set_bucket(bucket->clone());
 367
 368   if (!rgw::sal::User::empty(user.get())) {
 369     r = user->load_user(dpp, y);
 370     if (r < 0) {
 371       set_err_msg(err_msg, "failed to fetch user info");
 372       return r;
 373     }
 374   }
 375
 376   op_state.display_name = user->get_display_name();
 377
 378   clear_failure();
 379   return 0;
 380 }
 381
 382 bool rgw_find_bucket_by_id(const DoutPrefixProvider *dpp, CephContext *cct, rgw::sal::Store* store,
 383                            const string& marker, const string& bucket_id, rgw_bucket* bucket_out)
 384 {
 385   void *handle = NULL;
 386   bool truncated = false;
 387   string s;
 388
 389   int ret = store->meta_list_keys_init(dpp, "bucket.instance", marker, &handle);
 390   if (ret < 0) {
 391     cerr << "ERROR: can't get key: " << cpp_strerror(-ret) << std::endl;
 392     store->meta_list_keys_complete(handle);
 393     return -ret;
 394   }
 395   do {
 396       list<string> keys;
 397       ret = store->meta_list_keys_next(dpp, handle, 1000, keys, &truncated);
 398       if (ret < 0) {
 399         cerr << "ERROR: lists_keys_next(): " << cpp_strerror(-ret) << std::endl;
 400         store->meta_list_keys_complete(handle);
 401         return -ret;
 402       }
 403       for (list<string>::iterator iter = keys.begin(); iter != keys.end(); ++iter) {
 404         s = *iter;
 405         ret = rgw_bucket_parse_bucket_key(cct, s, bucket_out, nullptr);
 406         if (ret < 0) {
 407           continue;
 408         }
 409         if (bucket_id == bucket_out->bucket_id) {
 410           store->meta_list_keys_complete(handle);
 411           return true;
 412         }
 413       }
 414   } while (truncated);
 415   store->meta_list_keys_complete(handle);
 416   return false;
 417 }
 418
 419 int RGWBucket::chown(RGWBucketAdminOpState& op_state, const string& marker,
 420                      optional_yield y, const DoutPrefixProvider *dpp, std::string *err_msg)
 421 {
 422   int ret = bucket->chown(dpp, user.get(), user.get(), y, &marker);
 423   if (ret < 0) {
 424     set_err_msg(err_msg, "Failed to change object ownership: " + cpp_strerror(-ret));
 425   }
 426
 427   return ret;
 428 }
 429
 430 int RGWBucket::set_quota(RGWBucketAdminOpState& op_state, const DoutPrefixProvider *dpp, std::string *err_msg)
 431 {
 432   bucket = op_state.get_bucket()->clone();
 433
 434   bucket->get_info().quota = op_state.quota;
 435   int r = bucket->put_info(dpp, false, real_time());
 436   if (r < 0) {
 437     set_err_msg(err_msg, "ERROR: failed writing bucket instance info: " + cpp_strerror(-r));
 438     return r;
 439   }
 440   return r;
 441 }
 442
 443 int RGWBucket::remove_object(const DoutPrefixProvider *dpp, RGWBucketAdminOpState& op_state, std::string *err_msg)
 444 {
 445   std::string object_name = op_state.get_object_name();
 446
 447   rgw_obj_key key(object_name);
 448
 449   bucket = op_state.get_bucket()->clone();
 450
 451   int ret = rgw_remove_object(dpp, store, bucket.get(), key);
 452   if (ret < 0) {
 453     set_err_msg(err_msg, "unable to remove object" + cpp_strerror(-ret));
 454     return ret;
 455   }
 456
 457   return 0;
 458 }
 459
 460 static void dump_bucket_index(const vector<rgw_bucket_dir_entry>& objs,  Formatter *f)
 461 {
 462   for (auto iter = objs.begin(); iter != objs.end(); ++iter) {
 463     f->dump_string("object", iter->key.name);
 464   }
 465 }
 466
 467 static void dump_bucket_usage(map<RGWObjCategory, RGWStorageStats>& stats, Formatter *formatter)
 468 {
 469   map<RGWObjCategory, RGWStorageStats>::iterator iter;
 470
 471   formatter->open_object_section("usage");
 472   for (iter = stats.begin(); iter != stats.end(); ++iter) {
 473     RGWStorageStats& s = iter->second;
 474     formatter->open_object_section(to_string(iter->first));
 475     s.dump(formatter);
 476     formatter->close_section();
 477   }
 478   formatter->close_section();
 479 }
 480
 481 static void dump_index_check(map<RGWObjCategory, RGWStorageStats> existing_stats,
 482         map<RGWObjCategory, RGWStorageStats> calculated_stats,
 483         Formatter *formatter)
 484 {
 485   formatter->open_object_section("check_result");
 486   formatter->open_object_section("existing_header");
 487   dump_bucket_usage(existing_stats, formatter);
 488   formatter->close_section();
 489   formatter->open_object_section("calculated_header");
 490   dump_bucket_usage(calculated_stats, formatter);
 491   formatter->close_section();
 492   formatter->close_section();
 493 }
 494
 495 int RGWBucket::check_bad_index_multipart(RGWBucketAdminOpState& op_state,
 496                RGWFormatterFlusher& flusher,
 497                const DoutPrefixProvider *dpp, std::string *err_msg)
 498 {
 499   bool fix_index = op_state.will_fix_index();
 500
 501   bool is_truncated;
 502   map<string, bool> meta_objs;
 503   map<rgw_obj_index_key, string> all_objs;
 504
 505   bucket = op_state.get_bucket()->clone();
 506
 507   rgw::sal::Bucket::ListParams params;
 508
 509   params.list_versions = true;
 510   params.ns = RGW_OBJ_NS_MULTIPART;
 511
 512   do {
 513     rgw::sal::Bucket::ListResults results;
 514     int r = bucket->list(dpp, params, listing_max_entries, results, null_yield);
 515     if (r < 0) {
 516       set_err_msg(err_msg, "failed to list objects in bucket=" + bucket->get_name() +
 517               " err=" +  cpp_strerror(-r));
 518
 519       return r;
 520     }
 521     is_truncated = results.is_truncated;
 522
 523     vector<rgw_bucket_dir_entry>::iterator iter;
 524     for (iter = results.objs.begin(); iter != results.objs.end(); ++iter) {
 525       rgw_obj_index_key key = iter->key;
 526       rgw_obj obj(bucket->get_key(), key);
 527       string oid = obj.get_oid();
 528
 529       int pos = oid.find_last_of('.');
 530       if (pos < 0) {
 531         /* obj has no suffix */
 532         all_objs[key] = oid;
 533       } else {
 534         /* obj has suffix */
 535         string name = oid.substr(0, pos);
 536         string suffix = oid.substr(pos + 1);
 537
 538         if (suffix.compare("meta") == 0) {
 539           meta_objs[name] = true;
 540         } else {
 541           all_objs[key] = name;
 542         }
 543       }
 544     }
 545   } while (is_truncated);
 546
 547   list<rgw_obj_index_key> objs_to_unlink;
 548   Formatter *f =  flusher.get_formatter();
 549
 550   f->open_array_section("invalid_multipart_entries");
 551
 552   for (auto aiter = all_objs.begin(); aiter != all_objs.end(); ++aiter) {
 553     string& name = aiter->second;
 554
 555     if (meta_objs.find(name) == meta_objs.end()) {
 556       objs_to_unlink.push_back(aiter->first);
 557     }
 558
 559     if (objs_to_unlink.size() > listing_max_entries) {
 560       if (fix_index) {
 561         int r = bucket->remove_objs_from_index(dpp, objs_to_unlink);
 562         if (r < 0) {
 563           set_err_msg(err_msg, "ERROR: remove_obj_from_index() returned error: " +
 564                       cpp_strerror(-r));
 565           return r;
 566         }
 567       }
 568
 569       dump_mulipart_index_results(objs_to_unlink, flusher.get_formatter());
 570       flusher.flush();
 571       objs_to_unlink.clear();
 572     }
 573   }
 574
 575   if (fix_index) {
 576     int r = bucket->remove_objs_from_index(dpp, objs_to_unlink);
 577     if (r < 0) {
 578       set_err_msg(err_msg, "ERROR: remove_obj_from_index() returned error: " +
 579               cpp_strerror(-r));
 580
 581       return r;
 582     }
 583   }
 584
 585   dump_mulipart_index_results(objs_to_unlink, f);
 586   f->close_section();
 587   flusher.flush();
 588
 589   return 0;
 590 }
 591
 592 int RGWBucket::check_object_index(const DoutPrefixProvider *dpp,
 593                                   RGWBucketAdminOpState& op_state,
 594                                   RGWFormatterFlusher& flusher,
 595                                   optional_yield y,
 596                                   std::string *err_msg)
 597 {
 598
 599   bool fix_index = op_state.will_fix_index();
 600
 601   if (!fix_index) {
 602     set_err_msg(err_msg, "check-objects flag requires fix index enabled");
 603     return -EINVAL;
 604   }
 605
 606   bucket->set_tag_timeout(dpp, BUCKET_TAG_TIMEOUT);
 607
 608   rgw::sal::Bucket::ListResults results;
 609   results.is_truncated = true;
 610
 611   Formatter *formatter = flusher.get_formatter();
 612   formatter->open_object_section("objects");
 613   while (results.is_truncated) {
 614     rgw::sal::Bucket::ListParams params;
 615     params.marker = results.next_marker;
 616     params.force_check_filter = rgw_bucket_object_check_filter;
 617
 618     int r = bucket->list(dpp, params, listing_max_entries, results, y);
 619
 620     if (r == -ENOENT) {
 621       break;
 622     } else if (r < 0) {
 623       set_err_msg(err_msg, "ERROR: failed operation r=" + cpp_strerror(-r));
 624     }
 625
 626     dump_bucket_index(results.objs, formatter);
 627     flusher.flush();
 628   }
 629
 630   formatter->close_section();
 631
 632   bucket->set_tag_timeout(dpp, 0);
 633
 634   return 0;
 635 }
 636
 637
 638 int RGWBucket::check_index(const DoutPrefixProvider *dpp,
 639         RGWBucketAdminOpState& op_state,
 640         map<RGWObjCategory, RGWStorageStats>& existing_stats,
 641         map<RGWObjCategory, RGWStorageStats>& calculated_stats,
 642         std::string *err_msg)
 643 {
 644   bool fix_index = op_state.will_fix_index();
 645
 646   int r = bucket->check_index(dpp, existing_stats, calculated_stats);
 647   if (r < 0) {
 648     set_err_msg(err_msg, "failed to check index error=" + cpp_strerror(-r));
 649     return r;
 650   }
 651
 652   if (fix_index) {
 653     r = bucket->rebuild_index(dpp);
 654     if (r < 0) {
 655       set_err_msg(err_msg, "failed to rebuild index err=" + cpp_strerror(-r));
 656       return r;
 657     }
 658   }
 659
 660   return 0;
 661 }
 662
 663 int RGWBucket::sync(RGWBucketAdminOpState& op_state, const DoutPrefixProvider *dpp, std::string *err_msg)
 664 {
 665   if (!store->is_meta_master()) {
 666     set_err_msg(err_msg, "ERROR: failed to update bucket sync: only allowed on meta master zone");
 667     return -EINVAL;
 668   }
 669   bool sync = op_state.will_sync_bucket();
 670   if (sync) {
 671     bucket->get_info().flags &= ~BUCKET_DATASYNC_DISABLED;
 672   } else {
 673     bucket->get_info().flags |= BUCKET_DATASYNC_DISABLED;
 674   }
 675
 676   int r = bucket->put_info(dpp, false, real_time());
 677   if (r < 0) {
 678     set_err_msg(err_msg, "ERROR: failed writing bucket instance info:" + cpp_strerror(-r));
 679     return r;
 680   }
 681
 682   int shards_num = bucket->get_info().layout.current_index.layout.normal.num_shards? bucket->get_info().layout.current_index.layout.normal.num_shards : 1;
 683   int shard_id = bucket->get_info().layout.current_index.layout.normal.num_shards? 0 : -1;
 684
 685   if (!sync) {
 686     r = static_cast<rgw::sal::RadosStore*>(store)->svc()->bilog_rados->log_stop(dpp, bucket->get_info(), -1);
 687     if (r < 0) {
 688       set_err_msg(err_msg, "ERROR: failed writing stop bilog:" + cpp_strerror(-r));
 689       return r;
 690     }
 691   } else {
 692     r = static_cast<rgw::sal::RadosStore*>(store)->svc()->bilog_rados->log_start(dpp, bucket->get_info(), -1);
 693     if (r < 0) {
 694       set_err_msg(err_msg, "ERROR: failed writing resync bilog:" + cpp_strerror(-r));
 695       return r;
 696     }
 697   }
 698
 699   for (int i = 0; i < shards_num; ++i, ++shard_id) {
 700     r = static_cast<rgw::sal::RadosStore*>(store)->svc()->datalog_rados->add_entry(dpp, bucket->get_info(), shard_id);
 701     if (r < 0) {
 702       set_err_msg(err_msg, "ERROR: failed writing data log:" + cpp_strerror(-r));
 703       return r;
 704     }
 705   }
 706
 707   return 0;
 708 }
 709
 710
 711 int RGWBucket::policy_bl_to_stream(bufferlist& bl, ostream& o)
 712 {
 713   RGWAccessControlPolicy_S3 policy(g_ceph_context);
 714   int ret = decode_bl(bl, policy);
 715   if (ret < 0) {
 716     ldout(store->ctx(),0) << "failed to decode RGWAccessControlPolicy" << dendl;
 717   }
 718   policy.to_xml(o);
 719   return 0;
 720 }
 721
 722 int rgw_object_get_attr(const DoutPrefixProvider *dpp,
 723                         rgw::sal::Store* store, rgw::sal::Object* obj,
 724                         const char* attr_name, bufferlist& out_bl, optional_yield y)
 725 {
 726   RGWObjectCtx obj_ctx(store);
 727   std::unique_ptr<rgw::sal::Object::ReadOp> rop = obj->get_read_op(&obj_ctx);
 728
 729   return rop->get_attr(dpp, attr_name, out_bl, y);
 730 }
 731
 732 int RGWBucket::get_policy(RGWBucketAdminOpState& op_state, RGWAccessControlPolicy& policy, optional_yield y, const DoutPrefixProvider *dpp)
 733 {
 734   int ret;
 735   std::string object_name = op_state.get_object_name();
 736
 737   bucket = op_state.get_bucket()->clone();
 738
 739   if (!object_name.empty()) {
 740     bufferlist bl;
 741     std::unique_ptr<rgw::sal::Object> obj = bucket->get_object(rgw_obj_key(object_name));
 742
 743     ret = rgw_object_get_attr(dpp, store, obj.get(), RGW_ATTR_ACL, bl, y);
 744     if (ret < 0){
 745       return ret;
 746     }
 747
 748     ret = decode_bl(bl, policy);
 749     if (ret < 0) {
 750       ldout(store->ctx(),0) << "failed to decode RGWAccessControlPolicy" << dendl;
 751     }
 752     return ret;
 753   }
 754
 755   map<string, bufferlist>::iterator aiter = bucket->get_attrs().find(RGW_ATTR_ACL);
 756   if (aiter == bucket->get_attrs().end()) {
 757     return -ENOENT;
 758   }
 759
 760   ret = decode_bl(aiter->second, policy);
 761   if (ret < 0) {
 762     ldout(store->ctx(),0) << "failed to decode RGWAccessControlPolicy" << dendl;
 763   }
 764
 765   return ret;
 766 }
 767
 768
 769 int RGWBucketAdminOp::get_policy(rgw::sal::Store* store, RGWBucketAdminOpState& op_state,
 770                   RGWAccessControlPolicy& policy, const DoutPrefixProvider *dpp)
 771 {
 772   RGWBucket bucket;
 773
 774   int ret = bucket.init(store, op_state, null_yield, dpp);
 775   if (ret < 0)
 776     return ret;
 777
 778   ret = bucket.get_policy(op_state, policy, null_yield, dpp);
 779   if (ret < 0)
 780     return ret;
 781
 782   return 0;
 783 }
 784
 785 /* Wrappers to facilitate RESTful interface */
 786
 787
 788 int RGWBucketAdminOp::get_policy(rgw::sal::Store* store, RGWBucketAdminOpState& op_state,
 789                   RGWFormatterFlusher& flusher, const DoutPrefixProvider *dpp)
 790 {
 791   RGWAccessControlPolicy policy(store->ctx());
 792
 793   int ret = get_policy(store, op_state, policy, dpp);
 794   if (ret < 0)
 795     return ret;
 796
 797   Formatter *formatter = flusher.get_formatter();
 798
 799   flusher.start(0);
 800
 801   formatter->open_object_section("policy");
 802   policy.dump(formatter);
 803   formatter->close_section();
 804
 805   flusher.flush();
 806
 807   return 0;
 808 }
 809
 810 int RGWBucketAdminOp::dump_s3_policy(rgw::sal::Store* store, RGWBucketAdminOpState& op_state,
 811                   ostream& os, const DoutPrefixProvider *dpp)
 812 {
 813   RGWAccessControlPolicy_S3 policy(store->ctx());
 814
 815   int ret = get_policy(store, op_state, policy, dpp);
 816   if (ret < 0)
 817     return ret;
 818
 819   policy.to_xml(os);
 820
 821   return 0;
 822 }
 823
 824 int RGWBucketAdminOp::unlink(rgw::sal::Store* store, RGWBucketAdminOpState& op_state, const DoutPrefixProvider *dpp)
 825 {
 826   RGWBucket bucket;
 827
 828   int ret = bucket.init(store, op_state, null_yield, dpp);
 829   if (ret < 0)
 830     return ret;
 831
 832   return static_cast<rgw::sal::RadosStore*>(store)->ctl()->bucket->unlink_bucket(op_state.get_user_id(), op_state.get_bucket()->get_info().bucket, null_yield, dpp, true);
 833 }
 834
 835 int RGWBucketAdminOp::link(rgw::sal::Store* store, RGWBucketAdminOpState& op_state, const DoutPrefixProvider *dpp, string *err)
 836 {
 837   if (!op_state.is_user_op()) {
 838     set_err_msg(err, "empty user id");
 839     return -EINVAL;
 840   }
 841
 842   RGWBucket bucket;
 843   int ret = bucket.init(store, op_state, null_yield, dpp, err);
 844   if (ret < 0)
 845     return ret;
 846
 847   string bucket_id = op_state.get_bucket_id();
 848   std::string display_name = op_state.get_user_display_name();
 849   std::unique_ptr<rgw::sal::Bucket> loc_bucket;
 850   std::unique_ptr<rgw::sal::Bucket> old_bucket;
 851
 852   loc_bucket = op_state.get_bucket()->clone();
 853
 854   if (!bucket_id.empty() && bucket_id != loc_bucket->get_bucket_id()) {
 855     set_err_msg(err,
 856         "specified bucket id does not match " + loc_bucket->get_bucket_id());
 857     return -EINVAL;
 858   }
 859
 860   old_bucket = loc_bucket->clone();
 861
 862   loc_bucket->get_key().tenant = op_state.get_user_id().tenant;
 863
 864   if (!op_state.new_bucket_name.empty()) {
 865     auto pos = op_state.new_bucket_name.find('/');
 866     if (pos != string::npos) {
 867       loc_bucket->get_key().tenant = op_state.new_bucket_name.substr(0, pos);
 868       loc_bucket->get_key().name = op_state.new_bucket_name.substr(pos + 1);
 869     } else {
 870       loc_bucket->get_key().name = op_state.new_bucket_name;
 871     }
 872   }
 873
 874   RGWObjVersionTracker objv_tracker;
 875   RGWObjVersionTracker old_version = loc_bucket->get_info().objv_tracker;
 876
 877   map<string, bufferlist>::iterator aiter = loc_bucket->get_attrs().find(RGW_ATTR_ACL);
 878   if (aiter == loc_bucket->get_attrs().end()) {
 879         // should never happen; only pre-argonaut buckets lacked this.
 880     ldpp_dout(dpp, 0) << "WARNING: can't bucket link because no acl on bucket=" << old_bucket << dendl;
 881     set_err_msg(err,
 882         "While crossing the Anavros you have displeased the goddess Hera."
 883         "  You must sacrifice your ancient bucket " + loc_bucket->get_bucket_id());
 884     return -EINVAL;
 885   }
 886   bufferlist& aclbl = aiter->second;
 887   RGWAccessControlPolicy policy;
 888   ACLOwner owner;
 889   try {
 890    auto iter = aclbl.cbegin();
 891    decode(policy, iter);
 892    owner = policy.get_owner();
 893   } catch (buffer::error& e) {
 894     set_err_msg(err, "couldn't decode policy");
 895     return -EIO;
 896   }
 897
 898   int r = static_cast<rgw::sal::RadosStore*>(store)->ctl()->bucket->unlink_bucket(owner.get_id(), old_bucket->get_info().bucket, null_yield, dpp, false);
 899   if (r < 0) {
 900     set_err_msg(err, "could not unlink policy from user " + owner.get_id().to_str());
 901     return r;
 902   }
 903
 904   // now update the user for the bucket...
 905   if (display_name.empty()) {
 906     ldpp_dout(dpp, 0) << "WARNING: user " << op_state.get_user_id() << " has no display name set" << dendl;
 907   }
 908
 909   RGWAccessControlPolicy policy_instance;
 910   policy_instance.create_default(op_state.get_user_id(), display_name);
 911   owner = policy_instance.get_owner();
 912
 913   aclbl.clear();
 914   policy_instance.encode(aclbl);
 915
 916   bool exclusive = false;
 917   loc_bucket->get_info().owner = op_state.get_user_id();
 918   if (*loc_bucket != *old_bucket) {
 919     loc_bucket->get_info().bucket = loc_bucket->get_key();
 920     loc_bucket->get_info().objv_tracker.version_for_read()->ver = 0;
 921     exclusive = true;
 922   }
 923
 924   r = loc_bucket->put_info(dpp, exclusive, ceph::real_time());
 925   if (r < 0) {
 926     set_err_msg(err, "ERROR: failed writing bucket instance info: " + cpp_strerror(-r));
 927     return r;
 928   }
 929
 930   /* link to user */
 931   RGWBucketEntryPoint ep;
 932   ep.bucket = loc_bucket->get_info().bucket;
 933   ep.owner = op_state.get_user_id();
 934   ep.creation_time = loc_bucket->get_info().creation_time;
 935   ep.linked = true;
 936   rgw::sal::Attrs ep_attrs;
 937   rgw_ep_info ep_data{ep, ep_attrs};
 938
 939   r = static_cast<rgw::sal::RadosStore*>(store)->ctl()->bucket->link_bucket(op_state.get_user_id(), loc_bucket->get_info().bucket, loc_bucket->get_info().creation_time, null_yield, dpp, true, &ep_data);
 940   if (r < 0) {
 941     set_err_msg(err, "failed to relink bucket");
 942     return r;
 943   }
 944
 945   if (*loc_bucket != *old_bucket) {
 946     // like RGWRados::delete_bucket -- excepting no bucket_index work.
 947     r = static_cast<rgw::sal::RadosStore*>(store)->ctl()->bucket->remove_bucket_entrypoint_info(
 948                                         old_bucket->get_key(), null_yield, dpp,
 949                                         RGWBucketCtl::Bucket::RemoveParams()
 950                                         .set_objv_tracker(&ep_data.ep_objv));
 951     if (r < 0) {
 952       set_err_msg(err, "failed to unlink old bucket " + old_bucket->get_tenant() + "/" + old_bucket->get_name());
 953       return r;
 954     }
 955     r = static_cast<rgw::sal::RadosStore*>(store)->ctl()->bucket->remove_bucket_instance_info(
 956                                         old_bucket->get_key(), old_bucket->get_info(),
 957                                         null_yield, dpp,
 958                                         RGWBucketCtl::BucketInstance::RemoveParams()
 959                                         .set_objv_tracker(&ep_data.ep_objv));
 960     if (r < 0) {
 961       set_err_msg(err, "failed to unlink old bucket " + old_bucket->get_tenant() + "/" + old_bucket->get_name());
 962       return r;
 963     }
 964   }
 965
 966   return 0;
 967 }
 968
 969 int RGWBucketAdminOp::chown(rgw::sal::Store* store, RGWBucketAdminOpState& op_state, const string& marker, const DoutPrefixProvider *dpp, string *err)
 970 {
 971   RGWBucket bucket;
 972
 973   int ret = bucket.init(store, op_state, null_yield, dpp, err);
 974   if (ret < 0)
 975     return ret;
 976
 977   return bucket.chown(op_state, marker, null_yield, dpp, err);
 978
 979 }
 980
 981 int RGWBucketAdminOp::check_index(rgw::sal::Store* store, RGWBucketAdminOpState& op_state,
 982                   RGWFormatterFlusher& flusher, optional_yield y, const DoutPrefixProvider *dpp)
 983 {
 984   int ret;
 985   map<RGWObjCategory, RGWStorageStats> existing_stats;
 986   map<RGWObjCategory, RGWStorageStats> calculated_stats;
 987
 988
 989   RGWBucket bucket;
 990
 991   ret = bucket.init(store, op_state, null_yield, dpp);
 992   if (ret < 0)
 993     return ret;
 994
 995   Formatter *formatter = flusher.get_formatter();
 996   flusher.start(0);
 997
 998   ret = bucket.check_bad_index_multipart(op_state, flusher, dpp);
 999   if (ret < 0)
1000     return ret;
1001
1002   ret = bucket.check_object_index(dpp, op_state, flusher, y);
1003   if (ret < 0)
1004     return ret;
1005
1006   ret = bucket.check_index(dpp, op_state, existing_stats, calculated_stats);
1007   if (ret < 0)
1008     return ret;
1009
1010   dump_index_check(existing_stats, calculated_stats, formatter);
1011   flusher.flush();
1012
1013   return 0;
1014 }
1015
1016 int RGWBucketAdminOp::remove_bucket(rgw::sal::Store* store, RGWBucketAdminOpState& op_state,
1017                                     optional_yield y, const DoutPrefixProvider *dpp,
1018                                     bool bypass_gc, bool keep_index_consistent)
1019 {
1020   std::unique_ptr<rgw::sal::Bucket> bucket;
1021   std::unique_ptr<rgw::sal::User> user = store->get_user(op_state.get_user_id());
1022
1023   int ret = store->get_bucket(dpp, user.get(), user->get_tenant(), op_state.get_bucket_name(),
1024                               &bucket, y);
1025   if (ret < 0)
1026     return ret;
1027
1028   if (bypass_gc)
1029     ret = bucket->remove_bucket_bypass_gc(op_state.get_max_aio(), keep_index_consistent, y, dpp);
1030   else
1031     ret = bucket->remove_bucket(dpp, op_state.will_delete_children(),
1032                                 false, nullptr, y);
1033
1034   return ret;
1035 }
1036
1037 int RGWBucketAdminOp::remove_object(rgw::sal::Store* store, RGWBucketAdminOpState& op_state, const DoutPrefixProvider *dpp)
1038 {
1039   RGWBucket bucket;
1040
1041   int ret = bucket.init(store, op_state, null_yield, dpp);
1042   if (ret < 0)
1043     return ret;
1044
1045   return bucket.remove_object(dpp, op_state);
1046 }
1047
1048 int RGWBucketAdminOp::sync_bucket(rgw::sal::Store* store, RGWBucketAdminOpState& op_state, const DoutPrefixProvider *dpp, string *err_msg)
1049 {
1050   RGWBucket bucket;
1051   int ret = bucket.init(store, op_state, null_yield, dpp, err_msg);
1052   if (ret < 0)
1053   {
1054     return ret;
1055   }
1056   return bucket.sync(op_state, dpp, err_msg);
1057 }
1058
1059 static int bucket_stats(rgw::sal::Store* store,
1060                         const std::string& tenant_name,
1061                         const std::string& bucket_name,
1062                         Formatter *formatter,
1063                         const DoutPrefixProvider *dpp)
1064 {
1065   std::unique_ptr<rgw::sal::Bucket> bucket;
1066   map<RGWObjCategory, RGWStorageStats> stats;
1067
1068   real_time mtime;
1069   int ret = store->get_bucket(dpp, nullptr, tenant_name, bucket_name, &bucket, null_yield);
1070   if (ret < 0) {
1071     return ret;
1072   }
1073
1074   string bucket_ver, master_ver;
1075   string max_marker;
1076   ret = bucket->read_stats(dpp, RGW_NO_SHARD, &bucket_ver, &master_ver, stats, &max_marker);
1077   if (ret < 0) {
1078     cerr << "error getting bucket stats bucket=" << bucket->get_name() << " ret=" << ret << std::endl;
1079     return ret;
1080   }
1081
1082   utime_t ut(mtime);
1083   utime_t ctime_ut(bucket->get_creation_time());
1084
1085   formatter->open_object_section("stats");
1086   formatter->dump_string("bucket", bucket->get_name());
1087   formatter->dump_int("num_shards",
1088                       bucket->get_info().layout.current_index.layout.normal.num_shards);
1089   formatter->dump_string("tenant", bucket->get_tenant());
1090   formatter->dump_string("zonegroup", bucket->get_info().zonegroup);
1091   formatter->dump_string("placement_rule", bucket->get_info().placement_rule.to_str());
1092   ::encode_json("explicit_placement", bucket->get_key().explicit_placement, formatter);
1093   formatter->dump_string("id", bucket->get_bucket_id());
1094   formatter->dump_string("marker", bucket->get_marker());
1095   formatter->dump_stream("index_type") << bucket->get_info().layout.current_index.layout.type;
1096   ::encode_json("owner", bucket->get_info().owner, formatter);
1097   formatter->dump_string("ver", bucket_ver);
1098   formatter->dump_string("master_ver", master_ver);
1099   ut.gmtime(formatter->dump_stream("mtime"));
1100   ctime_ut.gmtime(formatter->dump_stream("creation_time"));
1101   formatter->dump_string("max_marker", max_marker);
1102   dump_bucket_usage(stats, formatter);
1103   encode_json("bucket_quota", bucket->get_info().quota, formatter);
1104
1105   // bucket tags
1106   auto iter = bucket->get_attrs().find(RGW_ATTR_TAGS);
1107   if (iter != bucket->get_attrs().end()) {
1108     RGWObjTagSet_S3 tagset;
1109     bufferlist::const_iterator piter{&iter->second};
1110     try {
1111       tagset.decode(piter);
1112       tagset.dump(formatter);
1113     } catch (buffer::error& err) {
1114       cerr << "ERROR: caught buffer:error, couldn't decode TagSet" << std::endl;
1115     }
1116   }
1117
1118   // TODO: bucket CORS
1119   // TODO: bucket LC
1120   formatter->close_section();
1121
1122   return 0;
1123 }
1124
1125 int RGWBucketAdminOp::limit_check(rgw::sal::Store* store,
1126                                   RGWBucketAdminOpState& op_state,
1127                                   const std::list<std::string>& user_ids,
1128                                   RGWFormatterFlusher& flusher, optional_yield y,
1129                                   const DoutPrefixProvider *dpp,
1130                                   bool warnings_only)
1131 {
1132   int ret = 0;
1133   const size_t max_entries =
1134     store->ctx()->_conf->rgw_list_buckets_max_chunk;
1135
1136   const size_t safe_max_objs_per_shard =
1137     store->ctx()->_conf->rgw_safe_max_objects_per_shard;
1138
1139   uint16_t shard_warn_pct =
1140     store->ctx()->_conf->rgw_shard_warning_threshold;
1141   if (shard_warn_pct > 100)
1142     shard_warn_pct = 90;
1143
1144   Formatter *formatter = flusher.get_formatter();
1145   flusher.start(0);
1146
1147   formatter->open_array_section("users");
1148
1149   for (const auto& user_id : user_ids) {
1150
1151     formatter->open_object_section("user");
1152     formatter->dump_string("user_id", user_id);
1153     formatter->open_array_section("buckets");
1154
1155     string marker;
1156     rgw::sal::BucketList buckets;
1157     do {
1158       std::unique_ptr<rgw::sal::User> user = store->get_user(rgw_user(user_id));
1159
1160       ret = user->list_buckets(dpp, marker, string(), max_entries, false, buckets, y);
1161
1162       if (ret < 0)
1163         return ret;
1164
1165       map<string, std::unique_ptr<rgw::sal::Bucket>>& m_buckets = buckets.get_buckets();
1166
1167       for (const auto& iter : m_buckets) {
1168         auto& bucket = iter.second;
1169         uint32_t num_shards = 1;
1170         uint64_t num_objects = 0;
1171
1172         marker = bucket->get_name(); /* Casey's location for marker update,
1173                                      * as we may now not reach the end of
1174                                      * the loop body */
1175
1176         ret = bucket->load_bucket(dpp, null_yield);
1177         if (ret < 0)
1178           continue;
1179
1180         /* need stats for num_entries */
1181         string bucket_ver, master_ver;
1182         std::map<RGWObjCategory, RGWStorageStats> stats;
1183         ret = bucket->read_stats(dpp, RGW_NO_SHARD, &bucket_ver, &master_ver, stats, nullptr);
1184
1185         if (ret < 0)
1186           continue;
1187
1188         for (const auto& s : stats) {
1189           num_objects += s.second.num_objects;
1190         }
1191
1192         num_shards = bucket->get_info().layout.current_index.layout.normal.num_shards;
1193         uint64_t objs_per_shard =
1194           (num_shards) ? num_objects/num_shards : num_objects;
1195         {
1196           bool warn;
1197           stringstream ss;
1198           uint64_t fill_pct = objs_per_shard * 100 / safe_max_objs_per_shard;
1199           if (fill_pct > 100) {
1200             ss << "OVER " << fill_pct << "%";
1201             warn = true;
1202           } else if (fill_pct >= shard_warn_pct) {
1203             ss << "WARN " << fill_pct << "%";
1204             warn = true;
1205           } else {
1206             ss << "OK";
1207             warn = false;
1208           }
1209
1210           if (warn || !warnings_only) {
1211             formatter->open_object_section("bucket");
1212             formatter->dump_string("bucket", bucket->get_name());
1213             formatter->dump_string("tenant", bucket->get_tenant());
1214             formatter->dump_int("num_objects", num_objects);
1215             formatter->dump_int("num_shards", num_shards);
1216             formatter->dump_int("objects_per_shard", objs_per_shard);
1217             formatter->dump_string("fill_status", ss.str());
1218             formatter->close_section();
1219           }
1220         }
1221       }
1222       formatter->flush(cout);
1223     } while (buckets.is_truncated()); /* foreach: bucket */
1224
1225     formatter->close_section();
1226     formatter->close_section();
1227     formatter->flush(cout);
1228
1229   } /* foreach: user_id */
1230
1231   formatter->close_section();
1232   formatter->flush(cout);
1233
1234   return ret;
1235 } /* RGWBucketAdminOp::limit_check */
1236
1237 int RGWBucketAdminOp::info(rgw::sal::Store* store,
1238                            RGWBucketAdminOpState& op_state,
1239                            RGWFormatterFlusher& flusher,
1240                            optional_yield y,
1241                            const DoutPrefixProvider *dpp)
1242 {
1243   RGWBucket bucket;
1244   int ret = 0;
1245   const std::string& bucket_name = op_state.get_bucket_name();
1246   if (!bucket_name.empty()) {
1247     ret = bucket.init(store, op_state, null_yield, dpp);
1248     if (-ENOENT == ret)
1249       return -ERR_NO_SUCH_BUCKET;
1250     else if (ret < 0)
1251       return ret;
1252   }
1253
1254   Formatter *formatter = flusher.get_formatter();
1255   flusher.start(0);
1256
1257   CephContext *cct = store->ctx();
1258
1259   const size_t max_entries = cct->_conf->rgw_list_buckets_max_chunk;
1260
1261   const bool show_stats = op_state.will_fetch_stats();
1262   const rgw_user& user_id = op_state.get_user_id();
1263   if (op_state.is_user_op()) {
1264     formatter->open_array_section("buckets");
1265
1266     rgw::sal::BucketList buckets;
1267     std::unique_ptr<rgw::sal::User> user = store->get_user(op_state.get_user_id());
1268     std::string marker;
1269     const std::string empty_end_marker;
1270     constexpr bool no_need_stats = false; // set need_stats to false
1271
1272     do {
1273       ret = user->list_buckets(dpp, marker, empty_end_marker, max_entries,
1274                               no_need_stats, buckets, y);
1275       if (ret < 0) {
1276         return ret;
1277       }
1278
1279       const std::string* marker_cursor = nullptr;
1280       map<string, std::unique_ptr<rgw::sal::Bucket>>& m = buckets.get_buckets();
1281
1282       for (const auto& i : m) {
1283         const std::string& obj_name = i.first;
1284         if (!bucket_name.empty() && bucket_name != obj_name) {
1285           continue;
1286         }
1287
1288         if (show_stats) {
1289           bucket_stats(store, user_id.tenant, obj_name, formatter, dpp);
1290         } else {
1291           formatter->dump_string("bucket", obj_name);
1292         }
1293
1294         marker_cursor = &obj_name;
1295       } // for loop
1296       if (marker_cursor) {
1297         marker = *marker_cursor;
1298       }
1299
1300       flusher.flush();
1301     } while (buckets.is_truncated());
1302
1303     formatter->close_section();
1304   } else if (!bucket_name.empty()) {
1305     ret = bucket_stats(store, user_id.tenant, bucket_name, formatter, dpp);
1306     if (ret < 0) {
1307       return ret;
1308     }
1309   } else {
1310     void *handle = nullptr;
1311     bool truncated = true;
1312
1313     formatter->open_array_section("buckets");
1314     ret = store->meta_list_keys_init(dpp, "bucket", string(), &handle);
1315     while (ret == 0 && truncated) {
1316       std::list<std::string> buckets;
1317       constexpr int max_keys = 1000;
1318       ret = store->meta_list_keys_next(dpp, handle, max_keys, buckets,
1319                                                    &truncated);
1320       for (auto& bucket_name : buckets) {
1321         if (show_stats) {
1322           bucket_stats(store, user_id.tenant, bucket_name, formatter, dpp);
1323         } else {
1324           formatter->dump_string("bucket", bucket_name);
1325         }
1326       }
1327     }
1328     store->meta_list_keys_complete(handle);
1329
1330     formatter->close_section();
1331   }
1332
1333   flusher.flush();
1334
1335   return 0;
1336 }
1337
1338 int RGWBucketAdminOp::set_quota(rgw::sal::Store* store, RGWBucketAdminOpState& op_state, const DoutPrefixProvider *dpp)
1339 {
1340   RGWBucket bucket;
1341
1342   int ret = bucket.init(store, op_state, null_yield, dpp);
1343   if (ret < 0)
1344     return ret;
1345   return bucket.set_quota(op_state, dpp);
1346 }
1347
1348 static int purge_bucket_instance(rgw::sal::Store* store, const RGWBucketInfo& bucket_info, const DoutPrefixProvider *dpp)
1349 {
1350   std::unique_ptr<rgw::sal::Bucket> bucket;
1351   int ret = store->get_bucket(nullptr, bucket_info, &bucket);
1352   if (ret < 0)
1353     return ret;
1354
1355   return bucket->purge_instance(dpp);
1356 }
1357
1358 inline auto split_tenant(const std::string& bucket_name){
1359   auto p = bucket_name.find('/');
1360   if(p != std::string::npos) {
1361     return std::make_pair(bucket_name.substr(0,p), bucket_name.substr(p+1));
1362   }
1363   return std::make_pair(std::string(), bucket_name);
1364 }
1365
1366 using bucket_instance_ls = std::vector<RGWBucketInfo>;
1367 void get_stale_instances(rgw::sal::Store* store, const std::string& bucket_name,
1368                          const vector<std::string>& lst,
1369                          bucket_instance_ls& stale_instances,
1370                          const DoutPrefixProvider *dpp)
1371 {
1372
1373   bucket_instance_ls other_instances;
1374 // first iterate over the entries, and pick up the done buckets; these
1375 // are guaranteed to be stale
1376   for (const auto& bucket_instance : lst){
1377     RGWBucketInfo binfo;
1378     std::unique_ptr<rgw::sal::Bucket> bucket;
1379     rgw_bucket rbucket;
1380     rgw_bucket_parse_bucket_key(store->ctx(), bucket_instance, &rbucket, nullptr);
1381     int r = store->get_bucket(dpp, nullptr, rbucket, &bucket, null_yield);
1382     if (r < 0){
1383       // this can only happen if someone deletes us right when we're processing
1384       ldpp_dout(dpp, -1) << "Bucket instance is invalid: " << bucket_instance
1385                           << cpp_strerror(-r) << dendl;
1386       continue;
1387     }
1388     binfo = bucket->get_info();
1389     if (binfo.reshard_status == cls_rgw_reshard_status::DONE)
1390       stale_instances.emplace_back(std::move(binfo));
1391     else {
1392       other_instances.emplace_back(std::move(binfo));
1393     }
1394   }
1395
1396   // Read the cur bucket info, if the bucket doesn't exist we can simply return
1397   // all the instances
1398   auto [tenant, bname] = split_tenant(bucket_name);
1399   RGWBucketInfo cur_bucket_info;
1400   std::unique_ptr<rgw::sal::Bucket> cur_bucket;
1401   int r = store->get_bucket(dpp, nullptr, tenant, bname, &cur_bucket, null_yield);
1402   if (r < 0) {
1403     if (r == -ENOENT) {
1404       // bucket doesn't exist, everything is stale then
1405       stale_instances.insert(std::end(stale_instances),
1406                              std::make_move_iterator(other_instances.begin()),
1407                              std::make_move_iterator(other_instances.end()));
1408     } else {
1409       // all bets are off if we can't read the bucket, just return the sureshot stale instances
1410       ldpp_dout(dpp, -1) << "error: reading bucket info for bucket: "
1411                           << bname << cpp_strerror(-r) << dendl;
1412     }
1413     return;
1414   }
1415
1416   // Don't process further in this round if bucket is resharding
1417   cur_bucket_info = cur_bucket->get_info();
1418   if (cur_bucket_info.reshard_status == cls_rgw_reshard_status::IN_PROGRESS)
1419     return;
1420
1421   other_instances.erase(std::remove_if(other_instances.begin(), other_instances.end(),
1422                                        [&cur_bucket_info](const RGWBucketInfo& b){
1423                                          return (b.bucket.bucket_id == cur_bucket_info.bucket.bucket_id ||
1424                                                  b.bucket.bucket_id == cur_bucket_info.new_bucket_instance_id);
1425                                        }),
1426                         other_instances.end());
1427
1428   // check if there are still instances left
1429   if (other_instances.empty()) {
1430     return;
1431   }
1432
1433   // Now we have a bucket with instances where the reshard status is none, this
1434   // usually happens when the reshard process couldn't complete, lockdown the
1435   // bucket and walk through these instances to make sure no one else interferes
1436   // with these
1437   {
1438     RGWBucketReshardLock reshard_lock(static_cast<rgw::sal::RadosStore*>(store), cur_bucket->get_info(), true);
1439     r = reshard_lock.lock(dpp);
1440     if (r < 0) {
1441       // most likely bucket is under reshard, return the sureshot stale instances
1442       ldpp_dout(dpp, 5) << __func__
1443                              << "failed to take reshard lock; reshard underway likey" << dendl;
1444       return;
1445     }
1446     auto sg = make_scope_guard([&reshard_lock](){ reshard_lock.unlock();} );
1447     // this should be fast enough that we may not need to renew locks and check
1448     // exit status?, should we read the values of the instances again?
1449     stale_instances.insert(std::end(stale_instances),
1450                            std::make_move_iterator(other_instances.begin()),
1451                            std::make_move_iterator(other_instances.end()));
1452   }
1453
1454   return;
1455 }
1456
1457 static int process_stale_instances(rgw::sal::Store* store, RGWBucketAdminOpState& op_state,
1458                                    RGWFormatterFlusher& flusher,
1459                                    const DoutPrefixProvider *dpp,
1460                                    std::function<void(const bucket_instance_ls&,
1461                                                       Formatter *,
1462                                                       rgw::sal::Store*)> process_f)
1463 {
1464   std::string marker;
1465   void *handle;
1466   Formatter *formatter = flusher.get_formatter();
1467   static constexpr auto default_max_keys = 1000;
1468
1469   int ret = store->meta_list_keys_init(dpp, "bucket.instance", marker, &handle);
1470   if (ret < 0) {
1471     cerr << "ERROR: can't get key: " << cpp_strerror(-ret) << std::endl;
1472     return ret;
1473   }
1474
1475   bool truncated;
1476
1477   formatter->open_array_section("keys");
1478   auto g = make_scope_guard([&store, &handle, &formatter]() {
1479                               store->meta_list_keys_complete(handle);
1480                               formatter->close_section(); // keys
1481                               formatter->flush(cout);
1482                             });
1483
1484   do {
1485     list<std::string> keys;
1486
1487     ret = store->meta_list_keys_next(dpp, handle, default_max_keys, keys, &truncated);
1488     if (ret < 0 && ret != -ENOENT) {
1489       cerr << "ERROR: lists_keys_next(): " << cpp_strerror(-ret) << std::endl;
1490       return ret;
1491     } if (ret != -ENOENT) {
1492       // partition the list of buckets by buckets as the listing is un sorted,
1493       // since it would minimize the reads to bucket_info
1494       std::unordered_map<std::string, std::vector<std::string>> bucket_instance_map;
1495       for (auto &key: keys) {
1496         auto pos = key.find(':');
1497         if(pos != std::string::npos)
1498           bucket_instance_map[key.substr(0,pos)].emplace_back(std::move(key));
1499       }
1500       for (const auto& kv: bucket_instance_map) {
1501         bucket_instance_ls stale_lst;
1502         get_stale_instances(store, kv.first, kv.second, stale_lst, dpp);
1503         process_f(stale_lst, formatter, store);
1504       }
1505     }
1506   } while (truncated);
1507
1508   return 0;
1509 }
1510
1511 int RGWBucketAdminOp::list_stale_instances(rgw::sal::Store* store,
1512                                            RGWBucketAdminOpState& op_state,
1513                                            RGWFormatterFlusher& flusher,
1514                                            const DoutPrefixProvider *dpp)
1515 {
1516   auto process_f = [](const bucket_instance_ls& lst,
1517                       Formatter *formatter,
1518                       rgw::sal::Store*){
1519                      for (const auto& binfo: lst)
1520                        formatter->dump_string("key", binfo.bucket.get_key());
1521                    };
1522   return process_stale_instances(store, op_state, flusher, dpp, process_f);
1523 }
1524
1525
1526 int RGWBucketAdminOp::clear_stale_instances(rgw::sal::Store* store,
1527                                             RGWBucketAdminOpState& op_state,
1528                                             RGWFormatterFlusher& flusher,
1529                                             const DoutPrefixProvider *dpp)
1530 {
1531   auto process_f = [dpp](const bucket_instance_ls& lst,
1532                       Formatter *formatter,
1533                       rgw::sal::Store* store){
1534                      for (const auto &binfo: lst) {
1535                        int ret = purge_bucket_instance(store, binfo, dpp);
1536                        if (ret == 0){
1537                          auto md_key = "bucket.instance:" + binfo.bucket.get_key();
1538                          ret = store->meta_remove(dpp, md_key, null_yield);
1539                        }
1540                        formatter->open_object_section("delete_status");
1541                        formatter->dump_string("bucket_instance", binfo.bucket.get_key());
1542                        formatter->dump_int("status", -ret);
1543                        formatter->close_section();
1544                      }
1545                    };
1546
1547   return process_stale_instances(store, op_state, flusher, dpp, process_f);
1548 }
1549
1550 static int fix_single_bucket_lc(rgw::sal::Store* store,
1551                                 const std::string& tenant_name,
1552                                 const std::string& bucket_name,
1553                                 const DoutPrefixProvider *dpp)
1554 {
1555   std::unique_ptr<rgw::sal::Bucket> bucket;
1556   int ret = store->get_bucket(dpp, nullptr, tenant_name, bucket_name, &bucket, null_yield);
1557   if (ret < 0) {
1558     // TODO: Should we handle the case where the bucket could've been removed between
1559     // listing and fetching?
1560     return ret;
1561   }
1562
1563   return rgw::lc::fix_lc_shard_entry(dpp, store, store->get_rgwlc()->get_lc(), bucket.get());
1564 }
1565
1566 static void format_lc_status(Formatter* formatter,
1567                              const std::string& tenant_name,
1568                              const std::string& bucket_name,
1569                              int status)
1570 {
1571   formatter->open_object_section("bucket_entry");
1572   std::string entry = tenant_name.empty() ? bucket_name : tenant_name + "/" + bucket_name;
1573   formatter->dump_string("bucket", entry);
1574   formatter->dump_int("status", status);
1575   formatter->close_section(); // bucket_entry
1576 }
1577
1578 static void process_single_lc_entry(rgw::sal::Store* store,
1579                                     Formatter *formatter,
1580                                     const std::string& tenant_name,
1581                                     const std::string& bucket_name,
1582                                     const DoutPrefixProvider *dpp)
1583 {
1584   int ret = fix_single_bucket_lc(store, tenant_name, bucket_name, dpp);
1585   format_lc_status(formatter, tenant_name, bucket_name, -ret);
1586 }
1587
1588 int RGWBucketAdminOp::fix_lc_shards(rgw::sal::Store* store,
1589                                     RGWBucketAdminOpState& op_state,
1590                                     RGWFormatterFlusher& flusher,
1591                                     const DoutPrefixProvider *dpp)
1592 {
1593   std::string marker;
1594   void *handle;
1595   Formatter *formatter = flusher.get_formatter();
1596   static constexpr auto default_max_keys = 1000;
1597
1598   bool truncated;
1599   if (const std::string& bucket_name = op_state.get_bucket_name();
1600       ! bucket_name.empty()) {
1601     const rgw_user user_id = op_state.get_user_id();
1602     process_single_lc_entry(store, formatter, user_id.tenant, bucket_name, dpp);
1603     formatter->flush(cout);
1604   } else {
1605     int ret = store->meta_list_keys_init(dpp, "bucket", marker, &handle);
1606     if (ret < 0) {
1607       std::cerr << "ERROR: can't get key: " << cpp_strerror(-ret) << std::endl;
1608       return ret;
1609     }
1610
1611     {
1612       formatter->open_array_section("lc_fix_status");
1613       auto sg = make_scope_guard([&store, &handle, &formatter](){
1614                                    store->meta_list_keys_complete(handle);
1615                                    formatter->close_section(); // lc_fix_status
1616                                    formatter->flush(cout);
1617                                  });
1618       do {
1619         list<std::string> keys;
1620         ret = store->meta_list_keys_next(dpp, handle, default_max_keys, keys, &truncated);
1621         if (ret < 0 && ret != -ENOENT) {
1622           std::cerr << "ERROR: lists_keys_next(): " << cpp_strerror(-ret) << std::endl;
1623           return ret;
1624         } if (ret != -ENOENT) {
1625           for (const auto &key:keys) {
1626             auto [tenant_name, bucket_name] = split_tenant(key);
1627             process_single_lc_entry(store, formatter, tenant_name, bucket_name, dpp);
1628           }
1629         }
1630         formatter->flush(cout); // regularly flush every 1k entries
1631       } while (truncated);
1632     }
1633
1634   }
1635   return 0;
1636
1637 }
1638
1639 static bool has_object_expired(const DoutPrefixProvider *dpp,
1640                                rgw::sal::Store* store,
1641                                rgw::sal::Bucket* bucket,
1642                                const rgw_obj_key& key, utime_t& delete_at)
1643 {
1644   std::unique_ptr<rgw::sal::Object> obj = bucket->get_object(key);
1645   bufferlist delete_at_bl;
1646
1647   int ret = rgw_object_get_attr(dpp, store, obj.get(), RGW_ATTR_DELETE_AT, delete_at_bl, null_yield);
1648   if (ret < 0) {
1649     return false;  // no delete at attr, proceed
1650   }
1651
1652   ret = decode_bl(delete_at_bl, delete_at);
1653   if (ret < 0) {
1654     return false;  // failed to parse
1655   }
1656
1657   if (delete_at <= ceph_clock_now() && !delete_at.is_zero()) {
1658     return true;
1659   }
1660
1661   return false;
1662 }
1663
1664 static int fix_bucket_obj_expiry(const DoutPrefixProvider *dpp,
1665                                  rgw::sal::Store* store,
1666                                  rgw::sal::Bucket* bucket,
1667                                  RGWFormatterFlusher& flusher, bool dry_run)
1668 {
1669   if (bucket->get_key().bucket_id == bucket->get_key().marker) {
1670     ldpp_dout(dpp, -1) << "Not a resharded bucket skipping" << dendl;
1671     return 0;  // not a resharded bucket, move along
1672   }
1673
1674   Formatter *formatter = flusher.get_formatter();
1675   formatter->open_array_section("expired_deletion_status");
1676   auto sg = make_scope_guard([&formatter] {
1677                                formatter->close_section();
1678                                formatter->flush(std::cout);
1679                              });
1680
1681   rgw::sal::Bucket::ListParams params;
1682   rgw::sal::Bucket::ListResults results;
1683
1684   params.list_versions = bucket->versioned();
1685   params.allow_unordered = true;
1686
1687   do {
1688     int ret = bucket->list(dpp, params, listing_max_entries, results, null_yield);
1689     if (ret < 0) {
1690       ldpp_dout(dpp, -1) << "ERROR failed to list objects in the bucket" << dendl;
1691       return ret;
1692     }
1693     for (const auto& obj : results.objs) {
1694       rgw_obj_key key(obj.key);
1695       utime_t delete_at;
1696       if (has_object_expired(dpp, store, bucket, key, delete_at)) {
1697         formatter->open_object_section("object_status");
1698         formatter->dump_string("object", key.name);
1699         formatter->dump_stream("delete_at") << delete_at;
1700
1701         if (!dry_run) {
1702           ret = rgw_remove_object(dpp, store, bucket, key);
1703           formatter->dump_int("status", ret);
1704         }
1705
1706         formatter->close_section();  // object_status
1707       }
1708     }
1709     formatter->flush(cout); // regularly flush every 1k entries
1710   } while (results.is_truncated);
1711
1712   return 0;
1713 }
1714
1715 int RGWBucketAdminOp::fix_obj_expiry(rgw::sal::Store* store,
1716                                      RGWBucketAdminOpState& op_state,
1717                                      RGWFormatterFlusher& flusher,
1718                                      const DoutPrefixProvider *dpp, bool dry_run)
1719 {
1720   RGWBucket admin_bucket;
1721   int ret = admin_bucket.init(store, op_state, null_yield, dpp);
1722   if (ret < 0) {
1723     ldpp_dout(dpp, -1) << "failed to initialize bucket" << dendl;
1724     return ret;
1725   }
1726   std::unique_ptr<rgw::sal::Bucket> bucket;
1727   ret = store->get_bucket(nullptr, admin_bucket.get_bucket_info(), &bucket);
1728   if (ret < 0) {
1729     return ret;
1730   }
1731
1732   return fix_bucket_obj_expiry(dpp, store, bucket.get(), flusher, dry_run);
1733 }
1734
1735 void RGWBucketCompleteInfo::dump(Formatter *f) const {
1736   encode_json("bucket_info", info, f);
1737   encode_json("attrs", attrs, f);
1738 }
1739
1740 void RGWBucketCompleteInfo::decode_json(JSONObj *obj) {
1741   JSONDecoder::decode_json("bucket_info", info, obj);
1742   JSONDecoder::decode_json("attrs", attrs, obj);
1743 }
1744
1745 class RGWBucketMetadataHandler : public RGWBucketMetadataHandlerBase {
1746 public:
1747   struct Svc {
1748     RGWSI_Bucket *bucket{nullptr};
1749   } svc;
1750
1751   struct Ctl {
1752     RGWBucketCtl *bucket{nullptr};
1753   } ctl;
1754
1755   RGWBucketMetadataHandler() {}
1756
1757   void init(RGWSI_Bucket *bucket_svc,
1758             RGWBucketCtl *bucket_ctl) override {
1759     base_init(bucket_svc->ctx(),
1760               bucket_svc->get_ep_be_handler().get());
1761     svc.bucket = bucket_svc;
1762     ctl.bucket = bucket_ctl;
1763   }
1764
1765   string get_type() override { return "bucket"; }
1766
1767   RGWMetadataObject *get_meta_obj(JSONObj *jo, const obj_version& objv, const ceph::real_time& mtime) override {
1768     RGWBucketEntryPoint be;
1769
1770     try {
1771       decode_json_obj(be, jo);
1772     } catch (JSONDecoder::err& e) {
1773       return nullptr;
1774     }
1775
1776     return new RGWBucketEntryMetadataObject(be, objv, mtime);
1777   }
1778
1779   int do_get(RGWSI_MetaBackend_Handler::Op *op, string& entry, RGWMetadataObject **obj, optional_yield y, const DoutPrefixProvider *dpp) override {
1780     RGWObjVersionTracker ot;
1781     RGWBucketEntryPoint be;
1782
1783     real_time mtime;
1784     map<string, bufferlist> attrs;
1785
1786     RGWSI_Bucket_EP_Ctx ctx(op->ctx());
1787
1788     int ret = svc.bucket->read_bucket_entrypoint_info(ctx, entry, &be, &ot, &mtime, &attrs, y, dpp);
1789     if (ret < 0)
1790       return ret;
1791
1792     RGWBucketEntryMetadataObject *mdo = new RGWBucketEntryMetadataObject(be, ot.read_version, mtime, std::move(attrs));
1793
1794     *obj = mdo;
1795
1796     return 0;
1797   }
1798
1799   int do_put(RGWSI_MetaBackend_Handler::Op *op, string& entry,
1800              RGWMetadataObject *obj,
1801              RGWObjVersionTracker& objv_tracker,
1802              optional_yield y,
1803              const DoutPrefixProvider *dpp,
1804              RGWMDLogSyncType type, bool from_remote_zone) override;
1805
1806   int do_remove(RGWSI_MetaBackend_Handler::Op *op, string& entry, RGWObjVersionTracker& objv_tracker,
1807                 optional_yield y, const DoutPrefixProvider *dpp) override {
1808     RGWBucketEntryPoint be;
1809
1810     real_time orig_mtime;
1811
1812     RGWSI_Bucket_EP_Ctx ctx(op->ctx());
1813
1814     int ret = svc.bucket->read_bucket_entrypoint_info(ctx, entry, &be, &objv_tracker, &orig_mtime, nullptr, y, dpp);
1815     if (ret < 0)
1816       return ret;
1817
1818     /*
1819      * We're unlinking the bucket but we don't want to update the entrypoint here - we're removing
1820      * it immediately and don't want to invalidate our cached objv_version or the bucket obj removal
1821      * will incorrectly fail.
1822      */
1823     ret = ctl.bucket->unlink_bucket(be.owner, be.bucket, y, dpp, false);
1824     if (ret < 0) {
1825       ldpp_dout(dpp, -1) << "could not unlink bucket=" << entry << " owner=" << be.owner << dendl;
1826     }
1827
1828     ret = svc.bucket->remove_bucket_entrypoint_info(ctx, entry, &objv_tracker, y, dpp);
1829     if (ret < 0) {
1830       ldpp_dout(dpp, -1) << "could not delete bucket=" << entry << dendl;
1831     }
1832     /* idempotent */
1833     return 0;
1834   }
1835
1836   int call(std::function<int(RGWSI_Bucket_EP_Ctx& ctx)> f) {
1837     return call(nullopt, f);
1838   }
1839
1840   int call(std::optional<RGWSI_MetaBackend_CtxParams> bectx_params,
1841            std::function<int(RGWSI_Bucket_EP_Ctx& ctx)> f) {
1842     return be_handler->call(bectx_params, [&](RGWSI_MetaBackend_Handler::Op *op) {
1843       RGWSI_Bucket_EP_Ctx ctx(op->ctx());
1844       return f(ctx);
1845     });
1846   }
1847 };
1848
1849 class RGWMetadataHandlerPut_Bucket : public RGWMetadataHandlerPut_SObj
1850 {
1851   RGWBucketMetadataHandler *bhandler;
1852   RGWBucketEntryMetadataObject *obj;
1853 public:
1854   RGWMetadataHandlerPut_Bucket(RGWBucketMetadataHandler *_handler,
1855                                RGWSI_MetaBackend_Handler::Op *op, string& entry,
1856                                RGWMetadataObject *_obj, RGWObjVersionTracker& objv_tracker,
1857                                optional_yield y,
1858                                RGWMDLogSyncType type, bool from_remote_zone) : RGWMetadataHandlerPut_SObj(_handler, op, entry, obj, objv_tracker, y, type, from_remote_zone),
1859                                                         bhandler(_handler) {
1860     obj = static_cast<RGWBucketEntryMetadataObject *>(_obj);
1861   }
1862   ~RGWMetadataHandlerPut_Bucket() {}
1863
1864   void encode_obj(bufferlist *bl) override {
1865     obj->get_ep().encode(*bl);
1866   }
1867
1868   int put_checked(const DoutPrefixProvider *dpp) override;
1869   int put_post(const DoutPrefixProvider *dpp) override;
1870 };
1871
1872 int RGWBucketMetadataHandler::do_put(RGWSI_MetaBackend_Handler::Op *op, string& entry,
1873                                      RGWMetadataObject *obj,
1874                                      RGWObjVersionTracker& objv_tracker,
1875                                      optional_yield y,
1876                                      const DoutPrefixProvider *dpp,
1877                                      RGWMDLogSyncType type, bool from_remote_zone)
1878 {
1879   RGWMetadataHandlerPut_Bucket put_op(this, op, entry, obj, objv_tracker, y, type, from_remote_zone);
1880   return do_put_operate(&put_op, dpp);
1881 }
1882
1883 int RGWMetadataHandlerPut_Bucket::put_checked(const DoutPrefixProvider *dpp)
1884 {
1885   RGWBucketEntryMetadataObject *orig_obj = static_cast<RGWBucketEntryMetadataObject *>(old_obj);
1886
1887   if (orig_obj) {
1888     obj->set_pattrs(&orig_obj->get_attrs());
1889   }
1890
1891   auto& be = obj->get_ep();
1892   auto mtime = obj->get_mtime();
1893   auto pattrs = obj->get_pattrs();
1894
1895   RGWSI_Bucket_EP_Ctx ctx(op->ctx());
1896
1897   return bhandler->svc.bucket->store_bucket_entrypoint_info(ctx, entry,
1898                                                            be,
1899                                                            false,
1900                                                            mtime,
1901                                                            pattrs,
1902                                                            &objv_tracker,
1903                                                            y,
1904                                                            dpp);
1905 }
1906
1907 int RGWMetadataHandlerPut_Bucket::put_post(const DoutPrefixProvider *dpp)
1908 {
1909   auto& be = obj->get_ep();
1910
1911   int ret;
1912
1913   /* link bucket */
1914   if (be.linked) {
1915     ret = bhandler->ctl.bucket->link_bucket(be.owner, be.bucket, be.creation_time, y, dpp, false);
1916   } else {
1917     ret = bhandler->ctl.bucket->unlink_bucket(be.owner, be.bucket, y, dpp, false);
1918   }
1919
1920   return ret;
1921 }
1922
1923 static void get_md5_digest(const RGWBucketEntryPoint *be, string& md5_digest) {
1924
1925    char md5[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 1];
1926    unsigned char m[CEPH_CRYPTO_MD5_DIGESTSIZE];
1927    bufferlist bl;
1928
1929    Formatter *f = new JSONFormatter(false);
1930    be->dump(f);
1931    f->flush(bl);
1932
1933    MD5 hash;
1934    // Allow use of MD5 digest in FIPS mode for non-cryptographic purposes
1935    hash.SetFlags(EVP_MD_CTX_FLAG_NON_FIPS_ALLOW);
1936    hash.Update((const unsigned char *)bl.c_str(), bl.length());
1937    hash.Final(m);
1938
1939    buf_to_hex(m, CEPH_CRYPTO_MD5_DIGESTSIZE, md5);
1940
1941    delete f;
1942
1943    md5_digest = md5;
1944 }
1945
1946 #define ARCHIVE_META_ATTR RGW_ATTR_PREFIX "zone.archive.info"
1947
1948 struct archive_meta_info {
1949   rgw_bucket orig_bucket;
1950
1951   bool from_attrs(CephContext *cct, map<string, bufferlist>& attrs) {
1952     auto iter = attrs.find(ARCHIVE_META_ATTR);
1953     if (iter == attrs.end()) {
1954       return false;
1955     }
1956
1957     auto bliter = iter->second.cbegin();
1958     try {
1959       decode(bliter);
1960     } catch (buffer::error& err) {
1961       ldout(cct, 0) << "ERROR: failed to decode archive meta info" << dendl;
1962       return false;
1963     }
1964
1965     return true;
1966   }
1967
1968   void store_in_attrs(map<string, bufferlist>& attrs) const {
1969     encode(attrs[ARCHIVE_META_ATTR]);
1970   }
1971
1972   void encode(bufferlist& bl) const {
1973     ENCODE_START(1, 1, bl);
1974     encode(orig_bucket, bl);
1975     ENCODE_FINISH(bl);
1976   }
1977
1978   void decode(bufferlist::const_iterator& bl) {
1979      DECODE_START(1, bl);
1980      decode(orig_bucket, bl);
1981      DECODE_FINISH(bl);
1982   }
1983 };
1984 WRITE_CLASS_ENCODER(archive_meta_info)
1985
1986 class RGWArchiveBucketMetadataHandler : public RGWBucketMetadataHandler {
1987 public:
1988   RGWArchiveBucketMetadataHandler() {}
1989
1990   int do_remove(RGWSI_MetaBackend_Handler::Op *op, string& entry, RGWObjVersionTracker& objv_tracker,
1991                 optional_yield y, const DoutPrefixProvider *dpp) override {
1992     auto cct = svc.bucket->ctx();
1993
1994     RGWSI_Bucket_EP_Ctx ctx(op->ctx());
1995
1996     ldpp_dout(dpp, 5) << "SKIP: bucket removal is not allowed on archive zone: bucket:" << entry << " ... proceeding to rename" << dendl;
1997
1998     string tenant_name, bucket_name;
1999     parse_bucket(entry, &tenant_name, &bucket_name);
2000     rgw_bucket entry_bucket;
2001     entry_bucket.tenant = tenant_name;
2002     entry_bucket.name = bucket_name;
2003
2004     real_time mtime;
2005
2006     /* read original entrypoint */
2007
2008     RGWBucketEntryPoint be;
2009     map<string, bufferlist> attrs;
2010     int ret = svc.bucket->read_bucket_entrypoint_info(ctx, entry, &be, &objv_tracker, &mtime, &attrs, y, dpp);
2011     if (ret < 0) {
2012         return ret;
2013     }
2014
2015     string bi_meta_name = RGWSI_Bucket::get_bi_meta_key(be.bucket);
2016
2017     /* read original bucket instance info */
2018
2019     map<string, bufferlist> attrs_m;
2020     ceph::real_time orig_mtime;
2021     RGWBucketInfo old_bi;
2022
2023     ret = ctl.bucket->read_bucket_instance_info(be.bucket, &old_bi, y, dpp, RGWBucketCtl::BucketInstance::GetParams()
2024                                                                     .set_mtime(&orig_mtime)
2025                                                                     .set_attrs(&attrs_m));
2026     if (ret < 0) {
2027         return ret;
2028     }
2029
2030     archive_meta_info ami;
2031
2032     if (!ami.from_attrs(svc.bucket->ctx(), attrs_m)) {
2033       ami.orig_bucket = old_bi.bucket;
2034       ami.store_in_attrs(attrs_m);
2035     }
2036
2037     /* generate a new bucket instance. We could have avoided this if we could just point a new
2038      * bucket entry point to the old bucket instance, however, due to limitation in the way
2039      * we index buckets under the user, bucket entrypoint and bucket instance of the same
2040      * bucket need to have the same name, so we need to copy the old bucket instance into
2041      * to a new entry with the new name
2042      */
2043
2044     string new_bucket_name;
2045
2046     RGWBucketInfo new_bi = old_bi;
2047     RGWBucketEntryPoint new_be = be;
2048
2049     string md5_digest;
2050
2051     get_md5_digest(&new_be, md5_digest);
2052     new_bucket_name = ami.orig_bucket.name + "-deleted-" + md5_digest;
2053
2054     new_bi.bucket.name = new_bucket_name;
2055     new_bi.objv_tracker.clear();
2056
2057     new_be.bucket.name = new_bucket_name;
2058
2059     ret = ctl.bucket->store_bucket_instance_info(be.bucket, new_bi, y, dpp, RGWBucketCtl::BucketInstance::PutParams()
2060                                                                     .set_exclusive(false)
2061                                                                     .set_mtime(orig_mtime)
2062                                                                     .set_attrs(&attrs_m)
2063                                                                     .set_orig_info(&old_bi));
2064     if (ret < 0) {
2065       ldpp_dout(dpp, 0) << "ERROR: failed to put new bucket instance info for bucket=" << new_bi.bucket << " ret=" << ret << dendl;
2066       return ret;
2067     }
2068
2069     /* store a new entrypoint */
2070
2071     RGWObjVersionTracker ot;
2072     ot.generate_new_write_ver(cct);
2073
2074     ret = svc.bucket->store_bucket_entrypoint_info(ctx, RGWSI_Bucket::get_entrypoint_meta_key(new_be.bucket),
2075                                                    new_be, true, mtime, &attrs, nullptr, y, dpp);
2076     if (ret < 0) {
2077       ldpp_dout(dpp, 0) << "ERROR: failed to put new bucket entrypoint for bucket=" << new_be.bucket << " ret=" << ret << dendl;
2078       return ret;
2079     }
2080
2081     /* link new bucket */
2082
2083     ret = ctl.bucket->link_bucket(new_be.owner, new_be.bucket, new_be.creation_time, y, dpp, false);
2084     if (ret < 0) {
2085       ldpp_dout(dpp, 0) << "ERROR: failed to link new bucket for bucket=" << new_be.bucket << " ret=" << ret << dendl;
2086       return ret;
2087     }
2088
2089     /* clean up old stuff */
2090
2091     ret = ctl.bucket->unlink_bucket(be.owner, entry_bucket, y, dpp, false);
2092     if (ret < 0) {
2093         ldpp_dout(dpp, -1) << "could not unlink bucket=" << entry << " owner=" << be.owner << dendl;
2094     }
2095
2096     // if (ret == -ECANCELED) it means that there was a race here, and someone
2097     // wrote to the bucket entrypoint just before we removed it. The question is
2098     // whether it was a newly created bucket entrypoint ...  in which case we
2099     // should ignore the error and move forward, or whether it is a higher version
2100     // of the same bucket instance ... in which we should retry
2101     ret = svc.bucket->remove_bucket_entrypoint_info(ctx,
2102                                                     RGWSI_Bucket::get_entrypoint_meta_key(be.bucket),
2103                                                     &objv_tracker,
2104                                                     y,
2105                                                     dpp);
2106     if (ret < 0) {
2107       ldpp_dout(dpp, 0) << "ERROR: failed to put new bucket entrypoint for bucket=" << new_be.bucket << " ret=" << ret << dendl;
2108       return ret;
2109     }
2110
2111     ret = ctl.bucket->remove_bucket_instance_info(be.bucket, old_bi, y, dpp);
2112     if (ret < 0) {
2113       ldpp_dout(dpp, -1) << "could not delete bucket=" << entry << dendl;
2114     }
2115
2116
2117     /* idempotent */
2118
2119     return 0;
2120   }
2121
2122   int do_put(RGWSI_MetaBackend_Handler::Op *op, string& entry,
2123              RGWMetadataObject *obj,
2124              RGWObjVersionTracker& objv_tracker,
2125              optional_yield y, const DoutPrefixProvider *dpp,
2126              RGWMDLogSyncType type, bool from_remote_zone) override {
2127     if (entry.find("-deleted-") != string::npos) {
2128       RGWObjVersionTracker ot;
2129       RGWMetadataObject *robj;
2130       int ret = do_get(op, entry, &robj, y, dpp);
2131       if (ret != -ENOENT) {
2132         if (ret < 0) {
2133           return ret;
2134         }
2135         ot.read_version = robj->get_version();
2136         delete robj;
2137
2138         ret = do_remove(op, entry, ot, y, dpp);
2139         if (ret < 0) {
2140           return ret;
2141         }
2142       }
2143     }
2144
2145     return RGWBucketMetadataHandler::do_put(op, entry, obj,
2146                                             objv_tracker, y, dpp, type, from_remote_zone);
2147   }
2148
2149 };
2150
2151 class RGWBucketInstanceMetadataHandler : public RGWBucketInstanceMetadataHandlerBase {
2152   int read_bucket_instance_entry(RGWSI_Bucket_BI_Ctx& ctx,
2153                                  const string& entry,
2154                                  RGWBucketCompleteInfo *bi,
2155                                  ceph::real_time *pmtime,
2156                                  optional_yield y,
2157                                  const DoutPrefixProvider *dpp) {
2158     return svc.bucket->read_bucket_instance_info(ctx,
2159                                                  entry,
2160                                                  &bi->info,
2161                                                  pmtime, &bi->attrs,
2162                                                  y,
2163                                                  dpp);
2164   }
2165
2166 public:
2167   struct Svc {
2168     RGWSI_Zone *zone{nullptr};
2169     RGWSI_Bucket *bucket{nullptr};
2170     RGWSI_BucketIndex *bi{nullptr};
2171   } svc;
2172
2173   RGWBucketInstanceMetadataHandler() {}
2174
2175   void init(RGWSI_Zone *zone_svc,
2176             RGWSI_Bucket *bucket_svc,
2177             RGWSI_BucketIndex *bi_svc) override {
2178     base_init(bucket_svc->ctx(),
2179               bucket_svc->get_bi_be_handler().get());
2180     svc.zone = zone_svc;
2181     svc.bucket = bucket_svc;
2182     svc.bi = bi_svc;
2183   }
2184
2185   string get_type() override { return "bucket.instance"; }
2186
2187   RGWMetadataObject *get_meta_obj(JSONObj *jo, const obj_version& objv, const ceph::real_time& mtime) override {
2188     RGWBucketCompleteInfo bci;
2189
2190     try {
2191       decode_json_obj(bci, jo);
2192     } catch (JSONDecoder::err& e) {
2193       return nullptr;
2194     }
2195
2196     return new RGWBucketInstanceMetadataObject(bci, objv, mtime);
2197   }
2198
2199   int do_get(RGWSI_MetaBackend_Handler::Op *op, string& entry, RGWMetadataObject **obj, optional_yield y, const DoutPrefixProvider *dpp) override {
2200     RGWBucketCompleteInfo bci;
2201     real_time mtime;
2202
2203     RGWSI_Bucket_BI_Ctx ctx(op->ctx());
2204
2205     int ret = svc.bucket->read_bucket_instance_info(ctx, entry, &bci.info, &mtime, &bci.attrs, y, dpp);
2206     if (ret < 0)
2207       return ret;
2208
2209     RGWBucketInstanceMetadataObject *mdo = new RGWBucketInstanceMetadataObject(bci, bci.info.objv_tracker.read_version, mtime);
2210
2211     *obj = mdo;
2212
2213     return 0;
2214   }
2215
2216   int do_put(RGWSI_MetaBackend_Handler::Op *op, string& entry,
2217              RGWMetadataObject *_obj, RGWObjVersionTracker& objv_tracker,
2218              optional_yield y, const DoutPrefixProvider *dpp,
2219              RGWMDLogSyncType sync_type, bool from_remote_zone) override;
2220
2221   int do_remove(RGWSI_MetaBackend_Handler::Op *op, string& entry, RGWObjVersionTracker& objv_tracker,
2222                 optional_yield y, const DoutPrefixProvider *dpp) override {
2223     RGWBucketCompleteInfo bci;
2224
2225     RGWSI_Bucket_BI_Ctx ctx(op->ctx());
2226
2227     int ret = read_bucket_instance_entry(ctx, entry, &bci, nullptr, y, dpp);
2228     if (ret < 0 && ret != -ENOENT)
2229       return ret;
2230
2231     return svc.bucket->remove_bucket_instance_info(ctx, entry, bci.info, &bci.info.objv_tracker, y, dpp);
2232   }
2233
2234   int call(std::function<int(RGWSI_Bucket_BI_Ctx& ctx)> f) {
2235     return call(nullopt, f);
2236   }
2237
2238   int call(std::optional<RGWSI_MetaBackend_CtxParams> bectx_params,
2239            std::function<int(RGWSI_Bucket_BI_Ctx& ctx)> f) {
2240     return be_handler->call(bectx_params, [&](RGWSI_MetaBackend_Handler::Op *op) {
2241       RGWSI_Bucket_BI_Ctx ctx(op->ctx());
2242       return f(ctx);
2243     });
2244   }
2245 };
2246
2247 class RGWMetadataHandlerPut_BucketInstance : public RGWMetadataHandlerPut_SObj
2248 {
2249   CephContext *cct;
2250   RGWBucketInstanceMetadataHandler *bihandler;
2251   RGWBucketInstanceMetadataObject *obj;
2252 public:
2253   RGWMetadataHandlerPut_BucketInstance(CephContext *_cct,
2254                                        RGWBucketInstanceMetadataHandler *_handler,
2255                                        RGWSI_MetaBackend_Handler::Op *_op, string& entry,
2256                                        RGWMetadataObject *_obj, RGWObjVersionTracker& objv_tracker,
2257                                        optional_yield y,
2258                                        RGWMDLogSyncType type, bool from_remote_zone) : RGWMetadataHandlerPut_SObj(_handler, _op, entry, obj, objv_tracker, y, type, from_remote_zone),
2259                                        cct(_cct), bihandler(_handler) {
2260     obj = static_cast<RGWBucketInstanceMetadataObject *>(_obj);
2261
2262     auto& bci = obj->get_bci();
2263     obj->set_pattrs(&bci.attrs);
2264   }
2265
2266   void encode_obj(bufferlist *bl) override {
2267     obj->get_bucket_info().encode(*bl);
2268   }
2269
2270   int put_check(const DoutPrefixProvider *dpp) override;
2271   int put_checked(const DoutPrefixProvider *dpp) override;
2272   int put_post(const DoutPrefixProvider *dpp) override;
2273 };
2274
2275 int RGWBucketInstanceMetadataHandler::do_put(RGWSI_MetaBackend_Handler::Op *op,
2276                                              string& entry,
2277                                              RGWMetadataObject *obj,
2278                                              RGWObjVersionTracker& objv_tracker,
2279                                              optional_yield y,
2280                                              const DoutPrefixProvider *dpp,
2281                                              RGWMDLogSyncType type, bool from_remote_zone)
2282 {
2283   RGWMetadataHandlerPut_BucketInstance put_op(svc.bucket->ctx(), this, op, entry, obj,
2284                                               objv_tracker, y, type, from_remote_zone);
2285   return do_put_operate(&put_op, dpp);
2286 }
2287
2288 void init_default_bucket_layout(CephContext *cct, rgw::BucketLayout& layout,
2289                                 const RGWZone& zone,
2290                                 std::optional<uint32_t> shards,
2291                                 std::optional<rgw::BucketIndexType> type) {
2292   layout.current_index.gen = 0;
2293   layout.current_index.layout.normal.hash_type = rgw::BucketHashType::Mod;
2294
2295   layout.current_index.layout.type =
2296     type.value_or(rgw::BucketIndexType::Normal);
2297
2298   if (shards) {
2299     layout.current_index.layout.normal.num_shards = *shards;
2300   } else if (cct->_conf->rgw_override_bucket_index_max_shards > 0) {
2301     layout.current_index.layout.normal.num_shards =
2302       cct->_conf->rgw_override_bucket_index_max_shards;
2303   } else {
2304     layout.current_index.layout.normal.num_shards =
2305       zone.bucket_index_max_shards;
2306   }
2307
2308   if (layout.current_index.layout.type == rgw::BucketIndexType::Normal) {
2309     layout.logs.push_back(log_layout_from_index(
2310                             layout.current_index.gen,
2311                             layout.current_index.layout.normal));
2312   }
2313 }
2314
2315 int RGWMetadataHandlerPut_BucketInstance::put_check(const DoutPrefixProvider *dpp)
2316 {
2317   int ret;
2318
2319   RGWBucketCompleteInfo& bci = obj->get_bci();
2320
2321   RGWBucketInstanceMetadataObject *orig_obj = static_cast<RGWBucketInstanceMetadataObject *>(old_obj);
2322
2323   RGWBucketCompleteInfo *old_bci = (orig_obj ? &orig_obj->get_bci() : nullptr);
2324
2325   const bool exists = (!!orig_obj);
2326
2327   if (from_remote_zone) {
2328     // don't sync bucket layout changes
2329     if (!exists) {
2330       auto& bci_index = bci.info.layout.current_index.layout;
2331       auto index_type = bci_index.type;
2332       auto num_shards = bci_index.normal.num_shards;
2333       init_default_bucket_layout(cct, bci.info.layout,
2334                                  bihandler->svc.zone->get_zone(),
2335                                  num_shards, index_type);
2336     } else {
2337       bci.info.layout = old_bci->info.layout;
2338     }
2339   }
2340
2341   if (!exists || old_bci->info.bucket.bucket_id != bci.info.bucket.bucket_id) {
2342     /* a new bucket, we need to select a new bucket placement for it */
2343     string tenant_name;
2344     string bucket_name;
2345     string bucket_instance;
2346     parse_bucket(entry, &tenant_name, &bucket_name, &bucket_instance);
2347
2348     RGWZonePlacementInfo rule_info;
2349     bci.info.bucket.name = bucket_name;
2350     bci.info.bucket.bucket_id = bucket_instance;
2351     bci.info.bucket.tenant = tenant_name;
2352     // if the sync module never writes data, don't require the zone to specify all placement targets
2353     if (bihandler->svc.zone->sync_module_supports_writes()) {
2354       ret = bihandler->svc.zone->select_bucket_location_by_rule(dpp, bci.info.placement_rule, &rule_info, y);
2355       if (ret < 0) {
2356         ldpp_dout(dpp, 0) << "ERROR: select_bucket_placement() returned " << ret << dendl;
2357         return ret;
2358       }
2359     }
2360     bci.info.layout.current_index.layout.type = rule_info.index_type;
2361   } else {
2362     /* existing bucket, keep its placement */
2363     bci.info.bucket.explicit_placement = old_bci->info.bucket.explicit_placement;
2364     bci.info.placement_rule = old_bci->info.placement_rule;
2365   }
2366
2367   /* record the read version (if any), store the new version */
2368   bci.info.objv_tracker.read_version = objv_tracker.read_version;
2369   bci.info.objv_tracker.write_version = objv_tracker.write_version;
2370
2371   return 0;
2372 }
2373
2374 int RGWMetadataHandlerPut_BucketInstance::put_checked(const DoutPrefixProvider *dpp)
2375 {
2376   RGWBucketInstanceMetadataObject *orig_obj = static_cast<RGWBucketInstanceMetadataObject *>(old_obj);
2377
2378   RGWBucketInfo *orig_info = (orig_obj ? &orig_obj->get_bucket_info() : nullptr);
2379
2380   auto& info = obj->get_bucket_info();
2381   auto mtime = obj->get_mtime();
2382   auto pattrs = obj->get_pattrs();
2383
2384   RGWSI_Bucket_BI_Ctx ctx(op->ctx());
2385
2386   return bihandler->svc.bucket->store_bucket_instance_info(ctx,
2387                                                          entry,
2388                                                          info,
2389                                                          orig_info,
2390                                                          false,
2391                                                          mtime,
2392                                                          pattrs,
2393                                                          y,
2394                                                          dpp);
2395 }
2396
2397 int RGWMetadataHandlerPut_BucketInstance::put_post(const DoutPrefixProvider *dpp)
2398 {
2399   RGWBucketCompleteInfo& bci = obj->get_bci();
2400
2401   objv_tracker = bci.info.objv_tracker;
2402
2403   int ret = bihandler->svc.bi->init_index(dpp, bci.info);
2404   if (ret < 0) {
2405     return ret;
2406   }
2407
2408   return STATUS_APPLIED;
2409 }
2410
2411 class RGWArchiveBucketInstanceMetadataHandler : public RGWBucketInstanceMetadataHandler {
2412 public:
2413   RGWArchiveBucketInstanceMetadataHandler() {}
2414
2415   int do_remove(RGWSI_MetaBackend_Handler::Op *op, string& entry, RGWObjVersionTracker& objv_tracker, optional_yield y, const DoutPrefixProvider *dpp) override {
2416     ldpp_dout(dpp, 0) << "SKIP: bucket instance removal is not allowed on archive zone: bucket.instance:" << entry << dendl;
2417     return 0;
2418   }
2419 };
2420
2421 RGWBucketCtl::RGWBucketCtl(RGWSI_Zone *zone_svc,
2422                            RGWSI_Bucket *bucket_svc,
2423                            RGWSI_Bucket_Sync *bucket_sync_svc,
2424                            RGWSI_BucketIndex *bi_svc) : cct(zone_svc->ctx())
2425 {
2426   svc.zone = zone_svc;
2427   svc.bucket = bucket_svc;
2428   svc.bucket_sync = bucket_sync_svc;
2429   svc.bi = bi_svc;
2430 }
2431
2432 void RGWBucketCtl::init(RGWUserCtl *user_ctl,
2433                         RGWBucketMetadataHandler *_bm_handler,
2434                         RGWBucketInstanceMetadataHandler *_bmi_handler,
2435                         RGWDataChangesLog *datalog,
2436                         const DoutPrefixProvider *dpp)
2437 {
2438   ctl.user = user_ctl;
2439
2440   bm_handler = _bm_handler;
2441   bmi_handler = _bmi_handler;
2442
2443   bucket_be_handler = bm_handler->get_be_handler();
2444   bi_be_handler = bmi_handler->get_be_handler();
2445
2446   datalog->set_bucket_filter(
2447     [this](const rgw_bucket& bucket, optional_yield y, const DoutPrefixProvider *dpp) {
2448       return bucket_exports_data(bucket, y, dpp);
2449     });
2450 }
2451
2452 int RGWBucketCtl::call(std::function<int(RGWSI_Bucket_X_Ctx& ctx)> f) {
2453   return bm_handler->call([&](RGWSI_Bucket_EP_Ctx& ep_ctx) {
2454     return bmi_handler->call([&](RGWSI_Bucket_BI_Ctx& bi_ctx) {
2455       RGWSI_Bucket_X_Ctx ctx{ep_ctx, bi_ctx};
2456       return f(ctx);
2457     });
2458   });
2459 }
2460
2461 int RGWBucketCtl::read_bucket_entrypoint_info(const rgw_bucket& bucket,
2462                                               RGWBucketEntryPoint *info,
2463                                               optional_yield y, const DoutPrefixProvider *dpp,
2464                                               const Bucket::GetParams& params)
2465 {
2466   return bm_handler->call(params.bectx_params, [&](RGWSI_Bucket_EP_Ctx& ctx) {
2467     return svc.bucket->read_bucket_entrypoint_info(ctx,
2468                                                    RGWSI_Bucket::get_entrypoint_meta_key(bucket),
2469                                                    info,
2470                                                    params.objv_tracker,
2471                                                    params.mtime,
2472                                                    params.attrs,
2473                                                    y,
2474                                                    dpp,
2475                                                    params.cache_info,
2476                                                    params.refresh_version);
2477   });
2478 }
2479
2480 int RGWBucketCtl::store_bucket_entrypoint_info(const rgw_bucket& bucket,
2481                                                RGWBucketEntryPoint& info,
2482                                                optional_yield y,
2483                                                const DoutPrefixProvider *dpp,
2484                                                const Bucket::PutParams& params)
2485 {
2486   return bm_handler->call([&](RGWSI_Bucket_EP_Ctx& ctx) {
2487     return svc.bucket->store_bucket_entrypoint_info(ctx,
2488                                                     RGWSI_Bucket::get_entrypoint_meta_key(bucket),
2489                                                     info,
2490                                                     params.exclusive,
2491                                                     params.mtime,
2492                                                     params.attrs,
2493                                                     params.objv_tracker,
2494                                                     y,
2495                                                     dpp);
2496   });
2497 }
2498
2499 int RGWBucketCtl::remove_bucket_entrypoint_info(const rgw_bucket& bucket,
2500                                                 optional_yield y,
2501                                                 const DoutPrefixProvider *dpp,
2502                                                 const Bucket::RemoveParams& params)
2503 {
2504   return bm_handler->call([&](RGWSI_Bucket_EP_Ctx& ctx) {
2505     return svc.bucket->remove_bucket_entrypoint_info(ctx,
2506                                                      RGWSI_Bucket::get_entrypoint_meta_key(bucket),
2507                                                      params.objv_tracker,
2508                                                      y,
2509                                                      dpp);
2510   });
2511 }
2512
2513 int RGWBucketCtl::read_bucket_instance_info(const rgw_bucket& bucket,
2514                                             RGWBucketInfo *info,
2515                                             optional_yield y,
2516                                             const DoutPrefixProvider *dpp,
2517                                             const BucketInstance::GetParams& params)
2518 {
2519   int ret = bmi_handler->call(params.bectx_params, [&](RGWSI_Bucket_BI_Ctx& ctx) {
2520     return svc.bucket->read_bucket_instance_info(ctx,
2521                                                  RGWSI_Bucket::get_bi_meta_key(bucket),
2522                                                  info,
2523                                                  params.mtime,
2524                                                  params.attrs,
2525                                                  y,
2526                                                  dpp,
2527                                                  params.cache_info,
2528                                                  params.refresh_version);
2529   });
2530
2531   if (ret < 0) {
2532     return ret;
2533   }
2534
2535   if (params.objv_tracker) {
2536     *params.objv_tracker = info->objv_tracker;
2537   }
2538
2539   return 0;
2540 }
2541
2542 int RGWBucketCtl::read_bucket_info(const rgw_bucket& bucket,
2543                                    RGWBucketInfo *info,
2544                                    optional_yield y,
2545                                    const DoutPrefixProvider *dpp,
2546                                    const BucketInstance::GetParams& params,
2547                                    RGWObjVersionTracker *ep_objv_tracker)
2548 {
2549   const rgw_bucket *b = &bucket;
2550
2551   std::optional<RGWBucketEntryPoint> ep;
2552
2553   if (b->bucket_id.empty()) {
2554     ep.emplace();
2555
2556     int r = read_bucket_entrypoint_info(*b, &(*ep), y, dpp, RGWBucketCtl::Bucket::GetParams()
2557                                                     .set_bectx_params(params.bectx_params)
2558                                                     .set_objv_tracker(ep_objv_tracker));
2559     if (r < 0) {
2560       return r;
2561     }
2562
2563     b = &ep->bucket;
2564   }
2565
2566   int ret = bmi_handler->call(params.bectx_params, [&](RGWSI_Bucket_BI_Ctx& ctx) {
2567     return svc.bucket->read_bucket_instance_info(ctx,
2568                                                  RGWSI_Bucket::get_bi_meta_key(*b),
2569                                                  info,
2570                                                  params.mtime,
2571                                                  params.attrs,
2572                                                  y, dpp,
2573                                                  params.cache_info,
2574                                                  params.refresh_version);
2575   });
2576
2577   if (ret < 0) {
2578     return ret;
2579   }
2580
2581   if (params.objv_tracker) {
2582     *params.objv_tracker = info->objv_tracker;
2583   }
2584
2585   return 0;
2586 }
2587
2588 int RGWBucketCtl::do_store_bucket_instance_info(RGWSI_Bucket_BI_Ctx& ctx,
2589                                                 const rgw_bucket& bucket,
2590                                                 RGWBucketInfo& info,
2591                                                 optional_yield y,
2592                                                 const DoutPrefixProvider *dpp,
2593                                                 const BucketInstance::PutParams& params)
2594 {
2595   if (params.objv_tracker) {
2596     info.objv_tracker = *params.objv_tracker;
2597   }
2598
2599   return svc.bucket->store_bucket_instance_info(ctx,
2600                                                 RGWSI_Bucket::get_bi_meta_key(bucket),
2601                                                 info,
2602                                                 params.orig_info,
2603                                                 params.exclusive,
2604                                                 params.mtime,
2605                                                 params.attrs,
2606                                                 y,
2607                                                 dpp);
2608 }
2609
2610 int RGWBucketCtl::store_bucket_instance_info(const rgw_bucket& bucket,
2611                                             RGWBucketInfo& info,
2612                                             optional_yield y,
2613                                             const DoutPrefixProvider *dpp,
2614                                             const BucketInstance::PutParams& params)
2615 {
2616   return bmi_handler->call([&](RGWSI_Bucket_BI_Ctx& ctx) {
2617     return do_store_bucket_instance_info(ctx, bucket, info, y, dpp, params);
2618   });
2619 }
2620
2621 int RGWBucketCtl::remove_bucket_instance_info(const rgw_bucket& bucket,
2622                                               RGWBucketInfo& info,
2623                                               optional_yield y,
2624                                               const DoutPrefixProvider *dpp,
2625                                               const BucketInstance::RemoveParams& params)
2626 {
2627   if (params.objv_tracker) {
2628     info.objv_tracker = *params.objv_tracker;
2629   }
2630
2631   return bmi_handler->call([&](RGWSI_Bucket_BI_Ctx& ctx) {
2632     return svc.bucket->remove_bucket_instance_info(ctx,
2633                                                    RGWSI_Bucket::get_bi_meta_key(bucket),
2634                                                    info,
2635                                                    &info.objv_tracker,
2636                                                    y,
2637                                                    dpp);
2638   });
2639 }
2640
2641 int RGWBucketCtl::do_store_linked_bucket_info(RGWSI_Bucket_X_Ctx& ctx,
2642                                               RGWBucketInfo& info,
2643                                               RGWBucketInfo *orig_info,
2644                                               bool exclusive, real_time mtime,
2645                                               obj_version *pep_objv,
2646                                               map<string, bufferlist> *pattrs,
2647                                               bool create_entry_point,
2648                                               optional_yield y, const DoutPrefixProvider *dpp)
2649 {
2650   bool create_head = !info.has_instance_obj || create_entry_point;
2651
2652   int ret = svc.bucket->store_bucket_instance_info(ctx.bi,
2653                                                    RGWSI_Bucket::get_bi_meta_key(info.bucket),
2654                                                    info,
2655                                                    orig_info,
2656                                                    exclusive,
2657                                                    mtime, pattrs,
2658                                                    y, dpp);
2659   if (ret < 0) {
2660     return ret;
2661   }
2662
2663   if (!create_head)
2664     return 0; /* done! */
2665
2666   RGWBucketEntryPoint entry_point;
2667   entry_point.bucket = info.bucket;
2668   entry_point.owner = info.owner;
2669   entry_point.creation_time = info.creation_time;
2670   entry_point.linked = true;
2671   RGWObjVersionTracker ot;
2672   if (pep_objv && !pep_objv->tag.empty()) {
2673     ot.write_version = *pep_objv;
2674   } else {
2675     ot.generate_new_write_ver(cct);
2676     if (pep_objv) {
2677       *pep_objv = ot.write_version;
2678     }
2679   }
2680   ret = svc.bucket->store_bucket_entrypoint_info(ctx.ep,
2681                                                  RGWSI_Bucket::get_entrypoint_meta_key(info.bucket),
2682                                                  entry_point,
2683                                                  exclusive,
2684                                                  mtime,
2685                                                  pattrs,
2686                                                  &ot,
2687                                                  y,
2688                                                  dpp);
2689   if (ret < 0)
2690     return ret;
2691
2692   return 0;
2693 }
2694 int RGWBucketCtl::convert_old_bucket_info(RGWSI_Bucket_X_Ctx& ctx,
2695                                           const rgw_bucket& bucket,
2696                                           optional_yield y,
2697                                           const DoutPrefixProvider *dpp)
2698 {
2699   RGWBucketEntryPoint entry_point;
2700   real_time ep_mtime;
2701   RGWObjVersionTracker ot;
2702   map<string, bufferlist> attrs;
2703   RGWBucketInfo info;
2704   auto cct = svc.bucket->ctx();
2705
2706   ldpp_dout(dpp, 10) << "RGWRados::convert_old_bucket_info(): bucket=" << bucket << dendl;
2707
2708   int ret = svc.bucket->read_bucket_entrypoint_info(ctx.ep,
2709                                                     RGWSI_Bucket::get_entrypoint_meta_key(bucket),
2710                                                     &entry_point, &ot, &ep_mtime, &attrs, y, dpp);
2711   if (ret < 0) {
2712     ldpp_dout(dpp, 0) << "ERROR: get_bucket_entrypoint_info() returned " << ret << " bucket=" << bucket << dendl;
2713     return ret;
2714   }
2715
2716   if (!entry_point.has_bucket_info) {
2717     /* already converted! */
2718     return 0;
2719   }
2720
2721   info = entry_point.old_bucket_info;
2722
2723   ot.generate_new_write_ver(cct);
2724
2725   ret = do_store_linked_bucket_info(ctx, info, nullptr, false, ep_mtime, &ot.write_version, &attrs, true, y, dpp);
2726   if (ret < 0) {
2727     ldpp_dout(dpp, 0) << "ERROR: failed to put_linked_bucket_info(): " << ret << dendl;
2728     return ret;
2729   }
2730
2731   return 0;
2732 }
2733
2734 int RGWBucketCtl::set_bucket_instance_attrs(RGWBucketInfo& bucket_info,
2735                                             map<string, bufferlist>& attrs,
2736                                             RGWObjVersionTracker *objv_tracker,
2737                                             optional_yield y,
2738                                             const DoutPrefixProvider *dpp)
2739 {
2740   return call([&](RGWSI_Bucket_X_Ctx& ctx) {
2741     rgw_bucket& bucket = bucket_info.bucket;
2742
2743     if (!bucket_info.has_instance_obj) {
2744       /* an old bucket object, need to convert it */
2745         int ret = convert_old_bucket_info(ctx, bucket, y, dpp);
2746         if (ret < 0) {
2747           ldpp_dout(dpp, 0) << "ERROR: failed converting old bucket info: " << ret << dendl;
2748           return ret;
2749         }
2750     }
2751
2752     return do_store_bucket_instance_info(ctx.bi,
2753                                          bucket,
2754                                          bucket_info,
2755                                          y,
2756                                          dpp,
2757                                          BucketInstance::PutParams().set_attrs(&attrs)
2758                                                                     .set_objv_tracker(objv_tracker)
2759                                                                     .set_orig_info(&bucket_info));
2760     });
2761 }
2762
2763
2764 int RGWBucketCtl::link_bucket(const rgw_user& user_id,
2765                               const rgw_bucket& bucket,
2766                               ceph::real_time creation_time,
2767                               optional_yield y,
2768                               const DoutPrefixProvider *dpp,
2769                               bool update_entrypoint,
2770                               rgw_ep_info *pinfo)
2771 {
2772   return bm_handler->call([&](RGWSI_Bucket_EP_Ctx& ctx) {
2773     return do_link_bucket(ctx, user_id, bucket, creation_time,
2774                           update_entrypoint, pinfo, y, dpp);
2775   });
2776 }
2777
2778 int RGWBucketCtl::do_link_bucket(RGWSI_Bucket_EP_Ctx& ctx,
2779                                  const rgw_user& user_id,
2780                                  const rgw_bucket& bucket,
2781                                  ceph::real_time creation_time,
2782                                  bool update_entrypoint,
2783                                  rgw_ep_info *pinfo,
2784                                  optional_yield y,
2785                                  const DoutPrefixProvider *dpp)
2786 {
2787   int ret;
2788
2789   RGWBucketEntryPoint ep;
2790   RGWObjVersionTracker ot;
2791   RGWObjVersionTracker& rot = (pinfo) ? pinfo->ep_objv : ot;
2792   map<string, bufferlist> attrs, *pattrs = nullptr;
2793   string meta_key;
2794
2795   if (update_entrypoint) {
2796     meta_key = RGWSI_Bucket::get_entrypoint_meta_key(bucket);
2797     if (pinfo) {
2798       ep = pinfo->ep;
2799       pattrs = &pinfo->attrs;
2800     } else {
2801       ret = svc.bucket->read_bucket_entrypoint_info(ctx,
2802                                                     meta_key,
2803                                                     &ep, &rot,
2804                                                     nullptr, &attrs,
2805                                                     y, dpp);
2806       if (ret < 0 && ret != -ENOENT) {
2807         ldpp_dout(dpp, 0) << "ERROR: store->get_bucket_entrypoint_info() returned: "
2808                       << cpp_strerror(-ret) << dendl;
2809       }
2810       pattrs = &attrs;
2811     }
2812   }
2813
2814   ret = ctl.user->add_bucket(dpp, user_id, bucket, creation_time, y);
2815   if (ret < 0) {
2816     ldpp_dout(dpp, 0) << "ERROR: error adding bucket to user directory:"
2817                   << " user=" << user_id
2818                   << " bucket=" << bucket
2819                   << " err=" << cpp_strerror(-ret)
2820                   << dendl;
2821     goto done_err;
2822   }
2823
2824   if (!update_entrypoint)
2825     return 0;
2826
2827   ep.linked = true;
2828   ep.owner = user_id;
2829   ep.bucket = bucket;
2830   ret = svc.bucket->store_bucket_entrypoint_info(
2831     ctx, meta_key, ep, false, real_time(), pattrs, &rot, y, dpp);
2832   if (ret < 0)
2833     goto done_err;
2834
2835   return 0;
2836
2837 done_err:
2838   int r = do_unlink_bucket(ctx, user_id, bucket, true, y, dpp);
2839   if (r < 0) {
2840     ldpp_dout(dpp, 0) << "ERROR: failed unlinking bucket on error cleanup: "
2841                            << cpp_strerror(-r) << dendl;
2842   }
2843   return ret;
2844 }
2845
2846 int RGWBucketCtl::unlink_bucket(const rgw_user& user_id, const rgw_bucket& bucket, optional_yield y, const DoutPrefixProvider *dpp, bool update_entrypoint)
2847 {
2848   return bm_handler->call([&](RGWSI_Bucket_EP_Ctx& ctx) {
2849     return do_unlink_bucket(ctx, user_id, bucket, update_entrypoint, y, dpp);
2850   });
2851 }
2852
2853 int RGWBucketCtl::do_unlink_bucket(RGWSI_Bucket_EP_Ctx& ctx,
2854                                    const rgw_user& user_id,
2855                                    const rgw_bucket& bucket,
2856                                    bool update_entrypoint,
2857                                    optional_yield y,
2858                                    const DoutPrefixProvider *dpp)
2859 {
2860   int ret = ctl.user->remove_bucket(dpp, user_id, bucket, y);
2861   if (ret < 0) {
2862     ldpp_dout(dpp, 0) << "ERROR: error removing bucket from directory: "
2863         << cpp_strerror(-ret)<< dendl;
2864   }
2865
2866   if (!update_entrypoint)
2867     return 0;
2868
2869   RGWBucketEntryPoint ep;
2870   RGWObjVersionTracker ot;
2871   map<string, bufferlist> attrs;
2872   string meta_key = RGWSI_Bucket::get_entrypoint_meta_key(bucket);
2873   ret = svc.bucket->read_bucket_entrypoint_info(ctx, meta_key, &ep, &ot, nullptr, &attrs, y, dpp);
2874   if (ret == -ENOENT)
2875     return 0;
2876   if (ret < 0)
2877     return ret;
2878
2879   if (!ep.linked)
2880     return 0;
2881
2882   if (ep.owner != user_id) {
2883     ldpp_dout(dpp, 0) << "bucket entry point user mismatch, can't unlink bucket: " << ep.owner << " != " << user_id << dendl;
2884     return -EINVAL;
2885   }
2886
2887   ep.linked = false;
2888   return svc.bucket->store_bucket_entrypoint_info(ctx, meta_key, ep, false, real_time(), &attrs, &ot, y, dpp);
2889 }
2890
2891 // TODO: remove RGWRados dependency for bucket listing
2892 int RGWBucketCtl::chown(rgw::sal::Store* store, rgw::sal::Bucket* bucket,
2893                         const rgw_user& user_id, const std::string& display_name,
2894                         const std::string& marker, optional_yield y, const DoutPrefixProvider *dpp)
2895 {
2896   map<string, bool> common_prefixes;
2897
2898   rgw::sal::Bucket::ListParams params;
2899   rgw::sal::Bucket::ListResults results;
2900
2901   params.list_versions = true;
2902   params.allow_unordered = true;
2903   params.marker = marker;
2904
2905   int count = 0;
2906   int max_entries = 1000;
2907
2908   //Loop through objects and update object acls to point to bucket owner
2909
2910   do {
2911     RGWObjectCtx obj_ctx(store);
2912     results.objs.clear();
2913     int ret = bucket->list(dpp, params, max_entries, results, y);
2914     if (ret < 0) {
2915       ldpp_dout(dpp, 0) << "ERROR: list objects failed: " << cpp_strerror(-ret) << dendl;
2916       return ret;
2917     }
2918
2919     params.marker = results.next_marker;
2920     count += results.objs.size();
2921
2922     for (const auto& obj : results.objs) {
2923       std::unique_ptr<rgw::sal::Object> r_obj = bucket->get_object(obj.key);
2924
2925       ret = r_obj->get_obj_attrs(&obj_ctx, y, dpp);
2926       if (ret < 0){
2927         ldpp_dout(dpp, 0) << "ERROR: failed to read object " << obj.key.name << cpp_strerror(-ret) << dendl;
2928         continue;
2929       }
2930       const auto& aiter = r_obj->get_attrs().find(RGW_ATTR_ACL);
2931       if (aiter == r_obj->get_attrs().end()) {
2932         ldpp_dout(dpp, 0) << "ERROR: no acls found for object " << obj.key.name << " .Continuing with next object." << dendl;
2933         continue;
2934       } else {
2935         bufferlist& bl = aiter->second;
2936         RGWAccessControlPolicy policy(store->ctx());
2937         ACLOwner owner;
2938         try {
2939           decode(policy, bl);
2940           owner = policy.get_owner();
2941         } catch (buffer::error& err) {
2942           ldpp_dout(dpp, 0) << "ERROR: decode policy failed" << err.what()
2943                                  << dendl;
2944           return -EIO;
2945         }
2946
2947         //Get the ACL from the policy
2948         RGWAccessControlList& acl = policy.get_acl();
2949
2950         //Remove grant that is set to old owner
2951         acl.remove_canon_user_grant(owner.get_id());
2952
2953         //Create a grant and add grant
2954         ACLGrant grant;
2955         grant.set_canon(user_id, display_name, RGW_PERM_FULL_CONTROL);
2956         acl.add_grant(&grant);
2957
2958         //Update the ACL owner to the new user
2959         owner.set_id(user_id);
2960         owner.set_name(display_name);
2961         policy.set_owner(owner);
2962
2963         bl.clear();
2964         encode(policy, bl);
2965
2966         r_obj->set_atomic(&obj_ctx);
2967         map<string, bufferlist> attrs;
2968         attrs[RGW_ATTR_ACL] = bl;
2969         ret = r_obj->set_obj_attrs(dpp, &obj_ctx, &attrs, nullptr, y);
2970         if (ret < 0) {
2971           ldpp_dout(dpp, 0) << "ERROR: modify attr failed " << cpp_strerror(-ret) << dendl;
2972           return ret;
2973         }
2974       }
2975     }
2976     cerr << count << " objects processed in " << bucket
2977         << ". Next marker " << params.marker.name << std::endl;
2978   } while(results.is_truncated);
2979   return 0;
2980 }
2981
2982 int RGWBucketCtl::read_bucket_stats(const rgw_bucket& bucket,
2983                                     RGWBucketEnt *result,
2984                                     optional_yield y,
2985                                     const DoutPrefixProvider *dpp)
2986 {
2987   return call([&](RGWSI_Bucket_X_Ctx& ctx) {
2988     return svc.bucket->read_bucket_stats(ctx, bucket, result, y, dpp);
2989   });
2990 }
2991
2992 int RGWBucketCtl::read_buckets_stats(map<string, RGWBucketEnt>& m,
2993                                      optional_yield y, const DoutPrefixProvider *dpp)
2994 {
2995   return call([&](RGWSI_Bucket_X_Ctx& ctx) {
2996     return svc.bucket->read_buckets_stats(ctx, m, y, dpp);
2997   });
2998 }
2999
3000 int RGWBucketCtl::sync_user_stats(const DoutPrefixProvider *dpp,
3001                                   const rgw_user& user_id,
3002                                   const RGWBucketInfo& bucket_info,
3003                                   optional_yield y,
3004                                   RGWBucketEnt* pent)
3005 {
3006   RGWBucketEnt ent;
3007   if (!pent) {
3008     pent = &ent;
3009   }
3010   int r = svc.bi->read_stats(dpp, bucket_info, pent, null_yield);
3011   if (r < 0) {
3012     ldpp_dout(dpp, 20) << __func__ << "(): failed to read bucket stats (r=" << r << ")" << dendl;
3013     return r;
3014   }
3015
3016   return ctl.user->flush_bucket_stats(dpp, user_id, *pent, y);
3017 }
3018
3019 int RGWBucketCtl::get_sync_policy_handler(std::optional<rgw_zone_id> zone,
3020                                           std::optional<rgw_bucket> bucket,
3021                                           RGWBucketSyncPolicyHandlerRef *phandler,
3022                                           optional_yield y,
3023                                           const DoutPrefixProvider *dpp)
3024 {
3025   int r = call([&](RGWSI_Bucket_X_Ctx& ctx) {
3026     return svc.bucket_sync->get_policy_handler(ctx, zone, bucket, phandler, y, dpp);
3027   });
3028   if (r < 0) {
3029     ldpp_dout(dpp, 20) << __func__ << "(): failed to get policy handler for bucket=" << bucket << " (r=" << r << ")" << dendl;
3030     return r;
3031   }
3032   return 0;
3033 }
3034
3035 int RGWBucketCtl::bucket_exports_data(const rgw_bucket& bucket,
3036                                       optional_yield y,
3037                                       const DoutPrefixProvider *dpp)
3038 {
3039
3040   RGWBucketSyncPolicyHandlerRef handler;
3041
3042   int r = get_sync_policy_handler(std::nullopt, bucket, &handler, y, dpp);
3043   if (r < 0) {
3044     return r;
3045   }
3046
3047   return handler->bucket_exports_data();
3048 }
3049
3050 int RGWBucketCtl::bucket_imports_data(const rgw_bucket& bucket,
3051                                       optional_yield y, const DoutPrefixProvider *dpp)
3052 {
3053
3054   RGWBucketSyncPolicyHandlerRef handler;
3055
3056   int r = get_sync_policy_handler(std::nullopt, bucket, &handler, y, dpp);
3057   if (r < 0) {
3058     return r;
3059   }
3060
3061   return handler->bucket_imports_data();
3062 }
3063
3064 RGWBucketMetadataHandlerBase *RGWBucketMetaHandlerAllocator::alloc()
3065 {
3066   return new RGWBucketMetadataHandler();
3067 }
3068
3069 RGWBucketInstanceMetadataHandlerBase *RGWBucketInstanceMetaHandlerAllocator::alloc()
3070 {
3071   return new RGWBucketInstanceMetadataHandler();
3072 }
3073
3074 RGWBucketMetadataHandlerBase *RGWArchiveBucketMetaHandlerAllocator::alloc()
3075 {
3076   return new RGWArchiveBucketMetadataHandler();
3077 }
3078
3079 RGWBucketInstanceMetadataHandlerBase *RGWArchiveBucketInstanceMetaHandlerAllocator::alloc()
3080 {
3081   return new RGWArchiveBucketInstanceMetadataHandler();
3082 }
3083
3084
3085 void RGWBucketEntryPoint::generate_test_instances(list<RGWBucketEntryPoint*>& o)
3086 {
3087   RGWBucketEntryPoint *bp = new RGWBucketEntryPoint();
3088   init_bucket(&bp->bucket, "tenant", "bucket", "pool", ".index.pool", "marker", "10");
3089   bp->owner = "owner";
3090   bp->creation_time = ceph::real_clock::from_ceph_timespec({ceph_le32(2), ceph_le32(3)});
3091
3092   o.push_back(bp);
3093   o.push_back(new RGWBucketEntryPoint);
3094 }
3095
3096 void RGWBucketEntryPoint::dump(Formatter *f) const
3097 {
3098   encode_json("bucket", bucket, f);
3099   encode_json("owner", owner, f);
3100   utime_t ut(creation_time);
3101   encode_json("creation_time", ut, f);
3102   encode_json("linked", linked, f);
3103   encode_json("has_bucket_info", has_bucket_info, f);
3104   if (has_bucket_info) {
3105     encode_json("old_bucket_info", old_bucket_info, f);
3106   }
3107 }
3108
3109 void RGWBucketEntryPoint::decode_json(JSONObj *obj) {
3110   JSONDecoder::decode_json("bucket", bucket, obj);
3111   JSONDecoder::decode_json("owner", owner, obj);
3112   utime_t ut;
3113   JSONDecoder::decode_json("creation_time", ut, obj);
3114   creation_time = ut.to_real_time();
3115   JSONDecoder::decode_json("linked", linked, obj);
3116   JSONDecoder::decode_json("has_bucket_info", has_bucket_info, obj);
3117   if (has_bucket_info) {
3118     JSONDecoder::decode_json("old_bucket_info", old_bucket_info, obj);
3119   }
3120 }
3121