ceph/src/cls/user/cls_user.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 "include/utime.h"
   7 #include "objclass/objclass.h"
   8
   9 #include "cls_user_ops.h"
  10
  11 CLS_VER(1,0)
  12 CLS_NAME(user)
  13
  14 static int write_entry(cls_method_context_t hctx, const string& key, const cls_user_bucket_entry& entry)
  15 {
  16   bufferlist bl;
  17   ::encode(entry, bl);
  18
  19   int ret = cls_cxx_map_set_val(hctx, key, &bl);
  20   if (ret < 0)
  21     return ret;
  22
  23   return 0;
  24 }
  25
  26 static int remove_entry(cls_method_context_t hctx, const string& key)
  27 {
  28   int ret = cls_cxx_map_remove_key(hctx, key);
  29   if (ret < 0)
  30     return ret;
  31
  32   return 0;
  33 }
  34
  35 static void get_key_by_bucket_name(const string& bucket_name, string *key)
  36 {
  37   *key = bucket_name;
  38 }
  39
  40 static int get_existing_bucket_entry(cls_method_context_t hctx, const string& bucket_name,
  41                                      cls_user_bucket_entry& entry)
  42 {
  43   if (bucket_name.empty()) {
  44     return -EINVAL;
  45   }
  46
  47   string key;
  48   get_key_by_bucket_name(bucket_name, &key);
  49
  50   bufferlist bl;
  51   int rc = cls_cxx_map_get_val(hctx, key, &bl);
  52   if (rc < 0) {
  53     CLS_LOG(10, "could not read entry %s", key.c_str());
  54     return rc;
  55   }
  56   try {
  57     bufferlist::iterator iter = bl.begin();
  58     ::decode(entry, iter);
  59   } catch (buffer::error& err) {
  60     CLS_LOG(0, "ERROR: failed to decode entry %s", key.c_str());
  61     return -EIO;
  62   }
  63
  64   return 0;
  65 }
  66
  67 static int read_header(cls_method_context_t hctx, cls_user_header *header)
  68 {
  69   bufferlist bl;
  70
  71   int ret = cls_cxx_map_read_header(hctx, &bl);
  72   if (ret < 0)
  73     return ret;
  74
  75   if (bl.length() == 0) {
  76     *header = cls_user_header();
  77     return 0;
  78   }
  79
  80   try {
  81     ::decode(*header, bl);
  82   } catch (buffer::error& err) {
  83     CLS_LOG(0, "ERROR: failed to decode user header");
  84     return -EIO;
  85   }
  86
  87   return 0;
  88 }
  89
  90 static void add_header_stats(cls_user_stats *stats, cls_user_bucket_entry& entry)
  91 {
  92   stats->total_entries += entry.count;
  93   stats->total_bytes += entry.size;
  94   stats->total_bytes_rounded += entry.size_rounded;
  95 }
  96
  97 static void dec_header_stats(cls_user_stats *stats, cls_user_bucket_entry& entry)
  98 {
  99   stats->total_bytes -= entry.size;
 100   stats->total_bytes_rounded -= entry.size_rounded;
 101   stats->total_entries -= entry.count;
 102 }
 103
 104 static void apply_entry_stats(const cls_user_bucket_entry& src_entry, cls_user_bucket_entry *target_entry)
 105 {
 106   target_entry->size = src_entry.size;
 107   target_entry->size_rounded = src_entry.size_rounded;
 108   target_entry->count = src_entry.count;
 109 }
 110
 111 static int cls_user_set_buckets_info(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 112 {
 113   bufferlist::iterator in_iter = in->begin();
 114
 115   cls_user_set_buckets_op op;
 116   try {
 117     ::decode(op, in_iter);
 118   } catch (buffer::error& err) {
 119     CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
 120     return -EINVAL;
 121   }
 122
 123   cls_user_header header;
 124   int ret = read_header(hctx, &header);
 125   if (ret < 0) {
 126     CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
 127     return ret;
 128   }
 129
 130   for (list<cls_user_bucket_entry>::iterator iter = op.entries.begin();
 131        iter != op.entries.end(); ++iter) {
 132     cls_user_bucket_entry& update_entry = *iter;
 133
 134     string key;
 135
 136     get_key_by_bucket_name(update_entry.bucket.name, &key);
 137
 138     cls_user_bucket_entry entry;
 139     ret = get_existing_bucket_entry(hctx, key, entry);
 140
 141     if (ret == -ENOENT) {
 142      if (!op.add)
 143       continue; /* racing bucket removal */
 144
 145      entry = update_entry;
 146
 147      ret = 0;
 148     } else if (op.add) {
 149       // bucket id may have changed (ie reshard)
 150       entry.bucket.bucket_id = update_entry.bucket.bucket_id;
 151     }
 152
 153     if (ret < 0) {
 154       CLS_LOG(0, "ERROR: get_existing_bucket_entry() key=%s returned %d", key.c_str(), ret);
 155       return ret;
 156     } else if (ret >= 0 && entry.user_stats_sync) {
 157       dec_header_stats(&header.stats, entry);
 158     }
 159
 160     CLS_LOG(20, "storing entry for key=%s size=%lld count=%lld",
 161             key.c_str(), (long long)update_entry.size, (long long)update_entry.count);
 162
 163     // sync entry stats when not an op.add, as when the case is op.add if its a
 164     // new entry we already have copied update_entry earlier, OTOH, for an existing entry
 165     // we end up clobbering the existing stats for the bucket
 166     if (!op.add){
 167       apply_entry_stats(update_entry, &entry);
 168     }
 169     entry.user_stats_sync = true;
 170
 171     ret = write_entry(hctx, key, entry);
 172     if (ret < 0)
 173       return ret;
 174
 175     add_header_stats(&header.stats, entry);
 176   }
 177
 178   bufferlist bl;
 179
 180   CLS_LOG(20, "header: total bytes=%lld entries=%lld", (long long)header.stats.total_bytes, (long long)header.stats.total_entries);
 181
 182   if (header.last_stats_update < op.time)
 183     header.last_stats_update = op.time;
 184
 185   ::encode(header, bl);
 186
 187   ret = cls_cxx_map_write_header(hctx, &bl);
 188   if (ret < 0)
 189     return ret;
 190
 191   return 0;
 192 }
 193
 194 static int cls_user_complete_stats_sync(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 195 {
 196   bufferlist::iterator in_iter = in->begin();
 197
 198   cls_user_complete_stats_sync_op op;
 199   try {
 200     ::decode(op, in_iter);
 201   } catch (buffer::error& err) {
 202     CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
 203     return -EINVAL;
 204   }
 205
 206   cls_user_header header;
 207   int ret = read_header(hctx, &header);
 208   if (ret < 0) {
 209     CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
 210     return ret;
 211   }
 212
 213   if (header.last_stats_sync < op.time)
 214     header.last_stats_sync = op.time;
 215
 216   bufferlist bl;
 217
 218   ::encode(header, bl);
 219
 220   ret = cls_cxx_map_write_header(hctx, &bl);
 221   if (ret < 0)
 222     return ret;
 223
 224   return 0;
 225 }
 226
 227 static int cls_user_remove_bucket(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 228 {
 229   bufferlist::iterator in_iter = in->begin();
 230
 231   cls_user_remove_bucket_op op;
 232   try {
 233     ::decode(op, in_iter);
 234   } catch (buffer::error& err) {
 235     CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
 236     return -EINVAL;
 237   }
 238
 239   cls_user_header header;
 240   int ret = read_header(hctx, &header);
 241   if (ret < 0) {
 242     CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
 243     return ret;
 244   }
 245
 246   string key;
 247
 248   get_key_by_bucket_name(op.bucket.name, &key);
 249
 250   cls_user_bucket_entry entry;
 251   ret = get_existing_bucket_entry(hctx, key, entry);
 252   if (ret == -ENOENT) {
 253     return 0; /* idempotent removal */
 254   }
 255   if (ret < 0) {
 256     CLS_LOG(0, "ERROR: get existing bucket entry, key=%s ret=%d", key.c_str(), ret);
 257     return ret;
 258   }
 259
 260   CLS_LOG(20, "removing entry at %s", key.c_str());
 261
 262   ret = remove_entry(hctx, key);
 263   if (ret < 0)
 264     return ret;
 265
 266   if (!entry.user_stats_sync) {
 267     return 0;
 268   }
 269
 270   dec_header_stats(&header.stats, entry);
 271
 272   CLS_LOG(20, "header: total bytes=%lld entries=%lld", (long long)header.stats.total_bytes, (long long)header.stats.total_entries);
 273
 274   bufferlist bl;
 275   encode(header, bl);
 276   return cls_cxx_map_write_header(hctx, &bl);
 277 }
 278
 279 static int cls_user_list_buckets(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 280 {
 281   bufferlist::iterator in_iter = in->begin();
 282
 283   cls_user_list_buckets_op op;
 284   try {
 285     ::decode(op, in_iter);
 286   } catch (buffer::error& err) {
 287     CLS_LOG(1, "ERROR: cls_user_list_op(): failed to decode op");
 288     return -EINVAL;
 289   }
 290
 291   map<string, bufferlist> keys;
 292
 293   const string& from_index = op.marker;
 294   const string& to_index = op.end_marker;
 295   const bool to_index_valid = !to_index.empty();
 296
 297 #define MAX_ENTRIES 1000
 298   size_t max_entries = op.max_entries;
 299   if (max_entries > MAX_ENTRIES)
 300     max_entries = MAX_ENTRIES;
 301
 302   string match_prefix;
 303   cls_user_list_buckets_ret ret;
 304
 305   int rc = cls_cxx_map_get_vals(hctx, from_index, match_prefix, max_entries, &keys, &ret.truncated);
 306   if (rc < 0)
 307     return rc;
 308
 309   CLS_LOG(20, "from_index=%s to_index=%s match_prefix=%s",
 310           from_index.c_str(),
 311           to_index.c_str(),
 312           match_prefix.c_str());
 313
 314   list<cls_user_bucket_entry>& entries = ret.entries;
 315   map<string, bufferlist>::iterator iter = keys.begin();
 316
 317   string marker;
 318
 319   for (; iter != keys.end(); ++iter) {
 320     const string& index = iter->first;
 321     marker = index;
 322
 323     if (to_index_valid && to_index.compare(index) <= 0) {
 324       ret.truncated = false;
 325       break;
 326     }
 327
 328     bufferlist& bl = iter->second;
 329     bufferlist::iterator biter = bl.begin();
 330     try {
 331       cls_user_bucket_entry e;
 332       ::decode(e, biter);
 333       entries.push_back(e);
 334     } catch (buffer::error& err) {
 335       CLS_LOG(0, "ERROR: cls_user_list: could not decode entry, index=%s", index.c_str());
 336     }
 337   }
 338
 339   if (ret.truncated) {
 340     ret.marker = marker;
 341   }
 342
 343   ::encode(ret, *out);
 344
 345   return 0;
 346 }
 347
 348 static int cls_user_get_header(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 349 {
 350   bufferlist::iterator in_iter = in->begin();
 351
 352   cls_user_get_header_op op;
 353   try {
 354     ::decode(op, in_iter);
 355   } catch (buffer::error& err) {
 356     CLS_LOG(1, "ERROR: cls_user_get_header_op(): failed to decode op");
 357     return -EINVAL;
 358   }
 359
 360   cls_user_get_header_ret op_ret;
 361
 362   int ret = read_header(hctx, &op_ret.header);
 363   if (ret < 0)
 364     return ret;
 365
 366   ::encode(op_ret, *out);
 367
 368   return 0;
 369 }
 370
 371 /// A method to reset the user.buckets header stats in accordance to the values
 372 /// seen in the user.buckets omap keys. This will not be equivalent to --sync-stats
 373 /// which requires comparing the values with actual bucket meta stats supplied
 374 /// by RGW
 375 static int cls_user_reset_stats(cls_method_context_t hctx, bufferlist *in, bufferlist *out /*ignore*/)
 376 {
 377   cls_user_reset_stats_op op;
 378
 379   try {
 380     auto bliter = in->begin();
 381     ::decode(op, bliter);
 382   } catch (buffer::error& err) {
 383     CLS_LOG(0, "ERROR: cls_user_reset_op(): failed to decode op");
 384     return -EINVAL;
 385   }
 386   cls_user_header header;
 387   bool truncated = false;
 388   string from_index, prefix;
 389   do {
 390     map<string, bufferlist> keys;
 391     int rc = cls_cxx_map_get_vals(hctx, from_index, prefix, MAX_ENTRIES, &keys, &truncated);
 392
 393     if (rc < 0)
 394       return rc;
 395
 396     for (const auto&kv : keys){
 397       cls_user_bucket_entry e;
 398       try {
 399         auto bl = kv.second;
 400         auto bliter = bl.begin();
 401         decode(e, bliter);
 402       } catch (buffer::error& err) {
 403         CLS_LOG(0, "ERROR: failed to decode bucket entry for %s", kv.first.c_str());
 404         return -EIO;
 405       }
 406       add_header_stats(&header.stats, e);
 407     }
 408   } while (truncated);
 409
 410   bufferlist bl;
 411   header.last_stats_update = op.time;
 412   ::encode(header, bl);
 413
 414   return cls_cxx_map_write_header(hctx, &bl);
 415 }
 416
 417 CLS_INIT(user)
 418 {
 419   CLS_LOG(1, "Loaded user class!");
 420
 421   cls_handle_t h_class;
 422   cls_method_handle_t h_user_set_buckets_info;
 423   cls_method_handle_t h_user_complete_stats_sync;
 424   cls_method_handle_t h_user_remove_bucket;
 425   cls_method_handle_t h_user_list_buckets;
 426   cls_method_handle_t h_user_get_header;
 427   cls_method_handle_t h_user_reset_stats;
 428
 429   cls_register("user", &h_class);
 430
 431   /* log */
 432   cls_register_cxx_method(h_class, "set_buckets_info", CLS_METHOD_RD | CLS_METHOD_WR,
 433                           cls_user_set_buckets_info, &h_user_set_buckets_info);
 434   cls_register_cxx_method(h_class, "complete_stats_sync", CLS_METHOD_RD | CLS_METHOD_WR,
 435                           cls_user_complete_stats_sync, &h_user_complete_stats_sync);
 436   cls_register_cxx_method(h_class, "remove_bucket", CLS_METHOD_RD | CLS_METHOD_WR, cls_user_remove_bucket, &h_user_remove_bucket);
 437   cls_register_cxx_method(h_class, "list_buckets", CLS_METHOD_RD, cls_user_list_buckets, &h_user_list_buckets);
 438   cls_register_cxx_method(h_class, "get_header", CLS_METHOD_RD, cls_user_get_header, &h_user_get_header);
 439   cls_register_cxx_method(h_class, "reset_user_stats", CLS_METHOD_RD | CLS_METHOD_WR, cls_user_reset_stats, &h_user_reset_stats);
 440   return;
 441 }
 442