rs_workflows/on_demand_processing.md

<< Back to index

Prefect flows and tasks for on-demand processing

build_and_generate_payload(logger, flow_env, task_table, dpr_input, retry_config)

Build and generate the payload file and the lineage source items

Source code in docs/rs-client-libraries/rs_workflows/on_demand_processing.py

def build_and_generate_payload(
    logger,
    flow_env: FlowEnv,
    task_table: dict[str, Any],
    dpr_input: DprProcessIn,
    retry_config: RetryConfig,
) -> tuple[PrefectFuture[PayloadSchema], dict[str, list[Item | str]]]:  # (payload, source_items)
    """Build and generate the payload file and the lineage source items"""

    # Validate tasktable contents against json schema
    with open(TASK_TABLE_SCHEMA, encoding="utf-8") as opened:
        schema = json.load(opened)
    jsonschema.validate(instance=task_table, schema=schema)

    # Dictionary of external variables values from flow's input
    external_variables = {
        "start_datetime": dpr_input.start_datetime,
        "end_datetime": dpr_input.end_datetime,
        "reference_date": dpr_input.reference_date,
        "instrument_mode": dpr_input.instrument_mode,
        "satellite": dpr_input.satellite,
    }

    # Replace configured external variables from the task table with the input parameters
    task_table = replace_external_variables(task_table, external_variables)

    processing_mode = list(dpr_input.processing_mode) if dpr_input.processing_mode else None
    unit_list = build_unit_list(
        tasktable=task_table,
        pipeline=dpr_input.pipeline,
        unit=dpr_input.unit,
        processing_mode=processing_mode,
    )

    tasks = []
    for unit in unit_list:
        # For each input_adfs element computed on STEP 1
        for input_adfs in unit["input_adfs"]:
            # HACK for S1-ARD: REFERENCE_DB is not staged, see below
            if input_adfs["name"] == "REFERENCE_DB":
                continue
            # For each specific input in case of multiplicity=one_per_input
            specific_input_name, product_stac_items = resolve_specific_input_product_stac_items(
                input_adfs,
                task_table,
                unit,
                dpr_input.input_products,
                flow_env.rs_client,
            )
            for specific_input_product_stac_item in product_stac_items:
                if specific_input_product_stac_item:
                    logger.info(
                        f"Submitting {input_adfs['name']} ADFS task for input {specific_input_product_stac_item}",
                    )
                tasks.append(
                    process_input_adfs.submit(
                        input_adfs,
                        dpr_input,
                        task_table,
                        (specific_input_name, specific_input_product_stac_item),
                        retry_config.staging_retries,
                        retry_config.staging_retry_delay,
                    ),
                )

    try:
        aux_items: list[tuple[str, str, tuple[bool, ItemCollection]]] = [
            result for t in tasks if (result := t.result()) is not None
        ]
    except (RuntimeError, KeyError) as err:
        raise err

    # A lineage source is either a staged ADF item or an input STAC self link.
    source_items: dict[str, list[Item | str]] = {}

    # Set of ADFS. Each tuple includes the adfs name, type and the s3/https storage path
    adfs: set[tuple[str, str, str]] = set()
    for name, adf_type, (status, item_collection) in aux_items:
        for item in item_collection.items:
            source_items.setdefault(name, []).append(item)

            if status:
                asset = next(iter(item.assets.values()))
                logger.info(f"ADFS '{name}' of type '{adf_type}': {asset.href}")
                adfs.add((name, adf_type, asset.href))
            else:
                raise ValueError(f"The adf input files {next(iter(item.assets.values()))} was not correctly staged")

    # HACK for S1-ARD: the reference pipeline writes the REFERENCE_DB in a new s3 folder
    if dpr_input.pipeline == "ARD_REFERENCE_PIPELINE":
        logger.warning(f"ARD HACK FOR PIPELINE {dpr_input.pipeline}")
        owner_id: str = dpr_input.env.owner_id
        output_collection = "s1-ard-reference-db"
        adf_type = "ARD_REFERENCE_DB"
        bucket_configuration = fetch_csv_from_endpoint(os.environ["RSPY_HOST_OSAM"] + "/internal/configuration")
        bucket_name = find_s3_output_bucket(bucket_configuration, owner_id, output_collection, adf_type)
        ref_db_path = os.path.join("s3://", bucket_name, owner_id, output_collection, str(uuid4()))
        logger.warning(f"ARD REFERENCE_DB: {ref_db_path}")
        # adf_type is not 'folder' on purpose, so that the path is not truncated by build_adfs
        adfs.add(("REFERENCE_DB", adf_type, ref_db_path))

    # generate the dpr payload file
    return (
        generate_payload.submit(
            flow_env,
            unit_list,
            list(adfs),
            dpr_input,
            external_modules=task_table.get("external_modules"),
        ),
        source_items,
    )

