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

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

// 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

# 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)


---

*Content truncated.*