PostgreSQL

Pull

Synopsis

Director connects outbound to a PostgreSQL server on a schedule and runs operator-defined read-only queries, emitting one record per returned row. Each query runs on its own cron schedule with its own state, and supports incremental (high-watermark) collection as well as contiguous time-window collection. Cursor and window positions are checkpointed persistently, so collection resumes where it left off after a restart.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: postgres
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    host: <string>
    port: <numeric>
    username: <string>
    password: <string>
    database: <string>
    connect_timeout: <numeric>
    encrypt: <boolean>
    insecure_skip_verify: <boolean>
    ca_name: <string>
    connection_string: <string>
  definitions:
    - name: postgres_custom_query_collector
      status: <boolean>
      inputs: <input[]>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
id	Y	-	Unique numeric identifier
name	Y	-	Device name
description	N	-	Optional description
type	Y	-	Must be postgres
tags	N	-	Optional tags
pipelines	N	-	Optional preprocessing pipelines
status	N	true	Enable/disable the device

Connection

Field	Required	Default	Description
host	Y	-	Hostname or IP address of the PostgreSQL server.
port	N	5432	Server port.
username	Y	-	Login used to authenticate.
password	N	-	Password for username.
database	N	-	Database name to connect to.
connect_timeout	N	30	Connection dial and ping timeout, in seconds.
connection_string	N	-	Raw DSN passthrough. When set it is authoritative and the structured connection properties are ignored. See Connection string.

TLS

Field	Required	Default	Description
encrypt	N	true	Encrypt the connection.
insecure_skip_verify	N	false	Accept the server certificate without verifying it.
ca_name	N	-	Custom CA used to verify the server certificate. Ignored when insecure_skip_verify is true.

note

TLS material fields (cert_name, key_name, ca_name, client_ca_name) accept any of the following:

• File name — resolved relative to the service root directory. Nested paths such as certs/prod/server.pem are supported.
• Absolute path — honored only if it resolves inside the service root. Any path that escapes the root is refused.
• Inline PEM content — used verbatim when the value contains -----BEGIN.
• Environment variable — ${ENV_VAR}.
• Vault reference — $secret{id=...} or $secret{store=...,ref=...}.

Custom Query Collector

Queries for this device are declared under the collector definition named postgres_custom_query_collector.

Queries are defined under the device's definitions: block. The definition name must match the device's collector constant, and each entry under inputs: is one query running on its own schedule with its own state.

Input properties

Inputs share the standard dataset input frame (id, name, status) plus the following properties:

Field	Required	Type	Default	Description
query	Y	string	—	The SQL statement to run. May contain the binding tokens described below.
cron	N	string	—	5-field cron expression. Empty or "0" runs the query on every stats interval tick.
timeout	N	int	120	Query timeout in seconds. A zero or negative value falls back to the default.
validate_query	N	boolean	true	Enforce that the query is a single read-only statement. See Query validation.
max_rows	N	int	0	Per-run row cap for bounded, chunked backfills; the remainder resumes on the next run. 0 means no cap. Ignored when the query uses window tokens.
max_retries	N	int	1	How many times a failed run retries on the next collection cycle before waiting for its next cron slot.
skippable	N	boolean	true	When true, a backlog of missed cron slots collapses into a single catch-up run. false replays every missed slot, one per cycle, until the schedule catches up.
pipeline_name	N	string	—	Route the collected records to a specific preprocessing pipeline by name.
tracking_column	N	string	—	Column whose maximum value becomes the next cursor. See Incremental collection.
tracking_column_type	N	string	numeric	numeric, timestamp, or string.
tracking_initial_value	N	string	—	First-run starting point. Defaults to 0, the Unix epoch, or the empty string per type.
tracking_rescan_margin	N	int	0	Re-scan this far behind the watermark on each run to catch late-committing rows. Numeric units, or seconds for timestamps.
window_format	N	string	datetime	How window bounds are bound: datetime, unix, unix_ms, or rfc3339. See Time-window collection.
initial_lookback	N	int	0	First-run backfill in seconds for window queries. 0 collects forward only.

Binding tokens

Three tokens may appear in a query. Each is rewritten to the database driver's own placeholder syntax and its value is passed as a bound parameter, so a token value can never alter the shape of the statement:

Token	Bound value
{{cursor}}	The last-seen value of the tracking column.
{{earliest}}	Lower bound of this run's time window.
{{latest}}	Upper bound of this run's time window.

The {{...}} spelling is the same on every database. Do not substitute a driver-native placeholder such as :cursor or ? — those are not recognized as tokens and reach the driver unbound.

Incremental collection

Setting tracking_column turns an input into a high-watermark collector: each run collects only rows past the largest value seen so far. The watermark is persisted after every successful run and survives restarts and device-ownership moves within a cluster.

query: "SELECT id, event_time, message FROM audit_log WHERE id > {{cursor}} ORDER BY id ASC"
tracking_column: id
tracking_column_type: numeric

Always ORDER BY the tracking column so that runs capped by max_rows, or interrupted partway, stay contiguous.

