Skip to content

Extending logdag

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

Extending logdag

This page is a developer how-to for the two most common extensions:

  1. Adding a causal-discovery method — a new algorithm adapter selectable via [dag] cause_algorithm.
  2. Adding a data source / evdb backend — a new feature-database backend selectable via [general] evdb.

Both extensions follow the same shape: logdag selects the implementation with an if/elif dispatcher keyed on a config option, so adding one means (a) writing a new module that satisfies a known interface and (b) wiring one new branch into the dispatcher. Read Architecture Overview first for the data model and the dispatch points; this page makes those concrete.

Grounding: the interfaces and steps below are taken from the existing adapters (pc_input.py, lingam_input.py), the dispatcher makedag.estimate_dag, the prior-knowledge feed pknowledge.py, the evdb factory source/evgen_common.py, the TimeSeriesDB ABC in source/sqlts.py, the existing backends (influx.py, influx3.py), and the contract suite tests/contract/test_tsdb_contract.py. Nothing here is speculative; signatures are copied from the code.

1. Adding a causal-discovery method

1.1 Where the dispatch happens

For each job, makedag.makedag_main builds the input and then calls:

graph = estimate_dag(conf, input_df, prior_knowledge)

makedag.estimate_dag is the single dispatch point. It:

  1. Returns an empty DAG immediately if input_df.shape[1] < 2 (fewer than two event variables) — your adapter is never called for trivial inputs.
  2. Reads cause_algorithm = conf.get("dag", "cause_algorithm") and a shared ci_func = conf.get("dag", "ci_func").
  3. Branches on cause_algorithm, pulls the options each method needs from the config, calls the adapter, and returns its graph. A value that is not a built-in is resolved as an out-of-tree plugin (§1.6); only if no plugin is registered does it raise ValueError("invalid dag.cause_algorithm").

The currently reachable values are pc, lingam, pc-corr, and lingam-corr.

1.2 The interface an adapter must satisfy

There is no shared abstract base class for method adapters. The contract is informal but firm, derived from what estimate_dag does with the return value and what the rest of the pipeline expects:

Adapter contract. A method adapter is a function that takes the input DataFrame (rows = time bins, columns = integer event ids / eid), plus its own algorithm parameters and an optional prior_knowledge, and returns a networkx.DiGraph whose nodes are the input column labels (the eids). Directed edges are single arcs; an undirected edge is encoded as a pair of opposite arcs between the same two nodes. Optionally attach a numeric weight edge attribute (LiNGAM does — it enables the ate_prune filter). Return None to signal estimation failure (makedag_main treats None as a failed job); return an empty DiGraph for "no edges".

Why these specifics matter:

  • Node labels = eids. showdag.LogDAG resolves every node back to an event via the evmap (node_evdef(eid)). If your graph relabels nodes to anything other than the input columns, node→event resolution breaks. Build the graph on data.columns (as pc_input and lingam_input do).
  • Undirected = two arcs. LogDAG.edge_isdirected, edges_by_direction, and the directed/undirected filters all infer directedness from the presence of the reverse arc. A PC CPDAG already uses this convention via pcalg.estimate_cpdag; if your method emits truly undirected edges, add both arcs.
  • weight is optional but useful. The ate_prune filter (Viewing and Filtering DAGs) prunes by abs(weight); methods without weights simply cannot be ATE-pruned.

1.3 Two existing adapters as templates

PC (pc_input.pc) — selects a CI test, optionally binarizes, builds an initial skeleton, and returns a CPDAG:

def pc(data, threshold, mode="gsq", skel_method="stable",
       pc_depth=None, verbose=False, prior_knowledge=None):
    if prior_knowledge:
        init_graph = prior_knowledge.pruned_initial_skeleton()
    else:
        init_graph = nx.complete_graph(data.columns)
    if mode == "gsq":
        from gsq.ci_tests import ci_test_bin
        func = ci_test_bin
        data = binarize_input(data)
    elif mode in ("fisherz", "fisherz_bin"):
        from citestfz.ci_tests import ci_test_gauss
        func = ci_test_gauss
    ...
    return estimate_dag(data, threshold, func, skel_method,
                        pc_depth, verbose, init_graph)   # -> CPDAG DiGraph

LiNGAM (lingam_input.estimate) — fits a LiNGAM model and emits a weighted DiGraph on data.columns:

def estimate(data, algorithm="ica", lower_limit=0.01,
             ica_max_iter=1000, prior_knowledge=None):
    import lingam
    ...                                   # fit ICALiNGAM / DirectLiNGAM
    g = nx.DiGraph()
    for i in range(adj.shape[0]):
        g.add_node(i)
    ...                                   # for each significant coefficient:
    g.add_edge(from_, to, weight=coef, label=str(round(coef, 2)))
    return g

Note the lazy import lingam / import pcalg inside the function: this keeps heavy / optional dependencies out of the import path for users who do not select that method. Follow the same pattern.

1.4 Prior knowledge: the two feeds