build_dask_dashboard_url_message(cluster_instance)

Build the Dask dashboard log message from the configured public gateway endpoint.

Source code in docs/rs-client-libraries/rs_workflows/on_demand_processing.py

def build_dask_dashboard_url_message(cluster_instance: str | None) -> str:
    """Build the Dask dashboard log message from the configured public gateway endpoint."""
    public_base = os.getenv("DASK_GATEWAY_PUBLIC", "")

    if not public_base or not cluster_instance:
        return "Dask cluster dashboard URL is unavailable"

    dashboard_url = f"{public_base.rstrip('/')}/clusters/{cluster_instance}/status"
    return f"Dask cluster dashboard URL: {dashboard_url}"

dpr_processing(dpr_input, retry_config=RetryConfig()) async

Prefect flow for dpr-process.

Parameters:

Name	Type	Description	Default
dpr_input	DprProcessIn	Input parameters for executing this flow	required
retry_config	RetryConfig	Staging retry config	RetryConfig()

Source code in docs/rs-client-libraries/rs_workflows/on_demand_processing.py

@flow(name="process-generic")
async def dpr_processing(
    dpr_input: DprProcessIn,
    retry_config: RetryConfig = RetryConfig(),  # type: ignore
) -> list[dict[str, Any]]:
    """
    Prefect flow for dpr-process.

    Args:
        dpr_input: Input parameters for executing this flow
        retry_config: Staging retry config
    """
    logger = get_logger()
    logger.info(f"Starting the DPR processing flow with processor: {dpr_input.processor_name}")
    # Init flow environment and opentelemetry span
    flow_env = FlowEnv(dpr_input.env)

    with flow_env.start_span(__name__, "dpr-processing"):

        cluster_info = ClusterInfo(
            jupyter_token=os.environ["JUPYTERHUB_API_TOKEN"],
            dask_gateway_address=os.environ["DASK_GATEWAY_ADDRESS"],
            cluster_label=dpr_input.dask_cluster_label,
            cluster_instance=dpr_input.dask_cluster_instance or "",
        )

        # read tasktable and construct list of processing units
        task_table: dict[str, Any] = flow_env.rs_client.get_dpr_client().get_process(
            dpr_input.processor_name,
            cluster_info,
        )

        # Persist the full task table as a Prefect artifact for later investigation.
        md = "# Task table\n\n```json\n" + json.dumps(task_table, indent=2) + "\n```"
        artifact_key_name: str = "dpr-task-table"
        await acreate_markdown_artifact(key=artifact_key_name, markdown=md, description="DPR task table")
        logger.info(f"📌 Artifact named '{artifact_key_name}' has been linked to this flow.")
        # Log the public Dask dashboard URL when the flow input provides the cluster instance.
        logger.info(build_dask_dashboard_url_message(cluster_info.cluster_instance))

        # Build and generate the payload file
        payload_task, source_items = build_and_generate_payload(
            logger,
            flow_env,
            task_table,
            dpr_input,
            retry_config,
        )
        generated_payload_res = payload_task.result()

        # Build lineage from the exact workflow that will be executed.
        lineage = build_output_lineage(generated_payload_res)
        # Reuse links resolved during payload generation; do not query the catalog again.
        if generated_payload_res.io:
            for input_product in generated_payload_res.io.input_products:
                source_items.setdefault(input_product.id, []).extend(input_product.source_item_hrefs)
        # create the generated payload as a dictionary, as it will be used for
        # the prefect artifact. the SecretStr will be masked here
        generated_payload_res_as_dict = generated_payload_res.dump()
        # create the YAML string first (synchronous). This will be used for writing both the artifact as well
        # as the tmp file
        # md = "# Payload file\n\n```json\n" + json.dumps(generated_payload_res_as_dict, indent=2) + "\n```"
        yaml_str = yaml.dump(generated_payload_res_as_dict, default_flow_style=False, sort_keys=False)
        # Write the payload as prefect artifact
        pretty_markdown = f"```yaml\n{yaml_str}\n```"
        artifact_key_name = "dpr-payload"
        await acreate_markdown_artifact(
            key=artifact_key_name,
            markdown=pretty_markdown,
            description="DPR Payload file",
        )
        logger.info(f"📌 Artifact named '{artifact_key_name}' has been linked to this flow.")

        # re-create the generated payload as a dictionary, as it will be used for
        # the payload file to upload to S3. here, the secrets are revealed
        generated_payload_res_with_secrets = generated_payload_res.dump(reveal_secrets=True)
        yaml_str = yaml.dump(generated_payload_res_with_secrets, default_flow_style=False, sort_keys=False)
        # upload the config payload contents to S3 or local file, without a temporary file
        logger.info(f"Writing the payload to file :\n {dpr_input.s3_payload_file}")
        await prefect_utils.upload_payload_bytes(yaml_str.encode("utf-8"), dpr_input.s3_payload_file)

        # Run the DPR processor
        processed_items = run_processor.submit(
            flow_env.serialize(),
            dpr_input.processor_name,
            generated_payload_res,
            cluster_info,
            dpr_input.s3_payload_file,
            dpr_input.input_products,
            wait_for=[payload_task],
        )
        try:
            processed_items.result()
        finally:
            logger.info("Deleting payload file")
            prefect_utils.delete_payload_file(dpr_input.s3_payload_file)

        processed = processed_items.result()

        # add derived_from links
        for processed_item in processed:
            output_id = processed_item.output_product_id

            if output_id not in lineage:
                raise ValueError(f"No payload lineage found for processed output '{output_id}'")

            added_hrefs: set[str] = set()
            for logical_source in sorted(lineage[output_id]):
                resolved_sources = source_items.get(logical_source, [])
                if not resolved_sources:
                    logger.warning("Skip lineage source '%s' because no STAC item was resolved", logical_source)
                    continue

                for source_item in resolved_sources:
                    source_href = source_item if isinstance(source_item, str) else source_item.get_self_href()

                    if not source_href:
                        logger.warning("Skip lineage source '%s' because it has no STAC self link", logical_source)
                        continue
                    parsed_href = urlsplit(source_href)
                    source_href = urlunsplit(("", "", parsed_href.path, parsed_href.query, parsed_href.fragment))
                    processed_item.stac_item.add_derived_from(source_href)
                    added_hrefs.add(source_href)

        # Publish processed items to the catalog
        published = catalog_flow.publish.submit(
            flow_env.serialize(),
            dpr_input.generated_product_to_collection_identifier,
            processed_items,
        )

        # Wait for last task to end.
        # NOTE: use .result() and not .wait() to unwrap and propagate exceptions, if any.
        published_items = published.result()  # type: ignore[unused-coroutine]
        return published_items

