diff --git a/Procfile b/Procfile index d79de9e8..620e6817 100755 --- a/Procfile +++ b/Procfile @@ -1,5 +1,6 @@ web: PYTHONUNBUFFERED=1 FLASK_ENV=development FLASK_DEBUG=1 FLASK_APP=agent.web:application SENTRY_DSN="https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17" ./env/bin/flask run -p 25052 redis: redis-server redis.conf -worker_1: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17' -worker_2: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17' +worker_1: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 signup high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17' +worker_2: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 signup high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17' worker_backup: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 backup high default low --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17' +worker_signup: PYTHONUNBUFFERED=1 ./repo/wait-for-it.sh redis://127.0.0.1:25025 && ./env/bin/rq worker --url redis://127.0.0.1:25025 signup --sentry-dsn 'https://e9b2a2274f2245daab48c6245fb69431@trace.frappe.cloud/17' diff --git a/agent/proxy.py b/agent/proxy.py index b004ce43..7da465d0 100644 --- a/agent/proxy.py +++ b/agent/proxy.py @@ -136,7 +136,7 @@ def _add_auto_scale_sites_to_upstream( def add_site_to_upstream_job(self, upstream, site): self.add_site_domain_to_upstream(upstream, site) - @job("Add Domain to Upstream") + @job("Add Domain to Upstream", priority="signup") def add_domain_to_upstream_job(self, upstream, domain): self.add_site_domain_to_upstream(upstream, domain) @@ -289,7 +289,7 @@ def update_site_status(self, upstream, site, status): with open(site_file, "w") as f: f.write(status) - @job("Setup Redirects on Hosts") + @job("Setup Redirects on Hosts", priority="signup") def setup_redirects_job(self, hosts, target): with self.proxy_config_modification_lock: if target in hosts: diff --git a/agent/server.py b/agent/server.py index f931babc..0f9de543 100644 --- a/agent/server.py +++ b/agent/server.py @@ -86,6 +86,10 @@ def press_url(self): def backup_workers(self) -> int: return self.config.get("backup_workers", 1) + @property + def signup_workers(self) -> int: + return self.config.get("signup_workers", 1) + def docker_login(self, registry): url = shlex.quote(registry["url"]) username = shlex.quote(registry["username"]) @@ -948,6 +952,10 @@ def update_agent_web(self, url=None, branch="master"): worker_name = f"agent:worker_backup-{worker}" self.execute(f"sudo supervisorctl restart {worker_name}", non_zero_throw=False) + for worker in range(self.signup_workers): + worker_name = f"agent:worker_signup-{worker}" + self.execute(f"sudo supervisorctl restart {worker_name}", non_zero_throw=False) + self.execute("sudo supervisorctl restart agent:web") run_patches() @@ -975,7 +983,7 @@ def update_agent_cli( # noqa: C901 # Stop required services if restart_rq_workers: - for group in ("worker", "worker_backup"): + for group in ("worker", "worker_backup", "worker_signup"): for worker_id in supervisor_status.get(group, {}): self.execute( f"sudo supervisorctl stop agent:{group}-{worker_id}", @@ -1020,6 +1028,8 @@ def update_agent_cli( # noqa: C901 self.execute(f"sudo supervisorctl start agent:worker-{i}") for i in range(self.backup_workers): self.execute(f"sudo supervisorctl start agent:worker_backup-{i}", non_zero_throw=False) + for i in range(self.signup_workers): + self.execute(f"sudo supervisorctl start agent:worker_signup-{i}", non_zero_throw=False) if restart_web_workers: self.execute("sudo supervisorctl start agent:web") @@ -1281,6 +1291,7 @@ def _generate_supervisor_config(self): "gunicorn_workers": self.config.get("gunicorn_workers", 2), "workers": self.config["workers"], "backup_workers": self.backup_workers, + "signup_workers": self.signup_workers, "directory": self.directory, "user": self.config["user"], "sentry_dsn": self.config.get("sentry_dsn"), diff --git a/agent/site.py b/agent/site.py index 16e559cc..2064106b 100644 --- a/agent/site.py +++ b/agent/site.py @@ -350,7 +350,7 @@ def update_config(self, value, remove=None): self.set_config(new_config) - @job("Add Domain", priority="high") + @job("Add Domain", priority="signup") def add_domain(self, domain): domains = set(self.config.get("domains", [])) domains.add(domain) @@ -501,7 +501,7 @@ def complete_setup_wizard(self, data): command = f"execute frappe.desk.page.setup_wizard.setup_wizard.setup_complete --kwargs {payload}" return self.bench_execute(command) - @job("Update Site Configuration", priority="high") + @job("Update Site Configuration", priority="signup") def update_config_job(self, value, remove): self.update_config(value, remove) diff --git a/agent/templates/agent/supervisor.conf.jinja2 b/agent/templates/agent/supervisor.conf.jinja2 index 4095f05c..2f280bff 100644 --- a/agent/templates/agent/supervisor.conf.jinja2 +++ b/agent/templates/agent/supervisor.conf.jinja2 @@ -18,7 +18,7 @@ user={{ user }} directory={{ directory }} [program:worker] -command=bash -c "{{ directory }}/repo/wait-for-it.sh redis://127.0.0.1:{{ redis_port }} && exec {{ directory }}/env/bin/rq worker {% if sentry_dsn %}--sentry-dsn '{{ sentry_dsn }}'{% endif %} --url redis://127.0.0.1:{{ redis_port }} high default low" +command=bash -c "{{ directory }}/repo/wait-for-it.sh redis://127.0.0.1:{{ redis_port }} && exec {{ directory }}/env/bin/rq worker {% if sentry_dsn %}--sentry-dsn '{{ sentry_dsn }}'{% endif %} --url redis://127.0.0.1:{{ redis_port }} signup high default low" environment=PYTHONUNBUFFERED=1 autostart=true autorestart=true @@ -49,6 +49,22 @@ stderr_logfile={{ directory }}/logs/worker_backup.error.log user={{ user }} directory={{ directory }} +{# Signup jobs get their own worker, so long jobs on the shared queues can't delay a signup. #} +{# It takes no other queue: a long job picked up while idle would hold it when a signup arrives. #} +[program:worker_signup] +command=bash -c "{{ directory }}/repo/wait-for-it.sh redis://127.0.0.1:{{ redis_port }} && exec {{ directory }}/env/bin/rq worker {% if sentry_dsn %}--sentry-dsn '{{ sentry_dsn }}'{% endif %} --url redis://127.0.0.1:{{ redis_port }} signup" +environment=PYTHONUNBUFFERED=1 +autostart=true +autorestart=true +stopwaitsecs=60 +killasgroup=true +numprocs={{ signup_workers }} +process_name=%(program_name)s-%(process_num)d +stdout_logfile={{ directory }}/logs/worker_signup.log +stderr_logfile={{ directory }}/logs/worker_signup.error.log +user={{ user }} +directory={{ directory }} + {% if is_proxy_server or is_standalone %} [program:nginx_reload_manager] command=bash -c "{{ directory }}/repo/wait-for-it.sh redis://127.0.0.1:{{ redis_port }} && exec {{directory}}/env/bin/python {{ directory }}/repo/agent/nginx_reload_manager.py" @@ -64,7 +80,7 @@ directory={{ directory }} [group:agent] -{% set programs = "web, redis, worker, worker_backup" %} +{% set programs = "web, redis, worker, worker_backup, worker_signup" %} {% if is_proxy_server or is_standalone %} {% set programs = programs + ", nginx_reload_manager" %} diff --git a/agent/tests/test_server.py b/agent/tests/test_server.py index d2f7800e..a9aade27 100644 --- a/agent/tests/test_server.py +++ b/agent/tests/test_server.py @@ -1,5 +1,6 @@ from __future__ import annotations +import configparser import json import os import shutil @@ -97,3 +98,49 @@ def test_a_rejected_source_leaves_the_stored_config_untouched(self): config = self.server.config self.assertEqual(config["ip_accept"], ["183.82.5.84/32"]) self.assertEqual(config["proxy_ip"], "10.3.1.0/24") + + +class TestServerSupervisorConfig(unittest.TestCase): + """Signup jobs get a worker that long jobs on the shared queues can't hold.""" + + def setUp(self): + self.directory = tempfile.mkdtemp() + with open(os.path.join(self.directory, "config.json"), "w") as f: + json.dump( + { + "name": "f1-mumbai.frappe.cloud", + "benches_directory": os.path.join(self.directory, "benches"), + "nginx_directory": os.path.join(self.directory, "nginx"), + "web_port": 25052, + "redis_port": 25025, + "workers": 2, + "user": "frappe", + }, + f, + ) + + self.server = Server(self.directory) + + def tearDown(self): + shutil.rmtree(self.directory) + + def rendered_supervisor_config(self): + self.server._generate_supervisor_config() + config = configparser.ConfigParser(interpolation=None) + config.read(os.path.join(self.directory, "supervisor.conf")) + return config + + def test_signup_worker_listens_only_on_the_signup_queue(self): + command = self.rendered_supervisor_config()["program:worker_signup"]["command"] + + self.assertTrue(command.endswith(' --url redis://127.0.0.1:25025 signup"')) + + def test_shared_workers_also_take_signup_jobs(self): + command = self.rendered_supervisor_config()["program:worker"]["command"] + + self.assertIn(" signup high default low", command) + + def test_signup_worker_is_in_the_agent_group(self): + programs = self.rendered_supervisor_config()["group:agent"]["programs"] + + self.assertIn("worker_signup", [program.strip() for program in programs.split(",")])