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