Expose error due to failure of local python pipeline node execution

elyra/pipeline/processor_local.py

@@ -21,7 +21,7 @@
 from elyra.pipeline import PipelineProcessor, PipelineProcessorResponse, Operation
 from elyra.util.path import get_absolute_path
 from notebook.gateway.managers import GatewayClient
-from subprocess import run
+from subprocess import run, CalledProcessError, PIPE
 from traitlets import log
 from typing import Dict
 
@@ -76,7 +76,7 @@ def process(self, pipeline):
                                        operation_name=operation.name,
                                        duration=(time.time() - t0))
             except Exception as ex:
-                raise RuntimeError(f'Error processing operation {operation.name}.') from ex
+                raise RuntimeError(f'Error processing operation {operation.name} {str(ex)}') from ex
 
         self.log_pipeline_info(pipeline.name, "pipeline processed", duration=(time.time() - t0_all))
 
@@ -205,8 +205,14 @@ def process(self, operation: Operation):
                 filepath,
                 **additional_kwargs
             )
+        except papermill.PapermillExecutionError as pmee:
+            self.log.error(f'Error executing {file_name} in cell {pmee.exec_count}: ' +
+                           f'{str(pmee.ename)} {str(pmee.evalue)}')
+            raise RuntimeError(f'({file_name}) in cell {pmee.exec_count}: ' +
+                               f'{str(pmee.ename)} {str(pmee.evalue)}') from pmee
         except Exception as ex:
-            raise RuntimeError(f'Internal error executing {filepath}: {ex}') from ex
+            self.log.error(f'Error executing {file_name}: {str(ex)}')
+            raise RuntimeError(f'({file_name})') from ex
 
         t1 = time.time()
         duration = (t1 - t0)
@@ -233,9 +239,19 @@ def process(self, operation: Operation):
         envs.update(operation.env_vars_as_dict())
         t0 = time.time()
         try:
-            run(argv, cwd=file_dir, env=envs, check=True)
+            run(argv, cwd=file_dir, env=envs, check=True, stderr=PIPE)
+        except CalledProcessError as cpe:
+            error_msg = str(cpe.stderr.decode())
+            self.log.error(f'Error executing {file_name}: {error_msg}')
+
+            error_trim_index = error_msg.rfind('\n', 0, error_msg.rfind('Error'))
+            if error_trim_index != -1:
+                raise RuntimeError(f'({file_name}): {error_msg[error_trim_index:].strip()}') from cpe
+            else:
+                raise RuntimeError(f'({file_name})') from cpe
         except Exception as ex:
-            raise RuntimeError(f'Internal error executing {filepath}: {ex}') from ex
+            self.log.error(f'Error executing {file_name}: {str(ex)}')
+            raise RuntimeError(f'({file_name})') from ex
 
         t1 = time.time()
         duration = (t1 - t0)

elyra/pipeline/tests/test_pipeline_processor_local.py

@@ -107,7 +107,7 @@ def test_pipeline_execution_bad_notebook(pipeline_dir):
 
     with pytest.raises(RuntimeError) as e:
         LocalPipelineProcessor(pipeline_dir).process(pipeline)
-    assert str(e.value) == 'Error processing operation node3.'
+    assert 'Error processing operation node3' in str(e.value)
 
     # Confirm outputs (and non-outputs)
     for node in processed_nodes:
@@ -133,7 +133,7 @@ def test_pipeline_execution_bad_python(pipeline_dir):
 
     with pytest.raises(RuntimeError) as e:
         LocalPipelineProcessor(pipeline_dir).process(pipeline)
-    assert str(e.value) == 'Error processing operation node2.'
+    assert 'Error processing operation node2' in str(e.value)
 
     # Confirm outputs (and non-outputs)
     for node in processed_nodes: