Merge pull request #2463 from ASFHyP3/develop

Release v8.0.0

.github/actions/deploy-hyp3/action.yml

@@ -30,7 +30,7 @@ inputs:
     description: "Comma separated list of Subnet IDs"
     required: true
   SECRET_ARN:
-    description: "ARN of Secret in AWS Secrets Manager containing all the credentials needed for job tasks"
+    description: "ARN of Secret in AWS Secrets Manager containing all the credentials needed for job steps"
     required: true
   CLOUDFORMATION_ROLE_ARN:
     description: "The CloudFormation role to use for this deployment"

.github/workflows/changelog.yml

@@ -13,4 +13,4 @@ on:
 
 jobs:
   call-changelog-check-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-changelog-check.yml@v0.11.2
+    uses: ASFHyP3/actions/.github/workflows/reusable-changelog-check.yml@v0.12.0

.github/workflows/create-jira-issue.yml

@@ -6,7 +6,7 @@ on:
 
 jobs:
   call-create-jira-issue-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-create-jira-issue.yml@v0.11.2
+    uses: ASFHyP3/actions/.github/workflows/reusable-create-jira-issue.yml@v0.12.0
     secrets:
       JIRA_BASE_URL: ${{ secrets.JIRA_BASE_URL }}
       JIRA_USER_EMAIL: ${{ secrets.JIRA_USER_EMAIL }}

.github/workflows/deploy-daac.yml

@@ -111,6 +111,6 @@ jobs:
   call-bump-version-workflow:
     if: github.ref == 'refs/heads/main'
     needs: deploy
-    uses: ASFHyP3/actions/.github/workflows/reusable-bump-version.yml@v0.11.2
+    uses: ASFHyP3/actions/.github/workflows/reusable-bump-version.yml@v0.12.0
     secrets:
       USER_TOKEN: ${{ secrets.TOOLS_BOT_PAK }}

.github/workflows/deploy-enterprise-test.yml

@@ -88,6 +88,7 @@ jobs:
             job_files: >-
               job_spec/INSAR_ISCE_BURST.yml
               job_spec/SRG_GSLC.yml
+              job_spec/SRG_TIME_SERIES.yml
             instance_types: r6id.xlarge,r6id.2xlarge,r6id.4xlarge,r6id.8xlarge,r6idn.xlarge,r6idn.2xlarge,r6idn.4xlarge,r6idn.8xlarge
             default_max_vcpus: 640
             expanded_max_vcpus: 640

.github/workflows/deploy-enterprise.yml

@@ -245,6 +245,7 @@ jobs:
             job_files: >-
               job_spec/INSAR_ISCE_BURST.yml
               job_spec/SRG_GSLC.yml
+              job_spec/SRG_TIME_SERIES.yml
             instance_types: r6id.xlarge,r6id.2xlarge,r6id.4xlarge,r6id.8xlarge,r6idn.xlarge,r6idn.2xlarge,r6idn.4xlarge,r6idn.8xlarge
             default_max_vcpus: 640
             expanded_max_vcpus: 640

.github/workflows/labeled-pr.yml

@@ -12,4 +12,4 @@ on:
 
 jobs:
   call-labeled-pr-check-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-labeled-pr-check.yml@v0.11.2
+    uses: ASFHyP3/actions/.github/workflows/reusable-labeled-pr-check.yml@v0.12.0

.github/workflows/release-template-comment.yml

@@ -7,7 +7,7 @@ on:
 
 jobs:
   call-release-checklist-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-release-checklist-comment.yml@v0.11.2
+    uses: ASFHyP3/actions/.github/workflows/reusable-release-checklist-comment.yml@v0.12.0
     permissions:
       pull-requests: write
     with:

.github/workflows/release.yml

@@ -7,7 +7,7 @@ on:
 
 jobs:
   call-release-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-release.yml@v0.11.2
+    uses: ASFHyP3/actions/.github/workflows/reusable-release.yml@v0.12.0
     with:
       release_prefix: HyP3
     secrets:

.github/workflows/static-analysis.yml

@@ -87,4 +87,4 @@ jobs:
           snyk iac test --severity-threshold=high
 
   call-secrets-analysis-workflow:
-    uses: ASFHyP3/actions/.github/workflows/reusable-secrets-analysis.yml@v0.11.2
+    uses: ASFHyP3/actions/.github/workflows/reusable-secrets-analysis.yml@v0.12.0

.gitignore

@@ -5,6 +5,7 @@ apps/api/src/hyp3_api/api-spec/job_parameters.yml
 apps/api/src/hyp3_api/job_validation_map.yml
 apps/step-function.json
 apps/**/*-cf.yml
