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
 167     entry.user_stats_sync = true;
 168
 169     ret = write_entry(hctx, key, entry);
 170     if (ret < 0)
 171       return ret;
 172
 173     add_header_stats(&header.stats, entry);
 174   }
 175
 176   bufferlist bl;
 177
 178   CLS_LOG(20, "header: total bytes=%lld entries=%lld", (long long)header.stats.total_bytes, (long long)header.stats.total_entries);
 179
 180   if (header.last_stats_update < op.time)
 181     header.last_stats_update = op.time;
 182
 183   ::encode(header, bl);
 184
 185   ret = cls_cxx_map_write_header(hctx, &bl);
 186   if (ret < 0)
 187     return ret;
 188
 189   return 0;
 190 }
 191
 192 static int cls_user_complete_stats_sync(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 193 {
 194   bufferlist::iterator in_iter = in->begin();
 195
 196   cls_user_complete_stats_sync_op op;
 197   try {
 198     ::decode(op, in_iter);
 199   } catch (buffer::error& err) {
 200     CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
 201     return -EINVAL;
 202   }
 203
 204   cls_user_header header;
 205   int ret = read_header(hctx, &header);
 206   if (ret < 0) {
 207     CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
 208     return ret;
 209   }
 210
 211   if (header.last_stats_sync < op.time)
 212     header.last_stats_sync = op.time;
 213
 214   bufferlist bl;
 215
 216   ::encode(header, bl);
 217
 218   ret = cls_cxx_map_write_header(hctx, &bl);
 219   if (ret < 0)
 220     return ret;
 221
 222   return 0;
 223 }
 224
 225 static int cls_user_remove_bucket(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 226 {
 227   bufferlist::iterator in_iter = in->begin();
 228
 229   cls_user_remove_bucket_op op;
 230   try {
 231     ::decode(op, in_iter);
 232   } catch (buffer::error& err) {
 233     CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
 234     return -EINVAL;
 235   }
 236
 237   cls_user_header header;
 238   int ret = read_header(hctx, &header);
 239   if (ret < 0) {
 240     CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
 241     return ret;
 242   }
 243
 244   string key;
 245
 246   get_key_by_bucket_name(op.bucket.name, &key);
 247
 248   cls_user_bucket_entry entry;
 249   ret = get_existing_bucket_entry(hctx, key, entry);
 250   if (ret == -ENOENT) {
 251     return 0; /* idempotent removal */
 252   }
 253   if (ret < 0) {
 254     CLS_LOG(0, "ERROR: get existing bucket entry, key=%s ret=%d", key.c_str(), ret);
 255     return ret;
 256   }
 257
 258   if (entry.user_stats_sync) {
 259     dec_header_stats(&header.stats, entry);
 260   }
 261
 262   CLS_LOG(20, "removing entry at %s", key.c_str());
 263
 264   ret = remove_entry(hctx, key);
 265   if (ret < 0)
 266     return ret;
 267
 268   return 0;
 269 }
 270
 271 static int cls_user_list_buckets(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 272 {
 273   bufferlist::iterator in_iter = in->begin();
 274
 275   cls_user_list_buckets_op op;
 276   try {
 277     ::decode(op, in_iter);
 278   } catch (buffer::error& err) {
 279     CLS_LOG(1, "ERROR: cls_user_list_op(): failed to decode op");
 280     return -EINVAL;
 281   }
 282
 283   map<string, bufferlist> keys;
 284
 285   const string& from_index = op.marker;
 286   const string& to_index = op.end_marker;
 287   const bool to_index_valid = !to_index.empty();
 288
 289 #define MAX_ENTRIES 1000
 290   size_t max_entries = op.max_entries;
 291   if (max_entries > MAX_ENTRIES)
 292     max_entries = MAX_ENTRIES;
 293
 294   string match_prefix;
 295   cls_user_list_buckets_ret ret;
 296
 297   int rc = cls_cxx_map_get_vals(hctx, from_index, match_prefix, max_entries, &keys, &ret.truncated);
 298   if (rc < 0)
 299     return rc;
 300
 301   CLS_LOG(20, "from_index=%s to_index=%s match_prefix=%s",
 302           from_index.c_str(),
 303           to_index.c_str(),
 304           match_prefix.c_str());
 305
 306   list<cls_user_bucket_entry>& entries = ret.entries;
 307   map<string, bufferlist>::iterator iter = keys.begin();
 308
 309   string marker;
 310
 311   for (; iter != keys.end(); ++iter) {
 312     const string& index = iter->first;
 313     marker = index;
 314
 315     if (to_index_valid && to_index.compare(index) <= 0) {
 316       ret.truncated = false;
 317       break;
 318     }
 319
 320     bufferlist& bl = iter->second;
 321     bufferlist::iterator biter = bl.begin();
 322     try {
 323       cls_user_bucket_entry e;
 324       ::decode(e, biter);
 325       entries.push_back(e);
 326     } catch (buffer::error& err) {
 327       CLS_LOG(0, "ERROR: cls_user_list: could not decode entry, index=%s", index.c_str());
 328     }
 329   }
 330
 331   if (ret.truncated) {
 332     ret.marker = marker;
 333   }
 334
 335   ::encode(ret, *out);
 336
 337   return 0;
 338 }
 339
 340 static int cls_user_get_header(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 341 {
 342   bufferlist::iterator in_iter = in->begin();
 343
 344   cls_user_get_header_op op;
 345   try {
 346     ::decode(op, in_iter);
 347   } catch (buffer::error& err) {
 348     CLS_LOG(1, "ERROR: cls_user_get_header_op(): failed to decode op");
 349     return -EINVAL;
 350   }
 351
 352   cls_user_get_header_ret op_ret;
 353
 354   int ret = read_header(hctx, &op_ret.header);
 355   if (ret < 0)
 356     return ret;
 357
 358   ::encode(op_ret, *out);
 359
 360   return 0;
 361 }
 362
 363 CLS_INIT(user)
 364 {
 365   CLS_LOG(1, "Loaded user class!");
 366
 367   cls_handle_t h_class;
 368   cls_method_handle_t h_user_set_buckets_info;
 369   cls_method_handle_t h_user_complete_stats_sync;
 370   cls_method_handle_t h_user_remove_bucket;
 371   cls_method_handle_t h_user_list_buckets;
 372   cls_method_handle_t h_user_get_header;
 373
 374   cls_register("user", &h_class);
 375
 376   /* log */
 377   cls_register_cxx_method(h_class, "set_buckets_info", CLS_METHOD_RD | CLS_METHOD_WR,
 378                           cls_user_set_buckets_info, &h_user_set_buckets_info);
 379   cls_register_cxx_method(h_class, "complete_stats_sync", CLS_METHOD_RD | CLS_METHOD_WR,
 380                           cls_user_complete_stats_sync, &h_user_complete_stats_sync);
 381   cls_register_cxx_method(h_class, "remove_bucket", CLS_METHOD_RD | CLS_METHOD_WR, cls_user_remove_bucket, &h_user_remove_bucket);
 382   cls_register_cxx_method(h_class, "list_buckets", CLS_METHOD_RD, cls_user_list_buckets, &h_user_list_buckets);
 383   cls_register_cxx_method(h_class, "get_header", CLS_METHOD_RD, cls_user_get_header, &h_user_get_header);
 384
 385   return;
 386 }
 387