feat(config): set maximum limit for number of Dask workers (reanahub#712

)
Alputer · Nov 11, 2024 · 4347c90 · 4347c90
1 parent 15d0b97
commit 4347c90
Show file tree

Hide file tree

Showing 4 changed files with 44 additions and 0 deletions.
diff --git a/docs/openapi.json b/docs/openapi.json
@@ -441,6 +441,10 @@
                   "title": "The maximum memory limit for Dask clusters created by users",
                   "value": "16Gi"
                 },
+                "dask_cluster_max_number_of_workers": {
+                  "title": "The maximum number of workers that users can ask for the single Dask cluster",
+                  "value": "20"
+                },
                 "dask_cluster_max_single_worker_memory": {
                   "title": "The maximum amount of memory that users can ask for the single Dask worker",
                   "value": "8Gi"
@@ -532,6 +536,17 @@
                   },
                   "type": "object"
                 },
+                "dask_cluster_max_number_of_workers": {
+                  "properties": {
+                    "title": {
+                      "type": "string"
+                    },
+                    "value": {
+                      "type": "string"
+                    }
+                  },
+                  "type": "object"
+                },
                 "dask_cluster_max_single_worker_memory": {
                   "properties": {
                     "title": {

diff --git a/reana_server/config.py b/reana_server/config.py
@@ -71,6 +71,11 @@
 )
 """Number of workers in Dask cluster by default """
 
+REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS = int(
+    os.getenv("REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS", 20)
+)
+"""Maximum number of workers in Dask cluster."""
+
 REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY = os.getenv(
     "REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY", "2Gi"
 )

diff --git a/reana_server/rest/info.py b/reana_server/rest/info.py
@@ -29,6 +29,7 @@
     REANA_DASK_CLUSTER_MAX_MEMORY_LIMIT,
     REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
     REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_MEMORY,
+    REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS,
 )
 from reana_server.decorators import signin_required
 
@@ -165,6 +166,13 @@ def info(user, **kwargs):  # noqa
                   value:
                     type: string
                 type: object
+              dask_cluster_max_number_of_workers:
+                properties:
+                  title:
+                    type: string
+                  value:
+                    type: string
+                type: object
             type: object
           examples:
             application/json:
@@ -225,6 +233,10 @@ def info(user, **kwargs):  # noqa
                     "title": "The maximum amount of memory that users can ask for the single Dask worker",
                     "value": "8Gi"
                 },
+                "dask_cluster_max_number_of_workers": {
+                    "title": "The maximum number of workers that users can ask for the single Dask cluster",
+                    "value": "20"
+                },
               }
         500:
           description: >-
@@ -299,6 +311,10 @@ def info(user, **kwargs):  # noqa
                 title="The maximum amount of memory that users can ask for the single Dask worker",
                 value=REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_MEMORY,
             )
+            cluster_information["dask_cluster_max_number_of_workers"] = dict(
+                title="The maximum number of workers that users can ask for the single Dask cluster",
+                value=REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS,
+            )
 
         return InfoSchema().dump(cluster_information)
 
@@ -349,3 +365,4 @@ class InfoSchema(Schema):
         dask_cluster_max_memory_limit = fields.Nested(StringInfoValue)
         dask_cluster_default_single_worker_memory = fields.Nested(StringInfoValue)
         dask_cluster_max_single_worker_memory = fields.Nested(StringInfoValue)
+        dask_cluster_max_number_of_workers = fields.Nested(StringInfoValue)
diff --git a/reana_server/validation.py b/reana_server/validation.py
@@ -26,6 +26,7 @@
     DASK_ENABLED,
     REANA_DASK_CLUSTER_MAX_MEMORY_LIMIT,
     REANA_DASK_CLUSTER_DEFAULT_NUMBER_OF_WORKERS,
+    REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS,
     REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
     REANA_DASK_CLUSTER_MAX_SINGLE_WORKER_MEMORY,
 )
@@ -188,6 +189,12 @@ def validate_dask_memory_and_cores_limits(reana_yaml: Dict) -> None:
                 "number_of_workers", REANA_DASK_CLUSTER_DEFAULT_NUMBER_OF_WORKERS
             )
         )
+
+        if number_of_workers > REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS:
+            raise REANAValidationError(
+                f'The "number_of_workers" provided in the dask resources exceeds the limit ({REANA_DASK_CLUSTER_MAX_NUMBER_OF_WORKERS}).'
+            )
+
         requested_dask_cluster_memory = (
             kubernetes_memory_to_bytes(single_worker_memory) * number_of_workers
         )