process_input_adfs(input_adfs, dpr_input, task_table, specific_input_product=(None, None), staging_retries=3, staging_retry_delay=60) async

Stage the ADFS inputs described in the task table for one processing unit input.

The task iterates through the ordered alternatives defined for a given ADFS input and stops at the first alternative that produces staged items.

For each alternative, the task: - builds the final AUX CQL2 request from the task table definition - resolves the target AUX collection identifier - runs AUX staging with retries - normalizes archived outputs when the staged assets still point to compressed archives - updates the catalog entries after normalization so downstream payload generation sees the final asset hrefs

Returns:

Type	Description
tuple[str, str, tuple[bool, ItemCollection]] | None	tuple[str, str, tuple[bool, ItemCollection]]: The input ADFS name and type together with the original staging status/item collection tuple shape expected by downstream code, or None if no alternative returned staged data for a non-mandatory ADFS.

Raises:

Type	Description
RuntimeError	If no alternative returns staged data for a mandatory ADFS, or if the task table content cannot be read as expected.

Source code in docs/rs-client-libraries/rs_workflows/on_demand_processing.py

@task(name="Process input ADFS")
async def process_input_adfs(
    input_adfs: dict[str, Any],
    dpr_input: DprProcessIn,
    task_table: dict[str, Any],
    specific_input_product: tuple[str | None, Item | None] = (None, None),
    staging_retries: int = 3,
    staging_retry_delay: int = 60,
) -> tuple[str, str, tuple[bool, ItemCollection]] | None:
    """
    Stage the ADFS inputs described in the task table for one processing unit input.

    The task iterates through the ordered alternatives defined for a given ADFS
    input and stops at the first alternative that produces staged items.

    For each alternative, the task:
    - builds the final AUX CQL2 request from the task table definition
    - resolves the target AUX collection identifier
    - runs AUX staging with retries
    - normalizes archived outputs when the staged assets still point to
      compressed archives
    - updates the catalog entries after normalization so downstream payload
      generation sees the final asset hrefs

    Returns:
        tuple[str, str, tuple[bool, ItemCollection]]:
            The input ADFS name and type together with the original staging status/item
            collection tuple shape expected by downstream code, or None if no alternative returned
            staged data for a non-mandatory ADFS.

    Raises:
        RuntimeError:
            If no alternative returns staged data for a mandatory ADFS, or if the task table content
            cannot be read as expected.
    """
    logger = get_logger()
    logger.info(f"🚧 Starting processing input ADFS for {input_adfs}")
    try:
        # For each "alternative" ( get it following the "order" )
        for alternative in input_adfs.get("alternatives", []):
            result = await _stage_input_adfs_alternative(
                alternative,
                input_adfs,
                dpr_input,
                task_table,
                specific_input_product,
                staging_retries,
                staging_retry_delay,
            )
            if result is not None:
                return result

        if not input_adfs.get("mandatory", True):
            logger.warning(f"⚠️ Searching for optional adfs input {input_adfs['name']} did not return any result")
            return None
        raise RuntimeError(f"Searching for adfs input {input_adfs['name']} did not return any result")

    except KeyError as kerr:
        raise RuntimeError(
            f"Unable to read / process tasktable and build cql2-json for: {json.dumps(input_adfs)}",
        ) from kerr