caution

Configuring tracking_column without placing {{cursor}} in the query is accepted but collects nothing incrementally: the watermark advances while the full result set is re-read on every run. A warning is logged when this combination is detected.

High-watermark tracking assumes the column is monotonic in commit order. A row that commits late with an already-passed value falls behind the watermark and is skipped. tracking_rescan_margin re-collects a safety margin behind the watermark on each run to catch those rows, at the cost of re-emitting the margin — deduplicate downstream when enabling it.

Time-window collection

A query referencing {{earliest}} or {{latest}} collects a contiguous time slice per run. Consecutive windows tile exactly, with no gap and no overlap: this run's upper bound becomes the next run's lower bound, persisted the same way as a cursor.

query: "SELECT * FROM login_events WHERE event_time >= {{earliest}} AND event_time < {{latest}}"
window_format: datetime
initial_lookback: 3600

Use >= on the lower bound and < on the upper bound so that a row landing exactly on a boundary is collected by exactly one window.

window_format must match how the column stores time. The default datetime binds a native timestamp; unix and unix_ms bind integer epoch values; rfc3339 binds a string.

max_rows is ignored for window queries, because a capped window would silently drop its remainder.

Query validation

With validate_query left at its default, a query must be a single statement beginning with SELECT, WITH, SHOW, EXPLAIN, DESCRIBE, or DESC. Multiple statements are rejected, as is a WITH or EXPLAIN statement containing INSERT, UPDATE, DELETE, or MERGE — and EXPLAIN ANALYZE, which executes its target rather than describing it. String literals, quoted identifiers, and comments are excluded from the check, so a semicolon or keyword inside them does not trigger a rejection. One trailing semicolon is stripped.

Set validate_query: false only for stored procedures or vendor read syntax the validator cannot prove read-only.

warning

Query validation guards against configuration accidents; it is not a security boundary. Pair it with a database login that has read-only permissions on the objects being queried.

Collected records

Each row becomes one JSON record whose fields are the query's column names, carrying native JSON types rather than stringified values:

Column type	Rendered as
Numeric, boolean	JSON number or boolean
Exact numeric (DECIMAL, NUMERIC, MONEY)	Unquoted JSON number with full precision preserved
Date and time	ISO-8601 string in UTC, millisecond precision
JSON, JSONB	Nested JSON object
Binary	Base64 string
NULL	null

A top-level @timestamp field is added to every record, which is what lets it flow through pipelines and into targets as a first-class document. If the query itself returns a column named @timestamp, that column is kept as-is and nothing is added.

Details

Connection string

Setting connection_string switches the device to raw DSN passthrough. The string is handed to the driver's own parser, and every structured property — host, port, username, password, database, and the TLS properties — is ignored. Transport security is then expressed entirely inside the DSN, for example through sslmode:

connection_string: "postgresql://user:pass@pg01:5432/mydb?sslmode=verify-full"

Checkpoints

Each input's cursor and window position is stored in Director's persistent state, keyed per input. After a restart the input resumes from its stored position rather than re-reading from the beginning, and the same state follows the device when ownership moves to another node in a cluster.

Examples

Basic Collection

Connecting to a PostgreSQL server and collecting active sessions every five minutes...

- id: 930000004 name: postgres-prod type: postgres status: true properties: host: pg01.example.com username: vmetric_reader password: "<password>" database: orders definitions: - name: postgres_custom_query_collector status: true inputs: - id: 4001 name: Active Sessions status: true properties: query: "SELECT pid, usename, datname, state FROM pg_stat_activity WHERE state <> 'idle'" cron: "*/5 * * * *"

Each returned row becomes one record, with the query's column names as fields and an added @timestamp...

{ "@timestamp": "2026-07-21T10:35:00.000Z", "pid": 20481, "usename": "app_rw", "datname": "orders", "state": "active" }

Incremental Collection

Collecting only audit rows committed since the last run, using a timestamp column as the high-watermark...

- id: 930000004 name: postgres-audit type: postgres status: true properties: host: pg01.example.com username: vmetric_reader password: "<password>" database: orders definitions: - name: postgres_custom_query_collector status: true inputs: - id: 4002 name: New Audit Rows status: true properties: query: "SELECT id, event_time, actor, action FROM audit_log WHERE event_time > {{cursor}} ORDER BY event_time ASC" tracking_column: event_time tracking_column_type: timestamp tracking_rescan_margin: 5 cron: "*/1 * * * *"

Time-Window Collection

Collecting login events in contiguous time slices, backfilling the first hour on the initial run...

- id: 930000004 name: postgres-logins type: postgres status: true properties: host: pg01.example.com username: vmetric_reader password: "<password>" database: orders encrypt: true ca_name: certs/pg-ca.pem definitions: - name: postgres_custom_query_collector status: true inputs: - id: 4003 name: Login Events Window status: true properties: query: "SELECT * FROM login_events WHERE event_time >= {{earliest}} AND event_time < {{latest}}" window_format: datetime initial_lookback: 3600 cron: "*/5 * * * *"