Pierre-Yves Ritschard <pyr at spootnik.org>
- Write-Riemann plugin.
- Write-Graphite plugin: Notification support.
+ - Write-Kafka plugin.
+ - Log-Logstash plugin.
+ - Normalization in the CPU plugin.
+ - Relative values in the Load plugin.
Piotr Hosowicz <the55 at wp.pl>
- SMF manifest for collectd.
requests. The transmitted data is either in a form understood by the
Exec plugin or formatted in JSON.
+ - write_kafka
+ Sends data to Apache Kafka, a distributed queue.
+
- write_mongodb
Sends data to MongoDB, a NoSQL database.
Used by the AMQP plugin for AMQP connections, for example to RabbitMQ.
<http://hg.rabbitmq.com/rabbitmq-c/>
+ * librdkafka (optional; also called “rdkafka”)
+ Used by the Kafka plugin for producing.
+ <https://github.com/edenhill/librdkafka>
+
* librouteros (optional)
Used by the `routeros' plugin to connect to a device running `RouterOS'.
<http://verplant.org/librouteros/>
uint8_t c[8];
double d;
- d = 8.642135e130;
+ d = 8.642135e130;
memcpy ((void *) &i0, (void *) &d, 8);
i1 = i0;
uint8_t c[8];
double d;
- d = 8.642135e130;
+ d = 8.642135e130;
memcpy ((void *) &i0, (void *) &d, 8);
i1 = endianflip (i0);
uint8_t c[8];
double d;
- d = 8.642135e130;
+ d = 8.642135e130;
memcpy ((void *) &i0, (void *) &d, 8);
i1 = intswap (i0);
have_htonll="yes"
AC_DEFINE(HAVE_HTONLL, 1, [Define if the function htonll exists.])
])
-
+
AC_MSG_RESULT([$have_htonll])
# Check for structures
AM_CONDITIONAL([BUILD_FREEBSD],[test "x$x$ac_system" = "xFreeBSD"])
-AM_CONDITIONAL([BUILD_AIX],[test "x$x$ac_system" = "xAIX"])
+AM_CONDITIONAL([BUILD_AIX],[test "x$x$ac_system" = "xAIX"])
if test "x$ac_system" = "xAIX"
then
else
SAVE_CPPFLAGS="$CPPFLAGS"
CPPFLAGS="$CPPFLAGS $with_snmp_cflags"
-
+
AC_CHECK_HEADERS(net-snmp/net-snmp-config.h, [], [with_libnetsnmp="no (net-snmp/net-snmp-config.h not found)"])
CPPFLAGS="$SAVE_CPPFLAGS"
then
SAVE_CPPFLAGS="$CPPFLAGS"
CPPFLAGS="$with_libowcapi_cppflags"
-
+
AC_CHECK_HEADERS(owcapi.h, [with_libowcapi="yes"], [with_libowcapi="no (owcapi.h not found)"])
CPPFLAGS="$SAVE_CPPFLAGS"
SAVE_CPPFLAGS="$CPPFLAGS"
LDFLAGS="$with_libowcapi_libs"
CPPFLAGS="$with_libowcapi_cppflags"
-
+
AC_CHECK_LIB(owcapi, OW_get, [with_libowcapi="yes"], [with_libowcapi="no (libowcapi not found)"])
LDFLAGS="$SAVE_LDFLAGS"
AM_CONDITIONAL(BUILD_WITH_LIBRABBITMQ, test "x$with_librabbitmq" = "xyes")
# }}}
+# --with-librdkafka {{{
+AC_ARG_WITH(librdkafka, [AS_HELP_STRING([--with-librdkafka@<:@=PREFIX@:>@], [Path to librdkafka.])],
+[
+ if test "x$withval" = "xno" && test "x$withval" != "xyes"
+ then
+ with_librdkafka_cppflags="-I$withval/include"
+ with_librdkafka_ldflags="-L$withval/lib"
+ with_librdkafka="yes"
+ else
+ with_librdkafka="$withval"
+ fi
+],
+[
+ with_librdkafka="yes"
+])
+SAVE_CPPFLAGS="$CPPFLAGS"
+SAVE_LDFLAGS="$LDFLAGS"
+
+if test "x$with_librdkafka" = "xyes"
+then
+ AC_CHECK_HEADERS(librdkafka/rdkafka.h, [with_librdkafka="yes"], [with_librdkafka="no (librdkafka/rdkafka.h not found)"])
+fi
+
+if test "x$with_librdkafka" = "xyes"
+then
+ AC_CHECK_LIB(rdkafka, rd_kafka_new, [with_librdkafka="yes"], [with_librdkafka="no (Symbol 'rd_kafka_new' not found)"])
+fi
+if test "x$with_librdkafka" = "xyes"
+then
+ BUILD_WITH_LIBRDKAFKA_CPPFLAGS="$with_librdkafka_cppflags"
+ BUILD_WITH_LIBRDKAFKA_LDFLAGS="$with_librdkafka_ldflags"
+ BUILD_WITH_LIBRDKAFKA_LIBS="-lrdkafka"
+ AC_SUBST(BUILD_WITH_LIBRDKAFKA_CPPFLAGS)
+ AC_SUBST(BUILD_WITH_LIBRDKAFKA_LDFLAGS)
+ AC_SUBST(BUILD_WITH_LIBRDKAFKA_LIBS)
+ AC_DEFINE(HAVE_LIBRDKAFKA, 1, [Define if librdkafka is present and usable.])
+fi
+CPPFLAGS="$SAVE_CPPFLAGS"
+LDFLAGS="$SAVE_LDFLAGS"
+AM_CONDITIONAL(BUILD_WITH_LIBRDKAFKA, test "x$with_librdkafka" = "xyes")
+
+# }}}
+
# --with-librouteros {{{
AC_ARG_WITH(librouteros, [AS_HELP_STRING([--with-librouteros@<:@=PREFIX@:>@], [Path to librouteros.])],
[
LDFLAGS="$SAVE_LDFLAGS"
if test "x$with_libtokyotyrant" = "xyes"
-then
+then
BUILD_WITH_LIBTOKYOTYRANT_CPPFLAGS="$with_libtokyotyrant_cppflags"
BUILD_WITH_LIBTOKYOTYRANT_LDFLAGS="$with_libtokyotyrant_ldflags"
BUILD_WITH_LIBTOKYOTYRANT_LIBS="$with_libtokyotyrant_libs"
AM_CONDITIONAL(BUILD_WITH_LIBTOKYOTYRANT, test "x$with_libtokyotyrant" = "xyes")
# }}}
+# --with-libudev {{{
+with_libudev_cflags=""
+with_libudev_ldflags=""
+AC_ARG_WITH(libudev, [AS_HELP_STRING([--with-libudev@<:@=PREFIX@:>@], [Path to libudev.])],
+[
+ if test "x$withval" = "xno"
+ then
+ with_libudev="no"
+ else
+ with_libudev="yes"
+ if test "x$withval" != "xyes"
+ then
+ with_libudev_cflags="-I$withval/include"
+ with_libudev_ldflags="-L$withval/lib"
+ with_libudev="yes"
+ fi
+ fi
+],
+[
+ if test "x$ac_system" = "xLinux"
+ then
+ with_libudev="yes"
+ else
+ with_libudev="no (Linux only library)"
+ fi
+])
+if test "x$with_libudev" = "xyes"
+then
+ SAVE_CPPFLAGS="$CPPFLAGS"
+ CPPFLAGS="$CPPFLAGS $with_libudev_cflags"
+
+ AC_CHECK_HEADERS(libudev.h, [], [with_libudev="no (libudev.h not found)"])
+
+ CPPFLAGS="$SAVE_CPPFLAGS"
+fi
+if test "x$with_libudev" = "xyes"
+then
+ SAVE_CPPFLAGS="$CPPFLAGS"
+ SAVE_LDFLAGS="$LDFLAGS"
+ CPPFLAGS="$CPPFLAGS $with_libudev_cflags"
+ LDFLAGS="$LDFLAGS $with_libudev_ldflags"
+
+ AC_CHECK_LIB(udev, udev_new,
+ [
+ AC_DEFINE(HAVE_LIBUDEV, 1, [Define to 1 if you have the udev library (-ludev).])
+ ],
+ [with_libudev="no (libudev not found)"])
+
+ CPPFLAGS="$SAVE_CPPFLAGS"
+ LDFLAGS="$SAVE_LDFLAGS"
+fi
+if test "x$with_libudev" = "xyes"
+then
+ BUILD_WITH_LIBUDEV_CFLAGS="$with_libudev_cflags"
+ BUILD_WITH_LIBUDEV_LDFLAGS="$with_libudev_ldflags"
+ AC_SUBST(BUILD_WITH_LIBUDEV_CFLAGS)
+ AC_SUBST(BUILD_WITH_LIBUDEV_LDFLAGS)
+fi
+AM_CONDITIONAL(BUILD_WITH_LIBUDEV, test "x$with_libudev" = "xyes")
+# }}}
+
# --with-libupsclient {{{
with_libupsclient_config=""
with_libupsclient_cflags=""
plugin_irq="no"
plugin_libvirt="no"
plugin_load="no"
+plugin_log_logstash="no"
plugin_memory="no"
plugin_multimeter="no"
plugin_nfs="no"
plugin_load="yes"
fi
+if test "x$with_libyajl" = "xyes"
+then
+ plugin_log_logstash="yes"
+fi
+
if test "x$c_cv_have_libperl$c_cv_have_perl_ithreads" = "xyesyes"
then
plugin_perl="yes"
AC_PLUGIN([libvirt], [$plugin_libvirt], [Virtual machine statistics])
AC_PLUGIN([load], [$plugin_load], [System load])
AC_PLUGIN([logfile], [yes], [File logging plugin])
+AC_PLUGIN([log_logstash], [$plugin_log_logstash], [Logstash json_event compatible logging])
AC_PLUGIN([lpar], [$with_perfstat], [AIX logical partitions statistics])
AC_PLUGIN([lvm], [$with_liblvm2app], [LVM statistics])
AC_PLUGIN([madwifi], [$have_linux_wireless_h], [Madwifi wireless statistics])
AC_PLUGIN([wireless], [$plugin_wireless], [Wireless statistics])
AC_PLUGIN([write_graphite], [yes], [Graphite / Carbon output plugin])
AC_PLUGIN([write_http], [$with_libcurl], [HTTP output plugin])
+AC_PLUGIN([write_kafka], [$with_librdkafka], [Kafka output plugin])
AC_PLUGIN([write_mongodb], [$with_libmongoc], [MongoDB output plugin])
AC_PLUGIN([write_redis], [$with_libcredis], [Redis output plugin])
AC_PLUGIN([write_riemann], [$have_protoc_c], [Riemann output plugin])
libpq . . . . . . . . $with_libpq
libpthread . . . . . $with_libpthread
librabbitmq . . . . . $with_librabbitmq
+ librdkafka . . . . . $with_librdkafka
librouteros . . . . . $with_librouteros
librrd . . . . . . . $with_librrd
libsensors . . . . . $with_libsensors
libsigrok . . . . . $with_libsigrok
libstatgrab . . . . . $with_libstatgrab
libtokyotyrant . . . $with_libtokyotyrant
+ libudev . . . . . . . $with_libudev
libupsclient . . . . $with_libupsclient
libvarnish . . . . . $with_libvarnish
libvirt . . . . . . . $with_libvirt
load . . . . . . . . $enable_load
logfile . . . . . . . $enable_logfile
lpar . . . . . . . . $enable_lpar
+ log_logstash . . . . $enable_log_logstash
lvm . . . . . . . . . $enable_lvm
madwifi . . . . . . . $enable_madwifi
match_empty_counter . $enable_match_empty_counter
wireless . . . . . . $enable_wireless
write_graphite . . . $enable_write_graphite
write_http . . . . . $enable_write_http
+ write_kafka . . . . . $enable_write_kafka
write_mongodb . . . . $enable_write_mongodb
write_redis . . . . . $enable_write_redis
write_riemann . . . . $enable_write_riemann
utils_subst.c utils_subst.h \
utils_tail.c utils_tail.h \
utils_time.c utils_time.h \
- types_list.c types_list.h
+ types_list.c types_list.h \
+ utils_threshold.c utils_threshold.h
+
collectd_CPPFLAGS = $(AM_CPPFLAGS) $(LTDLINCL)
collectd_CFLAGS = $(AM_CFLAGS)
collectd_tg_DEPENDENCIES = libcollectdclient/libcollectdclient.la
-pkglib_LTLIBRARIES =
+pkglib_LTLIBRARIES =
-BUILT_SOURCES =
+BUILT_SOURCES =
CLEANFILES =
if BUILD_PLUGIN_AGGREGATION
cpu_la_SOURCES = cpu.c
cpu_la_CFLAGS = $(AM_CFLAGS)
cpu_la_LDFLAGS = -module -avoid-version
-cpu_la_LIBADD =
+cpu_la_LIBADD =
if BUILD_WITH_LIBKSTAT
cpu_la_LIBADD += -lkstat
endif
disk_la_SOURCES = disk.c
disk_la_CFLAGS = $(AM_CFLAGS)
disk_la_LDFLAGS = -module -avoid-version
-disk_la_LIBADD =
+disk_la_LIBADD =
if BUILD_WITH_LIBKSTAT
disk_la_LIBADD += -lkstat
endif
disk_la_LDFLAGS += -framework IOKit
endif
if BUILD_WITH_LIBSTATGRAB
-disk_la_CFLAGS += $(BUILD_WITH_LIBSTATGRAB_CFLAGS)
+disk_la_CFLAGS += $(BUILD_WITH_LIBSTATGRAB_CFLAGS)
disk_la_LIBADD += $(BUILD_WITH_LIBSTATGRAB_LDFLAGS)
endif
+if BUILD_WITH_LIBUDEV
+disk_la_LIBADD += -ludev
+endif
if BUILD_WITH_PERFSTAT
disk_la_LIBADD += -lperfstat
endif
collectd_DEPENDENCIES += logfile.la
endif
+if BUILD_PLUGIN_LOG_LOGSTASH
+pkglib_LTLIBRARIES += log_logstash.la
+log_logstash_la_SOURCES = log_logstash.c
+log_logstash_la_CFLAGS = $(AM_CFLAGS)
+log_logstash_la_LDFLAGS = -module -avoid-version $(BUILD_WITH_LIBYAJL_LDFLAGS)
+log_logstash_la_CPPFLAGS = $(AM_CPPFLAGS) $(BUILD_WITH_LIBYAJL_CPPFLAGS)
+log_logstash_la_LIBADD = $(BUILD_WITH_LIBYAJL_LIBS)
+collectd_LDADD += "-dlopen" log_logstash.la
+collectd_DEPENDENCIES += log_logstash.la
+endif
+
if BUILD_PLUGIN_LPAR
pkglib_LTLIBRARIES += lpar.la
lpar_la_SOURCES = lpar.c
pkglib_LTLIBRARIES += olsrd.la
olsrd_la_SOURCES = olsrd.c
olsrd_la_LDFLAGS = -module -avoid-version
-olsrd_la_LIBADD =
+olsrd_la_LIBADD =
if BUILD_WITH_LIBSOCKET
olsrd_la_LIBADD += -lsocket
endif
unixsock_la_SOURCES = unixsock.c \
utils_cmd_flush.h utils_cmd_flush.c \
utils_cmd_getval.h utils_cmd_getval.c \
+ utils_cmd_getthreshold.h utils_cmd_getthreshold.c \
utils_cmd_listval.h utils_cmd_listval.c \
utils_cmd_putval.h utils_cmd_putval.c \
utils_cmd_putnotif.h utils_cmd_putnotif.c
collectd_DEPENDENCIES += write_http.la
endif
+if BUILD_PLUGIN_WRITE_KAFKA
+pkglib_LTLIBRARIES += write_kafka.la
+write_kafka_la_SOURCES = write_kafka.c \
+ utils_format_graphite.c utils_format_graphite.h \
+ utils_format_json.c utils_format_json.h \
+ utils_cmd_putval.c utils_cmd_putval.h \
+ utils_crc32.c utils_crc32.h
+write_kafka_la_LDFLAGS = -module -avoid-version $(BUILD_WITH_LIBRDKAFKA_LDFLAGS)
+write_kafka_la_LIBADD = $(BUILD_WITH_LIBRDKAFKA_LIBS)
+collectd_LDADD += "-dlopen" write_kafka.la
+collectd_DEPENDENCIES += write_kafka.la
+endif
+
if BUILD_PLUGIN_WRITE_MONGODB
pkglib_LTLIBRARIES += write_mongodb.la
write_mongodb_la_SOURCES = write_mongodb.c
if BUILD_PLUGIN_WRITE_RIEMANN
pkglib_LTLIBRARIES += write_riemann.la
-write_riemann_la_SOURCES = write_riemann.c
+write_riemann_la_SOURCES = write_riemann.c write_riemann_threshold.c
nodist_write_riemann_la_SOURCES = riemann.pb-c.c riemann.pb-c.h
write_riemann_la_LDFLAGS = -module -avoid-version
write_riemann_la_LIBADD = -lprotobuf-c
/* subscribe only */
char *exchange_type;
char *queue;
+ _Bool queue_durable;
+ _Bool queue_auto_delete;
amqp_connection_state_t connection;
pthread_mutex_t lock;
? amqp_cstring_bytes (conf->queue)
: AMQP_EMPTY_BYTES,
/* passive = */ 0,
- /* durable = */ 0,
+ /* durable = */ conf->queue_durable,
/* exclusive = */ 0,
- /* auto_delete = */ 1,
+ /* auto_delete = */ conf->queue_auto_delete,
/* arguments = */ AMQP_EMPTY_TABLE);
if (qd_ret == NULL)
{
{
camqp_config_t *conf = user_data->data;
char routing_key[6 * DATA_MAX_NAME_LEN];
- char buffer[4096];
+ char buffer[8192];
int status;
if ((ds == NULL) || (vl == NULL) || (conf == NULL))
/* subscribe only */
conf->exchange_type = NULL;
conf->queue = NULL;
+ conf->queue_durable = 0;
+ conf->queue_auto_delete = 1;
/* general */
conf->connection = NULL;
pthread_mutex_init (&conf->lock, /* attr = */ NULL);
status = cf_util_get_string (child, &conf->exchange_type);
else if ((strcasecmp ("Queue", child->key) == 0) && !publish)
status = cf_util_get_string (child, &conf->queue);
+ else if ((strcasecmp ("QueueDurable", child->key) == 0) && !publish)
+ status = cf_util_get_boolean (child, &conf->queue_durable);
+ else if ((strcasecmp ("QueueAutoDelete", child->key) == 0) && !publish)
+ status = cf_util_get_boolean (child, &conf->queue_auto_delete);
else if (strcasecmp ("RoutingKey", child->key) == 0)
status = cf_util_get_string (child, &conf->routing_key);
else if ((strcasecmp ("Persistent", child->key) == 0) && publish)
@LOAD_PLUGIN_SYSLOG@LoadPlugin syslog
@LOAD_PLUGIN_LOGFILE@LoadPlugin logfile
+@LOAD_PLUGIN_LOG_LOGSTASH@LoadPlugin log_logstash
#<Plugin logfile>
# LogLevel @DEFAULT_LOG_LEVEL@
# PrintSeverity false
#</Plugin>
+#<Plugin log_logstash>
+# LogLevel @DEFAULT_LOG_LEVEL@
+# File "@localstatedir@/log/@PACKAGE_NAME@.json.log"
+#</Plugin>
+
#<Plugin syslog>
# LogLevel @DEFAULT_LOG_LEVEL@
#</Plugin>
# ription of those options is available in the collectd.conf(5) manual page. #
##############################################################################
-#<Plugin "aggregation">
+#<Plugin aggregation>
# <Aggregation>
# #Host "unspecified"
# Plugin "cpu"
# </Aggregation>
#</Plugin>
-#<Plugin "amqp">
+#<Plugin amqp>
# <Publish "name">
# Host "localhost"
# Port "5672"
# CACert "/etc/ssl/ca.crt"
#</Plugin>
-#<Plugin "bind">
+#<Plugin bind>
# URL "http://localhost:8053/"
# ParseTime false
# OpCodes true
# IgnoreSelected false
#</Plugin>
+#<Plugin cpu>
+# ReportActive false
+# ReportByCpu true
+# ValuesPercentage false
+#</Plugin>
+#
#<Plugin csv>
# DataDir "@localstatedir@/lib/@PACKAGE_NAME@/csv"
# StoreRates false
# </URL>
#</Plugin>
-#<Plugin "curl_xml">
+#<Plugin curl_xml>
# <URL "http://localhost/stats.xml">
# Host "my_host"
# Instance "some_instance"
#<Plugin disk>
# Disk "/^[hs]d[a-f][0-9]?$/"
# IgnoreSelected false
+# UseBSDName false
+# UdevNameAttr "DEVNAME"
#</Plugin>
#<Plugin dns>
# </Directory>
#</Plugin>
-#<Plugin "gmond">
+#<Plugin gmond>
# MCReceiveFrom "239.2.11.71" "8649"
# <Metric "swap_total">
# Type "swap"
# IgnoreSelected true
#</Plugin>
-#<Plugin "java">
+#<Plugin java>
# JVMArg "-verbose:jni"
# JVMArg "-Djava.class.path=@prefix@/share/collectd/java/collectd-api.jar"
#
# TimerPercentile 90.0
#</Plugin>
-#<Plugin "swap">
+#<Plugin swap>
# ReportByDevice false
# ReportBytes true
# ValuesAbsolute true
# ValuesPercentage false
#</Plugin>
-#<Plugin "table">
+#<Plugin table>
# <Table "/proc/slabinfo">
# Instance "slabinfo"
# Separator " "
# </Table>
#</Plugin>
-#<Plugin "tail">
+#<Plugin tail>
# <File "/var/log/exim4/mainlog">
# Instance "exim"
# Interval 60
# </File>
#</Plugin>
-#<Plugin "tail_csv">
+#<Plugin tail_csv>
# <Metric "dropped">
# Type "percent"
# Instance "dropped"
##############################################################################
#@BUILD_PLUGIN_THRESHOLD_TRUE@LoadPlugin "threshold"
-#<Plugin "threshold">
+#<Plugin threshold>
# <Type "foo">
# WarningMin 0.00
# WarningMax 1000.00
Exchange "amq.fanout"
# ExchangeType "fanout"
# Queue "queue_name"
+ # QueueDurable false
+ # QueueAutoDelete true
# RoutingKey "collectd.#"
</Subscribe>
</Plugin>
=item B<Queue> I<Queue> (Subscribe only)
-Configures the I<queue> name to subscribe to. If no queue name was configures
+Configures the I<queue> name to subscribe to. If no queue name was configured
explicitly, a unique queue name will be created by the broker.
+=item B<QueueDurable> B<true>|B<false> (Subscribe only)
+
+Defines if the I<queue> subscribed to is durable (saved to persistent storage)
+or transient (will disappear if the AMQP broker is restarted). Defaults to
+"false".
+
+This option should be used in conjunction with the I<Persistent> option on the
+publish side.
+
+=item B<QueueAutoDelete> B<true>|B<false> (Subscribe only)
+
+Defines if the I<queue> subscribed to will be deleted once the last consumer
+unsubscribes. Defaults to "true".
+
=item B<RoutingKey> I<Key>
In I<Publish> blocks, this configures the routing key to set on all outgoing
set to B<false>, B<only> matching disks will be collected. If B<IgnoreSelected>
is set to B<true>, all disks are collected B<except> the ones matched.
+=item B<UseBSDName> B<true>|B<false>
+
+Whether to use the device's "BSD Name", on MacE<nbsp>OSE<nbsp>X, instead of the
+default major/minor numbers. Requires collectd to be built with Apple's
+IOKitLib support.
+
+=item B<UdevNameAttr> I<Attribute>
+
+Attempt to override disk instance name with the value of a specified udev
+attribute when built with B<libudev>. If the attribute is not defined for the
+given device, the default name is used. Example:
+
+ UdevNameAttr "DM_NAME"
+
=back
=head2 Plugin C<dns>
log file (e.E<nbsp>g. when rotating the logs). The plugin reopens the file
for each line it writes.
+=head2 Plugin C<log_logstash>
+
+The I<log logstash plugin> behaves like the logfile plugin but formats
+messages as JSON events for logstash to parse and input.
+
+=over 4
+
+=item B<LogLevel> B<debug|info|notice|warning|err>
+
+Sets the log-level. If, for example, set to B<notice>, then all events with
+severity B<notice>, B<warning>, or B<err> will be written to the logfile.
+
+Please note that B<debug> is only available if collectd has been compiled with
+debugging support.
+
+=item B<File> I<File>
+
+Sets the file to write log messages to. The special strings B<stdout> and
+B<stderr> can be used to write to the standard output and standard error
+channels, respectively. This, of course, only makes much sense when I<collectd>
+is running in foreground- or non-daemon-mode.
+
+=back
+
+B<Note>: There is no need to notify the daemon after moving or removing the
+log file (e.E<nbsp>g. when rotating the logs). The plugin reopens the file
+for each line it writes.
+
=head2 Plugin C<lpar>
The I<LPAR plugin> reads CPU statistics of I<Logical Partitions>, a
=back
+=head2 Plugin C<write_kafka>
+
+The I<write_kafka plugin> will send values to a I<Kafka> topic, a distributed
+queue.
+Synopsis:
+
+ <Plugin "write_kafka">
+ Property "metadata.broker.list" "broker1:9092,broker2:9092"
+ <Topic "collectd">
+ Format JSON
+ </Topic>
+ </Plugin>
+
+The following options are understood by the I<write_kafka plugin>:
+
+=over 4
+
+=item E<lt>B<Topic> I<Name>E<gt>
+
+The plugin's configuration consists of one or more B<Topic> blocks. Each block
+is given a unique I<Name> and specifies one kafka producer.
+Inside the B<Topic> block, the following per-topic options are
+understood:
+
+=over 4
+
+=item B<Property> I<String> I<String>
+
+Configure the named property for the current topic. Properties are
+forwarded to the kafka producer library B<librdkafka>.
+
+=item B<Key> I<String>
+
+Use the specified string as a partioning key for the topic. Kafka breaks
+topic into partitions and guarantees that for a given topology, the same
+consumer will be used for a specific key. The special (case insensitive)
+string B<Random> can be used to specify that an arbitrary partition should
+be used.
+
+=item B<Format> B<Command>|B<JSON>|B<Graphite>
+
+Selects the format in which messages are sent to the broker. If set to
+B<Command> (the default), values are sent as C<PUTVAL> commands which are
+identical to the syntax used by the I<Exec> and I<UnixSock plugins>.
+
+If set to B<JSON>, the values are encoded in the I<JavaScript Object Notation>,
+an easy and straight forward exchange format.
+
+If set to B<Graphite>, values are encoded in the I<Graphite> format, which is
+"<metric> <value> <timestamp>\n".
+
+=item B<StoreRates> B<true>|B<false>
+
+Determines whether or not C<COUNTER>, C<DERIVE> and C<ABSOLUTE> data sources
+are converted to a I<rate> (i.e. a C<GAUGE> value). If set to B<false> (the
+default), no conversion is performed. Otherwise the conversion is performed
+using the internal value cache.
+
+Please note that currently this option is only used if the B<Format> option has
+been set to B<JSON>.
+
+=item B<GraphitePrefix> (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> (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> (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>).
+
+=item B<GraphiteSeparateInstances> B<false>|B<true>
+
+If set to B<true>, the plugin instance and type instance will be in their own
+path component, for example C<host.cpu.0.cpu.idle>. If set to B<false> (the
+default), the plugin and plugin instance (and likewise the type and type
+instance) are put into one component, for example C<host.cpu-0.cpu-idle>.
+
+=item B<StoreRates> B<true>|B<false>
+
+If set to B<true> (the default), convert counter values to rates. If set to
+B<false> counter values are stored as is, i.e. as an increasing integer number.
+
+This will be reflected in the C<ds_type> tag: If B<StoreRates> is enabled,
+converted values will have "rate" appended to the data source type, e.g.
+C<ds_type:derive:rate>.
+
+=back
+
+=item B<Property> I<String> I<String>
+
+Configure the kafka producer through properties, you almost always will
+want to set B<metadata.broker.list> to your Kafka broker list.
+
+=back
+
=head2 Plugin C<write_riemann>
The I<write_riemann plugin> will send values to I<Riemann>, a powerful stream
know exactly what you're doing, you should only increase this setting from its
default value.
+=item B<Notifications> B<false>|B<true>
+
+If set to B<true>, create riemann events for notifications. This is B<true>
+by default. When processing thresholds from write_riemann, it might prove
+useful to avoid getting notification events.
+
+=item B<CheckThresholds> B<false>|B<true>
+
+If set to B<true>, attach state to events based on thresholds defined
+in the B<Threshold> plugin. Defaults to B<false>.
+
=back
=item B<Tag> I<String>
return -1;
}
- for (i = percents_cells; i < size; i++)
+ percents = tmp;
+
+ for (i = percents_cells ; i < size; i++)
memset(&percents[i], 0, sizeof(*percents));
- percents = tmp;
percents_cells = size;
return 0;
} /* cpu_states_grow */
# error "No applicable input method."
#endif
+#if HAVE_LIBUDEV
+#include <libudev.h>
+
+static char *conf_udev_name_attr = NULL;
+static struct udev *handle_udev;
+#endif
+
static const char *config_keys[] =
{
"Disk",
"UseBSDName",
- "IgnoreSelected"
+ "IgnoreSelected",
+ "UdevNameAttr"
};
static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
"on Mach / Mac OS X and will be ignored.");
#endif
}
+ else if (strcasecmp ("UdevNameAttr", key) == 0)
+ {
+#if HAVE_LIBUDEV
+ if (conf_udev_name_attr != NULL)
+ {
+ free (conf_udev_name_attr);
+ conf_udev_name_attr = NULL;
+ }
+ if ((conf_udev_name_attr = strdup (value)) == NULL)
+ return (1);
+#else
+ WARNING ("disk plugin: The \"UdevNameAttr\" option is only supported "
+ "if collectd is built with libudev support");
+#endif
+ }
else
{
return (-1);
}
#endif
+#if HAVE_LIBUDEV
+/**
+ * Attempt to provide an rename disk instance from an assigned udev attribute.
+ *
+ * On success, it returns a strduped char* to the desired attribute value.
+ * Otherwise it returns NULL.
+ */
+
+static char *disk_udev_attr_name (struct udev *udev, char *disk_name, const char *attr)
+{
+ struct udev_device *dev;
+ const char *prop;
+ char *output = NULL;
+
+ dev = udev_device_new_from_subsystem_sysname (udev, "block", disk_name);
+ if (dev != NULL)
+ {
+ prop = udev_device_get_property_value (dev, attr);
+ if (prop) {
+ output = strdup (prop);
+ DEBUG ("disk plugin: renaming %s => %s", disk_name, output);
+ }
+ udev_device_unref (dev);
+ }
+ return output;
+}
+#endif
+
#if HAVE_IOKIT_IOKITLIB_H
static signed long long dict_get_value (CFDictionaryRef dict, const char *key)
{
fieldshift = 1;
}
+#if HAVE_LIBUDEV
+ handle_udev = udev_new();
+#endif
+
while (fgets (buffer, sizeof (buffer), fh) != NULL)
{
char *disk_name;
+ char *output_name;
+ char *alt_name;
numfields = strsplit (buffer, fields, 32);
continue;
}
+ output_name = disk_name;
+
+#if HAVE_LIBUDEV
+ alt_name = disk_udev_attr_name (handle_udev, disk_name,
+ conf_udev_name_attr);
+#else
+ alt_name = NULL;
+#endif
+ if (alt_name != NULL)
+ output_name = alt_name;
+
if ((ds->read_bytes != 0) || (ds->write_bytes != 0))
- disk_submit (disk_name, "disk_octets",
+ disk_submit (output_name, "disk_octets",
ds->read_bytes, ds->write_bytes);
if ((ds->read_ops != 0) || (ds->write_ops != 0))
- disk_submit (disk_name, "disk_ops",
+ disk_submit (output_name, "disk_ops",
read_ops, write_ops);
if ((ds->avg_read_time != 0) || (ds->avg_write_time != 0))
- disk_submit (disk_name, "disk_time",
+ disk_submit (output_name, "disk_time",
ds->avg_read_time, ds->avg_write_time);
if (is_disk)
{
- disk_submit (disk_name, "disk_merged",
+ disk_submit (output_name, "disk_merged",
read_merged, write_merged);
} /* if (is_disk) */
+
+ /* release udev-based alternate name, if allocated */
+ free(alt_name);
} /* while (fgets (buffer, sizeof (buffer), fh) != NULL) */
+#if HAVE_LIBUDEV
+ udev_unref(handle_udev);
+#endif
+
fclose (fh);
/* #endif defined(KERNEL_LINUX) */
#undef SET_STRING
/* Set the `time' member. Java stores time in milliseconds. */
- status = ctoj_long (jvm_env, ((jlong) n->time) * ((jlong) 1000),
+ status = ctoj_long (jvm_env, (jlong) CDTIME_T_TO_MS (n->time),
c_notification, o_notification, "setTime");
if (status != 0)
{
return (-1);
}
/* Java measures time in milliseconds. */
- n->time = (time_t) (tmp_long / ((jlong) 1000));
+ n->time = MS_TO_CDTIME_T(tmp_long);
status = jtoc_int (jvm_env, &tmp_int,
class_ptr, object_ptr, "getSeverity");
--- /dev/null
+/**
+ * collectd - src/log_logstash.c
+ * Copyright (C) 2013 Pierre-Yves Ritschard
+ *
+ * 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:
+ * Pierre-Yves Ritschard <pyr at spootnik.org>
+ * Acknowledgements:
+ * This file is largely inspired by logfile.c
+ **/
+
+#include "collectd.h"
+#include "common.h"
+#include "plugin.h"
+
+#include <sys/types.h>
+#include <pthread.h>
+#include <yajl/yajl_common.h>
+#include <yajl/yajl_gen.h>
+#if HAVE_YAJL_YAJL_VERSION_H
+# include <yajl/yajl_version.h>
+#endif
+#if defined(YAJL_MAJOR) && (YAJL_MAJOR > 1)
+# define HAVE_YAJL_V2 1
+#endif
+
+#define DEFAULT_LOGFILE LOCALSTATEDIR"/log/"PACKAGE_NAME".json.log"
+
+#if COLLECT_DEBUG
+static int log_level = LOG_DEBUG;
+#else
+static int log_level = LOG_INFO;
+#endif /* COLLECT_DEBUG */
+
+static pthread_mutex_t file_lock = PTHREAD_MUTEX_INITIALIZER;
+
+static char *log_file = NULL;
+
+static const char *config_keys[] =
+{
+ "LogLevel",
+ "File"
+};
+static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
+
+static int log_logstash_config (const char *key, const char *value)
+{
+
+ if (0 == strcasecmp (key, "LogLevel")) {
+ log_level = parse_log_severity(value);
+ if (log_level < 0) {
+ log_level = LOG_INFO;
+ ERROR("log_logstash: invalid loglevel [%s] defaulting to 'info'",
+ value);
+ return 1;
+ }
+ }
+ else if (0 == strcasecmp (key, "File")) {
+ sfree (log_file);
+ log_file = strdup (value);
+ }
+ else {
+ return -1;
+ }
+ return 0;
+} /* int log_logstash_config (const char *, const char *) */
+
+static void log_logstash_print (yajl_gen g, int severity,
+ cdtime_t timestamp_time)
+{
+ FILE *fh;
+ _Bool do_close = 0;
+ struct tm timestamp_tm;
+ char timestamp_str[64];
+ const unsigned char *buf;
+ time_t tt;
+#if HAVE_YAJL_V2
+ size_t len;
+#else
+ unsigned int len;
+#endif
+
+ if (yajl_gen_string(g, (u_char *)"@level", strlen("@level")) !=
+ yajl_gen_status_ok)
+ goto err;
+
+ switch (severity) {
+ case LOG_ERR:
+ if (yajl_gen_string(g, (u_char *)"error", strlen("error")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ case LOG_WARNING:
+ if (yajl_gen_string(g, (u_char *)"warning",
+ strlen("warning")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ case LOG_NOTICE:
+ if (yajl_gen_string(g, (u_char *)"notice", strlen("notice")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ case LOG_INFO:
+ if (yajl_gen_string(g, (u_char *)"info", strlen("info")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ case LOG_DEBUG:
+ if (yajl_gen_string(g, (u_char *)"debug", strlen("debug")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ default:
+ if (yajl_gen_string(g, (u_char *)"unknown",
+ strlen("unknown")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ }
+
+ if (yajl_gen_string(g, (u_char *)"@timestamp", strlen("@timestamp")) !=
+ yajl_gen_status_ok)
+ goto err;
+
+ tt = CDTIME_T_TO_TIME_T (timestamp_time);
+ gmtime_r (&tt, ×tamp_tm);
+
+ /*
+ * format time as a UTC ISO 8601 compliant string
+ */
+ strftime (timestamp_str, sizeof (timestamp_str),
+ "%Y-%m-%d %H:%M:%SZ", ×tamp_tm);
+ timestamp_str[sizeof (timestamp_str) - 1] = '\0';
+
+ if (yajl_gen_string(g, (u_char *)timestamp_str,
+ strlen(timestamp_str)) !=
+ yajl_gen_status_ok)
+ goto err;
+
+ if (yajl_gen_map_close(g) != yajl_gen_status_ok)
+ goto err;
+
+ if (yajl_gen_get_buf(g, &buf, &len) != yajl_gen_status_ok)
+ goto err;
+ pthread_mutex_lock (&file_lock);
+
+ if (log_file == NULL)
+ {
+ fh = fopen (DEFAULT_LOGFILE, "a");
+ do_close = 1;
+ } else if (strcasecmp(log_file, "stdout") == 0) {
+ fh = stdout;
+ do_close = 0;
+ } else if (strcasecmp(log_file, "stderr") == 0) {
+ fh = stderr;
+ do_close = 0;
+ } else {
+ fh = fopen (log_file, "a");
+ do_close = 1;
+ }
+
+ if (fh == NULL)
+ {
+ char errbuf[1024];
+ fprintf (stderr, "log_logstash plugin: fopen (%s) failed: %s\n",
+ (log_file == NULL) ? DEFAULT_LOGFILE : log_file,
+ sstrerror (errno, errbuf, sizeof (errbuf)));
+ }
+ else
+ {
+ fprintf(fh, "%s\n", buf);
+ if (do_close) {
+ fclose (fh);
+ } else {
+ fflush(fh);
+ }
+ }
+ pthread_mutex_unlock (&file_lock);
+ yajl_gen_free(g);
+ return;
+
+ err:
+ yajl_gen_free(g);
+ fprintf(stderr, "Could not correctly generate JSON message\n");
+ return;
+} /* void log_logstash_print */
+
+static void log_logstash_log (int severity, const char *msg,
+ user_data_t __attribute__((unused)) *user_data)
+{
+ yajl_gen g;
+#if !defined(HAVE_YAJL_V2)
+ yajl_gen_config conf;
+
+ conf.beautify = 0;
+#endif
+
+ if (severity > log_level)
+ return;
+
+#if HAVE_YAJL_V2
+ g = yajl_gen_alloc(NULL);
+#else
+ g = yajl_gen_alloc(&conf, NULL);
+#endif
+
+ if (g == NULL) {
+ fprintf(stderr, "Could not allocate JSON generator.\n");
+ return;
+ }
+
+ if (yajl_gen_map_open(g) != yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_string(g, (u_char *)"@message", strlen("@message")) !=
+ yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_string(g, (u_char *)msg, strlen(msg)) !=
+ yajl_gen_status_ok)
+ goto err;
+
+ log_logstash_print (g, severity, cdtime ());
+ return;
+ err:
+ yajl_gen_free(g);
+ fprintf(stderr, "Could not generate JSON message preamble\n");
+ return;
+
+} /* void log_logstash_log (int, const char *) */
+
+static int log_logstash_notification (const notification_t *n,
+ user_data_t __attribute__((unused)) *user_data)
+{
+ yajl_gen g;
+#if HAVE_YAJL_V2
+ g = yajl_gen_alloc(NULL);
+#else
+ yajl_gen_config conf;
+
+ conf.beautify = 0;
+ g = yajl_gen_alloc(&conf, NULL);
+#endif
+
+ if (g == NULL) {
+ fprintf(stderr, "Could not allocate JSON generator.\n");
+ return (0);
+ }
+
+ if (yajl_gen_map_open(g) != yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_string(g, (u_char *)"@message", strlen("@message")) !=
+ yajl_gen_status_ok)
+ goto err;
+ if (strlen(n->message) > 0) {
+ if (yajl_gen_string(g, (u_char *)n->message,
+ strlen(n->message)) !=
+ yajl_gen_status_ok)
+ goto err;
+ } else {
+ if (yajl_gen_string(g, (u_char *)"notification without a message",
+ strlen("notification without a message")) !=
+ yajl_gen_status_ok)
+ goto err;
+ }
+
+
+ if (yajl_gen_string(g, (u_char *)"@fields", strlen("@fields")) !=
+ yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_map_open(g) !=
+ yajl_gen_status_ok)
+ goto err;
+
+ if (strlen(n->host) > 0) {
+ if (yajl_gen_string(g, (u_char *)"host", strlen("host")) !=
+ yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_string(g, (u_char *)n->host, strlen(n->host)) !=
+ yajl_gen_status_ok)
+ goto err;
+
+ }
+ if (strlen(n->plugin) > 0) {
+ if (yajl_gen_string(g, (u_char *)"plugin", strlen("plugin")) !=
+ yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_string(g, (u_char *)n->plugin, strlen(n->plugin)) !=
+ yajl_gen_status_ok)
+ goto err;
+ }
+ if (strlen(n->plugin_instance) > 0) {
+ if (yajl_gen_string(g, (u_char *)"plugin_instance",
+ strlen("plugin_instance")) !=
+ yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_string(g, (u_char *)n->plugin_instance,
+ strlen(n->plugin_instance)) !=
+ yajl_gen_status_ok)
+ goto err;
+ }
+ if (strlen(n->type) > 0) {
+ if (yajl_gen_string(g, (u_char *)"type", strlen("type")) !=
+ yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_string(g, (u_char *)n->type, strlen(n->type)) !=
+ yajl_gen_status_ok)
+ goto err;
+ }
+ if (strlen(n->type_instance) > 0) {
+ if (yajl_gen_string(g, (u_char *)"type_instance",
+ strlen("type_instance")) !=
+ yajl_gen_status_ok)
+ goto err;
+ if (yajl_gen_string(g, (u_char *)n->type_instance,
+ strlen(n->type_instance)) !=
+ yajl_gen_status_ok)
+ goto err;
+ }
+
+ if (yajl_gen_string(g, (u_char *)"severity",
+ strlen("severity")) !=
+ yajl_gen_status_ok)
+ goto err;
+
+ switch (n->severity) {
+ case NOTIF_FAILURE:
+ if (yajl_gen_string(g, (u_char *)"failure",
+ strlen("failure")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ case NOTIF_WARNING:
+ if (yajl_gen_string(g, (u_char *)"warning",
+ strlen("warning")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ case NOTIF_OKAY:
+ if (yajl_gen_string(g, (u_char *)"ok",
+ strlen("ok")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ default:
+ if (yajl_gen_string(g, (u_char *)"unknown",
+ strlen("unknown")) !=
+ yajl_gen_status_ok)
+ goto err;
+ break;
+ }
+ if (yajl_gen_map_close(g) != yajl_gen_status_ok)
+ goto err;
+
+ log_logstash_print (g, LOG_INFO, (n->time != 0) ? n->time : cdtime ());
+ return (0);
+
+ err:
+ yajl_gen_free(g);
+ fprintf(stderr, "Could not correctly generate JSON notification\n");
+ return (0);
+} /* int log_logstash_notification */
+
+void module_register (void)
+{
+ plugin_register_config ("log_logstash",
+ log_logstash_config,
+ config_keys,
+ config_keys_num);
+ plugin_register_log ("log_logstash",
+ log_logstash_log,
+ /* user_data = */ NULL);
+ plugin_register_notification ("log_logstash",
+ log_logstash_notification,
+ /* user_data = */ NULL);
+} /* void module_register (void) */
+
+/* vim: set sw=4 ts=4 tw=78 noexpandtab : */
{
if (0 == strcasecmp (key, "LogLevel")) {
log_level = parse_log_severity(value);
- if (log_level == -1) return 1; /* to keep previous behaviour */
+ if (log_level < 0) {
+ log_level = LOG_INFO;
+ ERROR ("logfile: invalid loglevel [%s] defaulting to 'info'", value);
+ return (1);
+ }
}
else if (0 == strcasecmp (key, "File")) {
sfree (log_file);
/*
* For an explanation on these fields please refer to
- * <http://code.sixapart.com/svn/memcached/trunk/server/doc/protocol.txt>
+ * <https://github.com/memcached/memcached/blob/master/doc/protocol.txt>
*/
/*
{
submit_gauge ("memcached_connections", "current", atof (fields[2]), st);
}
+ else if (FIELD_IS ("listen_disabled_num"))
+ {
+ submit_derive ("memcached_connections", "listen_disabled", atof (fields[2]), st);
+ }
/*
* Commands
{
log_level = parse_log_severity (value);
if (log_level < 0)
+ {
+ log_level = LOG_INFO;
+ ERROR ("syslog: invalid loglevel [%s] defaulting to 'info'", value);
return (1);
+ }
}
else if (strcasecmp (key, "NotifyLevel") == 0)
{
#include "plugin.h"
#include "utils_avltree.h"
#include "utils_cache.h"
+#include "utils_threshold.h"
#include <assert.h>
#include <pthread.h>
/*
- * Private data structures
- * {{{ */
-#define UT_FLAG_INVERT 0x01
-#define UT_FLAG_PERSIST 0x02
-#define UT_FLAG_PERCENTAGE 0x04
-#define UT_FLAG_INTERESTING 0x08
-#define UT_FLAG_PERSIST_OK 0x10
-typedef struct threshold_s
-{
- char host[DATA_MAX_NAME_LEN];
- char plugin[DATA_MAX_NAME_LEN];
- char plugin_instance[DATA_MAX_NAME_LEN];
- char type[DATA_MAX_NAME_LEN];
- char type_instance[DATA_MAX_NAME_LEN];
- char data_source[DATA_MAX_NAME_LEN];
- gauge_t warning_min;
- gauge_t warning_max;
- gauge_t failure_min;
- gauge_t failure_max;
- gauge_t hysteresis;
- unsigned int flags;
- int hits;
- struct threshold_s *next;
-} threshold_t;
-/* }}} */
-
-/*
- * Private (static) variables
- * {{{ */
-static c_avl_tree_t *threshold_tree = NULL;
-static pthread_mutex_t threshold_lock = PTHREAD_MUTEX_INITIALIZER;
-/* }}} */
-
-/*
* Threshold management
* ====================
* The following functions add, delete, search, etc. configured thresholds to
* the underlying AVL trees.
*/
-/*
- * threshold_t *threshold_get
- *
- * Retrieve one specific threshold configuration. For looking up a threshold
- * matching a value_list_t, see "threshold_search" below. Returns NULL if the
- * specified threshold doesn't exist.
- */
-static threshold_t *threshold_get (const char *hostname,
- const char *plugin, const char *plugin_instance,
- const char *type, const char *type_instance)
-{ /* {{{ */
- char name[6 * DATA_MAX_NAME_LEN];
- threshold_t *th = NULL;
-
- format_name (name, sizeof (name),
- (hostname == NULL) ? "" : hostname,
- (plugin == NULL) ? "" : plugin, plugin_instance,
- (type == NULL) ? "" : type, type_instance);
- name[sizeof (name) - 1] = '\0';
-
- if (c_avl_get (threshold_tree, name, (void *) &th) == 0)
- return (th);
- else
- return (NULL);
-} /* }}} threshold_t *threshold_get */
/*
* int ut_threshold_add
return (status);
} /* }}} int ut_threshold_add */
-/*
- * threshold_t *threshold_search
- *
- * Searches for a threshold configuration using all the possible variations of
- * "Host", "Plugin" and "Type" blocks. Returns NULL if no threshold could be
- * found.
- * XXX: This is likely the least efficient function in collectd.
- */
-static threshold_t *threshold_search (const value_list_t *vl)
-{ /* {{{ */
- threshold_t *th;
-
- if ((th = threshold_get (vl->host, vl->plugin, vl->plugin_instance,
- vl->type, vl->type_instance)) != NULL)
- return (th);
- else if ((th = threshold_get (vl->host, vl->plugin, vl->plugin_instance,
- vl->type, NULL)) != NULL)
- return (th);
- else if ((th = threshold_get (vl->host, vl->plugin, NULL,
- vl->type, vl->type_instance)) != NULL)
- return (th);
- else if ((th = threshold_get (vl->host, vl->plugin, NULL,
- vl->type, NULL)) != NULL)
- return (th);
- else if ((th = threshold_get (vl->host, "", NULL,
- vl->type, vl->type_instance)) != NULL)
- return (th);
- else if ((th = threshold_get (vl->host, "", NULL,
- vl->type, NULL)) != NULL)
- return (th);
- else if ((th = threshold_get ("", vl->plugin, vl->plugin_instance,
- vl->type, vl->type_instance)) != NULL)
- return (th);
- else if ((th = threshold_get ("", vl->plugin, vl->plugin_instance,
- vl->type, NULL)) != NULL)
- return (th);
- else if ((th = threshold_get ("", vl->plugin, NULL,
- vl->type, vl->type_instance)) != NULL)
- return (th);
- else if ((th = threshold_get ("", vl->plugin, NULL,
- vl->type, NULL)) != NULL)
- return (th);
- else if ((th = threshold_get ("", "", NULL,
- vl->type, vl->type_instance)) != NULL)
- return (th);
- else if ((th = threshold_get ("", "", NULL,
- vl->type, NULL)) != NULL)
- return (th);
-
- return (NULL);
-} /* }}} threshold_t *threshold_search */
-
/*
* Configuration
* =============
*
* Gets a list of matching thresholds and searches for the worst status by one
* of the thresholds. Then reports that status using the ut_report_state
- * function above.
+ * function above.
* Returns zero on success and if no threshold has been configured. Returns
* less than zero on failure.
*/
cdtime_t missing_time;
char identifier[6 * DATA_MAX_NAME_LEN];
notification_t n;
+ cdtime_t now;
if (threshold_tree == NULL)
return (0);
if ((th == NULL) || ((th->flags & UT_FLAG_INTERESTING) == 0))
return (0);
- missing_time = cdtime () - vl->time;
+ now = cdtime ();
+ missing_time = now - vl->time;
FORMAT_VL (identifier, sizeof (identifier), vl);
NOTIFICATION_INIT_VL (&n, vl);
ssnprintf (n.message, sizeof (n.message),
"%s has not been updated for %.3f seconds.",
identifier, CDTIME_T_TO_DOUBLE (missing_time));
+ n.time = now;
plugin_dispatch_notification (&n);
th.hits = 0;
th.hysteresis = 0;
th.flags = UT_FLAG_INTERESTING; /* interesting by default */
-
+
for (i = 0; i < ci->children_num; i++)
{
oconfig_item_t *option = ci->children + i;
#include "utils_cmd_flush.h"
#include "utils_cmd_getval.h"
+#include "utils_cmd_getthreshold.h"
#include "utils_cmd_listval.h"
#include "utils_cmd_putval.h"
#include "utils_cmd_putnotif.h"
{
handle_getval (fhout, buffer);
}
+ else if (strcasecmp (fields[0], "getthreshold") == 0)
+ {
+ handle_getthreshold (fhout, buffer);
+ }
else if (strcasecmp (fields[0], "putval") == 0)
{
handle_putval (fhout, buffer);
}
else
{
- plugin_flush (NULL, timeout, NULL);
+ plugin_flush (NULL, DOUBLE_TO_CDTIME_T (timeout), NULL);
print_to_socket (fh, "0 Done\n");
}
#include "common.h"
#include "plugin.h"
+#include "utils_avltree.h"
#include "utils_threshold.h"
#include "utils_parse_option.h" /* for `parse_string' */
#include "utils_cmd_getthreshold.h"
static int set_option_time (notification_t *n, const char *value)
{
- time_t tmp;
-
- tmp = (time_t) atoi (value);
- if (tmp <= 0)
+ char *endptr = NULL;
+ double tmp;
+
+ errno = 0;
+ tmp = strtod (value, &endptr);
+ if ((errno != 0) /* Overflow */
+ || (endptr == value) /* Invalid string */
+ || (endptr == NULL) /* This should not happen */
+ || (*endptr != 0)) /* Trailing chars */
return (-1);
- n->time = tmp;
+ n->time = DOUBLE_TO_CDTIME_T (tmp);
return (0);
} /* int set_option_time */
--- /dev/null
+/*
+ * COPYRIGHT (C) 1986 Gary S. Brown. You may use this program, or
+ * code or tables extracted from it, as desired without restriction.
+ *
+ * First, the polynomial itself and its table of feedback terms. The
+ * polynomial is
+ * X^32+X^26+X^23+X^22+X^16+X^12+X^11+X^10+X^8+X^7+X^5+X^4+X^2+X^1+X^0
+ *
+ * Note that we take it "backwards" and put the highest-order term in
+ * the lowest-order bit. The X^32 term is "implied"; the LSB is the
+ * X^31 term, etc. The X^0 term (usually shown as "+1") results in
+ * the MSB being 1
+ *
+ * Note that the usual hardware shift register implementation, which
+ * is what we're using (we're merely optimizing it by doing eight-bit
+ * chunks at a time) shifts bits into the lowest-order term. In our
+ * implementation, that means shifting towards the right. Why do we
+ * do it this way? Because the calculated CRC must be transmitted in
+ * order from highest-order term to lowest-order term. UARTs transmit
+ * characters in order from LSB to MSB. By storing the CRC this way
+ * we hand it to the UART in the order low-byte to high-byte; the UART
+ * sends each low-bit to hight-bit; and the result is transmission bit
+ * by bit from highest- to lowest-order term without requiring any bit
+ * shuffling on our part. Reception works similarly
+ *
+ * The feedback terms table consists of 256, 32-bit entries. Notes
+ *
+ * The table can be generated at runtime if desired; code to do so
+ * is shown later. It might not be obvious, but the feedback
+ * terms simply represent the results of eight shift/xor opera
+ * tions for all combinations of data and CRC register values
+ *
+ * The values must be right-shifted by eight bits by the "updcrc
+ * logic; the shift must be unsigned (bring in zeroes). On some
+ * hardware you could probably optimize the shift in assembler by
+ * using byte-swap instructions
+ * polynomial $edb88320
+ */
+
+#include <sys/types.h>
+
+u_int32_t crc32_buffer(const u_char *, size_t);
+static unsigned int crc32_tab[] = {
+ 0x00000000L, 0x77073096L, 0xee0e612cL, 0x990951baL, 0x076dc419L,
+ 0x706af48fL, 0xe963a535L, 0x9e6495a3L, 0x0edb8832L, 0x79dcb8a4L,
+ 0xe0d5e91eL, 0x97d2d988L, 0x09b64c2bL, 0x7eb17cbdL, 0xe7b82d07L,
+ 0x90bf1d91L, 0x1db71064L, 0x6ab020f2L, 0xf3b97148L, 0x84be41deL,
+ 0x1adad47dL, 0x6ddde4ebL, 0xf4d4b551L, 0x83d385c7L, 0x136c9856L,
+ 0x646ba8c0L, 0xfd62f97aL, 0x8a65c9ecL, 0x14015c4fL, 0x63066cd9L,
+ 0xfa0f3d63L, 0x8d080df5L, 0x3b6e20c8L, 0x4c69105eL, 0xd56041e4L,
+ 0xa2677172L, 0x3c03e4d1L, 0x4b04d447L, 0xd20d85fdL, 0xa50ab56bL,
+ 0x35b5a8faL, 0x42b2986cL, 0xdbbbc9d6L, 0xacbcf940L, 0x32d86ce3L,
+ 0x45df5c75L, 0xdcd60dcfL, 0xabd13d59L, 0x26d930acL, 0x51de003aL,
+ 0xc8d75180L, 0xbfd06116L, 0x21b4f4b5L, 0x56b3c423L, 0xcfba9599L,
+ 0xb8bda50fL, 0x2802b89eL, 0x5f058808L, 0xc60cd9b2L, 0xb10be924L,
+ 0x2f6f7c87L, 0x58684c11L, 0xc1611dabL, 0xb6662d3dL, 0x76dc4190L,
+ 0x01db7106L, 0x98d220bcL, 0xefd5102aL, 0x71b18589L, 0x06b6b51fL,
+ 0x9fbfe4a5L, 0xe8b8d433L, 0x7807c9a2L, 0x0f00f934L, 0x9609a88eL,
+ 0xe10e9818L, 0x7f6a0dbbL, 0x086d3d2dL, 0x91646c97L, 0xe6635c01L,
+ 0x6b6b51f4L, 0x1c6c6162L, 0x856530d8L, 0xf262004eL, 0x6c0695edL,
+ 0x1b01a57bL, 0x8208f4c1L, 0xf50fc457L, 0x65b0d9c6L, 0x12b7e950L,
+ 0x8bbeb8eaL, 0xfcb9887cL, 0x62dd1ddfL, 0x15da2d49L, 0x8cd37cf3L,
+ 0xfbd44c65L, 0x4db26158L, 0x3ab551ceL, 0xa3bc0074L, 0xd4bb30e2L,
+ 0x4adfa541L, 0x3dd895d7L, 0xa4d1c46dL, 0xd3d6f4fbL, 0x4369e96aL,
+ 0x346ed9fcL, 0xad678846L, 0xda60b8d0L, 0x44042d73L, 0x33031de5L,
+ 0xaa0a4c5fL, 0xdd0d7cc9L, 0x5005713cL, 0x270241aaL, 0xbe0b1010L,
+ 0xc90c2086L, 0x5768b525L, 0x206f85b3L, 0xb966d409L, 0xce61e49fL,
+ 0x5edef90eL, 0x29d9c998L, 0xb0d09822L, 0xc7d7a8b4L, 0x59b33d17L,
+ 0x2eb40d81L, 0xb7bd5c3bL, 0xc0ba6cadL, 0xedb88320L, 0x9abfb3b6L,
+ 0x03b6e20cL, 0x74b1d29aL, 0xead54739L, 0x9dd277afL, 0x04db2615L,
+ 0x73dc1683L, 0xe3630b12L, 0x94643b84L, 0x0d6d6a3eL, 0x7a6a5aa8L,
+ 0xe40ecf0bL, 0x9309ff9dL, 0x0a00ae27L, 0x7d079eb1L, 0xf00f9344L,
+ 0x8708a3d2L, 0x1e01f268L, 0x6906c2feL, 0xf762575dL, 0x806567cbL,
+ 0x196c3671L, 0x6e6b06e7L, 0xfed41b76L, 0x89d32be0L, 0x10da7a5aL,
+ 0x67dd4accL, 0xf9b9df6fL, 0x8ebeeff9L, 0x17b7be43L, 0x60b08ed5L,
+ 0xd6d6a3e8L, 0xa1d1937eL, 0x38d8c2c4L, 0x4fdff252L, 0xd1bb67f1L,
+ 0xa6bc5767L, 0x3fb506ddL, 0x48b2364bL, 0xd80d2bdaL, 0xaf0a1b4cL,
+ 0x36034af6L, 0x41047a60L, 0xdf60efc3L, 0xa867df55L, 0x316e8eefL,
+ 0x4669be79L, 0xcb61b38cL, 0xbc66831aL, 0x256fd2a0L, 0x5268e236L,
+ 0xcc0c7795L, 0xbb0b4703L, 0x220216b9L, 0x5505262fL, 0xc5ba3bbeL,
+ 0xb2bd0b28L, 0x2bb45a92L, 0x5cb36a04L, 0xc2d7ffa7L, 0xb5d0cf31L,
+ 0x2cd99e8bL, 0x5bdeae1dL, 0x9b64c2b0L, 0xec63f226L, 0x756aa39cL,
+ 0x026d930aL, 0x9c0906a9L, 0xeb0e363fL, 0x72076785L, 0x05005713L,
+ 0x95bf4a82L, 0xe2b87a14L, 0x7bb12baeL, 0x0cb61b38L, 0x92d28e9bL,
+ 0xe5d5be0dL, 0x7cdcefb7L, 0x0bdbdf21L, 0x86d3d2d4L, 0xf1d4e242L,
+ 0x68ddb3f8L, 0x1fda836eL, 0x81be16cdL, 0xf6b9265bL, 0x6fb077e1L,
+ 0x18b74777L, 0x88085ae6L, 0xff0f6a70L, 0x66063bcaL, 0x11010b5cL,
+ 0x8f659effL, 0xf862ae69L, 0x616bffd3L, 0x166ccf45L, 0xa00ae278L,
+ 0xd70dd2eeL, 0x4e048354L, 0x3903b3c2L, 0xa7672661L, 0xd06016f7L,
+ 0x4969474dL, 0x3e6e77dbL, 0xaed16a4aL, 0xd9d65adcL, 0x40df0b66L,
+ 0x37d83bf0L, 0xa9bcae53L, 0xdebb9ec5L, 0x47b2cf7fL, 0x30b5ffe9L,
+ 0xbdbdf21cL, 0xcabac28aL, 0x53b39330L, 0x24b4a3a6L, 0xbad03605L,
+ 0xcdd70693L, 0x54de5729L, 0x23d967bfL, 0xb3667a2eL, 0xc4614ab8L,
+ 0x5d681b02L, 0x2a6f2b94L, 0xb40bbe37L, 0xc30c8ea1L, 0x5a05df1bL,
+ 0x2d02ef8dL
+};
+
+/* Return a 32-bit CRC of the contents of the buffer. */
+
+u_int32_t
+crc32_buffer(const u_char *s, size_t len)
+{
+ size_t i;
+ u_int32_t ret;
+
+ ret = 0;
+ for (i = 0; i < len; i++)
+ ret = crc32_tab[(ret ^ s[i]) & 0xff] ^ (ret >> 8);
+ return ret;
+}
--- /dev/null
+/**
+ * collectd - src/utils_crc32.h
+ *
+ * Copyright (C) 2014 Pierre-Yves Ritschard
+ *
+ * Permission to use, copy, modify, and distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF MIND, USE, DATA OR PROFITS, WHETHER
+ * IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING
+ * OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ *
+ * Authors:
+ * Pierre-Yves Ritschard <pyr at spootnik.org>
+ */
+
+#ifndef UTILS_CRC32_H
+#define UTILS_CRC32_H 1
+
+u_int32_t crc32_buffer(const u_char *, size_t);
+
+#endif
#include "utils_cache.h"
#include "utils_parse_option.h"
-#define GRAPHITE_FORBIDDEN " \t\"\\:!/\n\r"
+#define GRAPHITE_FORBIDDEN " \t\"\\:!/()\n\r"
/* Utils functions to format data sets in graphite format.
* Largely taken from write_graphite.c as it remains the same formatting */
--- /dev/null
+/**
+ * collectd - src/utils_threshold.c
+ * Copyright (C) 2014 Pierre-Yves Ritschard
+ *
+ * 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:
+ * Pierre-Yves Ritschard <pyr at spootnik.org>
+ **/
+
+#include "collectd.h"
+#include "common.h"
+#include "utils_avltree.h"
+#include "utils_threshold.h"
+
+#include <pthread.h>
+
+/*
+ * Exported symbols
+ * {{{ */
+c_avl_tree_t *threshold_tree = NULL;
+pthread_mutex_t threshold_lock = PTHREAD_MUTEX_INITIALIZER;
+/* }}} */
+
+/*
+ * threshold_t *threshold_get
+ *
+ * Retrieve one specific threshold configuration. For looking up a threshold
+ * matching a value_list_t, see "threshold_search" below. Returns NULL if the
+ * specified threshold doesn't exist.
+ */
+threshold_t *threshold_get (const char *hostname,
+ const char *plugin, const char *plugin_instance,
+ const char *type, const char *type_instance)
+{ /* {{{ */
+ char name[6 * DATA_MAX_NAME_LEN];
+ threshold_t *th = NULL;
+
+ format_name (name, sizeof (name),
+ (hostname == NULL) ? "" : hostname,
+ (plugin == NULL) ? "" : plugin, plugin_instance,
+ (type == NULL) ? "" : type, type_instance);
+ name[sizeof (name) - 1] = '\0';
+
+ if (c_avl_get (threshold_tree, name, (void *) &th) == 0)
+ return (th);
+ else
+ return (NULL);
+} /* }}} threshold_t *threshold_get */
+
+/*
+ * threshold_t *threshold_search
+ *
+ * Searches for a threshold configuration using all the possible variations of
+ * "Host", "Plugin" and "Type" blocks. Returns NULL if no threshold could be
+ * found.
+ * XXX: This is likely the least efficient function in collectd.
+ */
+threshold_t *threshold_search (const value_list_t *vl)
+{ /* {{{ */
+ threshold_t *th;
+
+ if ((th = threshold_get (vl->host, vl->plugin, vl->plugin_instance,
+ vl->type, vl->type_instance)) != NULL)
+ return (th);
+ else if ((th = threshold_get (vl->host, vl->plugin, vl->plugin_instance,
+ vl->type, NULL)) != NULL)
+ return (th);
+ else if ((th = threshold_get (vl->host, vl->plugin, NULL,
+ vl->type, vl->type_instance)) != NULL)
+ return (th);
+ else if ((th = threshold_get (vl->host, vl->plugin, NULL,
+ vl->type, NULL)) != NULL)
+ return (th);
+ else if ((th = threshold_get (vl->host, "", NULL,
+ vl->type, vl->type_instance)) != NULL)
+ return (th);
+ else if ((th = threshold_get (vl->host, "", NULL,
+ vl->type, NULL)) != NULL)
+ return (th);
+ else if ((th = threshold_get ("", vl->plugin, vl->plugin_instance,
+ vl->type, vl->type_instance)) != NULL)
+ return (th);
+ else if ((th = threshold_get ("", vl->plugin, vl->plugin_instance,
+ vl->type, NULL)) != NULL)
+ return (th);
+ else if ((th = threshold_get ("", vl->plugin, NULL,
+ vl->type, vl->type_instance)) != NULL)
+ return (th);
+ else if ((th = threshold_get ("", vl->plugin, NULL,
+ vl->type, NULL)) != NULL)
+ return (th);
+ else if ((th = threshold_get ("", "", NULL,
+ vl->type, vl->type_instance)) != NULL)
+ return (th);
+ else if ((th = threshold_get ("", "", NULL,
+ vl->type, NULL)) != NULL)
+ return (th);
+
+ return (NULL);
+} /* }}} threshold_t *threshold_search */
+
+int ut_search_threshold (const value_list_t *vl, /* {{{ */
+ threshold_t *ret_threshold)
+{
+ threshold_t *t;
+
+ if (vl == NULL)
+ return (EINVAL);
+
+ /* Is this lock really necessary? */
+ pthread_mutex_lock (&threshold_lock);
+ t = threshold_search (vl);
+ if (t == NULL) {
+ pthread_mutex_unlock (&threshold_lock);
+ return (ENOENT);
+ }
+
+ memcpy (ret_threshold, t, sizeof (*ret_threshold));
+ pthread_mutex_unlock (&threshold_lock);
+
+ ret_threshold->next = NULL;
+
+ return (0);
+} /* }}} int ut_search_threshold */
+
+
--- /dev/null
+/**
+ * collectd - src/utils_threshold.h
+ * Copyright (C) 2014 Pierre-Yves Ritschard
+ *
+ * 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:
+ * Pierre-Yves Ritschard <pyr at spootnik.org>
+ **/
+
+#ifndef UTILS_THRESHOLD_H
+#define UTILS_THRESHOLD_H 1
+
+#define UT_FLAG_INVERT 0x01
+#define UT_FLAG_PERSIST 0x02
+#define UT_FLAG_PERCENTAGE 0x04
+#define UT_FLAG_INTERESTING 0x08
+#define UT_FLAG_PERSIST_OK 0x10
+typedef struct threshold_s
+{
+ char host[DATA_MAX_NAME_LEN];
+ char plugin[DATA_MAX_NAME_LEN];
+ char plugin_instance[DATA_MAX_NAME_LEN];
+ char type[DATA_MAX_NAME_LEN];
+ char type_instance[DATA_MAX_NAME_LEN];
+ char data_source[DATA_MAX_NAME_LEN];
+ gauge_t warning_min;
+ gauge_t warning_max;
+ gauge_t failure_min;
+ gauge_t failure_max;
+ gauge_t hysteresis;
+ unsigned int flags;
+ int hits;
+ struct threshold_s *next;
+} threshold_t;
+
+extern c_avl_tree_t *threshold_tree;
+extern pthread_mutex_t threshold_lock;
+
+threshold_t *threshold_get (const char *hostname,
+ const char *plugin, const char *plugin_instance,
+ const char *type, const char *type_instance);
+
+threshold_t *threshold_search (const value_list_t *vl);
+
+int ut_search_threshold (const value_list_t *vl,
+ threshold_t *ret_threshold);
+
+#endif /* UTILS_THRESHOLD_H */
+
+/* vim: set sw=2 sts=2 ts=8 : */
_Bool verify_peer;
_Bool verify_host;
char *cacert;
+ char *capath;
+ char *clientkey;
+ char *clientcert;
+ char *clientkeypass;
+ long sslversion;
_Bool store_rates;
#define WH_FORMAT_COMMAND 0
curl_easy_setopt (cb->curl, CURLOPT_SSL_VERIFYPEER, (long) cb->verify_peer);
curl_easy_setopt (cb->curl, CURLOPT_SSL_VERIFYHOST,
cb->verify_host ? 2L : 0L);
+ curl_easy_setopt (cb->curl, CURLOPT_SSLVERSION, cb->sslversion);
if (cb->cacert != NULL)
curl_easy_setopt (cb->curl, CURLOPT_CAINFO, cb->cacert);
+ if (cb->capath != NULL)
+ curl_easy_setopt (cb->curl, CURLOPT_CAPATH, cb->capath);
+
+ if (cb->clientkey != NULL && cb->clientcert != NULL)
+ {
+ curl_easy_setopt (cb->curl, CURLOPT_SSLKEY, cb->clientkey);
+ curl_easy_setopt (cb->curl, CURLOPT_SSLCERT, cb->clientcert);
+
+ if (cb->clientkeypass != NULL)
+ curl_easy_setopt (cb->curl, CURLOPT_SSLKEYPASSWD, cb->clientkeypass);
+ }
wh_reset_buffer (cb);
sfree (cb->pass);
sfree (cb->credentials);
sfree (cb->cacert);
+ sfree (cb->capath);
+ sfree (cb->clientkey);
+ sfree (cb->clientcert);
+ sfree (cb->clientkeypass);
sfree (cb);
} /* }}} void wh_callback_free */
return (-1);
}
memset (cb, 0, sizeof (*cb));
- cb->location = NULL;
- cb->user = NULL;
- cb->pass = NULL;
- cb->credentials = NULL;
cb->verify_peer = 1;
cb->verify_host = 1;
- cb->cacert = NULL;
cb->format = WH_FORMAT_COMMAND;
- cb->curl = NULL;
+ cb->sslversion = CURL_SSLVERSION_DEFAULT;
pthread_mutex_init (&cb->send_lock, /* attr = */ NULL);
cf_util_get_boolean (child, &cb->verify_host);
else if (strcasecmp ("CACert", child->key) == 0)
cf_util_get_string (child, &cb->cacert);
+ else if (strcasecmp ("CAPath", child->key) == 0)
+ cf_util_get_string (child, &cb->capath);
+ else if (strcasecmp ("ClientKey", child->key) == 0)
+ cf_util_get_string (child, &cb->clientkey);
+ else if (strcasecmp ("ClientCert", child->key) == 0)
+ cf_util_get_string (child, &cb->clientcert);
+ else if (strcasecmp ("ClientKeyPass", child->key) == 0)
+ cf_util_get_string (child, &cb->clientkeypass);
+ else if (strcasecmp ("SSLVersion", child->key) == 0)
+ {
+ char *value = NULL;
+
+ cf_util_get_string (child, &value);
+
+ if (value == NULL || strcasecmp ("default", value) == 0)
+ cb->sslversion = CURL_SSLVERSION_DEFAULT;
+ else if (strcasecmp ("SSLv2", value) == 0)
+ cb->sslversion = CURL_SSLVERSION_SSLv2;
+ else if (strcasecmp ("SSLv3", value) == 0)
+ cb->sslversion = CURL_SSLVERSION_SSLv3;
+ else if (strcasecmp ("TLSv1", value) == 0)
+ cb->sslversion = CURL_SSLVERSION_TLSv1;
+#if (LIBCURL_VERSION_MAJOR > 7) || (LIBCURL_VERSION_MAJOR == 7 && LIBCURL_VERSION_MINOR >= 34)
+ else if (strcasecmp ("TLSv1_0", value) == 0)
+ cb->sslversion = CURL_SSLVERSION_TLSv1_0;
+ else if (strcasecmp ("TLSv1_1", value) == 0)
+ cb->sslversion = CURL_SSLVERSION_TLSv1_1;
+ else if (strcasecmp ("TLSv1_2", value) == 0)
+ cb->sslversion = CURL_SSLVERSION_TLSv1_2;
+#endif
+ else
+ ERROR ("write_http plugin: Invalid SSLVersion "
+ "option: %s.", value);
+
+ sfree(value);
+ }
else if (strcasecmp ("Format", child->key) == 0)
config_set_format (cb, child);
else if (strcasecmp ("StoreRates", child->key) == 0)
--- /dev/null
+/**
+ * collectd - src/write_kafka.c
+ *
+ * Copyright (C) 2014 Pierre-Yves Ritschard
+ *
+ * Permission to use, copy, modify, and distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF MIND, USE, DATA OR PROFITS, WHETHER
+ * IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING
+ * OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ *
+ * Authors:
+ * Pierre-Yves Ritschard <pyr at spootnik.org>
+ */
+
+#include "collectd.h"
+#include "plugin.h"
+#include "common.h"
+#include "configfile.h"
+#include "utils_cache.h"
+#include "utils_cmd_putval.h"
+#include "utils_format_graphite.h"
+#include "utils_format_json.h"
+#include "utils_crc32.h"
+
+#include <sys/types.h>
+#include <librdkafka/rdkafka.h>
+#include <pthread.h>
+#include <zlib.h>
+
+struct kafka_topic_context {
+#define KAFKA_FORMAT_COMMAND 1
+#define KAFKA_FORMAT_GRAPHITE 2
+#define KAFKA_FORMAT_JSON 3
+ u_int8_t format;
+ unsigned int graphite_flags;
+ _Bool store_rates;
+ rd_kafka_topic_conf_t *conf;
+ rd_kafka_topic_t *topic;
+ rd_kafka_t *kafka;
+ int has_key;
+ u_int32_t key;
+ char *prefix;
+ char *postfix;
+ char escape_char;
+ char *topic_name;
+};
+
+static int kafka_write(const data_set_t *, const value_list_t *, user_data_t *);
+static int32_t kafka_partition(const rd_kafka_topic_t *, const void *, size_t,
+ int32_t, void *, void *);
+static void kafka_log(const rd_kafka_t *, int, const char *, const char *);
+
+static void kafka_log(const rd_kafka_t *rkt, int level,
+ const char *fac, const char *msg)
+{
+ plugin_log(level, "%s", msg);
+}
+
+static int32_t kafka_partition(const rd_kafka_topic_t *rkt,
+ const void *keydata, size_t keylen,
+ int32_t partition_cnt, void *p, void *m)
+{
+ u_int32_t key = *((u_int32_t *)keydata );
+
+ return key % partition_cnt;
+}
+
+static int kafka_write(const data_set_t *ds, /* {{{ */
+ const value_list_t *vl,
+ user_data_t *ud)
+{
+ int status = 0;
+ u_int32_t key;
+ char buffer[8192];
+ size_t bfree = sizeof(buffer);
+ size_t bfill = 0;
+ size_t blen = 0;
+ struct kafka_topic_context *ctx = ud->data;
+
+ if ((ds == NULL) || (vl == NULL) || (ctx == NULL))
+ return EINVAL;
+
+ bzero(buffer, sizeof(buffer));
+
+ switch (ctx->format) {
+ case KAFKA_FORMAT_COMMAND:
+ status = create_putval(buffer, sizeof(buffer), ds, vl);
+ if (status != 0) {
+ ERROR("write_kafka plugin: create_putval failed with status %i.",
+ status);
+ return status;
+ }
+ blen = strlen(buffer);
+ break;
+ case KAFKA_FORMAT_JSON:
+
+ format_json_initialize(buffer, &bfill, &bfree);
+ format_json_value_list(buffer, &bfill, &bfree, ds, vl,
+ ctx->store_rates);
+ format_json_finalize(buffer, &bfill, &bfree);
+ blen = strlen(buffer);
+ break;
+ case KAFKA_FORMAT_GRAPHITE:
+ status = format_graphite(buffer, sizeof(buffer), ds, vl,
+ ctx->prefix, ctx->postfix, ctx->escape_char,
+ ctx->graphite_flags);
+ if (status != 0) {
+ ERROR("write_kafka plugin: format_graphite failed with status %i.",
+ status);
+ return status;
+ }
+ blen = strlen(buffer);
+ break;
+ default:
+ ERROR("write_kafka plugin: invalid format %i.", ctx->format);
+ return -1;
+ }
+
+ /*
+ * We partition our stream by metric name
+ */
+ if (ctx->has_key)
+ key = ctx->key;
+ else
+ key = rand();
+
+ rd_kafka_produce(ctx->topic, RD_KAFKA_PARTITION_UA,
+ RD_KAFKA_MSG_F_COPY, buffer, blen,
+ &key, sizeof(key), NULL);
+
+ return status;
+} /* }}} int kafka_write */
+
+static void kafka_topic_context_free(void *p) /* {{{ */
+{
+ struct kafka_topic_context *ctx = p;
+
+ if (ctx == NULL)
+ return;
+
+ if (ctx->topic_name != NULL)
+ sfree(ctx->topic_name);
+ if (ctx->topic != NULL)
+ rd_kafka_topic_destroy(ctx->topic);
+ if (ctx->conf != NULL)
+ rd_kafka_topic_conf_destroy(ctx->conf);
+
+ sfree(ctx);
+} /* }}} void kafka_topic_context_free */
+
+static void kafka_config_topic(rd_kafka_conf_t *conf, oconfig_item_t *ci) /* {{{ */
+{
+ int status;
+ int i;
+ struct kafka_topic_context *tctx;
+ char *key;
+ char *val;
+ char callback_name[DATA_MAX_NAME_LEN];
+ char errbuf[1024];
+ user_data_t ud;
+ oconfig_item_t *child;
+ rd_kafka_conf_res_t ret;
+
+ if ((tctx = calloc(1, sizeof (*tctx))) == NULL) {
+ ERROR ("write_kafka plugin: calloc failed.");
+ return;
+ }
+
+ tctx->escape_char = '.';
+ tctx->store_rates = 1;
+
+ rd_kafka_conf_set_log_cb(conf, kafka_log);
+ if ((tctx->kafka = rd_kafka_new(RD_KAFKA_PRODUCER, conf,
+ errbuf, sizeof(errbuf))) == NULL) {
+ sfree(tctx);
+ ERROR("write_kafka plugin: cannot create kafka handle.");
+ return;
+ }
+ conf = NULL;
+
+ if ((tctx->conf = rd_kafka_topic_conf_new()) == NULL) {
+ rd_kafka_destroy(tctx->kafka);
+ sfree(tctx);
+ ERROR ("write_kafka plugin: cannot create topic configuration.");
+ return;
+ }
+
+ if (ci->values_num != 1) {
+ WARNING("kafka topic name needed.");
+ goto errout;
+ }
+
+ if (ci->values[0].type != OCONFIG_TYPE_STRING) {
+ WARNING("kafka topic needs a string argument.");
+ goto errout;
+ }
+
+ if ((tctx->topic_name = strdup(ci->values[0].value.string)) == NULL) {
+ ERROR("write_kafka plugin: cannot copy topic name.");
+ goto errout;
+ }
+
+ for (i = 0; i < ci->children_num; i++) {
+ /*
+ * The code here could be simplified but makes room
+ * for easy adding of new options later on.
+ */
+ child = &ci->children[i];
+ status = 0;
+
+ if (strcasecmp ("Property", child->key) == 0) {
+ if (child->values_num != 2) {
+ WARNING("kafka properties need both a key and a value.");
+ goto errout;
+ }
+ if (child->values[0].type != OCONFIG_TYPE_STRING ||
+ child->values[1].type != OCONFIG_TYPE_STRING) {
+ WARNING("kafka properties needs string arguments.");
+ goto errout;
+ }
+ key = child->values[0].value.string;
+ val = child->values[0].value.string;
+ ret = rd_kafka_topic_conf_set(tctx->conf,key, val,
+ errbuf, sizeof(errbuf));
+ if (ret != RD_KAFKA_CONF_OK) {
+ WARNING("cannot set kafka topic property %s to %s: %s.",
+ key, val, errbuf);
+ goto errout;
+ }
+
+ } else if (strcasecmp ("Key", child->key) == 0) {
+ char *tmp_buf = NULL;
+ status = cf_util_get_string(child, &tmp_buf);
+ if (status != 0) {
+ WARNING("write_kafka plugin: invalid key supplied");
+ break;
+ }
+
+ if (strcasecmp(tmp_buf, "Random") != 0) {
+ tctx->has_key = 1;
+ tctx->key = crc32_buffer((u_char *)tmp_buf, strlen(tmp_buf));
+ }
+ sfree(tmp_buf);
+
+ } else if (strcasecmp ("Format", child->key) == 0) {
+ status = cf_util_get_string(child, &key);
+ if (status != 0)
+ goto errout;
+
+ assert(key != NULL);
+
+ if (strcasecmp(key, "Command") == 0) {
+
+ tctx->format = KAFKA_FORMAT_COMMAND;
+
+ } else if (strcasecmp(key, "Graphite") == 0) {
+ tctx->format = KAFKA_FORMAT_GRAPHITE;
+
+ } else if (strcasecmp(key, "Json") == 0) {
+ tctx->format = KAFKA_FORMAT_JSON;
+
+ } else {
+ WARNING ("write_kafka plugin: Invalid format string: %s",
+ key);
+ }
+ sfree(key);
+
+ } else if (strcasecmp ("StoreRates", child->key) == 0) {
+ status = cf_util_get_boolean (child, &tctx->store_rates);
+ (void) cf_util_get_flag (child, &tctx->graphite_flags,
+ GRAPHITE_STORE_RATES);
+
+ } else if (strcasecmp ("GraphiteSeparateInstances", child->key) == 0) {
+ status = cf_util_get_flag (child, &tctx->graphite_flags,
+ GRAPHITE_SEPARATE_INSTANCES);
+
+ } else if (strcasecmp ("GraphiteAlwaysAppendDS", child->key) == 0) {
+ status = cf_util_get_flag (child, &tctx->graphite_flags,
+ GRAPHITE_ALWAYS_APPEND_DS);
+
+ } else if (strcasecmp ("GraphitePrefix", child->key) == 0) {
+ status = cf_util_get_string (child, &tctx->prefix);
+ } else if (strcasecmp ("GraphitePostfix", child->key) == 0) {
+ status = cf_util_get_string (child, &tctx->postfix);
+ } else if (strcasecmp ("GraphiteEscapeChar", child->key) == 0) {
+ char *tmp_buff = NULL;
+ status = cf_util_get_string (child, &tmp_buff);
+ if (strlen (tmp_buff) > 1)
+ WARNING ("write_kafka plugin: The option \"GraphiteEscapeChar\" handles "
+ "only one character. Others will be ignored.");
+ tctx->escape_char = tmp_buff[0];
+ sfree (tmp_buff);
+ } else {
+ WARNING ("write_kafka plugin: Invalid directive: %s.", child->key);
+ }
+
+ if (status != 0)
+ break;
+ }
+
+ rd_kafka_topic_conf_set_partitioner_cb(tctx->conf, kafka_partition);
+ rd_kafka_topic_conf_set_opaque(tctx->conf, tctx);
+
+ if ((tctx->topic = rd_kafka_topic_new(tctx->kafka, tctx->topic_name,
+ tctx->conf)) == NULL) {
+ ERROR("write_kafka plugin: cannot create topic.");
+ goto errout;
+ }
+ tctx->conf = NULL;
+
+ ssnprintf(callback_name, sizeof(callback_name),
+ "write_kafka/%s", tctx->topic_name);
+
+ ud.data = tctx;
+ ud.free_func = kafka_topic_context_free;
+
+ status = plugin_register_write (callback_name, kafka_write, &ud);
+ if (status != 0) {
+ WARNING ("write_kafka plugin: plugin_register_write (\"%s\") "
+ "failed with status %i.",
+ callback_name, status);
+ goto errout;
+ }
+ return;
+ errout:
+ if (conf != NULL)
+ rd_kafka_conf_destroy(conf);
+ if (tctx->kafka != NULL)
+ rd_kafka_destroy(tctx->kafka);
+ if (tctx->topic != NULL)
+ rd_kafka_topic_destroy(tctx->topic);
+ if (tctx->topic_name != NULL)
+ free(tctx->topic_name);
+ if (tctx->conf != NULL)
+ rd_kafka_topic_conf_destroy(tctx->conf);
+ sfree(tctx);
+} /* }}} int kafka_config_topic */
+
+static int kafka_config(oconfig_item_t *ci) /* {{{ */
+{
+ int i;
+ oconfig_item_t *child;
+ rd_kafka_conf_t *conf;
+ rd_kafka_conf_t *cloned;
+ rd_kafka_conf_res_t ret;
+ char errbuf[1024];
+
+ if ((conf = rd_kafka_conf_new()) == NULL) {
+ WARNING("cannot allocate kafka configuration.");
+ return -1;
+ }
+
+ for (i = 0; i < ci->children_num; i++) {
+ child = &ci->children[i];
+
+ if (strcasecmp("Topic", child->key) == 0) {
+ if ((cloned = rd_kafka_conf_dup(conf)) == NULL) {
+ WARNING("write_kafka plugin: cannot allocate memory for kafka config");
+ goto errout;
+ }
+ kafka_config_topic (cloned, child);
+ } else if (strcasecmp(child->key, "Property") == 0) {
+ char *key = NULL;
+ char *val = NULL;
+
+ if (child->values_num != 2) {
+ WARNING("kafka properties need both a key and a value.");
+ goto errout;
+ }
+ if (child->values[0].type != OCONFIG_TYPE_STRING ||
+ child->values[1].type != OCONFIG_TYPE_STRING) {
+ WARNING("kafka properties needs string arguments.");
+ goto errout;
+ }
+ if ((key = strdup(child->values[0].value.string)) == NULL) {
+ WARNING("cannot allocate memory for attribute key.");
+ goto errout;
+ }
+ if ((val = strdup(child->values[1].value.string)) == NULL) {
+ WARNING("cannot allocate memory for attribute value.");
+ goto errout;
+ }
+ ret = rd_kafka_conf_set(conf, key, val, errbuf, sizeof(errbuf));
+ if (ret != RD_KAFKA_CONF_OK) {
+ WARNING("cannot set kafka property %s to %s: %s",
+ key, val, errbuf);
+ goto errout;
+ }
+ sfree(key);
+ sfree(val);
+ } else {
+ WARNING ("write_kafka plugin: Ignoring unknown "
+ "configuration option \"%s\" at top level.",
+ child->key);
+ }
+ }
+ if (conf != NULL)
+ rd_kafka_conf_destroy(conf);
+ return (0);
+ errout:
+ if (conf != NULL)
+ rd_kafka_conf_destroy(conf);
+ return -1;
+} /* }}} int kafka_config */
+
+void module_register(void)
+{
+ plugin_register_complex_config ("write_kafka", kafka_config);
+}
+
+/* vim: set sw=8 sts=8 ts=8 noet : */
#define RIEMANN_PORT "5555"
#define RIEMANN_TTL_FACTOR 2.0
+int write_riemann_threshold_check(const data_set_t *, const value_list_t *, int *);
+
struct riemann_host {
char *name;
#define F_CONNECT 0x01
uint8_t flags;
pthread_mutex_t lock;
+ _Bool notifications;
+ _Bool check_thresholds;
_Bool store_rates;
_Bool always_append_ds;
char *node;
static Event *riemann_value_to_protobuf (struct riemann_host const *host, /* {{{ */
data_set_t const *ds,
value_list_t const *vl, size_t index,
- gauge_t const *rates)
+ gauge_t const *rates,
+ int status)
{
Event *event;
char name_buffer[5 * DATA_MAX_NAME_LEN];
event->time = CDTIME_T_TO_TIME_T (vl->time);
event->has_time = 1;
+ if (host->check_thresholds) {
+ switch (status) {
+ case STATE_OKAY:
+ event->state = strdup("ok");
+ break;
+ case STATE_ERROR:
+ event->state = strdup("critical");
+ break;
+ case STATE_WARNING:
+ event->state = strdup("warning");
+ break;
+ case STATE_MISSING:
+ event->state = strdup("unknown");
+ break;
+ }
+ }
+
ttl = CDTIME_T_TO_DOUBLE (vl->interval) * host->ttl_factor;
event->ttl = (float) ttl;
event->has_ttl = 1;
} /* }}} Event *riemann_value_to_protobuf */
static Msg *riemann_value_list_to_protobuf (struct riemann_host const *host, /* {{{ */
- data_set_t const *ds,
- value_list_t const *vl)
+ data_set_t const *ds,
+ value_list_t const *vl,
+ int *statuses)
{
Msg *msg;
size_t i;
for (i = 0; i < msg->n_events; i++)
{
msg->events[i] = riemann_value_to_protobuf (host, ds, vl,
- (int) i, rates);
+ (int) i, rates, statuses[i]);
if (msg->events[i] == NULL)
{
riemann_msg_protobuf_free (msg);
struct riemann_host *host = ud->data;
Msg *msg;
+ if (!host->notifications)
+ return 0;
+
msg = riemann_notification_to_protobuf (host, n);
if (msg == NULL)
return (-1);
user_data_t *ud)
{
int status;
+ int statuses[vl->values_len];
struct riemann_host *host = ud->data;
Msg *msg;
- msg = riemann_value_list_to_protobuf (host, ds, vl);
+ if (host->check_thresholds)
+ write_riemann_threshold_check(ds, vl, statuses);
+ msg = riemann_value_list_to_protobuf (host, ds, vl, statuses);
if (msg == NULL)
return (-1);
host->reference_count = 1;
host->node = NULL;
host->service = NULL;
+ host->notifications = 1;
+ host->check_thresholds = 0;
host->store_rates = 1;
host->always_append_ds = 0;
host->use_tcp = 0;
status = cf_util_get_string (child, &host->node);
if (status != 0)
break;
+ } else if (strcasecmp ("Notifications", child->key) == 0) {
+ status = cf_util_get_boolean(child, &host->notifications);
+ if (status != 0)
+ break;
+ } else if (strcasecmp ("CheckThresholds", child->key) == 0) {
+ status = cf_util_get_boolean(child, &host->check_thresholds);
+ if (status != 0)
+ break;
} else if (strcasecmp ("Port", child->key) == 0) {
status = cf_util_get_service (child, &host->service);
if (status != 0) {
child->key);
}
}
- return (0);
+ return 0;
} /* }}} int riemann_config */
void module_register(void)
--- /dev/null
+/**
+ * collectd - src/threshold.c
+ * Copyright (C) 2007-2010 Florian Forster
+ * Copyright (C) 2008-2009 Sebastian Harl
+ * Copyright (C) 2009 Andrés J. Díaz
+ * Copyright (C) 2014 Pierre-Yves Ritschard
+ *
+ * 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:
+ * Pierre-Yves Ritschard <pyr at spootnik.org>
+ * Florian octo Forster <octo at collectd.org>
+ * Sebastian Harl <sh at tokkee.org>
+ * Andrés J. Díaz <ajdiaz at connectical.com>
+ **/
+
+#include "collectd.h"
+#include "common.h"
+#include "plugin.h"
+#include "utils_avltree.h"
+#include "utils_cache.h"
+#include "utils_threshold.h"
+
+#include <assert.h>
+#include <ltdl.h>
+#include <pthread.h>
+
+/*
+ * Threshold management
+ * ====================
+ * The following functions add, delete, etc. configured thresholds to
+ * the underlying AVL trees.
+ */
+
+/*
+ * int ut_check_one_data_source
+ *
+ * Checks one data source against the given threshold configuration. If the
+ * `DataSource' option is set in the threshold, and the name does NOT match,
+ * `okay' is returned. If the threshold does match, its failure and warning
+ * min and max values are checked and `failure' or `warning' is returned if
+ * appropriate.
+ * Does not fail.
+ */
+static int ut_check_one_data_source (const data_set_t *ds,
+ const value_list_t __attribute__((unused)) *vl,
+ const threshold_t *th,
+ const gauge_t *values,
+ int ds_index)
+{ /* {{{ */
+ const char *ds_name;
+ int is_warning = 0;
+ int is_failure = 0;
+ int prev_state = STATE_OKAY;
+
+ /* check if this threshold applies to this data source */
+ if (ds != NULL)
+ {
+ ds_name = ds->ds[ds_index].name;
+ if ((th->data_source[0] != 0)
+ && (strcmp (ds_name, th->data_source) != 0))
+ return (STATE_OKAY);
+ }
+
+ if ((th->flags & UT_FLAG_INVERT) != 0)
+ {
+ is_warning--;
+ is_failure--;
+ }
+
+ /* XXX: This is an experimental code, not optimized, not fast, not reliable,
+ * and probably, do not work as you expect. Enjoy! :D */
+ if ( (th->hysteresis > 0) && ((prev_state = uc_get_state(ds,vl)) != STATE_OKAY) )
+ {
+ switch(prev_state)
+ {
+ case STATE_ERROR:
+ if ( (!isnan (th->failure_min) && ((th->failure_min + th->hysteresis) < values[ds_index])) ||
+ (!isnan (th->failure_max) && ((th->failure_max - th->hysteresis) > values[ds_index])) )
+ return (STATE_OKAY);
+ else
+ is_failure++;
+ case STATE_WARNING:
+ if ( (!isnan (th->warning_min) && ((th->warning_min + th->hysteresis) < values[ds_index])) ||
+ (!isnan (th->warning_max) && ((th->warning_max - th->hysteresis) > values[ds_index])) )
+ return (STATE_OKAY);
+ else
+ is_warning++;
+ }
+ }
+ else { /* no hysteresis */
+ if ((!isnan (th->failure_min) && (th->failure_min > values[ds_index]))
+ || (!isnan (th->failure_max) && (th->failure_max < values[ds_index])))
+ is_failure++;
+
+ if ((!isnan (th->warning_min) && (th->warning_min > values[ds_index]))
+ || (!isnan (th->warning_max) && (th->warning_max < values[ds_index])))
+ is_warning++;
+ }
+
+ if (is_failure != 0)
+ return (STATE_ERROR);
+
+ if (is_warning != 0)
+ return (STATE_WARNING);
+
+ return (STATE_OKAY);
+} /* }}} int ut_check_one_data_source */
+
+/*
+ * int ut_check_one_threshold
+ *
+ * Checks all data sources of a value list against the given threshold, using
+ * the ut_check_one_data_source function above. Returns the worst status,
+ * which is `okay' if nothing has failed.
+ * Returns less than zero if the data set doesn't have any data sources.
+ */
+static int ut_check_one_threshold (const data_set_t *ds,
+ const value_list_t *vl,
+ const threshold_t *th,
+ const gauge_t *values,
+ int *statuses)
+{ /* {{{ */
+ int ret = -1;
+ int i;
+ int status;
+ gauge_t values_copy[ds->ds_num];
+
+ memcpy (values_copy, values, sizeof (values_copy));
+
+ if ((th->flags & UT_FLAG_PERCENTAGE) != 0)
+ {
+ int num = 0;
+ gauge_t sum=0.0;
+
+ if (ds->ds_num == 1)
+ {
+ WARNING ("ut_check_one_threshold: The %s type has only one data "
+ "source, but you have configured to check this as a percentage. "
+ "That doesn't make much sense, because the percentage will always "
+ "be 100%%!", ds->type);
+ }
+
+ /* Prepare `sum' and `num'. */
+ for (i = 0; i < ds->ds_num; i++)
+ if (!isnan (values[i]))
+ {
+ num++;
+ sum += values[i];
+ }
+
+ if ((num == 0) /* All data sources are undefined. */
+ || (sum == 0.0)) /* Sum is zero, cannot calculate percentage. */
+ {
+ for (i = 0; i < ds->ds_num; i++)
+ values_copy[i] = NAN;
+ }
+ else /* We can actually calculate the percentage. */
+ {
+ for (i = 0; i < ds->ds_num; i++)
+ values_copy[i] = 100.0 * values[i] / sum;
+ }
+ } /* if (UT_FLAG_PERCENTAGE) */
+
+ for (i = 0; i < ds->ds_num; i++)
+ {
+ status = ut_check_one_data_source (ds, vl, th, values_copy, i);
+ if (status != -1) {
+ ret = 0;
+ if (statuses[i] < status)
+ statuses[i] = status;
+ }
+ } /* for (ds->ds_num) */
+
+ return (ret);
+} /* }}} int ut_check_one_threshold */
+
+/*
+ * int ut_check_threshold
+ *
+ * Gets a list of matching thresholds and searches for the worst status by one
+ * of the thresholds. Then reports that status using the ut_report_state
+ * function above.
+ * Returns zero on success and if no threshold has been configured. Returns
+ * less than zero on failure.
+ */
+int write_riemann_threshold_check (const data_set_t *ds, const value_list_t *vl,
+ int *statuses)
+{ /* {{{ */
+ threshold_t *th;
+ gauge_t *values;
+ int status;
+
+ memset(statuses, 0, vl->values_len * sizeof(*statuses));
+ if (threshold_tree == NULL)
+ return 0;
+
+ /* Is this lock really necessary? So far, thresholds are only inserted at
+ * startup. -octo */
+ pthread_mutex_lock (&threshold_lock);
+ th = threshold_search (vl);
+ pthread_mutex_unlock (&threshold_lock);
+ if (th == NULL)
+ return (0);
+
+ DEBUG ("ut_check_threshold: Found matching threshold(s)");
+
+ values = uc_get_rate (ds, vl);
+ if (values == NULL)
+ return (0);
+
+ while (th != NULL)
+ {
+ status = ut_check_one_threshold (ds, vl, th, values, statuses);
+ if (status < 0)
+ {
+ ERROR ("ut_check_threshold: ut_check_one_threshold failed.");
+ sfree (values);
+ return (-1);
+ }
+
+ th = th->next;
+ } /* while (th) */
+
+ sfree (values);
+
+ return (0);
+} /* }}} int ut_check_threshold */
+
+
+/* vim: set sw=2 ts=8 sts=2 tw=78 et fdm=marker : */
projects / collectd.git / commitdiff