render_aux_cql2(alternative, task_table, specific_input_product=(None, None))

Render the STAC/CQL2 request of one ADFS alternative from the task table queries.

Placeholders such as {input_name.property} are resolved from the properties of the STAC item of the specific input product (ADFS multiplicity 'one_per_input').

Returns:

Type	Description
tuple[dict, dict[str, Any]]	tuple[dict, dict]: the rendered STAC/CQL2 request and the resolved query parameters.

Source code in docs/rs-client-libraries/rs_workflows/on_demand_processing.py

def render_aux_cql2(
    alternative: dict[str, Any],
    task_table: dict[str, Any],
    specific_input_product: tuple[str | None, Item | None] = (None, None),
) -> tuple[dict, dict[str, Any]]:
    """
    Render the STAC/CQL2 request of one ADFS alternative from the task table queries.

    Placeholders such as ``{input_name.property}`` are resolved from the properties of the
    STAC item of the specific input product (ADFS multiplicity 'one_per_input').

    Returns:
        tuple[dict, dict]: the rendered STAC/CQL2 request and the resolved query parameters.
    """
    name = alternative["query"]["name"]
    specific_input_name, stac_item = specific_input_product
    parameters = {
        k: (
            stac_item.properties.get(m.group(2), v)
            if stac_item
            and isinstance(v, str)
            and (m := SPECIFIC_INPUT_PATTERN.match(v))
            and m.group(1) == specific_input_name
            else v
        )
        for k, v in deepcopy(alternative["query"]["parameters"]).items()
    }
    query = next(q for q in task_table["queries"] if q["name"] == name)
    return build_cql2_json(query, parameters), parameters

resolve_specific_input_product_stac_items(input_adfs, task_table, unit, provided_input_products, rs_client)

Resolve the STAC items of the input products an ADFS depends on, for the ADFS multiplicity 'one_per_input'.

Returns:

Type	Description
tuple[str, list[Item]] | tuple[None, list[None]]	The name of the referenced input product and the STAC items of the provided input products matching
tuple[str, list[Item]] | tuple[None, list[None]]	its regex. If the ADFS multiplicity is not 'one_per_input', (None, [None]) is returned so the ADFS
tuple[str, list[Item]] | tuple[None, list[None]]	is staged once.

Source code in docs/rs-client-libraries/rs_workflows/on_demand_processing.py

