feat(valkey): add valkey sentinel support

Author: duanhongyi

.woodpecker/chart.yaml

@@ -11,6 +11,7 @@ steps:
   - export APP_VERSION=$([ -z $CI_COMMIT_TAG ] && echo $CI_COMMIT_SHA || echo $VERSION)
   - export CHART_VERSION=$([ -z $CI_COMMIT_TAG ] && echo 1.0.0 || echo $VERSION)
   - sed -i "s/imageTag:\ \"canary\"/imageTag:\ $IMAGE_TAG/g" charts/$${CI_REPO_NAME}/values.yaml
+  - sed -i s#{{repository}}#oci://$DRYCC_REGISTRY/$([ -z $CI_COMMIT_TAG ] && echo charts-testing || echo charts)#g charts/$${CI_REPO_NAME}/Chart.yaml
   - helm package -u charts/$${CI_REPO_NAME} --version $CHART_VERSION --app-version $APP_VERSION
   - echo $CONTAINER_PASSWORD | helm registry login $DRYCC_REGISTRY -u $CONTAINER_USERNAME --password-stdin
   - helm push $${CI_REPO_NAME}-$CHART_VERSION.tgz oci://$DRYCC_REGISTRY/$([ -z $CI_COMMIT_TAG ] && echo charts-testing || echo charts)

charts/helmbroker/Chart.yaml

@@ -6,11 +6,8 @@ dependencies:
   - name: common
     repository: oci://registry.drycc.cc/charts
     version: ~1.1.2
-  - name: redis
-    repository: oci://registry.drycc.cc/charts
-    version: x.x.x
-  - name: rabbitmq
-    repository: oci://registry.drycc.cc/charts
+  - name: valkey
+    repository: {{repository}}
     version: x.x.x
 description: Drycc Workflow helmbroker.
 maintainers:

charts/helmbroker/templates/_helpers.tpl

@@ -7,39 +7,20 @@ env:
   value: {{ if .Values.username | default "" | ne "" }}{{ .Values.username }}{{ else }}{{ randAlphaNum 32 }}{{ end }}
 - name: HELMBROKER_PASSWORD
   value: {{ if .Values.password | default "" | ne "" }}{{ .Values.password }}{{ else }}{{ randAlphaNum 32 }}{{ end }}
-{{- if (.Values.rabbitmqUrl) }}
-- name: HELMBROKER_RABBITMQ_URL
-  value: {{ .Values.rabbitmqUrl }}
-{{- else if eq .Values.global.rabbitmqLocation "on-cluster" }}
-- name: "HELMBROKER_RABBITMQ_USERNAME"
+{{- if (.Values.valkeyUrl) }}
+- name: HELMBROKER_VALKEY_URL
   valueFrom:
     secretKeyRef:
-      name: rabbitmq-creds
-      key: username
-- name: "HELMBROKER_RABBITMQ_PASSWORD"
+      name: helmbroker-creds
+      key: valkey-url
+{{- else if eq .Values.global.valkeyLocation "on-cluster"  }}
+- name: VALKEY_PASSWORD
   valueFrom:
     secretKeyRef:
-      name: rabbitmq-creds
+      name: valkey-creds
       key: password
-- name: "HELMBROKER_RABBITMQ_URL"
-  value: "amqp://$(HELMBROKER_RABBITMQ_USERNAME):$(HELMBROKER_RABBITMQ_PASSWORD)@drycc-rabbitmq.{{$.Release.Namespace}}.svc.{{$.Values.global.clusterDomain}}:5672/helmbroker"
-{{- end }}
-{{- if (.Values.redisUrl) }}
-- name: HELMBROKER_REDIS_URL
-  value: {{ .Values.redisUrl }}
-{{- else if eq .Values.global.redisLocation "on-cluster" }}
-- name: "HELMBROKER_REDIS_ADDRS"
-  valueFrom:
-    secretKeyRef:
-      name: redis-creds
-      key: addrs
-- name: "HELMBROKER_REDIS_PASSWORD"
-  valueFrom:
-    secretKeyRef:
-      name: redis-creds
-      key: password
-- name: "HELMBROKER_REDIS_URL"
-  value: "redis://:$(HELMBROKER_REDIS_PASSWORD)@$(HELMBROKER_REDIS_ADDRS)/0"
+- name: HELMBROKER_VALKEY_URL
+  value: "redis://:$(VALKEY_PASSWORD)@drycc-valkey.{{.Release.Namespace}}.svc.{{.Values.global.clusterDomain}}:26379/0?master_set=drycc"
 {{- end }}
 {{- range $key, $value := .Values.environment }}
 - name: {{ $key }}

