JSONL for Log Processing

Node.js with Winston

Winston is the most popular logging library for Node.js with built-in JSONL support.

const winston = require('winston');

// Configure Winston for JSONL output
const logger = winston.createLogger({
  level: 'info',
  format: winston.format.json(),
  transports: [
    // Write all logs to application.jsonl
    new winston.transports.File({
      filename: 'application.jsonl',
      format: winston.format.combine(
        winston.format.timestamp(),
        winston.format.json()
      )
    }),
    // Also log to console for development
    new winston.transports.Console({
      format: winston.format.simple()
    })
  ]
});

// Add request context middleware (Express)
const addContext = (req, res, next) => {
  req.logger = logger.child({
    request_id: req.id,
    user_id: req.user?.id,
    ip: req.ip,
    user_agent: req.get('user-agent')
  });
  next();
};

// Log examples
logger.info('Server started', { port: 3000, env: 'production' });

logger.error('Database query failed', {
  query: 'SELECT * FROM users',
  error: err.message,
  stack: err.stack,
  duration_ms: 1250
});

logger.warn('High memory usage', {
  memory_used_mb: 512,
  memory_limit_mb: 1024,
  threshold_percent: 80
});

// With request context
app.get('/api/users', addContext, async (req, res) => {
  req.logger.info('Fetching users', {
    page: req.query.page,
    limit: req.query.limit
  });
  // ... handler code
});

Python with Structlog

Structlog makes structured logging in Python simple and performant.

import structlog
import logging
import sys

# Configure structlog for JSONL output
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

# Create logger
logger = structlog.get_logger()

# Basic logging
logger.info("server_started", port=8000, workers=4)

logger.error(
    "database_connection_failed",
    host="db.example.com",
    port=5432,
    retry_count=3,
    exc_info=True
)

# Add context for multiple log entries
log_with_context = logger.bind(
    user_id=12345,
    request_id="req_abc123",
    session_id="sess_xyz789"
)

log_with_context.info("user_action", action="login", ip="192.168.1.100")
log_with_context.info("user_action", action="view_profile")

# Django/Flask middleware example
class LoggingMiddleware:
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        request_id = environ.get('HTTP_X_REQUEST_ID', 'unknown')
        logger = structlog.get_logger().bind(
            request_id=request_id,
            path=environ.get('PATH_INFO'),
            method=environ.get('REQUEST_METHOD')
        )

        logger.info("request_started")
        try:
            return self.app(environ, start_response)
        finally:
            logger.info("request_completed")

Go with Zerolog

Zerolog is a zero-allocation JSON logger for Go with excellent performance.

package main

import (
    "os"
    "github.com/rs/zerolog"
    "github.com/rs/zerolog/log"
)

func main() {
    // Configure logger for JSONL file output
    file, err := os.OpenFile(
        "application.jsonl",
        os.O_APPEND|os.O_CREATE|os.O_WRONLY,
        0644,
    )
    if err != nil {
        panic(err)
    }
    defer file.Close()

    log.Logger = zerolog.New(file).With().Timestamp().Caller().Logger()

    // Basic logging
    log.Info().
        Str("event", "server_start").
        Int("port", 8080).
        Str("env", "production").
        Msg("Server starting")

    // Error logging with context
    err = connectDatabase()
    if err != nil {
        log.Error().
            Err(err).
            Str("host", "db.example.com").
            Int("port", 5432).
            Int("retry", 3).
            Msg("Database connection failed")
    }

    // HTTP middleware example
    func loggingMiddleware(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            start := time.Now()

            // Create logger with request context
            logger := log.With().
                Str("request_id", r.Header.Get("X-Request-ID")).
                Str("method", r.Method).
                Str("path", r.URL.Path).
                Str("remote_addr", r.RemoteAddr).
                Logger()

            logger.Info().Msg("request_started")

            // Wrap response writer to capture status
            wrapped := &statusWriter{ResponseWriter: w}
            next.ServeHTTP(wrapped, r)

            logger.Info().
                Int("status", wrapped.status).
                Dur("duration_ms", time.Since(start)).
                Msg("request_completed")
        })
    }
}

Java with Logback

Configure Logback for JSON structured logging in Spring Boot applications.

<!-- logback-spring.xml -->
<configuration>
    <appender name="FILE" class="ch.qos.logback.core.FileAppender">
        <file>application.jsonl</file>
        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
            <includeContext>false</includeContext>
            <timestampPattern>yyyy-MM-dd'T'HH:mm:ss.SSS'Z'</timestampPattern>
        </encoder>
    </appender>

    <root level="INFO">
        <appender-ref ref="FILE" />
    </root>
</configuration>

// Java application code
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

public class UserService {
    private static final Logger logger = LoggerFactory.getLogger(UserService.class);

    public User getUser(Long userId) {
        // Add context to MDC (Mapped Diagnostic Context)
        MDC.put("user_id", userId.toString());
        MDC.put("operation", "get_user");

        try {
            logger.info("Fetching user from database");
            User user = userRepository.findById(userId);

            if (user == null) {
                logger.warn("User not found");
                return null;
            }

            logger.info("User retrieved successfully");
            return user;
        } catch (Exception e) {
            logger.error("Error fetching user", e);
            throw e;
        } finally {
            MDC.clear();
        }
    }
}

Rust with Tracing

use tracing::{info, error, warn};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};

