fix(service): use separate queues for multi core service deployment (#…

…2602)

* rename namespace variable

helm-chart/renku-core/templates/deployment.yaml

@@ -1,4 +1,11 @@
 {{- range $version := .Values.versions }}
+{{- $queuesWithVersion := dict "datasetqueues" (list) "managementqueues" (list) -}}
+{{- range $queue := (split "," $.Values.datasetsWorkerQueues) -}}
+{{- $var := printf "%s.%s" $version.name $queue | append $queuesWithVersion.datasetqueues | set $queuesWithVersion "datasetqueues" -}}
+{{- end }}
+{{- range $queue := (split "," $.Values.managementWorkerQueues) -}}
+{{- $var := printf "%s.%s" $version.name $queue | append $queuesWithVersion.managementqueues | set $queuesWithVersion "managementqueues" -}}
+{{- end }}
 ---
 apiVersion: apps/v1
 kind: Deployment
@@ -69,7 +76,7 @@ spec:
               value: "0"
             - name: REDIS_PASSWORD
               value:
-            - name: WALRUS_NAMESPACE
+            - name: REDIS_NAMESPACE
               value: {{ $version.name }}
             - name: CACHE_DIR
               value: {{ $.Values.cacheDirectory }}
@@ -135,7 +142,7 @@ spec:
               value: "0"
             - name: REDIS_PASSWORD
               value:
-            - name: WALRUS_NAMESPACE
+            - name: REDIS_NAMESPACE
               value: {{ $version.name }}
             - name: RENKU_JWT_TOKEN_SECRET
               value: {{ $.Values.jwtTokenSecret }}
@@ -144,7 +151,7 @@ spec:
             - name: RENKU_SVC_CLEANUP_INTERVAL
               value: {{ $.Values.cleanupInterval | quote }}
             - name: RENKU_SVC_WORKER_QUEUES
-              value: {{ $.Values.datasetsWorkerQueues }}
+              value: {{ join "," $queuesWithVersion.datasetqueues }}
             - name: RENKU_SVC_CLEANUP_TTL_FILES
               value: {{ $.Values.cleanupFilesTTL | quote }}
             - name: RENKU_SVC_CLEANUP_TTL_PROJECTS
@@ -178,14 +185,14 @@ spec:
               value: "0"
             - name: REDIS_PASSWORD
               value:
-            - name: WALRUS_NAMESPACE
+            - name: REDIS_NAMESPACE
               value: {{ $version.name }}
             - name: CACHE_DIR
               value: {{ $.Values.cacheDirectory }}
             - name: RENKU_SVC_CLEANUP_INTERVAL
               value: {{ $.Values.cleanupInterval | quote }}
             - name: RENKU_SVC_WORKER_QUEUES
-              value: {{ $.Values.managementWorkerQueues }}
+              value: {{ join "," $queuesWithVersion.managementqueues }}
             - name: RENKU_SVC_CLEANUP_TTL_FILES
               value: {{ $.Values.cleanupFilesTTL | quote }}
             - name: RENKU_SVC_CLEANUP_TTL_PROJECTS
@@ -219,7 +226,7 @@ spec:
               value: "0"
             - name: REDIS_PASSWORD
               value:
-            - name: WALRUS_NAMESPACE
+            - name: REDIS_NAMESPACE
               value: {{ $version.name }}
             - name: CACHE_DIR
               value: {{ $.Values.cacheDirectory | quote }}

renku/service/.env-example

@@ -4,7 +4,7 @@ REDIS_DATABASE=0
 REDIS_PASSWORD=
 CACHE_HEALTH_CHECK_INTERVAL=60
 RENKU_DOMAIN=
-WALRUS_NAMESPACE=
+REDIS_NAMESPACE=
 
 # Server
 CACHE_DIR=/svc/cache

renku/service/cache/base.py

@@ -23,7 +23,7 @@
 from redis import RedisError
 from walrus import Database
 
-from renku.service.cache.config import REDIS_DATABASE, REDIS_HOST, REDIS_PASSWORD, REDIS_PORT, WALRUS_NAMESPACE
+from renku.service.cache.config import REDIS_DATABASE, REDIS_HOST, REDIS_NAMESPACE, REDIS_PASSWORD, REDIS_PORT
 
 
 class BaseCache:
@@ -40,7 +40,7 @@ class BaseCache:
 
     cache = redis.Redis(**config_)
     model_db = Database(**config_)
-    namespace = WALRUS_NAMESPACE
+    namespace = REDIS_NAMESPACE
 
     def set_record(self, name, key, value):
         """Insert a record to hash set."""

renku/service/cache/config.py

