ceph/src/cls/rgw/cls_rgw_client.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 "cls/rgw/cls_rgw_const.h"
   7 #include "cls/rgw/cls_rgw_client.h"
   8
   9 #include "common/debug.h"
  10
  11 using namespace librados;
  12
  13 const string BucketIndexShardsManager::KEY_VALUE_SEPARATOR = "#";
  14 const string BucketIndexShardsManager::SHARDS_SEPARATOR = ",";
  15
  16 /**
  17  * This class represents the bucket index object operation callback context.
  18  */
  19 template <typename T>
  20 class ClsBucketIndexOpCtx : public ObjectOperationCompletion {
  21 private:
  22   T *data;
  23   int *ret_code;
  24 public:
  25   ClsBucketIndexOpCtx(T* _data, int *_ret_code) : data(_data), ret_code(_ret_code) { ceph_assert(data); }
  26   ~ClsBucketIndexOpCtx() override {}
  27   void handle_completion(int r, bufferlist& outbl) override {
  28     if (r >= 0) {
  29       try {
  30         auto iter = outbl.cbegin();
  31         decode((*data), iter);
  32       } catch (buffer::error& err) {
  33         r = -EIO;
  34       }
  35     }
  36     if (ret_code) {
  37       *ret_code = r;
  38     }
  39   }
  40 };
  41
  42 void BucketIndexAioManager::do_completion(int id) {
  43   std::lock_guard l{lock};
  44
  45   map<int, librados::AioCompletion*>::iterator iter = pendings.find(id);
  46   ceph_assert(iter != pendings.end());
  47   completions[id] = iter->second;
  48   pendings.erase(iter);
  49
  50   // If the caller needs a list of finished objects, store them
  51   // for further processing
  52   map<int, string>::iterator miter = pending_objs.find(id);
  53   if (miter != pending_objs.end()) {
  54     completion_objs[id] = miter->second;
  55     pending_objs.erase(miter);
  56   }
  57
  58   cond.notify_all();
  59 }
  60
  61 bool BucketIndexAioManager::wait_for_completions(int valid_ret_code,
  62     int *num_completions, int *ret_code, map<int, string> *objs) {
  63   std::unique_lock locker{lock};
  64   if (pendings.empty() && completions.empty()) {
  65     return false;
  66   }
  67
  68   if (completions.empty()) {
  69     // Wait for AIO completion
  70     cond.wait(locker);
  71   }
  72
  73   // Clear the completed AIOs
  74   map<int, librados::AioCompletion*>::iterator iter = completions.begin();
  75   for (; iter != completions.end(); ++iter) {
  76     int r = iter->second->get_return_value();
  77     if (objs && r == 0) { /* update list of successfully completed objs */
  78       map<int, string>::iterator liter = completion_objs.find(iter->first);
  79       if (liter != completion_objs.end()) {
  80         (*objs)[liter->first] = liter->second;
  81       }
  82     }
  83     if (ret_code && (r < 0 && r != valid_ret_code))
  84       (*ret_code) = r;
  85     iter->second->release();
  86   }
  87   if (num_completions)
  88     (*num_completions) = completions.size();
  89   completions.clear();
  90
  91   return true;
  92 }
  93
  94 // note: currently only called by tesing code
  95 void cls_rgw_bucket_init_index(ObjectWriteOperation& o)
  96 {
  97   bufferlist in;
  98   o.exec(RGW_CLASS, RGW_BUCKET_INIT_INDEX, in);
  99 }
 100
 101 static bool issue_bucket_index_init_op(librados::IoCtx& io_ctx,
 102                                        const string& oid,
 103                                        BucketIndexAioManager *manager) {
 104   bufferlist in;
 105   librados::ObjectWriteOperation op;
 106   op.create(true);
 107   op.exec(RGW_CLASS, RGW_BUCKET_INIT_INDEX, in);
 108   return manager->aio_operate(io_ctx, oid, &op);
 109 }
 110
 111 static bool issue_bucket_index_clean_op(librados::IoCtx& io_ctx,
 112                                         const string& oid,
 113                                         BucketIndexAioManager *manager) {
 114   bufferlist in;
 115   librados::ObjectWriteOperation op;
 116   op.remove();
 117   return manager->aio_operate(io_ctx, oid, &op);
 118 }
 119
 120 static bool issue_bucket_set_tag_timeout_op(librados::IoCtx& io_ctx,
 121     const string& oid, uint64_t timeout, BucketIndexAioManager *manager) {
 122   bufferlist in;
 123   rgw_cls_tag_timeout_op call;
 124   call.tag_timeout = timeout;
 125   encode(call, in);
 126   ObjectWriteOperation op;
 127   op.exec(RGW_CLASS, RGW_BUCKET_SET_TAG_TIMEOUT, in);
 128   return manager->aio_operate(io_ctx, oid, &op);
 129 }
 130
 131 int CLSRGWIssueBucketIndexInit::issue_op(int shard_id, const string& oid)
 132 {
 133   return issue_bucket_index_init_op(io_ctx, oid, &manager);
 134 }
 135
 136 void CLSRGWIssueBucketIndexInit::cleanup()
 137 {
 138   // Do best effort removal
 139   for (auto citer = objs_container.begin(); citer != iter; ++citer) {
 140     io_ctx.remove(citer->second);
 141   }
 142 }
 143
 144 int CLSRGWIssueBucketIndexClean::issue_op(int shard_id, const string& oid)
 145 {
 146   return issue_bucket_index_clean_op(io_ctx, oid, &manager);
 147 }
 148
 149 int CLSRGWIssueSetTagTimeout::issue_op(int shard_id, const string& oid)
 150 {
 151   return issue_bucket_set_tag_timeout_op(io_ctx, oid, tag_timeout, &manager);
 152 }
 153
 154 void cls_rgw_bucket_update_stats(librados::ObjectWriteOperation& o,
 155                                  bool absolute,
 156                                  const map<RGWObjCategory, rgw_bucket_category_stats>& stats)
 157 {
 158   rgw_cls_bucket_update_stats_op call;
 159   call.absolute = absolute;
 160   call.stats = stats;
 161   bufferlist in;
 162   encode(call, in);
 163   o.exec(RGW_CLASS, RGW_BUCKET_UPDATE_STATS, in);
 164 }
 165
 166 void cls_rgw_bucket_prepare_op(ObjectWriteOperation& o, RGWModifyOp op, string& tag,
 167                                const cls_rgw_obj_key& key, const string& locator, bool log_op,
 168                                uint16_t bilog_flags, rgw_zone_set& zones_trace)
 169 {
 170   rgw_cls_obj_prepare_op call;
 171   call.op = op;
 172   call.tag = tag;
 173   call.key = key;
 174   call.locator = locator;
 175   call.log_op = log_op;
 176   call.bilog_flags = bilog_flags;
 177   call.zones_trace = zones_trace;
 178   bufferlist in;
 179   encode(call, in);
 180   o.exec(RGW_CLASS, RGW_BUCKET_PREPARE_OP, in);
 181 }
 182
 183 void cls_rgw_bucket_complete_op(ObjectWriteOperation& o, RGWModifyOp op, string& tag,
 184                                 rgw_bucket_entry_ver& ver,
 185                                 const cls_rgw_obj_key& key,
 186                                 rgw_bucket_dir_entry_meta& dir_meta,
 187                                 list<cls_rgw_obj_key> *remove_objs, bool log_op,
 188                                 uint16_t bilog_flags,
 189                                 rgw_zone_set *zones_trace)
 190 {
 191
 192   bufferlist in;
 193   rgw_cls_obj_complete_op call;
 194   call.op = op;
 195   call.tag = tag;
 196   call.key = key;
 197   call.ver = ver;
 198   call.meta = dir_meta;
 199   call.log_op = log_op;
 200   call.bilog_flags = bilog_flags;
 201   if (remove_objs)
 202     call.remove_objs = *remove_objs;
 203   if (zones_trace) {
 204     call.zones_trace = *zones_trace;
 205   }
 206   encode(call, in);
 207   o.exec(RGW_CLASS, RGW_BUCKET_COMPLETE_OP, in);
 208 }
 209
 210 void cls_rgw_bucket_list_op(librados::ObjectReadOperation& op,
 211                             const cls_rgw_obj_key& start_obj,
 212                             const std::string& filter_prefix,
 213                             const std::string& delimiter,
 214                             uint32_t num_entries,
 215                             bool list_versions,
 216                             rgw_cls_list_ret* result)
 217 {
 218   bufferlist in;
 219   rgw_cls_list_op call;
 220   call.start_obj = start_obj;
 221   call.filter_prefix = filter_prefix;
 222   call.delimiter = delimiter;
 223   call.num_entries = num_entries;
 224   call.list_versions = list_versions;
 225   encode(call, in);
 226
 227   op.exec(RGW_CLASS, RGW_BUCKET_LIST, in,
 228           new ClsBucketIndexOpCtx<rgw_cls_list_ret>(result, NULL));
 229 }
 230
 231 static bool issue_bucket_list_op(librados::IoCtx& io_ctx,
 232                                  const string& oid,
 233                                  const cls_rgw_obj_key& start_obj,
 234                                  const string& filter_prefix,
 235                                  const string& delimiter,
 236                                  uint32_t num_entries,
 237                                  bool list_versions,
 238                                  BucketIndexAioManager *manager,
 239                                  rgw_cls_list_ret *pdata)
 240 {
 241   librados::ObjectReadOperation op;
 242   cls_rgw_bucket_list_op(op,
 243                          start_obj, filter_prefix, delimiter,
 244                          num_entries, list_versions, pdata);
 245   return manager->aio_operate(io_ctx, oid, &op);
 246 }
 247
 248 int CLSRGWIssueBucketList::issue_op(int shard_id, const string& oid)
 249 {
 250   return issue_bucket_list_op(io_ctx, oid,
 251                               start_obj, filter_prefix, delimiter,
 252                               num_entries, list_versions, &manager,
 253                               &result[shard_id]);
 254 }
 255
 256 void cls_rgw_remove_obj(librados::ObjectWriteOperation& o, list<string>& keep_attr_prefixes)
 257 {
 258   bufferlist in;
 259   rgw_cls_obj_remove_op call;
 260   call.keep_attr_prefixes = keep_attr_prefixes;
 261   encode(call, in);
 262   o.exec(RGW_CLASS, RGW_OBJ_REMOVE, in);
 263 }
 264
 265 void cls_rgw_obj_store_pg_ver(librados::ObjectWriteOperation& o, const string& attr)
 266 {
 267   bufferlist in;
 268   rgw_cls_obj_store_pg_ver_op call;
 269   call.attr = attr;
 270   encode(call, in);
 271   o.exec(RGW_CLASS, RGW_OBJ_STORE_PG_VER, in);
 272 }
 273
 274 void cls_rgw_obj_check_attrs_prefix(librados::ObjectOperation& o, const string& prefix, bool fail_if_exist)
 275 {
 276   bufferlist in;
 277   rgw_cls_obj_check_attrs_prefix call;
 278   call.check_prefix = prefix;
 279   call.fail_if_exist = fail_if_exist;
 280   encode(call, in);
 281   o.exec(RGW_CLASS, RGW_OBJ_CHECK_ATTRS_PREFIX, in);
 282 }
 283
 284 void cls_rgw_obj_check_mtime(librados::ObjectOperation& o, const real_time& mtime, bool high_precision_time, RGWCheckMTimeType type)
 285 {
 286   bufferlist in;
 287   rgw_cls_obj_check_mtime call;
 288   call.mtime = mtime;
 289   call.high_precision_time = high_precision_time;
 290   call.type = type;
 291   encode(call, in);
 292   o.exec(RGW_CLASS, RGW_OBJ_CHECK_MTIME, in);
 293 }
 294
 295 int cls_rgw_bi_get(librados::IoCtx& io_ctx, const string oid,
 296                    BIIndexType index_type, cls_rgw_obj_key& key,
 297                    rgw_cls_bi_entry *entry)
 298 {
 299   bufferlist in, out;
 300   rgw_cls_bi_get_op call;
 301   call.key = key;
 302   call.type = index_type;
 303   encode(call, in);
 304   int r = io_ctx.exec(oid, RGW_CLASS, RGW_BI_GET, in, out);
 305   if (r < 0)
 306     return r;
 307
 308   rgw_cls_bi_get_ret op_ret;
 309   auto iter = out.cbegin();
 310   try {
 311     decode(op_ret, iter);
 312   } catch (buffer::error& err) {
 313     return -EIO;
 314   }
 315
 316   *entry = op_ret.entry;
 317
 318   return 0;
 319 }
 320
 321 int cls_rgw_bi_put(librados::IoCtx& io_ctx, const string oid, rgw_cls_bi_entry& entry)
 322 {
 323   bufferlist in, out;
 324   rgw_cls_bi_put_op call;
 325   call.entry = entry;
 326   encode(call, in);
 327   int r = io_ctx.exec(oid, RGW_CLASS, RGW_BI_PUT, in, out);
 328   if (r < 0)
 329     return r;
 330
 331   return 0;
 332 }
 333
 334 void cls_rgw_bi_put(ObjectWriteOperation& op, const string oid, rgw_cls_bi_entry& entry)
 335 {
 336   bufferlist in, out;
 337   rgw_cls_bi_put_op call;
 338   call.entry = entry;
 339   encode(call, in);
 340   op.exec(RGW_CLASS, RGW_BI_PUT, in);
 341 }
 342
 343 int cls_rgw_bi_list(librados::IoCtx& io_ctx, const string oid,
 344                    const string& name, const string& marker, uint32_t max,
 345                    list<rgw_cls_bi_entry> *entries, bool *is_truncated)
 346 {
 347   bufferlist in, out;
 348   rgw_cls_bi_list_op call;
 349   call.name = name;
 350   call.marker = marker;
 351   call.max = max;
 352   encode(call, in);
 353   int r = io_ctx.exec(oid, RGW_CLASS, RGW_BI_LIST, in, out);
 354   if (r < 0)
 355     return r;
 356
 357   rgw_cls_bi_list_ret op_ret;
 358   auto iter = out.cbegin();
 359   try {
 360     decode(op_ret, iter);
 361   } catch (buffer::error& err) {
 362     return -EIO;
 363   }
 364
 365   entries->swap(op_ret.entries);
 366   *is_truncated = op_ret.is_truncated;
 367
 368   return 0;
 369 }
 370
 371 int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid,
 372                             const cls_rgw_obj_key& key, bufferlist& olh_tag,
 373                             bool delete_marker, const string& op_tag, rgw_bucket_dir_entry_meta *meta,
 374                             uint64_t olh_epoch, ceph::real_time unmod_since, bool high_precision_time, bool log_op, rgw_zone_set& zones_trace)
 375 {
 376   librados::ObjectWriteOperation op;
 377   cls_rgw_bucket_link_olh(op, key, olh_tag, delete_marker, op_tag, meta,
 378                           olh_epoch, unmod_since, high_precision_time, log_op,
 379                           zones_trace);
 380
 381   return io_ctx.operate(oid, &op);
 382 }
 383
 384
 385 void cls_rgw_bucket_link_olh(librados::ObjectWriteOperation& op, const cls_rgw_obj_key& key,
 386                             bufferlist& olh_tag, bool delete_marker,
 387                             const string& op_tag, rgw_bucket_dir_entry_meta *meta,
 388                             uint64_t olh_epoch, ceph::real_time unmod_since, bool high_precision_time, bool log_op, rgw_zone_set& zones_trace)
 389 {
 390   bufferlist in, out;
 391   rgw_cls_link_olh_op call;
 392   call.key = key;
 393   call.olh_tag = string(olh_tag.c_str(), olh_tag.length());
 394   call.op_tag = op_tag;
 395   call.delete_marker = delete_marker;
 396   if (meta) {
 397     call.meta = *meta;
 398   }
 399   call.olh_epoch = olh_epoch;
 400   call.log_op = log_op;
 401   call.unmod_since = unmod_since;
 402   call.high_precision_time = high_precision_time;
 403   call.zones_trace = zones_trace;
 404   encode(call, in);
 405   op.exec(RGW_CLASS, RGW_BUCKET_LINK_OLH, in);
 406 }
 407
 408 int cls_rgw_bucket_unlink_instance(librados::IoCtx& io_ctx, const string& oid,
 409                                    const cls_rgw_obj_key& key, const string& op_tag,
 410                                    const string& olh_tag, uint64_t olh_epoch, bool log_op, rgw_zone_set& zones_trace)
 411 {
 412   librados::ObjectWriteOperation op;
 413   cls_rgw_bucket_unlink_instance(op, key, op_tag, olh_tag, olh_epoch, log_op, zones_trace);
 414   int r = io_ctx.operate(oid, &op);
 415   if (r < 0)
 416     return r;
 417
 418   return 0;
 419 }
 420
 421 void cls_rgw_bucket_unlink_instance(librados::ObjectWriteOperation& op,
 422                                    const cls_rgw_obj_key& key, const string& op_tag,
 423                                    const string& olh_tag, uint64_t olh_epoch, bool log_op, rgw_zone_set& zones_trace)
 424 {
 425   bufferlist in, out;
 426   rgw_cls_unlink_instance_op call;
 427   call.key = key;
 428   call.op_tag = op_tag;
 429   call.olh_epoch = olh_epoch;
 430   call.olh_tag = olh_tag;
 431   call.log_op = log_op;
 432   call.zones_trace = zones_trace;
 433   encode(call, in);
 434   op.exec(RGW_CLASS, RGW_BUCKET_UNLINK_INSTANCE, in);
 435 }
 436
 437 void cls_rgw_get_olh_log(librados::ObjectReadOperation& op, const cls_rgw_obj_key& olh, uint64_t ver_marker, const string& olh_tag, rgw_cls_read_olh_log_ret& log_ret, int& op_ret)
 438 {
 439   bufferlist in;
 440   rgw_cls_read_olh_log_op call;
 441   call.olh = olh;
 442   call.ver_marker = ver_marker;
 443   call.olh_tag = olh_tag;
 444   encode(call, in);
 445   op.exec(RGW_CLASS, RGW_BUCKET_READ_OLH_LOG, in, new ClsBucketIndexOpCtx<rgw_cls_read_olh_log_ret>(&log_ret, &op_ret));
 446 }
 447
 448 int cls_rgw_get_olh_log(IoCtx& io_ctx, string& oid, const cls_rgw_obj_key& olh, uint64_t ver_marker,
 449                         const string& olh_tag,
 450                         rgw_cls_read_olh_log_ret& log_ret)
 451 {
 452   int op_ret = 0;
 453   librados::ObjectReadOperation op;
 454   cls_rgw_get_olh_log(op, olh, ver_marker, olh_tag, log_ret, op_ret);
 455   int r = io_ctx.operate(oid, &op, NULL);
 456   if (r < 0) {
 457     return r;
 458   }
 459   if (op_ret < 0) {
 460     return op_ret;
 461   }
 462
 463  return r;
 464 }
 465
 466 void cls_rgw_trim_olh_log(librados::ObjectWriteOperation& op, const cls_rgw_obj_key& olh, uint64_t ver, const string& olh_tag)
 467 {
 468   bufferlist in;
 469   rgw_cls_trim_olh_log_op call;
 470   call.olh = olh;
 471   call.ver = ver;
 472   call.olh_tag = olh_tag;
 473   encode(call, in);
 474   op.exec(RGW_CLASS, RGW_BUCKET_TRIM_OLH_LOG, in);
 475 }
 476
 477 int cls_rgw_clear_olh(IoCtx& io_ctx, string& oid, const cls_rgw_obj_key& olh, const string& olh_tag)
 478 {
 479   librados::ObjectWriteOperation op;
 480   cls_rgw_clear_olh(op, olh, olh_tag);
 481
 482   return io_ctx.operate(oid, &op);
 483 }
 484
 485 void cls_rgw_clear_olh(librados::ObjectWriteOperation& op, const cls_rgw_obj_key& olh, const string& olh_tag)
 486 {
 487   bufferlist in;
 488   rgw_cls_bucket_clear_olh_op call;
 489   call.key = olh;
 490   call.olh_tag = olh_tag;
 491   encode(call, in);
 492   op.exec(RGW_CLASS, RGW_BUCKET_CLEAR_OLH, in);
 493 }
 494
 495 void cls_rgw_bilog_list(librados::ObjectReadOperation& op,
 496                         const std::string& marker, uint32_t max,
 497                         cls_rgw_bi_log_list_ret *pdata, int *ret)
 498 {
 499   cls_rgw_bi_log_list_op call;
 500   call.marker = marker;
 501   call.max = max;
 502
 503   bufferlist in;
 504   encode(call, in);
 505   op.exec(RGW_CLASS, RGW_BI_LOG_LIST, in, new ClsBucketIndexOpCtx<cls_rgw_bi_log_list_ret>(pdata, ret));
 506 }
 507
 508 static bool issue_bi_log_list_op(librados::IoCtx& io_ctx, const string& oid, int shard_id,
 509                                  BucketIndexShardsManager& marker_mgr, uint32_t max,
 510                                  BucketIndexAioManager *manager,
 511                                  cls_rgw_bi_log_list_ret *pdata)
 512 {
 513   librados::ObjectReadOperation op;
 514   cls_rgw_bilog_list(op, marker_mgr.get(shard_id, ""), max, pdata, nullptr);
 515   return manager->aio_operate(io_ctx, oid, &op);
 516 }
 517
 518 int CLSRGWIssueBILogList::issue_op(int shard_id, const string& oid)
 519 {
 520   return issue_bi_log_list_op(io_ctx, oid, shard_id, marker_mgr, max, &manager, &result[shard_id]);
 521 }
 522
 523 void cls_rgw_bilog_trim(librados::ObjectWriteOperation& op,
 524                         const std::string& start_marker,
 525                         const std::string& end_marker)
 526 {
 527   cls_rgw_bi_log_trim_op call;
 528   call.start_marker = start_marker;
 529   call.end_marker = end_marker;
 530
 531   bufferlist in;
 532   encode(call, in);
 533   op.exec(RGW_CLASS, RGW_BI_LOG_TRIM, in);
 534 }
 535
 536 static bool issue_bi_log_trim(librados::IoCtx& io_ctx, const string& oid, int shard_id,
 537                               BucketIndexShardsManager& start_marker_mgr,
 538                               BucketIndexShardsManager& end_marker_mgr, BucketIndexAioManager *manager) {
 539   cls_rgw_bi_log_trim_op call;
 540   librados::ObjectWriteOperation op;
 541   cls_rgw_bilog_trim(op, start_marker_mgr.get(shard_id, ""),
 542                      end_marker_mgr.get(shard_id, ""));
 543   return manager->aio_operate(io_ctx, oid, &op);
 544 }
 545
 546 int CLSRGWIssueBILogTrim::issue_op(int shard_id, const string& oid)
 547 {
 548   return issue_bi_log_trim(io_ctx, oid, shard_id, start_marker_mgr, end_marker_mgr, &manager);
 549 }
 550
 551 static bool issue_bucket_check_index_op(IoCtx& io_ctx, const string& oid, BucketIndexAioManager *manager,
 552     rgw_cls_check_index_ret *pdata) {
 553   bufferlist in;
 554   librados::ObjectReadOperation op;
 555   op.exec(RGW_CLASS, RGW_BUCKET_CHECK_INDEX, in, new ClsBucketIndexOpCtx<rgw_cls_check_index_ret>(
 556         pdata, NULL));
 557   return manager->aio_operate(io_ctx, oid, &op);
 558 }
 559
 560 int CLSRGWIssueBucketCheck::issue_op(int shard_id, const string& oid)
 561 {
 562   return issue_bucket_check_index_op(io_ctx, oid, &manager, &result[shard_id]);
 563 }
 564
 565 static bool issue_bucket_rebuild_index_op(IoCtx& io_ctx, const string& oid,
 566     BucketIndexAioManager *manager) {
 567   bufferlist in;
 568   librados::ObjectWriteOperation op;
 569   op.exec(RGW_CLASS, RGW_BUCKET_REBUILD_INDEX, in);
 570   return manager->aio_operate(io_ctx, oid, &op);
 571 }
 572
 573 int CLSRGWIssueBucketRebuild::issue_op(int shard_id, const string& oid)
 574 {
 575   return issue_bucket_rebuild_index_op(io_ctx, oid, &manager);
 576 }
 577
 578 void cls_rgw_encode_suggestion(char op, rgw_bucket_dir_entry& dirent, bufferlist& updates)
 579 {
 580   updates.append(op);
 581   encode(dirent, updates);
 582 }
 583
 584 void cls_rgw_suggest_changes(ObjectWriteOperation& o, bufferlist& updates)
 585 {
 586   o.exec(RGW_CLASS, RGW_DIR_SUGGEST_CHANGES, updates);
 587 }
 588
 589 int CLSRGWIssueGetDirHeader::issue_op(int shard_id, const string& oid)
 590 {
 591   cls_rgw_obj_key empty_key;
 592   string empty_prefix;
 593   string empty_delimiter;
 594   return issue_bucket_list_op(io_ctx, oid,
 595                               empty_key, empty_prefix, empty_delimiter,
 596                               0, false, &manager, &result[shard_id]);
 597 }
 598
 599 static bool issue_resync_bi_log(librados::IoCtx& io_ctx, const string& oid, BucketIndexAioManager *manager)
 600 {
 601   bufferlist in;
 602   librados::ObjectWriteOperation op;
 603   op.exec(RGW_CLASS, RGW_BI_LOG_RESYNC, in);
 604   return manager->aio_operate(io_ctx, oid, &op);
 605 }
 606
 607 int CLSRGWIssueResyncBucketBILog::issue_op(int shard_id, const string& oid)
 608 {
 609   return issue_resync_bi_log(io_ctx, oid, &manager);
 610 }
 611
 612 static bool issue_bi_log_stop(librados::IoCtx& io_ctx, const string& oid, BucketIndexAioManager *manager)
 613 {
 614   bufferlist in;
 615   librados::ObjectWriteOperation op;
 616   op.exec(RGW_CLASS, RGW_BI_LOG_STOP, in);
 617   return manager->aio_operate(io_ctx, oid, &op);
 618 }
 619
 620 int CLSRGWIssueBucketBILogStop::issue_op(int shard_id, const string& oid)
 621 {
 622   return issue_bi_log_stop(io_ctx, oid, &manager);
 623 }
 624
 625 class GetDirHeaderCompletion : public ObjectOperationCompletion {
 626   RGWGetDirHeader_CB *ret_ctx;
 627 public:
 628   explicit GetDirHeaderCompletion(RGWGetDirHeader_CB *_ctx) : ret_ctx(_ctx) {}
 629   ~GetDirHeaderCompletion() override {
 630     ret_ctx->put();
 631   }
 632   void handle_completion(int r, bufferlist& outbl) override {
 633     rgw_cls_list_ret ret;
 634     try {
 635       auto iter = outbl.cbegin();
 636       decode(ret, iter);
 637     } catch (buffer::error& err) {
 638       r = -EIO;
 639     }
 640
 641     ret_ctx->handle_response(r, ret.dir.header);
 642   }
 643 };
 644
 645 int cls_rgw_get_dir_header_async(IoCtx& io_ctx, string& oid, RGWGetDirHeader_CB *ctx)
 646 {
 647   bufferlist in, out;
 648   rgw_cls_list_op call;
 649   call.num_entries = 0;
 650   encode(call, in);
 651   ObjectReadOperation op;
 652   GetDirHeaderCompletion *cb = new GetDirHeaderCompletion(ctx);
 653   op.exec(RGW_CLASS, RGW_BUCKET_LIST, in, cb);
 654   AioCompletion *c = librados::Rados::aio_create_completion(nullptr, nullptr);
 655   int r = io_ctx.aio_operate(oid, c, &op, NULL);
 656   c->release();
 657   if (r < 0)
 658     return r;
 659
 660   return 0;
 661 }
 662
 663 int cls_rgw_usage_log_read(IoCtx& io_ctx, const string& oid, const string& user, const string& bucket,
 664                            uint64_t start_epoch, uint64_t end_epoch, uint32_t max_entries,
 665                            string& read_iter, map<rgw_user_bucket, rgw_usage_log_entry>& usage,
 666                            bool *is_truncated)
 667 {
 668   if (is_truncated)
 669     *is_truncated = false;
 670
 671   bufferlist in, out;
 672   rgw_cls_usage_log_read_op call;
 673   call.start_epoch = start_epoch;
 674   call.end_epoch = end_epoch;
 675   call.owner = user;
 676   call.max_entries = max_entries;
 677   call.bucket = bucket;
 678   call.iter = read_iter;
 679   encode(call, in);
 680   int r = io_ctx.exec(oid, RGW_CLASS, RGW_USER_USAGE_LOG_READ, in, out);
 681   if (r < 0)
 682     return r;
 683
 684   try {
 685     rgw_cls_usage_log_read_ret result;
 686     auto iter = out.cbegin();
 687     decode(result, iter);
 688     read_iter = result.next_iter;
 689     if (is_truncated)
 690       *is_truncated = result.truncated;
 691
 692     usage = result.usage;
 693   } catch (buffer::error& e) {
 694     return -EINVAL;
 695   }
 696
 697   return 0;
 698 }
 699
 700 int cls_rgw_usage_log_trim(IoCtx& io_ctx, const string& oid, const string& user, const string& bucket,
 701                            uint64_t start_epoch, uint64_t end_epoch)
 702 {
 703   bufferlist in;
 704   rgw_cls_usage_log_trim_op call;
 705   call.start_epoch = start_epoch;
 706   call.end_epoch = end_epoch;
 707   call.user = user;
 708   call.bucket = bucket;
 709   encode(call, in);
 710
 711   bool done = false;
 712   do {
 713     ObjectWriteOperation op;
 714     op.exec(RGW_CLASS, RGW_USER_USAGE_LOG_TRIM, in);
 715     int r = io_ctx.operate(oid, &op);
 716     if (r == -ENODATA)
 717       done = true;
 718     else if (r < 0)
 719       return r;
 720   } while (!done);
 721
 722   return 0;
 723 }
 724
 725 void cls_rgw_usage_log_trim(librados::ObjectWriteOperation& op, const string& user, const string& bucket, uint64_t start_epoch, uint64_t end_epoch)
 726 {
 727   bufferlist in;
 728   rgw_cls_usage_log_trim_op call;
 729   call.start_epoch = start_epoch;
 730   call.end_epoch = end_epoch;
 731   call.user = user;
 732   call.bucket = bucket;
 733   encode(call, in);
 734
 735   op.exec(RGW_CLASS, RGW_USER_USAGE_LOG_TRIM, in);
 736 }
 737
 738 void cls_rgw_usage_log_clear(ObjectWriteOperation& op)
 739 {
 740   bufferlist in;
 741   op.exec(RGW_CLASS, RGW_USAGE_LOG_CLEAR, in);
 742 }
 743
 744 void cls_rgw_usage_log_add(ObjectWriteOperation& op, rgw_usage_log_info& info)
 745 {
 746   bufferlist in;
 747   rgw_cls_usage_log_add_op call;
 748   call.info = info;
 749   encode(call, in);
 750   op.exec(RGW_CLASS, RGW_USER_USAGE_LOG_ADD, in);
 751 }
 752
 753 /* garbage collection */
 754
 755 void cls_rgw_gc_set_entry(ObjectWriteOperation& op, uint32_t expiration_secs, cls_rgw_gc_obj_info& info)
 756 {
 757   bufferlist in;
 758   cls_rgw_gc_set_entry_op call;
 759   call.expiration_secs = expiration_secs;
 760   call.info = info;
 761   encode(call, in);
 762   op.exec(RGW_CLASS, RGW_GC_SET_ENTRY, in);
 763 }
 764
 765 void cls_rgw_gc_defer_entry(ObjectWriteOperation& op, uint32_t expiration_secs, const string& tag)
 766 {
 767   bufferlist in;
 768   cls_rgw_gc_defer_entry_op call;
 769   call.expiration_secs = expiration_secs;
 770   call.tag = tag;
 771   encode(call, in);
 772   op.exec(RGW_CLASS, RGW_GC_DEFER_ENTRY, in);
 773 }
 774
 775 int cls_rgw_gc_list(IoCtx& io_ctx, string& oid, string& marker, uint32_t max, bool expired_only,
 776                     list<cls_rgw_gc_obj_info>& entries, bool *truncated, string& next_marker)
 777 {
 778   bufferlist in, out;
 779   cls_rgw_gc_list_op call;
 780   call.marker = marker;
 781   call.max = max;
 782   call.expired_only = expired_only;
 783   encode(call, in);
 784   int r = io_ctx.exec(oid, RGW_CLASS, RGW_GC_LIST, in, out);
 785   if (r < 0)
 786     return r;
 787
 788   cls_rgw_gc_list_ret ret;
 789   try {
 790     auto iter = out.cbegin();
 791     decode(ret, iter);
 792   } catch (buffer::error& err) {
 793     return -EIO;
 794   }
 795
 796   entries.swap(ret.entries);
 797
 798   if (truncated)
 799     *truncated = ret.truncated;
 800   next_marker = std::move(ret.next_marker);
 801   return r;
 802 }
 803
 804 void cls_rgw_gc_remove(librados::ObjectWriteOperation& op, const vector<string>& tags)
 805 {
 806   bufferlist in;
 807   cls_rgw_gc_remove_op call;
 808   call.tags = tags;
 809   encode(call, in);
 810   op.exec(RGW_CLASS, RGW_GC_REMOVE, in);
 811 }
 812
 813 int cls_rgw_lc_get_head(IoCtx& io_ctx, const string& oid, cls_rgw_lc_obj_head& head)
 814 {
 815   bufferlist in, out;
 816   int r = io_ctx.exec(oid, RGW_CLASS, RGW_LC_GET_HEAD, in, out);
 817   if (r < 0)
 818     return r;
 819
 820   cls_rgw_lc_get_head_ret ret;
 821   try {
 822     auto iter = out.cbegin();
 823     decode(ret, iter);
 824   } catch (buffer::error& err) {
 825     return -EIO;
 826   }
 827   head = ret.head;
 828
 829  return r;
 830 }
 831
 832 int cls_rgw_lc_put_head(IoCtx& io_ctx, const string& oid, cls_rgw_lc_obj_head& head)
 833 {
 834   bufferlist in, out;
 835   cls_rgw_lc_put_head_op call;
 836   call.head = head;
 837   encode(call, in);
 838   int r = io_ctx.exec(oid, RGW_CLASS, RGW_LC_PUT_HEAD, in, out);
 839   return r;
 840 }
 841
 842 int cls_rgw_lc_get_next_entry(IoCtx& io_ctx, const string& oid, string& marker,
 843                               cls_rgw_lc_entry& entry)
 844 {
 845   bufferlist in, out;
 846   cls_rgw_lc_get_next_entry_op call;
 847   call.marker = marker;
 848   encode(call, in);
 849   int r = io_ctx.exec(oid, RGW_CLASS, RGW_LC_GET_NEXT_ENTRY, in, out);
 850   if (r < 0)
 851     return r;
 852
 853   cls_rgw_lc_get_next_entry_ret ret;
 854   try {
 855     auto iter = out.cbegin();
 856     decode(ret, iter);
 857   } catch (buffer::error& err) {
 858     return -EIO;
 859   }
 860   entry = ret.entry;
 861
 862  return r;
 863 }
 864
 865 int cls_rgw_lc_rm_entry(IoCtx& io_ctx, const string& oid,
 866                         const cls_rgw_lc_entry& entry)
 867 {
 868   bufferlist in, out;
 869   cls_rgw_lc_rm_entry_op call;
 870   call.entry = entry;
 871   encode(call, in);
 872   int r = io_ctx.exec(oid, RGW_CLASS, RGW_LC_RM_ENTRY, in, out);
 873  return r;
 874 }
 875
 876 int cls_rgw_lc_set_entry(IoCtx& io_ctx, const string& oid,
 877                          const cls_rgw_lc_entry& entry)
 878 {
 879   bufferlist in, out;
 880   cls_rgw_lc_set_entry_op call;
 881   call.entry = entry;
 882   encode(call, in);
 883   int r = io_ctx.exec(oid, RGW_CLASS, RGW_LC_SET_ENTRY, in, out);
 884   return r;
 885 }
 886
 887 int cls_rgw_lc_get_entry(IoCtx& io_ctx, const string& oid,
 888                          const std::string& marker, cls_rgw_lc_entry& entry)
 889 {
 890   bufferlist in, out;
 891   cls_rgw_lc_get_entry_op call{marker};;
 892   encode(call, in);
 893   int r = io_ctx.exec(oid, RGW_CLASS, RGW_LC_GET_ENTRY, in, out);
 894
 895   if (r < 0) {
 896     return r;
 897   }
 898
 899   cls_rgw_lc_get_entry_ret ret;
 900   try {
 901     auto iter = out.cbegin();
 902     decode(ret, iter);
 903   } catch (buffer::error& err) {
 904     return -EIO;
 905   }
 906
 907   entry = std::move(ret.entry);
 908   return r;
 909 }
 910
 911 int cls_rgw_lc_list(IoCtx& io_ctx, const string& oid,
 912                     const string& marker,
 913                     uint32_t max_entries,
 914                     vector<cls_rgw_lc_entry>& entries)
 915 {
 916   bufferlist in, out;
 917   cls_rgw_lc_list_entries_op op;
 918
 919   entries.clear();
 920
 921   op.marker = marker;
 922   op.max_entries = max_entries;
 923
 924   encode(op, in);
 925
 926   int r = io_ctx.exec(oid, RGW_CLASS, RGW_LC_LIST_ENTRIES, in, out);
 927   if (r < 0)
 928     return r;
 929
 930   cls_rgw_lc_list_entries_ret ret;
 931   try {
 932     auto iter = out.cbegin();
 933     decode(ret, iter);
 934   } catch (buffer::error& err) {
 935     return -EIO;
 936   }
 937
 938   std::sort(std::begin(ret.entries), std::end(ret.entries),
 939             [](const cls_rgw_lc_entry& a, const cls_rgw_lc_entry& b)
 940               { return a.bucket < b.bucket; });
 941   entries = std::move(ret.entries);
 942   return r;
 943 }
 944
 945 void cls_rgw_reshard_add(librados::ObjectWriteOperation& op, const cls_rgw_reshard_entry& entry)
 946 {
 947   bufferlist in;
 948   cls_rgw_reshard_add_op call;
 949   call.entry = entry;
 950   encode(call, in);
 951   op.exec(RGW_CLASS, RGW_RESHARD_ADD, in);
 952 }
 953
 954 int cls_rgw_reshard_list(librados::IoCtx& io_ctx, const string& oid, string& marker, uint32_t max,
 955                          list<cls_rgw_reshard_entry>& entries, bool* is_truncated)
 956 {
 957   bufferlist in, out;
 958   cls_rgw_reshard_list_op call;
 959   call.marker = marker;
 960   call.max = max;
 961   encode(call, in);
 962   int r = io_ctx.exec(oid, RGW_CLASS, RGW_RESHARD_LIST, in, out);
 963   if (r < 0)
 964     return r;
 965
 966   cls_rgw_reshard_list_ret op_ret;
 967   auto iter = out.cbegin();
 968   try {
 969     decode(op_ret, iter);
 970   } catch (buffer::error& err) {
 971     return -EIO;
 972   }
 973
 974   entries.swap(op_ret.entries);
 975   *is_truncated = op_ret.is_truncated;
 976
 977   return 0;
 978 }
 979
 980 int cls_rgw_reshard_get(librados::IoCtx& io_ctx, const string& oid, cls_rgw_reshard_entry& entry)
 981 {
 982   bufferlist in, out;
 983   cls_rgw_reshard_get_op call;
 984   call.entry = entry;
 985   encode(call, in);
 986   int r = io_ctx.exec(oid, RGW_CLASS, RGW_RESHARD_GET, in, out);
 987   if (r < 0)
 988     return r;
 989
 990   cls_rgw_reshard_get_ret op_ret;
 991   auto iter = out.cbegin();
 992   try {
 993     decode(op_ret, iter);
 994   } catch (buffer::error& err) {
 995     return -EIO;
 996   }
 997
 998   entry = op_ret.entry;
 999
1000   return 0;
1001 }
1002
1003 void cls_rgw_reshard_remove(librados::ObjectWriteOperation& op, const cls_rgw_reshard_entry& entry)
1004 {
1005   bufferlist in;
1006   cls_rgw_reshard_remove_op call;
1007   call.tenant = entry.tenant;
1008   call.bucket_name = entry.bucket_name;
1009   call.bucket_id = entry.bucket_id;
1010   encode(call, in);
1011   op.exec(RGW_CLASS, RGW_RESHARD_REMOVE, in);
1012 }
1013
1014 int cls_rgw_set_bucket_resharding(librados::IoCtx& io_ctx, const string& oid,
1015                                   const cls_rgw_bucket_instance_entry& entry)
1016 {
1017   bufferlist in, out;
1018   cls_rgw_set_bucket_resharding_op call;
1019   call.entry = entry;
1020   encode(call, in);
1021   return io_ctx.exec(oid, RGW_CLASS, RGW_SET_BUCKET_RESHARDING, in, out);
1022 }
1023
1024 int cls_rgw_clear_bucket_resharding(librados::IoCtx& io_ctx, const string& oid)
1025 {
1026   bufferlist in, out;
1027   cls_rgw_clear_bucket_resharding_op call;
1028   encode(call, in);
1029   return io_ctx.exec(oid, RGW_CLASS, RGW_CLEAR_BUCKET_RESHARDING, in, out);
1030 }
1031
1032 int cls_rgw_get_bucket_resharding(librados::IoCtx& io_ctx, const string& oid,
1033                                   cls_rgw_bucket_instance_entry *entry)
1034 {
1035   bufferlist in, out;
1036   cls_rgw_get_bucket_resharding_op call;
1037   encode(call, in);
1038   int r= io_ctx.exec(oid, RGW_CLASS, RGW_GET_BUCKET_RESHARDING, in, out);
1039   if (r < 0)
1040     return r;
1041
1042   cls_rgw_get_bucket_resharding_ret op_ret;
1043   auto iter = out.cbegin();
1044   try {
1045     decode(op_ret, iter);
1046   } catch (buffer::error& err) {
1047     return -EIO;
1048   }
1049
1050   *entry = op_ret.new_instance;
1051
1052   return 0;
1053 }
1054
1055 void cls_rgw_guard_bucket_resharding(librados::ObjectOperation& op, int ret_err)
1056 {
1057   bufferlist in, out;
1058   cls_rgw_guard_bucket_resharding_op call;
1059   call.ret_err = ret_err;
1060   encode(call, in);
1061   op.exec(RGW_CLASS, RGW_GUARD_BUCKET_RESHARDING, in);
1062 }
1063
1064 static bool issue_set_bucket_resharding(librados::IoCtx& io_ctx, const string& oid,
1065                                         const cls_rgw_bucket_instance_entry& entry,
1066                                         BucketIndexAioManager *manager) {
1067   bufferlist in;
1068   cls_rgw_set_bucket_resharding_op call;
1069   call.entry = entry;
1070   encode(call, in);
1071   librados::ObjectWriteOperation op;
1072   op.exec(RGW_CLASS, RGW_SET_BUCKET_RESHARDING, in);
1073   return manager->aio_operate(io_ctx, oid, &op);
1074 }
1075
1076 int CLSRGWIssueSetBucketResharding::issue_op(int shard_id, const string& oid)
1077 {
1078   return issue_set_bucket_resharding(io_ctx, oid, entry, &manager);
1079 }