Centralized Logging¶

# mcp_mesh/logging.py
import logging
import json
import sys
import os
from datetime import datetime
from pythonjsonlogger import jsonlogger
from opentelemetry import trace
import asyncio
from contextvars import ContextVar

# Context variables for request tracking
request_id_var: ContextVar[str] = ContextVar('request_id', default='')
user_id_var: ContextVar[str] = ContextVar('user_id', default='')

class MCPMeshLogFormatter(jsonlogger.JsonFormatter):
    """Custom JSON formatter for MCP Mesh logs"""

    def add_fields(self, log_record, record, message_dict):
        super().add_fields(log_record, record, message_dict)

        # Add timestamp
        log_record['timestamp'] = datetime.utcnow().isoformat() + 'Z'

        # Add log level
        log_record['level'] = record.levelname

        # Add service information
        log_record['service'] = {
            'name': os.getenv('SERVICE_NAME', 'unknown'),
            'version': os.getenv('SERVICE_VERSION', '1.0.0'),
            'environment': os.getenv('ENVIRONMENT', 'development'),
            'instance_id': os.getenv('HOSTNAME', 'unknown')
        }

        # Add Kubernetes metadata
        log_record['kubernetes'] = {
            'namespace': os.getenv('K8S_NAMESPACE', 'default'),
            'pod_name': os.getenv('HOSTNAME', 'unknown'),
            'node_name': os.getenv('NODE_NAME', 'unknown')
        }

        # Add trace context
        span = trace.get_current_span()
        if span and span.get_span_context().is_valid:
            ctx = span.get_span_context()
            log_record['trace'] = {
                'trace_id': format(ctx.trace_id, '032x'),
                'span_id': format(ctx.span_id, '016x'),
                'trace_flags': format(ctx.trace_flags, '02x')
            }

        # Add request context
        request_id = request_id_var.get()
        if request_id:
            log_record['request_id'] = request_id

        user_id = user_id_var.get()
        if user_id:
            log_record['user_id'] = user_id

        # Add source location
        log_record['source'] = {
            'file': record.pathname,
            'line': record.lineno,
            'function': record.funcName
        }

        # Move message to correct field
        if 'message' in log_record:
            log_record['msg'] = log_record.pop('message')