+apps/start-execution-worker/src/batch_params_by_job_type.json
 lib/dynamo/dynamo/*.json
 lib/dynamo/dynamo/*.yml
 

CHANGELOG.md

@@ -4,6 +4,23 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [8.0.0]
+
+### Added
+- A job step can now be applied to every item in a list using a new `map: for <item> in <items>` syntax. For example, given a job spec with a `granules` parameter, a step that includes a `map: for granule in granules` field is applied to each item in the `granules` list and can refer to `Ref::granule` within its `command` field.
+- If a job contains a `map` step, the processing time value for that step (in the `processing_times` list in the job's API response) is a sub-list of processing times for the step's iterations, in the same order as the items in the input list.
+- A new `SRG_TIME_SERIES` job type has been added to the `hyp3-lavas` and `hyp3-lavas-test` deployments. This workflow uses the new `map` syntax described above to produce a GSLC for each level-0 Sentinel-1 granule passed via the `granules` parameter and then produces a time series product from the GSLCs. See the [HyP3 SRG](https://github.com/ASFHyP3/hyp3-srg) plugin.
+- The `SRG_GSLC` job type now includes parameter validation.
+
+### Changed
+- Changes to custom compute environments:
+  - Custom compute environments are now applied to individual job steps rather than to entire jobs. The `compute_environment` field is now provided at the step level rather than at the top level of the job spec.
+  - If the value of the `compute_environment` field is `Default`, then the step uses the deployment's default compute environment. Otherwise, the value must be the name of a custom compute environment defined in `job_spec/config/compute_environments.yml`.
+- Other changes to the job spec syntax:
+  - The `tasks` field has been renamed to `steps`.
+  - Job parameters no longer contain a top-level `default` field. The `default` field within each parameter's `api_schema` mapping is still supported.
+  - Job specs no longer explicitly define a `bucket_prefix` parameter. Instead, `bucket_prefix` is automatically defined and can still be referenced as `Ref::bucket_prefix` within each step's `command` field.
+
 ## [7.12.0]
 
 ### Changed

Makefile

@@ -1,4 +1,5 @@
 API = ${PWD}/apps/api/src
+APPS = ${PWD}/apps
 CHECK_PROCESSING_TIME = ${PWD}/apps/check-processing-time/src
 GET_FILES = ${PWD}/apps/get-files/src
 HANDLE_BATCH_EVENT = ${PWD}/apps/handle-batch-event/src
@@ -10,7 +11,7 @@ DISABLE_PRIVATE_DNS = ${PWD}/apps/disable-private-dns/src
 UPDATE_DB = ${PWD}/apps/update-db/src
 UPLOAD_LOG = ${PWD}/apps/upload-log/src
 DYNAMO = ${PWD}/lib/dynamo
-export PYTHONPATH = ${API}:${CHECK_PROCESSING_TIME}:${GET_FILES}:${HANDLE_BATCH_EVENT}:${SET_BATCH_OVERRIDES}:${SCALE_CLUSTER}:${START_EXECUTION_MANAGER}:${START_EXECUTION_WORKER}:${DISABLE_PRIVATE_DNS}:${UPDATE_DB}:${UPLOAD_LOG}:${DYNAMO}
+export PYTHONPATH = ${API}:${CHECK_PROCESSING_TIME}:${GET_FILES}:${HANDLE_BATCH_EVENT}:${SET_BATCH_OVERRIDES}:${SCALE_CLUSTER}:${START_EXECUTION_MANAGER}:${START_EXECUTION_WORKER}:${DISABLE_PRIVATE_DNS}:${UPDATE_DB}:${UPLOAD_LOG}:${DYNAMO}:${APPS}
 
 
 build: render
@@ -36,11 +37,12 @@ install:
 	python -m pip install -r requirements-all.txt
 
 files ?= job_spec/*.yml
+compute_env_file ?= job_spec/config/compute_environments.yml
 security_environment ?= ASF
 api_name ?= local
 cost_profile ?= DEFAULT
 render:
-	@echo rendering $(files) for API $(api_name) and security environment $(security_environment); python apps/render_cf.py -j $(files) -s $(security_environment) -n $(api_name) -c $(cost_profile)
+	@echo rendering $(files) for API $(api_name) and security environment $(security_environment); python apps/render_cf.py -j $(files) -e $(compute_env_file) -s $(security_environment) -n $(api_name) -c $(cost_profile)
 
 static: flake8 openapi-validate cfn-lint
 

apps/api/src/hyp3_api/api-spec/job_parameters.yml.j2

@@ -26,13 +26,13 @@ components:
       description: Parameters for running {{ job_type }} jobs
       type: object
       additionalProperties: false
-      {% for parameter, parameter_spec in job_spec['parameters'].items() if 'api_schema' in parameter_spec and parameter in job_spec.get('required_parameters', []) %}
+      {% for parameter, parameter_spec in job_spec['parameters'].items() if parameter in job_spec.get('required_parameters', []) %}
       {% if loop.first %}
       required:
       {% endif %}
         - {{ parameter }}
       {% endfor %}
-      {% for parameter, parameter_spec in job_spec['parameters'].items() if 'api_schema' in parameter_spec %}
+      {% for parameter, parameter_spec in job_spec['parameters'].items() %}
       {% if loop.first %}
       properties:
       {% endif %}

apps/api/src/hyp3_api/api-spec/openapi-spec.yml.j2

@@ -408,20 +408,25 @@ components:
 
     processing_times:
       description: >
-        A list of run times for the job's processing steps in the order that they were executed. For example,
-        a job comprised of a single processing step would yield a list containing one processing time, while a job
-        comprised of three processing steps would yield a list containing three processing times. An empty list
-        represents a failure to calculate processing times.
+        List of run times for the job's processing steps in the order that they were executed.
+        An empty list represents a failure to calculate processing times.
       type: array
       items:
-        $ref: '#/components/schemas/processing_time_in_seconds'
+        oneOf:
+          - type: array
+            items:
+              $ref: '#/components/schemas/processing_time_in_seconds'
+            example: [200, 100, 150]
+          - $ref: '#/components/schemas/processing_time_in_seconds'
+
 
     processing_time_in_seconds:
       description: >
-        Run time in seconds for a particular processing step's final attempt (regardless of whether it succeeded).
+        Run time in seconds for a processing step's final attempt (regardless of whether it succeeded).
         A value of zero indicates that there were no attempts.
       type: number
       minimum: 0
+      example: 50
 
   securitySchemes:
     EarthDataLogin:

apps/api/src/hyp3_api/handlers.py

@@ -6,7 +6,7 @@
 import dynamo
 from dynamo.exceptions import AccessCodeError, InsufficientCreditsError, UnexpectedApplicationStatusError
 from hyp3_api import util
-from hyp3_api.validation import GranuleValidationError, validate_jobs
+from hyp3_api.validation import BoundsValidationError, GranuleValidationError, validate_jobs
 
 
 def problem_format(status, message):
@@ -28,7 +28,7 @@ def post_jobs(body, user):
         validate_jobs(body['jobs'])
     except requests.HTTPError as e:
         print(f'WARN: CMR search failed: {e}')
-    except GranuleValidationError as e:
+    except (BoundsValidationError, GranuleValidationError) as e:
         abort(problem_format(400, str(e)))
 
     try:

apps/api/src/hyp3_api/validation.py

@@ -18,6 +18,10 @@ class GranuleValidationError(Exception):
     pass
 
 
+class BoundsValidationError(Exception):
+    pass
+
+
 with open(Path(__file__).parent / 'job_validation_map.yml') as f:
     JOB_VALIDATION_MAP = yaml.safe_load(f.read())
 
@@ -136,6 +140,62 @@ def get_multipolygon_from_geojson(input_file):
     return MultiPolygon(polygons)
 
 
+def check_bounds_formatting(job, _):
+    bounds = job['job_parameters']['bounds']
+    if bounds == [0.0, 0.0, 0.0, 0.0]:
+        return
+
+    if bounds[0] >= bounds[2] or bounds[1] >= bounds[3]:
+        raise BoundsValidationError(
+            'Invalid order for bounds. Bounds should be ordered [min lon, min lat, max lon, max lat].'
+        )
+
+    def bad_lat(lat):
+        return lat > 90 or lat < -90
+
+    def bad_lon(lon):
+        return lon > 180 or lon < -180
+
+    if any([bad_lon(bounds[0]), bad_lon(bounds[2]), bad_lat(bounds[1]), bad_lat(bounds[3])]):
+        raise BoundsValidationError(
+            'Invalid lon/lat value(s) in bounds. Bounds should be ordered [min lon, min lat, max lon, max lat].'
+        )
+
+
+def check_granules_intersecting_bounds(job, granule_metadata):
+    bounds = job['job_parameters']['bounds']
+    if bounds == [0.0, 0.0, 0.0, 0.0]:
+        bounds = granule_metadata[0]['polygon']
+    else:
+        bounds = Polygon.from_bounds(*bounds)
+    bad_granules = []
+    for granule in granule_metadata:
+        bbox = granule['polygon']
+        if not bbox.intersection(bounds):
+            bad_granules.append(granule['name'])
+    if bad_granules:
+        raise GranuleValidationError(
+            f'The following granules do not intersect the provided bounds: {bad_granules}.'
+        )
+
+
+def check_same_relative_orbits(job, granule_metadata):
+    previous_relative_orbit = None
+    for granule in granule_metadata:
+        name_split = granule['name'].split('_')
+        absolute_orbit = name_split[7]
+        # "Relationship between relative and absolute orbit numbers": https://sentiwiki.copernicus.eu/web/s1-products
+        offset = 73 if name_split[0] == 'S1A' else 27
+        relative_orbit = ((int(absolute_orbit) - offset) % 175) + 1
+        if not previous_relative_orbit:
+            previous_relative_orbit = relative_orbit
+        if relative_orbit != previous_relative_orbit:
+            raise GranuleValidationError(
+                f'Relative orbit number for {granule["name"]} does not match that of the previous granules: '
+                f'{relative_orbit} is not {previous_relative_orbit}.'
+            )
+
+
 def convert_single_burst_jobs(jobs: list[dict]) -> list[dict]:
     jobs = deepcopy(jobs)
     for job in jobs:

apps/check-processing-time/src/check_processing_time.py

@@ -1,4 +1,5 @@
 import json
+from typing import Union
 
 
 def get_time_from_attempts(attempts: list[dict]) -> float:
@@ -9,15 +10,18 @@ def get_time_from_attempts(attempts: list[dict]) -> float:
     return (final_attempt['StoppedAt'] - final_attempt['StartedAt']) / 1000
 
 
-def get_time_from_result(result: dict) -> float:
-    if 'Attempts' in result:
-        attempts = result['Attempts']
-    else:
-        attempts = json.loads(result['Cause'])['Attempts']
-    return get_time_from_attempts(attempts)
+def get_time_from_result(result: Union[list, dict]) -> Union[list, float]:
+    if isinstance(result, list):
+        return [get_time_from_result(item) for item in result]
 
+    if 'start' in result:
+        attempts = [{'StartedAt': start, 'StoppedAt': stop} for start, stop in zip(result['start'], result['stop'])]
+        return get_time_from_attempts(attempts)
 
-def lambda_handler(event, context) -> list[float]:
-    results_dict = event['processing_results']
-    results = [results_dict[key] for key in sorted(results_dict.keys())]
-    return list(map(get_time_from_result, results))
+    return get_time_from_attempts(json.loads(result['Cause'])['Attempts'])
+
+
+def lambda_handler(event, _) -> list[Union[list, float]]:
+    processing_results = event['processing_results']
+    result_list = [processing_results[key] for key in sorted(processing_results.keys())]
+    return get_time_from_result(result_list)

apps/compute-cf.yml.j2

@@ -29,15 +29,14 @@ Outputs:
   JobQueueArn:
     Value: !Ref BatchJobQueue
 
-  {% for job_type, job_spec in job_types.items()  if 'Default' != job_spec['compute_environment']['name'] %}
-  {% set name = job_spec['compute_environment']['name'] %}
+  {% for name in compute_envs %}
   {{ name }}ComputeEnvironmentArn:
     Value: !Ref {{ name }}ComputeEnvironment
 
   {{ name }}JobQueueArn:
     Value: !Ref {{ name }}JobQueue
-  {% endfor %}
 
+  {% endfor %}
   TaskRoleArn:
     Value: !GetAtt TaskRole.Arn
 
@@ -106,9 +105,7 @@ Resources:
   SchedulingPolicy:
     Type: AWS::Batch::SchedulingPolicy
 
-  {% for job_type, job_spec in job_types.items() if 'Default' != job_spec['compute_environment']['name'] %}
-  {% set env = job_spec['compute_environment'] %}
-  {% set name = env['name'] %}
+  {% for name, env in compute_envs.items() %}
   {% set instance_types = env['instance_types'].split(',') if 'instance_types' in env else '!Ref InstanceTypes' %}
   {% set ami_id = env['ami_id'] if 'ami_id' in env else '!Ref AmiId' %}
   {% set type = env['allocation_type'] if 'allocation_type' in env else 'SPOT' %}

apps/handle-batch-event/handle-batch-event-cf.yml.j2

@@ -5,8 +5,8 @@ Parameters:
   JobQueueArn:
     Type: String
 
-  {% for job_type, job_spec in job_types.items() if 'Default' != job_spec['compute_environment']['name'] %}
-  {{ job_spec['compute_environment']['name'] }}JobQueueArn:
+  {% for name in compute_envs %}
+  {{ name }}JobQueueArn:
     Type: String
   {% endfor %}
 
@@ -100,8 +100,8 @@ Resources:
         detail:
           jobQueue:
             - !Ref JobQueueArn
-            {% for job_type, job_spec in job_types.items() if 'Default' != job_spec['compute_environment']['name'] %}
-            - !Ref {{ job_spec['compute_environment']['name'] }}JobQueueArn
+            {% for name in compute_envs %}
+            - !Ref {{ name }}JobQueueArn
             {% endfor %}
           status:
             - RUNNING