Compose overlay for proxy + management API, with hot reload

deploy/cache-me-outside.env.example

@@ -116,3 +116,6 @@ CMO_CLUSTER_NODE_TIMEOUT=5000
 # Optional. Comma-separated CIDRs allowed to hit the published port.
 # Apply with deploy/ip-allowlist.sh (DOCKER-USER chain).
 # CMO_ALLOWED_CIDRS=10.0.0.0/8,203.0.113.7/32
+
+# CMO_MGMT_PORT=8088
+# CMO_TENANT_PORTS=7001-7020

deploy/docker-compose.tenants.yml

@@ -0,0 +1,82 @@
+# Optional: multi-tenant front door (proxy + management API/console).
+# Layer on top of docker-compose.yml:
+#
+#   docker compose -f docker-compose.yml -f docker-compose.tenants.yml up -d --build
+#
+# - cmo-mgmt: API, CLI target and web console on ${CMO_BIND_ADDRESS}:${CMO_MGMT_PORT:-8088}.
+#   Needs CMO_ADMIN_TOKEN (24+ chars) in the env file. Put TLS in front before
+#   exposing it beyond localhost.
+# - cmo-proxy: one port per database, 7001-7020 by default, limits enforced,
+#   reloads tenants.json within ~2 s of a change.
+# Both share the tenants volume. Clients connect to the proxy port with the
+# database's own user and password, not to 6379.
+
+services:
+  cmo-proxy:
+    image: python:3.12-slim
+    container_name: ${CMO_NAME_PREFIX:-cache-me-outside}-proxy
+    command: ["python3", "/opt/cmo/proxy/cmo_proxy.py", "--watch", "/var/lib/cmo/tenants.json"]
+    volumes:
+      - ../tools/proxy:/opt/cmo/proxy:ro
+      - cmo-tenants:/var/lib/cmo:ro
+    ports:
+      - "${CMO_BIND_ADDRESS:-127.0.0.1}:${CMO_TENANT_PORTS:-7001-7020}:${CMO_TENANT_PORTS:-7001-7020}"
+    depends_on:
+      cache-me-outside:
+        condition: service_healthy
+      cmo-tenants-init:
+        condition: service_completed_successfully
+    restart: unless-stopped
+    init: true
+    read_only: true
+    user: "65534:65534"
+    security_opt:
+      - no-new-privileges:true
+    cap_drop: [ALL]
+    mem_limit: ${CMO_PROXY_MEMORY:-128m}
+    networks:
+      cmo: {}
+
+  cmo-mgmt:
+    image: python:3.12-slim
+    container_name: ${CMO_NAME_PREFIX:-cache-me-outside}-mgmt
+    command: ["python3", "/opt/cmo/mgmt/cmo_mgmt.py",
+              "--config", "/var/lib/cmo/tenants.json",
+              "--server", "cache-me-outside:6379",
+              "--listen", "0.0.0.0:8088",
+              "--proxy-bind", "0.0.0.0"]
+    env_file:
+      - path: ${CMO_ENV_FILE:-.env}
+        required: false
+    volumes:
+      - ../tools/mgmt:/opt/cmo/mgmt:ro
+      - cmo-tenants:/var/lib/cmo
+    ports:
+      - "${CMO_BIND_ADDRESS:-127.0.0.1}:${CMO_MGMT_PORT:-8088}:8088"
+    depends_on:
+      cache-me-outside:
+        condition: service_healthy
+      cmo-tenants-init:
+        condition: service_completed_successfully
+    restart: unless-stopped
+    init: true
+    read_only: true
+    tmpfs: [/tmp]
+    user: "65534:65534"
+    security_opt:
+      - no-new-privileges:true
+    cap_drop: [ALL]
+    mem_limit: ${CMO_MGMT_MEMORY:-128m}
+    networks:
+      cmo: {}
+
+  cmo-tenants-init:
+    image: python:3.12-slim
+    command: ["sh", "-c", "chown 65534:65534 /var/lib/cmo"]
+    volumes:
+      - cmo-tenants:/var/lib/cmo
+    restart: "no"
+
+volumes:
+  cmo-tenants:
+    name: ${CMO_NAME_PREFIX:-cache-me-outside}-tenants

docs/ci/cmo-tenants-test.patch

@@ -0,0 +1,12 @@
+--- a/.github/workflows/cmo-ci.yml
++++ b/.github/workflows/cmo-ci.yml
+@@ -28,6 +28,9 @@
+       - name: Build server
+         run: make -j"$(nproc)" SERVER_CFLAGS='-Werror'
+
++      - name: Proxy reload + management tests
++        run: python3 tools/proxy/test_cmo_proxy.py && cd tools/mgmt && python3 test_cmo_mgmt.py && python3 test_cmo_browse.py
++
+       - name: Run a fast Valkey test subset
+         run: |
+           # introspection is omitted: its forkless-bgsave case is timing-sensitive

