Add error capturing and logging for job/task process failures

Author: JosepSampe

CHANGELOG.md

@@ -3,7 +3,7 @@
 ## [v3.4.1.dev0]
 
 ### Added
-- 
+- [Localhost] Added error capturing and logging for job/task process failures
 
 ### Changed
 - 
@@ -18,13 +18,9 @@
 - [Ceph] Added extra region parameter to Ceph backend
 
 ### Changed
-- [Setup] Moved IBM dependencies to lithops[ibm] extra
-- [Setup] Moved AWS dependencies to lithops[aws] extra
-- [Setup] Moved kubernetes dependencies to lithops[kubernetes] extra
-- [Setup] Moved knative dependencies to lithops[knative] extra
-- [Setup] Moved minio dependencies to lithops[minio] extra
-- [Setup] Moved ceph dependencies to lithops[ceph] extra
-- [Setup] Moved redis dependencies to lithops[redis] extra
+- [Setup] Moved IBM and AWS deps to lithops[ibm] and lithops[aws] extra
+- [Setup] Moved kubernetes and knative deps to lithops[kubernetes] and lithops[knative] extra
+- [Setup] Moved minio, ceph and redis deps to lithops[minio], lithops[ceph] and lithops[redis] extra
 - [Setup] Moved matplotlib, seaborn, numpy and pandas dependencies to lithops[plotting] extra
 - [Setup] Removed unused 'lxml', 'docker' and 'python-dateutil' packages from the setup.py
 - [Core] Detached progress bar from INFO logs

lithops/localhost/v1/localhost.py

@@ -114,7 +114,10 @@ def job_manager():
                 logger.debug(f'ExecutorID {executor_id} | JobID {job_id} - Running '
                              f'{total_calls} activations in the localhost worker')
                 process = self.env.run_job(job_key, job_filename)
-                process.communicate()  # blocks until the process finishes
+                stdout, stderr = process.communicate()  # blocks until the process finishes
+                if process.returncode != 0:
+                    logger.error(f"ExecutorID {executor_id} | JobID {job_id} - Job failed with return code {process.returncode}")
+                    logger.error(f"ExecutorID {executor_id} | JobID {job_id} - Error output from job process: {stderr}")
                 logger.debug(f'ExecutorID {executor_id} | JobID {job_id} - Execution finished')
 
                 if self.job_queue.empty():
@@ -301,7 +304,7 @@ def run_job(self, job_key, job_filename):
             self.setup()
 
         cmd = [self.runtime_name, RUNNER_FILE, 'run_job', job_filename]
-        process = sp.Popen(cmd, start_new_session=True)
+        process = sp.Popen(cmd, stdout=sp.PIPE, stderr=sp.PIPE, start_new_session=True)
         self.jobs[job_key] = process
 
         return process
@@ -369,7 +372,7 @@ def run_job(self, job_key, job_filename):
         cmd += f'--rm -v {tmp_path}:/tmp --entrypoint "python3" '
         cmd += f'{self.runtime_name} /tmp/{USER_TEMP_DIR}/localhost-runner.py run_job {job_filename}'
 
-        process = sp.Popen(shlex.split(cmd), start_new_session=True)
+        process = sp.Popen(shlex.split(cmd), stdout=sp.PIPE, stderr=sp.PIPE, start_new_session=True)
         self.jobs[job_key] = process
 
         return process

lithops/localhost/v2/localhost.py

@@ -321,9 +321,12 @@ def run_task(self, job_key, call_id):
 
         logger.debug(f"Going to execute task process {job_key_call_id}")
         cmd = [self.runtime_name, RUNNER_FILE, 'run_job', task_filename]
-        process = sp.Popen(cmd, start_new_session=True)
+        process = sp.Popen(cmd, stdout=sp.PIPE, stderr=sp.PIPE, start_new_session=True)
         self.task_processes[job_key_call_id] = process
-        process.communicate()  # blocks until the process finishes
+        stdout, stderr = process.communicate()  # blocks until the process finishes
+        if process.returncode != 0:
+            logger.error(f"Task process {job_key_call_id} failed with return code {process.returncode}")
+            logger.error(f"Error output from task process {job_key_call_id}: {stderr}")
         del self.task_processes[job_key_call_id]
         logger.debug(f"Task process {job_key_call_id} finished")
 
@@ -434,9 +437,12 @@ def run_task(self, job_key, call_id):
         cmd += f'"python3 /tmp/{USER_TEMP_DIR}/localhost-runner.py '
         cmd += f'run_job {docker_task_filename}"'
 
-        process = sp.Popen(shlex.split(cmd), start_new_session=True)
+        process = sp.Popen(shlex.split(cmd), stdout=sp.PIPE, stderr=sp.PIPE, start_new_session=True)
         self.task_processes[job_key_call_id] = process
-        process.communicate()  # blocks until the process finishes
+        stdout, stderr = process.communicate()  # blocks until the process finishes
+        if process.returncode != 0:
+            logger.error(f"Task process {job_key_call_id} failed with return code {process.returncode}")
+            logger.error(f"Error output from task process {job_key_call_id}: {stderr}")
         del self.task_processes[job_key_call_id]
         logger.debug(f"Task process {job_key_call_id} finished")