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