logging

Logging Package

Structured logging handlers with multi-destination support, NATS publishing, and graceful fallback behavior.

Features

• Multi-Handler Composition: Dispatch logs to multiple destinations simultaneously
• NATS Publishing: Publish logs to JetStream for real-time streaming
• Source Filtering: Exclude specific sources from NATS publishing (prefix matching)
• Graceful Degradation: NATS failures never block stdout logging
• Async Publishing: Non-blocking NATS publish to avoid logging latency
• Thread-Safe: Safe for concurrent use across goroutines

Installation

import "github.com/c360/semstreams/pkg/logging"

Quick Start

Basic Multi-Handler Setup

import (
    "log/slog"
    "os"
    
    "github.com/c360/semstreams/pkg/logging"
)

// Create stdout handler
stdoutHandler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
    Level: slog.LevelInfo,
})

// Create NATS handler
natsHandler := logging.NewNATSLogHandler(natsClient, logging.NATSLogHandlerConfig{
    MinLevel:       slog.LevelInfo,
    ExcludeSources: nil,
})

// Compose handlers - logs go to both stdout and NATS
multiHandler := logging.NewMultiHandler(stdoutHandler, natsHandler)
logger := slog.New(multiHandler)
slog.SetDefault(logger)

With Source Filtering

// Exclude WebSocket worker logs from NATS to prevent feedback loops
natsHandler := logging.NewNATSLogHandler(natsClient, logging.NATSLogHandlerConfig{
    MinLevel:       slog.LevelDebug,
    ExcludeSources: []string{"flow-service.websocket"},
})

// This goes to stdout only (source matches exclude prefix):
logger.With("source", "flow-service.websocket.health").Info("Health check")

// This goes to both stdout and NATS:
logger.With("source", "flow-service").Info("Flow started")

Handlers

MultiHandler

Composes multiple slog.Handler instances, dispatching log records to all of them.

multi := logging.NewMultiHandler(handler1, handler2, handler3)
logger := slog.New(multi)

// Logs dispatched to all three handlers
logger.Info("Hello world")

Key Behaviors:

• If one handler fails, others continue processing
• Enabled() returns true if ANY handler is enabled for the level
• WithAttrs() and WithGroup() create new instances (immutable)

NATSLogHandler

Publishes log records to NATS subjects in the format logs.{source}.{level}.

natsHandler := logging.NewNATSLogHandler(natsClient, logging.NATSLogHandlerConfig{
    MinLevel:       slog.LevelInfo,
    ExcludeSources: []string{"noisy-component", "debug-only"},
})

Configuration:

Source Extraction

NATSLogHandler extracts the source identifier from log attributes with priority:

1. source attribute (explicit)
2. component attribute (component-tagged logs)
3. service attribute (service-tagged logs)
4. "system" (default fallback)

// Explicit source
logger.With("source", "my-service.worker").Info("Working")
// → logs.my-service.worker.INFO

// Component attribute
logger.With("component", "udp-input").Info("Packet received")
// → logs.udp-input.INFO

// No source attributes
slog.Info("System message")
// → logs.system.INFO

Source Filtering

Exclude sources from NATS publishing using prefix matching:

ExcludeSources: []string{"flow-service.websocket"}

Matching Rules:

• "flow-service.websocket" → excluded
• "flow-service.websocket.health" → excluded (prefix match)
• "flow-service" → NOT excluded (different prefix)
• "flow-service.api" → NOT excluded (different prefix)

Use Case: Prevent feedback loops where WebSocket worker debug logs are sent over the same WebSocket connection.

NATS Subject Pattern

Logs are published to subjects following this pattern:

logs.{source}.{level}
  └── logs.udp-input.INFO
  └── logs.graph-processor.ERROR
  └── logs.system.WARN
  └── logs.flow-service.DEBUG

JetStream Stream Configuration

Configure a LOGS stream with TTL and size limits:

StreamConfig{
    Name:     "LOGS",
    Subjects: []string{"logs.>"},
    MaxAge:   1 * time.Hour,        // TTL: expire after 1 hour
    MaxBytes: 100 * 1024 * 1024,    // 100MB max storage
    Discard:  DiscardOld,           // Drop oldest when full
}

Log Entry Format

Logs published to NATS are JSON-encoded:

