Add script to rollback failed import#613
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new script, retry_failed_ingestion.py, along with its corresponding unit tests, to automate the rollback and retry process for failed Dataflow ingestion jobs. Feedback on the implementation highlights a critical correctness bug where the workflow could be retriggered even if a rollback fails, and a missing pagination implementation when fetching Dataflow job messages. Additionally, it is recommended to use posixpath instead of os.path for GCS path manipulations to avoid platform-specific path separator issues, and to clean up unused imports in the test file.
Not up to standards ⛔🟢 Issues
|
| Metric | Results |
|---|---|
| Complexity | 38 |
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
| from googleapiclient.discovery import build | ||
| from googleapiclient.errors import HttpError | ||
|
|
||
| FLAGS = flags.FLAGS |
There was a problem hiding this comment.
What do you think about putting this logic in the ingestion helper?
So that retrying a failed import, can just be executing a post call to the ingestion helper with a different URL param, and it would process all this logic you have here?
That way, it isnt a separate script/image, we can keep it well contained in the ingestion helper image, and use the workflow to trigger it just like everything else.
| optionally retriggers the Spanner ingestion workflow. | ||
|
|
||
| Usage: | ||
| python3 retry_failed_ingestion.py --job_id=<DATAFLOW_JOB_ID> \ |
There was a problem hiding this comment.
Is this the expected usage of this script? manual triggers?
Or is the expectation that the workflow automaitcally does this?
| "Dry run mode. Display actions without executing them.") | ||
|
|
||
|
|
||
| def get_failed_imports(dataflow, project_id, location, job_id): |
There was a problem hiding this comment.
I would probably recommend not just putting this as a single file script under scripts/
It's already almost 400 lines, and very likely to grow in the future. I would recommend keeping a main.py as the simple entry point, and helpers to navigate the complexity. This also seems complex enough that it warrants tests imo.
I opened anohter comment asking if this should just be in the ingestion-helper. Curious for your thoughts on that?
| logging.error(f"Error parsing importList parameter: {e}") | ||
|
|
||
| logging.info(f"Fetching error messages for Dataflow job {job_id}...") | ||
| request = dataflow.projects().locations().jobs().messages().list( |
There was a problem hiding this comment.
This seems to only be checking dataflow for failures. What if the failures came from elsewhere in the worfklow (spanner embeddings, ingestion helper issue, bq federation).
What's the long term plan for this, should we be checking all the stages? Should the workflow be modified to propoagate errors from dataflow? (if that's even possible)
| def _revert_txn(transaction): | ||
| # 1. Get current version from ImportStatus to know what to avoid. | ||
| status_sql = """ | ||
| SELECT LatestVersion FROM ImportStatus |
There was a problem hiding this comment.
Given the changes we discussed to IngestionHistory last week, isnt that a better place to read this from? ImportStatus seems brittle given that if we have started a new workflow since, that information is gone.
| """ | ||
| status_results = transaction.execute_sql( | ||
| status_sql, | ||
| params={'importName': short_name}, |
There was a problem hiding this comment.
How would this work for a multi-provenance import?
Like if I import WHO and ILO at once?
2 participants
No description provided.