tools/proxy/cmo_proxy.py

@@ -13,14 +13,14 @@
   {"tenants": {"acme": {"listen": "0.0.0.0:7001", "upstream": "127.0.0.1:6379",
                         "max_conns": 50, "ops_per_sec": 1000,
                         "bytes_per_sec": 5000000}}}
-Any limit may be 0 or omitted to mean unlimited. Optional spending caps:
-  "budget": {"url": "https://billing.example/api"} or {"file": "caps.json"}
-(see cmo_budget.py). Standard library only.
+Any limit may be 0 or omitted to mean unlimited. Standard library only.
 
   python3 cmo_proxy.py config.json
+  python3 cmo_proxy.py --watch config.json   # reload when the file changes
 """
 import asyncio
 import json
+import os
 import sys
 import time
 
@@ -186,21 +186,59 @@ async def serve(tenants):
     return servers
 
 
+class Reloader:
+    """Keeps listeners in step with the config file.
+
+    New or changed tenants get a fresh listener; removed tenants stop
+    listening. Connections already open keep running with the limits they
+    started with until the client disconnects.
+    """
+    def __init__(self, path, budget=None):
+        self.path, self.budget = path, budget
+        self.cfgs, self.servers, self.mtime = {}, {}, object()
+
+    async def check(self):
+        try:
+            m = os.stat(self.path).st_mtime_ns
+        except FileNotFoundError:
+            m = None
+        if m == self.mtime:
+            return False
+        self.mtime = m
+        cfg = {}
+        if m is not None:
+            with open(self.path) as f:
+                cfg = json.load(f).get("tenants", {})
+        for name in list(self.servers):
+            if cfg.get(name) != self.cfgs.get(name):
+                self.servers.pop(name).close()
+                self.cfgs.pop(name, None)
+        for name, c in cfg.items():
+            if name not in self.servers:
+                t = Tenant(name, c, budget=self.budget)
+                self.servers[name] = await asyncio.start_server(
+                    lambda r, w, t=t: handle(t, r, w), *t.listen)
+                self.cfgs[name] = c
+        return True
+
+    async def run(self, interval=2.0):
+        while True:
+            try:
+                await self.check()
+            except (ValueError, OSError, KeyError) as e:
+                print("config reload failed, keeping current listeners: %s" % e, file=sys.stderr)
+            await asyncio.sleep(interval)
+
+
 def main(argv):
-    if len(argv) != 2:
+    args = [a for a in argv[1:] if a != "--watch"]
+    if len(args) != 1:
         print(__doc__)
         return 1
-    with open(argv[1]) as f:
-        cfg = json.load(f)
-    budget = None
-    b = cfg.get("budget")
-    if b:
-        import cmo_budget
-        src = (cmo_budget.HttpCapSource(b["url"]) if "url" in b
-               else cmo_budget.FileCapSource(b["file"]))
-        budget = cmo_budget.BudgetGuard(src, ttl=b.get("ttl", 30),
-                                        fail_open=b.get("fail_open", True))
-    tenants = load(argv[1], budget=budget)
+    if "--watch" in argv:
+        asyncio.run(Reloader(args[0]).run())
+        return 0
+    tenants = load(args[0])
 
     async def run():
         servers = await serve(tenants)

tools/proxy/test_cmo_proxy.py

@@ -107,5 +107,28 @@ async def test_budget_blocks(self):
         w.close()
 
 
+class ReloadTest(unittest.IsolatedAsyncioTestCase):
+    async def test_add_change_remove(self):
+        import json, socket, tempfile
+        d = tempfile.mkdtemp(); path = os.path.join(d, "c.json")
+        r = p.Reloader(path)
+        self.assertTrue(await r.check())           # missing file: no listeners
+        self.assertEqual(r.servers, {})
+        s = socket.socket(); s.bind(("127.0.0.1", 0)); port = s.getsockname()[1]; s.close()
+        cfg = {"tenants": {"a": {"listen": "127.0.0.1:%d" % port, "upstream": "127.0.0.1:1"}}}
+        json.dump(cfg, open(path, "w"))
+        await r.check()
+        self.assertIn("a", r.servers)
+        first = r.servers["a"]
+        self.assertFalse(await r.check())          # unchanged
+        cfg["tenants"]["a"]["ops_per_sec"] = 5
+        json.dump(cfg, open(path, "w")); os.utime(path, ns=(1, 1))
+        await r.check()
+        self.assertIsNot(r.servers["a"], first)     # changed: new listener
+        json.dump({"tenants": {}}, open(path, "w")); os.utime(path, ns=(2, 2))
+        await r.check()
+        self.assertEqual(r.servers, {})
+
+
 if __name__ == "__main__":
     unittest.main()