Merge ~r00ta/maas:lp-2027735-3.4 into maas:3.4

[Jacopo Rota]

Cherry pick fdd4e6da22c3f23bde83cc3f138a00ab3db42f46 - bind each regiond process to a dedicated unix socket and let nginx load balancing the requests

Cherry pick fdd4e6da22c3f23bde83cc3f138a00ab3db42f46 - bind each regiond process to a dedicated unix socket and let nginx load balancing the requests

[Jacopo Rota (r00ta)]

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b lp-2027735-3.4 lp:~r00ta/maas/+git/maas into -b 3.4 lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/3057/console
COMMIT: 5ba89afc21bd8c0ddd4751c02b1ee3156617b004

review: Needs Fixing

[Jacopo Rota (r00ta)]

jenkins: !test

[MAAS Lander (maas-lander)]

UNIT TESTS
-b lp-2027735-3.4 lp:~r00ta/maas/+git/maas into -b 3.4 lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/3063/console
COMMIT: 5ba89afc21bd8c0ddd4751c02b1ee3156617b004

review: Needs Fixing

[Jacopo Rota (r00ta)]

jenkins: !test

[MAAS Lander (maas-lander)]

UNIT TESTS
-b lp-2027735-3.4 lp:~r00ta/maas/+git/maas into -b 3.4 lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 5ba89afc21bd8c0ddd4751c02b1ee3156617b004

review: Approve

diff --git a/snap/local/tree/bin/run-regiond b/snap/local/tree/bin/run-regiond
index e7c86ea..f3f8e7d 100755
--- a/snap/local/tree/bin/run-regiond
+++ b/snap/local/tree/bin/run-regiond
@@ -18,7 +18,7 @@ export MAAS_SYSLOG_LOG_DIR="$SNAP_COMMON/log"
 export MAAS_IMAGES_KEYRING_FILEPATH="/snap/maas/current/usr/share/keyrings/ubuntu-cloudimage-keyring.gpg"
 export MAAS_THIRD_PARTY_DRIVER_SETTINGS="$SNAP/etc/maas/drivers.yaml"
 export MAAS_HTTP_CONFIG_DIR="$SNAP_DATA/http"
-export MAAS_HTTP_SOCKET_PATH="$SNAP_DATA/maas-regiond-webapp.sock"
+export MAAS_HTTP_SOCKET_WORKER_BASE_PATH="$SNAP_DATA/maas-regiond-webapp.sock"
 
 # ensure these dirs exist here since the region creates config files for other
 # services
diff --git a/src/maasserver/regiondservices/http.py b/src/maasserver/regiondservices/http.py
index 0b8cb50..55c633a 100644
--- a/src/maasserver/regiondservices/http.py
+++ b/src/maasserver/regiondservices/http.py
@@ -18,6 +18,7 @@ from maasserver.regiondservices import certificate_expiration_check
 from maasserver.service_monitor import service_monitor
 from maasserver.utils import load_template
 from maasserver.utils.threads import deferToDatabase
+from maasserver.workers import WorkersService
 from provisioningserver.certificates import Certificate
 from provisioningserver.logger import LegacyLogger
 from provisioningserver.path import get_maas_data_path
@@ -29,8 +30,17 @@ from provisioningserver.utils.fs import atomic_write, get_root_path, snap
 
 log = LegacyLogger()
 
+REGIOND_SOCKET_PATH = os.getenv(
+    "MAAS_HTTP_SOCKET_WORKER_BASE_PATH",
+    get_maas_data_path("maas-regiond-webapp.sock"),
+)
+
 
 class RegionHTTPService(Service):
+    @staticmethod
+    def build_unix_socket_path_for_worker(worker_id: str) -> str:
+        return f"{REGIOND_SOCKET_PATH}.{worker_id}"
+
     def __init__(self, postgresListener: PostgresListenerService = None):
         super().__init__()
         self.listener = postgresListener
@@ -57,10 +67,12 @@ class RegionHTTPService(Service):
     def _configure(self, configuration):
         """Update the HTTP configuration for the region proxy service."""
         template = load_template("http", "regiond.nginx.conf.template")
