diff --git a/README.md b/README.md index 7d982e5b..e1a5d384 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,27 @@ To use PostgreSQL, set `db_uri` or the `GOSSEBIT_DB_URI` environment variable to postgres://user:password@host:5432/db_name ``` +### Multiple Workers + +By default, gooseBit runs with a single worker process. Running multiple workers (e.g. to utilize more +CPU cores) is supported if the following requirements are met: + +1. The in-memory object cache is disabled, since each worker would otherwise serve stale data + (set `cache: enabled: false` or `GOOSEBIT_CACHE__ENABLED=false`). +2. `secret_key` is set explicitly. Otherwise, each worker generates its own random key and user sessions + signed by one worker fail on the others. +3. PostgreSQL is recommended as the database. SQLite serializes writers via file locks, which is fine for + testing but not for concurrent load. +4. All workers share the same artifacts storage. Workers on one host share the `artifacts_dir` volume; + deployments spanning multiple hosts should use the S3 storage backend. + +With the Docker image, the worker count can then be set at runtime, for example: + +```txt +docker run -e GOOSEBIT_CACHE__ENABLED=false -e GOOSEBIT_SECRET_KEY= \ + -e GUNICORN_CMD_ARGS="--workers 4 --enable-stdio-inheritance" upstreamdata/goosebit +``` + ### Artifact Storage The software packages managed by gooseBit are either stored on the local filesystem (`artifacts_dir` setting) or an S3-compatible object storage. diff --git a/conftest.py b/conftest.py index 5065f8c5..55a5c214 100644 --- a/conftest.py +++ b/conftest.py @@ -5,7 +5,6 @@ from typing import Any, AsyncGenerator, Dict import pytest_asyncio -from aiocache import caches from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from tortoise import Tortoise @@ -17,7 +16,9 @@ from goosebit import app # noqa: E402 from goosebit.auth.permissions import GOOSEBIT_PERMISSIONS # noqa: E402 +from goosebit.cache import cache # noqa: E402 from goosebit.db.models import UpdateModeEnum, UpdateStateEnum # noqa: E402 +from goosebit.device_manager import DeviceManager # noqa: E402 from goosebit.settings import PWD_CXT # type: ignore[attr-defined] # noqa: E402 # Configure logging @@ -35,7 +36,9 @@ @pytest_asyncio.fixture(scope="function", autouse=True) async def clear_cache() -> AsyncGenerator[None, None]: - await caches.get("default").clear() + await cache.clear() + # class-level cache holding a row from the previous test's database + DeviceManager._hardware_default = None yield diff --git a/docker/Dockerfile b/docker/Dockerfile index a11ba6cd..171a091a 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -18,7 +18,9 @@ EXPOSE 60053 USER goosebit -# We currently do not fully support multiple workers. For more information, see: +# Multiple workers require the in-memory cache to be disabled (GOOSEBIT_CACHE__ENABLED=false) +# and an explicit GOOSEBIT_SECRET_KEY. Override GUNICORN_CMD_ARGS at runtime to raise the +# worker count. For more information, see: # https://github.com/UpstreamDataInc/goosebit/issues/125 ENV GUNICORN_CMD_ARGS="--workers 1 --enable-stdio-inheritance" diff --git a/docker/dev.dockerfile b/docker/dev.dockerfile index e3f46fa7..276ebae4 100644 --- a/docker/dev.dockerfile +++ b/docker/dev.dockerfile @@ -20,8 +20,9 @@ EXPOSE 60053 USER goosebit -# We currently do not fully support multiple workers. For more information, see: -# https://github.com/UpstreamDataInc/goosebit/issues/125 +# Multiple workers require the in-memory cache to be disabled (GOOSEBIT_CACHE__ENABLED=false) +# and an explicit GOOSEBIT_SECRET_KEY. Override GUNICORN_CMD_ARGS at runtime to raise the +# worker count. ENV GUNICORN_CMD_ARGS="--workers 1 --enable-stdio-inheritance" SHELL ["/bin/sh", "-c"] diff --git a/goosebit.yaml b/goosebit.yaml index 60e7feb0..5b3bd69c 100644 --- a/goosebit.yaml +++ b/goosebit.yaml @@ -29,6 +29,11 @@ poll_time: 00:01:00 # Limit the number of updates that run at the same time. Can be used to avoid overloading the update server. #max_concurrent_updates: 1000 +# In-memory cache for device/user objects. Must be disabled when running more than one worker +# process, otherwise workers serve stale data. See https://github.com/UpstreamDataInc/goosebit/issues/125 +#cache: +# enabled: true + # Whether to track the IP of the device when it polls. Useful for debugging, but can be turned off for privacy. #track_device_ip: true diff --git a/goosebit/cache.py b/goosebit/cache.py new file mode 100644 index 00000000..3010c2e8 --- /dev/null +++ b/goosebit/cache.py @@ -0,0 +1,45 @@ +from typing import Any + +from aiocache import SimpleMemoryCache +from aiocache.serializers import PickleSerializer + +from goosebit.settings import config + +CACHE_TTL = 600 + + +class Cache: + """Object cache honoring the `cache.enabled` setting. + + With caching disabled every lookup is a miss and writes are dropped, leaving the + database as the only source of truth. This is required when running multiple + workers, see https://github.com/UpstreamDataInc/goosebit/issues/125. + """ + + def __init__(self, enabled: bool): + self.enabled = enabled + self._backend = SimpleMemoryCache(serializer=PickleSerializer()) + + async def get(self, key: str) -> Any: + if not self.enabled: + return None + return await self._backend.get(key) + + async def set(self, key: str, value: Any) -> None: + if not self.enabled: + return + await self._backend.set(key, value, ttl=CACHE_TTL) + + async def delete(self, key: str) -> None: + if not self.enabled: + return + # missing keys are fine, the entry may have expired or was never cached by this worker + await self._backend.delete(key) + + async def clear(self) -> None: + if not self.enabled: + return + await self._backend.clear() + + +cache = Cache(enabled=config.cache.enabled) diff --git a/goosebit/device_manager.py b/goosebit/device_manager.py index e42729dc..1c62a11e 100644 --- a/goosebit/device_manager.py +++ b/goosebit/device_manager.py @@ -5,9 +5,9 @@ from enum import StrEnum from typing import Any, Awaitable, Callable, Optional -from aiocache import caches from fastapi.requests import Request +from goosebit.cache import cache from goosebit.db.models import ( Device, Hardware, @@ -18,16 +18,6 @@ ) from goosebit.schema.updates import UpdateChunk -caches.set_config( - { - "default": { - "cache": "aiocache.SimpleMemoryCache", - "serializer": {"class": "aiocache.serializers.PickleSerializer"}, - "ttl": 600, - }, - } -) - class HandlingType(StrEnum): SKIP = "skip" @@ -43,7 +33,6 @@ class DeviceManager: @staticmethod async def get_device(dev_id: str) -> Device: - cache = caches.get("default") device = await cache.get(dev_id) if device: return device # type: ignore[no-any-return] @@ -54,8 +43,7 @@ async def get_device(dev_id: str) -> Device: DeviceManager._hardware_default = hardware device = (await Device.get_or_create(id=dev_id, defaults={"hardware": hardware}))[0] - result = await cache.set(device.id, device, ttl=600) - assert result, "device being cached" + await cache.set(device.id, device) return device # type: ignore[no-any-return] @@ -64,8 +52,7 @@ async def save_device(device: Device, update_fields: list[str]) -> None: await device.save(update_fields=update_fields) # only update cache after a successful database save - result = await caches.get("default").set(device.id, device, ttl=600) - assert result, "device being cached" + await cache.set(device.id, device) @staticmethod async def update_auth_token(device: Device, auth_token: str) -> None: @@ -254,8 +241,7 @@ async def update_log(device: Device, log_data: str) -> None: async def delete_devices(ids: list[str]) -> None: await Device.filter(id__in=ids).delete() for dev_id in ids: - result = await caches.get("default").delete(dev_id) - assert result == 1, "device has been cached" + await cache.delete(dev_id) async def get_device(dev_id: str) -> Device: diff --git a/goosebit/settings/schema.py b/goosebit/settings/schema.py index 0b02e136..ba54d35a 100644 --- a/goosebit/settings/schema.py +++ b/goosebit/settings/schema.py @@ -1,5 +1,6 @@ import os from enum import StrEnum +from logging import getLogger from pathlib import Path from typing import Any @@ -15,6 +16,14 @@ from .const import CURRENT_DIR, GOOSEBIT_ROOT_DIR, LOGGING_DEFAULT +def _generate_secret_key() -> OctKey: + getLogger(__name__).warning( + "No secret_key configured, generated a random one. User sessions will not survive restarts " + "and logins will fail intermittently when running multiple workers. Set GOOSEBIT_SECRET_KEY." + ) + return OctKey.generate_key() + + class DeviceAuthMode(StrEnum): SETUP = "setup" # setup mode, any devices polling with an auth token that don't have one will save it STRICT = "strict" # all devices must have keys, and all keys must be set up with the API @@ -70,6 +79,12 @@ class StorageSettings(BaseModel): s3: S3StorageSettings | None = None +class CacheSettings(BaseModel): + # in-memory cache for device/user objects; must be disabled when running multiple workers, + # see https://github.com/UpstreamDataInc/goosebit/issues/125 + enabled: bool = True + + class GooseBitSettings(BaseSettings): model_config = SettingsConfigDict(env_prefix="GOOSEBIT_", extra="ignore", env_nested_delimiter="__") @@ -84,7 +99,7 @@ class GooseBitSettings(BaseSettings): device_auth: DeviceAuthSettings = DeviceAuthSettings() - secret_key: OctKey = Field(default_factory=OctKey.generate_key) + secret_key: OctKey = Field(default_factory=_generate_secret_key) plugins: list[str] = Field(default_factory=list) @@ -95,6 +110,8 @@ class GooseBitSettings(BaseSettings): storage: StorageSettings = StorageSettings() + cache: CacheSettings = CacheSettings() + metrics: MetricsSettings = MetricsSettings() logging: dict[str, Any] = LOGGING_DEFAULT diff --git a/goosebit/updater/controller/v1/routes.py b/goosebit/updater/controller/v1/routes.py index 1a9db13e..8b703f64 100644 --- a/goosebit/updater/controller/v1/routes.py +++ b/goosebit/updater/controller/v1/routes.py @@ -9,8 +9,9 @@ Response, StreamingResponse, ) +from tortoise.expressions import F -from goosebit.db.models import Device, Software, UpdateStateEnum +from goosebit.db.models import Device, Rollout, Software, UpdateStateEnum from goosebit.device_manager import DeviceManager, HandlingType, get_device from goosebit.settings import config from goosebit.storage import storage @@ -159,8 +160,8 @@ async def deployment_feedback( rollout = await DeviceManager.get_rollout(device) if rollout: if rollout.software == reported_software: - rollout.success_count += 1 - await rollout.save() + # atomic DB-side increment + await Rollout.filter(id=rollout.id).update(success_count=F("success_count") + 1) else: # edge case where device update mode got changed while update was running logging.warning( @@ -180,8 +181,8 @@ async def deployment_feedback( rollout = await DeviceManager.get_rollout(device) if rollout: if rollout.software == reported_software: - rollout.failure_count += 1 - await rollout.save() + # atomic DB-side increment + await Rollout.filter(id=rollout.id).update(failure_count=F("failure_count") + 1) else: # edge case where device update mode got changed while update was running logging.warning( diff --git a/goosebit/users/__init__.py b/goosebit/users/__init__.py index 9e2a909f..86ac359a 100644 --- a/goosebit/users/__init__.py +++ b/goosebit/users/__init__.py @@ -1,6 +1,5 @@ -from aiocache import caches - from goosebit.api.telemetry.metrics import users_count +from goosebit.cache import cache from goosebit.db.models import User from goosebit.settings import PWD_CXT # type: ignore[attr-defined] @@ -19,8 +18,7 @@ async def save_user(user: User, update_fields: list[str]) -> None: await user.save(update_fields=update_fields) # only update cache after a successful database save - result = await caches.get("default").set(user.username, user, ttl=600) - assert result, "user being cached" + await cache.set(user.username, user) @staticmethod async def update_enabled(user: User, enabled: bool) -> None: @@ -43,15 +41,13 @@ async def setup_user(cls, username: str, hashed_pwd: str, permissions: list[str] @staticmethod async def get_user(username: str) -> User: - cache = caches.get("default") user = await cache.get(username) if user: return user # type: ignore[no-any-return] user = await User.get_or_none(username=username) if user is not None: - result = await cache.set(user.username, user, ttl=600) - assert result, "user being cached" + await cache.set(user.username, user) return user # type: ignore[no-any-return] @@ -59,5 +55,5 @@ async def get_user(username: str) -> User: async def delete_users(usernames: list[str]) -> None: await User.filter(username__in=usernames).delete() for username in usernames: - await caches.get("default").delete(username) + await cache.delete(username) users_count.set(await User.all().count()) diff --git a/tests/e2e/docker/goosebit/Dockerfile b/tests/e2e/docker/goosebit/Dockerfile index a7143582..d8f21591 100644 --- a/tests/e2e/docker/goosebit/Dockerfile +++ b/tests/e2e/docker/goosebit/Dockerfile @@ -58,7 +58,9 @@ EXPOSE 60053 ENV PYTHONPATH=/app ENV PYTHONUNBUFFERED=1 -# We currently do not fully support multiple workers. For more information, see: +# Multiple workers require the in-memory cache to be disabled (GOOSEBIT_CACHE__ENABLED=false) +# and an explicit GOOSEBIT_SECRET_KEY. Override GUNICORN_CMD_ARGS at runtime to raise the +# worker count. For more information, see: # https://github.com/UpstreamDataInc/goosebit/issues/125 ENV GUNICORN_CMD_ARGS="--workers 1 --enable-stdio-inheritance" diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py new file mode 100644 index 00000000..c9d22561 --- /dev/null +++ b/tests/unit/test_cache.py @@ -0,0 +1,83 @@ +from typing import Any + +import pytest + +from goosebit.cache import Cache, cache +from goosebit.device_manager import DeviceManager +from goosebit.settings.schema import GooseBitSettings +from goosebit.users import UserManager, create_initial_user + + +@pytest.mark.asyncio +async def test_disabled_cache_is_noop() -> None: + disabled = Cache(enabled=False) + + await disabled.set("key", "value") + assert await disabled.get("key") is None + + # missing keys must not raise + await disabled.delete("key") + await disabled.clear() + + +@pytest.mark.asyncio +async def test_enabled_cache_round_trip() -> None: + enabled = Cache(enabled=True) + + await enabled.set("key", "value") + assert await enabled.get("key") == "value" + + await enabled.delete("key") + assert await enabled.get("key") is None + + # deleting a key that was never cached must not raise + await enabled.delete("never-cached") + + +@pytest.mark.asyncio +async def test_device_manager_with_disabled_cache(db: None, monkeypatch: Any) -> None: + monkeypatch.setattr(cache, "enabled", False) + + device = await DeviceManager.get_device("cache-test-device") + assert await cache.get(device.id) is None + + await DeviceManager.update_name(device, "renamed") + device = await DeviceManager.get_device("cache-test-device") + assert device.name == "renamed" + + await DeviceManager.delete_devices([device.id]) + recreated = await DeviceManager.get_device("cache-test-device") + assert recreated.name is None + + +@pytest.mark.asyncio +async def test_delete_device_that_was_never_cached(db: None) -> None: + device = await DeviceManager.get_device("uncached-device") + # simulate the entry expiring or the delete landing on a worker that never cached the device + await cache.clear() + + await DeviceManager.delete_devices([device.id]) + + +@pytest.mark.asyncio +async def test_user_manager_with_disabled_cache(db: None, monkeypatch: Any) -> None: + monkeypatch.setattr(cache, "enabled", False) + + await create_initial_user(username="cache@goosebit.test", hashed_pwd="hash") + user = await UserManager.get_user("cache@goosebit.test") + assert user is not None + assert await cache.get(user.username) is None + + await UserManager.update_enabled(user, False) + user = await UserManager.get_user("cache@goosebit.test") + assert user.enabled is False + + await UserManager.delete_users([user.username]) + assert await UserManager.get_user("cache@goosebit.test") is None + + +def test_cache_enabled_setting(monkeypatch: Any) -> None: + assert GooseBitSettings().cache.enabled is True + + monkeypatch.setenv("GOOSEBIT_CACHE__ENABLED", "false") + assert GooseBitSettings().cache.enabled is False