CWL CommandLineTool for stage-out

The CWL tool invokes the installed stage-out <catalog-directory> <bucket> <prefix> command in localhost/stage-out:latest. Existing job inputs and credential environment variable names are preserved.

cwlVersion: v1.2
$namespaces:
  s: https://schema.org/
# Training-example authors and creation date mirror codemeta.json.
s:name: Stage-out STAC results
s:description: Upload the results catalog and its assets to S3-compatible storage.
s:dateCreated: '2022-09-01'
s:license:
  s:name: Creative Commons Attribution-ShareAlike 4.0 International
  s:url: https://creativecommons.org/licenses/by-sa/4.0/
  s:identifier: CC-BY-SA-4.0
s:softwareHelp:
  s:name: Mastering Earth Observation Application Packaging with CWL
  s:url: https://eoap.github.io/mastering-app-package
s:publisher:
  s:name: EO Application Packaging
  s:identifier: https://github.com/eoap
s:author:
  - s:givenName: Jane
    s:familyName: Doe
    s:email: jane.doe@acme.earth
    s:affiliation:
      s:name: ACME
  - s:givenName: John
    s:familyName: Doe
    s:email: john.doe@acme.earth
    s:affiliation:
      s:name: ACME
s:softwareRequirements:
  - container runtime
  - cwl runner
s:softwareVersion: 2.0.2


class: CommandLineTool
id: stage-out

doc: "Stage-out the results to S3"
inputs:
  s3_bucket:
    type: string
    label: S3 bucket
    doc: Name of the destination S3 bucket.
  sub_path:
    type: string
    label: Object prefix
    doc: Object key prefix under which results are uploaded.
  aws_access_key_id:
    type: string
    label: S3 access key ID
    doc: Access key ID used to authenticate to the S3-compatible service.
  aws_secret_access_key:
    type: string
    label: S3 secret access key
    doc: Secret access key used to authenticate to the S3-compatible service.
  region_name:
    type: string
    label: S3 region
    doc: Region name used for S3 requests.
  endpoint_url:
    type: string
    label: S3 endpoint
    doc: URL of the S3-compatible service endpoint.
  stac_catalog:
    type: Directory
    label: Results STAC catalog
    doc: Directory containing the results STAC catalog and its referenced raster assets.
outputs:
  s3_catalog_output:
    outputBinding:
      outputEval: ${  return "s3://" + inputs.s3_bucket + "/" + inputs.sub_path + "/catalog.json"; }
    type: string
    label: Uploaded STAC catalog
    doc: S3 URI of the uploaded results catalog.
baseCommand: stage-out
arguments:
  - $( inputs.stac_catalog.path )
  - $( inputs.s3_bucket )
  - $( inputs.sub_path )
requirements:
  DockerRequirement:
    dockerPull: localhost/stage-out:latest
  InlineJavascriptRequirement: {}
  NetworkAccess:
    networkAccess: true
  EnvVarRequirement:
    envDef:
      aws_access_key_id: $( inputs.aws_access_key_id )
      aws_secret_access_key: $( inputs.aws_secret_access_key )
      aws_region_name: $( inputs.region_name )
      aws_endpoint_url: $( inputs.endpoint_url )
      PATH: /app/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
  ResourceRequirement: {}

The package copies the catalog into a unique temporary directory, uploads its assets, rewrites their links to S3, and publishes the catalog and item documents. It traverses nested catalogs recursively and leaves the input directory intact. Upload failures propagate; the command prints the resulting catalog URI to stdout.

Set aws_access_key_id, aws_secret_access_key, aws_region_name, and aws_endpoint_url through the CWL job inputs. The tool passes them as environment variables to boto3.

"""Stage a local STAC catalog and its assets to S3."""

from __future__ import annotations

import os
import posixpath
import shutil
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Any
from urllib.parse import unquote, urlparse

import boto3
import pystac
from loguru import logger
from pystac.stac_io import DefaultStacIO


class S3StacIO(DefaultStacIO):
    """Write STAC documents using the configured S3 client."""

    def __init__(self, client: Any) -> None:
        super().__init__()
        self.client = client

    def write_text(self, dest: Any, txt: str, *args: Any, **kwargs: Any) -> None:
        parsed = urlparse(str(dest))
        if parsed.scheme == "s3":
            self.client.put_object(
                Body=txt.encode("UTF-8"),
                Bucket=parsed.netloc,
                Key=parsed.path.lstrip("/"),
                ContentType="application/geo+json",
            )
        else:
            super().write_text(dest, txt, *args, **kwargs)


def asset_key(href: str, item_id: str, subfolder: str) -> str:
    """Keep relative file paths within the item's upload prefix."""
    parsed = urlparse(href)
    relative = (
        posixpath.basename(parsed.path)
        if parsed.scheme or posixpath.isabs(parsed.path)
        else parsed.path
    )
    relative = posixpath.normpath(relative)
    if relative == ".." or relative.startswith("../"):
        relative = posixpath.basename(relative)
    return posixpath.normpath(posixpath.join(subfolder, item_id, relative))


