Repository: microsoft/promptflow
Pull request: [PF run] Show error msg when pf.run failed
# Description
warning log
```
[2023-10-12 13:43:07,809][promptflow._sdk.operations._run_submitter][WARNING] - 1 out of 1 runs failed in bulk run. Please check out C:/Users/zhrua/.promptflow/.runs/failed_flow_variant_0_20231012_134245_423704 for more details.
```
Please add an informative description that covers that changes made by the pull request and link all relevant issues.

# All Promptflow Contribution checklist:
- [ ] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other significant changes.**
- [ ] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated review from promptflow team. Learn more: [suggested workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [ ] Title of the pull request is clear and informative.
- [ ] There are a small number of commits, each of which have an informative message. This means that previously merged commits do not appear in the history of the PR. For more information on cleaning up the commits in your PR, [see this page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [ ] Pull request includes test coverage for the included changes.


Problem statement:
[BUG] PFClient fails silently when there's an error running the flow
**Describe the bug**
A clear and concise description of the bug.

When the flow code raises an exception, the exception doesn't throw through PFClient.run(). It makes me think the run succeeded but it actually didn't. I have to look at `error.json` to know what went wrong.


**How To Reproduce the bug**
Steps to reproduce the behavior, how frequent can you experience the bug:
1. Manually raise an exception in a flow's code.
2. Run the flow with PFClient.run()
3. Observe the call succeeded without raising any exceptions.

**Expected behavior**
A clear and concise description of what you expected to happen.

An exception raised from PFClient.run() or it prints out the error messages.

**Screenshots**
If applicable, add screenshots to help explain your problem.

**Running Information(please complete the following information):**
 - Promptflow Package Version using `pf -v`: [e.g. 0.0.102309906]
 - Operating System: [e.g. Ubuntu 20.04, Windows 11]
 - Python Version using `python --version`: [e.g. python==3.10.12]

**Additional context**
Add any other context about the problem here.

Hints from issue discussion:
There is a bug with the flow run status, we will fix it and print an error log to alert customer that the flow run failed

Commit message: fix failed test case

Patch to review:
diff --git a/src/promptflow/promptflow/_sdk/operations/_run_submitter.py b/src/promptflow/promptflow/_sdk/operations/_run_submitter.py
index 7096ae16343..2b80330e044 100644
--- a/src/promptflow/promptflow/_sdk/operations/_run_submitter.py
+++ b/src/promptflow/promptflow/_sdk/operations/_run_submitter.py
@@ -25,6 +25,7 @@
     USE_VARIANTS,
     VARIANTS,
     ConnectionFields,
+    FlowRunProperties,
 )
 from promptflow._sdk._errors import InvalidFlowError
 from promptflow._sdk._load_functions import load_flow
@@ -295,16 +296,27 @@ def _submit_bulk_run(self, flow: Flow, run: Run, local_storage: LocalStorageOper
         run._dump()  # pylint: disable=protected-access
         try:
             bulk_result = flow_executor.exec_bulk(mapped_inputs, run_id=run_id)
+            # Filter the failed line result
+            failed_line_result = next(
+                filter(lambda result: result.run_info.status == Status.Failed, bulk_result.line_results), None)
+            if failed_line_result:
+                raise UserErrorException(**failed_line_result.run_info.error)
             # The bulk run is completed if the exec_bulk successfully completed.
             status = Status.Completed.value
         except Exception as e:
             # when run failed in executor, store the exception in result and dump to file
-            logger.warning(f"Run {run.name} failed when executing in executor.")
+            error_log = f"Run {run.name} failed when executing in executor."
+            if run.properties.get(FlowRunProperties.OUTPUT_PATH, None):
+                error_log = error_log + \
+                            f" Please check out {run.properties[FlowRunProperties.OUTPUT_PATH]} for more details."
+            logger.warning(error_log)
             exception = e
             # for user error, swallow stack trace and return failed run since user don't need the stack trace
             if not isinstance(e, UserErrorException):
                 # for other errors, raise it to user to help debug root cause.
                 raise e
+            else:
+                logger.warning(e.message)
             # won't raise the exception since it's already included in run object.
         finally:
             # persist snapshot and result
diff --git a/src/promptflow/tests/sdk_cli_test/e2etests/test_flow_run.py b/src/promptflow/tests/sdk_cli_test/e2etests/test_flow_run.py
index e53b7d4e9c8..027d8d4c9b6 100644
--- a/src/promptflow/tests/sdk_cli_test/e2etests/test_flow_run.py
+++ b/src/promptflow/tests/sdk_cli_test/e2etests/test_flow_run.py
@@ -279,11 +279,7 @@ def test_run_reference_failed_run(self, pf):
             data=f"{DATAS_DIR}/webClassification1.jsonl",
             column_mapping={"text": "${data.url}"},
         )
-        # "update" run status to failed since currently all run will be completed unless there's bug
-        pf.runs.update(
-            name=failed_run.name,
-            status="Failed",
-        )
+        assert failed_run.status == "Failed"
 
         run_name = str(uuid.uuid4())
         with pytest.raises(ValueError) as e:
@@ -300,9 +296,8 @@ def test_run_reference_failed_run(self, pf):
             pf.runs.get(name=run_name)
 
     def test_referenced_output_not_exist(self, pf):
-        # failed run won't generate output
         failed_run = pf.run(
-            flow=f"{FLOWS_DIR}/failed_flow",
+            flow=f"{FLOWS_DIR}/web_classification",
             data=f"{DATAS_DIR}/webClassification1.jsonl",
             column_mapping={"text": "${data.url}"},
         )
@@ -533,11 +528,7 @@ def test_incomplete_run_visualize(
             data=f"{DATAS_DIR}/webClassification1.jsonl",
             column_mapping={"text": "${data.url}"},
         )
-        # "update" run status to failed since currently all run will be completed unless there's bug
-        pf.runs.update(
-            name=failed_run.name,
-            status="Failed",
-        )
+        assert failed_run.status == "Failed"
 
         # patch logger.error to print, so that we can capture the error message using capfd
         from promptflow.azure.operations import _run_operations
@@ -707,8 +698,7 @@ def test_error_message_dump(self, pf):
             data=f"{DATAS_DIR}/webClassification1.jsonl",
             column_mapping={"text": "${data.url}"},
         )
-        # even if all lines failed, the bulk run's status is completed.
-        assert failed_run.status == "Completed"
+        assert failed_run.status == "Failed"
         # error messages will store in local
         local_storage = LocalStorageOperations(failed_run)
 


Code review comments:
diff_hunk: @@ -295,16 +296,27 @@ def _submit_bulk_run(self, flow: Flow, run: Run, local_storage: LocalStorageOper
         run._dump()  # pylint: disable=protected-access
         try:
             bulk_result = flow_executor.exec_bulk(mapped_inputs, run_id=run_id)
+            # Filter the failed line result
+            failed_line_result = next(
+                filter(lambda result: result.run_info.status == Status.Failed, bulk_result.line_results), None)
+            if failed_line_result:
+                raise UserErrorException(**failed_line_result.run_info.error)
path: src/promptflow/promptflow/_sdk/operations/_run_submitter.py
text: @user1:
This is a big change which means line run failure means whole batch run failure. Is this aligned with cloud run experience?

@author:
revert it, only log warning msg and not change batch run status to failed