LCLStreamer-API

Detector data streaming application for LCLS.

This is a FastAPI server that pushes datasets on command. The steps run as follows,

1. Receive a REST-API POST request to initiate a dataset push.

   • API request should follow LCLStreamer format.

2. A new transfer ID is created and a port allocated.

3. A separate LCLStreamer process is run via psik to send all images to the requested server.

4. The transfer ID / url combination is returned to the requestor.

5. The requestor should pull the data, e.g. using lclstream pull

6. While the transfer is active, the user can GET the transfer-id to read transfer state and statistics.

7. The pipeline terminates naturally, or the user can DELETE the transfer-id to cancel the producer.

Configuration

Create a configuration file in /etc/lclstream_api.yaml or $VIRTUAL_ENV/etc/lclstream_api.yaml like:

psik:
  prefix: "/sdf/home/r/rogersdd/psik"
  backends:
    default:
      type: "local"
    slurm:
      type: "slurm"
      project_name: "lcls:tmox42619"
      queue_name: "milano"

callback_url: "https://134.79.23.43:4433/v1/callback"

forwarder:
  ip: "134.79.23.43"
  start_port: 30001
  end_port: 31000

  jobspec:
    name: "fastcache"
    backend: "dtn"
    resources:
      duration: 360
      node_count: 1
      processes_per_node: 1
      cpu_cores_per_process: 1
    script: "/sdf/home/r/rogersdd/src/nng_stream/nng_cache"

replay:
  cache_fmt: None # disabled.
  #cache_fmt: "/sdf/home/r/rogersdd/lclstream_cache/%s"
  jobspec:
    name: "lclstream-push"
    backend: "default"
    resources:
      duration: 60
      node_count: 1
      processes_per_node: 1
      cpu_cores_per_process: 1
    script: "lclstream push --addr {url} --ndial 8 {pre}*.h5"

lclstreamer:
  jobspec:
    name: "lclstreamer"
    backend: "slurm"
    resources:
      duration: 60
      node_count: 1
      processes_per_node: 120
      cpu_cores_per_process: 1
    environment:
      CERTIFIED_CONFIG: "/sdf/home/r/rogersdd/venvs/lclstream_api/etc/certified"
    script: |
      pixi run --manifest-path /sdf/home/r/rogersdd/src/lclstreamer -e {psana_env} /sdf/group/lcls/ds/ana/sw/conda2/inst/envs/ps_20241122/bin/mpirun lclstreamer --config lclstreamer.json

Most of the important configuration parameters are psik.JobSpec objects explaining how to run producer and message forwarder processes.

• callback_url: The URL to which psik jobs should post their progress. This has to be accessible from the host running the lclstreamer job.

• psik: Configuration information for the psik client. This defines the local directory on the host running lclstream_api used to track jobs, as well as job execution backends for running the producer and message forwarder processes.

• lclstreamer: Setup information for the data source.

  • jobspec Defines the spec used to start the producers. SLURM configuration parameters go into the psik.backend definitions.

