Skip to content

Data Sources

Jump to bottom
sat edited this page Jul 15, 2026 · 1 revision

Data Sources

logdag reads event data from one or two input sources and stores extracted time-series features in a feature database (evdb). The logdag.source package owns everything related to ingestion, storage, and pre-processing.

See also: Overview and Pipeline · CLI Commands

Architecture overview

Input sources                    Feature DB (evdb)          DAG generation
─────────────────                ────────────────           ──────────────
amulog log DB  ──┐               ┌─ sql (SQLite/MySQL)  ──┐
                 ├─ evgen_log ──►│  influx (InfluxDB v1) │►  makedag
SNMP / RRD     ──┘  evgen_snmp  └─ influx3 (InfluxDB v3) ┘

The evdb backend is selected by [general] evdb in the logdag config file. When use_evdb = false (direct mode) the evdb step is skipped and amulog is queried at DAG generation time instead.

Log source: amulog (src_amulog.py, evgen_log.py)

AmulogLoader

logdag.source.src_amulog.AmulogLoader wraps an amulog LogData object and exposes the event and timestamp iteration interface used by evgen_log.

Key constructor parameters (wired in init_amulogloader and LogEventLoaderBase.__init__):

Parameter Config key Purpose
conf amulog config object (path from [database_amulog] source_conf) amulog config
dt_range [general] evdb_whole_term overall time window
gid_name [database_amulog] event_gid (ltid or ltgid) template ID type
use_mapping [database_amulog] use_anonymize_mapping anonymization support
host_tier [database_amulog] host_tier host stratification (see below)

The loader exposes:

  • iter_event(dt_range) — yields (host, gid) pairs that have log entries in the window. In stratified mode, yields (hgid, gid) pairs instead.
  • load(ev, dt_range) — returns a sorted list of timestamps for event ev.
  • iter_dt(ev, dt_range) — generator of timezone-aware datetime objects.
  • gid_instruction(gid) — human-readable description of the template.
  • group(gid) — amulog tag string for the template (used as event group).

LogEventLoader (evgen_log.py)

LogEventLoader (inherits EventLoader + LogEventLoaderBase) drives the full evdb write pipeline:

  1. Iterates over all (host, gid) pairs in the time window.
  2. Loads raw timestamps from amulog.
  3. Applies filter_log filters (if configured).
  4. Writes the surviving timestamp counts to the evdb under measurement log_feature.

The subcommand that triggers this is:

logdag.source make-evdb-log-all -c <config>

Direct mode (LogEventLoaderDirect) skips the evdb: evgen_log reads amulog on every query. Enabled by setting [general] use_evdb = false.

SNMP/RRD source (evgen_snmp.py, src_rrd.py)

SNMP-derived numeric metrics are stored in RRD files and loaded via the rrdtool Python binding (src_rrd.RRDLoader).

evgen_snmp defines two event types:

  • SNMPEventDefinition — one measurement per (host, mod_cls, mod_id, measure, direction).
  • SNMPVirtualEventDefinition — an aggregated "all" view (key = "all").

The relevant subcommands (under logdag.source) are:

logdag.source make-evdb-snmp       -c <config>   # single SNMP source
logdag.source make-evdb-snmp-all   -c <config>   # all sources
logdag.source make-evdb-snmp-org   -c <config>
logdag.source make-evdb-snmp-tests -c <config>
logdag.source show-snmp-stats      -c <config>

RRD configuration lives in [database_rrd]:

Option Purpose
rows RRD consolidation rows
cf Consolidation function (MAX, AVERAGE, …)
binsize Fetch resolution
correct_roundup Shift timestamps to align with syslog rounding

Feature DB (evdb) backends

The evdb backend is chosen by [general] evdb. All backends implement the TimeSeriesDB abstract base class defined in source/sqlts.py.

TimeSeriesDB contract

Every backend must implement:

Method Description
add(measure, d_tags, d_input, columns) Write time-series data
commit() Flush pending writes
get_df(measure, d_tags, fields, dt_range, ...) Fetch as a binned DataFrame
get_items(measure, d_tags, fields, dt_range) Fetch as (timestamp, values) pairs
get_count(measure, d_tags, fields, dt_range) Count entries
drop_measurement(measure) Remove a measurement
list_measurements() / list_series(measure) Introspection

sql — SQLite or MySQL (sqlts.py)

Selected when [general] evdb = sql (also accepted: sqlite, mysql).

Uses amulog's db_common layer for SQL query generation. The backend is self-contained (no extra install), which makes it the easiest choice for development and testing.

Config section: [database_sql].

[general]
evdb = sql

[database_sql]
log_dbname = log.db
snmp_dbname = snmp.db

Drop all features with:

logdag.source drop-features -c <config>

