From: Mehdi Abaakouk Date: Fri, 12 Feb 2016 08:13:36 +0000 (+0100) Subject: Cython Rados module X-Git-Tag: v10.1.0~369^2~5 X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=9997b32d0b8703e984a0b68c8dc828a442830832;p=ceph.git Cython Rados module Notable changes: * run_in_thread have disapeared * timeout argument of some methods are ignored * rados_create_write_op/rados_create_read_op returns WriteOp/ReadOp instead of the pointer address * rados_monitor_log callback 'arg' arguments was broken in previous python binding (callback was called with the pointer address instead pointed object) * object attributes that was pointer addresses are now private and not accessible in python Some tests have been added to cover all methods Signed-off-by: Mehdi Abaakouk --- diff --git a/src/pybind/Makefile.am b/src/pybind/Makefile.am index 78bc558e9162..fa5ff8d279e8 100644 --- a/src/pybind/Makefile.am +++ b/src/pybind/Makefile.am @@ -12,6 +12,7 @@ PY_DISTUTILS = \ ${PYTHON} ./setup.py include pybind/rbd/Makefile.am +include pybind/rados/Makefile.am endif endif diff --git a/src/pybind/rados/Makefile.am b/src/pybind/rados/Makefile.am new file mode 100644 index 000000000000..07b914b628b1 --- /dev/null +++ b/src/pybind/rados/Makefile.am @@ -0,0 +1,35 @@ +EXTRA_DIST += $(srcdir)/pybind/rados/setup.py $(srcdir)/pybind/rados/rados.pyx $(srcdir)/pybind/rados/rados.pxd + +rados-pybind-all: librados.la ${srcdir}/ceph_ver.h + cd $(srcdir)/pybind/rados; $(PY_DISTUTILS) build \ + --build-base $(shell readlink -f $(builddir))/build \ + --verbose + +rados-pybind-clean: ${srcdir}/ceph_ver.h + cd $(srcdir)/pybind/rados; $(PY_DISTUTILS) clean \ + --build-base $(shell readlink -f $(builddir))/build \ + --verbose + +rados-pybind-install-exec: ${srcdir}/ceph_ver.h + if test "$(DESTDIR)" ; then \ + if lsb_release -si | grep --quiet 'Ubuntu\|Debian\|Devuan' ; then \ + options=--install-layout=deb ; \ + else \ + options=--prefix=/usr ; \ + fi ; \ + root="--root=$(DESTDIR)" ; \ + else \ + options=--prefix=$(prefix) ; \ + fi ; \ + cd $(srcdir)/pybind; $(PY_DISTUTILS) build \ + --build-base $(shell readlink -f $(builddir))/build \ + install \ + $$options $$root \ + --single-version-externally-managed \ + --record /dev/null \ + --verbose + +LOCAL_ALL += rados-pybind-all +LOCAL_CLEAN += rados-pybind-clean +LOCAL_INSTALLEXEC += rados-pybind-install-exec + diff --git a/src/pybind/rados/rados.pxd b/src/pybind/rados/rados.pxd new file mode 100644 index 000000000000..d9ca4b4e6924 --- /dev/null +++ b/src/pybind/rados/rados.pxd @@ -0,0 +1,38 @@ +# cython: embedsignature=True +# +# Shared object for librbdpy +# +# Copyright 2016 Mehdi Abaakouk + + +cdef extern from "rados/librados.h" nogil: + ctypedef void* rados_t + ctypedef void* rados_config_t + ctypedef void* rados_ioctx_t + + +cdef class Rados(object): + cdef: + rados_t cluster + public object state + public object monitor_callback + public object parsed_args + public object conf_defaults + public object conffile + public object rados_id + + +cdef class Ioctx(object): + cdef: + rados_ioctx_t io + public char *name + public char *state + public object locator_key + public object nspace + + # TODO(sileht): we need to track leaving completion objects + # I guess we can do that in a lighter ways, but keep code simple + # as before for now + public object safe_completions + public object complete_completions + public object lock diff --git a/src/pybind/rados/rados.pyx b/src/pybind/rados/rados.pyx new file mode 100644 index 000000000000..81fd1dc21d2c --- /dev/null +++ b/src/pybind/rados/rados.pyx @@ -0,0 +1,3206 @@ +# cython: embedsignature=True +""" +This module is a thin wrapper around librados. + +Error codes from librados are turned into exceptions that subclass +:class:`Error`. Almost all methods may raise :class:`Error(the base class of all rados exceptions), :class:`PermissionError` +(the base class of all rados exceptions), :class:`PermissionError` +and :class:`IOError`, in addition to those documented for the +method. +""" +# Copyright 2011 Josh Durgin +# Copyright 2011, Hannu Valtonen +# Copyright 2015 Hector Martin +# Copyright 2016 Mehdi Abaakouk + +from cpython cimport PyObject, ref, exc, array +from cpython.dict cimport PyDict_New +from libc cimport errno +from libc.stdint cimport * +from libc.stdlib cimport malloc, realloc, free + +import array +import sys +import threading +import time + +from datetime import datetime +from functools import partial, wraps +from itertools import chain + +# Are we running Python 2.x +_python2 = sys.hexversion < 0x03000000 + +if _python2: + str_type = basestring +else: + str_type = str + + +cdef extern from "Python.h": + # These are in cpython/string.pxd, but use "object" types instead of + # PyObject*, which invokes assumptions in cpython that we need to + # legitimately break to implement zero-copy string buffers in Image.read(). + # This is valid use of the Python API and documented as a special case. + PyObject *PyBytes_FromStringAndSize(char *v, Py_ssize_t len) except NULL + char* PyBytes_AsString(PyObject *string) except NULL + int _PyBytes_Resize(PyObject **string, Py_ssize_t newsize) except -1 + void PyEval_InitThreads() + + +cdef extern from "time.h": + ctypedef long int time_t + ctypedef long int suseconds_t + + +cdef extern from "sys/time.h": + cdef struct timeval: + time_t tv_sec + suseconds_t tv_usec + + +cdef extern from "rados/rados_types.h" nogil: + cdef char* _LIBRADOS_ALL_NSPACES "LIBRADOS_ALL_NSPACES" + + +cdef extern from "rados/librados.h" nogil: + enum: + _LIBRADOS_OP_FLAG_EXCL "LIBRADOS_OP_FLAG_EXCL" + _LIBRADOS_OP_FLAG_FAILOK "LIBRADOS_OP_FLAG_FAILOK" + _LIBRADOS_OP_FLAG_FADVISE_RANDOM "LIBRADOS_OP_FLAG_FADVISE_RANDOM" + _LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL "LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL" + _LIBRADOS_OP_FLAG_FADVISE_WILLNEED "LIBRADOS_OP_FLAG_FADVISE_WILLNEED" + _LIBRADOS_OP_FLAG_FADVISE_DONTNEED "LIBRADOS_OP_FLAG_FADVISE_DONTNEED" + _LIBRADOS_OP_FLAG_FADVISE_NOCACHE "LIBRADOS_OP_FLAG_FADVISE_NOCACHE" + + + enum: + _LIBRADOS_OPERATION_NOFLAG "LIBRADOS_OPERATION_NOFLAG" + _LIBRADOS_OPERATION_BALANCE_READS "LIBRADOS_OPERATION_BALANCE_READS" + _LIBRADOS_OPERATION_LOCALIZE_READS "LIBRADOS_OPERATION_LOCALIZE_READS" + _LIBRADOS_OPERATION_ORDER_READS_WRITES "LIBRADOS_OPERATION_ORDER_READS_WRITES" + _LIBRADOS_OPERATION_IGNORE_CACHE "LIBRADOS_OPERATION_IGNORE_CACHE" + _LIBRADOS_OPERATION_SKIPRWLOCKS "LIBRADOS_OPERATION_SKIPRWLOCKS" + _LIBRADOS_OPERATION_IGNORE_OVERLAY "LIBRADOS_OPERATION_IGNORE_OVERLAY" + + cdef uint64_t _LIBRADOS_SNAP_HEAD "LIBRADOS_SNAP_HEAD" + + ctypedef void* rados_t + ctypedef void* rados_config_t + ctypedef void* rados_ioctx_t + ctypedef void* rados_xattrs_iter_t + ctypedef void* rados_omap_iter_t + ctypedef void* rados_list_ctx_t + ctypedef uint64_t rados_snap_t + ctypedef void *rados_write_op_t + ctypedef void *rados_read_op_t + ctypedef void *rados_completion_t + ctypedef void (*rados_callback_t)(rados_completion_t cb, void *arg) + ctypedef void (*rados_log_callback_t)(void *arg, const char *line, const char *who, + uint64_t sec, uint64_t nsec, uint64_t seq, const char *level, const char *msg) + + + cdef struct rados_cluster_stat_t: + uint64_t kb + uint64_t kb_used + uint64_t kb_avail + uint64_t num_objects + + cdef struct rados_pool_stat_t: + uint64_t num_bytes + uint64_t num_kb + uint64_t num_objects + uint64_t num_object_clones + uint64_t num_object_copies + uint64_t num_objects_missing_on_primary + uint64_t num_objects_unfound + uint64_t num_objects_degraded + uint64_t num_rd + uint64_t num_rd_kb + uint64_t num_wr + uint64_t num_wr_kb + + void rados_buffer_free(char *buf) + + void rados_version(int *major, int *minor, int *extra) + int rados_create2(rados_t *pcluster, const char *const clustername, + const char * const name, uint64_t flags) + int rados_connect(rados_t cluster) + void rados_shutdown(rados_t cluster) + int rados_conf_read_file(rados_t cluster, const char *path) + int rados_conf_parse_argv_remainder(rados_t cluster, int argc, const char **argv, const char **remargv) + int rados_conf_parse_env(rados_t cluster, const char *var) + int rados_conf_set(rados_t cluster, char *option, const char *value) + int rados_conf_get(rados_t cluster, char *option, char *buf, size_t len) + + int rados_ioctx_pool_stat(rados_ioctx_t io, rados_pool_stat_t *stats) + int64_t rados_pool_lookup(rados_t cluster, const char *pool_name) + int rados_pool_reverse_lookup(rados_t cluster, int64_t id, char *buf, size_t maxlen) + int rados_pool_create(rados_t cluster, const char *pool_name) + int rados_pool_create_with_auid(rados_t cluster, const char *pool_name, uint64_t auid) + int rados_pool_create_with_crush_rule(rados_t cluster, const char *pool_name, uint8_t crush_rule_num) + int rados_pool_create_with_all(rados_t cluster, const char *pool_name, uint64_t auid, uint8_t crush_rule_num) + int rados_pool_get_base_tier(rados_t cluster, int64_t pool, int64_t *base_tier) + int rados_pool_list(rados_t cluster, char *buf, size_t len) + int rados_pool_delete(rados_t cluster, const char *pool_name) + + int rados_cluster_stat(rados_t cluster, rados_cluster_stat_t *result) + int rados_cluster_fsid(rados_t cluster, char *buf, size_t len) + int rados_blacklist_add(rados_t cluster, char *client_address, uint32_t expire_seconds) + + int rados_ping_monitor(rados_t cluster, const char *mon_id, char **outstr, size_t *outstrlen) + int rados_mon_command(rados_t cluster, const char **cmd, size_t cmdlen, + const char *inbuf, size_t inbuflen, + char **outbuf, size_t *outbuflen, + char **outs, size_t *outslen) + int rados_mon_command_target(rados_t cluster, const char *name, const char **cmd, size_t cmdlen, + const char *inbuf, size_t inbuflen, + char **outbuf, size_t *outbuflen, + char **outs, size_t *outslen) + int rados_osd_command(rados_t cluster, int osdid, const char **cmd, size_t cmdlen, + const char *inbuf, size_t inbuflen, + char **outbuf, size_t *outbuflen, + char **outs, size_t *outslen) + int rados_pg_command(rados_t cluster, const char *pgstr, const char **cmd, size_t cmdlen, + const char *inbuf, size_t inbuflen, + char **outbuf, size_t *outbuflen, + char **outs, size_t *outslen) + int rados_monitor_log(rados_t cluster, const char *level, rados_log_callback_t cb, void *arg) + + int rados_wait_for_latest_osdmap(rados_t cluster) + + int rados_ioctx_create(rados_t cluster, const char *pool_name, rados_ioctx_t *ioctx) + void rados_ioctx_destroy(rados_ioctx_t io) + int rados_ioctx_pool_set_auid(rados_ioctx_t io, uint64_t auid) + void rados_ioctx_locator_set_key(rados_ioctx_t io, const char *key) + void rados_ioctx_set_namespace(rados_ioctx_t io, const char * nspace) + + uint64_t rados_get_last_version(rados_ioctx_t io) + int rados_stat(rados_ioctx_t io, const char *o, uint64_t *psize, time_t *pmtime) + int rados_write(rados_ioctx_t io, const char *oid, const char *buf, size_t len, uint64_t off) + int rados_write_full(rados_ioctx_t io, const char *oid, const char *buf, size_t len) + int rados_append(rados_ioctx_t io, const char *oid, const char *buf, size_t len) + int rados_read(rados_ioctx_t io, const char *oid, char *buf, size_t len, uint64_t off) + int rados_remove(rados_ioctx_t io, const char *oid) + int rados_trunc(rados_ioctx_t io, const char *oid, uint64_t size) + int rados_getxattr(rados_ioctx_t io, const char *o, const char *name, char *buf, size_t len) + int rados_setxattr(rados_ioctx_t io, const char *o, const char *name, const char *buf, size_t len) + int rados_rmxattr(rados_ioctx_t io, const char *o, const char *name) + int rados_getxattrs(rados_ioctx_t io, const char *oid, rados_xattrs_iter_t *iter) + int rados_getxattrs_next(rados_xattrs_iter_t iter, const char **name, const char **val, size_t *len) + void rados_getxattrs_end(rados_xattrs_iter_t iter) + + int rados_nobjects_list_open(rados_ioctx_t io, rados_list_ctx_t *ctx) + int rados_nobjects_list_next(rados_list_ctx_t ctx, const char **entry, const char **key, const char **nspace) + void rados_nobjects_list_close(rados_list_ctx_t ctx) + + int rados_ioctx_snap_rollback(rados_ioctx_t io, const char * oid, const char * snapname) + int rados_ioctx_snap_create(rados_ioctx_t io, const char * snapname) + int rados_ioctx_snap_remove(rados_ioctx_t io, const char * snapname) + int rados_ioctx_snap_lookup(rados_ioctx_t io, const char * name, rados_snap_t * id) + int rados_ioctx_snap_get_name(rados_ioctx_t io, rados_snap_t id, char * name, int maxlen) + void rados_ioctx_snap_set_read(rados_ioctx_t io, rados_snap_t snap) + int rados_ioctx_snap_list(rados_ioctx_t io, rados_snap_t * snaps, int maxlen) + int rados_ioctx_snap_get_stamp(rados_ioctx_t io, rados_snap_t id, time_t * t) + + int rados_lock_exclusive(rados_ioctx_t io, const char * oid, const char * name, + const char * cookie, const char * desc, + timeval * duration, uint8_t flags) + int rados_lock_shared(rados_ioctx_t io, const char * o, const char * name, + const char * cookie, const char * tag, const char * desc, + timeval * duration, uint8_t flags) + int rados_unlock(rados_ioctx_t io, const char * o, const char * name, const char * cookie) + + rados_write_op_t rados_create_write_op() + void rados_release_write_op(rados_write_op_t write_op) + + rados_read_op_t rados_create_read_op() + void rados_release_read_op(rados_read_op_t read_op) + + int rados_aio_create_completion(void * cb_arg, rados_callback_t cb_complete, rados_callback_t cb_safe, rados_completion_t * pc) + void rados_aio_release(rados_completion_t c) + int rados_aio_write(rados_ioctx_t io, const char * oid, rados_completion_t completion, const char * buf, size_t len, uint64_t off) + int rados_aio_append(rados_ioctx_t io, const char * oid, rados_completion_t completion, const char * buf, size_t len) + int rados_aio_write_full(rados_ioctx_t io, const char * oid, rados_completion_t completion, const char * buf, size_t len) + int rados_aio_remove(rados_ioctx_t io, const char * oid, rados_completion_t completion) + int rados_aio_read(rados_ioctx_t io, const char * oid, rados_completion_t completion, char * buf, size_t len, uint64_t off) + int rados_aio_flush(rados_ioctx_t io) + + int rados_aio_get_return_value(rados_completion_t c) + int rados_aio_wait_for_complete_and_cb(rados_completion_t c) + int rados_aio_wait_for_safe_and_cb(rados_completion_t c) + int rados_aio_wait_for_complete(rados_completion_t c) + int rados_aio_wait_for_safe(rados_completion_t c) + int rados_aio_is_complete(rados_completion_t c) + int rados_aio_is_safe(rados_completion_t c) + + int rados_exec(rados_ioctx_t io, const char * oid, const char * cls, const char * method, + const char * in_buf, size_t in_len, char * buf, size_t out_len) + + int rados_write_op_operate(rados_write_op_t write_op, rados_ioctx_t io, const char * oid, time_t * mtime, int flags) + void rados_write_op_omap_set(rados_write_op_t write_op, const char * const* keys, const char * const* vals, const size_t * lens, size_t num) + void rados_write_op_omap_rm_keys(rados_write_op_t write_op, const char * const* keys, size_t keys_len) + void rados_write_op_omap_clear(rados_write_op_t write_op) + void rados_read_op_omap_get_vals(rados_read_op_t read_op, const char * start_after, const char * filter_prefix, uint64_t max_return, rados_omap_iter_t * iter, int * prval) + void rados_read_op_omap_get_keys(rados_read_op_t read_op, const char * start_after, uint64_t max_return, rados_omap_iter_t * iter, int * prval) + void rados_read_op_omap_get_vals_by_keys(rados_read_op_t read_op, const char * const* keys, size_t keys_len, rados_omap_iter_t * iter, int * prval) + int rados_read_op_operate(rados_read_op_t read_op, rados_ioctx_t io, const char * oid, int flags) + int rados_omap_get_next(rados_omap_iter_t iter, const char * const* key, const char * const* val, size_t * len) + void rados_omap_get_end(rados_omap_iter_t iter) + + +LIBRADOS_OP_FLAG_EXCL = _LIBRADOS_OP_FLAG_EXCL +LIBRADOS_OP_FLAG_FAILOK = _LIBRADOS_OP_FLAG_FAILOK +LIBRADOS_OP_FLAG_FADVISE_RANDOM = _LIBRADOS_OP_FLAG_FADVISE_RANDOM +LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL = _LIBRADOS_OP_FLAG_FADVISE_SEQUENTIAL +LIBRADOS_OP_FLAG_FADVISE_WILLNEED = _LIBRADOS_OP_FLAG_FADVISE_WILLNEED +LIBRADOS_OP_FLAG_FADVISE_DONTNEED = _LIBRADOS_OP_FLAG_FADVISE_DONTNEED +LIBRADOS_OP_FLAG_FADVISE_NOCACHE = _LIBRADOS_OP_FLAG_FADVISE_NOCACHE + +LIBRADOS_SNAP_HEAD = _LIBRADOS_SNAP_HEAD + +LIBRADOS_OPERATION_NOFLAG = _LIBRADOS_OPERATION_NOFLAG +LIBRADOS_OPERATION_BALANCE_READS = _LIBRADOS_OPERATION_BALANCE_READS +LIBRADOS_OPERATION_LOCALIZE_READS = _LIBRADOS_OPERATION_LOCALIZE_READS +LIBRADOS_OPERATION_ORDER_READS_WRITES = _LIBRADOS_OPERATION_ORDER_READS_WRITES +LIBRADOS_OPERATION_IGNORE_CACHE = _LIBRADOS_OPERATION_IGNORE_CACHE +LIBRADOS_OPERATION_SKIPRWLOCKS = _LIBRADOS_OPERATION_SKIPRWLOCKS +LIBRADOS_OPERATION_IGNORE_OVERLAY = _LIBRADOS_OPERATION_IGNORE_OVERLAY + +LIBRADOS_ALL_NSPACES = _LIBRADOS_ALL_NSPACES.decode('utf-8') + +ANONYMOUS_AUID = 0xffffffffffffffff +ADMIN_AUID = 0 + +class Error(Exception): + """ `Error` class, derived from `Exception` """ + + +class InvalidArgument(Error): + pass + + +class InterruptedOrTimeoutError(Error): + """ `InterruptedOrTimeoutError` class, derived from `Error` """ + pass + + +class PermissionError(Error): + """ `PermissionError` class, derived from `Error` """ + pass + +class PermissionDeniedError(Error): + """ deal with EACCES related. """ + pass + +class ObjectNotFound(Error): + """ `ObjectNotFound` class, derived from `Error` """ + pass + + +class NoData(Error): + """ `NoData` class, derived from `Error` """ + pass + + +class ObjectExists(Error): + """ `ObjectExists` class, derived from `Error` """ + pass + + +class ObjectBusy(Error): + """ `ObjectBusy` class, derived from `Error` """ + pass + + +class IOError(Error): + """ `IOError` class, derived from `Error` """ + pass + + +class NoSpace(Error): + """ `NoSpace` class, derived from `Error` """ + pass + + +class IncompleteWriteError(Error): + """ `IncompleteWriteError` class, derived from `Error` """ + pass + + +class RadosStateError(Error): + """ `RadosStateError` class, derived from `Error` """ + pass + + +class IoctxStateError(Error): + """ `IoctxStateError` class, derived from `Error` """ + pass + +class ObjectStateError(Error): + """ `ObjectStateError` class, derived from `Error` """ + pass + + +class LogicError(Error): + """ `` class, derived from `Error` """ + pass + + +class TimedOut(Error): + """ `TimedOut` class, derived from `Error` """ + pass + + + +cdef errno_to_exception = { + errno.EPERM : PermissionError, + errno.ENOENT : ObjectNotFound, + errno.EIO : IOError, + errno.ENOSPC : NoSpace, + errno.EEXIST : ObjectExists, + errno.EBUSY : ObjectBusy, + errno.ENODATA : NoData, + errno.EINTR : InterruptedOrTimeoutError, + errno.ETIMEDOUT : TimedOut, + errno.EACCES : PermissionDeniedError +} + + +cdef make_ex(ret, msg): + """ + Translate a librados return code into an exception. + + :param ret: the return code + :type ret: int + :param msg: the error message to use + :type msg: str + :returns: a subclass of :class:`Error` + """ + ret = abs(ret) + if ret in errno_to_exception: + return errno_to_exception[ret](msg) + else: + return Error(msg + (": error code %d" % ret)) + + +# helper to specify an optional argument, where in addition to `cls`, `None` +# is also acceptable +def opt(cls): + return (cls, None) + + +# validate argument types of an instance method +# kwargs is an un-ordered dict, so use args instead +def requires(*types): + def is_type_of(v, t): + if t is None: + return v is None + else: + return isinstance(v, t) + + def check_type(val, arg_name, arg_type): + if isinstance(arg_type, tuple): + if any(is_type_of(val, t) for t in arg_type): + return + type_names = ' or '.join('None' if t is None else t.__name__ + for t in arg_type) + raise TypeError('%s must be %s' % (arg_name, type_names)) + else: + if is_type_of(val, arg_type): + return + assert(arg_type is not None) + raise TypeError('%s must be %s' % (arg_name, arg_type.__name__)) + + def wrapper(f): + # FIXME(sileht): this stop with + # AttributeError: 'method_descriptor' object has no attribute '__module__' + # @wraps(f) + def validate_func(*args, **kwargs): + # ignore the `self` arg + pos_args = zip(args[1:], types) + named_args = ((kwargs[name], (name, spec)) for name, spec in types + if name in kwargs) + for arg_val, (arg_name, arg_type) in chain(pos_args, named_args): + check_type(arg_val, arg_name, arg_type) + return f(*args, **kwargs) + return validate_func + return wrapper + + +def cstr(val, name, encoding="utf-8", opt=False): + """ + Create a byte string from a Python string + + :param basestring val: Python string + :param str name: Name of the string parameter, for exceptions + :param str encoding: Encoding to use + :param bool opt: If True, None is allowed + :rtype: bytes + :raises: :class:`InvalidArgument` + """ + if opt and val is None: + return None + if isinstance(val, bytes): + return val + elif isinstance(val, unicode): + return val.encode(encoding) + else: + raise TypeError('%s must be a string' % name) + + +def cstr_list(list_str, name, encoding="utf-8"): + return [cstr(s, name) for s in list_str] + + +def decode_cstr(val, encoding="utf-8"): + """ + Decode a byte string into a Python string. + + :param bytes val: byte string + :rtype: unicode or None + """ + if val is None: + return None + + return val.decode(encoding) + + +cdef char* opt_str(s) except? NULL: + if s is None: + return NULL + return s + + +cdef void* realloc_chk(void* ptr, size_t size) except NULL: + cdef void *ret = realloc(ptr, size) + if ret == NULL: + raise MemoryError("realloc failed") + return ret + + +cdef size_t * to_csize_t_array(list_int): + cdef size_t *ret = malloc(len(list_int) * sizeof(size_t)) + if ret == NULL: + raise MemoryError("malloc failed") + for i in xrange(len(list_int)): + ret[i] = list_int[i] + return ret + + +cdef char ** to_bytes_array(list_bytes): + cdef char **ret = malloc(len(list_bytes) * sizeof(char *)) + if ret == NULL: + raise MemoryError("malloc failed") + for i in xrange(len(list_bytes)): + ret[i] = list_bytes[i] + return ret + + + +cdef int __monitor_callback(void *arg, const char *line, const char *who, + uint64_t sec, uint64_t nsec, uint64_t seq, + const char *level, const char *msg) with gil: + cdef object cb_info = arg + cb_info[0](cb_info[1], line, who, sec, nsec, seq, level, msg) + return 0 + + +class Version(object): + """ Version information """ + def __init__(self, major, minor, extra): + self.major = major + self.minor = minor + self.extra = extra + + def __str__(self): + return "%d.%d.%d" % (self.major, self.minor, self.extra) + + +cdef class Rados(object): + """This class wraps librados functions""" + # NOTE(sileht): attributes declared in .pyd + + def __init__(self, *args, **kwargs): + PyEval_InitThreads() + self.__setup(*args, **kwargs) + + @requires(('rados_id', opt(str_type)), ('name', opt(str_type)), ('clustername', opt(str_type)), + ('conffile', opt(str_type))) + def __setup(self, rados_id=None, name=None, clustername=None, + conf_defaults=None, conffile=None, conf=None, flags=0): + self.monitor_callback = None + self.parsed_args = [] + self.conf_defaults = conf_defaults + self.conffile = conffile + self.rados_id = rados_id + + if rados_id and name: + raise Error("Rados(): can't supply both rados_id and name") + elif rados_id: + name = 'client.' + rados_id + elif name is None: + name = 'client.admin' + if clustername is None: + clustername = 'ceph' + + name = cstr(name, 'name') + clustername = cstr(clustername, 'clustername') + cdef: + char *_name = name + char *_clustername = clustername + int _flags = flags + int ret + + with nogil: + ret = rados_create2(&self.cluster, _clustername, _name, _flags) + if ret != 0: + raise Error("rados_initialize failed with error code: %d" % ret) + + self.state = "configuring" + # order is important: conf_defaults, then conffile, then conf + if conf_defaults: + for key, value in conf_defaults.items(): + self.conf_set(key, value) + if conffile is not None: + # read the default conf file when '' is given + if conffile == '': + conffile = None + self.conf_read_file(conffile) + if conf: + for key, value in conf.items(): + self.conf_set(key, value) + + def require_state(self, *args): + """ + Checks if the Rados object is in a special state + + :raises: RadosStateError + """ + if self.state in args: + return + raise RadosStateError("You cannot perform that operation on a \ +Rados object in state %s." % self.state) + + def shutdown(self): + """ + Disconnects from the cluster. Call this explicitly when a + Rados.connect()ed object is no longer used. + """ + if self.state != "shutdown": + with nogil: + rados_shutdown(self.cluster) + self.state = "shutdown" + + def __enter__(self): + self.connect() + return self + + def __exit__(self, type_, value, traceback): + self.shutdown() + return False + + def version(self): + """ + Get the version number of the ``librados`` C library. + + :returns: a tuple of ``(major, minor, extra)`` components of the + librados version + """ + cdef int major = 0 + cdef int minor = 0 + cdef int extra = 0 + with nogil: + rados_version(&major, &minor, &extra) + return Version(major, minor, extra) + + @requires(('path', opt(str_type))) + def conf_read_file(self, path=None): + """ + Configure the cluster handle using a Ceph config file. + + :param path: path to the config file + :type path: str + """ + self.require_state("configuring", "connected") + path = cstr(path, 'path', opt=True) + cdef: + char *_path = opt_str(path) + with nogil: + ret = rados_conf_read_file(self.cluster, _path) + if (ret != 0): + raise make_ex(ret, "error calling conf_read_file") + + def conf_parse_argv(self, args): + """ + Parse known arguments from args, and remove; returned + args contain only those unknown to ceph + """ + self.require_state("configuring", "connected") + if not args: + return + + cargs = cstr_list(args, 'args') + cdef: + int _argc = len(args) + char **_argv = to_bytes_array(cargs) + char **_remargv = NULL + + try: + _remargv = malloc(_argc * sizeof(char *)) + with nogil: + ret = rados_conf_parse_argv_remainder(self.cluster, _argc, + _argv, + _remargv) + if ret: + raise make_ex(ret, "error calling conf_parse_argv_remainder") + + # _remargv was allocated with fixed argc; collapse return + # list to eliminate any missing args + retargs = [decode_cstr(a) for a in _remargv[:_argc] + if a != NULL] + self.parsed_args = args + return retargs + finally: + free(_argv) + free(_remargv) + + def conf_parse_env(self, var='CEPH_ARGS'): + """ + Parse known arguments from an environment variable, normally + CEPH_ARGS. + """ + self.require_state("configuring", "connected") + if not var: + return + + var = cstr(var, 'var') + cdef: + char *_var = var + with nogil: + ret = rados_conf_parse_env(self.cluster, _var) + if (ret != 0): + raise make_ex(ret, "error calling conf_parse_env") + + @requires(('option', str_type)) + def conf_get(self, option): + """ + Get the value of a configuration option + + :param option: which option to read + :type option: str + + :returns: str - value of the option or None + :raises: :class:`TypeError` + """ + self.require_state("configuring", "connected") + option = cstr(option, 'option') + cdef: + char *_option = option + size_t length = 20 + char *ret_buf = NULL + + try: + while True: + ret_buf = realloc_chk(ret_buf, length) + with nogil: + ret = rados_conf_get(self.cluster, _option, ret_buf, length) + if (ret == 0): + return decode_cstr(ret_buf) + elif (ret == -errno.ENAMETOOLONG): + length = length * 2 + elif (ret == -errno.ENOENT): + return None + else: + raise make_ex(ret, "error calling conf_get") + finally: + free(ret_buf) + + @requires(('option', str_type), ('val', str_type)) + def conf_set(self, option, val): + """ + Set the value of a configuration option + + :param option: which option to set + :type option: str + :param option: value of the option + :type option: str + + :raises: :class:`TypeError`, :class:`ObjectNotFound` + """ + self.require_state("configuring", "connected") + option = cstr(option, 'option') + val = cstr(val, 'val') + cdef: + char *_option = option + char *_val = val + + with nogil: + ret = rados_conf_set(self.cluster, _option, _val) + if (ret != 0): + raise make_ex(ret, "error calling conf_set") + + def ping_monitor(self, mon_id): + """ + Ping a monitor to assess liveness + + May be used as a simply way to assess liveness, or to obtain + information about the monitor in a simple way even in the + absence of quorum. + + :param mon_id: the ID portion of the monitor's name (i.e., mon.) + :type mon_id: str + :returns: the string reply from the monitor + """ + + self.require_state("configuring", "connected") + + mon_id = cstr(mon_id, 'mon_id') + cdef: + char *_mon_id = mon_id + size_t outstrlen + char *outstr + + with nogil: + ret = rados_ping_monitor(self.cluster, _mon_id, &outstr, &outstrlen) + + if ret != 0: + raise make_ex(ret, "error calling ping_monitor") + + if outstrlen: + my_outstr = outstr[:outstrlen] + rados_buffer_free(outstr) + return decode_cstr(my_outstr) + + def connect(self, timeout=0): + """ + Connect to the cluster. Use shutdown() to release resources. + """ + self.require_state("configuring") + # NOTE(sileht): timeout was supported by old python API, + # but this is not something available in C API, so ignore + # for now and remove it later + with nogil: + ret = rados_connect(self.cluster) + if (ret != 0): + raise make_ex(ret, "error connecting to the cluster") + self.state = "connected" + + def get_cluster_stats(self): + """ + Read usage info about the cluster + + This tells you total space, space used, space available, and number + of objects. These are not updated immediately when data is written, + they are eventually consistent. + + :returns: dict - contains the following keys: + + - ``kb`` (int) - total space + + - ``kb_used`` (int) - space used + + - ``kb_avail`` (int) - free space available + + - ``num_objects`` (int) - number of objects + + """ + cdef: + rados_cluster_stat_t stats + + with nogil: + ret = rados_cluster_stat(self.cluster, &stats) + + if ret < 0: + raise make_ex( + ret, "Rados.get_cluster_stats(%s): get_stats failed" % self.rados_id) + return {'kb': stats.kb, + 'kb_used': stats.kb_used, + 'kb_avail': stats.kb_avail, + 'num_objects': stats.num_objects} + + @requires(('pool_name', str_type)) + def pool_exists(self, pool_name): + """ + Checks if a given pool exists. + + :param pool_name: name of the pool to check + :type pool_name: str + + :raises: :class:`TypeError`, :class:`Error` + :returns: bool - whether the pool exists, false otherwise. + """ + self.require_state("connected") + + pool_name = cstr(pool_name, 'pool_name') + cdef: + char *_pool_name = pool_name + + with nogil: + ret = rados_pool_lookup(self.cluster, _pool_name) + if (ret >= 0): + return True + elif (ret == -errno.ENOENT): + return False + else: + raise make_ex(ret, "error looking up pool '%s'" % pool_name) + + @requires(('pool_name', str_type)) + def pool_lookup(self, pool_name): + """ + Returns a pool's ID based on its name. + + :param pool_name: name of the pool to look up + :type pool_name: str + + :raises: :class:`TypeError`, :class:`Error` + :returns: int - pool ID, or None if it doesn't exist + """ + self.require_state("connected") + pool_name = cstr(pool_name, 'pool_name') + cdef: + char *_pool_name = pool_name + + with nogil: + ret = rados_pool_lookup(self.cluster, _pool_name) + if (ret >= 0): + return int(ret) + elif (ret == -errno.ENOENT): + return None + else: + raise make_ex(ret, "error looking up pool '%s'" % pool_name) + + @requires(('pool_id', int)) + def pool_reverse_lookup(self, pool_id): + """ + Returns a pool's name based on its ID. + + :param pool_id: ID of the pool to look up + :type pool_id: int + + :raises: :class:`TypeError`, :class:`Error` + :returns: string - pool name, or None if it doesn't exist + """ + self.require_state("connected") + cdef: + int64_t _pool_id = pool_id + size_t size = 512 + char *name = NULL + + try: + while True: + name = realloc_chk(name, size) + with nogil: + ret = rados_pool_reverse_lookup(self.cluster, _pool_id, name, size) + if ret >= 0: + break + elif ret != -errno.ERANGE and size <= 4096: + size *= 2 + elif ret == -errno.ENOENT: + return None + elif ret < 0: + raise make_ex(ret, "error reverse looking up pool '%s'" % pool_id) + + return decode_cstr(name) + + finally: + free(name) + + @requires(('pool_name', str_type), ('auid', opt(int)), ('crush_rule', opt(int))) + def create_pool(self, pool_name, auid=None, crush_rule=None): + """ + Create a pool: + - with default settings: if auid=None and crush_rule=None + - owned by a specific auid: auid given and crush_rule=None + - with a specific CRUSH rule: if auid=None and crush_rule given + - with a specific CRUSH rule and auid: if auid and crush_rule given + + :param pool_name: name of the pool to create + :type pool_name: str + :param auid: the id of the owner of the new pool + :type auid: int + :param crush_rule: rule to use for placement in the new pool + :type crush_rule: int + + :raises: :class:`TypeError`, :class:`Error` + """ + self.require_state("connected") + + pool_name = cstr(pool_name, 'pool_name') + cdef: + char *_pool_name = pool_name + uint8_t _crush_rule + uint64_t _auid + + if auid is None and crush_rule is None: + with nogil: + ret = rados_pool_create(self.cluster, _pool_name) + elif auid is None: + _crush_rule = crush_rule + with nogil: + ret = rados_pool_create_with_crush_rule(self.cluster, _pool_name, _crush_rule) + elif crush_rule is None: + _auid = auid + with nogil: + ret = rados_pool_create_with_auid(self.cluster, _pool_name, _auid) + else: + _auid = auid + _crush_rule = crush_rule + with nogil: + ret = rados_pool_create_with_all(self.cluster, _pool_name, _auid, _crush_rule) + if ret < 0: + raise make_ex(ret, "error creating pool '%s'" % pool_name) + + @requires(('pool_id', int)) + def get_pool_base_tier(self, pool_id): + """ + Get base pool + + :returns: base pool, or pool_id if tiering is not configured for the pool + """ + self.require_state("connected") + cdef: + int64_t base_tier = 0 + int64_t _pool_id = pool_id + + with nogil: + ret = rados_pool_get_base_tier(self.cluster, _pool_id, &base_tier) + if ret < 0: + raise make_ex(ret, "get_pool_base_tier(%d)" % pool_id) + return int(base_tier) + + @requires(('pool_name', str_type)) + def delete_pool(self, pool_name): + """ + Delete a pool and all data inside it. + + The pool is removed from the cluster immediately, + but the actual data is deleted in the background. + + :param pool_name: name of the pool to delete + :type pool_name: str + + :raises: :class:`TypeError`, :class:`Error` + """ + self.require_state("connected") + + pool_name = cstr(pool_name, 'pool_name') + cdef: + char *_pool_name = pool_name + + with nogil: + ret = rados_pool_delete(self.cluster, _pool_name) + if ret < 0: + raise make_ex(ret, "error deleting pool '%s'" % pool_name) + + def list_pools(self): + """ + Gets a list of pool names. + + :returns: list - of pool names. + """ + self.require_state("connected") + cdef: + size_t size = 512 + char *c_names = NULL + + try: + while True: + c_names = realloc_chk(c_names, size) + with nogil: + ret = rados_pool_list(self.cluster, c_names, size) + if ret > size: + size *= 2 + elif ret >= 0: + break + return [name for name in decode_cstr(c_names[:ret]).split('\0') + if name] + finally: + free(c_names) + + def get_fsid(self): + """ + Get the fsid of the cluster as a hexadecimal string. + + :raises: :class:`Error` + :returns: str - cluster fsid + """ + self.require_state("connected") + cdef: + char *ret_buf + size_t buf_len = 37 + PyObject* ret_s = NULL + + ret_s = PyBytes_FromStringAndSize(NULL, buf_len) + try: + ret_buf = PyBytes_AsString(ret_s) + with nogil: + ret = rados_cluster_fsid(self.cluster, ret_buf, buf_len) + if ret < 0: + raise make_ex(ret, "error getting cluster fsid") + if ret != buf_len: + _PyBytes_Resize(&ret_s, ret) + return ret_s + finally: + # We DECREF unconditionally: the cast to object above will have + # INCREFed if necessary. This also takes care of exceptions, + # including if _PyString_Resize fails (that will free the string + # itself and set ret_s to NULL, hence XDECREF). + ref.Py_XDECREF(ret_s) + + @requires(('ioctx_name', str_type)) + def open_ioctx(self, ioctx_name): + """ + Create an io context + + The io context allows you to perform operations within a particular + pool. + + :param ioctx_name: name of the pool + :type ioctx_name: str + + :raises: :class:`TypeError`, :class:`Error` + :returns: Ioctx - Rados Ioctx object + """ + self.require_state("connected") + ioctx_name = cstr(ioctx_name, 'ioctx_name') + cdef: + rados_ioctx_t ioctx + char *_ioctx_name = ioctx_name + with nogil: + ret = rados_ioctx_create(self.cluster, _ioctx_name, &ioctx) + if ret < 0: + raise make_ex(ret, "error opening pool '%s'" % ioctx_name) + io = Ioctx(ioctx_name) + io.io = ioctx + return io + + def mon_command(self, cmd, inbuf, timeout=0, target=None): + """ + mon_command[_target](cmd, inbuf, outbuf, outbuflen, outs, outslen) + returns (int ret, string outbuf, string outs) + """ + # NOTE(sileht): timeout is ignored because C API doesn't provide + # timeout argument, but we keep it for backward compat with old python binding + + self.require_state("connected") + cmd = cstr_list(cmd, 'c') + + if isinstance(target, int): + # NOTE(sileht): looks weird but test_monmap_dump pass int + target = str(target) + + target = cstr(target, 'target', opt=True) + inbuf = cstr(inbuf, 'inbuf') + + cdef: + char *_target = opt_str(target) + char **_cmd = to_bytes_array(cmd) + size_t _cmdlen = len(cmd) + + char *_inbuf = inbuf + size_t _inbuf_len = len(inbuf) + + char *_outbuf + size_t _outbuf_len + char *_outs + size_t _outs_len + + try: + if target: + with nogil: + ret = rados_mon_command_target(self.cluster, _target, + _cmd, _cmdlen, + _inbuf, _inbuf_len, + &_outbuf, &_outbuf_len, + &_outs, &_outs_len) + else: + with nogil: + ret = rados_mon_command(self.cluster, + _cmd, _cmdlen, + _inbuf, _inbuf_len, + &_outbuf, &_outbuf_len, + &_outs, &_outs_len) + + my_outs = decode_cstr(_outs[:_outs_len]) + my_outbuf = _outbuf[:_outbuf_len] + if _outs_len: + rados_buffer_free(_outs) + if _outbuf_len: + rados_buffer_free(_outbuf) + return (ret, my_outbuf, my_outs) + finally: + free(_cmd) + + def osd_command(self, osdid, cmd, inbuf, timeout=0): + """ + osd_command(osdid, cmd, inbuf, outbuf, outbuflen, outs, outslen) + returns (int ret, string outbuf, string outs) + """ + # NOTE(sileht): timeout is ignored because C API doesn't provide + # timeout argument, but we keep it for backward compat with old python binding + self.require_state("connected") + + cmd = cstr_list(cmd, 'cmd') + inbuf = cstr(inbuf, 'inbuf') + + cdef: + int _osdid = osdid + char **_cmd = to_bytes_array(cmd) + size_t _cmdlen = len(cmd) + + char *_inbuf = inbuf + size_t _inbuf_len = len(inbuf) + + char *_outbuf + size_t _outbuf_len + char *_outs + size_t _outs_len + + try: + with nogil: + ret = rados_osd_command(self.cluster, _osdid, + _cmd, _cmdlen, + _inbuf, _inbuf_len, + &_outbuf, &_outbuf_len, + &_outs, &_outs_len) + + my_outs = decode_cstr(_outs[:_outs_len]) + my_outbuf = _outbuf[:_outbuf_len] + if _outs_len: + rados_buffer_free(_outs) + if _outbuf_len: + rados_buffer_free(_outbuf) + return (ret, my_outbuf, my_outs) + finally: + free(_cmd) + + def pg_command(self, pgid, cmd, inbuf, timeout=0): + """ + pg_command(pgid, cmd, inbuf, outbuf, outbuflen, outs, outslen) + returns (int ret, string outbuf, string outs) + """ + # NOTE(sileht): timeout is ignored because C API doesn't provide + # timeout argument, but we keep it for backward compat with old python binding + self.require_state("connected") + + pgid = cstr(pgid, 'pgid') + cmd = cstr_list(cmd, 'cmd') + inbuf = cstr(inbuf, 'inbuf') + + cdef: + char *_pgid = pgid + char **_cmd = to_bytes_array(cmd) + size_t _cmdlen = len(cmd) + + char *_inbuf = inbuf + size_t _inbuf_len = len(inbuf) + + char *_outbuf + size_t _outbuf_len + char *_outs + size_t _outs_len + + try: + with nogil: + ret = rados_pg_command(self.cluster, _pgid, + _cmd, _cmdlen, + _inbuf, _inbuf_len, + &_outbuf, &_outbuf_len, + &_outs, &_outs_len) + + my_outs = decode_cstr(_outs[:_outs_len]) + my_outbuf = _outbuf[:_outbuf_len] + if _outs_len: + rados_buffer_free(_outs) + if _outbuf_len: + rados_buffer_free(_outbuf) + return (ret, my_outbuf, my_outs) + finally: + free(_cmd) + + def wait_for_latest_osdmap(self): + self.require_state("connected") + with nogil: + ret = rados_wait_for_latest_osdmap(self.cluster) + return ret + + def blacklist_add(self, client_address, expire_seconds=0): + """ + Blacklist a client from the OSDs + + :param client_address: client address + :type client_address: str + :param expire_seconds: number of seconds to blacklist + :type expire_seconds: int + + :raises: :class:`Error` + """ + self.require_state("connected") + client_address = cstr(client_address, 'client_address') + cdef: + uint32_t _expire_seconds = expire_seconds + char *_client_address = client_address + + with nogil: + ret = rados_blacklist_add(self.cluster, _client_address, _expire_seconds) + if ret < 0: + raise make_ex(ret, "error blacklisting client '%s'" % client_address) + + def monitor_log(self, level, callback, arg): + if level not in MONITOR_LEVELS: + raise LogicError("invalid monitor level " + level) + if callback is not None and not callable(callback): + raise LogicError("callback must be a callable function or None") + + level = cstr(level, 'level') + cdef char *_level = level + + if callback is None: + with nogil: + r = rados_monitor_log(self.cluster, _level, NULL, NULL) + self.monitor_callback = None + return + + cb = (callback, arg) + cdef PyObject* _arg = cb + with nogil: + r = rados_monitor_log(self.cluster, _level, + &__monitor_callback, _arg) + + if r: + raise make_ex(r, 'error calling rados_monitor_log') + # NOTE(sileht): Prevents the callback method from being garbage collected + self.monitor_callback = cb + + +cdef class OmapIterator(object): + """Omap iterator""" + + cdef public Ioctx ioctx + cdef rados_omap_iter_t ctx + + def __init__(self, Ioctx ioctx): + self.ioctx = ioctx + + def __iter__(self): + return self + + def next(self): + return self.__next__() + + def __next__(self): + """ + Get the next key-value pair in the object + :returns: next rados.OmapItem + """ + cdef: + char *key_ = NULL + char *val_ = NULL + size_t len_ + + with nogil: + ret = rados_omap_get_next(self.ctx, &key_, &val_, &len_) + + if (ret != 0): + raise make_ex(ret, "error iterating over the omap") + if key_ == NULL: + raise StopIteration() + key = decode_cstr(key_) + val = None + if val_ != NULL: + val = val_[:len_] + return (key, val) + + def __del__(self): + with nogil: + rados_omap_get_end(self.ctx) + + +cdef class ObjectIterator(object): + """rados.Ioctx Object iterator""" + + cdef rados_list_ctx_t ctx + + cdef public object ioctx + + def __cinit__(self, Ioctx ioctx): + self.ioctx = ioctx + + with nogil: + ret = rados_nobjects_list_open(ioctx.io, &self.ctx) + if ret < 0: + raise make_ex(ret, "error iterating over the objects in ioctx '%s'" + % self.ioctx.name) + + def __iter__(self): + return self + + def next(self): + return self.__next__() + + def __next__(self): + """ + Get the next object name and locator in the pool + + :raises: StopIteration + :returns: next rados.Ioctx Object + """ + cdef: + const char *key_ = NULL + const char *locator_ = NULL + const char *nspace_ = NULL + + with nogil: + ret = rados_nobjects_list_next(self.ctx, &key_, &locator_, &nspace_) + + if ret < 0: + raise StopIteration() + + key = decode_cstr(key_) + locator = decode_cstr(locator_) if locator_ != NULL else None + nspace = decode_cstr(nspace_) if nspace_ != NULL else None + return Object(self.ioctx, key, locator, nspace) + + def __del__(self): + with nogil: + rados_nobjects_list_close(self.ctx) + + +cdef class XattrIterator(object): + """Extended attribute iterator""" + + cdef rados_xattrs_iter_t it + cdef char* _oid + + cdef public Ioctx ioctx + cdef public object oid + + def __cinit__(self, Ioctx ioctx, oid): + self.ioctx = ioctx + self.oid = cstr(oid, 'oid') + self._oid = self.oid + + with nogil: + ret = rados_getxattrs(ioctx.io, self._oid, &self.it) + if ret != 0: + raise make_ex(ret, "Failed to get rados xattrs for object %r" % oid) + + def __iter__(self): + return self + + def next(self): + return self.__next__() + + def __next__(self): + """ + Get the next xattr on the object + + :raises: StopIteration + :returns: pair - of name and value of the next Xattr + """ + cdef: + const char *name_ = NULL + const char *val_ = NULL + size_t len_ = 0 + + with nogil: + ret = rados_getxattrs_next(self.it, &name_, &val_, &len_) + if (ret != 0): + raise make_ex(ret, "error iterating over the extended attributes \ +in '%s'" % self.oid) + if name_ == NULL: + raise StopIteration() + name = decode_cstr(name_) + val = val_[:len_] + return (name, val) + + def __del__(self): + with nogil: + rados_getxattrs_end(self.it) + + +cdef class SnapIterator(object): + """Snapshot iterator""" + + cdef public Ioctx ioctx + + cdef rados_snap_t *snaps + cdef int max_snap + cdef int cur_snap + + def __cinit__(self, Ioctx ioctx): + self.ioctx = ioctx + # We don't know how big a buffer we need until we've called the + # function. So use the exponential doubling strategy. + cdef int num_snaps = 10 + while True: + self.snaps = realloc_chk(self.snaps, + num_snaps * + sizeof(rados_snap_t)) + + with nogil: + ret = rados_ioctx_snap_list(ioctx.io, self.snaps, num_snaps) + if (ret >= 0): + self.max_snap = ret + break + elif (ret != -errno.ERANGE): + raise make_ex(ret, "error calling rados_snap_list for \ +ioctx '%s'" % self.ioctx.name) + num_snaps = num_snaps * 2 + self.cur_snap = 0 + + def __iter__(self): + return self + + def next(self): + return self.__next__() + + def __next__(self): + """ + Get the next Snapshot + + :raises: :class:`Error`, StopIteration + :returns: Snap - next snapshot + """ + if (self.cur_snap >= self.max_snap): + raise StopIteration + + cdef: + rados_snap_t snap_id = self.snaps[self.cur_snap] + int name_len = 10 + char *name = NULL + + try: + while True: + name = realloc_chk(name, name_len) + with nogil: + ret = rados_ioctx_snap_get_name(self.ioctx.io, snap_id, name, name_len) + if (ret == 0): + break + elif (ret != -errno.ERANGE): + raise make_ex(ret, "rados_snap_get_name error") + else: + name_len = name_len * 2 + + snap = Snap(self.ioctx, decode_cstr(name[:name_len]).rstrip('\0'), snap_id) + self.cur_snap = self.cur_snap + 1 + return snap + finally: + free(name) + + +cdef class Snap(object): + """Snapshot object""" + cdef public Ioctx ioctx + cdef public object name + + # NOTE(sileht): old API was storing the ctypes object + # instead of the value .... + cdef public rados_snap_t snap_id + + def __cinit__(self, Ioctx ioctx, object name, rados_snap_t snap_id): + self.ioctx = ioctx + self.name = name + self.snap_id = snap_id + + def __str__(self): + return "rados.Snap(ioctx=%s,name=%s,snap_id=%d)" \ + % (str(self.ioctx), self.name, self.snap_id) + + def get_timestamp(self): + """ + Find when a snapshot in the current pool occurred + + :raises: :class:`Error` + :returns: datetime - the data and time the snapshot was created + """ + cdef time_t snap_time + + with nogil: + ret = rados_ioctx_snap_get_stamp(self.ioctx.io, self.snap_id, &snap_time) + if (ret != 0): + raise make_ex(ret, "rados_ioctx_snap_get_stamp error") + return datetime.fromtimestamp(snap_time) + + +cdef class Completion(object): + """completion object""" + + cdef public: + Ioctx ioctx + object oncomplete + object onsafe + + cdef: + rados_callback_t complete_cb + rados_callback_t safe_cb + rados_completion_t rados_comp + + def __cinit__(self, Ioctx ioctx, object oncomplete, object onsafe): + self.oncomplete = oncomplete + self.onsafe = onsafe + self.ioctx = ioctx + + def is_safe(self): + """ + Is an asynchronous operation safe? + + This does not imply that the safe callback has finished. + + :returns: True if the operation is safe + """ + with nogil: + ret = rados_aio_is_safe(self.rados_comp) + return ret == 1 + + def is_complete(self): + """ + Has an asynchronous operation completed? + + This does not imply that the safe callback has finished. + + :returns: True if the operation is completed + """ + with nogil: + ret = rados_aio_is_complete(self.rados_comp) + return ret == 1 + + def wait_for_safe(self): + """ + Wait for an asynchronous operation to be marked safe + + This does not imply that the safe callback has finished. + """ + with nogil: + rados_aio_wait_for_safe(self.rados_comp) + + def wait_for_complete(self): + """ + Wait for an asynchronous operation to complete + + This does not imply that the complete callback has finished. + """ + with nogil: + rados_aio_wait_for_complete(self.rados_comp) + + def wait_for_safe_and_cb(self): + """ + Wait for an asynchronous operation to be marked safe and for + the safe callback to have returned + """ + with nogil: + rados_aio_wait_for_safe_and_cb(self.rados_comp) + + def wait_for_complete_and_cb(self): + """ + Wait for an asynchronous operation to complete and for the + complete callback to have returned + + :returns: whether the operation is completed + """ + with nogil: + ret = rados_aio_wait_for_complete_and_cb(self.rados_comp) + return ret + + def get_return_value(self): + """ + Get the return value of an asychronous operation + + The return value is set when the operation is complete or safe, + whichever comes first. + + :returns: int - return value of the operation + """ + with nogil: + ret = rados_aio_get_return_value(self.rados_comp) + return ret + + def __del__(self): + """ + Release a completion + + Call this when you no longer need the completion. It may not be + freed immediately if the operation is not acked and committed. + """ + with nogil: + rados_aio_release(self.rados_comp) + + def _complete(self): + self.oncomplete(self) + with self.ioctx.lock: + if self.oncomplete: + self.ioctx.complete_completions.remove(self) + + def _safe(self): + self.onsafe(self) + with self.ioctx.lock: + if self.onsafe: + self.ioctx.safe_completions.remove(self) + + def _cleanup(self): + with self.ioctx.lock: + if self.oncomplete: + self.ioctx.complete_completions.remove(self) + if self.onsafe: + self.ioctx.safe_completions.remove(self) + + +class OpCtx(object): + def __enter__(self): + return self.create() + + def __exit__(self, type, msg, traceback): + self.release() + + +cdef class WriteOp(object): + cdef rados_write_op_t write_op + + def create(self): + with nogil: + self.write_op = rados_create_write_op() + return self + + def release(self): + with nogil: + rados_release_write_op(self.write_op) + + +class WriteOpCtx(WriteOp, OpCtx): + """write operation context manager""" + + +cdef class ReadOp(object): + cdef rados_read_op_t read_op + + def create(self): + with nogil: + self.read_op = rados_create_read_op() + return self + + def release(self): + with nogil: + rados_release_read_op(self.read_op) + + +class ReadOpCtx(ReadOp, OpCtx): + """read operation context manager""" + + +cdef int __aio_safe_cb(rados_completion_t completion, void *args) with gil: + """ + Callback to onsafe() for asynchronous operations + """ + cdef object cb = args + cb.onsafe(cb) + return 0 + + +cdef int __aio_complete_cb(rados_completion_t completion, void *args) with gil: + """ + Callback to oncomplete() for asynchronous operations + """ + cdef object cb = args + cb.oncomplete(cb) + return 0 + + +cdef class Ioctx(object): + """rados.Ioctx object""" + # NOTE(sileht): attributes declared in .pyd + + def __init__(self, name): + self.name = name + self.state = "open" + + self.locator_key = "" + self.nspace = "" + self.lock = threading.Lock() + self.safe_completions = [] + self.complete_completions = [] + + def __enter__(self): + return self + + def __exit__(self, type_, value, traceback): + self.close() + return False + + def __del__(self): + self.close() + + def __track_completion(self, completion_obj): + if completion_obj.oncomplete: + with self.lock: + self.complete_completions.append(completion_obj) + if completion_obj.onsafe: + with self.lock: + self.safe_completions.append(completion_obj) + + def __get_completion(self, oncomplete, onsafe): + """ + Constructs a completion to use with asynchronous operations + + :param oncomplete: what to do when the write is safe and complete in memory + on all replicas + :type oncomplete: completion + :param onsafe: what to do when the write is safe and complete on storage + on all replicas + :type onsafe: completion + + :raises: :class:`Error` + :returns: completion object + """ + + completion_obj = Completion(self, oncomplete, onsafe) + + cdef: + rados_callback_t complete_cb = NULL + rados_callback_t safe_cb = NULL + rados_completion_t completion + PyObject* p_completion_obj= completion_obj + + if oncomplete: + complete_cb = &__aio_complete_cb + if onsafe: + safe_cb = &__aio_safe_cb + + with nogil: + ret = rados_aio_create_completion(p_completion_obj, complete_cb, safe_cb, + &completion) + if ret < 0: + raise make_ex(ret, "error getting a completion") + + completion_obj.rados_comp = completion + return completion_obj + + def aio_write(self, object_name, to_write, offset=0, + oncomplete=None, onsafe=None): + """ + Write data to an object asynchronously + + Queues the write and returns. + + :param object_name: name of the object + :type object_name: str + :param to_write: data to write + :type to_write: bytes + :param offset: byte offset in the object to begin writing at + :type offset: int + :param oncomplete: what to do when the write is safe and complete in memory + on all replicas + :type oncomplete: completion + :param onsafe: what to do when the write is safe and complete on storage + on all replicas + :type onsafe: completion + + :raises: :class:`Error` + :returns: completion object + """ + + object_name = cstr(object_name, 'object_name') + + cdef: + Completion completion + char* _object_name = object_name + char* _to_write = to_write + size_t size = len(to_write) + uint64_t _offset = offset + + completion = self.__get_completion(oncomplete, onsafe) + + with nogil: + ret = rados_aio_write(self.io, _object_name, completion.rados_comp, + _to_write, size, _offset) + if ret < 0: + raise make_ex(ret, "error writing object %s" % object_name) + self.__track_completion(completion) + return completion + + def aio_write_full(self, object_name, to_write, + oncomplete=None, onsafe=None): + """ + Asychronously write an entire object + + The object is filled with the provided data. If the object exists, + it is atomically truncated and then written. + Queues the write and returns. + + :param object_name: name of the object + :type object_name: str + :param to_write: data to write + :type to_write: str + :param oncomplete: what to do when the write is safe and complete in memory + on all replicas + :type oncomplete: completion + :param onsafe: what to do when the write is safe and complete on storage + on all replicas + :type onsafe: completion + + :raises: :class:`Error` + :returns: completion object + """ + + object_name = cstr(object_name, 'object_name') + + cdef: + Completion completion + char* _object_name = object_name + char* _to_write = to_write + size_t size = len(to_write) + + completion = self.__get_completion(oncomplete, onsafe) + with nogil: + ret = rados_aio_write_full(self.io, _object_name, + completion.rados_comp, + _to_write, size) + if ret < 0: + raise make_ex(ret, "error writing object %s" % object_name) + self.__track_completion(completion) + return completion + + def aio_append(self, object_name, to_append, oncomplete=None, onsafe=None): + """ + Asychronously append data to an object + + Queues the write and returns. + + :param object_name: name of the object + :type object_name: str + :param to_append: data to append + :type to_append: str + :param offset: byte offset in the object to begin writing at + :type offset: int + :param oncomplete: what to do when the write is safe and complete in memory + on all replicas + :type oncomplete: completion + :param onsafe: what to do when the write is safe and complete on storage + on all replicas + :type onsafe: completion + + :raises: :class:`Error` + :returns: completion object + """ + object_name = cstr(object_name, 'object_name') + + cdef: + Completion completion + char* _object_name = object_name + char* _to_append = to_append + size_t size = len(to_append) + + completion = self.__get_completion(oncomplete, onsafe) + with nogil: + ret = rados_aio_append(self.io, _object_name, + completion.rados_comp, + _to_append, size) + if ret < 0: + raise make_ex(ret, "error appending object %s" % object_name) + self.__track_completion(completion) + return completion + + def aio_flush(self): + """ + Block until all pending writes in an io context are safe + + :raises: :class:`Error` + """ + with nogil: + ret = rados_aio_flush(self.io) + if ret < 0: + raise make_ex(ret, "error flushing") + + def aio_read(self, object_name, length, offset, oncomplete): + """ + Asychronously read data from an object + + oncomplete will be called with the returned read value as + well as the completion: + + oncomplete(completion, data_read) + + :param object_name: name of the object to read from + :type object_name: str + :param length: the number of bytes to read + :type length: int + :param offset: byte offset in the object to begin reading from + :type offset: int + :param oncomplete: what to do when the read is complete + :type oncomplete: completion + + :raises: :class:`Error` + :returns: completion object + """ + + object_name = cstr(object_name, 'object_name') + + cdef: + Completion completion + char* _object_name = object_name + uint64_t _offset = offset + + char *ref_buf + size_t _length = length + PyObject* ret_s = NULL + + def oncomplete_(completion_v): + try: + return_value = completion_v.get_return_value() + if return_value != length: + _PyBytes_Resize(&ret_s, return_value) + return oncomplete(completion_v, ret_s if return_value >= 0 else None) + finally: + # We DECREF unconditionally: the cast to object above will have + # INCREFed if necessary. This also takes care of exceptions, + # including if _PyString_Resize fails (that will free the string + # itself and set ret_s to NULL, hence XDECREF). + ref.Py_XDECREF(ret_s) + + completion = self.__get_completion(oncomplete_, None) + ret_s = PyBytes_FromStringAndSize(NULL, length) + try: + ret_buf = PyBytes_AsString(ret_s) + with nogil: + ret = rados_aio_read(self.io, _object_name, completion.rados_comp, + ret_buf, _length, _offset) + if ret < 0: + raise make_ex(ret, "error reading %s" % object_name) + self.__track_completion(completion) + return completion + except Exception: + ref.Py_XDECREF(ret_s) + + def aio_remove(self, object_name, oncomplete=None, onsafe=None): + """ + Asychronously remove an object + + :param object_name: name of the object to remove + :type object_name: str + :param oncomplete: what to do when the remove is safe and complete in memory + on all replicas + :type oncomplete: completion + :param onsafe: what to do when the remove is safe and complete on storage + on all replicas + :type onsafe: completion + + :raises: :class:`Error` + :returns: completion object + """ + object_name = cstr(object_name, 'object_name') + + cdef: + Completion completion + char* _object_name = object_name + + completion = self.__get_completion(oncomplete, onsafe) + with nogil: + ret = rados_aio_remove(self.io, _object_name, + completion.rados_comp) + if ret < 0: + raise make_ex(ret, "error removing %s" % object_name) + self.__track_completion(completion) + return completion + + def require_ioctx_open(self): + """ + Checks if the rados.Ioctx object state is 'open' + + :raises: IoctxStateError + """ + if self.state != b"open": + raise IoctxStateError("The pool is %s" % self.state) + + def change_auid(self, auid): + """ + Attempt to change an io context's associated auid "owner." + + Requires that you have write permission on both the current and new + auid. + + :raises: :class:`Error` + """ + self.require_ioctx_open() + + cdef: + uint64_t _auid = auid + + with nogil: + ret = rados_ioctx_pool_set_auid(self.io, _auid) + if ret < 0: + raise make_ex(ret, "error changing auid of '%s' to %d" + % (self.name, auid)) + + @requires(('loc_key', str_type)) + def set_locator_key(self, loc_key): + """ + Set the key for mapping objects to pgs within an io context. + + The key is used instead of the object name to determine which + placement groups an object is put in. This affects all subsequent + operations of the io context - until a different locator key is + set, all objects in this io context will be placed in the same pg. + + :param loc_key: the key to use as the object locator, or NULL to discard + any previously set key + :type loc_key: str + + :raises: :class:`TypeError` + """ + self.require_ioctx_open() + cloc_key = cstr(loc_key, 'loc_key') + cdef char *_loc_key = cloc_key + with nogil: + rados_ioctx_locator_set_key(self.io, _loc_key) + self.locator_key = loc_key + + def get_locator_key(self): + """ + Get the locator_key of context + + :returns: locator_key + """ + return self.locator_key + + @requires(('snap_id', long)) + def set_read(self, snap_id): + """ + Set the snapshot for reading objects. + + To stop to read from snapshot, use set_read(LIBRADOS_SNAP_HEAD) + + :param snap_id: the snapshot Id + :type snap_id: int + + :raises: :class:`TypeError` + """ + self.require_ioctx_open() + cdef rados_snap_t _snap_id = snap_id + with nogil: + rados_ioctx_snap_set_read(self.io, _snap_id) + + @requires(('nspace', str_type)) + def set_namespace(self, nspace): + """ + Set the namespace for objects within an io context. + + The namespace in addition to the object name fully identifies + an object. This affects all subsequent operations of the io context + - until a different namespace is set, all objects in this io context + will be placed in the same namespace. + + :param nspace: the namespace to use, or None/"" for the default namespace + :type nspace: str + + :raises: :class:`TypeError` + """ + self.require_ioctx_open() + if nspace is None: + nspace = "" + cnspace = cstr(nspace, 'nspace') + cdef char *_nspace = cnspace + with nogil: + rados_ioctx_set_namespace(self.io, _nspace) + self.nspace = nspace + + def get_namespace(self): + """ + Get the namespace of context + + :returns: namespace + """ + return self.nspace + + def close(self): + """ + Close a rados.Ioctx object. + + This just tells librados that you no longer need to use the io context. + It may not be freed immediately if there are pending asynchronous + requests on it, but you should not use an io context again after + calling this function on it. + """ + if self.state == b"open": + self.require_ioctx_open() + with nogil: + rados_ioctx_destroy(self.io) + self.state = "closed" + + + @requires(('key', str_type), ('data', bytes)) + def write(self, key, data, offset=0): + """ + Write data to an object synchronously + + :param key: name of the object + :type key: str + :param data: data to write + :type data: bytes + :param offset: byte offset in the object to begin writing at + :type offset: int + + :raises: :class:`TypeError` + :raises: :class:`LogicError` + :returns: int - 0 on success + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + cdef: + char *_key = key + char *_data = data + size_t length = len(data) + uint64_t _offset = offset + + with nogil: + ret = rados_write(self.io, _key, _data, length, _offset) + if ret == 0: + return ret + elif ret < 0: + raise make_ex(ret, "Ioctx.write(%s): failed to write %s" + % (self.name, key)) + else: + raise LogicError("Ioctx.write(%s): rados_write \ +returned %d, but should return zero on success." % (self.name, ret)) + + @requires(('key', str_type), ('data', bytes)) + def write_full(self, key, data): + """ + Write an entire object synchronously. + + The object is filled with the provided data. If the object exists, + it is atomically truncated and then written. + + :param key: name of the object + :type key: str + :param data: data to write + :type data: bytes + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: int - 0 on success + """ + self.require_ioctx_open() + key = cstr(key, 'key') + cdef: + char *_key = key + char *_data = data + size_t length = len(data) + + with nogil: + ret = rados_write_full(self.io, _key, _data, length) + if ret == 0: + return ret + elif ret < 0: + raise make_ex(ret, "Ioctx.write_full(%s): failed to write %s" + % (self.name, key)) + else: + raise LogicError("Ioctx.write_full(%s): rados_write_full \ +returned %d, but should return zero on success." % (self.name, ret)) + + @requires(('key', str_type), ('data', bytes)) + def append(self, key, data): + """ + Append data to an object synchronously + + :param key: name of the object + :type key: str + :param data: data to write + :type data: bytes + + :raises: :class:`TypeError` + :raises: :class:`LogicError` + :returns: int - 0 on success + """ + self.require_ioctx_open() + key = cstr(key, 'key') + cdef: + char *_key = key + char *_data = data + size_t length = len(data) + + with nogil: + ret = rados_append(self.io, _key, _data, length) + if ret == 0: + return ret + elif ret < 0: + raise make_ex(ret, "Ioctx.append(%s): failed to append %s" + % (self.name, key)) + else: + raise LogicError("Ioctx.append(%s): rados_append \ +returned %d, but should return zero on success." % (self.name, ret)) + + @requires(('key', str_type)) + def read(self, key, length=8192, offset=0): + """ + Read data from an object synchronously + + :param key: name of the object + :type key: str + :param length: the number of bytes to read (default=8192) + :type length: int + :param offset: byte offset in the object to begin reading at + :type offset: int + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: str - data read from object + """ + self.require_ioctx_open() + key = cstr(key, 'key') + cdef: + char *_key = key + char *ret_buf + uint64_t _offset = offset + size_t _length = length + PyObject* ret_s = NULL + + ret_s = PyBytes_FromStringAndSize(NULL, length) + try: + ret_buf = PyBytes_AsString(ret_s) + with nogil: + ret = rados_read(self.io, _key, ret_buf, _length, _offset) + if ret < 0: + raise make_ex(ret, "Ioctx.read(%s): failed to read %s" % (self.name, key)) + + if ret != length: + _PyBytes_Resize(&ret_s, ret) + + return ret_s + finally: + # We DECREF unconditionally: the cast to object above will have + # INCREFed if necessary. This also takes care of exceptions, + # including if _PyString_Resize fails (that will free the string + # itself and set ret_s to NULL, hence XDECREF). + ref.Py_XDECREF(ret_s) + + @requires(('key', str_type), ('cls', str_type), ('method', str_type), ('data', bytes)) + def execute(self, key, cls, method, data, length=8192): + """ + Execute an OSD class method on an object. + + :param key: name of the object + :type key: str + :param cls: name of the object class + :type cls: str + :param method: name of the method + :type method: str + :param data: input data + :type data: bytes + :param length: size of output buffer in bytes (default=8291) + :type length: int + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: (ret, method output) + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + cls = cstr(cls, 'cls') + method = cstr(method, 'method') + cdef: + char *_key = key + char *_cls = cls + char *_method = method + char *_data = data + size_t _data_len = len(data) + + char *ref_buf + size_t _length = length + PyObject* ret_s = NULL + + ret_s = PyBytes_FromStringAndSize(NULL, length) + try: + ret_buf = PyBytes_AsString(ret_s) + with nogil: + ret = rados_exec(self.io, _key, _cls, _method, _data, + _data_len, ret_buf, _length) + if ret < 0: + raise make_ex(ret, "Ioctx.read(%s): failed to read %s" % (self.name, key)) + + if ret != length: + _PyBytes_Resize(&ret_s, ret) + + return ret, ret_s + finally: + # We DECREF unconditionally: the cast to object above will have + # INCREFed if necessary. This also takes care of exceptions, + # including if _PyString_Resize fails (that will free the string + # itself and set ret_s to NULL, hence XDECREF). + ref.Py_XDECREF(ret_s) + + def get_stats(self): + """ + Get pool usage statistics + + :returns: dict - contains the following keys: + + - ``num_bytes`` (int) - size of pool in bytes + + - ``num_kb`` (int) - size of pool in kbytes + + - ``num_objects`` (int) - number of objects in the pool + + - ``num_object_clones`` (int) - number of object clones + + - ``num_object_copies`` (int) - number of object copies + + - ``num_objects_missing_on_primary`` (int) - number of objets + missing on primary + + - ``num_objects_unfound`` (int) - number of unfound objects + + - ``num_objects_degraded`` (int) - number of degraded objects + + - ``num_rd`` (int) - bytes read + + - ``num_rd_kb`` (int) - kbytes read + + - ``num_wr`` (int) - bytes written + + - ``num_wr_kb`` (int) - kbytes written + """ + self.require_ioctx_open() + cdef rados_pool_stat_t stats + with nogil: + ret = rados_ioctx_pool_stat(self.io, &stats) + if ret < 0: + raise make_ex(ret, "Ioctx.get_stats(%s): get_stats failed" % self.name) + return {'num_bytes': stats.num_bytes, + 'num_kb': stats.num_kb, + 'num_objects': stats.num_objects, + 'num_object_clones': stats.num_object_clones, + 'num_object_copies': stats.num_object_copies, + "num_objects_missing_on_primary": stats.num_objects_missing_on_primary, + "num_objects_unfound": stats.num_objects_unfound, + "num_objects_degraded": stats.num_objects_degraded, + "num_rd": stats.num_rd, + "num_rd_kb": stats.num_rd_kb, + "num_wr": stats.num_wr, + "num_wr_kb": stats.num_wr_kb} + + @requires(('key', str_type)) + def remove_object(self, key): + """ + Delete an object + + This does not delete any snapshots of the object. + + :param key: the name of the object to delete + :type key: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: bool - True on success + """ + self.require_ioctx_open() + key = cstr(key, 'key') + cdef: + char *_key = key + + with nogil: + ret = rados_remove(self.io, _key) + if ret < 0: + raise make_ex(ret, "Failed to remove '%s'" % key) + return True + + @requires(('key', str_type)) + def trunc(self, key, size): + """ + Resize an object + + If this enlarges the object, the new area is logically filled with + zeroes. If this shrinks the object, the excess data is removed. + + :param key: the name of the object to resize + :type key: str + :param size: the new size of the object in bytes + :type size: int + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: int - 0 on success, otherwise raises error + """ + + self.require_ioctx_open() + key = cstr(key, 'key') + cdef: + char *_key = key + uint64_t _size = size + + with nogil: + ret = rados_trunc(self.io, _key, _size) + if ret < 0: + raise make_ex(ret, "Ioctx.trunc(%s): failed to truncate %s" % (self.name, key)) + return ret + + @requires(('key', str_type)) + def stat(self, key): + """ + Get object stats (size/mtime) + + :param key: the name of the object to get stats from + :type key: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: (size,timestamp) + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + cdef: + char *_key = key + uint64_t psize + time_t pmtime + + with nogil: + ret = rados_stat(self.io, _key, &psize, &pmtime) + if ret < 0: + raise make_ex(ret, "Failed to stat %r" % key) + return psize, time.localtime(pmtime) + + @requires(('key', str_type), ('xattr_name', str_type)) + def get_xattr(self, key, xattr_name): + """ + Get the value of an extended attribute on an object. + + :param key: the name of the object to get xattr from + :type key: str + :param xattr_name: which extended attribute to read + :type xattr_name: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: str - value of the xattr + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + xattr_name = cstr(xattr_name, 'xattr_name') + cdef: + char *_key = key + char *_xattr_name = xattr_name + size_t ret_length = 4096 + char *ret_buf = NULL + + try: + while ret_length < 4096 * 1024 * 1024: + ret_buf = realloc_chk(ret_buf, ret_length) + with nogil: + ret = rados_getxattr(self.io, _key, _xattr_name, ret_buf, ret_length) + if (ret == -errno.ERANGE): + ret_length *= 2 + elif ret < 0: + raise make_ex(ret, "Failed to get xattr %r" % xattr_name) + else: + break + return ret_buf[:ret] + finally: + free(ret_buf) + + @requires(('oid', str_type)) + def get_xattrs(self, oid): + """ + Start iterating over xattrs on an object. + + :param oid: the name of the object to get xattrs from + :type oid: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: XattrIterator + """ + self.require_ioctx_open() + return XattrIterator(self, oid) + + @requires(('key', str_type), ('xattr_name', str_type), ('xattr_value', bytes)) + def set_xattr(self, key, xattr_name, xattr_value): + """ + Set an extended attribute on an object. + + :param key: the name of the object to set xattr to + :type key: str + :param xattr_name: which extended attribute to set + :type xattr_name: str + :param xattr_value: the value of the extended attribute + :type xattr_value: bytes + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: bool - True on success, otherwise raise an error + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + xattr_name = cstr(xattr_name, 'xattr_name') + cdef: + char *_key = key + char *_xattr_name = xattr_name + char *_xattr_value = xattr_value + size_t _xattr_value_len = len(xattr_value) + + with nogil: + ret = rados_setxattr(self.io, _key, _xattr_name, + _xattr_value, _xattr_value_len) + if ret < 0: + raise make_ex(ret, "Failed to set xattr %r" % xattr_name) + return True + + @requires(('key', str_type), ('xattr_name', str_type)) + def rm_xattr(self, key, xattr_name): + """ + Removes an extended attribute on from an object. + + :param key: the name of the object to remove xattr from + :type key: str + :param xattr_name: which extended attribute to remove + :type xattr_name: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: bool - True on success, otherwise raise an error + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + xattr_name = cstr(xattr_name, 'xattr_name') + cdef: + char *_key = key + char *_xattr_name = xattr_name + + with nogil: + ret = rados_rmxattr(self.io, _key, _xattr_name) + if ret < 0: + raise make_ex(ret, "Failed to delete key %r xattr %r" % + (key, xattr_name)) + return True + + def list_objects(self): + """ + Get ObjectIterator on rados.Ioctx object. + + :returns: ObjectIterator + """ + self.require_ioctx_open() + return ObjectIterator(self) + + def list_snaps(self): + """ + Get SnapIterator on rados.Ioctx object. + + :returns: SnapIterator + """ + self.require_ioctx_open() + return SnapIterator(self) + + @requires(('snap_name', str_type)) + def create_snap(self, snap_name): + """ + Create a pool-wide snapshot + + :param snap_name: the name of the snapshot + :type snap_name: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + """ + self.require_ioctx_open() + snap_name = cstr(snap_name, 'snap_name') + cdef char *_snap_name = snap_name + + with nogil: + ret = rados_ioctx_snap_create(self.io, _snap_name) + if (ret != 0): + raise make_ex(ret, "Failed to create snap %s" % snap_name) + + @requires(('snap_name', str_type)) + def remove_snap(self, snap_name): + """ + Removes a pool-wide snapshot + + :param snap_name: the name of the snapshot + :type snap_name: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + """ + self.require_ioctx_open() + snap_name = cstr(snap_name, 'snap_name') + cdef char *_snap_name = snap_name + + with nogil: + ret = rados_ioctx_snap_remove(self.io, _snap_name) + if (ret != 0): + raise make_ex(ret, "Failed to remove snap %s" % snap_name) + + @requires(('snap_name', str_type)) + def lookup_snap(self, snap_name): + """ + Get the id of a pool snapshot + + :param snap_name: the name of the snapshot to lookop + :type snap_name: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + :returns: Snap - on success + """ + self.require_ioctx_open() + csnap_name = cstr(snap_name, 'snap_name') + cdef: + char *_snap_name = csnap_name + rados_snap_t snap_id + + with nogil: + ret = rados_ioctx_snap_lookup(self.io, _snap_name, &snap_id) + if (ret != 0): + raise make_ex(ret, "Failed to lookup snap %s" % snap_name) + return Snap(self, snap_name, int(snap_id)) + + @requires(('oid', str_type), ('snap_name', str_type)) + def snap_rollback(self, oid, snap_name): + """ + Rollback an object to a snapshot + + :param oid: the name of the object + :type oid: str + :param snap_name: the name of the snapshot + :type snap_name: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + """ + self.require_ioctx_open() + oid = cstr(oid, 'oid') + snap_name = cstr(snap_name, 'snap_name') + cdef: + char *_snap_name = snap_name + char *_oid = oid + + with nogil: + ret = rados_ioctx_snap_rollback(self.io, _oid, _snap_name) + if (ret != 0): + raise make_ex(ret, "Failed to rollback %s" % oid) + + def get_last_version(self): + """ + Return the version of the last object read or written to. + + This exposes the internal version number of the last object read or + written via this io context + + :returns: version of the last object used + """ + self.require_ioctx_open() + with nogil: + ret = rados_get_last_version(self.io) + return int(ret) + + def create_write_op(self): + """ + create write operation object. + need call release_write_op after use + """ + return WriteOp().create() + + def create_read_op(self): + """ + create read operation object. + need call release_read_op after use + """ + return ReadOp().create() + + def release_write_op(self, write_op): + """ + release memory alloc by create_write_op + """ + write_op.release() + + def release_read_op(self, read_op): + """ + release memory alloc by create_read_op + :para read_op: read_op object + :type: int + """ + read_op.release() + + @requires(('write_op', WriteOp), ('keys', tuple), ('values', tuple)) + def set_omap(self, write_op, keys, values): + """ + set keys values to write_op + :para write_op: write_operation object + :type write_op: WriteOp + :para keys: a tuple of keys + :type keys: tuple + :para values: a tuple of values + :type values: tuple + """ + + if len(keys) != len(values): + raise Error("Rados(): keys and values must have the same number of items") + + keys = cstr_list(keys, 'keys') + cdef: + WriteOp _write_op = write_op + size_t key_num = len(keys) + char **_keys = to_bytes_array(keys) + char **_values = to_bytes_array(values) + size_t *_lens = to_csize_t_array([len(v) for v in values]) + + try: + with nogil: + rados_write_op_omap_set(_write_op.write_op, + _keys, + _values, + _lens, key_num) + finally: + free(_keys) + free(_values) + free(_lens) + + @requires(('write_op', WriteOp), ('oid', str_type), ('mtime', opt(int)), ('flags', opt(int))) + def operate_write_op(self, write_op, oid, mtime=0, flags=LIBRADOS_OPERATION_NOFLAG): + """ + excute the real write operation + :para write_op: write operation object + :type write_op: WriteOp + :para oid: object name + :type oid: str + :para mtime: the time to set the mtime to, 0 for the current time + :type mtime: int + :para flags: flags to apply to the entire operation + :type flags: int + """ + + oid = cstr(oid, 'oid') + cdef: + WriteOp _write_op = write_op + char *_oid = oid + time_t _mtime = mtime + int _flags = flags + + with nogil: + rados_write_op_operate(_write_op.write_op, self.io, _oid, &_mtime, _flags) + + @requires(('read_op', ReadOp), ('oid', str_type), ('flag', opt(int))) + def operate_read_op(self, read_op, oid, flag=LIBRADOS_OPERATION_NOFLAG): + """ + excute the real read operation + :para read_op: read operation object + :type read_op: ReadOp + :para oid: object name + :type oid: str + :para flag: flags to apply to the entire operation + :type flag: int + """ + oid = cstr(oid, 'oid') + cdef: + ReadOp _read_op = read_op + char *_oid = oid + int _flag = flag + + with nogil: + rados_read_op_operate(_read_op.read_op, self.io, _oid, _flag) + + @requires(('read_op', ReadOp), ('start_after', str_type), ('filter_prefix', str_type), ('max_return', int)) + def get_omap_vals(self, read_op, start_after, filter_prefix, max_return): + """ + get the omap values + :para read_op: read operation object + :type read_op: ReadOp + :para start_after: list keys starting after start_after + :type start_after: str + :para filter_prefix: list only keys beginning with filter_prefix + :type filter_prefix: str + :para max_return: list no more than max_return key/value pairs + :type max_return: int + :returns: an iterator over the the requested omap values, return value from this action + """ + + start_after = cstr(start_after, 'start_after') if start_after else None + filter_prefix = cstr(filter_prefix, 'filter_prefix') if filter_prefix else None + cdef: + char *_start_after = opt_str(start_after) + char *_filter_prefix = opt_str(filter_prefix) + ReadOp _read_op = read_op + rados_omap_iter_t iter_addr = NULL + int _max_return = max_return + int prval = 0 + + with nogil: + rados_read_op_omap_get_vals(_read_op.read_op, _start_after, _filter_prefix, + _max_return, &iter_addr, &prval) + it = OmapIterator(self) + it.ctx = iter_addr + return it, int(prval) + + @requires(('read_op', ReadOp), ('start_after', str_type), ('max_return', int)) + def get_omap_keys(self, read_op, start_after, max_return): + """ + get the omap keys + :para read_op: read operation object + :type read_op: ReadOp + :para start_after: list keys starting after start_after + :type start_after: str + :para max_return: list no more than max_return key/value pairs + :type max_return: int + :returns: an iterator over the the requested omap values, return value from this action + """ + start_after = cstr(start_after, 'start_after') if start_after else None + cdef: + char *_start_after = opt_str(start_after) + ReadOp _read_op = read_op + rados_omap_iter_t iter_addr = NULL + int _max_return = max_return + int prval + + with nogil: + rados_read_op_omap_get_keys(_read_op.read_op, _start_after, + _max_return, &iter_addr, &prval) + it = OmapIterator(self) + it.ctx = iter_addr + return it, int(prval) + + @requires(('read_op', ReadOp), ('keys', tuple)) + def get_omap_vals_by_keys(self, read_op, keys): + """ + get the omap values by keys + :para read_op: read operation object + :type read_op: ReadOp + :para keys: input key tuple + :type keys: tuple + :returns: an iterator over the the requested omap values, return value from this action + """ + keys = cstr_list(keys, 'keys') + cdef: + ReadOp _read_op = read_op + rados_omap_iter_t iter_addr + char **_keys = to_bytes_array(keys) + size_t key_num = len(keys) + int prval + + try: + with nogil: + rados_read_op_omap_get_vals_by_keys(_read_op.read_op, + _keys, + key_num, &iter_addr, &prval) + it = OmapIterator(self) + it.ctx = iter_addr + return it, int(prval) + finally: + free(_keys) + + @requires(('write_op', WriteOp), ('keys', tuple)) + def remove_omap_keys(self, write_op, keys): + """ + remove omap keys specifiled + :para write_op: write operation object + :type write_op: WriteOp + :para keys: input key tuple + :type keys: tuple + """ + + keys = cstr_list(keys, 'keys') + cdef: + WriteOp _write_op = write_op + size_t key_num = len(keys) + char **_keys = to_bytes_array(keys) + + try: + with nogil: + rados_write_op_omap_rm_keys(_write_op.write_op, _keys, key_num) + finally: + free(_keys) + + @requires(('write_op', WriteOp)) + def clear_omap(self, write_op): + """ + Remove all key/value pairs from an object + :para write_op: write operation object + :type write_op: WriteOp + """ + + cdef: + WriteOp _write_op = write_op + + with nogil: + rados_write_op_omap_clear(_write_op.write_op) + + @requires(('key', str_type), ('name', str_type), ('cookie', str_type), ('desc', str_type), + ('duration', opt(int)), ('flags', int)) + def lock_exclusive(self, key, name, cookie, desc="", duration=None, flags=0): + + """ + Take an exclusive lock on an object + + :param key: name of the object + :type key: str + :param name: name of the lock + :type name: str + :param cookie: cookie of the lock + :type cookie: str + :param desc: description of the lock + :type desc: str + :param duration: duration of the lock in seconds + :type duration: int + :param flags: flags + :type flags: int + + :raises: :class:`TypeError` + :raises: :class:`Error` + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + name = cstr(name, 'name') + cookie = cstr(cookie, 'cookie') + desc = cstr(desc, 'desc') + + cdef: + char* _key = key + char* _name = name + char* _cookie = cookie + char* _desc = desc + uint8_t _flags = flags + timeval _duration + + if duration is None: + with nogil: + ret = rados_lock_exclusive(self.io, _key, _name, _cookie, _desc, + NULL, _flags) + else: + _duration.tv_sec = duration + with nogil: + ret = rados_lock_exclusive(self.io, _key, _name, _cookie, _desc, + &_duration, _flags) + + if ret < 0: + raise make_ex(ret, "Ioctx.rados_lock_exclusive(%s): failed to set lock %s on %s" % (self.name, name, key)) + + @requires(('key', str_type), ('name', str_type), ('cookie', str_type), ('tag', str_type), + ('desc', str_type), ('duration', opt(int)), ('flags', int)) + def lock_shared(self, key, name, cookie, tag, desc="", duration=None, flags=0): + + """ + Take a shared lock on an object + + :param key: name of the object + :type key: str + :param name: name of the lock + :type name: str + :param cookie: cookie of the lock + :type cookie: str + :param tag: tag of the lock + :type tag: str + :param desc: description of the lock + :type desc: str + :param duration: duration of the lock in seconds + :type duration: int + :param flags: flags + :type flags: int + + :raises: :class:`TypeError` + :raises: :class:`Error` + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + tag = cstr(tag, 'tag') + name = cstr(name, 'name') + cookie = cstr(cookie, 'cookie') + desc = cstr(desc, 'desc') + + cdef: + char* _key = key + char* _tag = tag + char* _name = name + char* _cookie = cookie + char* _desc = desc + uint8_t _flags = flags + timeval _duration + + if duration is None: + with nogil: + ret = rados_lock_shared(self.io, _key, _name, _cookie, _tag, _desc, + NULL, _flags) + else: + _duration.tv_sec = duration + with nogil: + ret = rados_lock_shared(self.io, _key, _name, _cookie, _tag, _desc, + &_duration, _flags) + if ret < 0: + raise make_ex(ret, "Ioctx.rados_lock_exclusive(%s): failed to set lock %s on %s" % (self.name, name, key)) + + @requires(('key', str_type), ('name', str_type), ('cookie', str_type)) + def unlock(self, key, name, cookie): + + """ + Release a shared or exclusive lock on an object + + :param key: name of the object + :type key: str + :param name: name of the lock + :type name: str + :param cookie: cookie of the lock + :type cookie: str + + :raises: :class:`TypeError` + :raises: :class:`Error` + """ + self.require_ioctx_open() + + key = cstr(key, 'key') + name = cstr(name, 'name') + cookie = cstr(cookie, 'cookie') + + cdef: + char* _key = key + char* _name = name + char* _cookie = cookie + + with nogil: + ret = rados_unlock(self.io, _key, _name, _cookie) + if ret < 0: + raise make_ex(ret, "Ioctx.rados_lock_exclusive(%s): failed to set lock %s on %s" % (self.name, name, key)) + + +def set_object_locator(func): + def retfunc(self, *args, **kwargs): + if self.locator_key is not None: + old_locator = self.ioctx.get_locator_key() + self.ioctx.set_locator_key(self.locator_key) + retval = func(self, *args, **kwargs) + self.ioctx.set_locator_key(old_locator) + return retval + else: + return func(self, *args, **kwargs) + return retfunc + + +def set_object_namespace(func): + def retfunc(self, *args, **kwargs): + if self.nspace is None: + raise LogicError("Namespace not set properly in context") + old_nspace = self.ioctx.get_namespace() + self.ioctx.set_namespace(self.nspace) + retval = func(self, *args, **kwargs) + self.ioctx.set_namespace(old_nspace) + return retval + return retfunc + + +class Object(object): + """Rados object wrapper, makes the object look like a file""" + def __init__(self, ioctx, key, locator_key=None, nspace=None): + self.key = key + self.ioctx = ioctx + self.offset = 0 + self.state = "exists" + self.locator_key = locator_key + self.nspace = "" if nspace is None else nspace + + def __str__(self): + return "rados.Object(ioctx=%s,key=%s,nspace=%s,locator=%s)" % \ + (str(self.ioctx), self.key, "--default--" + if self.nspace is "" else self.nspace, self.locator_key) + + def require_object_exists(self): + if self.state != "exists": + raise ObjectStateError("The object is %s" % self.state) + + @set_object_locator + @set_object_namespace + def read(self, length=1024 * 1024): + self.require_object_exists() + ret = self.ioctx.read(self.key, length, self.offset) + self.offset += len(ret) + return ret + + @set_object_locator + @set_object_namespace + def write(self, string_to_write): + self.require_object_exists() + ret = self.ioctx.write(self.key, string_to_write, self.offset) + if ret == 0: + self.offset += len(string_to_write) + return ret + + @set_object_locator + @set_object_namespace + def remove(self): + self.require_object_exists() + self.ioctx.remove_object(self.key) + self.state = "removed" + + @set_object_locator + @set_object_namespace + def stat(self): + self.require_object_exists() + return self.ioctx.stat(self.key) + + def seek(self, position): + self.require_object_exists() + self.offset = position + + @set_object_locator + @set_object_namespace + def get_xattr(self, xattr_name): + self.require_object_exists() + return self.ioctx.get_xattr(self.key, xattr_name) + + @set_object_locator + @set_object_namespace + def get_xattrs(self): + self.require_object_exists() + return self.ioctx.get_xattrs(self.key) + + @set_object_locator + @set_object_namespace + def set_xattr(self, xattr_name, xattr_value): + self.require_object_exists() + return self.ioctx.set_xattr(self.key, xattr_name, xattr_value) + + @set_object_locator + @set_object_namespace + def rm_xattr(self, xattr_name): + self.require_object_exists() + return self.ioctx.rm_xattr(self.key, xattr_name) + +MONITOR_LEVELS = [ + "debug", + "info", + "warn", "warning", + "err", "error", + "sec", + ] + + +class MonitorLog(object): + # NOTE(sileht): Keep this class for backward compat + # method moved to Rados.monitor_log() + """ + For watching cluster log messages. Instantiate an object and keep + it around while callback is periodically called. Construct with + 'level' to monitor 'level' messages (one of MONITOR_LEVELS). + arg will be passed to the callback. + + callback will be called with: + arg (given to __init__) + line (the full line, including timestamp, who, level, msg) + who (which entity issued the log message) + timestamp_sec (sec of a struct timespec) + timestamp_nsec (sec of a struct timespec) + seq (sequence number) + level (string representing the level of the log message) + msg (the message itself) + callback's return value is ignored + """ + def __init__(self, cluster, level, callback, arg): + self.level = level + self.callback = callback + self.arg = arg + self.cluster = cluster + self.cluster.monitor_log(level, callback, arg) + diff --git a/src/pybind/rados/setup.py b/src/pybind/rados/setup.py new file mode 100755 index 000000000000..6a7b9bd4ca91 --- /dev/null +++ b/src/pybind/rados/setup.py @@ -0,0 +1,51 @@ +# Largely taken from +# https://blog.kevin-brown.com/programming/2014/09/24/combining-autotools-and-setuptools.html +import os, sys, os.path + +from setuptools.command.egg_info import egg_info +from distutils.core import setup +from distutils.extension import Extension +from Cython.Build import cythonize + +def get_version(): + try: + for line in open(os.path.join(os.path.dirname(__file__), "..", "ceph_ver.h")): + if "CEPH_GIT_NICE_VER" in line: + return line.split()[2][1:-1] + else: + return "0" + except IOError: + return "0" + +class EggInfoCommand(egg_info): + def finalize_options(self): + egg_info.finalize_options(self) + if "build" in self.distribution.command_obj: + build_command = self.distribution.command_obj["build"] + self.egg_base = build_command.build_base + self.egg_info = os.path.join(self.egg_base, os.path.basename(self.egg_info)) + +# Disable cythonification if we're not really building anything +if (len(sys.argv) >= 2 and + any(i in sys.argv[1:] for i in ('--help', 'clean', 'egg_info', '--version') + )): + def cythonize(x, **kwargs): + return x + +setup( + name = 'rados', + version = get_version(), + description = "Python libraries for the Ceph librados library", + long_description = ( + "This package contains Python libraries for interacting with Ceph's " + "rados library."), + ext_modules = cythonize([ + Extension("rados", + ["rados.pyx"], + libraries=["rados"] + ) + ], build_dir=os.environ.get("CYTHON_BUILD_DIR", None)), + cmdclass={ + "egg_info": EggInfoCommand, + }, +) diff --git a/src/test/pybind/test_rados.py b/src/test/pybind/test_rados.py index b4d392bc8f28..5549d695cce4 100644 --- a/src/test/pybind/test_rados.py +++ b/src/test/pybind/test_rados.py @@ -3,7 +3,7 @@ from nose.tools import eq_ as eq, ok_ as ok, assert_raises from rados import (Rados, Error, RadosStateError, Object, ObjectExists, ObjectNotFound, ObjectBusy, requires, opt, ANONYMOUS_AUID, ADMIN_AUID, LIBRADOS_ALL_NSPACES, WriteOpCtx, ReadOpCtx, - LIBRADOS_SNAP_HEAD) + LIBRADOS_SNAP_HEAD, MonitorLog) import time import threading import json @@ -34,6 +34,11 @@ def test_ioctx_context_manager(): with conn.open_ioctx('rbd') as ioctx: pass +def test_parse_argv(): + args = ['osd', 'pool', 'delete', 'foobar', 'foobar', '--yes-i-really-really-mean-it'] + r = Rados() + eq(args, r.conf_parse_argv(args)) + def test_parse_argv_empty_str(): args = [''] r = Rados() @@ -133,6 +138,14 @@ class TestRados(object): def tearDown(self): self.rados.shutdown() + def test_ping_monitor(self): + assert_raises(ObjectNotFound, self.rados.ping_monitor, 'not_exists_monitor') + cmd = {'prefix': 'mon dump', 'format':'json'} + ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'') + for mon in json.loads(buf.decode('utf8'))['mons']: + buf = json.loads(self.rados.ping_monitor(mon['name'])) + assert buf.get('health') + def test_create(self): self.rados.create_pool('foo') self.rados.delete_pool('foo') @@ -230,6 +243,31 @@ class TestRados(object): def test_blacklist_add(self): self.rados.blacklist_add("1.2.3.4/123", 1) + def test_get_cluster_stats(self): + stats = self.rados.get_cluster_stats() + assert stats['kb'] > 0 + assert stats['kb_avail'] > 0 + assert stats['kb_used'] > 0 + assert stats['num_objects'] >= 0 + + def test_monitor_log(self): + lock = threading.Condition() + def cb(arg, line, who, sec, nsec, seq, level, msg): + # NOTE(sileht): the old pyrados API was received the pointer as int + # instead of the value of arg + eq(arg, "arg") + with lock: + lock.notify() + return 0 + + # NOTE(sileht): force don't save the monitor into local var + # to ensure all references are correctly tracked into the lib + MonitorLog(self.rados, "debug", cb, "arg") + with lock: + lock.wait() + MonitorLog(self.rados, "debug", None, None) + eq(None, self.rados.monitor_callback) + class TestIoctx(object): def setUp(self): @@ -246,6 +284,25 @@ class TestIoctx(object): self.rados.delete_pool('test_pool') self.rados.shutdown() + def test_get_last_version(self): + version = self.ioctx.get_last_version() + assert version >= 0 + + def test_get_stats(self): + stats = self.ioctx.get_stats() + eq(stats, {'num_objects_unfound': 0, + 'num_objects_missing_on_primary': 0, + 'num_object_clones': 0, + 'num_objects': 0, + 'num_object_copies': 0, + 'num_bytes': 0, + 'num_rd_kb': 0, + 'num_wr_kb': 0, + 'num_kb': 0, + 'num_wr': 0, + 'num_objects_degraded': 0, + 'num_rd': 0}) + def test_change_auid(self): self.ioctx.change_auid(ANONYMOUS_AUID) self.ioctx.change_auid(ADMIN_AUID) @@ -368,13 +425,13 @@ class TestIoctx(object): eq(self.ioctx.read("insnap"), b"contents1") self.ioctx.remove_snap("snap1") self.ioctx.remove_object("insnap") - + def test_snap_read(self): self.ioctx.write("insnap", b"contents1") self.ioctx.create_snap("snap1") self.ioctx.remove_object("insnap") snap = self.ioctx.lookup_snap("snap1") - self.ioctx.set_read(int(snap.snap_id.value)) + self.ioctx.set_read(snap.snap_id) eq(self.ioctx.read("insnap"), b"contents1") self.ioctx.set_read(LIBRADOS_SNAP_HEAD) self.ioctx.write("inhead", b"contents2") @@ -393,6 +450,15 @@ class TestIoctx(object): self.ioctx.operate_read_op(read_op, "hw") iter.next() eq(list(iter), [("2", b"bbb"), ("3", b"ccc"), ("4", b"\x04\x04\x04\x04")]) + with ReadOpCtx(self.ioctx) as read_op: + iter, ret = self.ioctx.get_omap_vals(read_op, "2", "", 4) + self.ioctx.operate_read_op(read_op, "hw") + eq(("3", b"ccc"), iter.next()) + eq(list(iter), [("4", b"\x04\x04\x04\x04")]) + with ReadOpCtx(self.ioctx) as read_op: + iter, ret = self.ioctx.get_omap_vals(read_op, "", "2", 4) + self.ioctx.operate_read_op(read_op, "hw") + eq(list(iter), [("2", b"bbb")]) def test_get_omap_vals_by_keys(self): keys = ("1", "2", "3", "4") @@ -468,6 +534,24 @@ class TestIoctx(object): eq(contents, b"bar") [i.remove() for i in self.ioctx.list_objects()] + def test_aio_write_no_comp_ref(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + # NOTE(sileht): force don't save the comp into local var + # to ensure all references are correctly tracked into the lib + self.ioctx.aio_write("foo", b"bar", 0, cb, cb) + with lock: + while count[0] < 2: + lock.wait() + contents = self.ioctx.read("foo") + eq(contents, b"bar") + [i.remove() for i in self.ioctx.list_objects()] + def test_aio_append(self): lock = threading.Condition() count = [0] @@ -607,13 +691,13 @@ class TestIoctx(object): assert_raises(ObjectNotFound, self.ioctx.unlock, "foo", "lock", "locker2") def test_execute(self): - self.ioctx.write("foo", "") # ensure object exists + self.ioctx.write("foo", b"") # ensure object exists - ret, buf = self.ioctx.execute("foo", "hello", "say_hello", "") - eq(buf, "Hello, world!") + ret, buf = self.ioctx.execute("foo", "hello", "say_hello", b"") + eq(buf, b"Hello, world!") - ret, buf = self.ioctx.execute("foo", "hello", "say_hello", "nose") - eq(buf, "Hello, nose!") + ret, buf = self.ioctx.execute("foo", "hello", "say_hello", b"nose") + eq(buf, b"Hello, nose!") class TestObject(object): @@ -711,3 +795,16 @@ class TestCommand(object): out = json.loads(err) eq(out['blocksize'], cmd['size']) eq(out['bytes_written'], cmd['count']) + + def test_ceph_osd_pool_create_utf8(self): + if _python2: + # Use encoded bytestring + poolname = b"\351\273\205" + else: + poolname = "\u9ec5" + + cmd = {"prefix": "osd pool create", "pg_num": 16, "pool": poolname} + ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'') + eq(ret, 0) + assert len(out) > 0 + eq(u"pool '\u9ec5' created", out)