[ceph.git] / ceph / src / cls / user / cls_user.cc

// -*- mode:C; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#include <iostream>

#include <string.h>
#include <stdlib.h>
#include <errno.h>

#include "include/types.h"
#include "include/utime.h"
#include "objclass/objclass.h"

#include "cls_user_types.h"
#include "cls_user_ops.h"

CLS_VER(1,0)
CLS_NAME(user)

static int write_entry(cls_method_context_t hctx, const string& key, const cls_user_bucket_entry& entry)
{
  bufferlist bl;
  ::encode(entry, bl);

  int ret = cls_cxx_map_set_val(hctx, key, &bl);
  if (ret < 0)
    return ret;

  return 0;
}

static int remove_entry(cls_method_context_t hctx, const string& key)
{
  int ret = cls_cxx_map_remove_key(hctx, key);
  if (ret < 0)
    return ret;

  return 0;
}

static void get_key_by_bucket_name(const string& bucket_name, string *key)
{
  *key = bucket_name;
}

static int get_existing_bucket_entry(cls_method_context_t hctx, const string& bucket_name,
                                     cls_user_bucket_entry& entry)
{
  if (bucket_name.empty()) {
    return -EINVAL;
  }

  string key;
  get_key_by_bucket_name(bucket_name, &key);

  bufferlist bl;
  int rc = cls_cxx_map_get_val(hctx, key, &bl);
  if (rc < 0) {
    CLS_LOG(10, "could not read entry %s", key.c_str());
    return rc;
  }
  try {
    bufferlist::iterator iter = bl.begin();
    ::decode(entry, iter);
  } catch (buffer::error& err) {
    CLS_LOG(0, "ERROR: failed to decode entry %s", key.c_str());
    return -EIO;
  }

  return 0;
}

static int read_header(cls_method_context_t hctx, cls_user_header *header)
{
  bufferlist bl;

  int ret = cls_cxx_map_read_header(hctx, &bl);
  if (ret < 0)
    return ret;

  if (bl.length() == 0) {
    *header = cls_user_header();
    return 0;
  }

  try {
    ::decode(*header, bl);
  } catch (buffer::error& err) {
    CLS_LOG(0, "ERROR: failed to decode user header");
    return -EIO;
  }

  return 0;
}

static void add_header_stats(cls_user_stats *stats, cls_user_bucket_entry& entry)
{
  stats->total_entries += entry.count;
  stats->total_bytes += entry.size;
  stats->total_bytes_rounded += entry.size_rounded;
}

static void dec_header_stats(cls_user_stats *stats, cls_user_bucket_entry& entry)
{
  stats->total_bytes -= entry.size;
  stats->total_bytes_rounded -= entry.size_rounded;
  stats->total_entries -= entry.count;
}

static void apply_entry_stats(const cls_user_bucket_entry& src_entry, cls_user_bucket_entry *target_entry)
{
  target_entry->size = src_entry.size;
  target_entry->size_rounded = src_entry.size_rounded;
  target_entry->count = src_entry.count;
}

