Logging Implementation

Logging and Observability Principles

⚠️ Prerequisite: All operations MUST be logged per Logging and Observability Mandate @.agents/rules/logging-and-observability-mandate.md. This guide provides implementation patterns only.

Logging Standards

Log Levels (Standard Priority)

Use consistent log levels across all services:

Logging Rules

1. Every request/operation must log:

// Start of operation
log.Info("creating task",
"correlationId", correlationID,
"userId", userID,
"title", task.Title,
)

// Success
log.Info("task created successfully",
"correlationId", correlationID,
"taskId", task.ID,
"duration", time.Since(start),
)

// Error
log.Error("failed to create task",
"correlationId", correlationID,
"error", err,
"userId", userID,
)

2. Always include context:

• correlationId: Trace requests across services (UUID)
• userId: Who triggered the action
• duration: Operation timing (milliseconds)
• error: Error details (if failed)

3. Structured logging only (no string formatting):

// ✅ Structured
log.Info("user login", "userId", userID, "ip", clientIP)

// ❌ String formatting
log.Info(fmt.Sprintf("User %s logged in from %s", userID, clientIP))

4. Security - Never log:

• Passwords or password hashes
• API keys or tokens
• Credit card numbers
• PII in production logs (email/phone only if necessary and sanitized)
• Full request/response bodies (unless DEBUG level in non-prod)

5. Performance - Never log in hot paths:

• Inside tight loops
• Per-item processing in batch operations (use summary instead)
• Synchronous logging in latency-critical paths

Best Practice: "Use logger middleware redaction (e.g., pino-redact, zap masking) rather than manual string manipulation."

Language-Specific Implementations

Go (using slog standard library)

import "log/slog"

// Configure logger
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
  Level: slog.LevelInfo, // Production default
}))

// Usage
logger.Info("operation started",
  "correlationId", correlationID,
  "userId", userID,
)

logger.Error("operation failed",
  "correlationId", correlationID,
  "error", err,
  "retryCount", retries,
)

TypeScript/Node.js (using pino)

import pino from 'pino';

const logger = pino({
  level: process.env.LOG_LEVEL || 'info',
});

logger.info({
  correlationId,
  userId,
  duration: Date.now() - startTime,
}, 'task created successfully');

logger.error({
  correlationId,
  error: err.message,
  stack: err.stack,
}, 'failed to create task');

Python (using structlog)

Setup — configure once at application startup:

import contextvars
import logging
import structlog

# Async-safe correlation ID propagation via contextvars (NOT thread-locals)
correlation_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
    "correlation_id", default=""
)

def add_correlation_id(
    logger: structlog.types.WrappedLogger,
    method_name: str,
    event_dict: structlog.types.EventDict,
) -> structlog.types.EventDict:
    """Injects the current context's correlation_id into every log entry."""
    if cid := correlation_id_var.get():
        event_dict["correlation_id"] = cid
    return event_dict

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,   # merge bind_contextvars() calls
        add_correlation_id,                         # inject from ContextVar
        structlog.stdlib.add_log_level,
        structlog.stdlib.add_logger_name,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        # Development: pretty console; Production: JSON
        structlog.dev.ConsoleRenderer()             # swap for JSONRenderer() in prod
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

ASGI Middleware — set correlation_id at the request boundary:

import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class CorrelationIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        cid = request.headers.get("X-Correlation-Id") or str(uuid.uuid4())
        token = correlation_id_var.set(cid)
        try:
            response = await call_next(request)
            response.headers["X-Correlation-Id"] = cid
            return response
        finally:
            correlation_id_var.reset(token)  # always restore context

# Register in app startup:
# app.add_middleware(CorrelationIdMiddleware)

Usage in handlers and services:

import structlog

logger = structlog.get_logger()

# correlation_id is automatically included via the ContextVar processor
logger.info("task_created",
    user_id=user_id,
    task_id=task.id,
    duration_ms=elapsed_ms,
)

logger.error("task_creation_failed",
    user_id=user_id,
    error=str(err),
)

Why contextvars not thread-locals? Thread-locals are unsafe in async code — coroutines can switch context between await points. contextvars.ContextVar is propagated correctly across await boundaries and asyncio.create_task() calls, making it the only correct approach for async Python logging.

Rust (using tracing + tracing-subscriber)

Basic tracing setup with subscriber configuration:

use tracing_subscriber::EnvFilter;

fn init_tracing() {
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("info"));

    tracing_subscriber::fmt()
        .with_env_filter(filter)
        .json() // Production: JSON format
        .with_target(true)
        .with_thread_ids(true)
        .init();
}

Environment-specific config (dev vs prod):

fn init_tracing(env: &str) {
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("info"));

    match env {
        "development" => {
            tracing_subscriber::fmt()
                .with_env_filter(EnvFilter::new("debug"))
                .pretty() // Pretty-printed for dev
                .init();
        }
        _ => {
            tracing_subscriber::fmt()
                .with_env_filter(filter)
                .json() // JSON for staging/prod
                .with_target(true)
                .init();
        }
    }
}

