collectd.spec: Don't enable XFS support on RHEL6
[collectd.git] / src / memcached.c
1 /**
2  * collectd - src/memcached.c, based on src/hddtemp.c
3  * Copyright (C) 2007       Antony Dovgal
4  * Copyright (C) 2007-2012  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
33 #include "common.h"
34 #include "plugin.h"
35
36 #include <netdb.h>
37 #include <netinet/in.h>
38 #include <netinet/tcp.h>
39 #include <sys/un.h>
40
41 #define MEMCACHED_DEF_HOST "127.0.0.1"
42 #define MEMCACHED_DEF_PORT "11211"
43
44 struct memcached_s {
45   char *name;
46   char *socket;
47   char *host;
48   char *port;
49 };
50 typedef struct memcached_s memcached_t;
51
52 static _Bool memcached_have_instances = 0;
53
54 static void memcached_free(void *arg) {
55   memcached_t *st = arg;
56   if (st == NULL)
57     return;
58
59   sfree(st->name);
60   sfree(st->socket);
61   sfree(st->host);
62   sfree(st->port);
63   sfree(st);
64 }
65
66 static int memcached_connect_unix(memcached_t *st) {
67   struct sockaddr_un serv_addr = {0};
68   int fd;
69
70   serv_addr.sun_family = AF_UNIX;
71   sstrncpy(serv_addr.sun_path, st->socket, 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 plugin: memcached_connect_unix: socket(2) failed: %s",
78           sstrerror(errno, errbuf, sizeof(errbuf)));
79     return (-1);
80   }
81
82   /* connect to the memcached daemon */
83   int status = connect(fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
84   if (status != 0) {
85     shutdown(fd, SHUT_RDWR);
86     close(fd);
87     fd = -1;
88   }
89
90   return (fd);
91 } /* int memcached_connect_unix */
92
93 static int memcached_connect_inet(memcached_t *st) {
94   const char *host;
95   const char *port;
96
97   struct addrinfo *ai_list;
98   int status;
99   int fd = -1;
100
101   host = (st->host != NULL) ? st->host : MEMCACHED_DEF_HOST;
102   port = (st->port != NULL) ? st->port : MEMCACHED_DEF_PORT;
103
104   struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
105                               .ai_flags = AI_ADDRCONFIG,
106                               .ai_socktype = SOCK_STREAM};
107
108   status = getaddrinfo(host, port, &ai_hints, &ai_list);
109   if (status != 0) {
110     char errbuf[1024];
111     ERROR("memcached plugin: memcached_connect_inet: "
112           "getaddrinfo(%s,%s) failed: %s",
113           host, port,
114           (status == EAI_SYSTEM) ? sstrerror(errno, errbuf, sizeof(errbuf))
115                                  : gai_strerror(status));
116     return (-1);
117   }
118
119   for (struct addrinfo *ai_ptr = ai_list; ai_ptr != NULL;
120        ai_ptr = ai_ptr->ai_next) {
121     /* create our socket descriptor */
122     fd = socket(ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
123     if (fd < 0) {
124       char errbuf[1024];
125       WARNING("memcached plugin: memcached_connect_inet: "
126               "socket(2) failed: %s",
127               sstrerror(errno, errbuf, sizeof(errbuf)));
128       continue;
129     }
130
131     /* connect to the memcached daemon */
132     status = (int)connect(fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
133     if (status != 0) {
134       shutdown(fd, SHUT_RDWR);
135       close(fd);
136       fd = -1;
137       continue;
138     }
139
140     /* A socket could be opened and connecting succeeded. We're done. */
141     break;
142   }
143
144   freeaddrinfo(ai_list);
145   return (fd);
146 } /* int memcached_connect_inet */
147
148 static int memcached_connect(memcached_t *st) {
149   if (st->socket != NULL)
150     return (memcached_connect_unix(st));
151   else
152     return (memcached_connect_inet(st));
153 }
154
155 static int memcached_query_daemon(char *buffer, size_t buffer_size,
156                                   memcached_t *st) {
157   int fd, status;
158   size_t buffer_fill;
159
160   fd = memcached_connect(st);
161   if (fd < 0) {
162     ERROR("memcached plugin: Instance \"%s\" could not connect to daemon.",
163           st->name);
164     return -1;
165   }
166
167   status = (int)swrite(fd, "stats\r\n", strlen("stats\r\n"));
168   if (status != 0) {
169     char errbuf[1024];
170     ERROR("memcached plugin: write(2) failed: %s",
171           sstrerror(errno, errbuf, sizeof(errbuf)));
172     shutdown(fd, SHUT_RDWR);
173     close(fd);
174     return (-1);
175   }
176
177   /* receive data from the memcached daemon */
178   memset(buffer, 0, buffer_size);
179
180   buffer_fill = 0;
181   while ((status = (int)recv(fd, buffer + buffer_fill,
182                              buffer_size - buffer_fill, /* flags = */ 0)) !=
183          0) {
184     char const end_token[5] = {'E', 'N', 'D', '\r', '\n'};
185     if (status < 0) {
186       char errbuf[1024];
187
188       if ((errno == EAGAIN) || (errno == EINTR))
189         continue;
190
191       ERROR("memcached: Error reading from socket: %s",
192             sstrerror(errno, errbuf, sizeof(errbuf)));
193       shutdown(fd, SHUT_RDWR);
194       close(fd);
195       return (-1);
196     }
197
198     buffer_fill += (size_t)status;
199     if (buffer_fill > buffer_size) {
200       buffer_fill = buffer_size;
201       WARNING("memcached plugin: Message was truncated.");
202       break;
203     }
204
205     /* If buffer ends in end_token, we have all the data. */
206     if (memcmp(buffer + buffer_fill - sizeof(end_token), end_token,
207                sizeof(end_token)) == 0)
208       break;
209   } /* while (recv) */
210
211   status = 0;
212   if (buffer_fill == 0) {
213     WARNING("memcached plugin: No data returned by memcached.");
214     status = -1;
215   }
216
217   shutdown(fd, SHUT_RDWR);
218   close(fd);
219   return (status);
220 } /* int memcached_query_daemon */
221
222 static void memcached_init_vl(value_list_t *vl, memcached_t const *st) {
223   char const *host = st->host;
224
225   /* Set vl->host to hostname_g, if:
226    * - Legacy mode is used.
227    * - "Socket" option is given (doc: "Host option is ignored").
228    * - "Host" option is not provided.
229    * - "Host" option is set to "localhost" or "127.0.0.1". */
230   if ((strcmp(st->name, "__legacy__") == 0) || (st->socket != NULL) ||
231       (st->host == NULL) || (strcmp("127.0.0.1", st->host) == 0) ||
232       (strcmp("localhost", st->host) == 0))
233     host = hostname_g;
234
235   sstrncpy(vl->plugin, "memcached", sizeof(vl->plugin));
236   sstrncpy(vl->host, host, sizeof(vl->host));
237   if (strcmp(st->name, "__legacy__") != 0)
238     sstrncpy(vl->plugin_instance, st->name, sizeof(vl->plugin_instance));
239 }
240
241 static void submit_derive(const char *type, const char *type_inst,
242                           derive_t value, memcached_t *st) {
243   value_t values[1];
244   value_list_t vl = VALUE_LIST_INIT;
245   memcached_init_vl(&vl, st);
246
247   values[0].derive = value;
248
249   vl.values = values;
250   vl.values_len = 1;
251   sstrncpy(vl.type, type, sizeof(vl.type));
252   if (type_inst != NULL)
253     sstrncpy(vl.type_instance, type_inst, sizeof(vl.type_instance));
254
255   plugin_dispatch_values(&vl);
256 }
257
258 static void submit_derive2(const char *type, const char *type_inst,
259                            derive_t value0, derive_t value1, memcached_t *st) {
260   value_t values[2];
261   value_list_t vl = VALUE_LIST_INIT;
262   memcached_init_vl(&vl, st);
263
264   values[0].derive = value0;
265   values[1].derive = value1;
266
267   vl.values = values;
268   vl.values_len = 2;
269   sstrncpy(vl.type, type, sizeof(vl.type));
270   if (type_inst != NULL)
271     sstrncpy(vl.type_instance, type_inst, sizeof(vl.type_instance));
272
273   plugin_dispatch_values(&vl);
274 }
275
276 static void submit_gauge(const char *type, const char *type_inst, gauge_t value,
277                          memcached_t *st) {
278   value_t values[1];
279   value_list_t vl = VALUE_LIST_INIT;
280   memcached_init_vl(&vl, st);
281
282   values[0].gauge = value;
283
284   vl.values = values;
285   vl.values_len = 1;
286   sstrncpy(vl.type, type, sizeof(vl.type));
287   if (type_inst != NULL)
288     sstrncpy(vl.type_instance, type_inst, sizeof(vl.type_instance));
289
290   plugin_dispatch_values(&vl);
291 }
292
293 static void submit_gauge2(const char *type, const char *type_inst,
294                           gauge_t value0, gauge_t value1, memcached_t *st) {
295   value_t values[2];
296   value_list_t vl = VALUE_LIST_INIT;
297   memcached_init_vl(&vl, st);
298
299   values[0].gauge = value0;
300   values[1].gauge = value1;
301
302   vl.values = values;
303   vl.values_len = 2;
304   sstrncpy(vl.type, type, sizeof(vl.type));
305   if (type_inst != NULL)
306     sstrncpy(vl.type_instance, type_inst, sizeof(vl.type_instance));
307
308   plugin_dispatch_values(&vl);
309 }
310
311 static int memcached_read(user_data_t *user_data) {
312   char buf[4096];
313   char *fields[3];
314   char *ptr;
315   char *line;
316   char *saveptr;
317   int fields_num;
318
319   gauge_t bytes_used = NAN;
320   gauge_t bytes_total = NAN;
321   gauge_t hits = NAN;
322   gauge_t gets = NAN;
323   gauge_t incr_hits = NAN;
324   derive_t incr = 0;
325   gauge_t decr_hits = NAN;
326   derive_t decr = 0;
327   derive_t rusage_user = 0;
328   derive_t rusage_syst = 0;
329   derive_t octets_rx = 0;
330   derive_t octets_tx = 0;
331
332   memcached_t *st;
333   st = user_data->data;
334
335   /* get data from daemon */
336   if (memcached_query_daemon(buf, sizeof(buf), st) < 0) {
337     return -1;
338   }
339
340 #define FIELD_IS(cnst)                                                         \
341   (((sizeof(cnst) - 1) == name_len) && (strcmp(cnst, fields[1]) == 0))
342
343   ptr = buf;
344   saveptr = NULL;
345   while ((line = strtok_r(ptr, "\n\r", &saveptr)) != NULL) {
346     int name_len;
347
348     ptr = NULL;
349
350     fields_num = strsplit(line, fields, 3);
351     if (fields_num != 3)
352       continue;
353
354     name_len = strlen(fields[1]);
355     if (name_len == 0)
356       continue;
357
358     /*
359      * For an explanation on these fields please refer to
360      * <https://github.com/memcached/memcached/blob/master/doc/protocol.txt>
361      */
362
363     /*
364      * CPU time consumed by the memcached process
365      */
366     if (FIELD_IS("rusage_user")) {
367       rusage_user = atoll(fields[2]);
368     } else if (FIELD_IS("rusage_system")) {
369       rusage_syst = atoll(fields[2]);
370     }
371
372     /*
373      * Number of threads of this instance
374      */
375     else if (FIELD_IS("threads")) {
376       submit_gauge2("ps_count", NULL, NAN, atof(fields[2]), st);
377     }
378
379     /*
380      * Number of items stored
381      */
382     else if (FIELD_IS("curr_items")) {
383       submit_gauge("memcached_items", "current", atof(fields[2]), st);
384     }
385
386     /*
387      * Number of bytes used and available (total - used)
388      */
389     else if (FIELD_IS("bytes")) {
390       bytes_used = atof(fields[2]);
391     } else if (FIELD_IS("limit_maxbytes")) {
392       bytes_total = atof(fields[2]);
393     }
394
395     /*
396      * Connections
397      */
398     else if (FIELD_IS("curr_connections")) {
399       submit_gauge("memcached_connections", "current", atof(fields[2]), st);
400     } else if (FIELD_IS("listen_disabled_num")) {
401       submit_derive("connections", "listen_disabled", atof(fields[2]), st);
402     }
403
404     /*
405      * Commands
406      */
407     else if ((name_len > 4) && (strncmp(fields[1], "cmd_", 4) == 0)) {
408       const char *name = fields[1] + 4;
409       submit_derive("memcached_command", name, atoll(fields[2]), st);
410       if (strcmp(name, "get") == 0)
411         gets = atof(fields[2]);
412     }
413
414     /*
415      * Increment/Decrement
416      */
417     else if (FIELD_IS("incr_misses")) {
418       derive_t incr_count = atoll(fields[2]);
419       submit_derive("memcached_ops", "incr_misses", incr_count, st);
420       incr += incr_count;
421     } else if (FIELD_IS("incr_hits")) {
422       derive_t incr_count = atoll(fields[2]);
423       submit_derive("memcached_ops", "incr_hits", incr_count, st);
424       incr_hits = atof(fields[2]);
425       incr += incr_count;
426     } else if (FIELD_IS("decr_misses")) {
427       derive_t decr_count = atoll(fields[2]);
428       submit_derive("memcached_ops", "decr_misses", decr_count, st);
429       decr += decr_count;
430     } else if (FIELD_IS("decr_hits")) {
431       derive_t decr_count = atoll(fields[2]);
432       submit_derive("memcached_ops", "decr_hits", decr_count, st);
433       decr_hits = atof(fields[2]);
434       decr += decr_count;
435     }
436
437     /*
438      * Operations on the cache, i. e. cache hits, cache misses and evictions of
439      * items
440      */
441     else if (FIELD_IS("get_hits")) {
442       submit_derive("memcached_ops", "hits", atoll(fields[2]), st);
443       hits = atof(fields[2]);
444     } else if (FIELD_IS("get_misses")) {
445       submit_derive("memcached_ops", "misses", atoll(fields[2]), st);
446     } else if (FIELD_IS("evictions")) {
447       submit_derive("memcached_ops", "evictions", atoll(fields[2]), st);
448     }
449
450     /*
451      * Network traffic
452      */
453     else if (FIELD_IS("bytes_read")) {
454       octets_rx = atoll(fields[2]);
455     } else if (FIELD_IS("bytes_written")) {
456       octets_tx = atoll(fields[2]);
457     }
458   } /* while ((line = strtok_r (ptr, "\n\r", &saveptr)) != NULL) */
459
460   if (!isnan(bytes_used) && !isnan(bytes_total) && (bytes_used <= bytes_total))
461     submit_gauge2("df", "cache", bytes_used, bytes_total - bytes_used, st);
462
463   if ((rusage_user != 0) || (rusage_syst != 0))
464     submit_derive2("ps_cputime", NULL, rusage_user, rusage_syst, st);
465
466   if ((octets_rx != 0) || (octets_tx != 0))
467     submit_derive2("memcached_octets", NULL, octets_rx, octets_tx, st);
468
469   if (!isnan(gets) && !isnan(hits)) {
470     gauge_t rate = NAN;
471
472     if (gets != 0.0)
473       rate = 100.0 * hits / gets;
474
475     submit_gauge("percent", "hitratio", rate, st);
476   }
477
478   if (!isnan(incr_hits) && incr != 0) {
479     gauge_t incr_rate = 100.0 * incr_hits / incr;
480     submit_gauge("percent", "incr_hitratio", incr_rate, st);
481     submit_derive("memcached_ops", "incr", incr, st);
482   }
483
484   if (!isnan(decr_hits) && decr != 0) {
485     gauge_t decr_rate = 100.0 * decr_hits / decr;
486     submit_gauge("percent", "decr_hitratio", decr_rate, st);
487     submit_derive("memcached_ops", "decr", decr, st);
488   }
489
490   return 0;
491 } /* int memcached_read */
492
493 static int memcached_add_read_callback(memcached_t *st) {
494   char callback_name[3 * DATA_MAX_NAME_LEN];
495   int status;
496
497   assert(st->name != NULL);
498   ssnprintf(callback_name, sizeof(callback_name), "memcached/%s", st->name);
499
500   user_data_t ud = {.data = st, .free_func = memcached_free};
501
502   status = plugin_register_complex_read(/* group = */ "memcached",
503                                         /* name      = */ callback_name,
504                                         /* callback  = */ memcached_read,
505                                         /* interval  = */ 0,
506                                         /* user_data = */ &ud);
507   return (status);
508 } /* int memcached_add_read_callback */
509
510 /* Configuration handling functiions
511  * <Plugin memcached>
512  *   <Instance "instance_name">
513  *     Host foo.zomg.com
514  *     Port "1234"
515  *   </Instance>
516  * </Plugin>
517  */
518 static int config_add_instance(oconfig_item_t *ci) {
519   memcached_t *st;
520   int status = 0;
521
522   /* Disable automatic generation of default instance in the init callback. */
523   memcached_have_instances = 1;
524
525   st = calloc(1, sizeof(*st));
526   if (st == NULL) {
527     ERROR("memcached plugin: calloc failed.");
528     return (-1);
529   }
530
531   st->name = NULL;
532   st->socket = NULL;
533   st->host = NULL;
534   st->port = NULL;
535
536   if (strcasecmp(ci->key, "Plugin") == 0) /* default instance */
537     st->name = sstrdup("__legacy__");
538   else /* <Instance /> block */
539     status = cf_util_get_string(ci, &st->name);
540   if (status != 0) {
541     sfree(st);
542     return (status);
543   }
544   assert(st->name != NULL);
545
546   for (int i = 0; i < ci->children_num; i++) {
547     oconfig_item_t *child = ci->children + i;
548
549     if (strcasecmp("Socket", child->key) == 0)
550       status = cf_util_get_string(child, &st->socket);
551     else if (strcasecmp("Host", child->key) == 0)
552       status = cf_util_get_string(child, &st->host);
553     else if (strcasecmp("Port", child->key) == 0)
554       status = cf_util_get_service(child, &st->port);
555     else {
556       WARNING("memcached plugin: Option `%s' not allowed here.", child->key);
557       status = -1;
558     }
559
560     if (status != 0)
561       break;
562   }
563
564   if (status == 0)
565     status = memcached_add_read_callback(st);
566
567   if (status != 0) {
568     memcached_free(st);
569     return (-1);
570   }
571
572   return (0);
573 }
574
575 static int memcached_config(oconfig_item_t *ci) {
576   int status = 0;
577   _Bool have_instance_block = 0;
578
579   for (int i = 0; i < ci->children_num; i++) {
580     oconfig_item_t *child = ci->children + i;
581
582     if (strcasecmp("Instance", child->key) == 0) {
583       config_add_instance(child);
584       have_instance_block = 1;
585     } else if (!have_instance_block) {
586       /* Non-instance option: Assume legacy configuration (without <Instance />
587        * blocks) and call config_add_instance() with the <Plugin /> block. */
588       return (config_add_instance(ci));
589     } else
590       WARNING("memcached plugin: The configuration option "
591               "\"%s\" is not allowed here. Did you "
592               "forget to add an <Instance /> block "
593               "around the configuration?",
594               child->key);
595   } /* for (ci->children) */
596
597   return (status);
598 }
599
600 static int memcached_init(void) {
601   memcached_t *st;
602   int status;
603
604   if (memcached_have_instances)
605     return (0);
606
607   /* No instances were configured, lets start a default instance. */
608   st = calloc(1, sizeof(*st));
609   if (st == NULL)
610     return (ENOMEM);
611   st->name = sstrdup("__legacy__");
612   st->socket = NULL;
613   st->host = NULL;
614   st->port = NULL;
615
616   status = memcached_add_read_callback(st);
617   if (status == 0)
618     memcached_have_instances = 1;
619   else
620     memcached_free(st);
621
622   return (status);
623 } /* int memcached_init */
624
625 void module_register(void) {
626   plugin_register_complex_config("memcached", memcached_config);
627   plugin_register_init("memcached", memcached_init);
628 }