Added exception handling to both api handlers and UI (#302)

* Added exception handling to both api handlers, and UI

* Ran lint

* keep network error state local to detail-view component

* fix lint

Co-authored-by: David L. Qiu <[email protected]>

Author: 3coins

CHANGELOG.md

@@ -12,7 +12,7 @@
 
 ### Bugs fixed
 
-- Fixed propagation of error message to UI  [#299](https://github.com/jupyter-server/jupyter-scheduler/pull/299) ([@3coins](https://github.com/3coins))
+- Fixed propagation of error message to UI [#299](https://github.com/jupyter-server/jupyter-scheduler/pull/299) ([@3coins](https://github.com/3coins))
 - correctly handle last page even when latest next_token is truthy [#292](https://github.com/jupyter-server/jupyter-scheduler/pull/292) ([@dlqqq](https://github.com/dlqqq))
 
 ### Contributors to this release

jupyter_scheduler/handlers.py

@@ -76,10 +76,15 @@ class JobDefinitionHandler(ExtensionHandlerMixin, JobHandlersMixin, APIHandler):
     @tornado.web.authenticated
     async def get(self, job_definition_id=None):
         if job_definition_id:
-            job_definition = await ensure_async(
-                self.scheduler.get_job_definition(job_definition_id)
-            )
-            self.finish(job_definition.json())
+            try:
+                job_definition = await ensure_async(
+                    self.scheduler.get_job_definition(job_definition_id)
+                )
+            except SchedulerError as e:
+                self.log.exception(e)
+                raise tornado.web.HTTPError(500, str(e)) from e
+            else:
+                self.finish(job_definition.json())
         else:
             create_time = self.get_query_argument("create_time", None)
             sort_by = compute_sort_model(self.get_query_arguments("sort_by"))
@@ -92,8 +97,13 @@ async def get(self, job_definition_id=None):
                 max_items=self.get_query_argument("max_items", DEFAULT_MAX_ITEMS),
                 next_token=self.get_query_argument("next_token", None),
             )
-            list_response = await ensure_async(self.scheduler.list_job_definitions(list_query))
-            self.finish(list_response.json(exclude_none=True))
+            try:
+                list_response = await ensure_async(self.scheduler.list_job_definitions(list_query))
+            except SchedulerError as e:
+                self.log.exception(e)
+                raise tornado.web.HTTPError(500, str(e)) from e
+            else:
+                self.finish(list_response.json(exclude_none=True))
 
     @tornado.web.authenticated
     async def post(self):
@@ -119,25 +129,42 @@ async def post(self):
     @tornado.web.authenticated
     async def patch(self, job_definition_id):
         payload = self.get_json_body()
-        await ensure_async(
-            self.scheduler.update_job_definition(job_definition_id, UpdateJobDefinition(**payload))
-        )
-        self.set_status(204)
-        self.finish()
+        try:
+            await ensure_async(
+                self.scheduler.update_job_definition(
+                    job_definition_id, UpdateJobDefinition(**payload)
+                )
+            )
+        except SchedulerError as e:
+            self.log.exception(e)
+            raise tornado.web.HTTPError(500, str(e)) from e
+        else:
+            self.set_status(204)
+            self.finish()
 
     @tornado.web.authenticated
     async def delete(self, job_definition_id):
-        await ensure_async(self.scheduler.delete_job_definition(job_definition_id))
-        self.set_status(204)
-        self.finish()
+        try:
+            await ensure_async(self.scheduler.delete_job_definition(job_definition_id))
+        except SchedulerError as e:
+            self.log.exception(e)
+            raise tornado.web.HTTPError(500, str(e)) from e
+        else:
+            self.set_status(204)
+            self.finish()
 
 
 class JobHandler(ExtensionHandlerMixin, JobHandlersMixin, APIHandler):
     @tornado.web.authenticated
     async def get(self, job_id=None):
         if job_id:
-            job = await ensure_async(self.scheduler.get_job(job_id))
-            self.finish(job.json())
+            try:
+                job = await ensure_async(self.scheduler.get_job(job_id))
+            except SchedulerError as e:
+                self.log.exception(e)
+                raise tornado.web.HTTPError(500, str(e)) from e
+            else:
+                self.finish(job.json())
         else:
             status = self.get_query_argument("status", None)
             start_time = self.get_query_argument("start_time", None)
@@ -152,9 +179,13 @@ async def get(self, job_id=None):
                 max_items=self.get_query_argument("max_items", DEFAULT_MAX_ITEMS),
                 next_token=self.get_query_argument("next_token", None),
             )
-            list_jobs_response = await ensure_async(self.scheduler.list_jobs(list_jobs_query))
-
-            self.finish(list_jobs_response.json(exclude_none=True))
+            try:
+                list_jobs_response = await ensure_async(self.scheduler.list_jobs(list_jobs_query))
+            except SchedulerError as e:
+                self.log.exception(e)
+                raise tornado.web.HTTPError(500, str(e)) from e
+            else:
+                self.finish(list_jobs_response.json(exclude_none=True))
 
     @tornado.web.authenticated
     async def post(self):
@@ -191,20 +222,28 @@ async def patch(self, job_id):
                 "Invalid value for field 'status'. Jobs can only be updated to status 'STOPPED' after creation.",
             )
 
-        if status:
-            await ensure_async(self.scheduler.stop_job(job_id))
+        try:
+            if status:
+                await ensure_async(self.scheduler.stop_job(job_id))
+            else:
+                await ensure_async(self.scheduler.update_job(job_id, UpdateJob(**payload)))
+        except SchedulerError as e:
+            self.log.exception(e)
+            raise tornado.web.HTTPError(500, str(e)) from e
         else:
-            await ensure_async(self.scheduler.update_job(job_id, UpdateJob(**payload)))
-
-        self.set_status(204)
-        self.finish()
+            self.set_status(204)
+            self.finish()
 
     @tornado.web.authenticated
     async def delete(self, job_id):
-        await ensure_async(self.scheduler.delete_job(job_id))
-
-        self.set_status(204)
-        self.finish()
+        try:
+            await ensure_async(self.scheduler.delete_job(job_id))
+        except SchedulerError as e:
+            self.log.exception(e)
+            raise tornado.web.HTTPError(500, str(e)) from e
+        else:
+            self.set_status(204)
+            self.finish()
 
 
 class JobFromDefinitionHandler(ExtensionHandlerMixin, JobHandlersMixin, APIHandler):
@@ -232,11 +271,15 @@ class BatchJobHandler(ExtensionHandlerMixin, JobHandlersMixin, APIHandler):
     @tornado.web.authenticated
     async def delete(self):
         job_ids = self.get_query_arguments("job_id")
-        for job_id in job_ids:
-            await ensure_async(self.scheduler.delete_job(job_id))
-
-        self.set_status(204)
-        self.finish()
+        try:
+            for job_id in job_ids:
+                await ensure_async(self.scheduler.delete_job(job_id))
+        except SchedulerError as e:
+            self.log.exception(e)
+            raise tornado.web.HTTPError(500, str(e)) from e
+        else:
+            self.set_status(204)
+            self.finish()
 
 
 class JobsCountHandler(ExtensionHandlerMixin, JobHandlersMixin, APIHandler):
@@ -246,9 +289,13 @@ async def get(self):
         count_jobs_query = CountJobsQuery(
             status=Status(status.upper()) if status else Status.IN_PROGRESS
         )
-        count = await ensure_async(self.scheduler.count_jobs(count_jobs_query))
-
-        self.finish(json.dumps(dict(count=count)))
+        try:
+            count = await ensure_async(self.scheduler.count_jobs(count_jobs_query))
+        except SchedulerError as e:
+            self.log.exception(e)
+            raise tornado.web.HTTPError(500, str(e)) from e
+        else:
+            self.finish(json.dumps(dict(count=count)))
 
 
 class RuntimeEnvironmentsHandler(ExtensionHandlerMixin, JobHandlersMixin, APIHandler):
@@ -304,7 +351,11 @@ def job_files_manager(self):
     @tornado.web.authenticated
     async def get(self, job_id):
         redownload = self.get_query_argument("redownload", False)
-        await self.job_files_manager.copy_from_staging(job_id=job_id, redownload=redownload)
-
-        self.set_status(204)
-        self.finish()
+        try:
+            await self.job_files_manager.copy_from_staging(job_id=job_id, redownload=redownload)
+        except Exception as e:
+            self.log.exception(e)
+            raise tornado.web.HTTPError(500, str(e)) from e
+        else:
+            self.set_status(204)
+            self.finish()

jupyter_scheduler/scheduler.py

@@ -1,4 +1,5 @@
 import os
+import random
 import shutil
 from multiprocessing import Process
 from typing import Dict, Optional, Type, Union
@@ -682,3 +683,99 @@ def get_staging_paths(self, model: Union[DescribeJob, DescribeJobDefinition]) ->
         staging_paths["input"] = os.path.join(self.staging_path, model.job_id, model.input_filename)
 
         return staging_paths
+
+
+class SchedulerWithErrors(Scheduler):
+    """
+    Use only for testing exceptions, not to be used in production
+
+    This scheduler uses the default `Scheduler`, but randomly
+    raises the `SchedulerError` to help view/test errors in the
+    UI. To use this, specify the fully classified class name at
+    at start up or add to one of the server config files.
+
+    Usage
+    -----
+    >> jupyter lab --SchedulerApp.scheduler_class=jupyter_scheduler.scheduler.SchedulerWithErrors
+    """
+
+    def _should_raise_error(self, probability=0.5):
+        return random.random() < probability
+
+    def create_job(self, model: CreateJob) -> str:
+        if self._should_raise_error():
+            raise SchedulerError("Failed create job because of a deliberate exception.")
+        else:
+            return super().create_job(model)
+
+    def update_job(self, job_id: str, model: UpdateJob):
+        if self._should_raise_error():
+            raise SchedulerError("Failed update job because of a deliberate exception.")
+        else:
+            super().update_job(job_id, model)
+
+    def list_jobs(self, query: ListJobsQuery) -> ListJobsResponse:
+        if self._should_raise_error():
+            raise SchedulerError("Failed list jobs because of a deliberate exception.")
+        else:
+            return super().list_jobs(query)
+
+    def count_jobs(self, query: CountJobsQuery) -> int:
+        if self._should_raise_error():
+            raise SchedulerError("Failed count jobs because of a deliberate exception.")
+        else:
+            return super().count_jobs(query)
+
+    def get_job(self, job_id: str, job_files: Optional[bool] = True) -> DescribeJob:
+        if self._should_raise_error():
+            raise SchedulerError("Failed get job because of a deliberate exception.")
+        else:
+            return super().get_job(job_id, job_files)
+
+    def delete_job(self, job_id: str):
+        if self._should_raise_error():
+            raise SchedulerError("Failed delete job because of a deliberate exception.")
+        else:
+            super().delete_job(job_id)
+
+    def stop_job(self, job_id: str):
+        if self._should_raise_error():
+            raise SchedulerError("Failed stop job because of a deliberate exception.")
+        else:
+            super().stop_job(job_id)
+
+    def create_job_definition(self, model: CreateJobDefinition) -> str:
+        if self._should_raise_error():
+            raise SchedulerError("Failed create job definition because of a deliberate exception.")
+        else:
+            return super().create_job_definition(model)
+
+    def update_job_definition(self, job_definition_id: str, model: UpdateJobDefinition):
+        if self._should_raise_error():
+            raise SchedulerError("Failed update job definition because of a deliberate exception.")
+        else:
+            super().update_job_definition(job_definition_id, model)
+
+    def delete_job_definition(self, job_definition_id: str):
+        if self._should_raise_error():
+            raise SchedulerError("Failed delete job definition because of a deliberate exception.")
+        else:
+            super().delete_job_definition(job_definition_id)
+
+    def get_job_definition(self, job_definition_id: str) -> DescribeJobDefinition:
+        if self._should_raise_error():
+            raise SchedulerError("Failed get job definition because of a deliberate exception.")
+        else:
+            return super().get_job_definition(job_definition_id)
+
+    def list_job_definitions(self, query: ListJobDefinitionsQuery) -> ListJobDefinitionsResponse:
+        if self._should_raise_error():
+            raise SchedulerError("Failed list job definitions because of a deliberate exception.")
+        else:
+            return super().list_job_definitions(query)
+
+    def create_job_from_definition(self, job_definition_id: str, model: CreateJobFromDefinition):
+        if self._should_raise_error():
+            raise SchedulerError("Failed list jobs because of a deliberate exception.")
+        else:
+            return super().create_job_from_definition(job_definition_id, model)