src/memcached.c

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