DPR processors (rs-dpr-service)

Running the DPR processors (Data PRocessors) is the main purpose of RS-Python. The DPR processors are the re-engineered Sentinel-1, Sentinel-2 and Sentinel-3 processors, written in Python by the EOPF (Earth Observation Processing Framework) project of ESA, on top of the EOPF Core Python Modules (CPM) (source code). They generate products in the Zarr format (see the EOPF sample notebooks). They are not part of RS-Python: RS-Python embeds them into Docker images and runs them on its Dask clusters.

rs-dpr-service is the web service that runs the DPR processors. It is not part of the rs-server repository because it must use the same Python and Dask versions as the processors, which are different from those of rs-server: isolating the processing in a dedicated service avoids library conflicts and Python version incompatibilities.

Source code	https://github.com/RS-PYTHON/rs-dpr-service
Swagger page	Processing service on the reference platform
Python API documentation	rs_dpr_service module of the Python API documentation
Task tables	config folder
Helm chart	rs-dpr-service chart

Endpoints

rs-dpr-service follows the OGC API - Processes standard, with pygeoapi, one process per processor:

Method	Endpoint	Description
GET	/dpr/processes	List the processes: s1_l0, s3_l0, s1_ard, s3_l1olci, s3_l2olci, mockup (a mockup processor for the tests) and conv_safe_zarr (conversion of legacy SAFE products to Zarr).
GET	/dpr/processes/{processor}	Get the task table of a processor.
POST	/dpr/processes/{processor}/execution	Run a processor. The request body contains the S3 location of the payload file and of its configuration files, the S3 folder for the processor reports, and the Dask cluster to use (Dask Gateway address and token, cluster label or instance). Returns the job identifier.
GET	/dpr/jobs, /dpr/jobs/{job_id}	List the jobs, get the status and progress of a job.
GET	/dpr/jobs/{job_id}/logs	Stream the logs of a job (server-sent events).
DELETE	/dpr/jobs/{job_id}	Cancel a job, see Cancel a DPR run.

In practice, these endpoints are called by the generic processing flow process-generic of the rs-client-libraries, that computes the payload file from the task table and the flow parameters. Calling them directly is mainly useful to follow or cancel a job.

How a processor is run

Two kinds of Docker images are involved:

Image	Content	Role
rs-dpr-service_py<python>-<dask>	The web service.	Receives the requests, connects to the Dask cluster as a client, submits the processor run, stores the jobs in a PostgreSQL database, streams the logs.
dask/<processor>/k8s	The DPR processor, EOPF CPM and their dependencies, plus the rs_dpr_service code sent to the workers.	The scheduler and workers of the Dask cluster of the processor, created through Dask Gateway (see Dask images).

flowchart LR
    flow["Prefect flow<br/>process-generic"]
    gateway["Envoy Gateway"]
    dpr["rs-dpr-service pod<br/>(Dask client)"]
    subgraph cluster["Dask cluster of the processor"]
        scheduler["Dask scheduler"]
        main["Dask worker:<br/>EOPF main code<br/>(eopf trigger, its own Dask client)"]
        workers["Dask workers:<br/>processing unit tasks"]
    end
    s3[("Object storage")]
    flow -- "POST /dpr/processes/{processor}/execution" --> gateway --> dpr
    dpr -- "1. submit one task" --> scheduler --> main
    main -- "2. submit the processing tasks" --> scheduler
    scheduler --> workers
    workers -- "read inputs, write outputs" --> s3

1. rs-dpr-service connects to the Dask cluster of the processor and submits a single task, that runs the EOPF triggering command (eopf trigger with the payload file) in a sub-process of a Dask worker, and forwards its logs. Before running the command, this task prepares the payload file (see Payload file).
2. EOPF reads the payload file and runs the processing units of the processor: it submits their tasks to the same Dask cluster, that distributes them on its workers. The workers read the inputs and write the outputs directly in the object storage.

In a typical Dask architecture, the main code that submits the tasks runs on the client side, i.e. in the rs-dpr-service pod. Here it runs on a worker of the Dask cluster, because the EOPF main code needs too many resources (memory, processor dependencies) to run in the web service.

The cancellation of a job sets a Dask event that makes the worker kill the EOPF sub-process.

Payload file

The payload files follow the format of EOPF CPM v3. Before triggering the processor, the Dask worker:

• writes the Dask cluster to use in the context_managers section of the payload, with the eopf.dask_utils.dask_context_manager.DaskContext context manager (Dask Gateway address, cluster instance and authentication),
• exports the S3 credentials of the input products (storage_options of their reader_params) as environment variables for the processor: S3_KEY, S3_SECRET, S3_URL, S3_REGION, and S3_SECRETS_JSON_BASE64 that defines the s3input alias. All the input products must use the same credentials,
• validates the payload file with the EOPF validate-payload command (preflight level), if the EOPF version is 3.1.0rc1 or later. An invalid payload stops the job before the processing starts.

Task tables

The task tables of the config folder follow the EOPF CPM v3 format: their io section is split into input, adfs and output lists, each product defining its engine (e.g. cpm_zarr, cpm_safe) and its reader_params or writer_params. They are all validated by the CI against the JSON schema schemas/TaskTable.schema.json, referenced by their $schema property.

Deployment

A Dask client must use the same Python and Dask versions as the Dask cluster it connects to. As the processors don't all use the same versions, rs-dpr-service is built and deployed once per Python and Dask version (rs-dpr-service-py-<python>-dask-<dask> applications of rs-server-deployment). Each instance serves the /dpr/processes/<processor> routes of the processors that use its versions, and the Envoy Gateway routes the requests to the right instance. The list of the Python and Dask versions used is given in the dask-cluster-versions.yml file of rs-workflow-env.

Experimental configurations

For tests and debugging, the request body can contain an experimental_config section (see settings.py):

• local_cluster: run the processor with a Dask LocalCluster created by EOPF inside a single worker of the Dask cluster (or, in local mode, inside the ghcr.io/rs-python/dask/<processor>/localcluster container, which contains both rs-dpr-service and the processor and allows to put breakpoints in the EOPF and processor code). By default, the LocalCluster has one worker per CPU, with 10 GiB of memory and a single thread per worker. It can also be enabled for all the jobs with the DPR_LOCAL_CLUSTER=true environment variable of rs-dpr-service,
• local_files: with a LocalCluster, download the inputs from the object storage to a local folder, run the processor on the local files, then upload the outputs.

Integrate a new processor

See DPR processor integration.