prior_knowledge is a pknowledge.PriorKnowledge instance (or None). There is no single uniform way methods consume it — pick the feed that fits your algorithm:

  • Skeleton-based methods (PC) call prior_knowledge.pruned_initial_skeleton(), which returns an undirected networkx.Graph over all node ids with no-edge pairs removed. Use it as the init_graph for skeleton estimation.
  • LiNGAM calls prior_knowledge.lingam_prior_knowledge() (optionally with node_ids=[i, j] for the pairwise estimate_corr variant), which returns a constraint matrix built by lingam.utils.make_prior_knowledge from the recorded paths / no-paths / exogenous / sink variables.

Your adapter should accept prior_knowledge=None and degrade gracefully (PC falls back to a complete graph; ICA-LiNGAM ignores prior knowledge with a warning). You do not need to modify pknowledge.py to add a method — it already exposes both feeds.

1.5 Step-by-step

  1. Create logdag/<name>_input.py with an estimate(...)-style function that satisfies the adapter contract (§1.2). Do heavy imports lazily inside the function. Build the graph on data.columns.

  2. Add a branch to makedag.estimate_dag for your cause_algorithm value. Read whatever options you need from config there and pass them positionally / by keyword to your adapter (mirror an existing branch):

    elif cause_algorithm == "<name>":
    from . import <name>_input
    my_opt = conf.getfloat("dag", "<your_option>")   # or a new [section]
    return <name>_input.estimate(input_df, my_opt,
                                 prior_knowledge=prior_knowledge)

  3. Declare any new config options in logdag/data/config.conf.default — add them to [dag] or to a dedicated section (LiNGAM uses [lingam]). Documenting defaults there makes them visible in logdag show-default-config and on Configuration Options.

  4. Choose a prior-knowledge feed (skeleton vs. matrix) and consume it, or accept None and ignore it.

  5. Verify with a small run (see Generating DAGs) using cause_algorithm = <name>, and confirm the result loads and resolves: logdag show-edge-list -c <conf> should print edges with event labels — that exercises node→event resolution and so validates that you built the graph on eid node labels.

1.6 Shipping a method out-of-tree (plugins)

You do not have to edit estimate_dag at all. If a cause_algorithm value is not one of the built-ins, estimate_dag resolves it through the logdag.cause_algorithm entry-point group: an installed package that registers an entry point in that group — whose value points to an estimate(conf, input_df, prior_knowledge=None) -> networkx.DiGraph callable — is invoked for that name. Only if no such plugin is registered does the value fall through to ValueError("invalid dag.cause_algorithm").

This lets a method that cannot live in the public tree (for example, a wrapper around a private library) be installed and selected as cause_algorithm = <name> without its code or name appearing in logdag. Unlike a built-in adapter — whose parameters estimate_dag extracts from config and passes in — a plugin receives the merged conf and reads its own [section] options itself; it must return a DiGraph satisfying the same return contract as a built-in (§1.2: nodes are the input eid columns, an undirected edge is a pair of opposite arcs, and a numeric weight is optional). Declare the entry point in your plugin package, e.g. in pyproject.toml:

[project.entry-points."logdag.cause_algorithm"]
<name> = "your_package.module:estimate"

2. Adding a data source / evdb backend

"Data source" can mean two different extension points; be clear which you need:

  • A new evdb backend (a new storage for extracted features) — the common case, covered here in detail. You implement the TimeSeriesDB contract and wire one branch into the evdb factory.
  • A new input source class (a new origin of raw events, beyond log and snmp) — a larger change touching the event-definition model and several dispatchers; outlined in §2.5.

2.1 The TimeSeriesDB contract

Every evdb backend subclasses the TimeSeriesDB ABC in source/sqlts.py and implements its abstract methods:

class TimeSeriesDB(ABC):
    @abstractmethod
    def list_measurements(self): ...
    @abstractmethod
    def list_series(self, measure): ...
    @abstractmethod
    def list_fields(self, measure): ...
    @abstractmethod
    def add(self, measure, d_tags, d_input, columns): ...
    @abstractmethod
    def commit(self): ...
    @abstractmethod
    def get_items(self, measure, d_tags, fields, dt_range): ...
    @abstractmethod
    def get_df(self, measure, d_tags, fields, dt_range,
               str_bin=None, func=None, fill=None, limit=None): ...
    @abstractmethod
    def get_count(self, measure, d_tags, fields, dt_range): ...
    @abstractmethod
    def drop_measurement(self, measure): ...

The ABC also provides static timestamp helpers (pdtimestamp, pdtimestamp_naive, pdtimestamps) delegating to source.convert, so all backends localize/convert time the same way — inherit them, do not re-implement.

