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