graflag.docker_ops

Docker operations for GraFlag using Docker SDK.

graflag.docker_ops.BUILD_ARG_KEYS = ('SOURCE_CODE', 'SOURCE_REF')

.env keys forwarded to docker build as --build-arg.

A method declares the value once, in its .env, and its Dockerfile reads it through an ARG of the same name. The alternative – writing the value into the Dockerfile – is what let a .env advertising one SOURCE_CODE sit next to an image cloning another, and it is why an upstream commit could not be pinned without editing two files in step.

Only keys named here cross the boundary. A method’s parameters are _-prefixed and belong to the run, not to the build.

graflag.docker_ops.GRAFLAG_LIBS_CHOICES = ('local', 'pypi')

Where a method image gets the GraFlag libraries.

Not a BUILD_ARG_KEYS entry, because it is not a property of the method. Whether the published wheel is current is a property of the deployment, and a method that could answer it differently from its neighbours would be a benchmark whose runs are not comparable.

graflag.docker_ops.IMAGE_NAME = re.compile('^[a-z0-9][a-z0-9._-]*$')

What a shared IMAGE= may be called.

The value is both a directory under images/ and half of an image reference, and Docker rejects an uppercase repository name – at push time, after the build has already run for minutes.

exception graflag.docker_ops.BuildFailed(message: str, log: str)

Bases: RuntimeError

A docker build that exited non-zero, with everything it printed.

The legacy builder prints each step’s output on stdout and leaves stderr the deprecation banner and “returned a non-zero code: 128” – which step failed, never why. The why was on stdout: for the rare method, git’s fatal: could not read Username for 'https://github.com' from cloning a private repository. An error built from stderr alone dropped it, and build.log was written only after a successful build, so a failed build kept no record of its cause. log is what build.log would have held.

class graflag.docker_ops.ImageNames(local: str, registry: str, owner: str, dockerfile: str, shared: bool)

Bases: NamedTuple

Every name one method’s image is known by, resolved together.

Built and deployed from the same call, so the tag that docker build produces and the one Swarm pulls cannot come apart.

local: str

<owner>:<tag> – the local tag on the manager.

registry: str

<manager>:5000/<owner>:<tag> – what Swarm pulls.

owner: str

the method itself, or the shared image named by IMAGE=. Never the same thing as METHOD_NAME at run time.

Type:

The method that owns the image

dockerfile: str

Remote path of the Dockerfile that builds it, already quoted.

shared: bool

True when the image is shared, i.e. IMAGE= was declared.

class graflag.docker_ops.ReservedEnvVars(*values)

Bases: Enum

Reserved environment variable names that should not be overridden by method parameters.

DATA = 'DATA'

EXP = 'EXP'

METHOD_NAME = 'METHOD_NAME'

COMMAND = 'COMMAND'

MONITOR_INTERVAL = 'MONITOR_INTERVAL'

GRAFLAG_PARAMS = 'GRAFLAG_PARAMS'

classmethod get_names()

graflag.docker_ops.TERMINAL_TASK_STATES = ('complete', 'failed', 'shutdown', 'rejected', 'orphaned')

Swarm task states after which a restart_policy=none task never runs again.

class graflag.docker_ops.DockerManager(ssh_manager, config, hosts_file: str = 'hosts.yml')

Bases: object

Handle Docker Swarm operations via Docker SDK with SSH tunnel.

property client: DockerClient

Lazy-initialize Docker client via SSH tunnel.

close()

Close Docker client and SSH tunnel.

Tolerates a partially-constructed instance: __del__ calls this, and if __init__ raised before setting these attributes the resulting AttributeError would surface as noise during garbage collection.

setup_swarm_manager()

Initialize Docker Swarm on manager node.

get_swarm_token() → str

Get Docker Swarm worker join token.

setup_workers(token: str)

Setup worker nodes to join the swarm (requires SSH to each worker).

get_nodes() → List[Dict]

Get list of swarm nodes.

setup_local_registry()

Setup local Docker registry service on manager.

build_method_image(method_name: str, tag: str = 'latest', method_env: Dict[str, str] | None = None, force_rm: bool = False) → str

Build method Docker image and push to local registry.

Uses SSH because the build context resides on the remote host.

Every BUILD_ARG_KEYS entry the method’s .env defines is passed through as --build-arg, so a Dockerfile can pin its upstream from the one place that already names it.

Parameters:

• method_name – Method whose methods/<name>/Dockerfile to build.

• tag – Image tag.

• method_env – The method’s parsed .env, when the caller already has it. Omitted, it is read from the manager – one extra SSH round trip, against a build that takes minutes.

• force_rm – Pass --force-rm, so docker removes the build’s intermediate containers even when it fails. Without it a failed build keeps its last container, and that container holds every layer the build had produced: 12.8 GB on the manager after one failed git clone of the rare method.

Returns:

Combined build and push log output.

build_evaluator_image() → str

Build graflag-evaluator image and push to registry.

Returns:

Registry image path.

create_service(exp_name: str, method_name: str, dataset: str, tag: str = 'latest', gpu_required: bool = True, method_params: dict = None) → str

Create Docker service for experiment.

create_evaluation_service(experiment_name: str) → str

Create Docker service to run evaluation.

list_services() → List[Dict]

List all Docker services.

The whole listing is the retry unit, not the first call in it: the drop usually lands on one of the per-service tasks() calls, and a svc handed out by the old client keeps pointing at the old client, so retrying just that call reuses the socket that already died.

get_service_names() → set

Get set of all service names.

stop_service(service_name: str)

Stop and remove a service.

cleanup_finished_service(service_name: str)

Remove a finished service (safe if it doesn’t exist).

remove_evaluation_service(experiment_name: str)

Remove evaluation service for an experiment.

get_service_logs(service_name: str, tail: int = 100) → List[str]

Get recent logs for a service.

Uses SSH + Docker CLI because the Docker SDK log streaming is unreliable for swarm services.

follow_service_logs(service_name: str, on_line=None)

Follow service logs in real-time until the task finishes.

Uses SSH + Docker CLI subprocess because the Docker SDK’s follow mode does not stream reliably for swarm services.

Parameters:

on_line – Optional callback receiving each streamed log line. When given, output is read here and forwarded rather than inherited, so callers can capture what the user sees.

Returns:

The terminal task state (‘complete’, ‘failed’, …), or None if the wait ended without observing one.

service_exists(service_name: str) → bool

Check if a Docker service exists.

service_task_state(service_name: str) → str | None

The newest task’s state (running, complete, …), lower case.

None while the service has no task yet. Raises ValueError when the service does not exist.

wait_for_service(service_name: str, poll: float = 3.0, timeout: float | None = None) → str | None

Wait until the service’s task ends, printing nothing.

The quiet counterpart of follow_service_logs(), for callers with nobody reading a terminal: the dashboard, and the MCP server, whose stdout is its protocol. Returns the terminal task state, or None if timeout seconds passed first.

is_service_failed(service_name: str) → bool

True when the service exists and no task reached a good end state.

Observed terminal states for a restart_policy=none service are complete on success and failed on error; shutdown and orphaned mean the task was stopped or its node went away, which is not a method failure but is not success either. This is only consulted when status.json is absent or unreadable – the runner’s own status wins.

get_cluster_status() → Dict

Get Docker Swarm cluster status.