Create health_ping.py

health_ping.py

@@ -0,0 +1,76 @@
+import threading
+import time
+import requests
+from datetime import datetime
+import os
+from config import config
+
+class HealthPinger:
+    def __init__(self):
+        self.running = False
+        self.thread = None
+        self.last_ping = None
+        self.ping_count = 0
+        self.space_url = config.SPACE_URL or os.getenv('SPACE_HOST', '')
+        self.interval = config.PING_INTERVAL
+        
+    def start(self):
+        """Start the health ping service"""
+        if self.running:
+            return
+        
+        self.running = True
+        self.thread = threading.Thread(target=self._ping_loop, daemon=True)
+        self.thread.start()
+        print(f"Health pinger started - Interval: {self.interval}s")
+    
+    def stop(self):
+        """Stop the health ping service"""
+        self.running = False
+        if self.thread:
+            self.thread.join(timeout=5)
+    
+    def _ping_loop(self):
+        """Main ping loop"""
+        while self.running:
+            try:
+                self._do_ping()
+            except Exception as e:
+                print(f"Ping error: {e}")
+            
+            time.sleep(self.interval)
+    
+    def _do_ping(self):
+        """Perform a health check ping"""
+        urls_to_ping = []
+        
+        # Add Space URL if available
+        if self.space_url:
+            if not self.space_url.startswith('http'):
+                self.space_url = f"https://{self.space_url}"
+            urls_to_ping.append(self.space_url)
+        
+        # Also ping localhost to keep internal services warm
+        urls_to_ping.append(f"http://localhost:{config.WEB_PORT}/api/health")
+        
+        for url in urls_to_ping:
+            try:
+                response = requests.get(url, timeout=30)
+                if response.status_code == 200:
+                    self.last_ping = datetime.now()
+                    self.ping_count += 1
+            except requests.RequestException:
+                pass
+    
+    def get_status(self):
+        """Get pinger status"""
+        return {
+            'running': self.running,
+            'last_ping': self.last_ping.isoformat() if self.last_ping else None,
+            'ping_count': self.ping_count,
+            'interval': self.interval,
+            'target_url': self.space_url
+        }
+
+# Global instance
+health_pinger = HealthPinger()