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