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