charts/helmbroker/templates/helmbroker-deployment.yaml

@@ -34,7 +34,7 @@ spec:
           - netcat
           - -v
           - -u
-          - $(HELMBROKER_REDIS_URL),$(HELMBROKER_RABBITMQ_URL)
+          - $(HELMBROKER_VALKEY_URL)
         {{- include "helmbroker.envs" . | indent 10 }}
         - name: drycc-helmbroker-fetch
           image: {{.Values.imageRegistry}}/{{.Values.imageOrg}}/helmbroker:{{.Values.imageTag}}

charts/helmbroker/templates/helmbroker-secret-creds.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: helmbroker-creds
+  labels:
+    heritage: drycc
+data:
+  {{- if (.Values.valkeyUrl) }}
+  valkey-url: {{ .Values.valkeyUrl | b64enc }}
+  {{- end }}

charts/helmbroker/values.yaml

@@ -23,8 +23,8 @@ diagnosticMode:
 
 ## config the helm-broker repositories
 repositories:
-- name: drycc-helm-broker
-  url: https://github.com/drycc/addons/releases/download/latest/index.yaml
+  - name: drycc-helm-broker
+    url: https://github.com/drycc/addons/releases/download/latest/index.yaml
 
 celeryReplicas: 1
 
@@ -33,11 +33,8 @@ celeryReplicas: 1
 username: admin
 password: admin
 
-# Configuring this will no longer use the built-in redis component
-redisUrl: ""
-
-# Configuring this will no longer use the built-in rabbitmq component
-rabbitmqUrl: ""
+# Configuring this will no longer use the built-in valkey component
+valkeyUrl: ""
 
 # Any custom controller environment variables
 # can be specified as key-value pairs under environment
@@ -51,7 +48,7 @@ api:
     key: "drycc.cc/node"
     type: "soft"
     values:
-    - "true"
+      - "true"
   podAffinityPreset:
     type: ""
     extraMatchLabels:
@@ -66,7 +63,7 @@ celery:
     key: "drycc.cc/node"
     type: "soft"
     values:
-    - "true"
+      - "true"
   podAffinityPreset:
     type: ""
     extraMatchLabels:
@@ -76,10 +73,6 @@ celery:
     extraMatchLabels:
       app: "drycc-helmbroker-celery"
 
-# drycc redis replicas must always be set to 1
-redis:
-  replicas: 1
-
 # Default override of addon values
 addonValues: {}
 
@@ -91,16 +84,11 @@ persistence:
   volumeName: ""
 
 global:
-  # Set the location of Workflow's redis instance
-  # Valid values are:
-  # - on-cluster: Run Redis within the Kubernetes cluster
-  # - off-cluster: Run Redis outside the Kubernetes cluster (configure in controller section)
-  redisLocation: "on-cluster"
-  # Set the location of Workflow's rabbitmq instance
+  # Set the location of Workflow's valkey instance
   # Valid values are:
-  # - on-cluster: Run Rabbitmq within the Kubernetes cluster
-  # - off-cluster: Run Rabbitmq outside the Kubernetes cluster (configure in controller section)
-  rabbitmqLocation: "on-cluster"
+  # - on-cluster: Run Valkey within the Kubernetes cluster
+  # - off-cluster: Run Valkey outside the Kubernetes cluster (configure in controller section)
+  valkeyLocation: "on-cluster"
   # Enable usage of RBAC authorization mode
   #
   # Valid values are:

