ska-src-ef-broker-client

Client library for the SRCNet computing broker: submit jobs, watch them, fetch their logs.

One dependency (requests). Installing the broker service instead would pull FastAPI, SQLAlchemy, Alembic, pika and the HTCondor bindings onto a machine that only wants to submit a job.

pip install ska-src-ef-broker-client

Run something

from ska_src_ef_broker_client import Broker

broker = Broker("https://broker.example.org", token=my_access_token)

job = broker.run("echo hello", image="registry.example.org/tools/app:1")
job.wait()                 # blocks until terminal, returns the concise status
print(job.state)           # COMPLETE
print(job.logs())

You never name a site: you say what to run and which data it needs, and the broker decides where that can happen.

Follow the data

Passing Rucio data identifiers makes the broker resolve where they live and restrict placement to the sites holding them:

job = broker.run(
    "./analyse.sh /srcnet/input",
    image="registry.example.org/tools/analysis:2",
    dids=["SKA-Mid.integration:EB-1234.product-abcd"],
    input_url="dav://storm.example.org/sa/inputs",
)

/srcnet/input exists only when the job declares an input area — run() omits --input otherwise rather than binding a path you never asked for.

A Dask cluster

The broker starts a scheduler and workers as sibling tasks on one site and hands your command the rendezvous file:

broker.run_dask(
    "python analysis.py",         # your code is the cluster's client
    image="registry.example.org/tools/analysis-dask:1",
    workers=4,
).wait(timeout=3600)

Your image runs all three roles, so it has to contain Dask itself — the scheduler and workers are dask scheduler / dask worker in your image, and the workers execute your code, so they need your libraries too. One layer is enough:

FROM registry.example.org/tools/analysis:1
RUN pip install --no-cache-dir "dask[array,distributed]==2026.7.1"

An image without Dask fails fast with a message naming it. See docs/dask-image-contract.md in the broker repository for the full contract and a pinned base image to derive from.

# inside analysis.py
import os
from dask.distributed import Client

client = Client(scheduler_file=os.environ["SKA_DASK_SCHEDULER_FILE"])

The cluster is torn down when your command exits.

A rapthor pipeline

broker.run_rapthor(
    data_dir="/srv/storage/site/sa/my-observation",
    image="registry.example.org/tools/rapthor:latest",
    cpu=8,
    threads=8,
)

Watch and inspect

A Job caches nothing, so a job id is all you need later — including after a notebook restart:

job = broker.job("job-abc123")
job.state            # current state
job.status()         # concise view
job.logs()           # printable log tails
job.next_action()    # triage hint when a job looks stuck
job.cancel()

broker.jobs(limit=20)          # recent jobs
broker.is_ready()              # broker health

Defaults can live on the client instead of every call:

broker = Broker(
    "https://broker.example.org",
    token=my_access_token,
    image="registry.example.org/tools/app:1",
    input_url="dav://storm.example.org/sa/inputs",
)
Broker.from_env()   # BROKER_URL / BROKER_TOKEN / BROKER_IMAGE / ...

The transport layer

BrokerClient is one method per HTTP endpoint, each returning a requests.Response, for callers who want the responses themselves:

from ska_src_ef_broker_client import BrokerClient

client = BrokerClient("https://broker.example.org", bearer_token=token)
response = client.get_job("job-abc123")
print(response.json()["state"])

Broker is built on it — it fills in the submit payload, gives jobs an object identity, and turns polling into job.wait().

Notes

• A token is required. Every call without one answers 401.
• TLS: for a private CA pass its bundle — verify="/path/ca.crt". verify=False disables verification altogether, which exposes the bearer token these calls carry to anyone on the path; keep it out of anything but a throwaway probe.
• A worked example of all three workload classes lives in the broker repository at demo/notebooks/broker_showcase.py (a marimo notebook).