-        socket_path = os.getenv(
-            "MAAS_HTTP_SOCKET_PATH",
-            get_maas_data_path("maas-regiond-webapp.sock"),
-        )
+
+        # Load balancing the unix sockets
+        worker_socket_paths = [
+            self.build_unix_socket_path_for_worker(worker_id)
+            for worker_id in WorkersService.get_worker_ids()
+        ]
 
         if configuration.tls_enabled:
             key_path, cert_path = self._create_cert_files(configuration.cert)
@@ -72,7 +84,7 @@ class RegionHTTPService(Service):
             "tls_port": configuration.port,
             "tls_key_path": key_path,
             "tls_cert_path": cert_path,
-            "socket_path": socket_path,
+            "worker_socket_paths": worker_socket_paths,
             "static_dir": str(get_root_path() / "usr/share/maas"),
         }
         rendered = template.substitute(environ).encode()
diff --git a/src/maasserver/regiondservices/tests/test_http.py b/src/maasserver/regiondservices/tests/test_http.py
index fc423b6..32f11a1 100644
--- a/src/maasserver/regiondservices/tests/test_http.py
+++ b/src/maasserver/regiondservices/tests/test_http.py
@@ -15,6 +15,7 @@ from maasserver.secrets import SecretManager
 from maasserver.testing.testcase import MAASTransactionServerTestCase
 from maasserver.triggers.testing import TransactionalHelpersMixin
 from maasserver.utils.threads import deferToDatabase
