Implemented fault tolerance based on Provenance module #169
Conversation
# Conflicts: # streamflow/cwl/step.py # streamflow/workflow/combinator.py # streamflow/workflow/step.py
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
There was a problem hiding this comment.
CodeQL found more than 10 potential problems in the proposed changes. Check the Files changed tab for more details.
| self.token_instances.pop(old_token_id) | ||
|
|
||
| # replace | ||
| pass |
Check warning
Code scanning / CodeQL
Unnecessary pass Warning
| # if DirectGraph.LAST_GRAPH_FLAG in self.graph[src] and len(self.graph[src]) > 1: | ||
| # self.graph[src].remove(DirectGraph.LAST_GRAPH_FLAG) | ||
| # logger.debug( | ||
| # f"DG: Remove {DirectGraph.LAST_GRAPH_FLAG} from {src} next elems" |
Check notice
Code scanning / CodeQL
Commented-out code Note
| # if not self.is_present(t_id): | ||
| # logger.debug(f"Remove orphan token {t_id}") | ||
| # self.token_available.pop(t_id, None) | ||
| # self.token_instances.pop(t_id, None) |
Check notice
Code scanning / CodeQL
Commented-out code Note
| # if token.persistent_id == old_token_id: | ||
| # return |
Check notice
Code scanning / CodeQL
Commented-out code Note
| # if not isinstance(t, JobToken): | ||
| # self.add(t, None) |
Check notice
Code scanning / CodeQL
Commented-out code Note
| # for t_id, next_t_ids in self.dag_tokens.items(): | ||
| # for next_t_id in next_t_ids: | ||
| # rdwp.new_add( | ||
| # self.info_tokens.get(t_id, None), | ||
| # self.info_tokens.get(next_t_id, None), |
Check notice
Code scanning / CodeQL
Commented-out code Note
GlassOfWhiskey
force-pushed
the
master
branch
from
d1735c2
to
e9bb106
Compare
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #169 +/- ##
==========================================
- Coverage 70.18% 62.15% -8.03%
==========================================
Files 83 83
Lines 10591 10528 -63
Branches 2504 2522 +18
==========================================
- Hits 7433 6544 -889
- Misses 2702 3575 +873
+ Partials 456 409 -47 ☔ View full report in Codecov by Sentry. |
| workflow.ports[port_recovery.port.name] = port_recovery.port | ||
| if len(job_token_names) == 1: | ||
| logger.debug(f"New-workflow {workflow.name} will be empty.") | ||
| pass |
Check warning
Code scanning / CodeQL
Unnecessary pass Warning
| f"\n\tis_avai: {info.is_available}\n\tport: {dict(info.port_row)}" | ||
| f"\n\tsteps: {[ s['name'] for s in info.step_rows] }" | ||
| ) | ||
| pass |
Check warning
Code scanning / CodeQL
Unnecessary pass Warning
| # if len(token_list) != 2: | ||
| # raise FailureHandlingException( | ||
| # f"Step recovery {failed_step.name} did not generate the right number of tokens: {len(token_list)}" |
Check notice
Code scanning / CodeQL
Commented-out code Note
| # if not isinstance(token_list[1], TerminationToken): | ||
| # raise FailureHandlingException( | ||
| # f"Step recovery {failed_step.name} did not work well. " | ||
| # f"It moved two tokens instead of one: {[t.persistent_id for t in token_list]}" |
Check notice
Code scanning / CodeQL
Commented-out code Note
| # for port in failed_step.get_input_ports().values(): | ||
| # if port.name not in new_workflow.ports.keys(): | ||
| # raise FailureHandlingException( | ||
| # f"La input port {port.name} dello step fallito {failed_step.name} " | ||
| # f"non è presente nel new_workflow {new_workflow.name}" |
Check notice
Code scanning / CodeQL
Commented-out code Note
| # for next_t_id in self.dag_tokens.succ(job_token.persistent_id): | ||
| # if not isinstance(next_t_id, int): | ||
| # continue | ||
| # step_rows = self.info_tokens[next_t_id].step_rows | ||
| # if is_there_step_type(step_rows, (ExecuteStep,)): | ||
| # output_token_ids.add(next_t_id) |
Check notice
Code scanning / CodeQL
Commented-out code Note
| # if not data_locations: | ||
| # # it is possible that src='' following a symbolic link | ||
| # # edit. tenere solo per debug. Ho aggiunto un controllo a monte dentro remotepath.py | ||
| # raise WorkflowTransferException( | ||
| # f"No data locations available {src_path if src_path else 'None'}" |
Check notice
Code scanning / CodeQL
Commented-out code Note
| @@ -7,6 +7,8 @@ | |||
| from importlib_resources import files | |||
|
|
|||
| from streamflow.core.data import DataLocation, DataManager, DataType | |||
| from streamflow.core.deployment import Connector | |||
| from streamflow.core.exception import WorkflowTransferException | |||
Check notice
Code scanning / CodeQL
Unused import Note
This PR changes the current fault tolerance module. When a step fails, the new fault tolerance module builds a sub-workflow composed of steps from the original workflow, and these steps are added to the sub-workflow if their output tokens are lost. The steps involved and the dependency are obtained using the Provenance module.