- netapp plugin.
- python plugin.
+Thomas Meson <zllak at hycik.org>
+ - Graphite support for the AMQP plugin.
+
Tomasz Pala <gotar at pld-linux.org>
- conntrack plugin.
if test "x$with_perfstat" = "xyes"
then
plugin_cpu="yes"
+ plugin_contextswitch="yes"
plugin_disk="yes"
plugin_memory="yes"
plugin_swap="yes"
plugin_interface="yes"
plugin_load="yes"
+ plugin_uptime="yes"
fi
if test "x$with_procinfo" = "xyes"
-----------
Manifest file for the Solaris SMF system and detailed information on how to
register collectd as a service with this system.
+
+collectd.service
+----------------
+ Service file for systemd. Please ship this file as
+ /lib/systemd/system/collectd.service in any linux package of collectd.
--- /dev/null
+[Unit]
+Description=statistics collection daemon
+Documentation=man:collectd(1)
+After=local-fs.target network.target
+Requires=local-fs.target network.target
+
+[Service]
+ExecStart=/usr/sbin/collectd -C /etc/collectd/collectd.conf -f
+Restart=always
+RestartSec=10
+StandardOutput=syslog
+StandardError=syslog
+
+[Install]
+WantedBy=multi-user.target
pkglib_LTLIBRARIES += amqp.la
amqp_la_SOURCES = amqp.c \
utils_cmd_putval.c utils_cmd_putval.h \
+ utils_format_graphite.c utils_format_graphite.h \
utils_format_json.c utils_format_json.h
amqp_la_LDFLAGS = -module -avoid-version $(BUILD_WITH_LIBRABBITMQ_LDFLAGS)
amqp_la_CPPFLAGS = $(AM_CPPFLAGS) $(BUILD_WITH_LIBRABBITMQ_CPPFLAGS)
pkglib_LTLIBRARIES += contextswitch.la
contextswitch_la_SOURCES = contextswitch.c
contextswitch_la_LDFLAGS = -module -avoid-version
+contextswitch_la_LIBADD =
+if BUILD_WITH_PERFSTAT
+contextswitch_la_LIBADD += -lperfstat
+endif
collectd_LDADD += "-dlopen" contextswitch.la
collectd_DEPENDENCIES += contextswitch.la
endif
if BUILD_WITH_LIBKSTAT
uptime_la_LIBADD += -lkstat
endif
+if BUILD_WITH_PERFSTAT
+uptime_la_LIBADD += -lperfstat
+endif
collectd_LDADD += "-dlopen" uptime.la
collectd_DEPENDENCIES += uptime.la
endif
if BUILD_PLUGIN_WRITE_GRAPHITE
pkglib_LTLIBRARIES += write_graphite.la
write_graphite_la_SOURCES = write_graphite.c \
- utils_format_json.c utils_format_json.h
+ utils_format_graphite.c utils_format_graphite.h \
+ utils_format_json.c utils_format_json.h
write_graphite_la_LDFLAGS = -module -avoid-version
collectd_LDADD += "-dlopen" write_graphite.la
collectd_DEPENDENCIES += write_graphite.la
#include "plugin.h"
#include "utils_cmd_putval.h"
#include "utils_format_json.h"
+#include "utils_format_graphite.h"
#include <pthread.h>
#define CAMQP_DM_VOLATILE 1
#define CAMQP_DM_PERSISTENT 2
-#define CAMQP_FORMAT_COMMAND 1
-#define CAMQP_FORMAT_JSON 2
+#define CAMQP_FORMAT_COMMAND 1
+#define CAMQP_FORMAT_JSON 2
+#define CAMQP_FORMAT_GRAPHITE 3
#define CAMQP_CHANNEL 1
uint8_t delivery_mode;
_Bool store_rates;
int format;
+ /* publish & graphite format only */
+ char *prefix;
+ char *postfix;
+ char escape_char;
/* subscribe only */
char *exchange_type;
sfree (conf->exchange_type);
sfree (conf->queue);
sfree (conf->routing_key);
+ sfree (conf->prefix);
+ sfree (conf->postfix);
+
sfree (conf);
} /* }}} void camqp_config_free */
props.content_type = amqp_cstring_bytes("text/collectd");
else if (conf->format == CAMQP_FORMAT_JSON)
props.content_type = amqp_cstring_bytes("application/json");
+ else if (conf->format == CAMQP_FORMAT_GRAPHITE)
+ props.content_type = amqp_cstring_bytes("text/graphite");
else
assert (23 == 42);
props.delivery_mode = conf->delivery_mode;
format_json_value_list (buffer, &bfill, &bfree, ds, vl, conf->store_rates);
format_json_finalize (buffer, &bfill, &bfree);
}
+ else if (conf->format == CAMQP_FORMAT_GRAPHITE)
+ {
+ status = format_graphite (buffer, sizeof (buffer), ds, vl,
+ conf->prefix, conf->postfix, conf->escape_char);
+ if (status != 0)
+ {
+ ERROR ("amqp plugin: format_graphite failed with status %i.",
+ status);
+ return (status);
+ }
+ }
else
{
ERROR ("amqp plugin: Invalid format (%i).", conf->format);
conf->format = CAMQP_FORMAT_COMMAND;
else if (strcasecmp ("JSON", string) == 0)
conf->format = CAMQP_FORMAT_JSON;
+ else if (strcasecmp ("Graphite", string) == 0)
+ conf->format = CAMQP_FORMAT_GRAPHITE;
else
{
WARNING ("amqp plugin: Invalid format string: %s",
/* publish only */
conf->delivery_mode = CAMQP_DM_VOLATILE;
conf->store_rates = 0;
+ /* publish & graphite only */
+ conf->prefix = NULL;
+ conf->postfix = NULL;
+ conf->escape_char = '_';
/* subscribe only */
conf->exchange_type = NULL;
conf->queue = NULL;
status = cf_util_get_boolean (child, &conf->store_rates);
else if ((strcasecmp ("Format", child->key) == 0) && publish)
status = camqp_config_set_format (child, conf);
+ else if ((strcasecmp ("GraphitePrefix", child->key) == 0) && publish)
+ status = cf_util_get_string (child, &conf->prefix);
+ else if ((strcasecmp ("GraphitePostfix", child->key) == 0) && publish)
+ status = cf_util_get_string (child, &conf->postfix);
+ else if ((strcasecmp ("GraphiteEscapeChar", child->key) == 0) && publish)
+ {
+ char *tmp_buff = NULL;
+ status = cf_util_get_string (child, &tmp_buff);
+ if (strlen (tmp_buff) > 1)
+ WARNING ("amqp plugin: The option \"GraphiteEscapeChar\" handles "
+ "only one character. Others will be ignored.");
+ conf->escape_char = tmp_buff[0];
+ sfree (tmp_buff);
+ }
else
WARNING ("amqp plugin: Ignoring unknown "
"configuration option \"%s\".", child->key);
#</Plugin>
#<Plugin memcached>
-# Host "127.0.0.1"
-# Port "11211"
+# <Instance "local">
+# Host "127.0.0.1"
+# Port "11211"
+# </Instance>
#</Plugin>
#<Plugin modbus>
# Host "localhost"
# Port 123
# ReverseLookups false
+# IncludeUnitID true
#</Plugin>
#<Plugin nut>
#<Plugin "swap">
# ReportByDevice false
+# ReportBytes true
#</Plugin>
#<Plugin "table">
# Persistent false
# Format "command"
# StoreRates false
+ # GraphitePrefix "collectd."
+ # GraphiteEscapeChar "_"
</Publish>
# Receive values from an AMQP broker
an easy and straight forward exchange format. The C<Content-Type> header field
will be set to C<application/json>.
+If set to B<Graphite>, values are encoded in the I<Graphite> format, which is
+"<metric> <value> <timestamp>\n". The C<Content-Type> header field will be set to
+C<text/graphite>.
+
A subscribing client I<should> use the C<Content-Type> header field to
determine how to decode the values. Currently, the I<AMQP plugin> itself can
only decode the B<Command> format.
Please note that currently this option is only used if the B<Format> option has
been set to B<JSON>.
+=item B<GraphitePrefix> (Publish and B<Format>=I<Graphite> only)
+
+A prefix can be added in the metric name when outputting in the I<Graphite> format.
+It's added before the I<Host> name.
+Metric name will be "<prefix><host><postfix><plugin><type><name>"
+
+=item B<GraphitePostfix> (Publish and B<Format>=I<Graphite> only)
+
+A postfix can be added in the metric name when outputting in the I<Graphite> format.
+It's added after the I<Host> name.
+Metric name will be "<prefix><host><postfix><plugin><type><name>"
+
+=item B<GraphiteEscapeChar> (Publish and B<Format>=I<Graphite> only)
+
+Specify a character to replace dots (.) in the host part of the metric name.
+In I<Graphite> metric name, dots are used as separators between different
+metric parts (host, plugin, type).
+Default is "_" (I<Underscore>).
+
=back
=head2 Plugin C<apache>
about cache utilization, memory and bandwidth used.
L<http://www.danga.com/memcached/>
+ <Plugin "memcached">
+ <Instance "name">
+ Host "memcache.example.com"
+ Port 11211
+ </Instance>
+ </Plugin>
+
+The plugin configuration consists of one or more B<Instance> blocks which
+specify one I<memcached> connection each. Within the B<Instance> blocks, the
+following options are allowed:
+
=over 4
=item B<Host> I<Hostname>
TCP-Port to connect to. Defaults to B<11211>.
+=item B<Socket> I<Path>
+
+Connect to I<memcached> using the UNIX domain socket at I<Path>. If this
+setting is given, the B<Host> and B<Port> settings are ignored.
+
=back
=head2 Plugin C<modbus>
lookups. The default is to do reverse lookups to preserve backwards
compatibility, though.
+=item B<IncludeUnitID> B<true>|B<false>
+
+When a peer is a refclock, include the unit ID in the I<type instance>.
+Defaults to B<false> for backward compatibility.
+
+If two refclock peers use the same driver and this is B<false>, the plugin will
+try to write simultaneous measurements from both to the same type instance.
+This will result in error messages in the log and only one set of measurements
+making it through.
+
=back
=head2 Plugin C<nut>
Defines the "database alias" or "service name" to connect to. Usually, these
names are defined in the file named C<$ORACLE_HOME/network/admin/tnsnames.ora>.
+=item B<Host> I<Host>
+
+Hostname to use when dispatching values for this database. Defaults to using
+the global hostname of the I<collectd> instance.
+
=item B<Username> I<Username>
Username used for authentication.
connections. Either a service name of a port number may be given. Please note
that numerical port numbers must be given as a string, too.
+=item B<Password> I<Password>
+
+Use I<Password> to authenticate when connecting to I<Redis>.
+
=item B<Timeout> I<Timeout in miliseconds>
The B<Timeout> option set the socket timeout for node response. Since the Redis
This option is only available if the I<Swap plugin> can read C</proc/swaps>
(under Linux) or use the L<swapctl(2)> mechanism (under I<Solaris>).
+=item B<ReportBytes> B<false>|B<true>
+
+When enabled, the I<swap I/O> is reported in bytes. When disabled, the default,
+I<swap I/O> is reported in pages. This option is available under Linux only.
+
=back
=head2 Plugin C<syslog>
/* no global variables */
/* #endif KERNEL_LINUX */
+#elif HAVE_PERFSTAT
+# include <sys/protosw.h>
+# include <libperfstat.h>
+/* #endif HAVE_PERFSTAT */
+
#else
# error "No applicable input method."
#endif
if (status == -2)
ERROR ("contextswitch plugin: Unable to find context switch value.");
-#endif /* KERNEL_LINUX */
+/* #endif KERNEL_LINUX */
+
+#elif HAVE_PERFSTAT
+ int status = 0;
+ perfstat_cpu_total_t perfcputotal;
+
+ status = perfstat_cpu_total(NULL, &perfcputotal, sizeof(perfstat_cpu_total_t), 1);
+ if (status < 0)
+ {
+ char errbuf[1024];
+ ERROR ("contextswitch plugin: perfstat_cpu_total: %s",
+ sstrerror (errno, errbuf, sizeof (errbuf)));
+ return (-1);
+ }
+
+ cs_submit(perfcputotal.pswitch);
+ status = 0;
+#endif /* defined(HAVE_PERFSTAT) */
return status;
}
/**
* collectd - src/memcached.c, based on src/hddtemp.c
* Copyright (C) 2007 Antony Dovgal
- * Copyright (C) 2007-2010 Florian Forster
+ * Copyright (C) 2007-2012 Florian Forster
* Copyright (C) 2009 Doug MacEachern
* Copyright (C) 2009 Franck Lombardi
+ * Copyright (C) 2012 Nicolas Szalay
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License as published by the
* Florian octo Forster <octo at collectd.org>
* Doug MacEachern <dougm at hyperic.com>
* Franck Lombardi
+ * Nicolas Szalay
**/
#include "collectd.h"
#include "plugin.h"
#include "configfile.h"
-# include <poll.h>
-# include <netdb.h>
-# include <sys/socket.h>
-# include <sys/un.h>
-# include <netinet/in.h>
-# include <netinet/tcp.h>
-
-/* Hack to work around the missing define in AIX */
-#ifndef MSG_DONTWAIT
-# define MSG_DONTWAIT MSG_NONBLOCK
-#endif
+#include <netdb.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
#define MEMCACHED_DEF_HOST "127.0.0.1"
#define MEMCACHED_DEF_PORT "11211"
-#define MEMCACHED_RETRY_COUNT 100
-
-static const char *config_keys[] =
+struct memcached_s
{
- "Socket",
- "Host",
- "Port"
+ char *name;
+ char *socket;
+ char *host;
+ char *port;
};
-static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
+typedef struct memcached_s memcached_t;
-static char *memcached_socket = NULL;
-static char *memcached_host = NULL;
-static char memcached_port[16];
+static _Bool memcached_have_instances = 0;
-static int memcached_query_daemon (char *buffer, int buffer_size) /* {{{ */
+static void memcached_free (memcached_t *st)
{
- int fd;
- ssize_t status;
- int buffer_fill;
- int i = 0;
-
- if (memcached_socket != NULL) {
- struct sockaddr_un serv_addr;
-
- memset (&serv_addr, 0, sizeof (serv_addr));
- serv_addr.sun_family = AF_UNIX;
- sstrncpy (serv_addr.sun_path, memcached_socket,
- sizeof (serv_addr.sun_path));
-
- /* create our socket descriptor */
- fd = socket (AF_UNIX, SOCK_STREAM, 0);
- if (fd < 0) {
- char errbuf[1024];
- ERROR ("memcached: unix socket: %s", sstrerror (errno, errbuf,
- sizeof (errbuf)));
- return -1;
- }
-
- /* connect to the memcached daemon */
- status = (ssize_t) connect (fd, (struct sockaddr *) &serv_addr,
- sizeof (serv_addr));
- if (status != 0) {
- shutdown (fd, SHUT_RDWR);
- close (fd);
- fd = -1;
- }
- }
- else { /* if (memcached_socket == NULL) */
- const char *host;
- const char *port;
-
- struct addrinfo ai_hints;
- struct addrinfo *ai_list, *ai_ptr;
- int ai_return = 0;
-
- memset (&ai_hints, '\0', sizeof (ai_hints));
- ai_hints.ai_flags = 0;
+ if (st == NULL)
+ return;
+
+ sfree (st->name);
+ sfree (st->socket);
+ sfree (st->host);
+ sfree (st->port);
+}
+
+static int memcached_connect_unix (memcached_t *st)
+{
+ struct sockaddr_un serv_addr;
+ int fd;
+
+ memset (&serv_addr, 0, sizeof (serv_addr));
+ serv_addr.sun_family = AF_UNIX;
+ sstrncpy (serv_addr.sun_path, st->socket,
+ sizeof (serv_addr.sun_path));
+
+ /* create our socket descriptor */
+ fd = socket (AF_UNIX, SOCK_STREAM, 0);
+ if (fd < 0)
+ {
+ char errbuf[1024];
+ ERROR ("memcached plugin: memcached_connect_unix: socket(2) failed: %s",
+ sstrerror (errno, errbuf, sizeof (errbuf)));
+ return (-1);
+ }
+
+ return (fd);
+} /* int memcached_connect_unix */
+
+static int memcached_connect_inet (memcached_t *st)
+{
+ char *host;
+ char *port;
+
+ struct addrinfo ai_hints;
+ struct addrinfo *ai_list, *ai_ptr;
+ int status;
+ int fd = -1;
+
+ memset (&ai_hints, 0, sizeof (ai_hints));
+ ai_hints.ai_flags = 0;
#ifdef AI_ADDRCONFIG
- /* ai_hints.ai_flags |= AI_ADDRCONFIG; */
+ ai_hints.ai_flags |= AI_ADDRCONFIG;
#endif
- ai_hints.ai_family = AF_INET;
- ai_hints.ai_socktype = SOCK_STREAM;
- ai_hints.ai_protocol = 0;
-
- host = memcached_host;
- if (host == NULL) {
- host = MEMCACHED_DEF_HOST;
- }
-
- port = memcached_port;
- if (strlen (port) == 0) {
- port = MEMCACHED_DEF_PORT;
- }
-
- if ((ai_return = getaddrinfo (host, port, &ai_hints, &ai_list)) != 0) {
- char errbuf[1024];
- ERROR ("memcached: getaddrinfo (%s, %s): %s",
- host, port,
- (ai_return == EAI_SYSTEM)
- ? sstrerror (errno, errbuf, sizeof (errbuf))
- : gai_strerror (ai_return));
- return -1;
- }
-
- fd = -1;
- for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) {
- /* create our socket descriptor */
- fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
- if (fd < 0) {
- char errbuf[1024];
- ERROR ("memcached: socket: %s", sstrerror (errno, errbuf, sizeof (errbuf)));
- continue;
- }
-
- /* connect to the memcached daemon */
- status = (ssize_t) connect (fd, (struct sockaddr *) ai_ptr->ai_addr, ai_ptr->ai_addrlen);
- if (status != 0) {
- shutdown (fd, SHUT_RDWR);
- close (fd);
- fd = -1;
- continue;
- }
-
- /* A socket could be opened and connecting succeeded. We're
- * done. */
- break;
- }
-
- freeaddrinfo (ai_list);
- }
-
- if (fd < 0) {
- ERROR ("memcached: Could not connect to daemon.");
- return -1;
- }
-
- if (send(fd, "stats\r\n", sizeof("stats\r\n") - 1, MSG_DONTWAIT) != (sizeof("stats\r\n") - 1)) {
- ERROR ("memcached: Could not send command to the memcached daemon.");
- return -1;
- }
-
- {
- struct pollfd p;
- int status;
-
- memset (&p, 0, sizeof (p));
- p.fd = fd;
- p.events = POLLIN | POLLERR | POLLHUP;
- p.revents = 0;
-
- status = poll (&p, /* nfds = */ 1,
- /* timeout = */ CDTIME_T_TO_MS (interval_g));
- if (status <= 0)
- {
- if (status == 0)
- {
- ERROR ("memcached: poll(2) timed out after %.3f seconds.",
- CDTIME_T_TO_DOUBLE (interval_g));
- }
- else
- {
- char errbuf[1024];
- ERROR ("memcached: poll(2) failed: %s",
- sstrerror (errno, errbuf, sizeof (errbuf)));
- }
- shutdown (fd, SHUT_RDWR);
- close (fd);
- return (-1);
- }
- }
-
- /* receive data from the memcached daemon */
- memset (buffer, '\0', buffer_size);
-
- buffer_fill = 0;
- while ((status = recv (fd, buffer + buffer_fill, buffer_size - buffer_fill, MSG_DONTWAIT)) != 0) {
- if (i > MEMCACHED_RETRY_COUNT) {
- ERROR("recv() timed out");
- break;
- }
- i++;
-
- if (status == -1) {
- char errbuf[1024];
-
- if (errno == EAGAIN) {
- continue;
- }
-
- ERROR ("memcached: Error reading from socket: %s",
- sstrerror (errno, errbuf, sizeof (errbuf)));
- shutdown(fd, SHUT_RDWR);
- close (fd);
- return -1;
- }
- buffer_fill += status;
-
- if (buffer_fill > 3 && buffer[buffer_fill-5] == 'E' && buffer[buffer_fill-4] == 'N' && buffer[buffer_fill-3] == 'D') {
- /* we got all the data */
- break;
- }
- }
-
- if (buffer_fill >= buffer_size) {
- buffer[buffer_size - 1] = '\0';
- WARNING ("memcached: Message from memcached has been truncated.");
- } else if (buffer_fill == 0) {
- WARNING ("memcached: Peer has unexpectedly shut down the socket. "
- "Buffer: `%s'", buffer);
- shutdown(fd, SHUT_RDWR);
- close(fd);
- return -1;
- }
-
- shutdown(fd, SHUT_RDWR);
- close(fd);
- return 0;
+ ai_hints.ai_family = AF_UNSPEC;
+ ai_hints.ai_socktype = SOCK_STREAM;
+ ai_hints.ai_protocol = 0;
+
+ host = (st->host != NULL) ? st->host : MEMCACHED_DEF_HOST;
+ port = (st->port != NULL) ? st->port : MEMCACHED_DEF_PORT;
+
+ ai_list = NULL;
+ status = getaddrinfo (host, port, &ai_hints, &ai_list);
+ if (status != 0)
+ {
+ char errbuf[1024];
+ ERROR ("memcached plugin: memcached_connect_inet: "
+ "getaddrinfo(%s,%s) failed: %s",
+ host, port,
+ (status == EAI_SYSTEM)
+ ? sstrerror (errno, errbuf, sizeof (errbuf))
+ : gai_strerror (status));
+ return (-1);
+ }
+
+ for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
+ {
+ /* create our socket descriptor */
+ fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
+ if (fd < 0)
+ {
+ char errbuf[1024];
+ WARNING ("memcached plugin: memcached_connect_inet: "
+ "socket(2) failed: %s",
+ sstrerror (errno, errbuf, sizeof (errbuf)));
+ continue;
+ }
+
+ /* connect to the memcached daemon */
+ status = (int) connect (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
+ if (status != 0)
+ {
+ shutdown (fd, SHUT_RDWR);
+ close (fd);
+ fd = -1;
+ continue;
+ }
+
+ /* A socket could be opened and connecting succeeded. We're done. */
+ break;
+ }
+
+ freeaddrinfo (ai_list);
+ return (fd);
+} /* int memcached_connect_inet */
+
+static int memcached_connect (memcached_t *st)
+{
+ if (st->socket != NULL)
+ return (memcached_connect_unix (st));
+ else
+ return (memcached_connect_inet (st));
}
-/* }}} */
-static int memcached_config (const char *key, const char *value) /* {{{ */
+static int memcached_query_daemon (char *buffer, size_t buffer_size, memcached_t *st)
{
- if (strcasecmp (key, "Socket") == 0) {
- if (memcached_socket != NULL) {
- free (memcached_socket);
- }
- memcached_socket = strdup (value);
- } else if (strcasecmp (key, "Host") == 0) {
- if (memcached_host != NULL) {
- free (memcached_host);
- }
- memcached_host = strdup (value);
- } else if (strcasecmp (key, "Port") == 0) {
- int port = (int) (atof (value));
- if ((port > 0) && (port <= 65535)) {
- ssnprintf (memcached_port, sizeof (memcached_port), "%i", port);
- } else {
- sstrncpy (memcached_port, value, sizeof (memcached_port));
- }
- } else {
- return -1;
- }
-
- return 0;
+ int fd = -1;
+ int status;
+ size_t buffer_fill;
+
+ fd = memcached_connect (st);
+ if (fd < 0) {
+ ERROR ("memcached plugin: Instance \"%s\" could not connect to daemon.",
+ st->name);
+ return -1;
+ }
+
+ status = (int) swrite (fd, "stats\r\n", strlen ("stats\r\n"));
+ if (status != 0)
+ {
+ char errbuf[1024];
+ ERROR ("memcached plugin: write(2) failed: %s",
+ sstrerror (errno, errbuf, sizeof (errbuf)));
+ shutdown(fd, SHUT_RDWR);
+ close (fd);
+ return (-1);
+ }
+
+ /* receive data from the memcached daemon */
+ memset (buffer, 0, buffer_size);
+
+ buffer_fill = 0;
+ while ((status = (int) recv (fd, buffer + buffer_fill,
+ buffer_size - buffer_fill, /* flags = */ 0)) != 0)
+ {
+ char const end_token[5] = {'E', 'N', 'D', '\r', '\n'};
+ if (status < 0)
+ {
+ char errbuf[1024];
+
+ if ((errno == EAGAIN) || (errno == EINTR))
+ continue;
+
+ ERROR ("memcached: Error reading from socket: %s",
+ sstrerror (errno, errbuf, sizeof (errbuf)));
+ shutdown(fd, SHUT_RDWR);
+ close (fd);
+ return (-1);
+ }
+
+ buffer_fill += (size_t) status;
+ if (buffer_fill > buffer_size)
+ {
+ buffer_fill = buffer_size;
+ WARNING ("memcached plugin: Message was truncated.");
+ break;
+ }
+
+ /* If buffer ends in end_token, we have all the data. */
+ if (memcmp (buffer + buffer_fill - sizeof (end_token),
+ end_token, sizeof (end_token)) == 0)
+ break;
+ } /* while (recv) */
+
+ status = 0;
+ if (buffer_fill == 0)
+ {
+ WARNING ("memcached plugin: No data returned by memcached.");
+ status = -1;
+ }
+
+ shutdown(fd, SHUT_RDWR);
+ close(fd);
+ return (status);
+} /* int memcached_query_daemon */
+
+static void memcached_init_vl (value_list_t *vl, memcached_t const *st)
+{
+ sstrncpy (vl->plugin, "memcached", sizeof (vl->plugin));
+ if (strcmp (st->name, "__legacy__") == 0) /* legacy mode */
+ {
+ sstrncpy (vl->host, hostname_g, sizeof (vl->host));
+ }
+ else
+ {
+ if (st->socket != NULL)
+ sstrncpy (vl->host, hostname_g, sizeof (vl->host));
+ else
+ sstrncpy (vl->host,
+ (st->host != NULL) ? st->host : MEMCACHED_DEF_HOST,
+ sizeof (vl->host));
+ sstrncpy (vl->plugin_instance, st->name, sizeof (vl->plugin_instance));
+ }
}
-/* }}} */
static void submit_derive (const char *type, const char *type_inst,
- derive_t value) /* {{{ */
+ derive_t value, memcached_t *st)
{
- value_t values[1];
- value_list_t vl = VALUE_LIST_INIT;
+ value_t values[1];
+ value_list_t vl = VALUE_LIST_INIT;
+ memcached_init_vl (&vl, st);
- values[0].derive = value;
+ values[0].derive = value;
- vl.values = values;
- vl.values_len = 1;
- sstrncpy (vl.host, hostname_g, sizeof (vl.host));
- sstrncpy (vl.plugin, "memcached", sizeof (vl.plugin));
- sstrncpy (vl.type, type, sizeof (vl.type));
- if (type_inst != NULL)
- sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
+ vl.values = values;
+ vl.values_len = 1;
+ sstrncpy (vl.type, type, sizeof (vl.type));
+ if (type_inst != NULL)
+ sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
- plugin_dispatch_values (&vl);
-} /* void memcached_submit_cmd */
-/* }}} */
+ plugin_dispatch_values (&vl);
+}
static void submit_derive2 (const char *type, const char *type_inst,
- derive_t value0, derive_t value1) /* {{{ */
+ derive_t value0, derive_t value1, memcached_t *st)
{
- value_t values[2];
- value_list_t vl = VALUE_LIST_INIT;
+ value_t values[2];
+ value_list_t vl = VALUE_LIST_INIT;
+ memcached_init_vl (&vl, st);
- values[0].derive = value0;
- values[1].derive = value1;
+ values[0].derive = value0;
+ values[1].derive = value1;
- vl.values = values;
- vl.values_len = 2;
- sstrncpy (vl.host, hostname_g, sizeof (vl.host));
- sstrncpy (vl.plugin, "memcached", sizeof (vl.plugin));
- sstrncpy (vl.type, type, sizeof (vl.type));
- if (type_inst != NULL)
- sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
+ vl.values = values;
+ vl.values_len = 2;
+ sstrncpy (vl.type, type, sizeof (vl.type));
+ if (type_inst != NULL)
+ sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
- plugin_dispatch_values (&vl);
-} /* void memcached_submit_cmd */
-/* }}} */
+ plugin_dispatch_values (&vl);
+}
static void submit_gauge (const char *type, const char *type_inst,
- gauge_t value) /* {{{ */
+ gauge_t value, memcached_t *st)
{
- value_t values[1];
- value_list_t vl = VALUE_LIST_INIT;
+ value_t values[1];
+ value_list_t vl = VALUE_LIST_INIT;
+ memcached_init_vl (&vl, st);
- values[0].gauge = value;
+ values[0].gauge = value;
- vl.values = values;
- vl.values_len = 1;
- sstrncpy (vl.host, hostname_g, sizeof (vl.host));
- sstrncpy (vl.plugin, "memcached", sizeof (vl.plugin));
- sstrncpy (vl.type, type, sizeof (vl.type));
- if (type_inst != NULL)
- sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
+ vl.values = values;
+ vl.values_len = 1;
+ sstrncpy (vl.type, type, sizeof (vl.type));
+ if (type_inst != NULL)
+ sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
- plugin_dispatch_values (&vl);
+ plugin_dispatch_values (&vl);
}
-/* }}} */
static void submit_gauge2 (const char *type, const char *type_inst,
- gauge_t value0, gauge_t value1) /* {{{ */
+ gauge_t value0, gauge_t value1, memcached_t *st)
{
- value_t values[2];
- value_list_t vl = VALUE_LIST_INIT;
+ value_t values[2];
+ value_list_t vl = VALUE_LIST_INIT;
+ memcached_init_vl (&vl, st);
- values[0].gauge = value0;
- values[1].gauge = value1;
+ values[0].gauge = value0;
+ values[1].gauge = value1;
- vl.values = values;
- vl.values_len = 2;
- sstrncpy (vl.host, hostname_g, sizeof (vl.host));
- sstrncpy (vl.plugin, "memcached", sizeof (vl.plugin));
- sstrncpy (vl.type, type, sizeof (vl.type));
- if (type_inst != NULL)
- sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
+ vl.values = values;
+ vl.values_len = 2;
+ sstrncpy (vl.type, type, sizeof (vl.type));
+ if (type_inst != NULL)
+ sstrncpy (vl.type_instance, type_inst, sizeof (vl.type_instance));
- plugin_dispatch_values (&vl);
+ plugin_dispatch_values (&vl);
}
-/* }}} */
-static int memcached_read (void) /* {{{ */
+static int memcached_read (user_data_t *user_data)
{
- char buf[4096];
- char *fields[3];
- char *ptr;
- char *line;
- char *saveptr;
- int fields_num;
-
- gauge_t bytes_used = NAN;
- gauge_t bytes_total = NAN;
- gauge_t hits = NAN;
- gauge_t gets = NAN;
- derive_t rusage_user = 0;
- derive_t rusage_syst = 0;
- derive_t octets_rx = 0;
- derive_t octets_tx = 0;
-
- /* get data from daemon */
- if (memcached_query_daemon (buf, sizeof (buf)) < 0) {
- return -1;
- }
+ char buf[4096];
+ char *fields[3];
+ char *ptr;
+ char *line;
+ char *saveptr;
+ int fields_num;
+
+ gauge_t bytes_used = NAN;
+ gauge_t bytes_total = NAN;
+ gauge_t hits = NAN;
+ gauge_t gets = NAN;
+ derive_t rusage_user = 0;
+ derive_t rusage_syst = 0;
+ derive_t octets_rx = 0;
+ derive_t octets_tx = 0;
+
+ memcached_t *st;
+ st = user_data->data;
+
+ /* get data from daemon */
+ if (memcached_query_daemon (buf, sizeof (buf), st) < 0) {
+ return -1;
+ }
#define FIELD_IS(cnst) \
- (((sizeof(cnst) - 1) == name_len) && (strcmp (cnst, fields[1]) == 0))
-
- ptr = buf;
- saveptr = NULL;
- while ((line = strtok_r (ptr, "\n\r", &saveptr)) != NULL)
- {
- int name_len;
-
- ptr = NULL;
-
- fields_num = strsplit(line, fields, 3);
- if (fields_num != 3)
- continue;
-
- name_len = strlen(fields[1]);
- if (name_len == 0)
- continue;
-
- /*
- * For an explanation on these fields please refer to
- * <http://code.sixapart.com/svn/memcached/trunk/server/doc/protocol.txt>
- */
-
- /*
- * CPU time consumed by the memcached process
- */
- if (FIELD_IS ("rusage_user"))
- {
- rusage_user = atoll (fields[2]);
- }
- else if (FIELD_IS ("rusage_system"))
- {
- rusage_syst = atoll(fields[2]);
- }
-
- /*
- * Number of threads of this instance
- */
- else if (FIELD_IS ("threads"))
- {
- submit_gauge2 ("ps_count", NULL, NAN, atof (fields[2]));
- }
-
- /*
- * Number of items stored
- */
- else if (FIELD_IS ("curr_items"))
- {
- submit_gauge ("memcached_items", "current", atof (fields[2]));
- }
-
- /*
- * Number of bytes used and available (total - used)
- */
- else if (FIELD_IS ("bytes"))
- {
- bytes_used = atof (fields[2]);
- }
- else if (FIELD_IS ("limit_maxbytes"))
- {
- bytes_total = atof(fields[2]);
- }
-
- /*
- * Connections
- */
- else if (FIELD_IS ("curr_connections"))
- {
- submit_gauge ("memcached_connections", "current", atof (fields[2]));
- }
-
- /*
- * Commands
- */
- else if ((name_len > 4) && (strncmp (fields[1], "cmd_", 4) == 0))
- {
- const char *name = fields[1] + 4;
- submit_derive ("memcached_command", name, atoll (fields[2]));
- if (strcmp (name, "get") == 0)
- gets = atof (fields[2]);
- }
-
- /*
- * Operations on the cache, i. e. cache hits, cache misses and evictions of items
- */
- else if (FIELD_IS ("get_hits"))
- {
- submit_derive ("memcached_ops", "hits", atoll (fields[2]));
- hits = atof (fields[2]);
- }
- else if (FIELD_IS ("get_misses"))
- {
- submit_derive ("memcached_ops", "misses", atoll (fields[2]));
- }
- else if (FIELD_IS ("evictions"))
- {
- submit_derive ("memcached_ops", "evictions", atoll (fields[2]));
- }
-
- /*
- * Network traffic
- */
- else if (FIELD_IS ("bytes_read"))
- {
- octets_rx = atoll (fields[2]);
- }
- else if (FIELD_IS ("bytes_written"))
- {
- octets_tx = atoll (fields[2]);
- }
- } /* while ((line = strtok_r (ptr, "\n\r", &saveptr)) != NULL) */
-
- if (!isnan (bytes_used) && !isnan (bytes_total) && (bytes_used <= bytes_total))
- submit_gauge2 ("df", "cache", bytes_used, bytes_total - bytes_used);
-
- if ((rusage_user != 0) || (rusage_syst != 0))
- submit_derive2 ("ps_cputime", NULL, rusage_user, rusage_syst);
-
- if ((octets_rx != 0) || (octets_tx != 0))
- submit_derive2 ("memcached_octets", NULL, octets_rx, octets_tx);
-
- if (!isnan (gets) && !isnan (hits))
- {
- gauge_t rate = NAN;
-
- if (gets != 0.0)
- rate = 100.0 * hits / gets;
-
- submit_gauge ("percent", "hitratio", rate);
- }
-
- return 0;
+ (((sizeof(cnst) - 1) == name_len) && (strcmp (cnst, fields[1]) == 0))
+
+ ptr = buf;
+ saveptr = NULL;
+ while ((line = strtok_r (ptr, "\n\r", &saveptr)) != NULL)
+ {
+ int name_len;
+
+ ptr = NULL;
+
+ fields_num = strsplit(line, fields, 3);
+ if (fields_num != 3)
+ continue;
+
+ name_len = strlen(fields[1]);
+ if (name_len == 0)
+ continue;
+
+ /*
+ * For an explanation on these fields please refer to
+ * <http://code.sixapart.com/svn/memcached/trunk/server/doc/protocol.txt>
+ */
+
+ /*
+ * CPU time consumed by the memcached process
+ */
+ if (FIELD_IS ("rusage_user"))
+ {
+ rusage_user = atoll (fields[2]);
+ }
+ else if (FIELD_IS ("rusage_system"))
+ {
+ rusage_syst = atoll(fields[2]);
+ }
+
+ /*
+ * Number of threads of this instance
+ */
+ else if (FIELD_IS ("threads"))
+ {
+ submit_gauge2 ("ps_count", NULL, NAN, atof (fields[2]), st);
+ }
+
+ /*
+ * Number of items stored
+ */
+ else if (FIELD_IS ("curr_items"))
+ {
+ submit_gauge ("memcached_items", "current", atof (fields[2]), st);
+ }
+
+ /*
+ * Number of bytes used and available (total - used)
+ */
+ else if (FIELD_IS ("bytes"))
+ {
+ bytes_used = atof (fields[2]);
+ }
+ else if (FIELD_IS ("limit_maxbytes"))
+ {
+ bytes_total = atof(fields[2]);
+ }
+
+ /*
+ * Connections
+ */
+ else if (FIELD_IS ("curr_connections"))
+ {
+ submit_gauge ("memcached_connections", "current", atof (fields[2]), st);
+ }
+
+ /*
+ * Commands
+ */
+ else if ((name_len > 4) && (strncmp (fields[1], "cmd_", 4) == 0))
+ {
+ const char *name = fields[1] + 4;
+ submit_derive ("memcached_command", name, atoll (fields[2]), st);
+ if (strcmp (name, "get") == 0)
+ gets = atof (fields[2]);
+ }
+
+ /*
+ * Operations on the cache, i. e. cache hits, cache misses and evictions of items
+ */
+ else if (FIELD_IS ("get_hits"))
+ {
+ submit_derive ("memcached_ops", "hits", atoll (fields[2]), st);
+ hits = atof (fields[2]);
+ }
+ else if (FIELD_IS ("get_misses"))
+ {
+ submit_derive ("memcached_ops", "misses", atoll (fields[2]), st);
+ }
+ else if (FIELD_IS ("evictions"))
+ {
+ submit_derive ("memcached_ops", "evictions", atoll (fields[2]), st);
+ }
+
+ /*
+ * Network traffic
+ */
+ else if (FIELD_IS ("bytes_read"))
+ {
+ octets_rx = atoll (fields[2]);
+ }
+ else if (FIELD_IS ("bytes_written"))
+ {
+ octets_tx = atoll (fields[2]);
+ }
+ } /* while ((line = strtok_r (ptr, "\n\r", &saveptr)) != NULL) */
+
+ if (!isnan (bytes_used) && !isnan (bytes_total) && (bytes_used <= bytes_total))
+ submit_gauge2 ("df", "cache", bytes_used, bytes_total - bytes_used, st);
+
+ if ((rusage_user != 0) || (rusage_syst != 0))
+ submit_derive2 ("ps_cputime", NULL, rusage_user, rusage_syst, st);
+
+ if ((octets_rx != 0) || (octets_tx != 0))
+ submit_derive2 ("memcached_octets", NULL, octets_rx, octets_tx, st);
+
+ if (!isnan (gets) && !isnan (hits))
+ {
+ gauge_t rate = NAN;
+
+ if (gets != 0.0)
+ rate = 100.0 * hits / gets;
+
+ submit_gauge ("percent", "hitratio", rate, st);
+ }
+
+ return 0;
+} /* int memcached_read */
+
+static int memcached_add_read_callback (memcached_t *st)
+{
+ user_data_t ud;
+ char callback_name[3*DATA_MAX_NAME_LEN];
+ int status;
+
+ memset (&ud, 0, sizeof (ud));
+ ud.data = st;
+ ud.free_func = (void *) memcached_free;
+
+ assert (st->name != NULL);
+ ssnprintf (callback_name, sizeof (callback_name), "memcached/%s", st->name);
+
+ status = plugin_register_complex_read (/* group = */ "memcached",
+ /* name = */ callback_name,
+ /* callback = */ memcached_read,
+ /* interval = */ NULL,
+ /* user_data = */ &ud);
+ return (status);
+} /* int memcached_add_read_callback */
+
+/* Configuration handling functiions
+ * <Plugin memcached>
+ * <Instance "instance_name">
+ * Host foo.zomg.com
+ * Port "1234"
+ * </Instance>
+ * </Plugin>
+ */
+static int config_add_instance(oconfig_item_t *ci)
+{
+ memcached_t *st;
+ int i;
+ int status = 0;
+
+ /* Disable automatic generation of default instance in the init callback. */
+ memcached_have_instances = 1;
+
+ st = malloc (sizeof (*st));
+ if (st == NULL)
+ {
+ ERROR ("memcached plugin: malloc failed.");
+ return (-1);
+ }
+
+ memset (st, 0, sizeof (*st));
+ st->name = NULL;
+ st->socket = NULL;
+ st->host = NULL;
+ st->port = NULL;
+
+ if (strcasecmp (ci->key, "Plugin") == 0) /* default instance */
+ st->name = sstrdup ("__legacy__");
+ else /* <Instance /> block */
+ status = cf_util_get_string (ci, &st->name);
+ if (status != 0)
+ {
+ sfree (st);
+ return (status);
+ }
+ assert (st->name != NULL);
+
+ for (i = 0; i < ci->children_num; i++)
+ {
+ oconfig_item_t *child = ci->children + i;
+
+ if (strcasecmp ("Socket", child->key) == 0)
+ status = cf_util_get_string (child, &st->socket);
+ else if (strcasecmp ("Host", child->key) == 0)
+ status = cf_util_get_string (child, &st->host);
+ else if (strcasecmp ("Port", child->key) == 0)
+ status = cf_util_get_service (child, &st->port);
+ else
+ {
+ WARNING ("memcached plugin: Option `%s' not allowed here.",
+ child->key);
+ status = -1;
+ }
+
+ if (status != 0)
+ break;
+ }
+
+ if (status == 0)
+ status = memcached_add_read_callback (st);
+
+ if (status != 0)
+ {
+ memcached_free(st);
+ return (-1);
+ }
+
+ return (0);
}
-/* }}} */
-void module_register (void) /* {{{ */
+static int memcached_config (oconfig_item_t *ci)
{
- plugin_register_config ("memcached", memcached_config, config_keys, config_keys_num);
- plugin_register_read ("memcached", memcached_read);
+ int status = 0;
+ _Bool have_instance_block = 0;
+ int i;
+
+ for (i = 0; i < ci->children_num; i++)
+ {
+ oconfig_item_t *child = ci->children + i;
+
+ if (strcasecmp ("Instance", child->key) == 0)
+ {
+ config_add_instance (child);
+ have_instance_block = 1;
+ }
+ else if (!have_instance_block)
+ {
+ /* Non-instance option: Assume legacy configuration (without <Instance />
+ * blocks) and call config_add_instance() with the <Plugin /> block. */
+ return (config_add_instance (ci));
+ }
+ else
+ WARNING ("memcached plugin: The configuration option "
+ "\"%s\" is not allowed here. Did you "
+ "forget to add an <Instance /> block "
+ "around the configuration?",
+ child->key);
+ } /* for (ci->children) */
+
+ return (status);
}
-/* }}} */
-
-/*
- * Local variables:
- * tab-width: 4
- * c-basic-offset: 4
- * End:
- * vim600: sw=4 ts=4 fdm=marker noexpandtab
- * vim<600: sw=4 ts=4 noexpandtab
- */
+static int memcached_init (void)
+{
+ memcached_t *st;
+ int status;
+
+ if (memcached_have_instances)
+ return (0);
+
+ /* No instances were configured, lets start a default instance. */
+ st = malloc (sizeof (*st));
+ if (st == NULL)
+ return (ENOMEM);
+ memset (st, 0, sizeof (*st));
+ st->name = sstrdup ("__legacy__");
+ st->socket = NULL;
+ st->host = NULL;
+ st->port = NULL;
+
+ status = memcached_add_read_callback (st);
+ if (status == 0)
+ memcached_have_instances = 1;
+ else
+ memcached_free (st);
+
+ return (status);
+} /* int memcached_init */
+
+void module_register (void)
+{
+ plugin_register_complex_config ("memcached", memcached_config);
+ plugin_register_init ("memcached", memcached_init);
+}
/**
* collectd - src/ntpd.c
- * Copyright (C) 2006-2007 Florian octo Forster
+ * Copyright (C) 2006-2012 Florian octo Forster
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License as published by the
* 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*
* Authors:
- * Florian octo Forster <octo at verplant.org>
+ * Florian octo Forster <octo at collectd.org>
**/
#define _BSD_SOURCE /* For NI_MAXHOST */
{
"Host",
"Port",
- "ReverseLookups"
+ "ReverseLookups",
+ "IncludeUnitID"
};
static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
-static int do_reverse_lookups = 1;
+static _Bool do_reverse_lookups = 1;
+
+/* This option only exists for backward compatibility. If it is false and two
+ * ntpd peers use the same refclock driver, the plugin will try to write
+ * simultaneous measurements from both to the same type instance. */
+static _Bool include_unit_id = 0;
# define NTPD_DEFAULT_HOST "localhost"
# define NTPD_DEFAULT_PORT "123"
else
do_reverse_lookups = 0;
}
+ else if (strcasecmp (key, "IncludeUnitID") == 0)
+ {
+ if (IS_TRUE (value))
+ include_unit_id = 1;
+ else
+ include_unit_id = 0;
+ }
else
{
return (-1);
plugin_dispatch_values (&vl);
}
+/* Each time a peer is polled, ntpd shifts the reach register to the left and
+ * sets the LSB based on whether the peer was reachable. If the LSB is zero,
+ * the values are out of date. */
+static void ntpd_submit_reach (char *type, char *type_inst, uint8_t reach,
+ double value)
+{
+ if (!(reach & 1))
+ value = NAN;
+
+ ntpd_submit (type, type_inst, value);
+}
+
static int ntpd_connect (void)
{
char *host;
return (val_double);
}
+static uint32_t ntpd_get_refclock_id (struct info_peer_summary const *peer_info)
+{
+ uint32_t addr = ntohl (peer_info->srcadr);
+ uint32_t refclock_id = (addr >> 8) & 0x00FF;
+
+ return (refclock_id);
+}
+
+static int ntpd_get_name_from_address (char *buffer, size_t buffer_size,
+ struct info_peer_summary const *peer_info, _Bool do_reverse_lookup)
+{
+ struct sockaddr_storage sa;
+ socklen_t sa_len;
+ int flags = 0;
+ int status;
+
+ memset (&sa, 0, sizeof (sa));
+
+ if (peer_info->v6_flag)
+ {
+ struct sockaddr_in6 sa6;
+
+ assert (sizeof (sa) >= sizeof (sa6));
+
+ memset (&sa6, 0, sizeof (sa6));
+ sa6.sin6_family = AF_INET6;
+ sa6.sin6_port = htons (123);
+ memcpy (&sa6.sin6_addr, &peer_info->srcadr6,
+ sizeof (struct in6_addr));
+ sa_len = sizeof (sa6);
+
+ memcpy (&sa, &sa6, sizeof (sa6));
+ }
+ else
+ {
+ struct sockaddr_in sa4;
+
+ assert (sizeof (sa) >= sizeof (sa4));
+
+ memset (&sa4, 0, sizeof (sa4));
+ sa4.sin_family = AF_INET;
+ sa4.sin_port = htons (123);
+ memcpy (&sa4.sin_addr, &peer_info->srcadr,
+ sizeof (struct in_addr));
+ sa_len = sizeof (sa4);
+
+ memcpy (&sa, &sa4, sizeof (sa4));
+ }
+
+ if (!do_reverse_lookup)
+ flags |= NI_NUMERICHOST;
+
+ status = getnameinfo ((struct sockaddr const *) &sa, sa_len,
+ buffer, buffer_size,
+ NULL, 0, /* No port name */
+ flags);
+ if (status != 0)
+ {
+ char errbuf[1024];
+ ERROR ("ntpd plugin: getnameinfo failed: %s",
+ (status == EAI_SYSTEM)
+ ? sstrerror (errno, errbuf, sizeof (errbuf))
+ : gai_strerror (status));
+ return (-1);
+ }
+
+ return (0);
+} /* ntpd_get_name_from_address */
+
+static int ntpd_get_name_refclock (char *buffer, size_t buffer_size,
+ struct info_peer_summary const *peer_info)
+{
+ uint32_t refclock_id = ntpd_get_refclock_id (peer_info);
+ uint32_t unit_id = ntohl (peer_info->srcadr) & 0x00FF;
+
+ if (refclock_id >= refclock_names_num)
+ return (ntpd_get_name_from_address (buffer, buffer_size,
+ peer_info,
+ /* do_reverse_lookup = */ 0));
+
+ if (include_unit_id)
+ ssnprintf (buffer, buffer_size, "%s-%"PRIu32,
+ refclock_names[refclock_id], unit_id);
+ else
+ sstrncpy (buffer, refclock_names[refclock_id], buffer_size);
+
+ return (0);
+} /* int ntpd_get_name_refclock */
+
+static int ntpd_get_name (char *buffer, size_t buffer_size,
+ struct info_peer_summary const *peer_info)
+{
+ uint32_t addr = ntohl (peer_info->srcadr);
+
+ if (!peer_info->v6_flag && ((addr & REFCLOCK_MASK) == REFCLOCK_ADDR))
+ return (ntpd_get_name_refclock (buffer, buffer_size,
+ peer_info));
+ else
+ return (ntpd_get_name_from_address (buffer, buffer_size,
+ peer_info, do_reverse_lookups));
+} /* int ntpd_addr_to_name */
+
static int ntpd_read (void)
{
struct info_kernel *ik;
double offset;
char peername[NI_MAXHOST];
- int refclock_id;
-
- ptr = ps + i;
- refclock_id = 0;
+ uint32_t refclock_id;
- /* Convert the `long floating point' offset value to double */
- M_LFPTOD (ntohl (ptr->offset_int), ntohl (ptr->offset_frc), offset);
+ ptr = ps + i;
- /* Special IP addresses for hardware clocks and stuff.. */
- if (!ptr->v6_flag
- && ((ntohl (ptr->srcadr) & REFCLOCK_MASK)
- == REFCLOCK_ADDR))
+ status = ntpd_get_name (peername, sizeof (peername), ptr);
+ if (status != 0)
{
- struct in_addr addr_obj;
- char *addr_str;
-
- refclock_id = (ntohl (ptr->srcadr) >> 8) & 0x000000FF;
-
- if (refclock_id < refclock_names_num)
- {
- sstrncpy (peername, refclock_names[refclock_id],
- sizeof (peername));
- }
- else
- {
- memset ((void *) &addr_obj, '\0', sizeof (addr_obj));
- addr_obj.s_addr = ptr->srcadr;
- addr_str = inet_ntoa (addr_obj);
-
- sstrncpy (peername, addr_str, sizeof (peername));
- }
+ ERROR ("ntpd plugin: Determining name of peer failed.");
+ continue;
}
- else /* Normal network host. */
- {
- struct sockaddr_storage sa;
- socklen_t sa_len;
- int flags = 0;
-
- memset (&sa, '\0', sizeof (sa));
-
- if (ptr->v6_flag)
- {
- struct sockaddr_in6 sa6;
-
- assert (sizeof (sa) >= sizeof (sa6));
-
- memset (&sa6, 0, sizeof (sa6));
- sa6.sin6_family = AF_INET6;
- sa6.sin6_port = htons (123);
- memcpy (&sa6.sin6_addr, &ptr->srcadr6,
- sizeof (struct in6_addr));
- sa_len = sizeof (sa6);
-
- memcpy (&sa, &sa6, sizeof (sa6));
- }
- else
- {
- struct sockaddr_in sa4;
-
- assert (sizeof (sa) >= sizeof (sa4));
- memset (&sa4, 0, sizeof (sa4));
- sa4.sin_family = AF_INET;
- sa4.sin_port = htons (123);
- memcpy (&sa4.sin_addr, &ptr->srcadr,
- sizeof (struct in_addr));
- sa_len = sizeof (sa4);
+ refclock_id = ntpd_get_refclock_id (ptr);
- memcpy (&sa, &sa4, sizeof (sa4));
- }
-
- if (do_reverse_lookups == 0)
- flags |= NI_NUMERICHOST;
-
- status = getnameinfo ((const struct sockaddr *) &sa,
- sa_len,
- peername, sizeof (peername),
- NULL, 0, /* No port name */
- flags);
- if (status != 0)
- {
- char errbuf[1024];
- ERROR ("ntpd plugin: getnameinfo failed: %s",
- (status == EAI_SYSTEM)
- ? sstrerror (errno, errbuf, sizeof (errbuf))
- : gai_strerror (status));
- continue;
- }
- }
+ /* Convert the `long floating point' offset value to double */
+ M_LFPTOD (ntohl (ptr->offset_int), ntohl (ptr->offset_frc), offset);
DEBUG ("peer %i:\n"
" peername = %s\n"
" srcadr = 0x%08x\n"
+ " reach = 0%03o\n"
" delay = %f\n"
" offset_int = %i\n"
" offset_frc = %i\n"
i,
peername,
ntohl (ptr->srcadr),
+ ptr->reach,
ntpd_read_fp (ptr->delay),
ntohl (ptr->offset_int),
ntohl (ptr->offset_frc),
ntpd_read_fp (ptr->dispersion));
if (refclock_id != 1) /* not the system clock (offset will always be zero.. */
- ntpd_submit ("time_offset", peername, offset);
- ntpd_submit ("time_dispersion", peername, ntpd_read_fp (ptr->dispersion));
+ ntpd_submit_reach ("time_offset", peername, ptr->reach,
+ offset);
+ ntpd_submit_reach ("time_dispersion", peername, ptr->reach,
+ ntpd_read_fp (ptr->dispersion));
if (refclock_id == 0) /* not a reference clock */
- ntpd_submit ("delay", peername, ntpd_read_fp (ptr->delay));
+ ntpd_submit_reach ("delay", peername, ptr->reach,
+ ntpd_read_fp (ptr->delay));
}
free (ps);
struct o_database_s
{
char *name;
+ char *host;
char *connect_id;
char *username;
char *password;
* </Plugin>
*/
-static int o_config_set_string (char **ret_string, /* {{{ */
- oconfig_item_t *ci)
-{
- char *string;
-
- if ((ci->values_num != 1)
- || (ci->values[0].type != OCONFIG_TYPE_STRING))
- {
- WARNING ("oracle plugin: The `%s' config option "
- "needs exactly one string argument.", ci->key);
- return (-1);
- }
-
- string = strdup (ci->values[0].value.string);
- if (string == NULL)
- {
- ERROR ("oracle plugin: strdup failed.");
- return (-1);
- }
-
- if (*ret_string != NULL)
- free (*ret_string);
- *ret_string = string;
-
- return (0);
-} /* }}} int o_config_set_string */
-
static int o_config_add_database (oconfig_item_t *ci) /* {{{ */
{
o_database_t *db;
return (-1);
}
memset (db, 0, sizeof (*db));
+ db->name = NULL;
+ db->host = NULL;
+ db->connect_id = NULL;
+ db->username = NULL;
+ db->password = NULL;
- status = o_config_set_string (&db->name, ci);
+ status = cf_util_get_string (ci, &db->name);
if (status != 0)
{
sfree (db);
oconfig_item_t *child = ci->children + i;
if (strcasecmp ("ConnectID", child->key) == 0)
- status = o_config_set_string (&db->connect_id, child);
+ status = cf_util_get_string (child, &db->connect_id);
+ else if (strcasecmp ("Host", child->key) == 0)
+ status = cf_util_get_string (child, &db->host);
else if (strcasecmp ("Username", child->key) == 0)
- status = o_config_set_string (&db->username, child);
+ status = cf_util_get_string (child, &db->username);
else if (strcasecmp ("Password", child->key) == 0)
- status = o_config_set_string (&db->password, child);
+ status = cf_util_get_string (child, &db->password);
else if (strcasecmp ("Query", child->key) == 0)
status = udb_query_pick_from_list (child, queries, queries_num,
&db->queries, &db->queries_num);
} /* for (j = 1; j <= param_counter; j++) */
/* }}} End of the ``define'' stuff. */
- status = udb_query_prepare_result (q, prep_area, hostname_g,
+ status = udb_query_prepare_result (q, prep_area,
+ (db->host != NULL) ? db->host : hostname_g,
/* plugin = */ "oracle", db->name, column_names, column_num,
/* interval = */ 0);
if (status != 0)
* Copyright (C) 2009 Andrés J. DÃaz
* Copyright (C) 2009 Manuel Sanmartin
* Copyright (C) 2010 Clément Stenac
+ * Copyright (C) 2012 Cosmin Ioiart
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License as published by the
* Andrés J. DÃaz <ajdiaz at connectical.com>
* Manuel Sanmartin
* Clément Stenac <clement.stenac at diwi.org>
+ * Cosmin Ioiart <cioiart at gmail.com>
**/
#include "collectd.h"
#define MAXARGLN 1024
/* #endif HAVE_PROCINFO_H */
+#elif KERNEL_SOLARIS
+# include <procfs.h>
+# include <dirent.h>
+/* #endif KERNEL_SOLARIS */
+
#else
# error "No applicable input method."
#endif
# include <regex.h>
#endif
+#if HAVE_KSTAT_H
+# include <kstat.h>
+#endif
+
#ifndef ARG_MAX
# define ARG_MAX 4096
#endif
#endif /* HAVE_PROCINFO_H */
/* put name of process from config to list_head_g tree
- list_head_g is a list of 'procstat_t' structs with
- processes names we want to watch */
+ * list_head_g is a list of 'procstat_t' structs with
+ * processes names we want to watch */
static void ps_list_register (const char *name, const char *regexp)
{
procstat_t *new;
}
DEBUG ("name = %s; num_proc = %lu; num_lwp = %lu; "
- "vmem_size = %lu; vmem_rss = %lu; vmem_data = %lu; "
+ "vmem_size = %lu; vmem_rss = %lu; vmem_data = %lu; "
"vmem_code = %lu; "
"vmem_minflt_counter = %"PRIi64"; vmem_majflt_counter = %"PRIi64"; "
"cpu_user_counter = %"PRIi64"; cpu_system_counter = %"PRIi64"; "
ps->io_rchar, ps->io_wchar, ps->io_syscr, ps->io_syscw);
} /* void ps_submit_proc_list */
+#if KERNEL_LINUX || KERNEL_SOLARIS
+static void ps_submit_fork_rate (derive_t value)
+{
+ value_t values[1];
+ value_list_t vl = VALUE_LIST_INIT;
+
+ values[0].derive = value;
+
+ vl.values = values;
+ vl.values_len = 1;
+ sstrncpy(vl.host, hostname_g, sizeof (vl.host));
+ sstrncpy(vl.plugin, "processes", sizeof (vl.plugin));
+ sstrncpy(vl.plugin_instance, "", sizeof (vl.plugin_instance));
+ sstrncpy(vl.type, "fork_rate", sizeof (vl.type));
+ sstrncpy(vl.type_instance, "", sizeof (vl.type_instance));
+
+ plugin_dispatch_values(&vl);
+}
+#endif /* KERNEL_LINUX || KERNEL_SOLARIS*/
+
/* ------- additional functions for KERNEL_LINUX/HAVE_THREAD_INFO ------- */
#if KERNEL_LINUX
static int ps_read_tasks (int pid)
continue;
numfields = strsplit (buffer, fields,
- STATIC_ARRAY_SIZE (fields));
+ STATIC_ARRAY_SIZE (fields));
if (numfields < 2)
continue;
return buf;
} /* char *ps_get_cmdline (...) */
-static unsigned long read_fork_rate ()
+static int read_fork_rate ()
{
FILE *proc_stat;
- char buf[1024];
- unsigned long result = 0;
- int numfields;
- char *fields[3];
+ char buffer[1024];
+ value_t value;
+ _Bool value_valid = 0;
- proc_stat = fopen("/proc/stat", "r");
- if (proc_stat == NULL) {
+ proc_stat = fopen ("/proc/stat", "r");
+ if (proc_stat == NULL)
+ {
char errbuf[1024];
ERROR ("processes plugin: fopen (/proc/stat) failed: %s",
sstrerror (errno, errbuf, sizeof (errbuf)));
- return ULONG_MAX;
+ return (-1);
}
- while (fgets (buf, sizeof(buf), proc_stat) != NULL)
+ while (fgets (buffer, sizeof (buffer), proc_stat) != NULL)
{
- char *endptr;
+ int status;
+ char *fields[3];
+ int fields_num;
- numfields = strsplit(buf, fields, STATIC_ARRAY_SIZE (fields));
- if (numfields != 2)
+ fields_num = strsplit (buffer, fields,
+ STATIC_ARRAY_SIZE (fields));
+ if (fields_num != 2)
continue;
if (strcmp ("processes", fields[0]) != 0)
continue;
- errno = 0;
- endptr = NULL;
- result = strtoul(fields[1], &endptr, /* base = */ 10);
- if ((endptr == fields[1]) || (errno != 0)) {
- ERROR ("processes plugin: Cannot parse fork rate: %s",
- fields[1]);
- result = ULONG_MAX;
- break;
- }
+ status = parse_value (fields[1], &value, DS_TYPE_DERIVE);
+ if (status == 0)
+ value_valid = 1;
break;
}
-
fclose(proc_stat);
- return result;
+ if (!value_valid)
+ return (-1);
+
+ ps_submit_fork_rate (value.derive);
+ return (0);
}
+#endif /*KERNEL_LINUX */
-static void ps_submit_fork_rate (unsigned long value)
+#if KERNEL_SOLARIS
+static const char *ps_get_cmdline (pid_t pid, /* {{{ */
+ char *buffer, size_t buffer_size)
{
- value_t values[1];
- value_list_t vl = VALUE_LIST_INIT;
+ char path[PATH_MAX];
+ psinfo_t info;
+ int status;
- values[0].derive = (derive_t) value;
+ snprintf(path, sizeof (path), "/proc/%i/psinfo", pid);
- vl.values = values;
- vl.values_len = 1;
- sstrncpy (vl.host, hostname_g, sizeof (vl.host));
- sstrncpy (vl.plugin, "processes", sizeof (vl.plugin));
- sstrncpy (vl.plugin_instance, "", sizeof (vl.plugin_instance));
- sstrncpy (vl.type, "fork_rate", sizeof (vl.type));
- sstrncpy (vl.type_instance, "", sizeof (vl.type_instance));
+ status = read_file_contents (path, (void *) &info, sizeof (info));
+ if (status != ((int) buffer_size))
+ {
+ ERROR ("processes plugin: Unexpected return value "
+ "while reading \"%s\": "
+ "Returned %i but expected %zu.",
+ path, status, buffer_size);
+ return (NULL);
+ }
- plugin_dispatch_values (&vl);
+ info.pr_psargs[sizeof (info.pr_psargs) - 1] = 0;
+ sstrncpy (buffer, info.pr_psargs, buffer_size);
+
+ return (buffer);
+} /* }}} int ps_get_cmdline */
+
+/*
+ * Reads process information on the Solaris OS. The information comes mainly from
+ * /proc/PID/status, /proc/PID/psinfo and /proc/PID/usage
+ * The values for input and ouput chars are calculated "by hand"
+ * Added a few "solaris" specific process states as well
+ */
+static int ps_read_process(int pid, procstat_t *ps, char *state)
+{
+ char filename[64];
+ char f_psinfo[64], f_usage[64];
+ char *buffer;
+
+ pstatus_t *myStatus;
+ psinfo_t *myInfo;
+ prusage_t *myUsage;
+
+ snprintf(filename, sizeof (filename), "/proc/%i/status", pid);
+ snprintf(f_psinfo, sizeof (f_psinfo), "/proc/%i/psinfo", pid);
+ snprintf(f_usage, sizeof (f_usage), "/proc/%i/usage", pid);
+
+
+ buffer = malloc(sizeof (pstatus_t));
+ memset(buffer, 0, sizeof (pstatus_t));
+ read_file_contents(filename, buffer, sizeof (pstatus_t));
+ myStatus = (pstatus_t *) buffer;
+
+ buffer = malloc(sizeof (psinfo_t));
+ memset(buffer, 0, sizeof(psinfo_t));
+ read_file_contents(f_psinfo, buffer, sizeof (psinfo_t));
+ myInfo = (psinfo_t *) buffer;
+
+ buffer = malloc(sizeof (prusage_t));
+ memset(buffer, 0, sizeof(prusage_t));
+ read_file_contents(f_usage, buffer, sizeof (prusage_t));
+ myUsage = (prusage_t *) buffer;
+
+ sstrncpy(ps->name, myInfo->pr_fname, sizeof (myInfo->pr_fname));
+ ps->num_lwp = myStatus->pr_nlwp;
+ if (myInfo->pr_wstat != 0) {
+ ps->num_proc = 0;
+ ps->num_lwp = 0;
+ *state = (char) 'Z';
+ return (0);
+ } else {
+ ps->num_proc = 1;
+ ps->num_lwp = myInfo->pr_nlwp;
+ }
+
+ /*
+ * Convert system time and user time from nanoseconds to microseconds
+ * for compatibility with the linux module
+ */
+ ps->cpu_system_counter = myStatus -> pr_stime.tv_nsec / 1000;
+ ps->cpu_user_counter = myStatus -> pr_utime.tv_nsec / 1000;
+
+ /*
+ * Convert rssize from KB to bytes to be consistent w/ the linux module
+ */
+ ps->vmem_rss = myInfo->pr_rssize * 1024;
+ ps->vmem_size = myInfo->pr_size * 1024;
+ ps->vmem_minflt_counter = myUsage->pr_minf;
+ ps->vmem_majflt_counter = myUsage->pr_majf;
+
+ /*
+ * TODO: Data and code segment calculations for Solaris
+ */
+
+ ps->vmem_data = -1;
+ ps->vmem_code = -1;
+ ps->stack_size = myStatus->pr_stksize;
+
+ /*
+ * Calculating input/ouput chars
+ * Formula used is total chars / total blocks => chars/block
+ * then convert input/output blocks to chars
+ */
+ ulong_t tot_chars = myUsage->pr_ioch;
+ ulong_t tot_blocks = myUsage->pr_inblk + myUsage->pr_oublk;
+ ulong_t chars_per_block = 1;
+ if (tot_blocks != 0)
+ chars_per_block = tot_chars / tot_blocks;
+ ps->io_rchar = myUsage->pr_inblk * chars_per_block;
+ ps->io_wchar = myUsage->pr_oublk * chars_per_block;
+ ps->io_syscr = myUsage->pr_sysc;
+ ps->io_syscw = myUsage->pr_sysc;
+
+
+ /*
+ * TODO: Find way of setting BLOCKED and PAGING status
+ */
+
+ *state = (char) 'R';
+ if (myStatus->pr_flags & PR_ASLEEP)
+ *state = (char) 'S';
+ else if (myStatus->pr_flags & PR_STOPPED)
+ *state = (char) 'T';
+ else if (myStatus->pr_flags & PR_DETACH)
+ *state = (char) 'E';
+ else if (myStatus->pr_flags & PR_DAEMON)
+ *state = (char) 'A';
+ else if (myStatus->pr_flags & PR_ISSYS)
+ *state = (char) 'Y';
+ else if (myStatus->pr_flags & PR_ORPHAN)
+ *state = (char) 'O';
+
+ sfree(myStatus);
+ sfree(myInfo);
+ sfree(myUsage);
+
+ return (0);
}
-#endif /* KERNEL_LINUX */
+/*
+ * Reads the number of threads created since the last reboot. On Solaris these
+ * are retrieved from kstat (module cpu, name sys, class misc, stat nthreads).
+ * The result is the sum for all the threads created on each cpu
+ */
+static int read_fork_rate()
+{
+ extern kstat_ctl_t *kc;
+ kstat_t *ksp_chain = NULL;
+ derive_t result = 0;
+
+ if (kc == NULL)
+ return (-1);
+
+ for (ksp_chain = kc->kc_chain;
+ ksp_chain != NULL;
+ ksp_chain = ksp_chain->ks_next)
+ {
+ if ((strcmp (ksp_chain->ks_module, "cpu") == 0)
+ && (strcmp (ksp_chain->ks_name, "sys") == 0)
+ && (strcmp (ksp_chain->ks_class, "misc") == 0))
+ {
+ long long tmp;
+
+ kstat_read (kc, ksp_chain, NULL);
+
+ tmp = get_kstat_value(ksp_chain, "nthreads");
+ if (tmp != -1LL)
+ result += tmp;
+ }
+ }
+
+ ps_submit_fork_rate (result);
+ return (0);
+}
+#endif /* KERNEL_SOLARIS */
#if HAVE_THREAD_INFO
static int mach_get_task_name (task_t t, int *pid, char *name, size_t name_max_len)
procstat_entry_t pse;
char state;
- unsigned long fork_rate;
-
procstat_t *ps_ptr;
running = sleeping = zombies = stopped = paging = blocked = 0;
for (ps_ptr = list_head_g; ps_ptr != NULL; ps_ptr = ps_ptr->next)
ps_submit_proc_list (ps_ptr);
- fork_rate = read_fork_rate();
- if (fork_rate != ULONG_MAX)
- ps_submit_fork_rate(fork_rate);
+ read_fork_rate();
/* #endif KERNEL_LINUX */
#elif HAVE_LIBKVM_GETPROCS && HAVE_STRUCT_KINFO_PROC_FREEBSD
kvm_t *kd;
char errbuf[1024];
- struct kinfo_proc *procs; /* array of processes */
+ struct kinfo_proc *procs; /* array of processes */
struct kinfo_proc *proc_ptr = NULL;
- int count; /* returns number of processes */
+ int count; /* returns number of processes */
int i;
procstat_t *ps_ptr;
if (procentry[i].pi_pid == 0)
cmdline = "swapper";
cargs = cmdline;
- }
+ }
else
{
if (getargs(&procentry[i], sizeof(struct procentry64), arglist, MAXARGLN) >= 0)
for (ps = list_head_g; ps != NULL; ps = ps->next)
ps_submit_proc_list (ps);
-#endif /* HAVE_PROCINFO_H */
+/* #endif HAVE_PROCINFO_H */
+
+#elif KERNEL_SOLARIS
+ /*
+ * The Solaris section adds a few more process states and removes some
+ * process states compared to linux. Most notably there is no "PAGING"
+ * and "BLOCKED" state for a process. The rest is similar to the linux
+ * code.
+ */
+ int running = 0;
+ int sleeping = 0;
+ int zombies = 0;
+ int stopped = 0;
+ int detached = 0;
+ int daemon = 0;
+ int system = 0;
+ int orphan = 0;
+
+ struct dirent *ent;
+ DIR *proc;
+
+ int status;
+ procstat_t *ps_ptr;
+ char state;
+
+ char cmdline[PRARGSZ];
+
+ ps_list_reset ();
+
+ proc = opendir ("/proc");
+ if (proc == NULL)
+ return (-1);
+
+ while ((ent = readdir(proc)) != NULL)
+ {
+ int pid;
+ struct procstat ps;
+ procstat_entry_t pse;
+
+ if (!isdigit ((int) ent->d_name[0]))
+ continue;
+
+ if ((pid = atoi (ent->d_name)) < 1)
+ continue;
+
+ status = ps_read_process (pid, &ps, &state);
+ if (status != 0)
+ {
+ DEBUG("ps_read_process failed: %i", status);
+ continue;
+ }
+
+ pse.id = pid;
+ pse.age = 0;
+
+ pse.num_proc = ps.num_proc;
+ pse.num_lwp = ps.num_lwp;
+ pse.vmem_size = ps.vmem_size;
+ pse.vmem_rss = ps.vmem_rss;
+ pse.vmem_data = ps.vmem_data;
+ pse.vmem_code = ps.vmem_code;
+ pse.stack_size = ps.stack_size;
+
+ pse.vmem_minflt = 0;
+ pse.vmem_minflt_counter = ps.vmem_minflt_counter;
+ pse.vmem_majflt = 0;
+ pse.vmem_majflt_counter = ps.vmem_majflt_counter;
+
+ pse.cpu_user = 0;
+ pse.cpu_user_counter = ps.cpu_user_counter;
+ pse.cpu_system = 0;
+ pse.cpu_system_counter = ps.cpu_system_counter;
+
+ pse.io_rchar = ps.io_rchar;
+ pse.io_wchar = ps.io_wchar;
+ pse.io_syscr = ps.io_syscr;
+ pse.io_syscw = ps.io_syscw;
+
+ switch (state)
+ {
+ case 'R': running++; break;
+ case 'S': sleeping++; break;
+ case 'E': detached++; break;
+ case 'Z': zombies++; break;
+ case 'T': stopped++; break;
+ case 'A': daemon++; break;
+ case 'Y': system++; break;
+ case 'O': orphan++; break;
+ }
+
+
+ ps_list_add (ps.name,
+ ps_get_cmdline ((pid_t) pid,
+ cmdline, sizeof (cmdline)),
+ &pse);
+ } /* while(readdir) */
+ closedir (proc);
+
+ ps_submit_state ("running", running);
+ ps_submit_state ("sleeping", sleeping);
+ ps_submit_state ("zombies", zombies);
+ ps_submit_state ("stopped", stopped);
+ ps_submit_state ("detached", detached);
+ ps_submit_state ("daemon", daemon);
+ ps_submit_state ("system", system);
+ ps_submit_state ("orphan", orphan);
+
+ for (ps_ptr = list_head_g; ps_ptr != NULL; ps_ptr = ps_ptr->next)
+ ps_submit_proc_list (ps_ptr);
+
+ read_fork_rate();
+#endif /* KERNEL_SOLARIS */
return (0);
} /* int ps_read */
{
char name[MAX_REDIS_NODE_NAME];
char host[HOST_NAME_MAX];
+ char passwd[HOST_NAME_MAX];
int port;
int timeout;
}
else if (strcasecmp ("Timeout", option->key) == 0)
status = cf_util_get_int (option, &rn.timeout);
+ else if (strcasecmp ("Password", option->key) == 0)
+ status = cf_util_get_string_buffer (option, rn.passwd, sizeof (rn.passwd));
else
WARNING ("redis plugin: Option `%s' not allowed inside a `Node' "
"block. I'll ignore this option.", option->key);
continue;
}
+ if (strlen (rn->passwd) > 0)
+ {
+ DEBUG ("redis plugin: authenticanting node `%s' passwd(%s).", rn->name, rn->passwd);
+ status = credis_auth(rh, rn->passwd);
+ if (status != 0)
+ {
+ WARNING ("redis plugin: unable to authenticate on node `%s'.", rn->name);
+ credis_close (rh);
+ continue;
+ }
+ }
+
memset (&info, 0, sizeof (info));
status = credis_info (rh, &info);
if (status != 0)
/**
* collectd - src/swap.c
- * Copyright (C) 2005-2010 Florian octo Forster
+ * Copyright (C) 2005-2012 Florian octo Forster
* Copyright (C) 2009 Stefan Völkel
* Copyright (C) 2009 Manuel Sanmartin
* Copyright (C) 2010 Aurélien Reynaud
#define MAX(x,y) ((x) > (y) ? (x) : (y))
#if KERNEL_LINUX
-# define SWAP_HAVE_CONFIG 1
-/* No global variables */
+# define SWAP_HAVE_REPORT_BY_DEVICE 1
+static derive_t pagesize;
+static _Bool report_bytes = 0;
+static _Bool report_by_device = 0;
/* #endif KERNEL_LINUX */
#elif HAVE_SWAPCTL && HAVE_SWAPCTL_TWO_ARGS
-# define SWAP_HAVE_CONFIG 1
+# define SWAP_HAVE_REPORT_BY_DEVICE 1
static derive_t pagesize;
+static _Bool report_by_device = 0;
/* #endif HAVE_SWAPCTL && HAVE_SWAPCTL_TWO_ARGS */
#elif defined(VM_SWAPUSAGE)
# error "No applicable input method."
#endif /* HAVE_LIBSTATGRAB */
-#if SWAP_HAVE_CONFIG
static const char *config_keys[] =
{
+ "ReportBytes",
"ReportByDevice"
};
static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
-static _Bool report_by_device = 0;
-
static int swap_config (const char *key, const char *value) /* {{{ */
{
- if (strcasecmp ("ReportByDevice", key) == 0)
+ if (strcasecmp ("ReportBytes", key) == 0)
+ {
+#if KERNEL_LINUX
+ report_bytes = IS_TRUE (value) ? 1 : 0;
+#else
+ WARNING ("swap plugin: The \"ReportBytes\" option is only "
+ "valid under Linux. "
+ "The option is going to be ignored.");
+#endif
+ }
+ else if (strcasecmp ("ReportByDevice", key) == 0)
{
+#if SWAP_HAVE_REPORT_BY_DEVICE
if (IS_TRUE (value))
report_by_device = 1;
else
report_by_device = 0;
+#else
+ WARNING ("swap plugin: The \"ReportByDevice\" option is not "
+ "supported on this platform. "
+ "The option is going to be ignored.");
+#endif /* SWAP_HAVE_REPORT_BY_DEVICE */
}
else
{
return (0);
} /* }}} int swap_config */
-#endif /* SWAP_HAVE_CONFIG */
static int swap_init (void) /* {{{ */
{
#if KERNEL_LINUX
- /* No init stuff */
+ pagesize = (derive_t) sysconf (_SC_PAGESIZE);
/* #endif KERNEL_LINUX */
#elif HAVE_SWAPCTL && HAVE_SWAPCTL_TWO_ARGS
swap_submit (plugin_instance, "swap", type_instance, v);
} /* }}} void swap_submit_gauge */
-#if KERNEL_LINUX
+#if KERNEL_LINUX || HAVE_PERFSTAT
static void swap_submit_derive (const char *plugin_instance, /* {{{ */
const char *type_instance, derive_t value)
{
v.derive = value;
swap_submit (plugin_instance, "swap_io", type_instance, v);
} /* }}} void swap_submit_derive */
+#endif
+#if KERNEL_LINUX
static int swap_read_separate (void) /* {{{ */
{
FILE *fh;
if (have_data != 0x03)
return (ENOENT);
+ if (report_bytes)
+ {
+ swap_in = swap_in * pagesize;
+ swap_out = swap_out * pagesize;
+ }
+
swap_submit_derive (NULL, "in", swap_in);
swap_submit_derive (NULL, "out", swap_out);
sstrerror (errno, errbuf, sizeof (errbuf)));
return (-1);
}
+
swap_submit_gauge (NULL, "used", (gauge_t) (pmemory.pgsp_total - pmemory.pgsp_free) * pagesize);
swap_submit_gauge (NULL, "free", (gauge_t) pmemory.pgsp_free * pagesize );
+ swap_submit_gauge (NULL, "reserved", (gauge_t) pmemory.pgsp_rsvd * pagesize);
+ swap_submit_derive (NULL, "in", (derive_t) pmemory.pgspins * pagesize);
+ swap_submit_derive (NULL, "out", (derive_t) pmemory.pgspouts * pagesize);
return (0);
} /* }}} int swap_read */
void module_register (void)
{
-#if SWAP_HAVE_CONFIG
- plugin_register_config ("swap", swap_config, config_keys, config_keys_num);
-#endif
+ plugin_register_config ("swap", swap_config,
+ config_keys, config_keys_num);
plugin_register_init ("swap", swap_init);
plugin_register_read ("swap", swap_read);
} /* void module_register */
#endif
#if KERNEL_LINUX
+# include <asm/types.h>
+/* sys/socket.h is necessary to compile when using netlink on older systems. */
+# include <sys/socket.h>
+# include <linux/netlink.h>
+# include <linux/inet_diag.h>
+# include <sys/socket.h>
+# include <arpa/inet.h>
/* #endif KERNEL_LINUX */
#elif HAVE_SYSCTLBYNAME
#endif /* KERNEL_AIX */
#if KERNEL_LINUX
+struct nlreq {
+ struct nlmsghdr nlh;
+ struct inet_diag_req r;
+};
+
static const char *tcp_state[] =
{
"", /* 0 */
static int port_collect_listening = 0;
static port_entry_t *port_list_head = NULL;
+#if KERNEL_LINUX
+static uint32_t sequence_number = 0;
+
+enum
+{
+ SRC_DUNNO,
+ SRC_NETLINK,
+ SRC_PROC
+} linux_source = SRC_DUNNO;
+#endif
+
static void conn_submit_port_entry (port_entry_t *pe)
{
value_t values[1];
} /* int conn_handle_ports */
#if KERNEL_LINUX
+/* Returns zero on success, less than zero on socket error and greater than
+ * zero on other errors. */
+static int conn_read_netlink (void)
+{
+ int fd;
+ struct sockaddr_nl nladdr;
+ struct nlreq req;
+ struct msghdr msg;
+ struct iovec iov;
+ struct inet_diag_msg *r;
+ char buf[8192];
+
+ /* If this fails, it's likely a permission problem. We'll fall back to
+ * reading this information from files below. */
+ fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_INET_DIAG);
+ if (fd < 0)
+ {
+ ERROR ("tcpconns plugin: conn_read_netlink: socket(AF_NETLINK, SOCK_RAW, "
+ "NETLINK_INET_DIAG) failed: %s",
+ sstrerror (errno, buf, sizeof (buf)));
+ return (-1);
+ }
+
+ memset(&nladdr, 0, sizeof(nladdr));
+ nladdr.nl_family = AF_NETLINK;
+
+ memset(&req, 0, sizeof(req));
+ req.nlh.nlmsg_len = sizeof(req);
+ req.nlh.nlmsg_type = TCPDIAG_GETSOCK;
+ /* NLM_F_ROOT: return the complete table instead of a single entry.
+ * NLM_F_MATCH: return all entries matching criteria (not implemented)
+ * NLM_F_REQUEST: must be set on all request messages */
+ req.nlh.nlmsg_flags = NLM_F_ROOT | NLM_F_MATCH | NLM_F_REQUEST;
+ req.nlh.nlmsg_pid = 0;
+ /* The sequence_number is used to track our messages. Since netlink is not
+ * reliable, we don't want to end up with a corrupt or incomplete old
+ * message in case the system is/was out of memory. */
+ req.nlh.nlmsg_seq = ++sequence_number;
+ req.r.idiag_family = AF_INET;
+ req.r.idiag_states = 0xfff;
+ req.r.idiag_ext = 0;
+
+ memset(&iov, 0, sizeof(iov));
+ iov.iov_base = &req;
+ iov.iov_len = sizeof(req);
+
+ memset(&msg, 0, sizeof(msg));
+ msg.msg_name = (void*)&nladdr;
+ msg.msg_namelen = sizeof(nladdr);
+ msg.msg_iov = &iov;
+ msg.msg_iovlen = 1;
+
+ if (sendmsg (fd, &msg, 0) < 0)
+ {
+ ERROR ("tcpconns plugin: conn_read_netlink: sendmsg(2) failed: %s",
+ sstrerror (errno, buf, sizeof (buf)));
+ close (fd);
+ return (-1);
+ }
+
+ iov.iov_base = buf;
+ iov.iov_len = sizeof(buf);
+
+ while (1)
+ {
+ int status;
+ struct nlmsghdr *h;
+
+ memset(&msg, 0, sizeof(msg));
+ msg.msg_name = (void*)&nladdr;
+ msg.msg_namelen = sizeof(nladdr);
+ msg.msg_iov = &iov;
+ msg.msg_iovlen = 1;
+
+ status = recvmsg(fd, (void *) &msg, /* flags = */ 0);
+ if (status < 0)
+ {
+ if ((errno == EINTR) || (errno == EAGAIN))
+ continue;
+
+ ERROR ("tcpconns plugin: conn_read_netlink: recvmsg(2) failed: %s",
+ sstrerror (errno, buf, sizeof (buf)));
+ close (fd);
+ return (-1);
+ }
+ else if (status == 0)
+ {
+ close (fd);
+ DEBUG ("tcpconns plugin: conn_read_netlink: Unexpected zero-sized "
+ "reply from netlink socket.");
+ return (0);
+ }
+
+ h = (struct nlmsghdr*)buf;
+ while (NLMSG_OK(h, status))
+ {
+ if (h->nlmsg_seq != sequence_number)
+ {
+ h = NLMSG_NEXT(h, status);
+ continue;
+ }
+
+ if (h->nlmsg_type == NLMSG_DONE)
+ {
+ close (fd);
+ return (0);
+ }
+ else if (h->nlmsg_type == NLMSG_ERROR)
+ {
+ struct nlmsgerr *msg_error;
+
+ msg_error = NLMSG_DATA(h);
+ WARNING ("tcpconns plugin: conn_read_netlink: Received error %i.",
+ msg_error->error);
+
+ close (fd);
+ return (1);
+ }
+
+ r = NLMSG_DATA(h);
+
+ /* This code does not (need to) distinguish between IPv4 and IPv6. */
+ conn_handle_ports (ntohs(r->id.idiag_sport),
+ ntohs(r->id.idiag_dport),
+ r->idiag_state);
+
+ h = NLMSG_NEXT(h, status);
+ } /* while (NLMSG_OK) */
+ } /* while (1) */
+
+ /* Not reached because the while() loop above handles the exit condition. */
+ return (0);
+} /* int conn_read_netlink */
+
static int conn_handle_line (char *buffer)
{
char *fields[32];
static int conn_read (void)
{
- int errors_num = 0;
+ int status;
conn_reset_port_entry ();
- if (conn_read_file ("/proc/net/tcp") != 0)
- errors_num++;
- if (conn_read_file ("/proc/net/tcp6") != 0)
- errors_num++;
-
- if (errors_num < 2)
+ if (linux_source == SRC_NETLINK)
{
- conn_submit_all ();
+ status = conn_read_netlink ();
}
- else
+ else if (linux_source == SRC_PROC)
{
- ERROR ("tcpconns plugin: Neither /proc/net/tcp nor /proc/net/tcp6 "
- "coult be read.");
- return (-1);
+ int errors_num = 0;
+
+ if (conn_read_file ("/proc/net/tcp") != 0)
+ errors_num++;
+ if (conn_read_file ("/proc/net/tcp6") != 0)
+ errors_num++;
+
+ if (errors_num < 2)
+ status = 0;
+ else
+ status = ENOENT;
+ }
+ else /* if (linux_source == SRC_DUNNO) */
+ {
+ /* Try to use netlink for getting this data, it is _much_ faster on systems
+ * with a large amount of connections. */
+ status = conn_read_netlink ();
+ if (status == 0)
+ {
+ INFO ("tcpconns plugin: Reading from netlink succeeded. "
+ "Will use the netlink method from now on.");
+ linux_source = SRC_NETLINK;
+ }
+ else
+ {
+ INFO ("tcpconns plugin: Reading from netlink failed. "
+ "Will read from /proc from now on.");
+ linux_source = SRC_PROC;
+
+ /* return success here to avoid the "plugin failed" message. */
+ return (0);
+ }
}
+ if (status == 0)
+ conn_submit_all ();
+ else
+ return (status);
+
return (0);
} /* int conn_read */
/* #endif KERNEL_LINUX */
/* Using sysctl interface to retrieve the boot time on *BSD / Darwin / OS X systems */
/* #endif HAVE_SYS_SYSCTL_H */
+#elif HAVE_PERFSTAT
+# include <sys/protosw.h>
+# include <libperfstat.h>
+/* Using perfstat_cpu_total to retrive the boot time in AIX */
+/* #endif HAVE_PERFSTAT */
+
#else
# error "No applicable input method."
#endif
"but `boottime' is zero!");
return (-1);
}
-#endif /* HAVE_SYS_SYSCTL_H */
+/* #endif HAVE_SYS_SYSCTL_H */
+
+#elif HAVE_PERFSTAT
+ int status;
+ perfstat_cpu_total_t cputotal;
+ int hertz;
+
+ status = perfstat_cpu_total(NULL, &cputotal,
+ sizeof(perfstat_cpu_total_t), 1);
+ if (status < 0)
+ {
+ char errbuf[1024];
+ ERROR ("uptime plugin: perfstat_cpu_total: %s",
+ sstrerror (errno, errbuf, sizeof (errbuf)));
+ return (-1);
+ }
+
+ hertz = sysconf(_SC_CLK_TCK);
+ if (hertz <= 0)
+ hertz = HZ;
+
+ boottime = time(NULL) - cputotal.lbolt / hertz;
+#endif /* HAVE_PERFSTAT */
return (0);
} /* }}} int uptime_init */
char **names = NULL;
cdtime_t *times = NULL;
size_t number = 0;
+ size_t size_arrays = 0;
int status = 0;
pthread_mutex_lock (&cache_lock);
+ size_arrays = (size_t) c_avl_size (cache_tree);
+ if (size_arrays < 1)
+ {
+ /* Handle the "no values" case here, to avoid the error message when
+ * calloc() returns NULL. */
+ pthread_mutex_unlock (&cache_lock);
+ return (0);
+ }
+
+ names = calloc (size_arrays, sizeof (*names));
+ times = calloc (size_arrays, sizeof (*times));
+ if ((names == NULL) || (times == NULL))
+ {
+ ERROR ("uc_get_names: calloc failed.");
+ sfree (names);
+ sfree (times);
+ pthread_mutex_unlock (&cache_lock);
+ return (ENOMEM);
+ }
+
iter = c_avl_get_iterator (cache_tree);
while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0)
{
- char **temp;
-
/* remove missing values when list values */
if (value->state == STATE_MISSING)
continue;
- if (ret_times != NULL)
- {
- cdtime_t *tmp_times;
+ /* c_avl_size does not return a number smaller than the number of elements
+ * returned by c_avl_iterator_next. */
+ assert (number < size_arrays);
- tmp_times = (cdtime_t *) realloc (times, sizeof (cdtime_t) * (number + 1));
- if (tmp_times == NULL)
- {
- status = -1;
- break;
- }
- times = tmp_times;
+ if (ret_times != NULL)
times[number] = value->last_time;
- }
- temp = (char **) realloc (names, sizeof (char *) * (number + 1));
- if (temp == NULL)
- {
- status = -1;
- break;
- }
- names = temp;
names[number] = strdup (key);
if (names[number] == NULL)
{
status = -1;
break;
}
+
number++;
} /* while (c_avl_iterator_next) */
--- /dev/null
+/**
+ * collectd - src/utils_format_graphite.c
+ * Copyright (C) 2012 Thomas Meson
+ * Copyright (C) 2012 Florian octo Forster
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by the
+ * Free Software Foundation; only version 2 of the License is applicable.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+ *
+ * Authors:
+ * Thomas Meson <zllak at hycik.org>
+ * Florian octo Forster <octo at collectd.org>
+ **/
+
+#include "collectd.h"
+#include "plugin.h"
+#include "common.h"
+
+#include "utils_cache.h"
+#include "utils_format_json.h"
+#include "utils_parse_option.h"
+
+/* Utils functions to format data sets in graphite format.
+ * Largely taken from write_graphite.c as it remains the same formatting */
+
+static int gr_format_values (char *ret, size_t ret_len,
+ int ds_num, const data_set_t *ds, const value_list_t *vl)
+{
+ size_t offset = 0;
+ int status;
+
+ assert (0 == strcmp (ds->type, vl->type));
+
+ memset (ret, 0, ret_len);
+
+#define BUFFER_ADD(...) do { \
+ status = ssnprintf (ret + offset, ret_len - offset, \
+ __VA_ARGS__); \
+ if (status < 1) \
+ { \
+ return (-1); \
+ } \
+ else if (((size_t) status) >= (ret_len - offset)) \
+ { \
+ return (-1); \
+ } \
+ else \
+ offset += ((size_t) status); \
+} while (0)
+
+ if (ds->ds[ds_num].type == DS_TYPE_GAUGE)
+ BUFFER_ADD ("%f", vl->values[ds_num].gauge);
+ else if (ds->ds[ds_num].type == DS_TYPE_COUNTER)
+ BUFFER_ADD ("%llu", vl->values[ds_num].counter);
+ else if (ds->ds[ds_num].type == DS_TYPE_DERIVE)
+ BUFFER_ADD ("%"PRIi64, vl->values[ds_num].derive);
+ else if (ds->ds[ds_num].type == DS_TYPE_ABSOLUTE)
+ BUFFER_ADD ("%"PRIu64, vl->values[ds_num].absolute);
+ else
+ {
+ ERROR ("gr_format_values plugin: Unknown data source type: %i",
+ ds->ds[ds_num].type);
+ return (-1);
+ }
+
+#undef BUFFER_ADD
+
+ return (0);
+}
+
+static void gr_copy_escape_part (char *dst, const char *src, size_t dst_len,
+ char escape_char)
+{
+ size_t i;
+
+ memset (dst, 0, dst_len);
+
+ if (src == NULL)
+ return;
+
+ for (i = 0; i < dst_len; i++)
+ {
+ if (src[i] == 0)
+ {
+ dst[i] = 0;
+ break;
+ }
+
+ if ((src[i] == '.')
+ || isspace ((int) src[i])
+ || iscntrl ((int) src[i]))
+ dst[i] = escape_char;
+ else
+ dst[i] = src[i];
+ }
+}
+
+static int gr_format_name (char *ret, int ret_len,
+ const value_list_t *vl,
+ const char *ds_name,
+ char *prefix,
+ char *postfix,
+ char escape_char)
+{
+ char n_host[DATA_MAX_NAME_LEN];
+ char n_plugin[DATA_MAX_NAME_LEN];
+ char n_plugin_instance[DATA_MAX_NAME_LEN];
+ char n_type[DATA_MAX_NAME_LEN];
+ char n_type_instance[DATA_MAX_NAME_LEN];
+
+ char tmp_plugin[2 * DATA_MAX_NAME_LEN + 1];
+ char tmp_type[2 * DATA_MAX_NAME_LEN + 1];
+
+ if (prefix == NULL)
+ prefix = "";
+
+ if (postfix == NULL)
+ postfix = "";
+
+ gr_copy_escape_part (n_host, vl->host,
+ sizeof (n_host), escape_char);
+ gr_copy_escape_part (n_plugin, vl->plugin,
+ sizeof (n_plugin), escape_char);
+ gr_copy_escape_part (n_plugin_instance, vl->plugin_instance,
+ sizeof (n_plugin_instance), escape_char);
+ gr_copy_escape_part (n_type, vl->type,
+ sizeof (n_type), escape_char);
+ gr_copy_escape_part (n_type_instance, vl->type_instance,
+ sizeof (n_type_instance), escape_char);
+
+ if (n_plugin_instance[0] != '\0')
+ ssnprintf (tmp_plugin, sizeof (tmp_plugin), "%s%c%s",
+ n_plugin,
+ '-',
+ n_plugin_instance);
+ else
+ sstrncpy (tmp_plugin, n_plugin, sizeof (tmp_plugin));
+
+ if (n_type_instance[0] != '\0')
+ ssnprintf (tmp_type, sizeof (tmp_type), "%s%c%s",
+ n_type,
+ '-',
+ n_type_instance);
+ else
+ sstrncpy (tmp_type, n_type, sizeof (tmp_type));
+
+ if (ds_name != NULL)
+ ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s",
+ prefix, n_host, postfix, tmp_plugin, tmp_type, ds_name);
+ else
+ ssnprintf (ret, ret_len, "%s%s%s.%s.%s",
+ prefix, n_host, postfix, tmp_plugin, tmp_type);
+
+ return (0);
+}
+
+int format_graphite (char *buffer, size_t buffer_size,
+ const data_set_t *ds, const value_list_t *vl, char *prefix,
+ char *postfix, char escape_char)
+{
+ int status = 0;
+ int i;
+ int buffer_pos = 0;
+
+ for (i = 0; i < ds->ds_num; i++)
+ {
+ const char *ds_name = NULL;
+ char key[10*DATA_MAX_NAME_LEN];
+ char values[512];
+ size_t message_len;
+ char message[1024];
+
+ ds_name = ds->ds[i].name;
+
+ /* Copy the identifier to `key' and escape it. */
+ status = gr_format_name (key, sizeof (key), vl, ds_name,
+ prefix, postfix, escape_char);
+ if (status != 0)
+ {
+ ERROR ("amqp plugin: error with gr_format_name");
+ return (status);
+ }
+
+ escape_string (key, sizeof (key));
+ /* Convert the values to an ASCII representation and put that into
+ * `values'. */
+ status = gr_format_values (values, sizeof (values), i, ds, vl);
+ if (status != 0)
+ {
+ ERROR ("format_graphite: error with gr_format_values");
+ return (status);
+ }
+
+ /* Compute the graphite command */
+ message_len = (size_t) ssnprintf (message, sizeof (message),
+ "%s %s %u\r\n",
+ key,
+ values,
+ (unsigned int) CDTIME_T_TO_TIME_T (vl->time));
+ if (message_len >= sizeof (message)) {
+ ERROR ("format_graphite: message buffer too small: "
+ "Need %zu bytes.", message_len + 1);
+ return (-ENOMEM);
+ }
+
+ /* Append it in case we got multiple data set */
+ if ((buffer_pos + message_len) >= buffer_size)
+ {
+ ERROR ("format_graphite: target buffer too small");
+ return (-ENOMEM);
+ }
+ memcpy((void *) (buffer + buffer_pos), message, message_len);
+ buffer_pos += message_len;
+ }
+ return (status);
+} /* int format_graphite */
+
+/* vim: set sw=2 sts=2 et fdm=marker : */
--- /dev/null
+/**
+ * collectd - src/utils_format_graphite.h
+ * Copyright (C) 2012 Thomas Meson
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by the
+ * Free Software Foundation; only version 2 of the License is applicable.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+ *
+ * Author:
+ * Thomas Meson <zllak at hycik.org>
+ **/
+
+#ifndef UTILS_FORMAT_GRAPHITE_H
+#define UTILS_FORMAT_GRAPHITE_H 1
+
+#include "collectd.h"
+#include "plugin.h"
+
+int format_graphite (char *buffer,
+ size_t buffer_size, const data_set_t *ds,
+ const value_list_t *vl, const char *prefix,
+ const char *postfix, const char escape_char);
+
+#endif /* UTILS_FORMAT_GRAPHITE_H */
#undef BUFFER_ADD
return (0);
-} /* }}} int buffer_add_string */
+} /* }}} int escape_string */
static int values_to_json (char *buffer, size_t buffer_size, /* {{{ */
const data_set_t *ds, const value_list_t *vl, int store_rates)
} /* }}} int values_to_json */
static int dstypes_to_json (char *buffer, size_t buffer_size, /* {{{ */
- const data_set_t *ds, const value_list_t *vl)
+ const data_set_t *ds)
{
size_t offset = 0;
int i;
} /* }}} int dstypes_to_json */
static int dsnames_to_json (char *buffer, size_t buffer_size, /* {{{ */
- const data_set_t *ds, const value_list_t *vl)
+ const data_set_t *ds)
{
size_t offset = 0;
int i;
return (0);
} /* }}} int dsnames_to_json */
+static int meta_data_to_json (char *buffer, size_t buffer_size, /* {{{ */
+ meta_data_t *meta)
+{
+ size_t offset = 0;
+ char **keys = NULL;
+ int keys_num;
+ int status;
+ int i;
+
+ memset (buffer, 0, buffer_size);
+
+#define BUFFER_ADD(...) do { \
+ status = ssnprintf (buffer + offset, buffer_size - offset, \
+ __VA_ARGS__); \
+ if (status < 1) \
+ return (-1); \
+ else if (((size_t) status) >= (buffer_size - offset)) \
+ return (-ENOMEM); \
+ else \
+ offset += ((size_t) status); \
+} while (0)
+
+ keys_num = meta_data_toc (meta, &keys);
+ for (i = 0; i < keys_num; ++i)
+ {
+ int type;
+ char *key = keys[i];
+
+ type = meta_data_type (meta, key);
+ if (type == MD_TYPE_STRING)
+ {
+ char *value = NULL;
+ if (meta_data_get_string (meta, key, &value) == 0)
+ {
+ char temp[512] = "";
+ escape_string (temp, sizeof (temp), value);
+ sfree (value);
+ BUFFER_ADD (",\"%s\":%s", key, temp);
+ }
+ }
+ else if (type == MD_TYPE_SIGNED_INT)
+ {
+ int64_t value = 0;
+ if (meta_data_get_signed_int (meta, key, &value) == 0)
+ BUFFER_ADD (",\"%s\":%"PRIi64, key, value);
+ }
+ else if (type == MD_TYPE_UNSIGNED_INT)
+ {
+ uint64_t value = 0;
+ if (meta_data_get_unsigned_int (meta, key, &value) == 0)
+ BUFFER_ADD (",\"%s\":%"PRIu64, key, value);
+ }
+ else if (type == MD_TYPE_DOUBLE)
+ {
+ double value = 0.0;
+ if (meta_data_get_double (meta, key, &value) == 0)
+ BUFFER_ADD (",\"%s\":%f", key, value);
+ }
+ else if (type == MD_TYPE_BOOLEAN)
+ {
+ _Bool value = 0;
+ if (meta_data_get_boolean (meta, key, &value) == 0)
+ BUFFER_ADD (",\"%s\":%s", key, value ? "true" : "false");
+ }
+
+ free (key);
+ } /* for (keys) */
+ free (keys);
+
+ if (offset <= 0)
+ return (ENOENT);
+
+ buffer[0] = '{'; /* replace leading ',' */
+ BUFFER_ADD ("}");
+
+#undef BUFFER_ADD
+
+ return (0);
+} /* int meta_data_to_json */
+
static int value_list_to_json (char *buffer, size_t buffer_size, /* {{{ */
const data_set_t *ds, const value_list_t *vl, int store_rates)
{
return (status);
BUFFER_ADD ("\"values\":%s", temp);
- status = dstypes_to_json (temp, sizeof (temp), ds, vl);
+ status = dstypes_to_json (temp, sizeof (temp), ds);
if (status != 0)
return (status);
BUFFER_ADD (",\"dstypes\":%s", temp);
- status = dsnames_to_json (temp, sizeof (temp), ds, vl);
+ status = dsnames_to_json (temp, sizeof (temp), ds);
if (status != 0)
return (status);
BUFFER_ADD (",\"dsnames\":%s", temp);
BUFFER_ADD_KEYVAL ("type", vl->type);
BUFFER_ADD_KEYVAL ("type_instance", vl->type_instance);
+ if (vl->meta != NULL)
+ {
+ char meta_buffer[buffer_size];
+ memset (meta_buffer, 0, sizeof (meta_buffer));
+ status = meta_data_to_json (meta_buffer, sizeof (meta_buffer), vl->meta);
+ if (status != 0)
+ return (status);
+
+ BUFFER_ADD (",\"meta\":%s", meta_buffer);
+ } /* if (vl->meta != NULL) */
+
BUFFER_ADD ("}");
#undef BUFFER_ADD_KEYVAL
#include "utils_cache.h"
#include "utils_parse_option.h"
+#include "utils_format_graphite.h"
/* Folks without pthread will need to disable this plugin. */
#include <pthread.h>
return (status);
}
-static int wg_format_values (char *ret, size_t ret_len,
- int ds_num, const data_set_t *ds, const value_list_t *vl,
- _Bool store_rates)
-{
- size_t offset = 0;
- int status;
- gauge_t *rates = NULL;
-
- assert (0 == strcmp (ds->type, vl->type));
-
- memset (ret, 0, ret_len);
-
-#define BUFFER_ADD(...) do { \
- status = ssnprintf (ret + offset, ret_len - offset, \
- __VA_ARGS__); \
- if (status < 1) \
- { \
- sfree (rates); \
- return (-1); \
- } \
- else if (((size_t) status) >= (ret_len - offset)) \
- { \
- sfree (rates); \
- return (-1); \
- } \
- else \
- offset += ((size_t) status); \
-} while (0)
-
- if (ds->ds[ds_num].type == DS_TYPE_GAUGE)
- BUFFER_ADD ("%f", vl->values[ds_num].gauge);
- else if (store_rates)
- {
- if (rates == NULL)
- rates = uc_get_rate (ds, vl);
- if (rates == NULL)
- {
- WARNING ("format_values: "
- "uc_get_rate failed.");
- return (-1);
- }
- BUFFER_ADD ("%g", rates[ds_num]);
- }
- else if (ds->ds[ds_num].type == DS_TYPE_COUNTER)
- BUFFER_ADD ("%llu", vl->values[ds_num].counter);
- else if (ds->ds[ds_num].type == DS_TYPE_DERIVE)
- BUFFER_ADD ("%"PRIi64, vl->values[ds_num].derive);
- else if (ds->ds[ds_num].type == DS_TYPE_ABSOLUTE)
- BUFFER_ADD ("%"PRIu64, vl->values[ds_num].absolute);
- else
- {
- ERROR ("format_values plugin: Unknown data source type: %i",
- ds->ds[ds_num].type);
- sfree (rates);
- return (-1);
- }
-
-#undef BUFFER_ADD
-
- sfree (rates);
- return (0);
-}
-
-static void wg_copy_escape_part (char *dst, const char *src, size_t dst_len,
- char escape_char)
-{
- size_t i;
-
- memset (dst, 0, dst_len);
-
- if (src == NULL)
- return;
-
- for (i = 0; i < dst_len; i++)
- {
- if (src[i] == 0)
- {
- dst[i] = 0;
- break;
- }
-
- if ((src[i] == '.')
- || isspace ((int) src[i])
- || iscntrl ((int) src[i]))
- dst[i] = escape_char;
- else
- dst[i] = src[i];
- }
-}
-
-static int wg_format_name (char *ret, int ret_len,
- const value_list_t *vl,
- const struct wg_callback *cb,
- const char *ds_name)
-{
- char n_host[DATA_MAX_NAME_LEN];
- char n_plugin[DATA_MAX_NAME_LEN];
- char n_plugin_instance[DATA_MAX_NAME_LEN];
- char n_type[DATA_MAX_NAME_LEN];
- char n_type_instance[DATA_MAX_NAME_LEN];
-
- char *prefix;
- char *postfix;
-
- char tmp_plugin[2 * DATA_MAX_NAME_LEN + 1];
- char tmp_type[2 * DATA_MAX_NAME_LEN + 1];
-
- prefix = cb->prefix;
- if (prefix == NULL)
- prefix = "";
-
- postfix = cb->postfix;
- if (postfix == NULL)
- postfix = "";
-
- wg_copy_escape_part (n_host, vl->host,
- sizeof (n_host), cb->escape_char);
- wg_copy_escape_part (n_plugin, vl->plugin,
- sizeof (n_plugin), cb->escape_char);
- wg_copy_escape_part (n_plugin_instance, vl->plugin_instance,
- sizeof (n_plugin_instance), cb->escape_char);
- wg_copy_escape_part (n_type, vl->type,
- sizeof (n_type), cb->escape_char);
- wg_copy_escape_part (n_type_instance, vl->type_instance,
- sizeof (n_type_instance), cb->escape_char);
-
- if (n_plugin_instance[0] != '\0')
- ssnprintf (tmp_plugin, sizeof (tmp_plugin), "%s%c%s",
- n_plugin,
- cb->separate_instances ? '.' : '-',
- n_plugin_instance);
- else
- sstrncpy (tmp_plugin, n_plugin, sizeof (tmp_plugin));
-
- if (n_type_instance[0] != '\0')
- ssnprintf (tmp_type, sizeof (tmp_type), "%s%c%s",
- n_type,
- cb->separate_instances ? '.' : '-',
- n_type_instance);
- else
- sstrncpy (tmp_type, n_type, sizeof (tmp_type));
-
- if (ds_name != NULL)
- ssnprintf (ret, ret_len, "%s%s%s.%s.%s.%s",
- prefix, n_host, postfix, tmp_plugin, tmp_type, ds_name);
- else
- ssnprintf (ret, ret_len, "%s%s%s.%s.%s",
- prefix, n_host, postfix, tmp_plugin, tmp_type);
-
- return (0);
-}
-
-static int wg_send_message (const char* key, const char* value,
- cdtime_t time, struct wg_callback *cb)
+static int wg_send_message (char const *message, struct wg_callback *cb)
{
int status;
size_t message_len;
- char message[1024];
-
- message_len = (size_t) ssnprintf (message, sizeof (message),
- "%s %s %u\r\n",
- key,
- value,
- (unsigned int) CDTIME_T_TO_TIME_T (time));
- if (message_len >= sizeof (message)) {
- ERROR ("write_graphite plugin: message buffer too small: "
- "Need %zu bytes.", message_len + 1);
- return (-1);
- }
+
+ message_len = strlen (message);
pthread_mutex_lock (&cb->send_lock);
static int wg_write_messages (const data_set_t *ds, const value_list_t *vl,
struct wg_callback *cb)
{
- char key[10*DATA_MAX_NAME_LEN];
- char values[512];
-
- int status, i;
+ char buffer[4096];
+ int status;
if (0 != strcmp (ds->type, vl->type))
{
return -1;
}
- for (i = 0; i < ds->ds_num; i++)
- {
- const char *ds_name = NULL;
-
- if (cb->always_append_ds || (ds->ds_num > 1))
- ds_name = ds->ds[i].name;
-
- /* Copy the identifier to `key' and escape it. */
- status = wg_format_name (key, sizeof (key), vl, cb, ds_name);
- if (status != 0)
- {
- ERROR ("write_graphite plugin: error with format_name");
- return (status);
- }
-
- escape_string (key, sizeof (key));
- /* Convert the values to an ASCII representation and put that into
- * `values'. */
- status = wg_format_values (values, sizeof (values), i, ds, vl,
- cb->store_rates);
- if (status != 0)
- {
- ERROR ("write_graphite plugin: error with "
- "wg_format_values");
- return (status);
- }
+ memset (buffer, 0, sizeof (buffer));
+ status = format_graphite (buffer, sizeof (buffer), ds, vl,
+ cb->prefix, cb->postfix, cb->escape_char);
+ if (status != 0) /* error message has been printed already. */
+ return (status);
- /* Send the message to graphite */
- status = wg_send_message (key, values, vl->time, cb);
- if (status != 0)
- {
- ERROR ("write_graphite plugin: error with "
- "wg_send_message");
- return (status);
- }
+ wg_send_message (buffer, cb);
+ if (status != 0)
+ {
+ ERROR ("write_graphite plugin: wg_send_message failed "
+ "with status %i.", status);
+ return (status);
}
return (0);
-}
+} /* int wg_write_messages */
static int wg_write (const data_set_t *ds, const value_list_t *vl,
user_data_t *user_data)