In rrd synchronous file creation use lock_file
[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/socket.h>
38 #include <sys/un.h>
39 #include <netinet/in.h>
40 #include <netinet/tcp.h>
41
42 #define MEMCACHED_DEF_HOST "127.0.0.1"
43 #define MEMCACHED_DEF_PORT "11211"
44
45 struct memcached_s
46 {
47   char *name;
48   char *socket;
49   char *host;
50   char *port;
51 };
52 typedef struct memcached_s memcached_t;
53
54 static _Bool memcached_have_instances = 0;
55
56 static void memcached_free (memcached_t *st)
57 {
58   if (st == NULL)
59     return;
60
61   sfree (st->name);
62   sfree (st->socket);
63   sfree (st->host);
64   sfree (st->port);
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      * <http://code.sixapart.com/svn/memcached/trunk/server/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
441     /*
442      * Commands
443      */
444     else if ((name_len > 4) && (strncmp (fields[1], "cmd_", 4) == 0))
445     {
446       const char *name = fields[1] + 4;
447       submit_derive ("memcached_command", name, atoll (fields[2]), st);
448       if (strcmp (name, "get") == 0)
449         gets = atof (fields[2]);
450     }
451
452     /*
453      * Increment/Decrement
454      */
455     else if (FIELD_IS("incr_misses"))
456     {
457       derive_t incr_count = atoll (fields[2]);
458       submit_derive ("memcached_ops", "incr_misses", incr_count, st);
459       incr += incr_count;
460     }
461     else if (FIELD_IS ("incr_hits"))
462     {
463       derive_t incr_count = atoll (fields[2]);
464       submit_derive ("memcached_ops", "incr_hits", incr_count, st);
465       incr_hits = atof (fields[2]);
466       incr += incr_count;
467     }
468     else if (FIELD_IS ("decr_misses"))
469     {
470       derive_t decr_count = atoll (fields[2]);
471       submit_derive ("memcached_ops", "decr_misses", decr_count, st);
472       decr += decr_count;
473     }
474     else if (FIELD_IS ("decr_hits"))
475     {
476       derive_t decr_count = atoll (fields[2]);
477       submit_derive ("memcached_ops", "decr_hits", decr_count, st);
478       decr_hits = atof (fields[2]);
479       decr += decr_count;
480     }
481
482     /*
483      * Operations on the cache, i. e. cache hits, cache misses and evictions of items
484      */
485     else if (FIELD_IS ("get_hits"))
486     {
487       submit_derive ("memcached_ops", "hits", atoll (fields[2]), st);
488       hits = atof (fields[2]);
489     }
490     else if (FIELD_IS ("get_misses"))
491     {
492       submit_derive ("memcached_ops", "misses", atoll (fields[2]), st);
493     }
494     else if (FIELD_IS ("evictions"))
495     {
496       submit_derive ("memcached_ops", "evictions", atoll (fields[2]), st);
497     }
498
499     /*
500      * Network traffic
501      */
502     else if (FIELD_IS ("bytes_read"))
503     {
504       octets_rx = atoll (fields[2]);
505     }
506     else if (FIELD_IS ("bytes_written"))
507     {
508       octets_tx = atoll (fields[2]);
509     }
510   } /* while ((line = strtok_r (ptr, "\n\r", &saveptr)) != NULL) */
511
512   if (!isnan (bytes_used) && !isnan (bytes_total) && (bytes_used <= bytes_total))
513     submit_gauge2 ("df", "cache", bytes_used, bytes_total - bytes_used, st);
514
515   if ((rusage_user != 0) || (rusage_syst != 0))
516     submit_derive2 ("ps_cputime", NULL, rusage_user, rusage_syst, st);
517
518   if ((octets_rx != 0) || (octets_tx != 0))
519     submit_derive2 ("memcached_octets", NULL, octets_rx, octets_tx, st);
520
521   if (!isnan (gets) && !isnan (hits))
522   {
523     gauge_t rate = NAN;
524
525     if (gets != 0.0)
526       rate = 100.0 * hits / gets;
527
528     submit_gauge ("percent", "hitratio", rate, st);
529   }
530
531   if (!isnan (incr_hits) && incr != 0)
532   {
533     gauge_t incr_rate = 100.0 * incr_hits / incr;
534     submit_gauge ("percent", "incr_hitratio", incr_rate, st);
535     submit_derive ("memcached_ops", "incr", incr, st);
536   }
537
538   if (!isnan (decr_hits) && decr != 0)
539   {
540     gauge_t decr_rate = 100.0 * decr_hits / decr;
541     submit_gauge ("percent", "decr_hitratio", decr_rate, st);
542     submit_derive ("memcached_ops", "decr", decr, st);
543   }
544
545   return 0;
546 } /* int memcached_read */
547
548 static int memcached_add_read_callback (memcached_t *st)
549 {
550   user_data_t ud;
551   char callback_name[3*DATA_MAX_NAME_LEN];
552   int status;
553
554   memset (&ud, 0, sizeof (ud));
555   ud.data = st;
556   ud.free_func = (void *) memcached_free;
557
558   assert (st->name != NULL);
559   ssnprintf (callback_name, sizeof (callback_name), "memcached/%s", st->name);
560
561   status = plugin_register_complex_read (/* group = */ "memcached",
562       /* name      = */ callback_name,
563       /* callback  = */ memcached_read,
564       /* interval  = */ NULL,
565       /* user_data = */ &ud);
566   return (status);
567 } /* int memcached_add_read_callback */
568
569 /* Configuration handling functiions
570  * <Plugin memcached>
571  *   <Instance "instance_name">
572  *     Host foo.zomg.com
573  *     Port "1234"
574  *   </Instance>
575  * </Plugin>
576  */
577 static int config_add_instance(oconfig_item_t *ci)
578 {
579   memcached_t *st;
580   int i;
581   int status = 0;
582
583   /* Disable automatic generation of default instance in the init callback. */
584   memcached_have_instances = 1;
585
586   st = malloc (sizeof (*st));
587   if (st == NULL)
588   {
589     ERROR ("memcached plugin: malloc failed.");
590     return (-1);
591   }
592
593   memset (st, 0, sizeof (*st));
594   st->name = NULL;
595   st->socket = NULL;
596   st->host = NULL;
597   st->port = NULL;
598
599   if (strcasecmp (ci->key, "Plugin") == 0) /* default instance */
600     st->name = sstrdup ("__legacy__");
601   else /* <Instance /> block */
602     status = cf_util_get_string (ci, &st->name);
603   if (status != 0)
604   {
605     sfree (st);
606     return (status);
607   }
608   assert (st->name != NULL);
609
610   for (i = 0; i < ci->children_num; i++)
611   {
612     oconfig_item_t *child = ci->children + i;
613
614     if (strcasecmp ("Socket", child->key) == 0)
615       status = cf_util_get_string (child, &st->socket);
616     else if (strcasecmp ("Host", child->key) == 0)
617       status = cf_util_get_string (child, &st->host);
618     else if (strcasecmp ("Port", child->key) == 0)
619       status = cf_util_get_service (child, &st->port);
620     else
621     {
622       WARNING ("memcached plugin: Option `%s' not allowed here.",
623           child->key);
624       status = -1;
625     }
626
627     if (status != 0)
628       break;
629   }
630
631   if (status == 0)
632     status = memcached_add_read_callback (st);
633
634   if (status != 0)
635   {
636     memcached_free(st);
637     return (-1);
638   }
639
640   return (0);
641 }
642
643 static int memcached_config (oconfig_item_t *ci)
644 {
645   int status = 0;
646   _Bool have_instance_block = 0;
647   int i;
648
649   for (i = 0; i < ci->children_num; i++)
650   {
651     oconfig_item_t *child = ci->children + i;
652
653     if (strcasecmp ("Instance", child->key) == 0)
654     {
655       config_add_instance (child);
656       have_instance_block = 1;
657     }
658     else if (!have_instance_block)
659     {
660       /* Non-instance option: Assume legacy configuration (without <Instance />
661        * blocks) and call config_add_instance() with the <Plugin /> block. */
662       return (config_add_instance (ci));
663     }
664     else
665       WARNING ("memcached plugin: The configuration option "
666           "\"%s\" is not allowed here. Did you "
667           "forget to add an <Instance /> block "
668           "around the configuration?",
669           child->key);
670   } /* for (ci->children) */
671
672   return (status);
673 }
674
675 static int memcached_init (void)
676 {
677   memcached_t *st;
678   int status;
679
680   if (memcached_have_instances)
681     return (0);
682
683   /* No instances were configured, lets start a default instance. */
684   st = malloc (sizeof (*st));
685   if (st == NULL)
686     return (ENOMEM);
687   memset (st, 0, sizeof (*st));
688   st->name = sstrdup ("__legacy__");
689   st->socket = NULL;
690   st->host = NULL;
691   st->port = NULL;
692
693   status = memcached_add_read_callback (st);
694   if (status == 0)
695     memcached_have_instances = 1;
696   else
697     memcached_free (st);
698
699   return (status);
700 } /* int memcached_init */
701
702 void module_register (void)
703 {
704   plugin_register_complex_config ("memcached", memcached_config);
705   plugin_register_init ("memcached", memcached_init);
706 }