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