static int cls_user_set_buckets_info(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
{
  bufferlist::iterator in_iter = in->begin();

  cls_user_set_buckets_op op;
  try {
    ::decode(op, in_iter);
  } catch (buffer::error& err) {
    CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
    return -EINVAL;
  }

  cls_user_header header;
  int ret = read_header(hctx, &header);
  if (ret < 0) {
    CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
    return ret;
  }

  for (list<cls_user_bucket_entry>::iterator iter = op.entries.begin();
       iter != op.entries.end(); ++iter) {
    cls_user_bucket_entry& update_entry = *iter;

    string key;

    get_key_by_bucket_name(update_entry.bucket.name, &key);

    cls_user_bucket_entry entry;
    ret = get_existing_bucket_entry(hctx, key, entry);

    if (ret == -ENOENT) {
     if (!op.add)
      continue; /* racing bucket removal */

     entry = update_entry;

     ret = 0;
    }

    if (ret < 0) {
      CLS_LOG(0, "ERROR: get_existing_bucket_entry() key=%s returned %d", key.c_str(), ret);
      return ret;
    } else if (ret >= 0 && entry.user_stats_sync) {
      dec_header_stats(&header.stats, entry);
    }

    CLS_LOG(20, "storing entry for key=%s size=%lld count=%lld",
            key.c_str(), (long long)update_entry.size, (long long)update_entry.count);

    // sync entry stats when not an op.add, as when the case is op.add if its a
    // new entry we already have copied update_entry earlier, OTOH, for an existing entry
    // we end up clobbering the existing stats for the bucket
    if (!op.add){
      apply_entry_stats(update_entry, &entry);
    }

    entry.user_stats_sync = true;

    ret = write_entry(hctx, key, entry);
    if (ret < 0)
      return ret;

    add_header_stats(&header.stats, entry);
  }

  bufferlist bl;

  CLS_LOG(20, "header: total bytes=%lld entries=%lld", (long long)header.stats.total_bytes, (long long)header.stats.total_entries);

  if (header.last_stats_update < op.time)
    header.last_stats_update = op.time;

  ::encode(header, bl);
  
  ret = cls_cxx_map_write_header(hctx, &bl);
  if (ret < 0)
    return ret;

  return 0;
}

static int cls_user_complete_stats_sync(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
{
  bufferlist::iterator in_iter = in->begin();

  cls_user_complete_stats_sync_op op;
  try {
    ::decode(op, in_iter);
  } catch (buffer::error& err) {
    CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
    return -EINVAL;
  }

  cls_user_header header;
  int ret = read_header(hctx, &header);
  if (ret < 0) {
    CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
    return ret;
  }

  if (header.last_stats_sync < op.time)
    header.last_stats_sync = op.time;

  bufferlist bl;

  ::encode(header, bl);

  ret = cls_cxx_map_write_header(hctx, &bl);
  if (ret < 0)
    return ret;

  return 0;
}

static int cls_user_remove_bucket(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
{
  bufferlist::iterator in_iter = in->begin();

  cls_user_remove_bucket_op op;
  try {
    ::decode(op, in_iter);
  } catch (buffer::error& err) {
    CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
    return -EINVAL;
  }

  cls_user_header header;
  int ret = read_header(hctx, &header);
  if (ret < 0) {
    CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
    return ret;
  }

  string key;

  get_key_by_bucket_name(op.bucket.name, &key);

  cls_user_bucket_entry entry;
  ret = get_existing_bucket_entry(hctx, key, entry);
  if (ret == -ENOENT) {
    return 0; /* idempotent removal */
  }
  if (ret < 0) {
    CLS_LOG(0, "ERROR: get existing bucket entry, key=%s ret=%d", key.c_str(), ret);
    return ret;
  }

  if (entry.user_stats_sync) {
    dec_header_stats(&header.stats, entry);
  }

  CLS_LOG(20, "removing entry at %s", key.c_str());

  ret = remove_entry(hctx, key);
  if (ret < 0)
    return ret;
  
  return 0;
}

static int cls_user_list_buckets(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
{
  bufferlist::iterator in_iter = in->begin();

  cls_user_list_buckets_op op;
  try {
    ::decode(op, in_iter);
  } catch (buffer::error& err) {
    CLS_LOG(1, "ERROR: cls_user_list_op(): failed to decode op");
    return -EINVAL;
  }

  map<string, bufferlist> keys;

  const string& from_index = op.marker;
  const string& to_index = op.end_marker;
  const bool to_index_valid = !to_index.empty();

#define MAX_ENTRIES 1000
  size_t max_entries = op.max_entries;
  if (max_entries > MAX_ENTRIES)
    max_entries = MAX_ENTRIES;

  string match_prefix;

  int rc = cls_cxx_map_get_vals(hctx, from_index, match_prefix, max_entries + 1, &keys);
  if (rc < 0)
    return rc;

  CLS_LOG(20, "from_index=%s to_index=%s match_prefix=%s",
          from_index.c_str(),
          to_index.c_str(),
          match_prefix.c_str());
  cls_user_list_buckets_ret ret;

  list<cls_user_bucket_entry>& entries = ret.entries;
  map<string, bufferlist>::iterator iter = keys.begin();

  bool done = false;
  string marker;

  size_t i;
  for (i = 0; i < max_entries && iter != keys.end(); ++i, ++iter) {
    const string& index = iter->first;
    marker = index;

    if (to_index_valid && to_index.compare(index) <= 0)
      break;

    bufferlist& bl = iter->second;
    bufferlist::iterator biter = bl.begin();
    try {
      cls_user_bucket_entry e;
      ::decode(e, biter);
      entries.push_back(e);
    } catch (buffer::error& err) {
      CLS_LOG(0, "ERROR: cls_user_list: could not decode entry, index=%s", index.c_str());
    }
  }

  if (iter == keys.end())
    done = true;
  else
    ret.marker = marker;

  ret.truncated = !done;

  ::encode(ret, *out);

  return 0;
}

static int cls_user_get_header(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
{
  bufferlist::iterator in_iter = in->begin();

  cls_user_get_header_op op;
  try {
    ::decode(op, in_iter);
  } catch (buffer::error& err) {
    CLS_LOG(1, "ERROR: cls_user_get_header_op(): failed to decode op");
    return -EINVAL;
  }

  cls_user_get_header_ret op_ret;

  int ret = read_header(hctx, &op_ret.header);
  if (ret < 0)
    return ret;

  ::encode(op_ret, *out);

  return 0;
}

CLS_INIT(user)
{
  CLS_LOG(1, "Loaded user class!");

  cls_handle_t h_class;
  cls_method_handle_t h_user_set_buckets_info;
  cls_method_handle_t h_user_complete_stats_sync;
  cls_method_handle_t h_user_remove_bucket;
  cls_method_handle_t h_user_list_buckets;
  cls_method_handle_t h_user_get_header;

  cls_register("user", &h_class);

  /* log */
  cls_register_cxx_method(h_class, "set_buckets_info", CLS_METHOD_RD | CLS_METHOD_WR,
                          cls_user_set_buckets_info, &h_user_set_buckets_info);
  cls_register_cxx_method(h_class, "complete_stats_sync", CLS_METHOD_RD | CLS_METHOD_WR,
                          cls_user_complete_stats_sync, &h_user_complete_stats_sync);
  cls_register_cxx_method(h_class, "remove_bucket", CLS_METHOD_RD | CLS_METHOD_WR, cls_user_remove_bucket, &h_user_remove_bucket);
  cls_register_cxx_method(h_class, "list_buckets", CLS_METHOD_RD, cls_user_list_buckets, &h_user_list_buckets);
  cls_register_cxx_method(h_class, "get_header", CLS_METHOD_RD, cls_user_get_header, &h_user_get_header);

  return;
}
Commit	Line	Data
7c673cae FG	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_user_types.h"
	15	#include "cls_user_ops.h"
	16
	17	CLS_VER(1,0)
	18	CLS_NAME(user)
	19
	20	static int write_entry(cls_method_context_t hctx, const string& key, const cls_user_bucket_entry& entry)
	21	{
	22	bufferlist bl;
	23	::encode(entry, bl);
	24
	25	int ret = cls_cxx_map_set_val(hctx, key, &bl);
	26	if (ret < 0)
	27	return ret;
	28
	29	return 0;
	30	}
	31
	32	static int remove_entry(cls_method_context_t hctx, const string& key)
	33	{
	34	int ret = cls_cxx_map_remove_key(hctx, key);
	35	if (ret < 0)
	36	return ret;
	37
	38	return 0;
	39	}
	40
	41	static void get_key_by_bucket_name(const string& bucket_name, string *key)
	42	{
	43	*key = bucket_name;
	44	}
	45
	46	static int get_existing_bucket_entry(cls_method_context_t hctx, const string& bucket_name,
	47	cls_user_bucket_entry& entry)
	48	{
	49	if (bucket_name.empty()) {
	50	return -EINVAL;
	51	}
	52
	53	string key;
	54	get_key_by_bucket_name(bucket_name, &key);
	55
	56	bufferlist bl;
	57	int rc = cls_cxx_map_get_val(hctx, key, &bl);
	58	if (rc < 0) {
	59	CLS_LOG(10, "could not read entry %s", key.c_str());
	60	return rc;
	61	}
	62	try {
	63	bufferlist::iterator iter = bl.begin();
	64	::decode(entry, iter);
65	} catch (buffer::error& err) {
66	CLS_LOG(0, "ERROR: failed to decode entry %s", key.c_str());
67	return -EIO;
68	}
69
70	return 0;
71	}
72
73	static int read_header(cls_method_context_t hctx, cls_user_header *header)
74	{
75	bufferlist bl;
76
77	int ret = cls_cxx_map_read_header(hctx, &bl);
78	if (ret < 0)
79	return ret;
80
81	if (bl.length() == 0) {
82	*header = cls_user_header();
83	return 0;
84	}
85
86	try {
87	::decode(*header, bl);
88	} catch (buffer::error& err) {
89	CLS_LOG(0, "ERROR: failed to decode user header");
90	return -EIO;
91	}
92
93	return 0;
94	}
95
96	static void add_header_stats(cls_user_stats *stats, cls_user_bucket_entry& entry)
97	{
98	stats->total_entries += entry.count;
99	stats->total_bytes += entry.size;
100	stats->total_bytes_rounded += entry.size_rounded;
101	}
102
103	static void dec_header_stats(cls_user_stats *stats, cls_user_bucket_entry& entry)
104	{
105	stats->total_bytes -= entry.size;
106	stats->total_bytes_rounded -= entry.size_rounded;
107	stats->total_entries -= entry.count;
108	}
109
110	static void apply_entry_stats(const cls_user_bucket_entry& src_entry, cls_user_bucket_entry *target_entry)
111	{
112	target_entry->size = src_entry.size;
113	target_entry->size_rounded = src_entry.size_rounded;
114	target_entry->count = src_entry.count;
115	}
116
117	static int cls_user_set_buckets_info(cls_method_context_t hctx, bufferlist in, bufferlist out)
118	{
119	bufferlist::iterator in_iter = in->begin();
120
121	cls_user_set_buckets_op op;
122	try {
123	::decode(op, in_iter);
124	} catch (buffer::error& err) {
125	CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
126	return -EINVAL;
127	}
128
129	cls_user_header header;
130	int ret = read_header(hctx, &header);
131	if (ret < 0) {
132	CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
133	return ret;
134	}
135
136	for (list<cls_user_bucket_entry>::iterator iter = op.entries.begin();
137	iter != op.entries.end(); ++iter) {
138	cls_user_bucket_entry& update_entry = *iter;
139
140	string key;
141
142	get_key_by_bucket_name(update_entry.bucket.name, &key);
143
144	cls_user_bucket_entry entry;
145	ret = get_existing_bucket_entry(hctx, key, entry);
146
147	if (ret == -ENOENT) {
148	if (!op.add)
149	continue; /* racing bucket removal */
150
151	entry = update_entry;
152
153	ret = 0;
154	}
155
156	if (ret < 0) {
157	CLS_LOG(0, "ERROR: get_existing_bucket_entry() key=%s returned %d", key.c_str(), ret);
158	return ret;
159	} else if (ret >= 0 && entry.user_stats_sync) {
160	dec_header_stats(&header.stats, entry);
161	}
162
163	CLS_LOG(20, "storing entry for key=%s size=%lld count=%lld",
164	key.c_str(), (long long)update_entry.size, (long long)update_entry.count);
165
166	// sync entry stats when not an op.add, as when the case is op.add if its a
167	// new entry we already have copied update_entry earlier, OTOH, for an existing entry
168	// we end up clobbering the existing stats for the bucket
169	if (!op.add){
170	apply_entry_stats(update_entry, &entry);
171	}
172
173	entry.user_stats_sync = true;
174
175	ret = write_entry(hctx, key, entry);
176	if (ret < 0)
177	return ret;
178
179	add_header_stats(&header.stats, entry);
180	}
181
182	bufferlist bl;
183
184	CLS_LOG(20, "header: total bytes=%lld entries=%lld", (long long)header.stats.total_bytes, (long long)header.stats.total_entries);
185
186	if (header.last_stats_update < op.time)
187	header.last_stats_update = op.time;
188
189	::encode(header, bl);
190
191	ret = cls_cxx_map_write_header(hctx, &bl);
192	if (ret < 0)
193	return ret;
194
195	return 0;
196	}
197
198	static int cls_user_complete_stats_sync(cls_method_context_t hctx, bufferlist in, bufferlist out)
199	{
200	bufferlist::iterator in_iter = in->begin();
201
202	cls_user_complete_stats_sync_op op;
203	try {
204	::decode(op, in_iter);
205	} catch (buffer::error& err) {
206	CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
207	return -EINVAL;
208	}
209
210	cls_user_header header;
211	int ret = read_header(hctx, &header);
212	if (ret < 0) {
213	CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
214	return ret;
215	}
216
217	if (header.last_stats_sync < op.time)
218	header.last_stats_sync = op.time;
219
220	bufferlist bl;
221
222	::encode(header, bl);
223
224	ret = cls_cxx_map_write_header(hctx, &bl);
225	if (ret < 0)
226	return ret;
227
228	return 0;
229	}
230
231	static int cls_user_remove_bucket(cls_method_context_t hctx, bufferlist in, bufferlist out)
232	{
233	bufferlist::iterator in_iter = in->begin();
234
235	cls_user_remove_bucket_op op;
236	try {
237	::decode(op, in_iter);
238	} catch (buffer::error& err) {
239	CLS_LOG(1, "ERROR: cls_user_add_op(): failed to decode op");
240	return -EINVAL;
241	}
242
243	cls_user_header header;
244	int ret = read_header(hctx, &header);
245	if (ret < 0) {
246	CLS_LOG(0, "ERROR: failed to read user info header ret=%d", ret);
247	return ret;
248	}
249
250	string key;
251
252	get_key_by_bucket_name(op.bucket.name, &key);
253
254	cls_user_bucket_entry entry;
255	ret = get_existing_bucket_entry(hctx, key, entry);
256	if (ret == -ENOENT) {
257	return 0; /* idempotent removal */
258	}
259	if (ret < 0) {
260	CLS_LOG(0, "ERROR: get existing bucket entry, key=%s ret=%d", key.c_str(), ret);
261	return ret;
262	}
263
264	if (entry.user_stats_sync) {
265	dec_header_stats(&header.stats, entry);
266	}
267
268	CLS_LOG(20, "removing entry at %s", key.c_str());
269
270	ret = remove_entry(hctx, key);
271	if (ret < 0)
272	return ret;
273
274	return 0;
275	}
276
277	static int cls_user_list_buckets(cls_method_context_t hctx, bufferlist in, bufferlist out)
278	{
279	bufferlist::iterator in_iter = in->begin();
280
281	cls_user_list_buckets_op op;
282	try {
283	::decode(op, in_iter);
284	} catch (buffer::error& err) {
285	CLS_LOG(1, "ERROR: cls_user_list_op(): failed to decode op");
286	return -EINVAL;
287	}
288
289	map<string, bufferlist> keys;
290
291	const string& from_index = op.marker;
292	const string& to_index = op.end_marker;
293	const bool to_index_valid = !to_index.empty();
294
295	#define MAX_ENTRIES 1000
296	size_t max_entries = op.max_entries;
297	if (max_entries > MAX_ENTRIES)
298	max_entries = MAX_ENTRIES;
299
300	string match_prefix;
301
302	int rc = cls_cxx_map_get_vals(hctx, from_index, match_prefix, max_entries + 1, &keys);
303	if (rc < 0)
304	return rc;
305
306	CLS_LOG(20, "from_index=%s to_index=%s match_prefix=%s",
307	from_index.c_str(),
308	to_index.c_str(),
309	match_prefix.c_str());
310	cls_user_list_buckets_ret ret;
311
312	list<cls_user_bucket_entry>& entries = ret.entries;
313	map<string, bufferlist>::iterator iter = keys.begin();
314
315	bool done = false;
316	string marker;
317
318	size_t i;
319	for (i = 0; i < max_entries && iter != keys.end(); ++i, ++iter) {
320	const string& index = iter->first;
321	marker = index;
322
323	if (to_index_valid && to_index.compare(index) <= 0)
324	break;
325
326	bufferlist& bl = iter->second;
327	bufferlist::iterator biter = bl.begin();
328	try {
329	cls_user_bucket_entry e;
330	::decode(e, biter);
331	entries.push_back(e);
332	} catch (buffer::error& err) {
333	CLS_LOG(0, "ERROR: cls_user_list: could not decode entry, index=%s", index.c_str());
334	}
335	}
336
337	if (iter == keys.end())
338	done = true;
339	else
340	ret.marker = marker;
341
342	ret.truncated = !done;
343
344	::encode(ret, *out);
345
346	return 0;
347	}
348
349	static int cls_user_get_header(cls_method_context_t hctx, bufferlist in, bufferlist out)
350	{
351	bufferlist::iterator in_iter = in->begin();
352
353	cls_user_get_header_op op;
354	try {
355	::decode(op, in_iter);
356	} catch (buffer::error& err) {
357	CLS_LOG(1, "ERROR: cls_user_get_header_op(): failed to decode op");
358	return -EINVAL;
359	}
360
361	cls_user_get_header_ret op_ret;
362
363	int ret = read_header(hctx, &op_ret.header);
364	if (ret < 0)
365	return ret;
366
367	::encode(op_ret, *out);
368
369	return 0;
370	}
371
372	CLS_INIT(user)
373	{
374	CLS_LOG(1, "Loaded user class!");
375
376	cls_handle_t h_class;
377	cls_method_handle_t h_user_set_buckets_info;
378	cls_method_handle_t h_user_complete_stats_sync;
379	cls_method_handle_t h_user_remove_bucket;
380	cls_method_handle_t h_user_list_buckets;
381	cls_method_handle_t h_user_get_header;
382
383	cls_register("user", &h_class);
384
385	/* log */
386	cls_register_cxx_method(h_class, "set_buckets_info", CLS_METHOD_RD \| CLS_METHOD_WR,
387	cls_user_set_buckets_info, &h_user_set_buckets_info);
388	cls_register_cxx_method(h_class, "complete_stats_sync", CLS_METHOD_RD \| CLS_METHOD_WR,
389	cls_user_complete_stats_sync, &h_user_complete_stats_sync);
390	cls_register_cxx_method(h_class, "remove_bucket", CLS_METHOD_RD \| CLS_METHOD_WR, cls_user_remove_bucket, &h_user_remove_bucket);
391	cls_register_cxx_method(h_class, "list_buckets", CLS_METHOD_RD, cls_user_list_buckets, &h_user_list_buckets);
392	cls_register_cxx_method(h_class, "get_header", CLS_METHOD_RD, cls_user_get_header, &h_user_get_header);
393
394	return;
395	}
396