Merge branch 'collectd-5.7' into collectd-5.8
[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 16
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   redis_query_t *next;
64 };
65
66 struct redis_node_s;
67 typedef struct redis_node_s redis_node_t;
68 struct redis_node_s {
69   char name[MAX_REDIS_NODE_NAME];
70   char host[HOST_NAME_MAX];
71   char passwd[MAX_REDIS_PASSWD_LENGTH];
72   int port;
73   struct timeval timeout;
74   redis_query_t *queries;
75
76   redis_node_t *next;
77 };
78
79 static redis_node_t *nodes_head = NULL;
80
81 static int redis_node_add(const redis_node_t *rn) /* {{{ */
82 {
83   redis_node_t *rn_copy;
84   redis_node_t *rn_ptr;
85
86   /* Check for duplicates first */
87   for (rn_ptr = nodes_head; rn_ptr != NULL; rn_ptr = rn_ptr->next)
88     if (strcmp(rn->name, rn_ptr->name) == 0)
89       break;
90
91   if (rn_ptr != NULL) {
92     ERROR("redis plugin: A node with the name `%s' already exists.", rn->name);
93     return -1;
94   }
95
96   rn_copy = malloc(sizeof(*rn_copy));
97   if (rn_copy == NULL) {
98     ERROR("redis plugin: malloc failed adding redis_node to the tree.");
99     return -1;
100   }
101
102   memcpy(rn_copy, rn, sizeof(*rn_copy));
103   rn_copy->next = NULL;
104
105   DEBUG("redis plugin: Adding node \"%s\".", rn->name);
106
107   if (nodes_head == NULL)
108     nodes_head = rn_copy;
109   else {
110     rn_ptr = nodes_head;
111     while (rn_ptr->next != NULL)
112       rn_ptr = rn_ptr->next;
113     rn_ptr->next = rn_copy;
114   }
115
116   return 0;
117 } /* }}} */
118
119 static redis_query_t *redis_config_query(oconfig_item_t *ci) /* {{{ */
120 {
121   redis_query_t *rq;
122   int status;
123
124   rq = calloc(1, sizeof(*rq));
125   if (rq == NULL) {
126     ERROR("redis plugin: calloc failed adding redis_query.");
127     return NULL;
128   }
129   status = cf_util_get_string_buffer(ci, rq->query, sizeof(rq->query));
130   if (status != 0)
131     goto err;
132
133   /*
134    * Default to a gauge type.
135    */
136   (void)strncpy(rq->type, "gauge", sizeof(rq->type));
137   (void)sstrncpy(rq->instance, rq->query, sizeof(rq->instance));
138   replace_special(rq->instance, sizeof(rq->instance));
139
140   for (int i = 0; i < ci->children_num; i++) {
141     oconfig_item_t *option = ci->children + i;
142
143     if (strcasecmp("Type", option->key) == 0) {
144       status = cf_util_get_string_buffer(option, rq->type, sizeof(rq->type));
145     } else if (strcasecmp("Instance", option->key) == 0) {
146       status =
147           cf_util_get_string_buffer(option, rq->instance, sizeof(rq->instance));
148     } else {
149       WARNING("redis plugin: unknown configuration option: %s", option->key);
150       status = -1;
151     }
152     if (status != 0)
153       goto err;
154   }
155   return rq;
156 err:
157   free(rq);
158   return NULL;
159 } /* }}} */
160
161 static int redis_config_node(oconfig_item_t *ci) /* {{{ */
162 {
163   redis_query_t *rq;
164   int status;
165   int timeout;
166
167   redis_node_t rn = {.port = REDIS_DEF_PORT,
168                      .timeout.tv_usec = REDIS_DEF_TIMEOUT};
169
170   sstrncpy(rn.host, REDIS_DEF_HOST, sizeof(rn.host));
171
172   status = cf_util_get_string_buffer(ci, rn.name, sizeof(rn.name));
173   if (status != 0)
174     return status;
175
176   for (int i = 0; i < ci->children_num; i++) {
177     oconfig_item_t *option = ci->children + i;
178
179     if (strcasecmp("Host", option->key) == 0)
180       status = cf_util_get_string_buffer(option, rn.host, sizeof(rn.host));
181     else if (strcasecmp("Port", option->key) == 0) {
182       status = cf_util_get_port_number(option);
183       if (status > 0) {
184         rn.port = status;
185         status = 0;
186       }
187     } else if (strcasecmp("Query", option->key) == 0) {
188       rq = redis_config_query(option);
189       if (rq == NULL) {
190         status = 1;
191       } else {
192         rq->next = rn.queries;
193         rn.queries = rq;
194       }
195     } else if (strcasecmp("Timeout", option->key) == 0) {
196       status = cf_util_get_int(option, &timeout);
197       if (status == 0)
198         rn.timeout.tv_usec = timeout;
199     } else if (strcasecmp("Password", option->key) == 0)
200       status = cf_util_get_string_buffer(option, rn.passwd, sizeof(rn.passwd));
201     else
202       WARNING("redis plugin: Option `%s' not allowed inside a `Node' "
203               "block. I'll ignore this option.",
204               option->key);
205
206     if (status != 0)
207       break;
208   }
209
210   if (status != 0)
211     return status;
212
213   return redis_node_add(&rn);
214 } /* }}} int redis_config_node */
215
216 static int redis_config(oconfig_item_t *ci) /* {{{ */
217 {
218   for (int i = 0; i < ci->children_num; i++) {
219     oconfig_item_t *option = ci->children + i;
220
221     if (strcasecmp("Node", option->key) == 0)
222       redis_config_node(option);
223     else
224       WARNING("redis plugin: Option `%s' not allowed in redis"
225               " configuration. It will be ignored.",
226               option->key);
227   }
228
229   if (nodes_head == NULL) {
230     ERROR("redis plugin: No valid node configuration could be found.");
231     return ENOENT;
232   }
233
234   return 0;
235 } /* }}} */
236
237 __attribute__((nonnull(2))) static void
238 redis_submit(char *plugin_instance, const char *type, const char *type_instance,
239              value_t value) /* {{{ */
240 {
241   value_list_t vl = VALUE_LIST_INIT;
242
243   vl.values = &value;
244   vl.values_len = 1;
245   sstrncpy(vl.plugin, "redis", sizeof(vl.plugin));
246   if (plugin_instance != NULL)
247     sstrncpy(vl.plugin_instance, plugin_instance, sizeof(vl.plugin_instance));
248   sstrncpy(vl.type, type, sizeof(vl.type));
249   if (type_instance != NULL)
250     sstrncpy(vl.type_instance, type_instance, sizeof(vl.type_instance));
251
252   plugin_dispatch_values(&vl);
253 } /* }}} */
254
255 static int redis_init(void) /* {{{ */
256 {
257   redis_node_t rn = {.name = "default",
258                      .host = REDIS_DEF_HOST,
259                      .port = REDIS_DEF_PORT,
260                      .timeout.tv_sec = 0,
261                      .timeout.tv_usec = REDIS_DEF_TIMEOUT,
262                      .next = NULL};
263
264   if (nodes_head == NULL)
265     redis_node_add(&rn);
266
267   return 0;
268 } /* }}} int redis_init */
269
270 static int redis_handle_info(char *node, char const *info_line,
271                              char const *type, char const *type_instance,
272                              char const *field_name, int ds_type) /* {{{ */
273 {
274   char *str = strstr(info_line, field_name);
275   static char buf[MAX_REDIS_VAL_SIZE];
276   value_t val;
277   if (str) {
278     int i;
279
280     str += strlen(field_name) + 1; /* also skip the ':' */
281     for (i = 0; (*str && (isdigit((unsigned char)*str) || *str == '.'));
282          i++, str++)
283       buf[i] = *str;
284     buf[i] = '\0';
285
286     if (parse_value(buf, &val, ds_type) == -1) {
287       WARNING("redis plugin: Unable to parse field `%s'.", field_name);
288       return -1;
289     }
290
291     redis_submit(node, type, type_instance, val);
292     return 0;
293   }
294   return -1;
295
296 } /* }}} int redis_handle_info */
297
298 static int redis_handle_query(redisContext *rh, redis_node_t *rn,
299                               redis_query_t *rq) /* {{{ */
300 {
301   redisReply *rr;
302   const data_set_t *ds;
303   value_t val;
304
305   ds = plugin_get_ds(rq->type);
306   if (!ds) {
307     ERROR("redis plugin: DataSet `%s' not defined.", rq->type);
308     return -1;
309   }
310
311   if (ds->ds_num != 1) {
312     ERROR("redis plugin: DS `%s' has too many types.", rq->type);
313     return -1;
314   }
315
316   if ((rr = redisCommand(rh, rq->query)) == NULL) {
317     WARNING("redis plugin: unable to carry out query `%s'.", rq->query);
318     return -1;
319   }
320
321   switch (rr->type) {
322   case REDIS_REPLY_INTEGER:
323     switch (ds->ds[0].type) {
324     case DS_TYPE_COUNTER:
325       val.counter = (counter_t)rr->integer;
326       break;
327     case DS_TYPE_GAUGE:
328       val.gauge = (gauge_t)rr->integer;
329       break;
330     case DS_TYPE_DERIVE:
331       val.gauge = (derive_t)rr->integer;
332       break;
333     case DS_TYPE_ABSOLUTE:
334       val.gauge = (absolute_t)rr->integer;
335       break;
336     }
337     break;
338   case REDIS_REPLY_STRING:
339     if (parse_value(rr->str, &val, ds->ds[0].type) == -1) {
340       WARNING("redis plugin: Unable to parse field `%s'.", rq->type);
341       freeReplyObject(rr);
342       return -1;
343     }
344     break;
345   default:
346     WARNING("redis plugin: Cannot coerce redis type.");
347     freeReplyObject(rr);
348     return -1;
349   }
350
351   redis_submit(rn->name, rq->type,
352                (strlen(rq->instance) > 0) ? rq->instance : NULL, val);
353   freeReplyObject(rr);
354   return 0;
355 } /* }}} int redis_handle_query */
356
357 static int redis_db_stats(char *node, char const *info_line) /* {{{ */
358 {
359   /* redis_db_stats parses and dispatches Redis database statistics,
360    * currently the number of keys for each database.
361    * info_line needs to have the following format:
362    *   db0:keys=4,expires=0,avg_ttl=0
363    */
364
365   for (int db = 0; db < REDIS_DEF_DB_COUNT; db++) {
366     static char buf[MAX_REDIS_VAL_SIZE];
367     static char field_name[11];
368     static char db_id[3];
369     value_t val;
370     char *str;
371     int i;
372
373     snprintf(field_name, sizeof(field_name), "db%d:keys=", db);
374
375     str = strstr(info_line, field_name);
376     if (!str)
377       continue;
378
379     str += strlen(field_name);
380     for (i = 0; (*str && isdigit((int)*str)); i++, str++)
381       buf[i] = *str;
382     buf[i] = '\0';
383
384     if (parse_value(buf, &val, DS_TYPE_GAUGE) != 0) {
385       WARNING("redis plugin: Unable to parse field `%s'.", field_name);
386       return -1;
387     }
388
389     snprintf(db_id, sizeof(db_id), "%d", db);
390     redis_submit(node, "records", db_id, val);
391   }
392   return 0;
393
394 } /* }}} int redis_db_stats */
395
396 static int redis_read(void) /* {{{ */
397 {
398   for (redis_node_t *rn = nodes_head; rn != NULL; rn = rn->next) {
399     redisContext *rh;
400     redisReply *rr;
401
402     DEBUG("redis plugin: querying info from node `%s' (%s:%d).", rn->name,
403           rn->host, rn->port);
404
405     rh = redisConnectWithTimeout((char *)rn->host, rn->port, rn->timeout);
406     if (rh == NULL) {
407       ERROR("redis plugin: unable to connect to node `%s' (%s:%d).", rn->name,
408             rn->host, rn->port);
409       continue;
410     }
411
412     if (strlen(rn->passwd) > 0) {
413       DEBUG("redis plugin: authenticating node `%s' passwd(%s).", rn->name,
414             rn->passwd);
415
416       if ((rr = redisCommand(rh, "AUTH %s", rn->passwd)) == NULL) {
417         WARNING("redis plugin: unable to authenticate on node `%s'.", rn->name);
418         goto redis_fail;
419       }
420
421       if (rr->type != REDIS_REPLY_STATUS) {
422         WARNING("redis plugin: invalid authentication on node `%s'.", rn->name);
423         goto redis_fail;
424       }
425
426       freeReplyObject(rr);
427     }
428
429     if ((rr = redisCommand(rh, "INFO")) == NULL) {
430       WARNING("redis plugin: unable to get info from node `%s'.", rn->name);
431       goto redis_fail;
432     }
433
434     redis_handle_info(rn->name, rr->str, "uptime", NULL, "uptime_in_seconds",
435                       DS_TYPE_GAUGE);
436     redis_handle_info(rn->name, rr->str, "current_connections", "clients",
437                       "connected_clients", DS_TYPE_GAUGE);
438     redis_handle_info(rn->name, rr->str, "blocked_clients", NULL,
439                       "blocked_clients", DS_TYPE_GAUGE);
440     redis_handle_info(rn->name, rr->str, "memory", NULL, "used_memory",
441                       DS_TYPE_GAUGE);
442     redis_handle_info(rn->name, rr->str, "memory_lua", NULL, "used_memory_lua",
443                       DS_TYPE_GAUGE);
444     /* changes_since_last_save: Deprecated in redis version 2.6 and above */
445     redis_handle_info(rn->name, rr->str, "volatile_changes", NULL,
446                       "changes_since_last_save", DS_TYPE_GAUGE);
447     redis_handle_info(rn->name, rr->str, "total_connections", NULL,
448                       "total_connections_received", DS_TYPE_DERIVE);
449     redis_handle_info(rn->name, rr->str, "total_operations", NULL,
450                       "total_commands_processed", DS_TYPE_DERIVE);
451     redis_handle_info(rn->name, rr->str, "operations_per_second", NULL,
452                       "instantaneous_ops_per_sec", DS_TYPE_GAUGE);
453     redis_handle_info(rn->name, rr->str, "expired_keys", NULL, "expired_keys",
454                       DS_TYPE_DERIVE);
455     redis_handle_info(rn->name, rr->str, "evicted_keys", NULL, "evicted_keys",
456                       DS_TYPE_DERIVE);
457     redis_handle_info(rn->name, rr->str, "pubsub", "channels",
458                       "pubsub_channels", DS_TYPE_GAUGE);
459     redis_handle_info(rn->name, rr->str, "pubsub", "patterns",
460                       "pubsub_patterns", DS_TYPE_GAUGE);
461     redis_handle_info(rn->name, rr->str, "current_connections", "slaves",
462                       "connected_slaves", DS_TYPE_GAUGE);
463     redis_handle_info(rn->name, rr->str, "cache_result", "hits",
464                       "keyspace_hits", DS_TYPE_DERIVE);
465     redis_handle_info(rn->name, rr->str, "cache_result", "misses",
466                       "keyspace_misses", DS_TYPE_DERIVE);
467     redis_handle_info(rn->name, rr->str, "total_bytes", "input",
468                       "total_net_input_bytes", DS_TYPE_DERIVE);
469     redis_handle_info(rn->name, rr->str, "total_bytes", "output",
470                       "total_net_output_bytes", DS_TYPE_DERIVE);
471
472     redis_db_stats(rn->name, rr->str);
473
474     for (redis_query_t *rq = rn->queries; rq != NULL; rq = rq->next)
475       redis_handle_query(rh, rn, rq);
476
477   redis_fail:
478     if (rr != NULL)
479       freeReplyObject(rr);
480     redisFree(rh);
481   }
482
483   return 0;
484 }
485 /* }}} */
486
487 void module_register(void) /* {{{ */
488 {
489   plugin_register_complex_config("redis", redis_config);
490   plugin_register_init("redis", redis_init);
491   plugin_register_read("redis", redis_read);
492   /* TODO: plugin_register_write: one redis list per value id with
493    * X elements */
494 }
495 /* }}} */