Release: develop -> main (maap-py v5, optional File/Directory inputs, PGT token hardening)

.github/workflows/docker-build.yml

@@ -14,12 +14,26 @@ on:
         type: boolean
 
 jobs:
+  test:
+    runs-on: ubuntu-24.04
+
+    steps:
+    - uses: actions/checkout@v7
+    - uses: actions/setup-python@v7
+      with:
+        python-version: '3.12'
+    - name: Install dependencies
+      run: pip install pytest cwltool==3.1.20250110105449 cwl_utils==0.35 PyYAML pystac==1.15.2 maap-py requests "urllib3>=2"
+    - name: Run tests
+      run: pytest tests
+
   build_and_push:
+    needs: test
 
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
 
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v7
     - name: Set repo name and tag
       id: vars
       run: |
@@ -42,24 +56,24 @@ jobs:
         echo "tag=$TAG_NAME" >> "$GITHUB_OUTPUT"
 
     - name: Log in to GitHub Container Registry
-      uses: docker/login-action@v2
+      uses: docker/login-action@v4
       with:
         registry: ghcr.io
         username: ${{ github.actor }}
         password: ${{ secrets.GITHUB_TOKEN }}
 
     - name: Log in to Private Registry
-      uses: docker/login-action@v2
+      uses: docker/login-action@v4
       with:
         registry: ${{ secrets.PRIVATE_REGISTRY_URL }}
         username: ${{ secrets.PRIVATE_REGISTRY_USERNAME }}
         password: ${{ secrets.PRIVATE_REGISTRY_PASSWORD }}
 
     - name: Set up Docker Buildx
-      uses: docker/setup-buildx-action@v3
+      uses: docker/setup-buildx-action@v4
 
     - name: Build and push
-      uses: docker/build-push-action@v6
+      uses: docker/build-push-action@v7
       with:
         platforms: linux/amd64
         push: true

create_inputs.py

@@ -1,5 +1,6 @@
 import json
 import pathlib
+import time
 
 import yaml
 import os
@@ -93,6 +94,9 @@ def convert_value_to_cwl_type(value, cwl_type_str):
         return value
 
 
+HTTP_RETRIES = 4
+
+
 def load_json_from_uri(uri):
     parsed = urllib.parse.urlparse(uri)
 
@@ -110,17 +114,20 @@ def load_json_from_uri(uri):
             print(f"Failed to parse JSON: {e}")
             return None
 
-    # http / https
-    try:
-        response = requests.get(uri)
-        response.raise_for_status()
-        return response.json()
-    except requests.exceptions.RequestException as e:
-        print(f"Request error: {e}")
-        return None
-    except ValueError as e:
-        print(f"Failed to parse JSON: {e}")
-        return None
+    # http / https: retry transient failures (e.g. CMR STAC 503s)
+    for attempt in range(1, HTTP_RETRIES + 1):
+        try:
+            response = requests.get(uri, timeout=60)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.RequestException as e:
+            print(f"Request error (attempt {attempt}/{HTTP_RETRIES}): {e}")
+            if attempt < HTTP_RETRIES:
+                time.sleep(2 ** attempt)
+        except ValueError as e:
+            print(f"Failed to parse JSON: {e}")
+            return None
+    return None
 
 
 def parse_cwl_inputs(cwl_file_path) -> [CommandInputParameter]:
