From 06bd66e7e43ae139bc6306f0326b6cb89fa5c29d Mon Sep 17 00:00:00 2001 From: Max Schettler Date: Tue, 17 Mar 2026 10:58:44 +0100 Subject: [PATCH 1/5] Modularize controller chart template --- chart/templates/controller.yaml | 176 --------------------- chart/templates/controller/configmap.yaml | 25 +++ chart/templates/controller/deployment.yaml | 84 ++++++++++ chart/templates/controller/rbac.yaml | 18 +++ chart/templates/controller/secret.yaml | 12 ++ chart/templates/controller/service.yaml | 33 ++++ 6 files changed, 172 insertions(+), 176 deletions(-) delete mode 100644 chart/templates/controller.yaml create mode 100644 chart/templates/controller/configmap.yaml create mode 100644 chart/templates/controller/deployment.yaml create mode 100644 chart/templates/controller/rbac.yaml create mode 100644 chart/templates/controller/secret.yaml create mode 100644 chart/templates/controller/service.yaml diff --git a/chart/templates/controller.yaml b/chart/templates/controller.yaml deleted file mode 100644 index 0021b8932..000000000 --- a/chart/templates/controller.yaml +++ /dev/null @@ -1,176 +0,0 @@ -apiVersion: v1 -kind: Secret -metadata: - name: vela-controller-secret -type: Opaque -data: - {{- $existingSecret := lookup "v1" "Secret" .Release.Namespace "vela-controller-secret" }} - {{- if $existingSecret }} - deployment-password-secret: {{ index $existingSecret.data "deployment-password-secret" }} - {{- else }} - deployment-password-secret: {{ randAlphaNum 32 | b64enc }} - {{- end }} ---- -apiVersion: v1 -kind: ServiceAccount -metadata: - name: vela-controller - namespace: {{ .Release.Namespace }} ---- -apiVersion: rbac.authorization.k8s.io/v1 -kind: ClusterRoleBinding -metadata: - name: vela-controller-{{ .Release.Namespace }}-admin -subjects: - - kind: ServiceAccount - name: vela-controller - namespace: {{ .Release.Namespace }} -roleRef: - apiGroup: rbac.authorization.k8s.io - kind: ClusterRole - name: cluster-admin ---- -apiVersion: apps/v1 -kind: Deployment -metadata: - name: vela-controller - labels: - app.kubernetes.io/name: vela-controller - app.kubernetes.io/instance: {{ .Release.Name }} -spec: - replicas: 1 - selector: - matchLabels: - app.kubernetes.io/name: vela-controller - template: - metadata: - labels: - app.kubernetes.io/name: vela-controller - app.kubernetes.io/instance: {{ .Release.Name }} - spec: - serviceAccountName: vela-controller - initContainers: -{{ include "vela.waitForPostgresInitContainer" (dict) | nindent 8 }} - containers: - - name: vela-controller - image: "{{ .Values.controller.image.repository }}:{{ .Values.controller.image.tag }}" - imagePullPolicy: Always - ports: - - name: http - containerPort: 8000 - protocol: TCP - envFrom: - - configMapRef: - name: vela-controller-config - env: - - name: VELA_GRAFANA_SECURITY_ADMIN_USER - valueFrom: - secretKeyRef: - name: vela-grafana-secret - key: VELA_GRAFANA_SECURITY_ADMIN_USER - - name: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD - valueFrom: - secretKeyRef: - name: vela-grafana-secret - key: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD - - name: DB_USER - valueFrom: - secretKeyRef: - name: database - key: superuser-username - - name: DB_PASSWORD - valueFrom: - secretKeyRef: - name: database - key: superuser-password - - name: VELA_DEPLOYMENT_PASSWORD_SECRET - valueFrom: - secretKeyRef: - name: vela-controller-secret - key: deployment-password-secret - - name: DB_HOST - value: database - - name: VELA_POSTGRES_URL - value: 'postgresql+asyncpg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres' - - name: VELA_GRAFANA_URL - value: "https://{{ .Values.domain }}:{{ .Values.port }}/grafana" - livenessProbe: - httpGet: - path: /health - port: http - periodSeconds: 5 - timeoutSeconds: 5 - failureThreshold: 3 - readinessProbe: - httpGet: - path: /health - port: http - initialDelaySeconds: 10 - {{- with .Values.containerSecurityContext }} - securityContext: - {{- toYaml . | nindent 12 }} - {{- end }} - {{- with .Values.podSecurityContext }} - securityContext: - {{- toYaml . | nindent 8 }} - {{- end }} ---- -apiVersion: v1 -kind: ConfigMap -metadata: - name: vela-controller-config -data: - VELA_ROOT_PATH: {{ .Values.controller.env.VELA_ROOT_PATH | quote }} - VELA_JWT_SECRET: {{ .Values.controller.env.VELA_JWT_SECRET | quote }} - VELA_GATEWAY_NAME: {{ .Values.gateway.name | quote }} - VELA_GATEWAY_NAMESPACE: {{ .Values.gateway.namespace | quote }} - VELA_PGMETA_CRYPTO_KEY: {{ .Values.controller.env.VELA_PGMETA_CRYPTO_KEY | quote }} - VELA_CORS_ORIGINS: {{ .Values.controller.env.VELA_CORS_ORIGINS | quote }} - VELA_KEYCLOAK_URL: {{ .Values.controller.env.VELA_KEYCLOAK_URL | quote }} - VELA_KEYCLOAK_ADMIN_NAME: {{ .Values.controller.env.VELA_KEYCLOAK_ADMIN_NAME | quote }} - VELA_KEYCLOAK_ADMIN_SECRET: {{ .Values.controller.env.VELA_KEYCLOAK_ADMIN_SECRET | quote }} - VELA_CLOUDFLARE__API_TOKEN: {{ .Values.controller.env.VELA_CLOUDFLARE_API_TOKEN | quote }} - VELA_CLOUDFLARE__ZONE_ID: {{ .Values.controller.env.VELA_CLOUDFLARE_ZONE_ID | quote }} - VELA_CLOUDFLARE__BRANCH_REF: {{ .Values.controller.env.VELA_BRANCH_REF | quote }} - VELA_CLOUDFLARE__BRANCH_DB_REF: {{ .Values.controller.env.VELA_BRANCH_DB_REF | quote }} - VELA_CLOUDFLARE__DOMAIN_SUFFIX: {{ .Values.domainSuffix | default .Values.domain | quote }} - VELA_DEPLOYMENT_NAMESPACE_PREFIX: {{ .Values.controller.env.VELA_DEPLOYMENT_NAMESPACE_PREFIX | quote }} - VELA_DEPLOYMENT_SERVICE_PORT: {{ .Values.port | quote }} - VELA_SERVER_ROOT_URL: "https://{{ .Values.domain }}:{{ .Values.port }}" - VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: {{ .Values.controller.env.VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER | quote }} - VELA_GRAFANA_URL: {{ .Values.controller.env.VELA_GRAFANA_URL | quote }} - VELA_SIMPLYBLOCK_CSI_NAMESPACE: {{ .Values.controller.env.VELA_SIMPLYBLOCK_CSI_NAMESPACE | quote }} ---- -apiVersion: v1 -kind: Service -metadata: - name: vela-controller-service -spec: - selector: - app.kubernetes.io/name: vela-controller - ports: - - name: http - port: 8000 - targetPort: 8000 - protocol: TCP - ---- -apiVersion: gateway.networking.k8s.io/v1 -kind: HTTPRoute -metadata: - name: vela-controller -spec: - parentRefs: - - name: {{ .Values.gateway.name }} - namespace: {{ .Values.gateway.namespace }} - hostnames: - - "{{ .Values.domain }}" - rules: - - matches: - - path: - type: PathPrefix - value: /vela - backendRefs: - - name: vela-controller-service - namespace: {{ .Release.Namespace }} - port: 8000 diff --git a/chart/templates/controller/configmap.yaml b/chart/templates/controller/configmap.yaml new file mode 100644 index 000000000..466252610 --- /dev/null +++ b/chart/templates/controller/configmap.yaml @@ -0,0 +1,25 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: vela-controller-config +data: + VELA_ROOT_PATH: {{ .Values.controller.env.VELA_ROOT_PATH | quote }} + VELA_JWT_SECRET: {{ .Values.controller.env.VELA_JWT_SECRET | quote }} + VELA_GATEWAY_NAME: {{ .Values.gateway.name | quote }} + VELA_GATEWAY_NAMESPACE: {{ .Values.gateway.namespace | quote }} + VELA_PGMETA_CRYPTO_KEY: {{ .Values.controller.env.VELA_PGMETA_CRYPTO_KEY | quote }} + VELA_CORS_ORIGINS: {{ .Values.controller.env.VELA_CORS_ORIGINS | quote }} + VELA_KEYCLOAK_URL: {{ .Values.controller.env.VELA_KEYCLOAK_URL | quote }} + VELA_KEYCLOAK_ADMIN_NAME: {{ .Values.controller.env.VELA_KEYCLOAK_ADMIN_NAME | quote }} + VELA_KEYCLOAK_ADMIN_SECRET: {{ .Values.controller.env.VELA_KEYCLOAK_ADMIN_SECRET | quote }} + VELA_CLOUDFLARE__API_TOKEN: {{ .Values.controller.env.VELA_CLOUDFLARE_API_TOKEN | quote }} + VELA_CLOUDFLARE__ZONE_ID: {{ .Values.controller.env.VELA_CLOUDFLARE_ZONE_ID | quote }} + VELA_CLOUDFLARE__BRANCH_REF: {{ .Values.controller.env.VELA_BRANCH_REF | quote }} + VELA_CLOUDFLARE__BRANCH_DB_REF: {{ .Values.controller.env.VELA_BRANCH_DB_REF | quote }} + VELA_CLOUDFLARE__DOMAIN_SUFFIX: {{ .Values.domainSuffix | default .Values.domain | quote }} + VELA_DEPLOYMENT_NAMESPACE_PREFIX: {{ .Values.controller.env.VELA_DEPLOYMENT_NAMESPACE_PREFIX | quote }} + VELA_DEPLOYMENT_SERVICE_PORT: {{ .Values.port | quote }} + VELA_SERVER_ROOT_URL: "https://{{ .Values.domain }}:{{ .Values.port }}" + VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: {{ .Values.controller.env.VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER | quote }} + VELA_GRAFANA_URL: {{ .Values.controller.env.VELA_GRAFANA_URL | quote }} + VELA_SIMPLYBLOCK_CSI_NAMESPACE: {{ .Values.controller.env.VELA_SIMPLYBLOCK_CSI_NAMESPACE | quote }} diff --git a/chart/templates/controller/deployment.yaml b/chart/templates/controller/deployment.yaml new file mode 100644 index 000000000..5fa1a9baf --- /dev/null +++ b/chart/templates/controller/deployment.yaml @@ -0,0 +1,84 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: vela-controller + labels: + app.kubernetes.io/name: vela-controller + app.kubernetes.io/instance: {{ .Release.Name }} +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: vela-controller + template: + metadata: + labels: + app.kubernetes.io/name: vela-controller + app.kubernetes.io/instance: {{ .Release.Name }} + spec: + serviceAccountName: vela-controller + initContainers: +{{ include "vela.waitForPostgresInitContainer" (dict) | nindent 8 }} + containers: + - name: vela-controller + image: "{{ .Values.controller.image.repository }}:{{ .Values.controller.image.tag }}" + imagePullPolicy: Always + ports: + - name: http + containerPort: 8000 + protocol: TCP + envFrom: + - configMapRef: + name: vela-controller-config + env: + - name: VELA_GRAFANA_SECURITY_ADMIN_USER + valueFrom: + secretKeyRef: + name: vela-grafana-secret + key: VELA_GRAFANA_SECURITY_ADMIN_USER + - name: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD + valueFrom: + secretKeyRef: + name: vela-grafana-secret + key: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD + - name: DB_USER + valueFrom: + secretKeyRef: + name: database + key: superuser-username + - name: DB_PASSWORD + valueFrom: + secretKeyRef: + name: database + key: superuser-password + - name: VELA_DEPLOYMENT_PASSWORD_SECRET + valueFrom: + secretKeyRef: + name: vela-controller-secret + key: deployment-password-secret + - name: DB_HOST + value: database + - name: VELA_POSTGRES_URL + value: 'postgresql+asyncpg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres' + - name: VELA_GRAFANA_URL + value: "https://{{ .Values.domain }}:{{ .Values.port }}/grafana" + livenessProbe: + httpGet: + path: /health + port: http + periodSeconds: 5 + timeoutSeconds: 5 + failureThreshold: 3 + readinessProbe: + httpGet: + path: /health + port: http + initialDelaySeconds: 10 + {{- with .Values.containerSecurityContext }} + securityContext: + {{- toYaml . | nindent 12 }} + {{- end }} + {{- with .Values.podSecurityContext }} + securityContext: + {{- toYaml . | nindent 8 }} + {{- end }} diff --git a/chart/templates/controller/rbac.yaml b/chart/templates/controller/rbac.yaml new file mode 100644 index 000000000..9bc42f6fd --- /dev/null +++ b/chart/templates/controller/rbac.yaml @@ -0,0 +1,18 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: vela-controller + namespace: {{ .Release.Namespace }} +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: vela-controller-{{ .Release.Namespace }}-admin +subjects: + - kind: ServiceAccount + name: vela-controller + namespace: {{ .Release.Namespace }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: cluster-admin diff --git a/chart/templates/controller/secret.yaml b/chart/templates/controller/secret.yaml new file mode 100644 index 000000000..f73794507 --- /dev/null +++ b/chart/templates/controller/secret.yaml @@ -0,0 +1,12 @@ +apiVersion: v1 +kind: Secret +metadata: + name: vela-controller-secret +type: Opaque +data: + {{- $existingSecret := lookup "v1" "Secret" .Release.Namespace "vela-controller-secret" }} + {{- if $existingSecret }} + deployment-password-secret: {{ index $existingSecret.data "deployment-password-secret" }} + {{- else }} + deployment-password-secret: {{ randAlphaNum 32 | b64enc }} + {{- end }} diff --git a/chart/templates/controller/service.yaml b/chart/templates/controller/service.yaml new file mode 100644 index 000000000..18532eaca --- /dev/null +++ b/chart/templates/controller/service.yaml @@ -0,0 +1,33 @@ +apiVersion: v1 +kind: Service +metadata: + name: vela-controller-service +spec: + selector: + app.kubernetes.io/name: vela-controller + ports: + - name: http + port: 8000 + targetPort: 8000 + protocol: TCP + +--- +apiVersion: gateway.networking.k8s.io/v1 +kind: HTTPRoute +metadata: + name: vela-controller +spec: + parentRefs: + - name: {{ .Values.gateway.name }} + namespace: {{ .Values.gateway.namespace }} + hostnames: + - "{{ .Values.domain }}" + rules: + - matches: + - path: + type: PathPrefix + value: /vela + backendRefs: + - name: vela-controller-service + namespace: {{ .Release.Namespace }} + port: 8000 From 5942b8707158f3844358bf0c9f1bc9c39485b745 Mon Sep 17 00:00:00 2001 From: Max Schettler Date: Tue, 17 Mar 2026 11:07:17 +0100 Subject: [PATCH 2/5] Add celery for task execution --- chart/templates/controller/deployment.yaml | 4 ++ chart/templates/controller/worker.yaml | 71 ++++++++++++++++++++++ containers/Containerfile | 4 ++ containers/compose-dev.yml | 24 +++++++- containers/compose.yml | 43 ++++++++++++- pyproject.toml | 1 + src/worker/__init__.py | 13 ++++ 7 files changed, 157 insertions(+), 3 deletions(-) create mode 100644 chart/templates/controller/worker.yaml create mode 100644 src/worker/__init__.py diff --git a/chart/templates/controller/deployment.yaml b/chart/templates/controller/deployment.yaml index 5fa1a9baf..d19dbfb48 100644 --- a/chart/templates/controller/deployment.yaml +++ b/chart/templates/controller/deployment.yaml @@ -62,6 +62,10 @@ spec: value: 'postgresql+asyncpg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres' - name: VELA_GRAFANA_URL value: "https://{{ .Values.domain }}:{{ .Values.port }}/grafana" + - name: VELA_BROKER_URL + value: 'sqla+postgresql+psycopg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres' + - name: VELA_RESULT_BACKEND + value: 'db+postgresql+psycopg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres' livenessProbe: httpGet: path: /health diff --git a/chart/templates/controller/worker.yaml b/chart/templates/controller/worker.yaml new file mode 100644 index 000000000..4f3b942c9 --- /dev/null +++ b/chart/templates/controller/worker.yaml @@ -0,0 +1,71 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: vela-controller-worker + labels: + app.kubernetes.io/name: vela-controller-worker + app.kubernetes.io/instance: {{ .Release.Name }} +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: vela-controller-worker + template: + metadata: + labels: + app.kubernetes.io/name: vela-controller-worker + app.kubernetes.io/instance: {{ .Release.Name }} + spec: + serviceAccountName: vela-controller + initContainers: +{{ include "vela.waitForPostgresInitContainer" (dict) | nindent 8 }} + containers: + - name: vela-controller-worker + image: "{{ .Values.controller.image.repository }}:{{ .Values.controller.image.tag }}" + imagePullPolicy: Always + command: ["celery", "-A", "simplyblock.vela.worker", "worker", "--loglevel=info"] + envFrom: + - configMapRef: + name: vela-controller-config + env: + - name: VELA_GRAFANA_SECURITY_ADMIN_USER + valueFrom: + secretKeyRef: + name: vela-grafana-secret + key: VELA_GRAFANA_SECURITY_ADMIN_USER + - name: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD + valueFrom: + secretKeyRef: + name: vela-grafana-secret + key: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD + - name: DB_USER + valueFrom: + secretKeyRef: + name: database + key: superuser-username + - name: DB_PASSWORD + valueFrom: + secretKeyRef: + name: database + key: superuser-password + - name: DB_HOST + value: database + - name: VELA_BROKER_URL + value: 'sqla+postgresql+psycopg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres' + - name: VELA_RESULT_BACKEND + value: 'db+postgresql+psycopg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres' + - name: VELA_DEPLOYMENT_PASSWORD_SECRET + valueFrom: + secretKeyRef: + name: vela-controller-secret + key: deployment-password-secret + - name: VELA_POSTGRES_URL + value: 'postgresql+asyncpg://$(DB_USER):$(DB_PASSWORD)@$(DB_HOST):5432/postgres' + {{- with .Values.containerSecurityContext }} + securityContext: + {{- toYaml . | nindent 12 }} + {{- end }} + {{- with .Values.podSecurityContext }} + securityContext: + {{- toYaml . | nindent 8 }} + {{- end }} diff --git a/containers/Containerfile b/containers/Containerfile index e0a05a2a2..316362ca4 100644 --- a/containers/Containerfile +++ b/containers/Containerfile @@ -17,6 +17,8 @@ RUN --mount=type=cache,target=/var/cache/apt \ # Installating via the community-apt repository failed when executed as a GitHub action RUN curl https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash +RUN useradd -m -u 1000 vela + WORKDIR /app COPY .. . @@ -24,4 +26,6 @@ COPY .. . RUN --mount=type=cache,target=/root/.cache/pip pip install . RUN rm -r /app/* +USER vela + CMD ["uvicorn", "simplyblock.vela.api:app", "--host", "0.0.0.0", "--proxy-headers"] diff --git a/containers/compose-dev.yml b/containers/compose-dev.yml index 523082502..296680c0b 100644 --- a/containers/compose-dev.yml +++ b/containers/compose-dev.yml @@ -4,13 +4,15 @@ services: context: .. dockerfile: containers/Containerfile volumes: - - ~/.kube/config:/root/.kube/config:ro + - ~/.kube/config:/home/vela/.kube/config:ro - ..:/app environment: VELA_POSTGRES_URL: 'postgresql+asyncpg://postgres:vela@database:5432/vela' VELA_JWT_SECRET: 'secret' VELA_CORS_ORIGINS: '["http://localhost:3000"]' VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: 'false' + VELA_BROKER_URL: 'sqla+postgresql+psycopg://postgres:vela@database:5432/vela' + VELA_RESULT_BACKEND: 'db+postgresql+psycopg://postgres:vela@database:5432/vela' ports: - 8000:8000 command: 'bash -c "python -m venv /tmp/venv && source /tmp/venv/bin/activate && pip install -e . && uvicorn --host 0.0.0.0 --reload simplyblock.vela.api:app"' @@ -25,6 +27,26 @@ services: restart: unless-stopped + controller-worker: + build: + context: .. + dockerfile: containers/Containerfile + volumes: + - ~/.kube/config:/home/vela/.kube/config:ro + - ..:/app + command: 'bash -c "python -m venv /tmp/venv && source /tmp/venv/bin/activate && pip install -e . && celery -A simplyblock.vela.worker worker --loglevel=info"' + environment: + VELA_BROKER_URL: 'sqla+postgresql+psycopg://postgres:vela@database:5432/vela' + VELA_RESULT_BACKEND: 'db+postgresql+psycopg://postgres:vela@database:5432/vela' + VELA_POSTGRES_URL: 'postgresql+asyncpg://postgres:vela@database:5432/vela' + VELA_JWT_SECRET: 'secret' + VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: 'false' + depends_on: + database: + condition: service_healthy + restart: unless-stopped + + database: image: docker.io/postgres:17 environment: diff --git a/containers/compose.yml b/containers/compose.yml index e05e1816d..fb163971d 100644 --- a/containers/compose.yml +++ b/containers/compose.yml @@ -18,7 +18,7 @@ services: context: .. dockerfile: containers/Containerfile volumes: - - ~/.kube/config:/root/.kube/config:ro + - ~/.kube/config:/home/vela/.kube/config:ro - ..:/app command: 'bash -c "python -m venv /tmp/venv && source /tmp/venv/bin/activate && pip install -e . && uvicorn --host 0.0.0.0 --reload simplyblock.vela.api:app"' environment: @@ -42,6 +42,8 @@ services: VELA_GRAFANA_SECURITY_ADMIN_PASSWORD: "password" VELA_SIMPLYBLOCK_CSI_NAMESPACE: 'simplyblock-csi' VELA_DEPLOYMENT_PASSWORD_SECRET: 'secret' + VELA_BROKER_URL: 'sqla+postgresql+psycopg://postgres:vela@database:5432/vela' + VELA_RESULT_BACKEND: 'db+postgresql+psycopg://postgres:vela@database:5432/vela' depends_on: database: @@ -54,6 +56,43 @@ services: restart: unless-stopped + controller-worker: + build: + context: .. + dockerfile: containers/Containerfile + volumes: + - ~/.kube/config:/home/vela/.kube/config:ro + - ..:/app + command: 'bash -c "python -m venv /tmp/venv && source /tmp/venv/bin/activate && pip install -e . && celery -A simplyblock.vela.worker worker --loglevel=info"' + environment: + VELA_BROKER_URL: 'sqla+postgresql+psycopg://postgres:vela@database:5432/vela' + VELA_RESULT_BACKEND: 'db+postgresql+psycopg://postgres:vela@database:5432/vela' + VELA_ROOT_PATH: '/vela' + VELA_JWT_SECRET: 'http://auth:8080/auth/realms/vela/protocol/openid-connect/certs' + VELA_POSTGRES_URL: 'postgresql+asyncpg://postgres:vela@database:5432/vela' + VELA_CORS_ORIGINS: '["http://localhost:3000"]' + VELA_PGMETA_CRYPTO_KEY: 'secret' + VELA_KEYCLOAK_URL: 'http://auth:8080/auth/' + VELA_KEYCLOAK_ADMIN_NAME: 'admin' + VELA_KEYCLOAK_ADMIN_SECRET: 'admin' + VELA_CLOUDFLARE__API_TOKEN: 'your_cloudflare_api_token' + VELA_CLOUDFLARE__ZONE_ID: 'your_cloudflare_zone_id' + VELA_CLOUDFLARE__BRANCH_REF: 'branch-ref.staging.vela.run' + VELA_CLOUDFLARE__BRANCH_DB_REF: 'branch-db-ref.staging.vela.run' + VELA_CLOUDFLARE__DOMAIN_SUFFIX: 'example.com' + VELA_DEPLOYMENT_NAMESPACE_PREFIX: 'vela' + VELA_GRAFANA_URL: 'http://grafana:3000' + VELA_ENABLE_DB_EXTERNAL_IPV6_LOADBALANCER: 'false' + VELA_GRAFANA_SECURITY_ADMIN_USER: "admin" + VELA_GRAFANA_SECURITY_ADMIN_PASSWORD: "password" + VELA_SIMPLYBLOCK_CSI_NAMESPACE: 'simplyblock-csi' + VELA_DEPLOYMENT_PASSWORD_SECRET: 'secret' + depends_on: + database: + condition: service_healthy + restart: unless-stopped + + auth: image: quay.io/keycloak/keycloak:26.4 environment: @@ -115,7 +154,7 @@ services: GF_AUTH_JWT_JWK_SET_URL: "https://caddy:8443/auth/realms/vela/protocol/openid-connect/certs" GF_AUTH_JWT_ALLOW_INSECURE_CERTIFICATE: "true" GF_AUTH_JWT_TLS_SKIP_VERIFY_INSECURE: "true" - user: "472:0" + user: "472:0" caddy: image: caddy:2 diff --git a/pyproject.toml b/pyproject.toml index 373639f7e..73c00c385 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,7 @@ dependencies = [ 'alembic', 'httpx', 'python-json-logger>=4.0', + 'celery[sqlalchemy]', ] [tool.setuptools] diff --git a/src/worker/__init__.py b/src/worker/__init__.py new file mode 100644 index 000000000..5d64ddd9a --- /dev/null +++ b/src/worker/__init__.py @@ -0,0 +1,13 @@ +from celery import Celery +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_prefix="vela_", case_sensitive=False) + broker_url: str + result_backend: str + + +_settings = Settings() # type: ignore[call-arg] + +app = Celery("vela", broker=_settings.broker_url, backend=_settings.result_backend) From e1e9f734a98f4777b3ed12988e3aa29f98149535 Mon Sep 17 00:00:00 2001 From: Max Schettler Date: Wed, 18 Mar 2026 00:15:24 +0100 Subject: [PATCH 3/5] Convert resize task to background worker --- pyproject.toml | 1 + src/api/__init__.py | 9 +- src/api/_util/backups.py | 2 +- src/api/_util/resourcelimit.py | 2 +- src/api/_util/role.py | 2 +- src/api/auth.py | 2 +- src/api/backup.py | 2 +- src/api/dependencies.py | 2 +- src/api/organization/member.py | 2 +- src/api/organization/project/__init__.py | 2 +- .../organization/project/branch/__init__.py | 260 ++++------------ .../project/branch/resize_tasks.py | 130 ++++++++ src/api/organization/project/branch/tasks.py | 86 ++++++ src/api/organization/role.py | 2 +- src/api/resources.py | 2 +- src/api/system.py | 2 +- src/api/user.py | 2 +- src/{api/db.py => database.py} | 25 +- src/deployment/kubernetes/__init__.py | 46 ++- src/deployment/monitors/health.py | 5 +- src/deployment/monitors/resize/__init__.py | 222 ------------- src/deployment/monitors/resize/pvc_resize.py | 291 ------------------ src/deployment/resize.py | 63 ++++ src/models/branch.py | 109 +------ ...50e_adapt_branch_model_to_external_task.py | 39 +++ src/worker/__init__.py | 11 + tests/branches/test_basic.py | 98 +----- 27 files changed, 479 insertions(+), 940 deletions(-) create mode 100644 src/api/organization/project/branch/resize_tasks.py create mode 100644 src/api/organization/project/branch/tasks.py rename src/{api/db.py => database.py} (58%) delete mode 100644 src/deployment/monitors/resize/__init__.py delete mode 100644 src/deployment/monitors/resize/pvc_resize.py create mode 100644 src/deployment/resize.py create mode 100644 src/models/migrations/versions/ad471311850e_adapt_branch_model_to_external_task.py diff --git a/pyproject.toml b/pyproject.toml index 73c00c385..90aa5cd04 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,7 @@ dependencies = [ 'httpx', 'python-json-logger>=4.0', 'celery[sqlalchemy]', + 'asgiref', ] [tool.setuptools] diff --git a/src/api/__init__.py b/src/api/__init__.py index af66fb7bc..fdd640472 100644 --- a/src/api/__init__.py +++ b/src/api/__init__.py @@ -13,12 +13,11 @@ from fastapi.routing import APIRoute from pydantic import BaseModel +from ..database import engine from ..deployment.monitors.health import vm_monitor -from ..deployment.monitors.resize import ResizeMonitor from ._util.role import create_access_rights_if_emtpy from .backup import router as backup_router from .backupmonitor import run_backup_monitor -from .db import engine from .organization import api as organization_api from .resources import api as resources_api from .resources import monitor_resources @@ -248,22 +247,18 @@ def health(): _use_route_names_as_operation_ids(app) -_resize_monitor = ResizeMonitor() - - @app.on_event("startup") async def on_startup(): await _populate_db() # start async background monitor asyncio.create_task(run_backup_monitor()) asyncio.create_task(monitor_resources()) - _resize_monitor.start() asyncio.create_task(vm_monitor.run()) @app.on_event("shutdown") async def on_shutdown(): - await _resize_monitor.stop() + pass __all__ = ["app"] diff --git a/src/api/_util/backups.py b/src/api/_util/backups.py index 057196bcf..8b8cc23b3 100644 --- a/src/api/_util/backups.py +++ b/src/api/_util/backups.py @@ -17,9 +17,9 @@ if TYPE_CHECKING: from ..._util import Identifier + from ...database import SessionDep from ...models.organization import Organization from ...models.project import Project - from ..db import SessionDep async def copy_branch_backup_schedules( diff --git a/src/api/_util/resourcelimit.py b/src/api/_util/resourcelimit.py index d58af4613..b6c82fa94 100644 --- a/src/api/_util/resourcelimit.py +++ b/src/api/_util/resourcelimit.py @@ -12,6 +12,7 @@ from sqlmodel.ext.asyncio.session import AsyncSession from ..._util import Identifier, single_or_none +from ...database import SessionDep from ...exceptions import VelaResourceLimitError from ...models.branch import Branch, BranchServiceStatus from ...models.organization import Organization @@ -28,7 +29,6 @@ ResourceUsageMinute, UsageCycle, ) -from ..db import SessionDep async def delete_branch_provisioning(session: SessionDep, branch_id: Identifier, *, commit: bool = True): diff --git a/src/api/_util/role.py b/src/api/_util/role.py index 5cb4c0206..a7d4b0462 100644 --- a/src/api/_util/role.py +++ b/src/api/_util/role.py @@ -5,9 +5,9 @@ from ulid import ULID from ..._util import Identifier +from ...database import SessionDep from ...models.branch import Branch from ...models.role import AccessRight, AccessRightPublic, Organization, Role, RoleAccessRight, RoleType, RoleUserLink -from ..db import SessionDep async def clone_user_role_assignment( diff --git a/src/api/auth.py b/src/api/auth.py index 9da6875d9..15b6f32e7 100644 --- a/src/api/auth.py +++ b/src/api/auth.py @@ -10,8 +10,8 @@ from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession +from ..database import SessionDep from ..models.user import JWT, User -from .db import SessionDep from .settings import get_settings # HTTPBearer returns 403 instead of 401. Avoid this by raising the error manually diff --git a/src/api/backup.py b/src/api/backup.py index 87d20bfad..9f30da70a 100644 --- a/src/api/backup.py +++ b/src/api/backup.py @@ -12,6 +12,7 @@ from sqlmodel import asc, delete, select from ulid import ULID +from ..database import SessionDep from ..models._util import Identifier from ..models.backups import ( BackupCreatePublic, @@ -38,7 +39,6 @@ create_branch_snapshot_bundle, delete_branch_snapshot, ) -from .db import SessionDep from .dependencies import OrganizationDep router = APIRouter(dependencies=[Depends(authenticated_user)], tags=["backup"]) diff --git a/src/api/dependencies.py b/src/api/dependencies.py index 50c20b165..37758d12e 100644 --- a/src/api/dependencies.py +++ b/src/api/dependencies.py @@ -6,6 +6,7 @@ from sqlmodel import select from .._util import Identifier +from ..database import SessionDep from ..models.backups import BackupEntry from ..models.branch import Branch, BranchApiKey, BranchServiceStatus from ..models.organization import Organization @@ -13,7 +14,6 @@ from ..models.role import Role from ..models.user import User from .auth import authenticated_user -from .db import SessionDep async def organization_lookup(session: SessionDep, organization_id: Identifier) -> Organization: diff --git a/src/api/organization/member.py b/src/api/organization/member.py index 36cb1d3dc..39030fd28 100644 --- a/src/api/organization/member.py +++ b/src/api/organization/member.py @@ -6,10 +6,10 @@ from fastapi.responses import JSONResponse from sqlalchemy.exc import IntegrityError +from ...database import SessionDep from ...models.user import UserID, UserPublic from .._util import Forbidden, NotFound, Unauthenticated from ..auth import authenticated_user, user_by_id -from ..db import SessionDep from ..dependencies import MemberDep, OrganizationDep, UserDep from ..user import public_list as public_user_list diff --git a/src/api/organization/project/__init__.py b/src/api/organization/project/__init__.py index e07bc8911..c9d8222f7 100644 --- a/src/api/organization/project/__init__.py +++ b/src/api/organization/project/__init__.py @@ -10,6 +10,7 @@ from sqlmodel import select from ...._util import Identifier +from ....database import SessionDep from ....deployment import delete_deployment, get_autoscaler_vm_identity from ....deployment.kubernetes.neonvm import set_virtualmachine_power_state from ....exceptions import VelaKubernetesError @@ -28,7 +29,6 @@ get_project_limit_totals, ) from ...auth import security -from ...db import SessionDep from ...dependencies import OrganizationDep, ProjectDep from . import branch as branch_module from .resources import api as resources_api diff --git a/src/api/organization/project/branch/__init__.py b/src/api/organization/project/branch/__init__.py index a274dede7..737532843 100644 --- a/src/api/organization/project/branch/__init__.py +++ b/src/api/organization/project/branch/__init__.py @@ -17,15 +17,14 @@ from fastapi.security import HTTPAuthorizationCredentials from keycloak.exceptions import KeycloakError from kubernetes_asyncio.client.exceptions import ApiException -from pydantic import AfterValidator, ValidationError +from pydantic import AfterValidator from sqlalchemy.exc import IntegrityError from sqlmodel import select from ....._util import DEFAULT_DB_NAME, DEFAULT_DB_USER, Identifier, storage_backend_bytes_to_db_bytes from ....._util.crypto import encrypt_with_passphrase, generate_keys +from .....database import AsyncSessionLocal, SessionDep from .....deployment import ( - AUTOSCALER_PVC_SUFFIX, - STORAGE_PVC_SUFFIX, DeploymentParameters, ResizeParameters, branch_api_domain, @@ -39,7 +38,6 @@ kube_service, resolve_branch_database_volume_size, update_branch_database_password, - update_branch_volume_iops, ) from .....deployment._util import deployment_namespace from .....deployment.health import ( @@ -64,9 +62,6 @@ BranchPgbouncerConfigStatus, BranchPgbouncerConfigUpdate, BranchPublic, - BranchResizeService, - BranchResizeStatus, - BranchResizeStatusEntry, BranchServiceStatus, BranchSourceDeploymentParameters, BranchStatus, @@ -75,7 +70,6 @@ CapaResizeKey, DatabaseInformation, PgbouncerConfig, - aggregate_resize_statuses, ) from .....models.resources import BranchAllocationPublic, ResourceLimitsPublic, ResourceType from ...._util import Conflict, Forbidden, NotFound, Unauthenticated, url_path_for @@ -96,12 +90,13 @@ SNAPSHOT_TIMEOUT_SEC as _SNAPSHOT_TIMEOUT_SECONDS, ) from ....backup_snapshots import branch_snapshots_used_size -from ....db import AsyncSessionLocal, SessionDep from ....dependencies import BranchDep, OrganizationDep, ProjectDep, backup_lookup, branch_lookup from ....keycloak import realm_admin from ....settings import get_settings as get_api_settings from .api_keys import api as api_key_api from .auth import api as auth_api +from .resize_tasks import dispatch_resize +from .tasks import task_api api = APIRouter(tags=["branch"]) @@ -120,11 +115,6 @@ BranchServiceStatus.RESIZING, } _PROTECTED_BRANCH_STATUSES: set[BranchServiceStatus] = {BranchServiceStatus.PAUSED} -_ACTIVE_RESIZE_STATUSES: set[BranchResizeStatus] = { - "PENDING", - "RESIZING", - "FILESYSTEM_RESIZE_PENDING", -} _CREATING_STATUS_ERROR_GRACE_PERIOD = timedelta(minutes=5) _STARTING_STATUS_ERROR_GRACE_PERIOD = timedelta(minutes=5) @@ -190,13 +180,19 @@ async def _cleanup_failed_branch_deployment(branch_id: Identifier) -> None: def _should_update_branch_status( current: BranchServiceStatus, derived: BranchServiceStatus, - resize_status: BranchResizeStatus, + *, + resize_in_progress: bool = True, ) -> bool: if current == derived: return False - if current == BranchServiceStatus.RESIZING and resize_status in _ACTIVE_RESIZE_STATUSES: - # Keep the explicit RESIZING while a resize is still in progress - # unless we detect a hard failure. + if current == BranchServiceStatus.RESIZING: + if not resize_in_progress: + return derived in { + BranchServiceStatus.ACTIVE_HEALTHY, + BranchServiceStatus.ACTIVE_UNHEALTHY, + BranchServiceStatus.STOPPED, + BranchServiceStatus.ERROR, + } return derived == BranchServiceStatus.ERROR if current == BranchServiceStatus.STARTING and derived == BranchServiceStatus.STOPPED: logger.debug("Ignoring STARTING -> STOPPED transition detected by branch status monitor") @@ -282,38 +278,14 @@ async def _refresh_branch_status(branch: Branch) -> BranchServiceStatus: status, ) - if _should_update_branch_status( - current_status, - status, - resize_status=branch.resize_status, - ): + resize_in_progress = branch.resize_task_id is not None + if _should_update_branch_status(current_status, status, resize_in_progress=resize_in_progress): branch.set_status(status) return status return current_status -def _normalize_resize_statuses(branch: Branch) -> dict[str, BranchResizeStatusEntry]: - statuses = branch.resize_statuses or {} - if not statuses: - return {} - - normalized: dict[str, BranchResizeStatusEntry] = {} - for service, entry in statuses.items(): - if isinstance(entry, BranchResizeStatusEntry): - normalized[service] = entry - continue - try: - normalized[service] = BranchResizeStatusEntry.model_validate(entry) - except ValidationError: - logger.warning( - "Skipping invalid resize status entry for branch %s service %s", - branch.id, - service, - ) - return normalized - - _DEFAULT_SERVICE_STATUS = BranchStatus( database=BranchServiceStatus.UNKNOWN, storage=BranchServiceStatus.UNKNOWN, @@ -717,96 +689,6 @@ async def _build_branch_entity( return entity -_PARAMETER_TO_SERVICE: dict[CapaResizeKey, BranchResizeService] = { - "database_size": "database_disk_resize", - "storage_size": "storage_api_disk_resize", - "milli_vcpu": "database_cpu_resize", - "memory_bytes": "database_memory_resize", - "iops": "database_iops_resize", -} - - -def _track_resize_change( - *, - parameter_key: CapaResizeKey, - new_value: int | None, - current_value: int | None, - statuses: dict[str, dict[str, Any]], - effective: dict[CapaResizeKey, int], - timestamp: str, -) -> None: - service_key = _PARAMETER_TO_SERVICE[parameter_key] - if new_value is None: - return - if new_value != current_value: - effective[parameter_key] = new_value - entry: dict[str, Any] = {"status": "PENDING", "timestamp": timestamp, "requested_at": timestamp} - statuses[service_key] = entry - elif statuses.get(service_key, {}).get("status") == "PENDING": - statuses.pop(service_key, None) - - -async def _apply_resize_operations( - session: SessionDep, - branch: Branch, - effective_parameters: dict[CapaResizeKey, int], -) -> None: - namespace, autoscaler_vm_name = get_autoscaler_vm_identity(branch.id) - if "database_size" in effective_parameters: - new_database_size = effective_parameters["database_size"] - pvc_name = f"{autoscaler_vm_name}{AUTOSCALER_PVC_SUFFIX}" - storage_size_bytes = str(new_database_size) - await kube_service.resize_pvc_storage(namespace, pvc_name, storage_size_bytes) - - if "storage_size" in effective_parameters: - new_storage_size = effective_parameters["storage_size"] - pvc_name = f"{autoscaler_vm_name}{STORAGE_PVC_SUFFIX}" - storage_size_bytes = str(new_storage_size) - await kube_service.resize_pvc_storage(namespace, pvc_name, storage_size_bytes) - - if "iops" in effective_parameters: - new_iops = effective_parameters["iops"] - await update_branch_volume_iops(branch.id, new_iops) - branch.iops = new_iops - await create_or_update_branch_provisioning( - session, - branch, - ResourceLimitsPublic(iops=new_iops), - commit=False, - ) - - milli_vcpu = effective_parameters.get("milli_vcpu", branch.milli_vcpu) - memory = effective_parameters.get("memory_bytes", branch.memory) - - cpu_changed = "milli_vcpu" in effective_parameters - memory_changed = "memory_bytes" in effective_parameters - - if cpu_changed or memory_changed: - await kube_service.resize_autoscaler_vm( - namespace, - autoscaler_vm_name, - cpu_milli=milli_vcpu, - memory_bytes=memory, - ) - if cpu_changed: - branch.milli_vcpu = milli_vcpu - await create_or_update_branch_provisioning( - session, - branch, - ResourceLimitsPublic(milli_vcpu=milli_vcpu), - commit=False, - ) - - if memory_changed: - branch.memory = memory - await create_or_update_branch_provisioning( - session, - branch, - ResourceLimitsPublic(ram=memory), - commit=False, - ) - - async def _deploy_branch_environment_task( *, organization_id: Identifier, @@ -1671,13 +1553,8 @@ async def status( _project: ProjectDep, branch: BranchDep, ) -> BranchStatusPublic: - normalized_resize_statuses = _normalize_resize_statuses(branch) service_status = await collect_branch_service_health(branch.id) - return BranchStatusPublic( - resize_status=branch.resize_status, - resize_statuses=normalized_resize_statuses, - service_status=service_status, - ) + return BranchStatusPublic(service_status=service_status) @instance_api.post( @@ -1927,15 +1804,14 @@ async def update_pgbouncer_config( ) async def resize( session: SessionDep, - _organization: OrganizationDep, - _project: ProjectDep, + request: Request, + organization: OrganizationDep, + project: ProjectDep, parameters: ResizeParameters, branch: BranchDep, ): - branch_in_session = await session.merge(branch) - if parameters.database_size is not None: - current_database_size = branch_in_session.database_size + current_database_size = branch.database_size requested_database_size = parameters.database_size if current_database_size is not None and requested_database_size < current_database_size: raise HTTPException( @@ -1947,7 +1823,7 @@ async def resize( ) if parameters.storage_size is not None: - current_storage = branch_in_session.storage_size + current_storage = branch.storage_size requested_storage = parameters.storage_size if current_storage is not None and requested_storage < current_storage: raise HTTPException( @@ -1958,72 +1834,41 @@ async def resize( ), ) - updated_statuses = dict(branch_in_session.resize_statuses or {}) - timestamp = datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + # Guard against concurrent resizes + if branch.resize_task_id is not None: + raise HTTPException(status_code=400, detail="A resize operation is already in progress") effective_parameters: dict[CapaResizeKey, int] = {} + for param_key, new_val, current_val in [ + ("database_size", parameters.database_size, branch.database_size), + ("storage_size", parameters.storage_size, branch.storage_size), + ("milli_vcpu", parameters.milli_vcpu, branch.milli_vcpu), + ("memory_bytes", parameters.memory_bytes, branch.memory), + ("iops", parameters.iops, branch.iops), + ]: + if new_val is not None and new_val != current_val: + effective_parameters[cast("CapaResizeKey", param_key)] = new_val - _track_resize_change( - parameter_key="database_size", - new_value=parameters.database_size, - current_value=branch_in_session.database_size, - statuses=updated_statuses, - effective=effective_parameters, - timestamp=timestamp, - ) - _track_resize_change( - parameter_key="storage_size", - new_value=parameters.storage_size, - current_value=branch_in_session.storage_size, - statuses=updated_statuses, - effective=effective_parameters, - timestamp=timestamp, - ) - _track_resize_change( - parameter_key="milli_vcpu", - new_value=parameters.milli_vcpu, - current_value=branch_in_session.milli_vcpu, - statuses=updated_statuses, - effective=effective_parameters, - timestamp=timestamp, - ) - _track_resize_change( - parameter_key="memory_bytes", - new_value=parameters.memory_bytes, - current_value=branch_in_session.memory, - statuses=updated_statuses, - effective=effective_parameters, - timestamp=timestamp, - ) - _track_resize_change( - parameter_key="iops", - new_value=parameters.iops, - current_value=branch_in_session.iops, - statuses=updated_statuses, - effective=effective_parameters, - timestamp=timestamp, - ) + if not effective_parameters: + return Response(status_code=202) + + await _ensure_resize_resource_limits(session, branch, effective_parameters) + + task_id = dispatch_resize(str(branch.id), dict(effective_parameters)) - if effective_parameters: - await _ensure_resize_resource_limits(session, branch_in_session, effective_parameters) - await _apply_resize_operations(session, branch_in_session, effective_parameters) - completion_timestamp = datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") - for param_key, service_key in ( - ("iops", "database_iops_resize"), - ("milli_vcpu", "database_cpu_resize"), - ("memory_bytes", "database_memory_resize"), - ): - if param_key in effective_parameters: - updated_statuses[service_key] = { - "status": "COMPLETED", - "timestamp": completion_timestamp, - } - - branch_in_session.resize_statuses = updated_statuses - branch_in_session.resize_status = aggregate_resize_statuses(updated_statuses) - branch_in_session.set_status(BranchServiceStatus.RESIZING) + branch.set_status(BranchServiceStatus.RESIZING) + branch.resize_task_id = task_id await session.commit() - return Response(status_code=202) + + task_url = url_path_for( + request, + "organizations:projects:branch:tasks:detail", + organization_id=await organization.awaitable_attrs.id, + project_id=await project.awaitable_attrs.id, + branch_id=await branch.awaitable_attrs.id, + task_id=task_id, + ) + return Response(status_code=202, headers={"Location": task_url}) _control_responses: dict[int | str, dict[str, Any]] = { @@ -2139,6 +1984,7 @@ async def control_branch( instance_api.include_router(auth_api, prefix="/auth") instance_api.include_router(api_key_api, prefix="/apikeys") +instance_api.include_router(task_api, prefix="/tasks") api.include_router(instance_api) diff --git a/src/api/organization/project/branch/resize_tasks.py b/src/api/organization/project/branch/resize_tasks.py new file mode 100644 index 000000000..d45d78105 --- /dev/null +++ b/src/api/organization/project/branch/resize_tasks.py @@ -0,0 +1,130 @@ +"""Celery chord tasks for proactive branch resize completion. + +The chord dispatches one sub-task per resource type (in the deployment layer) +and calls ``finalize_resize`` when all sub-tasks have settled. With +``task_chord_propagates = False`` the callback always fires, even on partial +failure, so succeeded fields are persisted and ``resize_task_id`` is cleared +regardless of individual sub-task outcomes. +""" + +import logging +from uuid import UUID + +from asgiref.sync import async_to_sync +from celery import chord +from ulid import ULID + +from .....database import AsyncSessionLocal +from .....deployment.health import collect_branch_service_health, derive_branch_status_from_services +from .....deployment.resize import resize_cpu_memory, resize_database_pvc, resize_iops, resize_storage_pvc +from .....models.branch import Branch, BranchServiceStatus +from .....models.resources import ResourceLimitsPublic +from .....worker import app +from ...._util.resourcelimit import create_or_update_branch_provisioning + +logger = logging.getLogger(__name__) + +_FIELD_TO_BRANCH_ATTR: dict[str, str] = { + "database_size": "database_size", + "storage_size": "storage_size", + "milli_vcpu": "milli_vcpu", + "memory_bytes": "memory", + "iops": "iops", +} + +_FIELD_TO_RESOURCE_LIMITS_KWARG: dict[str, str] = { + "database_size": "database_size", + "storage_size": "storage_size", + "milli_vcpu": "milli_vcpu", + "memory_bytes": "ram", + "iops": "iops", +} + + +async def _apply_succeeded_fields(session: object, branch: Branch, succeeded: dict[str, int]) -> None: + """Write each succeeded resize field onto the branch and update provisioning records.""" + for field, value in succeeded.items(): + branch_attr = _FIELD_TO_BRANCH_ATTR.get(field) + if branch_attr is None: + continue + if getattr(branch, branch_attr) == value: + continue + setattr(branch, branch_attr, value) + limits_kwarg = _FIELD_TO_RESOURCE_LIMITS_KWARG[field] + await create_or_update_branch_provisioning( + session, # type: ignore[arg-type] + branch, + ResourceLimitsPublic(**{limits_kwarg: value}), + commit=False, + ) + + +async def _async_finalize_resize( + job_results: list, + branch_id: str, + effective_parameters: dict, + job_fields: list[list[str]], +) -> dict: + succeeded: dict[str, int] = {} + errors: list[str] = [] + + for result, fields in zip(job_results, job_fields, strict=False): + if isinstance(result, Exception): + errors.append(str(result)) + else: + for field in fields: + if field in effective_parameters: + succeeded[field] = effective_parameters[field] + + async with AsyncSessionLocal() as session: + branch = await session.get(Branch, ULID.from_str(branch_id)) + if branch is None: + logger.error("Branch %s not found during finalize_resize", branch_id) + return {**succeeded, "errors": errors} + + await _apply_succeeded_fields(session, branch, succeeded) + + service_status = await collect_branch_service_health(branch.id) + if branch.status == BranchServiceStatus.RESIZING: + branch.set_status( + derive_branch_status_from_services(service_status, storage_enabled=branch.enable_file_storage) + ) + + branch.resize_task_id = None + await session.commit() + + return {**succeeded, "errors": errors} + + +@app.task(name="simplyblock.vela.branch.resize.finalize") +def finalize_resize( + job_results: list, + branch_id: str, + effective_parameters: dict, + job_fields: list[list[str]], +) -> dict: + """Chord callback: persist succeeded fields and clear resize_task_id.""" + return async_to_sync(_async_finalize_resize)(job_results, branch_id, effective_parameters, job_fields) + + +def dispatch_resize(branch_id: str, effective_parameters: dict) -> UUID: + """Build and dispatch a chord of resize sub-tasks; return the chord result UUID.""" + jobs = [] + job_fields: list[list[str]] = [] + + ep = effective_parameters + if "milli_vcpu" in ep or "memory_bytes" in ep: + jobs.append(resize_cpu_memory.s(branch_id, ep.get("milli_vcpu"), ep.get("memory_bytes"))) + job_fields.append([f for f in ("milli_vcpu", "memory_bytes") if f in ep]) + if "iops" in ep: + jobs.append(resize_iops.s(branch_id, ep["iops"])) + job_fields.append(["iops"]) + if "database_size" in ep: + jobs.append(resize_database_pvc.s(branch_id, ep["database_size"])) + job_fields.append(["database_size"]) + if "storage_size" in ep: + jobs.append(resize_storage_pvc.s(branch_id, ep["storage_size"])) + job_fields.append(["storage_size"]) + + result = chord(jobs)(finalize_resize.s(branch_id=branch_id, effective_parameters=ep, job_fields=job_fields)) + return UUID(result.id) diff --git a/src/api/organization/project/branch/tasks.py b/src/api/organization/project/branch/tasks.py new file mode 100644 index 000000000..38351a52b --- /dev/null +++ b/src/api/organization/project/branch/tasks.py @@ -0,0 +1,86 @@ +"""Branch task list/detail endpoints. + +Exposes Celery task state (currently resize only) under: + GET .../branches/{branch_id}/tasks + GET .../branches/{branch_id}/tasks/{task_id} +""" + +from datetime import datetime +from typing import Any +from uuid import UUID + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from ...._util import Forbidden, NotFound, Unauthenticated +from ....dependencies import BranchDep, OrganizationDep, ProjectDep +from .resize_tasks import finalize_resize + +task_api = APIRouter(tags=["branch"]) + +_CELERY_STATE_TO_STATUS: dict[str, str] = { + "PENDING": "PENDING", + "STARTED": "STARTED", + "SUCCESS": "COMPLETED", + "FAILURE": "FAILED", + "REVOKED": "FAILED", +} + + +class BranchTaskPublic(BaseModel): + id: UUID + task_type: str + status: str + parameters: dict + result: Any | None + error: str | None + date_done: datetime | None + + +def _build_task_public(task_id: UUID) -> BranchTaskPublic: + result = finalize_resize.AsyncResult(str(task_id)) + state = result.state + status = _CELERY_STATE_TO_STATUS.get(state, state) + kwargs: dict = result.kwargs or {} + return BranchTaskPublic( + id=task_id, + task_type="resize", + status=status, + parameters=kwargs.get("effective_parameters", {}), + result=result.result if state == "SUCCESS" else None, + error=str(result.traceback) if state == "FAILURE" and result.traceback else None, + date_done=result.date_done, + ) + + +@task_api.get( + "/", + name="organizations:projects:branch:tasks:list", + response_model=list[BranchTaskPublic], + responses={401: Unauthenticated, 403: Forbidden, 404: NotFound}, +) +async def list_tasks( + _organization: OrganizationDep, + _project: ProjectDep, + branch: BranchDep, +) -> list[BranchTaskPublic]: + if branch.resize_task_id is None: + return [] + return [_build_task_public(branch.resize_task_id)] + + +@task_api.get( + "/{task_id}", + name="organizations:projects:branch:tasks:detail", + response_model=BranchTaskPublic, + responses={401: Unauthenticated, 403: Forbidden, 404: NotFound}, +) +async def get_task( + _organization: OrganizationDep, + _project: ProjectDep, + branch: BranchDep, + task_id: UUID, +) -> BranchTaskPublic: + if branch.resize_task_id != task_id: + raise HTTPException(status_code=404, detail="Task not found") + return _build_task_public(task_id) diff --git a/src/api/organization/role.py b/src/api/organization/role.py index 16470d29f..9ccabdb6e 100644 --- a/src/api/organization/role.py +++ b/src/api/organization/role.py @@ -7,6 +7,7 @@ from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession +from ...database import SessionDep from ...models._util import Identifier from ...models.role import ( AccessRight, @@ -29,7 +30,6 @@ from .._util import Forbidden, NotFound, Unauthenticated from ..access_right_utils import check_access from ..auth import authenticated_user -from ..db import SessionDep from ..dependencies import OrganizationDep, RoleDep, branch_lookup, project_lookup api = APIRouter(dependencies=[Depends(authenticated_user)], tags=["role"]) diff --git a/src/api/resources.py b/src/api/resources.py index 9d2c7383b..72d9a83d4 100644 --- a/src/api/resources.py +++ b/src/api/resources.py @@ -12,6 +12,7 @@ from .._util import quantity_to_bytes, quantity_to_milli_cpu from ..check_branch_status import get_branch_status +from ..database import SessionDep from ..deployment import ( get_autoscaler_vm_identity, resolve_autoscaler_volume_identifiers, @@ -54,7 +55,6 @@ make_usage_cycle, ) from .auth import authenticated_user -from .db import SessionDep from .dependencies import BranchDep, OrganizationDep, ProjectDep from .organization.project.branch import refresh_branch_status from .settings import get_settings diff --git a/src/api/system.py b/src/api/system.py index 5256aa993..fa7fd47ee 100644 --- a/src/api/system.py +++ b/src/api/system.py @@ -21,10 +21,10 @@ VCPU_MILLIS_MIN, VCPU_MILLIS_STEP, ) +from ..database import SessionDep from ..models.resources import OrganizationLimitDefault, ResourceLimitDefinitionPublic, ResourceType from ..models.role import AccessRight from .auth import authenticated_user -from .db import SessionDep api = APIRouter(tags=["system"]) diff --git a/src/api/user.py b/src/api/user.py index 5d7c05a66..a0326c782 100644 --- a/src/api/user.py +++ b/src/api/user.py @@ -8,12 +8,12 @@ from pydantic import BaseModel, EmailStr, StrictBool from sqlmodel import and_, select +from ..database import SessionDep from ..models.membership import Membership from ..models.role import AccessRight, Role, RoleAccessRight, RoleUserLink, RoleUserLinkPublic, UserPermissionPublic from ..models.user import User, UserParameters, UserPublic from ._util import NotFound, Unauthenticated from .auth import authenticated_user -from .db import SessionDep from .keycloak import realm_admin api = APIRouter(dependencies=[Depends(authenticated_user)], tags=["user"]) diff --git a/src/api/db.py b/src/database.py similarity index 58% rename from src/api/db.py rename to src/database.py index d88a3d8e8..132c0c04a 100644 --- a/src/api/db.py +++ b/src/database.py @@ -1,3 +1,4 @@ +import os from collections.abc import AsyncGenerator from typing import Annotated @@ -5,17 +6,23 @@ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlmodel.ext.asyncio.session import AsyncSession -from .settings import get_settings +from .api.settings import get_settings -# Enable `pool_pre_ping` and periodic recycling so ASGI workers notice connections -# that Postgres closed while idle (seen as "connection is closed" during requests) -# and transparently reopen them before handing out a session. -engine = create_async_engine( - str(get_settings().postgres_url), - pool_pre_ping=True, - pool_recycle=3600, -) +def _is_worker() -> bool: + return bool(os.environ.get("VELA_BROKER_URL")) + + +if _is_worker(): + from sqlalchemy.pool import NullPool + + engine = create_async_engine(str(get_settings().postgres_url), poolclass=NullPool) +else: + engine = create_async_engine( + str(get_settings().postgres_url), + pool_pre_ping=True, + pool_recycle=3600, + ) AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) diff --git a/src/deployment/kubernetes/__init__.py b/src/deployment/kubernetes/__init__.py index f4e90b4a0..7b0b56715 100644 --- a/src/deployment/kubernetes/__init__.py +++ b/src/deployment/kubernetes/__init__.py @@ -1,5 +1,8 @@ +import asyncio import logging import math +import re +import time from collections.abc import Mapping from copy import deepcopy from typing import Any @@ -7,10 +10,21 @@ from aiohttp import ClientError from kubernetes_asyncio import client +from ..._util import quantity_to_bytes, storage_backend_bytes_to_db_bytes from ...exceptions import VelaKubernetesError from ._util import core_v1_client, custom_api_client, discovery_v1_client, storage_v1_client from .neonvm import NeonVM, get_neon_vm +POLL_INTERVAL_SECONDS = 5 +POLL_TIMEOUT_SECONDS = 30 * 60 # 30 minutes + +_FAILURE_PATTERN = re.compile( + r"\b(resize|resizing|resized)\w*\b.*\b(fail|failure|failed|failing|error|err)\w*\b" + r"|" + r"\b(fail|failure|failed|failing|error|err)\w*\b.*\b(resize|resizing|resized)\w*\b", + flags=re.IGNORECASE, +) + logger = logging.getLogger(__name__) @@ -292,7 +306,7 @@ async def get_persistent_volume_claim(self, namespace: str, name: str) -> Any: raise VelaKubernetesError(f"PersistentVolumeClaim {namespace!r}/{name!r} not found") from exc raise - async def resize_pvc_storage(self, namespace: str, name: str, storage: str) -> None: + async def resize_pvc_storage(self, namespace: str, name: str, storage: str, *, wait: bool = False) -> None: async with core_v1_client() as core_v1: try: await core_v1.patch_namespaced_persistent_volume_claim( @@ -308,6 +322,36 @@ async def resize_pvc_storage(self, namespace: str, name: str, storage: str) -> N logger.info("Resized PVC %s/%s to %s", namespace, name, storage) + if wait: + target = quantity_to_bytes(storage) + if target is None: + raise VelaKubernetesError(f"Cannot parse storage quantity {storage!r}") + await self._poll_pvc_until_complete(namespace, name, target) + + async def _poll_pvc_until_complete(self, namespace: str, name: str, target_bytes: int) -> int: + """Poll PVC status every 5s until resize completes or fails. Returns actual capacity.""" + start = time.monotonic() + async with core_v1_client() as core_v1: + while True: + pvc = await core_v1.read_namespaced_persistent_volume_claim(namespace=namespace, name=name) + + capacity_str = (pvc.status.capacity or {}).get("storage") + if capacity_str: + actual = quantity_to_bytes(capacity_str) + if actual and actual >= target_bytes: + return storage_backend_bytes_to_db_bytes(actual) + + for condition in pvc.status.conditions or []: + msg = condition.message or "" + if _FAILURE_PATTERN.search(msg): + raise RuntimeError(f"PVC {namespace}/{name} resize failed: {msg}") + + elapsed = time.monotonic() - start + if elapsed >= POLL_TIMEOUT_SECONDS: + raise TimeoutError(f"PVC {namespace}/{name} resize timed out after {POLL_TIMEOUT_SECONDS}s") + + await asyncio.sleep(POLL_INTERVAL_SECONDS) + async def get_persistent_volume(self, name: str) -> Any: async with core_v1_client() as core_v1: try: diff --git a/src/deployment/monitors/health.py b/src/deployment/monitors/health.py index a5a464e8c..c8cb4051a 100644 --- a/src/deployment/monitors/health.py +++ b/src/deployment/monitors/health.py @@ -3,10 +3,10 @@ import re from datetime import datetime, timedelta -from kubernetes_asyncio import client from pydantic import BaseModel from ulid import ULID +from ..kubernetes._util import custom_api_client from ..kubernetes.neonvm import NeonVM, Phase from ..settings import get_settings @@ -56,8 +56,7 @@ async def _check_status(self, vm: NeonVM): async def run(self): logger.info("Started VM monitor") - async with client.ApiClient() as api: - custom_api = client.CustomObjectsApi(api) + async with custom_api_client() as custom_api: while True: try: start = datetime.now() diff --git a/src/deployment/monitors/resize/__init__.py b/src/deployment/monitors/resize/__init__.py deleted file mode 100644 index 521ea306e..000000000 --- a/src/deployment/monitors/resize/__init__.py +++ /dev/null @@ -1,222 +0,0 @@ -"""Resize monitor orchestration. - -This module stitches together two complementary monitoring loops that ensure -branch resources reflect resize requests made through Vela: - -PVC resize pipeline -------------------- -1. ``stream_pvc_events`` opens a long-lived Kubernetes watch filtered to PVC - events. The watch feeds a bounded queue to decouple collection from - processing, maintaining a resourceVersion for informer-like continuity and - exponential backoff for resiliency. -2. A small worker pool (``_event_worker``) pulls events off the queue and calls - ``_handle_pvc_event`` which enriches the raw Kubernetes event, fetching PVC - capacity when necessary, and normalizes timestamps and statuses. -3. ``_apply_volume_status`` persists the synthesized status onto the owning - ``Branch`` record, updating aggregate resize state and capacity fields so that - API consumers observe consistent progress semantics. - -Control surface ---------------- -``run_resize_monitor`` drives both loops concurrently, coordinating shutdown via -``ResizeMonitor`` which offers a simple start/stop interface for long-lived -services. The design emphasizes: -* backpressure-aware queuing so surges of events do not starve handlers; -* idempotent database writes guarded by status transitions; -* graceful teardown of asyncio tasks to avoid leaking watches or sessions. -""" - -import asyncio -import logging -from contextlib import suppress -from typing import Any, cast - -from kubernetes_asyncio.client import CoreV1Api -from kubernetes_asyncio.client.models import CoreV1Event -from sqlmodel import select -from sqlmodel.ext.asyncio.session import AsyncSession -from ulid import ULID - -from ...._util import storage_backend_bytes_to_db_bytes -from ....api._util.resourcelimit import create_or_update_branch_provisioning -from ....api.db import engine -from ....deployment import deployment_branch -from ....exceptions import VelaDeploymentError, VelaKubernetesError -from ....models.branch import ( - RESIZE_STATUS_PRIORITY, - Branch, - BranchResizeStatus, - aggregate_resize_statuses, - should_transition_resize_status, -) -from ....models.resources import ResourceLimitsPublic -from ...health import collect_branch_service_health, derive_branch_status_from_services -from .pvc_resize import ( - INITIAL_BACKOFF_SECONDS, - VOLUME_SERVICE_MAP, - derive_status, - fetch_pvc_capacity, - normalize_iso_timestamp, - resource_from_pvc_name, - stream_pvc_events, -) - -logger = logging.getLogger(__name__) - - -async def _apply_volume_status( - *, - branch_id: ULID, - status: BranchResizeStatus | None, - event_payload: dict[str, str], - resource: str | None, - capacity: int | None, -) -> None: - """Persist the PVC resize status and size information on the branch record.""" - from sqlalchemy.exc import NoResultFound - - async with AsyncSession(engine) as session: - statement = select(Branch).where(Branch.id == branch_id) - result = await session.exec(statement) - try: - branch = result.one() - except NoResultFound: - return - - service_key = VOLUME_SERVICE_MAP.get(resource) if resource else None - statuses = dict(branch.resize_statuses or {}) - status_updated = False - - if status is not None and service_key: - current_entry = statuses.get(service_key) - current_status_value = None - if isinstance(current_entry, dict): - current_status_value = current_entry.get("status") - if isinstance(current_status_value, str) and current_status_value in RESIZE_STATUS_PRIORITY: - current_status = cast("BranchResizeStatus", current_status_value) - else: - current_status = None - - if should_transition_resize_status(current_status, status): - statuses[service_key] = { - "status": status, - "timestamp": event_payload["timestamp"], - } - branch.resize_statuses = statuses - branch.resize_status = aggregate_resize_statuses(statuses) - await set_branch_status(branch.resize_status, branch) - status_updated = True - elif status is not None and should_transition_resize_status(branch.resize_status, status): - branch.resize_status = status - await set_branch_status(branch.resize_status, branch) - status_updated = True - - if status_updated and status == "COMPLETED" and capacity is not None: - normalized_capacity = storage_backend_bytes_to_db_bytes(capacity) - if resource == "storage": - await create_or_update_branch_provisioning( - session, - branch, - ResourceLimitsPublic(storage_size=normalized_capacity), - commit=False, - ) - branch.storage_size = normalized_capacity - elif resource == "database": - await create_or_update_branch_provisioning( - session, - branch, - ResourceLimitsPublic(database_size=normalized_capacity), - commit=False, - ) - branch.database_size = normalized_capacity - - await session.commit() - - -async def set_branch_status(status: BranchResizeStatus, branch: Branch) -> None: - if status not in {"FAILED", "COMPLETED"}: - return - - service_status = await collect_branch_service_health(branch.id) - branch.set_status( - derive_branch_status_from_services( - service_status, - storage_enabled=branch.enable_file_storage, - ) - ) - - -async def _handle_pvc_event(core_v1: CoreV1Api, event: CoreV1Event) -> None: - """Map a raw Kubernetes event into branch status updates and capacity changes.""" - ref = event.involved_object - namespace = getattr(ref, "namespace", None) - name = getattr(ref, "name", None) - if not namespace or not name: - return - try: - branch_id = deployment_branch(namespace) - except VelaDeploymentError: - logger.debug("Ignoring resize event for namespace %s (not a branch namespace)", namespace) - return - resource = resource_from_pvc_name(name) - status = derive_status(event.reason, event.type, event.message) - capacity = None - if status == "COMPLETED": - capacity = await fetch_pvc_capacity(core_v1, namespace, name) - - status = cast("BranchResizeStatus", status) - timestamp_source = ( - event.last_timestamp or event.event_time or (event.metadata.creation_timestamp if event.metadata else None) - ) - payload = { - "timestamp": normalize_iso_timestamp(timestamp_source), - "type": event.reason or "", - "message": event.message or "", - } - await _apply_volume_status( - branch_id=branch_id, - status=status, - event_payload=payload, - resource=resource, - capacity=capacity, - ) - - -async def run_resize_monitor(stop_event: asyncio.Event) -> None: - """Drive PVC resize monitors until the caller signals shutdown.""" - while not stop_event.is_set(): - try: - await stream_pvc_events(stop_event, _handle_pvc_event) - except VelaKubernetesError as exc: - logger.warning("PVC resize monitor awaiting Kubernetes configuration: %s", exc) - await asyncio.sleep(INITIAL_BACKOFF_SECONDS) - except Exception: # pragma: no cover - defensive guard - logger.exception("PVC resize monitor unexpected failure; retrying") - await asyncio.sleep(INITIAL_BACKOFF_SECONDS) - - -class ResizeMonitor: - """Lightweight wrapper for starting and stopping the resize monitor loop.""" - - def __init__(self) -> None: - self._stop = asyncio.Event() - self._task: asyncio.Task[Any] | None = None - - def start(self) -> None: - """Ensure the background resize monitor task is running.""" - if self._task is None: - self._task = asyncio.create_task(run_resize_monitor(self._stop)) - - async def stop(self) -> None: - """Shut down the background task and reset internal state.""" - if self._task is None: - return - self._stop.set() - self._task.cancel() - with suppress(asyncio.CancelledError): - await self._task - self._task = None - self._stop = asyncio.Event() - - -__all__ = ["ResizeMonitor", "run_resize_monitor"] diff --git a/src/deployment/monitors/resize/pvc_resize.py b/src/deployment/monitors/resize/pvc_resize.py deleted file mode 100644 index 3e5309204..000000000 --- a/src/deployment/monitors/resize/pvc_resize.py +++ /dev/null @@ -1,291 +0,0 @@ -import asyncio -import logging -import re -from collections.abc import Awaitable, Callable -from contextlib import suppress -from datetime import UTC, datetime - -from aiohttp import ClientError -from kubernetes_asyncio import watch -from kubernetes_asyncio.client import CoreV1Api -from kubernetes_asyncio.client.exceptions import ApiException -from kubernetes_asyncio.client.models import CoreV1Event - -from ...._util import quantity_to_bytes -from ....deployment import AUTOSCALER_PVC_SUFFIX, DATABASE_PVC_SUFFIX, STORAGE_PVC_SUFFIX -from ....deployment.kubernetes._util import core_v1_client - -logger = logging.getLogger(__name__) - - -FAILURE_PATTERN = re.compile( - r"\b(resize|resizing|resized)\w*\b.*\b(fail|failure|failed|failing|error|err)\w*\b" - r"|" - r"\b(fail|failure|failed|failing|error|err)\w*\b.*\b(resize|resizing|resized)\w*\b", - flags=re.IGNORECASE, -) - -VOLUME_SERVICE_MAP = { - "database": "database_disk_resize", - "storage": "storage_api_disk_resize", -} - -INITIAL_BACKOFF_SECONDS = 5 -MAX_BACKOFF_SECONDS = 60 -PVC_EVENT_QUEUE_MAXSIZE = 2048 -PVC_WORKER_POOL_SIZE = 4 -PVC_WATCH_TIMEOUT_SECONDS = 60 -PVC_QUEUE_PUT_TIMEOUT_SECONDS = 2 - - -def normalize_iso_timestamp(value: datetime | None) -> str: - """Return a UTC ISO-8601 string for Kubernetes event timestamps.""" - if value is None: - value = datetime.now(UTC) - if value.tzinfo is None: - value = value.replace(tzinfo=UTC) - return value.astimezone(UTC).isoformat().replace("+00:00", "Z") - - -def resource_from_pvc_name(name: str) -> str | None: - """Infer which service a PVC belongs to based on the naming convention.""" - if name.endswith(STORAGE_PVC_SUFFIX): - return "storage" - if name.endswith(DATABASE_PVC_SUFFIX): - return "database" - if name.endswith(AUTOSCALER_PVC_SUFFIX): - return "database" - return None - - -def derive_status(reason: str | None, event_type: str | None, message: str | None) -> str | None: - """Translate a Kubernetes Event into a high-level resize status.""" - normalized_reason = (reason or "").upper() - normalized_message = message or "" - if normalized_reason in {"RESIZING", "EXTERNALEXPANDING"}: - return "RESIZING" - if normalized_reason == "FILESYSTEMRESIZEREQUIRED": - return "FILESYSTEM_RESIZE_PENDING" - if normalized_reason == "RESIZEFINISHED": - return "FILESYSTEM_RESIZE_PENDING" - if normalized_reason == "FILESYSTEMRESIZESUCCESSFUL": - return "COMPLETED" - if normalized_reason in {"VOLUMERESIZEFAILED", "FILESYSTEMRESIZEFAILED"}: - return "FAILED" - if (event_type or "").upper() == "WARNING" and FAILURE_PATTERN.search(normalized_message): - return "FAILED" - return None - - -async def fetch_pvc_capacity(core_v1: CoreV1Api, namespace: str, name: str) -> int | None: - """Fetch the latest PVC capacity in bytes via the Kubernetes API.""" - try: - pvc = await core_v1.read_namespaced_persistent_volume_claim(namespace=namespace, name=name) - except (ApiException, ClientError): - logger.exception("Failed to fetch PVC %s/%s for resize completion update", namespace, name) - return None - - status_capacity = getattr(pvc.status, "capacity", None) - if status_capacity: - capacity = quantity_to_bytes(status_capacity.get("storage")) - if capacity: - return capacity - - resources = getattr(pvc.spec, "resources", None) if pvc.spec else None - requests = getattr(resources, "requests", None) - if requests: - capacity = quantity_to_bytes(requests.get("storage")) - if capacity: - return capacity - - return None - - -EventHandler = Callable[[CoreV1Api, CoreV1Event], Awaitable[None]] - - -async def _enqueue_event( - queue: asyncio.Queue[CoreV1Event | None], - event: CoreV1Event, - stop_event: asyncio.Event, -) -> bool: - """Attempt to enqueue an event while honoring shutdown signals.""" - warned = False - while True: - if stop_event.is_set(): - return False - try: - await asyncio.wait_for(queue.put(event), timeout=PVC_QUEUE_PUT_TIMEOUT_SECONDS) - except TimeoutError: - if not warned: - warned = True - logger.warning( - "PVC resize monitor queue is saturated (%s/%s); delaying watcher", - queue.qsize(), - PVC_EVENT_QUEUE_MAXSIZE, - ) - continue - return True - - -async def _event_worker( - worker_id: int, - *, - queue: asyncio.Queue[CoreV1Event | None], - core_v1: CoreV1Api, - handler: EventHandler, - stop_event: asyncio.Event, -) -> None: - """Drain the PVC event queue and invoke the handler in a dedicated task.""" - worker_name = f"pvc-resize-worker-{worker_id}" - logger.debug("%s started", worker_name) - try: - while True: - try: - item = await queue.get() - except asyncio.CancelledError: - raise - - if item is None: - queue.task_done() - break - - try: - await handler(core_v1, item) - except asyncio.CancelledError: - raise - except Exception: - logger.exception("%s failed processing PVC event", worker_name) - finally: - queue.task_done() - - if stop_event.is_set() and queue.empty(): - logger.debug("%s draining queue and exiting", worker_name) - break - finally: - logger.debug("%s stopped", worker_name) - - -async def _consume_event_stream( - *, - core_v1: CoreV1Api, - field_selector: str, - resource_version: str | None, - stop_event: asyncio.Event, - queue: asyncio.Queue[CoreV1Event | None], -) -> str | None: - """Stream Kubernetes events until a shutdown signal or queue saturation occurs.""" - kube_watch = watch.Watch() - try: - async for item in kube_watch.stream( - core_v1.list_event_for_all_namespaces, - field_selector=field_selector, - resource_version=resource_version, - timeout_seconds=PVC_WATCH_TIMEOUT_SECONDS, - ): - if stop_event.is_set(): - break - - event: CoreV1Event = item["object"] - metadata = getattr(event, "metadata", None) - if metadata is not None: - resource_version = getattr(metadata, "resource_version", resource_version) - - if not await _enqueue_event(queue, event, stop_event): - break - return resource_version - finally: - kube_watch.stop() - - -async def _handle_stream_error( - exc: ApiException | ClientError, - *, - resource_version: str | None, - backoff_seconds: int, -) -> tuple[str | None, int]: - """Apply error handling and backoff policy for PVC event stream errors.""" - if isinstance(exc, ApiException) and exc.status == 410: - logger.info("PVC resize monitor resource version expired; resyncing from latest") - return "0", INITIAL_BACKOFF_SECONDS - - if isinstance(exc, ApiException): - logger.warning("PVC resize monitor API error: %s", exc) - else: - logger.warning("PVC resize monitor stream client error: %s", exc) - - await asyncio.sleep(backoff_seconds) - next_backoff = min(backoff_seconds * 2, MAX_BACKOFF_SECONDS) - return resource_version, next_backoff - - -async def _watch_pvc_event_stream( - *, - core_v1: CoreV1Api, - queue: asyncio.Queue[CoreV1Event | None], - stop_event: asyncio.Event, - field_selector: str, -) -> None: - """Watch PVC events until stopped, applying backoff and error handling.""" - resource_version: str | None = "0" - backoff_seconds = INITIAL_BACKOFF_SECONDS - while not stop_event.is_set(): - try: - resource_version = await _consume_event_stream( - core_v1=core_v1, - field_selector=field_selector, - resource_version=resource_version, - stop_event=stop_event, - queue=queue, - ) - backoff_seconds = INITIAL_BACKOFF_SECONDS - except TimeoutError: - continue - except asyncio.CancelledError: - raise - except (ApiException, ClientError) as exc: - resource_version, backoff_seconds = await _handle_stream_error( - exc, - resource_version=resource_version, - backoff_seconds=backoff_seconds, - ) - - -async def stream_pvc_events(stop_event: asyncio.Event, handler: EventHandler) -> None: - """Watch PVC-related events and push them through a worker-backed queue.""" - field_selector = "involvedObject.kind=PersistentVolumeClaim" - queue: asyncio.Queue[CoreV1Event | None] = asyncio.Queue(maxsize=PVC_EVENT_QUEUE_MAXSIZE) - async with core_v1_client() as core_v1: - workers = [ - asyncio.create_task( - _event_worker( - worker_id, - queue=queue, - core_v1=core_v1, - handler=handler, - stop_event=stop_event, - ) - ) - for worker_id in range(PVC_WORKER_POOL_SIZE) - ] - try: - await _watch_pvc_event_stream( - core_v1=core_v1, - queue=queue, - stop_event=stop_event, - field_selector=field_selector, - ) - finally: - # Allow workers to finish in-flight items before shutting down. - try: - await asyncio.wait_for(queue.join(), timeout=PVC_WATCH_TIMEOUT_SECONDS) - except TimeoutError: - logger.warning( - "PVC resize monitor queue did not drain within %s seconds; cancelling workers", - PVC_WATCH_TIMEOUT_SECONDS, - ) - for worker in workers: - worker.cancel() - for worker in workers: - with suppress(asyncio.CancelledError): - await worker diff --git a/src/deployment/resize.py b/src/deployment/resize.py new file mode 100644 index 000000000..f9fef0f15 --- /dev/null +++ b/src/deployment/resize.py @@ -0,0 +1,63 @@ +"""Celery task and dispatch helper for branch resize operations. + +The task lives in the deployment layer and has no DB access. The API layer +manages branch state; this module only orchestrates infrastructure changes and +returns the resulting sizes for the caller to persist. +""" + +import logging + +from asgiref.sync import async_to_sync +from ulid import ULID + +from ..worker import app +from . import ( + AUTOSCALER_PVC_SUFFIX, + STORAGE_PVC_SUFFIX, + get_autoscaler_vm_identity, + kube_service, + update_branch_volume_iops, +) + +logger = logging.getLogger(__name__) + + +async def _resize_cpu_memory(deployment_id: ULID, milli_vcpu: int | None, memory_bytes: int | None) -> None: + namespace, autoscaler_vm_name = get_autoscaler_vm_identity(deployment_id) + await kube_service.resize_autoscaler_vm( + namespace, autoscaler_vm_name, cpu_milli=milli_vcpu, memory_bytes=memory_bytes + ) + + +@app.task(name="simplyblock.vela.deployment.resize.cpu_memory") +def resize_cpu_memory(deployment_id: str, milli_vcpu: int | None, memory_bytes: int | None) -> None: + """Resize CPU and/or memory for a branch deployment.""" + async_to_sync(_resize_cpu_memory)(ULID.from_str(deployment_id), milli_vcpu, memory_bytes) + + +async def _resize_iops(deployment_id: ULID, iops: int) -> None: + await update_branch_volume_iops(deployment_id, iops) + + +@app.task(name="simplyblock.vela.deployment.resize.iops") +def resize_iops(deployment_id: str, iops: int) -> None: + """Resize IOPS for a branch deployment.""" + async_to_sync(_resize_iops)(ULID.from_str(deployment_id), iops) + + +async def _resize_pvc(deployment_id: ULID, pvc_suffix: str, target_size: int) -> None: + namespace, autoscaler_vm_name = get_autoscaler_vm_identity(deployment_id) + pvc_name = f"{autoscaler_vm_name}{pvc_suffix}" + await kube_service.resize_pvc_storage(namespace, pvc_name, str(target_size), wait=True) + + +@app.task(name="simplyblock.vela.deployment.resize.database_pvc") +def resize_database_pvc(deployment_id: str, database_size: int) -> None: + """Resize the database PVC for a branch deployment.""" + async_to_sync(_resize_pvc)(ULID.from_str(deployment_id), AUTOSCALER_PVC_SUFFIX, database_size) + + +@app.task(name="simplyblock.vela.deployment.resize.storage_pvc") +def resize_storage_pvc(deployment_id: str, storage_size: int) -> None: + """Resize the storage PVC for a branch deployment.""" + async_to_sync(_resize_pvc)(ULID.from_str(deployment_id), STORAGE_PVC_SUFFIX, storage_size) diff --git a/src/models/branch.py b/src/models/branch.py index b77059a10..d049981d5 100644 --- a/src/models/branch.py +++ b/src/models/branch.py @@ -1,9 +1,9 @@ -from collections.abc import Mapping +import uuid from datetime import UTC, datetime from enum import StrEnum from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Literal, Optional -from pydantic import BaseModel, ConfigDict, ValidationError, model_validator +from pydantic import BaseModel, model_validator from pydantic import Field as PydanticField from sqlalchemy import BigInteger, Boolean, Column, String, Text, UniqueConstraint, text from sqlalchemy.dialects.postgresql import JSONB @@ -102,19 +102,12 @@ class Branch(AsyncAttrs, Model, table=True): jwt_secret: Annotated[str, Field(default=None, sa_column=Column(Text, nullable=True))] anon_key: Annotated[str, Field(default=None, sa_column=Column(Text, nullable=True))] service_key: Annotated[str, Field(default=None, sa_column=Column(Text, nullable=True))] - resize_status: "BranchResizeStatus" = Field( - default="NONE", - sa_column=Column(String(length=48), nullable=False), - ) - resize_statuses: dict[str, dict[str, Any]] = Field( - default_factory=dict, - sa_column=Column(JSONB, nullable=False, server_default=text("'{}'::jsonb")), - ) resource_usage: dict[str, Any] = Field( default_factory=_default_resource_usage_payload, sa_column=Column(JSONB, nullable=False, server_default=text("'{}'::jsonb")), ) pitr_enabled: bool = Field(default=False, sa_column=Column(Boolean, nullable=False, server_default=text("false"))) + resize_task_id: uuid.UUID | None = Field(default=None, nullable=True) __table_args__ = (UniqueConstraint("project_id", "name", name="unique_branch_name_per_project"),) @@ -472,102 +465,8 @@ class BranchStatus(BaseModel): rest: BranchServiceStatus -BranchResizeStatus = Literal[ - "NONE", - "PENDING", - "RESIZING", - "FILESYSTEM_RESIZE_PENDING", - "COMPLETED", - "FAILED", -] - CapaResizeKey = Literal["database_size", "storage_size", "milli_vcpu", "memory_bytes", "iops"] -BranchResizeService = Literal[ - "database_disk_resize", - "storage_api_disk_resize", - "database_cpu_resize", - "database_memory_resize", - "database_iops_resize", -] - -RESIZE_STATUS_PRIORITY: dict[BranchResizeStatus, int] = { - "NONE": 0, - "PENDING": 1, - "RESIZING": 2, - "FILESYSTEM_RESIZE_PENDING": 3, - "COMPLETED": 4, - "FAILED": 5, -} - - -class BranchResizeStatusEntry(BaseModel): - """Single service's resize state and the timestamp when it was observed.""" - - model_config = ConfigDict(extra="allow") - - status: BranchResizeStatus - timestamp: str - - def timestamp_as_datetime(self) -> datetime: - value = self.timestamp - if not value: - return datetime.min.replace(tzinfo=UTC) - try: - normalized = value[:-1] + "+00:00" if value.endswith("Z") else value - parsed = datetime.fromisoformat(normalized) - except ValueError: - return datetime.min.replace(tzinfo=UTC) - if parsed.tzinfo is None: - parsed = parsed.replace(tzinfo=UTC) - return parsed.astimezone(UTC) - - -def should_transition_resize_status(current: BranchResizeStatus | None, proposed: BranchResizeStatus) -> bool: - """Return True when the proposed status should replace the current one.""" - - if proposed not in RESIZE_STATUS_PRIORITY: - return False - if current == proposed: - return False - if proposed == "FAILED": - return True - if current is None: - return True - return RESIZE_STATUS_PRIORITY[proposed] >= RESIZE_STATUS_PRIORITY[current] - - -def aggregate_resize_statuses( - statuses: Mapping[str, BranchResizeStatusEntry | Mapping[str, Any]] | None, -) -> BranchResizeStatus: - """ - Aggregate per-service resize statuses into a single branch-level status. - """ - highest: BranchResizeStatus = "NONE" - if not statuses: - return highest - - highest_timestamp = datetime.min.replace(tzinfo=UTC) - for entry in statuses.values(): - if isinstance(entry, BranchResizeStatusEntry): - snapshot = entry - else: - try: - snapshot = BranchResizeStatusEntry.model_validate(entry) - except ValidationError: - continue - - status = snapshot.status - if status not in RESIZE_STATUS_PRIORITY: - continue - timestamp = snapshot.timestamp_as_datetime() - if timestamp > highest_timestamp or ( - timestamp == highest_timestamp and should_transition_resize_status(highest, status) - ): - highest = status - highest_timestamp = timestamp - return highest - class BranchPublic(BaseModel): id: Identifier @@ -589,8 +488,6 @@ class BranchPublic(BaseModel): class BranchStatusPublic(BaseModel): - resize_status: BranchResizeStatus - resize_statuses: dict[str, BranchResizeStatusEntry] service_status: BranchStatus diff --git a/src/models/migrations/versions/ad471311850e_adapt_branch_model_to_external_task.py b/src/models/migrations/versions/ad471311850e_adapt_branch_model_to_external_task.py new file mode 100644 index 000000000..56db63c36 --- /dev/null +++ b/src/models/migrations/versions/ad471311850e_adapt_branch_model_to_external_task.py @@ -0,0 +1,39 @@ +"""Adapt branch model to external task + +Revision ID: ad471311850e +Revises: e8f3a2d51c9b +Create Date: 2026-03-17 21:55:53.413091 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +import sqlmodel +import sqlmodel.sql +from sqlalchemy.dialects import postgresql + + +# revision identifiers, used by Alembic. +revision: str = 'ad471311850e' +down_revision: Union[str, Sequence[str], None] = 'e8f3a2d51c9b' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.add_column('branch', sa.Column('resize_task_id', sa.Uuid(), nullable=True)) + op.drop_column('branch', 'resize_status') + op.drop_column('branch', 'resize_statuses') + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column('branch', 'resize_task_id') + op.add_column('branch', sa.Column('resize_statuses', postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), autoincrement=False, nullable=False)) + op.add_column('branch', sa.Column('resize_status', sa.VARCHAR(length=48), autoincrement=False, nullable=False)) + # ### end Alembic commands ### diff --git a/src/worker/__init__.py b/src/worker/__init__.py index 5d64ddd9a..fb20d4e15 100644 --- a/src/worker/__init__.py +++ b/src/worker/__init__.py @@ -11,3 +11,14 @@ class Settings(BaseSettings): _settings = Settings() # type: ignore[call-arg] app = Celery("vela", broker=_settings.broker_url, backend=_settings.result_backend) + +# Persist task name, args, and kwargs in celery_taskmeta so AsyncResult can +# reconstruct full task details without a custom model. +app.conf.result_extended = True + +# Chord callback always fires even when individual sub-tasks fail. +app.conf.task_chord_propagates = False + +# Register tasks — must be imported after `app` is defined. +from ..api.organization.project.branch import resize_tasks as _api_resize_tasks # noqa: E402, F401 +from ..deployment import resize as _deployment_resize # noqa: E402, F401 diff --git a/tests/branches/test_basic.py b/tests/branches/test_basic.py index 09026d2bf..33bb9ef81 100644 --- a/tests/branches/test_basic.py +++ b/tests/branches/test_basic.py @@ -124,92 +124,21 @@ def test_branch_list_contains(client, org, project, branch_id): def test_branch_status_endpoint(client, org, project, branch_id): r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/status") assert r.status_code == 200 - data = r.json() - assert "resize_status" in data def test_branch_resize(client, org, project, branch_id): + database_size = 6 * 1_000_000_000 # 6 GB, must be a multiple of 1 GB + storage_size = 2 * 1_000_000_000 # 2 GB, must be a multiple of 1 GB + memory_bytes = 2 * 1024 * 1024 * 1024 # 2 GiB, must be a multiple of 256 MiB r = client.post( f"organizations/{org}/projects/{project}/branches/{branch_id}/resize", - json={"iops": 2000}, - ) - assert r.status_code == 202 - wait_for_status( - client, - f"organizations/{org}/projects/{project}/branches/{branch_id}/", - "ACTIVE_HEALTHY", - BRANCH_TIMEOUT_SEC, - ) - r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/") - assert r.status_code == 200 - assert r.json()["max_resources"]["iops"] == 2000 - - -def test_branch_resize_cpu(client, org, project, branch_id): - r = client.post( - f"organizations/{org}/projects/{project}/branches/{branch_id}/resize", - json={"milli_vcpu": 1000}, - ) - assert r.status_code == 202 - wait_for_status( - client, - f"organizations/{org}/projects/{project}/branches/{branch_id}/", - "ACTIVE_HEALTHY", - BRANCH_TIMEOUT_SEC, - ) - r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/") - assert r.status_code == 200 - assert r.json()["max_resources"]["milli_vcpu"] == 1000 - - -def test_branch_resize_memory(client, org, project, branch_id): - # 2 GiB expressed in bytes (must be a multiple of 256 MiB) - two_gib = 2 * 1024 * 1024 * 1024 - r = client.post( - f"organizations/{org}/projects/{project}/branches/{branch_id}/resize", - json={"memory_bytes": two_gib}, - ) - assert r.status_code == 202 - wait_for_status( - client, - f"organizations/{org}/projects/{project}/branches/{branch_id}/", - "ACTIVE_HEALTHY", - BRANCH_TIMEOUT_SEC, - ) - r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/") - assert r.status_code == 200 - assert r.json()["max_resources"]["ram_bytes"] == two_gib - - -def test_branch_resize_database_size(client, org, project, branch_id): - # 6 GB expressed in bytes (must be a multiple of 1 GB) - six_gb = 6 * 1_000_000_000 - # due to issues related to round up - seven_gb = 7 * 1_000_000_000 - r = client.post( - f"organizations/{org}/projects/{project}/branches/{branch_id}/resize", - json={"database_size": six_gb}, - ) - assert r.status_code == 202 - wait_for_status( - client, - f"organizations/{org}/projects/{project}/branches/{branch_id}/", - "ACTIVE_HEALTHY", - BRANCH_TIMEOUT_SEC, - ) - r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/") - assert r.status_code == 200 - assert r.json()["max_resources"]["nvme_bytes"] == seven_gb - - -def test_branch_resize_storage_size(client, org, project, branch_id): - # 2 GB expressed in bytes (must be a multiple of 1 GB) - two_gb = 2 * 1_000_000_000 - # due to GiB rounding by the storage backend: 2 GB -> 2 GiB -> rounds up to 3 GB - three_gb = 3 * 1_000_000_000 - r = client.post( - f"organizations/{org}/projects/{project}/branches/{branch_id}/resize", - json={"storage_size": two_gb}, + json={ + "iops": 2000, + "milli_vcpu": 1000, + "memory_bytes": memory_bytes, + "database_size": database_size, + "storage_size": storage_size, + }, ) assert r.status_code == 202 wait_for_status( @@ -220,7 +149,12 @@ def test_branch_resize_storage_size(client, org, project, branch_id): ) r = client.get(f"organizations/{org}/projects/{project}/branches/{branch_id}/") assert r.status_code == 200 - assert r.json()["max_resources"]["storage_bytes"] == three_gb + resources = r.json()["max_resources"] + assert resources["iops"] == 2000 + assert resources["milli_vcpu"] == 1000 + assert resources["ram_bytes"] == memory_bytes + assert resources["nvme_bytes"] == database_size + assert resources["storage_bytes"] == storage_size def test_branch_password_reset(client, org, project, branch_id): From 9c9bb41f28372818ae63951a956ae17adc6de041 Mon Sep 17 00:00:00 2001 From: Max Schettler Date: Tue, 24 Mar 2026 16:47:27 +0100 Subject: [PATCH 4/5] Remove parse-quantity-wrapper --- src/_util/__init__.py | 10 ---------- src/api/backup_snapshots.py | 5 +++-- src/api/resources.py | 7 ++++--- src/deployment/kubernetes/__init__.py | 16 ++++++---------- src/deployment/kubernetes/neonvm.py | 17 ++++------------- 5 files changed, 17 insertions(+), 38 deletions(-) diff --git a/src/_util/__init__.py b/src/_util/__init__.py index 4be2db228..45fa6985b 100644 --- a/src/_util/__init__.py +++ b/src/_util/__init__.py @@ -243,16 +243,6 @@ def quantity_to_milli_cpu(value: str | Decimal | None) -> int | None: return int(quantity * Decimal(1000)) -def quantity_to_bytes(value: str | Decimal | None) -> int | None: - """Convert a Kubernetes-style quantity string (e.g. '10Gi', '512Mi') to bytes.""" - - quantity = _normalize_quantity(value) - if quantity is None: - return None - - return int(quantity) - - def permissive_numeric_timedelta(value: Any) -> Any: """Parses the given value into timedelta diff --git a/src/api/backup_snapshots.py b/src/api/backup_snapshots.py index 6e38bca59..04c7c8f26 100644 --- a/src/api/backup_snapshots.py +++ b/src/api/backup_snapshots.py @@ -9,8 +9,8 @@ from uuid import UUID import httpx +from kubernetes.utils import parse_quantity -from .._util import Identifier, quantity_to_bytes from ..deployment import AUTOSCALER_PVC_SUFFIX, AUTOSCALER_WAL_PVC_SUFFIX, get_autoscaler_vm_identity from ..deployment.kubernetes.snapshot import ( create_snapshot_from_pvc, @@ -26,6 +26,7 @@ if TYPE_CHECKING: from ulid import ULID + from .._util import Identifier from ..models.backups import BackupEntry logger = logging.getLogger(__name__) @@ -154,7 +155,7 @@ async def _create_snapshot_for_pvc( status = snapshot.get("status") or {} content_name_payload = status.get("boundVolumeSnapshotContentName") content_name = content_name_payload if isinstance(content_name_payload, str) else None - size_bytes = quantity_to_bytes(status.get("restoreSize")) + size_bytes = parse_quantity(status["restoreSize"]) try: if content_name is None: diff --git a/src/api/resources.py b/src/api/resources.py index 72d9a83d4..5ac88ad5f 100644 --- a/src/api/resources.py +++ b/src/api/resources.py @@ -6,11 +6,12 @@ from uuid import UUID from fastapi import APIRouter, Depends, HTTPException +from kubernetes.utils import parse_quantity from kubernetes_asyncio.client.exceptions import ApiException from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlmodel import select -from .._util import quantity_to_bytes, quantity_to_milli_cpu +from .._util import quantity_to_milli_cpu from ..check_branch_status import get_branch_status from ..database import SessionDep from ..deployment import ( @@ -315,9 +316,9 @@ def _parse_compute_usage(metrics: dict[str, Any]) -> tuple[int, int]: usage = cast("dict[str, Any]", compute_usage["usage"]) cpu_usage = quantity_to_milli_cpu(usage["cpu"]) - memory_usage = quantity_to_bytes(usage["memory"]) + memory_usage = parse_quantity(usage["memory"]) - if cpu_usage is None or memory_usage is None: + if cpu_usage is None: raise ValueError("Metrics API returned empty resource usage for compute container") return cpu_usage, memory_usage diff --git a/src/deployment/kubernetes/__init__.py b/src/deployment/kubernetes/__init__.py index 7b0b56715..f90302c98 100644 --- a/src/deployment/kubernetes/__init__.py +++ b/src/deployment/kubernetes/__init__.py @@ -5,12 +5,13 @@ import time from collections.abc import Mapping from copy import deepcopy +from decimal import Decimal from typing import Any from aiohttp import ClientError +from kubernetes.utils import parse_quantity from kubernetes_asyncio import client -from ..._util import quantity_to_bytes, storage_backend_bytes_to_db_bytes from ...exceptions import VelaKubernetesError from ._util import core_v1_client, custom_api_client, discovery_v1_client, storage_v1_client from .neonvm import NeonVM, get_neon_vm @@ -323,12 +324,9 @@ async def resize_pvc_storage(self, namespace: str, name: str, storage: str, *, w logger.info("Resized PVC %s/%s to %s", namespace, name, storage) if wait: - target = quantity_to_bytes(storage) - if target is None: - raise VelaKubernetesError(f"Cannot parse storage quantity {storage!r}") - await self._poll_pvc_until_complete(namespace, name, target) + await self._poll_pvc_until_complete(namespace, name, parse_quantity(storage)) - async def _poll_pvc_until_complete(self, namespace: str, name: str, target_bytes: int) -> int: + async def _poll_pvc_until_complete(self, namespace: str, name: str, target_bytes: Decimal) -> int: """Poll PVC status every 5s until resize completes or fails. Returns actual capacity.""" start = time.monotonic() async with core_v1_client() as core_v1: @@ -336,10 +334,8 @@ async def _poll_pvc_until_complete(self, namespace: str, name: str, target_bytes pvc = await core_v1.read_namespaced_persistent_volume_claim(namespace=namespace, name=name) capacity_str = (pvc.status.capacity or {}).get("storage") - if capacity_str: - actual = quantity_to_bytes(capacity_str) - if actual and actual >= target_bytes: - return storage_backend_bytes_to_db_bytes(actual) + if capacity_str and ((actual := parse_quantity(capacity_str)) >= target_bytes): + return int(actual) for condition in pvc.status.conditions or []: msg = condition.message or "" diff --git a/src/deployment/kubernetes/neonvm.py b/src/deployment/kubernetes/neonvm.py index 1b335a57c..f46ba3d0f 100644 --- a/src/deployment/kubernetes/neonvm.py +++ b/src/deployment/kubernetes/neonvm.py @@ -2,10 +2,11 @@ from typing import Annotated, Any, Literal from aiohttp.client_exceptions import ClientError +from kubernetes.utils import parse_quantity from kubernetes_asyncio.client.exceptions import ApiException from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, PlainSerializer, ValidationError -from ..._util import quantity_to_bytes, quantity_to_milli_cpu +from ..._util import quantity_to_milli_cpu from ...exceptions import VelaKubernetesError from ._util import custom_api_client @@ -19,16 +20,6 @@ def _require_int(value: Any, field: str) -> int: raise VelaKubernetesError(f"Autoscaler VM missing required integer field {field}") from None -def _require_quantity_bytes(value: Any, field: str) -> int: - if isinstance(value, (int, float)): - return int(value) - if isinstance(value, str): - parsed = quantity_to_bytes(value) - if parsed is not None: - return parsed - raise VelaKubernetesError(f"Autoscaler VM missing required quantity field {field}") - - def _require_cpu_millis(value: Any, field: str) -> int: if isinstance(value, (int, float)): return int(value) if value >= 1000 else int(value * 1000) @@ -117,13 +108,13 @@ class Port(CamelModel): class Guest(CamelModel): cpus: GuestCPUs memory_slots: MemorySlots - memory_slot_size: Any + memory_slot_size: str ports: list[Port] env: AutoscalerEnv = Field(default_factory=dict) @property def slot_size_bytes(self) -> int: - return _require_quantity_bytes(self.memory_slot_size, "guest.memorySlotSize") + return int(parse_quantity(self.memory_slot_size)) class NeonVMSpec(CamelModel): From 7ade63ba007fb3ceaea713971357d37e2197b78b Mon Sep 17 00:00:00 2001 From: Max Schettler Date: Tue, 24 Mar 2026 17:01:09 +0100 Subject: [PATCH 5/5] Simplify CPU specification handling --- src/_util/__init__.py | 2 +- src/deployment/kubernetes/__init__.py | 16 +++------------- src/deployment/kubernetes/neonvm.py | 20 ++++---------------- src/deployment/resize.py | 6 +++++- 4 files changed, 13 insertions(+), 31 deletions(-) diff --git a/src/_util/__init__.py b/src/_util/__init__.py index 45fa6985b..e82c80d8c 100644 --- a/src/_util/__init__.py +++ b/src/_util/__init__.py @@ -169,7 +169,7 @@ def validate_ulid(v: Any) -> ULID: ), ] -Quantity = Annotated[Decimal, BeforeValidator(parse_quantity)] +Quantity = Annotated[Decimal, BeforeValidator(parse_quantity), PlainSerializer(lambda d: str(d))] def bytes_to_kb(value: int) -> int: diff --git a/src/deployment/kubernetes/__init__.py b/src/deployment/kubernetes/__init__.py index f90302c98..ca96b6c23 100644 --- a/src/deployment/kubernetes/__init__.py +++ b/src/deployment/kubernetes/__init__.py @@ -362,7 +362,7 @@ async def resize_autoscaler_vm( namespace: str, name: str, *, - cpu_milli: int | None, + cpu: Decimal | None, memory_bytes: int | None, ) -> NeonVM: """ @@ -373,13 +373,8 @@ async def resize_autoscaler_vm( vm_manifest = _build_autoscaler_vm_manifest(vm.model_dump(by_alias=True), namespace, name) guest_spec = vm_manifest.setdefault("spec", {}).setdefault("guest", {}) - cpu_block = guest_spec.setdefault("cpus", {}) - min_milli = guest.cpus.min_milli - max_milli = guest.cpus.max_milli - limit_milli = cpu_milli if cpu_milli is not None else guest.cpus.use_milli - cpu_block["min"] = _milli_to_cores(min_milli) - cpu_block["max"] = _milli_to_cores(max_milli) - cpu_block["limit"] = _milli_to_cores(limit_milli) + if cpu is not None: + guest_spec.setdefault("cpus", {})["limit"] = str(cpu) if memory_bytes is not None: slot_size_bytes = guest.slot_size_bytes @@ -444,11 +439,6 @@ async def apply_autoscaler_vm(self, namespace: str, name: str, vm_manifest: dict return NeonVM.model_validate(applied) -def _milli_to_cores(value: int) -> int | float: - cores = value / 1000 - return int(cores) if cores.is_integer() else cores - - def _build_autoscaler_vm_manifest(vm_obj: dict[str, Any], namespace: str, name: str) -> dict[str, Any]: """ Prepare a clean autoscaler VM manifest for patching. diff --git a/src/deployment/kubernetes/neonvm.py b/src/deployment/kubernetes/neonvm.py index f46ba3d0f..4dfb7a293 100644 --- a/src/deployment/kubernetes/neonvm.py +++ b/src/deployment/kubernetes/neonvm.py @@ -6,7 +6,7 @@ from kubernetes_asyncio.client.exceptions import ApiException from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, PlainSerializer, ValidationError -from ..._util import quantity_to_milli_cpu +from ..._util import Quantity, quantity_to_milli_cpu from ...exceptions import VelaKubernetesError from ._util import custom_api_client @@ -57,21 +57,9 @@ class NeonVMStatus(CamelModel): class GuestCPUs(CamelModel): - min: Any - use: Any - max: Any - - @property - def use_milli(self) -> int: - return _require_cpu_millis(self.use, "guest.cpus.use") - - @property - def min_milli(self) -> int: - return _require_cpu_millis(self.min, "guest.cpus.min") - - @property - def max_milli(self) -> int: - return _require_cpu_millis(self.max, "guest.cpus.max") + min: Quantity + use: Quantity + max: Quantity class MemorySlots(CamelModel): diff --git a/src/deployment/resize.py b/src/deployment/resize.py index f9fef0f15..2508f6a9a 100644 --- a/src/deployment/resize.py +++ b/src/deployment/resize.py @@ -6,6 +6,7 @@ """ import logging +from decimal import Decimal from asgiref.sync import async_to_sync from ulid import ULID @@ -25,7 +26,10 @@ async def _resize_cpu_memory(deployment_id: ULID, milli_vcpu: int | None, memory_bytes: int | None) -> None: namespace, autoscaler_vm_name = get_autoscaler_vm_identity(deployment_id) await kube_service.resize_autoscaler_vm( - namespace, autoscaler_vm_name, cpu_milli=milli_vcpu, memory_bytes=memory_bytes + namespace, + autoscaler_vm_name, + cpu=Decimal(milli_vcpu) / 1000 if milli_vcpu is not None else None, + memory_bytes=memory_bytes, )