Improve Redis/Celery reliability to prevent task loss (#1853)

- Add visibility_timeout (12h) to prevent tasks from being redelivered
  while still running
- Enable task_acks_late and task_reject_on_worker_lost to ensure tasks
  are requeued if a worker crashes
- Set worker_prefetch_multiplier=1 to prevent task hoarding
- Add broker_heartbeat for faster connection problem detection
- Configure socket timeouts and retry_on_timeout for Redis connections
- Add exponential backoff retry (3 retries) for Redis client
- Add health_check_interval for periodic connection validation
- Add configurable task time limits (24h hard, 23h soft)
- Fix task_track_started from tuple (True,) to boolean True

New environment variables for configuration:
- CELERY_VISIBILITY_TIMEOUT, CELERY_BROKER_HEARTBEAT
- CELERY_TASK_TIME_LIMIT, CELERY_TASK_SOFT_TIME_LIMIT
- CELERY_BROKER_POOL_LIMIT, CELERY_REDIS_MAX_CONNECTIONS
- REDIS_SOCKET_TIMEOUT, REDIS_SOCKET_CONNECT_TIMEOUT
- REDIS_HEALTH_CHECK_INTERVAL

AI-assisted: Claude Code

Signed-off-by: Christian Berendt <[email protected]>

Author: berendt

osism/settings.py

@@ -21,6 +21,9 @@ def read_secret(secret_name):
 REDIS_HOST: str = os.getenv("REDIS_HOST", "redis")
 REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379"))
 REDIS_DB: int = int(os.getenv("REDIS_DB", "0"))
+REDIS_SOCKET_TIMEOUT: int = int(os.getenv("REDIS_SOCKET_TIMEOUT", "30"))
+REDIS_SOCKET_CONNECT_TIMEOUT: int = int(os.getenv("REDIS_SOCKET_CONNECT_TIMEOUT", "30"))
+REDIS_HEALTH_CHECK_INTERVAL: int = int(os.getenv("REDIS_HEALTH_CHECK_INTERVAL", "10"))
 
 
 NETBOX_URL = os.getenv("NETBOX_API", os.getenv("NETBOX_URL"))

osism/tasks/__init__.py

@@ -18,14 +18,63 @@
 
 
 class Config:
+    # Basic settings
     broker_connection_retry_on_startup = True
     enable_utc = True
     enable_ironic = os.environ.get("ENABLE_IRONIC", "True")
-    broker_url = "redis://redis"
-    result_backend = "redis://redis"
+
+    # Redis connection settings from environment
+    _redis_host = os.environ.get("REDIS_HOST", "redis")
+    _redis_port = os.environ.get("REDIS_PORT", "6379")
+    _redis_db = os.environ.get("REDIS_DB", "0")
+
+    broker_url = f"redis://{_redis_host}:{_redis_port}/{_redis_db}"
+    result_backend = f"redis://{_redis_host}:{_redis_port}/{_redis_db}"
+
+    # Connection pool settings
+    broker_pool_limit = int(os.environ.get("CELERY_BROKER_POOL_LIMIT", "10"))
+    redis_max_connections = int(os.environ.get("CELERY_REDIS_MAX_CONNECTIONS", "20"))
+
+    # Redis transport options for reliability
+    broker_transport_options = {
+        "visibility_timeout": int(
+            os.environ.get("CELERY_VISIBILITY_TIMEOUT", "43200")
+        ),  # 12 hours
+        "socket_timeout": 30,
+        "socket_connect_timeout": 30,
+        "retry_on_timeout": True,
+        "health_check_interval": 10,
+    }
+
+    # Result backend transport options
+    result_backend_transport_options = {
+        "socket_timeout": 30,
+        "socket_connect_timeout": 30,
+        "retry_on_timeout": True,
+    }
+
+    # Task acknowledgement settings - acknowledge after completion
+    task_acks_late = True
+    task_reject_on_worker_lost = True
+
+    # Worker settings - prevent task hoarding
+    worker_prefetch_multiplier = 1
+
+    # Heartbeat for connection detection
+    broker_heartbeat = int(os.environ.get("CELERY_BROKER_HEARTBEAT", "10"))
+
+    # Task time limits (can be overridden per-task)
+    task_time_limit = int(
+        os.environ.get("CELERY_TASK_TIME_LIMIT", "86400")
+    )  # 24h hard limit
+    task_soft_time_limit = int(
+        os.environ.get("CELERY_TASK_SOFT_TIME_LIMIT", "82800")
+    )  # 23h soft limit
+
+    # Queue settings
     task_create_missing_queues = True
     task_default_queue = "default"
-    task_track_started = (True,)
+    task_track_started = True
     task_routes = {
         "osism.tasks.ansible.*": {"queue": "osism-ansible"},
         "osism.tasks.ceph.*": {"queue": "ceph-ansible"},

osism/utils/__init__.py

@@ -13,6 +13,8 @@
 import pynetbox
 from pottery import Redlock
 from redis import Redis
+from redis.backoff import ExponentialBackoff
+from redis.retry import Retry
 import requests
 from requests.adapters import HTTPAdapter
 import urllib3
@@ -188,11 +190,19 @@ def get_netbox_connection(
     return nb
 
 
+# Redis retry configuration with exponential backoff
+_redis_retry = Retry(ExponentialBackoff(), retries=3)
+
 redis = Redis(
     host=settings.REDIS_HOST,
     port=settings.REDIS_PORT,
     db=settings.REDIS_DB,
     socket_keepalive=True,
+    socket_timeout=settings.REDIS_SOCKET_TIMEOUT,
+    socket_connect_timeout=settings.REDIS_SOCKET_CONNECT_TIMEOUT,
+    retry_on_timeout=True,
+    retry=_redis_retry,
+    health_check_interval=settings.REDIS_HEALTH_CHECK_INTERVAL,
 )
 redis.ping()