fn main() {
    // Configure JSON formatter
    let file = std::fs::File::create("application.jsonl")
        .expect("Failed to create log file");

    tracing_subscriber::registry()
        .with(
            tracing_subscriber::fmt::layer()
                .json()
                .with_writer(std::sync::Arc::new(file))
        )
        .init();

    info!(port = 8080, env = "production", "Server started");

    // Structured logging with context
    let user_id = 12345;
    let request_id = "req_abc123";

    info!(
        user_id = user_id,
        request_id = request_id,
        action = "login",
        "User logged in"
    );

    // Error logging
    if let Err(e) = connect_database() {
        error!(
            error = %e,
            host = "db.example.com",
            port = 5432,
            "Database connection failed"
        );
    }
}

Fluentd Configuration

# fluent.conf
<source>
  @type tail
  path /var/log/application.jsonl
  pos_file /var/log/td-agent/application.pos
  tag application.logs

  <parse>
    @type json
    time_key timestamp
    time_format %Y-%m-%dT%H:%M:%S%z
  </parse>
</source>

# Filter: Add hostname and enrich data
<filter application.logs>
  @type record_transformer
  <record>
    hostname ${hostname}
    tag ${tag}
    env production
  </record>
</filter>

# Filter: Parse user agent
<filter application.logs>
  @type parser
  key_name user_agent
  reserve_data true
  <parse>
    @type user_agent
  </parse>
</filter>

# Output to Elasticsearch
<match application.logs>
  @type elasticsearch
  host localhost
  port 9200
  index_name logs-%Y.%m.%d
  type_name _doc

  <buffer>
    @type file
    path /var/log/td-agent/buffer/logs
    flush_interval 10s
    chunk_limit_size 5M
  </buffer>
</match>

# Output to S3 for archival
<match application.logs>
  @type s3
  s3_bucket my-logs-bucket
  s3_region us-east-1
  path logs/%Y/%m/%d/

  <buffer time>
    @type file
    path /var/log/td-agent/buffer/s3
    timekey 3600  # 1 hour
    timekey_wait 10m
    chunk_limit_size 256m
  </buffer>

  <format>
    @type json
  </format>
</match>

Application Integration

// Node.js: fluent-logger
const logger = require('fluent-logger');

logger.configure('application', {
  host: 'localhost',
  port: 24224,
  timeout: 3.0
});

// Send structured logs
logger.emit('user.action', {
  user_id: 12345,
  action: 'login',
  ip: '192.168.1.100',
  timestamp: new Date().toISOString()
});

logger.emit('database.query', {
  query: 'SELECT * FROM users',
  duration_ms: 45,
  rows: 100
});

// Python: fluent-logger-python
from fluent import sender
from fluent import event

logger = sender.FluentSender('application', host='localhost', port=24224)

logger.emit('user.action', {
    'user_id': 12345,
    'action': 'login',
    'ip': '192.168.1.100'
})

logger.emit('database.query', {
    'query': 'SELECT * FROM users',
    'duration_ms': 45,
    'rows': 100
})

Streaming with Python

import json
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class LogFileHandler(FileSystemEventHandler):
    def __init__(self, filepath):
        self.filepath = filepath
        self.file = open(filepath, 'r')
        # Move to end of file
        self.file.seek(0, 2)

    def on_modified(self, event):
        if event.src_path == self.filepath:
            # Read new lines
            for line in self.file:
                try:
                    log_entry = json.loads(line)
                    self.process_log(log_entry)
                except json.JSONDecodeError:
                    pass

    def process_log(self, log_entry):
        # Real-time processing
        if log_entry.get('level') == 'error':
            self.alert_on_error(log_entry)

        if log_entry.get('duration_ms', 0) > 1000:
            self.alert_slow_request(log_entry)

    def alert_on_error(self, log):
        print(f"ERROR ALERT: {log['message']}")
        # Send to Slack, PagerDuty, etc.

    def alert_slow_request(self, log):
        print(f"SLOW REQUEST: {log.get('path')} took {log['duration_ms']}ms")

# Watch log file
handler = LogFileHandler('/var/log/application.jsonl')
observer = Observer()
observer.schedule(handler, path='/var/log', recursive=False)
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

Real-Time Alerting

import json
from collections import defaultdict
from datetime import datetime, timedelta

class LogAlerter:
    def __init__(self):
        self.error_counts = defaultdict(int)
        self.last_reset = datetime.now()

    def process_log_stream(self, log_file):
        with open(log_file, 'r') as f:
            # Follow file like 'tail -f'
            f.seek(0, 2)
            while True:
                line = f.readline()
                if not line:
                    time.sleep(0.1)
                    continue

                try:
                    log = json.loads(line)
                    self.analyze(log)
                except json.JSONDecodeError:
                    continue

    def analyze(self, log):
        # Reset counters every minute
        if datetime.now() - self.last_reset > timedelta(minutes=1):
            self.check_thresholds()
            self.error_counts.clear()
            self.last_reset = datetime.now()

        # Count errors by service
        if log.get('level') == 'error':
            service = log.get('service', 'unknown')
            self.error_counts[service] += 1

        # Immediate alerts
        if 'out of memory' in log.get('message', '').lower():
            self.send_alert('CRITICAL', 'Out of memory error detected', log)

        if log.get('status_code') == 500:
            self.send_alert('HIGH', '500 error detected', log)

    def check_thresholds(self):
        for service, count in self.error_counts.items():
            if count > 100:  # More than 100 errors per minute
                self.send_alert(
                    'HIGH',
                    f'{service} error rate exceeded: {count}/min',
                    {'service': service, 'count': count}
                )

    def send_alert(self, severity, message, context):
        # Integrate with alerting systems
        print(f"[{severity}] {message}")
        print(f"Context: {json.dumps(context, indent=2)}")
        # Send to Slack, PagerDuty, Email, etc.