• forwarder: The API needs the following information to start message-forwarding processes:

  • jobspec Defines the spec used to start the message-forwarding process. This is usually run on a data transfer node.

  • ip: The IP address used to construct the message forwarder URLs (tcp://IP:port)

  • start_port/end_port: The range of ports that will be used to start message forwarding servers on the ip above. Forwarding services always allocate two ports at once (receive,send -- in that order). Thus, each transfer is bound to a PortPair object at creation time.

• replay: if setup, can be used to implement a call to lclstream push which will forward all the h5 files in a given prefix. This is not tested at present due to the need to check file paths, but can be a future optimization when the the same data is requested twice.

The '{}' variables within the scripts are substituted with the appropriate values by lclstream_api during job creation. Also during job creation, the api ensures lclstreamer.json (as requested by the user's API call) is present in the job's run directory. So that path name should be left as-is.

Bootstrapping a New Installation

Installing this package brings with it the lclstream and the psik packages. The lclstream package is a client for lclstreamer. It can be used to run a test consumer. The psik package is invoked by lclstream_api to run the producer job. Before running transfers through the API, these two should be tested.

For setting up backends, create a $VIRTUAL_ENV/etc/psik.json file with backend configuration information,

{ "prefix": "/sdf/home/r/rogersdd/psik",
  "backends": {
    "default": {
        "type": "local"
    },
    "slurm": {
        "type": "slurm",
        "project_name": "lcls:tmox42619",
        "queue_name": "milano"
    }
}

Create the following yaml file specifying a psik job:

# slurm_test.yaml
name: lclstreamer_test
backend: slurm
resources:
  duration: 5
  node_count: 1
  processes_per_node: 1
  cpu_cores_per_process: 5
  gpu_cores_per_process: 0
  exclusive_node_use: false # not needed during testing
script: |
  cat >params.json <<.
  { "lclstreamer": {
    "source_identifier": "",
    "event_source": "InternalEventSource",
    "processing_pipeline": "BatchProcessingPipeline",
    "data_serializer": "Hdf5BinarySerializer",
    "data_handlers": [ "BinaryDataStreamingDataHandler" ],
    "skip_incomplete_events": false
  },
  "event_source": {
    "InternalEventSource": { "number_of_events_to_generate": 476 }
  },
  "data_sources": {
    "random": {
      "type": "GenericRandomNumpyArray",
      "array_shape": "20,2",
      "array_dtype": "float32"
    }
  },
  "data_serializer": {
    "Hdf5BinarySerializer": {
      "compression_level": 3,
      "fields": { "random": "/data/random" }
    }
  },
  "data_handlers": {
    "BinaryDataStreamingDataHandler": {
      "urls": [ "tcp://134.79.23.43:5001" ],
      "role": "client",
      "library": "nng",
      "socket_type": "push"
    }
  },
  "processing_pipeline": {
    "BatchProcessingPipeline": { "batch_size": 10 }
  }}
  .
  pixi run --manifest-path /sdf/home/r/rogersdd/src/lclstreamer -e psana1 mpiexec lclstreamer --config params.json

This creates an lclstreamer config, then executes lclstreamer within mpiexec. The producers will send their data to tcp://134.79.23.43:5001.

Before running this, execute a consumer on the host above (the IP above is sdfdtn003 in this case):

lclstream pull --listen tcp://134.79.23.43:5001 | tar tf -

Now put the job into the queue using

psik run slurm_test.yaml

If psik's slurm backend has been configured to run this job on the psana cluster, you should see data being received from the consumer side. To diagnose issues at this stage, look at the job's output using:

psik ls

and then reading the logs stored in this job's log directory. If there are errors, you can re-start the same job using

psik start <jobid>

rather than re-creating a copy of the same job. However, you'll need a new psik run command if you change the backend config. or the job spec (slurm_test.yaml).

Once this is working, both of the two pieces of information above (the psik config and the run-script line) should be pasted into $VIRTUAL_ENV/etc/lclstream_api.json described above.

Comments on psik backend configuration

It is simplest to use the SLURM backend to run jobs on the psana cluster. Additional backend configuration options are documented in psik.models.BackendConfig.

• queue_name : Optional[str] = None
• project_name : Optional[str] = None
• reservation_id : Optional[str] = None
• attributes : Dict[str,str] = {}

These are processed into the sbatch command-line when a psik job is launched. It is possible to use other backends or create new ones in order to start/cancel jobs differently.

Accessing psana from login and dtn nodes

Since we are running the lclstreamer-API from sdfdtn003, we need a way to manage SLURM jobs on the psana cluster from that system.

I did this with minimal hassle by creating wrapper scripts for sbatch and scancel in $HOME/swrap. Then my .bashrc contains the line

export PATH=$PATH:$HOME/swrap

This appends the swrap directory to the PATH. The wrappers are only visible on hosts without sbatch, since (e.g. on a psana node), the system's native sbatch command comes earlier in PATH.

#!/bin/bash
#sbatch
ssh psana cd $PWD '&&' sbatch $@

#!/bin/bash
#scancel
ssh psana scancel $@

Development

Install the code using uv sync --all-extras or pip install -e .[*].

Create a configuration file as explaned above, or in a local directory. If you set the LCLSTREAM_API_CONFIG environment variable to point at your own json file, it will override the one in $VIRTUAL_ENV/etc.

For testing, you can create a self-signed identity using

uv run certified init --host 127.0.0.1 test-cert

Manually run the server code with:

uv run uvicorn lclstream_api.server:app --reload

or

uv run certified serve lclstream_api.server:app https://127.0.0.1:4433

It is helpful to create a minimal $VIRTUAL_ENV/etc/psik.json setting

{ "prefix": "/path/to/prefix"}

so you can use psik ls.

Deployment Instructions

Install this package and its dependencies using pip install . (for deployment)

If you do not yet have a server certificate, create a server keypair using instructions from certified.

Run the server with the uvicorn launch command above, but specifying the key and certificate files as explained there.

certified serve lclstream_api.server:app https://0.0.0.0:4433

or

uvicorn --ssl-keyfile server.key --ssl-certfile server.pem \
        --ssl-cert-reqs 1 --ssl-ca-certs ca_root.pem \
        lclstream_api.server:app

Sign user certificates using:

certified introduce user.pem

or by installing and running signer.