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 **/
23 #include "collectd.h"
24 #include "common.h"
25 #include "plugin.h"
26 #include "configfile.h"
28 #include <sys/time.h>
29 #include <hiredis/hiredis.h>
31 #ifndef HOST_NAME_MAX
32 # define HOST_NAME_MAX _POSIX_HOST_NAME_MAX
33 #endif
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 MAX_REDIS_NODE_NAME 64
40 #define MAX_REDIS_PASSWD_LENGTH 512
41 #define MAX_REDIS_VAL_SIZE 256
42 #define MAX_REDIS_QUERY 2048
44 /* Redis plugin configuration example:
45 *
46 * <Plugin redis>
47 * <Node "mynode">
48 * Host "localhost"
49 * Port "6379"
50 * Timeout 2
51 * Password "foobar"
52 * </Node>
53 * </Plugin>
54 */
56 struct redis_query_s;
57 typedef struct redis_query_s redis_query_t;
58 struct redis_query_s
59 {
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 };
66 struct redis_node_s;
67 typedef struct redis_node_s redis_node_t;
68 struct redis_node_s
69 {
70 char name[MAX_REDIS_NODE_NAME];
71 char host[HOST_NAME_MAX];
72 char passwd[MAX_REDIS_PASSWD_LENGTH];
73 int port;
74 struct timeval timeout;
75 redis_query_t *queries;
77 redis_node_t *next;
78 };
80 static redis_node_t *nodes_head = NULL;
82 static int redis_node_add (const redis_node_t *rn) /* {{{ */
83 {
84 redis_node_t *rn_copy;
85 redis_node_t *rn_ptr;
87 /* Check for duplicates first */
88 for (rn_ptr = nodes_head; rn_ptr != NULL; rn_ptr = rn_ptr->next)
89 if (strcmp (rn->name, rn_ptr->name) == 0)
90 break;
92 if (rn_ptr != NULL)
93 {
94 ERROR ("redis plugin: A node with the name `%s' already exists.",
95 rn->name);
96 return (-1);
97 }
99 rn_copy = malloc (sizeof (*rn_copy));
100 if (rn_copy == NULL)
101 {
102 ERROR ("redis plugin: malloc failed adding redis_node to the tree.");
103 return (-1);
104 }
106 memcpy (rn_copy, rn, sizeof (*rn_copy));
107 rn_copy->next = NULL;
109 DEBUG ("redis plugin: Adding node \"%s\".", rn->name);
111 if (nodes_head == NULL)
112 nodes_head = rn_copy;
113 else
114 {
115 rn_ptr = nodes_head;
116 while (rn_ptr->next != NULL)
117 rn_ptr = rn_ptr->next;
118 rn_ptr->next = rn_copy;
119 }
121 return (0);
122 } /* }}} */
124 static redis_query_t *redis_config_query (oconfig_item_t *ci) /* {{{ */
125 {
126 redis_query_t *rq;
127 int status;
128 int i;
130 rq = calloc(1, sizeof(*rq));
131 if (rq == NULL) {
132 ERROR("redis plugin: calloc failed adding redis_query.");
133 return NULL;
134 }
135 status = cf_util_get_string_buffer(ci, rq->query, sizeof(rq->query));
136 if (status != 0)
137 goto err;
139 /*
140 * Default to a gauge type.
141 */
142 (void)strncpy(rq->type, "gauge", sizeof(rq->type));
143 (void)sstrncpy(rq->instance, rq->query, sizeof(rq->instance));
144 replace_special(rq->instance, sizeof(rq->instance));
146 for (i = 0; i < ci->children_num; i++) {
147 oconfig_item_t *option = ci->children + i;
149 if (strcasecmp("Type", option->key) == 0) {
150 status = cf_util_get_string_buffer(option, rq->type, sizeof(rq->type));
151 } else if (strcasecmp("Instance", option->key) == 0) {
152 status = cf_util_get_string_buffer(option, rq->instance, sizeof(rq->instance));
153 } else {
154 WARNING("redis plugin: unknown configuration option: %s", option->key);
155 status = -1;
156 }
157 if (status != 0)
158 goto err;
159 }
160 return rq;
161 err:
162 free(rq);
163 return NULL;
164 } /* }}} */
166 static int redis_config_node (oconfig_item_t *ci) /* {{{ */
167 {
168 redis_node_t rn;
169 redis_query_t *rq;
170 int i;
171 int status;
172 int timeout;
174 memset (&rn, 0, sizeof (rn));
175 sstrncpy (rn.host, REDIS_DEF_HOST, sizeof (rn.host));
176 rn.port = REDIS_DEF_PORT;
177 rn.timeout.tv_usec = REDIS_DEF_TIMEOUT;
178 rn.queries = NULL;
180 status = cf_util_get_string_buffer (ci, rn.name, sizeof (rn.name));
181 if (status != 0)
182 return (status);
184 for (i = 0; i < ci->children_num; i++)
185 {
186 oconfig_item_t *option = ci->children + i;
188 if (strcasecmp ("Host", option->key) == 0)
189 status = cf_util_get_string_buffer (option, rn.host, sizeof (rn.host));
190 else if (strcasecmp ("Port", option->key) == 0)
191 {
192 status = cf_util_get_port_number (option);
193 if (status > 0)
194 {
195 rn.port = status;
196 status = 0;
197 }
198 }
199 else if (strcasecmp ("Query", option->key) == 0)
200 {
201 rq = redis_config_query(option);
202 if (rq == NULL) {
203 status =1;
204 } else {
205 rq->next = rn.queries;
206 rn.queries = rq;
207 }
208 }
209 else if (strcasecmp ("Timeout", option->key) == 0)
210 {
211 status = cf_util_get_int (option, &timeout);
212 if (status == 0) rn.timeout.tv_usec = timeout;
213 }
214 else if (strcasecmp ("Password", option->key) == 0)
215 status = cf_util_get_string_buffer (option, rn.passwd, sizeof (rn.passwd));
216 else
217 WARNING ("redis plugin: Option `%s' not allowed inside a `Node' "
218 "block. I'll ignore this option.", option->key);
220 if (status != 0)
221 break;
222 }
224 if (status != 0)
225 return (status);
227 return (redis_node_add (&rn));
228 } /* }}} int redis_config_node */
230 static int redis_config (oconfig_item_t *ci) /* {{{ */
231 {
232 int i;
234 for (i = 0; i < ci->children_num; i++)
235 {
236 oconfig_item_t *option = ci->children + i;
238 if (strcasecmp ("Node", option->key) == 0)
239 redis_config_node (option);
240 else
241 WARNING ("redis plugin: Option `%s' not allowed in redis"
242 " configuration. It will be ignored.", option->key);
243 }
245 if (nodes_head == NULL)
246 {
247 ERROR ("redis plugin: No valid node configuration could be found.");
248 return (ENOENT);
249 }
251 return (0);
252 } /* }}} */
254 __attribute__ ((nonnull(2)))
255 static void redis_submit (char *plugin_instance,
256 const char *type, const char *type_instance,
257 value_t value) /* {{{ */
258 {
259 value_t values[1];
260 value_list_t vl = VALUE_LIST_INIT;
262 values[0] = value;
264 vl.values = values;
265 vl.values_len = 1;
266 sstrncpy (vl.host, hostname_g, sizeof (vl.host));
267 sstrncpy (vl.plugin, "redis", sizeof (vl.plugin));
268 if (plugin_instance != NULL)
269 sstrncpy (vl.plugin_instance, plugin_instance,
270 sizeof (vl.plugin_instance));
271 sstrncpy (vl.type, type, sizeof (vl.type));
272 if (type_instance != NULL)
273 sstrncpy (vl.type_instance, type_instance,
274 sizeof (vl.type_instance));
276 plugin_dispatch_values (&vl);
277 } /* }}} */
279 static int redis_init (void) /* {{{ */
280 {
281 redis_node_t rn = {
282 .name = "default",
283 .host = REDIS_DEF_HOST,
284 .port = REDIS_DEF_PORT,
285 .timeout.tv_sec = 0,
286 .timeout.tv_usec = REDIS_DEF_TIMEOUT,
287 .next = NULL
288 };
290 if (nodes_head == NULL)
291 redis_node_add (&rn);
293 return (0);
294 } /* }}} int redis_init */
296 static int redis_handle_info (char *node, char const *info_line, char const *type, char const *type_instance, char const *field_name, int ds_type) /* {{{ */
297 {
298 char *str = strstr (info_line, field_name);
299 static char buf[MAX_REDIS_VAL_SIZE];
300 value_t val;
301 if (str)
302 {
303 int i;
305 str += strlen (field_name) + 1; /* also skip the ':' */
306 for(i=0;(*str && (isdigit((unsigned char)*str) || *str == '.'));i++,str++)
307 buf[i] = *str;
308 buf[i] ='\0';
310 if(parse_value (buf, &val, ds_type) == -1)
311 {
312 WARNING ("redis plugin: Unable to parse field `%s'.", field_name);
313 return (-1);
314 }
316 redis_submit (node, type, type_instance, val);
317 return (0);
318 }
319 return (-1);
321 } /* }}} int redis_handle_info */
323 static int redis_handle_query (redisContext *rh, redis_node_t *rn, redis_query_t *rq) /* {{{ */
324 {
325 redisReply *rr;
326 const data_set_t *ds;
327 value_t val;
329 ds = plugin_get_ds (rq->type);
330 if (!ds) {
331 ERROR ("redis plugin: DataSet `%s' not defined.", rq->type);
332 return (-1);
333 }
335 if (ds->ds_num != 1) {
336 ERROR ("redis plugin: DS `%s' has too many types.", rq->type);
337 return (-1);
338 }
340 if ((rr = redisCommand(rh, rq->query)) == NULL) {
341 WARNING("redis plugin: unable to carry out query `%s'.", rq->query);
342 return (-1);
343 }
345 switch (rr->type) {
346 case REDIS_REPLY_INTEGER:
347 switch (ds->ds[0].type) {
348 case DS_TYPE_COUNTER:
349 val.counter = (counter_t)rr->integer;
350 break;
351 case DS_TYPE_GAUGE:
352 val.gauge = (gauge_t)rr->integer;
353 break;
354 case DS_TYPE_DERIVE:
355 val.gauge = (derive_t)rr->integer;
356 break;
357 case DS_TYPE_ABSOLUTE:
358 val.gauge = (absolute_t)rr->integer;
359 break;
360 }
361 break;
362 case REDIS_REPLY_STRING:
363 if (parse_value (rr->str, &val, ds->ds[0].type) == -1) {
364 WARNING("redis plugin: Unable to parse field `%s'.", rq->type);
365 freeReplyObject (rr);
366 return (-1);
367 }
368 break;
369 default:
370 WARNING("redis plugin: Cannot coerce redis type.");
371 freeReplyObject(rr);
372 return (-1);
373 }
375 redis_submit(rn->name, rq->type, (strlen(rq->instance) >0)?rq->instance:NULL, val);
376 freeReplyObject (rr);
377 return 0;
378 } /* }}} int redis_handle_query */
380 static int redis_read (void) /* {{{ */
381 {
382 redis_node_t *rn;
383 redis_query_t *rq;
385 for (rn = nodes_head; rn != NULL; rn = rn->next)
386 {
387 redisContext *rh;
388 redisReply *rr;
390 DEBUG ("redis plugin: querying info from node `%s' (%s:%d).", rn->name, rn->host, rn->port);
392 rh = redisConnectWithTimeout ((char *)rn->host, rn->port, rn->timeout);
393 if (rh == NULL)
394 {
395 ERROR ("redis plugin: unable to connect to node `%s' (%s:%d).", rn->name, rn->host, rn->port);
396 continue;
397 }
399 if (strlen (rn->passwd) > 0)
400 {
401 DEBUG ("redis plugin: authenticating node `%s' passwd(%s).", rn->name, rn->passwd);
403 if ((rr = redisCommand (rh, "AUTH %s", rn->passwd)) == NULL)
404 {
405 WARNING ("redis plugin: unable to authenticate on node `%s'.", rn->name);
406 goto redis_fail;
407 }
409 if (rr->type != REDIS_REPLY_STATUS)
410 {
411 WARNING ("redis plugin: invalid authentication on node `%s'.", rn->name);
412 goto redis_fail;
413 }
415 freeReplyObject (rr);
416 }
418 if ((rr = redisCommand(rh, "INFO")) == NULL)
419 {
420 WARNING ("redis plugin: unable to get info from node `%s'.", rn->name);
421 goto redis_fail;
422 }
424 redis_handle_info (rn->name, rr->str, "uptime", NULL, "uptime_in_seconds", DS_TYPE_GAUGE);
425 redis_handle_info (rn->name, rr->str, "current_connections", "clients", "connected_clients", DS_TYPE_GAUGE);
426 redis_handle_info (rn->name, rr->str, "blocked_clients", NULL, "blocked_clients", DS_TYPE_GAUGE);
427 redis_handle_info (rn->name, rr->str, "memory", NULL, "used_memory", DS_TYPE_GAUGE);
428 redis_handle_info (rn->name, rr->str, "memory_lua", NULL, "used_memory_lua", DS_TYPE_GAUGE);
429 /* changes_since_last_save: Deprecated in redis version 2.6 and above */
430 redis_handle_info (rn->name, rr->str, "volatile_changes", NULL, "changes_since_last_save", DS_TYPE_GAUGE);
431 redis_handle_info (rn->name, rr->str, "total_connections", NULL, "total_connections_received", DS_TYPE_DERIVE);
432 redis_handle_info (rn->name, rr->str, "total_operations", NULL, "total_commands_processed", DS_TYPE_DERIVE);
433 redis_handle_info (rn->name, rr->str, "operations_per_second", NULL, "instantaneous_ops_per_sec", DS_TYPE_GAUGE);
434 redis_handle_info (rn->name, rr->str, "expired_keys", NULL, "expired_keys", DS_TYPE_DERIVE);
435 redis_handle_info (rn->name, rr->str, "evicted_keys", NULL, "evicted_keys", DS_TYPE_DERIVE);
436 redis_handle_info (rn->name, rr->str, "pubsub", "channels", "pubsub_channels", DS_TYPE_GAUGE);
437 redis_handle_info (rn->name, rr->str, "pubsub", "patterns", "pubsub_patterns", DS_TYPE_GAUGE);
438 redis_handle_info (rn->name, rr->str, "current_connections", "slaves", "connected_slaves", DS_TYPE_GAUGE);
439 redis_handle_info (rn->name, rr->str, "cache_result", "hits", "keyspace_hits", DS_TYPE_DERIVE);
440 redis_handle_info (rn->name, rr->str, "cache_result", "misses", "keyspace_misses", DS_TYPE_DERIVE);
441 redis_handle_info (rn->name, rr->str, "total_bytes", "input", "total_net_input_bytes", DS_TYPE_DERIVE);
442 redis_handle_info (rn->name, rr->str, "total_bytes", "output", "total_net_output_bytes", DS_TYPE_DERIVE);
444 for (rq = rn->queries; rq != NULL; rq = rq->next)
445 redis_handle_query(rh, rn, rq);
447 redis_fail:
448 if (rr != NULL)
449 freeReplyObject (rr);
450 redisFree (rh);
451 }
453 return 0;
454 }
455 /* }}} */
457 void module_register (void) /* {{{ */
458 {
459 plugin_register_complex_config ("redis", redis_config);
460 plugin_register_init ("redis", redis_init);
461 plugin_register_read ("redis", redis_read);
462 /* TODO: plugin_register_write: one redis list per value id with
463 * X elements */
464 }
465 /* }}} */
467 /* vim: set sw=2 sts=2 et fdm=marker : */