def resolve_specific_input_product_stac_items(
    input_adfs: dict[str, Any],
    task_table: dict[str, Any],
    unit: dict[str, Any],
    provided_input_products: list[FlowInputProduct],
    rs_client: RsClient,
) -> tuple[str, list[Item]] | tuple[None, list[None]]:
    """
    Resolve the STAC items of the input products an ADFS depends on, for the ADFS multiplicity 'one_per_input'.

    Returns:
        The name of the referenced input product and the STAC items of the provided input products matching
        its regex. If the ADFS multiplicity is not 'one_per_input', (None, [None]) is returned so the ADFS
        is staged once.
    """
    input_adfs_io = search_by_name(get_io_entries(task_table, "adfs"), input_adfs["name"])
    if input_adfs_io.get("folder_multiplicity", None) == "one_per_input":
        logger = get_logger()
        input_product_names: set[str] = {product["name"] for product in unit["input_products"]}
        referenced_input_product_names = {
            match.group(1)
            for alt in input_adfs_io.get("alternatives", [])
            for value in alt.get("query", {}).get("parameters", {}).values()
            if isinstance(value, str)
            for match in [re.fullmatch(r"\{([^\.}]+)\.[^}]+\}", value)]
            if match and match.group(1) in input_product_names
        }
        if len(referenced_input_product_names) != 1:
            raise ValueError(
                f"ADFS multiplicy 'one_per_input' shall refer to a single input product, "
                f"found '{referenced_input_product_names}'",
            )
        referenced_input_product_name = next(iter(referenced_input_product_names))
        logger.info(f"ADFS multiplicity 'one_per_input' refers to input '{referenced_input_product_name}'")
        input_product_io: dict[str, Any] = search_by_name(
            get_io_entries(task_table, "input"),
            referenced_input_product_name,
        )
        input_product_regex: str = input_product_io.get("reader_params", {}).get("regex", None)
        if not input_product_regex:
            logger.warning(
                f"⚠️ Input product '{referenced_input_product_name}' should define a regex "
                "to perform reliable discrimination",
            )
        result = []
        catalog_client: CatalogClient = rs_client.get_catalog_client()
        for input_product in provided_input_products:
            stac_item, first_asset_path = resolve_stac_input_path(
                catalog_client,
                input_product.collection_name,
                input_product.item_id,
            )
            if not first_asset_path:
                continue
            # Check that first asset path matches input product regex
            if input_product_regex:
                logger.info(f"Checking asset path '{first_asset_path}' against regex '{input_product_regex}'")
                if not re.fullmatch(input_product_regex, first_asset_path):
                    logger.debug(f"Ignore path {first_asset_path} as it doesn't match regex '{input_product_regex}'")
                    continue
            else:
                logger.warning(f"Adding asset '{first_asset_path}' without regex validation")
            result.append(stac_item)
        if not result:
            logger.error(f"❌ Found no STAC item with assets matching the regex '{input_product_regex}'")
        return referenced_input_product_name, result
    return None, [None]

select_aux_collection_and_source(dpr_input, product_type)

Resolve the catalog collection identifier and STAC source for a requested AUX product type.

The selection logic mirrors the previous inline implementation used by process_input_adfs: - prefer the collection explicitly mapped to the requested product_type - otherwise fall back to the wildcard mapping (*) when present - otherwise derive the default convention-based collection name from the satellite and product type

Keeping this lookup in a dedicated helper makes the fallback chain easier to read and test independently from the staging flow logic.

Source code in docs/rs-client-libraries/rs_workflows/on_demand_processing.py

def select_aux_collection_and_source(
    dpr_input: DprProcessIn,
    product_type: str,
) -> tuple[str, AuxiliarySource, list[str] | None]:
    """
    Resolve the catalog collection identifier and STAC source for a requested AUX product type.

    The selection logic mirrors the previous inline implementation used by
    ``process_input_adfs``:
    - prefer the collection explicitly mapped to the requested ``product_type``
    - otherwise fall back to the wildcard mapping (``*``) when present
    - otherwise derive the default convention-based collection name from the
      satellite and product type

    Keeping this lookup in a dedicated helper makes the fallback chain easier to
    read and test independently from the staging flow logic.
    """
    default_aux_collection = f"{dpr_input.satellite}-aux-{product_type}"
    return next(
        (
            (p.collection_name, p.source, p.selected_assets)
            for p in dpr_input.auxiliary_product_to_collection_identifier
            if p.product_type == product_type
        ),
        next(
            (
                (p.collection_name, p.source, p.selected_assets)
                for p in dpr_input.auxiliary_product_to_collection_identifier
                if p.product_type == "*"
            ),
            (
                default_aux_collection,
                AuxiliarySource.AUXIP,
                None,
            ),
        ),
    )