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