ceph/src/cls/statelog/cls_statelog.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 <iostream>
   5
   6 #include <string.h>
   7 #include <stdlib.h>
   8 #include <errno.h>
   9
  10 #include "include/types.h"
  11 #include "include/utime.h"
  12 #include "objclass/objclass.h"
  13
  14 #include "cls_statelog_types.h"
  15 #include "cls_statelog_ops.h"
  16
  17 #include "global/global_context.h"
  18
  19 CLS_VER(1,0)
  20 CLS_NAME(statelog)
  21
  22 static string statelog_index_by_client_prefix = "1_";
  23 static string statelog_index_by_object_prefix = "2_";
  24
  25
  26 static int write_statelog_entry(cls_method_context_t hctx, const string& index, const cls_statelog_entry& entry)
  27 {
  28   bufferlist bl;
  29   ::encode(entry, bl);
  30
  31   int ret = cls_cxx_map_set_val(hctx, index, &bl);
  32   if (ret < 0)
  33     return ret;
  34
  35   return 0;
  36 }
  37
  38 static void get_index_by_client(const string& client_id, const string& op_id, string& index)
  39 {
  40   index = statelog_index_by_client_prefix;
  41   index.append(client_id + "_" + op_id);
  42 }
  43
  44 static void get_index_by_client(cls_statelog_entry& entry, string& index)
  45 {
  46   get_index_by_client(entry.client_id, entry.op_id, index);
  47 }
  48
  49 static void get_index_by_object(const string& object, const string& op_id, string& index)
  50 {
  51   char buf[16];
  52   snprintf(buf, sizeof(buf), "%d_", (int)object.size());
  53
  54   index = statelog_index_by_object_prefix + buf; /* append object length to ensure uniqueness */
  55   index.append(object + "_" + op_id);
  56 }
  57
  58 static void get_index_by_object(cls_statelog_entry& entry, string& index)
  59 {
  60   get_index_by_object(entry.object, entry.op_id, index);
  61 }
  62
  63 static int get_existing_entry(cls_method_context_t hctx, const string& client_id,
  64                                const string& op_id, const string& object,
  65                                cls_statelog_entry& entry)
  66 {
  67   if ((object.empty() && client_id.empty()) || op_id.empty()) {
  68     return -EINVAL;
  69   }
  70
  71   string obj_index;
  72   if (!object.empty()) {
  73     get_index_by_object(object, op_id, obj_index);
  74   } else {
  75     get_index_by_client(client_id, op_id, obj_index);
  76   }
  77
  78   bufferlist bl;
  79   int rc = cls_cxx_map_get_val(hctx, obj_index, &bl);
  80   if (rc < 0) {
  81     CLS_LOG(0, "could not find entry %s", obj_index.c_str());
  82     return rc;
  83   }
  84   try {
  85     bufferlist::iterator iter = bl.begin();
  86     ::decode(entry, iter);
  87   } catch (buffer::error& err) {
  88     CLS_LOG(0, "ERROR: failed to decode entry %s", obj_index.c_str());
  89     return -EIO;
  90   }
  91
  92   if ((!object.empty() && entry.object != object) ||
  93       (!client_id.empty() && entry.client_id != client_id)){
  94     /* ouch, we were passed inconsistent client_id / object */
  95     CLS_LOG(0, "data mismatch: object=%s client_id=%s entry: object=%s client_id=%s",
  96             object.c_str(), client_id.c_str(), entry.object.c_str(), entry.client_id.c_str());
  97     return -EINVAL;
  98   }
  99
 100   return 0;
 101 }
 102
 103 static int cls_statelog_add(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 104 {
 105   bufferlist::iterator in_iter = in->begin();
 106
 107   cls_statelog_add_op op;
 108   try {
 109     ::decode(op, in_iter);
 110   } catch (buffer::error& err) {
 111     CLS_LOG(1, "ERROR: cls_statelog_add_op(): failed to decode op");
 112     return -EINVAL;
 113   }
 114
 115   for (list<cls_statelog_entry>::iterator iter = op.entries.begin();
 116        iter != op.entries.end(); ++iter) {
 117     cls_statelog_entry& entry = *iter;
 118
 119     string index_by_client;
 120
 121     get_index_by_client(entry, index_by_client);
 122
 123     CLS_LOG(0, "storing entry by client/op at %s", index_by_client.c_str());
 124
 125     int ret = write_statelog_entry(hctx, index_by_client, entry);
 126     if (ret < 0)
 127       return ret;
 128
 129     string index_by_obj;
 130
 131     get_index_by_object(entry, index_by_obj);
 132
 133     CLS_LOG(0, "storing entry by object at %s", index_by_obj.c_str());
 134     ret = write_statelog_entry(hctx, index_by_obj, entry);
 135     if (ret < 0)
 136       return ret;
 137
 138   }
 139
 140   return 0;
 141 }
 142
 143 static int cls_statelog_list(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 144 {
 145   bufferlist::iterator in_iter = in->begin();
 146
 147   cls_statelog_list_op op;
 148   try {
 149     ::decode(op, in_iter);
 150   } catch (buffer::error& err) {
 151     CLS_LOG(1, "ERROR: cls_statelog_list_op(): failed to decode op");
 152     return -EINVAL;
 153   }
 154
 155   map<string, bufferlist> keys;
 156
 157   string from_index;
 158   string match_prefix;
 159
 160   if (!op.client_id.empty()) {
 161     get_index_by_client(op.client_id, op.op_id, match_prefix);
 162   } else if (!op.object.empty()) {
 163     get_index_by_object(op.object, op.op_id, match_prefix);
 164   } else {
 165     match_prefix = statelog_index_by_object_prefix;
 166   }
 167
 168   if (op.marker.empty()) {
 169     from_index = match_prefix;
 170   } else {
 171     from_index = op.marker;
 172   }
 173
 174 #define MAX_ENTRIES 1000
 175   size_t max_entries = op.max_entries;
 176   if (!max_entries || max_entries > MAX_ENTRIES)
 177     max_entries = MAX_ENTRIES;
 178
 179   int rc = cls_cxx_map_get_vals(hctx, from_index, match_prefix, max_entries + 1, &keys);
 180   if (rc < 0)
 181     return rc;
 182
 183   CLS_LOG(20, "from_index=%s match_prefix=%s", from_index.c_str(), match_prefix.c_str());
 184   cls_statelog_list_ret ret;
 185
 186   list<cls_statelog_entry>& entries = ret.entries;
 187   map<string, bufferlist>::iterator iter = keys.begin();
 188
 189   bool done = false;
 190   string marker;
 191
 192   size_t i;
 193   for (i = 0; i < max_entries && iter != keys.end(); ++i, ++iter) {
 194     const string& index = iter->first;
 195     marker = index;
 196
 197     bufferlist& bl = iter->second;
 198     bufferlist::iterator biter = bl.begin();
 199     try {
 200       cls_statelog_entry e;
 201       ::decode(e, biter);
 202       entries.push_back(e);
 203     } catch (buffer::error& err) {
 204       CLS_LOG(0, "ERROR: cls_statelog_list: could not decode entry, index=%s", index.c_str());
 205     }
 206   }
 207
 208   if (iter == keys.end())
 209     done = true;
 210   else
 211     ret.marker = marker;
 212
 213   ret.truncated = !done;
 214
 215   ::encode(ret, *out);
 216
 217   return 0;
 218 }
 219
 220 static int cls_statelog_remove(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 221 {
 222   bufferlist::iterator in_iter = in->begin();
 223
 224   cls_statelog_remove_op op;
 225   try {
 226     ::decode(op, in_iter);
 227   } catch (buffer::error& err) {
 228     CLS_LOG(1, "ERROR: cls_statelog_remove_op(): failed to decode op");
 229     return -EINVAL;
 230   }
 231
 232   cls_statelog_entry entry;
 233
 234   int rc = get_existing_entry(hctx, op.client_id, op.op_id, op.object, entry);
 235   if (rc < 0)
 236     return rc;
 237
 238   string obj_index;
 239   get_index_by_object(entry.object, entry.op_id, obj_index);
 240
 241   rc = cls_cxx_map_remove_key(hctx, obj_index);
 242   if (rc < 0) {
 243     CLS_LOG(0, "ERROR: failed to remove key");
 244     return rc;
 245   }
 246
 247   string client_index;
 248   get_index_by_client(entry.client_id, entry.op_id, client_index);
 249
 250   rc = cls_cxx_map_remove_key(hctx, client_index);
 251   if (rc < 0) {
 252     CLS_LOG(0, "ERROR: failed to remove key");
 253     return rc;
 254   }
 255
 256   return 0;
 257 }
 258
 259 static int cls_statelog_check_state(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 260 {
 261   bufferlist::iterator in_iter = in->begin();
 262
 263   cls_statelog_check_state_op op;
 264   try {
 265     ::decode(op, in_iter);
 266   } catch (buffer::error& err) {
 267     CLS_LOG(1, "ERROR: cls_statelog_check_state_op(): failed to decode op");
 268     return -EINVAL;
 269   }
 270
 271   if (op.object.empty() || op.op_id.empty()) {
 272     CLS_LOG(0, "object name or op id not specified");
 273     return -EINVAL;
 274   }
 275
 276
 277   cls_statelog_entry entry;
 278
 279   int rc = get_existing_entry(hctx, op.client_id, op.op_id, op.object, entry);
 280   if (rc < 0)
 281     return rc;
 282
 283   if (entry.state != op.state)
 284     return -ECANCELED;
 285
 286   return 0;
 287 }
 288
 289 CLS_INIT(statelog)
 290 {
 291   CLS_LOG(1, "Loaded log class!");
 292
 293   cls_handle_t h_class;
 294   cls_method_handle_t h_statelog_add;
 295   cls_method_handle_t h_statelog_list;
 296   cls_method_handle_t h_statelog_remove;
 297   cls_method_handle_t h_statelog_check_state;
 298
 299   cls_register("statelog", &h_class);
 300
 301   /* log */
 302   cls_register_cxx_method(h_class, "add", CLS_METHOD_RD | CLS_METHOD_WR, cls_statelog_add, &h_statelog_add);
 303   cls_register_cxx_method(h_class, "list", CLS_METHOD_RD, cls_statelog_list, &h_statelog_list);
 304   cls_register_cxx_method(h_class, "remove", CLS_METHOD_RD | CLS_METHOD_WR, cls_statelog_remove, &h_statelog_remove);
 305   cls_register_cxx_method(h_class, "check_state", CLS_METHOD_RD, cls_statelog_check_state, &h_statelog_check_state);
 306
 307   return;
 308 }
 309