X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fredis.c;h=f4de380050a6770a6af75b27510983b13e8b8361;hb=015bbda11c6760d2ce5d35fa3ad0896dddbcd8c1;hp=7875c7ea7d4593e243af1667807b8d22e9543317;hpb=01d23e3f5daf016d03f82d92a76be2fe3decdca4;p=collectd.git diff --git a/src/redis.c b/src/redis.c index 7875c7ea..f4de3800 100644 --- a/src/redis.c +++ b/src/redis.c @@ -35,7 +35,8 @@ #define REDIS_DEF_HOST "localhost" #define REDIS_DEF_PASSWD "" #define REDIS_DEF_PORT 6379 -#define REDIS_DEF_TIMEOUT 2000 +#define REDIS_DEF_TIMEOUT_SEC 2 +#define REDIS_DEF_DB_COUNT 256 #define MAX_REDIS_NODE_NAME 64 #define MAX_REDIS_PASSWD_LENGTH 512 #define MAX_REDIS_VAL_SIZE 256 @@ -59,6 +60,8 @@ struct redis_query_s { char query[MAX_REDIS_QUERY]; char type[DATA_MAX_NAME_LEN]; char instance[DATA_MAX_NAME_LEN]; + int database; + redis_query_t *next; }; @@ -75,7 +78,7 @@ struct redis_node_s { redis_node_t *next; }; -static redis_node_t *nodes_head = NULL; +static redis_node_t *nodes_head; static int redis_node_add(const redis_node_t *rn) /* {{{ */ { @@ -89,13 +92,13 @@ static int redis_node_add(const redis_node_t *rn) /* {{{ */ if (rn_ptr != NULL) { ERROR("redis plugin: A node with the name `%s' already exists.", rn->name); - return (-1); + return -1; } rn_copy = malloc(sizeof(*rn_copy)); if (rn_copy == NULL) { ERROR("redis plugin: malloc failed adding redis_node to the tree."); - return (-1); + return -1; } memcpy(rn_copy, rn, sizeof(*rn_copy)); @@ -112,7 +115,7 @@ static int redis_node_add(const redis_node_t *rn) /* {{{ */ rn_ptr->next = rn_copy; } - return (0); + return 0; } /* }}} */ static redis_query_t *redis_config_query(oconfig_item_t *ci) /* {{{ */ @@ -136,6 +139,8 @@ static redis_query_t *redis_config_query(oconfig_item_t *ci) /* {{{ */ (void)sstrncpy(rq->instance, rq->query, sizeof(rq->instance)); replace_special(rq->instance, sizeof(rq->instance)); + rq->database = 0; + for (int i = 0; i < ci->children_num; i++) { oconfig_item_t *option = ci->children + i; @@ -144,6 +149,13 @@ static redis_query_t *redis_config_query(oconfig_item_t *ci) /* {{{ */ } else if (strcasecmp("Instance", option->key) == 0) { status = cf_util_get_string_buffer(option, rq->instance, sizeof(rq->instance)); + } else if (strcasecmp("Database", option->key) == 0) { + status = cf_util_get_int(option, &rq->database); + if (rq->database < 0) { + WARNING("redis plugin: The \"Database\" option must be positive " + "integer or zero"); + status = -1; + } } else { WARNING("redis plugin: unknown configuration option: %s", option->key); status = -1; @@ -164,13 +176,13 @@ static int redis_config_node(oconfig_item_t *ci) /* {{{ */ int timeout; redis_node_t rn = {.port = REDIS_DEF_PORT, - .timeout.tv_usec = REDIS_DEF_TIMEOUT}; + .timeout.tv_sec = REDIS_DEF_TIMEOUT_SEC}; sstrncpy(rn.host, REDIS_DEF_HOST, sizeof(rn.host)); status = cf_util_get_string_buffer(ci, rn.name, sizeof(rn.name)); if (status != 0) - return (status); + return status; for (int i = 0; i < ci->children_num; i++) { oconfig_item_t *option = ci->children + i; @@ -193,8 +205,11 @@ static int redis_config_node(oconfig_item_t *ci) /* {{{ */ } } else if (strcasecmp("Timeout", option->key) == 0) { status = cf_util_get_int(option, &timeout); - if (status == 0) - rn.timeout.tv_usec = timeout; + if (status == 0) { + rn.timeout.tv_usec = timeout * 1000; + rn.timeout.tv_sec = rn.timeout.tv_usec / 1000000L; + rn.timeout.tv_usec %= 1000000L; + } } else if (strcasecmp("Password", option->key) == 0) status = cf_util_get_string_buffer(option, rn.passwd, sizeof(rn.passwd)); else @@ -207,9 +222,9 @@ static int redis_config_node(oconfig_item_t *ci) /* {{{ */ } if (status != 0) - return (status); + return status; - return (redis_node_add(&rn)); + return redis_node_add(&rn); } /* }}} int redis_config_node */ static int redis_config(oconfig_item_t *ci) /* {{{ */ @@ -227,10 +242,10 @@ static int redis_config(oconfig_item_t *ci) /* {{{ */ if (nodes_head == NULL) { ERROR("redis plugin: No valid node configuration could be found."); - return (ENOENT); + return ENOENT; } - return (0); + return 0; } /* }}} */ __attribute__((nonnull(2))) static void @@ -256,14 +271,13 @@ static int redis_init(void) /* {{{ */ redis_node_t rn = {.name = "default", .host = REDIS_DEF_HOST, .port = REDIS_DEF_PORT, - .timeout.tv_sec = 0, - .timeout.tv_usec = REDIS_DEF_TIMEOUT, + .timeout.tv_sec = REDIS_DEF_TIMEOUT_SEC, .next = NULL}; if (nodes_head == NULL) redis_node_add(&rn); - return (0); + return 0; } /* }}} int redis_init */ static int redis_handle_info(char *node, char const *info_line, @@ -284,13 +298,13 @@ static int redis_handle_info(char *node, char const *info_line, if (parse_value(buf, &val, ds_type) == -1) { WARNING("redis plugin: Unable to parse field `%s'.", field_name); - return (-1); + return -1; } redis_submit(node, type, type_instance, val); - return (0); + return 0; } - return (-1); + return -1; } /* }}} int redis_handle_info */ @@ -303,18 +317,26 @@ static int redis_handle_query(redisContext *rh, redis_node_t *rn, ds = plugin_get_ds(rq->type); if (!ds) { - ERROR("redis plugin: DataSet `%s' not defined.", rq->type); - return (-1); + ERROR("redis plugin: DS type `%s' not defined.", rq->type); + return -1; } if (ds->ds_num != 1) { - ERROR("redis plugin: DS `%s' has too many types.", rq->type); - return (-1); + ERROR("redis plugin: DS type `%s' has too many datasources. This is not " + "supported currently.", + rq->type); + return -1; + } + + if ((rr = redisCommand(rh, "SELECT %d", rq->database)) == NULL) { + WARNING("redis plugin: unable to switch to database `%d' on node `%s'.", + rq->database, rn->name); + return -1; } if ((rr = redisCommand(rh, rq->query)) == NULL) { WARNING("redis plugin: unable to carry out query `%s'.", rq->query); - return (-1); + return -1; } switch (rr->type) { @@ -336,15 +358,22 @@ static int redis_handle_query(redisContext *rh, redis_node_t *rn, break; case REDIS_REPLY_STRING: if (parse_value(rr->str, &val, ds->ds[0].type) == -1) { - WARNING("redis plugin: Unable to parse field `%s'.", rq->type); + WARNING("redis plugin: Query `%s': Unable to parse value.", rq->query); freeReplyObject(rr); - return (-1); + return -1; } break; + case REDIS_REPLY_ARRAY: + WARNING("redis plugin: Query `%s' should return string or integer. Arrays " + "are not supported.", + rq->query); + freeReplyObject(rr); + return -1; default: - WARNING("redis plugin: Cannot coerce redis type."); + WARNING("redis plugin: Query `%s': Cannot coerce redis type (%i).", + rq->query, rr->type); freeReplyObject(rr); - return (-1); + return -1; } redis_submit(rn->name, rq->type, @@ -353,6 +382,45 @@ static int redis_handle_query(redisContext *rh, redis_node_t *rn, return 0; } /* }}} int redis_handle_query */ +static int redis_db_stats(char *node, char const *info_line) /* {{{ */ +{ + /* redis_db_stats parses and dispatches Redis database statistics, + * currently the number of keys for each database. + * info_line needs to have the following format: + * db0:keys=4,expires=0,avg_ttl=0 + */ + + for (int db = 0; db < REDIS_DEF_DB_COUNT; db++) { + static char buf[MAX_REDIS_VAL_SIZE]; + static char field_name[12]; + static char db_id[4]; + value_t val; + char *str; + int i; + + snprintf(field_name, sizeof(field_name), "db%d:keys=", db); + + str = strstr(info_line, field_name); + if (!str) + continue; + + str += strlen(field_name); + for (i = 0; (*str && isdigit((int)*str)); i++, str++) + buf[i] = *str; + buf[i] = '\0'; + + if (parse_value(buf, &val, DS_TYPE_GAUGE) != 0) { + WARNING("redis plugin: Unable to parse field `%s'.", field_name); + return -1; + } + + snprintf(db_id, sizeof(db_id), "%d", db); + redis_submit(node, "records", db_id, val); + } + return 0; + +} /* }}} int redis_db_stats */ + static int redis_read(void) /* {{{ */ { for (redis_node_t *rn = nodes_head; rn != NULL; rn = rn->next) { @@ -364,8 +432,13 @@ static int redis_read(void) /* {{{ */ rh = redisConnectWithTimeout((char *)rn->host, rn->port, rn->timeout); if (rh == NULL) { - ERROR("redis plugin: unable to connect to node `%s' (%s:%d).", rn->name, - rn->host, rn->port); + ERROR("redis plugin: can't allocate redis context"); + continue; + } + if (rh->err) { + ERROR("redis plugin: unable to connect to node `%s' (%s:%d): %s.", + rn->name, rn->host, rn->port, rh->errstr); + redisFree(rh); continue; } @@ -429,6 +502,8 @@ static int redis_read(void) /* {{{ */ redis_handle_info(rn->name, rr->str, "total_bytes", "output", "total_net_output_bytes", DS_TYPE_DERIVE); + redis_db_stats(rn->name, rr->str); + for (redis_query_t *rq = rn->queries; rq != NULL; rq = rq->next) redis_handle_query(rh, rn, rq); @@ -451,5 +526,3 @@ void module_register(void) /* {{{ */ * X elements */ } /* }}} */ - -/* vim: set sw=2 sts=2 et fdm=marker : */