Merge pull request #62 from steamcmd/fix-redis-stability

Fix for Redis instability and changes health checks to be used for alternative health and readiness checks.

Author: jonakoudijs

Dockerfile

@@ -29,4 +29,16 @@ COPY --chown=$USER:$USER src/ $HOME/
 ##################### INSTALLATION END #####################
 
 # Set default container command
-CMD exec gunicorn web:app --max-requests 3000 --max-requests-jitter 150 --workers $WORKERS --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:$PORT
+CMD exec gunicorn web:app \
+    --max-requests 1000 \
+    --max-requests-jitter 100 \
+    --workers $WORKERS \
+    --worker-class uvicorn.workers.UvicornWorker \
+    --bind 0.0.0.0:$PORT \
+    --timeout 30 \
+    --keep-alive 2 \
+    --worker-connections 1000 \
+    --preload \
+    --access-logfile - \
+    --error-logfile - \
+    --log-level info

src/utils/redis.py

@@ -1,46 +1,90 @@
 import logging
 import config
 import redis
+from redis.connection import ConnectionPool
+import threading
+
+# Global connection pool - thread-safe singleton
+_connection_pool = None
+_pool_lock = threading.Lock()
+
+
+def get_connection_pool():
+    """
+    Get or create Redis connection pool (thread-safe singleton).
+    """
+    global _connection_pool
+
+    if _connection_pool is None:
+        with _pool_lock:
+            if _connection_pool is None:
+                try:
+                    # Connection pool configuration
+                    pool_kwargs = {
+                        "max_connections": 50,
+                        "retry_on_timeout": True,
+                        "socket_connect_timeout": 5,
+                        "socket_timeout": 5,
+                        "health_check_interval": 30,
+                    }
+
+                    # try connection string, or default to separate REDIS_* env vars
+                    if config.redis_url:
+                        _connection_pool = ConnectionPool.from_url(
+                            config.redis_url, db=config.redis_database, **pool_kwargs
+                        )
+                    elif config.redis_password:
+                        _connection_pool = ConnectionPool(
+                            host=config.redis_host,
+                            port=config.redis_port,
+                            password=config.redis_password,
+                            db=config.redis_database,
+                            **pool_kwargs,
+                        )
+                    else:
+                        _connection_pool = ConnectionPool(
+                            host=config.redis_host,
+                            port=config.redis_port,
+                            db=config.redis_database,
+                            **pool_kwargs,
+                        )
+
+                    logging.info("Redis connection pool created successfully")
+
+                except Exception as error:
+                    logging.error(f"Failed to create Redis connection pool: {error}")
+                    return None
+
+    return _connection_pool
 
 
 def connect():
     """
-    Parse redis config and connect.
+    Get Redis connection from pool.
     """
+    pool = get_connection_pool()
+    if pool is None:
+        return None
 
     try:
-        # try connection string, or default to separate REDIS_* env vars
-        if config.redis_url:
-            rds = redis.Redis.from_url(config.redis_url, db=config.redis_database)
-
-        elif config.redis_password:
-            rds = redis.Redis(
-                host=config.redis_host,
-                port=config.redis_port,
-                password=config.redis_password,
-                db=config.redis_database,
-            )
-        else:
-            rds = redis.Redis(
-                host=config.redis_host, port=config.redis_port, db=config.redis_database
-            )
-
+        return redis.Redis(connection_pool=pool)
     except Exception as error:
-        logging.error("Failed to connect to Redis with error: " + error)
-        return False
-
-    return rds
+        logging.error(f"Failed to get Redis connection from pool: {error}")
+        return None
 
 
 def read(key):
     """
     Read specified key from Redis.
     """
-
     rds = connect()
 
+    if rds is None:
+        logging.error(f"Failed to get Redis connection for key: {key}")
+        return False
+
     try:
-        # get info from cache
+        # get info from cache with timeout
         data = rds.get(key)
 
         # return False if not found
@@ -53,25 +97,40 @@ def read(key):
         # return data from Redis
         return data
 
+    except redis.ConnectionError as conn_error:
+        logging.error(
+            f"Redis connection error while reading key {key}: {conn_error}",
+            extra={"key": key, "error_type": "connection_error"},
+        )
+        return False
+    except redis.TimeoutError as timeout_error:
+        logging.error(
+            f"Redis timeout error while reading key {key}: {timeout_error}",
+            extra={"key": key, "error_type": "timeout_error"},
+        )
+        return False
     except Exception as redis_error:
-        # print query parse error and return empty dict
         logging.error(
-            "An error occured while trying to read and decode from Redis",
-            extra={"key": key, "error_msg": redis_error},
+            f"Unexpected error while reading from Redis key {key}: {redis_error}",
+            extra={
+                "key": key,
+                "error_type": "unexpected_error",
+                "error_msg": str(redis_error),
+            },
         )