Behavioural contract (asserted by tests/contract/test_tsdb_contract.py; these are the rules new backends must reproduce):

  • add(measure, d_tags, d_input, columns) writes rows. d_tags maps tag→value (e.g. {"host": ..., "key": ...}); d_input maps timestamp→row-of-values; columns are the field names. commit() flushes.
  • get_df(..., func=None) over an empty range returns None, not an empty DataFrame.
  • get_df(..., func="sum", str_bin=<dur>) returns a per-bin aggregated frame (the read path used by feature loading; see EventLoader.load).
  • get_df returns a wide frame: index is a tz-aware local DatetimeIndex, columns are the field names. (Backends that store naive UTC, like the Influx ones, must localize-UTC then convert to local.)
  • get_count(...) over an empty range returns 0 (an int, never None).
  • get_items(...) yields (datetime, numpy.array) pairs.
  • drop_measurement(measure) removes the measurement and is safe to call when the measurement does not exist (it must no-op, because drop_features iterates over every possible feature name regardless of what was actually written).

Asymmetry warning: some convenience methods (e.g. has_data) exist on the Influx backends but are not part of the ABC. Cross-backend code (and the EventLoader base) only relies on the abstract methods above. Do not assume a method exists on all backends — check the concrete class.

2.2 The factory dispatch

source/evgen_common.EventLoader._init_evdb is the single place a backend is chosen, keyed on [general] evdb:

def _init_evdb(self, conf, dbname_key):
    db_type = conf["general"]["evdb"]
    if db_type == "influx":
        dbname = conf["database_influx"][dbname_key]
        from . import influx
        return influx.init_influx(conf, dbname, df=False)
    elif db_type == "influx3":
        dbname = conf["database_influx3"][dbname_key]
        from . import influx3
        return influx3.init_influx_v3(conf, dbname)
    elif db_type in ("sql", "sqlite", "mysql"):
        from . import sqlts
        return sqlts.init_sqlts(conf)
    else:
        raise NotImplementedError

dbname_key is "log_dbname" or "snmp_dbname"; each loader passes the right one (e.g. LogEventLoader calls self._init_evdb(conf, "log_dbname")). Your factory receives the resolved database name and the config and returns a TimeSeriesDB instance.

2.3 InfluxDB v3 as the reference example

source/influx3.py is the most recent backend and the cleanest template, because it implements the whole TimeSeriesDB contract from scratch over a thin stdlib HTTP client (no third-party driver, hence no new dependency):

  • class InfluxDBv3(TimeSeriesDB) — writes via POST /api/v3/write_lp (Line Protocol) and reads via GET /api/v3/query_sql (DataFusion SQL, CSV).
  • init_influx_v3(conf, dbname) — the factory: reads [database_influx3] (host/port, optional token, batch_size), with the bearer token taken from the INFLUXDB_V3_TOKEN env var in preference to the config file, and returns the instance.

Its module docstring lists exactly which contract semantics it reproduces (empty-range None/0, local tz-aware index, tag isolation) — a good checklist when writing your own. The v3 backend is wired and contract-tested, but it is newer than the sql and influx (v1) backends and has had less production mileage; treat new backends as needing the same contract-test green bar before trusting them.

2.4 Step-by-step (new evdb backend)

  1. Create logdag/source/<backend>.py with a class MyTSDB(TimeSeriesDB) implementing all abstract methods and honoring the behavioural contract in §2.1.
  2. Add a factory init_<backend>(conf, dbname) that reads your config section and returns the instance (mirror init_influx_v3 / init_sqlts).
  3. Add a branch to evgen_common.EventLoader._init_evdb for your [general] evdb value, importing your module lazily and calling the factory with the right dbname_key.
  4. Add a config section (e.g. [database_<backend>]) with connection options and the log_dbname / snmp_dbname keys to logdag/data/config.conf.default, so [general] evdb = <backend> is self-describing on Configuration Options.
  5. Add the backend to the contract suite. tests/contract/test_tsdb_contract.py parameterizes over backends; wiring yours in is how you prove equivalence to sql / influx. Optional/network backends should skip cleanly when the server is absent (the repo convention is skip-by-default with a "required" gate, so the suite fails loud only where the backend is expected).
  6. Build an evdb and run a DAG end-to-end (logdag.source make-evdb-log-all then logdag make-dag; see Generating DAGs) with evdb = <backend> to confirm the write/read round-trip feeds DAG generation.

2.5 Adding a whole new input source (larger change)

Adding a new origin of events (beyond log / snmp) is more invasive because the source class threads through the event-definition model and several dispatchers. At minimum you would:

  • Add a source constant alongside log2event.SRCCLS_LOG / SRCCLS_SNMP and a branch in log2event.init_evloader (and the load_event_*_all helpers).
  • Implement a loader that subclasses source/evgen_common.EventLoader and a matching EventDefinition subclass (mirroring evgen_log.LogEventDefinition / LogEventLoader) that implements identifier, event(), series(), all_attr(), and iter_evdef().
  • Provide a raw-data adapter (mirroring src_amulog.AmulogLoader / src_rrd) for your origin.
  • Extend pknowledge.AdditionalSource / source-aware filters (subgraph_with_log / subgraph_with_snmp in showdag_filter.py) if the new source should participate in source-based pruning/filtering.

This is a genuine feature, not a one-branch edit; scope it accordingly.

Clone this wiki locally