From cf180de4179185f91a39389b1d82820503d2f47a Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Wed, 8 Dec 2021 16:25:08 -0800 Subject: [PATCH 01/13] nit Signed-off-by: Yee Hing Tong --- flytekit/clis/flyte_cli/main.py | 9 ++++++++- tests/flytekit/unit/remote/test_remote.py | 6 ++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/flytekit/clis/flyte_cli/main.py b/flytekit/clis/flyte_cli/main.py index 4e0390a041..4351d207d7 100644 --- a/flytekit/clis/flyte_cli/main.py +++ b/flytekit/clis/flyte_cli/main.py @@ -306,6 +306,7 @@ def _render_schedule_expr(lp): _CONFIG_FLAGS = ["-c", "--config"] _PRINCIPAL_FLAGS = ["-r", "--principal"] _INSECURE_FLAGS = ["-i", "--insecure"] +_CERTIFICATE_FLAGS = [, "--certificate"] _project_option = _click.option(*_PROJECT_FLAGS, required=True, help="The project namespace to query.") _optional_project_option = _click.option( @@ -339,7 +340,6 @@ def _render_schedule_expr(lp): ) _insecure_option = _click.option(*_INSECURE_FLAGS, is_flag=True, help="Do not use SSL") _urn_option = _click.option("-u", "--urn", required=True, help="The unique identifier for an entity.") - _optional_urn_option = _click.option("-u", "--urn", required=False, help="The unique identifier for an entity.") _host_option = _click.option( @@ -607,6 +607,13 @@ def make_context(self, cmd_name, args, parent=None): help="[Optional] The name to pass to the sub-command (if applicable) If set again in the sub-command, " "the sub-command's parameter takes precedence.", ) +@_click.option( + *_CERTIFICATE_FLAGS, + required=False, + type=str, + default=None, + help="[Optional] Local path to a certificate file. If specified, the file will be read and passed " +) @_insecure_option @_click.group("flyte-cli", deprecated=True) @_click.pass_context diff --git a/tests/flytekit/unit/remote/test_remote.py b/tests/flytekit/unit/remote/test_remote.py index 2bb8174069..e96eb4b624 100644 --- a/tests/flytekit/unit/remote/test_remote.py +++ b/tests/flytekit/unit/remote/test_remote.py @@ -180,3 +180,9 @@ def test_explicit_grpc_channel_credentials(mock_insecure, mock_url, mock_secure_ assert mock_secure_channel.called assert mock_secure_channel.call_args[0][1] == credentials assert not mock_ssl_channel_credentials.called + + +def test_vjkl(): + rr = FlyteRemote.from_config("flytesnacks", "development") + we = rr.fetch_workflow(name="core.control_flow.run_merge_sort.merge_sort?d", version="") + From c09b36e8763540d48251ae60a8ab3e368b4e92b0 Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Wed, 8 Dec 2021 16:25:18 -0800 Subject: [PATCH 02/13] remove Signed-off-by: Yee Hing Tong --- flytekit/clis/flyte_cli/main.py | 7 ------- 1 file changed, 7 deletions(-) diff --git a/flytekit/clis/flyte_cli/main.py b/flytekit/clis/flyte_cli/main.py index 4351d207d7..56261c7628 100644 --- a/flytekit/clis/flyte_cli/main.py +++ b/flytekit/clis/flyte_cli/main.py @@ -607,13 +607,6 @@ def make_context(self, cmd_name, args, parent=None): help="[Optional] The name to pass to the sub-command (if applicable) If set again in the sub-command, " "the sub-command's parameter takes precedence.", ) -@_click.option( - *_CERTIFICATE_FLAGS, - required=False, - type=str, - default=None, - help="[Optional] Local path to a certificate file. If specified, the file will be read and passed " -) @_insecure_option @_click.group("flyte-cli", deprecated=True) @_click.pass_context From 03668d95b67ee38ab5b957cccc7e1235888f4384 Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Thu, 9 Dec 2021 17:52:46 -0800 Subject: [PATCH 03/13] didn't crash Signed-off-by: Yee Hing Tong --- flytekit/clis/flyte_cli/main.py | 1 - flytekit/models/core/workflow.py | 2 +- flytekit/remote/component_nodes.py | 27 +++++++++++++++++++++++ flytekit/remote/nodes.py | 12 +++++++++- tests/flytekit/unit/remote/test_remote.py | 11 +++++++-- 5 files changed, 48 insertions(+), 5 deletions(-) diff --git a/flytekit/clis/flyte_cli/main.py b/flytekit/clis/flyte_cli/main.py index 56261c7628..081985ac3f 100644 --- a/flytekit/clis/flyte_cli/main.py +++ b/flytekit/clis/flyte_cli/main.py @@ -306,7 +306,6 @@ def _render_schedule_expr(lp): _CONFIG_FLAGS = ["-c", "--config"] _PRINCIPAL_FLAGS = ["-r", "--principal"] _INSECURE_FLAGS = ["-i", "--insecure"] -_CERTIFICATE_FLAGS = [, "--certificate"] _project_option = _click.option(*_PROJECT_FLAGS, required=True, help="The project namespace to query.") _optional_project_option = _click.option( diff --git a/flytekit/models/core/workflow.py b/flytekit/models/core/workflow.py index 6338a6695f..e9e68ce485 100644 --- a/flytekit/models/core/workflow.py +++ b/flytekit/models/core/workflow.py @@ -99,7 +99,7 @@ def else_node(self): def error(self): """ An error to throw in case none of the branches were taken. - :rtype: flytekit.models.core.errors.ContainerError + :rtype: flytekit.models.types.Error """ return self._error diff --git a/flytekit/remote/component_nodes.py b/flytekit/remote/component_nodes.py index 367cab8997..1389b7b4b7 100644 --- a/flytekit/remote/component_nodes.py +++ b/flytekit/remote/component_nodes.py @@ -124,3 +124,30 @@ def promote_from_model( raise _system_exceptions.FlyteSystemException( "Bad workflow node model, neither subworkflow nor launchplan specified." ) + + +class FlyteBranchNode(_workflow_model.BranchNode): + def __init__(self, if_else: _workflow_model.IfElseBlock): + super().__init__(if_else) + + @classmethod + def promote_from_model(cls, + base_model: _workflow_model.BranchNode, + sub_workflows: Dict[id_models.Identifier, _workflow_model.WorkflowTemplate], + node_launch_plans: Dict[id_models.Identifier, _launch_plan_model.LaunchPlanSpec], + tasks: Dict[id_models.Identifier, _task_model.TaskTemplate], + ) -> "FlyteBranchNode": + + from flytekit.remote.nodes import FlyteNode + + block = base_model.if_else + + else_node = None + if block.else_node: + else_node = FlyteNode.promote_from_model(block.else_node, sub_workflows, node_launch_plans, tasks) + + block.case._then_node = FlyteNode.promote_from_model(block.case.then_node, sub_workflows, node_launch_plans, tasks) + + new_if_else_block = _workflow_model.IfElseBlock(block.case, block.other, else_node, block.error) + + return cls(new_if_else_block) diff --git a/flytekit/remote/nodes.py b/flytekit/remote/nodes.py index f8ae1b2d6a..d29e1234f5 100644 --- a/flytekit/remote/nodes.py +++ b/flytekit/remote/nodes.py @@ -78,7 +78,7 @@ def promote_from_model( _logging.warning(f"Should not call promote from model on a start node or end node {model}") return None - flyte_task_node, flyte_workflow_node = None, None + flyte_task_node, flyte_workflow_node, flyte_branch_node = None, None, None if model.task_node is not None: flyte_task_node = _component_nodes.FlyteTaskNode.promote_from_model(model.task_node, tasks) elif model.workflow_node is not None: @@ -89,6 +89,8 @@ def promote_from_model( tasks, ) # TODO: Implement branch node https://github.com/flyteorg/flyte/issues/1116 + elif model.branch_node is not None: + flyte_branch_node = _component_nodes.FlyteBranchNode.promote_from_model(model.branch_node, sub_workflows, node_launch_plans, tasks) else: raise _system_exceptions.FlyteSystemException( f"Bad Node model, neither task nor workflow detected, node: {model}" @@ -132,6 +134,14 @@ def promote_from_model( raise _system_exceptions.FlyteSystemException( "Bad FlyteWorkflowNode model, both launch plan and workflow are None" ) + elif flyte_branch_node is not None: + return cls( + id=node_model_id, + upstream_nodes=[], # set downstream, model doesn't contain this information + bindings=model.inputs, + metadata=model.metadata, + flyte_branch=flyte_branch_node, + ) raise _system_exceptions.FlyteSystemException("Bad FlyteNode model, both task and workflow nodes are empty") @property diff --git a/tests/flytekit/unit/remote/test_remote.py b/tests/flytekit/unit/remote/test_remote.py index e96eb4b624..782b10cbf0 100644 --- a/tests/flytekit/unit/remote/test_remote.py +++ b/tests/flytekit/unit/remote/test_remote.py @@ -183,6 +183,13 @@ def test_explicit_grpc_channel_credentials(mock_insecure, mock_url, mock_secure_ def test_vjkl(): - rr = FlyteRemote.from_config("flytesnacks", "development") - we = rr.fetch_workflow(name="core.control_flow.run_merge_sort.merge_sort?d", version="") + rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") + wf = rr.fetch_workflow(name="core.control_flow.run_merge_sort.merge_sort", + version="931da9ead496000fc92379392d482295ab690988") + + +def test_jasilv(): + rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") + wf = rr.fetch_workflow(name="core.control_flow.run_conditions.nested_conditions", + version="931da9ead496000fc92379392d482295ab690988") From 38a8c5a395bc5c48b91ed1fb5879406adfc04033 Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Thu, 9 Dec 2021 18:05:08 -0800 Subject: [PATCH 04/13] add else and branch node to FlyteNode constructor Signed-off-by: Yee Hing Tong --- flytekit/remote/component_nodes.py | 20 +++++++++++++------- flytekit/remote/nodes.py | 18 +++++++++++------- tests/flytekit/unit/remote/test_remote.py | 12 +++++++----- 3 files changed, 31 insertions(+), 19 deletions(-) diff --git a/flytekit/remote/component_nodes.py b/flytekit/remote/component_nodes.py index 1389b7b4b7..c12dc38f90 100644 --- a/flytekit/remote/component_nodes.py +++ b/flytekit/remote/component_nodes.py @@ -131,12 +131,13 @@ def __init__(self, if_else: _workflow_model.IfElseBlock): super().__init__(if_else) @classmethod - def promote_from_model(cls, - base_model: _workflow_model.BranchNode, - sub_workflows: Dict[id_models.Identifier, _workflow_model.WorkflowTemplate], - node_launch_plans: Dict[id_models.Identifier, _launch_plan_model.LaunchPlanSpec], - tasks: Dict[id_models.Identifier, _task_model.TaskTemplate], - ) -> "FlyteBranchNode": + def promote_from_model( + cls, + base_model: _workflow_model.BranchNode, + sub_workflows: Dict[id_models.Identifier, _workflow_model.WorkflowTemplate], + node_launch_plans: Dict[id_models.Identifier, _launch_plan_model.LaunchPlanSpec], + tasks: Dict[id_models.Identifier, _task_model.TaskTemplate], + ) -> "FlyteBranchNode": from flytekit.remote.nodes import FlyteNode @@ -146,7 +147,12 @@ def promote_from_model(cls, if block.else_node: else_node = FlyteNode.promote_from_model(block.else_node, sub_workflows, node_launch_plans, tasks) - block.case._then_node = FlyteNode.promote_from_model(block.case.then_node, sub_workflows, node_launch_plans, tasks) + block.case._then_node = FlyteNode.promote_from_model( + block.case.then_node, sub_workflows, node_launch_plans, tasks + ) + + for o in block.other: + o._then_node = FlyteNode.promote_from_model(o.then_node, sub_workflows, node_launch_plans, tasks) new_if_else_block = _workflow_model.IfElseBlock(block.case, block.other, else_node, block.error) diff --git a/flytekit/remote/nodes.py b/flytekit/remote/nodes.py index d29e1234f5..97b49119f4 100644 --- a/flytekit/remote/nodes.py +++ b/flytekit/remote/nodes.py @@ -27,15 +27,18 @@ def __init__( flyte_task: Optional["FlyteTask"] = None, flyte_workflow: Optional["FlyteWorkflow"] = None, flyte_launch_plan: Optional["FlyteLaunchPlan"] = None, - flyte_branch=None, + flyte_branch_node: Optional["FlyteBranchNode"] = None, ): - non_none_entities = list(filter(None, [flyte_task, flyte_workflow, flyte_launch_plan, flyte_branch])) + # todo: flyte_branch_node is the only non-entity here, feels wrong, it should probably be a Condition + # or the other ones changed. + non_none_entities = list(filter(None, [flyte_task, flyte_workflow, flyte_launch_plan, flyte_branch_node])) if len(non_none_entities) != 1: raise _user_exceptions.FlyteAssertion( "An Flyte node must have one underlying entity specified at once. Received the following " "entities: {}".format(non_none_entities) ) - self._flyte_entity = flyte_task or flyte_workflow or flyte_launch_plan or flyte_branch + # todo: wip - flyte_branch_node is a hack kinda, it should be a Condition + self._flyte_entity = flyte_task or flyte_workflow or flyte_launch_plan or flyte_branch_node workflow_node = None if flyte_workflow is not None: @@ -46,7 +49,6 @@ def __init__( task_node = None if flyte_task: task_node = _component_nodes.FlyteTaskNode(flyte_task) - branch_node = None super(FlyteNode, self).__init__( id=id, @@ -56,7 +58,7 @@ def __init__( output_aliases=[], task_node=task_node, workflow_node=workflow_node, - branch_node=branch_node, + branch_node=flyte_branch_node, ) self._upstream = upstream_nodes @@ -90,7 +92,9 @@ def promote_from_model( ) # TODO: Implement branch node https://github.com/flyteorg/flyte/issues/1116 elif model.branch_node is not None: - flyte_branch_node = _component_nodes.FlyteBranchNode.promote_from_model(model.branch_node, sub_workflows, node_launch_plans, tasks) + flyte_branch_node = _component_nodes.FlyteBranchNode.promote_from_model( + model.branch_node, sub_workflows, node_launch_plans, tasks + ) else: raise _system_exceptions.FlyteSystemException( f"Bad Node model, neither task nor workflow detected, node: {model}" @@ -140,7 +144,7 @@ def promote_from_model( upstream_nodes=[], # set downstream, model doesn't contain this information bindings=model.inputs, metadata=model.metadata, - flyte_branch=flyte_branch_node, + flyte_branch_node=flyte_branch_node, ) raise _system_exceptions.FlyteSystemException("Bad FlyteNode model, both task and workflow nodes are empty") diff --git a/tests/flytekit/unit/remote/test_remote.py b/tests/flytekit/unit/remote/test_remote.py index 782b10cbf0..8a00b9e659 100644 --- a/tests/flytekit/unit/remote/test_remote.py +++ b/tests/flytekit/unit/remote/test_remote.py @@ -184,12 +184,14 @@ def test_explicit_grpc_channel_credentials(mock_insecure, mock_url, mock_secure_ def test_vjkl(): rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") - wf = rr.fetch_workflow(name="core.control_flow.run_merge_sort.merge_sort", - version="931da9ead496000fc92379392d482295ab690988") + wf = rr.fetch_workflow( + name="core.control_flow.run_merge_sort.merge_sort", version="931da9ead496000fc92379392d482295ab690988" + ) def test_jasilv(): rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") - wf = rr.fetch_workflow(name="core.control_flow.run_conditions.nested_conditions", - version="931da9ead496000fc92379392d482295ab690988") - + wf = rr.fetch_workflow( + name="core.control_flow.run_conditions.nested_conditions", version="931da9ead496000fc92379392d482295ab690988" + ) + print(wf) From da4161631de63a40b68d8696e10e0255111edce9 Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Mon, 13 Dec 2021 08:44:19 -0800 Subject: [PATCH 05/13] nit Signed-off-by: Yee Hing Tong --- tests/flytekit/unit/remote/test_remote.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/flytekit/unit/remote/test_remote.py b/tests/flytekit/unit/remote/test_remote.py index 8a00b9e659..b2be75362b 100644 --- a/tests/flytekit/unit/remote/test_remote.py +++ b/tests/flytekit/unit/remote/test_remote.py @@ -195,3 +195,6 @@ def test_jasilv(): name="core.control_flow.run_conditions.nested_conditions", version="931da9ead496000fc92379392d482295ab690988" ) print(wf) + + we = rr.fetch_workflow_execution(name="fahbj6avws") + rr.sync_workflow_execution(we, sync_nodes=True) From dff9462df7dc97e8c6dd94854ed587fad4c7aced Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Thu, 10 Feb 2022 09:21:21 -0800 Subject: [PATCH 06/13] no idea what's happening Signed-off-by: Yee Hing Tong --- flytekit/remote/component_nodes.py | 2 +- flytekit/remote/nodes.py | 3 ++- flytekit/remote/remote.py | 20 +++++++++++++------- tests/flytekit/unit/remote/test_remote.py | 7 +++---- 4 files changed, 19 insertions(+), 13 deletions(-) diff --git a/flytekit/remote/component_nodes.py b/flytekit/remote/component_nodes.py index 02b400d13b..2d6acca1ff 100644 --- a/flytekit/remote/component_nodes.py +++ b/flytekit/remote/component_nodes.py @@ -41,7 +41,7 @@ def promote_from_model( if base_model.reference_id in tasks: task = tasks[base_model.reference_id] - _logging.info(f"Found existing task template for {task.id}, will not retrieve from Admin") + _logging.debug(f"Found existing task template for {task.id}, will not retrieve from Admin") flyte_task = FlyteTask.promote_from_model(task) return cls(flyte_task) diff --git a/flytekit/remote/nodes.py b/flytekit/remote/nodes.py index 58bdf96d21..7f04185f4e 100644 --- a/flytekit/remote/nodes.py +++ b/flytekit/remote/nodes.py @@ -37,7 +37,8 @@ def __init__( "An Flyte node must have one underlying entity specified at once. Received the following " "entities: {}".format(non_none_entities) ) - # todo: wip - flyte_branch_node is a hack kinda, it should be a Condition + # todo: wip - flyte_branch_node is a hack, it should be a Condition, but backing out a Condition object from + # the compiled IfElseBlock is cumbersome, shouldn't do it if we can get away with it. self._flyte_entity = flyte_task or flyte_workflow or flyte_launch_plan or flyte_branch_node workflow_node = None diff --git a/flytekit/remote/remote.py b/flytekit/remote/remote.py index 93adc04db8..d3c08cdbdb 100644 --- a/flytekit/remote/remote.py +++ b/flytekit/remote/remote.py @@ -1201,13 +1201,9 @@ def sync_node_execution( for t in iterate_task_executions(self.client, execution.id) ] execution._interface = dynamic_flyte_wf.interface - else: - # If it does not, then it should be a static subworkflow - if not isinstance(execution._node.flyte_entity, FlyteWorkflow): - remote_logger.error( - f"NE {execution} entity should be a workflow, {type(execution._node)}, {execution._node}" - ) - raise Exception(f"Node entity has type {type(execution._node)}") + + # Handle the case where it's a static subworkflow + elif isinstance(execution._node.flyte_entity, FlyteWorkflow): sub_flyte_workflow = execution._node.flyte_entity sub_node_mapping = {n.id: n for n in sub_flyte_workflow.flyte_nodes} execution._underlying_node_executions = [ @@ -1216,6 +1212,16 @@ def sync_node_execution( ] execution._interface = sub_flyte_workflow.interface + # Handle the case where it's a branch node + elif execution._node.branch_node is not None: + print("hi") + else: + remote_logger.error( + f"NE {execution} undeterminable, {type(execution._node)}, {execution._node}" + ) + raise Exception(f"Node execution undeterminable, entity has type {type(execution._node)}") + + # This is the plain ol' task execution case else: execution._task_executions = [ diff --git a/tests/flytekit/unit/remote/test_remote.py b/tests/flytekit/unit/remote/test_remote.py index 0734f4db39..6c4f3e0a4a 100644 --- a/tests/flytekit/unit/remote/test_remote.py +++ b/tests/flytekit/unit/remote/test_remote.py @@ -185,16 +185,15 @@ def test_explicit_grpc_channel_credentials(mock_insecure, mock_url, mock_secure_ def test_vjkl(): rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") wf = rr.fetch_workflow( - name="core.control_flow.run_merge_sort.merge_sort", version="931da9ead496000fc92379392d482295ab690988" + name="core.control_flow.run_merge_sort.merge_sort", version="v0.3.39" ) def test_jasilv(): rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") wf = rr.fetch_workflow( - name="core.control_flow.run_conditions.nested_conditions", version="931da9ead496000fc92379392d482295ab690988" + name="core.control_flow.run_conditions.nested_conditions", version="v0.3.39" ) print(wf) - - we = rr.fetch_workflow_execution(name="fahbj6avws") + we = rr.fetch_workflow_execution(name="suss7xonol") rr.sync_workflow_execution(we, sync_nodes=True) From a3de6e0ed2b722686860926e2623bf8a25803b50 Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Mon, 14 Feb 2022 15:33:16 -0800 Subject: [PATCH 07/13] return early for now, can't figure out how to make branch nodes render properly Signed-off-by: Yee Hing Tong --- flytekit/remote/launch_plan.py | 7 ++++++ flytekit/remote/remote.py | 28 +++++++++++++---------- tests/flytekit/unit/remote/test_remote.py | 13 +++++------ 3 files changed, 29 insertions(+), 19 deletions(-) diff --git a/flytekit/remote/launch_plan.py b/flytekit/remote/launch_plan.py index b5dddea03f..e5d54f75ad 100644 --- a/flytekit/remote/launch_plan.py +++ b/flytekit/remote/launch_plan.py @@ -21,6 +21,13 @@ def __init__(self, id, *args, **kwargs): self._python_interface = None + # If fetched when creating this object, can store it here. + self._flyte_workflow = None + + @property + def flyte_workflow(self) -> Optional["FlyteWorkflow"]: + return self._flyte_workflow + @classmethod def promote_from_model( cls, id: id_models.Identifier, model: _launch_plan_models.LaunchPlanSpec diff --git a/flytekit/remote/remote.py b/flytekit/remote/remote.py index d3c08cdbdb..526e734d8a 100644 --- a/flytekit/remote/remote.py +++ b/flytekit/remote/remote.py @@ -440,6 +440,7 @@ def fetch_launch_plan( wf_id = flyte_launch_plan.workflow_id workflow = self.fetch_workflow(wf_id.project, wf_id.domain, wf_id.name, wf_id.version) flyte_launch_plan._interface = workflow.interface + flyte_launch_plan._flyte_workflow = workflow flyte_launch_plan.guessed_python_interface = Interface( inputs=TypeEngine.guess_python_types(flyte_launch_plan.interface.inputs), outputs=TypeEngine.guess_python_types(flyte_launch_plan.interface.outputs), @@ -1053,6 +1054,7 @@ def sync_workflow_execution( if execution.spec.launch_plan.resource_type == ResourceType.TASK: # This condition is only true for single-task executions flyte_entity = self.fetch_task(lp_id.project, lp_id.domain, lp_id.name, lp_id.version) + node_interface = flyte_entity.interface if sync_nodes: # Need to construct the mapping. There should've been returned exactly three nodes, a start, # an end, and a task node. @@ -1080,10 +1082,10 @@ def sync_workflow_execution( ) else: # This is the default case, an execution of a normal workflow through a launch plan - wf_id = self.fetch_launch_plan(lp_id.project, lp_id.domain, lp_id.name, lp_id.version).workflow_id - flyte_entity = self.fetch_workflow(wf_id.project, wf_id.domain, wf_id.name, wf_id.version) - execution._flyte_workflow = flyte_entity - node_mapping = flyte_entity._node_map + fetched_lp = self.fetch_launch_plan(lp_id.project, lp_id.domain, lp_id.name, lp_id.version) + node_interface = fetched_lp.flyte_workflow.interface + execution._flyte_workflow = fetched_lp.flyte_workflow + node_mapping = fetched_lp.flyte_workflow._node_map # update node executions (if requested), and inputs/outputs if sync_nodes: @@ -1091,10 +1093,12 @@ def sync_workflow_execution( for n in underlying_node_executions: node_execs[n.id.node_id] = self.sync_node_execution(n, node_mapping) execution._node_executions = node_execs - return self._assign_inputs_and_outputs(execution, execution_data, flyte_entity.interface) + return self._assign_inputs_and_outputs(execution, execution_data, node_interface) def sync_node_execution( - self, execution: FlyteNodeExecution, node_mapping: typing.Dict[str, FlyteNode] + self, + execution: FlyteNodeExecution, + node_mapping: typing.Dict[str, FlyteNode], ) -> FlyteNodeExecution: """ Get data backing a node execution. These FlyteNodeExecution objects should've come from Admin with the model @@ -1190,7 +1194,9 @@ def sync_node_execution( dynamic_flyte_wf = FlyteWorkflow.promote_from_closure(compiled_wf, node_launch_plans) execution._underlying_node_executions = [ - self.sync_node_execution(FlyteNodeExecution.promote_from_model(cne), dynamic_flyte_wf._node_map) + self.sync_node_execution( + FlyteNodeExecution.promote_from_model(cne), dynamic_flyte_wf._node_map + ) for cne in child_node_executions ] # This is copied from below - dynamic tasks have both task executions (executions of the parent @@ -1214,14 +1220,12 @@ def sync_node_execution( # Handle the case where it's a branch node elif execution._node.branch_node is not None: - print("hi") + remote_logger.debug(f"Skipping for now") + return execution else: - remote_logger.error( - f"NE {execution} undeterminable, {type(execution._node)}, {execution._node}" - ) + remote_logger.error(f"NE {execution} undeterminable, {type(execution._node)}, {execution._node}") raise Exception(f"Node execution undeterminable, entity has type {type(execution._node)}") - # This is the plain ol' task execution case else: execution._task_executions = [ diff --git a/tests/flytekit/unit/remote/test_remote.py b/tests/flytekit/unit/remote/test_remote.py index 6c4f3e0a4a..45218e2b3f 100644 --- a/tests/flytekit/unit/remote/test_remote.py +++ b/tests/flytekit/unit/remote/test_remote.py @@ -184,16 +184,15 @@ def test_explicit_grpc_channel_credentials(mock_insecure, mock_url, mock_secure_ def test_vjkl(): rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") - wf = rr.fetch_workflow( - name="core.control_flow.run_merge_sort.merge_sort", version="v0.3.39" - ) + wf = rr.fetch_workflow(name="core.control_flow.run_merge_sort.merge_sort", version="v0.3.39") def test_jasilv(): rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") - wf = rr.fetch_workflow( - name="core.control_flow.run_conditions.nested_conditions", version="v0.3.39" - ) - print(wf) + # wf = rr.fetch_workflow( + # name="core.control_flow.run_conditions.nested_conditions", version="v0.3.39" + # ) + # print(wf) we = rr.fetch_workflow_execution(name="suss7xonol") rr.sync_workflow_execution(we, sync_nodes=True) + print(rr) From 196e7cfc78d8b0eeb4df87ff58cf3a7cee48413f Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Mon, 14 Feb 2022 15:37:07 -0800 Subject: [PATCH 08/13] nit Signed-off-by: Yee Hing Tong --- flytekit/remote/nodes.py | 1 - tests/flytekit/unit/remote/test_remote.py | 28 +++++++++++------------ 2 files changed, 14 insertions(+), 15 deletions(-) diff --git a/flytekit/remote/nodes.py b/flytekit/remote/nodes.py index 7f04185f4e..e7ee0dc487 100644 --- a/flytekit/remote/nodes.py +++ b/flytekit/remote/nodes.py @@ -91,7 +91,6 @@ def promote_from_model( node_launch_plans, tasks, ) - # TODO: Implement branch node https://github.com/flyteorg/flyte/issues/1116 elif model.branch_node is not None: flyte_branch_node = _component_nodes.FlyteBranchNode.promote_from_model( model.branch_node, sub_workflows, node_launch_plans, tasks diff --git a/tests/flytekit/unit/remote/test_remote.py b/tests/flytekit/unit/remote/test_remote.py index 45218e2b3f..1688612437 100644 --- a/tests/flytekit/unit/remote/test_remote.py +++ b/tests/flytekit/unit/remote/test_remote.py @@ -182,17 +182,17 @@ def test_explicit_grpc_channel_credentials(mock_insecure, mock_url, mock_secure_ assert mock_ssl_channel_credentials.call_count == 1 -def test_vjkl(): - rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") - wf = rr.fetch_workflow(name="core.control_flow.run_merge_sort.merge_sort", version="v0.3.39") - - -def test_jasilv(): - rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") - # wf = rr.fetch_workflow( - # name="core.control_flow.run_conditions.nested_conditions", version="v0.3.39" - # ) - # print(wf) - we = rr.fetch_workflow_execution(name="suss7xonol") - rr.sync_workflow_execution(we, sync_nodes=True) - print(rr) +# def test_vjkl(): +# rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") +# wf = rr.fetch_workflow(name="core.control_flow.run_merge_sort.merge_sort", version="v0.3.39") +# +# +# def test_jasilv(): +# rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") +# # wf = rr.fetch_workflow( +# # name="core.control_flow.run_conditions.nested_conditions", version="v0.3.39" +# # ) +# # print(wf) +# we = rr.fetch_workflow_execution(name="suss7xonol") +# rr.sync_workflow_execution(we, sync_nodes=True) +# print(rr) From 7d87723a788e29d5d75d578eba9a8c229c9f6d9b Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Mon, 14 Feb 2022 15:45:10 -0800 Subject: [PATCH 09/13] empty From 0b8e1f5ace0e9987013b0a1978bd8d569ae21281 Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Mon, 20 Aug 2018 20:19:25 +0100 Subject: [PATCH 10/13] empty From 2e22ede75d958b21fd37a46a10c295bc7322492e Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Mon, 20 Aug 2018 20:19:25 +0100 Subject: [PATCH 11/13] nit Signed-off-by: Yee Hing Tong --- flytekit/remote/remote.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/flytekit/remote/remote.py b/flytekit/remote/remote.py index 526e734d8a..9e76019012 100644 --- a/flytekit/remote/remote.py +++ b/flytekit/remote/remote.py @@ -1194,9 +1194,7 @@ def sync_node_execution( dynamic_flyte_wf = FlyteWorkflow.promote_from_closure(compiled_wf, node_launch_plans) execution._underlying_node_executions = [ - self.sync_node_execution( - FlyteNodeExecution.promote_from_model(cne), dynamic_flyte_wf._node_map - ) + self.sync_node_execution(FlyteNodeExecution.promote_from_model(cne), dynamic_flyte_wf._node_map) for cne in child_node_executions ] # This is copied from below - dynamic tasks have both task executions (executions of the parent From 315323106d6e9b294c035331a2321cb97b3b5ae4 Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Tue, 15 Feb 2022 18:32:22 -0800 Subject: [PATCH 12/13] nit Signed-off-by: Yee Hing Tong --- flytekit/remote/remote.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flytekit/remote/remote.py b/flytekit/remote/remote.py index 9e76019012..50d29e8e23 100644 --- a/flytekit/remote/remote.py +++ b/flytekit/remote/remote.py @@ -1218,7 +1218,7 @@ def sync_node_execution( # Handle the case where it's a branch node elif execution._node.branch_node is not None: - remote_logger.debug(f"Skipping for now") + remote_logger.debug("Skipping remote node execution for now") return execution else: remote_logger.error(f"NE {execution} undeterminable, {type(execution._node)}, {execution._node}") From 6b2e868d050c8ceba3974391d8fa3d43757a31de Mon Sep 17 00:00:00 2001 From: Yee Hing Tong Date: Thu, 17 Feb 2022 16:41:02 -0800 Subject: [PATCH 13/13] nit Signed-off-by: Yee Hing Tong --- tests/flytekit/unit/remote/test_remote.py | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/tests/flytekit/unit/remote/test_remote.py b/tests/flytekit/unit/remote/test_remote.py index 1688612437..cd80d166e2 100644 --- a/tests/flytekit/unit/remote/test_remote.py +++ b/tests/flytekit/unit/remote/test_remote.py @@ -180,19 +180,3 @@ def test_explicit_grpc_channel_credentials(mock_insecure, mock_url, mock_secure_ assert mock_secure_channel.called assert mock_secure_channel.call_args[0][1] == credentials assert mock_ssl_channel_credentials.call_count == 1 - - -# def test_vjkl(): -# rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") -# wf = rr.fetch_workflow(name="core.control_flow.run_merge_sort.merge_sort", version="v0.3.39") -# -# -# def test_jasilv(): -# rr = FlyteRemote.from_config("flytesnacks", "development", config_file_path="/Users/ytong/.flyte/local_sandbox") -# # wf = rr.fetch_workflow( -# # name="core.control_flow.run_conditions.nested_conditions", version="v0.3.39" -# # ) -# # print(wf) -# we = rr.fetch_workflow_execution(name="suss7xonol") -# rr.sync_workflow_execution(we, sync_nodes=True) -# print(rr)