Structured logging with operation context (using #[instrument]):

use tracing::{info, error, instrument};
use uuid::Uuid;

#[instrument(
    skip(service),
    fields(correlation_id = %Uuid::new_v4(), user_id = %user_id)
)]
pub async fn create_task(
    service: &TaskService,
    user_id: &str,
    request: CreateTaskRequest,
) -> Result<Task, AppError> {
    info!(title = %request.title, "creating task");
    let start = std::time::Instant::now();

    let task = service.create(request).await.map_err(|e| {
        error!(error = %e, "failed to create task");
        e
    })?;

    info!(
        task_id = %task.id,
        duration_ms = start.elapsed().as_millis() as u64,
        "task created successfully"
    );
    Ok(task)
}

Propagating spans across async tasks:

use tracing::Instrument;

// ✅ Propagate trace context into spawned tasks
tokio::spawn(
    async move { process_background_job(job_id).await }
        .instrument(tracing::info_span!("background_job", job_id = %job_id))
);

For async runtime debugging (task starvation, blocked tasks), use tokio-console with console-subscriber. For distributed tracing, bridge tracing to OpenTelemetry via tracing-opentelemetry.

Log Patterns by Operation Type

API Request/Response

// Request received
log.Info("request received",
  "method", r.Method,
  "path", r.URL.Path,
  "correlationId", correlationID,
  "userId", userID,
)

// Request completed
log.Info("request completed",
  "correlationId", correlationID,
  "status", statusCode,
  "duration", duration,
)

Database Operations

// Query start (DEBUG level)
log.Debug("executing query",
  "correlationId", correlationID,
  "query", "SELECT * FROM tasks WHERE user_id = $1",
)

// Query success (DEBUG level)
log.Debug("query completed",
  "correlationId", correlationID,
  "rowsReturned", len(results),
  "duration", duration,
)

// Query error (ERROR level)
log.Error("query failed",
  "correlationId", correlationID,
  "error", err,
  "query", "SELECT * FROM tasks WHERE user_id = $1",
)

External API Calls

// Call start
log.Info("calling external API",
  "correlationId", correlationID,
  "service", "email-provider",
  "endpoint", "/send",
)

// Retry (WARN level)
log.Warn("retrying external API call",
  "correlationId", correlationID,
  "service", "email-provider",
  "attempt", retryCount,
  "error", err,
)

// Circuit breaker open (WARN level)
log.Warn("circuit breaker opened",
  "correlationId", correlationID,
  "service", "email-provider",
  "failureCount", failures,
)

Background Jobs

// Job start
log.Info("job started",
  "jobId", jobID,
  "jobType", "email-digest",
)

// Progress (INFO level - periodic, not per-item)
log.Info("job progress",
  "jobId", jobID,
  "processed", 1000,
  "total", 5000,
  "percentComplete", 20,
)

// Job complete
log.Info("job completed",
  "jobId", jobID,
  "duration", duration,
  "itemsProcessed", count,
)

Error Scenarios

// Recoverable error (ERROR level)
log.Error("validation failed",
  "correlationId", correlationID,
  "userId", userID,
  "error", "invalid email format",
  "input", sanitizedInput, // Sanitized!
)

// Fatal error (FATAL level)
log.Fatal("critical dependency unavailable",
  "error", err,
  "dependency", "database",
  "action", "shutting down",
)

Environment-Specific Configuration

Configuration (Go example):

func configureLogger() *slog.Logger {
var handler slog.Handler

    level := slog.LevelInfo
    if os.Getenv("ENV") == "development" {
        level = slog.LevelDebug
        handler = slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
            Level: level,
        })
    } else {
        handler = slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
            Level: level,
        })
    }
    
    return slog.New(handler)
    }

Configuration (Rust example):

fn configure_logger() {
    let env = std::env::var("ENV").unwrap_or_else(|_| "development".into());
    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));

    if env == "development" {
        tracing_subscriber::fmt()
            .with_env_filter(tracing_subscriber::EnvFilter::new("debug"))
            .pretty()
            .init();
    } else {
        tracing_subscriber::fmt()
            .with_env_filter(filter)
            .json()
            .init();
    }
}

Testing Logs

Unit tests: Capture and assert on log output

// Go example
func TestUserLogin(t *testing.T) {
var buf bytes.Buffer
logger := slog.New(slog.NewJSONHandler(&buf, nil))

    // Test operation
    service := NewUserService(logger, mockStore)
    err := service.Login(ctx, email, password)
    
    // Assert logs
    require.NoError(t, err)
    logs := buf.String()
    assert.Contains(t, logs, "user login successful")
    assert.Contains(t, logs, email)
    }

Monitoring Integration

Correlation IDs:

• Generate at ingress (API gateway, first handler)
• Propagate through all services
• Include in all logs, errors, and traces
• Format: UUID v4

Log aggregation:

• Ship to centralized system (CloudWatch, GCP Logs, Datadog)
• Index by: correlationId, userId, level, timestamp
• Alert on ERROR/FATAL patterns
• Dashboard: request rates, error rates, latency

Checklist for Every Feature

• All public operations log INFO on start
• All operations log INFO/ERROR on complete/failure
• All logs include correlationId
• No sensitive data in logs
• Structured logging (key-value pairs)
• Appropriate log level used
• Error logs include error details
• Performance-critical paths use DEBUG level

Related Principles

• Logging and Observability Mandate @.agents/rules/logging-and-observability-mandate.md
• Monitoring and Alerting Principles @.agents/rules/monitoring-and-alerting-principles.md
• Error Handling Principles @.agents/rules/error-handling-principles.md
• Security Mandate @security-mandate.md
• Security Principles @security-principles.md
• API Design Principles @.agents/rules/api-design-principles.md