Enhance Redis connection handling with retries; improve logging in encoding process

video_encoder/config.py

@@ -2,12 +2,13 @@ import os
 from typing import List, Dict
 
 import os
-from redis import from_url
+from redis import from_url, RedisError
 from dotenv import load_dotenv
+import time
+import logging
 
 load_dotenv()
 
-import logging
 class EncodingConfig:
     DEBUG = os.getenv("DEBUG", "False") == "True"
     TEMP_DIR = "./tmp"
@@ -24,11 +25,18 @@ class RedisConfig:
     @classmethod
     def get_connection(cls):
         if not cls.CONN_POOL:
-            cls.CONN_POOL = from_url(cls.URL, max_connections=20)
-            logger.info("Successfully connected to Redis")
+            while True:
+                try:
+                    cls.CONN_POOL = from_url(cls.URL, max_connections=20, ssl_cert_reqs=None)
+                    logger.info("Successfully connected to Redis")
+                    break
+                except RedisError as e:
+                    logger.error(f"Could not connect to Redis instance: {e}. Retrying in 2 seconds...")
+                    time.sleep(2)
         return cls.CONN_POOL
+
     HMAC_SECRET = os.getenv("VIDEO_HMAC_SECRET", "default-secret-change-me")
-    
+
     RESOLUTIONS: List[Dict] = [
         {
             "name": "1080p",
@@ -41,7 +49,7 @@ class RedisConfig:
             "preset": "slow"
         },
         {
-            "name": "720p", 
+            "name": "720p",
             "width": 1280,
             "height": 720,
             "video_bitrate": "2500k",
@@ -65,13 +73,13 @@ class RedisConfig:
             "width": 426,
             "height": 240,
             "video_bitrate": "400k",
-            "audio_bitrate": "64k", 
+            "audio_bitrate": "64k",
             "codec": "libx264",
             "profile": "baseline",
             "preset": "veryfast"
         }
     ]
-    
+
     @classmethod
     def validate_resolutions(cls):
         for res in cls.RESOLUTIONS:

video_encoder/worker/ffmpeg.py

@@ -39,18 +39,25 @@ class FFmpegEncoder:
             commands.append(cmd)
         return commands
 
-    def encode(self) -> str:
-        """Execute encoding commands and return master playlist path"""
-        master_playlist = os.path.join(self.output_dir, f"{self.base_name}_master.m3u8")
-        
-        with open(master_playlist, "w") as f:
-            f.write("#EXTM3U\n")
-            for res in reversed(EncodingConfig.RESOLUTIONS):
-                f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={res['video_bitrate'].replace('k', '000')},"
-                        f"RESOLUTION={res['width']}x{res['height']}\n")
-                f.write(f"{self.base_name}_{res['name']}.m3u8\n")
-        
-        for cmd in self.generate_commands():
-            subprocess.run(cmd, check=True, capture_output=True)
-            
-        return master_playlist
+import logging
+logger = logging.getLogger(__name__)
+
+def encode(self) -> str:
+    """Execute encoding commands and return master playlist path"""
+    master_playlist = os.path.join(self.output_dir, f"{self.base_name}_master.m3u8")
+
+    with open(master_playlist, "w") as f:
+        f.write("#EXTM3U\n")
+        for res in reversed(EncodingConfig.RESOLUTIONS):
+            f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={res['video_bitrate'].replace('k', '000')},"
+                    f"RESOLUTION={res['width']}x{res['height']}\n")
+            f.write(f"{self.base_name}_{res['name']}.m3u8\n")
+
+    for cmd in self.generate_commands():
+        result = subprocess.run(cmd, capture_output=True)
+        logger.debug(f"FFmpeg command: {' '.join(cmd)}")
+        logger.debug(f"FFmpeg output: {result.stdout.decode()}")
+        logger.debug(f"FFmpeg error: {result.stderr.decode()}")
+        result.check_returncode()
+
+    return master_playlist

video_encoder/worker/tasks.py

@@ -9,6 +9,7 @@ from typing import Optional
 from .ffmpeg import FFmpegEncoder
 from ..config import EncodingConfig
 from datetime import datetime
+import time
 
 logger = logging.getLogger(__name__)
 
@@ -26,21 +27,27 @@ def update_job_progress(job_id: str, progress: float):
     RedisConfig.get_connection().hset(f"job:{job_id}", "progress", progress)
 
 def encode_video_task(job_id: str, input_path: str) -> Optional[str]:
-    try:
-        logger.info(f"Starting encoding for job {job_id}")
-        output_dir = os.path.join(EncodingConfig.TEMP_DIR, job_id)
-        encoder = FFmpegEncoder(input_path, output_dir)
+    max_retries = 3
+    for attempt in range(max_retries):
+        try:
+            logger.info(f"Starting encoding for job {job_id} (attempt {attempt + 1})")
+            output_dir = os.path.join(EncodingConfig.TEMP_DIR, job_id)
+            encoder = FFmpegEncoder(input_path, output_dir)
 
-        update_job_progress(job_id, 0.1)
-        master_playlist = encoder.encode()
-        update_job_progress(job_id, 1.0)
+            update_job_progress(job_id, 0.1)
+            master_playlist = encoder.encode()
+            update_job_progress(job_id, 1.0)
 
-        logger.info(f"Encoding completed for job {job_id}")
-        return master_playlist
-    except Exception as e:
-        logger.error(f"Encoding failed for job {job_id}: {str(e)}")
-        update_job_progress(job_id, -1.0)
-        raise
-    finally:
-        # Cleanup temporary files after 1 hour
-        shutil.rmtree(output_dir, ignore_errors=True)
+            logger.info(f"Encoding completed for job {job_id}")
+            return master_playlist
+        except Exception as e:
+            logger.error(f"Encoding failed for job {job_id} (attempt {attempt + 1}): {str(e)}")
+            update_job_progress(job_id, -1.0)
+            if attempt < max_retries - 1:
+                logger.info(f"Retrying encoding for job {job_id} in 5 seconds...")
+                time.sleep(5)
+            else:
+                raise
+        finally:
+            # Cleanup temporary files after 1 hour
+            shutil.rmtree(output_dir, ignore_errors=True)