+from maasserver.workers import WorkersService
 from maastesting.crochet import wait_for
 from provisioningserver.testing.certificates import (
     get_sample_cert_with_cacerts,
@@ -85,6 +86,10 @@ class TestRegionHTTPService(
         mock_cert_check.assert_called_once_with()
 
     def test_configure_not_snap(self):
+        # MAASDataFixture updates `MAAS_DATA` in the environment to point to this new location.
+        data_path = os.getenv("MAAS_DATA")
+        http.REGIOND_SOCKET_PATH = f"{data_path}/maas-regiond-webapp.sock"
+
         tempdir = self.make_dir()
         nginx_conf = Path(tempdir) / "regiond.nginx.conf"
         service = http.RegionHTTPService()
@@ -97,10 +102,14 @@ class TestRegionHTTPService(
 
         service._configure(http._Configuration(self.cert, port=5443))
 
-        # MAASDataFixture updates `MAAS_DATA` in the environment to point to this new location.
-        data_path = os.getenv("MAAS_DATA")
         nginx_config = nginx_conf.read_text()
-        self.assertIn(f"{data_path}/maas-regiond-webapp.sock;", nginx_config)
+
+        worker_ids = WorkersService.get_worker_ids()
+        for worker_id in worker_ids:
+            self.assertIn(
+                f"{data_path}/maas-regiond-webapp.sock.{worker_id};",
+                nginx_config,
+            )
         self.assertIn("root /usr/share/maas/web/static;", nginx_config)
         self.assertIn("listen 5443 ssl http2;", nginx_config)
         self.assertIn("ssl_certificate cert_path;", nginx_config)
@@ -111,11 +120,12 @@ class TestRegionHTTPService(
             os,
             "environ",
             {
-                "MAAS_HTTP_SOCKET_PATH": "/snap/maas/maas-regiond-webapp.sock",
                 "SNAP": "/snap/maas/5443",
                 "MAAS_HTTP_CONFIG_DIR": os.getenv("MAAS_DATA"),
             },
         )
+        http.REGIOND_SOCKET_PATH = "/snap/maas/maas-regiond-webapp.sock"
+
         tempdir = self.make_dir()
         nginx_conf = Path(tempdir) / "regiond.nginx.conf"
         service = http.RegionHTTPService()
@@ -129,9 +139,12 @@ class TestRegionHTTPService(
         service._configure(http._Configuration(cert=self.cert, port=5443))
 
         nginx_config = nginx_conf.read_text()
-        self.assertIn(
-            "server unix:/snap/maas/maas-regiond-webapp.sock;", nginx_config
-        )
+        worker_ids = WorkersService.get_worker_ids()
+        for worker_id in worker_ids:
+            self.assertIn(
+                f"server unix:/snap/maas/maas-regiond-webapp.sock.{worker_id};",
+                nginx_config,
+            )
         self.assertIn(
             "root /snap/maas/5443/usr/share/maas/web/static;", nginx_config
         )
diff --git a/src/maasserver/templates/http/regiond.nginx.conf.template b/src/maasserver/templates/http/regiond.nginx.conf.template
index 1c0c539..681d6e6 100644
--- a/src/maasserver/templates/http/regiond.nginx.conf.template
+++ b/src/maasserver/templates/http/regiond.nginx.conf.template
@@ -1,7 +1,9 @@
 # -*- mode: nginx -*-
 
 upstream regiond-webapp {
-    server unix:{{socket_path}};
+{{for worker_socket_path in worker_socket_paths}}
+    server unix:{{worker_socket_path}};
+{{endfor}}
 }
 
 proxy_http_version 1.1;
diff --git a/src/maasserver/tests/test_workers.py b/src/maasserver/tests/test_workers.py
index 2da7ac8..8276a6b 100644
--- a/src/maasserver/tests/test_workers.py
+++ b/src/maasserver/tests/test_workers.py
@@ -18,7 +18,6 @@ from maasserver.workers import (
     WorkersService,
 )
 from maastesting.crochet import wait_for
-from maastesting.matchers import MockCalledOnceWith, MockCallsMatch
 from maastesting.testcase import MAASTestCase
 from provisioningserver.utils.twisted import DeferredValue
 
@@ -29,14 +28,14 @@ class TestWorkersCount(MAASTestCase):
     def test_MAX_WORKERS_COUNT_default_cpucount(self):
         from maasserver.workers import MAX_WORKERS_COUNT
 
-        self.assertEqual(os.cpu_count(), MAX_WORKERS_COUNT)
+        assert os.cpu_count() == MAX_WORKERS_COUNT
 
     def test_set_max_workers_count(self):
         worker_count = random.randint(1, 8)
         set_max_workers_count(worker_count)
         from maasserver.workers import MAX_WORKERS_COUNT
 
-        self.assertEqual(worker_count, MAX_WORKERS_COUNT)
+        assert worker_count == MAX_WORKERS_COUNT
 
 
 class TestWorkersService(MAASTestCase):
@@ -47,31 +46,61 @@ class TestWorkersService(MAASTestCase):
 
         from maasserver.workers import MAX_WORKERS_COUNT
 
-        self.assertEqual(MAX_WORKERS_COUNT, service.worker_count)
-        self.assertEqual(sys.argv[0], service.worker_cmd)
+        assert MAX_WORKERS_COUNT == len(service.get_worker_ids())
+        assert sys.argv[0] == service.worker_cmd
 
     def test_calls_spawnWorkers_on_start(self):
         service = WorkersService(reactor)
         self.patch(service, "spawnWorkers")
         service.startService()
-        self.assertThat(service.spawnWorkers, MockCalledOnceWith())
+        service.spawnWorkers.assert_called_once()
 
     def test_spawnWorkers_calls__spawnWorker_for_missing_workers(self):
         worker_count = random.randint(2, 16)
-        service = WorkersService(reactor, worker_count=worker_count)
+        set_max_workers_count(worker_count)
+        service = WorkersService(reactor)
         self.patch(service, "_spawnWorker")
         pid = random.randint(1, 500)
-        service.workers[pid] = WorkerProcess(service)
+        service.workers[pid] = WorkerProcess(service, worker_id="0")
+        service.missing_worker_ids.remove("0")
         service.spawnWorkers()
-        calls = [call(runningImport=True)] + [
-            call() for _ in range(worker_count - 2)
+        calls = [call("1", runningImport=True)] + [
+            call(str(worker_id)) for worker_id in range(2, worker_count)
         ]
-        self.assertThat(service._spawnWorker, MockCallsMatch(*calls))
+        service._spawnWorker.assert_has_calls(calls)
+
+    def test_registerWorker(self):
+        worker_count = 2
+        set_max_workers_count(worker_count)
+        service = WorkersService(reactor)
+        self.patch(service, "_spawnWorker")
+
+        worker = WorkerProcess(service, worker_id="0")
+        worker.pid = 100
+        service.registerWorker(worker)
+
+        service.spawnWorkers()
+        calls = [call("1", runningImport=True)]
+        service._spawnWorker.assert_has_calls(calls)
+
+    def test_unregisterWorker(self):
+        worker_count = 2
+        set_max_workers_count(worker_count)
+        service = WorkersService(reactor)
+        self.patch(service, "_spawnWorker")
+
+        worker = WorkerProcess(service, worker_id="0")
+        worker.pid = 100
+        service.registerWorker(worker)
+        service.unregisterWorker(worker, None)
+        calls = [call("1", runningImport=True), call("0")]
+        service._spawnWorker.assert_has_calls(calls)
 
     @wait_for_reactor
     @inlineCallbacks
     def test_killWorker_spawns_another(self):
-        service = WorkersService(reactor, worker_count=1, worker_cmd="cat")
+        set_max_workers_count(1)
+        service = WorkersService(reactor, worker_cmd="cat")
 
         dv = DeferredValue()
         original_unregisterWorker = service.unregisterWorker
@@ -91,15 +120,16 @@ class TestWorkersService(MAASTestCase):
             pid = list(service.workers.keys())[0]
             service.killWorker(pid)
             yield dv.get(timeout=2)
-            self.assertNotIn(pid, service.workers)
-            self.assertEqual(1, len(service.workers))
+            assert pid not in service.workers
+            assert len(service.workers) == 1
         finally:
             service.stopService()
 
     @wait_for_reactor
     @inlineCallbacks
     def test_termWorker_spawns_another(self):
-        service = WorkersService(reactor, worker_count=1, worker_cmd="cat")
+        set_max_workers_count(1)
+        service = WorkersService(reactor, worker_cmd="cat")
 
         dv = DeferredValue()
         original_unregisterWorker = service.unregisterWorker
@@ -119,15 +149,16 @@ class TestWorkersService(MAASTestCase):
             pid = list(service.workers.keys())[0]
             service.termWorker(pid)
             yield dv.get(timeout=2)
-            self.assertNotIn(pid, service.workers)
-            self.assertEqual(1, len(service.workers))
+            assert pid not in service.workers
+            assert len(service.workers) == 1
         finally:
             service.stopService()
 
     @wait_for_reactor
     @inlineCallbacks
     def test_stopService_doesnt(self):
-        service = WorkersService(reactor, worker_count=1, worker_cmd="cat")
+        set_max_workers_count(1)
+        service = WorkersService(reactor, worker_cmd="cat")
 
         dv = DeferredValue()
         original_unregisterWorker = service.unregisterWorker
@@ -142,9 +173,9 @@ class TestWorkersService(MAASTestCase):
 
         try:
             service.startService()
-            self.assertEqual(1, len(service.workers))
+            assert len(service.workers) == 1
         finally:
             service.stopService()
 
         yield dv.get(timeout=2)
-        self.assertEqual({}, service.workers)
+        assert len(service.workers) == 0
diff --git a/src/maasserver/webapp.py b/src/maasserver/webapp.py
index 143364e..5363847 100644
--- a/src/maasserver/webapp.py
+++ b/src/maasserver/webapp.py
@@ -20,6 +20,7 @@ from twisted.web.server import Request, Site
 from twisted.web.wsgi import WSGIResource
 
 from maasserver import concurrency
+from maasserver.regiondservices.http import RegionHTTPService
 from maasserver.utils.threads import deferToDatabase
 from maasserver.utils.views import WebApplicationHandler
 from maasserver.websockets.protocol import WebSocketFactory
@@ -29,7 +30,6 @@ from maasserver.websockets.websockets import (
 )
 from metadataserver.api_twisted import StatusHandlerResource
 from provisioningserver.logger import LegacyLogger
-from provisioningserver.path import get_maas_data_path
 from provisioningserver.utils.twisted import (
     asynchronous,
     reducedWebLogFormatter,
@@ -227,16 +227,18 @@ class WebApplicationService(StreamServerEndpointService):
     def _makeEndpoint(self):
         """Make the endpoint for the webapp."""
 
-        socket_path = os.getenv(
-            "MAAS_HTTP_SOCKET_PATH",
-            get_maas_data_path("maas-regiond-webapp.sock"),
+        worker_id = os.getenv("MAAS_REGIOND_WORKER_ID", "")
+        worker_socket_path = (
+            RegionHTTPService.build_unix_socket_path_for_worker(worker_id)
         )
 
         s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        if os.path.exists(socket_path):
-            os.unlink(socket_path)
+        try:
+            os.unlink(worker_socket_path)
+        except FileNotFoundError:
+            pass
 
-        s.bind(socket_path)
+        s.bind(worker_socket_path)
         # Use a backlog of 50, which seems to be fairly common.
         s.listen(50)
 
diff --git a/src/maasserver/workers.py b/src/maasserver/workers.py
index 9e9a0b1..2f40068 100644
--- a/src/maasserver/workers.py
+++ b/src/maasserver/workers.py
@@ -26,9 +26,10 @@ def set_max_workers_count(worker_count):
 
 
 class WorkerProcess(protocol.ProcessProtocol):
-    def __init__(self, service, runningImport=False):
+    def __init__(self, service, worker_id: str, runningImport=False):
         super().__init__()
         self.service = service
+        self.worker_id = worker_id
         self.runningImport = runningImport
 
     def connectionMade(self):
@@ -54,17 +55,19 @@ class WorkersService(service.Service):
     Manages the lifecycle of the workers.
     """
 
-    def __init__(self, reactor, *, worker_count=None, worker_cmd=None):
+    @staticmethod
+    def get_worker_ids() -> list[str]:
+        return [str(worker_id) for worker_id in range(MAX_WORKERS_COUNT)]
+
+    def __init__(self, reactor, *, worker_cmd=None):
         super().__init__()
         self.reactor = reactor
         self.stopping = False
-        self.worker_count = worker_count
-        if self.worker_count is None:
-            self.worker_count = MAX_WORKERS_COUNT
         self.worker_cmd = worker_cmd
         if self.worker_cmd is None:
             self.worker_cmd = sys.argv[0]
         self.workers = {}
+        self.missing_worker_ids = self.get_worker_ids()
 
     def startService(self):
         """Start the workers."""
@@ -85,27 +88,30 @@ class WorkersService(service.Service):
         if self.stopping:
             # Don't spwan new workers if the service is stopping.
             return
-        missing = self.worker_count - len(self.workers)
         if self.workers:
             runningImport = max(
                 worker.runningImport for worker in self.workers.values()
             )
         else:
             runningImport = False
-        for _ in range(missing):
+
+        # Work on a copy to avoid races
+        for worker_id in list(self.missing_worker_ids):
             if not runningImport:
-                self._spawnWorker(runningImport=True)
+                self._spawnWorker(worker_id, runningImport=True)
                 runningImport = True
             else:
-                self._spawnWorker()
+                self._spawnWorker(worker_id)
 
     def registerWorker(self, worker):
         """Register the worker."""
         self.workers[worker.pid] = worker
+        self.missing_worker_ids.remove(worker.worker_id)
 
     def unregisterWorker(self, worker, status):
         """Worker has died."""
         del self.workers[worker.pid]
+        self.missing_worker_ids.append(worker.worker_id)
         self.spawnWorkers()
 
     def termWorker(self, pid):
@@ -122,11 +128,14 @@ class WorkersService(service.Service):
             log.msg("Killing worker pid:%d." % pid)
             worker.signal("KILL")
 
-    def _spawnWorker(self, runningImport=False):
+    def _spawnWorker(self, worker_id: str, runningImport: bool = False):
         """Spawn a new worker."""
-        worker = WorkerProcess(self, runningImport=runningImport)
+        worker = WorkerProcess(
+            self, worker_id=worker_id, runningImport=runningImport
+        )
         env = os.environ.copy()
         env["MAAS_REGIOND_PROCESS_MODE"] = "worker"
+        env["MAAS_REGIOND_WORKER_ID"] = worker_id
         env["MAAS_REGIOND_WORKER_COUNT"] = str(MAX_WORKERS_COUNT)
         if runningImport:
             env["MAAS_REGIOND_RUN_IMPORTER_SERVICE"] = "true"