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