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