def local_asset_path(asset: pystac.Asset) -> Path:
    """Resolve filesystem paths and decoded local file URIs."""
    location = urlparse(asset.get_absolute_href())
    if location.scheme not in ("", "file") or location.netloc not in ("", "localhost"):
        raise ValueError("Stage-out requires local filesystem assets")
    return Path(unquote(location.path))


def upload_directory(local: Path, client: Any, bucket: str, prefix: str) -> str:
    """Upload complete stores even when the asset references a subgroup."""
    store = next(
        (
            part
            for part in (local, *local.parents)
            if part.name.endswith(".zarr")
            and any((part / marker).is_file() for marker in ("zarr.json", ".zgroup"))
        ),
        local,
    )
    store_key = posixpath.join(prefix, store.name)
    files = sorted(path for path in store.rglob("*") if path.is_file())
    if not files:
        raise ValueError(f"Empty directory asset: {store}")
    for path in files:
        client.upload_file(
            str(path),
            bucket,
            posixpath.join(store_key, path.relative_to(store).as_posix()),
        )
    suffix = local.relative_to(store).as_posix()
    return store_key if suffix == "." else posixpath.join(store_key, suffix)


def upload_assets(item: pystac.Item, client: Any, bucket: str, subfolder: str) -> None:
    """Upload assets before rewriting their references."""
    if Path(item.id).name != item.id or item.id in {"", ".", ".."}:
        raise ValueError("STAC item id must be a single directory name")
    for asset in item.assets.values():
        key = asset_key(asset.href, item.id, subfolder)
        local = local_asset_path(asset)
        logger.info("Uploading asset to s3://{}/{}", bucket, key)
        if local.is_dir():
            key = upload_directory(
                local, client, bucket, posixpath.join(subfolder, item.id)
            )
        else:
            client.upload_file(str(local), bucket, key)
        asset.href = f"s3://{bucket}/{key}"


def upload_catalog(
    catalog: pystac.Catalog, client: Any, bucket: str, subfolder: str
) -> str:
    """Upload assets and publish the rewritten catalog documents."""
    catalog.set_root(catalog)
    items = list(catalog.get_items(recursive=True))
    for item in items:
        upload_assets(item, client, bucket, subfolder)
    destination = f"s3://{bucket}/{subfolder}"
    catalog.normalize_hrefs(destination)
    stac_io = S3StacIO(client)
    catalog.save(catalog_type=pystac.CatalogType.SELF_CONTAINED, stac_io=stac_io)
    logger.info("Published catalog to {}/catalog.json", destination)
    return f"{destination}/catalog.json"


def execute(catalog: Path, bucket: str, subfolder: str) -> str:
    """Copy the input catalog, upload it, and return its S3 URL."""
    client = boto3.client(
        "s3",
        aws_access_key_id=os.environ["aws_access_key_id"],
        aws_secret_access_key=os.environ["aws_secret_access_key"],
        region_name=os.environ["aws_region_name"],
        endpoint_url=os.environ["aws_endpoint_url"],
    )
    with TemporaryDirectory(prefix="stage-out-") as directory:
        copied = Path(directory) / "catalog"
        shutil.copytree(catalog, copied)
        loaded = pystac.read_file(str(copied / "catalog.json"))
        if not isinstance(loaded, pystac.Catalog):
            raise ValueError("The input must contain a STAC catalog or collection")
        return upload_catalog(loaded, client, bucket, subfolder)

Build the staging images:

export WORKSPACE=/workspace/mastering-app-package

podman \
    build \
    --format docker \
    -t localhost/stage:latest \
    ${WORKSPACE}/water-bodies/command-line-tools/stage

# Current checkout uses separate installed staging commands.
podman build --format docker -t localhost/stage-in:latest ${WORKSPACE}/water-bodies/command-line-tools/stage-in
podman build --format docker -t localhost/stage-out:latest ${WORKSPACE}/water-bodies/command-line-tools/stage-out

Run stage-out:

cwltool \
    --podman \
    ${WORKSPACE}/cwl-cli/stage-out.cwl \
    --aws_access_key_id $( cat ~/.aws/credentials | grep aws_access_key_id | cut -d "=" -f 2 ) \
    --aws_secret_access_key $( cat ~/.aws/credentials | grep aws_secret_access_key | cut -d "=" -f 2 ) \
    --endpoint_url $( cat ~/.aws/config | grep endpoint_url | head -n 1 | cut -d "=" -f 2 ) \
    --region_name $( cat ~/.aws/config | grep region | cut -d "=" -f 2 ) \
    --s3_bucket processing-results \
    --sub_path processing-results/$( cat /proc/sys/kernel/random/uuid ) \
    --stac_catalog ${WORKSPACE}/runs