.

analyze_analytics.py

@@ -0,0 +1,241 @@
+#!/usr/bin/env python3
+"""
+Analytics Analysis Tool for Neural OS Multi-GPU System
+
+This script analyzes the structured analytics logs to generate reports and insights.
+Usage: python analyze_analytics.py [--since HOURS] [--type TYPE]
+"""
+
+import json
+import argparse
+import glob
+import time
+from collections import defaultdict, Counter
+from datetime import datetime, timedelta
+import statistics
+
+class AnalyticsAnalyzer:
+    def __init__(self, since_hours=24):
+        self.since_timestamp = time.time() - (since_hours * 3600)
+        self.data = {
+            'gpu_metrics': [],
+            'connection_events': [],
+            'queue_metrics': [],
+            'ip_stats': []
+        }
+        self.load_data()
+    
+    def load_data(self):
+        """Load all analytics data files"""
+        file_types = {
+            'gpu_metrics': 'gpu_metrics_*.jsonl',
+            'connection_events': 'connection_events_*.jsonl', 
+            'queue_metrics': 'queue_metrics_*.jsonl',
+            'ip_stats': 'ip_stats_*.jsonl'
+        }
+        
+        for data_type, pattern in file_types.items():
+            files = glob.glob(pattern)
+            for file_path in files:
+                try:
+                    with open(file_path, 'r') as f:
+                        for line in f:
+                            try:
+                                record = json.loads(line.strip())
+                                if record.get('type') != 'metadata' and record.get('timestamp', 0) >= self.since_timestamp:
+                                    self.data[data_type].append(record)
+                            except json.JSONDecodeError:
+                                continue
+                except FileNotFoundError:
+                    continue
+        
+        print(f"Loaded data from the last {(time.time() - self.since_timestamp) / 3600:.1f} hours:")
+        for data_type, records in self.data.items():
+            print(f"  {data_type}: {len(records)} records")
+        print()
+    
+    def analyze_gpu_utilization(self):
+        """Analyze GPU utilization patterns"""
+        print("🖥️  GPU UTILIZATION ANALYSIS")
+        print("=" * 40)
+        
+        gpu_records = [r for r in self.data['gpu_metrics'] if r.get('type') == 'gpu_status']
+        if not gpu_records:
+            print("No GPU utilization data found.")
+            return
+        
+        utilizations = [r['utilization_percent'] for r in gpu_records]
+        total_gpus = gpu_records[-1].get('total_gpus', 0)
+        
+        print(f"Total GPUs: {total_gpus}")
+        print(f"Average utilization: {statistics.mean(utilizations):.1f}%")
+        print(f"Peak utilization: {max(utilizations):.1f}%")
+        print(f"Minimum utilization: {min(utilizations):.1f}%")
+        print(f"Utilization std dev: {statistics.stdev(utilizations) if len(utilizations) > 1 else 0:.1f}%")
+        
+        # Utilization distribution
+        high_util = sum(1 for u in utilizations if u >= 80)
+        med_util = sum(1 for u in utilizations if 40 <= u < 80)
+        low_util = sum(1 for u in utilizations if u < 40)
+        
+        print(f"\nUtilization distribution:")
+        print(f"  High (≥80%): {high_util} samples ({high_util/len(utilizations)*100:.1f}%)")
+        print(f"  Medium (40-79%): {med_util} samples ({med_util/len(utilizations)*100:.1f}%)")
+        print(f"  Low (<40%): {low_util} samples ({low_util/len(utilizations)*100:.1f}%)")
+        print()
+    
+    def analyze_connections(self):
+        """Analyze connection patterns"""
+        print("🔗 CONNECTION ANALYSIS")
+        print("=" * 40)
+        
+        opens = [r for r in self.data['connection_events'] if r.get('type') == 'connection_open']
+        closes = [r for r in self.data['connection_events'] if r.get('type') == 'connection_close']
+        
+        if not opens and not closes:
+            print("No connection data found.")
+            return
+        
+        print(f"Total connections opened: {len(opens)}")
+        print(f"Total connections closed: {len(closes)}")
+        
+        if closes:
+            durations = [r['duration'] for r in closes]
+            interactions = [r['interactions'] for r in closes]
+            reasons = [r['reason'] for r in closes]
+            
+            print(f"\nSession durations:")
+            print(f"  Average: {statistics.mean(durations):.1f}s")
+            print(f"  Median: {statistics.median(durations):.1f}s")
+            print(f"  Max: {max(durations):.1f}s")
+            print(f"  Min: {min(durations):.1f}s")
+            
+            print(f"\nInteractions per session:")
+            print(f"  Average: {statistics.mean(interactions):.1f}")
+            print(f"  Median: {statistics.median(interactions):.1f}")
+            print(f"  Max: {max(interactions)}")
+            
+            print(f"\nSession end reasons:")
+            reason_counts = Counter(reasons)
+            for reason, count in reason_counts.most_common():
+                print(f"  {reason}: {count} ({count/len(closes)*100:.1f}%)")
+        print()
+    
+    def analyze_queue_performance(self):
+        """Analyze queue performance"""
+        print("📝 QUEUE PERFORMANCE ANALYSIS")
+        print("=" * 40)
+        
+        bypasses = [r for r in self.data['queue_metrics'] if r.get('type') == 'queue_bypass']
+        waits = [r for r in self.data['queue_metrics'] if r.get('type') == 'queue_wait']
+        statuses = [r for r in self.data['queue_metrics'] if r.get('type') == 'queue_status']
+        
+        total_users = len(bypasses) + len(waits)
+        if total_users == 0:
+            print("No queue data found.")
+            return
+        
+        print(f"Total users processed: {total_users}")
+        print(f"Users bypassed queue: {len(bypasses)} ({len(bypasses)/total_users*100:.1f}%)")
+        print(f"Users waited in queue: {len(waits)} ({len(waits)/total_users*100:.1f}%)")
+        
+        if waits:
+            wait_times = [r['wait_time'] for r in waits]
+            positions = [r['queue_position'] for r in waits]
+            
+            print(f"\nWait time statistics:")
+            print(f"  Average wait: {statistics.mean(wait_times):.1f}s")
+            print(f"  Median wait: {statistics.median(wait_times):.1f}s")
+            print(f"  Max wait: {max(wait_times):.1f}s")
+            print(f"  Average queue position: {statistics.mean(positions):.1f}")
+        
+        if statuses:
+            queue_sizes = [r['queue_size'] for r in statuses]
+            estimated_waits = [r['estimated_wait'] for r in statuses if r['queue_size'] > 0]
+            
+            print(f"\nQueue size statistics:")
+            print(f"  Average queue size: {statistics.mean(queue_sizes):.1f}")
+            print(f"  Max queue size: {max(queue_sizes)}")
+            
+            if estimated_waits:
+                print(f"  Average estimated wait: {statistics.mean(estimated_waits):.1f}s")
+        print()
+    
+    def analyze_ip_usage(self):
+        """Analyze IP address usage patterns"""
+        print("🌍 IP USAGE ANALYSIS") 
+        print("=" * 40)
+        
+        ip_records = self.data['ip_stats']
+        if not ip_records:
+            print("No IP usage data found.")
+            return
+        
+        # Get latest connection counts per IP
+        latest_ip_data = {}
+        for record in ip_records:
+            if record.get('type') == 'ip_update':
+                ip = record['ip_address']
+                latest_ip_data[ip] = record['connection_count']
+        
+        if not latest_ip_data:
+            print("No IP connection data found.")
+            return
+        
+        total_connections = sum(latest_ip_data.values())
+        unique_ips = len(latest_ip_data)
+        
+        print(f"Total unique IP addresses: {unique_ips}")
+        print(f"Total connections: {total_connections}")
+        print(f"Average connections per IP: {total_connections/unique_ips:.1f}")
+        
+        print(f"\nTop IP addresses by connection count:")
+        sorted_ips = sorted(latest_ip_data.items(), key=lambda x: x[1], reverse=True)
+        for i, (ip, count) in enumerate(sorted_ips[:10], 1):
+            percentage = count / total_connections * 100
+            print(f"  {i:2d}. {ip}: {count} connections ({percentage:.1f}%)")
+        print()
+    
+    def generate_summary_report(self):
+        """Generate a comprehensive summary report"""
+        print("📊 SYSTEM SUMMARY REPORT")
+        print("=" * 50)
+        
+        # Time range
+        start_time = datetime.fromtimestamp(self.since_timestamp)
+        end_time = datetime.now()
+        duration_hours = (end_time.timestamp() - self.since_timestamp) / 3600
+        
+        print(f"Report period: {start_time.strftime('%Y-%m-%d %H:%M:%S')} to {end_time.strftime('%Y-%m-%d %H:%M:%S')}")
+        print(f"Duration: {duration_hours:.1f} hours")
+        print()
+        
+        self.analyze_gpu_utilization()
+        self.analyze_connections()
+        self.analyze_queue_performance()
+        self.analyze_ip_usage()
+
+def main():
+    parser = argparse.ArgumentParser(description='Analyze Neural OS analytics data')
+    parser.add_argument('--since', type=float, default=24, 
+                       help='Analyze data from the last N hours (default: 24)')
+    parser.add_argument('--type', choices=['gpu', 'connections', 'queue', 'ip', 'summary'],
+                       default='summary', help='Type of analysis to perform')
+    
+    args = parser.parse_args()
+    
+    analyzer = AnalyticsAnalyzer(since_hours=args.since)
+    
+    if args.type == 'gpu':
+        analyzer.analyze_gpu_utilization()
+    elif args.type == 'connections':
+        analyzer.analyze_connections()
+    elif args.type == 'queue':
+        analyzer.analyze_queue_performance()
+    elif args.type == 'ip':
+        analyzer.analyze_ip_usage()
+    else:
+        analyzer.generate_summary_report()
+
+if __name__ == '__main__':
+    main() 

