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 #define REDIS_DEF_HOST "localhost"
  32 #define REDIS_DEF_PASSWD ""
  33 #define REDIS_DEF_PORT 6379
  34 #define REDIS_DEF_TIMEOUT_SEC 2
  35 #define REDIS_DEF_DB_COUNT 256
  36 #define MAX_REDIS_VAL_SIZE 256
  37 #define MAX_REDIS_QUERY 2048
  38
  39 /* Redis plugin configuration example:
  40  *
  41  * <Plugin redis>
  42  *   <Node "mynode">
  43  *     Host "localhost"
  44  *     Port "6379"
  45  *     Timeout 2
  46  *     Password "foobar"
  47  *   </Node>
  48  * </Plugin>
  49  */
  50
  51 struct redis_query_s;
  52 typedef struct redis_query_s redis_query_t;
  53 struct redis_query_s {
  54   char query[MAX_REDIS_QUERY];
  55   char type[DATA_MAX_NAME_LEN];
  56   char instance[DATA_MAX_NAME_LEN];
  57   int database;
  58
  59   redis_query_t *next;
  60 };
  61
  62 struct prev_s {
  63   derive_t keyspace_hits;
  64   derive_t keyspace_misses;
  65 };
  66 typedef struct prev_s prev_t;
  67
  68 struct redis_node_s;
  69 typedef struct redis_node_s redis_node_t;
  70 struct redis_node_s {
  71   char *name;
  72   char *host;
  73   char *passwd;
  74   int port;
  75   struct timeval timeout;
  76   bool report_command_stats;
  77   bool report_cpu_usage;
  78   redisContext *redisContext;
  79   redis_query_t *queries;
  80   prev_t prev;
  81
  82   redis_node_t *next;
  83 };
  84
  85 static bool redis_have_instances;
  86 static int redis_read(user_data_t *user_data);
  87
  88 static void redis_node_free(void *arg) {
  89   redis_node_t *rn = arg;
  90   if (rn == NULL)
  91     return;
  92
  93   redis_query_t *rq = rn->queries;
  94   while (rq != NULL) {
  95     redis_query_t *next = rq->next;
  96     sfree(rq);
  97     rq = next;
  98   }
  99
 100   if (rn->redisContext)
 101     redisFree(rn->redisContext);
 102   sfree(rn->name);
 103   sfree(rn->host);
 104   sfree(rn->passwd);
 105   sfree(rn);
 106 } /* void redis_node_free */
 107
 108 static int redis_node_add(redis_node_t *rn) /* {{{ */
 109 {
 110   DEBUG("redis plugin: Adding node \"%s\".", rn->name);
 111
 112   /* Disable automatic generation of default instance in the init callback. */
 113   redis_have_instances = true;
 114
 115   char cb_name[sizeof("redis/") + DATA_MAX_NAME_LEN];
 116   snprintf(cb_name, sizeof(cb_name), "redis/%s", rn->name);
 117
 118   return plugin_register_complex_read(
 119       /* group = */ "redis",
 120       /* name      = */ cb_name,
 121       /* callback  = */ redis_read,
 122       /* interval  = */ 0,
 123       &(user_data_t){
 124           .data = rn, .free_func = redis_node_free,
 125       });
 126 } /* }}} */
 127
 128 static redis_query_t *redis_config_query(oconfig_item_t *ci) /* {{{ */
 129 {
 130   redis_query_t *rq;
 131   int status;
 132
 133   rq = calloc(1, sizeof(*rq));
 134   if (rq == NULL) {
 135     ERROR("redis plugin: calloc failed adding redis_query.");
 136     return NULL;
 137   }
 138   status = cf_util_get_string_buffer(ci, rq->query, sizeof(rq->query));
 139   if (status != 0)
 140     goto err;
 141
 142   /*
 143    * Default to a gauge type.
 144    */
 145   (void)strncpy(rq->type, "gauge", sizeof(rq->type));
 146   (void)sstrncpy(rq->instance, rq->query, sizeof(rq->instance));
 147   replace_special(rq->instance, sizeof(rq->instance));
 148
 149   rq->database = 0;
 150
 151   for (int i = 0; i < ci->children_num; i++) {
 152     oconfig_item_t *option = ci->children + i;
 153
 154     if (strcasecmp("Type", option->key) == 0) {
 155       status = cf_util_get_string_buffer(option, rq->type, sizeof(rq->type));
 156     } else if (strcasecmp("Instance", option->key) == 0) {
 157       status =
 158           cf_util_get_string_buffer(option, rq->instance, sizeof(rq->instance));
 159     } else if (strcasecmp("Database", option->key) == 0) {
 160       status = cf_util_get_int(option, &rq->database);
 161       if (rq->database < 0) {
 162         WARNING("redis plugin: The \"Database\" option must be positive "
 163                 "integer or zero");
 164         status = -1;
 165       }
 166     } else {
 167       WARNING("redis plugin: unknown configuration option: %s", option->key);
 168       status = -1;
 169     }
 170     if (status != 0)
 171       goto err;
 172   }
 173   return rq;
 174 err:
 175   free(rq);
 176   return NULL;
 177 } /* }}} */
 178
 179 static int redis_config_node(oconfig_item_t *ci) /* {{{ */
 180 {
 181   redis_node_t *rn = calloc(1, sizeof(*rn));
 182   if (rn == NULL) {
 183     ERROR("redis plugin: calloc failed adding node.");
 184     return ENOMEM;
 185   }
 186
 187   rn->port = REDIS_DEF_PORT;
 188   rn->timeout.tv_sec = REDIS_DEF_TIMEOUT_SEC;
 189   rn->report_cpu_usage = true;
 190
 191   rn->host = strdup(REDIS_DEF_HOST);
 192   if (rn->host == NULL) {
 193     ERROR("redis plugin: strdup failed adding node.");
 194     sfree(rn);
 195     return ENOMEM;
 196   }
 197
 198   int status = cf_util_get_string(ci, &rn->name);
 199   if (status != 0) {
 200     sfree(rn->host);
 201     sfree(rn);
 202     return status;
 203   }
 204
 205   for (int i = 0; i < ci->children_num; i++) {
 206     oconfig_item_t *option = ci->children + i;
 207
 208     if (strcasecmp("Host", option->key) == 0)
 209       status = cf_util_get_string(option, &rn->host);
 210     else if (strcasecmp("Port", option->key) == 0) {
 211       status = cf_util_get_port_number(option);
 212       if (status > 0) {
 213         rn->port = status;
 214         status = 0;
 215       }
 216     } else if (strcasecmp("Query", option->key) == 0) {
 217       redis_query_t *rq = redis_config_query(option);
 218       if (rq == NULL) {
 219         status = 1;
 220       } else {
 221         rq->next = rn->queries;
 222         rn->queries = rq;
 223       }
 224     } else if (strcasecmp("Timeout", option->key) == 0) {
 225       int timeout;
 226       status = cf_util_get_int(option, &timeout);
 227       if (status == 0) {
 228         rn->timeout.tv_usec = timeout * 1000;
 229         rn->timeout.tv_sec = rn->timeout.tv_usec / 1000000L;
 230         rn->timeout.tv_usec %= 1000000L;
 231       }
 232     } else if (strcasecmp("Password", option->key) == 0)
 233       status = cf_util_get_string(option, &rn->passwd);
 234     else if (strcasecmp("ReportCommandStats", option->key) == 0)
 235       status = cf_util_get_boolean(option, &rn->report_command_stats);
 236     else if (strcasecmp("ReportCpuUsage", option->key) == 0)
 237       status = cf_util_get_boolean(option, &rn->report_cpu_usage);
 238     else
 239       WARNING("redis plugin: Option `%s' not allowed inside a `Node' "
 240               "block. I'll ignore this option.",
 241               option->key);
 242
 243     if (status != 0)
 244       break;
 245   }
 246
 247   if (status != 0) {
 248     redis_node_free(rn);
 249     return status;
 250   }
 251
 252   return redis_node_add(rn);
 253 } /* }}} int redis_config_node */
 254
 255 static int redis_config(oconfig_item_t *ci) /* {{{ */
 256 {
 257   for (int i = 0; i < ci->children_num; i++) {
 258     oconfig_item_t *option = ci->children + i;
 259
 260     if (strcasecmp("Node", option->key) == 0)
 261       redis_config_node(option);
 262     else
 263       WARNING("redis plugin: Option `%s' not allowed in redis"
 264               " configuration. It will be ignored.",
 265               option->key);
 266   }
 267
 268   return 0;
 269 } /* }}} */
 270
 271 __attribute__((nonnull(2))) static void
 272 redis_submit(const char *plugin_instance, const char *type,
 273              const char *type_instance, value_t value) /* {{{ */
 274 {
 275   value_list_t vl = VALUE_LIST_INIT;
 276
 277   vl.values = &value;
 278   vl.values_len = 1;
 279   sstrncpy(vl.plugin, "redis", sizeof(vl.plugin));
 280   if (plugin_instance != NULL)
 281     sstrncpy(vl.plugin_instance, plugin_instance, sizeof(vl.plugin_instance));
 282   sstrncpy(vl.type, type, sizeof(vl.type));
 283   if (type_instance != NULL)
 284     sstrncpy(vl.type_instance, type_instance, sizeof(vl.type_instance));
 285
 286   plugin_dispatch_values(&vl);
 287 } /* }}} */
 288
 289 __attribute__((nonnull(2))) static void
 290 redis_submit2(const char *plugin_instance, const char *type,
 291               const char *type_instance, value_t value0,
 292               value_t value1) /* {{{ */
 293 {
 294   value_list_t vl = VALUE_LIST_INIT;
 295   value_t values[] = {value0, value1};
 296
 297   vl.values = values;
 298   vl.values_len = STATIC_ARRAY_SIZE(values);
 299
 300   sstrncpy(vl.plugin, "redis", sizeof(vl.plugin));
 301   sstrncpy(vl.type, type, sizeof(vl.type));
 302
 303   if (plugin_instance != NULL)
 304     sstrncpy(vl.plugin_instance, plugin_instance, sizeof(vl.plugin_instance));
 305
 306   if (type_instance != NULL)
 307     sstrncpy(vl.type_instance, type_instance, sizeof(vl.type_instance));
 308
 309   plugin_dispatch_values(&vl);
 310 } /* }}} */
 311
 312 static int redis_init(void) /* {{{ */
 313 {
 314   if (redis_have_instances)
 315     return 0;
 316
 317   redis_node_t *rn = calloc(1, sizeof(*rn));
 318   if (rn == NULL)
 319     return ENOMEM;
 320
 321   rn->port = REDIS_DEF_PORT;
 322   rn->timeout.tv_sec = REDIS_DEF_TIMEOUT_SEC;
 323
 324   rn->name = strdup("default");
 325   rn->host = strdup(REDIS_DEF_HOST);
 326
 327   if (rn->name == NULL || rn->host == NULL) {
 328     sfree(rn->name);
 329     sfree(rn->host);
 330     sfree(rn);
 331     return ENOMEM;
 332   }
 333
 334   return redis_node_add(rn);
 335 } /* }}} int redis_init */
 336
 337 static void *c_redisCommand(redis_node_t *rn, const char *format, ...) {
 338   redisContext *c = rn->redisContext;
 339
 340   if (c == NULL)
 341     return NULL;
 342
 343   va_list ap;
 344   va_start(ap, format);
 345   void *reply = redisvCommand(c, format, ap);
 346   va_end(ap);
 347
 348   if (reply == NULL) {
 349     ERROR("redis plugin: Connection error: %s", c->errstr);
 350     redisFree(rn->redisContext);
 351     rn->redisContext = NULL;
 352   }
 353
 354   return reply;
 355 } /* void c_redisCommand */
 356
 357 static int redis_get_info_value(char const *info_line, char const *field_name,
 358                                 int ds_type, value_t *val) {
 359   char *str = strstr(info_line, field_name);
 360   static char buf[MAX_REDIS_VAL_SIZE];
 361   if (str) {
 362     int i;
 363
 364     str += strlen(field_name) + 1; /* also skip the ':' */
 365     for (i = 0; (*str && (isdigit((unsigned char)*str) || *str == '.'));
 366          i++, str++)
 367       buf[i] = *str;
 368     buf[i] = '\0';
 369
 370     if (parse_value(buf, val, ds_type) == -1) {
 371       WARNING("redis plugin: Unable to parse field `%s'.", field_name);
 372       return -1;
 373     }
 374
 375     return 0;
 376   }
 377   return -1;
 378 } /* int redis_get_info_value */
 379
 380 static int redis_handle_info(char *node, char const *info_line,
 381                              char const *type, char const *type_instance,
 382                              char const *field_name, int ds_type) /* {{{ */
 383 {
 384   value_t val;
 385   if (redis_get_info_value(info_line, field_name, ds_type, &val) != 0)
 386     return -1;
 387
 388   redis_submit(node, type, type_instance, val);
 389   return 0;
 390 } /* }}} int redis_handle_info */
 391
 392 static int redis_handle_query(redis_node_t *rn, redis_query_t *rq) /* {{{ */
 393 {
 394   redisReply *rr;
 395   const data_set_t *ds;
 396   value_t val;
 397
 398   ds = plugin_get_ds(rq->type);
 399   if (!ds) {
 400     ERROR("redis plugin: DS type `%s' not defined.", rq->type);
 401     return -1;
 402   }
 403
 404   if (ds->ds_num != 1) {
 405     ERROR("redis plugin: DS type `%s' has too many datasources. This is not "
 406           "supported currently.",
 407           rq->type);
 408     return -1;
 409   }
 410
 411   if ((rr = c_redisCommand(rn, "SELECT %d", rq->database)) == NULL) {
 412     WARNING("redis plugin: unable to switch to database `%d' on node `%s'.",
 413             rq->database, rn->name);
 414     return -1;
 415   }
 416
 417   if ((rr = c_redisCommand(rn, rq->query)) == NULL) {
 418     WARNING("redis plugin: unable to carry out query `%s'.", rq->query);
 419     return -1;
 420   }
 421
 422   switch (rr->type) {
 423   case REDIS_REPLY_INTEGER:
 424     switch (ds->ds[0].type) {
 425     case DS_TYPE_COUNTER:
 426       val.counter = (counter_t)rr->integer;
 427       break;
 428     case DS_TYPE_GAUGE:
 429       val.gauge = (gauge_t)rr->integer;
 430       break;
 431     case DS_TYPE_DERIVE:
 432       val.gauge = (derive_t)rr->integer;
 433       break;
 434     case DS_TYPE_ABSOLUTE:
 435       val.gauge = (absolute_t)rr->integer;
 436       break;
 437     }
 438     break;
 439   case REDIS_REPLY_STRING:
 440     if (parse_value(rr->str, &val, ds->ds[0].type) == -1) {
 441       WARNING("redis plugin: Query `%s': Unable to parse value.", rq->query);
 442       freeReplyObject(rr);
 443       return -1;
 444     }
 445     break;
 446   case REDIS_REPLY_ERROR:
 447     WARNING("redis plugin: Query `%s' failed: %s.", rq->query, rr->str);
 448     freeReplyObject(rr);
 449     return -1;
 450   case REDIS_REPLY_ARRAY:
 451     WARNING("redis plugin: Query `%s' should return string or integer. Arrays "
 452             "are not supported.",
 453             rq->query);
 454     freeReplyObject(rr);
 455     return -1;
 456   default:
 457     WARNING("redis plugin: Query `%s': Cannot coerce redis type (%i).",
 458             rq->query, rr->type);
 459     freeReplyObject(rr);
 460     return -1;
 461   }
 462
 463   redis_submit(rn->name, rq->type,
 464                (strlen(rq->instance) > 0) ? rq->instance : NULL, val);
 465   freeReplyObject(rr);
 466   return 0;
 467 } /* }}} int redis_handle_query */
 468
 469 static int redis_db_stats(const char *node, char const *info_line) /* {{{ */
 470 {
 471   /* redis_db_stats parses and dispatches Redis database statistics,
 472    * currently the number of keys for each database.
 473    * info_line needs to have the following format:
 474    *   db0:keys=4,expires=0,avg_ttl=0
 475    */
 476
 477   for (int db = 0; db < REDIS_DEF_DB_COUNT; db++) {
 478     static char buf[MAX_REDIS_VAL_SIZE];
 479     static char field_name[12];
 480     static char db_id[4];
 481     value_t val;
 482     char *str;
 483     int i;
 484
 485     snprintf(field_name, sizeof(field_name), "db%d:keys=", db);
 486
 487     str = strstr(info_line, field_name);
 488     if (!str)
 489       continue;
 490
 491     str += strlen(field_name);
 492     for (i = 0; (*str && isdigit((int)*str)); i++, str++)
 493       buf[i] = *str;
 494     buf[i] = '\0';
 495
 496     if (parse_value(buf, &val, DS_TYPE_GAUGE) != 0) {
 497       WARNING("redis plugin: Unable to parse field `%s'.", field_name);
 498       return -1;
 499     }
 500
 501     snprintf(db_id, sizeof(db_id), "%d", db);
 502     redis_submit(node, "records", db_id, val);
 503   }
 504   return 0;
 505
 506 } /* }}} int redis_db_stats */
 507
 508 static void redis_cpu_usage(const char *node, char const *info_line) {
 509   while (42) {
 510     value_t rusage_user;
 511     value_t rusage_syst;
 512
 513     if (redis_get_info_value(info_line, "used_cpu_user", DS_TYPE_GAUGE,
 514                              &rusage_user) != 0)
 515       break;
 516
 517     if (redis_get_info_value(info_line, "used_cpu_sys", DS_TYPE_GAUGE,
 518                              &rusage_syst) != 0)
 519       break;
 520
 521     redis_submit2(node, "ps_cputime", "daemon",
 522                   (value_t){.derive = rusage_user.gauge * 1000000},
 523                   (value_t){.derive = rusage_syst.gauge * 1000000});
 524     break;
 525   }
 526
 527   while (42) {
 528     value_t rusage_user;
 529     value_t rusage_syst;
 530
 531     if (redis_get_info_value(info_line, "used_cpu_user_children", DS_TYPE_GAUGE,
 532                              &rusage_user) != 0)
 533       break;
 534
 535     if (redis_get_info_value(info_line, "used_cpu_sys_children", DS_TYPE_GAUGE,
 536                              &rusage_syst) != 0)
 537       break;
 538
 539     redis_submit2(node, "ps_cputime", "children",
 540                   (value_t){.derive = rusage_user.gauge * 1000000},
 541                   (value_t){.derive = rusage_syst.gauge * 1000000});
 542     break;
 543   }
 544 } /* void redis_cpu_usage */
 545
 546 static gauge_t calculate_ratio_percent(derive_t part1, derive_t part2,
 547                                        derive_t *prev1, derive_t *prev2) {
 548   if ((*prev1 == 0) || (*prev2 == 0) || (part1 < *prev1) || (part2 < *prev2)) {
 549     *prev1 = part1;
 550     *prev2 = part2;
 551     return NAN;
 552   }
 553
 554   derive_t num = part1 - *prev1;
 555   derive_t denom = part2 - *prev2 + num;
 556
 557   *prev1 = part1;
 558   *prev2 = part2;
 559
 560   if (denom == 0)
 561     return NAN;
 562
 563   if (num == 0)
 564     return 0;
 565
 566   return 100.0 * (gauge_t)num / (gauge_t)denom;
 567 } /* gauge_t calculate_ratio_percent */
 568
 569 static void redis_keyspace_usage(redis_node_t *rn, char const *info_line) {
 570   value_t hits, misses;
 571
 572   if (redis_get_info_value(info_line, "keyspace_hits", DS_TYPE_DERIVE, &hits) !=
 573       0)
 574     return;
 575
 576   if (redis_get_info_value(info_line, "keyspace_misses", DS_TYPE_DERIVE,
 577                            &misses) != 0)
 578     return;
 579
 580   redis_submit(rn->name, "cache_result", "hits", hits);
 581   redis_submit(rn->name, "cache_result", "misses", misses);
 582
 583   prev_t *prev = &rn->prev;
 584   gauge_t ratio = calculate_ratio_percent(
 585       hits.derive, misses.derive, &prev->keyspace_hits, &prev->keyspace_misses);
 586   redis_submit(rn->name, "percent", "hitratio", (value_t){.gauge = ratio});
 587
 588 } /* void redis_keyspace_usage */
 589
 590 static void redis_check_connection(redis_node_t *rn) {
 591   if (rn->redisContext)
 592     return;
 593
 594   redisContext *rh = redisConnectWithTimeout(rn->host, rn->port, rn->timeout);
 595
 596   if (rh == NULL) {
 597     ERROR("redis plugin: can't allocate redis context");
 598     return;
 599   }
 600   if (rh->err) {
 601     ERROR("redis plugin: unable to connect to node `%s' (%s:%d): %s.", rn->name,
 602           rn->host, rn->port, rh->errstr);
 603     redisFree(rh);
 604     return;
 605   }
 606
 607   rn->redisContext = rh;
 608
 609   if (rn->passwd) {
 610     redisReply *rr;
 611
 612     DEBUG("redis plugin: authenticating node `%s' passwd(%s).", rn->name,
 613           rn->passwd);
 614
 615     if ((rr = c_redisCommand(rn, "AUTH %s", rn->passwd)) == NULL) {
 616       WARNING("redis plugin: unable to authenticate on node `%s'.", rn->name);
 617       return;
 618     }
 619
 620     if (rr->type != REDIS_REPLY_STATUS) {
 621       WARNING("redis plugin: invalid authentication on node `%s'.", rn->name);
 622       freeReplyObject(rr);
 623       redisFree(rn->redisContext);
 624       rn->redisContext = NULL;
 625       return;
 626     }
 627
 628     freeReplyObject(rr);
 629   }
 630   return;
 631 } /* void redis_check_connection */
 632
 633 static void redis_read_server_info(redis_node_t *rn) {
 634   redisReply *rr;
 635
 636   if ((rr = c_redisCommand(rn, "INFO")) == NULL) {
 637     WARNING("redis plugin: unable to get INFO from node `%s'.", rn->name);
 638     return;
 639   }
 640
 641   redis_handle_info(rn->name, rr->str, "uptime", NULL, "uptime_in_seconds",
 642                     DS_TYPE_GAUGE);
 643   redis_handle_info(rn->name, rr->str, "current_connections", "clients",
 644                     "connected_clients", DS_TYPE_GAUGE);
 645   redis_handle_info(rn->name, rr->str, "blocked_clients", NULL,
 646                     "blocked_clients", DS_TYPE_GAUGE);
 647   redis_handle_info(rn->name, rr->str, "memory", NULL, "used_memory",
 648                     DS_TYPE_GAUGE);
 649   redis_handle_info(rn->name, rr->str, "memory_lua", NULL, "used_memory_lua",
 650                     DS_TYPE_GAUGE);
 651   /* changes_since_last_save: Deprecated in redis version 2.6 and above */
 652   redis_handle_info(rn->name, rr->str, "volatile_changes", NULL,
 653                     "changes_since_last_save", DS_TYPE_GAUGE);
 654   redis_handle_info(rn->name, rr->str, "total_connections", NULL,
 655                     "total_connections_received", DS_TYPE_DERIVE);
 656   redis_handle_info(rn->name, rr->str, "total_operations", NULL,
 657                     "total_commands_processed", DS_TYPE_DERIVE);
 658   redis_handle_info(rn->name, rr->str, "expired_keys", NULL, "expired_keys",
 659                     DS_TYPE_DERIVE);
 660   redis_handle_info(rn->name, rr->str, "evicted_keys", NULL, "evicted_keys",
 661                     DS_TYPE_DERIVE);
 662   redis_handle_info(rn->name, rr->str, "pubsub", "channels", "pubsub_channels",
 663                     DS_TYPE_GAUGE);
 664   redis_handle_info(rn->name, rr->str, "pubsub", "patterns", "pubsub_patterns",
 665                     DS_TYPE_GAUGE);
 666   redis_handle_info(rn->name, rr->str, "current_connections", "slaves",
 667                     "connected_slaves", DS_TYPE_GAUGE);
 668   redis_handle_info(rn->name, rr->str, "total_bytes", "input",
 669                     "total_net_input_bytes", DS_TYPE_DERIVE);
 670   redis_handle_info(rn->name, rr->str, "total_bytes", "output",
 671                     "total_net_output_bytes", DS_TYPE_DERIVE);
 672
 673   redis_keyspace_usage(rn, rr->str);
 674
 675   redis_db_stats(rn->name, rr->str);
 676
 677   if (rn->report_cpu_usage)
 678     redis_cpu_usage(rn->name, rr->str);
 679
 680   freeReplyObject(rr);
 681 } /* void redis_read_server_info */
 682
 683 static void redis_read_command_stats(redis_node_t *rn) {
 684   redisReply *rr;
 685
 686   if ((rr = c_redisCommand(rn, "INFO commandstats")) == NULL) {
 687     WARNING("redis plugin: node `%s': unable to get `INFO commandstats'.",
 688             rn->name);
 689     return;
 690   }
 691
 692   if (rr->type != REDIS_REPLY_STRING) {
 693     WARNING("redis plugin: node `%s' `INFO commandstats' returned unsupported "
 694             "redis type %i.",
 695             rn->name, rr->type);
 696     freeReplyObject(rr);
 697     return;
 698   }
 699
 700   char *command;
 701   char *line;
 702   char *ptr = rr->str;
 703   char *saveptr = NULL;
 704   while ((line = strtok_r(ptr, "\n\r", &saveptr)) != NULL) {
 705     ptr = NULL;
 706
 707     if (line[0] == '#')
 708       continue;
 709
 710     /* command name */
 711     if (strstr(line, "cmdstat_") != line) {
 712       ERROR("redis plugin: not found 'cmdstat_' prefix in line '%s'", line);
 713       continue;
 714     }
 715
 716     char *values = strstr(line, ":");
 717     if (values == NULL) {
 718       ERROR("redis plugin: not found ':' separator in line '%s'", line);
 719       continue;
 720     }
 721
 722     /* Null-terminate command token */
 723     values[0] = '\0';
 724     command = line + strlen("cmdstat_");
 725     values++;
 726
 727     /* parse values */
 728     /* cmdstat_publish:calls=20795774,usec=111039258,usec_per_call=5.34 */
 729     char *field;
 730     char *saveptr_field = NULL;
 731     while ((field = strtok_r(values, "=", &saveptr_field)) != NULL) {
 732       values = NULL;
 733
 734       const char *type;
 735       /* only these are supported */
 736       if (strcmp(field, "calls") == 0)
 737         type = "commands";
 738       else if (strcmp(field, "usec") == 0)
 739         type = "redis_command_cputime";
 740       else
 741         continue;
 742
 743       if ((field = strtok_r(NULL, ",", &saveptr_field)) == NULL)
 744         continue;
 745
 746       char *endptr = NULL;
 747       errno = 0;
 748       derive_t value = strtoll(field, &endptr, 0);
 749
 750       if ((endptr == field) || (errno != 0))
 751         continue;
 752
 753       redis_submit(rn->name, type, command, (value_t){.derive = value});
 754     }
 755   }
 756   freeReplyObject(rr);
 757 } /* void redis_read_command_stats */
 758
 759 static int redis_read(user_data_t *user_data) /* {{{ */
 760 {
 761   redis_node_t *rn = user_data->data;
 762
 763   DEBUG("redis plugin: querying info from node `%s' (%s:%d).", rn->name,
 764         rn->host, rn->port);
 765
 766   redis_check_connection(rn);
 767
 768   if (!rn->redisContext) /* no connection */
 769     return -1;
 770
 771   redis_read_server_info(rn);
 772
 773   if (!rn->redisContext) /* connection lost */
 774     return -1;
 775
 776   if (rn->report_command_stats) {
 777     redis_read_command_stats(rn);
 778
 779     if (!rn->redisContext) /* connection lost */
 780       return -1;
 781   }
 782
 783   for (redis_query_t *rq = rn->queries; rq != NULL; rq = rq->next) {
 784     redis_handle_query(rn, rq);
 785     if (!rn->redisContext) /* connection lost */
 786       return -1;
 787   }
 788
 789   return 0;
 790 }
 791 /* }}} */
 792
 793 void module_register(void) /* {{{ */
 794 {
 795   plugin_register_complex_config("redis", redis_config);
 796   plugin_register_init("redis", redis_init);
 797 }
 798 /* }}} */