Skip to content

Commit 2f7a176

Browse files
mikedorfmanmikedorfman
authored
Merge pull request #4206 from nasa/mikedorfman/CUMULUS-4382
mikedorfman/CUMULUS-4382
2 parents 8bd9b1c + 81e8b17 commit 2f7a176

13 files changed

Lines changed: 1741 additions & 0 deletions

File tree

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,3 +50,4 @@ yarn.lock
5050
*.tsbuildinfo
5151
unit-logs/
5252
test_output.txt
53+
__pycache__/

‎CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,8 @@ Please complete the following steps before upgrading Cumulus.
5151
By setting `db_log_min_duration_ms` to a positive value (in milliseconds) and `enabled_cloudwatch_logs_exports`
5252
to `["postgresql"]`, RDS will log and export any database queries that take longer than that threshold.
5353
The module also configures the required RDS extensions and parameters necessary for slow query instrumentation.
54+
- **CUMULUS-4382**
55+
- Migrated the granule-invalidator task to the `tasks` directory as part of a coreification task in support of providing rolling archive functionality.
5456

5557
### Changed
5658

‎tasks/granule-invalidator/README.md‎

Lines changed: 351 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,351 @@
1+
# @cumulus/granule-invalidator
2+
3+
This task queries the Cumulus API for granules that fulfill certain criteria as defined in `src/invalidations.py`. Currently, the criteria that can be optionally specified via config (see Configuration section) are:
4+
5+
- science_date - any granules that have a science date older than the specified time
6+
- ingest_date - any granules that have an ingest date older than the specified time
7+
- cross_collection - any granules that have corresponding granules in another collection that match with the same begin/end time
8+
9+
This task uses the Cumulus Message Adapter and is intended for use in a Cumulus workflow. This task is meant to be run against a single collection and version. Importantly, this repository does not alter the database, it only identifies a list of granules that fulfills the criteria.
10+
11+
## Usage
12+
13+
This lambda takes the following input and config objects, derived from workflow configuration using the [Cumulus Message Adapter](https://github.com/nasa/cumulus-message-adapter/blob/master/CONTRACT.md) to drive configuration from the full cumulus message. The output from the task follows the Cumulus Message Adapter contract and provides the information detailed below.
14+
15+
### Configuration
16+
17+
| field name | type | default | required | values | description
18+
| ---------- | ---- | ------- | -------- | ------ | -----------
19+
| collection | string | N/A | yes | N/A | The collection shortname to apply invalidations to
20+
| version | string | N/A | yes | N/A | The version of the collection to apply invalidations to
21+
| page_length_ms | integer | 604800000 | no | N/A | Time window in milliseconds for pagination; defaults to 7 days
22+
| granule_invalidations | array | N/A | yes | N/A | Array of invalidation criteria objects
23+
| granule_invalidations[].type | string | N/A | yes | `science_date`, `ingest_date`, `cross_collection` | The type of invalidation to apply
24+
| granule_invalidations[].maximum_minutes_old | integer | N/A | conditional | N/A | Maximum age in minutes; required for `science_date` and `ingest_date` types
25+
| granule_invalidations[].invalidating_collection | string | N/A | conditional | N/A | Collection shortname of granules to match against; required for `cross_collection` type
26+
| granule_invalidations[].invalidating_version | string | N/A | conditional | N/A | Collection version of granules to match against; required for `cross_collection` type
27+
28+
The following sample configuration lists all AMSR_E_L2_Rain v13 granules older than 525600 minutes and all granules that have a corresponding granule in the AMSR_E_L2_Ocean v13 collection that match beginning and end time:
29+
30+
```json
31+
{
32+
"collection": "AMSR_E_L2_Rain",
33+
"version": "13",
34+
"granule_invalidations": [
35+
{
36+
"type": "science_date",
37+
"maximum_minutes_old": 525600
38+
},
39+
{
40+
"type": "cross_collection",
41+
"invalidating_collection": "AMSR_E_L2_Ocean",
42+
"invalidating_version": "13"
43+
}
44+
]
45+
}
46+
```
47+
48+
### Input
49+
50+
The input field for this task is expected to be empty.
51+
52+
### Output
53+
54+
In general, the output can be used to identify specifically which granules are to be removed, in the structure that a Cumulus bulk delete API call accepts and provide human-readable stats to verify validity of this set of granules. It is often useful to store the `granules` field in a remote message, as this list can easily exceed the maximum length of a StepFunctions payload and allows for easy inspection by operators. The following is an example of the output of this task:
55+
56+
```json
57+
{
58+
"granules": [
59+
{
60+
"granuleId": "AMSR_E_L2_Rain_V13_200206020859_D",
61+
"collectionId": "AMSR_E_L2_Rain___13"
62+
},
63+
{
64+
"granuleId": "AMSR_E_L2_Rain_V13_200206021100_D",
65+
"collectionId": "AMSR_E_L2_Rain___13"
66+
},
67+
{
68+
"granuleId": "AMSR_E_L2_Rain_V13_200206021430_D",
69+
"collectionId": "AMSR_E_L2_Rain___13"
70+
}
71+
],
72+
"forceRemoveFromCmr": true,
73+
"granules_to_be_deleted_count": 3,
74+
"aggregated_stats": "Total number of granules to be removed: 3\nTotal number of granules to be retained: 97\nGranules to be removed by invalidation type:\nscience_date - 2 granules\ncross_collection - 1 granules\n"
75+
}
76+
```
77+
78+
79+
### Example workflow configuration
80+
81+
This lambda only fulfils part of the overarching goal - which is to identify, then remove granules from Cumulus. To achieve the goal, you may set up a scheduled Cumulus rule. This workflow might:
82+
83+
* Run the granules-to-delete identification task to identify which granules should be removed
84+
* Using an SNS topic/endpoint, send an email with details about what should be deleted based on configuration passed into the task
85+
* Wait for a configurable amount of time for ops intervention if they identify something that should not be deleted in that list
86+
* Call the Cumulus Private API lambda from the workflow and use the Bulk Delete endpoint to delete these granules from Cumulus and CMR
87+
88+
A sample definition of this workflow could look like:
89+
90+
```json
91+
{
92+
"StartAt": "Is Wait Time Set?",
93+
"States": {
94+
"Is Wait Time Set?": {
95+
"Type": "Choice",
96+
"Choices": [
97+
{
98+
"Not": {
99+
"Variable": "$.meta.waitSeconds",
100+
"IsPresent": true
101+
},
102+
"Next": "Set Default Wait Time"
103+
}
104+
],
105+
"Default": "Identify Granules to Delete"
106+
},
107+
"Set Default Wait Time": {
108+
"Type": "Pass",
109+
"Result": "${default_wait_s}",
110+
"ResultPath": "$.meta.waitSeconds",
111+
"Next": "Identify Granules to Delete"
112+
},
113+
"Identify Granules to Delete": {
114+
"Type": "Task",
115+
"Resource": "${granule_invalidator_arn}",
116+
"Next": "Are there granules to delete?",
117+
"Parameters": {
118+
"cma": {
119+
"event.$": "$",
120+
"ReplaceConfig": {
121+
"Path": "$.payload.granules"
122+
},
123+
"task_config": {
124+
"granule_invalidations": "{$.meta.collection.meta.granule_invalidations}",
125+
"collection": "{$.meta.collection.name}",
126+
"version": "{$.meta.collection.version}",
127+
"cumulus_message": {
128+
"outputs": [
129+
{
130+
"source": "{$.aggregated_stats}",
131+
"destination": "{$.meta.aggregated_stats}"
132+
},
133+
{
134+
"source": "{$.granules_to_be_deleted_count}",
135+
"destination": "{$.meta.granules_to_be_deleted_count}"
136+
},
137+
{
138+
"source": "{$.granules}",
139+
"destination": "{$.payload.granules}"
140+
},
141+
{
142+
"source": "{$.forceRemoveFromCmr}",
143+
"destination": "{$.payload.forceRemoveFromCmr}"
144+
}
145+
]
146+
}
147+
}
148+
}
149+
},
150+
"Catch": [
151+
{
152+
"ErrorEquals": ["States.ALL"],
153+
"Next": "Notify Ops of Failure",
154+
"ResultPath": "$.error"
155+
}
156+
],
157+
"Assign": {
158+
"deletion_count.$": "$.meta.granules_to_be_deleted_count"
159+
}
160+
},
161+
"Are there granules to delete?": {
162+
"Type": "Choice",
163+
"Choices": [
164+
{
165+
"Variable": "$deletion_count",
166+
"NumericGreaterThan": 0,
167+
"Next": "Notify Ops of Deletion Information"
168+
}
169+
],
170+
"Default": "Notify Ops of Nothing to Delete"
171+
},
172+
"Notify Ops of Deletion Information": {
173+
"Type": "Task",
174+
"Resource": "arn:aws:states:::sns:publish",
175+
"Parameters": {
176+
"Subject.$": "States.Format('QuickLook Deletion - Scheduled for {} granules for {}.{}', $deletion_count, $.meta.collection.name, $.meta.collection.version)",
177+
"Message.$": ${jsonencode(notification_template)},
178+
"TopicArn": "${ops_notification_arn}"
179+
},
180+
"ResultPath": null,
181+
"Next": "Wait for Ops Intervention"
182+
},
183+
"Notify Ops of Nothing to Delete": {
184+
"Type": "Task",
185+
"Resource": "arn:aws:states:::sns:publish",
186+
"Parameters": {
187+
"Subject.$": "States.Format('QuickLook Deletion - Nothing to delete for {}.{}', $.meta.collection.name, $.meta.collection.version)",
188+
"Message.$": ${jsonencode(noop_template)},
189+
"TopicArn": "${ops_notification_arn}"
190+
},
191+
"ResultPath": null,
192+
"End": true
193+
},
194+
"Wait for Ops Intervention": {
195+
"Type": "Wait",
196+
"SecondsPath": "$.meta.waitSeconds",
197+
"Next": "Reload Message"
198+
},
199+
"Reload Message": {
200+
"Type": "Task",
201+
"Resource": "${cumulus_message_manipulator_arn}",
202+
"Next": "Bulk Delete",
203+
"Parameters": {
204+
"cma": {
205+
"event.$": "$",
206+
"task_config": {
207+
"manipulation_action": "noop",
208+
"manipulation_params": {}
209+
}
210+
}
211+
},
212+
"Catch": [
213+
{
214+
"ErrorEquals": ["States.ALL"],
215+
"Next": "Notify Ops of Failure",
216+
"ResultPath": "$.error"
217+
}
218+
]
219+
},
220+
"Bulk Delete": {
221+
"Type": "Task",
222+
"Resource": "${cumulus_internal_api_arn}",
223+
"Next": "Wait for Bulk Delete to Finish",
224+
"Parameters": {
225+
"httpMethod": "POST",
226+
"resource": "{proxy+}",
227+
"path": "/granules/bulkDelete",
228+
"headers": {
229+
"Content-Type": "application/json"
230+
},
231+
"body.$": "States.JsonToString($.payload)"
232+
},
233+
"Catch": [
234+
{
235+
"ErrorEquals": ["States.ALL"],
236+
"Next": "Notify Ops of Failure",
237+
"ResultPath": "$.error"
238+
}
239+
],
240+
"ResultSelector": {
241+
"api_return.$": "States.StringToJson($.body)"
242+
},
243+
"ResultPath": "$.payload"
244+
},
245+
"Check Bulk Delete Status": {
246+
"Type": "Task",
247+
"Resource": "${cumulus_internal_api_arn}",
248+
"Parameters": {
249+
"httpMethod": "GET",
250+
"resource": "{proxy+}",
251+
"path.$": "States.Format('/asyncOperations/{}', $.payload.api_return.id)",
252+
"headers": {
253+
"Content-Type": "application/json"
254+
}
255+
},
256+
"ResultSelector": {
257+
"api_return.$": "States.StringToJson($.body)"
258+
},
259+
"ResultPath": "$.payload",
260+
"Next": "Is bulk delete done?"
261+
},
262+
"Is bulk delete done?": {
263+
"Type": "Choice",
264+
"Choices": [
265+
{
266+
"Variable": "$.payload.api_return.status",
267+
"StringMatches": "RUNNING",
268+
"Next": "Wait for Bulk Delete to Finish"
269+
},
270+
{
271+
"Variable": "$.payload.api_return.status",
272+
"StringMatches": "SUCCEEDED",
273+
"Next": "Notify Ops of Successful Deletion"
274+
}
275+
],
276+
"Default": "Notify Ops of Failure"
277+
},
278+
"Wait for Bulk Delete to Finish": {
279+
"Type": "Wait",
280+
"Seconds": 60,
281+
"Next": "Check Bulk Delete Status"
282+
},
283+
"Notify Ops of Successful Deletion": {
284+
"Type": "Task",
285+
"Resource": "arn:aws:states:::sns:publish",
286+
"Parameters": {
287+
"Subject.$": "States.Format('QuickLook Deletion - Successfully deleted {} granules for {}.{}', $deletion_count, $.meta.collection.name, $.meta.collection.version)",
288+
"Message.$": ${jsonencode(success_template)},
289+
"TopicArn": "${ops_notification_arn}"
290+
},
291+
"ResultPath": null,
292+
"End": true
293+
},
294+
"Notify Ops of Failure": {
295+
"Type": "Task",
296+
"Resource": "arn:aws:states:::sns:publish",
297+
"Parameters": {
298+
"Subject.$": "States.Format('QuickLook Deletion - Delete Failed for {}.{}', $.meta.collection.name, $.meta.collection.version)",
299+
"Message.$": ${jsonencode(failure_template)},
300+
"TopicArn": "${ops_notification_arn}"
301+
},
302+
"Next": "Workflow Failed",
303+
"ResultPath": null
304+
},
305+
"Workflow Failed": {
306+
"Type": "Fail"
307+
}
308+
}
309+
}
310+
```
311+
312+
## Architecture
313+
```mermaid
314+
graph TB
315+
A["AWS Lambda Handler<br/>lambda_handler"] -->|calls| B["Lambda Adapter<br/>lambda_adapter"]
316+
317+
B -->|fetches granules<br/>via async requests| C["Cumulus API<br/>CumulusApi.list_granules"]
318+
C -->|processes sequentially| F1["science_date<br/>invalidations.py"]
319+
320+
F1 -->|identifies old by<br/>productionDateTime| F2["ingest_date<br/>invalidations.py"]
321+
F2 -->|identifies old by<br/>createdAt| D["Cumulus API<br/>CumulusApi.list_granules<br/>for invalidating collection"]
322+
D -->|fetches granules<br/>from other collection| F3["cross_collection<br/>invalidations.py"]
323+
F3 -->|matches by<br/>begin/end time| J["Aggregated Valid &<br/>Invalid Granules"]
324+
325+
J -->|aggregates| K["Output<br/>Granule List<br/>Statistics"]
326+
327+
K -->|returns| L["Cumulus Workflow<br/>Response"]
328+
```
329+
330+
### Internal Dependencies
331+
332+
This task relies on access to the Cumulus API via the Private API lambda. This API interaction is run through the cumulus-api python client which requires the `PRIVATE_API_LAMBDA_ARN` environment variable to be set. Additionally, the task must have permissions to invoke the Private API lambda.
333+
334+
### External Dependencies
335+
336+
This task does not have any runtime external dependencies, however it is dependent on the following github packages when building:
337+
- https://github.com/ghrcdaac/cumulus-api
338+
- https://github.com/nasa/cumulus-message-adapter
339+
- https://github.com/nasa/cumulus-message-adapter-python
340+
341+
The typical workflow configuration for this task does contain an indirect dependency on the Common Metadata Repository (CMR) - when running the bulk delete, the granules are also removed from CMR.
342+
343+
## Contributing
344+
345+
To make a contribution, please [see our Cumulus contributing guidelines](https://github.com/nasa/cumulus/blob/master/CONTRIBUTING.md) and our documentation on [adding a task](https://nasa.github.io/cumulus/docs/adding-a-task)
346+
347+
## About Cumulus
348+
349+
Cumulus is a cloud-based data ingest, archive, distribution and management prototype for NASA's future Earth science data streams.
350+
351+
[Cumulus Documentation](https://nasa.github.io/cumulus)

‎tasks/granule-invalidator/generate_dirs.bash‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
#!/bin/bash
2+
3+
# Script to generate standard task directory structure
4+
5+
TARGET_DIR="."
6+
7+
# Create all directories
8+
mkdir -p "$TARGET_DIR"/{bin,src/python_reference_task,deploy,tests/integration_tests,tests/unit_tests/python_reference_task}
9+
10+
echo "Directory structure created at: $TARGET_DIR"

0 commit comments

Comments
 (0)