graflag package

GraFlag - Graph Anomaly Detection Benchmarking Tool

A tool for benchmarking Graph Anomaly Detection methods using Docker Swarm across multiple nodes with shared NFS storage.

class graflag.GraFlag(config_file: str | None = None, interactive: bool = True)

Bases: object

Main GraFlag orchestration class.

All public methods return structured data. No direct printing to stdout (except follow_logs which streams in real time).

setup()

Setup GraFlag cluster: initialize swarm and setup workers.

status() → ClusterInfo

Get cluster status.

Returns:

ClusterInfo with nodes, services, and shared directory info.

static make_experiment_name(method_name: str, dataset: str) → str

Build the canonical experiment directory name.

Callers that need the name before the (blocking) run starts – the GUI, for instance – must use this and pass the result to run(), otherwise their name and the real directory disagree on both casing and timestamp.

run(method_name: str, dataset: str, tag: str = 'latest', build: bool = False, gpu: bool = True, method_params: dict = None, exp_name: str = None, keep_service: bool = False, force_rm: bool = False, follow: bool = True) → str

Run experiment.

Blocks until the run has ended either way.

Parameters:

• exp_name – Pre-computed experiment name (see make_experiment_name()). Generated when omitted.

• follow – Stream the container’s output to stdout while waiting, as the CLI does. False waits by polling the experiment’s status instead and prints nothing – for the dashboard and the MCP server, where stdout is not a terminal (for the MCP server it is the protocol).

• keep_service – Leave the finished Swarm service in place. Useful when you want docker service ps to inspect the task.

• force_rm – With build, remove the build’s intermediate containers even when it fails (docker build --force-rm).

Returns:

Experiment name.

Raises:

GraFlagError – If run fails.

wait_for_experiment(experiment_name: str, poll: float = 5.0, timeout: float | None = None) → ExperimentInfo | None

Wait until an experiment reaches a terminal status, printing nothing.

Polls the same probe graflag list experiments uses, so “finished” means what the listing says it means, including the task that died before the runner could write status.json. An experiment whose directory holds no status and whose service is gone (“unknown”) also ends the wait: nothing is left that could change it.

Returns the last ExperimentInfo, or None when the experiment does not exist. With timeout, returns the current state once it has passed, terminal or not.

register_metric(result_type: str, metric_func: Callable, experiment: str = None)

Register a custom metric as a plugin file on the cluster.

The function source is extracted via inspect.getsource and written to a .py plugin file that the evaluator loads at runtime.

Parameters:

• result_type – Result type the metric applies to (e.g. "EDGE_STREAM_ANOMALY_SCORES").

• metric_func – A function with signature (scores, ground_truth, **kwargs) -> Dict[str, float].

• experiment – If given, the plugin is scoped to that experiment (custom_metrics/ inside the experiment directory). Otherwise it is saved to the global plugins directory.

Raises:

GraFlagError – If the function source cannot be extracted or the file cannot be written.

evaluate(experiment_name: str, follow: bool = True)

Evaluate an experiment: compute metrics and generate plots.

Parameters:

follow – Stream the evaluator’s output to stdout, as the CLI does. False waits for it silently (see run()).

Raises:

GraFlagError – If evaluation fails.

verify(experiment_name: str) → VerificationReport

Check that a finished experiment published a result worth believing.

The last of the four integration gates, after evaluate(): it compares what the method says it measured with what it published. See graflag.verify for the checks. Reads the experiment in one remote call and never moves the scores to the client.

Raises:

GraFlagError – If the experiment does not exist or cannot be read.

list_methods() → List[MethodInfo]

List available methods.

Returns:

List of MethodInfo objects.

list_datasets() → List[DatasetInfo]

List available datasets.

Returns:

List of DatasetInfo objects.

count_experiments() → int

How many experiment directories exist on the share.

list_experiments takes a limit and gives no indication when it truncates, so a dashboard showing 50 of 78 looked complete. This is the cheap other half: one ls | wc -l, no per-experiment probing, so it can be called alongside the list without doubling its cost.

Prefers the total captured by the most recent list_experiments, which gets it from the same remote call it already makes. Asking separately doubled the cold cost of the GUI’s most-polled endpoint (measured: 437ms -> 1.06s) for a single integer.

Returns:

The directory count, or 0 when the experiments root is absent.

list_experiments(limit: int = 50, offset: int = 0) → List[ExperimentInfo]

List recent experiments.

