redis plugin: fix issue found by master aggregation tests
[collectd.git] / src / redis.c
1 /**
2  * collectd - src/redis.c, based on src/memcached.c
3  * Copyright (C) 2010       Andrés J. Díaz <ajdiaz@connectical.com>
4  *
5  * This program is free software; you can redistribute it and/or modify it
6  * under the terms of the GNU General Public License as published by the
7  * Free Software Foundation; either version 2 of the License, or (at your
8  * option) any later version.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License for more details.
14  *
15  * You should have received a copy of the GNU General Public License along
16  * with this program; if not, write to the Free Software Foundation, Inc.,
17  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
18  *
19  * Authors:
20  *   Andrés J. Díaz <ajdiaz@connectical.com>
21  **/
22
23 #include "collectd.h"
24
25 #include "common.h"
26 #include "plugin.h"
27
28 #include <hiredis/hiredis.h>
29 #include <sys/time.h>
30
31 #ifndef HOST_NAME_MAX
32 #define HOST_NAME_MAX _POSIX_HOST_NAME_MAX
33 #endif
34
35 #define REDIS_DEF_HOST "localhost"
36 #define REDIS_DEF_PASSWD ""
37 #define REDIS_DEF_PORT 6379
38 #define REDIS_DEF_TIMEOUT 2000
39 #define REDIS_DEF_DB_COUNT 256
40 #define MAX_REDIS_NODE_NAME 64
41 #define MAX_REDIS_PASSWD_LENGTH 512
42 #define MAX_REDIS_VAL_SIZE 256
43 #define MAX_REDIS_QUERY 2048
44
45 /* Redis plugin configuration example:
46  *
47  * <Plugin redis>
48  *   <Node "mynode">
49  *     Host "localhost"
50  *     Port "6379"
51  *     Timeout 2
52  *     Password "foobar"
53  *   </Node>
54  * </Plugin>
55  */
56
57 struct redis_query_s;
58 typedef struct redis_query_s redis_query_t;
59 struct redis_query_s {
60   char query[MAX_REDIS_QUERY];
61   char type[DATA_MAX_NAME_LEN];
62   char instance[DATA_MAX_NAME_LEN];
63   int database;
64
65   redis_query_t *next;
66 };
67
68 struct redis_node_s;
69 typedef struct redis_node_s redis_node_t;
70 struct redis_node_s {
71   char name[MAX_REDIS_NODE_NAME];
72   char host[HOST_NAME_MAX];
73   char passwd[MAX_REDIS_PASSWD_LENGTH];
74   int port;
75   struct timeval timeout;
76   redis_query_t *queries;
77
78   redis_node_t *next;
79 };
80
81 static redis_node_t *nodes_head;
82
83 static int redis_node_add(const redis_node_t *rn) /* {{{ */
84 {
85   redis_node_t *rn_copy;
86   redis_node_t *rn_ptr;
87
88   /* Check for duplicates first */
89   for (rn_ptr = nodes_head; rn_ptr != NULL; rn_ptr = rn_ptr->next)
90     if (strcmp(rn->name, rn_ptr->name) == 0)
91       break;
92
93   if (rn_ptr != NULL) {
94     ERROR("redis plugin: A node with the name `%s' already exists.", rn->name);
95     return -1;
96   }
97
98   rn_copy = malloc(sizeof(*rn_copy));
99   if (rn_copy == NULL) {
100     ERROR("redis plugin: malloc failed adding redis_node to the tree.");
101     return -1;
102   }
103
104   memcpy(rn_copy, rn, sizeof(*rn_copy));
105   rn_copy->next = NULL;
106
107   DEBUG("redis plugin: Adding node \"%s\".", rn->name);
108
109   if (nodes_head == NULL)
110     nodes_head = rn_copy;
111   else {
112     rn_ptr = nodes_head;
113     while (rn_ptr->next != NULL)
114       rn_ptr = rn_ptr->next;
115     rn_ptr->next = rn_copy;
116   }
117
118   return 0;
119 } /* }}} */
120
121 static redis_query_t *redis_config_query(oconfig_item_t *ci) /* {{{ */
122 {
123   redis_query_t *rq;
124   int status;
125
126   rq = calloc(1, sizeof(*rq));
127   if (rq == NULL) {
128     ERROR("redis plugin: calloc failed adding redis_query.");
129     return NULL;
130   }
131   status = cf_util_get_string_buffer(ci, rq->query, sizeof(rq->query));
132   if (status != 0)
133     goto err;
134
135   /*
136    * Default to a gauge type.
137    */
138   (void)strncpy(rq->type, "gauge", sizeof(rq->type));
139   (void)sstrncpy(rq->instance, rq->query, sizeof(rq->instance));
140   replace_special(rq->instance, sizeof(rq->instance));
141
142   rq->database = 0;
143
144   for (int i = 0; i < ci->children_num; i++) {
145     oconfig_item_t *option = ci->children + i;
146
147     if (strcasecmp("Type", option->key) == 0) {
148       status = cf_util_get_string_buffer(option, rq->type, sizeof(rq->type));
149     } else if (strcasecmp("Instance", option->key) == 0) {
150       status =
151           cf_util_get_string_buffer(option, rq->instance, sizeof(rq->instance));
152     } else if (strcasecmp("Database", option->key) == 0) {
153       status = cf_util_get_int(option, &rq->database);
154       if (rq->database < 0) {
155         WARNING("redis plugin: The \"Database\" option must be positive "
156                 "integer or zero");
157         status = -1;
158       }
159     } else {
160       WARNING("redis plugin: unknown configuration option: %s", option->key);
161       status = -1;
162     }
163     if (status != 0)
164       goto err;
165   }
166   return rq;
167 err:
168   free(rq);
169   return NULL;
170 } /* }}} */
171
172 static int redis_config_node(oconfig_item_t *ci) /* {{{ */
173 {
174   redis_query_t *rq;
175   int status;
176   int timeout;
177
178   redis_node_t rn = {.port = REDIS_DEF_PORT,
179                      .timeout.tv_usec = REDIS_DEF_TIMEOUT};
180
181   sstrncpy(rn.host, REDIS_DEF_HOST, sizeof(rn.host));
182
183   status = cf_util_get_string_buffer(ci, rn.name, sizeof(rn.name));
184   if (status != 0)
185     return status;
186
187   for (int i = 0; i < ci->children_num; i++) {
188     oconfig_item_t *option = ci->children + i;
189
190     if (strcasecmp("Host", option->key) == 0)
191       status = cf_util_get_string_buffer(option, rn.host, sizeof(rn.host));
192     else if (strcasecmp("Port", option->key) == 0) {
193       status = cf_util_get_port_number(option);
194       if (status > 0) {
195         rn.port = status;
196         status = 0;
197       }
198     } else if (strcasecmp("Query", option->key) == 0) {
199       rq = redis_config_query(option);
200       if (rq == NULL) {
201         status = 1;
202       } else {
203         rq->next = rn.queries;
204         rn.queries = rq;
205       }
206     } else if (strcasecmp("Timeout", option->key) == 0) {
207       status = cf_util_get_int(option, &timeout);
208       if (status == 0)
209         rn.timeout.tv_usec = timeout;
210     } else if (strcasecmp("Password", option->key) == 0)
211       status = cf_util_get_string_buffer(option, rn.passwd, sizeof(rn.passwd));
212     else
213       WARNING("redis plugin: Option `%s' not allowed inside a `Node' "
214               "block. I'll ignore this option.",
215               option->key);
216
217     if (status != 0)
218       break;
219   }
220
221   if (status != 0)
222     return status;
223
224   return redis_node_add(&rn);
225 } /* }}} int redis_config_node */
226
227 static int redis_config(oconfig_item_t *ci) /* {{{ */
228 {
229   for (int i = 0; i < ci->children_num; i++) {
230     oconfig_item_t *option = ci->children + i;
231
232     if (strcasecmp("Node", option->key) == 0)
233       redis_config_node(option);
234     else
235       WARNING("redis plugin: Option `%s' not allowed in redis"
236               " configuration. It will be ignored.",
237               option->key);
238   }
239
240   if (nodes_head == NULL) {
241     ERROR("redis plugin: No valid node configuration could be found.");
242     return ENOENT;
243   }
244
245   return 0;
246 } /* }}} */
247
248 __attribute__((nonnull(2))) static void
249 redis_submit(char *plugin_instance, const char *type, const char *type_instance,
250              value_t value) /* {{{ */
251 {
252   value_list_t vl = VALUE_LIST_INIT;
253
254   vl.values = &value;
255   vl.values_len = 1;
256   sstrncpy(vl.plugin, "redis", sizeof(vl.plugin));
257   if (plugin_instance != NULL)
258     sstrncpy(vl.plugin_instance, plugin_instance, sizeof(vl.plugin_instance));
259   sstrncpy(vl.type, type, sizeof(vl.type));
260   if (type_instance != NULL)
261     sstrncpy(vl.type_instance, type_instance, sizeof(vl.type_instance));
262
263   plugin_dispatch_values(&vl);
264 } /* }}} */
265
266 static int redis_init(void) /* {{{ */
267 {
268   redis_node_t rn = {.name = "default",
269                      .host = REDIS_DEF_HOST,
270                      .port = REDIS_DEF_PORT,
271                      .timeout.tv_sec = 0,
272                      .timeout.tv_usec = REDIS_DEF_TIMEOUT,
273                      .next = NULL};
274
275   if (nodes_head == NULL)
276     redis_node_add(&rn);
277
278   return 0;
279 } /* }}} int redis_init */
280
281 static int redis_handle_info(char *node, char const *info_line,
282                              char const *type, char const *type_instance,
283                              char const *field_name, int ds_type) /* {{{ */
284 {
285   char *str = strstr(info_line, field_name);
286   static char buf[MAX_REDIS_VAL_SIZE];
287   value_t val;
288   if (str) {
289     int i;
290
291     str += strlen(field_name) + 1; /* also skip the ':' */
292     for (i = 0; (*str && (isdigit((unsigned char)*str) || *str == '.'));
293          i++, str++)
294       buf[i] = *str;
295     buf[i] = '\0';
296
297     if (parse_value(buf, &val, ds_type) == -1) {
298       WARNING("redis plugin: Unable to parse field `%s'.", field_name);
299       return -1;
300     }
301
302     redis_submit(node, type, type_instance, val);
303     return 0;
304   }
305   return -1;
306
307 } /* }}} int redis_handle_info */
308
309 static int redis_handle_query(redisContext *rh, redis_node_t *rn,
310                               redis_query_t *rq) /* {{{ */
311 {
312   redisReply *rr;
313   const data_set_t *ds;
314   value_t val;
315
316   ds = plugin_get_ds(rq->type);
317   if (!ds) {
318     ERROR("redis plugin: DataSet `%s' not defined.", rq->type);
319     return -1;
320   }
321
322   if (ds->ds_num != 1) {
323     ERROR("redis plugin: DS `%s' has too many types.", rq->type);
324     return -1;
325   }
326
327   if ((rr = redisCommand(rh, "SELECT %d", rq->database)) == NULL) {
328     WARNING("redis plugin: unable to switch to database `%d' on node `%s'.",
329             rq->database, rn->name);
330     return -1;
331   }
332
333   if ((rr = redisCommand(rh, rq->query)) == NULL) {
334     WARNING("redis plugin: unable to carry out query `%s'.", rq->query);
335     return -1;
336   }
337
338   switch (rr->type) {
339   case REDIS_REPLY_INTEGER:
340     switch (ds->ds[0].type) {
341     case DS_TYPE_COUNTER:
342       val.counter = (counter_t)rr->integer;
343       break;
344     case DS_TYPE_GAUGE:
345       val.gauge = (gauge_t)rr->integer;
346       break;
347     case DS_TYPE_DERIVE:
348       val.gauge = (derive_t)rr->integer;
349       break;
350     case DS_TYPE_ABSOLUTE:
351       val.gauge = (absolute_t)rr->integer;
352       break;
353     }
354     break;
355   case REDIS_REPLY_STRING:
356     if (parse_value(rr->str, &val, ds->ds[0].type) == -1) {
357       WARNING("redis plugin: Unable to parse field `%s'.", rq->type);
358       freeReplyObject(rr);
359       return -1;
360     }
361     break;
362   default:
363     WARNING("redis plugin: Cannot coerce redis type.");
364     freeReplyObject(rr);
365     return -1;
366   }
367
368   redis_submit(rn->name, rq->type,
369                (strlen(rq->instance) > 0) ? rq->instance : NULL, val);
370   freeReplyObject(rr);
371   return 0;
372 } /* }}} int redis_handle_query */
373
374 static int redis_db_stats(char *node, char const *info_line) /* {{{ */
375 {
376   /* redis_db_stats parses and dispatches Redis database statistics,
377    * currently the number of keys for each database.
378    * info_line needs to have the following format:
379    *   db0:keys=4,expires=0,avg_ttl=0
380    */
381
382   for (int db = 0; db < REDIS_DEF_DB_COUNT; db++) {
383     static char buf[MAX_REDIS_VAL_SIZE];
384     static char field_name[12];
385     static char db_id[4];
386     value_t val;
387     char *str;
388     int i;
389
390     snprintf(field_name, sizeof(field_name), "db%d:keys=", db);
391
392     str = strstr(info_line, field_name);
393     if (!str)
394       continue;
395
396     str += strlen(field_name);
397     for (i = 0; (*str && isdigit((int)*str)); i++, str++)
398       buf[i] = *str;
399     buf[i] = '\0';
400
401     if (parse_value(buf, &val, DS_TYPE_GAUGE) != 0) {
402       WARNING("redis plugin: Unable to parse field `%s'.", field_name);
403       return -1;
404     }
405
406     snprintf(db_id, sizeof(db_id), "%d", db);
407     redis_submit(node, "records", db_id, val);
408   }
409   return 0;
410
411 } /* }}} int redis_db_stats */
412
413 static int redis_read(void) /* {{{ */
414 {
415   for (redis_node_t *rn = nodes_head; rn != NULL; rn = rn->next) {
416     redisContext *rh;
417     redisReply *rr;
418
419     DEBUG("redis plugin: querying info from node `%s' (%s:%d).", rn->name,
420           rn->host, rn->port);
421
422     rh = redisConnectWithTimeout((char *)rn->host, rn->port, rn->timeout);
423     if (rh == NULL) {
424       ERROR("redis plugin: unable to connect to node `%s' (%s:%d).", rn->name,
425             rn->host, rn->port);
426       continue;
427     }
428
429     if (strlen(rn->passwd) > 0) {
430       DEBUG("redis plugin: authenticating node `%s' passwd(%s).", rn->name,
431             rn->passwd);
432
433       if ((rr = redisCommand(rh, "AUTH %s", rn->passwd)) == NULL) {
434         WARNING("redis plugin: unable to authenticate on node `%s'.", rn->name);
435         goto redis_fail;
436       }
437
438       if (rr->type != REDIS_REPLY_STATUS) {
439         WARNING("redis plugin: invalid authentication on node `%s'.", rn->name);
440         goto redis_fail;
441       }
442
443       freeReplyObject(rr);
444     }
445
446     if ((rr = redisCommand(rh, "INFO")) == NULL) {
447       WARNING("redis plugin: unable to get info from node `%s'.", rn->name);
448       goto redis_fail;
449     }
450
451     redis_handle_info(rn->name, rr->str, "uptime", NULL, "uptime_in_seconds",
452                       DS_TYPE_GAUGE);
453     redis_handle_info(rn->name, rr->str, "current_connections", "clients",
454                       "connected_clients", DS_TYPE_GAUGE);
455     redis_handle_info(rn->name, rr->str, "blocked_clients", NULL,
456                       "blocked_clients", DS_TYPE_GAUGE);
457     redis_handle_info(rn->name, rr->str, "memory", NULL, "used_memory",
458                       DS_TYPE_GAUGE);
459     redis_handle_info(rn->name, rr->str, "memory_lua", NULL, "used_memory_lua",
460                       DS_TYPE_GAUGE);
461     /* changes_since_last_save: Deprecated in redis version 2.6 and above */
462     redis_handle_info(rn->name, rr->str, "volatile_changes", NULL,
463                       "changes_since_last_save", DS_TYPE_GAUGE);
464     redis_handle_info(rn->name, rr->str, "total_connections", NULL,
465                       "total_connections_received", DS_TYPE_DERIVE);
466     redis_handle_info(rn->name, rr->str, "total_operations", NULL,
467                       "total_commands_processed", DS_TYPE_DERIVE);
468     redis_handle_info(rn->name, rr->str, "operations_per_second", NULL,
469                       "instantaneous_ops_per_sec", DS_TYPE_GAUGE);
470     redis_handle_info(rn->name, rr->str, "expired_keys", NULL, "expired_keys",
471                       DS_TYPE_DERIVE);
472     redis_handle_info(rn->name, rr->str, "evicted_keys", NULL, "evicted_keys",
473                       DS_TYPE_DERIVE);
474     redis_handle_info(rn->name, rr->str, "pubsub", "channels",
475                       "pubsub_channels", DS_TYPE_GAUGE);
476     redis_handle_info(rn->name, rr->str, "pubsub", "patterns",
477                       "pubsub_patterns", DS_TYPE_GAUGE);
478     redis_handle_info(rn->name, rr->str, "current_connections", "slaves",
479                       "connected_slaves", DS_TYPE_GAUGE);
480     redis_handle_info(rn->name, rr->str, "cache_result", "hits",
481                       "keyspace_hits", DS_TYPE_DERIVE);
482     redis_handle_info(rn->name, rr->str, "cache_result", "misses",
483                       "keyspace_misses", DS_TYPE_DERIVE);
484     redis_handle_info(rn->name, rr->str, "total_bytes", "input",
485                       "total_net_input_bytes", DS_TYPE_DERIVE);
486     redis_handle_info(rn->name, rr->str, "total_bytes", "output",
487                       "total_net_output_bytes", DS_TYPE_DERIVE);
488
489     redis_db_stats(rn->name, rr->str);
490
491     for (redis_query_t *rq = rn->queries; rq != NULL; rq = rq->next)
492       redis_handle_query(rh, rn, rq);
493
494   redis_fail:
495     if (rr != NULL)
496       freeReplyObject(rr);
497     redisFree(rh);
498   }
499
500   return 0;
501 }
502 /* }}} */
503
504 void module_register(void) /* {{{ */
505 {
506   plugin_register_complex_config("redis", redis_config);
507   plugin_register_init("redis", redis_init);
508   plugin_register_read("redis", redis_read);
509   /* TODO: plugin_register_write: one redis list per value id with
510    * X elements */
511 }
512 /* }}} */