rootfs/helmbroker/celery.py

@@ -1,35 +1,40 @@
 import os
+from urllib.parse import urlparse, parse_qs, urlencode
 from kombu import Exchange, Queue
 from celery import Celery
+from .config import VALKEY_URL
 
 
 class Config(object):
     # Celery Configuration Options
-    timezone = "Asia/Shanghai"
     enable_utc = True
     task_serializer = 'pickle'
     accept_content = frozenset([
-       'application/data',
-       'application/text',
-       'application/json',
-       'application/x-python-serialize',
+        'application/data',
+        'application/text',
+        'application/json',
+        'application/x-python-serialize',
     ])
     task_track_started = True
     task_time_limit = 30 * 60
     worker_max_tasks_per_child = 200
+    worker_prefetch_multiplier = 1
     result_expires = 24 * 60 * 60
-    broker_url = os.environ.get("HELMBROKER_RABBITMQ_URL", 'amqp://guest:guest@127.0.0.1:5672/')
-    broker_connection_retry_on_startup = True
-    task_default_queue = 'helmbroker.low'
+    cache_backend = 'django-cache'
+    task_default_queue = 'helmbroker.middle'
     task_default_exchange = 'helmbroker.priority'
-    task_default_routing_key = 'helmbroker.priority.low'
+    task_default_routing_key = 'helmbroker.priority.middle'
+    broker_transport_options = {"queue_order_strategy": "sorted"}
+    task_create_missing_queues = True
+    task_inherit_parent_priority = True
     broker_connection_retry_on_startup = True
     worker_cancel_long_running_tasks_on_connection_loss = True
 
 
 app = Celery('helmbroker')
 app.config_from_object(Config())
 app.conf.update(
+    timezone=os.environ.get('TZ', 'UTC'),
     task_routes={
         'helmbroker.tasks.provision': {
             'queue': 'helmbroker.high',
@@ -51,22 +56,36 @@ class Config(object):
     task_queues=(
         Queue(
             'helmbroker.low', exchange=Exchange('helmbroker.priority', type="direct"),
-            routing_key='helmbroker.priority.low', queue_arguments={'x-queue-type': 'quorum'},
+            routing_key='helmbroker.priority.low',
         ),
         Queue(
             'helmbroker.high', exchange=Exchange('helmbroker.priority', type="direct"),
-            routing_key='helmbroker.priority.high', queue_arguments={'x-queue-type': 'quorum'},
+            routing_key='helmbroker.priority.high',
         ),
         Queue(
             'helmbroker.middle', exchange=Exchange('helmbroker.priority', type="direct"),
-            routing_key='helmbroker.priority.middle', queue_arguments={'x-queue-type': 'quorum'},
+            routing_key='helmbroker.priority.middle',
         ),
     ),
 )
 app.autodiscover_tasks(("helmbroker.tasks",))
-
-
-app.config_from_object(Config())
+url = urlparse(VALKEY_URL)
+query = parse_qs(url.query)
+broker_transport_options = {"queue_order_strategy": "sorted", "visibility_timeout": 43200}
+result_backend_transport_options = {}
+if 'master_set' in query:
+    master_name = query.pop('master_set')[0]
+    password = url.netloc.split("@")[0].split(":")[1]
+    kwargs = {'sentinel_kwargs': {'password': password}, 'master_name': master_name}
+    broker_transport_options.update(kwargs)
+    result_backend_transport_options.update(kwargs)
+    VALKEY_URL = f"sentinel://{url.netloc}{url.path}?{urlencode(query)}"
+app.conf.update(
+    broker_url=VALKEY_URL,
+    result_backend=VALKEY_URL,
+    broker_transport_options=broker_transport_options,
+    result_backend_transport_options=result_backend_transport_options,
+)
 
 if __name__ == '__main__':
     app.start()

rootfs/helmbroker/config.py

@@ -10,7 +10,7 @@
 USERNAME = os.environ.get('HELMBROKER_USERNAME')
 PASSWORD = os.environ.get('HELMBROKER_PASSWORD')
 
-REDIS_URL = os.environ.get("HELMBROKER_REDIS_URL", 'redis://localhost:6379/0')
+VALKEY_URL = os.environ.get("HELMBROKER_VALKEY_URL", 'redis://localhost:6379/0')
 
 
 class Config:

rootfs/helmbroker/database/fetch.py

@@ -58,7 +58,7 @@ def _fetch_addon(url, dest):
         os.makedirs(dest, exist_ok=True)
         with tarfile.open(fileobj=tgz_file, mode="r:gz") as tarobj:
             for tarinfo in tarobj:
-                tarobj.extract(tarinfo.name, dest)
+                tarobj.extract(tarinfo.name, dest, filter='data')
         filename1 = os.path.join(dest, "meta.yaml")
         with open(filename1, "r") as f1:
             meta = yaml.load(stream=f1, Loader=yaml.Loader)

rootfs/helmbroker/database/metadata.py

@@ -4,8 +4,8 @@
 import logging
 import jsonschema
 
-from redis import client
-from ..config import ADDONS_PATH, REDIS_URL
+from ..utils import get_valkey_client
+from ..config import ADDONS_PATH
 
 logger = logging.getLogger(__name__)
 
@@ -116,21 +116,20 @@ def save_instance_meta(instance_id, data):
     json_data = json.dumps(data, sort_keys=True, indent=2)
     with open(file, "w") as f:
         f.write(json_data)
-    redis = client.Redis.from_url(REDIS_URL)
-    redis.set(cache_key, json_data)
+    get_valkey_client().set(cache_key, json_data)
 
 
 def load_instance_meta(instance_id):
     cache_key = f"helmbroker:instance:{instance_id}"
-    redis = client.Redis.from_url(REDIS_URL)
+    valkey = get_valkey_client()
 
-    json_data = redis.get(cache_key)
+    json_data = valkey.get(cache_key)
     if not json_data:
         from .query import get_instance_file
         file = get_instance_file(instance_id)
         with open(file) as f:
             json_data = f.read()
-            redis.set(cache_key, json_data)
+            valkey.set(cache_key, json_data)
     return json.loads(json_data)
 
 
@@ -144,20 +143,19 @@ def save_binding_meta(instance_id, data):
     json_data = json.dumps(data, sort_keys=True, indent=2)
     with open(file, "w") as f:
         f.write(json_data)
-    redis = client.Redis.from_url(REDIS_URL)
-    redis.set(cache_key, json_data)
+    get_valkey_client().set(cache_key, json_data)
 
 
 def load_binding_meta(instance_id):
     from .query import get_binding_file
     cache_key = f"helmbroker:binding:{instance_id}"
-    redis = client.Redis.from_url(REDIS_URL)
-    json_data = redis.get(cache_key)
+    valkey = get_valkey_client()
+    json_data = valkey.get(cache_key)
     if not json_data:
         file = get_binding_file(instance_id)
         with open(file, 'r') as f:
             json_data = f.read()
-            redis.set(cache_key, json_data)
+            valkey.set(cache_key, json_data)
     return json.loads(json_data)
 
 
@@ -170,18 +168,17 @@ def save_addons_meta(data):
     json_data = json.dumps(data, sort_keys=True, indent=2)
     with open(file, "w") as f:
         f.write(json_data)
-    redis = client.Redis.from_url(REDIS_URL)
-    redis.set(cache_key, json_data)
+    get_valkey_client().set(cache_key, json_data)
 
 
 def load_addons_meta():
     cache_key = "helmbroker:addons"
-    redis = client.Redis.from_url(REDIS_URL)
+    valkey = get_valkey_client()
 
-    json_data = redis.get(cache_key)
+    json_data = valkey.get(cache_key)
     if not json_data:
         file = os.path.join(ADDONS_PATH, "addons.json")
         with open(file, 'r') as f:
             json_data = f.read()
-            redis.set(cache_key, json_data)
+            valkey.set(cache_key, json_data)
     return json.loads(json_data)