Table of contents

• Description
• Installation
• Operating Instructions
• Jupyter Notebook Examples
• Apache Airflow Support

Description

The American Science Cloud Infrastructure Services Resource Orchestration Toolkit (AmSC-ISRO-Toolkit [AmSCROT]) provides infrastructure orchestration for AmSC use.

Installation

pip install amscrot-py

Operation Instructions

Credentials

AmSCROT reads provider credentials from ~/.amscrot/credentials.yml. Each section key corresponds to a service type or profile name:

esnet-iri:
  api_key: <token>
  api_endpoint: https://iri.es.net/api/v1

nersc-iri:
  api_key: <token>
  api_endpoint: https://api.iri.nersc.gov/api/v1

amsc-iro:
  api_key: <token>
  api_endpoint: https://...

An example credentials file is provided in credentials-template.yml.

Core Concepts

Basic Usage

1. Set up a Client and ServiceClient

from amscrot.client.client import Client
from amscrot.serviceclient import ServiceClient
from amscrot.util.constants import Constants

client = Client()

# Choose a provider: KUBE, AMSC_IRI, AMSC_IRO
svc = ServiceClient.create(
    type=Constants.ServiceType.AMSC_IRI,
    name="nersc-compute",
    profile="nersc-iri"           # matches credentials.yml section
)
client.add_service_client(svc)

2. Discover Available Resources

result = svc.discover()      # returns DiscoveryResult

# Iterate typed resources
for item in result.by_type("compute"):
    print(item.data["id"], item.data["name"])

# Normalized Facility objects (provider-agnostic)
for facility in result.facilities:
    print(facility.name, [c.cores for c in (facility.compute or [])])

3. Define a Job

from amscrot.client.job import Job, JobSpec, JobType, JobServiceType

spec = JobSpec(
    executable="python",
    arguments=["-c", "print('hello')"],
    resources={
            "node_count": 1,
            "process_count": 1,
            "processes_per_node": 1,
            "cpu_cores_per_process": 1,
            "exclusive_node_use": False,
            "memory": 268435456
    },
    attributes={
        "container": {"image": "python:3.12-slim"},  # provider image
        "resource_id": "<compute-resource-id>"       # from discovery
    }
)

job = Job(
    name="my-job",
    type=JobType.COMPUTE,
    service_type=JobServiceType.BATCH,
    service_client=svc,
    job_spec=spec
)

4. Create a Session and Submit

session = client.create_session("my-session")
session.add_job(job)

# Validate (raises PlanError on failure)
session.plan(verbose=True)

# Submit all jobs
session.apply()

5. Wait for Completion

from amscrot.client.job import JobState

results = session.wait(
    timeout=300,
    interval=5,
    verbose=True,
)

for job_name, status in results.items():
    print(f"{job_name}: {status.state}  message={status.message}")

session.wait() polls until all jobs reach a terminal state (COMPLETED, FAILED, or CANCELED). Pass jobs=[job1, job2] to wait on a subset.

6. Clean Up

session.destroy()   # cancels running jobs and removes session state

Sessions are persisted to ~/.amscrot/sessions/<session-name>/ so they survive process restarts. An existing session is restored automatically on client.create_session(name).

7. Fetch Output Files (IRI providers)

After a job completes, stdout/stderr can be downloaded from the remote filesystem:

# Include stdout/stderr paths in the job spec attributes
spec = JobSpec(
    executable="python",
    arguments=["-c", "print('done')"],
    attributes={
        "resource_id": "<compute-resource-id>",
        "directory": "/path/to/workdir",        # remote working directory
        "stdout_path": "/path/to/workdir/out.log",
        "stderr_path": "/path/to/workdir/err.log",
    }
)

# After session.wait() returns COMPLETED:
fetched = session.fetch_output_files(jobs=[job])
# fetched == {"my-job": {"stdout": "/local/.amscrot/sessions/my-session/files/my-job/stdout.log",
#                        "stderr": "/local/.amscrot/sessions/my-session/files/my-job/stderr.log"}}

Files are written to ~/.amscrot/sessions/<session-name>/files/<job-name>/ by default. Pass output_path= to override.

8. Direct Filesystem Access (IRI providers)

ESnet IRI and NERSC IRI service clients expose an IriFilesystem interface for direct file operations independent of job submission:

fs = svc.filesystem          # IriFilesystem instance (None if client unavailable)

# Upload a local file to remote storage
fs.upload(storage_resource_id, local_path="/tmp/input.txt", remote_path="/scratch/input.txt")

# Download a remote file
fs.download(storage_resource_id, remote_path="/scratch/out.log", local_path="/tmp/out.log")

# List a remote directory
entries = fs.list(storage_resource_id, remote_path="/scratch/")

# Compute checksum of a remote file
checksum = fs.checksum(storage_resource_id, remote_path="/scratch/data.tar")

# Create a tar archive of a remote directory
fs.compress(storage_resource_id, remote_path="/scratch/results/", archive_path="/scratch/results.tar.gz")

storage_resource_id is the UUID of a storage resource from svc.discover(). For most IRI deployments, the home storage resource is auto-resolved when calling session.fetch_output_files().

Kubernetes / Kueue Jobs

spec = JobSpec(
    executable="sleep",
    arguments=["30"],
    resources={"requests": {"cpu": "1", "memory": "1Gi"}},
    attributes={
        "container": {"image": "busybox"},
        "namespace": "default",
        "labels": {"kueue.x-k8s.io/queue-name": "compute-queue"},
        "completions": 1,
        "restartPolicy": "Never"
    }
)

See scripts/kube/setup-kueue.sh to install Kueue and create the required ResourceFlavor, ClusterQueue, LocalQueue, and PriorityClass resources on your cluster.

Jupyter Notebook Examples

Interactive notebooks are provided under examples/notebooks/client/.

Apache Airflow Support

The airflow/ subdirectory provides custom Airflow operators for submitting and monitoring IRI compute jobs as part of larger data pipelines.

Operators

• IriJobSubmitOperator — Plans, submits, and waits for an IRI job. Accepts service_type, profile, executable, resources, and attributes. Pushes job_id to XCom on completion.
• IriFetchOutputOperator — Downloads stdout/stderr from a completed job to a local directory.

Quick Start

cd airflow/
bash setup_airflow.sh --start   # installs deps, initialises DB, launches standalone Airflow

Open http://localhost:8080, configure credentials in ~/.amscrot/credentials.yml, then trigger the demo DAG (esnet_iri_example or gpt2_training_job) from the UI or via the REST API:

curl -X POST http://localhost:8080/api/v2/dags/esnet_iri_example/dagRuns \
     -H "Content-Type: application/json" \
     -u "admin:<password>" -d '{}'

See airflow/README.md for the full operator reference and configuration options.