@@ -195,14 +202,26 @@ def map_inputs_from_context(job_inputs: dict, cwl_inputs: [CommandInputParameter
         input_name = get_id_from_uri(input.id)
         base_type, is_optional = parse_cwl_type(input_type)
 
+        if base_type in ("Directory", "File"):
+            raw_value = job_inputs.get(input_name)
+            if raw_value is None or raw_value == "None" or raw_value == "":
+                if is_optional:
+                    # Omit so cwltool does not pass the flag to the algorithm
+                    print(f"No value provided for optional {base_type} input '{input_name}', skipping.")
+                    continue
+                raise ValueError(f"No value provided for required {base_type} input '{input_name}'.")
+
         if base_type == "Directory":
             print("Encountered STAC input, localizing assets.")
             # Load STAC JSON directly from the URI rather than using the HySDS-localized copy.
             # HySDS caches localized files by URI path — if the JSON content changes at the same
             # URI, HySDS serves the stale cached version. Fetching from the URI at runtime
             # guarantees we always get the current content.
             # TODO: Update the HySDS spec builder to not map this input as a localize field.
-            stac_json = load_json_from_uri(job_inputs.get(input_name))
+            stac_uri = job_inputs.get(input_name)
+            stac_json = load_json_from_uri(stac_uri)
+            if stac_json is None:
+                raise RuntimeError(f"Could not load STAC JSON for input '{input_name}' from {stac_uri}")
             output_dir = str(pathlib.Path(input_name).resolve())
             stage_in.download_stac_assets_from_json(stac_json, output_dir)
             workflow_inputs[input_name] = {"class": "Directory", "path": output_dir}

docker/Dockerfile

@@ -21,9 +21,10 @@ RUN pip install --no-cache-dir \
     cwl_utils==0.35 \
     PyYAML \
     awscli \
-    pystac \
-    maap-py \
-    requests
+    pystac==1.15.2 \
+    maap-py==5.1.0 \
+    requests \
+    "urllib3>=2"
 
 # Copy only what you need
 COPY docker_stats/docker-stats-on-exit-shim /docker-stats-on-exit-shim

dps_wrapper.sh

@@ -4,7 +4,7 @@ USERNAME=$(python /app/get_username.py)
 source /maap-dps.env
 export MAAP_API_HOST
 export DPS_MACHINE_TOKEN
-MAAP_PGT=$(python /app/get_maap_pgt_token.py)
+MAAP_PGT=$(python /app/get_maap_pgt_token.py) || exit $?
 export MAAP_PGT
 unset DPS_MACHINE_TOKEN
 

environment.yaml

@@ -3,5 +3,5 @@ dependencies:
   - cwltool==3.1.20250110105449
   - cwl_utils==0.35
   - nodejs
-  - pystac
-  - maap-py
+  - pystac==1.15.2
+  - maap-py==5.1.0

get_maap_pgt_token.py

@@ -1,24 +1,104 @@
 import json
 import os
+import sys
+
 import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry  # backoff_jitter requires urllib3>=2
+
+TIMEOUT = (5, 30)  # (connect, read) seconds
+MAX_RETRIES = 5  # i.e. up to 6 attempts
+BACKOFF_FACTOR = 1.0  # sleeps ~ factor * 2**(n-1) seconds between retries, plus jitter
+BACKOFF_JITTER = 1.0
+BACKOFF_MAX = 30.0
+RETRY_STATUSES = tuple(range(500, 600))
+EXIT_CODE_PGT_FAILURE = 78  # distinct job exit status for PGT token failures
+
+
+class PgtTokenError(Exception):
+    """Raised when the MAAP PGT token cannot be obtained."""
 
 
-def get_maap_pgt_token(username, maap_api_host=os.environ.get("MAAP_API_HOST")):
+def _session():
+    """Session retrying connect errors, timeouts and 5xx with exponential backoff and jitter."""
+    retry = Retry(
+        total=MAX_RETRIES,
+        backoff_factor=BACKOFF_FACTOR,
+        backoff_jitter=BACKOFF_JITTER,
+        backoff_max=BACKOFF_MAX,
+        status_forcelist=RETRY_STATUSES,
+        raise_on_status=False,  # hand back the last response so we can raise a clear error
+    )
+    session = requests.Session()
+    adapter = HTTPAdapter(max_retries=retry)
+    session.mount("https://", adapter)
+    session.mount("http://", adapter)
+    return session
+
+
+def get_maap_pgt_token(username, maap_api_host=None):
+    maap_api_host = maap_api_host or os.environ.get("MAAP_API_HOST")
     url = f"https://{maap_api_host}/api/members/{username}"
     headers = {"dps-token": os.environ.get("DPS_MACHINE_TOKEN")}
-    response = requests.get(url, headers=headers)
-    maap_pgt_token = response.json()["session_key"]
-    return maap_pgt_token
+
+    try:
+        with _session() as session:
+            response = session.get(url, headers=headers, timeout=TIMEOUT)
+    except requests.RequestException as e:
+        raise PgtTokenError(
+            f"Could not get MAAP_PGT for user '{username}' from {url} "
+            f"after {MAX_RETRIES + 1} attempts; last error: {type(e).__name__}: {e}"
+        ) from e
+
+    if response.status_code >= 500:
+        raise PgtTokenError(
+            f"Could not get MAAP_PGT for user '{username}' from {url} "
+            f"after {MAX_RETRIES + 1} attempts; last error: HTTP {response.status_code}"
+        )
+
+    try:
+        response.raise_for_status()
+    except requests.HTTPError as e:
+        # 4xx is permanent (e.g. bad token, unknown user) and was not retried
+        raise PgtTokenError(
+            f"Request for user '{username}' to {url} failed: HTTP {response.status_code}"
+        ) from e
+
+    try:
+        token = response.json()["session_key"]
+    except (ValueError, KeyError, TypeError) as e:
+        raise PgtTokenError(
+            f"Response from {url} for user '{username}' has no usable 'session_key'"
+        ) from e
+    if not isinstance(token, str) or not token.strip():
+        raise PgtTokenError(
+            f"Response from {url} for user '{username}' has an empty 'session_key'"
+        )
+    return token
+
+
+def report_failure(error, job_dir="."):
+    """Surface the failure to HySDS.
+
+    stderr lands in _stderr.txt; _alt_error.txt overrides the job's `error`
+    field (job_worker reads it from the job dir). Keep it a short single line.
+    """
+    message = " ".join(str(error).split())
+    print(f"Unable to get user MAAP_PGT token: {message}", file=sys.stderr)
+    with open(os.path.join(job_dir, "_alt_error.txt"), "w") as f:
+        f.write(f"MAAP_PGT token failure: {message}\n")
 
 
 if __name__ == '__main__':
     with open("_job.json") as fr:
         job = json.loads(fr.read())
-        username = job.get("username")
-        if username:
-            try:
-                maap_pgt_token = get_maap_pgt_token(username)
-                print(maap_pgt_token)
-            except Exception as e:
-                print(f"Unable to get user MAAP_PGT token: {e}")
-                raise e
+    username = job.get("username")
+    try:
+        if not username:
+            # an empty MAAP_PGT would only surface later as a 401 in the user's code
+            raise PgtTokenError("no 'username' in _job.json, cannot request a MAAP_PGT")
+        print(get_maap_pgt_token(username))
+    except PgtTokenError as e:
+        # stdout is captured into MAAP_PGT by dps_wrapper.sh, so report on stderr
+        report_failure(e)
+        sys.exit(EXIT_CODE_PGT_FAILURE)

stage_in.py

@@ -7,19 +7,27 @@
 from datetime import datetime
 from maap.maap import MAAP
 
-maap = MAAP()
+_maap = None
+
+
+def _get_maap():
+    """Create the MAAP client on first use so jobs without file downloads make no API call."""
+    global _maap
+    if _maap is None:
+        _maap = MAAP()
+    return _maap
 
 
 def _download_asset(href, destination_path):
-    """Route asset download: MAAP S3 buckets use boto3, everything else uses maap.downloadGranule."""
+    """Route asset download: MAAP S3 buckets use boto3, everything else uses maap.download_granule."""
     if href.startswith("s3://maap-"):
         parts = href[5:].split("/", 1)
         bucket, key = parts[0], parts[1]
         filename = os.path.basename(key)
         dest = os.path.join(destination_path, filename)
         boto3.client("s3").download_file(bucket, key, dest)
         return dest
-    return maap.downloadGranule(
+    return _get_maap().download_granule(
         online_access_url=href,
         destination_path=destination_path,
         overwrite=True,