logging-and-monitoring-for-agentic-workflows

# 📊 Logging and Monitoring for Agentic Workflows ## 🔴 AI FIRST Quality Principle > **Apply the AI FIRST principle: never accept first-pass quality. Minimum 2 iterations. Read all output, improve every section. No shortcuts.** ## 📋 Overview This skill provides comprehensive patterns for implementing observability in GitHub Agentic Workflows. It covers structured logging architectures, metrics collection, alerting strategies, debugging techniques, and production monitoring best practices for autonomous agent systems. ## 🎯 Core Concepts ### Observability Architecture ```mermaid graph TB subgraph "Agent Execution" A[Agent Start] --> B[Task Execution] B --> C[MCP Operations] C --> D[Agent Completion] end subgraph "Logging Layer" B --> E[Structured Logs] C --> E D --> E E --> F[Log Aggregation] end subgraph "Metrics Layer" B --> G[Performance Metrics] C --> G D --> G G --> H[Time Series DB] end subgraph "Alerting Layer" F --> I[Alert Rules] H --> I I --> J[Notifications] end subgraph "Visualization" F --> K[Log Explorer] H --> L[Dashboards] K --> M[Insights] L --> M end style E fill:#00d9ff style G fill:#ff006e style I fill:#ffbe0b ``` ### Three Pillars of Observability 1. **Logs**: Detailed event records with context 2. **Metrics**: Quantitative measurements over time 3. **Traces**: Request flow through system components ## 📝 Structured Logging ### 1. Logging Architecture #### JSON Structured Logging Format ```javascript // scripts/agents/lib/logger.js import winston from 'winston'; import { v4 as uuidv4 } from 'uuid'; /** * Structured logger for agentic workflows * Outputs JSON logs with consistent schema */ class AgentLogger { constructor(options = {}) { this.agentId = options.agentId || process.env.AGENT_ID || uuidv4(); this.sessionId = options.sessionId || uuidv4(); this.environment = process.env.NODE_ENV || 'development'; this.logger = winston.createLogger({ level: options.level || process.env.LOG_LEVEL || 'info', format: winston.format.combine( winston.format.timestamp({ format: 'ISO' }), winston.format.errors({ stack: true }), winston.format.json() ), defaultMeta: { agent_id: this.agentId, session_id: this.sessionId, environment: this.environment, service: 'agentic-workflow', version: process.env.APP_VERSION || '1.0.0' }, transports: [ // Console output (GitHub Actions) new winston.transports.Console({ format: winston.format.combine( winston.format.colorize(), winston.format.printf(this.formatConsoleOutput.bind(this)) ) }), // File output (local development) new winston.transports.File({ filename: 'logs/agent-errors.log', level: 'error', maxsize: 10485760, // 10MB maxFiles: 5, tailable: true }), new winston.transports.File({ filename: 'logs/agent-combined.log', maxsize: 10485760, maxFiles: 10, tailable: true }) ], exitOnError: false }); } /** * Format console output for human readability */ formatConsoleOutput(info) { const { timestamp, level, message, ...meta } = info; const metaStr = Object.keys(meta).length ? JSON.stringify(meta, null, 2) : ''; return `${timestamp} [${level}] ${message}${metaStr ? '\n' + metaStr : ''}`; } /** * Log agent lifecycle events */ logAgentStart(taskType, context = {}) { this.logger.info('Agent execution started', { event_type: 'agent_start', task_type: taskType, ...context }); } logAgentComplete(taskType, result, metrics = {}) { this.logger.info('Agent execution completed', { event_type: 'agent_complete', task_type: taskType, status: result.status, duration_ms: metrics.duration, tokens_used: metrics.tokens, ...result }); } logAgentError(taskType, error, context = {}) { this.logger.error('Agent execution failed', { event_type: 'agent_error', task_type: taskType, error_message: error.message, error_stack: error.stack, error_code: error.code, ...context }); } /** * Log MCP operations */ logMCPCall(toolName, params, context = {}) { this.logger.debug('MCP tool call', { event_type: 'mcp_call', tool_name: toolName, params: this.sanitizeParams(params), ...context }); } logMCPResponse(toolName, response, duration) { this.logger.debug('MCP tool response', { event_type: 'mcp_response', tool_name: toolName, duration_ms: duration, response_size: JSON.stringify(response).length, status: 'success' }); } logMCPError(toolName, error, duration) { this.logger.error('MCP tool error', { event_type: 'mcp_error', tool_name: toolName, duration_ms: duration, error_message: error.message, error_code: error.code }); } /** * Log performance metrics */ logPerformance(operation, metrics) { this.logger.info('Performance metrics', { event_type: 'performance', operation, duration_ms: metrics.duration, memory_mb: metrics.memory, cpu_percent: metrics.cpu }); } /** * Log security events */ logSecurityEvent(eventType, details) { this.logger.warn('Security event', { event_type: 'security', security_event: eventType, ...details }); } /** * Sanitize sensitive data from logs */ sanitizeParams(params) { const sensitive = ['token', 'password', 'secret', 'key', 'api_key']; const sanitized = { ...params }; for (const key of Object.keys(sanitized)) { if (sensitive.some(s => key.toLowerCase().includes(s))) { sanitized[key] = '***REDACTED***'; } } return sanitized; } /** * Create child logger with additional context */ child(metadata) { const childLogger = Object.create(this); childLogger.logger = this.logger.child(metadata); return childLogger; } } export default AgentLogger; // Usage example const logger = new AgentLogger({ agentId: 'pr-analyzer', sessionId: process.env.GITHUB_RUN_ID }); logger.logAgentStart('pr-analysis', { pr_number: 123, repository: 'owner/repo' }); ``` #### Python Structured Logging ```python # scripts/agents/lib/logger.py import logging import json import sys import os from datetime import datetime from typing import Any, Dict, Optional import uuid class JSONFormatter(logging.Formatter): """ JSON formatter for structured logging """ def __init__(self): super().__init__() self.agent_id = os.getenv('AGENT_ID', str(uuid.uuid4())) self.session_id = os.getenv('GITHUB_RUN_ID', str(uuid.uuid4())) self.environment = os.getenv('ENVIRONMENT', 'development') def format(self, record: logging.LogRecord) -> str: """Format log record as JSON""" log_data = { 'timestamp': datetime.utcnow().isoformat() + 'Z', 'level': record.levelname, 'message': record.getMessage(), 'agent_id': self.agent_id, 'session_id': self.session_id, 'environment': self.environment, 'service': 'agentic-workflow', 'logger': record.name, 'module': record.module, 'function': record.funcName, 'line': record.lineno } # Add exception info if present if record.exc_info: log_data['exception'] = { 'type': record.exc_info[0].__name__, 'message': str(record.exc_info[1]), 'traceback': self.formatException(record.exc_info) } # Add extra fields if hasattr(record, 'extra_fields'): log_data.update(record.extra_fields) return json.dumps(log_data) class AgentLogger: """ Structured logger for agentic workflows """ def __init__(self, name: str, level: str = 'INFO'): self.logger = logging.getLogger(name) self.logger.setLevel(getattr(logging, level.upper())) # Console handler with JSON format console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(JSONFormatter()) self.logger.addHandler(console_handler) # File handler for errors if not os.path.exists('logs'): os.makedirs('logs') error_handler = logging.FileHandler('logs/agent-errors.log') error_handler.setLevel(logging.ERROR) error_handler.setFormatter(JSONFormatter()) self.logger.addHandler(error_handler) def log_agent_start(self, task_type: str, context: Dict[str, Any] = None): """Log agent start event""" extra = { 'extra_fields': { 'event_type': 'agent_start', 'task_type': task_type, **(context or {}) } } self.logger.info('Agent execution started', extra=extra) def log_agent_complete(self, task_type: str, result: Dict[str, Any], metrics: Dict[str, Any] = None): """Log agent completion event""" extra = { 'extra_fields': { 'event_type': 'agent_complete', 'task_type': task_type, 'status': result.get('status'), 'duration_ms': metrics.get('duration') if metrics else None, **result } } self.logger.info('Agent execution completed', extra=extra) def log_agent_error(self, task_type: str, error: Exception, context: Dict[str, Any] = None): """Log agent error event""" extra = { 'extra_fields': { 'event_type': 'agent_error', 'task_type': task_type, 'error_type': type(error).__name__, 'error_message': str(error), **(context or {}) } } self.logger.error('Agent execution failed', extra=extra, exc_info=error) def log_mcp_call(self, tool_name: str, params: Dict[str, Any], context: Dict[str, Any] = None): """Log MCP tool call""" extra = { 'extra_fields': { 'event_type': 'mcp_call', 'tool_name': tool_name, 'params': self._sanitize_params(params), **(context or {}) } } self.logger.debug('MCP tool call', extra=extra) def _sanitize_params(self, params: Dict[str, Any]) -> Dict[str, Any]: """Sanitize sensitive data""" sensitive = ['token', 'password', 'secret', 'key', 'api_key'] sanitized = params.copy() for key in sanitized.keys(): if any(s in key.lower() for s in sensitive): sanitized[key] = '***REDACTED***' return sanitized # Usage logger = AgentLogger('pr-analyzer', level='INFO') logger.log_agent_start('pr-analysis', {'pr_number': 123}) ``` ### 2. Log Levels and When to Use Them #### Log Level Guidelines