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