{
    "timestamp": "2024-01-15T10:30:00.123456789Z",
    "level": "INFO",
    "source": "udp-input",
    "message": "Packet received",
    "fields": {
        "bytes": 1024,
        "remote_addr": "192.168.1.1:5000"
    }
}

Architecture

Data Flow

┌─────────────────────────────────────────────────────────────────┐
│                        slog.Logger                               │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                       MultiHandler                               │
│  ┌──────────────────────┐    ┌──────────────────────────────┐   │
│  │   TextHandler        │    │      NATSLogHandler          │   │
│  │   (stdout)           │    │                              │   │
│  │                      │    │  ┌────────────────────────┐  │   │
│  │                      │    │  │ Check MinLevel         │  │   │
│  │                      │    │  │ Extract Source         │  │   │
│  │                      │    │  │ Check ExcludeSources   │  │   │
│  │                      │    │  │ Async Publish to NATS  │  │   │
│  │                      │    │  └────────────────────────┘  │   │
│  └──────────────────────┘    └──────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘
         │                                    │
         ▼                                    ▼
    ┌─────────┐                    ┌──────────────────┐
    │ stdout  │                    │  NATS JetStream  │
    └─────────┘                    │  (LOGS stream)   │
                                   └──────────────────┘
                                            │
                                            ▼
                                   ┌──────────────────┐
                                   │  WebSocket       │
                                   │  Status Stream   │
                                   └──────────────────┘

Graceful Degradation

The architecture ensures logging never fails due to NATS issues:

Performance

Benchmarks (M3 MacBook Pro)

At 10,000 logs/second: ~1.5ms total overhead per second.

Memory

• MultiHandler: O(n) where n is number of handlers
• NATSLogHandler: O(1) for handler state
• Per log: O(m) where m is number of attributes
• Async goroutines: Short-lived, minimal impact

Integration with WebSocket Status Stream

This package is designed to work with the service package's WebSocket status stream:

1. Application logs → Published to NATS via NATSLogHandler
2. LOGS JetStream stream → Stores logs with TTL (1hr default)
3. WebSocket clients → Subscribe to logs.> subjects
4. Real-time streaming → No slog interception timing issues

The ExcludeSources config prevents feedback loops where WebSocket worker logs would be sent over the WebSocket, generating more logs.

Testing

# Run tests with race detector
go test -race ./pkg/logging

# Run benchmarks
go test -bench=. ./pkg/logging

Example: Complete Application Setup

package main

import (
    "context"
    "log/slog"
    "os"

    "github.com/c360/semstreams/config"
    "github.com/c360/semstreams/natsclient"
    "github.com/c360/semstreams/pkg/logging"
)

func main() {
    // Connect to NATS
    natsClient, err := natsclient.NewClient("nats://localhost:4222")
    if err != nil {
        slog.Error("Failed to connect to NATS", "error", err)
        os.Exit(1)
    }
    defer natsClient.Close(context.Background())

    // Ensure LOGS stream exists
    streamsManager := config.NewStreamsManager(natsClient, slog.Default())
    if err := streamsManager.EnsureStreams(context.Background(), cfg); err != nil {
        slog.Error("Failed to ensure streams", "error", err)
        os.Exit(1)
    }

    // Setup multi-destination logging
    level := slog.LevelInfo
    
    stdoutHandler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
        Level: level,
    })
    
    natsHandler := logging.NewNATSLogHandler(natsClient, logging.NATSLogHandlerConfig{
        MinLevel:       level,
        ExcludeSources: []string{"flow-service.websocket"},
    })
    
    multiHandler := logging.NewMultiHandler(stdoutHandler, natsHandler)
    logger := slog.New(multiHandler)
    slog.SetDefault(logger)

    // Application logs now go to both stdout and NATS
    slog.Info("Application started", "version", "1.0.0")

    // Component-tagged logs
    componentLogger := slog.With("component", "processor")
    componentLogger.Info("Processing started", "batch_size", 100)

    // Source-tagged logs for fine-grained filtering
    workerLogger := slog.With("source", "flow-service.websocket")
    workerLogger.Debug("WebSocket worker tick") // Goes to stdout only
}

Related Packages

• service: WebSocket status stream that consumes NATS logs
• config: LOGS stream configuration with TTL and size limits
• natsclient: NATS client used for publishing

License

See LICENSE file in repository root.