X-Git-Url: https://git.verplant.org/?a=blobdiff_plain;f=src%2Fceph.c;h=64d87855fbb0f820ea1a8e8f6d09eeb61894154c;hb=4d370741101aeb037ae52f3529a4a0869e0dc08a;hp=d762754d51733326aa58d466c60f29404675d762;hpb=4b8c1289003f0d60728f06a43bfcc9bc3de7004c;p=collectd.git diff --git a/src/ceph.c b/src/ceph.c index d762754d..64d87855 100644 --- a/src/ceph.c +++ b/src/ceph.c @@ -27,6 +27,7 @@ #define _BSD_SOURCE #include "collectd.h" + #include "common.h" #include "plugin.h" @@ -37,6 +38,9 @@ #if HAVE_YAJL_YAJL_VERSION_H #include #endif +#ifdef HAVE_SYS_CAPABILITY_H +# include +#endif #include #include @@ -96,7 +100,7 @@ enum ceph_dset_type_d }; /** Valid types for ceph defined in types.db */ -const char * ceph_dset_types [CEPH_DSET_TYPES_NUM] = +static const char * const ceph_dset_types [CEPH_DSET_TYPES_NUM] = {"ceph_latency", "ceph_bytes", "ceph_rate"}; /******* ceph_daemon *******/ @@ -134,11 +138,10 @@ struct yajl_struct { node_handler_t handler; void * handler_arg; - struct { - char key[DATA_MAX_NAME_LEN]; - int key_len; - } state[YAJL_MAX_DEPTH]; - int depth; + + char *key; + char *stack[YAJL_MAX_DEPTH]; + size_t depth; }; typedef struct yajl_struct yajl_struct; @@ -164,7 +167,7 @@ static int convert_special_metrics = 1; static struct ceph_daemon **g_daemons = NULL; /** Number of elements in g_daemons */ -static int g_num_daemons = 0; +static size_t g_num_daemons = 0; /** * A set of data that we build up in memory while parsing the JSON. @@ -257,68 +260,87 @@ static int ceph_cb_boolean(void *ctx, int bool_val) return CEPH_CB_CONTINUE; } +#define BUFFER_ADD(dest, src) do { \ + size_t dest_size = sizeof (dest); \ + size_t dest_len = strlen (dest); \ + if (dest_size > dest_len) { \ + sstrncpy ((dest) + dest_len, (src), dest_size - dest_len); \ + } \ + (dest)[dest_size - 1] = 0; \ +} while (0) + static int ceph_cb_number(void *ctx, const char *number_val, yajl_len_t number_len) { - yajl_struct *yajl = (yajl_struct*)ctx; + yajl_struct *state = (yajl_struct*) ctx; char buffer[number_len+1]; - int i, latency_type = 0, result; - char key[128]; + char key[2 * DATA_MAX_NAME_LEN] = { 0 }; + _Bool latency_type = 0; + int status; memcpy(buffer, number_val, number_len); - buffer[sizeof(buffer) - 1] = 0; + buffer[sizeof(buffer) - 1] = '\0'; - ssnprintf(key, yajl->state[0].key_len, "%s", yajl->state[0].key); - for(i = 1; i < yajl->depth; i++) + for (size_t i = 0; i < state->depth; i++) { - if((i == yajl->depth-1) && ((strcmp(yajl->state[i].key,"avgcount") == 0) - || (strcmp(yajl->state[i].key,"sum") == 0))) + if (state->stack[i] == NULL) + continue; + + if (strlen (key) != 0) + BUFFER_ADD (key, "."); + BUFFER_ADD (key, state->stack[i]); + } + + /* Special case for latency metrics. */ + if ((strcmp ("avgcount", state->key) == 0) + || (strcmp ("sum", state->key) == 0)) + { + latency_type = 1; + + /* depth >= 2 => (stack[-1] != NULL && stack[-2] != NULL) */ + assert ((state->depth < 2) + || ((state->stack[state->depth - 1] != NULL) + && (state->stack[state->depth - 2] != NULL))); + + /* Super-special case for filestore.journal_wr_bytes.avgcount: For + * some reason, Ceph schema encodes this as a count/sum pair while all + * other "Bytes" data (excluding used/capacity bytes for OSD space) uses + * a single "Derive" type. To spare further confusion, keep this KPI as + * the same type of other "Bytes". Instead of keeping an "average" or + * "rate", use the "sum" in the pair and assign that to the derive + * value. */ + if (convert_special_metrics && (state->depth >= 2) + && (strcmp("filestore", state->stack[state->depth - 2]) == 0) + && (strcmp("journal_wr_bytes", state->stack[state->depth - 1]) == 0) + && (strcmp("avgcount", state->key) == 0)) { - if(convert_special_metrics) - { - /** - * Special case for filestore:JournalWrBytes. For some reason, - * Ceph schema encodes this as a count/sum pair while all - * other "Bytes" data (excluding used/capacity bytes for OSD - * space) uses a single "Derive" type. To spare further - * confusion, keep this KPI as the same type of other "Bytes". - * Instead of keeping an "average" or "rate", use the "sum" in - * the pair and assign that to the derive value. - */ - if((strcmp(yajl->state[i-1].key, "journal_wr_bytes") == 0) && - (strcmp(yajl->state[i-2].key,"filestore") == 0) && - (strcmp(yajl->state[i].key,"avgcount") == 0)) - { - DEBUG("ceph plugin: Skipping avgcount for filestore.JournalWrBytes"); - yajl->depth = (yajl->depth - 1); - return CEPH_CB_CONTINUE; - } - } - //probably a avgcount/sum pair. if not - we'll try full key later - latency_type = 1; - break; + DEBUG("ceph plugin: Skipping avgcount for filestore.JournalWrBytes"); + return CEPH_CB_CONTINUE; } - strncat(key, ".", 1); - strncat(key, yajl->state[i].key, yajl->state[i].key_len+1); + } + else /* not a latency type */ + { + BUFFER_ADD (key, "."); + BUFFER_ADD (key, state->key); } - result = yajl->handler(yajl->handler_arg, buffer, key); - - if((result == RETRY_AVGCOUNT) && latency_type) + status = state->handler(state->handler_arg, buffer, key); + if((status == RETRY_AVGCOUNT) && latency_type) { - strncat(key, ".", 1); - strncat(key, yajl->state[yajl->depth-1].key, - yajl->state[yajl->depth-1].key_len+1); - result = yajl->handler(yajl->handler_arg, buffer, key); + /* Add previously skipped part of the key, either "avgcount" or "sum", + * and try again. */ + BUFFER_ADD (key, "."); + BUFFER_ADD (key, state->key); + + status = state->handler(state->handler_arg, buffer, key); } - if(result == -ENOMEM) + if (status != 0) { - ERROR("ceph plugin: memory allocation failed"); + ERROR("ceph plugin: JSON handler failed with status %d.", status); return CEPH_CB_ABORT; } - yajl->depth = (yajl->depth - 1); return CEPH_CB_CONTINUE; } @@ -330,37 +352,52 @@ static int ceph_cb_string(void *ctx, const unsigned char *string_val, static int ceph_cb_start_map(void *ctx) { + yajl_struct *state = (yajl_struct*) ctx; + + /* Push key to the stack */ + if (state->depth == YAJL_MAX_DEPTH) + return CEPH_CB_ABORT; + + state->stack[state->depth] = state->key; + state->depth++; + state->key = NULL; + return CEPH_CB_CONTINUE; } -static int -ceph_cb_map_key(void *ctx, const unsigned char *key, yajl_len_t string_len) +static int ceph_cb_end_map(void *ctx) { - yajl_struct *yajl = (yajl_struct*)ctx; + yajl_struct *state = (yajl_struct*) ctx; - if((yajl->depth+1) >= YAJL_MAX_DEPTH) - { - ERROR("ceph plugin: depth exceeds max, aborting."); + /* Pop key from the stack */ + if (state->depth == 0) return CEPH_CB_ABORT; - } - - char buffer[string_len+1]; - - memcpy(buffer, key, string_len); - buffer[sizeof(buffer) - 1] = 0; - snprintf(yajl->state[yajl->depth].key, sizeof(buffer), "%s", buffer); - yajl->state[yajl->depth].key_len = sizeof(buffer); - yajl->depth = (yajl->depth + 1); + sfree (state->key); + state->depth--; + state->key = state->stack[state->depth]; + state->stack[state->depth] = NULL; return CEPH_CB_CONTINUE; } -static int ceph_cb_end_map(void *ctx) +static int +ceph_cb_map_key(void *ctx, const unsigned char *key, yajl_len_t string_len) { - yajl_struct *yajl = (yajl_struct*)ctx; + yajl_struct *state = (yajl_struct*) ctx; + size_t sz = ((size_t) string_len) + 1; + + sfree (state->key); + state->key = malloc (sz); + if (state->key == NULL) + { + ERROR ("ceph plugin: malloc failed."); + return CEPH_CB_ABORT; + } + + memmove (state->key, key, sz - 1); + state->key[sz - 1] = 0; - yajl->depth = (yajl->depth - 1); return CEPH_CB_CONTINUE; } @@ -395,8 +432,7 @@ static void ceph_daemon_print(const struct ceph_daemon *d) static void ceph_daemons_print(void) { - int i; - for(i = 0; i < g_num_daemons; ++i) + for(size_t i = 0; i < g_num_daemons; ++i) { ceph_daemon_print(g_daemons[i]); } @@ -404,15 +440,15 @@ static void ceph_daemons_print(void) static void ceph_daemon_free(struct ceph_daemon *d) { - int i = 0; - for(; i < d->last_idx; i++) + for(int i = 0; i < d->last_idx; i++) { sfree(d->last_poll_data[i]); } sfree(d->last_poll_data); d->last_poll_data = NULL; d->last_idx = 0; - for(i = 0; i < d->ds_num; i++) + + for(int i = 0; i < d->ds_num; i++) { sfree(d->ds_names[i]); } @@ -521,10 +557,9 @@ static _Bool has_suffix (char const *str, char const *suffix) /* count_parts returns the number of elements a "foo.bar.baz" style key has. */ static size_t count_parts (char const *key) { - char const *ptr; size_t parts_num = 0; - for (ptr = key; ptr != NULL; ptr = strchr (ptr + 1, '.')) + for (const char *ptr = key; ptr != NULL; ptr = strchr (ptr + 1, '.')) parts_num++; return parts_num; @@ -566,7 +601,6 @@ static int ceph_daemon_add_ds_entry(struct ceph_daemon *d, const char *name, { uint32_t type; char ds_name[DATA_MAX_NAME_LEN]; - memset(ds_name, 0, sizeof(ds_name)); if(convert_special_metrics) { @@ -596,7 +630,7 @@ static int ceph_daemon_add_ds_entry(struct ceph_daemon *d, const char *name, return -ENOMEM; } - d->ds_names[d->ds_num] = malloc(sizeof(char) * DATA_MAX_NAME_LEN); + d->ds_names[d->ds_num] = malloc(DATA_MAX_NAME_LEN); if(!d->ds_names[d->ds_num]) { return -ENOMEM; @@ -657,10 +691,9 @@ static int cc_handle_bool(struct oconfig_item_s *item, int *dest) static int cc_add_daemon_config(oconfig_item_t *ci) { - int ret, i; - struct ceph_daemon *nd, cd; + int ret; + struct ceph_daemon *nd, cd = { 0 }; struct ceph_daemon **tmp; - memset(&cd, 0, sizeof(struct ceph_daemon)); if((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING)) { @@ -675,7 +708,7 @@ static int cc_add_daemon_config(oconfig_item_t *ci) return ret; } - for(i=0; i < ci->children_num; i++) + for(int i=0; i < ci->children_num; i++) { oconfig_item_t *child = ci->children + i; @@ -720,21 +753,22 @@ static int cc_add_daemon_config(oconfig_item_t *ci) } g_daemons = tmp; - nd = malloc(sizeof(*nd)); + nd = malloc(sizeof (*nd)); if(!nd) { return ENOMEM; } memcpy(nd, &cd, sizeof(*nd)); - g_daemons[g_num_daemons++] = nd; + g_daemons[g_num_daemons] = nd; + g_num_daemons++; return 0; } static int ceph_config(oconfig_item_t *ci) { - int ret, i; + int ret; - for(i = 0; i < ci->children_num; ++i) + for(int i = 0; i < ci->children_num; ++i) { oconfig_item_t *child = ci->children + i; if(strcasecmp("Daemon", child->key) == 0) @@ -818,7 +852,7 @@ node_handler_define_schema(void *arg, const char *val, const char *key) static int add_last(struct ceph_daemon *d, const char *ds_n, double cur_sum, uint64_t cur_count) { - d->last_poll_data[d->last_idx] = malloc(1 * sizeof(struct last_data)); + d->last_poll_data[d->last_idx] = malloc(sizeof (*d->last_poll_data[d->last_idx])); if(!d->last_poll_data[d->last_idx]) { return -ENOMEM; @@ -846,7 +880,7 @@ static int update_last(struct ceph_daemon *d, const char *ds_n, int index, if(!d->last_poll_data) { - d->last_poll_data = malloc(1 * sizeof(struct last_data *)); + d->last_poll_data = malloc(sizeof (*d->last_poll_data)); if(!d->last_poll_data) { return -ENOMEM; @@ -871,8 +905,7 @@ static int update_last(struct ceph_daemon *d, const char *ds_n, int index, */ static int backup_search_for_last_avg(struct ceph_daemon *d, const char *ds_n) { - int i = 0; - for(; i < d->last_idx; i++) + for(int i = 0; i < d->last_idx; i++) { if(strcmp(d->last_poll_data[i]->ds_name, ds_n) == 0) { @@ -932,12 +965,11 @@ static double get_last_avg(struct ceph_daemon *d, const char *ds_n, int index, */ static uint32_t backup_search_for_type(struct ceph_daemon *d, char *ds_name) { - int idx = 0; - for(; idx < d->ds_num; idx++) + for(int i = 0; i < d->ds_num; i++) { - if(strcmp(d->ds_names[idx], ds_name) == 0) + if(strcmp(d->ds_names[i], ds_name) == 0) { - return d->ds_types[idx]; + return d->ds_types[i]; } } return DSET_TYPE_UNFOUND; @@ -956,7 +988,6 @@ static int node_handler_fetch_data(void *arg, const char *val, const char *key) int index = vtmp->index; char ds_name[DATA_MAX_NAME_LEN]; - memset(ds_name, 0, sizeof(ds_name)); if (parse_keys (ds_name, sizeof (ds_name), key)) { @@ -1059,7 +1090,7 @@ static int node_handler_fetch_data(void *arg, const char *val, const char *key) static int cconn_connect(struct cconn *io) { - struct sockaddr_un address; + struct sockaddr_un address = { 0 }; int flags, fd, err; if(io->state != CSTATE_UNCONNECTED) { @@ -1069,12 +1100,11 @@ static int cconn_connect(struct cconn *io) fd = socket(PF_UNIX, SOCK_STREAM, 0); if(fd < 0) { - int err = -errno; + err = -errno; ERROR("ceph plugin: cconn_connect: socket(PF_UNIX, SOCK_STREAM, 0) " "failed: error %d", err); return err; } - memset(&address, 0, sizeof(struct sockaddr_un)); address.sun_family = AF_UNIX; snprintf(address.sun_path, sizeof(address.sun_path), "%s", io->d->asok_path); @@ -1434,19 +1464,26 @@ static int milli_diff(const struct timeval *t1, const struct timeval *t2) */ static int cconn_main_loop(uint32_t request_type) { - int i, ret, some_unreachable = 0; + int ret, some_unreachable = 0; struct timeval end_tv; struct cconn io_array[g_num_daemons]; - DEBUG("ceph plugin: entering cconn_main_loop(request_type = %d)", request_type); + DEBUG ("ceph plugin: entering cconn_main_loop(request_type = %"PRIu32")", request_type); + + if (g_num_daemons < 1) + { + ERROR ("ceph plugin: No daemons configured. See the \"Daemon\" config option."); + return ENOENT; + } /* create cconn array */ - memset(io_array, 0, sizeof(io_array)); - for(i = 0; i < g_num_daemons; ++i) + for (size_t i = 0; i < g_num_daemons; i++) { - io_array[i].d = g_daemons[i]; - io_array[i].request_type = request_type; - io_array[i].state = CSTATE_UNCONNECTED; + io_array[i] = (struct cconn) { + .d = g_daemons[i], + .request_type = request_type, + .state = CSTATE_UNCONNECTED, + }; } /** Calculate the time at which we should give up */ @@ -1461,13 +1498,13 @@ static int cconn_main_loop(uint32_t request_type) struct pollfd fds[g_num_daemons]; memset(fds, 0, sizeof(fds)); nfds = 0; - for(i = 0; i < g_num_daemons; ++i) + for(size_t i = 0; i < g_num_daemons; ++i) { struct cconn *io = io_array + i; ret = cconn_prepare(io, fds + nfds); if(ret < 0) { - WARNING("ceph plugin: cconn_prepare(name=%s,i=%d,st=%d)=%d", + WARNING("ceph plugin: cconn_prepare(name=%s,i=%zu,st=%d)=%d", io->d->name, i, io->state, ret); cconn_close(io); io->request_type = ASOK_REQ_NONE; @@ -1499,13 +1536,14 @@ static int cconn_main_loop(uint32_t request_type) ERROR("ceph plugin: poll(2) error: %d", ret); goto done; } - for(i = 0; i < nfds; ++i) + for(int i = 0; i < nfds; ++i) { struct cconn *io = polled_io_array[i]; int revents = fds[i].revents; if(revents == 0) { /* do nothing */ + continue; } else if(cconn_validate_revents(io, revents)) { @@ -1518,7 +1556,7 @@ static int cconn_main_loop(uint32_t request_type) } else { - int ret = cconn_handle_event(io); + ret = cconn_handle_event(io); if(ret) { WARNING("ceph plugin: cconn_handle_event(name=%s," @@ -1530,7 +1568,7 @@ static int cconn_main_loop(uint32_t request_type) } } } - done: for(i = 0; i < g_num_daemons; ++i) + done: for(size_t i = 0; i < g_num_daemons; ++i) { cconn_close(io_array + i); } @@ -1553,18 +1591,35 @@ static int ceph_read(void) /******* lifecycle *******/ static int ceph_init(void) { - int ret; +#if defined(HAVE_SYS_CAPABILITY_H) && defined(CAP_DAC_OVERRIDE) + if (check_capability (CAP_DAC_OVERRIDE) != 0) + { + if (getuid () == 0) + WARNING ("ceph plugin: Running collectd as root, but the " + "CAP_DAC_OVERRIDE capability is missing. The plugin's read " + "function will probably fail. Is your init system dropping " + "capabilities?"); + else + WARNING ("ceph plugin: collectd doesn't have the CAP_DAC_OVERRIDE " + "capability. If you don't want to run collectd as root, try running " + "\"setcap cap_dac_override=ep\" on the collectd binary."); + } +#endif + ceph_daemons_print(); - ret = cconn_main_loop(ASOK_REQ_VERSION); + if (g_num_daemons < 1) + { + ERROR ("ceph plugin: No daemons configured. See the \"Daemon\" config option."); + return ENOENT; + } - return (ret) ? ret : 0; + return cconn_main_loop(ASOK_REQ_VERSION); } static int ceph_shutdown(void) { - int i; - for(i = 0; i < g_num_daemons; ++i) + for(size_t i = 0; i < g_num_daemons; ++i) { ceph_daemon_free(g_daemons[i]); } @@ -1582,3 +1637,4 @@ void module_register(void) plugin_register_read("ceph", ceph_read); plugin_register_shutdown("ceph", ceph_shutdown); } +/* vim: set sw=4 sts=4 et : */