-        logging.error(redis_error)
-
-    # return failed status
-    return False
+        return False
 
 
 def write(key, data):
     """
     Write specified key to Redis.
     """
-
     rds = connect()
 
+    if rds is None:
+        logging.error(f"Failed to get Redis connection for key: {key}")
+        return False
+
     # write data and set ttl
     try:
         expiration = int(config.cache_expiration)
@@ -82,43 +141,70 @@ def write(key, data):
         else:
             rds.set(key, data, ex=expiration)
 
-        # return succes status
+        # return success status
         return True
 
+    except redis.ConnectionError as conn_error:
+        logging.error(
+            f"Redis connection error while writing key {key}: {conn_error}",
+            extra={"key": key, "error_type": "connection_error"},
+        )
+        return False
+    except redis.TimeoutError as timeout_error:
+        logging.error(
+            f"Redis timeout error while writing key {key}: {timeout_error}",
+            extra={"key": key, "error_type": "timeout_error"},
+        )
+        return False
     except Exception as redis_error:
-        # print query parse error and return empty dict
         logging.error(
-            "An error occured while trying to write to Redis cache",
-            extra={"key": key, "error_msg": redis_error},
+            f"Unexpected error while writing to Redis key {key}: {redis_error}",
+            extra={
+                "key": key,
+                "error_type": "unexpected_error",
+                "error_msg": str(redis_error),
+            },
         )
-        logging.error(redis_error)
-
-    # return fail status
-    return False
+        return False
 
 
 def increment(key, amount=1):
     """
-    Increment value of amount to
-    specified key to Redis.
+    Increment value of amount to specified key in Redis.
     """
-
     rds = connect()
 
+    if rds is None:
+        logging.error(f"Failed to get Redis connection for key: {key}")
+        return False
+
     # increment data of key
     try:
         # increment and set new value
         rds.incrby(key, amount)
 
-        # return succes status
+        # return success status
         return True
 
+    except redis.ConnectionError as conn_error:
+        logging.error(
+            f"Redis connection error while incrementing key {key}: {conn_error}",
+            extra={"key": key, "error_type": "connection_error"},
+        )
+        return False
+    except redis.TimeoutError as timeout_error:
+        logging.error(
+            f"Redis timeout error while incrementing key {key}: {timeout_error}",
+            extra={"key": key, "error_type": "timeout_error"},
+        )
+        return False
     except Exception as redis_error:
-        # print query parse error and return empty dict
         logging.error(
-            "An error occured while trying to increment value in Redis cache",
-            extra={"key": key, "error_msg": redis_error},
+            f"Unexpected error while incrementing Redis key {key}: {redis_error}",
+            extra={
+                "key": key,
+                "error_type": "unexpected_error",
+                "error_msg": str(redis_error),
+            },
         )
-
-    # return fail status
-    return False
+        return False

src/web.py

@@ -42,9 +42,6 @@ def read_app(app_id: int, pretty: bool = False):
             logging.info(
                 "App info could not be found in cache", extra={"apps": str([app_id])}
             )
-            info = utils.steam.get_apps_info([app_id])
-            data = json.dumps(info)
-            utils.redis.write("app." + str(app_id), data)
         else:
             info = json.loads(info)
             logging.info(
@@ -95,3 +92,47 @@ def read_item(pretty: bool = False):
             "data": "Something went wrong while retrieving and parsing the current API version. Please try again later",
             "pretty": pretty,
         }
+
+
+@app.get("/health")
+def health_check():
+    """
+    Health check endpoint for Kubernetes liveness and readiness probes.
+    """
+    try:
+        # Test Redis connection if cache is enabled
+        if config.cache == "True":
+            rds = utils.redis.connect()
+            if rds is None:
+                return {"status": "unhealthy", "error": "Redis connection failed"}, 503
+
+            # Test Redis with a simple ping
+            rds.ping()
+
+        return {"status": "healthy", "timestamp": "2024-01-01T00:00:00Z"}
+
+    except Exception as e:
+        logging.error(f"Health check failed: {e}")
+        return {"status": "unhealthy", "error": str(e)}, 503
+
+
+@app.get("/ready")
+def readiness_check():
+    """
+    Readiness check endpoint for Kubernetes readiness probes.
+    """
+    try:
+        # Test Redis connection if cache is enabled
+        if config.cache == "True":
+            rds = utils.redis.connect()
+            if rds is None:
+                return {"status": "not_ready", "error": "Redis connection failed"}, 503
+
+            # Test Redis with a simple ping
+            rds.ping()
+
+        return {"status": "ready"}
+
+    except Exception as e:
+        logging.error(f"Readiness check failed: {e}")
+        return {"status": "not_ready", "error": str(e)}, 503