From: Marc Fournier Date: Mon, 24 Nov 2014 08:44:45 +0000 (+0100) Subject: Merge branch 'collectd-5.4' X-Git-Tag: collectd-5.5.0~127 X-Git-Url: https://git.octo.it/?p=collectd.git;a=commitdiff_plain;h=e1812f8e5c5f91328398c66b602d07ee7c6ede51;hp=-c Merge branch 'collectd-5.4' Conflicts: src/Makefile.am src/cpu.c src/swap.c src/write_redis.c NB: dropped change to src/swap.c made in 1a146775d4, as suggested in the commit message. --- e1812f8e5c5f91328398c66b602d07ee7c6ede51 diff --combined src/collectd.conf.in index 5196979e,1e0f78c1..a25cce55 --- a/src/collectd.conf.in +++ b/src/collectd.conf.in @@@ -25,13 -25,6 +25,13 @@@ #AutoLoadPlugin false #----------------------------------------------------------------------------# +# When enabled, internal statistics are collected, using "collectd" as the # +# plugin name. # +# Disabled by default. # +#----------------------------------------------------------------------------# +#CollectInternalStats false + +#----------------------------------------------------------------------------# # Interval at which to query values. This may be overwritten on a per-plugin # # base by using the 'Interval' option of the LoadPlugin block: # # # @@@ -40,10 -33,9 +40,10 @@@ #----------------------------------------------------------------------------# #Interval 10 -#Timeout 2 -#ReadThreads 5 -#WriteThreads 5 +#MaxReadInterval 86400 +#Timeout 2 +#ReadThreads 5 +#WriteThreads 5 # Limit the size of the write queue. Default is no limit. Setting up a limit is # recommended for servers handling a high volume of traffic. @@@ -60,7 -52,6 +60,7 @@@ @LOAD_PLUGIN_SYSLOG@LoadPlugin syslog @LOAD_PLUGIN_LOGFILE@LoadPlugin logfile +@LOAD_PLUGIN_LOG_LOGSTASH@LoadPlugin log_logstash # # LogLevel @DEFAULT_LOG_LEVEL@ @@@ -69,11 -60,6 +69,11 @@@ # PrintSeverity false # +# +# LogLevel @DEFAULT_LOG_LEVEL@ +# File "@localstatedir@/log/@PACKAGE_NAME@.json.log" +# + # # LogLevel @DEFAULT_LOG_LEVEL@ # @@@ -95,7 -81,6 +95,7 @@@ #@BUILD_PLUGIN_APPLE_SENSORS_TRUE@LoadPlugin apple_sensors #@BUILD_PLUGIN_AQUAERO_TRUE@LoadPlugin aquaero #@BUILD_PLUGIN_ASCENT_TRUE@LoadPlugin ascent +#@BUILD_PLUGIN_BAROMETER_TRUE@LoadPlugin barometer #@BUILD_PLUGIN_BATTERY_TRUE@LoadPlugin battery #@BUILD_PLUGIN_BIND_TRUE@LoadPlugin bind #@BUILD_PLUGIN_CONNTRACK_TRUE@LoadPlugin conntrack @@@ -111,7 -96,6 +111,7 @@@ #@BUILD_PLUGIN_DF_TRUE@LoadPlugin df #@BUILD_PLUGIN_DISK_TRUE@LoadPlugin disk #@BUILD_PLUGIN_DNS_TRUE@LoadPlugin dns +#@BUILD_PLUGIN_DRBD_TRUE@LoadPlugin drbd #@BUILD_PLUGIN_EMAIL_TRUE@LoadPlugin email #@BUILD_PLUGIN_ENTROPY_TRUE@LoadPlugin entropy #@BUILD_PLUGIN_ETHSTAT_TRUE@LoadPlugin ethstat @@@ -126,6 -110,7 +126,6 @@@ #@BUILD_PLUGIN_IPVS_TRUE@LoadPlugin ipvs #@BUILD_PLUGIN_IRQ_TRUE@LoadPlugin irq #@BUILD_PLUGIN_JAVA_TRUE@LoadPlugin java -#@BUILD_PLUGIN_LIBVIRT_TRUE@LoadPlugin libvirt @BUILD_PLUGIN_LOAD_TRUE@@BUILD_PLUGIN_LOAD_TRUE@LoadPlugin load #@BUILD_PLUGIN_LPAR_TRUE@LoadPlugin lpar #@BUILD_PLUGIN_LVM_TRUE@LoadPlugin lvm @@@ -150,7 -135,6 +150,7 @@@ #@BUILD_PLUGIN_NUT_TRUE@LoadPlugin nut #@BUILD_PLUGIN_OLSRD_TRUE@LoadPlugin olsrd #@BUILD_PLUGIN_ONEWIRE_TRUE@LoadPlugin onewire +#@BUILD_PLUGIN_OPENLDAP_TRUE@LoadPlugin openldap #@BUILD_PLUGIN_OPENVPN_TRUE@LoadPlugin openvpn #@BUILD_PLUGIN_ORACLE_TRUE@LoadPlugin oracle #@BUILD_PLUGIN_PERL_TRUE@ @@@ -172,7 -156,6 +172,7 @@@ #@BUILD_PLUGIN_SENSORS_TRUE@LoadPlugin sensors #@BUILD_PLUGIN_SERIAL_TRUE@LoadPlugin serial #@BUILD_PLUGIN_SIGROK_TRUE@LoadPlugin sigrok +#@BUILD_PLUGIN_SMART_TRUE@LoadPlugin smart #@BUILD_PLUGIN_SNMP_TRUE@LoadPlugin snmp #@BUILD_PLUGIN_STATSD_TRUE@LoadPlugin statsd #@BUILD_PLUGIN_SWAP_TRUE@LoadPlugin swap @@@ -191,17 -174,14 +191,17 @@@ #@BUILD_PLUGIN_UUID_TRUE@LoadPlugin uuid #@BUILD_PLUGIN_VARNISH_TRUE@LoadPlugin varnish #@BUILD_PLUGIN_MIC_TRUE@LoadPlugin mic +#@BUILD_PLUGIN_VIRT_TRUE@LoadPlugin virt #@BUILD_PLUGIN_VMEM_TRUE@LoadPlugin vmem #@BUILD_PLUGIN_VSERVER_TRUE@LoadPlugin vserver #@BUILD_PLUGIN_WIRELESS_TRUE@LoadPlugin wireless #@BUILD_PLUGIN_WRITE_GRAPHITE_TRUE@LoadPlugin write_graphite #@BUILD_PLUGIN_WRITE_HTTP_TRUE@LoadPlugin write_http +#@BUILD_PLUGIN_WRITE_KAFKA_TRUE@LoadPlugin write_kafka #@BUILD_PLUGIN_WRITE_MONGODB_TRUE@LoadPlugin write_mongodb #@BUILD_PLUGIN_WRITE_REDIS_TRUE@LoadPlugin write_redis #@BUILD_PLUGIN_WRITE_RIEMANN_TRUE@LoadPlugin write_riemann +#@BUILD_PLUGIN_WRITE_TSDB_TRUE@LoadPlugin write_tsdb #@BUILD_PLUGIN_XMMS_TRUE@LoadPlugin xmms #@BUILD_PLUGIN_ZFS_ARC_TRUE@LoadPlugin zfs_arc @@@ -212,7 -192,7 +212,7 @@@ # ription of those options is available in the collectd.conf(5) manual page. # ############################################################################## -# +# # # #Host "unspecified" # Plugin "cpu" @@@ -232,7 -212,7 +232,7 @@@ # # -# +# # # Host "localhost" # Port "5672" @@@ -272,21 -252,6 +272,21 @@@ # CACert "/etc/ssl/ca.crt" # +# +# Device "/dev/i2c-0"; +# Oversampling 512 +# PressureOffset 0.0 +# TemperatureOffset 0.0 +# Normalization 2 +# Altitude 238.0 +# TemperatureSensor "myserver/onewire-F10FCA000800/temperature" +# + +# +# ValuesPercentage false +# ReportDegraded +# + # # URL "http://localhost:8053/" # ParseTime false @@@ -312,12 -277,6 +312,12 @@@ # IgnoreSelected false # +# +# ReportByCpu true +# ReportByState true +# ValuesPercentage false +# +# # # DataDir "@localstatedir@/lib/@PACKAGE_NAME@/csv" # StoreRates false @@@ -328,15 -287,7 +328,15 @@@ # URL "http://finance.google.com/finance?q=NYSE%3AAMD" # User "foo" # Password "bar" +# Digest false +# VerifyPeer true +# VerifyHost true +# CACert "/path/to/ca.crt" +# Header "X-Custom-Header: foobar" +# Post "foo=bar" +# # MeasureResponseTime false +# MeasureResponseCode false # # Regex "]*> *([0-9]*\\.[0-9]+) *" # DSType "GaugeAverage" @@@ -347,6 -298,41 +347,41 @@@ # # + # + # Instance "test_http_json" + # + # Type "gauge" + # # Expect: 1 + # + # + # Type "gauge" + # # Expect: 2 + # + # + # Type "gauge" + # # Expect: 3 + # + # + # Type "gauge" + # # Expect: 4 + # + # + # Type "gauge" + # # Expect: 5 + # + # + # Type "gauge" + # # Expect: 6 + # + # + # put this as test.json on your webserver, the above config demonstraces + # how to match them. + # { + # "testArray":[1,2], + # "testArrayInbetween":[{"blarg":3},{"blub":4}], + # "testDirectHit":5, + # "testSubLevelHit":{"oneMoreLevel":6} + # } ## See: http://wiki.apache.org/couchdb/Runtime_Statistics # # Instance "httpd" @@@ -377,18 -363,15 +412,18 @@@ # # -# +# # # Host "my_host" # Instance "some_instance" # User "collectd" # Password "thaiNg0I" +# Digest false # VerifyPeer true # VerifyHost true # CACert "/path/to/ca.crt" +# Header "X-Custom-Header: foobar" +# Post "foo=bar" # # # Type "magic_level" @@@ -437,8 -420,6 +472,8 @@@ # # Disk "/^[hs]d[a-f][0-9]?$/" # IgnoreSelected false +# UseBSDName false +# UdevNameAttr "DEVNAME" # # @@@ -477,7 -458,7 +512,7 @@@ # # -# +# # MCReceiveFrom "239.2.11.71" "8649" # # Type "swap" @@@ -521,7 -502,7 +556,7 @@@ # IgnoreSelected true # -# +# # JVMArg "-verbose:jni" # JVMArg "-Djava.class.path=@prefix@/share/collectd/java/collectd-api.jar" # @@@ -531,8 -512,15 +566,8 @@@ # # -# -# Connection "xen:///" -# RefreshInterval 60 -# Domain "name" -# BlockDevice "name:device" -# InterfaceDevice "name:device" -# IgnoreSelected false -# HostnameFormat name -# InterfaceFormat name +# +# ReportRelative true # # @@@ -582,11 -570,6 +617,11 @@@ # # +# +# ValuesAbsolute true +# ValuesPercentage false +# + # # # RegisterBase 1234 @@@ -614,12 -597,9 +649,12 @@@ # Password "secret" # Database "db_name" # MasterStats true +# ConnectTimeout 10 +# InnodbStats true # # # +# Alias "squeeze" # Host "localhost" # Socket "/var/run/mysql/mysqld.sock" # SlaveStats true @@@ -696,7 -676,6 +731,7 @@@ # Username "user" # Password "secret" # Interface "eth0" +# ResolveInterval 14400 @LOAD_PLUGIN_NETWORK@ # TimeToLive "128" # @@@ -769,17 -748,6 +804,17 @@@ # IgnoreSelected false # +# +# +# URL "ldap://localhost:389" +# StartTLS false +# VerifyHost true +# CACert "/path/to/ca.crt" +# Timeout -1 +# Version 3 +# +# + # # StatusFile "/etc/openvpn/openvpn-status.log" # ImprovedNamingSchema false @@@ -984,11 -952,6 +1019,11 @@@ # # +# +# Disk "/^[hs]d[a-f][0-9]?$/" +# IgnoreSelected false +# + # # # Type "voltage" @@@ -1039,16 -1002,20 +1074,22 @@@ # DeleteGauges false # DeleteSets false # TimerPercentile 90.0 + # TimerPercentile 95.0 + # TimerPercentile 99.0 + # TimerLower false + # TimerUpper false + # TimerSum false + # TimerCount false # -# +# # ReportByDevice false # ReportBytes true +# ValuesAbsolute true +# ValuesPercentage false # -# +# # # Instance "slabinfo" # Separator " " @@@ -1067,10 -1034,9 +1108,10 @@@ #
#
-# +# # # Instance "exim" +# Interval 60 # # Regex "S=([1-9][0-9]*)" # DSType "CounterAdd" @@@ -1087,7 -1053,7 +1128,7 @@@ # # -# +# # # Type "percent" # Instance "dropped" @@@ -1116,7 -1082,6 +1157,7 @@@ # # ListeningPorts false +# AllPortsSummary false # LocalPort "25" # RemotePort "25" # @@@ -1178,42 -1143,29 +1219,42 @@@ # If you prefer defining another instance you can do # so by using # -# CollectCache true # CollectBackend true -# CollectBan false # Varnish 3 only +# CollectBan false # Varnish 3 and above +# CollectCache true # CollectConnections true # CollectDirectorDNS false # Varnish 3 only -# CollectSHM true # CollectESI false # CollectFetch false # CollectHCB false # CollectObjects false # CollectPurge false # Varnish 2 only # CollectSession false +# CollectSHM true # CollectSMA false # Varnish 2 only # CollectSMS false # CollectSM false # Varnish 2 only # CollectStruct false # CollectTotals false -# CollectUptime false +# CollectUptime false # Varnish 3 and above # CollectVCL false +# CollectVSM false # Varnish 4 only # CollectWorkers false # # +# +# Connection "xen:///" +# RefreshInterval 60 +# Domain "name" +# BlockDevice "name:device" +# InterfaceDevice "name:device" +# IgnoreSelected false +# HostnameFormat name +# InterfaceFormat name +# PluginInstanceFormat name +# + # # Verbose false # @@@ -1239,24 -1191,11 +1280,24 @@@ # VerifyPeer true # VerifyHost true # CACert "/etc/ssl/ca.crt" +# CAPath "/etc/ssl/certs/" +# ClientKey "/etc/ssl/client.pem" +# ClientCert "/etc/ssl/client.crt" +# ClientKeyPass "secret" +# SSLVersion "TLSv1" # Format "Command" # StoreRates false +# BufferSize 4096 #
#
+# +# Property "metadata.broker.list" "localhost:9092" +# +# Format JSON +# +# + # # # Host "localhost" @@@ -1285,20 -1224,8 +1326,20 @@@ # StoreRates true # AlwaysAppendDS false # TTLFactor 2.0 +# EventServicePrefix "" # # Tag "foobar" +# Attribute "foo" "bar" +# + +# +# +# Host "localhost" +# Port "4242" +# HostTags "status=production" +# StoreRates false +# AlwaysAppendDS false +# # ############################################################################## @@@ -1341,7 -1268,7 +1382,7 @@@ ############################################################################## #@BUILD_PLUGIN_THRESHOLD_TRUE@LoadPlugin "threshold" -# +# # # WarningMin 0.00 # WarningMax 1000.00 diff --combined src/collectd.conf.pod index 20c61260,03a14443..a65d432b --- a/src/collectd.conf.pod +++ b/src/collectd.conf.pod @@@ -9,17 -9,17 +9,17 @@@ collectd.conf - Configuration for the s BaseDir "/var/lib/collectd" PIDFile "/run/collectd.pid" Interval 10.0 - + LoadPlugin cpu LoadPlugin load - + Interval 3600 ValuesPercentage true - + LoadPlugin ping Host "example.org" @@@ -146,18 -146,6 +146,18 @@@ BPluginE...E> block act B statement. B statements are still required for plugins that don't provide any configuration, e.g. the I. +=item B B|B + +When set to B, various statistics about the I daemon will be +collected, with "collectd" as the I. Defaults to B. + +The "write_queue" I reports the number of elements currently +queued and the number of elements dropped off the queue by the +B/B mechanism. + +The "cache" I reports the number of elements in the value list +cache (the cache you can interact with using L). + =item B I [I] If I points to a file, includes that file. If I points to a @@@ -230,14 -218,6 +230,14 @@@ B You should set this once an I or know some serious RRDtool magic! (Assuming you're using the I or I plugin.) +=item B I + +Read plugin doubles interval between queries after each failed attempt +to get data. + +This options limits the maximum value of the interval. The default value is +B<86400>. + =item B I Consider a value list "missing" when no update has been read or received for @@@ -288,11 -268,8 +288,11 @@@ If B is set to non unset, the latter will default to half of B. If you do not want to randomly drop values when the queue size is between -I and I, set If B and -B to same value. +I and I, set B and B +to the same value. + +Enabling the B option is of great help to figure out the +values to set B and B to. =item B I @@@ -370,10 -347,10 +370,10 @@@ The full example configuration looks li Plugin "cpu" Type "cpu" - + GroupBy "Host" GroupBy "TypeInstance" - + CalculateSum true CalculateAverage true @@@ -449,13 -426,13 +449,13 @@@ The following example calculates the av Plugin "cpu" PluginInstance "/[0,2,4,6,8]$/" Type "cpu" - + SetPlugin "cpu" SetPluginInstance "even-%{aggregation}" - + GroupBy "Host" GroupBy "TypeInstance" - + CalculateAverage true @@@ -526,7 -503,7 +526,7 @@@ possibly filtering or messages # GraphiteSeparateInstances false # GraphiteAlwaysAppendDS false - + # Receive values from an AMQP broker Host "localhost" @@@ -537,8 -514,6 +537,8 @@@ Exchange "amq.fanout" # ExchangeType "fanout" # Queue "queue_name" + # QueueDurable false + # QueueAutoDelete true # RoutingKey "collectd.#" @@@ -591,23 -566,9 +591,23 @@@ be bound to this exchange =item B I (Subscribe only) -Configures the I name to subscribe to. If no queue name was configures +Configures the I name to subscribe to. If no queue name was configured explicitly, a unique queue name will be created by the broker. +=item B B|B (Subscribe only) + +Defines if the I 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 option on the +publish side. + +=item B B|B (Subscribe only) + +Defines if the I subscribed to will be deleted once the last consumer +unsubscribes. Defaults to "true". + =item B I In I blocks, this configures the routing key to set on all outgoing @@@ -856,165 -817,6 +856,165 @@@ and are checked by default depends on t =back +=head2 Plugin C + +This plugin reads absolute air pressure using digital barometer sensor MPL115A2 +or MPL3115 from Freescale (sensor attached to any I2C bus available in +the computer, for HW details see +I or +I). +The sensor type - one fo these two - is detected automatically by the plugin +and indicated in the plugin_instance (typically you will see subdirectory +"barometer-mpl115" or "barometer-mpl3115"). + +The plugin provides absolute barometric pressure, air pressure reduced to sea +level (several possible approximations) and as an auxiliary value also internal +sensor temperature. It uses (expects/provides) typical metric units - pressure +in [hPa], temperature in [C], altitude in [m]. + +It was developed and tested under Linux only. The only platform dependency is +the standard Linux i2c-dev interface (the particular bus driver has to +support the SM Bus command subset). + +The reduction or normalization to mean sea level pressure requires (depedning on +selected method/approximation) also altitude and reference to temperature sensor(s). +When multiple temperature sensors are configured the minumum of their values is +always used (expecting that the warmer ones are affected by e.g. direct sun light +at that moment). + +Synopsis: + + + Device "/dev/i2c-0"; + Oversampling 512 + PressureOffset 0.0 + TemperatureOffset 0.0 + Normalization 2 + Altitude 238.0 + TemperatureSensor "myserver/onewire-F10FCA000800/temperature" + + +=over 4 + +=item B I + +Device name of the I2C bus to which the sensor is connected. Note that typically +you need to have loaded the i2c-dev module. +Using i2c-tools you can check/list i2c buses available on your system by: + + i2cdetect -l + +Then you can scan for devices on given bus. E.g. to scan the whole bus 0 use: + + i2cdetect -y -a 0 + +This way you should be able to verify that the pressure sensor (either type) is +connected and detected on address 0x60. + +=item B I + +For MPL115 this is the size of the averaging window. To filter out sensor noise +a simple averaging using floating window of configurable size is used. The plugin +will use average of the last C measurements (value of 1 means no averaging). +Minimal size is 1, maximal 1024. + +For MPL3115 this is the oversampling value. The actual oversampling is performed +by the sensor and the higher value the higher accuracy and longer conversion time +(although nothing to worry about in the collectd context). Supported values are: +1, 2, 4, 8, 16, 32, 64 and 128. Any other value is adjusted by the plugin to +the closest supported one. Default is 128. + +=item B I + +You can further calibrate the sensor by supplying pressure and/or temperature offsets. +This is added to the measured/caclulated value (i.e. if the measured value is too high +then use negative offset). +In hPa, default is 0.0. + +=item B I + +You can further calibrate the sensor by supplying pressure and/or temperature offsets. +This is added to the measured/caclulated value (i.e. if the measured value is too high +then use negative offset). +In C, default is 0.0. + +=item B I + +Normalization method - what approximation/model is used to compute mean sea +level pressure from the air absolute pressure. + +Supported values of the C (integer between from 0 to 2) are: + +=over 5 + +=item B<0> - no conversion, absolute pressrure is simply copied over. For this method you + do not need to configure C or C. + +=item B<1> - international formula for conversion , +See I. +For this method you have to configure C but do not need C +(uses fixed global temperature average instead). + +=item B<2> - formula as recommended by the Deutsche Wetterdienst (German +Meteorological Service). +See I +For this method you have to configure both C and C. + +=back + + +=item B I + +The altitude (in meters) of the location where you meassure the pressure. + +=item B I + +Temperature sensor which should be used as a reference when normalizing the pressure. +When specified more sensors a minumum is found and uses each time. +The temperature reading directly from this pressure sensor/plugin +is typically not suitable as the pressure sensor +will be probably inside while we want outside temperature. +The collectd reference name is something like +/-/- +( is usually omitted when there is just single value type). +Or you can figure it out from the path of the output data files. + +=back + +=head2 Plugin C + +The I reports the remaining capacity, power and voltage of +laptop batteries. + +=over 4 + +=item B B|B + +When enabled, remaining capacity is reported as a percentage, e.g. "42% +capacity remaining". Otherwise the capacity is stored as reported by the +battery, most likely in "Wh". This option does not work with all input methods, +in particular when only C is available on an old Linux system. +Defaults to B. + +=item B B|B + +Typical laptop batteries degrade over time, meaning the capacity decreases with +recharge cycles. The maximum charge of the previous charge cycle is tracked as +"last full capacity" and used to determine that a battery is "fully charged". + +When this option is set to B, the default, the I will +only report the remaining capacity. If the B option is +enabled, the relative remaining capacity is calculated as the ratio of the +"remaining capacity" and the "last full capacity". This is what most tools, +such as the status bar of desktop environments, also do. + +When set to B, the battery plugin will report three values: B +(remaining capacity), B (difference between "last full capacity" +and "remaining capacity") and B (difference between "design capacity" +and "last full capacity"). + +=back + =head2 Plugin C Starting with BIND 9.5.0, the most widely used DNS server software provides @@@ -1041,17 -843,17 +1041,17 @@@ Synopsis ParseTime false OpCodes true QTypes true - + ServerStats true ZoneMaintStats true ResolverStats false MemoryStats true - + QTypes true ResolverStats true CacheRRSets true - + Zone "127.in-addr.arpa/IN" @@@ -1192,68 -994,6 +1192,68 @@@ at all, B cgroups are selected =back +=head2 Plugin C + +This plugin collects IP conntrack statistics. + +=over 4 + +=item B + +Assume the B and B files to be found in +F instead of F. + +=back + +=head2 Plugin C + +The I collects CPU usage metrics. By default, CPU usage is reported +as Jiffies, using the C type. Two aggregations are available: + +=over 4 + +=item + +Sum, per-state, over all CPUs installed in the system; and + +=item + +Sum, per-CPU, over all non-idle states of a CPU, creating an "active" state. + +=back + +The two aggregations can be combined, leading to I only emitting a +single "active" metric for the entire system. As soon as one of these +aggregations (or both) is enabled, the I will report a percentage, +rather than Jiffies. In addition, you can request individual, per-state, +per-CPU metrics to be reported as percentage. + +The following configuration options are available: + +=over 4 + +=item B B|B + +When set to B, the default, reports per-state metrics, e.g. "system", +"user" and "idle". +When set to B, aggregates (sums) all I states into one +"active" metric. + +=item B B|B + +When set to B, the default, reports per-CPU (per-core) metrics. +When set to B, instead of reporting metrics for individual CPUs, only a +global sum of CPU states is emitted. + +=item B B|B + +This option is only considered when both, B and B +are set to B. In this case, by default, metrics will be reported as +Jiffies. By setting this option to B, you can request percentage values +in the un-aggregated (per-CPU, per-state) mode as well. + +=back + =head2 Plugin C This plugin doesn't have any options. It reads @@@ -1296,16 -1036,6 +1296,16 @@@ finance page and dispatch the value to URL "http://finance.google.com/finance?q=NYSE%3AAMD" User "foo" Password "bar" + Digest false + VerifyPeer true + VerifyHost true + CACert "/path/to/ca.crt" + Header "X-Custom-Header: foobar" + Post "foo=bar" + + MeasureResponseTime false + MeasureResponseCode false + Regex "]*> *([0-9]*\\.[0-9]+) *" DSType "GaugeAverage" @@@ -1337,10 -1067,6 +1337,10 @@@ Username to use if authorization is req Password to use if authorization is required to read the page. +=item B B|B + +Enable HTTP digest authentication. + =item B B|B Enable or disable peer SSL certificate verification. See @@@ -1378,19 -1104,13 +1378,19 @@@ C) Measure response time for the request. If this setting is enabled, B blocks (see below) are optional. Disabled by default. +=item B B|B + +Measure response code for the request. If this setting is enabled, B +blocks (see below) are optional. Disabled by default. + =item BMatchE> One or more B blocks that define how to match information in the data returned by C. The C plugin uses the same infrastructure that's used by the C plugin, so please see the documentation of the C -plugin below on how matches are defined. If the B option -is set to B, B blocks are optional. +plugin below on how matches are defined. If the B or +B options are set to B, B blocks are +optional. =back @@@ -1457,25 -1177,12 +1457,25 @@@ The following options are valid within Sets the plugin instance to I. +=item B I + +Sets the interval (in seconds) in which the values will be collected from this +URL. By default the global B setting will be used. + =item B I + =item B I + +=item B B|B + =item B B|B + =item B B|B + =item B I + =item B
I
+ =item B I These options behave exactly equivalent to the appropriate options of the @@@ -1513,8 -1220,6 +1513,8 @@@ The B uses B VerifyPeer true VerifyHost true CACert "/path/to/ca.crt" + Header "X-Custom-Header: foobar" + Post "foo=bar" Type "magic_level" @@@ -1566,8 -1271,6 +1566,8 @@@ Examples =item B I +=item B B|B + =item B B|B =item B B|B @@@ -1807,16 -1510,6 +1807,16 @@@ it should be able to handle integer an There must be at least one B option inside each B block. +=item B [I I ...] + +Names the columns whose content is used as metadata for the data sets +that are dispatched to the daemon. + +The actual data type in the columns is not that important. The plugin will +automatically cast the values to the right type if it know how to do that. So +it should be able to handle integer an floating point types, as well as strings +(if they include a number at the beginning). + =back =head3 B blocks @@@ -1931,17 -1624,17 +1931,17 @@@ transfer agents and web caches =item B B|B -Enables or disables reporting of free, used and used disk space in 1K-blocks. -Defaults to true. +Enables or disables reporting of free and used disk space in 1K-blocks. +Defaults to B. -=item B B|B +=item B B|B -Enables or disables reporting of free, used and used disk space in percentage. -Defaults to false. +Enables or disables reporting of free and used disk space in percentage. +Defaults to B. -This is useful for deploying collectd on the cloud, where machines with -different disk size may exist. Then it is more practical to configure thresholds -based on relative disk size. +This is useful for deploying I on the cloud, where machines with +different disk size may exist. Then it is more practical to configure +thresholds based on relative disk size. =back @@@ -1977,20 -1670,6 +1977,20 @@@ collected. If at least one B opti set to B, B matching disks will be collected. If B is set to B, all disks are collected B the ones matched. +=item B B|B + +Whether to use the device's "BSD Name", on MacEOSEX, instead of the +default major/minor numbers. Requires collectd to be built with Apple's +IOKitLib support. + +=item B I + +Attempt to override disk instance name with the value of a specified udev +attribute when built with B. If the attribute is not defined for the +given device, the default name is used. Example: + + UdevNameAttr "DM_NAME" + =back =head2 Plugin C @@@ -2454,63 -2133,94 +2454,63 @@@ independent from the I argum =back -=head2 Plugin C +=head2 Plugin C -This plugin allows CPU, disk and network load to be collected for virtualized -guests on the machine. This means that these characteristics can be collected -for guest systems without installing any software on them - collectd only runs -on the hosting system. The statistics are collected through libvirt -(L). +The I collects the system load. These numbers give a rough overview +over the utilization of a machine. The system load is defined as the number of +runnable tasks in the run-queue and is provided by many operating systems as a +one, five or fifteen minute average. -Only I is required. +The following configuration options are available: =over 4 -=item B I - -Connect to the hypervisor given by I. For example if using Xen use: - - Connection "xen:///" - -Details which URIs allowed are given at L. - -=item B I - -Refresh the list of domains and devices every I. The default is 60 -seconds. Setting this to be the same or smaller than the I will cause -the list of domains and devices to be refreshed on every iteration. - -Refreshing the devices in particular is quite a costly operation, so if your -virtualization setup is static you might consider increasing this. If this -option is set to 0, refreshing is disabled completely. - -=item B I - -=item B I - -=item B I - -=item B I|I +=item B B|B -Select which domains and devices are collected. +When enabled, system load divided by number of available CPU cores is reported +for intervals 1 min, 5 min and 15 min. Defaults to false. -If I is not given or I then only the listed domains and -disk/network devices are collected. +=back -If I is I then the test is reversed and the listed -domains and disk/network devices are ignored, while the rest are collected. -The domain name and device names may use a regular expression, if the name is -surrounded by I and collectd was compiled with support for regexps. +=head2 Plugin C -The default is to collect statistics for all domains and all their devices. +=over 4 -Example: +=item B B - BlockDevice "/:hdb/" - IgnoreSelected "true" +Sets the log-level. If, for example, set to B, then all events with +severity B, B, or B will be written to the logfile. -Ignore all I devices on any domain, but other block devices (eg. I) -will be collected. +Please note that B is only available if collectd has been compiled with +debugging support. -=item B B +=item B I -When the libvirt plugin logs data, it sets the hostname of the collected data -according to this setting. The default is to use the guest name as provided by -the hypervisor, which is equal to setting B. +Sets the file to write log messages to. The special strings B and +B can be used to write to the standard output and standard error +channels, respectively. This, of course, only makes much sense when I +is running in foreground- or non-daemon-mode. -B means use the guest's UUID. This is useful if you want to track the -same guest across migrations. +=item B B|B -B means to use the global B setting, which is probably not -useful on its own because all guests will appear to have the same name. +Prefix all lines printed by the current time. Defaults to B. -You can also specify combinations of these fields. For example B -means to concatenate the guest name and UUID (with a literal colon character -between, thus I<"foo:1234-1234-1234-1234">). +=item B B|B -=item B B|B
+When enabled, all lines are prefixed by the severity of the log message, for +example "warning". Defaults to B. -When the libvirt plugin logs interface data, it sets the name of the collected -data according to this setting. The default is to use the path as provided by -the hypervisor (the "dev" property of the target node), which is equal to -setting B. +=back -B
means use the interface's mac address. This is useful since the -interface path might change between reboots of a guest or across migrations. +B: There is no need to notify the daemon after moving or removing the +log file (e.Eg. when rotating the logs). The plugin reopens the file +for each line it writes. -=back +=head2 Plugin C -=head2 Plugin C +The I behaves like the logfile plugin but formats +messages as JSON events for logstash to parse and input. =over 4 @@@ -2529,6 -2239,15 +2529,6 @@@ B can be used to write to the s channels, respectively. This, of course, only makes much sense when I is running in foreground- or non-daemon-mode. -=item B B|B - -Prefix all lines printed by the current time. Defaults to B. - -=item B B|B - -When enabled, all lines are prefixed by the severity of the log message, for -example "warning". Defaults to B. - =back B: There is no need to notify the daemon after moving or removing the @@@ -2708,7 -2427,7 +2708,7 @@@ B The following options are valid inside the Bmic> block: @@@ -2726,7 -2445,7 +2726,7 @@@ =item B B|B -If enabled (the default) a sum of the CPU usage accross all cores is reported. +If enabled (the default) a sum of the CPU usage across all cores is reported. =item B B|B @@@ -2774,7 -2493,7 +2774,7 @@@ Fan I =item fout -Fan Out +Fan Out =item vccp @@@ -2815,11 -2534,11 +2815,11 @@@ Known power names are =item total0 -Total power utilization averaged over Time Window 0 (uWatts). +Total power utilization averaged over Time Window 0 (uWatts). =item total1 -Total power utilization averaged over Time Window 0 (uWatts). +Total power utilization averaged over Time Window 0 (uWatts). =item inst @@@ -2827,55 -2546,34 +2827,55 @@@ Instantaneous power (uWatts) =item imax -Max instantaneous power (uWatts). +Max instantaneous power (uWatts). =item pcie -PCI-E connector power (uWatts). +PCI-E connector power (uWatts). =item c2x3 -2x3 connector power (uWatts). +2x3 connector power (uWatts). =item c2x4 -2x4 connector power (uWatts). +2x4 connector power (uWatts). =item vccp -Core rail (uVolts). +Core rail (uVolts). =item vddg -Uncore rail (uVolts). +Uncore rail (uVolts). =item vddq -Memory subsystem rail (uVolts). +Memory subsystem rail (uVolts). + +=back =back +=head2 Plugin C + +The I provides the following configuration options: + +=over 4 + +=item B B|B + +Enables or disables reporting of physical memory usage in absolute numbers, +i.e. bytes. Defaults to B. + +=item B B|B + +Enables or disables reporting of physical memory usage in percentages, e.g. +percent of physical memory used. Defaults to B. + +This is useful for deploying I in a heterogeneous environment in +which the sizes of physical memory vary. + =back =head2 Plugin C @@@ -2893,19 -2591,19 +2893,19 @@@ B - + RegisterBase 2 RegisterType float Type voltage Instance "input-2" - + Address "192.168.0.42" Port "502" Interval 60 - + Instance "power-supply" Collect "voltage-input-1" @@@ -3036,11 -2734,9 +3036,11 @@@ Synopsis Password "password" Port "3306" MasterStats true + ConnectTimeout 10 + Alias "squeeze" Host "localhost" Socket "/var/run/mysql/mysqld.sock" SlaveStats true @@@ -3055,11 -2751,6 +3055,11 @@@ section "mysql_real_connect()" in the B =over 4 +=item B I + +Alias to use as sender instead of hostname when reporting. This may be useful +when having cryptic hostnames. + =item B I Hostname of the database server. Defaults to B. @@@ -3098,11 -2789,6 +3098,11 @@@ only has any effect, if B is set Otherwise, use the B option above. See the documentation for the C function for details. +=item B I + +If enabled, metrics about the InnoDB storage engine are collected. +Disabled by default. + =item B I =item B I @@@ -3116,10 -2802,6 +3116,10 @@@ privileges. See the B documentati If enabled, the plugin sends a notification if the replication slave I/O and / or SQL threads are not running. +=item B I + +Sets the connect timeout for the MySQL client. + =back =head2 Plugin C @@@ -3154,7 -2836,7 +3154,7 @@@ Required capabilities are documented be User "username" Password "aef4Aebe" Interval 30 - + Interval 30 GetNameCache true @@@ -3162,12 -2844,12 +3162,12 @@@ GetBufferCache true GetInodeCache true - + Interval 30 GetBusy true - + Interval 30 GetIO "volume0" @@@ -3177,7 -2859,7 +3177,7 @@@ GetLatency "volume0" IgnoreSelectedLatency false - + Interval 30 GetCapacity "vol0" @@@ -3187,15 -2869,15 +3187,15 @@@ GetSnapshot "vol3" IgnoreSelectedSnapshot false - + Interval 60 - + Interval 30 - + Interval 30 GetCPULoad true @@@ -3756,7 -3438,7 +3756,7 @@@ signature) # Export to an internal server # (demonstrates usage without additional options) Server "collectd.internal.tld" - + # Export to an external server # (demonstrates usage with signature options) @@@ -3819,12 -3501,6 +3819,12 @@@ behavior is to let the kernel choose th that the manual selection of an interface for unicast traffic is only necessary in rare cases. +=item B I + +Sets the interval at which to re-resolve the DNS for the I. This is +useful to force a regular DNS lookup to support a high availability setup. If +not specified, re-resolves are never attempted. + =back =item BListen> I [I]B> @@@ -4151,36 -3827,13 +4151,36 @@@ B See notes below The C plugin uses the B library from the B project L to read sensors connected via the onewire bus. -Currently only temperature sensors (sensors with the family code C<10>, -e.Eg. DS1820, DS18S20, DS1920) can be read. If you have other sensors you -would like to have included, please send a sort request to the mailing list. +It can be used in two possible modes - standard or advanced. + +In the standard mode only temperature sensors (sensors with the family code +C<10>, C<22> and C<28> - e.g. DS1820, DS18S20, DS1920) can be read. If you have +other sensors you would like to have included, please send a sort request to +the mailing list. You can select sensors to be read or to be ignored depending +on the option B). When no list is provided the whole bus is +walked and all sensors are read. Hubs (the DS2409 chips) are working, but read the note, why this plugin is experimental, below. +In the advanced mode you can configure any sensor to be read (only numerical +value) using full OWFS path (e.g. "/uncached/10.F10FCA000800/temperature"). +In this mode you have to list all the sensors. Neither default bus walk nor +B are used here. Address and type (file) is extracted from +the path automatically and should produce compatible structure with the "standard" +mode (basically the path is expected as for example +"/uncached/10.F10FCA000800/temperature" where it would extract address part +"F10FCA000800" and the rest after the slash is considered the type - here +"temperature"). +There are two advantages to this mode - you can access virtually any sensor +(not just temperature), select whether to use cached or directly read values +and it is slighlty faster. The downside is more complex configuration. + +The two modes are distinguished automatically by the format of the address. +It is not possible to mix the two modes. Once a full path is detected in any +B then the whole addressing (all sensors) is considered to be this way +(and as standard addresses will fail parsing they will be ignored). + =over 4 =item B I @@@ -4201,23 -3854,14 +4201,23 @@@ This directive is B and does =item B I -Selects sensors to collect or to ignore, depending on B, see -below. Sensors are specified without the family byte at the beginning, to you'd -use C, and B include the leading C<10.> family byte and -point. +In the standard mode selects sensors to collect or to ignore +(depending on B, see below). Sensors are specified without +the family byte at the beginning, so you have to use for example C, +and B include the leading C<10.> family byte and point. +When no B is configured the whole Onewire bus is walked and all supported +sensors (see above) are read. + +In the advanced mode the B specifies full OWFS path - e.g. +C (or when cached values are OK +C). B is not used. + +As there can be multiple devices on the bus you can list multiple sensor (use +multiple B elements). =item B I|I -If no configuration if given, the B plugin will collect data from all +If no configuration is given, the B plugin will collect data from all sensors found. This may not be practical, especially if sensors are added and removed regularly. Sometimes, however, it's easier/preferred to collect only specific sensors or all sensors I a few specified ones. This option @@@ -4225,8 -3869,6 +4225,8 @@@ enables you to do that: By setting B is inverted: All selected interfaces are ignored and all other interfaces are collected. +Used only in the standard mode - see above. + =item B I Sets the interval in which all sensors should be read. If not specified, the @@@ -4245,70 -3887,6 +4245,70 @@@ short: If it works for you: Great! But change, though this is unlikely. Oh, and if you want to help improving this plugin, just send a short notice to the mailing list. ThanksE:) +=head2 Plugin C + +To use the C plugin you first need to configure the I +server correctly. The backend database C needs to be loaded and +working. See slapd-monitor(5) for the details. + +The configuration of the C plugin consists of one or more B +blocks. Each block requires one string argument as the instance name. For +example: + + + + URL "ldap://localhost/" + + + URL "ldaps://localhost/" + + + +The instance name will be used as the I. To emulate the old +(versionE4) behavior, you can use an empty string (""). In order for the +plugin to work correctly, each instance name must be unique. This is not +enforced by the plugin and it is your responsibility to ensure it is. + +The following options are accepted within each B block: + +=over 4 + +=item B I + +Sets the URL to use to connect to the I server. This option is +I. + +=item B B + +Defines whether TLS must be used when connecting to the I server. +Disabled by default. + +=item B B + +Enables or disables peer host name verification. If enabled, the plugin checks +if the C or a C field of the SSL +certificate matches the host name provided by the B option. If this +identity check fails, the connection is aborted. Enabled by default. + +=item B I + +File that holds one or more SSL certificates. If you want to use TLS/SSL you +may possibly need this option. What CA certificates are checked by default +depends on the distribution you use and can be changed with the usual ldap +client configuration mechanisms. See ldap.conf(5) for the details. + +=item B I + +Sets the timeout value for ldap operations. Defaults to B<-1> which results in +an infinite timeout. + +=item B I + +An integer which sets the LDAP protocol version number to use when connecting +to the I server. Defaults to B<3> for using I. + +=back + =head2 Plugin C The OpenVPN plugin reads a status file maintained by OpenVPN and gathers @@@ -4962,15 -4540,18 +4962,15 @@@ Specify the username to be used when co Specify the password to be used when connecting to the server. +=item B I + +Skip expired values in query output. + =item B I|I|I|I Specify whether to use an SSL connection when contacting the server. The following modes are supported: -=item B I - -Specify the plugin instance name that should be used instead of the database -name (which is the default, if this option has not been specified). This -allows to query multiple databases of the same name on the same host (e.g. -when running multiple database server versions in parallel). - =over 4 =item I @@@ -4991,13 -4572,6 +4991,13 @@@ Use SSL only =back +=item B I + +Specify the plugin instance name that should be used instead of the database +name (which is the default, if this option has not been specified). This +allows to query multiple databases of the same name on the same host (e.g. +when running multiple database server versions in parallel). + =item B I Specify the Kerberos service name to use when authenticating with Kerberos 5 @@@ -5337,10 -4911,6 +5337,10 @@@ which configures the connection paramet Host "localhost" Port "6379" Timeout 2000 + + Type "queue_length" + Instance "myqueue" + @@@ -5378,22 -4948,6 +5378,22 @@@ read function is blocking, you should k in mind that the sum of all B values for all B should be lower than B defined globally. +=item B I + +The B block identifies a query to execute against the redis server. +There may be an arbitrary number of queries to execute. + +=item B I + +Within a query definition, a valid collectd type to use as when submitting +the result of the query. When not supplied, will default to B. + +=item B I + +Within a query definition, an optional type instance to use when submitting +the result of the query. When not supplied will default to the escaped +command, up to 64 chars. + =back =head2 Plugin C @@@ -5727,40 -5281,6 +5727,40 @@@ measurements are discarded =back +=head2 Plugin C + +The C plugin collects SMART information from physical +disks. Values collectd include temperature, power cycle count, poweron +time and bad sectors. Also, all SMART attributes are collected along +with the normalized current value, the worst value, the threshold and +a human readable value. + +Using the following two options you can ignore some disks or configure the +collection only of specific disks. + +=over 4 + +=item B I + +Select the disk I. Whether it is collected or ignored depends on the +B setting, see below. As with other plugins that use the +daemon's ignorelist functionality, a string that starts and ends with a slash +is interpreted as a regular expression. Examples: + + Disk "sdd" + Disk "/hda[34]/" + +=item B B|B + +Sets whether selected disks, i.Ee. the ones matches by any of the B +statements, are ignored or if all other disks are ignored. The behavior +(hopefully) is intuitive: If no B option is configured, all disks are +collected. If at least one B option is given and no B or +set to B, B matching disks will be collected. If B +is set to B, all disks are collected B the ones matched. + +=back + =head2 Plugin C Since the configuration of the C is a little more complicated than @@@ -5812,7 -5332,20 +5812,20 @@@ that I of all reported timers computed latency. This is useful for cutting off the long tail latency, as it's often done in I (SLAs). - If not specified, no percentile is calculated / dispatched. + Different percentiles can be calculated by setting this option several times. + If none are specified, no percentiles are calculated / dispatched. + + =item B B|B + + =item B B|B + + =item B B|B + + =item B B|B + + Calculate and dispatch various values out of I metrics received during + an interval. If set to B, the default, these values aren't calculated / + dispatched. =back @@@ -5838,19 -5371,6 +5851,19 @@@ This option is only available if the I< When enabled, the I is reported in bytes. When disabled, the default, I is reported in pages. This option is available under Linux only. +=item B B|B + +Enables or disables reporting of absolute swap metrics, i.e. number of I +available and used. Defaults to B. + +=item B B|B + +Enables or disables reporting of relative swap metrics, i.e. I +available and free. Defaults to B. + +This is useful for deploying I in a heterogeneous environment, where +swap sizes differ and you want to specify generic thresholds or similar. + =back =head2 Plugin C @@@ -5986,7 -5506,6 +5999,7 @@@ user using (extended) regular expressio Instance "exim" + Interval 60 Regex "S=([1-9][0-9]*)" DSType "CounterAdd" @@@ -6013,9 -5532,6 +6026,9 @@@ This plugin instance is for all B option. This way you can extract several plugin instances from one logfile, handy when parsing syslog and the like. +The B option allows you to define the length of time between reads. If +this is not set, the default Interval will be used. + Each B block has the following options to describe how the match should be performed: @@@ -6298,11 -5814,6 +6311,11 @@@ connections a mail server or news serve how many connections a web proxy holds to web servers. You have to give the port in numeric form. +=item B I|I + +If this option is set to I a summary of statistics from all connections +are collectd. This option defaults to I. + =back =head2 Plugin C @@@ -6432,35 -5943,22 +6445,35 @@@ Take the UUID from the given file (defa =head2 Plugin C The I collects information about Varnish, an HTTP accelerator. +It collects a subset of the values displayed by L, and +organizes them in categories which can be enabled or disabled. Currently only +metrics shown in L's I
section are collected. The exact +meaning of each metric can be found in L. Synopsis: + CollectBackend true + CollectBan false CollectCache true CollectConnections true - CollectBackend true - CollectSHM true + CollectDirectorDNS false CollectESI false CollectFetch false CollectHCB false + CollectObjects false + CollectPurge false + CollectSession false + CollectSHM true CollectSMA false CollectSMS false CollectSM false + CollectStruct false CollectTotals false + CollectUptime false + CollectVCL false + CollectVSM false CollectWorkers false @@@ -6474,24 -5972,29 +6487,24 @@@ Inside each EBE block =over 4 -=item B B|B - -Cache hits and misses. True by default. - -=item B B|B - -Number of client connections received, accepted and dropped. True by default. - =item B B|B Back-end connection statistics, such as successful, reused, and closed connections. True by default. -=item B B|B - -Statistics about the shared memory log, a memory region to store -log messages which is flushed to disk when full. True by default. - =item B B|B Statistics about ban operations, such as number of bans added, retired, and number of objects tested against ban operations. Only available with Varnish -3.x. False by default. +3.x and above. False by default. + +=item B B|B + +Cache hits and misses. True by default. + +=item B B|B + +Number of client connections received, accepted and dropped. True by default. =item B B|B @@@ -6525,14 -6028,7 +6538,14 @@@ number of objects tested against purge =item B B|B Client session statistics. Number of past and current sessions, session herd and -linger counters, etc. False by default. +linger counters, etc. False by default. Note that if using Varnish 4.x, some +metrics found in the Connections and Threads sections with previous versions of +Varnish have been moved here. + +=item B B|B + +Statistics about the shared memory log, a memory region to store +log messages which is flushed to disk when full. True by default. =item B B|B @@@ -6563,118 -6059,18 +6576,118 @@@ the number of requests and bytes transf =item B B|B -Varnish uptime. False by default. +Varnish uptime. Only available with Varnish 3.x and above. False by default. =item B B|B Number of total (available + discarded) VCL (config files). False by default. +=item B B|B + +Collect statistics about Varnish's shared memory usage (used by the logging and +statistics subsystems). Only available with Varnish 4.x. False by default. + =item B B|B Collect statistics about worker threads. False by default. =back +=head2 Plugin C + +This plugin allows CPU, disk and network load to be collected for virtualized +guests on the machine. This means that these metrics can be collected for guest +systems without installing any software on them - I only runs on the +host system. The statistics are collected through libvirt +(L). + +Only I is required. + +=over 4 + +=item B I + +Connect to the hypervisor given by I. For example if using Xen use: + + Connection "xen:///" + +Details which URIs allowed are given at L. + +=item B I + +Refresh the list of domains and devices every I. The default is 60 +seconds. Setting this to be the same or smaller than the I will cause +the list of domains and devices to be refreshed on every iteration. + +Refreshing the devices in particular is quite a costly operation, so if your +virtualization setup is static you might consider increasing this. If this +option is set to 0, refreshing is disabled completely. + +=item B I + +=item B I + +=item B I + +=item B B|B + +Select which domains and devices are collected. + +If I is not given or B then only the listed domains and +disk/network devices are collected. + +If I is B then the test is reversed and the listed +domains and disk/network devices are ignored, while the rest are collected. + +The domain name and device names may use a regular expression, if the name is +surrounded by I and collectd was compiled with support for regexps. + +The default is to collect statistics for all domains and all their devices. + +Example: + + BlockDevice "/:hdb/" + IgnoreSelected "true" + +Ignore all I devices on any domain, but other block devices (eg. I) +will be collected. + +=item B B + +When the virt plugin logs data, it sets the hostname of the collected data +according to this setting. The default is to use the guest name as provided by +the hypervisor, which is equal to setting B. + +B means use the guest's UUID. This is useful if you want to track the +same guest across migrations. + +B means to use the global B setting, which is probably not +useful on its own because all guests will appear to have the same name. + +You can also specify combinations of these fields. For example B +means to concatenate the guest name and UUID (with a literal colon character +between, thus I<"foo:1234-1234-1234-1234">). + +=item B B|B
+ +When the virt plugin logs interface data, it sets the name of the collected +data according to this setting. The default is to use the path as provided by +the hypervisor (the "dev" property of the target node), which is equal to +setting B. + +B
means use the interface's mac address. This is useful since the +interface path might change between reboots of a guest or across migrations. + +=item B B + +When the virt plugin logs data, it sets the plugin_instance of the collected +data according to this setting. The default is to use the guest name as provided +by the hypervisor, which is equal to setting B. + +B means use the guest's UUID. + +=back + =head2 Plugin C The C plugin collects information about the usage of virtual memory. @@@ -6791,59 -6187,6 +6804,59 @@@ more than one DS =back +=head2 Plugin C + +The C plugin writes data to I, a scalable open-source +time series database. The plugin connects to a I, a masterless, no shared +state daemon that ingests metrics and stores them in HBase. The plugin uses +I over the "line based" protocol with a default port 4242. The data will +be sent in blocks of at most 1428 bytes to minimize the number of network +packets. + +Synopsis: + + + + Host "tsd-1.my.domain" + Port "4242" + HostTags "status=production" + + + +The configuration consists of one or more EBEIE +blocks. Inside the B blocks, the following options are recognized: + +=over 4 + +=item B I
+ +Hostname or address to connect to. Defaults to C. + +=item B I + +Service name or port number to connect to. Defaults to C<4242>. + + +=item B I + +When set, I is added to the end of the metric. It is intended to be +used for name=value pairs that the TSD will tag the metric with. Dots and +whitespace are I escaped in this string. + +=item B B|B + +If set to B, convert counter values to rates. If set to B +(the default) counter values are stored as is, as an increasing +integer number. + +=item B B|B + +If set the B, append the name of the I (DS) to the "metric" +identifier. If set to B (the default), this is only done when there is +more than one DS. + +=back + =head2 Plugin C The I will send values to I, a schema-less @@@ -6899,9 -6242,8 +6912,9 @@@ want to use authentication all three fi =head2 Plugin C -This output plugin submits values to an http server by POST them using the -PUTVAL plain-text protocol. Each destination you want to post data to needs to +This output plugin submits values to an HTTP server using POST requests and +encoding metrics with JSON or using the C command described in +L. Each destination you want to post data to needs to have one B block, within which the destination can be configured further, for example by specifying authentication data. @@@ -6911,7 -6253,6 +6924,7 @@@ Synopsis User "collectd" Password "weCh3ik0" + Format JSON @@@ -6947,33 -6288,6 +6960,33 @@@ File that holds one or more SSL certifi possibly need this option. What CA certificates come bundled with C and are checked by default depends on the distribution you use. +=item B I + +Directory holding one or more CA certificate files. You can use this if for +some reason all the needed CA certificates aren't in the same file and can't be +pointed to using the B option. Requires C to be built against +OpenSSL. + +=item B I + +File that holds the private key in PEM format to be used for certificate-based +authentication. + +=item B I + +File that holds the SSL certificate to be used for certificate-based +authentication. + +=item B I + +Password required to load the private key in B. + +=item B B|B|B|B|B|B + +Define which SSL protocol version must be used. By default C will +attempt to figure out the remote SSL protocol version. See +L for more details. + =item B B|B Format of the output to generate. If set to B, will create output that @@@ -6985,129 -6299,69 +6998,184 @@@ Defaults to B =item B B If set to B, convert counter values to rates. If set to B (the -default) counter values are stored as is, i.Ee. as an increasing integer -number. +default) counter values are stored as is, i.e. as an increasing integer number. + +=item B I + +Sets the send buffer size to I. By increasing this buffer, less HTTP +requests will be generated, but more metrics will be batched / metrics are +cached for longer before being sent, introducing additional delay until they +are available on the server side. I must be at least 1024 and cannot +exceed the size of an C, i.e. 2EGByte. +Defaults to C<4096>. + +=back + +=head2 Plugin C + +The I will send values to a I topic, a distributed +queue. +Synopsis: + + + Property "metadata.broker.list" "broker1:9092,broker2:9092" + + Format JSON + + + +The following options are understood by the I: + +=over 4 + +=item EB IE + +The plugin's configuration consists of one or more B blocks. Each block +is given a unique I and specifies one kafka producer. +Inside the B block, the following per-topic options are +understood: + +=over 4 + +=item B I I + +Configure the named property for the current topic. Properties are +forwarded to the kafka producer library B. + +=item B I + +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 can be used to specify that an arbitrary partition should +be used. + +=item B B|B|B + +Selects the format in which messages are sent to the broker. If set to +B (the default), values are sent as C commands which are +identical to the syntax used by the I and I. + +If set to B, the values are encoded in the I, +an easy and straight forward exchange format. + +If set to B, values are encoded in the I format, which is +CmetricE EvalueE EtimestampE\n>. + +=item B B|B + +Determines whether or not C, C and C data sources +are converted to a I (i.e. a C value). If set to B (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 option has +been set to B. + +=item B (B=I only) + +A prefix can be added in the metric name when outputting in the I +format. It's added before the I name. +Metric name will be +CprefixEEhostEEpostfixEEpluginEEtypeEEnameE> + +=item B (B=I only) + +A postfix can be added in the metric name when outputting in the I +format. It's added after the I name. +Metric name will be +CprefixEEhostEEpostfixEEpluginEEtypeEEnameE> + +=item B (B=I only) + +Specify a character to replace dots (.) in the host part of the metric name. +In I metric name, dots are used as separators between different +metric parts (host, plugin, type). +Default is C<_> (I). + +=item B B|B + +If set to B, the plugin instance and type instance will be in their own +path component, for example C. If set to B (the +default), the plugin and plugin instance (and likewise the type and type +instance) are put into one component, for example C. + +=item B B|B + +If set to B (the default), convert counter values to rates. If set to +B counter values are stored as is, i.e. as an increasing integer number. + +This will be reflected in the C tag: If B is enabled, +converted values will have "rate" appended to the data source type, e.g. +C. + +=back + +=item B I I + +Configure the kafka producer through properties, you almost always will +want to set B to your Kafka broker list. =back + =head2 Plugin C + + The I submits values to I, a data structure server. + + Synopsis: + + + + Host "localhost" + Port "6379" + Timeout 1000 + + + + Values are submitted to I, using the metric name as the key, and + the timestamp as the score. Retrieving a date range can then be done using the + C I command. Additionnally, all the identifiers of these + I are kept in a I called C and can be + retrieved using the C I command. See + L and L for + details. + + The information shown in the synopsis above is the I + which is used by the plugin if no configuration is present. + + The plugin can send values to multiple instances of I by specifying + one B block for each instance. Within the B blocks, the following + options are available: + + =over 4 + + =item B I + + The B block identifies a new I node, that is a new I + instance running in an specified host and port. The name for node is a + canonical identifier which is used as I. It is limited to + 64Echaracters in length. + + =item B I + + The B option is the hostname or IP-address where the I instance is + running on. + + =item B I + + The B option is the TCP port on which the Redis instance accepts + 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 I + + The B option sets the socket connection timeout, in milliseconds. + + =back + =head2 Plugin C -The I will send values to I, a powerfull stream +The I will send values to I, a powerful stream aggregation and monitoring system. The plugin sends I encoded data to I using UDP packets. @@@ -7123,7 -6377,6 +7191,7 @@@ Synopsis TTLFactor 2.0 Tag "foobar" + Attribute "foo" "bar" The following options are understood by the I: @@@ -7152,20 -6405,6 +7220,20 @@@ Service name or port number to connect Specify the protocol to use when communicating with I. Defaults to B. +=item B B|B + +If set to B and B is set to B, +events will be batched in memory and flushed at +regular intervals or when B is exceeded. + +Notifications are not batched and sent as soon as possible. + +Defaults to false + +=item B I + +Maximum payload size for a riemann packet. Defaults to 8192 + =item B B|B If set to B (the default), convert counter values to rates. If set to @@@ -7191,23 -6430,6 +7259,23 @@@ interval is multiplied to set the TTL. know exactly what you're doing, you should only increase this setting from its default value. +=item B B|B + +If set to B, create riemann events for notifications. This is B +by default. When processing thresholds from write_riemann, it might prove +useful to avoid getting notification events. + +=item B B|B + +If set to B, attach state to events based on thresholds defined +in the B plugin. Defaults to B. + +=item B I + +Add the given string as a prefix to the event service name. +If B not set or set to an empty string (""), +no prefix will be used. + =back =item B I @@@ -7215,11 -6437,6 +7283,11 @@@ Add the given string as an additional tag to the metric being sent to I. +=item B I I + +Consider the two given strings to be the key and value of an additional +attribute for each metric being sent out to I. + =back =head1 THRESHOLD CONFIGURATION @@@ -7891,7 -7108,7 +7959,7 @@@ Example Max 100 Satisfy "All" - + # Match if the value of any data source is outside the range of 0 - 100. Min 0 @@@ -8073,7 -7290,7 +8141,7 @@@ Example # Replace "example.net" with "example.com" Host "\\" "example.com" - + # Strip "www." from hostnames Host "\\ @@@ -8156,6 -7373,6 +8224,6 @@@ Locto@verplant.orgE +Florian Forster Eocto@collectd.orgE =cut diff --combined src/cpu.c index d1978f96,1c4e5f6b..2fb4918b --- a/src/cpu.c +++ b/src/cpu.c @@@ -1,10 -1,9 +1,10 @@@ /** * collectd - src/cpu.c - * Copyright (C) 2005-2010 Florian octo Forster + * Copyright (C) 2005-2014 Florian octo Forster * Copyright (C) 2008 Oleg King * Copyright (C) 2009 Simon Kuhnle * Copyright (C) 2009 Manuel Sanmartin + * Copyright (C) 2013-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 @@@ -20,11 -19,10 +20,11 @@@ * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA * * Authors: - * Florian octo Forster + * Florian octo Forster * Oleg King * Simon Kuhnle * Manuel Sanmartin + * Pierre-Yves Ritschard **/ #include "collectd.h" @@@ -88,18 -86,6 +88,18 @@@ # define CAN_USE_SYSCTL 0 #endif +#define CPU_STATE_USER 0 +#define CPU_STATE_SYSTEM 1 +#define CPU_STATE_WAIT 2 +#define CPU_STATE_NICE 3 +#define CPU_STATE_SWAP 4 +#define CPU_STATE_INTERRUPT 5 +#define CPU_STATE_SOFTIRQ 6 +#define CPU_STATE_STEAL 7 +#define CPU_STATE_IDLE 8 +#define CPU_STATE_ACTIVE 9 /* sum of (!idle) */ +#define CPU_STATE_MAX 10 /* #states */ + #if HAVE_STATGRAB_H # include #endif @@@ -114,29 -100,10 +114,23 @@@ # error "No applicable input method." #endif +static const char *cpu_state_names[] = { + "user", + "system", + "wait", + "nice", + "swap", + "interrupt", + "softirq", + "steal", + "idle", + "active" +}; + #ifdef PROCESSOR_CPU_LOAD_INFO static mach_port_t port_host; static processor_port_array_t cpu_list; static mach_msg_type_number_t cpu_list_len; - - #if PROCESSOR_TEMPERATURE - static int cpu_temp_retry_counter = 0; - static int cpu_temp_retry_step = 1; - static int cpu_temp_retry_max = 1; - #endif /* PROCESSOR_TEMPERATURE */ /* #endif PROCESSOR_CPU_LOAD_INFO */ #elif defined(KERNEL_LINUX) @@@ -172,57 -139,9 +166,57 @@@ static int numcpu static int pnumcpu; #endif /* HAVE_PERFSTAT */ +#define RATE_ADD(sum, val) do { \ + if (isnan (sum)) \ + (sum) = (val); \ + else if (!isnan (val)) \ + (sum) += (val); \ +} while (0) + +struct cpu_state_s +{ + value_to_rate_state_t conv; + gauge_t rate; + _Bool has_value; +}; +typedef struct cpu_state_s cpu_state_t; + +static cpu_state_t *cpu_states = NULL; +static size_t cpu_states_num = 0; /* #cpu_states allocated */ + +/* Highest CPU number in the current iteration. Used by the dispatch logic to + * determine how many CPUs there were. Reset to 0 by cpu_reset(). */ +static size_t global_cpu_num = 0; + +static _Bool report_by_cpu = 1; +static _Bool report_by_state = 1; +static _Bool report_percent = 0; + +static const char *config_keys[] = +{ + "ReportByCpu", + "ReportByState", + "ValuesPercentage" +}; +static int config_keys_num = STATIC_ARRAY_SIZE (config_keys); + +static int cpu_config (char const *key, char const *value) /* {{{ */ +{ + if (strcasecmp (key, "ReportByCpu") == 0) + report_by_cpu = IS_TRUE (value) ? 1 : 0; + else if (strcasecmp (key, "ValuesPercentage") == 0) + report_percent = IS_TRUE (value) ? 1 : 0; + else if (strcasecmp (key, "ReportByState") == 0) + report_by_state = IS_TRUE (value) ? 1 : 0; + else + return (-1); + + return (0); +} /* }}} int cpu_config */ + static int init (void) { - #if PROCESSOR_CPU_LOAD_INFO || PROCESSOR_TEMPERATURE + #if PROCESSOR_CPU_LOAD_INFO kern_return_t status; port_host = mach_host_self (); @@@ -316,294 -235,48 +310,288 @@@ return (0); } /* int init */ -static void submit (int cpu_num, const char *type_instance, derive_t value) +static void submit_value (int cpu_num, int cpu_state, const char *type, value_t value) { value_t values[1]; value_list_t vl = VALUE_LIST_INIT; - values[0].derive = value; + memcpy(&values[0], &value, sizeof(value)); vl.values = values; vl.values_len = 1; + sstrncpy (vl.host, hostname_g, sizeof (vl.host)); sstrncpy (vl.plugin, "cpu", sizeof (vl.plugin)); - ssnprintf (vl.plugin_instance, sizeof (vl.plugin_instance), - "%i", cpu_num); - sstrncpy (vl.type, "cpu", sizeof (vl.type)); - sstrncpy (vl.type_instance, type_instance, sizeof (vl.type_instance)); + sstrncpy (vl.type, type, sizeof (vl.type)); + sstrncpy (vl.type_instance, cpu_state_names[cpu_state], + sizeof (vl.type_instance)); + if (cpu_num >= 0) { + ssnprintf (vl.plugin_instance, sizeof (vl.plugin_instance), + "%i", cpu_num); + } plugin_dispatch_values (&vl); } +static void submit_percent(int cpu_num, int cpu_state, gauge_t percent) +{ + value_t value; + + /* This function is called for all known CPU states, but each read + * method will only report a subset. The remaining states are left as + * NAN and we ignore them here. */ + if (isnan (percent)) + return; + + value.gauge = percent; + submit_value (cpu_num, cpu_state, "percent", value); +} + +static void submit_derive(int cpu_num, int cpu_state, derive_t derive) +{ + value_t value; + + value.derive = derive; + submit_value (cpu_num, cpu_state, "cpu", value); +} + +/* Takes the zero-index number of a CPU and makes sure that the module-global + * cpu_states buffer is large enough. Returne ENOMEM on erorr. */ +static int cpu_states_alloc (size_t cpu_num) /* {{{ */ +{ + cpu_state_t *tmp; + size_t sz; + + sz = (((size_t) cpu_num) + 1) * CPU_STATE_MAX; + assert (sz > 0); + + /* We already have enough space. */ + if (cpu_states_num >= sz) + return 0; + + tmp = realloc (cpu_states, sz * sizeof (*cpu_states)); + if (tmp == NULL) + { + ERROR ("cpu plugin: realloc failed."); + return (ENOMEM); + } + cpu_states = tmp; + tmp = cpu_states + cpu_states_num; + + memset (tmp, 0, (sz - cpu_states_num) * sizeof (*cpu_states)); + cpu_states_num = sz; + return 0; +} /* }}} cpu_states_alloc */ + +static cpu_state_t *get_cpu_state (size_t cpu_num, size_t state) /* {{{ */ +{ + size_t index = ((cpu_num * CPU_STATE_MAX) + state); + + if (index >= cpu_states_num) + return (NULL); + + return (&cpu_states[index]); +} /* }}} cpu_state_t *get_cpu_state */ + +/* Populates the per-CPU CPU_STATE_ACTIVE rate and the global rate_by_state + * array. */ +static void aggregate (gauge_t *sum_by_state) /* {{{ */ +{ + size_t cpu_num; + size_t state; + + for (state = 0; state < CPU_STATE_MAX; state++) + sum_by_state[state] = NAN; + + for (cpu_num = 0; cpu_num < global_cpu_num; cpu_num++) + { + cpu_state_t *this_cpu_states = get_cpu_state (cpu_num, 0); + + this_cpu_states[CPU_STATE_ACTIVE].rate = NAN; + + for (state = 0; state < CPU_STATE_ACTIVE; state++) + { + if (!this_cpu_states[state].has_value) + continue; + + RATE_ADD (sum_by_state[state], this_cpu_states[state].rate); + if (state != CPU_STATE_IDLE) + RATE_ADD (this_cpu_states[CPU_STATE_ACTIVE].rate, this_cpu_states[state].rate); + } + + RATE_ADD (sum_by_state[CPU_STATE_ACTIVE], this_cpu_states[CPU_STATE_ACTIVE].rate); + } +} /* }}} void aggregate */ + +/* Commits (dispatches) the values for one CPU or the global aggregation. + * cpu_num is the index of the CPU to be committed or -1 in case of the global + * aggregation. rates is a pointer to CPU_STATE_MAX gauge_t values holding the + * current rate; each rate may be NAN. Calculates the percentage of each state + * and dispatches the metric. */ +static void cpu_commit_one (int cpu_num, /* {{{ */ + gauge_t rates[static CPU_STATE_MAX]) +{ + size_t state; + gauge_t sum; + + sum = rates[CPU_STATE_ACTIVE]; + RATE_ADD (sum, rates[CPU_STATE_IDLE]); + + if (!report_by_state) + { + gauge_t percent = 100.0 * rates[CPU_STATE_ACTIVE] / sum; + submit_percent (cpu_num, CPU_STATE_ACTIVE, percent); + return; + } + + for (state = 0; state < CPU_STATE_ACTIVE; state++) + { + gauge_t percent = 100.0 * rates[state] / sum; + submit_percent (cpu_num, state, percent); + } +} /* }}} void cpu_commit_one */ + +/* Resets the internal aggregation. This is called by the read callback after + * each iteration / after each call to cpu_commit(). */ +static void cpu_reset (void) /* {{{ */ +{ + size_t i; + + for (i = 0; i < cpu_states_num; i++) + cpu_states[i].has_value = 0; + + global_cpu_num = 0; +} /* }}} void cpu_reset */ + +/* Legacy behavior: Dispatches the raw derive values without any aggregation. */ +static void cpu_commit_without_aggregation (void) /* {{{ */ +{ + int state; + + for (state = 0; state < CPU_STATE_ACTIVE; state++) + { + size_t cpu_num; + if (report_by_cpu) { + for (cpu_num = 0; cpu_num < global_cpu_num; cpu_num++) + { + cpu_state_t *s = get_cpu_state (cpu_num, state); + + if (!s->has_value) + continue; + + submit_derive ((int) cpu_num, (int) state, s->conv.last_value.derive); + } + } else { + derive_t derive_total = 0; + for (cpu_num = 0; cpu_num < global_cpu_num; cpu_num++) + { + cpu_state_t *s = get_cpu_state (cpu_num, state); + + if (!s->has_value) + continue; + + derive_total += s->conv.last_value.derive; + + } + submit_derive (-1, (int) state, derive_total); + } + } +} /* }}} void cpu_commit_without_aggregation */ + +/* Aggregates the internal state and dispatches the metrics. */ +static void cpu_commit (void) /* {{{ */ +{ + gauge_t global_rates[CPU_STATE_MAX] = { + NAN, NAN, NAN, NAN, NAN, NAN, NAN, NAN, NAN, NAN + }; + size_t cpu_num; + + if (report_by_state && !report_percent) + { + cpu_commit_without_aggregation (); + return; + } + + aggregate (global_rates); + + if (!report_by_cpu) + { + cpu_commit_one (-1, global_rates); + return; + } + + for (cpu_num = 0; cpu_num < global_cpu_num; cpu_num++) + { + cpu_state_t *this_cpu_states = get_cpu_state (cpu_num, 0); + gauge_t local_rates[CPU_STATE_MAX] = { + NAN, NAN, NAN, NAN, NAN, NAN, NAN, NAN, NAN, NAN + }; + size_t state; + + for (state = 0; state < CPU_STATE_ACTIVE; state++) + if (this_cpu_states[state].has_value) + local_rates[state] = this_cpu_states[state].rate; + + cpu_commit_one ((int) cpu_num, local_rates); + } +} /* }}} void cpu_commit */ + +/* Adds a derive value to the internal state. This should be used by each read + * function for each state. At the end of the iteration, the read function + * should call cpu_commit(). */ +static int cpu_stage (size_t cpu_num, size_t state, derive_t value, cdtime_t now) /* {{{ */ +{ + int status; + cpu_state_t *s; + value_t v; + + if (state >= CPU_STATE_ACTIVE) + return (EINVAL); + + status = cpu_states_alloc (cpu_num); + if (status != 0) + return (status); + + if (global_cpu_num <= cpu_num) + global_cpu_num = cpu_num + 1; + + s = get_cpu_state (cpu_num, state); + + v.gauge = NAN; + status = value_to_rate (&v, value, &s->conv, DS_TYPE_DERIVE, now); + if (status != 0) + return (status); + + s->rate = v.gauge; + s->has_value = 1; + return (0); +} /* }}} int cpu_stage */ + static int cpu_read (void) { -#if PROCESSOR_CPU_LOAD_INFO + cdtime_t now = cdtime (); + - #if PROCESSOR_CPU_LOAD_INFO || PROCESSOR_TEMPERATURE /* {{{ */ ++#if PROCESSOR_CPU_LOAD_INFO /* {{{ */ int cpu; kern_return_t status; - + - #if PROCESSOR_CPU_LOAD_INFO processor_cpu_load_info_data_t cpu_info; mach_msg_type_number_t cpu_info_len; - #endif - #if PROCESSOR_TEMPERATURE - processor_info_data_t cpu_temp; - mach_msg_type_number_t cpu_temp_len; - #endif host_t cpu_host; for (cpu = 0; cpu < cpu_list_len; cpu++) { - #if PROCESSOR_CPU_LOAD_INFO cpu_host = 0; cpu_info_len = PROCESSOR_BASIC_INFO_COUNT; - if ((status = processor_info (cpu_list[cpu], - PROCESSOR_CPU_LOAD_INFO, &cpu_host, - (processor_info_t) &cpu_info, &cpu_info_len)) != KERN_SUCCESS) + status = processor_info (cpu_list[cpu], PROCESSOR_CPU_LOAD_INFO, &cpu_host, + (processor_info_t) &cpu_info, &cpu_info_len); + if (status != KERN_SUCCESS) { - ERROR ("cpu plugin: processor_info failed with status %i", (int) status); + ERROR ("cpu plugin: processor_info (PROCESSOR_CPU_LOAD_INFO) failed: %s", + mach_error_string (status)); continue; } @@@ -613,59 -286,17 +601,15 @@@ continue; } - submit (cpu, "user", (derive_t) cpu_info.cpu_ticks[CPU_STATE_USER]); - submit (cpu, "nice", (derive_t) cpu_info.cpu_ticks[CPU_STATE_NICE]); - submit (cpu, "system", (derive_t) cpu_info.cpu_ticks[CPU_STATE_SYSTEM]); - submit (cpu, "idle", (derive_t) cpu_info.cpu_ticks[CPU_STATE_IDLE]); + cpu_stage (cpu, CPU_STATE_USER, (derive_t) cpu_info.cpu_ticks[CPU_STATE_USER], now); + cpu_stage (cpu, CPU_STATE_NICE, (derive_t) cpu_info.cpu_ticks[CPU_STATE_NICE], now); + cpu_stage (cpu, CPU_STATE_SYSTEM, (derive_t) cpu_info.cpu_ticks[CPU_STATE_SYSTEM], now); + cpu_stage (cpu, CPU_STATE_IDLE, (derive_t) cpu_info.cpu_ticks[CPU_STATE_IDLE], now); - #endif /* PROCESSOR_CPU_LOAD_INFO */ - - #if PROCESSOR_TEMPERATURE - /* - * Not all Apple computers do have this ability. To minimize - * the messages sent to the syslog we do an exponential - * stepback if `processor_info' fails. We still try ~once a day - * though.. - */ - if (cpu_temp_retry_counter > 0) - { - cpu_temp_retry_counter--; - continue; - } - - cpu_temp_len = PROCESSOR_INFO_MAX; - - status = processor_info (cpu_list[cpu], - PROCESSOR_TEMPERATURE, - &cpu_host, - cpu_temp, &cpu_temp_len); - if (status != KERN_SUCCESS) - { - ERROR ("cpu plugin: processor_info failed: %s", - mach_error_string (status)); - - cpu_temp_retry_counter = cpu_temp_retry_step; - cpu_temp_retry_step *= 2; - if (cpu_temp_retry_step > cpu_temp_retry_max) - cpu_temp_retry_step = cpu_temp_retry_max; - - continue; - } - - if (cpu_temp_len != 1) - { - DEBUG ("processor_info (PROCESSOR_TEMPERATURE) returned %i elements..?", - (int) cpu_temp_len); - continue; - } - - cpu_temp_retry_counter = 0; - cpu_temp_retry_step = 1; - #endif /* PROCESSOR_TEMPERATURE */ } -/* #endif PROCESSOR_CPU_LOAD_INFO */ +/* }}} #endif PROCESSOR_CPU_LOAD_INFO */ -#elif defined(KERNEL_LINUX) +#elif defined(KERNEL_LINUX) /* {{{ */ int cpu; - derive_t user, nice, syst, idle; - derive_t wait, intr, sitr; /* sitr == soft interrupt */ FILE *fh; char buf[1024]; @@@ -692,27 -323,37 +636,27 @@@ continue; cpu = atoi (fields[0] + 3); - user = atoll (fields[1]); - nice = atoll (fields[2]); - syst = atoll (fields[3]); - idle = atoll (fields[4]); - submit (cpu, "user", user); - submit (cpu, "nice", nice); - submit (cpu, "system", syst); - submit (cpu, "idle", idle); + cpu_stage (cpu, CPU_STATE_USER, (derive_t) atoll(fields[1]), now); + cpu_stage (cpu, CPU_STATE_NICE, (derive_t) atoll(fields[2]), now); + cpu_stage (cpu, CPU_STATE_SYSTEM, (derive_t) atoll(fields[3]), now); + cpu_stage (cpu, CPU_STATE_IDLE, (derive_t) atoll(fields[4]), now); if (numfields >= 8) { - wait = atoll (fields[5]); - intr = atoll (fields[6]); - sitr = atoll (fields[7]); - - submit (cpu, "wait", wait); - submit (cpu, "interrupt", intr); - submit (cpu, "softirq", sitr); + cpu_stage (cpu, CPU_STATE_WAIT, (derive_t) atoll(fields[5]), now); + cpu_stage (cpu, CPU_STATE_INTERRUPT, (derive_t) atoll(fields[6]), now); + cpu_stage (cpu, CPU_STATE_SOFTIRQ, (derive_t) atoll(fields[7]), now); if (numfields >= 9) - submit (cpu, "steal", atoll (fields[8])); + cpu_stage (cpu, CPU_STATE_STEAL, (derive_t) atoll(fields[8]), now); } } - fclose (fh); -/* #endif defined(KERNEL_LINUX) */ +/* }}} #endif defined(KERNEL_LINUX) */ -#elif defined(HAVE_LIBKSTAT) +#elif defined(HAVE_LIBKSTAT) /* {{{ */ int cpu; - derive_t user, syst, idle, wait; static cpu_stat_t cs; if (kc == NULL) @@@ -723,14 -364,19 +667,14 @@@ if (kstat_read (kc, ksp[cpu], &cs) == -1) continue; /* error message? */ - idle = (derive_t) cs.cpu_sysinfo.cpu[CPU_IDLE]; - user = (derive_t) cs.cpu_sysinfo.cpu[CPU_USER]; - syst = (derive_t) cs.cpu_sysinfo.cpu[CPU_KERNEL]; - wait = (derive_t) cs.cpu_sysinfo.cpu[CPU_WAIT]; - - submit (ksp[cpu]->ks_instance, "user", user); - submit (ksp[cpu]->ks_instance, "system", syst); - submit (ksp[cpu]->ks_instance, "idle", idle); - submit (ksp[cpu]->ks_instance, "wait", wait); + cpu_stage (ksp[cpu]->ks_instance, CPU_STATE_IDLE, (derive_t) cs.cpu_sysinfo.cpu[CPU_IDLE], now); + cpu_stage (ksp[cpu]->ks_instance, CPU_STATE_USER, (derive_t) cs.cpu_sysinfo.cpu[CPU_USER], now); + cpu_stage (ksp[cpu]->ks_instance, CPU_STATE_SYSTEM, (derive_t) cs.cpu_sysinfo.cpu[CPU_KERNEL], now); + cpu_stage (ksp[cpu]->ks_instance, CPU_STATE_WAIT, (derive_t) cs.cpu_sysinfo.cpu[CPU_WAIT], now); } -/* #endif defined(HAVE_LIBKSTAT) */ +/* }}} #endif defined(HAVE_LIBKSTAT) */ -#elif CAN_USE_SYSCTL +#elif CAN_USE_SYSCTL /* {{{ */ uint64_t cpuinfo[numcpu][CPUSTATES]; size_t cpuinfo_size; int status; @@@ -786,15 -432,14 +730,15 @@@ } for (i = 0; i < numcpu; i++) { - submit (i, "user", cpuinfo[i][CP_USER]); - submit (i, "nice", cpuinfo[i][CP_NICE]); - submit (i, "system", cpuinfo[i][CP_SYS]); - submit (i, "idle", cpuinfo[i][CP_IDLE]); - submit (i, "interrupt", cpuinfo[i][CP_INTR]); + cpu_stage (i, CPU_STATE_USER, (derive_t) cpuinfo[i][CP_USER], now); + cpu_stage (i, CPU_STATE_NICE, (derive_t) cpuinfo[i][CP_NICE], now); + cpu_stage (i, CPU_STATE_SYSTEM, (derive_t) cpuinfo[i][CP_SYS], now); + cpu_stage (i, CPU_STATE_IDLE, (derive_t) cpuinfo[i][CP_IDLE], now); + cpu_stage (i, CPU_STATE_INTERRUPT, (derive_t) cpuinfo[i][CP_INTR], now); } -/* #endif CAN_USE_SYSCTL */ -#elif defined(HAVE_SYSCTLBYNAME) && defined(HAVE_SYSCTL_KERN_CP_TIMES) +/* }}} #endif CAN_USE_SYSCTL */ + +#elif defined(HAVE_SYSCTLBYNAME) && defined(HAVE_SYSCTL_KERN_CP_TIMES) /* {{{ */ long cpuinfo[maxcpu][CPUSTATES]; size_t cpuinfo_size; int i; @@@ -811,15 -456,14 +755,15 @@@ } for (i = 0; i < numcpu; i++) { - submit (i, "user", cpuinfo[i][CP_USER]); - submit (i, "nice", cpuinfo[i][CP_NICE]); - submit (i, "system", cpuinfo[i][CP_SYS]); - submit (i, "idle", cpuinfo[i][CP_IDLE]); - submit (i, "interrupt", cpuinfo[i][CP_INTR]); + cpu_stage (i, CPU_STATE_USER, (derive_t) cpuinfo[i][CP_USER], now); + cpu_stage (i, CPU_STATE_NICE, (derive_t) cpuinfo[i][CP_NICE], now); + cpu_stage (i, CPU_STATE_SYSTEM, (derive_t) cpuinfo[i][CP_SYS], now); + cpu_stage (i, CPU_STATE_IDLE, (derive_t) cpuinfo[i][CP_IDLE], now); + cpu_stage (i, CPU_STATE_INTERRUPT, (derive_t) cpuinfo[i][CP_INTR], now); } -/* #endif HAVE_SYSCTL_KERN_CP_TIMES */ -#elif defined(HAVE_SYSCTLBYNAME) +/* }}} #endif HAVE_SYSCTL_KERN_CP_TIMES */ + +#elif defined(HAVE_SYSCTLBYNAME) /* {{{ */ long cpuinfo[CPUSTATES]; size_t cpuinfo_size; @@@ -833,14 -477,14 +777,14 @@@ return (-1); } - submit (0, "user", cpuinfo[CP_USER]); - submit (0, "nice", cpuinfo[CP_NICE]); - submit (0, "system", cpuinfo[CP_SYS]); - submit (0, "idle", cpuinfo[CP_IDLE]); - submit (0, "interrupt", cpuinfo[CP_INTR]); -/* #endif HAVE_SYSCTLBYNAME */ + cpu_stage (0, CPU_STATE_USER, (derive_t) cpuinfo[CP_USER], now); + cpu_stage (0, CPU_STATE_NICE, (derive_t) cpuinfo[CP_NICE], now); + cpu_stage (0, CPU_STATE_SYSTEM, (derive_t) cpuinfo[CP_SYS], now); + cpu_stage (0, CPU_STATE_IDLE, (derive_t) cpuinfo[CP_IDLE], now); + cpu_stage (0, CPU_STATE_INTERRUPT, (derive_t) cpuinfo[CP_INTR], now); +/* }}} #endif HAVE_SYSCTLBYNAME */ -#elif defined(HAVE_LIBSTATGRAB) +#elif defined(HAVE_LIBSTATGRAB) /* {{{ */ sg_cpu_stats *cs; cs = sg_get_cpu_stats (); @@@ -850,15 -494,15 +794,15 @@@ return (-1); } - submit (0, "idle", (derive_t) cs->idle); - submit (0, "nice", (derive_t) cs->nice); - submit (0, "swap", (derive_t) cs->swap); - submit (0, "system", (derive_t) cs->kernel); - submit (0, "user", (derive_t) cs->user); - submit (0, "wait", (derive_t) cs->iowait); -/* #endif HAVE_LIBSTATGRAB */ + cpu_state (0, CPU_STATE_IDLE, (derive_t) cs->idle); + cpu_state (0, CPU_STATE_NICE, (derive_t) cs->nice); + cpu_state (0, CPU_STATE_SWAP, (derive_t) cs->swap); + cpu_state (0, CPU_STATE_SYSTEM, (derive_t) cs->kernel); + cpu_state (0, CPU_STATE_USER, (derive_t) cs->user); + cpu_state (0, CPU_STATE_WAIT, (derive_t) cs->iowait); +/* }}} #endif HAVE_LIBSTATGRAB */ -#elif defined(HAVE_PERFSTAT) +#elif defined(HAVE_PERFSTAT) /* {{{ */ perfstat_id_t id; int i, cpus; @@@ -870,10 -514,10 +814,10 @@@ sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); } - - if (pnumcpu != numcpu || perfcpu == NULL) + + if (pnumcpu != numcpu || perfcpu == NULL) { - if (perfcpu != NULL) + if (perfcpu != NULL) free(perfcpu); perfcpu = malloc(numcpu * sizeof(perfstat_cpu_t)); } @@@ -888,25 -532,20 +832,25 @@@ return (-1); } - for (i = 0; i < cpus; i++) + for (i = 0; i < cpus; i++) { - submit (i, "idle", (derive_t) perfcpu[i].idle); - submit (i, "system", (derive_t) perfcpu[i].sys); - submit (i, "user", (derive_t) perfcpu[i].user); - submit (i, "wait", (derive_t) perfcpu[i].wait); + cpu_stage (i, CPU_STATE_IDLE, (derive_t) perfcpu[i].idle, now); + cpu_stage (i, CPU_STATE_SYSTEM, (derive_t) perfcpu[i].sys, now); + cpu_stage (i, CPU_STATE_USER, (derive_t) perfcpu[i].user, now); + cpu_stage (i, CPU_STATE_WAIT, (derive_t) perfcpu[i].wait, now); } -#endif /* HAVE_PERFSTAT */ +#endif /* }}} HAVE_PERFSTAT */ + cpu_commit (); + cpu_reset (); return (0); } void module_register (void) { plugin_register_init ("cpu", init); + plugin_register_config ("cpu", cpu_config, config_keys, config_keys_num); plugin_register_read ("cpu", cpu_read); } /* void module_register */ + +/* vim: set sw=8 sts=8 noet fdm=marker : */ diff --combined src/curl_json.c index 8a084fed,6dec89e5..e99500aa --- a/src/curl_json.c +++ b/src/curl_json.c @@@ -71,13 -71,11 +71,13 @@@ struct cj_s /* {{{ * char *user; char *pass; char *credentials; + _Bool digest; _Bool verify_peer; _Bool verify_host; char *cacert; struct curl_slist *headers; char *post_body; + cdtime_t interval; CURL *curl; char curl_errbuf[CURL_ERROR_SIZE]; @@@ -234,9 -232,14 +234,14 @@@ static int cj_cb_number (void *ctx if (key != NULL) NOTICE ("curl_json plugin: Found \"%s\", but the configuration expects" " a map.", buffer); - cj_cb_inc_array_index (ctx, /* update_key = */ 0); - return (CJ_CB_CONTINUE); - } else { + cj_cb_inc_array_index (ctx, /* update_key = */ 1); + key = db->state[db->depth].key; + if (key == NULL) { + return (CJ_CB_CONTINUE); + } + } + else + { cj_cb_inc_array_index (ctx, /* update_key = */ 1); } @@@ -276,10 -279,21 +281,21 @@@ static int cj_cb_map_key (void *ctx memcpy (name, in_name, name_len); name[name_len] = 0; - if (c_avl_get (tree, name, (void *) &value) == 0) - db->state[db->depth].key = value; + if (c_avl_get (tree, name, (void *) &value) == 0) { + if (CJ_IS_KEY((cj_key_t*)value)) { + db->state[db->depth].key = value; + } + else { + db->state[db->depth].tree = (c_avl_tree_t*) value; + } + } else if (c_avl_get (tree, CJ_ANY, (void *) &value) == 0) - db->state[db->depth].key = value; + if (CJ_IS_KEY((cj_key_t*)value)) { + db->state[db->depth].key = value; + } + else { + db->state[db->depth].tree = (c_avl_tree_t*) value; + } else db->state[db->depth].key = NULL; } @@@ -589,17 -603,13 +605,17 @@@ static int cj_init_curl (cj_t *db) /* { curl_easy_setopt (db->curl, CURLOPT_NOSIGNAL, 1L); curl_easy_setopt (db->curl, CURLOPT_WRITEFUNCTION, cj_curl_callback); curl_easy_setopt (db->curl, CURLOPT_WRITEDATA, db); - curl_easy_setopt (db->curl, CURLOPT_USERAGENT, - PACKAGE_NAME"/"PACKAGE_VERSION); + curl_easy_setopt (db->curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT); curl_easy_setopt (db->curl, CURLOPT_ERRORBUFFER, db->curl_errbuf); curl_easy_setopt (db->curl, CURLOPT_URL, db->url); if (db->user != NULL) { +#ifdef HAVE_CURLOPT_USERNAME + curl_easy_setopt (db->curl, CURLOPT_USERNAME, db->user); + curl_easy_setopt (db->curl, CURLOPT_PASSWORD, + (db->pass == NULL) ? "" : db->pass); +#else size_t credentials_size; credentials_size = strlen (db->user) + 2; @@@ -616,10 -626,6 +632,10 @@@ ssnprintf (db->credentials, credentials_size, "%s:%s", db->user, (db->pass == NULL) ? "" : db->pass); curl_easy_setopt (db->curl, CURLOPT_USERPWD, db->credentials); +#endif + + if (db->digest) + curl_easy_setopt (db->curl, CURLOPT_HTTPAUTH, CURLAUTH_DIGEST); } curl_easy_setopt (db->curl, CURLOPT_SSL_VERIFYPEER, (long) db->verify_peer); @@@ -686,8 -692,6 +702,8 @@@ static int cj_config_add_url (oconfig_i status = cf_util_get_string (child, &db->user); else if (db->url && strcasecmp ("Password", child->key) == 0) status = cf_util_get_string (child, &db->pass); + else if (strcasecmp ("Digest", child->key) == 0) + status = cf_util_get_boolean (child, &db->digest); else if (db->url && strcasecmp ("VerifyPeer", child->key) == 0) status = cf_util_get_boolean (child, &db->verify_peer); else if (db->url && strcasecmp ("VerifyHost", child->key) == 0) @@@ -700,8 -704,6 +716,8 @@@ status = cf_util_get_string (child, &db->post_body); else if (strcasecmp ("Key", child->key) == 0) status = cj_config_add_key (db, child); + else if (strcasecmp ("Interval", child->key) == 0) + status = cf_util_get_cdtime(child, &db->interval); else { WARNING ("curl_json plugin: Option `%s' not allowed here.", child->key); @@@ -729,9 -731,6 +745,9 @@@ { user_data_t ud; char *cb_name; + struct timespec interval = { 0, 0 }; + + CDTIME_T_TO_TIMESPEC (db->interval, &interval); if (db->instance == NULL) db->instance = strdup("default"); @@@ -746,9 -745,8 +762,9 @@@ cb_name = ssnprintf_alloc ("curl_json-%s-%s", db->instance, db->url ? db->url : db->sock); - plugin_register_complex_read (/* group = */ "curl_json", cb_name, cj_read, - /* interval = */ NULL, &ud); + plugin_register_complex_read (/* group = */ NULL, cb_name, cj_read, + /* interval = */ (db->interval > 0) ? &interval : NULL, + &ud); sfree (cb_name); } else @@@ -819,10 -817,11 +835,10 @@@ static void cj_submit (cj_t *db, cj_key if (key->instance == NULL) { - if ((db->depth == 0) || (strcmp ("", db->state[db->depth-1].name) == 0)) - sstrncpy (vl.type_instance, db->state[db->depth].name, sizeof (vl.type_instance)); - else - ssnprintf (vl.type_instance, sizeof (vl.type_instance), "%s-%s", - db->state[db->depth-1].name, db->state[db->depth].name); + int i, len = 0; + for (i = 0; i < db->depth; i++) + len += ssnprintf(vl.type_instance+len, sizeof(vl.type_instance)-len, + i ? "-%s" : "%s", db->state[i+1].name); } else sstrncpy (vl.type_instance, key->instance, sizeof (vl.type_instance)); @@@ -832,9 -831,6 +848,9 @@@ sstrncpy (vl.plugin_instance, db->instance, sizeof (vl.plugin_instance)); sstrncpy (vl.type, key->type, sizeof (vl.type)); + if (db->interval > 0) + vl.interval = db->interval; + plugin_dispatch_values (&vl); } /* }}} int cj_submit */ diff --combined src/daemon/common.c index 535dfad3,00000000..e2b618fb mode 100644,000000..100644 --- a/src/daemon/common.c +++ b/src/daemon/common.c @@@ -1,1650 -1,0 +1,1650 @@@ +/** + * collectd - src/common.c + * Copyright (C) 2005-2014 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + * Niki W. Waibel + * Sebastian Harl + * Michał Mirosław +**/ + +#if HAVE_CONFIG_H +# include "config.h" +#endif + +#include "collectd.h" +#include "common.h" +#include "plugin.h" +#include "utils_cache.h" + +#if HAVE_PTHREAD_H +# include +#endif + +#ifdef HAVE_MATH_H +# include +#endif + +/* for getaddrinfo */ +#include +#include +#include + +#if HAVE_NETINET_IN_H +# include +#endif + +/* for ntohl and htonl */ +#if HAVE_ARPA_INET_H +# include +#endif + +#ifdef HAVE_LIBKSTAT +extern kstat_ctl_t *kc; +#endif + +#if !HAVE_GETPWNAM_R +static pthread_mutex_t getpwnam_r_lock = PTHREAD_MUTEX_INITIALIZER; +#endif + +#if !HAVE_STRERROR_R +static pthread_mutex_t strerror_r_lock = PTHREAD_MUTEX_INITIALIZER; +#endif + +char *sstrncpy (char *dest, const char *src, size_t n) +{ + strncpy (dest, src, n); + dest[n - 1] = '\0'; + + return (dest); +} /* char *sstrncpy */ + +int ssnprintf (char *dest, size_t n, const char *format, ...) +{ + int ret = 0; + va_list ap; + + va_start (ap, format); + ret = vsnprintf (dest, n, format, ap); + dest[n - 1] = '\0'; + va_end (ap); + + return (ret); +} /* int ssnprintf */ + +char *ssnprintf_alloc (char const *format, ...) /* {{{ */ +{ + char static_buffer[1024] = ""; + char *alloc_buffer; + size_t alloc_buffer_size; + int status; + va_list ap; + + /* Try printing into the static buffer. In many cases it will be + * sufficiently large and we can simply return a strdup() of this + * buffer. */ + va_start (ap, format); + status = vsnprintf (static_buffer, sizeof (static_buffer), format, ap); + va_end (ap); + if (status < 0) + return (NULL); + + /* "status" does not include the null byte. */ + alloc_buffer_size = (size_t) (status + 1); + if (alloc_buffer_size <= sizeof (static_buffer)) + return (strdup (static_buffer)); + + /* Allocate a buffer large enough to hold the string. */ + alloc_buffer = malloc (alloc_buffer_size); + if (alloc_buffer == NULL) + return (NULL); + memset (alloc_buffer, 0, alloc_buffer_size); + + /* Print again into this new buffer. */ + va_start (ap, format); + status = vsnprintf (alloc_buffer, alloc_buffer_size, format, ap); + va_end (ap); + if (status < 0) + { + sfree (alloc_buffer); + return (NULL); + } + + return (alloc_buffer); +} /* }}} char *ssnprintf_alloc */ + +char *sstrdup (const char *s) +{ + char *r; + size_t sz; + + if (s == NULL) + return (NULL); + + /* Do not use `strdup' here, because it's not specified in POSIX. It's + * ``only'' an XSI extension. */ + sz = strlen (s) + 1; + r = (char *) malloc (sizeof (char) * sz); + if (r == NULL) + { + ERROR ("sstrdup: Out of memory."); + exit (3); + } + memcpy (r, s, sizeof (char) * sz); + + return (r); +} /* char *sstrdup */ + +/* Even though Posix requires "strerror_r" to return an "int", + * some systems (e.g. the GNU libc) return a "char *" _and_ + * ignore the second argument ... -tokkee */ +char *sstrerror (int errnum, char *buf, size_t buflen) +{ + buf[0] = '\0'; + +#if !HAVE_STRERROR_R + { + char *temp; + + pthread_mutex_lock (&strerror_r_lock); + + temp = strerror (errnum); + sstrncpy (buf, temp, buflen); + + pthread_mutex_unlock (&strerror_r_lock); + } +/* #endif !HAVE_STRERROR_R */ + +#elif STRERROR_R_CHAR_P + { + char *temp; + temp = strerror_r (errnum, buf, buflen); + if (buf[0] == '\0') + { + if ((temp != NULL) && (temp != buf) && (temp[0] != '\0')) + sstrncpy (buf, temp, buflen); + else + sstrncpy (buf, "strerror_r did not return " + "an error message", buflen); + } + } +/* #endif STRERROR_R_CHAR_P */ + +#else + if (strerror_r (errnum, buf, buflen) != 0) + { + ssnprintf (buf, buflen, "Error #%i; " + "Additionally, strerror_r failed.", + errnum); + } +#endif /* STRERROR_R_CHAR_P */ + + return (buf); +} /* char *sstrerror */ + +void *smalloc (size_t size) +{ + void *r; + + if ((r = malloc (size)) == NULL) + { + ERROR ("Not enough memory."); + exit (3); + } + + return (r); +} /* void *smalloc */ + +#if 0 +void sfree (void **ptr) +{ + if (ptr == NULL) + return; + + if (*ptr != NULL) + free (*ptr); + + *ptr = NULL; +} +#endif + +ssize_t sread (int fd, void *buf, size_t count) +{ + char *ptr; + size_t nleft; + ssize_t status; + + ptr = (char *) buf; + nleft = count; + + while (nleft > 0) + { + status = read (fd, (void *) ptr, nleft); + + if ((status < 0) && ((errno == EAGAIN) || (errno == EINTR))) + continue; + + if (status < 0) + return (status); + + if (status == 0) + { + DEBUG ("Received EOF from fd %i. " + "Closing fd and returning error.", + fd); + close (fd); + return (-1); + } + + assert ((0 > status) || (nleft >= (size_t)status)); + + nleft = nleft - status; + ptr = ptr + status; + } + + return (0); +} + + +ssize_t swrite (int fd, const void *buf, size_t count) +{ + const char *ptr; + size_t nleft; + ssize_t status; + + ptr = (const char *) buf; + nleft = count; + + while (nleft > 0) + { + status = write (fd, (const void *) ptr, nleft); + + if ((status < 0) && ((errno == EAGAIN) || (errno == EINTR))) + continue; + + if (status < 0) + return (status); + + nleft = nleft - status; + ptr = ptr + status; + } + + return (0); +} + +int strsplit (char *string, char **fields, size_t size) +{ + size_t i; + char *ptr; + char *saveptr; + + i = 0; + ptr = string; + saveptr = NULL; + while ((fields[i] = strtok_r (ptr, " \t\r\n", &saveptr)) != NULL) + { + ptr = NULL; + i++; + + if (i >= size) + break; + } + + return ((int) i); +} + +int strjoin (char *dst, size_t dst_len, + char **fields, size_t fields_num, + const char *sep) +{ + size_t field_len; + size_t sep_len; + int i; + + memset (dst, '\0', dst_len); + + if (fields_num <= 0) + return (-1); + + sep_len = 0; + if (sep != NULL) + sep_len = strlen (sep); + + for (i = 0; i < (int)fields_num; i++) + { + if ((i > 0) && (sep_len > 0)) + { + if (dst_len <= sep_len) + return (-1); + + strncat (dst, sep, dst_len); + dst_len -= sep_len; + } + + field_len = strlen (fields[i]); + + if (dst_len <= field_len) + return (-1); + + strncat (dst, fields[i], dst_len); + dst_len -= field_len; + } + + return (strlen (dst)); +} + +int strsubstitute (char *str, char c_from, char c_to) +{ + int ret; + + if (str == NULL) + return (-1); + + ret = 0; + while (*str != '\0') + { + if (*str == c_from) + { + *str = c_to; + ret++; + } + str++; + } + + return (ret); +} /* int strsubstitute */ + +int escape_string (char *buffer, size_t buffer_size) +{ + char *temp; + size_t i; + size_t j; + + /* Check if we need to escape at all first */ + temp = strpbrk (buffer, " \t\"\\"); + if (temp == NULL) + return (0); + + if (buffer_size < 3) + return (EINVAL); + + temp = (char *) malloc (buffer_size); + if (temp == NULL) + return (ENOMEM); + memset (temp, 0, buffer_size); + + temp[0] = '"'; + j = 1; + + for (i = 0; i < buffer_size; i++) + { + if (buffer[i] == 0) + { + break; + } + else if ((buffer[i] == '"') || (buffer[i] == '\\')) + { + if (j > (buffer_size - 4)) + break; + temp[j] = '\\'; + temp[j + 1] = buffer[i]; + j += 2; + } + else + { + if (j > (buffer_size - 3)) + break; + temp[j] = buffer[i]; + j++; + } + } + + assert ((j + 1) < buffer_size); + temp[j] = '"'; + temp[j + 1] = 0; + + sstrncpy (buffer, temp, buffer_size); + sfree (temp); + return (0); +} /* int escape_string */ + +int strunescape (char *buf, size_t buf_len) +{ + size_t i; + + for (i = 0; (i < buf_len) && (buf[i] != '\0'); ++i) + { + if (buf[i] != '\\') + continue; + + if (((i + 1) >= buf_len) || (buf[i + 1] == 0)) { + ERROR ("string unescape: backslash found at end of string."); + /* Ensure null-byte at the end of the buffer. */ + buf[i] = 0; + return (-1); + } + + switch (buf[i + 1]) { + case 't': + buf[i] = '\t'; + break; + case 'n': + buf[i] = '\n'; + break; + case 'r': + buf[i] = '\r'; + break; + default: + buf[i] = buf[i + 1]; + break; + } + + /* Move everything after the position one position to the left. + * Add a null-byte as last character in the buffer. */ + memmove (buf + i + 1, buf + i + 2, buf_len - i - 2); + buf[buf_len - 1] = 0; + } + return (0); +} /* int strunescape */ + +size_t strstripnewline (char *buffer) +{ + size_t buffer_len = strlen (buffer); + + while (buffer_len > 0) + { + if ((buffer[buffer_len - 1] != '\n') + && (buffer[buffer_len - 1] != '\r')) + break; - buffer[buffer_len] = 0; + buffer_len--; ++ buffer[buffer_len] = 0; + } + + return (buffer_len); +} /* size_t strstripnewline */ + +int escape_slashes (char *buffer, size_t buffer_size) +{ + int i; + size_t buffer_len; + + buffer_len = strlen (buffer); + + if (buffer_len <= 1) + { + if (strcmp ("/", buffer) == 0) + { + if (buffer_size < 5) + return (-1); + sstrncpy (buffer, "root", buffer_size); + } + return (0); + } + + /* Move one to the left */ + if (buffer[0] == '/') + { + memmove (buffer, buffer + 1, buffer_len); + buffer_len--; + } + + for (i = 0; i < buffer_len - 1; i++) + { + if (buffer[i] == '/') + buffer[i] = '_'; + } + + return (0); +} /* int escape_slashes */ + +void replace_special (char *buffer, size_t buffer_size) +{ + size_t i; + + for (i = 0; i < buffer_size; i++) + { + if (buffer[i] == 0) + return; + if ((!isalnum ((int) buffer[i])) && (buffer[i] != '-')) + buffer[i] = '_'; + } +} /* void replace_special */ + +int timeval_cmp (struct timeval tv0, struct timeval tv1, struct timeval *delta) +{ + struct timeval *larger; + struct timeval *smaller; + + int status; + + NORMALIZE_TIMEVAL (tv0); + NORMALIZE_TIMEVAL (tv1); + + if ((tv0.tv_sec == tv1.tv_sec) && (tv0.tv_usec == tv1.tv_usec)) + { + if (delta != NULL) { + delta->tv_sec = 0; + delta->tv_usec = 0; + } + return (0); + } + + if ((tv0.tv_sec < tv1.tv_sec) + || ((tv0.tv_sec == tv1.tv_sec) && (tv0.tv_usec < tv1.tv_usec))) + { + larger = &tv1; + smaller = &tv0; + status = -1; + } + else + { + larger = &tv0; + smaller = &tv1; + status = 1; + } + + if (delta != NULL) { + delta->tv_sec = larger->tv_sec - smaller->tv_sec; + + if (smaller->tv_usec <= larger->tv_usec) + delta->tv_usec = larger->tv_usec - smaller->tv_usec; + else + { + --delta->tv_sec; + delta->tv_usec = 1000000 + larger->tv_usec - smaller->tv_usec; + } + } + + assert ((delta == NULL) + || ((0 <= delta->tv_usec) && (delta->tv_usec < 1000000))); + + return (status); +} /* int timeval_cmp */ + +int check_create_dir (const char *file_orig) +{ + struct stat statbuf; + + char file_copy[512]; + char dir[512]; + int dir_len = 512; + char *fields[16]; + int fields_num; + char *ptr; + char *saveptr; + int last_is_file = 1; + int path_is_absolute = 0; + size_t len; + int i; + + /* + * Sanity checks first + */ + if (file_orig == NULL) + return (-1); + + if ((len = strlen (file_orig)) < 1) + return (-1); + else if (len >= sizeof (file_copy)) + return (-1); + + /* + * If `file_orig' ends in a slash the last component is a directory, + * otherwise it's a file. Act accordingly.. + */ + if (file_orig[len - 1] == '/') + last_is_file = 0; + if (file_orig[0] == '/') + path_is_absolute = 1; + + /* + * Create a copy for `strtok_r' to destroy + */ + sstrncpy (file_copy, file_orig, sizeof (file_copy)); + + /* + * Break into components. This will eat up several slashes in a row and + * remove leading and trailing slashes.. + */ + ptr = file_copy; + saveptr = NULL; + fields_num = 0; + while ((fields[fields_num] = strtok_r (ptr, "/", &saveptr)) != NULL) + { + ptr = NULL; + fields_num++; + + if (fields_num >= 16) + break; + } + + /* + * For each component, do.. + */ + for (i = 0; i < (fields_num - last_is_file); i++) + { + /* + * Do not create directories that start with a dot. This + * prevents `../../' attacks and other likely malicious + * behavior. + */ + if (fields[i][0] == '.') + { + ERROR ("Cowardly refusing to create a directory that " + "begins with a `.' (dot): `%s'", file_orig); + return (-2); + } + + /* + * Join the components together again + */ + dir[0] = '/'; + if (strjoin (dir + path_is_absolute, dir_len - path_is_absolute, + fields, i + 1, "/") < 0) + { + ERROR ("strjoin failed: `%s', component #%i", file_orig, i); + return (-1); + } + + while (42) { + if ((stat (dir, &statbuf) == -1) + && (lstat (dir, &statbuf) == -1)) + { + if (errno == ENOENT) + { + if (mkdir (dir, S_IRWXU | S_IRWXG | S_IRWXO) == 0) + break; + + /* this might happen, if a different thread created + * the directory in the meantime + * => call stat() again to check for S_ISDIR() */ + if (EEXIST == errno) + continue; + + char errbuf[1024]; + ERROR ("check_create_dir: mkdir (%s): %s", dir, + sstrerror (errno, + errbuf, sizeof (errbuf))); + return (-1); + } + else + { + char errbuf[1024]; + ERROR ("check_create_dir: stat (%s): %s", dir, + sstrerror (errno, errbuf, + sizeof (errbuf))); + return (-1); + } + } + else if (!S_ISDIR (statbuf.st_mode)) + { + ERROR ("check_create_dir: `%s' exists but is not " + "a directory!", dir); + return (-1); + } + break; + } + } + + return (0); +} /* check_create_dir */ + +#ifdef HAVE_LIBKSTAT +int get_kstat (kstat_t **ksp_ptr, char *module, int instance, char *name) +{ + char ident[128]; + + *ksp_ptr = NULL; + + if (kc == NULL) + return (-1); + + ssnprintf (ident, sizeof (ident), "%s,%i,%s", module, instance, name); + + *ksp_ptr = kstat_lookup (kc, module, instance, name); + if (*ksp_ptr == NULL) + { + ERROR ("get_kstat: Cound not find kstat %s", ident); + return (-1); + } + + if ((*ksp_ptr)->ks_type != KSTAT_TYPE_NAMED) + { + ERROR ("get_kstat: kstat %s has wrong type", ident); + *ksp_ptr = NULL; + return (-1); + } + +#ifdef assert + assert (*ksp_ptr != NULL); + assert ((*ksp_ptr)->ks_type == KSTAT_TYPE_NAMED); +#endif + + if (kstat_read (kc, *ksp_ptr, NULL) == -1) + { + ERROR ("get_kstat: kstat %s could not be read", ident); + return (-1); + } + + if ((*ksp_ptr)->ks_type != KSTAT_TYPE_NAMED) + { + ERROR ("get_kstat: kstat %s has wrong type", ident); + return (-1); + } + + return (0); +} + +long long get_kstat_value (kstat_t *ksp, char *name) +{ + kstat_named_t *kn; + long long retval = -1LL; + + if (ksp == NULL) + { + ERROR ("get_kstat_value (\"%s\"): ksp is NULL.", name); + return (-1LL); + } + else if (ksp->ks_type != KSTAT_TYPE_NAMED) + { + ERROR ("get_kstat_value (\"%s\"): ksp->ks_type (%#x) " + "is not KSTAT_TYPE_NAMED (%#x).", + name, + (unsigned int) ksp->ks_type, + (unsigned int) KSTAT_TYPE_NAMED); + return (-1LL); + } + + if ((kn = (kstat_named_t *) kstat_data_lookup (ksp, name)) == NULL) + return (-1LL); + + if (kn->data_type == KSTAT_DATA_INT32) + retval = (long long) kn->value.i32; + else if (kn->data_type == KSTAT_DATA_UINT32) + retval = (long long) kn->value.ui32; + else if (kn->data_type == KSTAT_DATA_INT64) + retval = (long long) kn->value.i64; /* According to ANSI C99 `long long' must hold at least 64 bits */ + else if (kn->data_type == KSTAT_DATA_UINT64) + retval = (long long) kn->value.ui64; /* XXX: Might overflow! */ + else + WARNING ("get_kstat_value: Not a numeric value: %s", name); + + return (retval); +} +#endif /* HAVE_LIBKSTAT */ + +#ifndef HAVE_HTONLL +unsigned long long ntohll (unsigned long long n) +{ +#if BYTE_ORDER == BIG_ENDIAN + return (n); +#else + return (((unsigned long long) ntohl (n)) << 32) + ntohl (n >> 32); +#endif +} /* unsigned long long ntohll */ + +unsigned long long htonll (unsigned long long n) +{ +#if BYTE_ORDER == BIG_ENDIAN + return (n); +#else + return (((unsigned long long) htonl (n)) << 32) + htonl (n >> 32); +#endif +} /* unsigned long long htonll */ +#endif /* HAVE_HTONLL */ + +#if FP_LAYOUT_NEED_NOTHING +/* Well, we need nothing.. */ +/* #endif FP_LAYOUT_NEED_NOTHING */ + +#elif FP_LAYOUT_NEED_ENDIANFLIP || FP_LAYOUT_NEED_INTSWAP +# if FP_LAYOUT_NEED_ENDIANFLIP +# define FP_CONVERT(A) ((((uint64_t)(A) & 0xff00000000000000LL) >> 56) | \ + (((uint64_t)(A) & 0x00ff000000000000LL) >> 40) | \ + (((uint64_t)(A) & 0x0000ff0000000000LL) >> 24) | \ + (((uint64_t)(A) & 0x000000ff00000000LL) >> 8) | \ + (((uint64_t)(A) & 0x00000000ff000000LL) << 8) | \ + (((uint64_t)(A) & 0x0000000000ff0000LL) << 24) | \ + (((uint64_t)(A) & 0x000000000000ff00LL) << 40) | \ + (((uint64_t)(A) & 0x00000000000000ffLL) << 56)) +# else +# define FP_CONVERT(A) ((((uint64_t)(A) & 0xffffffff00000000LL) >> 32) | \ + (((uint64_t)(A) & 0x00000000ffffffffLL) << 32)) +# endif + +double ntohd (double d) +{ + union + { + uint8_t byte[8]; + uint64_t integer; + double floating; + } ret; + + ret.floating = d; + + /* NAN in x86 byte order */ + if ((ret.byte[0] == 0x00) && (ret.byte[1] == 0x00) + && (ret.byte[2] == 0x00) && (ret.byte[3] == 0x00) + && (ret.byte[4] == 0x00) && (ret.byte[5] == 0x00) + && (ret.byte[6] == 0xf8) && (ret.byte[7] == 0x7f)) + { + return (NAN); + } + else + { + uint64_t tmp; + + tmp = ret.integer; + ret.integer = FP_CONVERT (tmp); + return (ret.floating); + } +} /* double ntohd */ + +double htond (double d) +{ + union + { + uint8_t byte[8]; + uint64_t integer; + double floating; + } ret; + + if (isnan (d)) + { + ret.byte[0] = ret.byte[1] = ret.byte[2] = ret.byte[3] = 0x00; + ret.byte[4] = ret.byte[5] = 0x00; + ret.byte[6] = 0xf8; + ret.byte[7] = 0x7f; + return (ret.floating); + } + else + { + uint64_t tmp; + + ret.floating = d; + tmp = FP_CONVERT (ret.integer); + ret.integer = tmp; + return (ret.floating); + } +} /* double htond */ +#endif /* FP_LAYOUT_NEED_ENDIANFLIP || FP_LAYOUT_NEED_INTSWAP */ + +int format_name (char *ret, int ret_len, + const char *hostname, + const char *plugin, const char *plugin_instance, + const char *type, const char *type_instance) +{ + char *buffer; + size_t buffer_size; + + buffer = ret; + buffer_size = (size_t) ret_len; + +#define APPEND(str) do { \ + size_t l = strlen (str); \ + if (l >= buffer_size) \ + return (ENOBUFS); \ + memcpy (buffer, (str), l); \ + buffer += l; buffer_size -= l; \ +} while (0) + + assert (plugin != NULL); + assert (type != NULL); + + APPEND (hostname); + APPEND ("/"); + APPEND (plugin); + if ((plugin_instance != NULL) && (plugin_instance[0] != 0)) + { + APPEND ("-"); + APPEND (plugin_instance); + } + APPEND ("/"); + APPEND (type); + if ((type_instance != NULL) && (type_instance[0] != 0)) + { + APPEND ("-"); + APPEND (type_instance); + } + assert (buffer_size > 0); + buffer[0] = 0; + +#undef APPEND + return (0); +} /* int format_name */ + +int format_values (char *ret, size_t ret_len, /* {{{ */ + const data_set_t *ds, const value_list_t *vl, + _Bool store_rates) +{ + size_t offset = 0; + int status; + int i; + 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) + + BUFFER_ADD ("%.3f", CDTIME_T_TO_DOUBLE (vl->time)); + + for (i = 0; i < ds->ds_num; i++) + { + if (ds->ds[i].type == DS_TYPE_GAUGE) + BUFFER_ADD (":%f", vl->values[i].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[i]); + } + else if (ds->ds[i].type == DS_TYPE_COUNTER) + BUFFER_ADD (":%llu", vl->values[i].counter); + else if (ds->ds[i].type == DS_TYPE_DERIVE) + BUFFER_ADD (":%"PRIi64, vl->values[i].derive); + else if (ds->ds[i].type == DS_TYPE_ABSOLUTE) + BUFFER_ADD (":%"PRIu64, vl->values[i].absolute); + else + { + ERROR ("format_values plugin: Unknown data source type: %i", + ds->ds[i].type); + sfree (rates); + return (-1); + } + } /* for ds->ds_num */ + +#undef BUFFER_ADD + + sfree (rates); + return (0); +} /* }}} int format_values */ + +int parse_identifier (char *str, char **ret_host, + char **ret_plugin, char **ret_plugin_instance, + char **ret_type, char **ret_type_instance) +{ + char *hostname = NULL; + char *plugin = NULL; + char *plugin_instance = NULL; + char *type = NULL; + char *type_instance = NULL; + + hostname = str; + if (hostname == NULL) + return (-1); + + plugin = strchr (hostname, '/'); + if (plugin == NULL) + return (-1); + *plugin = '\0'; plugin++; + + type = strchr (plugin, '/'); + if (type == NULL) + return (-1); + *type = '\0'; type++; + + plugin_instance = strchr (plugin, '-'); + if (plugin_instance != NULL) + { + *plugin_instance = '\0'; + plugin_instance++; + } + + type_instance = strchr (type, '-'); + if (type_instance != NULL) + { + *type_instance = '\0'; + type_instance++; + } + + *ret_host = hostname; + *ret_plugin = plugin; + *ret_plugin_instance = plugin_instance; + *ret_type = type; + *ret_type_instance = type_instance; + return (0); +} /* int parse_identifier */ + +int parse_identifier_vl (const char *str, value_list_t *vl) /* {{{ */ +{ + char str_copy[6 * DATA_MAX_NAME_LEN]; + char *host = NULL; + char *plugin = NULL; + char *plugin_instance = NULL; + char *type = NULL; + char *type_instance = NULL; + int status; + + if ((str == NULL) || (vl == NULL)) + return (EINVAL); + + sstrncpy (str_copy, str, sizeof (str_copy)); + + status = parse_identifier (str_copy, &host, + &plugin, &plugin_instance, + &type, &type_instance); + if (status != 0) + return (status); + + sstrncpy (vl->host, host, sizeof (vl->host)); + sstrncpy (vl->plugin, plugin, sizeof (vl->plugin)); + sstrncpy (vl->plugin_instance, + (plugin_instance != NULL) ? plugin_instance : "", + sizeof (vl->plugin_instance)); + sstrncpy (vl->type, type, sizeof (vl->type)); + sstrncpy (vl->type_instance, + (type_instance != NULL) ? type_instance : "", + sizeof (vl->type_instance)); + + return (0); +} /* }}} int parse_identifier_vl */ + +int parse_value (const char *value_orig, value_t *ret_value, int ds_type) +{ + char *value; + char *endptr = NULL; + size_t value_len; + + if (value_orig == NULL) + return (EINVAL); + + value = strdup (value_orig); + if (value == NULL) + return (ENOMEM); + value_len = strlen (value); + + while ((value_len > 0) && isspace ((int) value[value_len - 1])) + { + value[value_len - 1] = 0; + value_len--; + } + + switch (ds_type) + { + case DS_TYPE_COUNTER: + ret_value->counter = (counter_t) strtoull (value, &endptr, 0); + break; + + case DS_TYPE_GAUGE: + ret_value->gauge = (gauge_t) strtod (value, &endptr); + break; + + case DS_TYPE_DERIVE: + ret_value->derive = (derive_t) strtoll (value, &endptr, 0); + break; + + case DS_TYPE_ABSOLUTE: + ret_value->absolute = (absolute_t) strtoull (value, &endptr, 0); + break; + + default: + sfree (value); + ERROR ("parse_value: Invalid data source type: %i.", ds_type); + return -1; + } + + if (value == endptr) { + ERROR ("parse_value: Failed to parse string as %s: %s.", + DS_TYPE_TO_STRING (ds_type), value); + sfree (value); + return -1; + } + else if ((NULL != endptr) && ('\0' != *endptr)) + INFO ("parse_value: Ignoring trailing garbage \"%s\" after %s value. " + "Input string was \"%s\".", + endptr, DS_TYPE_TO_STRING (ds_type), value_orig); + + sfree (value); + return 0; +} /* int parse_value */ + +int parse_values (char *buffer, value_list_t *vl, const data_set_t *ds) +{ + int i; + char *dummy; + char *ptr; + char *saveptr; + + i = -1; + dummy = buffer; + saveptr = NULL; + while ((ptr = strtok_r (dummy, ":", &saveptr)) != NULL) + { + dummy = NULL; + + if (i >= vl->values_len) + { + /* Make sure i is invalid. */ + i = vl->values_len + 1; + break; + } + + if (i == -1) + { + if (strcmp ("N", ptr) == 0) + vl->time = cdtime (); + else + { + char *endptr = NULL; + double tmp; + + errno = 0; + tmp = strtod (ptr, &endptr); + if ((errno != 0) /* Overflow */ + || (endptr == ptr) /* Invalid string */ + || (endptr == NULL) /* This should not happen */ + || (*endptr != 0)) /* Trailing chars */ + return (-1); + + vl->time = DOUBLE_TO_CDTIME_T (tmp); + } + } + else + { + if ((strcmp ("U", ptr) == 0) && (ds->ds[i].type == DS_TYPE_GAUGE)) + vl->values[i].gauge = NAN; + else if (0 != parse_value (ptr, &vl->values[i], ds->ds[i].type)) + return -1; + } + + i++; + } /* while (strtok_r) */ + + if ((ptr != NULL) || (i != vl->values_len)) + return (-1); + return (0); +} /* int parse_values */ + +#if !HAVE_GETPWNAM_R +int getpwnam_r (const char *name, struct passwd *pwbuf, char *buf, + size_t buflen, struct passwd **pwbufp) +{ + int status = 0; + struct passwd *pw; + + memset (pwbuf, '\0', sizeof (struct passwd)); + + pthread_mutex_lock (&getpwnam_r_lock); + + do + { + pw = getpwnam (name); + if (pw == NULL) + { + status = (errno != 0) ? errno : ENOENT; + break; + } + +#define GETPWNAM_COPY_MEMBER(member) \ + if (pw->member != NULL) \ + { \ + int len = strlen (pw->member); \ + if (len >= buflen) \ + { \ + status = ENOMEM; \ + break; \ + } \ + sstrncpy (buf, pw->member, buflen); \ + pwbuf->member = buf; \ + buf += (len + 1); \ + buflen -= (len + 1); \ + } + GETPWNAM_COPY_MEMBER(pw_name); + GETPWNAM_COPY_MEMBER(pw_passwd); + GETPWNAM_COPY_MEMBER(pw_gecos); + GETPWNAM_COPY_MEMBER(pw_dir); + GETPWNAM_COPY_MEMBER(pw_shell); + + pwbuf->pw_uid = pw->pw_uid; + pwbuf->pw_gid = pw->pw_gid; + + if (pwbufp != NULL) + *pwbufp = pwbuf; + } while (0); + + pthread_mutex_unlock (&getpwnam_r_lock); + + return (status); +} /* int getpwnam_r */ +#endif /* !HAVE_GETPWNAM_R */ + +int notification_init (notification_t *n, int severity, const char *message, + const char *host, + const char *plugin, const char *plugin_instance, + const char *type, const char *type_instance) +{ + memset (n, '\0', sizeof (notification_t)); + + n->severity = severity; + + if (message != NULL) + sstrncpy (n->message, message, sizeof (n->message)); + if (host != NULL) + sstrncpy (n->host, host, sizeof (n->host)); + if (plugin != NULL) + sstrncpy (n->plugin, plugin, sizeof (n->plugin)); + if (plugin_instance != NULL) + sstrncpy (n->plugin_instance, plugin_instance, + sizeof (n->plugin_instance)); + if (type != NULL) + sstrncpy (n->type, type, sizeof (n->type)); + if (type_instance != NULL) + sstrncpy (n->type_instance, type_instance, + sizeof (n->type_instance)); + + return (0); +} /* int notification_init */ + +int walk_directory (const char *dir, dirwalk_callback_f callback, + void *user_data, int include_hidden) +{ + struct dirent *ent; + DIR *dh; + int success; + int failure; + + success = 0; + failure = 0; + + if ((dh = opendir (dir)) == NULL) + { + char errbuf[1024]; + ERROR ("walk_directory: Cannot open '%s': %s", dir, + sstrerror (errno, errbuf, sizeof (errbuf))); + return -1; + } + + while ((ent = readdir (dh)) != NULL) + { + int status; + + if (include_hidden) + { + if ((strcmp (".", ent->d_name) == 0) + || (strcmp ("..", ent->d_name) == 0)) + continue; + } + else /* if (!include_hidden) */ + { + if (ent->d_name[0]=='.') + continue; + } + + status = (*callback) (dir, ent->d_name, user_data); + if (status != 0) + failure++; + else + success++; + } + + closedir (dh); + + if ((success == 0) && (failure > 0)) + return (-1); + return (0); +} + +ssize_t read_file_contents (const char *filename, char *buf, size_t bufsize) +{ + FILE *fh; + ssize_t ret; + + fh = fopen (filename, "r"); + if (fh == NULL) + return (-1); + + ret = (ssize_t) fread (buf, 1, bufsize, fh); + if ((ret == 0) && (ferror (fh) != 0)) + { + ERROR ("read_file_contents: Reading file \"%s\" failed.", + filename); + ret = -1; + } + + fclose(fh); + return (ret); +} + +counter_t counter_diff (counter_t old_value, counter_t new_value) +{ + counter_t diff; + + if (old_value > new_value) + { + if (old_value <= 4294967295U) + diff = (4294967295U - old_value) + new_value; + else + diff = (18446744073709551615ULL - old_value) + + new_value; + } + else + { + diff = new_value - old_value; + } + + return (diff); +} /* counter_t counter_diff */ + +int rate_to_value (value_t *ret_value, gauge_t rate, /* {{{ */ + rate_to_value_state_t *state, + int ds_type, cdtime_t t) +{ + gauge_t delta_gauge; + cdtime_t delta_t; + + if (ds_type == DS_TYPE_GAUGE) + { + state->last_value.gauge = rate; + state->last_time = t; + + *ret_value = state->last_value; + return (0); + } + + /* Counter and absolute can't handle negative rates. Reset "last time" + * to zero, so that the next valid rate will re-initialize the + * structure. */ + if ((rate < 0.0) + && ((ds_type == DS_TYPE_COUNTER) + || (ds_type == DS_TYPE_ABSOLUTE))) + { + memset (state, 0, sizeof (*state)); + return (EINVAL); + } + + /* Another invalid state: The time is not increasing. */ + if (t <= state->last_time) + { + memset (state, 0, sizeof (*state)); + return (EINVAL); + } + + delta_t = t - state->last_time; + delta_gauge = (rate * CDTIME_T_TO_DOUBLE (delta_t)) + state->residual; + + /* Previous value is invalid. */ + if (state->last_time == 0) /* {{{ */ + { + if (ds_type == DS_TYPE_DERIVE) + { + state->last_value.derive = (derive_t) rate; + state->residual = rate - ((gauge_t) state->last_value.derive); + } + else if (ds_type == DS_TYPE_COUNTER) + { + state->last_value.counter = (counter_t) rate; + state->residual = rate - ((gauge_t) state->last_value.counter); + } + else if (ds_type == DS_TYPE_ABSOLUTE) + { + state->last_value.absolute = (absolute_t) rate; + state->residual = rate - ((gauge_t) state->last_value.absolute); + } + else + { + assert (23 == 42); + } + + state->last_time = t; + return (EAGAIN); + } /* }}} */ + + if (ds_type == DS_TYPE_DERIVE) + { + derive_t delta_derive = (derive_t) delta_gauge; + + state->last_value.derive += delta_derive; + state->residual = delta_gauge - ((gauge_t) delta_derive); + } + else if (ds_type == DS_TYPE_COUNTER) + { + counter_t delta_counter = (counter_t) delta_gauge; + + state->last_value.counter += delta_counter; + state->residual = delta_gauge - ((gauge_t) delta_counter); + } + else if (ds_type == DS_TYPE_ABSOLUTE) + { + absolute_t delta_absolute = (absolute_t) delta_gauge; + + state->last_value.absolute = delta_absolute; + state->residual = delta_gauge - ((gauge_t) delta_absolute); + } + else + { + assert (23 == 42); + } + + state->last_time = t; + *ret_value = state->last_value; + return (0); +} /* }}} value_t rate_to_value */ + +int value_to_rate (value_t *ret_rate, derive_t value, /* {{{ */ + value_to_rate_state_t *state, + int ds_type, cdtime_t t) +{ + double interval; + + /* Another invalid state: The time is not increasing. */ + if (t <= state->last_time) + { + memset (state, 0, sizeof (*state)); + return (EINVAL); + } + + interval = CDTIME_T_TO_DOUBLE(t - state->last_time); + + /* Previous value is invalid. */ + if (state->last_time == 0) /* {{{ */ + { + if (ds_type == DS_TYPE_DERIVE) + { + state->last_value.derive = value; + } + else if (ds_type == DS_TYPE_COUNTER) + { + state->last_value.counter = (counter_t) value; + } + else if (ds_type == DS_TYPE_ABSOLUTE) + { + state->last_value.absolute = (absolute_t) value; + } + else + { + assert (23 == 42); + } + + state->last_time = t; + return (EAGAIN); + } /* }}} */ + + if (ds_type == DS_TYPE_DERIVE) + { + ret_rate->gauge = (value - state->last_value.derive) / interval; + state->last_value.derive = value; + } + else if (ds_type == DS_TYPE_COUNTER) + { + ret_rate->gauge = (((counter_t)value) - state->last_value.counter) / interval; + state->last_value.counter = (counter_t) value; + } + else if (ds_type == DS_TYPE_ABSOLUTE) + { + ret_rate->gauge = (((absolute_t)value) - state->last_value.absolute) / interval; + state->last_value.absolute = (absolute_t) value; + } + else + { + assert (23 == 42); + } + + state->last_time = t; + return (0); +} /* }}} value_t rate_to_value */ + +int service_name_to_port_number (const char *service_name) +{ + struct addrinfo *ai_list; + struct addrinfo *ai_ptr; + struct addrinfo ai_hints; + int status; + int service_number; + + if (service_name == NULL) + return (-1); + + ai_list = NULL; + memset (&ai_hints, 0, sizeof (ai_hints)); + ai_hints.ai_family = AF_UNSPEC; + + status = getaddrinfo (/* node = */ NULL, service_name, + &ai_hints, &ai_list); + if (status != 0) + { + ERROR ("service_name_to_port_number: getaddrinfo failed: %s", + gai_strerror (status)); + return (-1); + } + + service_number = -1; + for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) + { + if (ai_ptr->ai_family == AF_INET) + { + struct sockaddr_in *sa; + + sa = (void *) ai_ptr->ai_addr; + service_number = (int) ntohs (sa->sin_port); + } + else if (ai_ptr->ai_family == AF_INET6) + { + struct sockaddr_in6 *sa; + + sa = (void *) ai_ptr->ai_addr; + service_number = (int) ntohs (sa->sin6_port); + } + + if ((service_number > 0) && (service_number <= 65535)) + break; + } + + freeaddrinfo (ai_list); + + if ((service_number > 0) && (service_number <= 65535)) + return (service_number); + return (-1); +} /* int service_name_to_port_number */ + +int strtoderive (const char *string, derive_t *ret_value) /* {{{ */ +{ + derive_t tmp; + char *endptr; + + if ((string == NULL) || (ret_value == NULL)) + return (EINVAL); + + errno = 0; + endptr = NULL; + tmp = (derive_t) strtoll (string, &endptr, /* base = */ 0); + if ((endptr == string) || (errno != 0)) + return (-1); + + *ret_value = tmp; + return (0); +} /* }}} int strtoderive */ + +int strtogauge (const char *string, gauge_t *ret_value) /* {{{ */ +{ + gauge_t tmp; + char *endptr = NULL; + + if ((string == NULL) || (ret_value == NULL)) + return (EINVAL); + + errno = 0; + endptr = NULL; + tmp = (gauge_t) strtod (string, &endptr); + if (errno != 0) + return (errno); + else if ((endptr == NULL) || (*endptr != 0)) + return (EINVAL); + + *ret_value = tmp; + return (0); +} /* }}} int strtogauge */ + +int strarray_add (char ***ret_array, size_t *ret_array_len, char const *str) /* {{{ */ +{ + char **array; + size_t array_len = *ret_array_len; + + if (str == NULL) + return (EINVAL); + + array = realloc (*ret_array, + (array_len + 1) * sizeof (*array)); + if (array == NULL) + return (ENOMEM); + *ret_array = array; + + array[array_len] = strdup (str); + if (array[array_len] == NULL) + return (ENOMEM); + + array_len++; + *ret_array_len = array_len; + return (0); +} /* }}} int strarray_add */ + +void strarray_free (char **array, size_t array_len) /* {{{ */ +{ + size_t i; + + for (i = 0; i < array_len; i++) + sfree (array[i]); + sfree (array); +} /* }}} void strarray_free */ diff --combined src/daemon/filter_chain.c index c87b8773,00000000..b93435fd mode 100644,000000..100644 --- a/src/daemon/filter_chain.c +++ b/src/daemon/filter_chain.c @@@ -1,1058 -1,0 +1,1079 @@@ +/** + * collectd - src/filter_chain.c + * Copyright (C) 2008-2010 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + **/ + +#include "collectd.h" +#include "configfile.h" +#include "plugin.h" +#include "utils_complain.h" +#include "common.h" +#include "filter_chain.h" + +/* + * Data types + */ +/* List of matches, used in fc_rule_t and for the global `match_list_head' + * variable. */ +struct fc_match_s; +typedef struct fc_match_s fc_match_t; /* {{{ */ +struct fc_match_s +{ + char name[DATA_MAX_NAME_LEN]; + match_proc_t proc; + void *user_data; + fc_match_t *next; +}; /* }}} */ + +/* List of targets, used in fc_rule_t and for the global `target_list_head' + * variable. */ +struct fc_target_s; +typedef struct fc_target_s fc_target_t; /* {{{ */ +struct fc_target_s +{ + char name[DATA_MAX_NAME_LEN]; + void *user_data; + target_proc_t proc; + fc_target_t *next; +}; /* }}} */ + +/* List of rules, used in fc_chain_t */ +struct fc_rule_s; +typedef struct fc_rule_s fc_rule_t; /* {{{ */ +struct fc_rule_s +{ + char name[DATA_MAX_NAME_LEN]; + fc_match_t *matches; + fc_target_t *targets; + fc_rule_t *next; +}; /* }}} */ + +/* List of chains, used for `chain_list_head' */ +struct fc_chain_s /* {{{ */ +{ + char name[DATA_MAX_NAME_LEN]; + fc_rule_t *rules; + fc_target_t *targets; + fc_chain_t *next; +}; /* }}} */ + ++/* Writer configuration. */ ++struct fc_writer_s; ++typedef struct fc_writer_s fc_writer_t; /* {{{ */ ++struct fc_writer_s ++{ ++ char *plugin; ++ c_complain_t complaint; ++}; /* }}} */ ++ +/* + * Global variables + */ +static fc_match_t *match_list_head; +static fc_target_t *target_list_head; +static fc_chain_t *chain_list_head; + +/* + * Private functions + */ +static void fc_free_matches (fc_match_t *m) /* {{{ */ +{ + if (m == NULL) + return; + + if (m->proc.destroy != NULL) + (*m->proc.destroy) (&m->user_data); + else if (m->user_data != NULL) + { + ERROR ("Filter subsystem: fc_free_matches: There is user data, but no " + "destroy functions has been specified. " + "Memory will probably be lost!"); + } + + if (m->next != NULL) + fc_free_matches (m->next); + + free (m); +} /* }}} void fc_free_matches */ + +static void fc_free_targets (fc_target_t *t) /* {{{ */ +{ + if (t == NULL) + return; + + if (t->proc.destroy != NULL) + (*t->proc.destroy) (&t->user_data); + else if (t->user_data != NULL) + { + ERROR ("Filter subsystem: fc_free_targets: There is user data, but no " + "destroy functions has been specified. " + "Memory will probably be lost!"); + } + + if (t->next != NULL) + fc_free_targets (t->next); + + free (t); +} /* }}} void fc_free_targets */ + +static void fc_free_rules (fc_rule_t *r) /* {{{ */ +{ + if (r == NULL) + return; + + fc_free_matches (r->matches); + fc_free_targets (r->targets); + + if (r->next != NULL) + fc_free_rules (r->next); + + free (r); +} /* }}} void fc_free_rules */ + +static void fc_free_chains (fc_chain_t *c) /* {{{ */ +{ + if (c == NULL) + return; + + fc_free_rules (c->rules); + fc_free_targets (c->targets); + + if (c->next != NULL) + fc_free_chains (c->next); + + free (c); +} /* }}} void fc_free_chains */ + +static char *fc_strdup (const char *orig) /* {{{ */ +{ + size_t sz; + char *dest; + + if (orig == NULL) + return (NULL); + + sz = strlen (orig) + 1; + dest = (char *) malloc (sz); + if (dest == NULL) + return (NULL); + + memcpy (dest, orig, sz); + + return (dest); +} /* }}} char *fc_strdup */ + +/* + * Configuration. + * + * The configuration looks somewhat like this: + * + * + * + * + * Plugin "^mysql$" + * Type "^mysql_command$" + * TypeInstance "^show_" + * + * + * + * + * + * + * Plugin "rrdtool" + * + * + */ +static int fc_config_add_match (fc_match_t **matches_head, /* {{{ */ + oconfig_item_t *ci) +{ + fc_match_t *m; + fc_match_t *ptr; + int status; + + if ((ci->values_num != 1) + || (ci->values[0].type != OCONFIG_TYPE_STRING)) + { + WARNING ("Filter subsystem: `Match' blocks require " + "exactly one string argument."); + return (-1); + } + + ptr = match_list_head; + while (ptr != NULL) + { + if (strcasecmp (ptr->name, ci->values[0].value.string) == 0) + break; + ptr = ptr->next; + } + + if (ptr == NULL) + { + WARNING ("Filter subsystem: Cannot find a \"%s\" match. " + "Did you load the appropriate plugin?", + ci->values[0].value.string); + return (-1); + } + + m = (fc_match_t *) malloc (sizeof (*m)); + if (m == NULL) + { + ERROR ("fc_config_add_match: malloc failed."); + return (-1); + } + memset (m, 0, sizeof (*m)); + + sstrncpy (m->name, ptr->name, sizeof (m->name)); + memcpy (&m->proc, &ptr->proc, sizeof (m->proc)); + m->user_data = NULL; + m->next = NULL; + + if (m->proc.create != NULL) + { + status = (*m->proc.create) (ci, &m->user_data); + if (status != 0) + { + WARNING ("Filter subsystem: Failed to create a %s match.", + m->name); + fc_free_matches (m); + return (-1); + } + } + + if (*matches_head != NULL) + { + ptr = *matches_head; + while (ptr->next != NULL) + ptr = ptr->next; + + ptr->next = m; + } + else + { + *matches_head = m; + } + + return (0); +} /* }}} int fc_config_add_match */ + +static int fc_config_add_target (fc_target_t **targets_head, /* {{{ */ + oconfig_item_t *ci) +{ + fc_target_t *t; + fc_target_t *ptr; + int status; + + if ((ci->values_num != 1) + || (ci->values[0].type != OCONFIG_TYPE_STRING)) + { + WARNING ("Filter subsystem: `Target' blocks require " + "exactly one string argument."); + return (-1); + } + + ptr = target_list_head; + while (ptr != NULL) + { + if (strcasecmp (ptr->name, ci->values[0].value.string) == 0) + break; + ptr = ptr->next; + } + + if (ptr == NULL) + { + WARNING ("Filter subsystem: Cannot find a \"%s\" target. " + "Did you load the appropriate plugin?", + ci->values[0].value.string); + return (-1); + } + + t = (fc_target_t *) malloc (sizeof (*t)); + if (t == NULL) + { + ERROR ("fc_config_add_target: malloc failed."); + return (-1); + } + memset (t, 0, sizeof (*t)); + + sstrncpy (t->name, ptr->name, sizeof (t->name)); + memcpy (&t->proc, &ptr->proc, sizeof (t->proc)); + t->user_data = NULL; + t->next = NULL; + + if (t->proc.create != NULL) + { + status = (*t->proc.create) (ci, &t->user_data); + if (status != 0) + { + WARNING ("Filter subsystem: Failed to create a %s target.", + t->name); + fc_free_targets (t); + return (-1); + } + } + else + { + t->user_data = NULL; + } + + if (*targets_head != NULL) + { + ptr = *targets_head; + while (ptr->next != NULL) + ptr = ptr->next; + + ptr->next = t; + } + else + { + *targets_head = t; + } + + return (0); +} /* }}} int fc_config_add_target */ + +static int fc_config_add_rule (fc_chain_t *chain, /* {{{ */ + oconfig_item_t *ci) +{ + fc_rule_t *rule; + char rule_name[2*DATA_MAX_NAME_LEN] = "Unnamed rule"; + int status = 0; + int i; + + if (ci->values_num > 1) + { + WARNING ("Filter subsystem: `Rule' blocks have at most one argument."); + return (-1); + } + else if ((ci->values_num == 1) + && (ci->values[0].type != OCONFIG_TYPE_STRING)) + { + WARNING ("Filter subsystem: `Rule' blocks expect one string argument " + "or no argument at all."); + return (-1); + } + + rule = (fc_rule_t *) malloc (sizeof (*rule)); + if (rule == NULL) + { + ERROR ("fc_config_add_rule: malloc failed."); + return (-1); + } + memset (rule, 0, sizeof (*rule)); + rule->next = NULL; + + if (ci->values_num == 1) + { + sstrncpy (rule->name, ci->values[0].value.string, sizeof (rule->name)); + ssnprintf (rule_name, sizeof (rule_name), "Rule \"%s\"", + ci->values[0].value.string); + } + + for (i = 0; i < ci->children_num; i++) + { + oconfig_item_t *option = ci->children + i; + status = 0; + + if (strcasecmp ("Match", option->key) == 0) + status = fc_config_add_match (&rule->matches, option); + else if (strcasecmp ("Target", option->key) == 0) + status = fc_config_add_target (&rule->targets, option); + else + { + WARNING ("Filter subsystem: %s: Option `%s' not allowed " + "inside a block.", rule_name, option->key); + status = -1; + } + + if (status != 0) + break; + } /* for (ci->children) */ + + /* Additional sanity checking. */ + while (status == 0) + { + if (rule->targets == NULL) + { + WARNING ("Filter subsystem: %s: No target has been specified.", + rule_name); + status = -1; + break; + } + + break; + } /* while (status == 0) */ + + if (status != 0) + { + fc_free_rules (rule); + return (-1); + } + + if (chain->rules != NULL) + { + fc_rule_t *ptr; + + ptr = chain->rules; + while (ptr->next != NULL) + ptr = ptr->next; + + ptr->next = rule; + } + else + { + chain->rules = rule; + } + + return (0); +} /* }}} int fc_config_add_rule */ + +static int fc_config_add_chain (const oconfig_item_t *ci) /* {{{ */ +{ + fc_chain_t *chain = NULL; + int status = 0; + int i; + int new_chain = 1; + + if ((ci->values_num != 1) + || (ci->values[0].type != OCONFIG_TYPE_STRING)) + { + WARNING ("Filter subsystem: blocks require exactly one " + "string argument."); + return (-1); + } + + if (chain_list_head != NULL) + { + if ((chain = fc_chain_get_by_name (ci->values[0].value.string)) != NULL) + new_chain = 0; + } + + if (chain == NULL) + { + chain = (fc_chain_t *) malloc (sizeof (*chain)); + if (chain == NULL) + { + ERROR ("fc_config_add_chain: malloc failed."); + return (-1); + } + memset (chain, 0, sizeof (*chain)); + sstrncpy (chain->name, ci->values[0].value.string, sizeof (chain->name)); + chain->rules = NULL; + chain->targets = NULL; + chain->next = NULL; + } + + for (i = 0; i < ci->children_num; i++) + { + oconfig_item_t *option = ci->children + i; + status = 0; + + if (strcasecmp ("Rule", option->key) == 0) + status = fc_config_add_rule (chain, option); + else if (strcasecmp ("Target", option->key) == 0) + status = fc_config_add_target (&chain->targets, option); + else + { + WARNING ("Filter subsystem: Chain %s: Option `%s' not allowed " + "inside a block.", chain->name, option->key); + status = -1; + } + + if (status != 0) + break; + } /* for (ci->children) */ + + if (status != 0) + { + fc_free_chains (chain); + return (-1); + } + + if (chain_list_head != NULL) + { + if (!new_chain) + return (0); + + fc_chain_t *ptr; + + ptr = chain_list_head; + while (ptr->next != NULL) + ptr = ptr->next; + + ptr->next = chain; + } + else + { + chain_list_head = chain; + } + + return (0); +} /* }}} int fc_config_add_chain */ + +/* + * Built-in target "jump" + * + * Prefix `bit' like `_b_uilt-_i_n _t_arget' + */ +static int fc_bit_jump_create (const oconfig_item_t *ci, /* {{{ */ + void **user_data) +{ + oconfig_item_t *ci_chain; + + if (ci->children_num != 1) + { + ERROR ("Filter subsystem: The built-in target `jump' needs exactly " + "one `Chain' argument!"); + return (-1); + } + + ci_chain = ci->children; + if (strcasecmp ("Chain", ci_chain->key) != 0) + { + ERROR ("Filter subsystem: The built-in target `jump' does not " + "support the configuration option `%s'.", + ci_chain->key); + return (-1); + } + + if ((ci_chain->values_num != 1) + || (ci_chain->values[0].type != OCONFIG_TYPE_STRING)) + { + ERROR ("Filter subsystem: Built-in target `jump': The `Chain' option " + "needs exactly one string argument."); + return (-1); + } + + *user_data = fc_strdup (ci_chain->values[0].value.string); + if (*user_data == NULL) + { + ERROR ("fc_bit_jump_create: fc_strdup failed."); + return (-1); + } + + return (0); +} /* }}} int fc_bit_jump_create */ + +static int fc_bit_jump_destroy (void **user_data) /* {{{ */ +{ + if (user_data != NULL) + { + free (*user_data); + *user_data = NULL; + } + + return (0); +} /* }}} int fc_bit_jump_destroy */ + +static int fc_bit_jump_invoke (const data_set_t *ds, /* {{{ */ + value_list_t *vl, notification_meta_t __attribute__((unused)) **meta, + void **user_data) +{ + char *chain_name; + fc_chain_t *chain; + int status; + + chain_name = *user_data; + + for (chain = chain_list_head; chain != NULL; chain = chain->next) + if (strcasecmp (chain_name, chain->name) == 0) + break; + + if (chain == NULL) + { + ERROR ("Filter subsystem: Built-in target `jump': There is no chain " + "named `%s'.", chain_name); + return (-1); + } + + status = fc_process_chain (ds, vl, chain); + if (status < 0) + return (status); + else if (status == FC_TARGET_STOP) + return (FC_TARGET_STOP); + else + return (FC_TARGET_CONTINUE); +} /* }}} int fc_bit_jump_invoke */ + +static int fc_bit_stop_invoke (const data_set_t __attribute__((unused)) *ds, /* {{{ */ + value_list_t __attribute__((unused)) *vl, + notification_meta_t __attribute__((unused)) **meta, + void __attribute__((unused)) **user_data) +{ + return (FC_TARGET_STOP); +} /* }}} int fc_bit_stop_invoke */ + +static int fc_bit_return_invoke (const data_set_t __attribute__((unused)) *ds, /* {{{ */ + value_list_t __attribute__((unused)) *vl, + notification_meta_t __attribute__((unused)) **meta, + void __attribute__((unused)) **user_data) +{ + return (FC_TARGET_RETURN); +} /* }}} int fc_bit_return_invoke */ + +static int fc_bit_write_create (const oconfig_item_t *ci, /* {{{ */ + void **user_data) +{ + int i; + - char **plugin_list; - size_t plugin_list_len; - - plugin_list = NULL; - plugin_list_len = 0; ++ fc_writer_t *plugin_list = NULL; ++ size_t plugin_list_len = 0; + + for (i = 0; i < ci->children_num; i++) + { + oconfig_item_t *child = ci->children + i; - char **temp; ++ fc_writer_t *temp; + int j; + + if (strcasecmp ("Plugin", child->key) != 0) + { + ERROR ("Filter subsystem: The built-in target `write' does not " + "support the configuration option `%s'.", + child->key); + continue; + } + + for (j = 0; j < child->values_num; j++) + { ++ char *plugin; ++ + if (child->values[j].type != OCONFIG_TYPE_STRING) + { + ERROR ("Filter subsystem: Built-in target `write': " + "The `Plugin' option accepts only string arguments."); + continue; + } ++ plugin = child->values[j].value.string; + - temp = (char **) realloc (plugin_list, (plugin_list_len + 2) ++ temp = (fc_writer_t *) realloc (plugin_list, (plugin_list_len + 2) + * (sizeof (*plugin_list))); + if (temp == NULL) + { + ERROR ("fc_bit_write_create: realloc failed."); + continue; + } + plugin_list = temp; + - plugin_list[plugin_list_len] = fc_strdup (child->values[j].value.string); - if (plugin_list[plugin_list_len] == NULL) ++ plugin_list[plugin_list_len].plugin = fc_strdup (plugin); ++ if (plugin_list[plugin_list_len].plugin == NULL) + { + ERROR ("fc_bit_write_create: fc_strdup failed."); + continue; + } ++ C_COMPLAIN_INIT (&plugin_list[plugin_list_len].complaint); + plugin_list_len++; - plugin_list[plugin_list_len] = NULL; ++ plugin_list[plugin_list_len].plugin = NULL; + } /* for (j = 0; j < child->values_num; j++) */ + } /* for (i = 0; i < ci->children_num; i++) */ + + *user_data = plugin_list; + + return (0); +} /* }}} int fc_bit_write_create */ + +static int fc_bit_write_destroy (void **user_data) /* {{{ */ +{ - char **plugin_list; ++ fc_writer_t *plugin_list; + size_t i; + + if ((user_data == NULL) || (*user_data == NULL)) + return (0); + + plugin_list = *user_data; + - for (i = 0; plugin_list[i] != NULL; i++) - free (plugin_list[i]); ++ for (i = 0; plugin_list[i].plugin != NULL; i++) ++ free (plugin_list[i].plugin); + free (plugin_list); + + return (0); +} /* }}} int fc_bit_write_destroy */ + +static int fc_bit_write_invoke (const data_set_t *ds, /* {{{ */ + value_list_t *vl, notification_meta_t __attribute__((unused)) **meta, + void **user_data) +{ - char **plugin_list; ++ fc_writer_t *plugin_list; + int status; + + plugin_list = NULL; + if (user_data != NULL) + plugin_list = *user_data; + - if ((plugin_list == NULL) || (plugin_list[0] == NULL)) ++ if ((plugin_list == NULL) || (plugin_list[0].plugin == NULL)) + { - static c_complain_t enoent_complaint = C_COMPLAIN_INIT_STATIC; ++ static c_complain_t write_complaint = C_COMPLAIN_INIT_STATIC; + + status = plugin_write (/* plugin = */ NULL, ds, vl); + if (status == ENOENT) + { + /* in most cases this is a permanent error, so use the complain + * mechanism rather than spamming the logs */ - c_complain (LOG_INFO, &enoent_complaint, ++ c_complain (LOG_INFO, &write_complaint, + "Filter subsystem: Built-in target `write': Dispatching value to " + "all write plugins failed with status %i (ENOENT). " + "Most likely this means you didn't load any write plugins.", + status); + } + else if (status != 0) + { - INFO ("Filter subsystem: Built-in target `write': Dispatching value to " ++ /* often, this is a permanent error (e.g. target system unavailable), ++ * so use the complain mechanism rather than spamming the logs */ ++ c_complain (LOG_INFO, &write_complaint, ++ "Filter subsystem: Built-in target `write': Dispatching value to " + "all write plugins failed with status %i.", status); + } + else + { + assert (status == 0); - c_release (LOG_INFO, &enoent_complaint, "Filter subsystem: " ++ c_release (LOG_INFO, &write_complaint, "Filter subsystem: " + "Built-in target `write': Some write plugin is back to normal " + "operation. `write' succeeded."); + } + } + else + { + size_t i; + - for (i = 0; plugin_list[i] != NULL; i++) ++ for (i = 0; plugin_list[i].plugin != NULL; i++) + { - status = plugin_write (plugin_list[i], ds, vl); ++ status = plugin_write (plugin_list[i].plugin, ds, vl); + if (status != 0) + { - INFO ("Filter subsystem: Built-in target `write': Dispatching value to " - "the `%s' plugin failed with status %i.", plugin_list[i], status); ++ c_complain (LOG_INFO, &plugin_list[i].complaint, ++ "Filter subsystem: Built-in target `write': Dispatching value to " ++ "the `%s' plugin failed with status %i.", ++ plugin_list[i].plugin, status); ++ } ++ else ++ { ++ c_release (LOG_INFO, &plugin_list[i].complaint, ++ "Filter subsystem: Built-in target `write': Plugin `%s' is back " ++ "to normal operation. `write' succeeded.", plugin_list[i].plugin); + } + } /* for (i = 0; plugin_list[i] != NULL; i++) */ + } + + return (FC_TARGET_CONTINUE); +} /* }}} int fc_bit_write_invoke */ + +static int fc_init_once (void) /* {{{ */ +{ + static int done = 0; + target_proc_t tproc; + + if (done != 0) + return (0); + + memset (&tproc, 0, sizeof (tproc)); + tproc.create = fc_bit_jump_create; + tproc.destroy = fc_bit_jump_destroy; + tproc.invoke = fc_bit_jump_invoke; + fc_register_target ("jump", tproc); + + memset (&tproc, 0, sizeof (tproc)); + tproc.create = NULL; + tproc.destroy = NULL; + tproc.invoke = fc_bit_stop_invoke; + fc_register_target ("stop", tproc); + + memset (&tproc, 0, sizeof (tproc)); + tproc.create = NULL; + tproc.destroy = NULL; + tproc.invoke = fc_bit_return_invoke; + fc_register_target ("return", tproc); + + memset (&tproc, 0, sizeof (tproc)); + tproc.create = fc_bit_write_create; + tproc.destroy = fc_bit_write_destroy; + tproc.invoke = fc_bit_write_invoke; + fc_register_target ("write", tproc); + + done++; + return (0); +} /* }}} int fc_init_once */ + +/* + * Public functions + */ +/* Add a match to list of available matches. */ +int fc_register_match (const char *name, match_proc_t proc) /* {{{ */ +{ + fc_match_t *m; + + DEBUG ("fc_register_match (%s);", name); + + m = (fc_match_t *) malloc (sizeof (*m)); + if (m == NULL) + return (-ENOMEM); + memset (m, 0, sizeof (*m)); + + sstrncpy (m->name, name, sizeof (m->name)); + memcpy (&m->proc, &proc, sizeof (m->proc)); + m->next = NULL; + + if (match_list_head == NULL) + { + match_list_head = m; + } + else + { + fc_match_t *ptr; + + ptr = match_list_head; + while (ptr->next != NULL) + ptr = ptr->next; + + ptr->next = m; + } + + return (0); +} /* }}} int fc_register_match */ + +/* Add a target to list of available targets. */ +int fc_register_target (const char *name, target_proc_t proc) /* {{{ */ +{ + fc_target_t *t; + + DEBUG ("fc_register_target (%s);", name); + + t = (fc_target_t *) malloc (sizeof (*t)); + if (t == NULL) + return (-ENOMEM); + memset (t, 0, sizeof (*t)); + + sstrncpy (t->name, name, sizeof (t->name)); + memcpy (&t->proc, &proc, sizeof (t->proc)); + t->next = NULL; + + if (target_list_head == NULL) + { + target_list_head = t; + } + else + { + fc_target_t *ptr; + + ptr = target_list_head; + while (ptr->next != NULL) + ptr = ptr->next; + + ptr->next = t; + } + + return (0); +} /* }}} int fc_register_target */ + +fc_chain_t *fc_chain_get_by_name (const char *chain_name) /* {{{ */ +{ + fc_chain_t *chain; + + if (chain_name == NULL) + return (NULL); + + for (chain = chain_list_head; chain != NULL; chain = chain->next) + if (strcasecmp (chain_name, chain->name) == 0) + return (chain); + + return (NULL); +} /* }}} int fc_chain_get_by_name */ + +int fc_process_chain (const data_set_t *ds, value_list_t *vl, /* {{{ */ + fc_chain_t *chain) +{ + fc_rule_t *rule; + fc_target_t *target; + int status; + + if (chain == NULL) + return (-1); + + DEBUG ("fc_process_chain (chain = %s);", chain->name); + + status = FC_TARGET_CONTINUE; + for (rule = chain->rules; rule != NULL; rule = rule->next) + { + fc_match_t *match; + + if (rule->name[0] != 0) + { + DEBUG ("fc_process_chain (%s): Testing the `%s' rule.", + chain->name, rule->name); + } + + /* N. B.: rule->matches may be NULL. */ + for (match = rule->matches; match != NULL; match = match->next) + { + /* FIXME: Pass the meta-data to match targets here (when implemented). */ + status = (*match->proc.match) (ds, vl, /* meta = */ NULL, + &match->user_data); + if (status < 0) + { + WARNING ("fc_process_chain (%s): A match failed.", chain->name); + break; + } + else if (status != FC_MATCH_MATCHES) + break; + } + + /* for-loop has been aborted: Either error or no match. */ + if (match != NULL) + { + status = FC_TARGET_CONTINUE; + continue; + } + + if (rule->name[0] != 0) + { + DEBUG ("fc_process_chain (%s): Rule `%s' matches.", + chain->name, rule->name); + } + + for (target = rule->targets; target != NULL; target = target->next) + { + /* If we get here, all matches have matched the value. Execute the + * target. */ + /* FIXME: Pass the meta-data to match targets here (when implemented). */ + status = (*target->proc.invoke) (ds, vl, /* meta = */ NULL, + &target->user_data); + if (status < 0) + { + WARNING ("fc_process_chain (%s): A target failed.", chain->name); + continue; + } + else if (status == FC_TARGET_CONTINUE) + continue; + else if (status == FC_TARGET_STOP) + break; + else if (status == FC_TARGET_RETURN) + break; + else + { + WARNING ("fc_process_chain (%s): Unknown return value " + "from target `%s': %i", + chain->name, target->name, status); + } + } + + if ((status == FC_TARGET_STOP) + || (status == FC_TARGET_RETURN)) + { + if (rule->name[0] != 0) + { + DEBUG ("fc_process_chain (%s): Rule `%s' signaled " + "the %s condition.", + chain->name, rule->name, + (status == FC_TARGET_STOP) ? "stop" : "return"); + } + break; + } + else + { + status = FC_TARGET_CONTINUE; + } + } /* for (rule) */ + + if (status == FC_TARGET_STOP) + return (FC_TARGET_STOP); + else if (status == FC_TARGET_RETURN) + return (FC_TARGET_CONTINUE); + + /* for-loop has been aborted: A target returned `FC_TARGET_STOP' */ + if (rule != NULL) + return (FC_TARGET_CONTINUE); + + DEBUG ("fc_process_chain (%s): Executing the default targets.", + chain->name); + + status = FC_TARGET_CONTINUE; + for (target = chain->targets; target != NULL; target = target->next) + { + /* If we get here, all matches have matched the value. Execute the + * target. */ + /* FIXME: Pass the meta-data to match targets here (when implemented). */ + status = (*target->proc.invoke) (ds, vl, /* meta = */ NULL, + &target->user_data); + if (status < 0) + { + WARNING ("fc_process_chain (%s): The default target failed.", + chain->name); + } + else if (status == FC_TARGET_CONTINUE) + continue; + else if (status == FC_TARGET_STOP) + break; + else if (status == FC_TARGET_RETURN) + break; + else + { + WARNING ("fc_process_chain (%s): Unknown return value " + "from target `%s': %i", + chain->name, target->name, status); + } + } + + if ((status == FC_TARGET_STOP) + || (status == FC_TARGET_RETURN)) + { + assert (target != NULL); + DEBUG ("fc_process_chain (%s): Default target `%s' signaled " + "the %s condition.", + chain->name, target->name, + (status == FC_TARGET_STOP) ? "stop" : "return"); + if (status == FC_TARGET_STOP) + return (FC_TARGET_STOP); + else + return (FC_TARGET_CONTINUE); + } + + DEBUG ("fc_process_chain (%s): Signaling `continue' at end of chain.", + chain->name); + + return (FC_TARGET_CONTINUE); +} /* }}} int fc_process_chain */ + +/* Iterate over all rules in the chain and execute all targets for which all + * matches match. */ +int fc_default_action (const data_set_t *ds, value_list_t *vl) /* {{{ */ +{ + /* FIXME: Pass the meta-data to match targets here (when implemented). */ + return (fc_bit_write_invoke (ds, vl, + /* meta = */ NULL, /* user_data = */ NULL)); +} /* }}} int fc_default_action */ + +int fc_configure (const oconfig_item_t *ci) /* {{{ */ +{ + fc_init_once (); + + if (ci == NULL) + return (-EINVAL); + + if (strcasecmp ("Chain", ci->key) == 0) + return (fc_config_add_chain (ci)); + + WARNING ("Filter subsystem: Unknown top level config option `%s'.", + ci->key); + + return (-1); +} /* }}} int fc_configure */ + +/* vim: set sw=2 sts=2 et fdm=marker : */ diff --combined src/daemon/plugin.c index 97b794f7,00000000..6c6ab63c mode 100644,000000..100644 --- a/src/daemon/plugin.c +++ b/src/daemon/plugin.c @@@ -1,2652 -1,0 +1,2652 @@@ +/** + * collectd - src/plugin.c + * Copyright (C) 2005-2014 Florian octo Forster + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * Authors: + * Florian octo Forster + * Sebastian Harl + **/ + +#include "collectd.h" +#include "common.h" +#include "plugin.h" +#include "configfile.h" +#include "filter_chain.h" +#include "utils_avltree.h" +#include "utils_cache.h" +#include "utils_complain.h" +#include "utils_llist.h" +#include "utils_heap.h" +#include "utils_time.h" +#include "utils_random.h" + +#include + +/* + * Private structures + */ +struct callback_func_s +{ + void *cf_callback; + user_data_t cf_udata; + plugin_ctx_t cf_ctx; +}; +typedef struct callback_func_s callback_func_t; + +#define RF_SIMPLE 0 +#define RF_COMPLEX 1 +#define RF_REMOVE 65535 +struct read_func_s +{ + /* `read_func_t' "inherits" from `callback_func_t'. + * The `rf_super' member MUST be the first one in this structure! */ +#define rf_callback rf_super.cf_callback +#define rf_udata rf_super.cf_udata +#define rf_ctx rf_super.cf_ctx + callback_func_t rf_super; + char rf_group[DATA_MAX_NAME_LEN]; + char *rf_name; + int rf_type; + cdtime_t rf_interval; + cdtime_t rf_effective_interval; + cdtime_t rf_next_read; +}; +typedef struct read_func_s read_func_t; + +struct write_queue_s; +typedef struct write_queue_s write_queue_t; +struct write_queue_s +{ + value_list_t *vl; + plugin_ctx_t ctx; + write_queue_t *next; +}; + +/* + * Private variables + */ +static c_avl_tree_t *plugins_loaded = NULL; + +static llist_t *list_init; +static llist_t *list_write; +static llist_t *list_flush; +static llist_t *list_missing; +static llist_t *list_shutdown; +static llist_t *list_log; +static llist_t *list_notification; + +static fc_chain_t *pre_cache_chain = NULL; +static fc_chain_t *post_cache_chain = NULL; + +static c_avl_tree_t *data_sets; + +static char *plugindir = NULL; + +#ifndef DEFAULT_MAX_READ_INTERVAL +# define DEFAULT_MAX_READ_INTERVAL TIME_T_TO_CDTIME_T (86400) +#endif +static c_heap_t *read_heap = NULL; +static llist_t *read_list; +static int read_loop = 1; +static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t read_cond = PTHREAD_COND_INITIALIZER; +static pthread_t *read_threads = NULL; +static int read_threads_num = 0; +static cdtime_t max_read_interval = DEFAULT_MAX_READ_INTERVAL; + +static write_queue_t *write_queue_head; +static write_queue_t *write_queue_tail; +static long write_queue_length = 0; +static _Bool write_loop = 1; +static pthread_mutex_t write_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t write_cond = PTHREAD_COND_INITIALIZER; +static pthread_t *write_threads = NULL; +static size_t write_threads_num = 0; + +static pthread_key_t plugin_ctx_key; +static _Bool plugin_ctx_key_initialized = 0; + +static long write_limit_high = 0; +static long write_limit_low = 0; + +static derive_t stats_values_dropped = 0; +static _Bool record_statistics = 0; + +/* + * Static functions + */ +static int plugin_dispatch_values_internal (value_list_t *vl); + +static const char *plugin_get_dir (void) +{ + if (plugindir == NULL) + return (PLUGINDIR); + else + return (plugindir); +} + +static void plugin_update_internal_statistics (void) { /* {{{ */ + derive_t copy_write_queue_length; + value_list_t vl = VALUE_LIST_INIT; + value_t values[2]; + + copy_write_queue_length = write_queue_length; + + /* Initialize `vl' */ + vl.values = values; + vl.values_len = 2; + vl.time = 0; + sstrncpy (vl.host, hostname_g, sizeof (vl.host)); + sstrncpy (vl.plugin, "collectd", sizeof (vl.plugin)); + + vl.type_instance[0] = 0; + vl.values_len = 1; + + /* Write queue */ + sstrncpy (vl.plugin_instance, "write_queue", + sizeof (vl.plugin_instance)); + + /* Write queue : queue length */ + vl.values[0].gauge = (gauge_t) copy_write_queue_length; + sstrncpy (vl.type, "queue_length", sizeof (vl.type)); + vl.type_instance[0] = 0; + plugin_dispatch_values (&vl); + + /* Write queue : Values dropped (queue length > low limit) */ + vl.values[0].derive = (derive_t) stats_values_dropped; + sstrncpy (vl.type, "derive", sizeof (vl.type)); + sstrncpy (vl.type_instance, "dropped", sizeof (vl.type_instance)); + plugin_dispatch_values (&vl); + + /* Cache */ + sstrncpy (vl.plugin_instance, "cache", + sizeof (vl.plugin_instance)); + + /* Cache : Nb entry in cache tree */ + vl.values[0].gauge = (gauge_t) uc_get_size(); + sstrncpy (vl.type, "cache_size", sizeof (vl.type)); + vl.type_instance[0] = 0; + plugin_dispatch_values (&vl); + + return; +} /* }}} void plugin_update_internal_statistics */ + +static void destroy_callback (callback_func_t *cf) /* {{{ */ +{ + if (cf == NULL) + return; + + if ((cf->cf_udata.data != NULL) && (cf->cf_udata.free_func != NULL)) + { + cf->cf_udata.free_func (cf->cf_udata.data); + cf->cf_udata.data = NULL; + cf->cf_udata.free_func = NULL; + } + sfree (cf); +} /* }}} void destroy_callback */ + +static void destroy_all_callbacks (llist_t **list) /* {{{ */ +{ + llentry_t *le; + + if (*list == NULL) + return; + + le = llist_head (*list); + while (le != NULL) + { + llentry_t *le_next; + + le_next = le->next; + + sfree (le->key); + destroy_callback (le->value); + le->value = NULL; + + le = le_next; + } + + llist_destroy (*list); + *list = NULL; +} /* }}} void destroy_all_callbacks */ + +static void destroy_read_heap (void) /* {{{ */ +{ + if (read_heap == NULL) + return; + + while (42) + { + callback_func_t *cf; + + cf = c_heap_get_root (read_heap); + if (cf == NULL) + break; + + destroy_callback (cf); + } + + c_heap_destroy (read_heap); + read_heap = NULL; +} /* }}} void destroy_read_heap */ + +static int register_callback (llist_t **list, /* {{{ */ + const char *name, callback_func_t *cf) +{ + llentry_t *le; + char *key; + + if (*list == NULL) + { + *list = llist_create (); + if (*list == NULL) + { + ERROR ("plugin: register_callback: " + "llist_create failed."); + destroy_callback (cf); + return (-1); + } + } + + key = strdup (name); + if (key == NULL) + { + ERROR ("plugin: register_callback: strdup failed."); + destroy_callback (cf); + return (-1); + } + + le = llist_search (*list, name); + if (le == NULL) + { + le = llentry_create (key, cf); + if (le == NULL) + { + ERROR ("plugin: register_callback: " + "llentry_create failed."); + free (key); + destroy_callback (cf); + return (-1); + } + + llist_append (*list, le); + } + else + { + callback_func_t *old_cf; + + old_cf = le->value; + le->value = cf; + + WARNING ("plugin: register_callback: " + "a callback named `%s' already exists - " + "overwriting the old entry!", name); + + destroy_callback (old_cf); + sfree (key); + } + + return (0); +} /* }}} int register_callback */ + +static int create_register_callback (llist_t **list, /* {{{ */ + const char *name, void *callback, user_data_t *ud) +{ + callback_func_t *cf; + + cf = (callback_func_t *) malloc (sizeof (*cf)); + if (cf == NULL) + { + ERROR ("plugin: create_register_callback: malloc failed."); + return (-1); + } + memset (cf, 0, sizeof (*cf)); + + cf->cf_callback = callback; + if (ud == NULL) + { + cf->cf_udata.data = NULL; + cf->cf_udata.free_func = NULL; + } + else + { + cf->cf_udata = *ud; + } + + cf->cf_ctx = plugin_get_ctx (); + + return (register_callback (list, name, cf)); +} /* }}} int create_register_callback */ + +static int plugin_unregister (llist_t *list, const char *name) /* {{{ */ +{ + llentry_t *e; + + if (list == NULL) + return (-1); + + e = llist_search (list, name); + if (e == NULL) + return (-1); + + llist_remove (list, e); + + sfree (e->key); + destroy_callback (e->value); + + llentry_destroy (e); + + return (0); +} /* }}} int plugin_unregister */ + +/* + * (Try to) load the shared object `file'. Won't complain if it isn't a shared + * object, but it will bitch about a shared object not having a + * ``module_register'' symbol.. + */ +static int plugin_load_file (char *file, uint32_t flags) +{ + lt_dlhandle dlh; + void (*reg_handle) (void); + + lt_dlinit (); + lt_dlerror (); /* clear errors */ + +#if LIBTOOL_VERSION == 2 + if (flags & PLUGIN_FLAGS_GLOBAL) { + lt_dladvise advise; + lt_dladvise_init(&advise); + lt_dladvise_global(&advise); + dlh = lt_dlopenadvise(file, advise); + lt_dladvise_destroy(&advise); + } else { + dlh = lt_dlopen (file); + } +#else /* if LIBTOOL_VERSION == 1 */ + if (flags & PLUGIN_FLAGS_GLOBAL) + WARNING ("plugin_load_file: The global flag is not supported, " + "libtool 2 is required for this."); + dlh = lt_dlopen (file); +#endif + + if (dlh == NULL) + { + char errbuf[1024] = ""; + + ssnprintf (errbuf, sizeof (errbuf), + "lt_dlopen (\"%s\") failed: %s. " - "The most common cause for this problem are " ++ "The most common cause for this problem is " + "missing dependencies. Use ldd(1) to check " + "the dependencies of the plugin " + "/ shared object.", + file, lt_dlerror ()); + + ERROR ("%s", errbuf); + /* Make sure this is printed to STDERR in any case, but also + * make sure it's printed only once. */ + if (list_log != NULL) + fprintf (stderr, "ERROR: %s\n", errbuf); + + return (1); + } + + if ((reg_handle = (void (*) (void)) lt_dlsym (dlh, "module_register")) == NULL) + { + WARNING ("Couldn't find symbol \"module_register\" in \"%s\": %s\n", + file, lt_dlerror ()); + lt_dlclose (dlh); + return (-1); + } + + (*reg_handle) (); + + return (0); +} + +static void *plugin_read_thread (void __attribute__((unused)) *args) +{ + while (read_loop != 0) + { + read_func_t *rf; + plugin_ctx_t old_ctx; + cdtime_t now; + int status; + int rf_type; + int rc; + + /* Get the read function that needs to be read next. + * We don't need to hold "read_lock" for the heap, but we need + * to call c_heap_get_root() and pthread_cond_wait() in the + * same protected block. */ + pthread_mutex_lock (&read_lock); + rf = c_heap_get_root (read_heap); + if (rf == NULL) + { + pthread_cond_wait (&read_cond, &read_lock); + pthread_mutex_unlock (&read_lock); + continue; + } + pthread_mutex_unlock (&read_lock); + + if (rf->rf_interval == 0) + { + /* this should not happen, because the interval is set + * for each plugin when loading it + * XXX: issue a warning? */ + rf->rf_interval = plugin_get_interval (); + rf->rf_effective_interval = rf->rf_interval; + + rf->rf_next_read = cdtime (); + } + + /* sleep until this entry is due, + * using pthread_cond_timedwait */ + pthread_mutex_lock (&read_lock); + /* In pthread_cond_timedwait, spurious wakeups are possible + * (and really happen, at least on NetBSD with > 1 CPU), thus + * we need to re-evaluate the condition every time + * pthread_cond_timedwait returns. */ + rc = 0; + while ((read_loop != 0) + && (cdtime () < rf->rf_next_read) + && rc == 0) + { + struct timespec ts = { 0 }; + + CDTIME_T_TO_TIMESPEC (rf->rf_next_read, &ts); + + rc = pthread_cond_timedwait (&read_cond, &read_lock, + &ts); + } + + /* Must hold `read_lock' when accessing `rf->rf_type'. */ + rf_type = rf->rf_type; + pthread_mutex_unlock (&read_lock); + + /* Check if we're supposed to stop.. This may have interrupted + * the sleep, too. */ + if (read_loop == 0) + { + /* Insert `rf' again, so it can be free'd correctly */ + c_heap_insert (read_heap, rf); + break; + } + + /* The entry has been marked for deletion. The linked list + * entry has already been removed by `plugin_unregister_read'. + * All we have to do here is free the `read_func_t' and + * continue. */ + if (rf_type == RF_REMOVE) + { + DEBUG ("plugin_read_thread: Destroying the `%s' " + "callback.", rf->rf_name); + sfree (rf->rf_name); + destroy_callback ((callback_func_t *) rf); + rf = NULL; + continue; + } + + DEBUG ("plugin_read_thread: Handling `%s'.", rf->rf_name); + + old_ctx = plugin_set_ctx (rf->rf_ctx); + + if (rf_type == RF_SIMPLE) + { + int (*callback) (void); + + callback = rf->rf_callback; + status = (*callback) (); + } + else + { + plugin_read_cb callback; + + assert (rf_type == RF_COMPLEX); + + callback = rf->rf_callback; + status = (*callback) (&rf->rf_udata); + } + + plugin_set_ctx (old_ctx); + + /* If the function signals failure, we will increase the + * intervals in which it will be called. */ + if (status != 0) + { + rf->rf_effective_interval *= 2; + if (rf->rf_effective_interval > max_read_interval) + rf->rf_effective_interval = max_read_interval; + + NOTICE ("read-function of plugin `%s' failed. " + "Will suspend it for %.3f seconds.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_effective_interval)); + } + else + { + /* Success: Restore the interval, if it was changed. */ + rf->rf_effective_interval = rf->rf_interval; + } + + /* update the ``next read due'' field */ + now = cdtime (); + + DEBUG ("plugin_read_thread: Effective interval of the " + "%s plugin is %.3f seconds.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_effective_interval)); + + /* Calculate the next (absolute) time at which this function + * should be called. */ + rf->rf_next_read += rf->rf_effective_interval; + + /* Check, if `rf_next_read' is in the past. */ + if (rf->rf_next_read < now) + { + /* `rf_next_read' is in the past. Insert `now' + * so this value doesn't trail off into the + * past too much. */ + rf->rf_next_read = now; + } + + DEBUG ("plugin_read_thread: Next read of the %s plugin at %.3f.", + rf->rf_name, + CDTIME_T_TO_DOUBLE (rf->rf_next_read)); + + /* Re-insert this read function into the heap again. */ + c_heap_insert (read_heap, rf); + } /* while (read_loop) */ + + pthread_exit (NULL); + return ((void *) 0); +} /* void *plugin_read_thread */ + +static void start_read_threads (int num) +{ + int i; + + if (read_threads != NULL) + return; + + read_threads = (pthread_t *) calloc (num, sizeof (pthread_t)); + if (read_threads == NULL) + { + ERROR ("plugin: start_read_threads: calloc failed."); + return; + } + + read_threads_num = 0; + for (i = 0; i < num; i++) + { + if (pthread_create (read_threads + read_threads_num, NULL, + plugin_read_thread, NULL) == 0) + { + read_threads_num++; + } + else + { + ERROR ("plugin: start_read_threads: pthread_create failed."); + return; + } + } /* for (i) */ +} /* void start_read_threads */ + +static void stop_read_threads (void) +{ + int i; + + if (read_threads == NULL) + return; + + INFO ("collectd: Stopping %i read threads.", read_threads_num); + + pthread_mutex_lock (&read_lock); + read_loop = 0; + DEBUG ("plugin: stop_read_threads: Signalling `read_cond'"); + pthread_cond_broadcast (&read_cond); + pthread_mutex_unlock (&read_lock); + + for (i = 0; i < read_threads_num; i++) + { + if (pthread_join (read_threads[i], NULL) != 0) + { + ERROR ("plugin: stop_read_threads: pthread_join failed."); + } + read_threads[i] = (pthread_t) 0; + } + sfree (read_threads); + read_threads_num = 0; +} /* void stop_read_threads */ + +static void plugin_value_list_free (value_list_t *vl) /* {{{ */ +{ + if (vl == NULL) + return; + + meta_data_destroy (vl->meta); + sfree (vl->values); + sfree (vl); +} /* }}} void plugin_value_list_free */ + +static value_list_t *plugin_value_list_clone (value_list_t const *vl_orig) /* {{{ */ +{ + value_list_t *vl; + + if (vl_orig == NULL) + return (NULL); + + vl = malloc (sizeof (*vl)); + if (vl == NULL) + return (NULL); + memcpy (vl, vl_orig, sizeof (*vl)); + + vl->values = calloc (vl_orig->values_len, sizeof (*vl->values)); + if (vl->values == NULL) + { + plugin_value_list_free (vl); + return (NULL); + } + memcpy (vl->values, vl_orig->values, + vl_orig->values_len * sizeof (*vl->values)); + + vl->meta = meta_data_clone (vl->meta); + if ((vl_orig->meta != NULL) && (vl->meta == NULL)) + { + plugin_value_list_free (vl); + return (NULL); + } + + if (vl->time == 0) + vl->time = cdtime (); + + /* Fill in the interval from the thread context, if it is zero. */ + if (vl->interval == 0) + { + plugin_ctx_t ctx = plugin_get_ctx (); + + if (ctx.interval != 0) + vl->interval = ctx.interval; + else + { + char name[6 * DATA_MAX_NAME_LEN]; + FORMAT_VL (name, sizeof (name), vl); + ERROR ("plugin_value_list_clone: Unable to determine " + "interval from context for " + "value list \"%s\". " + "This indicates a broken plugin. " + "Please report this problem to the " + "collectd mailing list or at " + ".", name); + vl->interval = cf_get_default_interval (); + } + } + + return (vl); +} /* }}} value_list_t *plugin_value_list_clone */ + +static int plugin_write_enqueue (value_list_t const *vl) /* {{{ */ +{ + write_queue_t *q; + + q = malloc (sizeof (*q)); + if (q == NULL) + return (ENOMEM); + q->next = NULL; + + q->vl = plugin_value_list_clone (vl); + if (q->vl == NULL) + { + sfree (q); + return (ENOMEM); + } + + /* Store context of caller (read plugin); otherwise, it would not be + * available to the write plugins when actually dispatching the + * value-list later on. */ + q->ctx = plugin_get_ctx (); + + pthread_mutex_lock (&write_lock); + + if (write_queue_tail == NULL) + { + write_queue_head = q; + write_queue_tail = q; + write_queue_length = 1; + } + else + { + write_queue_tail->next = q; + write_queue_tail = q; + write_queue_length += 1; + } + + pthread_cond_signal (&write_cond); + pthread_mutex_unlock (&write_lock); + + return (0); +} /* }}} int plugin_write_enqueue */ + +static value_list_t *plugin_write_dequeue (void) /* {{{ */ +{ + write_queue_t *q; + value_list_t *vl; + + pthread_mutex_lock (&write_lock); + + while (write_loop && (write_queue_head == NULL)) + pthread_cond_wait (&write_cond, &write_lock); + + if (write_queue_head == NULL) + { + pthread_mutex_unlock (&write_lock); + return (NULL); + } + + q = write_queue_head; + write_queue_head = q->next; + write_queue_length -= 1; + if (write_queue_head == NULL) { + write_queue_tail = NULL; + assert(0 == write_queue_length); + } + + pthread_mutex_unlock (&write_lock); + + (void) plugin_set_ctx (q->ctx); + + vl = q->vl; + sfree (q); + return (vl); +} /* }}} value_list_t *plugin_write_dequeue */ + +static void *plugin_write_thread (void __attribute__((unused)) *args) /* {{{ */ +{ + while (write_loop) + { + value_list_t *vl = plugin_write_dequeue (); + if (vl == NULL) + continue; + + plugin_dispatch_values_internal (vl); + + plugin_value_list_free (vl); + } + + pthread_exit (NULL); + return ((void *) 0); +} /* }}} void *plugin_write_thread */ + +static void start_write_threads (size_t num) /* {{{ */ +{ + size_t i; + + if (write_threads != NULL) + return; + + write_threads = (pthread_t *) calloc (num, sizeof (pthread_t)); + if (write_threads == NULL) + { + ERROR ("plugin: start_write_threads: calloc failed."); + return; + } + + write_threads_num = 0; + for (i = 0; i < num; i++) + { + int status; + + status = pthread_create (write_threads + write_threads_num, + /* attr = */ NULL, + plugin_write_thread, + /* arg = */ NULL); + if (status != 0) + { + char errbuf[1024]; + ERROR ("plugin: start_write_threads: pthread_create failed " + "with status %i (%s).", status, + sstrerror (status, errbuf, sizeof (errbuf))); + return; + } + + write_threads_num++; + } /* for (i) */ +} /* }}} void start_write_threads */ + +static void stop_write_threads (void) /* {{{ */ +{ + write_queue_t *q; + int i; + + if (write_threads == NULL) + return; + + INFO ("collectd: Stopping %zu write threads.", write_threads_num); + + pthread_mutex_lock (&write_lock); + write_loop = 0; + DEBUG ("plugin: stop_write_threads: Signalling `write_cond'"); + pthread_cond_broadcast (&write_cond); + pthread_mutex_unlock (&write_lock); + + for (i = 0; i < write_threads_num; i++) + { + if (pthread_join (write_threads[i], NULL) != 0) + { + ERROR ("plugin: stop_write_threads: pthread_join failed."); + } + write_threads[i] = (pthread_t) 0; + } + sfree (write_threads); + write_threads_num = 0; + + pthread_mutex_lock (&write_lock); + i = 0; + for (q = write_queue_head; q != NULL; ) + { + write_queue_t *q1 = q; + plugin_value_list_free (q->vl); + q = q->next; + sfree (q1); + i++; + } + write_queue_head = NULL; + write_queue_tail = NULL; + write_queue_length = 0; + pthread_mutex_unlock (&write_lock); + + if (i > 0) + { + WARNING ("plugin: %i value list%s left after shutting down " + "the write threads.", + i, (i == 1) ? " was" : "s were"); + } +} /* }}} void stop_write_threads */ + +/* + * Public functions + */ +void plugin_set_dir (const char *dir) +{ + if (plugindir != NULL) + free (plugindir); + + if (dir == NULL) + plugindir = NULL; + else if ((plugindir = strdup (dir)) == NULL) + { + char errbuf[1024]; + ERROR ("strdup failed: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + } +} + +static _Bool plugin_is_loaded (char const *name) +{ + int status; + + if (plugins_loaded == NULL) + plugins_loaded = c_avl_create ((void *) strcasecmp); + assert (plugins_loaded != NULL); + + status = c_avl_get (plugins_loaded, name, /* ret_value = */ NULL); + return (status == 0); +} + +static int plugin_mark_loaded (char const *name) +{ + char *name_copy; + int status; + + name_copy = strdup (name); + if (name_copy == NULL) + return (ENOMEM); + + status = c_avl_insert (plugins_loaded, + /* key = */ name_copy, /* value = */ NULL); + return (status); +} + +static void plugin_free_loaded () +{ + void *key; + void *value; + + if (plugins_loaded == NULL) + return; + + while (c_avl_pick (plugins_loaded, &key, &value) == 0) + { + sfree (key); + assert (value == NULL); + } + + c_avl_destroy (plugins_loaded); + plugins_loaded = NULL; +} + +#define BUFSIZE 512 +int plugin_load (char const *plugin_name, uint32_t flags) +{ + DIR *dh; + const char *dir; + char filename[BUFSIZE] = ""; + char typename[BUFSIZE]; + int typename_len; + int ret; + struct stat statbuf; + struct dirent *de; + int status; + + if (plugin_name == NULL) + return (EINVAL); + + /* Check if plugin is already loaded and don't do anything in this + * case. */ + if (plugin_is_loaded (plugin_name)) + return (0); + + dir = plugin_get_dir (); + ret = 1; + + /* + * XXX: Magic at work: + * + * Some of the language bindings, for example the Python and Perl + * plugins, need to be able to export symbols to the scripts they run. + * For this to happen, the "Globals" flag needs to be set. + * Unfortunately, this technical detail is hard to explain to the + * average user and she shouldn't have to worry about this, ideally. + * So in order to save everyone's sanity use a different default for a + * handful of special plugins. --octo + */ + if ((strcasecmp ("perl", plugin_name) == 0) + || (strcasecmp ("python", plugin_name) == 0)) + flags |= PLUGIN_FLAGS_GLOBAL; + + /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the + * type when matching the filename */ + status = ssnprintf (typename, sizeof (typename), "%s.so", plugin_name); + if ((status < 0) || ((size_t) status >= sizeof (typename))) + { + WARNING ("plugin_load: Filename too long: \"%s.so\"", plugin_name); + return (-1); + } + typename_len = strlen (typename); + + if ((dh = opendir (dir)) == NULL) + { + char errbuf[1024]; + ERROR ("plugin_load: opendir (%s) failed: %s", dir, + sstrerror (errno, errbuf, sizeof (errbuf))); + return (-1); + } + + while ((de = readdir (dh)) != NULL) + { + if (strncasecmp (de->d_name, typename, typename_len)) + continue; + + status = ssnprintf (filename, sizeof (filename), + "%s/%s", dir, de->d_name); + if ((status < 0) || ((size_t) status >= sizeof (filename))) + { + WARNING ("plugin_load: Filename too long: \"%s/%s\"", + dir, de->d_name); + continue; + } + + if (lstat (filename, &statbuf) == -1) + { + char errbuf[1024]; + WARNING ("plugin_load: stat (\"%s\") failed: %s", + filename, + sstrerror (errno, errbuf, sizeof (errbuf))); + continue; + } + else if (!S_ISREG (statbuf.st_mode)) + { + /* don't follow symlinks */ + WARNING ("plugin_load: %s is not a regular file.", + filename); + continue; + } + + status = plugin_load_file (filename, flags); + if (status == 0) + { + /* success */ + plugin_mark_loaded (plugin_name); + ret = 0; + break; + } + else + { + ERROR ("plugin_load: Load plugin \"%s\" failed with " + "status %i.", plugin_name, status); + } + } + + closedir (dh); + + if (filename[0] == 0) + ERROR ("plugin_load: Could not find plugin \"%s\" in %s", + plugin_name, dir); + + return (ret); +} + +/* + * The `register_*' functions follow + */ +int plugin_register_config (const char *name, + int (*callback) (const char *key, const char *val), + const char **keys, int keys_num) +{ + cf_register (name, callback, keys, keys_num); + return (0); +} /* int plugin_register_config */ + +int plugin_register_complex_config (const char *type, + int (*callback) (oconfig_item_t *)) +{ + return (cf_register_complex (type, callback)); +} /* int plugin_register_complex_config */ + +int plugin_register_init (const char *name, + int (*callback) (void)) +{ + return (create_register_callback (&list_init, name, (void *) callback, + /* user_data = */ NULL)); +} /* plugin_register_init */ + +static int plugin_compare_read_func (const void *arg0, const void *arg1) +{ + const read_func_t *rf0; + const read_func_t *rf1; + + rf0 = arg0; + rf1 = arg1; + + if (rf0->rf_next_read < rf1->rf_next_read) + return (-1); + else if (rf0->rf_next_read > rf1->rf_next_read) + return (1); + else + return (0); +} /* int plugin_compare_read_func */ + +/* Add a read function to both, the heap and a linked list. The linked list if + * used to look-up read functions, especially for the remove function. The heap + * is used to determine which plugin to read next. */ +static int plugin_insert_read (read_func_t *rf) +{ + int status; + llentry_t *le; + + rf->rf_next_read = cdtime (); + rf->rf_effective_interval = rf->rf_interval; + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + read_list = llist_create (); + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: read_list failed."); + return (-1); + } + } + + if (read_heap == NULL) + { + read_heap = c_heap_create (plugin_compare_read_func); + if (read_heap == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: c_heap_create failed."); + return (-1); + } + } + + le = llist_search (read_list, rf->rf_name); + if (le != NULL) + { + pthread_mutex_unlock (&read_lock); + WARNING ("The read function \"%s\" is already registered. " + "Check for duplicate \"LoadPlugin\" lines " + "in your configuration!", + rf->rf_name); + return (EINVAL); + } + + le = llentry_create (rf->rf_name, rf); + if (le == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: llentry_create failed."); + return (-1); + } + + status = c_heap_insert (read_heap, rf); + if (status != 0) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: c_heap_insert failed."); + llentry_destroy (le); + return (-1); + } + + /* This does not fail. */ + llist_append (read_list, le); + + /* Wake up all the read threads. */ + pthread_cond_broadcast (&read_cond); + pthread_mutex_unlock (&read_lock); + return (0); +} /* int plugin_insert_read */ + +int plugin_register_read (const char *name, + int (*callback) (void)) +{ + read_func_t *rf; + int status; + + rf = malloc (sizeof (*rf)); + if (rf == NULL) + { + ERROR ("plugin_register_read: malloc failed."); + return (ENOMEM); + } + + memset (rf, 0, sizeof (read_func_t)); + rf->rf_callback = (void *) callback; + rf->rf_udata.data = NULL; + rf->rf_udata.free_func = NULL; + rf->rf_ctx = plugin_get_ctx (); + rf->rf_group[0] = '\0'; + rf->rf_name = strdup (name); + rf->rf_type = RF_SIMPLE; + rf->rf_interval = plugin_get_interval (); + + status = plugin_insert_read (rf); + if (status != 0) + sfree (rf); + + return (status); +} /* int plugin_register_read */ + +int plugin_register_complex_read (const char *group, const char *name, + plugin_read_cb callback, + const struct timespec *interval, + user_data_t *user_data) +{ + read_func_t *rf; + int status; + + rf = malloc (sizeof (*rf)); + if (rf == NULL) + { + ERROR ("plugin_register_complex_read: malloc failed."); + return (ENOMEM); + } + + memset (rf, 0, sizeof (read_func_t)); + rf->rf_callback = (void *) callback; + if (group != NULL) + sstrncpy (rf->rf_group, group, sizeof (rf->rf_group)); + else + rf->rf_group[0] = '\0'; + rf->rf_name = strdup (name); + rf->rf_type = RF_COMPLEX; + if (interval != NULL) + rf->rf_interval = TIMESPEC_TO_CDTIME_T (interval); + else + rf->rf_interval = plugin_get_interval (); + + /* Set user data */ + if (user_data == NULL) + { + rf->rf_udata.data = NULL; + rf->rf_udata.free_func = NULL; + } + else + { + rf->rf_udata = *user_data; + } + + rf->rf_ctx = plugin_get_ctx (); + + status = plugin_insert_read (rf); + if (status != 0) + sfree (rf); + + return (status); +} /* int plugin_register_complex_read */ + +int plugin_register_write (const char *name, + plugin_write_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_write, name, + (void *) callback, ud)); +} /* int plugin_register_write */ + +int plugin_register_flush (const char *name, + plugin_flush_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_flush, name, + (void *) callback, ud)); +} /* int plugin_register_flush */ + +int plugin_register_missing (const char *name, + plugin_missing_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_missing, name, + (void *) callback, ud)); +} /* int plugin_register_missing */ + +int plugin_register_shutdown (const char *name, + int (*callback) (void)) +{ + return (create_register_callback (&list_shutdown, name, + (void *) callback, /* user_data = */ NULL)); +} /* int plugin_register_shutdown */ + +static void plugin_free_data_sets (void) +{ + void *key; + void *value; + + if (data_sets == NULL) + return; + + while (c_avl_pick (data_sets, &key, &value) == 0) + { + data_set_t *ds = value; + /* key is a pointer to ds->type */ + + sfree (ds->ds); + sfree (ds); + } + + c_avl_destroy (data_sets); + data_sets = NULL; +} /* void plugin_free_data_sets */ + +int plugin_register_data_set (const data_set_t *ds) +{ + data_set_t *ds_copy; + int i; + + if ((data_sets != NULL) + && (c_avl_get (data_sets, ds->type, NULL) == 0)) + { + NOTICE ("Replacing DS `%s' with another version.", ds->type); + plugin_unregister_data_set (ds->type); + } + else if (data_sets == NULL) + { + data_sets = c_avl_create ((int (*) (const void *, const void *)) strcmp); + if (data_sets == NULL) + return (-1); + } + + ds_copy = (data_set_t *) malloc (sizeof (data_set_t)); + if (ds_copy == NULL) + return (-1); + memcpy(ds_copy, ds, sizeof (data_set_t)); + + ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t) + * ds->ds_num); + if (ds_copy->ds == NULL) + { + free (ds_copy); + return (-1); + } + + for (i = 0; i < ds->ds_num; i++) + memcpy (ds_copy->ds + i, ds->ds + i, sizeof (data_source_t)); + + return (c_avl_insert (data_sets, (void *) ds_copy->type, (void *) ds_copy)); +} /* int plugin_register_data_set */ + +int plugin_register_log (const char *name, + plugin_log_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_log, name, + (void *) callback, ud)); +} /* int plugin_register_log */ + +int plugin_register_notification (const char *name, + plugin_notification_cb callback, user_data_t *ud) +{ + return (create_register_callback (&list_notification, name, + (void *) callback, ud)); +} /* int plugin_register_log */ + +int plugin_unregister_config (const char *name) +{ + cf_unregister (name); + return (0); +} /* int plugin_unregister_config */ + +int plugin_unregister_complex_config (const char *name) +{ + cf_unregister_complex (name); + return (0); +} /* int plugin_unregister_complex_config */ + +int plugin_unregister_init (const char *name) +{ + return (plugin_unregister (list_init, name)); +} + +int plugin_unregister_read (const char *name) /* {{{ */ +{ + llentry_t *le; + read_func_t *rf; + + if (name == NULL) + return (-ENOENT); + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + return (-ENOENT); + } + + le = llist_search (read_list, name); + if (le == NULL) + { + pthread_mutex_unlock (&read_lock); + WARNING ("plugin_unregister_read: No such read function: %s", + name); + return (-ENOENT); + } + + llist_remove (read_list, le); + + rf = le->value; + assert (rf != NULL); + rf->rf_type = RF_REMOVE; + + pthread_mutex_unlock (&read_lock); + + llentry_destroy (le); + + DEBUG ("plugin_unregister_read: Marked `%s' for removal.", name); + + return (0); +} /* }}} int plugin_unregister_read */ + +static int compare_read_func_group (llentry_t *e, void *ud) /* {{{ */ +{ + read_func_t *rf = e->value; + char *group = ud; + + return strcmp (rf->rf_group, (const char *)group); +} /* }}} int compare_read_func_group */ + +int plugin_unregister_read_group (const char *group) /* {{{ */ +{ + llentry_t *le; + read_func_t *rf; + + int found = 0; + + if (group == NULL) + return (-ENOENT); + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + return (-ENOENT); + } + + while (42) + { + le = llist_search_custom (read_list, + compare_read_func_group, (void *)group); + + if (le == NULL) + break; + + ++found; + + llist_remove (read_list, le); + + rf = le->value; + assert (rf != NULL); + rf->rf_type = RF_REMOVE; + + llentry_destroy (le); + + DEBUG ("plugin_unregister_read_group: " + "Marked `%s' (group `%s') for removal.", + rf->rf_name, group); + } + + pthread_mutex_unlock (&read_lock); + + if (found == 0) + { + WARNING ("plugin_unregister_read_group: No such " + "group of read function: %s", group); + return (-ENOENT); + } + + return (0); +} /* }}} int plugin_unregister_read_group */ + +int plugin_unregister_write (const char *name) +{ + return (plugin_unregister (list_write, name)); +} + +int plugin_unregister_flush (const char *name) +{ + return (plugin_unregister (list_flush, name)); +} + +int plugin_unregister_missing (const char *name) +{ + return (plugin_unregister (list_missing, name)); +} + +int plugin_unregister_shutdown (const char *name) +{ + return (plugin_unregister (list_shutdown, name)); +} + +int plugin_unregister_data_set (const char *name) +{ + data_set_t *ds; + + if (data_sets == NULL) + return (-1); + + if (c_avl_remove (data_sets, name, NULL, (void *) &ds) != 0) + return (-1); + + sfree (ds->ds); + sfree (ds); + + return (0); +} /* int plugin_unregister_data_set */ + +int plugin_unregister_log (const char *name) +{ + return (plugin_unregister (list_log, name)); +} + +int plugin_unregister_notification (const char *name) +{ + return (plugin_unregister (list_notification, name)); +} + +void plugin_init_all (void) +{ + char const *chain_name; + long write_threads_num; + llentry_t *le; + int status; + + /* Init the value cache */ + uc_init (); + + if (IS_TRUE (global_option_get ("CollectInternalStats"))) + record_statistics = 1; + + chain_name = global_option_get ("PreCacheChain"); + pre_cache_chain = fc_chain_get_by_name (chain_name); + + chain_name = global_option_get ("PostCacheChain"); + post_cache_chain = fc_chain_get_by_name (chain_name); + + write_limit_high = global_option_get_long ("WriteQueueLimitHigh", + /* default = */ 0); + if (write_limit_high < 0) + { + ERROR ("WriteQueueLimitHigh must be positive or zero."); + write_limit_high = 0; + } + + write_limit_low = global_option_get_long ("WriteQueueLimitLow", + /* default = */ write_limit_high / 2); + if (write_limit_low < 0) + { + ERROR ("WriteQueueLimitLow must be positive or zero."); + write_limit_low = write_limit_high / 2; + } + else if (write_limit_low > write_limit_high) + { + ERROR ("WriteQueueLimitLow must not be larger than " + "WriteQueueLimitHigh."); + write_limit_low = write_limit_high; + } + + write_threads_num = global_option_get_long ("WriteThreads", + /* default = */ 5); + if (write_threads_num < 1) + { + ERROR ("WriteThreads must be positive."); + write_threads_num = 5; + } + + start_write_threads ((size_t) write_threads_num); + + if ((list_init == NULL) && (read_heap == NULL)) + return; + + /* Calling all init callbacks before checking if read callbacks + * are available allows the init callbacks to register the read + * callback. */ + le = llist_head (list_init); + while (le != NULL) + { + callback_func_t *cf; + plugin_init_cb callback; + plugin_ctx_t old_ctx; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + status = (*callback) (); + plugin_set_ctx (old_ctx); + + if (status != 0) + { + ERROR ("Initialization of plugin `%s' " + "failed with status %i. " + "Plugin will be unloaded.", + le->key, status); + /* Plugins that register read callbacks from the init + * callback should take care of appropriate error + * handling themselves. */ + /* FIXME: Unload _all_ functions */ + plugin_unregister_read (le->key); + } + + le = le->next; + } + + max_read_interval = global_option_get_time ("MaxReadInterval", + DEFAULT_MAX_READ_INTERVAL); + + /* Start read-threads */ + if (read_heap != NULL) + { + const char *rt; + int num; + + rt = global_option_get ("ReadThreads"); + num = atoi (rt); + if (num != -1) + start_read_threads ((num > 0) ? num : 5); + } +} /* void plugin_init_all */ + +/* TODO: Rename this function. */ +void plugin_read_all (void) +{ + if(record_statistics) { + plugin_update_internal_statistics (); + } + uc_check_timeout (); + + return; +} /* void plugin_read_all */ + +/* Read function called when the `-T' command line argument is given. */ +int plugin_read_all_once (void) +{ + int status; + int return_status = 0; + + if (read_heap == NULL) + { + NOTICE ("No read-functions are registered."); + return (0); + } + + while (42) + { + read_func_t *rf; + plugin_ctx_t old_ctx; + + rf = c_heap_get_root (read_heap); + if (rf == NULL) + break; + + old_ctx = plugin_set_ctx (rf->rf_ctx); + + if (rf->rf_type == RF_SIMPLE) + { + int (*callback) (void); + + callback = rf->rf_callback; + status = (*callback) (); + } + else + { + plugin_read_cb callback; + + callback = rf->rf_callback; + status = (*callback) (&rf->rf_udata); + } + + plugin_set_ctx (old_ctx); + + if (status != 0) + { + NOTICE ("read-function of plugin `%s' failed.", + rf->rf_name); + return_status = -1; + } + + destroy_callback ((void *) rf); + } + + return (return_status); +} /* int plugin_read_all_once */ + +int plugin_write (const char *plugin, /* {{{ */ + const data_set_t *ds, const value_list_t *vl) +{ + llentry_t *le; + int status; + + if (vl == NULL) + return (EINVAL); + + if (list_write == NULL) + return (ENOENT); + + if (ds == NULL) + { + ds = plugin_get_ds (vl->type); + if (ds == NULL) + { + ERROR ("plugin_write: Unable to lookup type `%s'.", vl->type); + return (ENOENT); + } + } + + if (plugin == NULL) + { + int success = 0; + int failure = 0; + + le = llist_head (list_write); + while (le != NULL) + { + callback_func_t *cf = le->value; + plugin_write_cb callback; + + /* do not switch plugin context; rather keep the context (interval) + * information of the calling read plugin */ + + DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); + callback = cf->cf_callback; + status = (*callback) (ds, vl, &cf->cf_udata); + if (status != 0) + failure++; + else + success++; + + le = le->next; + } + + if ((success == 0) && (failure != 0)) + status = -1; + else + status = 0; + } + else /* plugin != NULL */ + { + callback_func_t *cf; + plugin_write_cb callback; + + le = llist_head (list_write); + while (le != NULL) + { + if (strcasecmp (plugin, le->key) == 0) + break; + + le = le->next; + } + + if (le == NULL) + return (ENOENT); + + cf = le->value; + + /* do not switch plugin context; rather keep the context (interval) + * information of the calling read plugin */ + + DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); + callback = cf->cf_callback; + status = (*callback) (ds, vl, &cf->cf_udata); + } + + return (status); +} /* }}} int plugin_write */ + +int plugin_flush (const char *plugin, cdtime_t timeout, const char *identifier) +{ + llentry_t *le; + + if (list_flush == NULL) + return (0); + + le = llist_head (list_flush); + while (le != NULL) + { + callback_func_t *cf; + plugin_flush_cb callback; + plugin_ctx_t old_ctx; + + if ((plugin != NULL) + && (strcmp (plugin, le->key) != 0)) + { + le = le->next; + continue; + } + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + (*callback) (timeout, identifier, &cf->cf_udata); + + plugin_set_ctx (old_ctx); + + le = le->next; + } + return (0); +} /* int plugin_flush */ + +void plugin_shutdown_all (void) +{ + llentry_t *le; + + stop_read_threads (); + + destroy_all_callbacks (&list_init); + + pthread_mutex_lock (&read_lock); + llist_destroy (read_list); + read_list = NULL; + pthread_mutex_unlock (&read_lock); + + destroy_read_heap (); + + plugin_flush (/* plugin = */ NULL, + /* timeout = */ 0, + /* identifier = */ NULL); + + le = NULL; + if (list_shutdown != NULL) + le = llist_head (list_shutdown); + + while (le != NULL) + { + callback_func_t *cf; + plugin_shutdown_cb callback; + plugin_ctx_t old_ctx; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + /* Advance the pointer before calling the callback allows + * shutdown functions to unregister themselves. If done the + * other way around the memory `le' points to will be freed + * after callback returns. */ + le = le->next; + + (*callback) (); + + plugin_set_ctx (old_ctx); + } + + stop_write_threads (); + + /* Write plugins which use the `user_data' pointer usually need the + * same data available to the flush callback. If this is the case, set + * the free_function to NULL when registering the flush callback and to + * the real free function when registering the write callback. This way + * the data isn't freed twice. */ + destroy_all_callbacks (&list_flush); + destroy_all_callbacks (&list_missing); + destroy_all_callbacks (&list_write); + + destroy_all_callbacks (&list_notification); + destroy_all_callbacks (&list_shutdown); + destroy_all_callbacks (&list_log); + + plugin_free_loaded (); + plugin_free_data_sets (); +} /* void plugin_shutdown_all */ + +int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */ +{ + llentry_t *le; + + if (list_missing == NULL) + return (0); + + le = llist_head (list_missing); + while (le != NULL) + { + callback_func_t *cf; + plugin_missing_cb callback; + plugin_ctx_t old_ctx; + int status; + + cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + callback = cf->cf_callback; + + status = (*callback) (vl, &cf->cf_udata); + plugin_set_ctx (old_ctx); + if (status != 0) + { + if (status < 0) + { + ERROR ("plugin_dispatch_missing: Callback function \"%s\" " + "failed with status %i.", + le->key, status); + return (status); + } + else + { + return (0); + } + } + + le = le->next; + } + return (0); +} /* int }}} plugin_dispatch_missing */ + +static int plugin_dispatch_values_internal (value_list_t *vl) +{ + int status; + static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC; + + value_t *saved_values; + int saved_values_len; + + data_set_t *ds; + + int free_meta_data = 0; + + if ((vl == NULL) || (vl->type[0] == 0) + || (vl->values == NULL) || (vl->values_len < 1)) + { + ERROR ("plugin_dispatch_values: Invalid value list " + "from plugin %s.", vl->plugin); + return (-1); + } + + /* Free meta data only if the calling function didn't specify any. In + * this case matches and targets may add some and the calling function + * may not expect (and therefore free) that data. */ + if (vl->meta == NULL) + free_meta_data = 1; + + if (list_write == NULL) + c_complain_once (LOG_WARNING, &no_write_complaint, + "plugin_dispatch_values: No write callback has been " + "registered. Please load at least one output plugin, " + "if you want the collected data to be stored."); + + if (data_sets == NULL) + { + ERROR ("plugin_dispatch_values: No data sets registered. " + "Could the types database be read? Check " + "your `TypesDB' setting!"); + return (-1); + } + + if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0) + { + char ident[6 * DATA_MAX_NAME_LEN]; + + FORMAT_VL (ident, sizeof (ident), vl); + INFO ("plugin_dispatch_values: Dataset not found: %s " + "(from \"%s\"), check your types.db!", + vl->type, ident); + return (-1); + } + + /* Assured by plugin_value_list_clone(). The time is determined at + * _enqueue_ time. */ + assert (vl->time != 0); + assert (vl->interval != 0); + + DEBUG ("plugin_dispatch_values: time = %.3f; interval = %.3f; " + "host = %s; " + "plugin = %s; plugin_instance = %s; " + "type = %s; type_instance = %s;", + CDTIME_T_TO_DOUBLE (vl->time), + CDTIME_T_TO_DOUBLE (vl->interval), + vl->host, + vl->plugin, vl->plugin_instance, + vl->type, vl->type_instance); + +#if COLLECT_DEBUG + assert (0 == strcmp (ds->type, vl->type)); +#else + if (0 != strcmp (ds->type, vl->type)) + WARNING ("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)", + ds->type, vl->type); +#endif + +#if COLLECT_DEBUG + assert (ds->ds_num == vl->values_len); +#else + if (ds->ds_num != vl->values_len) + { + ERROR ("plugin_dispatch_values: ds->type = %s: " + "(ds->ds_num = %i) != " + "(vl->values_len = %i)", + ds->type, ds->ds_num, vl->values_len); + return (-1); + } +#endif + + escape_slashes (vl->host, sizeof (vl->host)); + escape_slashes (vl->plugin, sizeof (vl->plugin)); + escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance)); + escape_slashes (vl->type, sizeof (vl->type)); + escape_slashes (vl->type_instance, sizeof (vl->type_instance)); + + /* Copy the values. This way, we can assure `targets' that they get + * dynamically allocated values, which they can free and replace if + * they like. */ + if ((pre_cache_chain != NULL) || (post_cache_chain != NULL)) + { + saved_values = vl->values; + saved_values_len = vl->values_len; + + vl->values = (value_t *) calloc (vl->values_len, + sizeof (*vl->values)); + if (vl->values == NULL) + { + ERROR ("plugin_dispatch_values: calloc failed."); + vl->values = saved_values; + return (-1); + } + memcpy (vl->values, saved_values, + vl->values_len * sizeof (*vl->values)); + } + else /* if ((pre == NULL) && (post == NULL)) */ + { + saved_values = NULL; + saved_values_len = 0; + } + + if (pre_cache_chain != NULL) + { + status = fc_process_chain (ds, vl, pre_cache_chain); + if (status < 0) + { + WARNING ("plugin_dispatch_values: Running the " + "pre-cache chain failed with " + "status %i (%#x).", + status, status); + } + else if (status == FC_TARGET_STOP) + { + /* Restore the state of the value_list so that plugins + * don't get confused.. */ + if (saved_values != NULL) + { + free (vl->values); + vl->values = saved_values; + vl->values_len = saved_values_len; + } + return (0); + } + } + + /* Update the value cache */ + uc_update (ds, vl); + + if (post_cache_chain != NULL) + { + status = fc_process_chain (ds, vl, post_cache_chain); + if (status < 0) + { + WARNING ("plugin_dispatch_values: Running the " + "post-cache chain failed with " + "status %i (%#x).", + status, status); + } + } + else + fc_default_action (ds, vl); + + /* Restore the state of the value_list so that plugins don't get + * confused.. */ + if (saved_values != NULL) + { + free (vl->values); + vl->values = saved_values; + vl->values_len = saved_values_len; + } + + if ((free_meta_data != 0) && (vl->meta != NULL)) + { + meta_data_destroy (vl->meta); + vl->meta = NULL; + } + + return (0); +} /* int plugin_dispatch_values_internal */ + +static double get_drop_probability (void) /* {{{ */ +{ + long pos; + long size; + long wql; + + pthread_mutex_lock (&write_lock); + wql = write_queue_length; + pthread_mutex_unlock (&write_lock); + + if (wql < write_limit_low) + return (0.0); + if (wql >= write_limit_high) + return (1.0); + + pos = 1 + wql - write_limit_low; + size = 1 + write_limit_high - write_limit_low; + + return (((double) pos) / ((double) size)); +} /* }}} double get_drop_probability */ + +static _Bool check_drop_value (void) /* {{{ */ +{ + static cdtime_t last_message_time = 0; + static pthread_mutex_t last_message_lock = PTHREAD_MUTEX_INITIALIZER; + + double p; + double q; + int status; + + if (write_limit_high == 0) + return (0); + + p = get_drop_probability (); + if (p == 0.0) + return (0); + + status = pthread_mutex_trylock (&last_message_lock); + if (status == 0) + { + cdtime_t now; + + now = cdtime (); + if ((now - last_message_time) > TIME_T_TO_CDTIME_T (1)) + { + last_message_time = now; + ERROR ("plugin_dispatch_values: Low water mark " + "reached. Dropping %.0f%% of metrics.", + 100.0 * p); + } + pthread_mutex_unlock (&last_message_lock); + } + + if (p == 1.0) + return (1); + + q = cdrand_d (); + if (q > p) + return (1); + else + return (0); +} /* }}} _Bool check_drop_value */ + +int plugin_dispatch_values (value_list_t const *vl) +{ + int status; + static pthread_mutex_t statistics_lock = PTHREAD_MUTEX_INITIALIZER; + + if (check_drop_value ()) { + if(record_statistics) { + pthread_mutex_lock(&statistics_lock); + stats_values_dropped++; + pthread_mutex_unlock(&statistics_lock); + } + return (0); + } + + status = plugin_write_enqueue (vl); + if (status != 0) + { + char errbuf[1024]; + ERROR ("plugin_dispatch_values: plugin_write_enqueue failed " + "with status %i (%s).", status, + sstrerror (status, errbuf, sizeof (errbuf))); + return (status); + } + + return (0); +} + +__attribute__((sentinel)) +int plugin_dispatch_multivalue (value_list_t const *template, /* {{{ */ + _Bool store_percentage, ...) +{ + value_list_t *vl; + int failed = 0; + gauge_t sum = 0.0; + va_list ap; + + assert (template->values_len == 1); + + va_start (ap, store_percentage); + while (42) + { + char const *name; + gauge_t value; + + name = va_arg (ap, char const *); + if (name == NULL) + break; + + value = va_arg (ap, gauge_t); + if (!isnan (value)) + sum += value; + } + va_end (ap); + + vl = plugin_value_list_clone (template); + /* plugin_value_list_clone makes sure vl->time is set to non-zero. */ + if (store_percentage) + sstrncpy (vl->type, "percent", sizeof (vl->type)); + + va_start (ap, store_percentage); + while (42) + { + char const *name; + int status; + + /* Set the type instance. */ + name = va_arg (ap, char const *); + if (name == NULL) + break; + sstrncpy (vl->type_instance, name, sizeof (vl->type_instance)); + + /* Set the value. */ + vl->values[0].gauge = va_arg (ap, gauge_t); + if (store_percentage) + vl->values[0].gauge *= 100.0 / sum; + + status = plugin_write_enqueue (vl); + if (status != 0) + failed++; + } + va_end (ap); + + plugin_value_list_free (vl); + return (failed); +} /* }}} int plugin_dispatch_multivalue */ + +int plugin_dispatch_notification (const notification_t *notif) +{ + llentry_t *le; + /* Possible TODO: Add flap detection here */ + + DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; " + "time = %.3f; host = %s;", + notif->severity, notif->message, + CDTIME_T_TO_DOUBLE (notif->time), notif->host); + + /* Nobody cares for notifications */ + if (list_notification == NULL) + return (-1); + + le = llist_head (list_notification); + while (le != NULL) + { + callback_func_t *cf; + plugin_notification_cb callback; + int status; + + /* do not switch plugin context; rather keep the context + * (interval) information of the calling plugin */ + + cf = le->value; + callback = cf->cf_callback; + status = (*callback) (notif, &cf->cf_udata); + if (status != 0) + { + WARNING ("plugin_dispatch_notification: Notification " + "callback %s returned %i.", + le->key, status); + } + + le = le->next; + } + + return (0); +} /* int plugin_dispatch_notification */ + +void plugin_log (int level, const char *format, ...) +{ + char msg[1024]; + va_list ap; + llentry_t *le; + +#if !COLLECT_DEBUG + if (level >= LOG_DEBUG) + return; +#endif + + va_start (ap, format); + vsnprintf (msg, sizeof (msg), format, ap); + msg[sizeof (msg) - 1] = '\0'; + va_end (ap); + + if (list_log == NULL) + { + fprintf (stderr, "%s\n", msg); + return; + } + + le = llist_head (list_log); + while (le != NULL) + { + callback_func_t *cf; + plugin_log_cb callback; + + cf = le->value; + callback = cf->cf_callback; + + /* do not switch plugin context; rather keep the context + * (interval) information of the calling plugin */ + + (*callback) (level, msg, &cf->cf_udata); + + le = le->next; + } +} /* void plugin_log */ + +int parse_log_severity (const char *severity) +{ + int log_level = -1; + + if ((0 == strcasecmp (severity, "emerg")) + || (0 == strcasecmp (severity, "alert")) + || (0 == strcasecmp (severity, "crit")) + || (0 == strcasecmp (severity, "err"))) + log_level = LOG_ERR; + else if (0 == strcasecmp (severity, "warning")) + log_level = LOG_WARNING; + else if (0 == strcasecmp (severity, "notice")) + log_level = LOG_NOTICE; + else if (0 == strcasecmp (severity, "info")) + log_level = LOG_INFO; +#if COLLECT_DEBUG + else if (0 == strcasecmp (severity, "debug")) + log_level = LOG_DEBUG; +#endif /* COLLECT_DEBUG */ + + return (log_level); +} /* int parse_log_severity */ + +int parse_notif_severity (const char *severity) +{ + int notif_severity = -1; + + if (strcasecmp (severity, "FAILURE") == 0) + notif_severity = NOTIF_FAILURE; + else if (strcmp (severity, "OKAY") == 0) + notif_severity = NOTIF_OKAY; + else if ((strcmp (severity, "WARNING") == 0) + || (strcmp (severity, "WARN") == 0)) + notif_severity = NOTIF_WARNING; + + return (notif_severity); +} /* int parse_notif_severity */ + +const data_set_t *plugin_get_ds (const char *name) +{ + data_set_t *ds; + + if (data_sets == NULL) + { + ERROR ("plugin_get_ds: No data sets are defined yet."); + return (NULL); + } + + if (c_avl_get (data_sets, name, (void *) &ds) != 0) + { + DEBUG ("No such dataset registered: %s", name); + return (NULL); + } + + return (ds); +} /* data_set_t *plugin_get_ds */ + +static int plugin_notification_meta_add (notification_t *n, + const char *name, + enum notification_meta_type_e type, + const void *value) +{ + notification_meta_t *meta; + notification_meta_t *tail; + + if ((n == NULL) || (name == NULL) || (value == NULL)) + { + ERROR ("plugin_notification_meta_add: A pointer is NULL!"); + return (-1); + } + + meta = (notification_meta_t *) malloc (sizeof (notification_meta_t)); + if (meta == NULL) + { + ERROR ("plugin_notification_meta_add: malloc failed."); + return (-1); + } + memset (meta, 0, sizeof (notification_meta_t)); + + sstrncpy (meta->name, name, sizeof (meta->name)); + meta->type = type; + + switch (type) + { + case NM_TYPE_STRING: + { + meta->nm_value.nm_string = strdup ((const char *) value); + if (meta->nm_value.nm_string == NULL) + { + ERROR ("plugin_notification_meta_add: strdup failed."); + sfree (meta); + return (-1); + } + break; + } + case NM_TYPE_SIGNED_INT: + { + meta->nm_value.nm_signed_int = *((int64_t *) value); + break; + } + case NM_TYPE_UNSIGNED_INT: + { + meta->nm_value.nm_unsigned_int = *((uint64_t *) value); + break; + } + case NM_TYPE_DOUBLE: + { + meta->nm_value.nm_double = *((double *) value); + break; + } + case NM_TYPE_BOOLEAN: + { + meta->nm_value.nm_boolean = *((_Bool *) value); + break; + } + default: + { + ERROR ("plugin_notification_meta_add: Unknown type: %i", type); + sfree (meta); + return (-1); + } + } /* switch (type) */ + + meta->next = NULL; + tail = n->meta; + while ((tail != NULL) && (tail->next != NULL)) + tail = tail->next; + + if (tail == NULL) + n->meta = meta; + else + tail->next = meta; + + return (0); +} /* int plugin_notification_meta_add */ + +int plugin_notification_meta_add_string (notification_t *n, + const char *name, + const char *value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_STRING, value)); +} + +int plugin_notification_meta_add_signed_int (notification_t *n, + const char *name, + int64_t value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_SIGNED_INT, &value)); +} + +int plugin_notification_meta_add_unsigned_int (notification_t *n, + const char *name, + uint64_t value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_UNSIGNED_INT, &value)); +} + +int plugin_notification_meta_add_double (notification_t *n, + const char *name, + double value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_DOUBLE, &value)); +} + +int plugin_notification_meta_add_boolean (notification_t *n, + const char *name, + _Bool value) +{ + return (plugin_notification_meta_add (n, name, NM_TYPE_BOOLEAN, &value)); +} + +int plugin_notification_meta_copy (notification_t *dst, + const notification_t *src) +{ + notification_meta_t *meta; + + assert (dst != NULL); + assert (src != NULL); + assert (dst != src); + assert ((src->meta == NULL) || (src->meta != dst->meta)); + + for (meta = src->meta; meta != NULL; meta = meta->next) + { + if (meta->type == NM_TYPE_STRING) + plugin_notification_meta_add_string (dst, meta->name, + meta->nm_value.nm_string); + else if (meta->type == NM_TYPE_SIGNED_INT) + plugin_notification_meta_add_signed_int (dst, meta->name, + meta->nm_value.nm_signed_int); + else if (meta->type == NM_TYPE_UNSIGNED_INT) + plugin_notification_meta_add_unsigned_int (dst, meta->name, + meta->nm_value.nm_unsigned_int); + else if (meta->type == NM_TYPE_DOUBLE) + plugin_notification_meta_add_double (dst, meta->name, + meta->nm_value.nm_double); + else if (meta->type == NM_TYPE_BOOLEAN) + plugin_notification_meta_add_boolean (dst, meta->name, + meta->nm_value.nm_boolean); + } + + return (0); +} /* int plugin_notification_meta_copy */ + +int plugin_notification_meta_free (notification_meta_t *n) +{ + notification_meta_t *this; + notification_meta_t *next; + + if (n == NULL) + { + ERROR ("plugin_notification_meta_free: n == NULL!"); + return (-1); + } + + this = n; + while (this != NULL) + { + next = this->next; + + if (this->type == NM_TYPE_STRING) + { + free ((char *)this->nm_value.nm_string); + this->nm_value.nm_string = NULL; + } + sfree (this); + + this = next; + } + + return (0); +} /* int plugin_notification_meta_free */ + +static void plugin_ctx_destructor (void *ctx) +{ + sfree (ctx); +} /* void plugin_ctx_destructor */ + +static plugin_ctx_t ctx_init = { /* interval = */ 0 }; + +static plugin_ctx_t *plugin_ctx_create (void) +{ + plugin_ctx_t *ctx; + + ctx = malloc (sizeof (*ctx)); + if (ctx == NULL) { + char errbuf[1024]; + ERROR ("Failed to allocate plugin context: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return NULL; + } + + *ctx = ctx_init; + assert (plugin_ctx_key_initialized); + pthread_setspecific (plugin_ctx_key, ctx); + DEBUG("Created new plugin context."); + return (ctx); +} /* int plugin_ctx_create */ + +void plugin_init_ctx (void) +{ + pthread_key_create (&plugin_ctx_key, plugin_ctx_destructor); + plugin_ctx_key_initialized = 1; +} /* void plugin_init_ctx */ + +plugin_ctx_t plugin_get_ctx (void) +{ + plugin_ctx_t *ctx; + + assert (plugin_ctx_key_initialized); + ctx = pthread_getspecific (plugin_ctx_key); + + if (ctx == NULL) { + ctx = plugin_ctx_create (); + /* this must no happen -- exit() instead? */ + if (ctx == NULL) + return ctx_init; + } + + return (*ctx); +} /* plugin_ctx_t plugin_get_ctx */ + +plugin_ctx_t plugin_set_ctx (plugin_ctx_t ctx) +{ + plugin_ctx_t *c; + plugin_ctx_t old; + + assert (plugin_ctx_key_initialized); + c = pthread_getspecific (plugin_ctx_key); + + if (c == NULL) { + c = plugin_ctx_create (); + /* this must no happen -- exit() instead? */ + if (c == NULL) + return ctx_init; + } + + old = *c; + *c = ctx; + + return (old); +} /* void plugin_set_ctx */ + +cdtime_t plugin_get_interval (void) +{ + cdtime_t interval; + + interval = plugin_get_ctx().interval; + if (interval > 0) + return interval; + + return cf_get_default_interval (); +} /* cdtime_t plugin_get_interval */ + +typedef struct { + plugin_ctx_t ctx; + void *(*start_routine) (void *); + void *arg; +} plugin_thread_t; + +static void *plugin_thread_start (void *arg) +{ + plugin_thread_t *plugin_thread = arg; + + void *(*start_routine) (void *) = plugin_thread->start_routine; + void *plugin_arg = plugin_thread->arg; + + plugin_set_ctx (plugin_thread->ctx); + + free (plugin_thread); + + return start_routine (plugin_arg); +} /* void *plugin_thread_start */ + +int plugin_thread_create (pthread_t *thread, const pthread_attr_t *attr, + void *(*start_routine) (void *), void *arg) +{ + plugin_thread_t *plugin_thread; + + plugin_thread = malloc (sizeof (*plugin_thread)); + if (plugin_thread == NULL) + return -1; + + plugin_thread->ctx = plugin_get_ctx (); + plugin_thread->start_routine = start_routine; + plugin_thread->arg = arg; + + return pthread_create (thread, attr, + plugin_thread_start, plugin_thread); +} /* int plugin_thread_create */ + +/* vim: set sw=8 ts=8 noet fdm=marker : */ diff --combined src/write_graphite.c index d3d52027,504ea9d4..41451a8c --- a/src/write_graphite.c +++ b/src/write_graphite.c @@@ -49,6 -49,7 +49,6 @@@ #include "utils_cache.h" #include "utils_complain.h" -#include "utils_parse_option.h" #include "utils_format_graphite.h" /* Folks without pthread will need to disable this plugin. */ @@@ -82,10 -83,6 +82,10 @@@ # define WG_SEND_BUF_SIZE 1428 #endif +#ifndef WG_MIN_RECONNECT_INTERVAL +# define WG_MIN_RECONNECT_INTERVAL TIME_T_TO_CDTIME_T (1) +#endif + /* * Private variables */ @@@ -112,7 -109,6 +112,7 @@@ struct wg_callbac pthread_mutex_t send_lock; c_complain_t init_complaint; + cdtime_t last_connect_time; }; @@@ -190,23 -186,17 +190,25 @@@ static int wg_callback_init (struct wg_ struct addrinfo ai_hints; struct addrinfo *ai_list; struct addrinfo *ai_ptr; + cdtime_t now; int status; const char *node = cb->node ? cb->node : WG_DEFAULT_NODE; const char *service = cb->service ? cb->service : WG_DEFAULT_SERVICE; const char *protocol = cb->protocol ? cb->protocol : WG_DEFAULT_PROTOCOL; + char connerr[1024] = ""; + if (cb->sock_fd > 0) return (0); + /* Don't try to reconnect too often. By default, one reconnection attempt + * is made per second. */ + now = cdtime (); + if ((now - cb->last_connect_time) < WG_MIN_RECONNECT_INTERVAL) + return (EAGAIN); + cb->last_connect_time = now; + memset (&ai_hints, 0, sizeof (ai_hints)); #ifdef AI_ADDRCONFIG ai_hints.ai_flags |= AI_ADDRCONFIG; @@@ -233,12 -223,19 +235,19 @@@ { cb->sock_fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol); - if (cb->sock_fd < 0) + if (cb->sock_fd < 0) { + char errbuf[1024]; + snprintf (connerr, sizeof (connerr), "failed to open socket: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); continue; + } status = connect (cb->sock_fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen); if (status != 0) { + char errbuf[1024]; + snprintf (connerr, sizeof (connerr), "failed to connect to remote " + "host: %s", sstrerror (errno, errbuf, sizeof (errbuf))); close (cb->sock_fd); cb->sock_fd = -1; continue; @@@ -251,11 -248,12 +260,12 @@@ if (cb->sock_fd < 0) { - char errbuf[1024]; + if (connerr[0] == '\0') + /* this should not happen but try to get a message anyway */ + sstrerror (errno, connerr, sizeof (connerr)); c_complain (LOG_ERR, &cb->init_complaint, - "write_graphite plugin: Connecting to %s:%s via %s failed. " - "The last error was: %s", node, service, protocol, - sstrerror (errno, errbuf, sizeof (errbuf))); + "write_graphite plugin: Connecting to %s:%s via %s failed. " + "The last error was: %s", node, service, protocol, connerr); return (-1); } else