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