duct.outputs

duct.outputs.bosun

class duct.outputs.bosun.Bosun(*a)[source]

Bases: OpenTSDB

Bosun HTTP API output

Parameters:

  • url (str.) – URL (default: http://localhost:4242)

  • maxsize (int.) – Maximum queue backlog size (default: 250000, 0 disables)

  • maxrate (int.) – Maximum rate of documents added to index (default: 100)

  • interval (int.) – Queue check interval in seconds (default: 1.0)

  • user (str.) – Optional basic auth username

  • password (str.) – Optional basic auth password

async createMetadata(metas)[source]

Create metadata objects for new service keys

async sendEvents(events)[source]

Send a list of events to OpenTSDB

duct.outputs.elasticsearch

class duct.outputs.elasticsearch.ElasticSearch(*a)[source]

Bases: Output

ElasticSearch HTTP API output

Parameters:

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

  • maxsize (int) – Maximum queue backlog size (default: 250000, 0 disables)

  • maxrate (int) – Maximum rate of documents added to index (default: 100)

  • interval (int) – Queue check interval in seconds (default: 1.0)

  • user (str) – Optional basic auth username

  • password (str) – Optional basic auth password

  • index (str) – Index name format (default: duct-%Y.%m.%d)

async createClient()[source]

Sets up HTTP connector and starts queue drain task

async sendEvents(events)[source]

Send events to Elasticsearch in bulk

async stop()[source]

Called when the service shuts down

transformEvent(event)[source]

Transform an event into a format suitable for Elasticsearch

duct.outputs.elasticsearch.ElasticSearchLog

alias of ElasticSearch

duct.outputs.graphite

class duct.outputs.graphite.Graphite(*a, **kw)[source]

Bases: Output

Graphite output. By default events are named {prefix}.{hostname}.{source service} unless prefix_hostname is set to False

Parameters:

  • server (str.) – Graphite server hostname (default: localhost)

  • port (int.) – Graphite server port (default: 2003)

  • prefix (str.) – Metric path prefix (default: None)

  • prefix_hostname (bool.) – Whether to append the event source hostname field to the path (default: True)

async eventsReceived(events: list[Event])[source]

Called when events are routed to this output

async push_batch_to_graphite(metrics)[source]

Sends a batch of metric lines to Graphite via a single TCP connection.

duct.outputs.influxdb3

class duct.outputs.influxdb3.InfluxDB3(*a)[source]

Bases: Output

InfluxDB 3 HTTP API output using line protocol.

Events are written as measurements named after the event service. Tags include the source hostname, state, and any event attributes. The metric value is stored in the value field.

Configuration arguments:

Parameters:

  • url (str) – InfluxDB 3 server URL (default: http://localhost:8086)

  • database (str) – Target database / bucket name (default: duct)

  • token (str) – Authentication token (default: None)

  • precision (str) – Timestamp precision - nanosecond, microsecond, millisecond, or second (default: nanosecond)

  • maxsize (int) – Maximum queue backlog size (default: 250000, 0 disables)

  • maxrate (int) – Maximum events flushed per interval (default: 1000, 0 disables)

  • interval (float) – Queue drain interval in seconds (default: 1.0)

async createClient()[source]

Set up the InfluxDB 3 client and start the drain loop.

async sendEvents(events)[source]

Write a batch of events to InfluxDB 3.

async stop()[source]

Called when the service shuts down

transformEvent(ev)[source]

Convert an event to an InfluxDB line protocol string.

duct.outputs.logger

class duct.outputs.logger.Logger(*a, **kw)[source]

Bases: Output

Logger output

Parameters:

logfile (str) – Logfile (default: write to standard log)

async eventsReceived(events)[source]

Receives a list of events and queues them

async stop()[source]

Called when the service shuts down

duct.outputs.nats

class duct.outputs.nats.Nats(*a, **kw)[source]

Bases: Output

NATS output

Parameters:

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

  • prefix (str) – Prefix added to topics (default: “”)

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

  • interval (float) – Queue drain interval in seconds (default: 1.0)

  • jetstream (bool) – Publish via JetStream instead of core NATS (default: false)

  • 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 createClient() → Client[source]

Coroutine that sets up the output connection

async sendEvents(events: list[Event])[source]

Send batches of events to NATS or JetStream

async stop()[source]

Called when the service shuts down

duct.outputs.null

class duct.outputs.null.Null(config, duct)[source]

Bases: Output

Null output throws your events away

async eventsReceived(events)[source]

Receives a list of events and queues them

duct.outputs.opentsdb

class duct.outputs.opentsdb.OpenTSDB(*a)[source]

Bases: Output

OpenTSDB HTTP API output

Parameters:

  • url (str) – URL (default: http://localhost:4242)

  • maxsize (int) – Maximum queue backlog size (default: 250000, 0 disables)

  • maxrate (int) – Maximum rate of documents added to index (default: 100)

  • interval (int) – Queue check interval in seconds (default: 1.0)

  • user (str) – Optional basic auth username

  • password (str) – Optional basic auth password

async createClient()[source]

Coroutine that sets up the output connection

async sendEvents(events)[source]

Send a list of events to OpenTSDB

async stop()[source]

Called when the service shuts down

transformEvent(ev)[source]

Convert an event object into OpenTSDB format

duct.outputs.prometheus

class duct.outputs.prometheus.Prometheus(*a)[source]

Bases: Output

Prometheus scrape-endpoint output

Parameters:

  • port (int.) – Listening port (default: 9100)

  • metric_path (str.) – Metrics path (default: metrics)

  • prefix (str.) – Prometheus metric prefix (default: duct_)

async createClient()[source]

Coroutine that sets up the output connection

async eventsReceived(events)[source]

Receives a list of events and queues them

async stop()[source]

Called when the service shuts down

duct.outputs.riemann

class duct.outputs.riemann.RiemannTCP(*a)[source]

Bases: Output

Riemann TCP output

Parameters:

  • server (str.) – Riemann server hostname (default: localhost)

  • port (int.) – Riemann server port (default: 5555)

  • failover (bool.) – Enable server failover; server may be a list

  • maxrate (int.) – Maximum de-queue rate (0 is no limit)

  • maxsize (int.) – Maximum queue size (default: 250000)

  • interval (float.) – De-queue interval in seconds (default: 1.0)

  • pressure (int.) – Maximum backpressure (-1 is no limit)

  • tls (bool.) – Use TLS (default: false)

  • cert (str.) – Client certificate path

  • key (str.) – Client private key path

  • allow_nan (bool.) – Send events with None metric value (default: true)

async createClient()[source]

Create a TCP connection to Riemann and start the drain loop

async emptyQueue()[source]

Drain the event queue to Riemann

async stop()[source]

Called when the service shuts down

class duct.outputs.riemann.RiemannUDP(*a)[source]

Bases: Output

Riemann UDP output (spray-and-pray mode)

Parameters:

  • server (str.) – Riemann server IP address (default: 127.0.0.1)

  • port (int.) – Riemann server port (default: 5555)

async createClient()[source]

Create a UDP connection to Riemann

async eventsReceived(events)[source]

Receives a list of events and transmits them to Riemann