src/memcached.c

   1 /**
   2  * collectd - src/memcached.c, based on src/hddtemp.c
   3  * Copyright (C) 2007       Antony Dovgal
   4  * Copyright (C) 2007-2010  Florian Forster
   5  * Copyright (C) 2009       Doug MacEachern
   6  * Copyright (C) 2009       Franck Lombardi
   7  * Copyright (C) 2012       Nicolas Szalay
   8  *
   9  * This program is free software; you can redistribute it and/or modify it
  10  * under the terms of the GNU General Public License as published by the
  11  * Free Software Foundation; either version 2 of the License, or (at your
  12  * option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful, but
  15  * WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU General Public License along
  20  * with this program; if not, write to the Free Software Foundation, Inc.,
  21  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  22  *
  23  * Authors:
  24  *   Antony Dovgal <tony at daylessday dot org>
  25  *   Florian octo Forster <octo at collectd.org>
  26  *   Doug MacEachern <dougm at hyperic.com>
  27  *   Franck Lombardi
  28  *   Nicolas Szalay
  29  **/
  30
  31 #include "collectd.h"
  32 #include "common.h"
  33 #include "plugin.h"
  34 #include "configfile.h"
  35
  36 # include <poll.h>
  37 # include <netdb.h>
  38 # include <sys/socket.h>
  39 # include <sys/un.h>
  40 # include <netinet/in.h>
  41 # include <netinet/tcp.h>
  42
  43 /* Hack to work around the missing define in AIX */
  44 #ifndef MSG_DONTWAIT
  45 # define MSG_DONTWAIT MSG_NONBLOCK
  46 #endif
  47
  48 #define MEMCACHED_DEF_HOST "127.0.0.1"
  49 #define MEMCACHED_DEF_PORT "11211"
  50
  51 #define MEMCACHED_RETRY_COUNT 100
  52
  53 struct memcached_s
  54 {
  55   char *name;
  56   char *socket;
  57   char *host;
  58   char *port;
  59 };
  60
  61 typedef struct memcached_s memcached_t;
  62
  63 static int memcached_read (user_data_t *user_data);
  64
  65 static void memcached_free (memcached_t *st)
  66 {
  67   if (st == NULL)
  68     return;
  69
  70   sfree (st->name);
  71   sfree (st->socket);
  72   sfree (st->host);
  73   sfree (st->port);
  74 }
  75
  76 static int memcached_query_daemon (char *buffer, int buffer_size, user_data_t *user_data)
  77 {
  78   int fd;
  79   ssize_t status;
  80   int buffer_fill;
  81   int i = 0;
  82
  83   memcached_t *st;
  84   st = user_data->data;
  85   if (st->socket != NULL) {
  86     struct sockaddr_un serv_addr;
  87
  88      memset (&serv_addr, 0, sizeof (serv_addr));
  89      serv_addr.sun_family = AF_UNIX;
  90      sstrncpy (serv_addr.sun_path, st->socket,
  91      sizeof (serv_addr.sun_path));
  92
  93      /* create our socket descriptor */
  94      fd = socket (AF_UNIX, SOCK_STREAM, 0);
  95      if (fd < 0) {
  96        char errbuf[1024];
  97        ERROR ("memcached: unix socket: %s", sstrerror (errno, errbuf,
  98        sizeof (errbuf)));
  99        return -1;
 100      }
 101   }
 102   else {
 103     const char *host;
 104     const char *port;
 105
 106     struct addrinfo  ai_hints;
 107     struct addrinfo *ai_list, *ai_ptr;
 108     int              ai_return = 0;
 109
 110     memset (&ai_hints, '\0', sizeof (ai_hints));
 111     ai_hints.ai_flags    = 0;
 112 #ifdef AI_ADDRCONFIG
 113     ai_hints.ai_flags   |= AI_ADDRCONFIG;
 114 #endif
 115     ai_hints.ai_family   = AF_UNSPEC;
 116     ai_hints.ai_socktype = SOCK_STREAM;
 117     ai_hints.ai_protocol = 0;
 118
 119     host = (st->host != NULL) ? st->host : MEMCACHED_DEF_HOST;
 120     port = (st->port != NULL) ? st->port : MEMCACHED_DEF_PORT;
 121
 122     if ((ai_return = getaddrinfo (host, port, &ai_hints, &ai_list)) != 0) {
 123       char errbuf[1024];
 124       ERROR ("memcached: getaddrinfo (%s, %s): %s",
 125           host, port,
 126           (ai_return == EAI_SYSTEM)
 127           ? sstrerror (errno, errbuf, sizeof (errbuf))
 128           : gai_strerror (ai_return));
 129       return -1;
 130     }
 131
 132     for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) {
 133       /* create our socket descriptor */
 134       fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
 135       if (fd < 0) {
 136         char errbuf[1024];
 137         ERROR ("memcached: socket: %s", sstrerror (errno, errbuf, sizeof (errbuf)));
 138         continue;
 139       }
 140
 141       /* connect to the memcached daemon */
 142       status = (ssize_t) connect (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 143       if (status != 0) {
 144         shutdown (fd, SHUT_RDWR);
 145         close (fd);
 146         fd = -1;
 147         continue;
 148       }
 149
 150       /* A socket could be opened and connecting succeeded. We're
 151        * done. */
 152       break;
 153     }
 154
 155     freeaddrinfo (ai_list);
 156   }
 157
 158   if (fd < 0) {
 159     ERROR ("memcached: Could not connect to daemon.");
 160     return -1;
 161   }
 162
 163   if (send(fd, "stats\r\n", sizeof("stats\r\n") - 1, MSG_DONTWAIT) != (sizeof("stats\r\n") - 1)) {
 164     ERROR ("memcached: Could not send command to the memcached daemon.");
 165     return -1;
 166   }
 167
 168   {
 169     struct pollfd p;
 170     int status;
 171
 172     memset (&p, 0, sizeof (p));
 173     p.fd = fd;
 174     p.events = POLLIN | POLLERR | POLLHUP;
 175     p.revents = 0;
 176
 177     status = poll (&p, /* nfds = */ 1,
 178         /* timeout = */ CDTIME_T_TO_MS (interval_g));
 179     if (status <= 0)
 180     {
 181       if (status == 0)
 182       {
 183         ERROR ("memcached: poll(2) timed out after %.3f seconds.",
 184             CDTIME_T_TO_DOUBLE (interval_g));
 185       }
 186       else
 187       {
 188         char errbuf[1024];
 189         ERROR ("memcached: poll(2) failed: %s",
 190             sstrerror (errno, errbuf, sizeof (errbuf)));
 191       }
 192       shutdown (fd, SHUT_RDWR);
 193       close (fd);
 194       return (-1);
 195     }
 196   }
 197
 198   /* receive data from the memcached daemon */
 199   memset (buffer, '\0', buffer_size);
 200
 201   buffer_fill = 0;
 202   while ((status = recv (fd, buffer + buffer_fill, buffer_size - buffer_fill, MSG_DONTWAIT)) != 0) {
 203     if (i > MEMCACHED_RETRY_COUNT) {
 204       ERROR("recv() timed out");
 205       break;
 206     }
 207     i++;
 208
 209     if (status == -1) {
 210       char errbuf[1024];
 211
 212       if (errno == EAGAIN) {
 213         continue;
 214       }
 215
 216       ERROR ("memcached: Error reading from socket: %s",
 217           sstrerror (errno, errbuf, sizeof (errbuf)));
 218       shutdown(fd, SHUT_RDWR);
 219       close (fd);
 220       return -1;
 221     }
 222     buffer_fill += status;
 223
 224     if (buffer_fill > 3 && buffer[buffer_fill-5] == 'E' && buffer[buffer_fill-4] == 'N' && buffer[buffer_fill-3] == 'D') {
 225       /* we got all the data */
 226       break;
 227     }
 228   }
 229
 230   if (buffer_fill >= buffer_size) {
 231     buffer[buffer_size - 1] = '\0';
 232     WARNING ("memcached: Message from memcached has been truncated.");
 233   } else if (buffer_fill == 0) {
 234     WARNING ("memcached: Peer has unexpectedly shut down the socket. "
 235         "Buffer: `%s'", buffer);
 236     shutdown(fd, SHUT_RDWR);
 237     close(fd);
 238     return -1;
 239   }
 240
 241   shutdown(fd, SHUT_RDWR);
 242   close(fd);
 243   return 0;
 244 }
 245
 246 /* Configuration handling functiions
 247  * <Plugin memcached>
 248  *   <Instance "instance_name">
 249  *     Host foo.zomg.com
 250  *     Port "1234"
 251  *   </Instance>
 252  * </Plugin>
 253  */
 254 static int config_add_instance(oconfig_item_t *ci)
 255 {
 256   memcached_t *st;
 257   int i;
 258   int status;
 259
 260   if ((ci->values_num != 1)
 261     || (ci->values[0].type != OCONFIG_TYPE_STRING))
 262   {
 263     WARNING ("memcached plugin: The `%s' config option "
 264       "needs exactly one string argument.", ci->key);
 265     return (-1);
 266   }
 267
 268   st = (memcached_t *) malloc (sizeof (*st));
 269   if (st == NULL)
 270   {
 271     ERROR ("memcached plugin: malloc failed.");
 272     return (-1);
 273   }
 274
 275   st->name = NULL;
 276   st->socket = NULL;
 277   st->host = NULL;
 278   st->port = NULL;
 279   memset (st, 0, sizeof (*st));
 280
 281   status = cf_util_get_string (ci, &st->name);
 282   if (status != 0)
 283   {
 284     sfree (st);
 285     return (status);
 286   }
 287   assert (st->name != NULL);
 288
 289   for (i = 0; i < ci->children_num; i++)
 290   {
 291     oconfig_item_t *child = ci->children + i;
 292
 293     if (strcasecmp ("Socket", child->key) == 0)
 294       status = cf_util_get_string (child, &st->socket);
 295     else if (strcasecmp ("Host", child->key) == 0)
 296       status = cf_util_get_string (child, &st->host);
 297     else if (strcasecmp ("Port", child->key) == 0)
 298       status = cf_util_get_service (child, &st->port);
 299     else
 300     {
 301       WARNING ("memcached plugin: Option `%s' not allowed here.",
 302           child->key);
 303       status = -1;
 304     }
 305
 306     if (status != 0)
 307       break;
 308   }
 309
 310   if (status == 0)
 311   {
 312     user_data_t ud;
 313     char callback_name[3*DATA_MAX_NAME_LEN];
 314
 315     memset (&ud, 0, sizeof (ud));
 316     ud.data = st;
 317     ud.free_func = (void *) memcached_free;
 318
 319     memset (callback_name, 0, sizeof (callback_name));
 320     ssnprintf (callback_name, sizeof (callback_name),
 321         "memcached/%s/%s",
 322         (st->host != NULL) ? st->host : MEMCACHED_DEF_HOST,
 323         (st->port != NULL) ? st->port : MEMCACHED_DEF_PORT)
 324
 325     status = plugin_register_complex_read (/* group = */ "memcached",
 326         /* name      = */ callback_name,
 327         /* callback  = */ memcached_read,
 328         /* interval  = */ NULL,
 329         /* user_data = */ &ud);
 330   }
 331
 332   if (status != 0)
 333   {
 334     memcached_free(st);
 335     return (-1);
 336   }
 337
 338   return (0);
 339 }
 340
 341 static int memcached_config (oconfig_item_t *ci)
 342 {
 343   int status = 0;
 344   int i;
 345
 346   for (i = 0; i < ci->children_num; i++)
 347   {
 348     oconfig_item_t *child = ci->children + i;
 349
 350     if (strcasecmp ("Instance", child->key) == 0)
 351       config_add_instance (child);
 352     else
 353       WARNING ("memcached plugin: The configuration option "
 354           "\"%s\" is not allowed here. Did you "
 355           "forget to add an <Instance /> block "
 356           "around the configuration?",
 357           child->key);
 358   } /* for (ci->children) */
 359
 360   return (status);
 361 }
 362
 363 static void submit_derive (const char *type, const char *type_inst,
 364     derive_t value, memcached_t *st)
 365 {
 366   value_t values[1];
 367   value_list_t vl = VALUE_LIST_INIT;
 368
 369   values[0].derive = value;
 370
 371   vl.values = values;
 372   vl.values_len = 1;
 373   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 374   sstrncpy (vl.plugin, "memcached", sizeof (vl.plugin));
 375   if (st->name != NULL)
 376     sstrncpy (vl.plugin_instance, st->name,  sizeof (vl.plugin_instance));
 377   sstrncpy (vl.type, type, sizeof (vl.type));
 378   if (type_inst != NULL)
 379     sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
 380
 381   plugin_dispatch_values (&vl);
 382 }
 383
 384 static void submit_derive2 (const char *type, const char *type_inst,
 385     derive_t value0, derive_t value1, memcached_t *st)
 386 {
 387   value_t values[2];
 388   value_list_t vl = VALUE_LIST_INIT;
 389
 390   values[0].derive = value0;
 391   values[1].derive = value1;
 392
 393   vl.values = values;
 394   vl.values_len = 2;
 395   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 396   sstrncpy (vl.plugin, "memcached", sizeof (vl.plugin));
 397   if (st->name != NULL)
 398     sstrncpy (vl.plugin_instance, st->name,  sizeof (vl.plugin_instance));
 399   sstrncpy (vl.type, type, sizeof (vl.type));
 400   if (type_inst != NULL)
 401     sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
 402
 403   plugin_dispatch_values (&vl);
 404 }
 405
 406 static void submit_gauge (const char *type, const char *type_inst,
 407     gauge_t value, memcached_t *st)
 408 {
 409   value_t values[1];
 410   value_list_t vl = VALUE_LIST_INIT;
 411
 412   values[0].gauge = value;
 413
 414   vl.values = values;
 415   vl.values_len = 1;
 416   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 417   sstrncpy (vl.plugin, "memcached", sizeof (vl.plugin));
 418   if (st->name != NULL)
 419     sstrncpy (vl.plugin_instance, st->name,  sizeof (vl.plugin_instance));
 420   sstrncpy (vl.type, type, sizeof (vl.type));
 421   if (type_inst != NULL)
 422     sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
 423
 424   plugin_dispatch_values (&vl);
 425 }
 426
 427 static void submit_gauge2 (const char *type, const char *type_inst,
 428     gauge_t value0, gauge_t value1, memcached_t *st)
 429 {
 430   value_t values[2];
 431   value_list_t vl = VALUE_LIST_INIT;
 432
 433   values[0].gauge = value0;
 434   values[1].gauge = value1;
 435
 436   vl.values = values;
 437   vl.values_len = 2;
 438   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 439   sstrncpy (vl.plugin, "memcached", sizeof (vl.plugin));
 440   if (st->name != NULL)
 441     sstrncpy (vl.plugin_instance, st->name,  sizeof (vl.plugin_instance));
 442   sstrncpy (vl.type, type, sizeof (vl.type));
 443   if (type_inst != NULL)
 444     sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
 445
 446   plugin_dispatch_values (&vl);
 447 }
 448
 449 static int memcached_read (user_data_t *user_data)
 450 {
 451   char buf[4096];
 452   char *fields[3];
 453   char *ptr;
 454   char *line;
 455   char *saveptr;
 456   int fields_num;
 457
 458   gauge_t bytes_used = NAN;
 459   gauge_t bytes_total = NAN;
 460   gauge_t hits = NAN;
 461   gauge_t gets = NAN;
 462   derive_t rusage_user = 0;
 463   derive_t rusage_syst = 0;
 464   derive_t octets_rx = 0;
 465   derive_t octets_tx = 0;
 466
 467   memcached_t *st;
 468   st = user_data->data;
 469
 470   /* get data from daemon */
 471   if (memcached_query_daemon (buf, sizeof (buf), user_data) < 0) {
 472     return -1;
 473   }
 474
 475 #define FIELD_IS(cnst) \
 476   (((sizeof(cnst) - 1) == name_len) && (strcmp (cnst, fields[1]) == 0))
 477
 478   ptr = buf;
 479   saveptr = NULL;
 480   while ((line = strtok_r (ptr, "\n\r", &saveptr)) != NULL)
 481   {
 482     int name_len;
 483
 484     ptr = NULL;
 485
 486     fields_num = strsplit(line, fields, 3);
 487     if (fields_num != 3)
 488       continue;
 489
 490     name_len = strlen(fields[1]);
 491     if (name_len == 0)
 492       continue;
 493
 494     /*
 495      * For an explanation on these fields please refer to
 496      * <http://code.sixapart.com/svn/memcached/trunk/server/doc/protocol.txt>
 497      */
 498
 499     /*
 500      * CPU time consumed by the memcached process
 501      */
 502     if (FIELD_IS ("rusage_user"))
 503     {
 504       rusage_user = atoll (fields[2]);
 505     }
 506     else if (FIELD_IS ("rusage_system"))
 507     {
 508       rusage_syst = atoll(fields[2]);
 509     }
 510
 511     /*
 512      * Number of threads of this instance
 513      */
 514     else if (FIELD_IS ("threads"))
 515     {
 516       submit_gauge2 ("ps_count", NULL, NAN, atof (fields[2]), st);
 517     }
 518
 519     /*
 520      * Number of items stored
 521      */
 522     else if (FIELD_IS ("curr_items"))
 523     {
 524       submit_gauge ("memcached_items", "current", atof (fields[2]), st);
 525     }
 526
 527     /*
 528      * Number of bytes used and available (total - used)
 529      */
 530     else if (FIELD_IS ("bytes"))
 531     {
 532       bytes_used = atof (fields[2]);
 533     }
 534     else if (FIELD_IS ("limit_maxbytes"))
 535     {
 536       bytes_total = atof(fields[2]);
 537     }
 538
 539     /*
 540      * Connections
 541      */
 542     else if (FIELD_IS ("curr_connections"))
 543     {
 544       submit_gauge ("memcached_connections", "current", atof (fields[2]), st);
 545     }
 546
 547     /*
 548      * Commands
 549      */
 550     else if ((name_len > 4) && (strncmp (fields[1], "cmd_", 4) == 0))
 551     {
 552       const char *name = fields[1] + 4;
 553       submit_derive ("memcached_command", name, atoll (fields[2]), st);
 554       if (strcmp (name, "get") == 0)
 555         gets = atof (fields[2]);
 556     }
 557
 558     /*
 559      * Operations on the cache, i. e. cache hits, cache misses and evictions of items
 560      */
 561     else if (FIELD_IS ("get_hits"))
 562     {
 563       submit_derive ("memcached_ops", "hits", atoll (fields[2]), st);
 564       hits = atof (fields[2]);
 565     }
 566     else if (FIELD_IS ("get_misses"))
 567     {
 568       submit_derive ("memcached_ops", "misses", atoll (fields[2]), st);
 569     }
 570     else if (FIELD_IS ("evictions"))
 571     {
 572       submit_derive ("memcached_ops", "evictions", atoll (fields[2]), st);
 573     }
 574
 575     /*
 576      * Network traffic
 577      */
 578     else if (FIELD_IS ("bytes_read"))
 579     {
 580       octets_rx = atoll (fields[2]);
 581     }
 582     else if (FIELD_IS ("bytes_written"))
 583     {
 584       octets_tx = atoll (fields[2]);
 585     }
 586   } /* while ((line = strtok_r (ptr, "\n\r", &saveptr)) != NULL) */
 587
 588   if (!isnan (bytes_used) && !isnan (bytes_total) && (bytes_used <= bytes_total))
 589     submit_gauge2 ("df", "cache", bytes_used, bytes_total - bytes_used, st);
 590
 591   if ((rusage_user != 0) || (rusage_syst != 0))
 592     submit_derive2 ("ps_cputime", NULL, rusage_user, rusage_syst, st);
 593
 594   if ((octets_rx != 0) || (octets_tx != 0))
 595     submit_derive2 ("memcached_octets", NULL, octets_rx, octets_tx, st);
 596
 597   if (!isnan (gets) && !isnan (hits))
 598   {
 599     gauge_t rate = NAN;
 600
 601     if (gets != 0.0)
 602       rate = 100.0 * hits / gets;
 603
 604     submit_gauge ("percent", "hitratio", rate, st);
 605   }
 606
 607   return 0;
 608 }
 609
 610 void module_register (void)
 611 {
 612   plugin_register_complex_config ("memcached", memcached_config);
 613 }