dispatcher.py

@@ -32,28 +32,75 @@ class SystemAnalytics:
         self.users_waited_in_queue = 0  # Users who had to wait
         self.gpu_utilization_samples = deque(maxlen=100)  # GPU utilization over time
         self.queue_size_samples = deque(maxlen=100)  # Queue size over time
+        
+        # File handles for different analytics
         self.log_file = None
-        self._init_log_file()
+        self.gpu_metrics_file = None
+        self.connection_events_file = None
+        self.queue_metrics_file = None
+        self.ip_stats_file = None
+        self._init_log_files()
     
-    def _init_log_file(self):
-        """Initialize the system log file"""
+    def _init_log_files(self):
+        """Initialize all analytics log files"""
         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-        log_filename = f"system_analytics_{timestamp}.log"
-        self.log_file = log_filename
+        
+        # Main human-readable log
+        self.log_file = f"system_analytics_{timestamp}.log"
         self._write_log("="*80)
         self._write_log("NEURAL OS MULTI-GPU SYSTEM ANALYTICS")
         self._write_log("="*80)
         self._write_log(f"System started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
         self._write_log("")
+        
+        # Structured data files for analysis
+        self.gpu_metrics_file = f"gpu_metrics_{timestamp}.jsonl"
+        self.connection_events_file = f"connection_events_{timestamp}.jsonl"
+        self.queue_metrics_file = f"queue_metrics_{timestamp}.jsonl"
+        self.ip_stats_file = f"ip_stats_{timestamp}.jsonl"
+        
+        # Initialize with headers/metadata
+        self._write_json_log(self.gpu_metrics_file, {
+            "type": "metadata",
+            "timestamp": time.time(),
+            "description": "GPU utilization metrics",
+            "fields": ["timestamp", "total_gpus", "active_gpus", "available_gpus", "utilization_percent"]
+        })
+        
+        self._write_json_log(self.connection_events_file, {
+            "type": "metadata", 
+            "timestamp": time.time(),
+            "description": "Connection lifecycle events",
+            "fields": ["timestamp", "event_type", "client_id", "ip_address", "duration", "interactions", "reason"]
+        })
+        
+        self._write_json_log(self.queue_metrics_file, {
+            "type": "metadata",
+            "timestamp": time.time(), 
+            "description": "Queue performance metrics",
+            "fields": ["timestamp", "queue_size", "estimated_wait", "bypass_rate", "avg_wait_time"]
+        })
+        
+        self._write_json_log(self.ip_stats_file, {
+            "type": "metadata",
+            "timestamp": time.time(),
+            "description": "IP address usage statistics", 
+            "fields": ["timestamp", "ip_address", "connection_count", "total_unique_ips"]
+        })
     
     def _write_log(self, message):
-        """Write message to log file and console"""
+        """Write message to main log file and console"""
         timestamp = datetime.now().strftime("%H:%M:%S")
         log_message = f"[{timestamp}] {message}"
         print(log_message)
         with open(self.log_file, "a") as f:
             f.write(log_message + "\n")
     
+    def _write_json_log(self, filename, data):
+        """Write structured data to JSON lines file"""
+        with open(filename, "a") as f:
+            f.write(json.dumps(data) + "\n")
+    
     def log_new_connection(self, client_id: str, ip: str):
         """Log new connection"""
         self.total_connections += 1
@@ -61,8 +108,30 @@ class SystemAnalytics:
         self.ip_addresses[ip] += 1
         
         unique_ips = len(self.ip_addresses)
+        timestamp = time.time()
+        
+        # Human-readable log
         self._write_log(f"🔗 NEW CONNECTION: {client_id} from {ip}")
         self._write_log(f"   📊 Total connections: {self.total_connections} | Active: {self.active_connections} | Unique IPs: {unique_ips}")
+        
+        # Structured data logs
+        self._write_json_log(self.connection_events_file, {
+            "type": "connection_open",
+            "timestamp": timestamp,
+            "client_id": client_id,
+            "ip_address": ip,
+            "total_connections": self.total_connections,
+            "active_connections": self.active_connections,
+            "unique_ips": unique_ips
+        })
+        
+        self._write_json_log(self.ip_stats_file, {
+            "type": "ip_update",
+            "timestamp": timestamp,
+            "ip_address": ip,
+            "connection_count": self.ip_addresses[ip],
+            "total_unique_ips": unique_ips
+        })
     
     def log_connection_closed(self, client_id: str, duration: float, interactions: int, reason: str = "normal"):
         """Log connection closed"""
@@ -71,17 +140,44 @@ class SystemAnalytics:
         self.session_durations.append(duration)
         
         avg_duration = sum(self.session_durations) / len(self.session_durations) if self.session_durations else 0
+        timestamp = time.time()
         
+        # Human-readable log
         self._write_log(f"🚪 CONNECTION CLOSED: {client_id}")
         self._write_log(f"   ⏱️  Duration: {duration:.1f}s | Interactions: {interactions} | Reason: {reason}")
         self._write_log(f"   📊 Active connections: {self.active_connections} | Avg session duration: {avg_duration:.1f}s")
+        
+        # Structured data log
+        self._write_json_log(self.connection_events_file, {
+            "type": "connection_close",
+            "timestamp": timestamp,
+            "client_id": client_id,
+            "duration": duration,
+            "interactions": interactions,
+            "reason": reason,
+            "active_connections": self.active_connections,
+            "avg_session_duration": avg_duration
+        })
     
     def log_queue_bypass(self, client_id: str):
         """Log when user bypasses queue (gets GPU immediately)"""
         self.users_bypassed_queue += 1
         bypass_rate = (self.users_bypassed_queue / self.total_connections) * 100 if self.total_connections > 0 else 0
+        timestamp = time.time()
+        
+        # Human-readable log
         self._write_log(f"⚡ QUEUE BYPASS: {client_id} got GPU immediately")
         self._write_log(f"   📊 Bypass rate: {bypass_rate:.1f}% ({self.users_bypassed_queue}/{self.total_connections})")
+        
+        # Structured data log
+        self._write_json_log(self.queue_metrics_file, {
+            "type": "queue_bypass",
+            "timestamp": timestamp,
+            "client_id": client_id,
+            "bypass_rate": bypass_rate,
+            "users_bypassed": self.users_bypassed_queue,
+            "total_connections": self.total_connections
+        })
     
     def log_queue_wait(self, client_id: str, wait_time: float, queue_position: int):
         """Log when user had to wait in queue"""
@@ -90,9 +186,23 @@ class SystemAnalytics:
         
         avg_wait = sum(self.waiting_times) / len(self.waiting_times) if self.waiting_times else 0
         wait_rate = (self.users_waited_in_queue / self.total_connections) * 100 if self.total_connections > 0 else 0
+        timestamp = time.time()
         
+        # Human-readable log
         self._write_log(f"⏳ QUEUE WAIT: {client_id} waited {wait_time:.1f}s (was #{queue_position})")
         self._write_log(f"   📊 Wait rate: {wait_rate:.1f}% | Avg wait time: {avg_wait:.1f}s")
+        
+        # Structured data log
+        self._write_json_log(self.queue_metrics_file, {
+            "type": "queue_wait",
+            "timestamp": timestamp,
+            "client_id": client_id,
+            "wait_time": wait_time,
+            "queue_position": queue_position,
+            "wait_rate": wait_rate,
+            "avg_wait_time": avg_wait,
+            "users_waited": self.users_waited_in_queue
+        })
     
     def log_gpu_status(self, total_gpus: int, active_gpus: int, available_gpus: int):
         """Log GPU utilization"""
@@ -100,9 +210,22 @@ class SystemAnalytics:
         self.gpu_utilization_samples.append(utilization)
         
         avg_utilization = sum(self.gpu_utilization_samples) / len(self.gpu_utilization_samples) if self.gpu_utilization_samples else 0
+        timestamp = time.time()
         
+        # Human-readable log
         self._write_log(f"🖥️  GPU STATUS: {active_gpus}/{total_gpus} in use ({utilization:.1f}% utilization)")
         self._write_log(f"   📊 Available: {available_gpus} | Avg utilization: {avg_utilization:.1f}%")
+        
+        # Structured data log
+        self._write_json_log(self.gpu_metrics_file, {
+            "type": "gpu_status",
+            "timestamp": timestamp,
+            "total_gpus": total_gpus,
+            "active_gpus": active_gpus,
+            "available_gpus": available_gpus,
+            "utilization_percent": utilization,
+            "avg_utilization_percent": avg_utilization
+        })
     
     def log_worker_registered(self, worker_id: str, gpu_id: int, endpoint: str):
         """Log when a worker registers"""
@@ -122,7 +245,18 @@ class SystemAnalytics:
         self.queue_size_samples.append(queue_size)
         
         avg_queue_size = sum(self.queue_size_samples) / len(self.queue_size_samples) if self.queue_size_samples else 0
+        timestamp = time.time()
+        
+        # Always log to structured data for analysis
+        self._write_json_log(self.queue_metrics_file, {
+            "type": "queue_status",
+            "timestamp": timestamp,
+            "queue_size": queue_size,
+            "estimated_wait": estimated_wait,
+            "avg_queue_size": avg_queue_size
+        })
         
+        # Only log to human-readable if there's a queue
         if queue_size > 0:
             self._write_log(f"📝 QUEUE STATUS: {queue_size} users waiting | Est. wait: {estimated_wait:.1f}s")
             self._write_log(f"   📊 Avg queue size: {avg_queue_size:.1f}")

start_system.sh

@@ -131,14 +131,26 @@ for ((i=0; i<NUM_GPUS; i++)); do
 done
 echo ""
 echo "📋 Log files:"
-echo "   System analytics: system_analytics_*.log (real-time monitoring)"
-echo "   Dispatcher: dispatcher.log"
-echo "   Workers summary: workers.log"
+echo "   📊 Analytics (human-readable): system_analytics_*.log"
+echo "   🖥️  GPU metrics (JSON): gpu_metrics_*.jsonl"
+echo "   🔗 Connection events (JSON): connection_events_*.jsonl"
+echo "   📝 Queue metrics (JSON): queue_metrics_*.jsonl" 
+echo "   🌍 IP statistics (JSON): ip_stats_*.jsonl"
+echo "   🎯 Dispatcher: dispatcher.log"
+echo "   🔧 Workers summary: workers.log"
 for ((i=0; i<NUM_GPUS; i++)); do
-    echo "   GPU $i worker: worker_gpu_$i.log"
+    echo "   🖥️  GPU $i worker: worker_gpu_$i.log"
 done
 echo ""
-echo "💡 Monitor system in real-time: tail -f system_analytics_*.log"
+echo "💡 Real-time monitoring:"
+echo "   Human-readable: tail -f system_analytics_*.log"
+echo "   GPU utilization: tail -f gpu_metrics_*.jsonl"
+echo "   Connection events: tail -f connection_events_*.jsonl"
+echo ""
+echo "📈 Data analysis:"
+echo "   Summary report: python analyze_analytics.py"
+echo "   Last 6 hours: python analyze_analytics.py --since 6"
+echo "   GPU analysis only: python analyze_analytics.py --type gpu"
 echo "Press Ctrl+C to stop the system"
 echo "================================"
 

tail_workers.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+"""
+Script to tail all worker log files simultaneously.
+Usage: python tail_workers.py [--num-gpus N]
+"""
+
+import argparse
+import os
+import time
+import sys
+from typing import Dict
+
+def tail_all_workers(num_gpus: int):
+    """Tail all worker log files simultaneously"""
+    print(f"Tailing logs for {num_gpus} GPU workers...")
+    print("=" * 60)
+    
+    # Keep track of file positions
+    log_positions: Dict[int, int] = {}
+    for i in range(num_gpus):
+        log_positions[i] = 0
+    
+    try:
+        while True:
+            has_new_output = False
+            
+            for i in range(num_gpus):
+                log_file = f"worker_gpu_{i}.log"
+                
+                try:
+                    if os.path.exists(log_file):
+                        with open(log_file, 'r') as f:
+                            f.seek(log_positions[i])
+                            new_lines = f.readlines()
+                            
+                            if new_lines:
+                                has_new_output = True
+                                for line in new_lines:
+                                    timestamp = time.strftime("%H:%M:%S")
+                                    print(f"[{timestamp}] [GPU {i}] {line.rstrip()}")
+                                
+                            log_positions[i] = f.tell()
+                    else:
+                        # File doesn't exist yet, check if we should show a message
+                        if log_positions[i] == 0:
+                            print(f"[INFO] Waiting for {log_file} to be created...")
+                            log_positions[i] = -1  # Mark as checked
+                            
+                except Exception as e:
+                    print(f"[ERROR] Error reading {log_file}: {e}")
+            
+            # Only sleep if there was no new output to keep it responsive
+            if not has_new_output:
+                time.sleep(0.1)
+                
+    except KeyboardInterrupt:
+        print("\nStopping log monitoring...")
+
+def main():
+    parser = argparse.ArgumentParser(description="Tail all worker log files")
+    parser.add_argument("--num-gpus", type=int, default=2, 
+                       help="Number of GPU workers to monitor (default: 2)")
+    
+    args = parser.parse_args()
+    
+    if args.num_gpus < 1:
+        print("Error: Number of GPUs must be at least 1")
+        sys.exit(1)
+    
+    tail_all_workers(args.num_gpus)
+
+if __name__ == "__main__":
+    main()