+++ /dev/null
-# 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 <hannu.valtonen@ormod.com>
-# Copyright 2015 Hector Martin <marcan@marcan.st>
-# Copyright 2016 Mehdi Abaakouk <sileht@redhat.com>
-
-from cpython cimport PyObject, ref
-from cpython.pycapsule cimport *
-from libc cimport errno
-from libc.stdint cimport *
-from libc.stdlib cimport malloc, realloc, free
-
-import sys
-import threading
-import time
-
-from collections import Callable
-from datetime import datetime
-from functools import partial, wraps
-from itertools import chain
-
-# Are we running Python 2.x
-if sys.version_info[0] < 3:
- 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 Ioctx.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"
- _LIBRADOS_CREATE_EXCLUSIVE "LIBRADOS_CREATE_EXCLUSIVE"
- _LIBRADOS_CREATE_IDEMPOTENT "LIBRADOS_CREATE_IDEMPOTENT"
-
- 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)
- ctypedef void (*rados_log_callback2_t)(void *arg, const char *line, const char *channel, const char *who, const char *name,
- 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_create_with_context(rados_t *cluster, rados_config_t cct)
- 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_inconsistent_pg_list(rados_t cluster, int64_t pool, char *buf, size_t len)
-
- 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_application_enable(rados_ioctx_t io, const char *app_name,
- int force)
- int rados_application_list(rados_ioctx_t io, char *values,
- size_t *values_len)
- int rados_application_metadata_get(rados_ioctx_t io, const char *app_name,
- const char *key, char *value,
- size_t *value_len)
- int rados_application_metadata_set(rados_ioctx_t io, const char *app_name,
- const char *key, const char *value)
- int rados_application_metadata_remove(rados_ioctx_t io,
- const char *app_name, const char *key)
- int rados_application_metadata_list(rados_ioctx_t io,
- const char *app_name, char *keys,
- size_t *key_len, char *values,
- size_t *value_len)
- 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_mgr_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_monitor_log2(rados_t cluster, const char *level, rados_log_callback2_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_stat(rados_ioctx_t io, const char *oid, rados_completion_t completion, uint64_t *psize, time_t *pmtime)
- 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_aio_exec(rados_ioctx_t io, const char * oid, rados_completion_t completion, 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)
- int rados_aio_write_op_operate(rados_write_op_t write_op, rados_ioctx_t io, rados_completion_t completion, 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_write_op_set_flags(rados_write_op_t write_op, int flags)
-
- void rados_write_op_create(rados_write_op_t write_op, int exclusive, const char *category)
- void rados_write_op_append(rados_write_op_t write_op, const char *buffer, size_t len)
- void rados_write_op_write_full(rados_write_op_t write_op, const char *buffer, size_t len)
- void rados_write_op_write(rados_write_op_t write_op, const char *buffer, size_t len, uint64_t offset)
- void rados_write_op_remove(rados_write_op_t write_op)
- void rados_write_op_truncate(rados_write_op_t write_op, uint64_t offset)
- void rados_write_op_zero(rados_write_op_t write_op, uint64_t offset, uint64_t len)
-
- void rados_read_op_omap_get_vals2(rados_read_op_t read_op, const char * start_after, const char * filter_prefix, uint64_t max_return, rados_omap_iter_t * iter, unsigned char *pmore, int * prval)
- void rados_read_op_omap_get_keys2(rados_read_op_t read_op, const char * start_after, uint64_t max_return, rados_omap_iter_t * iter, unsigned char *pmore, 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_aio_read_op_operate(rados_read_op_t read_op, rados_ioctx_t io, rados_completion_t completion, const char *oid, int flags)
- void rados_read_op_set_flags(rados_read_op_t read_op, 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')
-
-LIBRADOS_CREATE_EXCLUSIVE = _LIBRADOS_CREATE_EXCLUSIVE
-LIBRADOS_CREATE_IDEMPOTENT = _LIBRADOS_CREATE_IDEMPOTENT
-
-ANONYMOUS_AUID = 0xffffffffffffffff
-ADMIN_AUID = 0
-
-
-class Error(Exception):
- """ `Error` class, derived from `Exception` """
- pass
-
-
-class InvalidArgumentError(Error):
- pass
-
-
-class OSError(Error):
- """ `OSError` class, derived from `Error` """
- def __init__(self, message, errno=None):
- super(OSError, self).__init__(message)
- self.errno = errno
-
- def __str__(self):
- msg = super(OSError, self).__str__()
- if self.errno is None:
- return msg
- return '[errno {0}] {1}'.format(self.errno, msg)
-
- def __reduce__(self):
- return (self.__class__, (self.message, self.errno))
-
-class InterruptedOrTimeoutError(OSError):
- """ `InterruptedOrTimeoutError` class, derived from `OSError` """
- pass
-
-
-class PermissionError(OSError):
- """ `PermissionError` class, derived from `OSError` """
- pass
-
-
-class PermissionDeniedError(OSError):
- """ deal with EACCES related. """
- pass
-
-
-class ObjectNotFound(OSError):
- """ `ObjectNotFound` class, derived from `OSError` """
- pass
-
-
-class NoData(OSError):
- """ `NoData` class, derived from `OSError` """
- pass
-
-
-class ObjectExists(OSError):
- """ `ObjectExists` class, derived from `OSError` """
- pass
-
-
-class ObjectBusy(OSError):
- """ `ObjectBusy` class, derived from `IOError` """
- pass
-
-
-class IOError(OSError):
- """ `ObjectBusy` class, derived from `OSError` """
- pass
-
-
-class NoSpace(OSError):
- """ `NoSpace` class, derived from `OSError` """
- 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(OSError):
- """ `TimedOut` class, derived from `OSError` """
- pass
-
-
-IF UNAME_SYSNAME == "FreeBSD":
- cdef errno_to_exception = {
- errno.EPERM : PermissionError,
- errno.ENOENT : ObjectNotFound,
- errno.EIO : IOError,
- errno.ENOSPC : NoSpace,
- errno.EEXIST : ObjectExists,
- errno.EBUSY : ObjectBusy,
- errno.ENOATTR : NoData,
- errno.EINTR : InterruptedOrTimeoutError,
- errno.ETIMEDOUT : TimedOut,
- errno.EACCES : PermissionDeniedError,
- errno.EINVAL : InvalidArgumentError,
- }
-ELSE:
- 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,
- errno.EINVAL : InvalidArgumentError,
- }
-
-
-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, errno=ret)
- else:
- return OSError(msg, errno=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 = <size_t *>malloc(len(list_int) * sizeof(size_t))
- if ret == NULL:
- raise MemoryError("malloc failed")
- for i in xrange(len(list_int)):
- ret[i] = <size_t>list_int[i]
- return ret
-
-
-cdef char ** to_bytes_array(list_bytes):
- cdef char **ret = <char **>malloc(len(list_bytes) * sizeof(char *))
- if ret == NULL:
- raise MemoryError("malloc failed")
- for i in xrange(len(list_bytes)):
- ret[i] = <char *>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 = <object>arg
- cb_info[0](cb_info[1], line, who, sec, nsec, seq, level, msg)
- return 0
-
-cdef int __monitor_callback2(void *arg, const char *line, const char *channel,
- const char *who,
- const char *name,
- uint64_t sec, uint64_t nsec, uint64_t seq,
- const char *level, const char *msg) with gil:
- cdef object cb_info = <object>arg
- cb_info[0](cb_info[1], line, channel, name, 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,
- context=None):
- self.monitor_callback = None
- self.monitor_callback2 = 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 = ''
-
- name = cstr(name, 'name')
- clustername = cstr(clustername, 'clustername')
- cdef:
- char *_name = name
- char *_clustername = clustername
- int _flags = flags
- int ret
-
- if context:
- # Unpack void* (aka rados_config_t) from capsule
- rados_config = <rados_config_t> PyCapsule_GetPointer(context, NULL)
- with nogil:
- ret = rados_create_with_context(&self.cluster, rados_config)
- else:
- 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 = <char **>malloc(_argc * sizeof(char *))
- with nogil:
- ret = rados_conf_parse_argv_remainder(self.cluster, _argc,
- <const char**>_argv,
- <const char**>_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 = <char *>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.<ID>)
- :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 = 0
- 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 = <char *>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)
-
- @requires(('pool_id', int))
- def get_inconsistent_pgs(self, pool_id):
- """
- List inconsistent placement groups in the given pool
-
- :param pool_id: ID of the pool in which PGs are listed
- :type pool_id: int
- :returns: list - inconsistent placement groups
- """
- self.require_state("connected")
- cdef:
- int64_t pool = pool_id
- size_t size = 512
- char *pgs = NULL
-
- try:
- while True:
- pgs = <char *>realloc_chk(pgs, size);
- with nogil:
- ret = rados_inconsistent_pg_list(self.cluster, pool,
- pgs, size)
- if ret > <int>size:
- size *= 2
- elif ret >= 0:
- break
- else:
- raise make_ex(ret, "error calling inconsistent_pg_list")
- return [pg for pg in decode_cstr(pgs[:ret]).split('\0') if pg]
- finally:
- free(pgs)
-
- 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 = <char *>realloc_chk(c_names, size)
- with nogil:
- ret = rados_pool_list(self.cluster, c_names, size)
- if ret > <int>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 != <int>buf_len:
- _PyBytes_Resize(&ret_s, ret)
- return <object>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,
- <const char **>_cmd, _cmdlen,
- <const char*>_inbuf, _inbuf_len,
- &_outbuf, &_outbuf_len,
- &_outs, &_outs_len)
- else:
- with nogil:
- ret = rados_mon_command(self.cluster,
- <const char **>_cmd, _cmdlen,
- <const char*>_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,
- <const char **>_cmd, _cmdlen,
- <const char*>_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 mgr_command(self, cmd, inbuf, timeout=0):
- """
- 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:
- 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_mgr_command(self.cluster,
- <const char **>_cmd, _cmdlen,
- <const char*>_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,
- <const char **>_cmd, _cmdlen,
- <const char *>_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, <const char*>_level, NULL, NULL)
- self.monitor_callback = None
- self.monitor_callback2 = None
- return
-
- cb = (callback, arg)
- cdef PyObject* _arg = <PyObject*>cb
- with nogil:
- r = rados_monitor_log(self.cluster, <const char*>_level,
- <rados_log_callback_t>&__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
- self.monitor_callback2 = None
-
- def monitor_log2(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_log2(self.cluster, <const char*>_level, NULL, NULL)
- self.monitor_callback = None
- self.monitor_callback2 = None
- return
-
- cb = (callback, arg)
- cdef PyObject* _arg = <PyObject*>cb
- with nogil:
- r = rados_monitor_log2(self.cluster, <const char*>_level,
- <rados_log_callback2_t>&__monitor_callback2, _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 = None
- self.monitor_callback2 = cb
-
-
-cdef class OmapIterator(object):
- """Omap iterator"""
-
- cdef public Ioctx ioctx
- cdef rados_omap_iter_t ctx
-
- def __cinit__(self, Ioctx ioctx):
- self.ioctx = ioctx
-
- def __iter__(self):
- return self
-
- 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 __dealloc__(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):
- """
- 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 __dealloc__(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):
- """
- 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 __dealloc__(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 = <rados_snap_t*>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):
- """
- 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 = <char *>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
- PyObject* buf
-
- 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 __dealloc__(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.
- """
- ref.Py_XDECREF(self.buf)
- self.buf = NULL
- if self.rados_comp != NULL:
- with nogil:
- rados_aio_release(self.rados_comp)
- self.rados_comp = NULL
-
- 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)
-
- @requires(('exclusive', opt(int)))
- def new(self, exclusive=None):
- """
- Create the object.
- """
-
- cdef:
- int _exclusive = exclusive
-
- with nogil:
- rados_write_op_create(self.write_op, _exclusive, NULL)
-
-
- def remove(self):
- """
- Remove object.
- """
- with nogil:
- rados_write_op_remove(self.write_op)
-
- @requires(('flags', int))
- def set_flags(self, flags=LIBRADOS_OPERATION_NOFLAG):
- """
- Set flags for the last operation added to this write_op.
- :para flags: flags to apply to the last operation
- :type flags: int
- """
-
- cdef:
- int _flags = flags
-
- with nogil:
- rados_write_op_set_flags(self.write_op, _flags)
-
- @requires(('to_write', bytes))
- def append(self, to_write):
- """
- Append data to an object synchronously
- :param to_write: data to write
- :type to_write: bytes
- """
-
- cdef:
- char *_to_write = to_write
- size_t length = len(to_write)
-
- with nogil:
- rados_write_op_append(self.write_op, _to_write, length)
-
- @requires(('to_write', bytes))
- def write_full(self, to_write):
- """
- Write whole object, atomically replacing it.
- :param to_write: data to write
- :type to_write: bytes
- """
-
- cdef:
- char *_to_write = to_write
- size_t length = len(to_write)
-
- with nogil:
- rados_write_op_write_full(self.write_op, _to_write, length)
-
- @requires(('to_write', bytes), ('offset', int))
- def write(self, to_write, offset=0):
- """
- Write to offset.
- :param to_write: data to write
- :type to_write: bytes
- :param offset: byte offset in the object to begin writing at
- :type offset: int
- """
-
- cdef:
- char *_to_write = to_write
- size_t length = len(to_write)
- uint64_t _offset = offset
-
- with nogil:
- rados_write_op_write(self.write_op, _to_write, length, _offset)
-
- @requires(('offset', int), ('length', int))
- def zero(self, offset, length):
- """
- Zero part of an object.
- :param offset: byte offset in the object to begin writing at
- :type offset: int
- :param offset: number of zero to write
- :type offset: int
- """
-
- cdef:
- size_t _length = length
- uint64_t _offset = offset
-
- with nogil:
- rados_write_op_zero(self.write_op, _length, _offset)
-
- @requires(('offset', int))
- def truncate(self, offset):
- """
- Truncate an object.
- :param offset: byte offset in the object to begin truncating at
- :type offset: int
- """
-
- cdef:
- uint64_t _offset = offset
-
- with nogil:
- rados_write_op_truncate(self.write_op, _offset)
-
-
-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)
-
- @requires(('flags', int))
- def set_flags(self, flags=LIBRADOS_OPERATION_NOFLAG):
- """
- Set flags for the last operation added to this read_op.
- :para flags: flags to apply to the last operation
- :type flags: int
- """
-
- cdef:
- int _flags = flags
-
- with nogil:
- rados_read_op_set_flags(self.read_op, _flags)
-
-
-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 = <object>args
- cb._safe()
- return 0
-
-
-cdef int __aio_complete_cb(rados_completion_t completion, void *args) with gil:
- """
- Callback to oncomplete() for asynchronous operations
- """
- cdef object cb = <object>args
- cb._complete()
- 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 __dealloc__(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= <PyObject*>completion_obj
-
- if oncomplete:
- complete_cb = <rados_callback_t>&__aio_complete_cb
- if onsafe:
- safe_cb = <rados_callback_t>&__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
-
- @requires(('object_name', str_type), ('oncomplete', opt(Callable)))
- def aio_stat(self, object_name, oncomplete):
- """
- Asynchronously get object stats (size/mtime)
-
- oncomplete will be called with the returned size and mtime
- as well as the completion:
-
- oncomplete(completion, size, mtime)
-
- :param object_name: the name of the object to get stats from
- :type object_name: str
- :param oncomplete: what to do when the stat 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 psize
- time_t pmtime
-
- def oncomplete_(completion_v):
- cdef Completion _completion_v = completion_v
- return_value = _completion_v.get_return_value()
- if return_value >= 0:
- return oncomplete(_completion_v, psize, time.localtime(pmtime))
- else:
- return oncomplete(_completion_v, None, None)
-
- completion = self.__get_completion(oncomplete_, None)
- self.__track_completion(completion)
- with nogil:
- ret = rados_aio_stat(self.io, _object_name, completion.rados_comp,
- &psize, &pmtime)
-
- if ret < 0:
- completion._cleanup()
- raise make_ex(ret, "error stating %s" % object_name)
- return completion
-
- @requires(('object_name', str_type), ('to_write', bytes), ('offset', int),
- ('oncomplete', opt(Callable)), ('onsafe', opt(Callable)))
- 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)
- self.__track_completion(completion)
- with nogil:
- ret = rados_aio_write(self.io, _object_name, completion.rados_comp,
- _to_write, size, _offset)
- if ret < 0:
- completion._cleanup()
- raise make_ex(ret, "error writing object %s" % object_name)
- return completion
-
- @requires(('object_name', str_type), ('to_write', bytes), ('oncomplete', opt(Callable)),
- ('onsafe', opt(Callable)))
- 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)
- self.__track_completion(completion)
- with nogil:
- ret = rados_aio_write_full(self.io, _object_name,
- completion.rados_comp,
- _to_write, size)
- if ret < 0:
- completion._cleanup()
- raise make_ex(ret, "error writing object %s" % object_name)
- return completion
-
- @requires(('object_name', str_type), ('to_append', bytes), ('oncomplete', opt(Callable)),
- ('onsafe', opt(Callable)))
- 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)
- self.__track_completion(completion)
- with nogil:
- ret = rados_aio_append(self.io, _object_name,
- completion.rados_comp,
- _to_append, size)
- if ret < 0:
- completion._cleanup()
- raise make_ex(ret, "error appending object %s" % object_name)
- 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")
-
- @requires(('object_name', str_type), ('length', int), ('offset', int),
- ('oncomplete', opt(Callable)))
- 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
-
- def oncomplete_(completion_v):
- cdef Completion _completion_v = completion_v
- return_value = _completion_v.get_return_value()
- if return_value > 0 and return_value != length:
- _PyBytes_Resize(&_completion_v.buf, return_value)
- return oncomplete(_completion_v, <object>_completion_v.buf if return_value >= 0 else None)
-
- completion = self.__get_completion(oncomplete_, None)
- completion.buf = PyBytes_FromStringAndSize(NULL, length)
- ret_buf = PyBytes_AsString(completion.buf)
- self.__track_completion(completion)
- with nogil:
- ret = rados_aio_read(self.io, _object_name, completion.rados_comp,
- ret_buf, _length, _offset)
- if ret < 0:
- completion._cleanup()
- raise make_ex(ret, "error reading %s" % object_name)
- return completion
-
- @requires(('object_name', str_type), ('cls', str_type), ('method', str_type),
- ('data', bytes), ('length', int),
- ('oncomplete', opt(Callable)), ('onsafe', opt(Callable)))
- def aio_execute(self, object_name, cls, method, data,
- length=8192, oncomplete=None, onsafe=None):
- """
- Asynchronously execute an OSD class method on an object.
-
- oncomplete and onsafe will be called with the data returned from
- the plugin as well as the completion:
-
- oncomplete(completion, data)
- onsafe(completion, data)
-
- :param object_name: name of the object
- :type object_name: 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=8192)
- :type length: int
- :param oncomplete: what to do when the execution is complete
- :type oncomplete: completion
- :param onsafe: what to do when the execution is safe and complete
- :type onsafe: completion
-
- :raises: :class:`Error`
- :returns: completion object
- """
-
- object_name = cstr(object_name, 'object_name')
- cls = cstr(cls, 'cls')
- method = cstr(method, 'method')
- cdef:
- Completion completion
- char *_object_name = object_name
- char *_cls = cls
- char *_method = method
- char *_data = data
- size_t _data_len = len(data)
-
- char *ref_buf
- size_t _length = length
-
- def oncomplete_(completion_v):
- cdef Completion _completion_v = completion_v
- return_value = _completion_v.get_return_value()
- if return_value > 0 and return_value != length:
- _PyBytes_Resize(&_completion_v.buf, return_value)
- return oncomplete(_completion_v, <object>_completion_v.buf if return_value >= 0 else None)
-
- def onsafe_(completion_v):
- cdef Completion _completion_v = completion_v
- return_value = _completion_v.get_return_value()
- return onsafe(_completion_v, <object>_completion_v.buf if return_value >= 0 else None)
-
- completion = self.__get_completion(oncomplete_ if oncomplete else None, onsafe_ if onsafe else None)
- completion.buf = PyBytes_FromStringAndSize(NULL, length)
- ret_buf = PyBytes_AsString(completion.buf)
- self.__track_completion(completion)
- with nogil:
- ret = rados_aio_exec(self.io, _object_name, completion.rados_comp,
- _cls, _method, _data, _data_len, ret_buf, _length)
- if ret < 0:
- completion._cleanup()
- raise make_ex(ret, "error executing %s::%s on %s" % (cls, method, object_name))
- return completion
-
- @requires(('object_name', str_type), ('oncomplete', opt(Callable)), ('onsafe', opt(Callable)))
- 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)
- self.__track_completion(completion)
- with nogil:
- ret = rados_aio_remove(self.io, _object_name,
- completion.rados_comp)
- if ret < 0:
- completion._cleanup()
- raise make_ex(ret, "error removing %s" % object_name)
- return completion
-
- def require_ioctx_open(self):
- """
- Checks if the rados.Ioctx object state is 'open'
-
- :raises: IoctxStateError
- """
- if self.state != "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 == "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 <object>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=8192)
- :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, <object>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 = <char *>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,
- <const char**>_keys,
- <const char**>_values,
- <const size_t*>_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:
- ret = rados_write_op_operate(_write_op.write_op, self.io, _oid, &_mtime, _flags)
- if ret != 0:
- raise make_ex(ret, "Failed to operate write op for oid %s" % oid)
-
- @requires(('write_op', WriteOp), ('oid', str_type), ('oncomplete', opt(Callable)), ('onsafe', opt(Callable)), ('mtime', opt(int)), ('flags', opt(int)))
- def operate_aio_write_op(self, write_op, oid, oncomplete=None, onsafe=None, mtime=0, flags=LIBRADOS_OPERATION_NOFLAG):
- """
- excute the real write operation asynchronously
- :para write_op: write operation object
- :type write_op: WriteOp
- :para oid: object name
- :type oid: 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
- :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
-
- :raises: :class:`Error`
- :returns: completion object
- """
-
- oid = cstr(oid, 'oid')
- cdef:
- WriteOp _write_op = write_op
- char *_oid = oid
- Completion completion
- time_t _mtime = mtime
- int _flags = flags
-
- completion = self.__get_completion(oncomplete, onsafe)
- self.__track_completion(completion)
-
- with nogil:
- ret = rados_aio_write_op_operate(_write_op.write_op, self.io, completion.rados_comp, _oid,
- &_mtime, _flags)
- if ret != 0:
- completion._cleanup()
- raise make_ex(ret, "Failed to operate aio write op for oid %s" % oid)
- return completion
-
- @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:
- ret = rados_read_op_operate(_read_op.read_op, self.io, _oid, _flag)
- if ret != 0:
- raise make_ex(ret, "Failed to operate read op for oid %s" % oid)
-
- @requires(('read_op', ReadOp), ('oid', str_type), ('oncomplete', opt(Callable)), ('onsafe', opt(Callable)), ('flag', opt(int)))
- def operate_aio_read_op(self, read_op, oid, oncomplete=None, onsafe=None, 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
- :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
- :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
- Completion completion
- int _flag = flag
-
- completion = self.__get_completion(oncomplete, onsafe)
- self.__track_completion(completion)
-
- with nogil:
- ret = rados_aio_read_op_operate(_read_op.read_op, self.io, completion.rados_comp, _oid, _flag)
- if ret != 0:
- completion._cleanup()
- raise make_ex(ret, "Failed to operate aio read op for oid %s" % oid)
- return completion
-
- @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 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_vals2(_read_op.read_op, _start_after, _filter_prefix,
- _max_return, &iter_addr, NULL, &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 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 = 0
-
- with nogil:
- rados_read_op_omap_get_keys2(_read_op.read_op, _start_after,
- _max_return, &iter_addr, NULL, &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 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 = 0
-
- try:
- with nogil:
- rados_read_op_omap_get_vals_by_keys(_read_op.read_op,
- <const char**>_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, <const char**>_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 application_enable(self, app_name, force=False):
- """
- Enable an application on an OSD pool
-
- :param app_name: application name
- :type app_name: str
- :param force: False if only a single app should exist per pool
- :type expire_seconds: boool
-
- :raises: :class:`Error`
- """
- app_name = cstr(app_name, 'app_name')
- cdef:
- char *_app_name = app_name
- int _force = (1 if force else 0)
-
- with nogil:
- ret = rados_application_enable(self.io, _app_name, _force)
- if ret < 0:
- raise make_ex(ret, "error enabling application")
-
- def application_list(self):
- """
- Returns a list of enabled applications
-
- :returns: list of app name string
- """
- cdef:
- size_t length = 128
- char *apps = NULL
-
- try:
- while True:
- apps = <char *>realloc_chk(apps, length)
- with nogil:
- ret = rados_application_list(self.io, apps, &length)
- if ret == 0:
- return [decode_cstr(app) for app in
- apps[:length].split(b'\0') if app]
- elif ret == -errno.ENOENT:
- return None
- elif ret == -errno.ERANGE:
- pass
- else:
- raise make_ex(ret, "error listing applications")
- finally:
- free(apps)
-
- def application_metadata_set(self, app_name, key, value):
- """
- Sets application metadata on an OSD pool
-
- :param app_name: application name
- :type app_name: str
- :param key: metadata key
- :type key: str
- :param value: metadata value
- :type value: str
-
- :raises: :class:`Error`
- """
- app_name = cstr(app_name, 'app_name')
- key = cstr(key, 'key')
- value = cstr(value, 'value')
- cdef:
- char *_app_name = app_name
- char *_key = key
- char *_value = value
-
- with nogil:
- ret = rados_application_metadata_set(self.io, _app_name, _key,
- _value)
- if ret < 0:
- raise make_ex(ret, "error setting application metadata")
-
- def application_metadata_remove(self, app_name, key):
- """
- Remove application metadata from an OSD pool
-
- :param app_name: application name
- :type app_name: str
- :param key: metadata key
- :type key: str
-
- :raises: :class:`Error`
- """
- app_name = cstr(app_name, 'app_name')
- key = cstr(key, 'key')
- cdef:
- char *_app_name = app_name
- char *_key = key
-
- with nogil:
- ret = rados_application_metadata_remove(self.io, _app_name, _key)
- if ret < 0:
- raise make_ex(ret, "error removing application metadata")
-
- def application_metadata_list(self, app_name):
- """
- Returns a list of enabled applications
-
- :param app_name: application name
- :type app_name: str
- :returns: list of key/value tuples
- """
- app_name = cstr(app_name, 'app_name')
- cdef:
- char *_app_name = app_name
- size_t key_length = 128
- size_t val_length = 128
- char *c_keys = NULL
- char *c_vals = NULL
-
- try:
- while True:
- c_keys = <char *>realloc_chk(c_keys, key_length)
- c_vals = <char *>realloc_chk(c_vals, val_length)
- with nogil:
- ret = rados_application_metadata_list(self.io, _app_name,
- c_keys, &key_length,
- c_vals, &val_length)
- if ret == 0:
- keys = [decode_cstr(key) for key in
- c_keys[:key_length].split(b'\0') if key]
- vals = [decode_cstr(val) for val in
- c_vals[:val_length].split(b'\0') if val]
- return zip(keys, vals)
- elif ret == -errno.ERANGE:
- pass
- else:
- raise make_ex(ret, "error listing application metadata")
- finally:
- free(c_keys)
- free(c_vals)
-
-
-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)
-
Code Review / stor4nfv.git / blobdiff