diff --git a/.travis.yml b/.travis.yml index 879cb65c..a1d9bf44 100644 --- a/.travis.yml +++ b/.travis.yml @@ -10,7 +10,7 @@ before_script: - python tr_sys/manage.py migrate - docker run -d -p 5672:5672 rabbitmq - sleep 10 # give some time for server to come up - - cd tr_sys; celery -A tr_sys worker -l info & + - cd tr_sys; celery -A tr_sys worker -l info -Q celery,heavy & - sleep 10 # give some time for server to come up - cd ..; python tr_sys/manage.py runserver --noreload & - sleep 10 # give some time for server to come up diff --git a/config/start_celery.sh b/config/start_celery.sh index a3a51cff..1f0737b3 100755 --- a/config/start_celery.sh +++ b/config/start_celery.sh @@ -3,7 +3,10 @@ cd /ars/tr_sys # Start beat with log file celery -A tr_sys beat -l info -f /var/log/celerybeat.log & -# Start worker in foreground - -celery -A tr_sys worker -l info --concurrency=10 --prefetch-multiplier=1 +# Start worker in foreground. Single-worker deployments (docker-compose, the +# legacy deploy/ chart) must consume BOTH queues: merge-and-post-process and +# ingest-ara-response are routed to `heavy` (tr_sys/celery.py task_routes) and +# would otherwise sit in RabbitMQ with no consumer. helm/ars runs a separate +# worker pool per queue instead. +celery -A tr_sys worker -l info -Q celery,heavy --concurrency=10 --prefetch-multiplier=1 diff --git a/config/tr_sys_settings.py b/config/tr_sys_settings.py index 881e98aa..bbae193a 100755 --- a/config/tr_sys_settings.py +++ b/config/tr_sys_settings.py @@ -42,7 +42,6 @@ 'django.contrib.messages', 'django.contrib.staticfiles', 'corsheaders', - #'channels', 'tr_ars.apps.ARSConfig', #'tr_ars.default_ars_app.ars_app.AppConfig', 'tr_ara_aragorn.aragorn_app.AppConfig', @@ -100,17 +99,6 @@ ] WSGI_APPLICATION = 'tr_sys.wsgi.application' -# Channels -ASGI_APPLICATION = 'tr_sys.routing.application' - -CHANNEL_LAYERS = { - 'default': { - 'BACKEND': 'channels_redis.core.RedisChannelLayer', - 'CONFIG': { - "hosts": [('redis', 6379)], - }, - }, -} # Database # https://docs.djangoproject.com/en/1.11/ref/settings/#databases diff --git a/docker-compose.yml b/docker-compose.yml index 3c62da00..6915029b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -34,12 +34,6 @@ services: - ./config/tr_sys_settings.py:/ars/tr_sys/tr_sys/settings.py depends_on: - arsdb - - redis - redis: - image: redis:6.2 - container_name: ars_redis - expose: - - 6379 celery_worker: command: sh -c "wait-for rabbitmq:5672 -t 30 && wait-for arsserver:8000 -t 60 -- /bin/bash /ars/config/start_celery.sh" container_name: celery_worker diff --git a/helm/ars/Chart.yaml b/helm/ars/Chart.yaml new file mode 100644 index 00000000..70835cec --- /dev/null +++ b/helm/ars/Chart.yaml @@ -0,0 +1,11 @@ +apiVersion: v2 +name: ars +description: >- + Helm chart for the ARS (Autonomous Relay System) and its backing services. + Unlike deploy/ (the Jenkins-driven production chart), this chart is + self-contained: it deploys MySQL and RabbitMQ alongside the app so a + dev instance can be installed on any cluster with no externally supplied + config files. +type: application +version: 0.2.0 +appVersion: "1.42.1" diff --git a/helm/ars/README.md b/helm/ars/README.md new file mode 100644 index 00000000..a2f8af1f --- /dev/null +++ b/helm/ars/README.md @@ -0,0 +1,72 @@ +# ars (self-contained dev chart) + +A Helm chart that deploys the ARS **and all of its backing services** — MySQL +and RabbitMQ — so a dev instance comes up on any Kubernetes cluster with +no externally supplied files. It is the outcome of `deploy/HELM_DEPENDENCIES.md`; +the Jenkins-driven production chart in `deploy/` is untouched. + +Differences from `deploy/`: + +| | `deploy/` | `helm/ars/` | +|---|---|---| +| Topology | one pod, 4 containers over `localhost` | one pod per component (server / heavy worker / light worker / beat / mysql / rabbitmq) | +| MySQL | external, invisible to the chart | in-cluster StatefulSet + PVC by default, or `mysql.enabled=false` + `mysql.external.*` | +| RabbitMQ | sidecar, no volume | own pod with a PVC (celery is configured for persistent delivery) | +| Redis | sidecar (token gate + unused Channels layer) | gone; nothing in the app uses it anymore | +| `settings.py` | injected by Jenkins, `sed`-substituted | in the chart (`files/settings.py`), reads env vars; credentials via a Secret | +| Migrations | `manage.py migrate` in the server container command | post-install/pre-upgrade hook Job (keep migrations backward compatible) | +| Celery beat | started inside the worker script | own single-replica Deployment | +| Scaling | impossible (replicating the pod duplicates the broker) | `arsserver.replicas` / `celeryworkers..replicas` | + +## Quick start (kind / minikube / any dev cluster) + +```sh +# 1. Build the app image and make it available to the cluster +docker build -t relay_ars:latest . +kind load docker-image relay_ars:latest # or: minikube image load relay_ars:latest + +# 2. Install +helm install ars ./helm/ars -n ars --create-namespace + +# 3. Talk to it +kubectl -n ars port-forward svc/ars 8000:80 +curl http://localhost:8000/ars/api/health +``` + +On a cluster with ECR access, skip the build and use the CI image: + +```sh +helm install ars ./helm/ars -n ars --create-namespace \ + --set image.repository=853771734544.dkr.ecr.us-east-1.amazonaws.com/translator-ars \ + --set image.tag= --set image.pullPolicy=Always +``` + +## Using external backing services + +```yaml +mysql: + enabled: false + external: + host: my-rds.cluster-xyz.us-east-1.rds.amazonaws.com + port: 3306 + database: arsdb + username: ars +``` + +Passwords always come from the release Secret — set `secrets.mysqlPassword` +etc., or point `secrets.existingSecret` at a Secret you manage (keys: +`django-secret-key`, `aes-master-key`, `mysql-root-password`, `mysql-password`, +`rabbitmq-password`). + +## Notes and limits + +- **Dev credentials ship in `values.yaml`.** Override every `secrets.*` value + (or use `secrets.existingSecret`) for any instance others can reach. +- The in-cluster MySQL is a single node with no backups — fine for dev, not for + anything you'd miss. +- Queries fan out to the live ARAs/KPs selected by `env.TR_ENV` (SmartAPI + maturity level). +- OTEL trace export points at `otel.jaegerHost`; if no collector exists at that + address the exporter logs connection errors and the app carries on. +- `affinity`/`tolerations` are empty by default so pods schedule anywhere; copy + the node pinning from `deploy/values.yaml` if an environment needs it. diff --git a/helm/ars/files/settings.py b/helm/ars/files/settings.py new file mode 100644 index 00000000..cbdfa72d --- /dev/null +++ b/helm/ars/files/settings.py @@ -0,0 +1,201 @@ +""" +Django settings for tr_sys, as deployed by the helm/ars chart. + +This file is mounted over tr_sys/tr_sys/settings.py (same mechanism as the +production deploy/ chart) but takes every connection parameter and credential +from environment variables, which the chart supplies from values.yaml and its +Secret. It should track tr_sys/tr_sys/settings.py apart from those env reads. +""" + +import os +from .otel_config import configure_opentelemetry + +configure_opentelemetry() + +# Build paths inside the project like this: os.path.join(BASE_DIR, ...) +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +SECRET_KEY = os.environ.get('DJANGO_SECRET_KEY', 'ars-dev-insecure-secret-key') + +DEBUG = os.environ.get('DJANGO_DEBUG', 'true').lower() in ('1', 'true', 'yes') + +ALLOWED_HOSTS = [h.strip() for h in os.environ.get('DJANGO_ALLOWED_HOSTS', '*').split(',') if h.strip()] + +SECURE_PROXY_SSL_HEADER = ('HTTP_X_FORWARDED_PROTO', 'https') + +# Application definition + +INSTALLED_APPS = [ + 'django.contrib.admin', + 'django.contrib.auth', + 'django.contrib.contenttypes', + 'django.contrib.sessions', + 'django.contrib.messages', + 'django.contrib.staticfiles', + 'corsheaders', + 'tr_ars.apps.ARSConfig', + 'tr_ara_aragorn.aragorn_app.AppConfig', + 'tr_ara_arax.arax_app.AppConfig', + 'tr_ara_bte.bte_app.AppConfig', + 'tr_ara_improving.improving_app.AppConfig', + 'tr_ara_unsecret.unsecret_app.AppConfig', + 'tr_ara_wfr.wfr_app.AppConfig', + 'tr_ara_cqs.cqs_app.AppConfig', + 'tr_ara_shepherd_aragorn.shepherd_aragorn_app.AppConfig', + 'tr_ara_shepherd_arax.shepherd_arax_app.AppConfig', + 'tr_ara_shepherd_bte.shepherd_bte_app.AppConfig', + 'tr_kp_genetics.genetics_app.AppConfig', + 'tr_kp_clinical.clinical_app.AppConfig', + 'tr_kp_drug.drug_app.AppConfig', + 'tr_kp_molecular.molecular_app.AppConfig', + 'tr_kp_cam.cam_app.AppConfig', + 'tr_kp_textmining.textmining_app.AppConfig', + 'tr_kp_openpredict.openpredict_app.AppConfig', + 'tr_kp_cohd.cohd_app.AppConfig', + 'tr_kp_chp.chp_app.AppConfig', + 'django_celery_results', + 'markdownify', + 'django_celery_beat', +] + +MIDDLEWARE = [ + 'django.middleware.security.SecurityMiddleware', + 'django.contrib.sessions.middleware.SessionMiddleware', + 'corsheaders.middleware.CorsMiddleware', + 'django.middleware.common.CommonMiddleware', + 'django.middleware.csrf.CsrfViewMiddleware', + 'django.contrib.auth.middleware.AuthenticationMiddleware', + 'django.contrib.messages.middleware.MessageMiddleware', + 'django.middleware.clickjacking.XFrameOptionsMiddleware', +] + +ROOT_URLCONF = 'tr_sys.urls' + +CORS_ORIGIN_ALLOW_ALL = True # for further customization see https://pypi.org/project/django-cors-headers/ + +TEMPLATES = [ + { + 'BACKEND': 'django.template.backends.django.DjangoTemplates', + 'DIRS': [os.path.join(BASE_DIR, 'tr_ars','templates')], + 'APP_DIRS': True, + 'OPTIONS': { + 'context_processors': [ + 'django.template.context_processors.debug', + 'django.template.context_processors.request', + 'django.contrib.auth.context_processors.auth', + 'django.contrib.messages.context_processors.messages', + ], + }, + }, +] + +WSGI_APPLICATION = 'tr_sys.wsgi.application' + +# Database +# https://docs.djangoproject.com/en/1.11/ref/settings/#databases + +DATABASES = { + 'default': { + 'ENGINE': 'django.db.backends.mysql', + 'NAME': os.environ.get('ARS_DB_NAME', 'arsdb'), + 'USER': os.environ.get('ARS_DB_USER', 'ars'), + 'PASSWORD': os.environ.get('ARS_DB_PASSWORD', ''), + 'HOST': os.environ.get('ARS_DB_HOST', 'localhost'), + 'PORT': os.environ.get('ARS_DB_PORT', '3306'), + } +} + +DJANGO_LOG_LEVEL = os.environ.get('DJANGO_LOG_LEVEL', 'DEBUG') +LOGGING = { + 'formatters': { + 'simple': { + 'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}', + 'style': '{', + } + }, + 'version': 1, + 'disable_existing_loggers': False, + 'handlers': { + 'console': { + 'class': 'logging.StreamHandler', + 'formatter': 'simple' + } + }, + 'root': { + 'handlers': ['console'], + 'level': DJANGO_LOG_LEVEL, + }, + 'loggers': { + 'tr_ars.tasks': { + 'level': DJANGO_LOG_LEVEL, + 'handlers': ['console'], + }, + 'tr_ars.default_ars_app.api': { + 'level': DJANGO_LOG_LEVEL, + 'handlers': ['console'] + } + } +} + +# Password validation +# https://docs.djangoproject.com/en/1.11/ref/settings/#auth-password-validators + +AUTH_PASSWORD_VALIDATORS = [ + { + 'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator', + }, + { + 'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator', + }, + { + 'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator', + }, + { + 'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator', + }, +] + + +# Internationalization +# https://docs.djangoproject.com/en/1.11/topics/i18n/ + +LANGUAGE_CODE = 'en-us' + +TIME_ZONE = 'UTC' + +USE_I18N = True + +USE_L10N = True + +USE_TZ = True + +# markdowninfy + +MARKDOWNIFY_STRIP = False +MARKDOWNIFY_WHITELIST_TAGS = { + 'a', 'p', + 'h1', 'h2', 'h3','h4', 'h5', 'h6', 'h7', + 'ul', 'li', 'span', +} + +# Static files (CSS, JavaScript, Images) +# https://docs.djangoproject.com/en/1.11/howto/static-files/ + +STATIC_URL = '/static/' +STATIC_ROOT = os.path.join(BASE_DIR, "static") + +# Celery settings + +CELERY_RESULT_BACKEND = 'django-db' +CELERY_CACHE_BACKEND = 'django-cache' +CELERY_BROKER_URL = os.environ.get('CELERY_BROKER_URL', 'amqp://localhost') +CELERY_IMPORTS = [ + 'tr_ars.tasks', +] +# Other important shared settings +DATA_UPLOAD_MAX_MEMORY_SIZE=1073741824 +CELERY_TASK_ALWAYS_EAGER=False +CELERY_TASK_ACKS_LATE = True + +USE_CELERY = True +DEFAULT_HOST = os.environ.get('ARS_DEFAULT_HOST', 'http://localhost:8000') diff --git a/helm/ars/ncats-values.yaml b/helm/ars/ncats-values.yaml new file mode 100644 index 00000000..b16e9eea --- /dev/null +++ b/helm/ars/ncats-values.yaml @@ -0,0 +1,130 @@ +# NCATS / transltr.io overlay for the helm/ars chart. +# +# Mirrors the environment-specific settings from deploy/values.yaml (the +# Jenkins-driven chart): ECR images, prod-sized gunicorn, the application=ars +# node pinning, and the ALB ingress. Apply on top of the defaults: +# +# helm upgrade --install ars ./helm/ars -n ars -f ./helm/ars/ncats-values.yaml +# +# UPPERCASE_VALUE tokens are deliberately the same placeholder style the +# existing deploy/prepare.sh + deploy.sh sed flow substitutes, so that flow (or +# --set on the command line) carries over unchanged. Everything not listed here +# inherits the dev defaults from values.yaml. + +image: + repository: 853771734544.dkr.ecr.us-east-1.amazonaws.com/translator-ars + tag: DOCKER_VERSION_VALUE + pullPolicy: Always + +env: + # maturity per environment: ci=staging, test=testing, prod=production + TR_ENV: TR_ENV_VALUE + ANNOTATOR_QUERY_BACKEND: elasticsearch + +django: + debug: "false" + allowedHosts: "ars.transltr.io,ars-dev.transltr.io,ARS_HOSTNAME_VALUE" + logLevel: DEBUG + +secrets: + # In NCATS clusters credentials should come from a managed Secret, not from + # a values file. Create it with the keys: django-secret-key, aes-master-key, + # mysql-root-password, mysql-password, rabbitmq-password. + existingSecret: ars-secrets + +arsserver: + replicas: 1 + gunicorn: + # Prod sizing from deploy/values.yaml (concurrency ~= workers * threads). + # 8 processes rather than one per vCPU because the celery workers share the + # same nodes and do the CPU-heavy merge work. + workerClass: gthread + workers: 8 + threads: 4 + timeout: 360 + gracefulTimeout: 120 + maxRequests: 1000 + maxRequestsJitter: 100 + resources: + requests: + cpu: "2" + memory: 4Gi + limits: + cpu: "4" + memory: 8Gi + +celeryworkers: + heavy: + queues: heavy + replicas: 4 + concurrency: 5 + maxTasksPerChild: 20 + resources: + requests: + cpu: "2" + memory: 4Gi + limits: + cpu: "4" + memory: 12Gi + light: + queues: celery + replicas: 2 + concurrency: 4 + resources: + requests: + cpu: 250m + memory: 1Gi + limits: + cpu: "1" + memory: 2Gi + +mysql: + # NCATS environments use the external (RDS) database, as the Jenkins deploy does. + enabled: false + external: + host: ARS_DB_HOST_VALUE + port: 3306 + database: ARS_DB_NAME_VALUE + username: ARS_DB_USER_VALUE + +rabbitmq: + enabled: true + image: + repository: 853771734544.dkr.ecr.us-east-1.amazonaws.com/translator-rabbitmq + tag: "4.0" + pullPolicy: Always + +ingress: + enabled: true + className: alb + annotations: + alb.ingress.kubernetes.io/scheme: internet-facing + alb.ingress.kubernetes.io/tags: Name=ARS_ALB_TAG_VALUE,Environment=ENVIRONMENT_TAG_VALUE + alb.ingress.kubernetes.io/security-groups: ARS_ALB_SG_VALUE + alb.ingress.kubernetes.io/ssl-policy: ELBSecurityPolicy-TLS-1-2-Ext-2018-06 + alb.ingress.kubernetes.io/listen-ports: '[{"HTTP": 80}, {"HTTPS":443}]' + # AWS LB Controller >= 2.2 replacement for the legacy actions.ssl-redirect + # action-service pair the deploy/ chart used + alb.ingress.kubernetes.io/ssl-redirect: '443' + alb.ingress.kubernetes.io/success-codes: '200' + alb.ingress.kubernetes.io/target-type: ip + alb.ingress.kubernetes.io/load-balancer-attributes: idle_timeout.timeout_seconds=600 + host: ARS_HOSTNAME_VALUE + path: / + pathType: Prefix + +# Pin every pod to the ARS node group, as deploy/values.yaml does. +affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - key: application + operator: In + values: + - ars +tolerations: + - key: "transltr" + value: "ars" + operator: "Equal" + effect: "NoSchedule" diff --git a/helm/ars/renci-values.yaml b/helm/ars/renci-values.yaml new file mode 100644 index 00000000..7193b7df --- /dev/null +++ b/helm/ars/renci-values.yaml @@ -0,0 +1,161 @@ +# RENCI Sterling overlay for the helm/ars chart. +# +# Conventions here follow the unencrypted parts of +# https://github.com/helxplatform/translator-devops (the *-populated.yaml files +# there are git-crypt encrypted, so anything environment-secret is a fill-in): +# - .apps.renci.org hostnames, cert-manager cluster-issuer "letsencrypt", +# TLS secret named "-tls" +# - nginx ingress with long proxy timeouts (Translator queries are long-running) +# - storageClass "basic" for ReadWriteOnce PVCs +# (https://wiki.renci.org/index.php?title=Kubernetes_Cloud/Sterling#Persistent_Storage) +# - Jaeger deployed from the translator-devops jaeger chart; the ARS code +# default (http://jaeger-otel-collector.sri:4317) already points at it +# +# helm upgrade --install ars ./helm/ars -n -f ./helm/ars/renci-values.yaml +# +# Everything not listed here inherits the dev defaults from values.yaml, +# including the in-cluster MySQL/RabbitMQ (Sterling has no RDS). + +image: + # There is no public ARS image and Sterling cannot pull from the NCATS ECR, + # so push a build somewhere Sterling can reach, e.g. RENCI Harbor: + # docker build -t containers.renci.org/translator/relay-ars: . + # docker push containers.renci.org/translator/relay-ars: + repository: containers.renci.org/translator/relay-ars + tag: 0.0.6 + pullPolicy: IfNotPresent + +# Needed if the Harbor project is private (kubectl create secret docker-registry ...) +imagePullSecrets: [] +# - name: renci-harbor + +env: + # SmartAPI maturity: RENCI instances run against development-maturity ARAs/KPs + TR_ENV: development + +otel: + # jaeger chart from translator-devops. On RENCI the collector lives in the same + # namespace as the ARS (translator-exp), so the bare service name resolves; the + # chart default targets "sri" instead, which is where NCATS runs it. If the ARS + # ever moves to a different namespace than the collector, qualify this + # (e.g. "http://jaeger-otel-collector.translator-exp") -- a bare name will not + # resolve across namespaces, and the exporter fails silently into the logs. + jaegerHost: "http://jaeger-otel-collector" + jaegerPort: "4317" + +django: + debug: "false" + allowedHosts: "ars.apps.renci.org" + logLevel: INFO + +secrets: + # Keep real credentials out of this file (translator-devops encrypts its + # populated values with git-crypt; we don't have that set up here). + # Create the Secret out-of-band with keys: django-secret-key, aes-master-key, + # mysql-root-password, mysql-password, rabbitmq-password. + existingSecret: ars-secrets + +# Moderate sizing -- Sterling namespaces carry resource quotas +arsserver: + replicas: 2 + gunicorn: + workers: 8 + threads: 4 + resources: + requests: + cpu: "2" + memory: 3Gi + limits: + cpu: "4" + memory: 6Gi + +celeryworkers: + heavy: + queues: heavy + replicas: 3 + concurrency: 5 + maxTasksPerChild: 20 + resources: + requests: + cpu: "3" + memory: 10Gi + limits: + cpu: "6" + memory: 16Gi + light: + queues: celery + replicas: 2 + concurrency: 4 + resources: + requests: + cpu: 250m + memory: 1Gi + limits: + cpu: "1" + memory: 2Gi + +mysql: + enabled: true + persistence: + # Messages are stored as compressed TRAPI blobs. Pruning of non-retained + # messages (see Message.retain) is handled outside this repo, so it may not + # be wired up for this in-cluster database -- size for unbounded growth + # until that is confirmed. "basic" may not support online expansion. + size: 120Gi + storageClass: basic + # The busiest component: all 15 heavy worker slots and 16 gunicorn workers + # contend here, and merges hold a row lock while reading and writing multi-MB + # blobs. The MySQL default buffer pool is 128MB, which sends nearly every one + # of those reads to disk; keep it at roughly half the memory limit. + extraArgs: + - "--innodb-buffer-pool-size=4G" + resources: + requests: + cpu: "1" + memory: 6Gi + limits: + cpu: "4" + memory: 8Gi + +rabbitmq: + enabled: true + persistence: + size: 30Gi + storageClass: basic + resources: + requests: + cpu: 500m + memory: 1Gi + limits: + cpu: "2" + memory: 4Gi + +ingress: + enabled: true + # Sterling's nginx ingress controller is the cluster default + className: "" + annotations: + cert-manager.io/cluster-issuer: letsencrypt + # Long-running Translator queries: same timeout set the other + # translator-devops charts use (aragorn, answer-appraiser) + nginx.ingress.kubernetes.io/proxy-connect-timeout: "86400" + nginx.ingress.kubernetes.io/proxy-read-timeout: "86400" + nginx.ingress.kubernetes.io/proxy-send-timeout: "86400" + nginx.ingress.kubernetes.io/send-timeout: "86400" + nginx.ingress.kubernetes.io/connection-proxy-header: keep-alive + nginx.ingress.kubernetes.io/service-upstream: "true" + nginx.ingress.kubernetes.io/upstream-keepalive-timeout: "86400" + # TRAPI messages are large + nginx.ingress.kubernetes.io/proxy-body-size: 512m + host: ars.apps.renci.org + path: / + pathType: ImplementationSpecific + tls: + - secretName: ars.apps.renci.org-tls + hosts: + - ars.apps.renci.org + +# No node pinning on Sterling (the application=ars affinity is an NCATS EKS thing) +affinity: {} +tolerations: [] +nodeSelector: {} \ No newline at end of file diff --git a/helm/ars/templates/NOTES.txt b/helm/ars/templates/NOTES.txt new file mode 100644 index 00000000..bec99921 --- /dev/null +++ b/helm/ars/templates/NOTES.txt @@ -0,0 +1,30 @@ +ARS {{ .Chart.AppVersion }} deployed as release {{ .Release.Name }} in namespace {{ .Release.Namespace }}. + +Components: + {{ include "ars.fullname" . }}-server Django/gunicorn API ({{ .Values.arsserver.replicas }} replica(s)) +{{- range $name, $pool := .Values.celeryworkers }} + {{ include "ars.fullname" $ }}-worker-{{ $name }} celery worker, queue(s) {{ $pool.queues }} ({{ $pool.replicas }} x {{ $pool.concurrency }} slots) +{{- end }} + {{ include "ars.fullname" . }}-beat celery beat (always 1) +{{- if .Values.mysql.enabled }} + {{ include "ars.fullname" . }}-mysql MySQL {{ .Values.mysql.image.tag }} (in-cluster, PVC {{ .Values.mysql.persistence.size }}) +{{- else }} + external MySQL at {{ .Values.mysql.external.host }}:{{ .Values.mysql.external.port }} +{{- end }} +{{- if .Values.rabbitmq.enabled }} + {{ include "ars.fullname" . }}-rabbitmq RabbitMQ {{ .Values.rabbitmq.image.tag }} (in-cluster, PVC {{ .Values.rabbitmq.persistence.size }}) +{{- end }} + +Migrations run automatically via the {{ include "ars.fullname" . }}-migrate Job +(post-install/pre-upgrade hook); on a first install give it a minute before use. + +To reach the API: +{{- if .Values.ingress.enabled }} + http://{{ .Values.ingress.host }}{{ .Values.ingress.path }} +{{- else }} + kubectl -n {{ .Release.Namespace }} port-forward svc/{{ include "ars.fullname" . }} 8000:{{ .Values.service.port }} + curl http://localhost:8000/ars/api/health +{{- end }} + +NOTE: the default credentials in values.yaml (secrets.*) are for throwaway dev +instances only -- override them or set secrets.existingSecret for anything shared. diff --git a/helm/ars/templates/_helpers.tpl b/helm/ars/templates/_helpers.tpl new file mode 100644 index 00000000..e8cba668 --- /dev/null +++ b/helm/ars/templates/_helpers.tpl @@ -0,0 +1,209 @@ +{{/* +Expand the name of the chart. +*/}} +{{- define "ars.name" -}} +{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Create a default fully qualified app name (63-char limited per DNS naming spec). +*/}} +{{- define "ars.fullname" -}} +{{- if .Values.fullnameOverride }} +{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }} +{{- else }} +{{- $name := default .Chart.Name .Values.nameOverride }} +{{- if contains $name .Release.Name }} +{{- .Release.Name | trunc 63 | trimSuffix "-" }} +{{- else }} +{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }} +{{- end }} +{{- end }} +{{- end }} + +{{/* +Chart name and version as used by the chart label. +*/}} +{{- define "ars.chart" -}} +{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Common labels +*/}} +{{- define "ars.labels" -}} +helm.sh/chart: {{ include "ars.chart" . }} +{{ include "ars.selectorLabels" . }} +{{- if .Chart.AppVersion }} +app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} +{{- end }} +app.kubernetes.io/managed-by: {{ .Release.Service }} +{{- end }} + +{{/* +Selector labels (shared across components; add app.kubernetes.io/component per workload) +*/}} +{{- define "ars.selectorLabels" -}} +app.kubernetes.io/name: {{ include "ars.name" . }} +app.kubernetes.io/instance: {{ .Release.Name }} +{{- end }} + +{{/* +Name of the Secret holding all credentials. +*/}} +{{- define "ars.secretName" -}} +{{- if .Values.secrets.existingSecret }} +{{- .Values.secrets.existingSecret }} +{{- else }} +{{- printf "%s-secrets" (include "ars.fullname" .) }} +{{- end }} +{{- end }} + +{{/* ---- dependency coordinates: in-cluster Service DNS or the external value ---- */}} + +{{- define "ars.mysqlHost" -}} +{{- if .Values.mysql.enabled }} +{{- printf "%s-mysql" (include "ars.fullname" .) }} +{{- else }} +{{- required "mysql.external.host is required when mysql.enabled=false" .Values.mysql.external.host }} +{{- end }} +{{- end }} + +{{- define "ars.mysqlPort" -}} +{{- if .Values.mysql.enabled }}3306{{- else }}{{ .Values.mysql.external.port }}{{- end }} +{{- end }} + +{{- define "ars.mysqlDatabase" -}} +{{- if .Values.mysql.enabled }}{{ .Values.mysql.auth.database }}{{- else }}{{ .Values.mysql.external.database }}{{- end }} +{{- end }} + +{{- define "ars.mysqlUser" -}} +{{- if .Values.mysql.enabled }}{{ .Values.mysql.auth.username }}{{- else }}{{ .Values.mysql.external.username }}{{- end }} +{{- end }} + +{{- define "ars.rabbitmqHost" -}} +{{- if .Values.rabbitmq.enabled }} +{{- printf "%s-rabbitmq" (include "ars.fullname" .) }} +{{- else }} +{{- required "rabbitmq.external.host is required when rabbitmq.enabled=false" .Values.rabbitmq.external.host }} +{{- end }} +{{- end }} + +{{- define "ars.rabbitmqPort" -}} +{{- if .Values.rabbitmq.enabled }}5672{{- else }}{{ .Values.rabbitmq.external.port }}{{- end }} +{{- end }} + +{{- define "ars.rabbitmqUser" -}} +{{- if .Values.rabbitmq.enabled }}{{ .Values.rabbitmq.auth.username }}{{- else }}{{ .Values.rabbitmq.external.username }}{{- end }} +{{- end }} + +{{/* +Environment shared by every app container (web, worker, beat, migrate job). +Secrets come in via secretKeyRef; the broker URL is assembled with $(VAR) +expansion so the password never appears in the manifest. +*/}} +{{- define "ars.commonEnv" -}} +- name: ARS_DB_HOST + value: {{ include "ars.mysqlHost" . | quote }} +- name: ARS_DB_PORT + value: {{ include "ars.mysqlPort" . | quote }} +- name: ARS_DB_NAME + value: {{ include "ars.mysqlDatabase" . | quote }} +- name: ARS_DB_USER + value: {{ include "ars.mysqlUser" . | quote }} +- name: ARS_DB_PASSWORD + valueFrom: + secretKeyRef: + name: {{ include "ars.secretName" . }} + key: mysql-password +- name: DJANGO_SECRET_KEY + valueFrom: + secretKeyRef: + name: {{ include "ars.secretName" . }} + key: django-secret-key +- name: DJANGO_DEBUG + value: {{ .Values.django.debug | quote }} +- name: POD_IP + valueFrom: + fieldRef: + fieldPath: status.podIP +# django.allowedHosts only needs the PUBLIC hostname(s). The in-cluster addresses +# are appended here: kubelet probes reach the pod by IP, and the app calls itself +# by service name (see ARS_DEFAULT_HOST below). Django rejects any Host header not +# in this list with DisallowedHost, which would fail the readiness probe forever. +- name: DJANGO_ALLOWED_HOSTS + value: "{{ .Values.django.allowedHosts }},{{ include "ars.fullname" . }},{{ include "ars.fullname" . }}.{{ .Release.Namespace }}.svc.cluster.local,localhost,127.0.0.1,$(POD_IP)" +- name: DJANGO_LOG_LEVEL + value: {{ .Values.django.logLevel | quote }} +- name: RABBITMQ_USER + value: {{ include "ars.rabbitmqUser" . | quote }} +- name: RABBITMQ_PASSWORD + valueFrom: + secretKeyRef: + name: {{ include "ars.secretName" . }} + key: rabbitmq-password +- name: CELERY_BROKER_URL + value: "amqp://$(RABBITMQ_USER):$(RABBITMQ_PASSWORD)@{{ include "ars.rabbitmqHost" . }}:{{ include "ars.rabbitmqPort" . }}//" +- name: AES_MASTER_KEY + valueFrom: + secretKeyRef: + name: {{ include "ars.secretName" . }} + key: aes-master-key +# Base URL the ARS uses to reach ITSELF. Two things are built from it (tr_ars/tasks.py): +# 1. outbound: agent URIs are relative ("/ara-aragorn/api/"), so this + the path +# hits this deployment's own per-agent proxy views, which forward to the real ARA; +# 2. the callback URL handed to each ARA so it can POST results back -- and +# default_ars_app/api.py forwards that callback verbatim to the ARA. +# Because of (2) the ARAs must be able to resolve this address. The default is the +# cluster-internal FQDN, which resolves from ANY namespace in the cluster (a bare +# service name only resolves within this one, so ARAs in another namespace would +# silently fail to call back and every child would sit at Running until +# catch_timeout marks it 598). For ARAs outside the cluster set django.defaultHost +# to the public URL instead -- it is one or the other, not both. +- name: ARS_DEFAULT_HOST + value: {{ .Values.django.defaultHost | default (printf "http://%s.%s.svc.cluster.local:%v" (include "ars.fullname" .) .Release.Namespace .Values.service.port) | quote }} +- name: ARS_MERGE_LOCK_MAX_RETRIES + value: {{ .Values.merge.lockMaxRetries | quote }} +- name: ARS_MERGE_ERROR_MAX_RETRIES + value: {{ .Values.merge.errorMaxRetries | quote }} +- name: JAEGER_HOST + value: {{ .Values.otel.jaegerHost | quote }} +- name: JAEGER_PORT + value: {{ .Values.otel.jaegerPort | quote }} +{{- range $key, $value := .Values.env }} +- name: {{ $key }} + value: {{ $value | quote }} +{{- end }} +{{- with .Values.extraEnv }} +{{ toYaml . }} +{{- end }} +{{- end }} + +{{/* +Pod scheduling knobs shared by every pod in the release. +*/}} +{{- define "ars.podScheduling" -}} +{{- with .Values.imagePullSecrets }} +imagePullSecrets: + {{- toYaml . | nindent 2 }} +{{- end }} +{{- with .Values.nodeSelector }} +nodeSelector: + {{- toYaml . | nindent 2 }} +{{- end }} +{{- with .Values.affinity }} +affinity: + {{- toYaml . | nindent 2 }} +{{- end }} +{{- with .Values.tolerations }} +tolerations: + {{- toYaml . | nindent 2 }} +{{- end }} +{{- end }} + +{{/* +The settings.py volume + a checksum helper for pod-restart-on-change. +*/}} +{{- define "ars.settingsChecksum" -}} +{{ .Files.Get "files/settings.py" | sha256sum }} +{{- end }} diff --git a/helm/ars/templates/arsserver.yaml b/helm/ars/templates/arsserver.yaml new file mode 100644 index 00000000..2d834290 --- /dev/null +++ b/helm/ars/templates/arsserver.yaml @@ -0,0 +1,91 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "ars.fullname" . }}-server + labels: + {{- include "ars.labels" . | nindent 4 }} + app.kubernetes.io/component: server +spec: + replicas: {{ .Values.arsserver.replicas }} + strategy: + type: Recreate + selector: + matchLabels: + {{- include "ars.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: server + template: + metadata: + labels: + {{- include "ars.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: server + annotations: + checksum/settings: {{ include "ars.settingsChecksum" . }} + spec: + {{- include "ars.podScheduling" . | nindent 6 }} + # Keep above gunicorn's --graceful-timeout so in-flight requests can finish. + # With Recreate, new pods only start once old ones are gone, so a longer + # grace period slows the upgrade but never overlaps pods against the quota. + terminationGracePeriodSeconds: {{ .Values.arsserver.terminationGracePeriodSeconds | default 130 }} + initContainers: + - name: wait-for-mysql + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + command: ["/bin/wait-for", "{{ include "ars.mysqlHost" . }}:{{ include "ars.mysqlPort" . }}", "-t", "300"] + - name: wait-for-rabbitmq + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + command: ["/bin/wait-for", "{{ include "ars.rabbitmqHost" . }}:{{ include "ars.rabbitmqPort" . }}", "-t", "300"] + containers: + - name: arsserver + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + # Migrations run in the migrate Job (templates/migrate-job.yaml), not here, + # so replicas > 1 never race on the schema. + command: ["/bin/sh"] + args: + - "-c" + - >- + exec gunicorn tr_sys.wsgi:application + --chdir /ars/tr_sys + --bind 0.0.0.0:{{ .Values.arsserver.containerPort }} + --worker-class {{ .Values.arsserver.gunicorn.workerClass | default "gthread" }} + --workers {{ .Values.arsserver.gunicorn.workers | default 4 }} + --threads {{ .Values.arsserver.gunicorn.threads | default 8 }} + --timeout {{ .Values.arsserver.gunicorn.timeout | default 360 }} + --graceful-timeout {{ .Values.arsserver.gunicorn.gracefulTimeout | default 120 }} + --max-requests {{ .Values.arsserver.gunicorn.maxRequests | default 1000 }} + --max-requests-jitter {{ .Values.arsserver.gunicorn.maxRequestsJitter | default 100 }} + --access-logfile - --error-logfile - + ports: + - name: http + containerPort: {{ .Values.arsserver.containerPort }} + env: + {{- include "ars.commonEnv" . | nindent 12 }} + volumeMounts: + - name: settings + mountPath: /ars/tr_sys/tr_sys/settings.py + subPath: settings.py + livenessProbe: + tcpSocket: + port: http + initialDelaySeconds: 30 + periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 3 + readinessProbe: + # Checks DB connectivity and an in-process celery ping + httpGet: + path: /ars/api/health + port: http + initialDelaySeconds: 20 + periodSeconds: 30 + timeoutSeconds: 10 + failureThreshold: 3 + {{- with .Values.arsserver.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + volumes: + - name: settings + configMap: + name: {{ include "ars.fullname" . }}-settings diff --git a/helm/ars/templates/celerybeat.yaml b/helm/ars/templates/celerybeat.yaml new file mode 100644 index 00000000..364b4774 --- /dev/null +++ b/helm/ars/templates/celerybeat.yaml @@ -0,0 +1,54 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "ars.fullname" . }}-beat + labels: + {{- include "ars.labels" . | nindent 4 }} + app.kubernetes.io/component: beat +spec: + # Never scale this: N beat schedulers fire every periodic task N times. + replicas: 1 + strategy: + type: Recreate + selector: + matchLabels: + {{- include "ars.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: beat + template: + metadata: + labels: + {{- include "ars.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: beat + annotations: + checksum/settings: {{ include "ars.settingsChecksum" . }} + spec: + {{- include "ars.podScheduling" . | nindent 6 }} + initContainers: + - name: wait-for-rabbitmq + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + command: ["/bin/wait-for", "{{ include "ars.rabbitmqHost" . }}:{{ include "ars.rabbitmqPort" . }}", "-t", "300"] + containers: + - name: celerybeat + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + command: ["/bin/sh", "-c"] + args: + - >- + cd /ars/tr_sys && + exec celery -A tr_sys beat -l info + -s /tmp/celerybeat-schedule + env: + {{- include "ars.commonEnv" . | nindent 12 }} + volumeMounts: + - name: settings + mountPath: /ars/tr_sys/tr_sys/settings.py + subPath: settings.py + {{- with .Values.celerybeat.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + volumes: + - name: settings + configMap: + name: {{ include "ars.fullname" . }}-settings diff --git a/helm/ars/templates/celeryworker.yaml b/helm/ars/templates/celeryworker.yaml new file mode 100644 index 00000000..ff0f6723 --- /dev/null +++ b/helm/ars/templates/celeryworker.yaml @@ -0,0 +1,87 @@ +{{- /* +One Deployment per worker pool in .Values.celeryworkers. Each pool consumes only +the queues in its `queues` value, so the `heavy` pool's replicas x concurrency is +the cluster-wide cap on concurrent memory-heavy work. +Beat runs in its own single-replica Deployment. +*/ -}} +{{- $root := . -}} +{{- range $name, $pool := .Values.celeryworkers }} +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "ars.fullname" $root }}-worker-{{ $name }} + labels: + {{- include "ars.labels" $root | nindent 4 }} + app.kubernetes.io/component: worker + app.kubernetes.io/pool: {{ $name }} +spec: + replicas: {{ $pool.replicas }} + strategy: + type: Recreate + selector: + matchLabels: + {{- include "ars.selectorLabels" $root | nindent 6 }} + app.kubernetes.io/component: worker + app.kubernetes.io/pool: {{ $name }} + template: + metadata: + labels: + {{- include "ars.selectorLabels" $root | nindent 8 }} + app.kubernetes.io/component: worker + app.kubernetes.io/pool: {{ $name }} + annotations: + checksum/settings: {{ include "ars.settingsChecksum" $root }} + spec: + {{- include "ars.podScheduling" $root | nindent 6 }} + # On SIGTERM celery stops taking tasks and finishes the running ones; anything + # still running at the end of this is SIGKILLed and redelivered (acks_late). + # With Recreate, new pods only start once old ones are gone, so a longer + # grace period slows the upgrade but never overlaps pods against the quota. + terminationGracePeriodSeconds: {{ $pool.terminationGracePeriodSeconds | default 60 }} + initContainers: + - name: wait-for-mysql + image: "{{ $root.Values.image.repository }}:{{ $root.Values.image.tag }}" + imagePullPolicy: {{ $root.Values.image.pullPolicy }} + command: ["/bin/wait-for", "{{ include "ars.mysqlHost" $root }}:{{ include "ars.mysqlPort" $root }}", "-t", "300"] + - name: wait-for-rabbitmq + image: "{{ $root.Values.image.repository }}:{{ $root.Values.image.tag }}" + imagePullPolicy: {{ $root.Values.image.pullPolicy }} + command: ["/bin/wait-for", "{{ include "ars.rabbitmqHost" $root }}:{{ include "ars.rabbitmqPort" $root }}", "-t", "300"] + containers: + - name: celeryworker + image: "{{ $root.Values.image.repository }}:{{ $root.Values.image.tag }}" + imagePullPolicy: {{ $root.Values.image.pullPolicy }} + # `exec` makes celery PID 1: sh as PID 1 doesn't forward SIGTERM, so without + # it celery never gets a warm shutdown and is SIGKILLed at the grace period. + # No livenessProbe: once celery is PID 1 the container exits if it dies. + command: ["/bin/sh", "-c"] + args: + - >- + cd /ars/tr_sys && + exec celery -A tr_sys worker -l info + -Q {{ $pool.queues }} + -n {{ $name }}@%h + --concurrency={{ $pool.concurrency }} + --prefetch-multiplier=1 + {{- if $pool.maxTasksPerChild }} + --max-tasks-per-child={{ $pool.maxTasksPerChild }} + {{- end }} + {{- if $pool.maxMemoryPerChildKb }} + --max-memory-per-child={{ $pool.maxMemoryPerChildKb }} + {{- end }} + env: + {{- include "ars.commonEnv" $root | nindent 12 }} + volumeMounts: + - name: settings + mountPath: /ars/tr_sys/tr_sys/settings.py + subPath: settings.py + {{- with $pool.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + volumes: + - name: settings + configMap: + name: {{ include "ars.fullname" $root }}-settings +{{- end }} \ No newline at end of file diff --git a/helm/ars/templates/configmap.yaml b/helm/ars/templates/configmap.yaml new file mode 100644 index 00000000..80d3f77e --- /dev/null +++ b/helm/ars/templates/configmap.yaml @@ -0,0 +1,9 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: {{ include "ars.fullname" . }}-settings + labels: + {{- include "ars.labels" . | nindent 4 }} +data: + settings.py: | +{{ .Files.Get "files/settings.py" | indent 4 }} diff --git a/helm/ars/templates/ingress.yaml b/helm/ars/templates/ingress.yaml new file mode 100644 index 00000000..05add0d5 --- /dev/null +++ b/helm/ars/templates/ingress.yaml @@ -0,0 +1,31 @@ +{{- if .Values.ingress.enabled }} +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: {{ include "ars.fullname" . }} + labels: + {{- include "ars.labels" . | nindent 4 }} + {{- with .Values.ingress.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + {{- with .Values.ingress.className }} + ingressClassName: {{ . }} + {{- end }} + {{- with .Values.ingress.tls }} + tls: + {{- toYaml . | nindent 4 }} + {{- end }} + rules: + - host: {{ .Values.ingress.host | quote }} + http: + paths: + - path: {{ .Values.ingress.path }} + pathType: {{ .Values.ingress.pathType }} + backend: + service: + name: {{ include "ars.fullname" . }} + port: + number: {{ .Values.service.port }} +{{- end }} diff --git a/helm/ars/templates/migrate-job.yaml b/helm/ars/templates/migrate-job.yaml new file mode 100644 index 00000000..4dcd0946 --- /dev/null +++ b/helm/ars/templates/migrate-job.yaml @@ -0,0 +1,57 @@ +{{- if .Values.migrations.enabled }} +apiVersion: batch/v1 +kind: Job +metadata: + name: {{ include "ars.fullname" . }}-migrate + labels: + {{- include "ars.labels" . | nindent 4 }} + app.kubernetes.io/component: migrate + annotations: + # install: post-install, because the in-cluster MySQL doesn't exist yet + # pre-install. The web pods start in parallel; on a fresh database requests + # that touch not-yet-created tables can 500 until this completes. + # upgrade: pre-upgrade, so it runs against the MySQL that is already up and + # BEFORE any manifest changes. As post-upgrade it raced the StatefulSet + # rollout (an upgrade that touches mysql.* restarts that pod) and hung in + # wait-for-mysql; it also let new app code query columns not yet created. + # Consequence: migrations must stay backward compatible with the previous + # release's code (additive), since old pods keep running against them. + helm.sh/hook: post-install,pre-upgrade + helm.sh/hook-weight: "0" + helm.sh/hook-delete-policy: before-hook-creation,hook-succeeded +spec: + backoffLimit: 3 + template: + metadata: + labels: + {{- include "ars.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: migrate + spec: + {{- include "ars.podScheduling" . | nindent 6 }} + restartPolicy: Never + initContainers: + - name: wait-for-mysql + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + command: ["/bin/wait-for", "{{ include "ars.mysqlHost" . }}:{{ include "ars.mysqlPort" . }}", "-t", "300"] + containers: + - name: migrate + image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" + imagePullPolicy: {{ .Values.image.pullPolicy }} + workingDir: /ars + command: ["python", "tr_sys/manage.py", "migrate", "--noinput"] + env: + {{- include "ars.commonEnv" . | nindent 12 }} + {{- with .Values.migrations.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + volumeMounts: + - name: settings + mountPath: /ars/tr_sys/tr_sys/settings.py + subPath: settings.py + volumes: + - name: settings + configMap: + name: {{ include "ars.fullname" . }}-settings +{{- end }} diff --git a/helm/ars/templates/mysql.yaml b/helm/ars/templates/mysql.yaml new file mode 100644 index 00000000..919c6510 --- /dev/null +++ b/helm/ars/templates/mysql.yaml @@ -0,0 +1,113 @@ +{{- if .Values.mysql.enabled }} +apiVersion: v1 +kind: Service +metadata: + name: {{ include "ars.fullname" . }}-mysql + labels: + {{- include "ars.labels" . | nindent 4 }} + app.kubernetes.io/component: mysql +spec: + ports: + - port: 3306 + targetPort: mysql + name: mysql + selector: + {{- include "ars.selectorLabels" . | nindent 4 }} + app.kubernetes.io/component: mysql +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: {{ include "ars.fullname" . }}-mysql + labels: + {{- include "ars.labels" . | nindent 4 }} + app.kubernetes.io/component: mysql +spec: + serviceName: {{ include "ars.fullname" . }}-mysql + replicas: 1 + selector: + matchLabels: + {{- include "ars.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: mysql + template: + metadata: + labels: + {{- include "ars.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: mysql + spec: + {{- include "ars.podScheduling" . | nindent 6 }} + terminationGracePeriodSeconds: 300 + containers: + - name: mysql + image: "{{ .Values.mysql.image.repository }}:{{ .Values.mysql.image.tag }}" + imagePullPolicy: {{ .Values.mysql.image.pullPolicy }} + # Match docker-compose.yml: PyMySQL-friendly auth plugin + args: + - "--default-authentication-plugin=mysql_native_password" + {{- range .Values.mysql.extraArgs }} + - {{ . | quote }} + {{- end }} + env: + - name: MYSQL_ROOT_PASSWORD + valueFrom: + secretKeyRef: + name: {{ include "ars.secretName" . }} + key: mysql-root-password + - name: MYSQL_DATABASE + value: {{ .Values.mysql.auth.database | quote }} + - name: MYSQL_USER + value: {{ .Values.mysql.auth.username | quote }} + - name: MYSQL_PASSWORD + valueFrom: + secretKeyRef: + name: {{ include "ars.secretName" . }} + key: mysql-password + ports: + - name: mysql + containerPort: 3306 + volumeMounts: + - name: data + mountPath: /var/lib/mysql + startupProbe: + exec: + command: + - sh + - -c + - mysqladmin ping -h 127.0.0.1 -u"$MYSQL_USER" -p"$MYSQL_PASSWORD" --silent + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 180 + livenessProbe: + exec: + command: + - sh + - -c + - mysqladmin ping -h 127.0.0.1 -u"$MYSQL_USER" -p"$MYSQL_PASSWORD" --silent + periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 3 + readinessProbe: + exec: + command: + - sh + - -c + - mysql -h 127.0.0.1 -u"$MYSQL_USER" -p"$MYSQL_PASSWORD" -e 'SELECT 1' + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 6 + {{- with .Values.mysql.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: ["ReadWriteOnce"] + {{- with .Values.mysql.persistence.storageClass }} + storageClassName: {{ . | quote }} + {{- end }} + resources: + requests: + storage: {{ .Values.mysql.persistence.size }} +{{- end }} diff --git a/helm/ars/templates/rabbitmq.yaml b/helm/ars/templates/rabbitmq.yaml new file mode 100644 index 00000000..5aca106f --- /dev/null +++ b/helm/ars/templates/rabbitmq.yaml @@ -0,0 +1,88 @@ +{{- if .Values.rabbitmq.enabled }} +apiVersion: v1 +kind: Service +metadata: + name: {{ include "ars.fullname" . }}-rabbitmq + labels: + {{- include "ars.labels" . | nindent 4 }} + app.kubernetes.io/component: rabbitmq +spec: + ports: + - port: 5672 + targetPort: amqp + name: amqp + selector: + {{- include "ars.selectorLabels" . | nindent 4 }} + app.kubernetes.io/component: rabbitmq +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: {{ include "ars.fullname" . }}-rabbitmq + labels: + {{- include "ars.labels" . | nindent 4 }} + app.kubernetes.io/component: rabbitmq +spec: + serviceName: {{ include "ars.fullname" . }}-rabbitmq + replicas: 1 + selector: + matchLabels: + {{- include "ars.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: rabbitmq + template: + metadata: + labels: + {{- include "ars.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: rabbitmq + spec: + {{- include "ars.podScheduling" . | nindent 6 }} + containers: + - name: rabbitmq + image: "{{ .Values.rabbitmq.image.repository }}:{{ .Values.rabbitmq.image.tag }}" + imagePullPolicy: {{ .Values.rabbitmq.image.pullPolicy }} + env: + # Applied on first initialization of the data dir only + - name: RABBITMQ_DEFAULT_USER + value: {{ .Values.rabbitmq.auth.username | quote }} + - name: RABBITMQ_DEFAULT_PASS + valueFrom: + secretKeyRef: + name: {{ include "ars.secretName" . }} + key: rabbitmq-password + ports: + - name: amqp + containerPort: 5672 + volumeMounts: + # Celery is configured with persistent delivery + acks-late + # (tr_sys/tr_sys/celery.py); a volume is what makes that real. + - name: data + mountPath: /var/lib/rabbitmq + livenessProbe: + tcpSocket: + port: amqp + initialDelaySeconds: 30 + periodSeconds: 20 + timeoutSeconds: 5 + failureThreshold: 3 + readinessProbe: + exec: + command: ["rabbitmq-diagnostics", "-q", "ping"] + initialDelaySeconds: 20 + periodSeconds: 15 + timeoutSeconds: 10 + {{- with .Values.rabbitmq.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: ["ReadWriteOnce"] + {{- with .Values.rabbitmq.persistence.storageClass }} + storageClassName: {{ . | quote }} + {{- end }} + resources: + requests: + storage: {{ .Values.rabbitmq.persistence.size }} +{{- end }} diff --git a/helm/ars/templates/secret.yaml b/helm/ars/templates/secret.yaml new file mode 100644 index 00000000..149a2b59 --- /dev/null +++ b/helm/ars/templates/secret.yaml @@ -0,0 +1,15 @@ +{{- if not .Values.secrets.existingSecret }} +apiVersion: v1 +kind: Secret +metadata: + name: {{ include "ars.fullname" . }}-secrets + labels: + {{- include "ars.labels" . | nindent 4 }} +type: Opaque +stringData: + django-secret-key: {{ .Values.secrets.djangoSecretKey | quote }} + aes-master-key: {{ .Values.secrets.aesMasterKey | quote }} + mysql-root-password: {{ .Values.secrets.mysqlRootPassword | quote }} + mysql-password: {{ .Values.secrets.mysqlPassword | quote }} + rabbitmq-password: {{ .Values.secrets.rabbitmqPassword | quote }} +{{- end }} diff --git a/helm/ars/templates/service.yaml b/helm/ars/templates/service.yaml new file mode 100644 index 00000000..d6387949 --- /dev/null +++ b/helm/ars/templates/service.yaml @@ -0,0 +1,17 @@ +apiVersion: v1 +kind: Service +metadata: + name: {{ include "ars.fullname" . }} + labels: + {{- include "ars.labels" . | nindent 4 }} + app.kubernetes.io/component: server +spec: + type: {{ .Values.service.type }} + ports: + - port: {{ .Values.service.port }} + targetPort: http + protocol: TCP + name: http + selector: + {{- include "ars.selectorLabels" . | nindent 4 }} + app.kubernetes.io/component: server diff --git a/helm/ars/values.yaml b/helm/ars/values.yaml new file mode 100644 index 00000000..a1d65846 --- /dev/null +++ b/helm/ars/values.yaml @@ -0,0 +1,240 @@ +# Dev-oriented defaults for the self-contained ARS chart. +# +# Everything needed for `helm install` on a plain cluster is here: in-cluster +# MySQL/RabbitMQ, dev credentials, and an app image name that matches a +# local `docker build -t relay_ars .` (load it into kind/minikube, or override +# image.repository/tag with an ECR image). +# +# The secrets block holds DEV-ONLY defaults. Override them (or set +# secrets.existingSecret) for any instance that is reachable by anyone else. + +nameOverride: "" +fullnameOverride: "" + +image: + # The ARS app image, used by the web server, celery worker, celery beat and + # the migration job. `relay_ars` matches the docker-compose build tag; for a + # shared cluster point this at the ECR repo instead, e.g. + # 853771734544.dkr.ecr.us-east-1.amazonaws.com/translator-ars + repository: relay_ars + tag: latest + pullPolicy: IfNotPresent + +imagePullSecrets: [] + +# Translator environment wiring (external HTTP services -- not deployed here) +env: + # SmartAPI maturity level: production | staging | testing | development + TR_ENV: development + # biothings_annotator backend: "biothings" (the package default) or + # "elasticsearch". With "elasticsearch" the package connects to + # ELASTICSEARCH_CONNECTION, which defaults to "in_cluster" + # (http://elasticsearch.es-core-components.svc.cluster.local:9200). + ANNOTATOR_QUERY_BACKEND: biothings +# Extra env vars appended verbatim to all app containers (list of EnvVar) +extraEnv: [] + +# OTLP trace export (tr_sys/tr_sys/otel_config.py). If the collector does not +# exist in your cluster the exporter just logs connection errors; harmless. +otel: + jaegerHost: "http://jaeger-otel-collector.sri" + jaegerPort: "4317" + +django: + debug: "true" + allowedHosts: "*" + logLevel: DEBUG + # Base URL the ARAs use to call this ARS back with results (see + # ARS_DEFAULT_HOST in _helpers.tpl). Leave EMPTY when the ARAs run in this + # cluster, it then defaults to the internal FQDN, which resolves from any + # namespace. Set it to the public URL (e.g. "https://ars.apps.renci.org") + # only when the ARAs are genuinely outside the cluster; that routes callbacks + # and proxy calls out through the ingress and back in. + defaultHost: "" + +secrets: + # If set, no Secret is rendered and this pre-existing Secret is used instead. + # It must contain the same keys the template creates (see templates/secret.yaml). + existingSecret: "" + # ------ dev-only defaults below ------ + djangoSecretKey: "ars-dev-insecure-secret-key-change-me" + # base64-encoded 32-byte AES key (the app base64-decodes this; see tr_ars/tasks.py) + aesMasterKey: "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=" + mysqlRootPassword: "ars-dev-root-password" + mysqlPassword: "ars-dev-password" + rabbitmqPassword: "ars-dev-password" + +arsserver: + replicas: 1 + containerPort: 8000 + gunicorn: + workerClass: gthread + # Dev-sized. Production (deploy/values.yaml) runs 8x4; concurrency ~= workers * threads. + workers: 4 + threads: 8 + timeout: 360 + gracefulTimeout: 120 + maxRequests: 1000 + maxRequestsJitter: 100 + # Keep above gunicorn.gracefulTimeout so in-flight requests can finish on shutdown + terminationGracePeriodSeconds: 130 + resources: + requests: + cpu: 500m + memory: 1Gi + limits: + cpu: "2" + memory: 4Gi + +# Two celery worker pools, one Deployment each (templates/celeryworker.yaml). +# Memory-heavy tasks are routed to the `heavy` queue (task_routes in +# everything else stays on the default `celery` queue. +# heavy.replicas * heavy.concurrency is the cluster-wide cap on concurrent heavy +# work; size memory per slot (limit / concurrency ~ one large merge or ingest). +celeryworkers: + heavy: + queues: heavy + replicas: 1 + concurrency: 4 + # Recycle a child process after this many tasks. Prefork children keep the + # RSS of the largest message they ever parsed, so without this a pod's + # memory only ratchets upward (same idea as gunicorn's maxRequests). + maxTasksPerChild: 20 + # How long a stopping worker gets to finish its running merges/ingests before + # they're SIGKILLed and redelivered. Size it to the longest merge. + terminationGracePeriodSeconds: 300 + resources: + requests: + cpu: 500m + memory: 2Gi + limits: + cpu: "2" + memory: 8Gi + light: + queues: celery + replicas: 1 + concurrency: 4 + terminationGracePeriodSeconds: 60 + resources: + requests: + cpu: 100m + memory: 512Mi + limits: + cpu: "1" + memory: 1Gi + +# Beat is always exactly 1 replica (a second scheduler would double-fire tasks) +celerybeat: + # Beat's workload is tiny (one scheduler tick, catch_timeout every 3 min) but it + # loads the same app image as the workers: Django + tr_ars.tasks/utils pulls in + # reasoner-pydantic, biothings_annotator, zstandard and the OTEL instrumentation + # for ~200MB RSS before it does anything. Anything under ~512Mi OOMs at import. + resources: + requests: + cpu: 100m + memory: 512Mi + limits: + cpu: 500m + memory: 1Gi + + +# merge_and_post_process retry tuning (tr_sys/tr_ars/utils.py) +merge: + # Retry cap for waiting on the per-parent merge db lock; constant 2-5s backoff + # between attempts, so ~6 min worst case. Each retry only republishes pks. + lockMaxRetries: "100" + # Retry cap for real errors during merge, tracked separately from lock + # retries; exponential backoff 1,2,4,8,16s then capped at 30s (~2 min total). + errorMaxRetries: "8" + +migrations: + # Run `manage.py migrate` as a post-install/pre-upgrade hook Job + enabled: true + # Same story as celerybeat: the job does almost no work but loads the whole + # app (~200MB RSS) to run Django's migrate. Without an explicit request the + # namespace LimitRange default applies, which is often small enough to OOM + # the job before it reaches the database -- and a failed hook fails the release. + resources: + requests: + cpu: 100m + memory: 512Mi + limits: + cpu: "1" + memory: 1Gi + +mysql: + # true: deploy a single-node MySQL StatefulSet (dev). false: use external.* + enabled: true + image: + repository: mysql + tag: "8.0" + pullPolicy: IfNotPresent + auth: + database: arsdb + username: ars + persistence: + size: 8Gi + storageClass: "" + # Extra mysqld flags. The merge path reads and writes multi-MB compressed + # message blobs under a row lock, so on any instance carrying real traffic set + # --innodb-buffer-pool-size to roughly half the memory limit below; the MySQL + # default is 128MB, which sends nearly every blob read to disk. + extraArgs: [] + resources: + requests: + cpu: 250m + memory: 512Mi + limits: + cpu: "1" + memory: 2Gi + # Used when enabled=false (e.g. RDS). Password comes from the Secret either way. + external: + host: "" + port: 3306 + database: arsdb + username: ars + +rabbitmq: + # true: deploy a single-node RabbitMQ StatefulSet with a persistent volume + enabled: true + image: + repository: rabbitmq + tag: "3.13" + pullPolicy: IfNotPresent + auth: + # guest/guest only works over localhost, so a split-pod deploy needs a real user + username: ars + persistence: + size: 4Gi + storageClass: "" + resources: + requests: + cpu: 250m + memory: 512Mi + limits: + cpu: "1" + memory: 1Gi + external: + host: "" + port: 5672 + username: "" + +service: + type: ClusterIP + port: 80 + +ingress: + enabled: false + className: "" + annotations: {} + host: ars.local + path: / + pathType: Prefix + tls: [] + +# Applied to every pod in the release. Left empty so the chart schedules on any +# cluster; the production node pinning (application=ars label + transltr taint) +# lives in deploy/values.yaml and can be copied here per-environment. +affinity: {} +tolerations: [] +nodeSelector: {} diff --git a/requirements.txt b/requirements.txt index 66bedc29..fd9f60af 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,7 +9,6 @@ bleach==6.1.0 Brotli==1.1.0 celery==5.5.3 certifi==2025.4.26 -channels==4.2.2 charset-normalizer==3.4.2 click==8.1.8 click-didyoumean==0.3.1 @@ -118,7 +117,6 @@ rich~=13.3.2 mock~=4.0.3 zstandard==0.23.0 botocore~=1.27.75 -redis~=5.0.0 sqlalchemy~=1.4.47 isodate~=0.6.1 boto3~=1.24.75 @@ -147,7 +145,6 @@ validators~=0.20.0 twisted~=24.7.0 pexpect~=4.8.0 h2~=4.1.0 -daphne~=3.0.2 simplejson~=3.17.6 notebook~=7.0.0 prompt_toolkit~=3.0.20 @@ -159,4 +156,3 @@ future~=0.18.2 flask==2.2.5 djangorestframework~=3.14.0 pandas~=1.5.3 -redis>=5.0,<6 diff --git a/tr_sys/README.md b/tr_sys/README.md index c4be5e05..aaa81bdb 100644 --- a/tr_sys/README.md +++ b/tr_sys/README.md @@ -3,7 +3,7 @@ ARS implementation ### Prerequisites -Require python 3.10+, django 3.1+ and channels packages. +Require python 3.10+ and django 4.2+. Test your python environment ```bash diff --git a/tr_sys/requirements.txt b/tr_sys/requirements.txt index 10ad5365..4fbd3fdf 100644 --- a/tr_sys/requirements.txt +++ b/tr_sys/requirements.txt @@ -8,7 +8,6 @@ bleach==6.1.0 Brotli==1.2.0 celery==5.5.3 certifi==2025.4.26 -channels==4.2.2 charset-normalizer==3.4.2 click==8.1.8 click-didyoumean==0.3.1 @@ -89,7 +88,6 @@ wrapt==1.17.2 wsproto==1.2.0 zipp==3.20.2 zstandard -redis git+https://github.com/biothings/biothings_annotator#egg=biothings_annotator opentelemetry-api==1.43.0 opentelemetry-sdk==1.43.0 diff --git a/tr_sys/routing.py b/tr_sys/routing.py deleted file mode 100644 index 7be2b379..00000000 --- a/tr_sys/routing.py +++ /dev/null @@ -1,10 +0,0 @@ -from channels.routing import ProtocolTypeRouter - -application = ProtocolTypeRouter({ - # (http->django views is added by default) - 'websocket': AuthMiddlewareStack( - URLRouter( - tr_ars.routing.ws_urlpatterns - ) - ), -}) diff --git a/tr_sys/tests/unit/test_ingest_ara_response.py b/tr_sys/tests/unit/test_ingest_ara_response.py new file mode 100644 index 00000000..03538ebd --- /dev/null +++ b/tr_sys/tests/unit/test_ingest_ara_response.py @@ -0,0 +1,224 @@ +""" +Unit tests for ingest_ara_response, the celery task that now does everything the +ARA-callback view used to do after receiving the body (parse, result count, +pre_merge_process, validation, notifications, enqueueing the merge). + +DB, notifications and the merge internals are mocked; transaction.on_commit is +patched to run its callback immediately so the merge enqueue can be asserted. +""" +from unittest.mock import patch, MagicMock + +import pytest + +from tr_ars import tasks +from tr_ars import utils +from tr_ars.models import Message + + +CHILD_PK = "child-pk-1" +PARENT_PK = "parent-pk-1" + + +def _message(results, agent_name="ara-shepherd-test", params=None, nodes=None): + mesg = MagicMock() + mesg.pk = CHILD_PK + mesg.status = "R" + mesg.ref_id = PARENT_PK + mesg.actor_id = 7 + mesg.params = params if params is not None else {} + mesg.updated_at = "2026-09-02T00:00:00" + if nodes is None: + nodes = {"n0": {"categories": ["biolink:Gene"]}} + body = {"message": {"results": results, "knowledge_graph": {"nodes": nodes, "edges": {}}}} \ + if results is not None else {"message": {}} + mesg.decompress_dict.return_value = body + return mesg + + +@pytest.fixture +def env(): + parent = MagicMock() + actor = MagicMock() + actor.inforesid = "infores:shepherd-test" + actor.agent.name = "ara-shepherd-test" + with patch.object(tasks, "Message") as message_cls, \ + patch.object(tasks, "Actor") as actor_cls, \ + patch.object(tasks, "get_object_or_404", return_value=parent), \ + patch.object(tasks, "transaction") as txn, \ + patch.object(utils, "pre_merge_process") as pre_merge, \ + patch.object(utils, "remove_phantom_support_graphs"), \ + patch.object(utils, "validate", return_value=True) as validate, \ + patch.object(utils, "ScoreStatCalc", return_value={"stat": 1}), \ + patch.object(utils.merge_and_post_process, "apply_async") as merge_enqueue: + actor_cls.objects.get.return_value = actor + # run on_commit callbacks immediately so the enqueue is observable + txn.on_commit.side_effect = lambda fn: fn() + yield {"message_cls": message_cls, "actor": actor, "parent": parent, + "pre_merge": pre_merge, "validate": validate, "merge_enqueue": merge_enqueue} + + +def _events(parent): + return [c.args[0]["event_type"] for c in parent.notify_subscribers.call_args_list] + + +def test_results_are_processed_and_merge_is_enqueued(env): + mesg = _message(results=[{"a": 1}, {"b": 2}]) + env["message_cls"].objects.get.return_value = mesg + + result = tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + + assert result.state == "SUCCESS" + assert _events(env["parent"]) == ["ara_response_complete"] + complete = env["parent"].notify_subscribers.call_args_list[0].args[0] + assert complete["ara_n_results"] == 2 + assert complete["child_uuid"] == CHILD_PK + assert mesg.result_count == 2 + assert mesg.status == "D" + assert mesg.code == 200 + env["pre_merge"].assert_called_once() + # the processed body is persisted before the merge task is enqueued + mesg.save_compressed_dict.assert_called_once() + env["merge_enqueue"].assert_called_once_with((PARENT_PK, CHILD_PK, "ara-shepherd-test")) + + +def test_status_header_is_applied(env): + mesg = _message(results=[{"a": 1}]) + env["message_cls"].objects.get.return_value = mesg + tasks.ingest_ara_response.apply(args=(CHILD_PK, "S")) + assert mesg.status == "S" + complete = env["parent"].notify_subscribers.call_args_list[0].args[0] + assert complete["ara_response_status"] == "S" + + +def test_invalid_response_is_marked_422_and_not_merged(env): + mesg = _message(results=[{"a": 1}]) + env["message_cls"].objects.get.return_value = mesg + env["validate"].return_value = False + + result = tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + + assert result.state == "SUCCESS" + assert mesg.status == "E" + assert mesg.code == 422 + assert _events(env["parent"]) == ["ara_response_complete", "ara_failed_validation"] + env["merge_enqueue"].assert_not_called() + + +def test_validation_is_skipped_when_params_disable_it(env): + mesg = _message(results=[{"a": 1}], params={"validate": False}) + env["message_cls"].objects.get.return_value = mesg + env["validate"].return_value = False # would fail if consulted + tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + env["validate"].assert_not_called() + env["merge_enqueue"].assert_called_once() + + +def test_no_results_records_zero_and_does_not_merge(env): + mesg = _message(results=None) + env["message_cls"].objects.get.return_value = mesg + + tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + + assert mesg.result_count == 0 + assert mesg.status == "D" + complete = env["parent"].notify_subscribers.call_args_list[0].args[0] + # no KG nodes either, so the Pathfinder empty-KG check reports 0 rather than None + assert complete["ara_n_results"] == 0 + env["pre_merge"].assert_not_called() + env["merge_enqueue"].assert_not_called() + + +def test_results_without_kg_nodes_are_treated_as_empty(env): + """Pathfinder can return one "result" with an empty KG; that must not be merged.""" + mesg = _message(results=[{"a": 1}], nodes={}) + env["message_cls"].objects.get.return_value = mesg + + tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + + assert mesg.result_count == 0 + assert mesg.status == "D" + complete = env["parent"].notify_subscribers.call_args_list[0].args[0] + assert complete["ara_n_results"] == 0 + env["pre_merge"].assert_not_called() + env["merge_enqueue"].assert_not_called() + + +def test_non_ara_agent_is_processed_but_not_merged(env): + mesg = _message(results=[{"a": 1}]) + env["message_cls"].objects.get.return_value = mesg + env["actor"].agent.name = "kp-test" + tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + assert mesg.result_count == 1 + env["merge_enqueue"].assert_not_called() + + +def test_failure_marks_message_error_500(env): + mesg = _message(results=[{"a": 1}]) + mesg.decompress_dict.side_effect = RuntimeError("corrupt body") + env["message_cls"].objects.get.return_value = mesg + + result = tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + + assert result.state == "SUCCESS" # handled, not raised + assert mesg.status == "E" + assert mesg.code == 500 + mesg.save.assert_called() + env["merge_enqueue"].assert_not_called() + + +def test_message_no_longer_running_is_skipped(env): + """e.g. catch_timeout already marked it 598 while the task sat in the queue.""" + mesg = _message(results=[{"a": 1}]) + mesg.status = "E" + env["message_cls"].objects.get.return_value = mesg + + result = tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + + assert result.state == "SUCCESS" + mesg.decompress_dict.assert_not_called() + mesg.save.assert_not_called() + env["parent"].notify_subscribers.assert_not_called() + env["merge_enqueue"].assert_not_called() + + +@pytest.mark.parametrize("decoded", [{}, None, ["not", "a", "dict"]]) +def test_undecodable_response_is_marked_error_400(env, decoded): + """decompress_dict returns {} on a JSON error; that must not become Done with 0 results.""" + mesg = _message(results=[{"a": 1}]) + mesg.decompress_dict.return_value = decoded + env["message_cls"].objects.get.return_value = mesg + + result = tasks.ingest_ara_response.apply(args=(CHILD_PK, "D")) + + assert result.state == "SUCCESS" + assert mesg.status == "E" + assert mesg.code == 400 + mesg.save.assert_called_once_with(update_fields=["status", "code", "updated_at"]) + env["parent"].notify_subscribers.assert_not_called() + env["pre_merge"].assert_not_called() + env["merge_enqueue"].assert_not_called() + + +def _bare_message(): + stub = MagicMock() + stub.pk = "round-trip" + stub.data = None + return stub + + +def test_save_compressed_bytes_round_trips_without_parsing(): + """The view stores raw bytes; decompress_dict must read them back as the dict.""" + raw = b'{"message": {"results": [{"x": 1}]}, "logs": []}' + mesg = _bare_message() + Message.save_compressed_bytes(mesg, raw) + assert mesg.data.startswith(b'\x28\xb5\x2f\xfd') # zstd magic + assert Message.decompress_dict(mesg) == {"message": {"results": [{"x": 1}]}, "logs": []} + + +def test_save_compressed_bytes_passes_through_already_compressed_data(): + raw = b'{"a": 1}' + first = _bare_message() + Message.save_compressed_bytes(first, raw) + second = _bare_message() + Message.save_compressed_bytes(second, first.data) + assert second.data == first.data \ No newline at end of file diff --git a/tr_sys/tests/unit/test_merge_retry.py b/tr_sys/tests/unit/test_merge_retry.py new file mode 100644 index 00000000..3cab415b --- /dev/null +++ b/tr_sys/tests/unit/test_merge_retry.py @@ -0,0 +1,168 @@ +""" +Unit tests for the retry semantics of merge_and_post_process. + +The DB and merge internals are all mocked out; these tests exercise only the +retry/backoff decision logic. Celery's eager mode re-executes retries +synchronously (Task.apply resubmits on Retry), so contention scenarios run +their full retry budget inline and we can assert exact attempt counts. +""" +from contextlib import nullcontext +from unittest.mock import patch, MagicMock + +import pytest +from celery.exceptions import Retry, MaxRetriesExceededError + +from tr_ars import utils +from tr_sys.celery_gates.backoff import MERGE_LOCK_MAX_RETRIES + + +TASK_ARGS = ("parent-pk-1", "child-pk-1", "ara-test") + + +@pytest.fixture +def mock_parent(): + parent = MagicMock() + parent.merge_semaphore = False + parent.merged_versions_list = [] + return parent + + +@pytest.fixture +def merge_env(mock_parent): + """Patch the DB and merge internals; yield the mocks for per-test tuning.""" + merged = MagicMock() + merged.id = "merged-pk-1" + merged.pk = "merged-pk-1" + # the child row the task loads its payload from, keyed by child_pk + child = MagicMock() + child.pk = "child-pk-1" + child.decompress_dict.return_value = {"message": {"knowledge_graph": {}}} + env = {"parent": mock_parent, "merged": merged, "child": child} + with patch.object(utils, "transaction") as txn, \ + patch.object(utils, "get_object_or_404", return_value=mock_parent), \ + patch.object(utils, "Message") as message_cls, \ + patch.object(utils, "try_lock_merge", return_value=True) as try_lock, \ + patch.object(utils, "unlock_merge"), \ + patch.object(utils, "merge_received", return_value=(merged, mock_parent, {})) as merge_received, \ + patch.object(utils, "post_process", return_value=(merged, 200, "D")) as post_process, \ + patch.object(utils, "record_error"), \ + patch.object(utils, "exp_backoff_with_jitter", return_value=0), \ + patch.object(utils, "constant_backoff_with_jitter", return_value=0): + txn.atomic.return_value = nullcontext() + message_cls.objects.filter.return_value.first.return_value = mock_parent + message_cls.objects.get.return_value = child + message_cls.DoesNotExist = type("DoesNotExist", (Exception,), {}) + env.update(try_lock=try_lock, merge_received=merge_received, + post_process=post_process, message_cls=message_cls) + yield env + + +def _failed_events(mock_parent): + return [c.args[0]["event_type"] for c in mock_parent.notify_subscribers.call_args_list + if c.args[0].get("event_type") == "merged_version_failed"] + + +def test_happy_path_merges(merge_env): + result = utils.merge_and_post_process.apply(args=TASK_ARGS) + assert result.state == "SUCCESS" + merge_env["merge_received"].assert_called_once() + merge_env["post_process"].assert_called_once() + + +def test_lock_contention_retries_then_succeeds(merge_env): + merge_env["try_lock"].side_effect = [False, False, True] + result = utils.merge_and_post_process.apply(args=TASK_ARGS) + assert result.state == "SUCCESS" + assert merge_env["try_lock"].call_count == 3 + merge_env["merge_received"].assert_called_once() + + +def test_lock_exhaustion_fails_loudly_after_full_budget(merge_env, mock_parent): + """Budget exhaustion must fail (not silently return), after exactly the budgeted attempts.""" + merge_env["try_lock"].return_value = False + result = utils.merge_and_post_process.apply(args=TASK_ARGS) + assert result.state == "FAILURE" + assert isinstance(result.result, MaxRetriesExceededError) + # initial attempt + MERGE_LOCK_MAX_RETRIES retries + assert merge_env["try_lock"].call_count == MERGE_LOCK_MAX_RETRIES + 1 + merge_env["merge_received"].assert_not_called() + # exhaustion is surfaced to API consumers, not just logs + assert _failed_events(mock_parent) + + +def test_error_retries_stop_at_cap(merge_env): + """A persistently failing merge is attempted exactly MERGE_ERROR_MAX_RETRIES + 1 times.""" + merge_env["merge_received"].side_effect = RuntimeError("boom") + result = utils.merge_and_post_process.apply(args=TASK_ARGS) + assert result.state == "FAILURE" + assert isinstance(result.result, RuntimeError) + assert merge_env["merge_received"].call_count == utils.MERGE_ERROR_MAX_RETRIES + 1 + + +def test_error_retry_recovers_after_transient_failure(merge_env, mock_parent): + merged = merge_env["merged"] + merge_env["merge_received"].side_effect = [RuntimeError("boom"), RuntimeError("boom"), + (merged, mock_parent, {})] + result = utils.merge_and_post_process.apply(args=TASK_ARGS) + assert result.state == "SUCCESS" + assert merge_env["merge_received"].call_count == 3 + + +def test_error_retry_preserves_caller_kwargs(merge_env): + """The error retry merges into request.kwargs instead of replacing them wholesale.""" + merge_env["merge_received"].side_effect = RuntimeError("boom") + retry_spy = MagicMock(side_effect=Retry("requested retry")) + kwargs = {"parent_pk": "parent-pk-1", "child_pk": "child-pk-1", "agent_name": "ara-test"} + with patch.object(utils.merge_and_post_process, "retry", retry_spy): + result = utils.merge_and_post_process.apply(kwargs=kwargs) + assert result.state == "RETRY" + retry_kwargs = retry_spy.call_args.kwargs["kwargs"] + assert retry_kwargs["error_retries"] == 1 + for key, value in kwargs.items(): + assert retry_kwargs[key] == value + + +def test_error_retry_increments_existing_counter(merge_env): + merge_env["merge_received"].side_effect = RuntimeError("boom") + retry_spy = MagicMock(side_effect=Retry("requested retry")) + with patch.object(utils.merge_and_post_process, "retry", retry_spy): + utils.merge_and_post_process.apply( + kwargs={"parent_pk": "p", "child_pk": "c", "agent_name": "a", + "error_retries": 3}) + assert retry_spy.call_args.kwargs["kwargs"]["error_retries"] == 4 + + +def test_direct_call_error_retry_survives_missing_request_kwargs(merge_env): + """Called directly (not via a worker), request.kwargs is None; the retry must not TypeError.""" + merge_env["merge_received"].side_effect = RuntimeError("boom") + retry_spy = MagicMock(side_effect=Retry("requested retry")) + with patch.object(utils.merge_and_post_process, "retry", retry_spy): + with pytest.raises(Retry): + utils.merge_and_post_process(*TASK_ARGS) + assert retry_spy.call_args.kwargs["kwargs"]["error_retries"] == 1 + + +def test_payload_is_loaded_from_child_pk_not_the_task_args(merge_env): + """The task takes a pk and reads the body from the DB, not from the celery payload.""" + result = utils.merge_and_post_process.apply(args=TASK_ARGS) + assert result.state == "SUCCESS" + merge_env["message_cls"].objects.get.assert_called_once_with(pk="child-pk-1") + merge_env["child"].decompress_dict.assert_called_once() + # merge_received gets the inner "message", matching what callers used to pass inline + passed = merge_env["merge_received"].call_args.args[1] + assert passed == {"knowledge_graph": {}} + + +def test_missing_child_is_not_retried(merge_env): + """A child row that no longer exists is permanent; retrying cannot fix it.""" + merge_env["message_cls"].objects.get.side_effect = merge_env["message_cls"].DoesNotExist + result = utils.merge_and_post_process.apply(args=TASK_ARGS) + assert result.state == "SUCCESS" # returns cleanly rather than erroring + merge_env["merge_received"].assert_not_called() + + +def test_child_without_message_key_is_skipped(merge_env): + merge_env["child"].decompress_dict.return_value = {"nothing": "here"} + result = utils.merge_and_post_process.apply(args=TASK_ARGS) + assert result.state == "SUCCESS" + merge_env["merge_received"].assert_not_called() \ No newline at end of file diff --git a/tr_sys/tests/unit/test_message_callback_view.py b/tr_sys/tests/unit/test_message_callback_view.py new file mode 100644 index 00000000..e8642bc0 --- /dev/null +++ b/tr_sys/tests/unit/test_message_callback_view.py @@ -0,0 +1,118 @@ +""" +Unit tests for the ARA callback view (POST /ars/api/messages/). + +The view must never parse the body: it applies the duplicate guards +from the row alone, stores the raw bytes compressed, enqueues +ingest_ara_response on commit, and returns a small acknowledgement rather than +serializing the whole message back out. +""" +import json +from unittest.mock import patch, MagicMock + +import pytest +from django.test import RequestFactory + +from tr_ars import api + + +KEY = "2c1a9a5e-3a4e-4c0e-9d1c-1b1a0c3b2f11" +BODY = b'{"message": {"results": [{"a": 1}], "knowledge_graph": {}}}' + + +def _mesg(status="R", result_count=None, code=202): + mesg = MagicMock() + mesg.pk = KEY + mesg.status = status + mesg.result_count = result_count + mesg.code = code + return mesg + + +@pytest.fixture +def env(): + # Replace the module's `json` reference (not the global json module, which the + # tests themselves use) so any parse of the body inside the view fails loudly. + json_stub = MagicMock() + json_stub.loads.side_effect = AssertionError("view must not parse the body") + with patch.object(api, "Message") as message_cls, \ + patch.object(api, "transaction") as txn, \ + patch.object(api.tasks.ingest_ara_response, "apply_async") as enqueue, \ + patch.object(api, "json", json_stub): + message_cls.DoesNotExist = type("DoesNotExist", (Exception,), {}) + txn.on_commit.side_effect = lambda fn: fn() + yield {"message_cls": message_cls, "enqueue": enqueue} + + +def _post(body=BODY, headers=None): + req = RequestFactory().post(f"/ars/api/messages/{KEY}", data=body, + content_type="application/json") + if headers: + req.headers = headers + return req + + +def test_callback_is_stored_raw_and_ingest_is_enqueued(env): + mesg = _mesg() + env["message_cls"].objects.get.return_value = mesg + + resp = api.message(_post(), KEY) + + assert resp.status_code == 201 + mesg.save_compressed_bytes.assert_called_once_with(BODY) + mesg.save.assert_called_once() + env["enqueue"].assert_called_once_with((KEY, "D")) + # a small ack, not the whole message serialized back out + payload = json.loads(resp.content) + assert payload == {"pk": KEY, "status": "Accepted", "code": 202} + mesg.to_dict.assert_not_called() + + +def test_status_header_is_forwarded_to_the_task(env): + env["message_cls"].objects.get.return_value = _mesg() + resp = api.message(_post(headers={"tr_ars.message.status": "S"}), KEY) + assert resp.status_code == 201 + env["enqueue"].assert_called_once_with((KEY, "S")) + + +@pytest.mark.parametrize("mesg,expected_status", [ + (_mesg(status="D"), 200), + (_mesg(status="R", result_count=5), 409), + (_mesg(status="E", code=500), 400), +]) +def test_duplicate_guards_reject_without_storing_or_enqueueing(env, mesg, expected_status): + env["message_cls"].objects.get.return_value = mesg + resp = api.message(_post(), KEY) + assert resp.status_code == expected_status + mesg.save_compressed_bytes.assert_not_called() + mesg.save.assert_not_called() + env["enqueue"].assert_not_called() + + +def test_claim_lost_to_a_timeout_is_rejected(env): + """The row passed the status guards but was marked 598 before the atomic claim.""" + mesg = _mesg() + env["message_cls"].objects.get.return_value = mesg + claim = env["message_cls"].objects.filter.return_value.exclude + claim.return_value.update.return_value = 0 + + resp = api.message(_post(), KEY) + + claim.assert_called_once_with(status__in=["D", "E"]) + assert resp.status_code == 200 + mesg.save_compressed_bytes.assert_not_called() + env["enqueue"].assert_not_called() + + +def test_unknown_pk_is_404(env): + env["message_cls"].objects.get.side_effect = env["message_cls"].DoesNotExist + resp = api.message(_post(), KEY) + assert resp.status_code == 404 + env["enqueue"].assert_not_called() + + +def test_large_body_is_never_parsed_in_the_request(env): + """The whole point of the change: the body size must not matter to the view.""" + env["message_cls"].objects.get.return_value = _mesg() + big = b'{"message": {"results": [' + b'{"a": 1},' * 200000 + b'{"a": 1}]}}' + resp = api.message(_post(body=big), KEY) + assert resp.status_code == 201 # json.loads is patched to raise if called \ No newline at end of file diff --git a/tr_sys/tests/unit/test_message_save.py b/tr_sys/tests/unit/test_message_save.py new file mode 100644 index 00000000..87e9ee83 --- /dev/null +++ b/tr_sys/tests/unit/test_message_save.py @@ -0,0 +1,85 @@ +""" +Unit tests for Message.save() and the compressed data it writes. + +The ARA callback view and ingest_ara_response both load a child message, write new +data with save_compressed_bytes / save_compressed_dict and save it, and +merge_and_post_process then reads that row back. So save() must write the new data, +not the bytes the message was loaded with. + +No DB: Model.save is patched to record the data that would be written. +""" +import json +import uuid +from unittest.mock import patch + +import pytest +import zstandard as zstd +from django.db import models as django_models + +from tr_ars.models import Actor, Agent, Message + + +QUERY = {"message": {"query_graph": {}}} +RESPONSE = {"message": {"results": [{"x": 1}], "knowledge_graph": {}}} + + +def _compress(d): + return zstd.ZstdCompressor().compress(json.dumps(d).encode("utf-8")) + + +def _decompress(data): + return json.loads(zstd.ZstdDecompressor().decompress(data)) + + +def _with_actor(mesg): + # save_compressed_dict logs the agent name + mesg.actor = Actor(agent=Agent(name="ara-test")) + return mesg + + +def _loaded(data): + """A message built the way a queryset builds one, holding data already in the DB.""" + return _with_actor(Message.from_db("default", ["id", "data"], [uuid.uuid4(), _compress(data)])) + + +@pytest.fixture +def written(): + stored = [] + + def record(self, *args, **kwargs): + stored.append(self.data) + + with patch.object(django_models.Model, "save", record), \ + patch.object(Message, "should_notify", return_value=False): + yield stored + + +def test_save_writes_bytes_stored_on_a_loaded_message(written): + """The callback view: the stored query is replaced by the ARA's response body.""" + mesg = _loaded(QUERY) + mesg.save_compressed_bytes(json.dumps(RESPONSE).encode("utf-8")) + mesg.save() + assert _decompress(written[-1]) == RESPONSE + + +def test_save_writes_dict_stored_on_a_loaded_message(written): + """ingest_ara_response: the raw body is replaced by the pre_merge_process output.""" + processed = {"message": dict(RESPONSE["message"], processed=True)} + mesg = _loaded(RESPONSE) + mesg.save_compressed_dict(processed) + mesg.save() + assert _decompress(written[-1]) == processed + + +def test_save_leaves_loaded_data_unchanged_when_not_rewritten(written): + mesg = _loaded(QUERY) + loaded = mesg.data + mesg.status = "R" + mesg.save() + assert written[-1] == loaded + + +def test_save_compresses_a_dict_passed_to_the_constructor(written): + mesg = _with_actor(Message(data=RESPONSE)) + mesg.save() + assert _decompress(written[-1]) == RESPONSE \ No newline at end of file diff --git a/tr_sys/tr_ars/api.py b/tr_sys/tr_ars/api.py index 08cfc929..0174dc19 100644 --- a/tr_sys/tr_ars/api.py +++ b/tr_sys/tr_ars/api.py @@ -490,126 +490,71 @@ def message(req, key): elif req.method == 'POST': with tracer.start_as_current_span('message') as span: span.set_attribute("pk", str(key)) + claimed = 0 try: - data = json.loads(req.body) - #if 'query_graph' not in data or 'knowledge_graph' not in data or 'results' not in data: - # return HttpResponse('Not a valid Translator API json', status=400) - mesg = get_object_or_404(Message.objects.filter(pk=key)) + mesg = Message.objects.get(pk=key) status = 'D' - code = 200 if 'tr_ars.message.status' in req.headers: status = req.headers['tr_ars.message.status'] - res=utils.get_safe(data,"message","results") - #kg = utils.get_safe(data,"message", "knowledge_graph") - - nodes = utils.get_safe(data,"message","knowledge_graph","nodes") - #Checking for a lack of nodes and setting results to an empty list if we have none - #This corrects an issue with PF queries coming back with one "result" but no nodes or edges - #Which messes up post-processing/finishing queries - if not nodes: - res=[] - actor = Actor.objects.get(pk=mesg.actor_id) - inforesid =actor.inforesid - parent=get_object_or_404(Message.objects.filter(pk=mesg.ref_id)) - if res is not None: - result_length = len(res) - else: - result_length = None - notification = { - "event_type":"ara_response_complete", - "ara_name":actor.inforesid, - "child_uuid":str(mesg.pk), - "ara_response_status": status, - "ara_n_results":result_length - } - parent.notify_subscribers(notification) - span.set_attribute("agent", inforesid) - logging.info('received msg from agent: %s with parent pk: %s and result: %s' % (str(inforesid), str(mesg.ref_id),str(result_length))) - if mesg.status=='D': - return HttpResponse('ARS has already received %s results from pk: %s' % (str(result_length), str(key))) - if mesg.result_count is not None and mesg.result_count >0: - return HttpResponse('ARS already has a response with: %s results for pk %s \nWe are temporarily ' - 'disallowing subsequent updates to PKs which already have results\n' - % (str(result_length), str(key)),status=409) - - if mesg.status=='E': - return HttpResponse("Response received but Message is already in state "+str(mesg.code)+". Response rejected\n",status=400) - if res is not None and result_length>0: - mesg.result_count = result_length - scorestat = utils.ScoreStatCalc(res) - mesg.result_stat = scorestat - #before we do basically anything else, we normalize - parent_pk = mesg.ref_id - #message_to_merge =utils.get_safe(data,"message") - message_to_merge = data - agent_name = str(mesg.actor.agent.name) - logger.info("Running pre_merge_process for agent %s with %s" % (agent_name, result_length)) - utils.pre_merge_process(message_to_merge,key, agent_name, inforesid) - if mesg.data and 'results' in mesg.data and mesg.data['results'] != None and len(mesg.data['results']) > 0: - mesg = Message.create(name=mesg.name, status=status, actor=mesg.actor, ref=mesg) - if "validate" in mesg.params.keys() and not mesg.params["validate"]: - valid = True - else: - #utils.remove_phantom_support_graphs(message_to_merge) - valid = utils.validate(message_to_merge) - if valid: - if agent_name.startswith('ara-'): - logger.info("pre async call for agent %s" % agent_name) - #utils.merge_and_post_process(parent_pk,message_to_merge['message'],agent_name) - utils.merge_and_post_process.apply_async((parent_pk,message_to_merge['message'],agent_name)) - logger.info("post async call for agent %s" % agent_name) - else: - logger.debug("Validation problem found for agent %s with pk %s" % (agent_name, str(mesg.ref_id))) - code = 422 - status = 'E' - mesg.status = status - mesg.code = code - mesg.save_compressed_dict(data) - mesg.save() - notification = { - "event_type":"ara_failed_validation", - "ara_name":actor.inforesid, - "child_uuid":str(mesg.pk), - "ara_response_status": status, - "ara_n_results":result_length - } - parent.notify_subscribers(notification) - return HttpResponse("Problem with TRAPI Validation", - status=422) - - mesg.status = status - mesg.code = code - mesg.save_compressed_dict(data) - #06-09-2026: Making a design choice that None results means result_count of 0 - #saves us from having to check for None all over the place. - if res is None: - mesg.result_count = 0 - mesg.save() - - return HttpResponse(json.dumps(mesg.to_dict(), indent=2), - status=201) + + # Duplicate / late-response guards + if mesg.status == 'D': + return HttpResponse('ARS has already received results from pk: %s' % str(key)) + if mesg.result_count is not None and mesg.result_count > 0: + return HttpResponse('ARS already has a response for pk %s \nWe are temporarily ' + 'disallowing subsequent updates to PKs which already have results\n' + % str(key), status=409) + if mesg.status == 'E': + return HttpResponse("Response received but Message is already in state "+str(mesg.code)+". Response rejected\n", status=400) + + # Record the time that the message is received. + # This also checks if there is already a message with this pk with a received_at stamp, + # another way to reject duplicate ARA responses. Excluding D/E repeats the + # status guards above atomically, in case catch_timeout marked it 598 since. + received_at_timestamp = timezone.now() + claimed = (Message.objects.filter(pk=key, received_at__isnull=True) + .exclude(status__in=['D', 'E']) + .update(received_at=received_at_timestamp)) + if not claimed: + return HttpResponse('ARS has already received a response for pk %s, ' + 'or is no longer accepting one' % str(key)) + + # Previously we had a pre-merge phase that ran here. It deserialized the json, + # ran some operations on it, serialized it again, and then queued it up for celery. + # We also then encoded decompressed and serialized the whole message again to send it back to the ARA, + # which it never actually uses anymore. (mesg.save_compressed_dict(data) AND json.dumps(mesg.to_dict())) + # That was a bit inefficient but also it's unpacking the full TRAPI message here in the ARS api server, + # which can cost quite a bit of memory for large ARA responses. Now, we push all of those pre-merge + # operations to be processed by celery workers after the task queue, so we can keep the ARS server + # quick and focus our resource allocation on the celery workers. + + # Store the body exactly as sent, compressed, never parsed here, + # and hand everything that needs the parsed message (result count, + # pre_merge_process, validation, notifications, the merge) to + # ingest_ara_response. + mesg.save_compressed_bytes(req.body) + # update_fields: a full save would write this instance's received_at (None) back + mesg.save(update_fields=['data', 'updated_at']) + child_pk = str(mesg.pk) + # Use on_commit so a worker can't read the row too early + transaction.on_commit( + lambda: tasks.ingest_ara_response.apply_async((child_pk, status))) + # Return a small JsonResponse instead of the previous large mesg.to_dict() + return JsonResponse({"pk": child_pk, "status": "Accepted", "code": 202}, status=201) except Message.DoesNotExist: return HttpResponse('Unknown state reference %s' % key, status=404) - except json.decoder.JSONDecodeError as e: - return HttpResponse('Can not decode json:
\n%s for the pk: %s' % (req.body, key), status=500) - except Exception as e: - mesg.status = 'E' - mesg.code = 500 - log_entry = { - "message":"Internal ARS Server Error", - "timestamp":mesg.updated_at, - "level":"ERROR" - } - if 'logs' in data.keys(): - data['logs'].append(log_entry) - else: - data['logs'] = [log_entry] - mesg.save_compressed_dict(data) - mesg.save() logger.error("Unexpected error 12: {} with the pk: %s".format(traceback.format_exception(type(e), e, e.__traceback__), key)) + if claimed: + # If we get here some unexpected error happened on our end, + # like failing to queue the ingest_ara_response task. + # Here we set received_at back to None so that an ARA could + # see this 500 and retry successfully. Setting received_at=None + # also allows this message to get timed out appropriately, it + # will just look like no response was ever received. + Message.objects.filter(pk=key).update(received_at=None) return HttpResponse('Internal server error', status=500) else: return HttpResponse('Method %s not supported!' % req.method, status=400) diff --git a/tr_sys/tr_ars/consumers.py b/tr_sys/tr_ars/consumers.py deleted file mode 100644 index 63182a59..00000000 --- a/tr_sys/tr_ars/consumers.py +++ /dev/null @@ -1,18 +0,0 @@ -from channels.generic.websocket import WebsocketConsumer -import json - -class ARSConsumer(WebsocketConsumer): - def connect(self): - self.accept() - - def disconnect(self, close_code): - pass - - def receive(self, text_data): - text_data_json = json.loads(text_data) - message = text_data_json['message'] - - self.send(text_data=json.dumps({ - 'message': message - })) - diff --git a/tr_sys/tr_ars/migrations/0015_message_received_at.py b/tr_sys/tr_ars/migrations/0015_message_received_at.py new file mode 100644 index 00000000..615932b9 --- /dev/null +++ b/tr_sys/tr_ars/migrations/0015_message_received_at.py @@ -0,0 +1,16 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('tr_ars', '0014_auto_20250121_2122'), + ] + + operations = [ + migrations.AddField( + model_name='message', + name='received_at', + field=models.DateTimeField(blank=True, null=True, verbose_name="time the agent's response was received"), + ), + ] \ No newline at end of file diff --git a/tr_sys/tr_ars/models.py b/tr_sys/tr_ars/models.py index b68cd1bf..3009036e 100644 --- a/tr_sys/tr_ars/models.py +++ b/tr_sys/tr_ars/models.py @@ -35,7 +35,7 @@ class Agent(ARSModel): contact = models.EmailField(null=True) registered = models.DateTimeField(default=timezone.now) updated = models.DateTimeField(auto_now=True) - + def __str__(self): return 'agent{name:%s, uri:%s}' % (self.name, self.uri) @@ -83,7 +83,7 @@ class Message(ARSModel): ('W', 'Waiting'), ('U', 'Unknown') ) - + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False, db_index=True) name = models.SlugField('Message name', null=False) code = models.PositiveSmallIntegerField('HTTP status code', @@ -92,6 +92,7 @@ class Message(ARSModel): actor = models.ForeignKey(Actor, null=False, on_delete=models.CASCADE) timestamp = models.DateTimeField(auto_now_add=True,db_index=True) updated_at = models.DateTimeField(auto_now=True) + received_at = models.DateTimeField("time the agent's response was received", null=True, blank=True) data = models.BinaryField('data payload', null=True) url = models.URLField('location of data', max_length=256, null=True) ref = models.ForeignKey('self', null=True, blank=True, @@ -110,21 +111,11 @@ class Message(ARSModel): def __str__(self): return "message[%s]{name:%s, status:%s}" % (self.id, self.name, self.status) - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - # # Decompress the compressed data when initializing the model instance - if self.data and self.data is not None: - self.original_data = self.data - else: - self.original_data = {} - def save(self, *args, **kwargs): - # Compress the data before saving logger.info("Entering save") - if self.original_data: + if self.data is not None and not isinstance(self.data, (bytes, bytearray, memoryview)): logger.info('Compressing the data at save call') - self.save_compressed_dict(self.original_data) - self.original_data = {} # Clear original data to avoid redundancy + self.save_compressed_dict(self.data) super().save(*args, **kwargs) if self.should_notify() and self.ref is None: @@ -162,6 +153,26 @@ def save_compressed_dict(self, data): logger.error("❌ failed to compress data %s"% e) print("Error compressing data:", e) + def save_compressed_bytes(self, raw): + """Compress an already-serialized JSON body without parsing it. + + For an ARA callback parsing a multi-MB body only to re-serialize it is + expensive and isn't needed anymore. Instead, we use this to compress + the message as is, and parse it later when we actually do the work on it. + decompress_dict() reads what this writes. + """ + if raw is None: + return + if isinstance(raw, str): + raw = raw.encode('utf-8') + if raw.startswith(b'\x28\xb5\x2f\xfd'): + self.data = raw + return + start = t.time() + self.data = zstd.ZstdCompressor().compress(raw) + logger.debug("compressed %s raw bytes for pk %s in %.2fs", + len(raw), str(self.pk), t.time() - start) + def decompress_dict(self): try: # Check if self.data is already a dictionary @@ -208,7 +219,7 @@ def decompress_dict(self): logger.info("✅ Parsed dict keys: %s; message keys: %s", list(original_data.keys()), list(message.keys())) else: logger.warning("'message' is not a dict. Top-level keys: %s", list(original_data.keys())) - + total_time = t.time() - start logging.info(f"TOTAL TIME TAKEN for agent {self.actor.agent.name} to decompress: {total_time:.2f} seconds") return original_data diff --git a/tr_sys/tr_ars/routing.py b/tr_sys/tr_ars/routing.py deleted file mode 100644 index efcbaab2..00000000 --- a/tr_sys/tr_ars/routing.py +++ /dev/null @@ -1,7 +0,0 @@ -from django.urls import re_path - -from . import consumers - -ws_urlpatterns = [ - re_path(r'ws/ars/(?P\w+)/$', consumers.ARSConsumer), -] diff --git a/tr_sys/tr_ars/tasks.py b/tr_sys/tr_ars/tasks.py index 54368582..1dc3608f 100644 --- a/tr_sys/tr_ars/tasks.py +++ b/tr_sys/tr_ars/tasks.py @@ -13,6 +13,8 @@ import traceback from django.utils import timezone from django.shortcuts import get_object_or_404 +from django.db import transaction +from django.db.models import Q from opentelemetry import trace import time as sleeptime from .api import decrypt_secret @@ -21,20 +23,10 @@ import os import hashlib from requests.exceptions import RequestException, Timeout -from tr_sys.celery_gates.context import (expensive_section) logger = get_task_logger(__name__) tracer = trace.get_tracer(__name__) -# @shared_task(bind=True, acks_late=True, max_retries=50, default_retry_delay=2) -# def test_expensive(self, seconds=10): -# logger.info(f"[{self.request.id}] before expensive section") -# with expensive_section(self): -# logger.info(f"[{self.request.id}] ✅ ENTER expensive section") -# time.sleep(seconds) -# logger.info(f"[{self.request.id}] ✅ EXIT expensive section") -# logger.info(f"[{self.request.id}] after expensive section") - @shared_task(name="send-message-to-actor") def send_message(actor_dict, mesg_dict, timeout=300): infores=actor_dict['fields']['inforesid'] @@ -147,6 +139,9 @@ def send_message(actor_dict, mesg_dict, timeout=300): child_pk=str(mesg.pk) logger.info("Running pre_merge_process for agent %s with %s" % (agent_name, len(results))) utils.pre_merge_process(message_to_merge,child_pk, agent_name, inforesid) + if "validate" not in mesg.params.keys() or mesg.params["validate"]: + # preserving previous behavior, only remove phantom support graphs when validate=true + utils.remove_phantom_support_graphs(message_to_merge) #Whether we did any additional processing or not, we need to save what we have mesg.code = status_code mesg.status = status @@ -201,7 +196,6 @@ def send_message(actor_dict, mesg_dict, timeout=300): if "validate" in mesg.params.keys() and not mesg.params["validate"]: valid = True else: - utils.remove_phantom_support_graphs(message_to_merge) valid = utils.validate(message_to_merge) if valid: if agent_name.startswith('ara-'): @@ -214,8 +208,14 @@ def send_message(actor_dict, mesg_dict, timeout=300): #parent = get_object_or_404(Message.objects.filter(pk=parent_pk)) #logging.info(f'parent merged_versions_list before going into merge&post-process for pk: %s are %s' % (parent_pk,parent.merged_versions_list)) #utils.merge_and_post_process(parent_pk,message_to_merge['message'],agent_name) - #utils.merge_and_post_process(parent_pk,message_to_merge['message'],agent_name) - utils.merge_and_post_process.apply_async((parent_pk,message_to_merge['message'],agent_name)) + + # The task loads the body from this row instead of taking it in the + # celery payload (rdata was already saved to it above), so enqueue + # on_commit to guarantee the row is visible when a worker picks it up. + merge_child_pk = mesg.pk + transaction.on_commit( + lambda: utils.merge_and_post_process.apply_async( + (parent_pk, merge_child_pk, agent_name))) logger.info("post async call for agent %s" % agent_name) else: logger.debug("Validation problem found for agent %s with pk %s" % (agent_name, str(mesg.ref_id))) @@ -228,6 +228,120 @@ def send_message(actor_dict, mesg_dict, timeout=300): logging.debug("Skipping merge and post for "+str(mesg.pk)+ " because the contributing message is in state: "+str(mesg.code)) +@shared_task(name="ingest-ara-response", bind=True, acks_late=True) +def ingest_ara_response(self, child_pk, status='D'): + """Everything the ARA-callback django view used to do after receiving the body. + + The view now stores the raw bytes and returns 201 without parsing them. + Parsing, pre_merge_process, validation and the resulting notifications all + happen here, so a multi-MB (sometimes tens of MB) callback never expands into + a large parsed dict inside an ARS api server gunicorn worker. Doing it here + means it gets handled by the celery workers instead. + + `status` is the ARA-supplied tr_ars.message.status header, defaulting to 'D'. + """ + with tracer.start_as_current_span("ingest_ara_response") as span: + span.set_attribute("pk", str(child_pk)) + mesg = Message.objects.get(pk=child_pk) + if mesg.status != 'R': + # Already timed out by catch_timeout (or otherwise finished) while this + # task sat in the queue; don't flip it back to Done or merge it. + logger.info("Skipping ingest for pk %s, message is already in state %s (code %s)", + child_pk, mesg.status, mesg.code) + return + actor = Actor.objects.get(pk=mesg.actor_id) + inforesid = actor.inforesid + agent_name = str(actor.agent.name) + parent = get_object_or_404(Message.objects.filter(pk=mesg.ref_id)) + span.set_attribute("agent", inforesid) + code = 200 + data = None + try: + data = mesg.decompress_dict() + if not isinstance(data, dict) or not data: + # decompress_dict swallows JSON errors and returns {}. The view no longer + # parses the body (so can't 500 the ARA like it used to), so catch a + # malformed response here instead of recording it as Done with 0 results. + # Most likely the ARA posted an error string/empty body; the raw bytes + # stay in data for inspection. + logger.error("Could not decode ARA response for pk %s from agent %s", child_pk, inforesid) + mesg.status = 'E' + mesg.code = 400 + mesg.save(update_fields=['status', 'code', 'updated_at']) + return + res = utils.get_safe(data, "message", "results") + nodes = utils.get_safe(data, "message", "knowledge_graph", "nodes") + #Checking for a lack of nodes and setting results to an empty list if we have none + #This corrects an issue with PF queries coming back with one "result" but no nodes or edges + #Which messes up post-processing/finishing queries + if not nodes: + res = [] + result_length = len(res) if res is not None else None + span.set_attribute("ara_n_results", result_length if result_length is not None else -1) + parent.notify_subscribers({ + "event_type": "ara_response_complete", + "ara_name": inforesid, + "child_uuid": str(mesg.pk), + "ara_response_status": status, + "ara_n_results": result_length, + }) + logger.info('received msg from agent: %s with parent pk: %s and result: %s', + inforesid, mesg.ref_id, result_length) + + if res is not None and result_length > 0: + mesg.result_count = result_length + mesg.result_stat = utils.ScoreStatCalc(res) + utils.pre_merge_process(data, child_pk, agent_name, inforesid) + if "validate" in mesg.params.keys() and not mesg.params["validate"]: + valid = True + else: + #utils.remove_phantom_support_graphs(data) + valid = utils.validate(data) + if not valid: + logger.debug("Validation problem found for agent %s with pk %s", agent_name, mesg.ref_id) + mesg.status = 'E' + mesg.code = 422 + mesg.save_compressed_dict(data) + mesg.save() + parent.notify_subscribers({ + "event_type": "ara_failed_validation", + "ara_name": inforesid, + "child_uuid": str(mesg.pk), + "ara_response_status": 'E', + "ara_n_results": result_length, + }) + return + mesg.status = status + mesg.code = code + # pre_merge_process mutated data in place; the merge task loads the + # child row, so persist the processed form before enqueueing it. + mesg.save_compressed_dict(data) + mesg.save() + if agent_name.startswith('ara-'): + parent_pk = mesg.ref_id + transaction.on_commit( + lambda: utils.merge_and_post_process.apply_async( + (parent_pk, child_pk, agent_name))) + return + + # No results: record that explicitly (None results means result_count 0) + mesg.status = status + mesg.code = code + mesg.result_count = 0 + mesg.save() + except Exception as e: + logger.error("Unexpected error ingesting ARA response for pk %s: %s", + child_pk, traceback.format_exception(type(e), e, e.__traceback__)) + mesg.status = 'E' + mesg.code = 500 + if isinstance(data, dict): + log_entry = {"message": "Internal ARS Server Error", + "timestamp": str(mesg.updated_at), "level": "ERROR"} + data.setdefault('logs', []).append(log_entry) + mesg.save_compressed_dict(data) + mesg.save() + + @shared_task(name="catch_timeout") def catch_timeout_async(): now =timezone.now() @@ -236,14 +350,23 @@ def catch_timeout_async(): max_time = now-timezone.timedelta(minutes=5) max_time_merged=now-timezone.timedelta(minutes=8) max_time_pathfinder = now-timezone.timedelta(minutes=5) + # Upper bound on how long a received ARA response may wait for ingest-ara-response + # to finish. Generous so a heavy-queue backlog doesn't time out real responses, + # but finite so a lost or crashed ingest task can't leave the parent waiting forever. + max_time_ingest = now-timezone.timedelta(minutes=20) #retrieving last 15 min running records might become overwhelming, so we might need to refine this filter to grab records between 4 min< x < 15 min or have 2 sets (for standard/pathfinder) queires - messages = Message.objects.filter(timestamp__gt=time_threshold, status__in='R').values_list('actor','id','timestamp','updated_at','params') + # Received-but-unfinished messages (received_at is set) are included regardless of timestamp, + # otherwise one whose ingest never completes would age out of the window before its deadline. + messages = Message.objects.filter( + Q(timestamp__gt=time_threshold) | Q(received_at__isnull=False), + status__in='R').values_list('actor','id','timestamp','updated_at','params','received_at') for mesg in messages: mpk=mesg[0] id = mesg[1] actor = Agent.objects.get(pk=mpk) timestamp=mesg[2] + received_at=mesg[5] query_type=mesg[4]['query_type'] if actor.name != 'ars-ars-agent' else None logging.info(f'actor: {actor} id: {mesg[1]} timestamp: {mesg[2]} updated_at {mesg[3]} query_type {query_type}') @@ -263,6 +386,20 @@ def catch_timeout_async(): message.save(update_fields=['status','code','updated_at']) else: continue + elif received_at is not None: + # If received_at is set, the agent has already responded in time + # and the message is headed to ingest-ara-response. We don't want + # to time out ARA responses based on our own concurrency + # bottleneck, so let it continue on unless ingest has clearly + # been lost, in which case time it out so the parent can finish. + if received_at < max_time_ingest: + logging.info(f'for actor: {actor.name}, and pk {str(id)}, response was received at {received_at} but ingest has not finished after 20 min, setting code to 598') + message = get_object_or_404(Message.objects.filter(pk=id)) + message.code = 598 + message.status = 'E' + message.updated_at = timezone.now() + message.save(update_fields=['status','code','updated_at']) + continue else: if query_type == 'standard' and timestamp < max_time: logging.info(f'for actor: {actor.name}, and pk {str(id)} of query type: {query_type}, the status is still "Running" after 5 min, setting code to 598') diff --git a/tr_sys/tr_ars/utils.py b/tr_sys/tr_ars/utils.py index e9159ab9..dbb8d497 100644 --- a/tr_sys/tr_ars/utils.py +++ b/tr_sys/tr_ars/utils.py @@ -35,14 +35,15 @@ from opentelemetry import trace from opentelemetry.trace import Status, StatusCode -from tr_sys.celery_gates.expensive_gate import exp_backoff_with_jitter +from tr_sys.celery_gates.backoff import (exp_backoff_with_jitter, constant_backoff_with_jitter, + MERGE_LOCK_MAX_RETRIES) tracer = trace.get_tracer(__name__) import asyncio import zstandard as zstd -from tr_sys.celery_gates.context import (expensive_section) +from celery.exceptions import Retry, MaxRetriesExceededError from tr_sys.otel_config import count_error, record_error -from celery.exceptions import Retry + ARS_ACTOR = { 'channel': [], @@ -57,6 +58,7 @@ NORMALIZER_URL=os.getenv("TR_NORMALIZER") if os.getenv("TR_NORMALIZER") is not None else "https://nodenorm-es.ci.transltr.io/get_normalized_nodes" ANNOTATOR_URL=os.getenv("TR_ANNOTATOR") if os.getenv("TR_ANNOTATOR") is not None else "https://biothings.ncats.io/curie" APPRAISER_URL=os.getenv("TR_APPRAISE") if os.getenv("TR_APPRAISE") is not None else "https://answerappraiser.ci.transltr.io/get_appraisal" +MERGE_ERROR_MAX_RETRIES = int(os.getenv("ARS_MERGE_ERROR_MAX_RETRIES", "8")) class QueryGraph(): def __init__(self,qg): @@ -649,15 +651,38 @@ def unlock_merge(message: Message) -> None: logging.exception("Failed to release merge_semaphore for message %s", getattr(message, "pk", "")) raise -@shared_task(name="merge-and-post-process", bind=True, acks_late=True, max_retries=20) -def merge_and_post_process(self, parent_pk,message_to_merge, agent_name): + +@shared_task(name="merge-and-post-process", bind=True, acks_late=True, max_retries=None) +def merge_and_post_process(self, parent_pk, child_pk, agent_name, error_retries=0): """ - Safe merge & post-process task: - - Acquire expensive token (expensive_section) first + Safe merge & post-process task (runs on the `heavy` queue, see celery.py task_routes; + the heavy worker pool's concurrency is what bounds how many of these run at once): + - Load the child message to merge from child_pk - Acquire DB boolean lock (inside short select_for_update atomic) - Do merge_received(), post_process() and notifications - Always release DB boolean lock in finally - - Use self.request.retries for retry/backoff + + child_pk is the PK of the child Message holding the data to merge; the task + loads and decompresses it itself. It used to receive that whole TRAPI message + inline, and since every retry republishes the payload, contention retries were + pushing multi-MB bodies through the broker over and over. Callers MUST commit + the child's data before enqueueing (use transaction.on_commit) or the task will + not find it. + + Retry functionality: + - The built-in Celery retry functionality, tracked by the counter self.request.retries and enforced by Celery, + is used for all tasks, contention is the main cause for a retry (the merge db lock is occupied). + This is done by passing max_retries=MERGE_LOCK_MAX_RETRIES after a failed merge lock acquisition. + When this kind of retry occurs we use a quick delay to avoid pointless downtime waiting on ourselves. + - Retries for errors in merging are handled separately by error_retries, an arg based counter + for actual merge_and_post_process errors, capped by MERGE_ERROR_MAX_RETRIES, enforced by the code here + and not celery's internal retry check. These use exponential backoff (1, 2, 4, 8, 16s, then capped at 30s) + to give downstream services time to recover, so the default cap of 8 buys about two minutes of patience — + enough to outlast an annotator/appraiser restart or a rolling deploy. + - Note that max_retries=None in the merge_and_post_process decorator signature means celery imposes no + limit of its own: contention retry sites must pass max_retries explicitly, while the error path + intentionally omits it and relies on the error_retries cap instead. Future changes should preserve that + split — a retry with neither limit would loop forever. """ merged=None stats={} @@ -667,128 +692,158 @@ def merge_and_post_process(self, parent_pk,message_to_merge, agent_name): attempt = (getattr(self.request, "retries", 0) or 0) + 1 outcome = "unknown" task_span.set_attribute("merge.parent_pk", str(parent_pk)) + task_span.set_attribute("merge.child_pk", str(child_pk)) task_span.set_attribute("merge.agent", agent_name) task_span.set_attribute("merge.attempt", attempt) logging.info(f"🚀Starting merge for %s with parent PK: %s"% (agent_name,parent_pk)) try: - #Acquire an expensive token so we don't hold DB locked while waiting - with expensive_section(self): - logging.info("[%s] 🟢 acquired expensive token", self.request.id) - - # short critical section: lock row + decide if we can merge + # Load the payload before the row lock (this decompresses a multi-MB + # body) so we don't hold the parent locked while doing it. + try: + with tracer.start_as_current_span("merge.load_child") as load_span: + load_span.set_attribute("merge.child_pk", str(child_pk)) + child = Message.objects.get(pk=child_pk) + message_to_merge = get_safe(child.decompress_dict(), "message") + except Message.DoesNotExist: + # The child row is gone; retrying cannot bring it back. + outcome = "child_missing" + logging.error("Child message %s no longer exists for agent %s (parent=%s); skipping merge.", + child_pk, agent_name, parent_pk) + return + if message_to_merge is None: + outcome = "child_empty" + logging.error("Child message %s for agent %s has no 'message' to merge; skipping.", + child_pk, agent_name) + return + + # short critical section: lock row + decide if we can merge + try: + with transaction.atomic(): + # a span of its own because select_for_update can block here + with tracer.start_as_current_span("merge.lock.acquire") as lock_span: + parent = get_object_or_404(Message.objects.select_for_update().filter(pk=parent_pk)) + logging.info("the merge semaphore for agent %s is %s"% (agent_name, parent.merge_semaphore)) + # how many merges this parent has already absorbed, read off the row we just loaded + version_index = len(parent.merged_versions_list or []) + task_span.set_attribute("merge.version_index", version_index) + lock_span.set_attribute("merge.version_index", version_index) + lock_span.set_attribute("merge.agent", agent_name) + lock_span.set_attribute("merge.attempt", attempt) + #try to acquire DB boolean lock + lock_acquired = try_lock_merge(parent) + lock_span.set_attribute("merge.lock.acquired", lock_acquired) + task_span.set_attribute("merge.lock.acquired", lock_acquired) + if not lock_acquired: + # Another task holds the merge lock. Retry after a quick delay. + delay = constant_backoff_with_jitter() + outcome = "lock_contended" + task_span.set_attribute("merge.lock.retry_delay_seconds", delay) + logging.info("🔄 merge lock held for %s (parent=%s); retrying in %.1fs (retries=%s)", + agent_name, parent_pk, delay, self.request.retries) + # raise retry — requeues the task and frees the worker slot. + raise self.retry(countdown=delay, max_retries=MERGE_LOCK_MAX_RETRIES) + else: + logging.info(" the merge semaphore for agent %s is %s"% (agent_name, parent.merge_semaphore)) + locked = True + # Perform the merge and persist minimal durable state inside the transaction + logging.info(f"[{self.request.id}] ✅ ENTER expensive section") + logging.info("Before merging for %s with parent PK: %s", agent_name, parent_pk) + #merging + with tracer.start_as_current_span("merge.merge_received") as merge_span: + merge_span.set_attribute("merge.agent", agent_name) + merge_span.set_attribute("merge.version_index", version_index) + merged, parent, stats = merge_received(parent, message_to_merge, agent_name) + # sizes of what we just produced, straight off the stats merge_received already computed + if isinstance(stats, dict): + for stat in ("results", "knowledge_graph_nodes", "knowledge_graph_edges"): + if stat in stats: + merge_span.set_attribute("merge.%s" % stat, stats[stat]) + task_span.set_attribute("merge.%s" % stat, stats[stat]) + logging.info(f"After merging for %s with parent PK: %s"% (agent_name,parent_pk)) + parent.save(update_fields=['params','merged_versions_list','merged_version']) + + if merged is None: + outcome = "merge_returned_none" + logging.info("merge_received returned None for %s %s", agent_name, parent_pk) + return + + notification={ + "event_type":"merged_version_begun", + "complete":False, + "merged_versions_list":parent.merged_versions_list if parent.merged_versions_list is not None else [] + } + parent.notify_subscribers(notification) + except Message.DoesNotExist: + outcome = "parent_missing" + logging.info("Message %s does not exist. Skipping merge.🚨" % parent_pk) + return + + # At this point: transaction committed, merged + parent persisted. + # Release the DB boolean lock immediately so other merges can proceed. + if locked and parent is not None: try: - with transaction.atomic(): - # a span of its own because select_for_update can block here - with tracer.start_as_current_span("merge.lock.acquire") as lock_span: - parent = get_object_or_404(Message.objects.select_for_update().filter(pk=parent_pk)) - logging.info("the merge semaphore for agent %s is %s"% (agent_name, parent.merge_semaphore)) - # how many merges this parent has already absorbed, read off the row we just loaded - version_index = len(parent.merged_versions_list or []) - task_span.set_attribute("merge.version_index", version_index) - lock_span.set_attribute("merge.version_index", version_index) - lock_span.set_attribute("merge.agent", agent_name) - lock_span.set_attribute("merge.attempt", attempt) - #try to acquire DB boolean lock - lock_acquired = try_lock_merge(parent) - lock_span.set_attribute("merge.lock.acquired", lock_acquired) - task_span.set_attribute("merge.lock.acquired", lock_acquired) - if not lock_acquired: - #someone else already locked it & possibly going through merge, retry ... - retries = self.request.retries - if retries < 10: - delay = 5 - outcome = "lock_contended" - task_span.set_attribute("merge.lock.retry_delay_seconds", delay) - if retries >= 7: - logging.warning("⚠️ High retry count (%s) for merge lock on parent=%s agent=%s.Retrying in %ss",retries, parent_pk, agent_name, delay) - else: - logging.info(" 🔄 Merged_version locked for %s. Attempt %s. Retrying in %ss",agent_name, retries, delay) - # raise retry — this will release the expensive token (expensive_section finally), stop the task, requeues it and free the worker - raise self.retry(countdown=delay) - else: - outcome = "lock_budget_exhausted" - task_span.set_status(Status(StatusCode.ERROR, "gave up waiting for the per-parent merge lock")) - logging.info("❌ Merging failed for %s %s after retries", agent_name, parent_pk) - return - else: - logging.info(" the merge semaphore for agent %s is %s"% (agent_name, parent.merge_semaphore)) - locked = True - # Perform the merge and persist minimal durable state inside the transaction - logging.info(f"[{self.request.id}] ✅ ENTER expensive section") - logging.info("Before merging for %s with parent PK: %s", agent_name, parent_pk) - #merging - with tracer.start_as_current_span("merge.merge_received") as merge_span: - merge_span.set_attribute("merge.agent", agent_name) - merge_span.set_attribute("merge.version_index", version_index) - merged, parent, stats = merge_received(parent, message_to_merge, agent_name) - # sizes of what we just produced, straight off the stats merge_received already computed - if isinstance(stats, dict): - for stat in ("results", "knowledge_graph_nodes", "knowledge_graph_edges"): - if stat in stats: - merge_span.set_attribute("merge.%s" % stat, stats[stat]) - task_span.set_attribute("merge.%s" % stat, stats[stat]) - logging.info(f"After merging for %s with parent PK: %s"% (agent_name,parent_pk)) - parent.save(update_fields=['params','merged_versions_list','merged_version']) - - if merged is None: - outcome = "merge_returned_none" - logging.info("merge_received returned None for %s %s", agent_name, parent_pk) - return - - notification={ - "event_type":"merged_version_begun", - "complete":False, - "merged_versions_list":parent.merged_versions_list if parent.merged_versions_list is not None else [] - } - parent.notify_subscribers(notification) - except Message.DoesNotExist: - outcome = "parent_missing" - logging.info("Message %s does not exist. Skipping merge.🚨" % parent_pk) - return - - # At this point: transaction committed, merged + parent persisted. - # Release the DB boolean lock immediately so other merges can proceed. - if locked and parent is not None: - try: - unlock_merge(parent) - locked = False - except Exception as e: - logging.exception("Failed to release DB merge_semaphore for parent %s", parent_pk) - record_error(e) + unlock_merge(parent) + locked = False + except Exception as e: + logging.exception("Failed to release DB merge_semaphore for parent %s", parent_pk) + record_error(e) - logging.info('merged data for agent %s with pk %s is returned & ready to be preprocessed' % (agent_name, str(merged.id))) - #post-process - # span is nested inside expensive_gate.hold, so the trace shows the token being held across the annotator/appraiser calls - with tracer.start_as_current_span("merge.post_process") as post_span: - post_span.set_attribute("merge.agent", agent_name) - post_span.set_attribute("merge.merged_pk", str(merged.id)) - merged, code, status = post_process(merged, merged.id, agent_name) - post_span.set_attribute("merge.post_process.code", code) - post_span.set_attribute("merge.post_process.status", status) - logging.info('post processing complete for agent %s with pk %s is returned & ready to be preprocessed' % (agent_name, str(merged.id))) - notification= {"event_type": "merged_version_available", - "complete": False, - "merged_version": str(merged.pk), - "merged_versions_list": parent.merged_versions_list if parent.merged_versions_list is not None else [], - 'stats': stats} - parent.notify_subscribers(notification) - - merged.status = status - merged.code = code - merged.save() - outcome = "merged" - logging.info(f"[{self.request.id}] ✅ EXIT expensive section") + logging.info('merged data for agent %s with pk %s is returned & ready to be preprocessed' % (agent_name, str(merged.id))) + #post-process + with tracer.start_as_current_span("merge.post_process") as post_span: + post_span.set_attribute("merge.agent", agent_name) + post_span.set_attribute("merge.merged_pk", str(merged.id)) + merged, code, status = post_process(merged, merged.id, agent_name) + post_span.set_attribute("merge.post_process.code", code) + post_span.set_attribute("merge.post_process.status", status) + logging.info('post processing complete for agent %s with pk %s is returned & ready to be preprocessed' % (agent_name, str(merged.id))) + notification= {"event_type": "merged_version_available", + "complete": False, + "merged_version": str(merged.pk), + "merged_versions_list": parent.merged_versions_list if parent.merged_versions_list is not None else [], + 'stats': stats} + parent.notify_subscribers(notification) + + merged.status = status + merged.code = code + merged.save() + outcome = "merged" except Retry: - # If we bubble here, expensive_section requested a retry before acquiring a token, - # or we explicitly raised retry while inside; allow it to propagate (worker freed). - if outcome == "unknown": - outcome = "gate_unavailable" + # lock contention retry raised above; let it propagate (worker slot freed) logging.info("Task retry requested — requeueing (parent=%s, retries=%s)",parent_pk, self.request.retries) raise + except MaxRetriesExceededError: + # Lock retry budget exhausted (MERGE_LOCK_MAX_RETRIES). + # This is not a real error, so keep it out of the catch-all Exception handling below. + outcome = "retry_budget_exhausted" + task_span.set_status(Status(StatusCode.ERROR, "exhausted the retry budget waiting on contention")) + logging.error("❌ Merge gave up after exhausting the retry budget for agent %s pk %s (retries=%s).", + agent_name, parent_pk, self.request.retries) + # surface the failure to API consumers, not just logs/OTEL + try: + if parent is None: + parent = Message.objects.filter(pk=parent_pk).first() + if parent is not None: + parent.notify_subscribers({ + "event_type": "merged_version_failed", + "agent_name": agent_name, + "reason": "retry_budget_exhausted", + "merged_versions_list": parent.merged_versions_list if parent.merged_versions_list is not None else [] + }) + except Exception as notify_error: + logging.exception("Failed to notify subscribers of merge retry exhaustion for parent %s", parent_pk) + record_error(notify_error) + raise + except Exception as e: + # TODO - it would be best to categorize and handle different kinds of errors in different ways here. + # If there is a deterministic error (ie something wrong with the data) retrying is a waste of time and + # hogs resources for no reason. Transient errors such as issues with postgres or service calls to annotator + # or appraiser are legitimate reasons to retry and should go through the retry process. It would also be + # good to log the kinds of errors occurring and/or make them visible to OTEL. outcome = "error" logging.info("Problem with merger for agent %s pk: %s " % (agent_name, (parent_pk))) logging.info(e, exc_info=True) @@ -798,9 +853,14 @@ def merge_and_post_process(self, parent_pk,message_to_merge, agent_name): merged.status='E' merged.code = 422 merged.save() - # retry on post_process failure - delay = exp_backoff_with_jitter(self.request.retries) - raise self.retry(exc=e, countdown=delay) + if error_retries >= MERGE_ERROR_MAX_RETRIES: + logging.error("Merge for agent %s pk %s failed after %s error retries; giving up.", + agent_name, parent_pk, error_retries) + raise + delay = exp_backoff_with_jitter(error_retries) + raise self.retry( + kwargs={**(self.request.kwargs or {}), "error_retries": error_retries + 1}, + exc=e, countdown=delay) finally: task_span.set_attribute("merge.outcome", outcome) diff --git a/tr_sys/tr_sys/celery.py b/tr_sys/tr_sys/celery.py index c68b3eaa..36d0d6fa 100644 --- a/tr_sys/tr_sys/celery.py +++ b/tr_sys/tr_sys/celery.py @@ -37,6 +37,16 @@ def debug_task(self): 'schedule': crontab(minute='*/3'), }, } +# Memory-heavy tasks ie a merge, or ingesting an ARA callback; both expand a +# multi-MB TRAPI message ~10x in memory. They go to their own queue, consumed only +# by the `heavy` worker pool. That pool's replicas x concurrency IS the +# cluster-wide cap on concurrent heavy work (it replaced the redis token gate). +# Everything else stays on the default queue so cheap tasks never wait on heavy tasks. +app.conf.task_routes = { + "merge-and-post-process": {"queue": "heavy"}, + "ingest-ara-response": {"queue": "heavy"}, +} + # this make sure that celery and rabbitMQ can reprocess the unacknowledged messages app.conf.update( task_acks_late=True, #task messages will be acknowledged after the task has been executed diff --git a/tr_sys/tr_sys/celery_gates/backoff.py b/tr_sys/tr_sys/celery_gates/backoff.py new file mode 100644 index 00000000..2624c497 --- /dev/null +++ b/tr_sys/tr_sys/celery_gates/backoff.py @@ -0,0 +1,44 @@ +""" +Retry backoff helpers and the merge-lock retry budget. + +These used to live next to the redis "expensive token" gate in expensive_gate.py. +The gate is gone: memory-heavy tasks are bounded by the `heavy` celery queue's +worker concurrency instead (see task_routes in tr_sys/celery.py). The helpers +stay because merge_and_post_process still retries on two things: + - the per-parent merge db lock (constant backoff, MERGE_LOCK_MAX_RETRIES) + - real errors (exponential backoff, MERGE_ERROR_MAX_RETRIES in tr_ars/utils.py) +""" +import os +import random + +# Celery retry budget for waiting on the per-parent merge db lock. Each attempt +# waits 2-5s (constant_backoff_with_jitter), so the default is ~6 min worst case. +# Now a lock retry republishes only pks, so a generous budget is cheap. +MERGE_LOCK_MAX_RETRIES = int(os.getenv("ARS_MERGE_LOCK_MAX_RETRIES", "100")) + + +def exp_backoff_with_jitter(retries: int, base: int = 1, max_delay: int = 30) -> int: + """ + Exponential backoff with jitter. + `retries` is the current retry count (0..). + + Use this for retrying genuine failures (e.g. a downstream error), NOT for + lock contention. Use constant_backoff_with_jitter for that. + """ + rcount = max(0, int(retries or 0)) + delay = min(max_delay, base * (2 ** rcount)) + delay = int(delay * random.uniform(0.8, 1.2)) # first retry ~1s, next ~2s, ... capping at 30s + return max(1, delay) + + +def constant_backoff_with_jitter(base: float = 2, + spread: float = 3) -> float: + """ + Short, roughly-constant backoff with jitter, for lock contention. + Unlike exponential backoff this keeps the task re-checking on a short, steady + cadence so a lock freed by a finishing merge is picked up within a few seconds + instead of sitting idle for the longer delays used for failure retries. The + delay is 2-5 seconds by default. Don't go much faster than that: each retry is a + full round trip through the celery broker. + """ + return base + random.random() * spread \ No newline at end of file diff --git a/tr_sys/tr_sys/celery_gates/context.py b/tr_sys/tr_sys/celery_gates/context.py deleted file mode 100644 index fb5d82a2..00000000 --- a/tr_sys/tr_sys/celery_gates/context.py +++ /dev/null @@ -1,78 +0,0 @@ -# this code holds context manager, imprts try_acqurie/release from expensive_gate.py - -# Relay/tr_sys/tr_sys/celery_gates/context.py -from contextlib import contextmanager -from celery.exceptions import Retry, MaxRetriesExceededError -from opentelemetry import trace -from opentelemetry.trace import Status, StatusCode -from .expensive_gate import try_acquire, release, LeaseRenewer, exp_backoff_with_jitter, ARS_EXPENSIVE_TOKEN_LIMIT -from celery.utils.log import get_task_logger -logger = get_task_logger(__name__) -tracer = trace.get_tracer(__name__) - -@contextmanager -def expensive_section(task_self, limit: int = ARS_EXPENSIVE_TOKEN_LIMIT): - """ - Usage: inside a Celery task with `bind=True`: - with expensive_section(self): - # do expensive work - If acquire fails -> raises self.retry(...) to requeue quickly. - """ - # require task_self (bind=True) - task_id = str(task_self.request.id) - retries = getattr(task_self.request, "retries", 0) or 0 - delay = exp_backoff_with_jitter(getattr(task_self.request, "retries", 0)) - # We're intentionally creating two different spans here: - # `expensive_gate.acquire` says whether this attempt got a token and how many attempts it took - with tracer.start_as_current_span( - "expensive_gate.acquire", record_exception=False, set_status_on_exception=False - ) as span: - span.set_attribute("gate.task_id", task_id) - span.set_attribute("gate.limit", limit) - span.set_attribute("gate.attempt", retries + 1) - try: - # Try to acquire; if fail -> retry right away with backoff - acquired = try_acquire(task_id, limit=limit) - span.set_attribute("gate.acquired", bool(acquired)) - span.set_attribute("gate.result", "acquired" if acquired else "unavailable") - if not acquired: - span.set_attribute("gate.retry_delay_seconds", delay) - # Use task_self.request.retries (celery increments retries on retry()). - logger.debug("Task %s could not acquire token; retrying in %ss (retries=%s)", task_id, delay, getattr(task_self.request, "retries", 0)) - # We raise self.retry so Celery releases the worker and requeues. - raise task_self.retry(countdown=delay) - except MaxRetriesExceededError: - # Out of retries — fail gracefully and log - span.set_attribute("gate.result", "budget_exhausted") - span.set_status(Status(StatusCode.ERROR, "exceeded max retries waiting for an expensive token")) - logger.warning("Task %s exceeded max retries while waiting for expensive token; failing.", task_id) - # Optionally: raise or return. We raise to mark task as failed. - raise - - - # We have the token; start renewer to extend lease periodically - renewer = LeaseRenewer(task_id) - renewer.start() - - # The other span `expensive_gate.hold` shows how long this task held a token once acquired - with tracer.start_as_current_span( - "expensive_gate.hold", record_exception=False, set_status_on_exception=False - ) as span: - span.set_attribute("gate.task_id", task_id) - span.set_attribute("gate.limit", limit) - span.set_attribute("gate.attempt", retries + 1) - try: - yield - span.set_attribute("gate.result", "completed") - except Retry: - span.set_attribute("gate.result", "retry") - raise - except Exception as e: - span.set_attribute("gate.result", "error") - span.record_exception(e) - span.set_status(Status(StatusCode.ERROR, str(e))) - raise - finally: - # stop renewing and release token (always run) - renewer.stop() - release(task_id) \ No newline at end of file diff --git a/tr_sys/tr_sys/celery_gates/expensive_gate.py b/tr_sys/tr_sys/celery_gates/expensive_gate.py deleted file mode 100644 index 52b84cee..00000000 --- a/tr_sys/tr_sys/celery_gates/expensive_gate.py +++ /dev/null @@ -1,127 +0,0 @@ -#this code holds Redis connection, Lua script registration, try_acquire(), renew(), and release(), backoff helper and lease renew thread - -# Relay/tr_sys/tr_sys/celery_gates/expensive_gate.py -import os -import time -import random -import threading -from typing import Optional -import redis -from celery.utils.log import get_task_logger -logger = get_task_logger(__name__) - -# Config via env (set these in your Helm values / CI) -REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0") -ZKEY = os.getenv("ARS_EXPENSIVE_ZKEY", "ars:expensive_tokens") -ARS_EXPENSIVE_TOKEN_LIMIT = int(os.getenv("ARS_EXPENSIVE_LIMIT", "12")) # default token limit -LEASE_MS = int(os.getenv("ARS_EXPENSIVE_LEASE_MS", "180000")) # default lease 3 minutes (ms) -RENEW_EVERY_SEC = int(os.getenv("ARS_EXPENSIVE_RENEW_SEC", "15")) # renew interval - -# Redis client -r = redis.Redis.from_url(REDIS_URL, decode_responses=True) - -# Inline Lua script: cleanup expired, check limit, add member -_ACQUIRE_LUA = """ -local zkey = KEYS[1] -local now = tonumber(ARGV[1]) -local lease_ms = tonumber(ARGV[2]) -local limit = tonumber(ARGV[3]) -local member = ARGV[4] - --- remove expired tokens -redis.call("ZREMRANGEBYSCORE", zkey, "-inf", now) - --- check capacity -local current = redis.call("ZCARD", zkey) -if tonumber(current) >= limit then - return 0 -end - --- add member with expiry score -redis.call("ZADD", zkey, now + lease_ms, member) -return 1 -""" - -_ACQUIRE_SCRIPT = None - - -def _load_acquire_script(): - global _ACQUIRE_SCRIPT - if _ACQUIRE_SCRIPT is None: - _ACQUIRE_SCRIPT = r.register_script(_ACQUIRE_LUA) - return _ACQUIRE_SCRIPT - - -def now_ms() -> int: - return int(time.time() * 1000) - - -def try_acquire(task_id: str, limit: int = ARS_EXPENSIVE_TOKEN_LIMIT) -> bool: - """ - Atomically attempt to acquire a lease token for task_id. - Returns True when token claimed, False otherwise. - """ - script = _load_acquire_script() - try: - res = script(keys=[ZKEY], args=[now_ms(), LEASE_MS, limit, task_id]) - logger.warning("try_acquire task=%s limit=%s res=%s redis=%s zkey=%s", - task_id, limit, res, REDIS_URL, ZKEY) - return bool(res) - except redis.RedisError: - # On Redis failure, be conservative and deny (or you can choose to allow) - return False - - -def renew(task_id: str) -> None: - """Extend the lease for this task if it still exists.""" - score = r.zscore(ZKEY, task_id) - if score is not None: - # set new score = now + lease_ms - r.zadd(ZKEY, {task_id: now_ms() + LEASE_MS}) - - -def release(task_id: str) -> None: - """Release the token (remove the member).""" - try: - r.zrem(ZKEY, task_id) - except redis.RedisError: - # best-effort - pass - - -def exp_backoff_with_jitter(retries: int, base: int = 1, max_delay: int = 30) -> int: - """ - Exponential backoff with jitter. - `retries` is the current retry count (0..). - """ - rcount = max(0, int(retries or 0)) - delay = min(max_delay, base * (2 ** rcount)) - delay = int(delay * random.uniform(0.8, 1.2)) #makes first retry ~2sec, next ~4sec, etc. capping at 30 sec - return max(1, delay) - - -class LeaseRenewer: - """ - Background renewer thread to keep a token alive while the task runs. - Call start() right after acquiring, and stop() in finally. - """ - def __init__(self, task_id: str, interval: Optional[int] = None): - self.task_id = task_id - self.interval = interval or RENEW_EVERY_SEC - self._stop = threading.Event() - self._thread = threading.Thread(target=self._loop, daemon=True) - - def start(self): - self._thread.start() - - def stop(self): - self._stop.set() - # thread will exit soon - - def _loop(self): - while not self._stop.wait(self.interval): - try: - renew(self.task_id) - except Exception: - # best-effort: ignore and continue - pass \ No newline at end of file diff --git a/tr_sys/tr_sys/otel_config.py b/tr_sys/tr_sys/otel_config.py index eb200c79..98e78f28 100644 --- a/tr_sys/tr_sys/otel_config.py +++ b/tr_sys/tr_sys/otel_config.py @@ -18,13 +18,13 @@ # Don't trace health checks and other noise endpoints. # Deployment env can override with its own comma separated list with OTEL_PYTHON_DJANGO_EXCLUDED_URLS -DEFAULT_EXCLUDED_URLS = "ars/api/health,ars/api/retain" +DEFAULT_EXCLUDED_URLS = "ars/api/health,ars/api/retain,ars/api/get_status" # Periodic tasks whose spans are identical every time they run: catch_timeout # every 3 minutes and celery's own daily result-backend housekeeping. Celery # traces each one twice, as apply_async/ and run/. They're not # very useful and are noisy, so just drop them. -DONT_TRACE_TASKS = ("catch_timeout", "celery.backend_cleanup") +DONT_TRACE_TASKS = ("catch_timeout", "celery.backend_cleanup", "health_ping") # Calls with these methods aren't interesting for what we care about for the ARS DONT_TRACE_METHODS = frozenset(("GET", "HEAD", "OPTIONS")) diff --git a/tr_sys/tr_sys/routing.py b/tr_sys/tr_sys/routing.py deleted file mode 100644 index 96d6284f..00000000 --- a/tr_sys/tr_sys/routing.py +++ /dev/null @@ -1,5 +0,0 @@ -from channels.routing import ProtocolTypeRouter - -application = ProtocolTypeRouter({ - # (http->django views is added by default) -}) diff --git a/tr_sys/tr_sys/settings.py b/tr_sys/tr_sys/settings.py index ef92bb18..220ba271 100644 --- a/tr_sys/tr_sys/settings.py +++ b/tr_sys/tr_sys/settings.py @@ -41,7 +41,6 @@ 'django.contrib.messages', 'django.contrib.staticfiles', 'corsheaders', -# 'channels', 'tr_ars.apps.ARSConfig', #'tr_ars.default_ars_app.ars_app.AppConfig', 'tr_ara_aragorn.aragorn_app.AppConfig', @@ -100,17 +99,6 @@ ] WSGI_APPLICATION = 'tr_sys.wsgi.application' -# Channels -ASGI_APPLICATION = 'tr_sys.routing.application' - -CHANNEL_LAYERS = { - 'default': { - 'BACKEND': 'channels_redis.core.RedisChannelLayer', - 'CONFIG': { - "hosts": [('127.0.0.1', 6379)], - }, - }, -} # Database # https://docs.djangoproject.com/en/1.11/ref/settings/#databases