ceph/src/cls/log/cls_log.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 "include/types.h"
   5 #include "include/utime.h"
   6 #include "objclass/objclass.h"
   7
   8 #include "cls_log_types.h"
   9 #include "cls_log_ops.h"
  10
  11 #include "global/global_context.h"
  12 #include "include/compat.h"
  13
  14 CLS_VER(1,0)
  15 CLS_NAME(log)
  16
  17 static string log_index_prefix = "1_";
  18
  19
  20 static int write_log_entry(cls_method_context_t hctx, string& index, cls_log_entry& entry)
  21 {
  22   bufferlist bl;
  23   encode(entry, bl);
  24
  25   int ret = cls_cxx_map_set_val(hctx, index, &bl);
  26   if (ret < 0)
  27     return ret;
  28
  29   return 0;
  30 }
  31
  32 static void get_index_time_prefix(utime_t& ts, string& index)
  33 {
  34   char buf[32];
  35   snprintf(buf, sizeof(buf), "%010ld.%06ld_", (long)ts.sec(), (long)ts.usec());
  36
  37   index = log_index_prefix + buf;
  38 }
  39
  40 static int read_header(cls_method_context_t hctx, cls_log_header& header)
  41 {
  42   bufferlist header_bl;
  43
  44   int ret = cls_cxx_map_read_header(hctx, &header_bl);
  45   if (ret < 0)
  46     return ret;
  47
  48   if (header_bl.length() == 0) {
  49     header = cls_log_header();
  50     return 0;
  51   }
  52
  53   auto iter = header_bl.cbegin();
  54   try {
  55     decode(header, iter);
  56   } catch (buffer::error& err) {
  57     CLS_LOG(0, "ERROR: read_header(): failed to decode header");
  58   }
  59
  60   return 0;
  61 }
  62
  63 static int write_header(cls_method_context_t hctx, cls_log_header& header)
  64 {
  65   bufferlist header_bl;
  66   encode(header, header_bl);
  67
  68   int ret = cls_cxx_map_write_header(hctx, &header_bl);
  69   if (ret < 0)
  70     return ret;
  71
  72   return 0;
  73 }
  74
  75 static void get_index(cls_method_context_t hctx, utime_t& ts, string& index)
  76 {
  77   get_index_time_prefix(ts, index);
  78
  79   string unique_id;
  80
  81   cls_cxx_subop_version(hctx, &unique_id);
  82
  83   index.append(unique_id);
  84 }
  85
  86 static int cls_log_add(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
  87 {
  88   auto in_iter = in->cbegin();
  89
  90   cls_log_add_op op;
  91   try {
  92     decode(op, in_iter);
  93   } catch (buffer::error& err) {
  94     CLS_LOG(1, "ERROR: cls_log_add_op(): failed to decode op");
  95     return -EINVAL;
  96   }
  97
  98   cls_log_header header;
  99
 100   int ret = read_header(hctx, header);
 101   if (ret < 0)
 102     return ret;
 103
 104   for (list<cls_log_entry>::iterator iter = op.entries.begin();
 105        iter != op.entries.end(); ++iter) {
 106     cls_log_entry& entry = *iter;
 107
 108     string index;
 109
 110     utime_t timestamp = entry.timestamp;
 111     if (op.monotonic_inc && timestamp < header.max_time)
 112       timestamp = header.max_time;
 113     else if (timestamp > header.max_time)
 114       header.max_time = timestamp;
 115
 116     if (entry.id.empty()) {
 117       get_index(hctx, timestamp, index);
 118       entry.id = index;
 119     } else {
 120       index = entry.id;
 121     }
 122
 123     CLS_LOG(20, "storing entry at %s", index.c_str());
 124
 125
 126     if (index > header.max_marker)
 127       header.max_marker = index;
 128
 129     ret = write_log_entry(hctx, index, entry);
 130     if (ret < 0)
 131       return ret;
 132   }
 133
 134   ret = write_header(hctx, header);
 135   if (ret < 0)
 136     return ret;
 137
 138   return 0;
 139 }
 140
 141 static int cls_log_list(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 142 {
 143   auto in_iter = in->cbegin();
 144
 145   cls_log_list_op op;
 146   try {
 147     decode(op, in_iter);
 148   } catch (buffer::error& err) {
 149     CLS_LOG(1, "ERROR: cls_log_list_op(): failed to decode op");
 150     return -EINVAL;
 151   }
 152
 153   map<string, bufferlist> keys;
 154
 155   string from_index;
 156   string to_index;
 157
 158   if (op.marker.empty()) {
 159     get_index_time_prefix(op.from_time, from_index);
 160   } else {
 161     from_index = op.marker;
 162   }
 163   bool use_time_boundary = (!op.from_time.is_zero() && (op.to_time >= op.from_time));
 164
 165   if (use_time_boundary)
 166     get_index_time_prefix(op.to_time, to_index);
 167
 168 #define MAX_ENTRIES 1000
 169   size_t max_entries = op.max_entries;
 170   if (!max_entries || max_entries > MAX_ENTRIES)
 171     max_entries = MAX_ENTRIES;
 172
 173   cls_log_list_ret ret;
 174
 175   int rc = cls_cxx_map_get_vals(hctx, from_index, log_index_prefix, max_entries, &keys, &ret.truncated);
 176   if (rc < 0)
 177     return rc;
 178
 179   list<cls_log_entry>& entries = ret.entries;
 180   map<string, bufferlist>::iterator iter = keys.begin();
 181
 182   string marker;
 183
 184   for (; iter != keys.end(); ++iter) {
 185     const string& index = iter->first;
 186     marker = index;
 187     if (use_time_boundary && index.compare(0, to_index.size(), to_index) >= 0) {
 188       ret.truncated = false;
 189       break;
 190     }
 191
 192     bufferlist& bl = iter->second;
 193     auto biter = bl.cbegin();
 194     try {
 195       cls_log_entry e;
 196       decode(e, biter);
 197       entries.push_back(e);
 198     } catch (buffer::error& err) {
 199       CLS_LOG(0, "ERROR: cls_log_list: could not decode entry, index=%s", index.c_str());
 200     }
 201   }
 202
 203   ret.marker = marker;
 204
 205   encode(ret, *out);
 206
 207   return 0;
 208 }
 209
 210
 211 static int cls_log_trim(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 212 {
 213   auto in_iter = in->cbegin();
 214
 215   cls_log_trim_op op;
 216   try {
 217     decode(op, in_iter);
 218   } catch (buffer::error& err) {
 219     CLS_LOG(0, "ERROR: cls_log_list_op(): failed to decode entry");
 220     return -EINVAL;
 221   }
 222
 223   map<string, bufferlist> keys;
 224
 225   string from_index;
 226   string to_index;
 227
 228   if (op.from_marker.empty()) {
 229     get_index_time_prefix(op.from_time, from_index);
 230   } else {
 231     from_index = op.from_marker;
 232   }
 233   if (op.to_marker.empty()) {
 234     get_index_time_prefix(op.to_time, to_index);
 235   } else {
 236     to_index = op.to_marker;
 237   }
 238
 239 #define MAX_TRIM_ENTRIES 1000
 240   size_t max_entries = MAX_TRIM_ENTRIES;
 241   bool more;
 242
 243   int rc = cls_cxx_map_get_vals(hctx, from_index, log_index_prefix, max_entries, &keys, &more);
 244   if (rc < 0)
 245     return rc;
 246
 247   map<string, bufferlist>::iterator iter = keys.begin();
 248
 249   bool removed = false;
 250   for (; iter != keys.end(); ++iter) {
 251     const string& index = iter->first;
 252
 253     CLS_LOG(20, "index=%s to_index=%s", index.c_str(), to_index.c_str());
 254
 255     if (index.compare(0, to_index.size(), to_index) > 0)
 256       break;
 257
 258     CLS_LOG(20, "removing key: index=%s", index.c_str());
 259
 260     int rc = cls_cxx_map_remove_key(hctx, index);
 261     if (rc < 0) {
 262       CLS_LOG(1, "ERROR: cls_cxx_map_remove_key failed rc=%d", rc);
 263       return -EINVAL;
 264     }
 265     removed = true;
 266   }
 267
 268   if (!removed)
 269     return -ENODATA;
 270
 271   return 0;
 272 }
 273
 274 static int cls_log_info(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 275 {
 276   auto in_iter = in->cbegin();
 277
 278   cls_log_info_op op;
 279   try {
 280     decode(op, in_iter);
 281   } catch (buffer::error& err) {
 282     CLS_LOG(1, "ERROR: cls_log_add_op(): failed to decode op");
 283     return -EINVAL;
 284   }
 285
 286   cls_log_info_ret ret;
 287
 288   int rc = read_header(hctx, ret.header);
 289   if (rc < 0)
 290     return rc;
 291
 292   encode(ret, *out);
 293
 294   return 0;
 295 }
 296
 297 CLS_INIT(log)
 298 {
 299   CLS_LOG(1, "Loaded log class!");
 300
 301   cls_handle_t h_class;
 302   cls_method_handle_t h_log_add;
 303   cls_method_handle_t h_log_list;
 304   cls_method_handle_t h_log_trim;
 305   cls_method_handle_t h_log_info;
 306
 307   cls_register("log", &h_class);
 308
 309   /* log */
 310   cls_register_cxx_method(h_class, "add", CLS_METHOD_RD | CLS_METHOD_WR, cls_log_add, &h_log_add);
 311   cls_register_cxx_method(h_class, "list", CLS_METHOD_RD, cls_log_list, &h_log_list);
 312   cls_register_cxx_method(h_class, "trim", CLS_METHOD_RD | CLS_METHOD_WR, cls_log_trim, &h_log_trim);
 313   cls_register_cxx_method(h_class, "info", CLS_METHOD_RD, cls_log_info, &h_log_info);
 314
 315   return;
 316 }
 317