integrations/atom_communication_memory_monitoring.py

"""
ATOM Communication Memory Production Monitoring System
Real-time monitoring, alerting, and performance tracking
"""

import asyncio
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta
import json
import logging
import time
from typing import Any, Dict, List, Optional

from integrations.atom_communication_ingestion_pipeline import ingestion_pipeline, memory_manager

logger = logging.getLogger(__name__)

@dataclass
class MonitoringMetric:
    """Monitoring metric data structure"""
    name: str
    value: float
    unit: str
    timestamp: datetime
    tags: Dict[str, str]
    threshold: Optional[float] = None

@dataclass
class Alert:
    """Alert data structure"""
    id: str
    severity: str  # info, warning, error, critical
    title: str
    message: str
    timestamp: datetime
    resolved: bool = False
    resolved_at: Optional[datetime] = None
    tags: Dict[str, str]

class AtomCommunicationMemoryMonitoring:
    """Production monitoring system for ATOM communication memory"""
    
    def __init__(self):
        self.metrics: List[MonitoringMetric] = []
        self.alerts: List[Alert] = []
        self.is_running = False
        self.monitoring_interval = 60  # seconds
        self.alert_thresholds = {
            'ingestion_rate': 0.1,  # messages per second
            'error_rate': 0.05,  # 5% error rate
            'memory_usage': 0.8,  # 80% memory usage
            'search_latency': 1.0,  # 1 second
            'database_size': 100_000_000_000  # 100GB
        }
    
    async def start_monitoring(self):
        """Start the monitoring system"""
        self.is_running = True
        logger.info("Starting ATOM communication memory monitoring")
        
        while self.is_running:
            try:
                await self.collect_metrics()
                await self.check_alerts()
                await asyncio.sleep(self.monitoring_interval)
            except Exception as e:
                logger.error(f"Error in monitoring loop: {str(e)}")
                await asyncio.sleep(60)  # Wait longer on error
    
    def stop_monitoring(self):
        """Stop the monitoring system"""
        self.is_running = False
        logger.info("Stopping ATOM communication memory monitoring")
    
    async def collect_metrics(self):
        """Collect monitoring metrics"""
        try:
            timestamp = datetime.now()
            
            # Get ingestion stats
            stats = ingestion_pipeline.get_ingestion_stats()
            
            # Database metrics
            db_metrics = await self._collect_database_metrics(timestamp)
            
            # Ingestion metrics
            ingestion_metrics = await self._collect_ingestion_metrics(stats, timestamp)
            
            # Performance metrics
            performance_metrics = await self._collect_performance_metrics(timestamp)
            
            # Add all metrics
            self.metrics.extend(db_metrics + ingestion_metrics + performance_metrics)
            
            # Keep only last 24 hours of metrics
            cutoff_time = timestamp - timedelta(hours=24)
            self.metrics = [m for m in self.metrics if m.timestamp > cutoff_time]
            
            logger.info(f"Collected {len(db_metrics + ingestion_metrics + performance_metrics)} metrics")
            
        except Exception as e:
            logger.error(f"Error collecting metrics: {str(e)}")
    
    async def _collect_database_metrics(self, timestamp: datetime) -> List[MonitoringMetric]:
        """Collect database-related metrics"""
        metrics = []
        
        try:
            if memory_manager.connections_table:
                # Get record count
                df = memory_manager.connections_table.to_pandas()
                record_count = len(df)
                
                metrics.append(MonitoringMetric(
                    name="database_record_count",
                    value=record_count,
                    unit="records",
                    timestamp=timestamp,
                    tags={"table": "atom_communications"},
                    threshold=self.alert_thresholds['database_size']
                ))
                
                # Get database size (estimated)
                estimated_size = record_count * 1024  # Estimate 1KB per record
                metrics.append(MonitoringMetric(
                    name="database_size",
                    value=estimated_size,
                    unit="bytes",
                    timestamp=timestamp,
                    tags={"table": "atom_communications"},
                    threshold=self.alert_thresholds['database_size']
                ))
                
                # App distribution
                app_dist = df["app_type"].value_counts().to_dict()
                for app, count in app_dist.items():
                    metrics.append(MonitoringMetric(
                        name=f"records_{app}",
                        value=count,
                        unit="records",
                        timestamp=timestamp,
                        tags={"app": app, "metric": "record_count"}
                    ))
        
        except Exception as e:
            logger.error(f"Error collecting database metrics: {str(e)}")
        
        return metrics
    
    async def _collect_ingestion_metrics(self, stats: Dict[str, Any], timestamp: datetime) -> List[MonitoringMetric]:
        """Collect ingestion-related metrics"""
        metrics = []
        
        try:
            # Total messages
            total_messages = stats.get('total_messages', 0)
            metrics.append(MonitoringMetric(
                name="total_messages_ingested",
                value=total_messages,
                unit="messages",
                timestamp=timestamp,
                tags={"metric": "total_ingestion"}
            ))
            
            # Active streams
            active_streams = len(stats.get('active_streams', []))
            metrics.append(MonitoringMetric(
                name="active_real_time_streams",
                value=active_streams,
                unit="streams",
                timestamp=timestamp,
                tags={"metric": "active_streams"}
            ))
            
            # Configured apps
            configured_apps = len(stats.get('configured_apps', []))
            metrics.append(MonitoringMetric(
                name="configured_apps",
                value=configured_apps,
                unit="apps",
                timestamp=timestamp,
                tags={"metric": "configured_apps"}
            ))
            
        except Exception as e:
            logger.error(f"Error collecting ingestion metrics: {str(e)}")
        
        return metrics
    
    async def _collect_performance_metrics(self, timestamp: datetime) -> List[MonitoringMetric]:
        """Collect performance-related metrics"""
        metrics = []
        
        try:
            # Ingestion rate (simplified)
            recent_metrics = [m for m in self.metrics 
                             if m.name == "total_messages_ingested" 
                             and (timestamp - m.timestamp).total_seconds() < 300]  # Last 5 minutes
            
            if len(recent_metrics) >= 2:
                recent_metrics.sort(key=lambda x: x.timestamp)
                latest_count = recent_metrics[-1].value
                earliest_count = recent_metrics[0].value
                time_diff = (recent_metrics[-1].timestamp - recent_metrics[0].timestamp).total_seconds()
                
                if time_diff > 0:
                    ingestion_rate = (latest_count - earliest_count) / time_diff
                    metrics.append(MonitoringMetric(
                        name="ingestion_rate",
                        value=ingestion_rate,
                        unit="messages/second",
                        timestamp=timestamp,
                        tags={"metric": "performance"},
                        threshold=self.alert_thresholds['ingestion_rate']
                    ))
            
            # Memory usage (simplified - would need actual monitoring)
            import psutil
            memory_percent = psutil.virtual_memory().percent / 100
            metrics.append(MonitoringMetric(
                name="memory_usage",
                value=memory_percent,
                unit="fraction",
                timestamp=timestamp,
                tags={"metric": "performance"},
                threshold=self.alert_thresholds['memory_usage']
            ))
            
        except Exception as e:
            logger.error(f"Error collecting performance metrics: {str(e)}")
        
        return metrics
    
    async def check_alerts(self):
        """Check thresholds and generate alerts"""
        try:
            timestamp = datetime.now()
            
            # Get latest metrics for each metric name
            latest_metrics = {}
            for metric in self.metrics:
                if metric.name not in latest_metrics or metric.timestamp > latest_metrics[metric.name].timestamp:
                    latest_metrics[metric.name] = metric
            
            # Check thresholds
            for metric_name, metric in latest_metrics.items():
                if metric.threshold and metric.value > metric.threshold:
                    await self._create_alert(
                        severity="warning",
                        title=f"Threshold exceeded for {metric_name}",
                        message=f"{metric_name}: {metric.value:.2f} {metric.unit} (threshold: {metric.threshold})",
                        timestamp=timestamp,
                        tags=metric.tags
                    )
            
            # Check for system health
            if not memory_manager.db:
                await self._create_alert(
                    severity="critical",
                    title="Database connection lost",
                    message="LanceDB database connection is not available",
                    timestamp=timestamp,
                    tags={"component": "database"}
                )
            
        except Exception as e:
            logger.error(f"Error checking alerts: {str(e)}")
    
    async def _create_alert(self, severity: str, title: str, message: str, 
                           timestamp: datetime, tags: Dict[str, str]):
        """Create a new alert"""
        alert_id = f"alert_{int(timestamp.timestamp())}_{len(self.alerts)}"
        
        # Check if similar alert already exists
        existing_alert = next((a for a in self.alerts if not a.resolved and a.title == title), None)
        
        if existing_alert:
            # Update existing alert
            existing_alert.timestamp = timestamp
            existing_alert.message = message
        else:
            # Create new alert
            alert = Alert(
                id=alert_id,
                severity=severity,
                title=title,
                message=message,
                timestamp=timestamp,
                tags=tags
            )
            
            self.alerts.append(alert)
            logger.warning(f"Alert created: {severity} - {title}")
    
    def get_metrics_summary(self, time_window: int = 3600) -> Dict[str, Any]:
        """Get summary of metrics for the last N seconds"""
        try:
            cutoff_time = datetime.now() - timedelta(seconds=time_window)
            recent_metrics = [m for m in self.metrics if m.timestamp > cutoff_time]
            
            # Group metrics by name
            metrics_by_name = {}
            for metric in recent_metrics:
                if metric.name not in metrics_by_name:
                    metrics_by_name[metric.name] = []
                metrics_by_name[metric.name].append(metric)
            
            # Calculate summaries
            summary = {
                "time_window": time_window,
                "metric_count": len(recent_metrics),
                "metrics": {}
            }
            
            for name, metric_list in metrics_by_name.items():
                values = [m.value for m in metric_list]
                summary["metrics"][name] = {
                    "latest": values[-1] if values else None,
                    "average": sum(values) / len(values) if values else None,
                    "min": min(values) if values else None,
                    "max": max(values) if values else None,
                    "count": len(values),
                    "unit": metric_list[0].unit if metric_list else None
                }
            
            return summary
            
        except Exception as e:
            logger.error(f"Error getting metrics summary: {str(e)}")
            return {"error": str(e)}
    
    def get_alerts_summary(self, include_resolved: bool = False) -> Dict[str, Any]:
        """Get summary of alerts"""
        try:
            alerts = self.alerts if include_resolved else [a for a in self.alerts if not a.resolved]
            
            # Count by severity
            severity_counts = {}
            for alert in alerts:
                severity_counts[alert.severity] = severity_counts.get(alert.severity, 0) + 1
            
            return {
                "total_alerts": len(alerts),
                "unresolved_alerts": len([a for a in alerts if not a.resolved]),
                "severity_distribution": severity_counts,
                "recent_alerts": [
                    {
                        "id": alert.id,
                        "severity": alert.severity,
                        "title": alert.title,
                        "message": alert.message,
                        "timestamp": alert.timestamp.isoformat(),
                        "resolved": alert.resolved
                    }
                    for alert in sorted(alerts, key=lambda x: x.timestamp, reverse=True)[:10]
                ]
            }
            
        except Exception as e:
            logger.error(f"Error getting alerts summary: {str(e)}")
            return {"error": str(e)}
    
    def get_health_status(self) -> Dict[str, Any]:
        """Get overall system health status"""
        try:
            # Check critical components
            health_checks = {
                "database": memory_manager.db is not None,
                "ingestion_pipeline": len(ingestion_pipeline.ingestion_configs) > 0,
                "monitoring": self.is_running
            }
            
            # Check recent errors
            recent_alerts = [a for a in self.alerts 
                             if not a.resolved 
                             and a.severity in ["error", "critical"]
                             and (datetime.now() - a.timestamp).total_seconds() < 3600]
            
            overall_status = "healthy"
            if not all(health_checks.values()):
                overall_status = "unhealthy"
            elif recent_alerts:
                overall_status = "degraded"
            
            return {
                "overall_status": overall_status,
                "timestamp": datetime.now().isoformat(),
                "health_checks": health_checks,
                "recent_critical_alerts": len(recent_alerts),
                "monitoring_active": self.is_running
            }
            
        except Exception as e:
            logger.error(f"Error getting health status: {str(e)}")
            return {"error": str(e), "overall_status": "unknown"}

# Create global monitoring instance
atom_memory_monitoring = AtomCommunicationMemoryMonitoring()

# Export for use
__all__ = [
    'AtomCommunicationMemoryMonitoring',
    'atom_memory_monitoring',
    'MonitoringMetric',
    'Alert'
]