duct.sources

duct.sources.apache

class duct.sources.apache.Apache(config, queueBack, duct)

Bases: Source

Reads Apache mod_status output

Parameters:

url (str.) – URL to fetch status from

Metrics:

(service name).conns.active:

Active connections at this time

(service name).conns.writing:

Writing connections

(service name).conns.closing:

Closing connections

(service name).conns.keep_alive:

KeepAlive connections

(service name).accesses:

Accepted connections

(service name).uptime:

Server uptime

(service name).cpu_load:

CPU load

(service name).total_kbytes:

Total kbytes served

(service name).request_rate:

Requests per second

(service name).bytes_rate:

Bytes per second

(service name).bytes_req:

Bytes per request

(service name).workers.busy:

Busy workers

(service name).workers.idle:

Idle workers

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.database.elasticsearch

class duct.sources.database.elasticsearch.ElasticSearch(*a, **kw)

Bases: Source

Reads elasticsearch metrics

Parameters:

• url (str.) – Elasticsearch base URL (default: http://localhost:9200)

• user (str.) – Basic auth username

• password (str.) – Password

Metrics:

(service name).cluster.status:

Cluster status (Red=0, Yellow=1, Green=2)

(service name).cluster.nodes:

Cluster node count

(service name).indices:

Total indices in cluster

(service name).shards.total:

Total number of shards

(service name).shards.primary:

Number of primary shards

(service name).documents.total:

Total documents

(service name).documents.rate:

Documents per second

(service name).documents.size:

Size of document store in bytes

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.database.memcache

class duct.sources.database.memcache.Memcache(*a, **kw)

Bases: Source

Reads memcache metrics

Parameters:

• host (str.) – Database host (default localhost)

• port (int.) – Database port (default 11211)

Metrics:

(service name).(metrics):

Metrics from memcached

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.database.postgresql

class duct.sources.database.postgresql.PostgreSQL(*a, **kw)

Bases: Source

Reads PostgreSQL metrics

Parameters:

• host (str.) – Database host

• port (int.) – Database port

• user (str.) – Username

• password (str.) – Password

Metrics:

(service name).(database name).(metrics):

Metrics from pg_stat_database

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.docker

class duct.sources.docker.ContainerStats(*a, **kw)

Bases: Source

Returns stats for Docker containers on this host

Parameters:

url (str.) – Docker stats URL

Metrics:

(service name).(container name).mem_limit:

Maximum memory for container

(service name).(container name).mem_used:

Memory used by container

(service name).(container name).cpu:

Percentage of system CPU in use

(service name).(container name).io_read:

IO reads per second

(service name).(container name).io_write:

IO writes per second

(service name).(container name).io_sync:

IO synchronous op/s

(service name).(container name).io_async:

IO asynchronous op/s

(service name).(container name).io_total:

Total IOPS

Note. If a MARATHON_APP_ID environment variable exists on the container then container name will be used instead of that.

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.generator

class duct.sources.generator.Function(config, queueBack, duct)

Bases: Source

Produces an arbitrary function

Functions can contain the functions sin, cos, sinh, cosh, tan, tanh, asin, acos, atan, asinh, acosh, atanh, log(n, [base|e]), abs

Or the constants e, pi, and variable x

Parameters:

• dx (float.) – Resolution with time (steps of x)

• function (string.) – Function to produce

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.haproxy

class duct.sources.haproxy.HAProxy(*a, **kw)

Bases: Source

Reads Nginx stub_status

Parameters:

• url (str.) – URL to fetch stats from

• user (str.) – Username

• password (str.) – Password

Metrics:

(service name).(backend|frontend|nodes).(stats):

Various statistics

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.linux.basic

class duct.sources.linux.basic.CPU(*a)

Bases: Source

Reports system CPU utilisation as a percentage/100

Metrics:

(service name):

Percentage CPU utilisation

(service name).(type):

Percentage CPU utilisation by type

(service name).coreX:

Percentage CPU utilisation per core

(service name).coreX.(type):

Percentage CPU utilisation per core by type

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

async sshGet()

Get method used when use_ssh is enabled

class duct.sources.linux.basic.DiskFree(config, queueBack, duct)

Bases: Source

Returns the free space for all mounted filesystems

Parameters:

disks (list.) – List of devices to check (optional)

Metrics:

(service name).(device).used:

Used space (%)

(service name).(device).bytes:

Used space (kbytes)

(service name).(device).free:

Free space (kbytes)

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

class duct.sources.linux.basic.DiskIO(*a, **kw)

Bases: Source

Reports disk IO statistics per device

Parameters:

devices (list.) – List of devices to check (optional)

Metrics:

(service name).(device name).reads:

Number of completed reads

(service name).(device name).read_bytes:

Bytes read per second

(service name).(device name).read_latency:

Disk read latency

(service name).(device name).writes:

Number of completed writes

(service name).(device name).write_bytes:

Bytes written per second

(service name).(device name).write_latency:

Disk write latency

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

async sshGet()

Get method used when use_ssh is enabled

class duct.sources.linux.basic.LoadAverage(config, queueBack, duct)

Bases: Source

Reports system load average for the current host

Metrics:

(service name):

Load average

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

async sshGet()

Get method used when use_ssh is enabled

class duct.sources.linux.basic.Memory(config, queueBack, duct)

Bases: Source

Reports system memory utilisation as a percentage/100

Metrics:

(service name):

Percentage memory utilisation

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

async sshGet()

Get method used when use_ssh is enabled

class duct.sources.linux.basic.Network(config, queueBack, duct)

Bases: Source

Returns all network interface statistics

Parameters:

interfaces (list.) – List of interfaces to check (optional)

Metrics:

(service name).(device).tx_bytes:

Bytes transmitted

(service name).(device).tx_packets:

Packets transmitted

(service name).(device).tx_errors:

Errors

(service name).(device).rx_bytes:

Bytes received

(service name).(device).rx_packets:

Packets received

(service name).(device).rx_errors:

Errors

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

async sshGet()

Get method used when use_ssh is enabled

duct.sources.linux.ipsec

class duct.sources.linux.ipsec.StrongSwan(config, queueBack, duct)

Bases: Source

Returns the status of strongSwan IPSec tunnels

Metrics:

(service name).(peer name):

Tunnel status

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.linux.process

class duct.sources.linux.process.ProcessCount(config, queueBack, duct)

Bases: Source

Returns the ps count on the system

Metrics:

(service name):

Number of processes

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

class duct.sources.linux.process.ProcessStats(config, queueBack, duct)

Bases: Source

Returns memory used by each active parent process

Metrics:

(service name).proc.(process name).cpu:

Per process CPU usage

(service name).proc.(process name).memory:

Per process memory use

(service name).proc.(process name).age:

Per process age

(service name).user.(user name).cpu:

Per user CPU usage

(service name).user.(user name).memory:

Per user memory use

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.linux.sensors

class duct.sources.linux.sensors.LMSensors(config, queueBack, duct)

Bases: Source

Returns lm-sensors output

This does the exact same thing as the Sensors class but uses lm-sensors.

Metrics:

(service name).(adapter).(sensor):

Sensor value

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

class duct.sources.linux.sensors.SMART(*a, **kw)

Bases: Source

Returns SMART output for all disks

Metrics:

(service name).(disk).(sensor):

Sensor value

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

class duct.sources.linux.sensors.Sensors(config, queueBack, duct)

Bases: Source

Returns hwmon sensors info

Note: There is no transformation done on values, they may be in thousands

Metrics:

(service name).(adapter).(sensor):

Sensor value

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.mail

duct.sources.media.libav

class duct.sources.media.libav.DarwinRTSP(config, queueBack, duct)

Bases: Source

Makes avprobe requests of a Darwin RTSP sample stream (sample_100kbit.mp4)

Parameters:

destination – Host name or IP address to check

Metrics: :(service name): Time to complete request

You can also override the hostname argument to make it match metrics from that host.

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.munin

class duct.sources.munin.MuninClient(host, port)

Bases: object

asyncio-based munin-node client

async connect()

Open connection to munin-node.

async disconnect()

Close connection to munin-node.

async send_command(command, multiline=False)

Send a command to munin-node and return the response.

class duct.sources.munin.MuninNode(config, queueBack, duct)

Bases: Source

Connects to munin-node and retrieves all metrics

Parameters:

• host (str.) – munin-node hostname (probably localhost)

• port (int.) – munin-node port (probably 4949)

Metrics:

(service name).(plugin name).(keys…):

A dot separated tree of munin plugin keys

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.nats

class duct.sources.nats.Nats(*a, **kw)

Bases: Source

NATS source

Parameters:

• servers (list) – List of NATS URIs (default: [“nats://localhost:4222”])

• topics (list) – List of topics to subscribe to (default: [“>”])

• format (str) – Serialisation format - json, senml-json, senml-cbor (default: senml-json)

• jetstream (bool) – Subscribe via JetStream with a durable consumer (default: false)

• durable (str) – Durable consumer name for JetStream subscriptions (default: “ducted”)

• credentials_file (str) – Path to NATS credentials (.creds) file

• nkey_seed_file (str) – Path to NKey seed file

• tls_ca_file (str) – Path to CA certificate file for TLS

• tls_cert_file (str) – Path to client certificate file for mTLS

• tls_key_file (str) – Path to client private key file for mTLS

async get()

Uses async queue

async startTimer()

Starts the polling loop for this source

async stopTimer()

Stops the polling loop for this source

duct.sources.network

class duct.sources.network.HTTP(config, queueBack, duct)

Bases: Source

Performs an HTTP request

Parameters:

• url (str.) – HTTP URL

• method (str.) – HTTP request method to use (default GET)

• match (str.) – A text string to match in the document when it is correct

• useragent (str.) – User-Agent header to use

• timeout (int.) – Timeout for connection (default 60s)

Metrics:

(service name).latency:

Time to complete request

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

class duct.sources.network.Ping(config, queueBack, duct)

Bases: Source

Performs an Ping checks against a destination

Parameters:

destination (str.) – Host name or IP address to ping

Metrics:

(service name).latency:

Ping latency

(service name).loss:

Packet loss

You can also override the hostname argument to make it match metrics from that host.

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.nginx

class duct.sources.nginx.Nginx(config, queueBack, duct)

Bases: Source

Reads Nginx stub_status

Parameters:

stats_url (str.) – URL to fetch stub_status from

Metrics:

(service name).active:

Active connections at this time

(service name).accepts:

Accepted connections

(service name).handled:

Handled connections

(service name).requests:

Total client requests

(service name).reading:

Reading requests

(service name).writing:

Writing responses

(service name).waiting:

Waiting connections

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

class duct.sources.nginx.NginxLog(*a)

Bases: Source

Tails Nginx log files, parses them and returns log events for outputs which support them.

Parameters:

• log_format (str.) – Log format passed to parser, same as the config definition (default: combined)

• file (str.) – Log file

• max_lines (int.) – Maximum number of log lines to read per interval to prevent overwhelming Duct when reading large logs (default 2000)

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

got_eventlog(event)

Received event log

class duct.sources.nginx.NginxLogMetrics(*a)

Bases: Source

Tails Nginx log files, parses them and returns metrics for data usage and requests against other fields.

Parameters:

• log_format (str.) – Log format passed to parser, same as the config definition

• file (str.) – Log file

• max_lines (int.) – Maximum number of log lines to read per interval to prevent overwhelming Duct when reading large logs (default 2000)

• resolution (int.) – Aggregate bucket resolution in seconds (default 10)

• history (bool.) – Read the entire file from scratch if we’ve never seen it (default false)

Metrics:

(service name).total_rbytes:

Bytes total for all requests

(service name).total_requests:

Total request count

(service name).stats.(code).(requests|rbytes):

Metrics by status code

(service name).user-agent.(agent).(requests|rbytes):

Metrics by user agent

(service name).client.(ip).(requests|rbytes):

Metrics by client IP

(service name).request.(request path).(requests|rbytes):

Metrics by request path

dumpEvents(ts)

Dump collected events for the time bucket

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

got_line(line)

Line received

duct.sources.postfix

class duct.sources.postfix.Postfix(*a, **kw)

Bases: Source

Postfix checks

Parameters:

spool (str.) – Postfix spool directory (default: /var/spool/postfix)

Metrics:

(service_name).active:

(service_name).deferred:

(service_name).maildrop:

(service_name).incoming:

(service_name).corrupt:

(service_name).hold:

(service_name).bounce:

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.power.nut

class duct.sources.power.nut.UPS(*a, **kw)

Bases: Source

Query UPS statistics directly from a NUT server over TCP

Parameters:

• ups (str.) – UPS name as configured in NUT (default: ‘ups’)

• host (str.) – NUT server host (default: ‘localhost’)

• port (int.) – NUT server port (default: 3493)

• timeout (int.) – Connection timeout in seconds (default: 10)

Metrics:

(service_name).battery.charge:

Battery charge percentage

(service_name).battery.runtime:

Battery runtime remaining in seconds

(service_name).battery.voltage:

Battery voltage

(service_name).input.voltage:

Input mains voltage

(service_name).input.frequency:

Input mains frequency

(service_name).output.voltage:

Output voltage

(service_name).ups.load:

UPS load percentage

(service_name).ups.temperature:

UPS temperature (if supported)

(service_name).ups.realpower:

Real power in watts (if supported)

(service_name).ups.status:

UPS status string (OL, OB, LB, etc.)

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.power.raritan_px

class duct.sources.power.raritan_px.RaritanPX(*a, **kw)

Bases: SNMP

Monitors a Raritan PX PDU via SNMP using the PDU2-MIB.

Requires compiled PDU2-MIB Python modules in mib_path.

Parameters:

• ip (str.) – PDU hostname or IP (default: 127.0.0.1)

• port (int.) – SNMP port (default: 161)

• community (str.) – SNMP read community (default: public)

• mib_path (str.) – Directory containing compiled MIB modules (default: /var/lib/duct/mibs)

Metrics emitted (prefixed by service name):

Outlet.N.voltage:

Outlet N RMS voltage (V)

Outlet.N.current:

Outlet N RMS current (A)

Outlet.N.power:

Outlet N active power (W)

Outlet.N.apparent_power:

Outlet N apparent power (VA)

Outlet.N.pf:

Outlet N power factor

Outlet.N.energy:

Outlet N active energy (Wh)

Inlet.N.*:

Same metrics for inlet N

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.python.uwsgi

class duct.sources.python.uwsgi.Emperor(config, queueBack, duct)

Bases: Source

Connects to UWSGI Emperor stats and creates useful metrics

Parameters:

• host (str.) – Hostname (default localhost)

• port (int.) – Port

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.rabbitmq

class duct.sources.rabbitmq.Queues(*a, **kw)

Bases: Source

Returns Queue information for a particular vhost

Parameters:

vhost (str.) – Vhost name

Metrics:

(service_name).(queue).ready:

Ready messages for queue

(service_name).(queue).unack:

Unacknowledged messages for queue

(service_name).(queue).ready_rate:

Ready rate of change per second

(service_name).(queue).unack_rate:

Unacknowledge rate of change per second

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.redis

class duct.sources.redis.Queues(*a, **kw)

Bases: Source

Query llen from redis-cli

Parameters:

• queue (str.) – Queue name (defaults to ‘celery’, just because)

• db (int.) – DB number

• clipath (str.) – Path to redis-cli (default: /usr/bin/redis-cli)

Metrics:

(service_name):

Queue length

(service_name):

Queue rate

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.riak

class duct.sources.riak.RiakStats(config, queueBack, duct)

Bases: Source

Returns GET/PUT rates for a Riak node

Parameters:

• url (str.) – Riak stats URL

• useragent (str.) – User-Agent header to use

Metrics:

(service name).latency:

Time to complete request

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

duct.sources.riemann

class duct.sources.riemann.RiemannTCP(*a, **kw)

Bases: Source

Provides a listening Riemann TCP server that accepts metrics and proxies them to the queue.

Parameters:

port (int.) – Port to listen on (default: 5555)

get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

async startTimer()

Creates a Riemann TCP server instead of a polling timer

async stopTimer()

Stops the polling loop for this source

duct.sources.sensors.environment.ds18b20

class duct.sources.sensors.environment.ds18b20.DS18B20(*a, **kw)

Bases: Source

DS18B20 1-wire temperature sensor. This module reads sensors connected with the wire kernel module typically used on a Raspberry Pi with the w1-gpio device tree overlay. These can be slow to read if you have a lot of them, so make sure you have a sufficient interval set.

Parameters:

• device_path (str) – Path for w1_therm and w1_gpio devices

• device_map (dict) – A dictionary structure re-mapping sensor serials to custom names

• ignore_unmapped (bool) – If True any sensor not listed in the device_map is ignored (default: False)

async get()

Polls all sensors and sends metrics in a batch every POLL_INTERVAL seconds.

read_sensors(bus)

Read all sensors on the device path

read_temp(bus, sensor_id)

Reads the temperature from a 1-Wire sensor device.

duct.sources.sensors.environment.mpl115

class duct.sources.sensors.environment.mpl115.MPL115(*a, **kw)

Bases: Source

MPL115 temperature and pressure sensor

This source was based on: https://github.com/FaBoPlatform/FaBoBarometer-MPL115-Python

Parameters:

• address (hex.) – I2C address of the sensor (default 0x60)

• smbus (int.) – Bus number (default 1)

• altitude (float.) – Altitude for correction (default 0)

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

async readCoefficients()

Read coefficients from sensor

async readData()

Read data from the sensor returns dict containing ‘hpa’ and ‘temp’ values

duct.sources.sflow

class duct.sources.sflow.sFlow(*a, **kw)

Bases: Source

Provides an sFlow UDP server Source.

Parameters:

• port (int.) – UDP port to listen on (default: 6343)

• dnslookup (bool.) – Enable reverse DNS lookup for device IPs (default: True)

async get()

sFlow does not poll; data arrives via UDP.

async startTimer()

Starts the polling loop for this source

async stopTimer()

Stops the polling loop for this source

class duct.sources.sflow.sFlowReceiver(source)

Bases: DatagramReceiver

sFlow datagram protocol

process_convo_queue(queue, host, idx, deltaIn, tDelta)

Process the conversation queue

receive_counter(counter, host)

Called when a counter is received

receive_flow(flow, sample, host)

Called when a flow is received

duct.sources.snmp

class duct.sources.snmp.SNMP(*a, **kw)

Bases: Source

Connects to an SNMP agent and retrieves OIDs.

Parameters:

• ip (str.) – SNMP agent host (default: 127.0.0.1)

• port (int.) – SNMP port (default: 161)

• community (str.) – SNMP read community

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

async getCounter(soid)

Walk SNMP OID and return list of (oid, value) tuples.

async getIfMetrics()

Retrieve interface metrics via SNMP and return Event list.

class duct.sources.snmp.SNMPCisco837(*a, **kw)

Bases: SNMP

Connects to a Cisco 837 and makes metrics.

Parameters:

• ip – SNMP agent host (default: 127.0.0.1)

• port – SNMP port (default: 161)

• community – SNMP read community

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.

class duct.sources.snmp.SNMPConnection(host, port, community)

Bases: object

A wrapper class for PySNMP asyncio functions.

Parameters:

• host – SNMP agent host

• port – SNMP port

• community – SNMP read community

async walk(oid)

Walk OIDs under oid. Returns a list of (oid, value) tuples.

async walkMib(mib_name, obj_name, mib_dirs=None)

Walk a MIB table by symbolic name. Returns a list of (oid, value) tuples.

Parameters:

• mib_name – MIB module name (e.g. ‘PDU2-MIB’)

• obj_name – Object name within the MIB (e.g. ‘measurementsOutletSensorValue’)

• mib_dirs – List of directories containing compiled MIB modules

duct.sources.unbound

class duct.sources.unbound.Stats(*a, **kw)

Bases: Source

Returns stats from unbound-control

Parameters:

executable (str.) – Path to unbound-control executable (default: /usr/sbin/unbound-control)

async get()

Get method called every self.inter seconds. Should return a list of Event objects or None.