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