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     }
 149
 150     if (ret < 0) {
 151       CLS_LOG(0, "ERROR: get_existing_bucket_entry() key=%s returned %d", key.c_str(), ret);
 152       return ret;
 153     } else if (ret >= 0 && entry.user_stats_sync) {
 154       dec_header_stats(&header.stats, entry);
 155     }
 156
 157     CLS_LOG(20, "storing entry for key=%s size=%lld count=%lld",
 158             key.c_str(), (long long)update_entry.size, (long long)update_entry.count);
 159
 160     // sync entry stats when not an op.add, as when the case is op.add if its a
 161     // new entry we already have copied update_entry earlier, OTOH, for an existing entry
 162     // we end up clobbering the existing stats for the bucket
 163     if (!op.add){
 164       apply_entry_stats(update_entry, &entry);
 165     }
 166     entry.user_stats_sync = true;
 167
 168     ret = write_entry(hctx, key, entry);
 169     if (ret < 0)
 170       return ret;
 171
 172     add_header_stats(&header.stats, entry);
 173   }
 174
 175   bufferlist bl;
 176
 177   CLS_LOG(20, "header: total bytes=%lld entries=%lld", (long long)header.stats.total_bytes, (long long)header.stats.total_entries);
 178
 179   if (header.last_stats_update < op.time)
 180     header.last_stats_update = op.time;
 181
 182   ::encode(header, bl);
 183
 184   ret = cls_cxx_map_write_header(hctx, &bl);
 185   if (ret < 0)
 186     return ret;
 187
 188   return 0;
 189 }
 190
 191 static int cls_user_complete_stats_sync(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 192 {
 193   bufferlist::iterator in_iter = in->begin();
 194
 195   cls_user_complete_stats_sync_op op;
 196   try {
 197     ::decode(op, in_iter);
 198   } catch (buffer::error& err) {
 199     CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
 200     return -EINVAL;
 201   }
 202
 203   cls_user_header header;
 204   int ret = read_header(hctx, &header);
 205   if (ret < 0) {
 206     CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
 207     return ret;
 208   }
 209
 210   if (header.last_stats_sync < op.time)
 211     header.last_stats_sync = op.time;
 212
 213   bufferlist bl;
 214
 215   ::encode(header, bl);
 216
 217   ret = cls_cxx_map_write_header(hctx, &bl);
 218   if (ret < 0)
 219     return ret;
 220
 221   return 0;
 222 }
 223
 224 static int cls_user_remove_bucket(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 225 {
 226   bufferlist::iterator in_iter = in->begin();
 227
 228   cls_user_remove_bucket_op op;
 229   try {
 230     ::decode(op, in_iter);
 231   } catch (buffer::error& err) {
 232     CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
 233     return -EINVAL;
 234   }
 235
 236   cls_user_header header;
 237   int ret = read_header(hctx, &header);
 238   if (ret < 0) {
 239     CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
 240     return ret;
 241   }
 242
 243   string key;
 244
 245   get_key_by_bucket_name(op.bucket.name, &key);
 246
 247   cls_user_bucket_entry entry;
 248   ret = get_existing_bucket_entry(hctx, key, entry);
 249   if (ret == -ENOENT) {
 250     return 0; /* idempotent removal */
 251   }
 252   if (ret < 0) {
 253     CLS_LOG(0, "ERROR: get existing bucket entry, key=%s ret=%d", key.c_str(), ret);
 254     return ret;
 255   }
 256
 257   if (entry.user_stats_sync) {
 258     dec_header_stats(&header.stats, entry);
 259   }
 260
 261   CLS_LOG(20, "removing entry at %s", key.c_str());
 262
 263   ret = remove_entry(hctx, key);
 264   if (ret < 0)
 265     return ret;
 266
 267   return 0;
 268 }
 269
 270 static int cls_user_list_buckets(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 271 {
 272   bufferlist::iterator in_iter = in->begin();
 273
 274   cls_user_list_buckets_op op;
 275   try {
 276     ::decode(op, in_iter);
 277   } catch (buffer::error& err) {
 278     CLS_LOG(1, "ERROR: cls_user_list_op(): failed to decode op");
 279     return -EINVAL;
 280   }
 281
 282   map<string, bufferlist> keys;
 283
 284   const string& from_index = op.marker;
 285   const string& to_index = op.end_marker;
 286   const bool to_index_valid = !to_index.empty();
 287
 288 #define MAX_ENTRIES 1000
 289   size_t max_entries = op.max_entries;
 290   if (max_entries > MAX_ENTRIES)
 291     max_entries = MAX_ENTRIES;
 292
 293   string match_prefix;
 294   cls_user_list_buckets_ret ret;
 295
 296   int rc = cls_cxx_map_get_vals(hctx, from_index, match_prefix, max_entries, &keys, &ret.truncated);
 297   if (rc < 0)
 298     return rc;
 299
 300   CLS_LOG(20, "from_index=%s to_index=%s match_prefix=%s",
 301           from_index.c_str(),
 302           to_index.c_str(),
 303           match_prefix.c_str());
 304
 305   list<cls_user_bucket_entry>& entries = ret.entries;
 306   map<string, bufferlist>::iterator iter = keys.begin();
 307
 308   string marker;
 309
 310   for (; iter != keys.end(); ++iter) {
 311     const string& index = iter->first;
 312     marker = index;
 313
 314     if (to_index_valid && to_index.compare(index) <= 0) {
 315       ret.truncated = false;
 316       break;
 317     }
 318
 319     bufferlist& bl = iter->second;
 320     bufferlist::iterator biter = bl.begin();
 321     try {
 322       cls_user_bucket_entry e;
 323       ::decode(e, biter);
 324       entries.push_back(e);
 325     } catch (buffer::error& err) {
 326       CLS_LOG(0, "ERROR: cls_user_list: could not decode entry, index=%s", index.c_str());
 327     }
 328   }
 329
 330   if (ret.truncated) {
 331     ret.marker = marker;
 332   }
 333
 334   ::encode(ret, *out);
 335
 336   return 0;
 337 }
 338
 339 static int cls_user_get_header(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 340 {
 341   bufferlist::iterator in_iter = in->begin();
 342
 343   cls_user_get_header_op op;
 344   try {
 345     ::decode(op, in_iter);
 346   } catch (buffer::error& err) {
 347     CLS_LOG(1, "ERROR: cls_user_get_header_op(): failed to decode op");
 348     return -EINVAL;
 349   }
 350
 351   cls_user_get_header_ret op_ret;
 352
 353   int ret = read_header(hctx, &op_ret.header);
 354   if (ret < 0)
 355     return ret;
 356
 357   ::encode(op_ret, *out);
 358
 359   return 0;
 360 }
 361
 362 /// A method to reset the user.buckets header stats in accordance to the values
 363 /// seen in the user.buckets omap keys. This will not be equivalent to --sync-stats
 364 /// which requires comparing the values with actual bucket meta stats supplied
 365 /// by RGW
 366 static int cls_user_reset_stats(cls_method_context_t hctx, bufferlist *in, bufferlist *out /*ignore*/)
 367 {
 368   cls_user_reset_stats_op op;
 369
 370   try {
 371     auto bliter = in->begin();
 372     ::decode(op, bliter);
 373   } catch (buffer::error& err) {
 374     CLS_LOG(0, "ERROR: cls_user_reset_op(): failed to decode op");
 375     return -EINVAL;
 376   }
 377   cls_user_header header;
 378   bool truncated = false;
 379   string from_index, prefix;
 380   do {
 381     map<string, bufferlist> keys;
 382     int rc = cls_cxx_map_get_vals(hctx, from_index, prefix, MAX_ENTRIES, &keys, &truncated);
 383
 384     if (rc < 0)
 385       return rc;
 386
 387     for (const auto&kv : keys){
 388       cls_user_bucket_entry e;
 389       try {
 390         auto bl = kv.second;
 391         auto bliter = bl.begin();
 392         decode(e, bliter);
 393       } catch (buffer::error& err) {
 394         CLS_LOG(0, "ERROR: failed to decode bucket entry for %s", kv.first.c_str());
 395         return -EIO;
 396       }
 397       add_header_stats(&header.stats, e);
 398     }
 399   } while (truncated);
 400
 401   bufferlist bl;
 402   header.last_stats_update = op.time;
 403   ::encode(header, bl);
 404
 405   return cls_cxx_map_write_header(hctx, &bl);
 406 }
 407
 408 CLS_INIT(user)
 409 {
 410   CLS_LOG(1, "Loaded user class!");
 411
 412   cls_handle_t h_class;
 413   cls_method_handle_t h_user_set_buckets_info;
 414   cls_method_handle_t h_user_complete_stats_sync;
 415   cls_method_handle_t h_user_remove_bucket;
 416   cls_method_handle_t h_user_list_buckets;
 417   cls_method_handle_t h_user_get_header;
 418   cls_method_handle_t h_user_reset_stats;
 419
 420   cls_register("user", &h_class);
 421
 422   /* log */
 423   cls_register_cxx_method(h_class, "set_buckets_info", CLS_METHOD_RD | CLS_METHOD_WR,
 424                           cls_user_set_buckets_info, &h_user_set_buckets_info);
 425   cls_register_cxx_method(h_class, "complete_stats_sync", CLS_METHOD_RD | CLS_METHOD_WR,
 426                           cls_user_complete_stats_sync, &h_user_complete_stats_sync);
 427   cls_register_cxx_method(h_class, "remove_bucket", CLS_METHOD_RD | CLS_METHOD_WR, cls_user_remove_bucket, &h_user_remove_bucket);
 428   cls_register_cxx_method(h_class, "list_buckets", CLS_METHOD_RD, cls_user_list_buckets, &h_user_list_buckets);
 429   cls_register_cxx_method(h_class, "get_header", CLS_METHOD_RD, cls_user_get_header, &h_user_get_header);
 430   cls_register_cxx_method(h_class, "reset_user_stats", CLS_METHOD_RD | CLS_METHOD_WR, cls_user_reset_stats, &h_user_reset_stats);
 431   return;
 432 }
 433