influx — InfluxDB v1 (influx.py)

Selected when [general] evdb = influx.

Uses the influxdb Python client (installed via pip install -e .[influx]). Communicates over InfluxQL. Compatible with InfluxDB v1.x and the v1 compatibility API of InfluxDB v2.

Config section: [database_influx]:

[general]
evdb = influx

[database_influx]
host = localhost
port = 8086
log_dbname = logdag_log
snmp_dbname = logdag_snmp

The database must exist before logdag is run (logdag does not create it).

influx3 — InfluxDB v3 Core (influx3.py)

Selected when [general] evdb = influx3.

Uses the InfluxDB v3 HTTP API (POST /api/v3/write_lp, GET /api/v3/query_sql) via Python's standard library urllib — no additional package is required. Queries are written in SQL (DataFusion dialect) rather than InfluxQL.

Key design points:

  • Single-level database namespace (no retention policy concept).
  • Authentication is optional: set the environment variable INFLUXDB_V3_TOKEN or leave it empty when running without auth (--without-auth dev servers).
  • Auto-creates the database on first write when create_if_missing = True.

Config section: [database_influx3]:

[general]
evdb = influx3

[database_influx3]
host = http://localhost:8181
log_dbname = logdag_log
snmp_dbname = logdag_snmp

Note — maturity: The influx3 backend is wired in current code and its contract tests pass, but it is newer and less battle-tested than the sql and influx backends. Use with caution in production and confirm its behaviour against your InfluxDB v3 deployment.

Event filtering during ingestion (filter_log.py, period.py)

Before a log event series is written to the evdb, it is optionally passed through a chain of filter rules configured in [filter] rules.

Available rules (defined in filter_log.FUNCTIONS):

Rule Effect
sizetest Skip events with fewer than pre_count occurrences or a span shorter than pre_term. If the test fails the event is left unchanged rather than removed (it guards against applying Fourier analysis to trivially short series).
filter_periodic Detect a periodic component via FFT and subtract it, returning the aperiodic residual. The series is shrunk but not removed entirely.
remove_periodic Detect periodicity via FFT; if found, remove the entire series (return None).
remove_corr Detect periodicity via autocorrelation; if found, remove the series.
remove_linear Remove series whose cumulative count closely follows a uniform (linear) growth curve.

Rules are applied in order. If any rule (except sizetest) returns None or an empty list, the event is dropped from the evdb.

Fourier and autocorrelation parameters (thresholds, window sizes) are configured in the [filter] section. Refer to Configuration Options for the full table.

Post-processing numeric features (evpost.py)

evpost provides optional transformations applied to SNMP numeric features before they are fed into the causal algorithm. Available transforms:

Function Description
fillzero Replace NaN with 0
fillavg Replace NaN with the column mean
norm_fillavg Z-score normalise then fill NaN with 0
root_square_diff First difference scaled by current value
diff_abs Absolute first difference
getnan Binary indicator for missing values
convolve Moving average
outlier Binary indicator: 1 where value exceeds median + threshold
outlier_median_absdev Same using median absolute deviation
anomaly_lof Local Outlier Factor anomaly detection (requires scikit-learn)
anomaly_if Isolation Forest anomaly detection (requires scikit-learn)

Host stratification (host_tier)

When [database_amulog] host_tier is set to a non-empty string, AmulogLoader activates amulog's host_group mechanism.

Host stratification is the host axis of event-node aggregation; its log-template-axis counterpart is event_gid (ltid vs ltgid, in the AmulogLoader parameter table above). Both control how many — and how coarse — the event nodes fed to causal discovery are.

How it works:

  1. On first use, AmulogLoader._ensure_hg_map builds a bidirectional map between original hosts and their group IDs (hgid) at the chosen tier.
  2. iter_event replaces each (host, gid) pair with (hgid, gid) and deduplicates, so that hosts in the same tier group are treated as a single logical node.
  3. _iter_lines (used by load) fetches log lines from all original hosts that belong to the hgid, merging them transparently.

This means the evdb stores features under the hgid label rather than individual host names. No change to the amulog database is needed; the grouping is resolved at query time.

Requirements:

  • [manager] host_group_filename must be set in the amulog config to point to a valid host-group definition file.
  • The tier name given in host_tier must be a tier defined in that file. AmulogLoader.__init__ raises ValueError with a list of valid tiers if the name is not found.

Example config fragment:

[database_amulog]
source_conf   = /path/to/amulog.conf
event_gid     = ltgid
host_tier     = site

Dropping stored features

To clear the evdb and start fresh:

logdag.source drop-features -c <config>

This calls drop_measurement for each measurement (log_feature, SNMP measurements) on whichever backend is configured.

Clone this wiki locally