Probes every experiment in a single remote command. The previous implementation issued one SSH connection per experiment directory – and there is no ControlMaster, so each was a fresh TCP and auth handshake. With 50 experiments that was 52 connections, repeated every two seconds by the GUI’s background updater. limit was also applied only after probing everything.

Parameters:

• limit – how many to return.

• offset – how many of the newest to skip. Paging was done in the browser over whatever the API had already capped, so the last page was the end of the cap rather than the end of the data.

Returns:

List of ExperimentInfo (most recent first).

list_services() → List[Dict]

List running Docker services.

Returns:

List of service dicts with name, replicas, image, status.

get_logs(experiment_name: str, tail: int = 100) → List[str]

Get experiment logs (non-streaming).

Tries Docker service logs first, then falls back to method_output.txt.

Returns:

List of log lines.

follow_logs(experiment_name: str, tee_file: str = None)

Follow logs for an experiment (streams to stdout).

Shows build log (if exists) + service logs. Falls back to method_output.txt if the service is gone.

show_logs(experiment_name: str, tee_file: str = None)

Show logs (non-follow mode) — prints to stdout.

stop(experiment_name: str, remove: bool = False)

Stop a running experiment/service.

Parameters:

• experiment_name – Name of the experiment

• remove – If True, also delete the experiment directory

TERMINAL_STATUSES = ('completed', 'failed', 'stopped')

WAIT_MISSES = 3

Consecutive failed probes after which wait_for_experiment() concludes that the experiment is gone rather than that ssh blinked.

cleanup_services(experiment: str = None, dry_run: bool = False) → List[ServiceCleanupResult]

Remove Swarm services for experiments that have finished.

Swarm services are created with restart_policy=none and were never removed, so every run left one behind forever. They are what docker service ls and graflag list services show, and they hold task records on the manager indefinitely.

Two safety rules, because removing a service is not reversible:

1. Only terminal experiments are touched. Anything still running or building keeps its service.

2. The service is kept unless the run’s output is already on disk. graflag logs falls back to method_output.txt once the service is gone, but a run that died before the runner wrote that file has its Docker logs as the only diagnostic – removing the service would destroy the evidence for the failure you most need to read.

Parameters:

• experiment – Limit to one experiment. Default: sweep all.

• dry_run – Report what would be removed without removing it.

Returns:

One ServiceCleanupResult per candidate considered.

OWNED_DIRS = ('datasets', 'experiments', 'images', 'libs', 'methods')

The five directories GraFlag owns at the root of the share. They are named here as well as in the survey so that a widened rule in one place cannot reach them: the survey decides what is collected, this decides what may be deleted.

clear(apply: bool = False, share: bool = True, images: bool = True, collect: bool = False) → ClearReport

Remove what is on the cluster but no longer belongs to anything.

graflag cleanup sweeps finished services; this sweeps the storage they leave behind. An item is removed only on evidence that nothing owns it any more, never because its name looks disposable:

• an experiment whose methods/<name> directory is gone. The method was deleted or renamed, so the run can no longer be reproduced or even read against its definition.

• a dataset referenced only by such experiments. A dataset no experiment has used yet is kept – it is there to be run, and “used by nothing” is its normal state before the first run.

• an image, local or in the registry, whose owner is not a method. The owner is IMAGE= from the method’s .env when it declares one, so the seventeen bond_* methods keep bond_base alive between them.

• a stray at the root of the share: anything beside the five directories GraFlag owns.

Nothing is removed unless apply is true. The dry run reports the same rows with removed false, so the two are diffable.

collect additionally garbage-collects the registry’s blobs, which is what actually reclaims the disk – deleting a repository removes its manifests and leaves every layer behind. It is a separate flag because it cannot be done safely while the registry is serving: garbage collection deletes blobs the running registry still answers HEAD for from its cache, so the next docker push reports “Layer already exists”, uploads nothing, and produces an image the manager can run and a worker cannot pull. This scales the service to zero first and back up afterwards, in a finally, so an error during collection cannot leave the cluster without a registry.

Parameters:

• apply – Remove. Default is to report only.

• share – Consider experiments, datasets and strays.

• images – Consider local images and registry repositories.

• collect – Garbage-collect registry blobs (implies images).

Returns:

A ClearReport. Failures to remove an individual item are collected in errors rather than raised – clearing is housekeeping and must not fail the caller.

get_experiment_results(experiment_name: str) → ExperimentResults | None

Get experiment results from results.json.

get_evaluation_results(experiment_name: str) → EvaluationResults | None

Get evaluation results from eval/evaluation.json.