@@ -23,4 +23,4 @@
 REDIS_DATABASE = int(os.getenv("REDIS_DATABASE", 0))
 REDIS_PASSWORD = os.getenv("REDIS_PASSWORD")
 
-WALRUS_NAMESPACE = os.getenv("WALRUS_NAMESPACE")
+REDIS_NAMESPACE = os.getenv("REDIS_NAMESPACE")

renku/service/controllers/api/mixins.py

@@ -29,6 +29,7 @@
 from renku.core.management import RENKU_HOME
 from renku.core.metadata.repository import Repository
 from renku.core.utils.contexts import click_context
+from renku.service.cache.config import REDIS_NAMESPACE
 from renku.service.cache.models.job import Job
 from renku.service.cache.models.project import Project
 from renku.service.cache.models.user import User
@@ -145,7 +146,7 @@ def execute_op(self):
 
             job = self.cache.make_job(self.user, job_data={"ctrl_context": {**self.context, **ctrl_cls}})
 
-            with enqueue_retry(f"delayed.ctrl.{ctrl_cls['renku_ctrl']}") as queue:
+            with enqueue_retry(f"{REDIS_NAMESPACE}.delayed.ctrl.{ctrl_cls['renku_ctrl']}") as queue:
                 queue.enqueue(delayed_ctrl_job, self.context, self.user_data, job.job_id, **ctrl_cls)
 
             return job

renku/service/jobs/queues.py

@@ -21,22 +21,23 @@
 from rq import Queue
 
 from renku.service.cache.base import BaseCache
+from renku.service.cache.config import REDIS_NAMESPACE
 
-CLEANUP_QUEUE_FILES = "cache.cleanup.files"
-CLEANUP_QUEUE_PROJECTS = "cache.cleanup.projects"
+CLEANUP_QUEUE_FILES = f"{REDIS_NAMESPACE}.cache.cleanup.files"
+CLEANUP_QUEUE_PROJECTS = f"{REDIS_NAMESPACE}.cache.cleanup.projects"
 
-DATASETS_JOB_QUEUE = "datasets.jobs"
-MIGRATIONS_JOB_QUEUE = "project.migrations"
-GRAPH_JOB_QUEUE = "graph.jobs"
+DATASETS_JOB_QUEUE = f"{REDIS_NAMESPACE}.datasets.jobs"
+MIGRATIONS_JOB_QUEUE = f"{REDIS_NAMESPACE}.project.migrations"
+GRAPH_JOB_QUEUE = f"{REDIS_NAMESPACE}.graph.jobs"
 
-DELAYED_CTRL_DATASETS_CREATE = "delayed.ctrl.DatasetsCreateCtrl"
-DELAYED_CTRL_DATASETS_ADD = "delayed.ctrl.DatasetsAddFileCtrl"
-DELAYED_CTRL_DATASETS_REMOVE = "delayed.ctrl.DatasetsRemoveCtrl"
-DELAYED_CTRL_DATASETS_IMPORT = "delayed.ctrl.DatasetsImportCtrl"
-DELAYED_CTRL_DATASETS_EDIT = "delayed.ctrl.DatasetsEditCtrl"
-DELAYED_CTRL_DATASETS_UNLINK = "delayed.ctrl.DatasetsUnlinkCtrl"
-DELAYED_CTRL_PROJECT_MIGRATE = "delayed.ctrl.MigrateProjectCtrl"
-DELAYED_CTRL_CONFIG_SET = "delayed.ctrl.SetConfigCtrl"
+DELAYED_CTRL_DATASETS_CREATE = f"{REDIS_NAMESPACE}.delayed.ctrl.DatasetsCreateCtrl"
+DELAYED_CTRL_DATASETS_ADD = f"{REDIS_NAMESPACE}.delayed.ctrl.DatasetsAddFileCtrl"
+DELAYED_CTRL_DATASETS_REMOVE = f"{REDIS_NAMESPACE}.delayed.ctrl.DatasetsRemoveCtrl"
+DELAYED_CTRL_DATASETS_IMPORT = f"{REDIS_NAMESPACE}.delayed.ctrl.DatasetsImportCtrl"
+DELAYED_CTRL_DATASETS_EDIT = f"{REDIS_NAMESPACE}.delayed.ctrl.DatasetsEditCtrl"
+DELAYED_CTRL_DATASETS_UNLINK = f"{REDIS_NAMESPACE}.delayed.ctrl.DatasetsUnlinkCtrl"
+DELAYED_CTRL_PROJECT_MIGRATE = f"{REDIS_NAMESPACE}.delayed.ctrl.MigrateProjectCtrl"
+DELAYED_CTRL_CONFIG_SET = f"{REDIS_NAMESPACE}.delayed.ctrl.SetConfigCtrl"
 
 
 QUEUES = [