def setup_logging(service_name: str, log_level: str = "INFO"):
    """Set up structured logging for MCP Mesh service"""

    # Create logger
    logger = logging.getLogger('mcp_mesh')
    logger.setLevel(getattr(logging, log_level.upper()))

    # Remove existing handlers
    logger.handlers = []

    # Create console handler with JSON formatter
    handler = logging.StreamHandler(sys.stdout)
    formatter = MCPMeshLogFormatter(
        '%(timestamp)s %(level)s %(name)s %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    # Add exception hook for unhandled exceptions
    def handle_exception(exc_type, exc_value, exc_traceback):
        if issubclass(exc_type, KeyboardInterrupt):
            sys.__excepthook__(exc_type, exc_value, exc_traceback)
            return

        logger.critical(
            "Unhandled exception",
            exc_info=(exc_type, exc_value, exc_traceback),
            extra={
                'exception_type': exc_type.__name__,
                'exception_message': str(exc_value)
            }
        )

    sys.excepthook = handle_exception

    return logger

# Logging utilities
class LogContext:
    """Context manager for adding fields to logs"""

    def __init__(self, **kwargs):
        self.fields = kwargs
        self.tokens = []

    def __enter__(self):
        # Set context variables
        if 'request_id' in self.fields:
            self.tokens.append(request_id_var.set(self.fields['request_id']))
        if 'user_id' in self.fields:
            self.tokens.append(user_id_var.set(self.fields['user_id']))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Reset context variables
        for token in self.tokens:
            request_id_var.reset(token)

# Enhanced logger for agents
class AgentLogger:
    """Agent-specific logger with additional functionality"""

    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.logger = setup_logging(f"mcp-mesh-{agent_name}")
        self._metrics = {
            'requests_total': 0,
            'errors_total': 0,
            'warnings_total': 0
        }

    def _log(self, level: str, message: str, **kwargs):
        """Internal log method with agent context"""
        extra = {
            'agent': {
                'name': self.agent_name,
                'type': 'mcp-mesh'
            }
        }
        extra.update(kwargs)

        # Update metrics
        if level == 'ERROR':
            self._metrics['errors_total'] += 1
        elif level == 'WARNING':
            self._metrics['warnings_total'] += 1

        getattr(self.logger, level.lower())(message, extra=extra)

    def info(self, message: str, **kwargs):
        self._log('INFO', message, **kwargs)

    def warning(self, message: str, **kwargs):
        self._log('WARNING', message, **kwargs)

    def error(self, message: str, **kwargs):
        self._log('ERROR', message, **kwargs)

    def debug(self, message: str, **kwargs):
        self._log('DEBUG', message, **kwargs)

    def request(self, method: str, params: dict = None,
               duration_ms: float = None, status: str = "success"):
        """Log API request with standard fields"""
        self._metrics['requests_total'] += 1

        self.info(
            f"Request processed: {method}",
            request={
                'method': method,
                'params': params or {},
                'duration_ms': duration_ms,
                'status': status
            },
            metrics={
                'duration_ms': duration_ms
            }
        )

    def exception(self, message: str, exc: Exception, **kwargs):
        """Log exception with full context"""
        import traceback

        self.error(
            message,
            exception={
                'type': type(exc).__name__,
                'message': str(exc),
                'traceback': traceback.format_exc()
            },
            **kwargs
        )

    def audit(self, action: str, resource: str, result: str, **kwargs):
        """Log audit event"""
        self.info(
            f"Audit: {action} {resource}",
            audit={
                'action': action,
                'resource': resource,
                'result': result,
                'timestamp': datetime.utcnow().isoformat()
            },
            **kwargs
        )

# Example usage in an agent
class LoggedWeatherAgent:
    def __init__(self):
        self.logger = AgentLogger("weather-agent")
        self.cache = {}

    async def get_forecast(self, location: str, days: int = 7):
        """Get weather forecast with comprehensive logging"""
        start_time = asyncio.get_event_loop().time()

        # Create request context
        request_id = f"req-{start_time}"

        with LogContext(request_id=request_id):
            self.logger.info(
                f"Getting forecast for {location}",
                location=location,
                days=days,
                cache_keys=list(self.cache.keys())
            )

            try:
                # Check cache
                cache_key = f"{location}:{days}"
                if cache_key in self.cache:
                    self.logger.debug(
                        "Cache hit",
                        cache_key=cache_key,
                        cache_size=len(self.cache)
                    )
                    return self.cache[cache_key]

                # Call external API
                self.logger.info(
                    "Calling weather API",
                    api_endpoint="https://api.weather.com/forecast",
                    timeout=30
                )

                # Simulate API call
                await asyncio.sleep(0.1)
                result = {"temp": 72, "conditions": "sunny"}

                # Cache result
                self.cache[cache_key] = result

                # Calculate duration
                duration_ms = (asyncio.get_event_loop().time() - start_time) * 1000

                # Log successful request
                self.logger.request(
                    method="get_forecast",
                    params={"location": location, "days": days},
                    duration_ms=duration_ms,
                    status="success"
                )

                # Audit log
                self.logger.audit(
                    action="forecast_retrieved",
                    resource=f"location:{location}",
                    result="success",
                    user_id="api-user-123"
                )

                return result

            except Exception as e:
                duration_ms = (asyncio.get_event_loop().time() - start_time) * 1000

                self.logger.exception(
                    f"Failed to get forecast for {location}",
                    exc=e,
                    location=location,
                    duration_ms=duration_ms
                )

                self.logger.request(
                    method="get_forecast",
                    params={"location": location, "days": days},
                    duration_ms=duration_ms,
                    status="error"
                )

                raise

# log_analysis.py
from elasticsearch import Elasticsearch
from datetime import datetime, timedelta
import json

class LogAnalyzer:
    """Analyze logs from Elasticsearch"""

    def __init__(self, es_host: str = "localhost:9200",
                 username: str = "elastic",
                 password: str = "changeme"):
        self.es = Elasticsearch(
            [es_host],
            basic_auth=(username, password)
        )
        self.index_pattern = "mcp-mesh-logs-*"

    def search_by_trace_id(self, trace_id: str):
        """Find all logs for a specific trace"""
        query = {
            "query": {
                "term": {
                    "trace.trace_id": trace_id
                }
            },
            "sort": [
                {"timestamp": "asc"}
            ],
            "size": 1000
        }

        response = self.es.search(
            index=self.index_pattern,
            body=query
        )

        return [hit["_source"] for hit in response["hits"]["hits"]]

    def find_errors_by_agent(self, agent_name: str,
                            hours: int = 24):
        """Find recent errors for an agent"""
        query = {
            "query": {
                "bool": {
                    "must": [
                        {"term": {"agent.name": agent_name}},
                        {"term": {"level": "ERROR"}},
                        {"range": {
                            "timestamp": {
                                "gte": f"now-{hours}h"
                            }
                        }}
                    ]
                }
            },
            "aggs": {
                "error_types": {
                    "terms": {
                        "field": "exception.type",
                        "size": 10
                    }
                },
                "error_timeline": {
                    "date_histogram": {
                        "field": "timestamp",
                        "fixed_interval": "1h"
                    }
                }
            },
            "size": 100
        }

        response = self.es.search(
            index=self.index_pattern,
            body=query
        )

        return {
            "errors": [hit["_source"] for hit in response["hits"]["hits"]],
            "error_types": response["aggregations"]["error_types"]["buckets"],
            "timeline": response["aggregations"]["error_timeline"]["buckets"]
        }

    def analyze_request_patterns(self, hours: int = 1):
        """Analyze request patterns across all agents"""
        query = {
            "query": {
                "bool": {
                    "must": [
                        {"exists": {"field": "request.method"}},
                        {"range": {
                            "timestamp": {
                                "gte": f"now-{hours}h"
                            }
                        }}
                    ]
                }
            },
            "aggs": {
                "requests_by_agent": {
                    "terms": {
                        "field": "agent.name",
                        "size": 20
                    },
                    "aggs": {
                        "methods": {
                            "terms": {
                                "field": "request.method",
                                "size": 10
                            }
                        },
                        "avg_duration": {
                            "avg": {
                                "field": "request.duration_ms"
                            }
                        },
                        "error_rate": {
                            "terms": {
                                "field": "request.status",
                                "size": 2
                            }
                        }
                    }
                },
                "slow_requests": {
                    "top_hits": {
                        "sort": [
                            {"request.duration_ms": "desc"}
                        ],
                        "size": 10,
                        "_source": ["timestamp", "agent.name",
                                   "request.method", "request.duration_ms"]
                    }
                }
            },
            "size": 0
        }

        response = self.es.search(
            index=self.index_pattern,
            body=query
        )

        return response["aggregations"]

    def detect_anomalies(self, agent_name: str):
        """Detect anomalous patterns in logs"""
        # Use machine learning features if available
        query = {
            "query": {
                "bool": {
                    "must": [
                        {"term": {"agent.name": agent_name}},
                        {"range": {
                            "timestamp": {
                                "gte": "now-1h"
                            }
                        }}
                    ]
                }
            },
            "aggs": {
                "log_rate": {
                    "date_histogram": {
                        "field": "timestamp",
                        "fixed_interval": "1m"
                    },
                    "aggs": {
                        "level_breakdown": {
                            "terms": {
                                "field": "level"
                            }
                        }
                    }
                },
                "unique_errors": {
                    "cardinality": {
                        "field": "exception.message.keyword"
                    }
                },
                "rare_terms": {
                    "rare_terms": {
                        "field": "msg.keyword",
                        "max_doc_count": 2
                    }
                }
            }
        }

        response = self.es.search(
            index=self.index_pattern,
            body=query
        )

        # Analyze for anomalies
        anomalies = []

        # Check for sudden spike in errors
        for bucket in response["aggregations"]["log_rate"]["buckets"]:
            error_count = next(
                (b["doc_count"] for b in bucket["level_breakdown"]["buckets"]
                 if b["key"] == "ERROR"), 0
            )
            if error_count > 10:  # Threshold
                anomalies.append({
                    "type": "error_spike",
                    "timestamp": bucket["key_as_string"],
                    "count": error_count
                })

        # Check for new error types
        unique_errors = response["aggregations"]["unique_errors"]["value"]
        if unique_errors > 5:  # Threshold
            anomalies.append({
                "type": "high_error_variety",
                "unique_count": unique_errors
            })

        return anomalies

    def export_logs(self, query: dict, output_file: str):
        """Export logs matching query to file"""
        from elasticsearch.helpers import scan

        with open(output_file, 'w') as f:
            for doc in scan(
                self.es,
                index=self.index_pattern,
                query=query,
                size=1000
            ):
                f.write(json.dumps(doc["_source"]) + "\n")

# Usage example
analyzer = LogAnalyzer()

# Find logs for a trace
trace_logs = analyzer.search_by_trace_id("abc123def456")
for log in trace_logs:
    print(f"{log['timestamp']} [{log['level']}] {log['msg']}")

# Analyze errors
errors = analyzer.find_errors_by_agent("weather-agent")
print(f"Found {len(errors['errors'])} errors")
for error_type in errors['error_types']:
    print(f"  {error_type['key']}: {error_type['doc_count']}")

# Detect anomalies
anomalies = analyzer.detect_anomalies("payment-agent")
if anomalies:
    print("Anomalies detected:")
    for anomaly in anomalies:
        print(f"  {anomaly['type']}: {anomaly}")

# performance_logs.py
def analyze_performance_logs(es_client, agent_name: str):
    """Analyze performance from logs"""

    query = {
        "query": {
            "bool": {
                "must": [
                    {"term": {"agent.name": agent_name}},
                    {"exists": {"field": "metrics.duration_ms"}},
                    {"range": {"timestamp": {"gte": "now-1h"}}}
                ]
            }
        },
        "aggs": {
            "percentiles": {
                "percentiles": {
                    "field": "metrics.duration_ms",
                    "percents": [50, 95, 99]
                }
            },
            "outliers": {
                "terms": {
                    "field": "request.method",
                    "size": 10,
                    "order": {"max_duration": "desc"}
                },
                "aggs": {
                    "max_duration": {
                        "max": {"field": "metrics.duration_ms"}
                    },
                    "sample_logs": {
                        "top_hits": {
                            "size": 1,
                            "sort": [{"metrics.duration_ms": "desc"}]
                        }
                    }
                }
            }
        }
    }

    result = es_client.search(index="mcp-mesh-logs-*", body=query)

    print(f"Performance Analysis for {agent_name}:")
    print(f"P50: {result['aggregations']['percentiles']['values']['50.0']}ms")
    print(f"P95: {result['aggregations']['percentiles']['values']['95.0']}ms")
    print(f"P99: {result['aggregations']['percentiles']['values']['99.0']}ms")

    print("\nSlowest Operations:")
    for bucket in result['aggregations']['outliers']['buckets']:
        print(f"  {bucket['key']}: {bucket['max_duration']['value']}ms")

# Use sampling for high-frequency logs
if random.random() < 0.01:  # 1% sampling
    logger.debug("High frequency event", sampled=True)

# Aggregate before logging
if time.time() - self.last_log > 60:  # Log every minute
    logger.info(f"Processed {self.count} requests", count=self.count)
    self.count = 0
    self.last_log = time.time()