copy_files(source_paths, dest_path: str, recursive: bool = False, from_remote: bool = False)

Copy files/directories bidirectionally.

mount_nfs(shared_dir: str)

Mount NFS share on local machine.

sync(local_path: str, is_lib: bool = False)

Sync a local method or library directory to remote shared storage.

get_experiment(experiment_name: str) → ExperimentInfo | None

One experiment’s status, from a single remote probe.

None when the experiment does not exist (or the manager could not be reached – the probe cannot tell the two apart).

exception graflag.GraFlagError

Bases: Exception

Custom exception for GraFlag errors.

class graflag.GraflagConfig(config_file: str | None = None)

Bases: object

Handle configuration loading and validation for GraFlag.

get(key: str, default: str | None = None) → str | None

Get configuration value.

property remote_shared_dir: str

property manager_ip: str

property ssh_port: str

property ssh_key: str | None

property nfs_port: str

property hosts_file: str | None

property graflag_libs: str

class graflag.ClusterInfo(manager_ip: str, is_connected: bool, swarm_initialized: bool, worker_nodes: Dict[str, str]]=<factory>, shared_dir: str = '', shared_contents: List[str] = <factory>, services: List[Dict] = <factory>, error: str | None = None)

Bases: object

Cluster status information.

manager_ip: str

is_connected: bool

swarm_initialized: bool

worker_nodes: List[Dict[str, str]]

shared_dir: str = ''

shared_contents: List[str]

services: List[Dict]

error: str | None = None

to_dict() → dict

class graflag.MethodInfo(name: str, description: str = '', source_code: str = '', integration: str = '', image: str = '', supported_data: str = '', parameters: Dict[str, ~typing.Any]=<factory>, has_dockerfile: bool = False, has_env: bool = False)

Bases: object

Method metadata.

name: str

description: str = ''

source_code: str = ''

integration: str = ''

upstream (the image runs the authors’ implementation) or reimplementation (this repository’s code, written from the paper). Empty for a method whose .env predates the key – rendered as “unstated”, which is what it is, rather than assumed to be either.

image: str = ''

The shared image this method builds from, from IMAGE= in its .env. Empty when the method builds its own – which is the common case; the seventeen bond_* methods share bond_base.

supported_data: str = ''

parameters: Dict[str, Any]

has_dockerfile: bool = False

has_env: bool = False

to_dict() → dict

class graflag.DatasetInfo(name: str, path: str = '', size_mb: float = 0.0, file_count: int = 0)

Bases: object

Dataset metadata.

name: str

path: str = ''

size_mb: float = 0.0

file_count: int = 0

to_dict() → dict

class graflag.ExperimentInfo(name: str, method: str, dataset: str, timestamp: str, status: str, has_results: bool = False, has_evaluation: bool = False, results_path: str | None = None, evaluation_path: str | None = None, service_name: str | None = None)

Bases: object

Experiment metadata and status.

name: str

method: str

dataset: str

timestamp: str

status: str

has_results: bool = False

has_evaluation: bool = False

results_path: str | None = None

evaluation_path: str | None = None

service_name: str | None = None

to_dict() → dict

class graflag.ExperimentResults(experiment_name: str, method_name: str, dataset: str, metadata: Dict[str, ~typing.Any]=<factory>, execution_time_ms: float | None = None, peak_memory_mb: float | None = None, peak_gpu_memory_mb: float | None = None, result_type: str | None = None, scores_available: bool = False)

Bases: object

Parsed experiment results.

experiment_name: str

method_name: str

dataset: str

metadata: Dict[str, Any]

execution_time_ms: float | None = None

peak_memory_mb: float | None = None

peak_gpu_memory_mb: float | None = None

result_type: str | None = None

scores_available: bool = False

to_dict() → dict

class graflag.EvaluationResults(experiment_name: str, metrics: Dict[str, float]=<factory>, plots_available: List[str] = <factory>, evaluation_path: str | None = None)

Bases: object

Parsed evaluation results.

experiment_name: str

metrics: Dict[str, float]

plots_available: List[str]

evaluation_path: str | None = None

to_dict() → dict

class graflag.RunProgress(experiment_name: str, status: str, message: str = '', log_lines: List[str] = <factory>)

Bases: object

Progress information for run execution.

experiment_name: str

status: str

message: str = ''

log_lines: List[str]

to_dict() → dict

class graflag.ServiceCleanupResult(experiment: str, removed: bool, reason: str)

Bases: object

Outcome of considering one finished service for removal.

experiment: str

removed: bool

reason: str

to_dict() → dict