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
   3
   4 #include <errno.h>
   5
   6 #include <string>
   7 #include <map>
   8 #include <sstream>
   9
  10 #include <boost/utility/string_ref.hpp>
  11 #include <boost/format.hpp>
  12
  13 #include "common/errno.h"
  14 #include "common/ceph_json.h"
  15 #include "rgw_rados.h"
  16 #include "rgw_acl.h"
  17 #include "rgw_acl_s3.h"
  18
  19 #include "include/types.h"
  20 #include "rgw_bucket.h"
  21 #include "rgw_user.h"
  22 #include "rgw_string.h"
  23 #include "rgw_multi.h"
  24
  25 #include "include/rados/librados.hpp"
  26 // until everything is moved from rgw_common
  27 #include "rgw_common.h"
  28
  29 #include "cls/user/cls_user_types.h"
  30
  31 #define dout_context g_ceph_context
  32 #define dout_subsys ceph_subsys_rgw
  33
  34 #define BUCKET_TAG_TIMEOUT 30
  35
  36 using namespace std;
  37
  38 static RGWMetadataHandler *bucket_meta_handler = NULL;
  39 static RGWMetadataHandler *bucket_instance_meta_handler = NULL;
  40
  41 // define as static when RGWBucket implementation compete
  42 void rgw_get_buckets_obj(const rgw_user& user_id, string& buckets_obj_id)
  43 {
  44   buckets_obj_id = user_id.to_str();
  45   buckets_obj_id += RGW_BUCKETS_OBJ_SUFFIX;
  46 }
  47
  48 /*
  49  * Note that this is not a reversal of parse_bucket(). That one deals
  50  * with the syntax we need in metadata and such. This one deals with
  51  * the representation in RADOS pools. We chose '/' because it's not
  52  * acceptable in bucket names and thus qualified buckets cannot conflict
  53  * with the legacy or S3 buckets.
  54  */
  55 std::string rgw_make_bucket_entry_name(const std::string& tenant_name,
  56                                        const std::string& bucket_name) {
  57   std::string bucket_entry;
  58
  59   if (bucket_name.empty()) {
  60     bucket_entry.clear();
  61   } else if (tenant_name.empty()) {
  62     bucket_entry = bucket_name;
  63   } else {
  64     bucket_entry = tenant_name + "/" + bucket_name;
  65   }
  66
  67   return bucket_entry;
  68 }
  69
  70 /*
  71  * Tenants are separated from buckets in URLs by a colon in S3.
  72  * This function is not to be used on Swift URLs, not even for COPY arguments.
  73  */
  74 void rgw_parse_url_bucket(const string &bucket, const string& auth_tenant,
  75                           string &tenant_name, string &bucket_name) {
  76
  77   int pos = bucket.find(':');
  78   if (pos >= 0) {
  79     /*
  80      * N.B.: We allow ":bucket" syntax with explicit empty tenant in order
  81      * to refer to the legacy tenant, in case users in new named tenants
  82      * want to access old global buckets.
  83      */
  84     tenant_name = bucket.substr(0, pos);
  85     bucket_name = bucket.substr(pos + 1);
  86   } else {
  87     tenant_name = auth_tenant;
  88     bucket_name = bucket;
  89   }
  90 }
  91
  92 /**
  93  * Get all the buckets owned by a user and fill up an RGWUserBuckets with them.
  94  * Returns: 0 on success, -ERR# on failure.
  95  */
  96 int rgw_read_user_buckets(RGWRados * store,
  97                           const rgw_user& user_id,
  98                           RGWUserBuckets& buckets,
  99                           const string& marker,
 100                           const string& end_marker,
 101                           uint64_t max,
 102                           bool need_stats,
 103                           bool *is_truncated,
 104                           uint64_t default_amount)
 105 {
 106   int ret;
 107   buckets.clear();
 108   string buckets_obj_id;
 109   rgw_get_buckets_obj(user_id, buckets_obj_id);
 110   rgw_raw_obj obj(store->get_zone_params().user_uid_pool, buckets_obj_id);
 111   list<cls_user_bucket_entry> entries;
 112
 113   bool truncated = false;
 114   string m = marker;
 115
 116   uint64_t total = 0;
 117
 118   if (!max) {
 119     max = default_amount;
 120   }
 121
 122   do {
 123     ret = store->cls_user_list_buckets(obj, m, end_marker, max - total, entries, &m, &truncated);
 124     if (ret == -ENOENT)
 125       ret = 0;
 126
 127     if (ret < 0)
 128       return ret;
 129
 130     for (const auto& entry : entries) {
 131       buckets.add(RGWBucketEnt(user_id, entry));
 132       total++;
 133     }
 134
 135   } while (truncated && total < max);
 136
 137   if (is_truncated != nullptr) {
 138     *is_truncated = truncated;
 139   }
 140
 141   if (need_stats) {
 142     map<string, RGWBucketEnt>& m = buckets.get_buckets();
 143     ret = store->update_containers_stats(m);
 144     if (ret < 0 && ret != -ENOENT) {
 145       ldout(store->ctx(), 0) << "ERROR: could not get stats for buckets" << dendl;
 146       return ret;
 147     }
 148   }
 149   return 0;
 150 }
 151
 152 int rgw_bucket_sync_user_stats(RGWRados *store, const rgw_user& user_id, const RGWBucketInfo& bucket_info)
 153 {
 154   string buckets_obj_id;
 155   rgw_get_buckets_obj(user_id, buckets_obj_id);
 156   rgw_raw_obj obj(store->get_zone_params().user_uid_pool, buckets_obj_id);
 157
 158   return store->cls_user_sync_bucket_stats(obj, bucket_info);
 159 }
 160
 161 int rgw_bucket_sync_user_stats(RGWRados *store, const string& tenant_name, const string& bucket_name)
 162 {
 163   RGWBucketInfo bucket_info;
 164   RGWObjectCtx obj_ctx(store);
 165   int ret = store->get_bucket_info(obj_ctx, tenant_name, bucket_name, bucket_info, NULL);
 166   if (ret < 0) {
 167     ldout(store->ctx(), 0) << "ERROR: could not fetch bucket info: ret=" << ret << dendl;
 168     return ret;
 169   }
 170
 171   ret = rgw_bucket_sync_user_stats(store, bucket_info.owner, bucket_info);
 172   if (ret < 0) {
 173     ldout(store->ctx(), 0) << "ERROR: could not sync user stats for bucket " << bucket_name << ": ret=" << ret << dendl;
 174     return ret;
 175   }
 176
 177   return 0;
 178 }
 179
 180 int rgw_link_bucket(RGWRados *store, const rgw_user& user_id, rgw_bucket& bucket, real_time creation_time, bool update_entrypoint)
 181 {
 182   int ret;
 183   string& tenant_name = bucket.tenant;
 184   string& bucket_name = bucket.name;
 185
 186   cls_user_bucket_entry new_bucket;
 187
 188   RGWBucketEntryPoint ep;
 189   RGWObjVersionTracker ot;
 190
 191   bucket.convert(&new_bucket.bucket);
 192   new_bucket.size = 0;
 193   if (real_clock::is_zero(creation_time))
 194     new_bucket.creation_time = real_clock::now();
 195   else
 196     new_bucket.creation_time = creation_time;
 197
 198   map<string, bufferlist> attrs;
 199   RGWObjectCtx obj_ctx(store);
 200
 201   if (update_entrypoint) {
 202     ret = store->get_bucket_entrypoint_info(obj_ctx, tenant_name, bucket_name, ep, &ot, NULL, &attrs);
 203     if (ret < 0 && ret != -ENOENT) {
 204       ldout(store->ctx(), 0) << "ERROR: store->get_bucket_entrypoint_info() returned: "
 205                              << cpp_strerror(-ret) << dendl;
 206     }
 207   }
 208
 209   string buckets_obj_id;
 210   rgw_get_buckets_obj(user_id, buckets_obj_id);
 211
 212   rgw_raw_obj obj(store->get_zone_params().user_uid_pool, buckets_obj_id);
 213   ret = store->cls_user_add_bucket(obj, new_bucket);
 214   if (ret < 0) {
 215     ldout(store->ctx(), 0) << "ERROR: error adding bucket to directory: "
 216                            << cpp_strerror(-ret) << dendl;
 217     goto done_err;
 218   }
 219
 220   if (!update_entrypoint)
 221     return 0;
 222
 223   ep.linked = true;
 224   ep.owner = user_id;
 225   ep.bucket = bucket;
 226   ret = store->put_bucket_entrypoint_info(tenant_name, bucket_name, ep, false, ot, real_time(), &attrs);
 227   if (ret < 0)
 228     goto done_err;
 229
 230   return 0;
 231 done_err:
 232   int r = rgw_unlink_bucket(store, user_id, bucket.tenant, bucket.name);
 233   if (r < 0) {
 234     ldout(store->ctx(), 0) << "ERROR: failed unlinking bucket on error cleanup: "
 235                            << cpp_strerror(-r) << dendl;
 236   }
 237   return ret;
 238 }
 239
 240 int rgw_unlink_bucket(RGWRados *store, const rgw_user& user_id, const string& tenant_name, const string& bucket_name, bool update_entrypoint)
 241 {
 242   int ret;
 243
 244   string buckets_obj_id;
 245   rgw_get_buckets_obj(user_id, buckets_obj_id);
 246
 247   cls_user_bucket bucket;
 248   bucket.name = bucket_name;
 249   rgw_raw_obj obj(store->get_zone_params().user_uid_pool, buckets_obj_id);
 250   ret = store->cls_user_remove_bucket(obj, bucket);
 251   if (ret < 0) {
 252     ldout(store->ctx(), 0) << "ERROR: error removing bucket from directory: "
 253         << cpp_strerror(-ret)<< dendl;
 254   }
 255
 256   if (!update_entrypoint)
 257     return 0;
 258
 259   RGWBucketEntryPoint ep;
 260   RGWObjVersionTracker ot;
 261   map<string, bufferlist> attrs;
 262   RGWObjectCtx obj_ctx(store);
 263   ret = store->get_bucket_entrypoint_info(obj_ctx, tenant_name, bucket_name, ep, &ot, NULL, &attrs);
 264   if (ret == -ENOENT)
 265     return 0;
 266   if (ret < 0)
 267     return ret;
 268
 269   if (!ep.linked)
 270     return 0;
 271
 272   if (ep.owner != user_id) {
 273     ldout(store->ctx(), 0) << "bucket entry point user mismatch, can't unlink bucket: " << ep.owner << " != " << user_id << dendl;
 274     return -EINVAL;
 275   }
 276
 277   ep.linked = false;
 278   return store->put_bucket_entrypoint_info(tenant_name, bucket_name, ep, false, ot, real_time(), &attrs);
 279 }
 280
 281 int rgw_bucket_store_info(RGWRados *store, const string& bucket_name, bufferlist& bl, bool exclusive,
 282                           map<string, bufferlist> *pattrs, RGWObjVersionTracker *objv_tracker,
 283                           real_time mtime) {
 284   return store->meta_mgr->put_entry(bucket_meta_handler, bucket_name, bl, exclusive, objv_tracker, mtime, pattrs);
 285 }
 286
 287 int rgw_bucket_instance_store_info(RGWRados *store, string& entry, bufferlist& bl, bool exclusive,
 288                           map<string, bufferlist> *pattrs, RGWObjVersionTracker *objv_tracker,
 289                           real_time mtime) {
 290   return store->meta_mgr->put_entry(bucket_instance_meta_handler, entry, bl, exclusive, objv_tracker, mtime, pattrs);
 291 }
 292
 293 int rgw_bucket_instance_remove_entry(RGWRados *store, string& entry, RGWObjVersionTracker *objv_tracker) {
 294   return store->meta_mgr->remove_entry(bucket_instance_meta_handler, entry, objv_tracker);
 295 }
 296
 297 // 'tenant/' is used in bucket instance keys for sync to avoid parsing ambiguity
 298 // with the existing instance[:shard] format. once we parse the shard, the / is
 299 // replaced with a : to match the [tenant:]instance format
 300 void rgw_bucket_instance_key_to_oid(string& key)
 301 {
 302   // replace tenant/ with tenant:
 303   auto c = key.find('/');
 304   if (c != string::npos) {
 305     key[c] = ':';
 306   }
 307 }
 308
 309 // convert bucket instance oids back to the tenant/ format for metadata keys.
 310 // it's safe to parse 'tenant:' only for oids, because they won't contain the
 311 // optional :shard at the end
 312 void rgw_bucket_instance_oid_to_key(string& oid)
 313 {
 314   // find first : (could be tenant:bucket or bucket:instance)
 315   auto c = oid.find(':');
 316   if (c != string::npos) {
 317     // if we find another :, the first one was for tenant
 318     if (oid.find(':', c + 1) != string::npos) {
 319       oid[c] = '/';
 320     }
 321   }
 322 }
 323
 324 int rgw_bucket_parse_bucket_instance(const string& bucket_instance, string *target_bucket_instance, int *shard_id)
 325 {
 326   ssize_t pos = bucket_instance.rfind(':');
 327   if (pos < 0) {
 328     return -EINVAL;
 329   }
 330
 331   string first = bucket_instance.substr(0, pos);
 332   string second = bucket_instance.substr(pos + 1);
 333
 334   if (first.find(':') == string::npos) {
 335     *shard_id = -1;
 336     *target_bucket_instance = bucket_instance;
 337     return 0;
 338   }
 339
 340   *target_bucket_instance = first;
 341   string err;
 342   *shard_id = strict_strtol(second.c_str(), 10, &err);
 343   if (!err.empty()) {
 344     return -EINVAL;
 345   }
 346
 347   return 0;
 348 }
 349
 350 // parse key in format: [tenant/]name:instance[:shard_id]
 351 int rgw_bucket_parse_bucket_key(CephContext *cct, const string& key,
 352                                 rgw_bucket *bucket, int *shard_id)
 353 {
 354   boost::string_ref name{key};
 355   boost::string_ref instance;
 356
 357   // split tenant/name
 358   auto pos = name.find('/');
 359   if (pos != boost::string_ref::npos) {
 360     auto tenant = name.substr(0, pos);
 361     bucket->tenant.assign(tenant.begin(), tenant.end());
 362     name = name.substr(pos + 1);
 363   }
 364
 365   // split name:instance
 366   pos = name.find(':');
 367   if (pos != boost::string_ref::npos) {
 368     instance = name.substr(pos + 1);
 369     name = name.substr(0, pos);
 370   }
 371   bucket->name.assign(name.begin(), name.end());
 372
 373   // split instance:shard
 374   pos = instance.find(':');
 375   if (pos == boost::string_ref::npos) {
 376     bucket->bucket_id.assign(instance.begin(), instance.end());
 377     *shard_id = -1;
 378     return 0;
 379   }
 380
 381   // parse shard id
 382   auto shard = instance.substr(pos + 1);
 383   string err;
 384   auto id = strict_strtol(shard.data(), 10, &err);
 385   if (!err.empty()) {
 386     ldout(cct, 0) << "ERROR: failed to parse bucket shard '"
 387         << instance.data() << "': " << err << dendl;
 388     return -EINVAL;
 389   }
 390
 391   *shard_id = id;
 392   instance = instance.substr(0, pos);
 393   bucket->bucket_id.assign(instance.begin(), instance.end());
 394   return 0;
 395 }
 396
 397 int rgw_bucket_set_attrs(RGWRados *store, RGWBucketInfo& bucket_info,
 398                          map<string, bufferlist>& attrs,
 399                          RGWObjVersionTracker *objv_tracker)
 400 {
 401   rgw_bucket& bucket = bucket_info.bucket;
 402
 403   if (!bucket_info.has_instance_obj) {
 404     /* an old bucket object, need to convert it */
 405     RGWObjectCtx obj_ctx(store);
 406     int ret = store->convert_old_bucket_info(obj_ctx, bucket.tenant, bucket.name);
 407     if (ret < 0) {
 408       ldout(store->ctx(), 0) << "ERROR: failed converting old bucket info: " << ret << dendl;
 409       return ret;
 410     }
 411   }
 412
 413   /* we want the bucket instance name without the oid prefix cruft */
 414   string key = bucket.get_key();
 415   bufferlist bl;
 416
 417   ::encode(bucket_info, bl);
 418
 419   return rgw_bucket_instance_store_info(store, key, bl, false, &attrs, objv_tracker, real_time());
 420 }
 421
 422 static void dump_mulipart_index_results(list<rgw_obj_index_key>& objs_to_unlink,
 423         Formatter *f)
 424 {
 425   // make sure that an appropiately titled header has been opened previously
 426   auto oiter = objs_to_unlink.begin();
 427
 428   f->open_array_section("invalid_multipart_entries");
 429
 430   for ( ; oiter != objs_to_unlink.end(); ++oiter) {
 431     f->dump_string("object",  oiter->name);
 432   }
 433
 434   f->close_section();
 435 }
 436
 437 void check_bad_user_bucket_mapping(RGWRados *store, const rgw_user& user_id,
 438                                    bool fix)
 439 {
 440   RGWUserBuckets user_buckets;
 441   bool is_truncated = false;
 442   string marker;
 443
 444   CephContext *cct = store->ctx();
 445
 446   size_t max_entries = cct->_conf->rgw_list_buckets_max_chunk;
 447
 448   do {
 449     int ret = rgw_read_user_buckets(store, user_id, user_buckets, marker,
 450                                     string(), max_entries, false,
 451                                     &is_truncated);
 452     if (ret < 0) {
 453       ldout(store->ctx(), 0) << "failed to read user buckets: "
 454                              << cpp_strerror(-ret) << dendl;
 455       return;
 456     }
 457
 458     map<string, RGWBucketEnt>& buckets = user_buckets.get_buckets();
 459     for (map<string, RGWBucketEnt>::iterator i = buckets.begin();
 460          i != buckets.end();
 461          ++i) {
 462       marker = i->first;
 463
 464       RGWBucketEnt& bucket_ent = i->second;
 465       rgw_bucket& bucket = bucket_ent.bucket;
 466
 467       RGWBucketInfo bucket_info;
 468       real_time mtime;
 469       RGWObjectCtx obj_ctx(store);
 470       int r = store->get_bucket_info(obj_ctx, user_id.tenant, bucket.name, bucket_info, &mtime);
 471       if (r < 0) {
 472         ldout(store->ctx(), 0) << "could not get bucket info for bucket=" << bucket << dendl;
 473         continue;
 474       }
 475
 476       rgw_bucket& actual_bucket = bucket_info.bucket;
 477
 478       if (actual_bucket.name.compare(bucket.name) != 0 ||
 479           actual_bucket.tenant.compare(bucket.tenant) != 0 ||
 480           actual_bucket.marker.compare(bucket.marker) != 0 ||
 481           actual_bucket.bucket_id.compare(bucket.bucket_id) != 0) {
 482         cout << "bucket info mismatch: expected " << actual_bucket << " got " << bucket << std::endl;
 483         if (fix) {
 484           cout << "fixing" << std::endl;
 485           r = rgw_link_bucket(store, user_id, actual_bucket, bucket_info.creation_time);
 486           if (r < 0) {
 487             cerr << "failed to fix bucket: " << cpp_strerror(-r) << std::endl;
 488           }
 489         }
 490       }
 491     }
 492   } while (is_truncated);
 493 }
 494
 495 static bool bucket_object_check_filter(const string& oid)
 496 {
 497   rgw_obj_key key;
 498   string ns;
 499   return rgw_obj_key::oid_to_key_in_ns(oid, &key, ns);
 500 }
 501
 502 int rgw_remove_object(RGWRados *store, RGWBucketInfo& bucket_info, rgw_bucket& bucket, rgw_obj_key& key)
 503 {
 504   RGWObjectCtx rctx(store);
 505
 506   if (key.instance.empty()) {
 507     key.instance = "null";
 508   }
 509
 510   rgw_obj obj(bucket, key);
 511
 512   return store->delete_obj(rctx, bucket_info, obj, bucket_info.versioning_status());
 513 }
 514
 515 int rgw_remove_bucket(RGWRados *store, rgw_bucket& bucket, bool delete_children)
 516 {
 517   int ret;
 518   map<RGWObjCategory, RGWStorageStats> stats;
 519   std::vector<rgw_bucket_dir_entry> objs;
 520   map<string, bool> common_prefixes;
 521   RGWBucketInfo info;
 522   RGWObjectCtx obj_ctx(store);
 523
 524   string bucket_ver, master_ver;
 525
 526   ret = store->get_bucket_info(obj_ctx, bucket.tenant, bucket.name, info, NULL);
 527   if (ret < 0)
 528     return ret;
 529
 530   ret = store->get_bucket_stats(info, RGW_NO_SHARD, &bucket_ver, &master_ver, stats, NULL);
 531   if (ret < 0)
 532     return ret;
 533
 534   RGWRados::Bucket target(store, info);
 535   RGWRados::Bucket::List list_op(&target);
 536   CephContext *cct = store->ctx();
 537   int max = 1000;
 538
 539   list_op.params.list_versions = true;
 540
 541   do {
 542     objs.clear();
 543
 544     ret = list_op.list_objects(max, &objs, &common_prefixes, NULL);
 545     if (ret < 0)
 546       return ret;
 547
 548     if (!objs.empty() && !delete_children) {
 549       lderr(store->ctx()) << "ERROR: could not remove non-empty bucket " << bucket.name << dendl;
 550       return -ENOTEMPTY;
 551     }
 552
 553     for (const auto& obj : objs) {
 554       rgw_obj_key key(obj.key);
 555       ret = rgw_remove_object(store, info, bucket, key);
 556       if (ret < 0)
 557         return ret;
 558     }
 559
 560   } while (!objs.empty());
 561
 562   string prefix, delimiter;
 563
 564   ret = abort_bucket_multiparts(store, cct, info, prefix, delimiter);
 565   if (ret < 0) {
 566     return ret;
 567   }
 568
 569   ret = rgw_bucket_sync_user_stats(store, bucket.tenant, info);
 570   if ( ret < 0) {
 571      dout(1) << "WARNING: failed sync user stats before bucket delete. ret=" <<  ret << dendl;
 572   }
 573
 574   RGWObjVersionTracker objv_tracker;
 575
 576   ret = store->delete_bucket(info, objv_tracker);
 577   if (ret < 0) {
 578     lderr(store->ctx()) << "ERROR: could not remove bucket " << bucket.name << dendl;
 579     return ret;
 580   }
 581
 582   ret = rgw_unlink_bucket(store, info.owner, bucket.tenant, bucket.name, false);
 583   if (ret < 0) {
 584     lderr(store->ctx()) << "ERROR: unable to remove user bucket information" << dendl;
 585   }
 586
 587   return ret;
 588 }
 589
 590 static int aio_wait(librados::AioCompletion *handle)
 591 {
 592   librados::AioCompletion *c = (librados::AioCompletion *)handle;
 593   c->wait_for_safe();
 594   int ret = c->get_return_value();
 595   c->release();
 596   return ret;
 597 }
 598
 599 static int drain_handles(list<librados::AioCompletion *>& pending)
 600 {
 601   int ret = 0;
 602   while (!pending.empty()) {
 603     librados::AioCompletion *handle = pending.front();
 604     pending.pop_front();
 605     int r = aio_wait(handle);
 606     if (r < 0) {
 607       ret = r;
 608     }
 609   }
 610   return ret;
 611 }
 612
 613 int rgw_remove_bucket_bypass_gc(RGWRados *store, rgw_bucket& bucket,
 614                                 int concurrent_max, bool keep_index_consistent)
 615 {
 616   int ret;
 617   map<RGWObjCategory, RGWStorageStats> stats;
 618   std::vector<rgw_bucket_dir_entry> objs;
 619   map<string, bool> common_prefixes;
 620   RGWBucketInfo info;
 621   RGWObjectCtx obj_ctx(store);
 622   CephContext *cct = store->ctx();
 623
 624   string bucket_ver, master_ver;
 625
 626   ret = store->get_bucket_info(obj_ctx, bucket.tenant, bucket.name, info, NULL);
 627   if (ret < 0)
 628     return ret;
 629
 630   ret = store->get_bucket_stats(info, RGW_NO_SHARD, &bucket_ver, &master_ver, stats, NULL);
 631   if (ret < 0)
 632     return ret;
 633
 634   string prefix, delimiter;
 635
 636   ret = abort_bucket_multiparts(store, cct, info, prefix, delimiter);
 637   if (ret < 0) {
 638     return ret;
 639   }
 640
 641   RGWRados::Bucket target(store, info);
 642   RGWRados::Bucket::List list_op(&target);
 643
 644   list_op.params.list_versions = true;
 645
 646   std::list<librados::AioCompletion*> handles;
 647
 648   int max = 1000;
 649   int max_aio = concurrent_max;
 650   ret = list_op.list_objects(max, &objs, &common_prefixes, NULL);
 651   if (ret < 0)
 652     return ret;
 653
 654   while (!objs.empty()) {
 655     std::vector<rgw_bucket_dir_entry>::iterator it = objs.begin();
 656     for (; it != objs.end(); ++it) {
 657       RGWObjState *astate = NULL;
 658       rgw_obj obj(bucket, (*it).key);
 659
 660       ret = store->get_obj_state(&obj_ctx, info, obj, &astate, false);
 661       if (ret == -ENOENT) {
 662         dout(1) << "WARNING: cannot find obj state for obj " << obj.get_oid() << dendl;
 663         continue;
 664       }
 665       if (ret < 0) {
 666         lderr(store->ctx()) << "ERROR: get obj state returned with error " << ret << dendl;
 667         return ret;
 668       }
 669
 670       if (astate->has_manifest) {
 671         RGWObjManifest& manifest = astate->manifest;
 672         RGWObjManifest::obj_iterator miter = manifest.obj_begin();
 673         rgw_obj head_obj = manifest.get_obj();
 674         rgw_raw_obj raw_head_obj;
 675         store->obj_to_raw(info.placement_rule, head_obj, &raw_head_obj);
 676
 677
 678         for (; miter != manifest.obj_end() && max_aio--; ++miter) {
 679           if (!max_aio) {
 680             ret = drain_handles(handles);
 681             if (ret < 0) {
 682               lderr(store->ctx()) << "ERROR: could not drain handles as aio completion returned with " << ret << dendl;
 683               return ret;
 684             }
 685             max_aio = concurrent_max;
 686           }
 687
 688           rgw_raw_obj last_obj = miter.get_location().get_raw_obj(store);
 689           if (last_obj == raw_head_obj) {
 690             // have the head obj deleted at the end
 691             continue;
 692           }
 693
 694           ret = store->delete_raw_obj_aio(last_obj, handles);
 695           if (ret < 0) {
 696             lderr(store->ctx()) << "ERROR: delete obj aio failed with " << ret << dendl;
 697             return ret;
 698           }
 699         } // for all shadow objs
 700
 701         ret = store->delete_obj_aio(head_obj, info, astate, handles, keep_index_consistent);
 702         if (ret < 0) {
 703           lderr(store->ctx()) << "ERROR: delete obj aio failed with " << ret << dendl;
 704           return ret;
 705         }
 706       }
 707
 708       if (!max_aio) {
 709         ret = drain_handles(handles);
 710         if (ret < 0) {
 711           lderr(store->ctx()) << "ERROR: could not drain handles as aio completion returned with " << ret << dendl;
 712           return ret;
 713         }
 714         max_aio = concurrent_max;
 715       }
 716     } // for all RGW objects
 717     objs.clear();
 718
 719     ret = list_op.list_objects(max, &objs, &common_prefixes, NULL);
 720     if (ret < 0)
 721       return ret;
 722   }
 723
 724   ret = drain_handles(handles);
 725   if (ret < 0) {
 726     lderr(store->ctx()) << "ERROR: could not drain handles as aio completion returned with " << ret << dendl;
 727     return ret;
 728   }
 729
 730   ret = rgw_bucket_sync_user_stats(store, bucket.tenant, info);
 731   if (ret < 0) {
 732      dout(1) << "WARNING: failed sync user stats before bucket delete. ret=" <<  ret << dendl;
 733   }
 734
 735   RGWObjVersionTracker objv_tracker;
 736
 737   ret = rgw_bucket_delete_bucket_obj(store, bucket.tenant, bucket.name, objv_tracker);
 738   if (ret < 0) {
 739     lderr(store->ctx()) << "ERROR: could not remove bucket " << bucket.name << "with ret as " << ret << dendl;
 740     return ret;
 741   }
 742
 743   if (!store->is_syncing_bucket_meta(bucket)) {
 744     RGWObjVersionTracker objv_tracker;
 745     string entry = bucket.get_key();
 746     ret = rgw_bucket_instance_remove_entry(store, entry, &objv_tracker);
 747     if (ret < 0) {
 748       lderr(store->ctx()) << "ERROR: could not remove bucket instance entry" << bucket.name << "with ret as " << ret << dendl;
 749       return ret;
 750     }
 751   }
 752
 753   ret = rgw_unlink_bucket(store, info.owner, bucket.tenant, bucket.name, false);
 754   if (ret < 0) {
 755     lderr(store->ctx()) << "ERROR: unable to remove user bucket information" << dendl;
 756   }
 757
 758   return ret;
 759 }
 760
 761 int rgw_bucket_delete_bucket_obj(RGWRados *store,
 762                                  const string& tenant_name,
 763                                  const string& bucket_name,
 764                                  RGWObjVersionTracker& objv_tracker)
 765 {
 766   string key;
 767
 768   rgw_make_bucket_entry_name(tenant_name, bucket_name, key);
 769   return store->meta_mgr->remove_entry(bucket_meta_handler, key, &objv_tracker);
 770 }
 771
 772 static void set_err_msg(std::string *sink, std::string msg)
 773 {
 774   if (sink && !msg.empty())
 775     *sink = msg;
 776 }
 777
 778 int RGWBucket::init(RGWRados *storage, RGWBucketAdminOpState& op_state)
 779 {
 780   if (!storage)
 781     return -EINVAL;
 782
 783   store = storage;
 784
 785   rgw_user user_id = op_state.get_user_id();
 786   tenant = user_id.tenant;
 787   bucket_name = op_state.get_bucket_name();
 788   RGWUserBuckets user_buckets;
 789   RGWObjectCtx obj_ctx(store);
 790
 791   if (bucket_name.empty() && user_id.empty())
 792     return -EINVAL;
 793
 794   if (!bucket_name.empty()) {
 795     int r = store->get_bucket_info(obj_ctx, tenant, bucket_name, bucket_info, NULL);
 796     if (r < 0) {
 797       ldout(store->ctx(), 0) << "could not get bucket info for bucket=" << bucket_name << dendl;
 798       return r;
 799     }
 800
 801     op_state.set_bucket(bucket_info.bucket);
 802   }
 803
 804   if (!user_id.empty()) {
 805     int r = rgw_get_user_info_by_uid(store, user_id, user_info);
 806     if (r < 0)
 807       return r;
 808
 809     op_state.display_name = user_info.display_name;
 810   }
 811
 812   clear_failure();
 813   return 0;
 814 }
 815
 816 int RGWBucket::link(RGWBucketAdminOpState& op_state, std::string *err_msg)
 817 {
 818   if (!op_state.is_user_op()) {
 819     set_err_msg(err_msg, "empty user id");
 820     return -EINVAL;
 821   }
 822
 823   string bucket_id = op_state.get_bucket_id();
 824   if (bucket_id.empty()) {
 825     set_err_msg(err_msg, "empty bucket instance id");
 826     return -EINVAL;
 827   }
 828
 829   std::string display_name = op_state.get_user_display_name();
 830   rgw_bucket bucket = op_state.get_bucket();
 831
 832   const rgw_pool& root_pool = store->get_zone_params().domain_root;
 833   rgw_raw_obj obj(root_pool, bucket.name);
 834   RGWObjVersionTracker objv_tracker;
 835
 836   map<string, bufferlist> attrs;
 837   RGWBucketInfo bucket_info;
 838
 839   string key = bucket.name + ":" + bucket_id;
 840   RGWObjectCtx obj_ctx(store);
 841   int r = store->get_bucket_instance_info(obj_ctx, key, bucket_info, NULL, &attrs);
 842   if (r < 0) {
 843     return r;
 844   }
 845
 846   rgw_user user_id = op_state.get_user_id();
 847
 848   map<string, bufferlist>::iterator aiter = attrs.find(RGW_ATTR_ACL);
 849   if (aiter != attrs.end()) {
 850     bufferlist aclbl = aiter->second;
 851     RGWAccessControlPolicy policy;
 852     ACLOwner owner;
 853     try {
 854      bufferlist::iterator iter = aclbl.begin();
 855      ::decode(policy, iter);
 856      owner = policy.get_owner();
 857     } catch (buffer::error& err) {
 858       set_err_msg(err_msg, "couldn't decode policy");
 859       return -EIO;
 860     }
 861
 862     r = rgw_unlink_bucket(store, owner.get_id(), bucket.tenant, bucket.name, false);
 863     if (r < 0) {
 864       set_err_msg(err_msg, "could not unlink policy from user " + owner.get_id().to_str());
 865       return r;
 866     }
 867
 868     // now update the user for the bucket...
 869     if (display_name.empty()) {
 870       ldout(store->ctx(), 0) << "WARNING: user " << user_info.user_id << " has no display name set" << dendl;
 871     }
 872     policy.create_default(user_info.user_id, display_name);
 873
 874     owner = policy.get_owner();
 875     r = store->set_bucket_owner(bucket_info.bucket, owner);
 876     if (r < 0) {
 877       set_err_msg(err_msg, "failed to set bucket owner: " + cpp_strerror(-r));
 878       return r;
 879     }
 880
 881     // ...and encode the acl
 882     aclbl.clear();
 883     policy.encode(aclbl);
 884
 885     r = store->system_obj_set_attr(NULL, obj, RGW_ATTR_ACL, aclbl, &objv_tracker);
 886     if (r < 0) {
 887       return r;
 888     }
 889
 890     RGWAccessControlPolicy policy_instance;
 891     policy_instance.create_default(user_info.user_id, display_name);
 892     aclbl.clear();
 893     policy_instance.encode(aclbl);
 894
 895     string oid_bucket_instance = RGW_BUCKET_INSTANCE_MD_PREFIX + key;
 896     rgw_raw_obj obj_bucket_instance(root_pool, oid_bucket_instance);
 897     r = store->system_obj_set_attr(NULL, obj_bucket_instance, RGW_ATTR_ACL, aclbl, &objv_tracker);
 898     if (r < 0) {
 899       return r;
 900     }
 901
 902     r = rgw_link_bucket(store, user_info.user_id, bucket_info.bucket, real_time());
 903     if (r < 0) {
 904       return r;
 905     }
 906   }
 907
 908   return 0;
 909 }
 910
 911 int RGWBucket::unlink(RGWBucketAdminOpState& op_state, std::string *err_msg)
 912 {
 913   rgw_bucket bucket = op_state.get_bucket();
 914
 915   if (!op_state.is_user_op()) {
 916     set_err_msg(err_msg, "could not fetch user or user bucket info");
 917     return -EINVAL;
 918   }
 919
 920   int r = rgw_unlink_bucket(store, user_info.user_id, bucket.tenant, bucket.name);
 921   if (r < 0) {
 922     set_err_msg(err_msg, "error unlinking bucket" + cpp_strerror(-r));
 923   }
 924
 925   return r;
 926 }
 927
 928 int RGWBucket::remove(RGWBucketAdminOpState& op_state, bool bypass_gc,
 929                       bool keep_index_consistent, std::string *err_msg)
 930 {
 931   bool delete_children = op_state.will_delete_children();
 932   rgw_bucket bucket = op_state.get_bucket();
 933   int ret;
 934
 935   if (bypass_gc) {
 936     if (delete_children) {
 937       ret = rgw_remove_bucket_bypass_gc(store, bucket, op_state.get_max_aio(), keep_index_consistent);
 938     } else {
 939       set_err_msg(err_msg, "purge objects should be set for gc to be bypassed");
 940       return -EINVAL;
 941     }
 942   } else {
 943     ret = rgw_remove_bucket(store, bucket, delete_children);
 944   }
 945
 946   if (ret < 0) {
 947     set_err_msg(err_msg, "unable to remove bucket" + cpp_strerror(-ret));
 948     return ret;
 949   }
 950
 951   return 0;
 952 }
 953
 954 int RGWBucket::remove_object(RGWBucketAdminOpState& op_state, std::string *err_msg)
 955 {
 956   rgw_bucket bucket = op_state.get_bucket();
 957   std::string object_name = op_state.get_object_name();
 958
 959   rgw_obj_key key(object_name);
 960
 961   int ret = rgw_remove_object(store, bucket_info, bucket, key);
 962   if (ret < 0) {
 963     set_err_msg(err_msg, "unable to remove object" + cpp_strerror(-ret));
 964     return ret;
 965   }
 966
 967   return 0;
 968 }
 969
 970 static void dump_bucket_index(map<string, rgw_bucket_dir_entry> result,  Formatter *f)
 971 {
 972   map<string, rgw_bucket_dir_entry>::iterator iter;
 973   for (iter = result.begin(); iter != result.end(); ++iter) {
 974     f->dump_string("object", iter->first);
 975    }
 976 }
 977
 978 static void dump_bucket_usage(map<RGWObjCategory, RGWStorageStats>& stats, Formatter *formatter)
 979 {
 980   map<RGWObjCategory, RGWStorageStats>::iterator iter;
 981
 982   formatter->open_object_section("usage");
 983   for (iter = stats.begin(); iter != stats.end(); ++iter) {
 984     RGWStorageStats& s = iter->second;
 985     const char *cat_name = rgw_obj_category_name(iter->first);
 986     formatter->open_object_section(cat_name);
 987     s.dump(formatter);
 988     formatter->close_section();
 989   }
 990   formatter->close_section();
 991 }
 992
 993 static void dump_index_check(map<RGWObjCategory, RGWStorageStats> existing_stats,
 994         map<RGWObjCategory, RGWStorageStats> calculated_stats,
 995         Formatter *formatter)
 996 {
 997   formatter->open_object_section("check_result");
 998   formatter->open_object_section("existing_header");
 999   dump_bucket_usage(existing_stats, formatter);
1000   formatter->close_section();
1001   formatter->open_object_section("calculated_header");
1002   dump_bucket_usage(calculated_stats, formatter);
1003   formatter->close_section();
1004   formatter->close_section();
1005 }
1006
1007 int RGWBucket::check_bad_index_multipart(RGWBucketAdminOpState& op_state,
1008         list<rgw_obj_index_key>& objs_to_unlink, std::string *err_msg)
1009 {
1010   bool fix_index = op_state.will_fix_index();
1011   rgw_bucket bucket = op_state.get_bucket();
1012
1013   int max = 1000;
1014
1015   map<string, bool> common_prefixes;
1016
1017   bool is_truncated;
1018   map<string, bool> meta_objs;
1019   map<rgw_obj_index_key, string> all_objs;
1020
1021   RGWBucketInfo bucket_info;
1022   RGWObjectCtx obj_ctx(store);
1023   int r = store->get_bucket_instance_info(obj_ctx, bucket, bucket_info, nullptr, nullptr);
1024   if (r < 0) {
1025     ldout(store->ctx(), 0) << "ERROR: " << __func__ << "(): get_bucket_instance_info(bucket=" << bucket << ") returned r=" << r << dendl;
1026     return r;
1027   }
1028
1029   RGWRados::Bucket target(store, bucket_info);
1030   RGWRados::Bucket::List list_op(&target);
1031
1032   list_op.params.list_versions = true;
1033   list_op.params.ns = RGW_OBJ_NS_MULTIPART;
1034
1035   do {
1036     vector<rgw_bucket_dir_entry> result;
1037     int r = list_op.list_objects(max, &result, &common_prefixes, &is_truncated);
1038     if (r < 0) {
1039       set_err_msg(err_msg, "failed to list objects in bucket=" + bucket.name +
1040               " err=" +  cpp_strerror(-r));
1041
1042       return r;
1043     }
1044
1045     vector<rgw_bucket_dir_entry>::iterator iter;
1046     for (iter = result.begin(); iter != result.end(); ++iter) {
1047       rgw_obj_index_key key = iter->key;
1048       rgw_obj obj(bucket, key);
1049       string oid = obj.get_oid();
1050
1051       int pos = oid.find_last_of('.');
1052       if (pos < 0) {
1053         /* obj has no suffix */
1054         all_objs[key] = oid;
1055       } else {
1056         /* obj has suffix */
1057         string name = oid.substr(0, pos);
1058         string suffix = oid.substr(pos + 1);
1059
1060         if (suffix.compare("meta") == 0) {
1061           meta_objs[name] = true;
1062         } else {
1063           all_objs[key] = name;
1064         }
1065       }
1066     }
1067
1068   } while (is_truncated);
1069
1070   for (auto aiter = all_objs.begin(); aiter != all_objs.end(); ++aiter) {
1071     string& name = aiter->second;
1072
1073     if (meta_objs.find(name) == meta_objs.end()) {
1074       objs_to_unlink.push_back(aiter->first);
1075     }
1076   }
1077
1078   if (objs_to_unlink.empty())
1079     return 0;
1080
1081   if (fix_index) {
1082     int r = store->remove_objs_from_index(bucket_info, objs_to_unlink);
1083     if (r < 0) {
1084       set_err_msg(err_msg, "ERROR: remove_obj_from_index() returned error: " +
1085               cpp_strerror(-r));
1086
1087       return r;
1088     }
1089   }
1090
1091   return 0;
1092 }
1093
1094 int RGWBucket::check_object_index(RGWBucketAdminOpState& op_state,
1095                                   RGWFormatterFlusher& flusher,
1096                                   std::string *err_msg)
1097 {
1098
1099   bool fix_index = op_state.will_fix_index();
1100
1101   rgw_bucket bucket = op_state.get_bucket();
1102
1103   if (!fix_index) {
1104     set_err_msg(err_msg, "check-objects flag requires fix index enabled");
1105     return -EINVAL;
1106   }
1107
1108   store->cls_obj_set_bucket_tag_timeout(bucket_info, BUCKET_TAG_TIMEOUT);
1109
1110   string prefix;
1111   rgw_obj_index_key marker;
1112   bool is_truncated = true;
1113
1114   Formatter *formatter = flusher.get_formatter();
1115   formatter->open_object_section("objects");
1116   while (is_truncated) {
1117     map<string, rgw_bucket_dir_entry> result;
1118
1119     int r = store->cls_bucket_list(bucket_info, RGW_NO_SHARD, marker, prefix, 1000, true,
1120                                    result, &is_truncated, &marker,
1121                                    bucket_object_check_filter);
1122     if (r == -ENOENT) {
1123       break;
1124     } else if (r < 0 && r != -ENOENT) {
1125       set_err_msg(err_msg, "ERROR: failed operation r=" + cpp_strerror(-r));
1126     }
1127
1128
1129     dump_bucket_index(result, formatter);
1130     flusher.flush();
1131
1132   }
1133
1134   formatter->close_section();
1135
1136   store->cls_obj_set_bucket_tag_timeout(bucket_info, 0);
1137
1138   return 0;
1139 }
1140
1141
1142 int RGWBucket::check_index(RGWBucketAdminOpState& op_state,
1143         map<RGWObjCategory, RGWStorageStats>& existing_stats,
1144         map<RGWObjCategory, RGWStorageStats>& calculated_stats,
1145         std::string *err_msg)
1146 {
1147   rgw_bucket bucket = op_state.get_bucket();
1148   bool fix_index = op_state.will_fix_index();
1149
1150   int r = store->bucket_check_index(bucket_info, &existing_stats, &calculated_stats);
1151   if (r < 0) {
1152     set_err_msg(err_msg, "failed to check index error=" + cpp_strerror(-r));
1153     return r;
1154   }
1155
1156   if (fix_index) {
1157     r = store->bucket_rebuild_index(bucket_info);
1158     if (r < 0) {
1159       set_err_msg(err_msg, "failed to rebuild index err=" + cpp_strerror(-r));
1160       return r;
1161     }
1162   }
1163
1164   return 0;
1165 }
1166
1167
1168 int RGWBucket::policy_bl_to_stream(bufferlist& bl, ostream& o)
1169 {
1170   RGWAccessControlPolicy_S3 policy(g_ceph_context);
1171   bufferlist::iterator iter = bl.begin();
1172   try {
1173     policy.decode(iter);
1174   } catch (buffer::error& err) {
1175     dout(0) << "ERROR: caught buffer::error, could not decode policy" << dendl;
1176     return -EIO;
1177   }
1178   policy.to_xml(o);
1179   return 0;
1180 }
1181
1182 static int policy_decode(RGWRados *store, bufferlist& bl, RGWAccessControlPolicy& policy)
1183 {
1184   bufferlist::iterator iter = bl.begin();
1185   try {
1186     policy.decode(iter);
1187   } catch (buffer::error& err) {
1188     ldout(store->ctx(), 0) << "ERROR: caught buffer::error, could not decode policy" << dendl;
1189     return -EIO;
1190   }
1191   return 0;
1192 }
1193
1194 int RGWBucket::get_policy(RGWBucketAdminOpState& op_state, RGWAccessControlPolicy& policy)
1195 {
1196   std::string object_name = op_state.get_object_name();
1197   rgw_bucket bucket = op_state.get_bucket();
1198   RGWObjectCtx obj_ctx(store);
1199
1200   RGWBucketInfo bucket_info;
1201   map<string, bufferlist> attrs;
1202   int ret = store->get_bucket_info(obj_ctx, bucket.tenant, bucket.name, bucket_info, NULL, &attrs);
1203   if (ret < 0) {
1204     return ret;
1205   }
1206
1207   if (!object_name.empty()) {
1208     bufferlist bl;
1209     rgw_obj obj(bucket, object_name);
1210
1211     RGWRados::Object op_target(store, bucket_info, obj_ctx, obj);
1212     RGWRados::Object::Read rop(&op_target);
1213
1214     int ret = rop.get_attr(RGW_ATTR_ACL, bl);
1215     if (ret < 0)
1216       return ret;
1217
1218     return policy_decode(store, bl, policy);
1219   }
1220
1221   map<string, bufferlist>::iterator aiter = attrs.find(RGW_ATTR_ACL);
1222   if (aiter == attrs.end()) {
1223     return -ENOENT;
1224   }
1225
1226   return policy_decode(store, aiter->second, policy);
1227 }
1228
1229
1230 int RGWBucketAdminOp::get_policy(RGWRados *store, RGWBucketAdminOpState& op_state,
1231                   RGWAccessControlPolicy& policy)
1232 {
1233   RGWBucket bucket;
1234
1235   int ret = bucket.init(store, op_state);
1236   if (ret < 0)
1237     return ret;
1238
1239   ret = bucket.get_policy(op_state, policy);
1240   if (ret < 0)
1241     return ret;
1242
1243   return 0;
1244 }
1245
1246 /* Wrappers to facilitate RESTful interface */
1247
1248
1249 int RGWBucketAdminOp::get_policy(RGWRados *store, RGWBucketAdminOpState& op_state,
1250                   RGWFormatterFlusher& flusher)
1251 {
1252   RGWAccessControlPolicy policy(store->ctx());
1253
1254   int ret = get_policy(store, op_state, policy);
1255   if (ret < 0)
1256     return ret;
1257
1258   Formatter *formatter = flusher.get_formatter();
1259
1260   flusher.start(0);
1261
1262   formatter->open_object_section("policy");
1263   policy.dump(formatter);
1264   formatter->close_section();
1265
1266   flusher.flush();
1267
1268   return 0;
1269 }
1270
1271 int RGWBucketAdminOp::dump_s3_policy(RGWRados *store, RGWBucketAdminOpState& op_state,
1272                   ostream& os)
1273 {
1274   RGWAccessControlPolicy_S3 policy(store->ctx());
1275
1276   int ret = get_policy(store, op_state, policy);
1277   if (ret < 0)
1278     return ret;
1279
1280   policy.to_xml(os);
1281
1282   return 0;
1283 }
1284
1285 int RGWBucketAdminOp::unlink(RGWRados *store, RGWBucketAdminOpState& op_state)
1286 {
1287   RGWBucket bucket;
1288
1289   int ret = bucket.init(store, op_state);
1290   if (ret < 0)
1291     return ret;
1292
1293   return bucket.unlink(op_state);
1294 }
1295
1296 int RGWBucketAdminOp::link(RGWRados *store, RGWBucketAdminOpState& op_state, string *err)
1297 {
1298   RGWBucket bucket;
1299
1300   int ret = bucket.init(store, op_state);
1301   if (ret < 0)
1302     return ret;
1303
1304   return bucket.link(op_state, err);
1305
1306 }
1307
1308 int RGWBucketAdminOp::check_index(RGWRados *store, RGWBucketAdminOpState& op_state,
1309                   RGWFormatterFlusher& flusher)
1310 {
1311   int ret;
1312   map<string, rgw_bucket_dir_entry> result;
1313   map<RGWObjCategory, RGWStorageStats> existing_stats;
1314   map<RGWObjCategory, RGWStorageStats> calculated_stats;
1315   list<rgw_obj_index_key> objs_to_unlink;
1316
1317   RGWBucket bucket;
1318
1319   ret = bucket.init(store, op_state);
1320   if (ret < 0)
1321     return ret;
1322
1323   Formatter *formatter = flusher.get_formatter();
1324   flusher.start(0);
1325
1326   ret = bucket.check_bad_index_multipart(op_state, objs_to_unlink);
1327   if (ret < 0)
1328     return ret;
1329
1330   dump_mulipart_index_results(objs_to_unlink, formatter);
1331   flusher.flush();
1332
1333   ret = bucket.check_object_index(op_state, flusher);
1334   if (ret < 0)
1335     return ret;
1336
1337   ret = bucket.check_index(op_state, existing_stats, calculated_stats);
1338   if (ret < 0)
1339     return ret;
1340
1341   dump_index_check(existing_stats, calculated_stats, formatter);
1342   flusher.flush();
1343
1344   return 0;
1345 }
1346
1347 int RGWBucketAdminOp::remove_bucket(RGWRados *store, RGWBucketAdminOpState& op_state,
1348                                     bool bypass_gc, bool keep_index_consistent)
1349 {
1350   RGWBucket bucket;
1351
1352   int ret = bucket.init(store, op_state);
1353   if (ret < 0)
1354     return ret;
1355
1356   return bucket.remove(op_state, bypass_gc, keep_index_consistent);
1357 }
1358
1359 int RGWBucketAdminOp::remove_object(RGWRados *store, RGWBucketAdminOpState& op_state)
1360 {
1361   RGWBucket bucket;
1362
1363   int ret = bucket.init(store, op_state);
1364   if (ret < 0)
1365     return ret;
1366
1367   return bucket.remove_object(op_state);
1368 }
1369
1370 static int bucket_stats(RGWRados *store, const std::string& tenant_name, std::string&  bucket_name, Formatter *formatter)
1371 {
1372   RGWBucketInfo bucket_info;
1373   map<RGWObjCategory, RGWStorageStats> stats;
1374
1375   real_time mtime;
1376   RGWObjectCtx obj_ctx(store);
1377   int r = store->get_bucket_info(obj_ctx, tenant_name, bucket_name, bucket_info, &mtime);
1378   if (r < 0)
1379     return r;
1380
1381   rgw_bucket& bucket = bucket_info.bucket;
1382
1383   string bucket_ver, master_ver;
1384   string max_marker;
1385   int ret = store->get_bucket_stats(bucket_info, RGW_NO_SHARD, &bucket_ver, &master_ver, stats, &max_marker);
1386   if (ret < 0) {
1387     cerr << "error getting bucket stats ret=" << ret << std::endl;
1388     return ret;
1389   }
1390
1391   utime_t ut(mtime);
1392
1393   formatter->open_object_section("stats");
1394   formatter->dump_string("bucket", bucket.name);
1395   formatter->dump_string("zonegroup", bucket_info.zonegroup);
1396   formatter->dump_string("placement_rule", bucket_info.placement_rule);
1397   ::encode_json("explicit_placement", bucket.explicit_placement, formatter);
1398   formatter->dump_string("id", bucket.bucket_id);
1399   formatter->dump_string("marker", bucket.marker);
1400   formatter->dump_stream("index_type") << bucket_info.index_type;
1401   ::encode_json("owner", bucket_info.owner, formatter);
1402   formatter->dump_string("ver", bucket_ver);
1403   formatter->dump_string("master_ver", master_ver);
1404   formatter->dump_stream("mtime") << ut;
1405   formatter->dump_string("max_marker", max_marker);
1406   dump_bucket_usage(stats, formatter);
1407   encode_json("bucket_quota", bucket_info.quota, formatter);
1408   formatter->close_section();
1409
1410   return 0;
1411 }
1412
1413 int RGWBucketAdminOp::limit_check(RGWRados *store,
1414                                   RGWBucketAdminOpState& op_state,
1415                                   const std::list<std::string>& user_ids,
1416                                   RGWFormatterFlusher& flusher,
1417                                   bool warnings_only)
1418 {
1419   int ret = 0;
1420   const size_t max_entries =
1421     store->ctx()->_conf->rgw_list_buckets_max_chunk;
1422
1423   const size_t safe_max_objs_per_shard =
1424     store->ctx()->_conf->rgw_safe_max_objects_per_shard;
1425
1426   uint16_t shard_warn_pct =
1427     store->ctx()->_conf->rgw_shard_warning_threshold;
1428   if (shard_warn_pct > 100)
1429     shard_warn_pct = 90;
1430
1431   Formatter *formatter = flusher.get_formatter();
1432   flusher.start(0);
1433
1434   formatter->open_array_section("users");
1435
1436   for (const auto& user_id : user_ids) {
1437     formatter->open_object_section("user");
1438     formatter->dump_string("user_id", user_id);
1439     bool done;
1440     formatter->open_array_section("buckets");
1441     do {
1442       RGWUserBuckets buckets;
1443       string marker;
1444       bool is_truncated;
1445
1446       ret = rgw_read_user_buckets(store, user_id, buckets,
1447                                   marker, string(), max_entries, false,
1448                                   &is_truncated);
1449       if (ret < 0)
1450         return ret;
1451
1452       map<string, RGWBucketEnt>& m_buckets = buckets.get_buckets();
1453
1454       for (const auto& iter : m_buckets) {
1455         auto& bucket = iter.second.bucket;
1456         uint32_t num_shards = 1;
1457         uint64_t num_objects = 0;
1458
1459         /* need info for num_shards */
1460         RGWBucketInfo info;
1461         RGWObjectCtx obj_ctx(store);
1462
1463         marker = bucket.name; /* Casey's location for marker update,
1464                                * as we may now not reach the end of
1465                                * the loop body */
1466
1467         ret = store->get_bucket_info(obj_ctx, bucket.tenant, bucket.name,
1468                                      info, nullptr);
1469         if (ret < 0)
1470           continue;
1471
1472         /* need stats for num_entries */
1473         string bucket_ver, master_ver;
1474         std::map<RGWObjCategory, RGWStorageStats> stats;
1475         ret = store->get_bucket_stats(info, RGW_NO_SHARD, &bucket_ver,
1476                                       &master_ver, stats, nullptr);
1477
1478         if (ret < 0)
1479           continue;
1480
1481         for (const auto& s : stats) {
1482             num_objects += s.second.num_objects;
1483         }
1484
1485         num_shards = info.num_shards;
1486         uint64_t objs_per_shard =
1487           (num_shards) ? num_objects/num_shards : num_objects;
1488         {
1489           bool warn = false;
1490           stringstream ss;
1491           if (objs_per_shard > safe_max_objs_per_shard) {
1492             double over =
1493               100 - (safe_max_objs_per_shard/objs_per_shard * 100);
1494               ss << boost::format("OVER %4f%%") % over;
1495               warn = true;
1496           } else {
1497             double fill_pct =
1498               objs_per_shard / safe_max_objs_per_shard * 100;
1499             if (fill_pct >= shard_warn_pct) {
1500               ss << boost::format("WARN %4f%%") % fill_pct;
1501               warn = true;
1502             } else {
1503               ss << "OK";
1504             }
1505           }
1506
1507           if (warn || (! warnings_only)) {
1508             formatter->open_object_section("bucket");
1509             formatter->dump_string("bucket", bucket.name);
1510             formatter->dump_string("tenant", bucket.tenant);
1511             formatter->dump_int("num_objects", num_objects);
1512             formatter->dump_int("num_shards", num_shards);
1513             formatter->dump_int("objects_per_shard", objs_per_shard);
1514             formatter->dump_string("fill_status", ss.str());
1515             formatter->close_section();
1516           }
1517         }
1518       }
1519
1520       done = (m_buckets.size() < max_entries);
1521     } while (!done); /* foreach: bucket */
1522
1523     formatter->close_section();
1524     formatter->close_section();
1525     formatter->flush(cout);
1526
1527   } /* foreach: user_id */
1528
1529   formatter->close_section();
1530   formatter->flush(cout);
1531
1532   return ret;
1533 } /* RGWBucketAdminOp::limit_check */
1534
1535 int RGWBucketAdminOp::info(RGWRados *store, RGWBucketAdminOpState& op_state,
1536                   RGWFormatterFlusher& flusher)
1537 {
1538   RGWBucket bucket;
1539   int ret;
1540
1541   string bucket_name = op_state.get_bucket_name();
1542
1543   if (!bucket_name.empty()) {
1544     ret = bucket.init(store, op_state);
1545     if (ret < 0)
1546       return ret;
1547   }
1548
1549   Formatter *formatter = flusher.get_formatter();
1550   flusher.start(0);
1551
1552   CephContext *cct = store->ctx();
1553
1554   const size_t max_entries = cct->_conf->rgw_list_buckets_max_chunk;
1555
1556   bool show_stats = op_state.will_fetch_stats();
1557   rgw_user user_id = op_state.get_user_id();
1558   if (op_state.is_user_op()) {
1559     formatter->open_array_section("buckets");
1560
1561     RGWUserBuckets buckets;
1562     string marker;
1563     bool is_truncated = false;
1564
1565     do {
1566       ret = rgw_read_user_buckets(store, op_state.get_user_id(), buckets,
1567                                   marker, string(), max_entries, false,
1568                                   &is_truncated);
1569       if (ret < 0)
1570         return ret;
1571
1572       map<string, RGWBucketEnt>& m = buckets.get_buckets();
1573       map<string, RGWBucketEnt>::iterator iter;
1574
1575       for (iter = m.begin(); iter != m.end(); ++iter) {
1576         std::string  obj_name = iter->first;
1577         if (show_stats)
1578           bucket_stats(store, user_id.tenant, obj_name, formatter);
1579         else
1580           formatter->dump_string("bucket", obj_name);
1581
1582         marker = obj_name;
1583       }
1584
1585       flusher.flush();
1586     } while (is_truncated);
1587
1588     formatter->close_section();
1589   } else if (!bucket_name.empty()) {
1590     bucket_stats(store, user_id.tenant, bucket_name, formatter);
1591   } else {
1592     RGWAccessHandle handle;
1593
1594     formatter->open_array_section("buckets");
1595     if (store->list_buckets_init(&handle) >= 0) {
1596       rgw_bucket_dir_entry obj;
1597       while (store->list_buckets_next(obj, &handle) >= 0) {
1598         if (show_stats)
1599           bucket_stats(store, user_id.tenant, obj.key.name, formatter);
1600         else
1601           formatter->dump_string("bucket", obj.key.name);
1602       }
1603     }
1604
1605     formatter->close_section();
1606   }
1607
1608   flusher.flush();
1609
1610   return 0;
1611 }
1612
1613
1614 void rgw_data_change::dump(Formatter *f) const
1615 {
1616   string type;
1617   switch (entity_type) {
1618     case ENTITY_TYPE_BUCKET:
1619       type = "bucket";
1620       break;
1621     default:
1622       type = "unknown";
1623   }
1624   encode_json("entity_type", type, f);
1625   encode_json("key", key, f);
1626   utime_t ut(timestamp);
1627   encode_json("timestamp", ut, f);
1628 }
1629
1630 void rgw_data_change::decode_json(JSONObj *obj) {
1631   string s;
1632   JSONDecoder::decode_json("entity_type", s, obj);
1633   if (s == "bucket") {
1634     entity_type = ENTITY_TYPE_BUCKET;
1635   } else {
1636     entity_type = ENTITY_TYPE_UNKNOWN;
1637   }
1638   JSONDecoder::decode_json("key", key, obj);
1639   utime_t ut;
1640   JSONDecoder::decode_json("timestamp", ut, obj);
1641   timestamp = ut.to_real_time();
1642 }
1643
1644 void rgw_data_change_log_entry::dump(Formatter *f) const
1645 {
1646   encode_json("log_id", log_id, f);
1647   utime_t ut(log_timestamp);
1648   encode_json("log_timestamp", ut, f);
1649   encode_json("entry", entry, f);
1650 }
1651
1652 void rgw_data_change_log_entry::decode_json(JSONObj *obj) {
1653   JSONDecoder::decode_json("log_id", log_id, obj);
1654   utime_t ut;
1655   JSONDecoder::decode_json("log_timestamp", ut, obj);
1656   log_timestamp = ut.to_real_time();
1657   JSONDecoder::decode_json("entry", entry, obj);
1658 }
1659
1660 int RGWDataChangesLog::choose_oid(const rgw_bucket_shard& bs) {
1661     const string& name = bs.bucket.name;
1662     int shard_shift = (bs.shard_id > 0 ? bs.shard_id : 0);
1663     uint32_t r = (ceph_str_hash_linux(name.c_str(), name.size()) + shard_shift) % num_shards;
1664
1665     return (int)r;
1666 }
1667
1668 int RGWDataChangesLog::renew_entries()
1669 {
1670   if (!store->need_to_log_data())
1671     return 0;
1672
1673   /* we can't keep the bucket name as part of the cls_log_entry, and we need
1674    * it later, so we keep two lists under the map */
1675   map<int, pair<list<rgw_bucket_shard>, list<cls_log_entry> > > m;
1676
1677   lock.Lock();
1678   map<rgw_bucket_shard, bool> entries;
1679   entries.swap(cur_cycle);
1680   lock.Unlock();
1681
1682   map<rgw_bucket_shard, bool>::iterator iter;
1683   string section;
1684   real_time ut = real_clock::now();
1685   for (iter = entries.begin(); iter != entries.end(); ++iter) {
1686     const rgw_bucket_shard& bs = iter->first;
1687
1688     int index = choose_oid(bs);
1689
1690     cls_log_entry entry;
1691
1692     rgw_data_change change;
1693     bufferlist bl;
1694     change.entity_type = ENTITY_TYPE_BUCKET;
1695     change.key = bs.get_key();
1696     change.timestamp = ut;
1697     ::encode(change, bl);
1698
1699     store->time_log_prepare_entry(entry, ut, section, change.key, bl);
1700
1701     m[index].first.push_back(bs);
1702     m[index].second.emplace_back(std::move(entry));
1703   }
1704
1705   map<int, pair<list<rgw_bucket_shard>, list<cls_log_entry> > >::iterator miter;
1706   for (miter = m.begin(); miter != m.end(); ++miter) {
1707     list<cls_log_entry>& entries = miter->second.second;
1708
1709     real_time now = real_clock::now();
1710
1711     int ret = store->time_log_add(oids[miter->first], entries, NULL);
1712     if (ret < 0) {
1713       /* we don't really need to have a special handling for failed cases here,
1714        * as this is just an optimization. */
1715       lderr(cct) << "ERROR: store->time_log_add() returned " << ret << dendl;
1716       return ret;
1717     }
1718
1719     real_time expiration = now;
1720     expiration += make_timespan(cct->_conf->rgw_data_log_window);
1721
1722     list<rgw_bucket_shard>& buckets = miter->second.first;
1723     list<rgw_bucket_shard>::iterator liter;
1724     for (liter = buckets.begin(); liter != buckets.end(); ++liter) {
1725       update_renewed(*liter, expiration);
1726     }
1727   }
1728
1729   return 0;
1730 }
1731
1732 void RGWDataChangesLog::_get_change(const rgw_bucket_shard& bs, ChangeStatusPtr& status)
1733 {
1734   assert(lock.is_locked());
1735   if (!changes.find(bs, status)) {
1736     status = ChangeStatusPtr(new ChangeStatus);
1737     changes.add(bs, status);
1738   }
1739 }
1740
1741 void RGWDataChangesLog::register_renew(rgw_bucket_shard& bs)
1742 {
1743   Mutex::Locker l(lock);
1744   cur_cycle[bs] = true;
1745 }
1746
1747 void RGWDataChangesLog::update_renewed(rgw_bucket_shard& bs, real_time& expiration)
1748 {
1749   Mutex::Locker l(lock);
1750   ChangeStatusPtr status;
1751   _get_change(bs, status);
1752
1753   ldout(cct, 20) << "RGWDataChangesLog::update_renewd() bucket_name=" << bs.bucket.name << " shard_id=" << bs.shard_id << " expiration=" << expiration << dendl;
1754   status->cur_expiration = expiration;
1755 }
1756
1757 int RGWDataChangesLog::get_log_shard_id(rgw_bucket& bucket, int shard_id) {
1758   rgw_bucket_shard bs(bucket, shard_id);
1759
1760   return choose_oid(bs);
1761 }
1762
1763 int RGWDataChangesLog::add_entry(rgw_bucket& bucket, int shard_id) {
1764   if (!store->need_to_log_data())
1765     return 0;
1766
1767   rgw_bucket_shard bs(bucket, shard_id);
1768
1769   int index = choose_oid(bs);
1770   mark_modified(index, bs);
1771
1772   lock.Lock();
1773
1774   ChangeStatusPtr status;
1775   _get_change(bs, status);
1776
1777   lock.Unlock();
1778
1779   real_time now = real_clock::now();
1780
1781   status->lock->Lock();
1782
1783   ldout(cct, 20) << "RGWDataChangesLog::add_entry() bucket.name=" << bucket.name << " shard_id=" << shard_id << " now=" << now << " cur_expiration=" << status->cur_expiration << dendl;
1784
1785   if (now < status->cur_expiration) {
1786     /* no need to send, recently completed */
1787     status->lock->Unlock();
1788
1789     register_renew(bs);
1790     return 0;
1791   }
1792
1793   RefCountedCond *cond;
1794
1795   if (status->pending) {
1796     cond = status->cond;
1797
1798     assert(cond);
1799
1800     status->cond->get();
1801     status->lock->Unlock();
1802
1803     int ret = cond->wait();
1804     cond->put();
1805     if (!ret) {
1806       register_renew(bs);
1807     }
1808     return ret;
1809   }
1810
1811   status->cond = new RefCountedCond;
1812   status->pending = true;
1813
1814   string& oid = oids[index];
1815   real_time expiration;
1816
1817   int ret;
1818
1819   do {
1820     status->cur_sent = now;
1821
1822     expiration = now;
1823     expiration += ceph::make_timespan(cct->_conf->rgw_data_log_window);
1824
1825     status->lock->Unlock();
1826
1827     bufferlist bl;
1828     rgw_data_change change;
1829     change.entity_type = ENTITY_TYPE_BUCKET;
1830     change.key = bs.get_key();
1831     change.timestamp = now;
1832     ::encode(change, bl);
1833     string section;
1834
1835     ldout(cct, 20) << "RGWDataChangesLog::add_entry() sending update with now=" << now << " cur_expiration=" << expiration << dendl;
1836
1837     ret = store->time_log_add(oid, now, section, change.key, bl);
1838
1839     now = real_clock::now();
1840
1841     status->lock->Lock();
1842
1843   } while (!ret && real_clock::now() > expiration);
1844
1845   cond = status->cond;
1846
1847   status->pending = false;
1848   status->cur_expiration = status->cur_sent; /* time of when operation started, not completed */
1849   status->cur_expiration += make_timespan(cct->_conf->rgw_data_log_window);
1850   status->cond = NULL;
1851   status->lock->Unlock();
1852
1853   cond->done(ret);
1854   cond->put();
1855
1856   return ret;
1857 }
1858
1859 int RGWDataChangesLog::list_entries(int shard, const real_time& start_time, const real_time& end_time, int max_entries,
1860                                     list<rgw_data_change_log_entry>& entries,
1861                                     const string& marker,
1862                                     string *out_marker,
1863                                     bool *truncated) {
1864   if (shard >= num_shards)
1865     return -EINVAL;
1866
1867   list<cls_log_entry> log_entries;
1868
1869   int ret = store->time_log_list(oids[shard], start_time, end_time,
1870                                  max_entries, log_entries, marker,
1871                                  out_marker, truncated);
1872   if (ret < 0)
1873     return ret;
1874
1875   list<cls_log_entry>::iterator iter;
1876   for (iter = log_entries.begin(); iter != log_entries.end(); ++iter) {
1877     rgw_data_change_log_entry log_entry;
1878     log_entry.log_id = iter->id;
1879     real_time rt = iter->timestamp.to_real_time();
1880     log_entry.log_timestamp = rt;
1881     bufferlist::iterator liter = iter->data.begin();
1882     try {
1883       ::decode(log_entry.entry, liter);
1884     } catch (buffer::error& err) {
1885       lderr(cct) << "ERROR: failed to decode data changes log entry" << dendl;
1886       return -EIO;
1887     }
1888     entries.push_back(log_entry);
1889   }
1890
1891   return 0;
1892 }
1893
1894 int RGWDataChangesLog::list_entries(const real_time& start_time, const real_time& end_time, int max_entries,
1895              list<rgw_data_change_log_entry>& entries, LogMarker& marker, bool *ptruncated) {
1896   bool truncated;
1897   entries.clear();
1898
1899   for (; marker.shard < num_shards && (int)entries.size() < max_entries;
1900        marker.shard++, marker.marker.clear()) {
1901     int ret = list_entries(marker.shard, start_time, end_time, max_entries - entries.size(), entries,
1902                            marker.marker, NULL, &truncated);
1903     if (ret == -ENOENT) {
1904       continue;
1905     }
1906     if (ret < 0) {
1907       return ret;
1908     }
1909     if (truncated) {
1910       *ptruncated = true;
1911       return 0;
1912     }
1913   }
1914
1915   *ptruncated = (marker.shard < num_shards);
1916
1917   return 0;
1918 }
1919
1920 int RGWDataChangesLog::get_info(int shard_id, RGWDataChangesLogInfo *info)
1921 {
1922   if (shard_id >= num_shards)
1923     return -EINVAL;
1924
1925   string oid = oids[shard_id];
1926
1927   cls_log_header header;
1928
1929   int ret = store->time_log_info(oid, &header);
1930   if ((ret < 0) && (ret != -ENOENT))
1931     return ret;
1932
1933   info->marker = header.max_marker;
1934   info->last_update = header.max_time.to_real_time();
1935
1936   return 0;
1937 }
1938
1939 int RGWDataChangesLog::trim_entries(int shard_id, const real_time& start_time, const real_time& end_time,
1940                                     const string& start_marker, const string& end_marker)
1941 {
1942   int ret;
1943
1944   if (shard_id > num_shards)
1945     return -EINVAL;
1946
1947   ret = store->time_log_trim(oids[shard_id], start_time, end_time, start_marker, end_marker);
1948
1949   if (ret == -ENOENT || ret == -ENODATA)
1950     ret = 0;
1951
1952   return ret;
1953 }
1954
1955 int RGWDataChangesLog::trim_entries(const real_time& start_time, const real_time& end_time,
1956                                     const string& start_marker, const string& end_marker)
1957 {
1958   for (int shard = 0; shard < num_shards; shard++) {
1959     int ret = store->time_log_trim(oids[shard], start_time, end_time, start_marker, end_marker);
1960     if (ret == -ENOENT || ret == -ENODATA) {
1961       continue;
1962     }
1963     if (ret < 0)
1964       return ret;
1965   }
1966
1967   return 0;
1968 }
1969
1970 bool RGWDataChangesLog::going_down()
1971 {
1972   return down_flag;
1973 }
1974
1975 RGWDataChangesLog::~RGWDataChangesLog() {
1976   down_flag = true;
1977   renew_thread->stop();
1978   renew_thread->join();
1979   delete renew_thread;
1980   delete[] oids;
1981 }
1982
1983 void *RGWDataChangesLog::ChangesRenewThread::entry() {
1984   do {
1985     dout(2) << "RGWDataChangesLog::ChangesRenewThread: start" << dendl;
1986     int r = log->renew_entries();
1987     if (r < 0) {
1988       dout(0) << "ERROR: RGWDataChangesLog::renew_entries returned error r=" << r << dendl;
1989     }
1990
1991     if (log->going_down())
1992       break;
1993
1994     int interval = cct->_conf->rgw_data_log_window * 3 / 4;
1995     lock.Lock();
1996     cond.WaitInterval(lock, utime_t(interval, 0));
1997     lock.Unlock();
1998   } while (!log->going_down());
1999
2000   return NULL;
2001 }
2002
2003 void RGWDataChangesLog::ChangesRenewThread::stop()
2004 {
2005   Mutex::Locker l(lock);
2006   cond.Signal();
2007 }
2008
2009 void RGWDataChangesLog::mark_modified(int shard_id, const rgw_bucket_shard& bs)
2010 {
2011   auto key = bs.get_key();
2012   modified_lock.get_read();
2013   map<int, set<string> >::iterator iter = modified_shards.find(shard_id);
2014   if (iter != modified_shards.end()) {
2015     set<string>& keys = iter->second;
2016     if (keys.find(key) != keys.end()) {
2017       modified_lock.unlock();
2018       return;
2019     }
2020   }
2021   modified_lock.unlock();
2022
2023   RWLock::WLocker wl(modified_lock);
2024   modified_shards[shard_id].insert(key);
2025 }
2026
2027 void RGWDataChangesLog::read_clear_modified(map<int, set<string> > &modified)
2028 {
2029   RWLock::WLocker wl(modified_lock);
2030   modified.swap(modified_shards);
2031   modified_shards.clear();
2032 }
2033
2034 void RGWBucketCompleteInfo::dump(Formatter *f) const {
2035   encode_json("bucket_info", info, f);
2036   encode_json("attrs", attrs, f);
2037 }
2038
2039 void RGWBucketCompleteInfo::decode_json(JSONObj *obj) {
2040   JSONDecoder::decode_json("bucket_info", info, obj);
2041   JSONDecoder::decode_json("attrs", attrs, obj);
2042 }
2043
2044 class RGWBucketMetadataHandler : public RGWMetadataHandler {
2045
2046 public:
2047   string get_type() override { return "bucket"; }
2048
2049   int get(RGWRados *store, string& entry, RGWMetadataObject **obj) override {
2050     RGWObjVersionTracker ot;
2051     RGWBucketEntryPoint be;
2052
2053     real_time mtime;
2054     map<string, bufferlist> attrs;
2055     RGWObjectCtx obj_ctx(store);
2056
2057     string tenant_name, bucket_name;
2058     parse_bucket(entry, &tenant_name, &bucket_name);
2059     int ret = store->get_bucket_entrypoint_info(obj_ctx, tenant_name, bucket_name, be, &ot, &mtime, &attrs);
2060     if (ret < 0)
2061       return ret;
2062
2063     RGWBucketEntryMetadataObject *mdo = new RGWBucketEntryMetadataObject(be, ot.read_version, mtime);
2064
2065     *obj = mdo;
2066
2067     return 0;
2068   }
2069
2070   int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker,
2071           real_time mtime, JSONObj *obj, sync_type_t sync_type) override {
2072     RGWBucketEntryPoint be, old_be;
2073     try {
2074       decode_json_obj(be, obj);
2075     } catch (JSONDecoder::err& e) {
2076       return -EINVAL;
2077     }
2078
2079     real_time orig_mtime;
2080     map<string, bufferlist> attrs;
2081
2082     RGWObjVersionTracker old_ot;
2083     RGWObjectCtx obj_ctx(store);
2084
2085     string tenant_name, bucket_name;
2086     parse_bucket(entry, &tenant_name, &bucket_name);
2087     int ret = store->get_bucket_entrypoint_info(obj_ctx, tenant_name, bucket_name, old_be, &old_ot, &orig_mtime, &attrs);
2088     if (ret < 0 && ret != -ENOENT)
2089       return ret;
2090
2091     // are we actually going to perform this put, or is it too old?
2092     if (ret != -ENOENT &&
2093         !check_versions(old_ot.read_version, orig_mtime,
2094                         objv_tracker.write_version, mtime, sync_type)) {
2095       return STATUS_NO_APPLY;
2096     }
2097
2098     objv_tracker.read_version = old_ot.read_version; /* maintain the obj version we just read */
2099
2100     ret = store->put_bucket_entrypoint_info(tenant_name, bucket_name, be, false, objv_tracker, mtime, &attrs);
2101     if (ret < 0)
2102       return ret;
2103
2104     /* link bucket */
2105     if (be.linked) {
2106       ret = rgw_link_bucket(store, be.owner, be.bucket, be.creation_time, false);
2107     } else {
2108       ret = rgw_unlink_bucket(store, be.owner, be.bucket.tenant, be.bucket.name, false);
2109     }
2110
2111     return ret;
2112   }
2113
2114   struct list_keys_info {
2115     RGWRados *store;
2116     RGWListRawObjsCtx ctx;
2117   };
2118
2119   int remove(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker) override {
2120     RGWBucketEntryPoint be;
2121     RGWObjectCtx obj_ctx(store);
2122
2123     string tenant_name, bucket_name;
2124     parse_bucket(entry, &tenant_name, &bucket_name);
2125     int ret = store->get_bucket_entrypoint_info(obj_ctx, tenant_name, bucket_name, be, &objv_tracker, NULL, NULL);
2126     if (ret < 0)
2127       return ret;
2128
2129     /*
2130      * We're unlinking the bucket but we don't want to update the entrypoint here - we're removing
2131      * it immediately and don't want to invalidate our cached objv_version or the bucket obj removal
2132      * will incorrectly fail.
2133      */
2134     ret = rgw_unlink_bucket(store, be.owner, tenant_name, bucket_name, false);
2135     if (ret < 0) {
2136       lderr(store->ctx()) << "could not unlink bucket=" << entry << " owner=" << be.owner << dendl;
2137     }
2138
2139     ret = rgw_bucket_delete_bucket_obj(store, tenant_name, bucket_name, objv_tracker);
2140     if (ret < 0) {
2141       lderr(store->ctx()) << "could not delete bucket=" << entry << dendl;
2142     }
2143     /* idempotent */
2144     return 0;
2145   }
2146
2147   void get_pool_and_oid(RGWRados *store, const string& key, rgw_pool& pool, string& oid) override {
2148     oid = key;
2149     pool = store->get_zone_params().domain_root;
2150   }
2151
2152   int list_keys_init(RGWRados *store, void **phandle) override
2153   {
2154     list_keys_info *info = new list_keys_info;
2155
2156     info->store = store;
2157
2158     *phandle = (void *)info;
2159
2160     return 0;
2161   }
2162
2163   int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated) override {
2164     list_keys_info *info = static_cast<list_keys_info *>(handle);
2165
2166     string no_filter;
2167
2168     keys.clear();
2169
2170     RGWRados *store = info->store;
2171
2172     list<string> unfiltered_keys;
2173
2174     int ret = store->list_raw_objects(store->get_zone_params().domain_root, no_filter,
2175                                       max, info->ctx, unfiltered_keys, truncated);
2176     if (ret < 0 && ret != -ENOENT)
2177       return ret;
2178     if (ret == -ENOENT) {
2179       if (truncated)
2180         *truncated = false;
2181       return 0;
2182     }
2183
2184     // now filter out the system entries
2185     list<string>::iterator iter;
2186     for (iter = unfiltered_keys.begin(); iter != unfiltered_keys.end(); ++iter) {
2187       string& k = *iter;
2188
2189       if (k[0] != '.') {
2190         keys.push_back(k);
2191       }
2192     }
2193
2194     return 0;
2195   }
2196
2197   void list_keys_complete(void *handle) override {
2198     list_keys_info *info = static_cast<list_keys_info *>(handle);
2199     delete info;
2200   }
2201 };
2202
2203 class RGWBucketInstanceMetadataHandler : public RGWMetadataHandler {
2204
2205 public:
2206   string get_type() override { return "bucket.instance"; }
2207
2208   int get(RGWRados *store, string& oid, RGWMetadataObject **obj) override {
2209     RGWBucketCompleteInfo bci;
2210
2211     real_time mtime;
2212     RGWObjectCtx obj_ctx(store);
2213
2214     int ret = store->get_bucket_instance_info(obj_ctx, oid, bci.info, &mtime, &bci.attrs);
2215     if (ret < 0)
2216       return ret;
2217
2218     RGWBucketInstanceMetadataObject *mdo = new RGWBucketInstanceMetadataObject(bci, bci.info.objv_tracker.read_version, mtime);
2219
2220     *obj = mdo;
2221
2222     return 0;
2223   }
2224
2225   int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker,
2226           real_time mtime, JSONObj *obj, sync_type_t sync_type) override {
2227     RGWBucketCompleteInfo bci, old_bci;
2228     try {
2229       decode_json_obj(bci, obj);
2230     } catch (JSONDecoder::err& e) {
2231       return -EINVAL;
2232     }
2233
2234     real_time orig_mtime;
2235     RGWObjectCtx obj_ctx(store);
2236
2237     int ret = store->get_bucket_instance_info(obj_ctx, entry, old_bci.info,
2238             &orig_mtime, &old_bci.attrs);
2239     bool exists = (ret != -ENOENT);
2240     if (ret < 0 && exists)
2241       return ret;
2242
2243     if (!exists || old_bci.info.bucket.bucket_id != bci.info.bucket.bucket_id) {
2244       /* a new bucket, we need to select a new bucket placement for it */
2245       auto key(entry);
2246       rgw_bucket_instance_oid_to_key(key);
2247       string tenant_name;
2248       string bucket_name;
2249       string bucket_instance;
2250       parse_bucket(key, &tenant_name, &bucket_name, &bucket_instance);
2251
2252       RGWZonePlacementInfo rule_info;
2253       bci.info.bucket.name = bucket_name;
2254       bci.info.bucket.bucket_id = bucket_instance;
2255       bci.info.bucket.tenant = tenant_name;
2256       ret = store->select_bucket_location_by_rule(bci.info.placement_rule, &rule_info);
2257       if (ret < 0) {
2258         ldout(store->ctx(), 0) << "ERROR: select_bucket_placement() returned " << ret << dendl;
2259         return ret;
2260       }
2261       bci.info.index_type = rule_info.index_type;
2262     } else {
2263       /* existing bucket, keep its placement */
2264       bci.info.bucket.explicit_placement = old_bci.info.bucket.explicit_placement;
2265       bci.info.placement_rule = old_bci.info.placement_rule;
2266     }
2267
2268     // are we actually going to perform this put, or is it too old?
2269     if (exists &&
2270         !check_versions(old_bci.info.objv_tracker.read_version, orig_mtime,
2271                         objv_tracker.write_version, mtime, sync_type)) {
2272       objv_tracker.read_version = old_bci.info.objv_tracker.read_version;
2273       return STATUS_NO_APPLY;
2274     }
2275
2276     /* record the read version (if any), store the new version */
2277     bci.info.objv_tracker.read_version = old_bci.info.objv_tracker.read_version;
2278     bci.info.objv_tracker.write_version = objv_tracker.write_version;
2279
2280     ret = store->put_bucket_instance_info(bci.info, false, mtime, &bci.attrs);
2281     if (ret < 0)
2282       return ret;
2283
2284     objv_tracker = bci.info.objv_tracker;
2285
2286     ret = store->init_bucket_index(bci.info, bci.info.num_shards);
2287     if (ret < 0)
2288       return ret;
2289
2290     return STATUS_APPLIED;
2291   }
2292
2293   struct list_keys_info {
2294     RGWRados *store;
2295     RGWListRawObjsCtx ctx;
2296   };
2297
2298   int remove(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker) override {
2299     RGWBucketInfo info;
2300     RGWObjectCtx obj_ctx(store);
2301
2302     int ret = store->get_bucket_instance_info(obj_ctx, entry, info, NULL, NULL);
2303     if (ret < 0 && ret != -ENOENT)
2304       return ret;
2305
2306     return rgw_bucket_instance_remove_entry(store, entry, &info.objv_tracker);
2307   }
2308
2309   void get_pool_and_oid(RGWRados *store, const string& key, rgw_pool& pool, string& oid) override {
2310     oid = RGW_BUCKET_INSTANCE_MD_PREFIX + key;
2311     rgw_bucket_instance_key_to_oid(oid);
2312     pool = store->get_zone_params().domain_root;
2313   }
2314
2315   int list_keys_init(RGWRados *store, void **phandle) override
2316   {
2317     list_keys_info *info = new list_keys_info;
2318
2319     info->store = store;
2320
2321     *phandle = (void *)info;
2322
2323     return 0;
2324   }
2325
2326   int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated) override {
2327     list_keys_info *info = static_cast<list_keys_info *>(handle);
2328
2329     string no_filter;
2330
2331     keys.clear();
2332
2333     RGWRados *store = info->store;
2334
2335     list<string> unfiltered_keys;
2336
2337     int ret = store->list_raw_objects(store->get_zone_params().domain_root, no_filter,
2338                                       max, info->ctx, unfiltered_keys, truncated);
2339     if (ret < 0 && ret != -ENOENT)
2340       return ret;
2341     if (ret == -ENOENT) {
2342       if (truncated)
2343         *truncated = false;
2344       return 0;
2345     }
2346
2347     constexpr int prefix_size = sizeof(RGW_BUCKET_INSTANCE_MD_PREFIX) - 1;
2348     // now filter in the relevant entries
2349     list<string>::iterator iter;
2350     for (iter = unfiltered_keys.begin(); iter != unfiltered_keys.end(); ++iter) {
2351       string& k = *iter;
2352
2353       if (k.compare(0, prefix_size, RGW_BUCKET_INSTANCE_MD_PREFIX) == 0) {
2354         auto oid = k.substr(prefix_size);
2355         rgw_bucket_instance_oid_to_key(oid);
2356         keys.emplace_back(std::move(oid));
2357       }
2358     }
2359
2360     return 0;
2361   }
2362
2363   void list_keys_complete(void *handle) override {
2364     list_keys_info *info = static_cast<list_keys_info *>(handle);
2365     delete info;
2366   }
2367
2368   /*
2369    * hash entry for mdlog placement. Use the same hash key we'd have for the bucket entry
2370    * point, so that the log entries end up at the same log shard, so that we process them
2371    * in order
2372    */
2373   void get_hash_key(const string& section, const string& key, string& hash_key) override {
2374     string k;
2375     int pos = key.find(':');
2376     if (pos < 0)
2377       k = key;
2378     else
2379       k = key.substr(0, pos);
2380     hash_key = "bucket:" + k;
2381   }
2382 };
2383
2384 void rgw_bucket_init(RGWMetadataManager *mm)
2385 {
2386   bucket_meta_handler = new RGWBucketMetadataHandler;
2387   mm->register_handler(bucket_meta_handler);
2388   bucket_instance_meta_handler = new RGWBucketInstanceMetadataHandler;
2389   mm->register_handler(bucket_instance_meta_handler);
2390 }