feat(dask): make Dask autoscaler configurable (reanahub#702)

Author: Alputer

docs/openapi.json

@@ -429,6 +429,10 @@
                     "slurmcern"
                   ]
                 },
+                "dask_autoscaler_enabled": {
+                  "title": "Dask autoscaler enabled in the cluster",
+                  "value": "False"
+                },
                 "dask_cluster_default_number_of_workers": {
                   "title": "The number of Dask workers created by default",
                   "value": "2Gi"
@@ -499,6 +503,17 @@
                   },
                   "type": "object"
                 },
+                "dask_autoscaler_enabled": {
+                  "properties": {
+                    "title": {
+                      "type": "string"
+                    },
+                    "value": {
+                      "type": "string"
+                    }
+                  },
+                  "type": "object"
+                },
                 "dask_cluster_default_number_of_workers": {
                   "properties": {
                     "title": {

reana_server/config.py

@@ -61,6 +61,9 @@
 DASK_ENABLED = strtobool(os.getenv("DASK_ENABLED", "true"))
 """Whether Dask is enabled in the cluster or not"""
 
+DASK_AUTOSCALER_ENABLED = os.getenv("DASK_AUTOSCALER_ENABLED", "true").lower() == "true"
+"""Whether Dask autoscaler is enabled in the cluster or not"""
+
 REANA_DASK_CLUSTER_MAX_MEMORY_LIMIT = os.getenv(
     "REANA_DASK_CLUSTER_MAX_MEMORY_LIMIT", "16Gi"
 )

reana_server/rest/info.py

@@ -25,6 +25,7 @@
     REANA_KUBERNETES_JOBS_MAX_USER_TIMEOUT_LIMIT,
     REANA_INTERACTIVE_SESSION_MAX_INACTIVITY_PERIOD,
     DASK_ENABLED,
+    DASK_AUTOSCALER_ENABLED,
     REANA_DASK_CLUSTER_DEFAULT_NUMBER_OF_WORKERS,
     REANA_DASK_CLUSTER_MAX_MEMORY_LIMIT,
     REANA_DASK_CLUSTER_DEFAULT_SINGLE_WORKER_MEMORY,
@@ -137,6 +138,13 @@ def info(user, **kwargs):  # noqa
                   value:
                     type: string
                 type: object
+              dask_autoscaler_enabled:
+                properties:
+                  title:
+                    type: string
+                  value:
+                    type: string
+                type: object
               dask_cluster_max_memory_limit:
                 properties:
                   title:
@@ -209,6 +217,10 @@ def info(user, **kwargs):  # noqa
                     "title": "Dask workflows allowed in the cluster",
                     "value": "False"
                 },
+                "dask_autoscaler_enabled": {
+                    "title": "Dask autoscaler enabled in the cluster",
+                    "value": "False"
+                },
                 "dask_cluster_max_memory_limit": {
                     "title": "The maximum memory limit for Dask clusters created by users",
                     "value": "16Gi"
@@ -283,6 +295,10 @@ def info(user, **kwargs):  # noqa
             ),
         )
         if DASK_ENABLED:
+            cluster_information["dask_autoscaler_enabled"] = dict(
+                title="Dask autoscaler enabled in the cluster",
+                value=bool(DASK_AUTOSCALER_ENABLED),
+            )
             cluster_information["dask_cluster_default_number_of_workers"] = dict(
                 title="The number of Dask workers created by default",
                 value=REANA_DASK_CLUSTER_DEFAULT_NUMBER_OF_WORKERS,
@@ -345,6 +361,7 @@ class InfoSchema(Schema):
     kubernetes_max_memory_limit = fields.Nested(StringInfoValue)
     dask_enabled = fields.Nested(StringInfoValue)
     if DASK_ENABLED:
+        dask_autoscaler_enabled = fields.Nested(StringInfoValue)
         dask_cluster_default_number_of_workers = fields.Nested(StringInfoValue)
         dask_cluster_max_memory_limit = fields.Nested(StringInfoValue)
         dask_cluster_default_single_worker_memory = fields.Nested(StringInfoValue)