JSONL for Log Processing

Node.js with Winston

Winston is a widely used Node.js logging library with built-in JSON output, one object per line.

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

# stdlib logging carries the lines, so give it a handler and a level first
logging.basicConfig(format="%(message)s", stream=sys.stdout, level=logging.INFO)

# 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 (
    "errors"
    "net/http"
    "os"
    "time"

    "github.com/rs/zerolog"
    "github.com/rs/zerolog/log"
)

func connectDatabase() error { return errors.New("connection refused") } // stand-in

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")
    }
}

// statusWriter records the status code a handler writes
type statusWriter struct {
    http.ResponseWriter
    status int
}

func (w *statusWriter) WriteHeader(code int) {
    w.status = code
    w.ResponseWriter.WriteHeader(code)
}

// HTTP middleware example (package level - Go has no nested func declarations)
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, status: http.StatusOK}
        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/fluent/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>

# Fan out to Elasticsearch and S3. Only the first matching <match> block
# receives an event, so two destinations go inside one copy output.
<match application.logs>
  @type copy

  <store>
    @type elasticsearch
    host localhost
    port 9200
    index_name logs-%Y.%m.%d

    <buffer>
      @type file
      path /var/log/fluent/buffer/logs
      flush_interval 10s
      chunk_limit_size 5M
    </buffer>
  </store>

  # Archive to S3
  <store>
    @type s3
    s3_bucket my-logs-bucket
    s3_region us-east-1
    path logs/%Y/%m/%d/

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

    <format>
      @type json
    </format>
  </store>
</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
})

Grafana Loki: Label-Based Indexing

Loki builds no full-text index. It indexes a small set of metadata labels - namespace, application, environment - and stores raw log lines in compressed chunks. A query selects a stream by its labels first, then parses the lines in that stream at read time.

The mapping from JSONL is direct: the entire JSON object stays as the log line, and only stable, low-cardinality fields become labels. High-cardinality values - request ids, user ids, trace ids, status codes - must stay inside the line, because every distinct label combination creates a stream and too many streams degrade ingestion and queries alike. Loki 3.0 added structured metadata for values that fit neither place.

# LogQL: labels pick the stream, the parser reads the JSONL line

# 1. Stream selector (indexed) then JSON parsing (read time)
{namespace="checkout", container="api"}
  | json
  | level = "error"
  | duration_ms > 1000

# 2. High-cardinality fields live IN the line, not in labels.
#    This is correct - it filters after parsing:
{namespace="checkout"} | json | request_id = "req_abc123"

#    This would be wrong - request_id as a label creates one
#    stream per request and blows up the index:
#    {namespace="checkout", request_id="req_abc123"}

# 3. Pull a nested field out with an explicit label expression
{namespace="checkout"} | json service_name="service.name", err="error.type"
  | err = "UpstreamTimeout"

# 4. Metric query built from the parsed line
sum by (service_name) (
  rate({namespace="checkout"} | json service_name="service.name" | level = "error" [5m])
)

Source: Grafana Loki documentation - Understand labels

OpenSearch: The Same Bulk Envelope

OpenSearch is the community fork of Elasticsearch, and for log ingestion the important detail is how little changed. The Bulk API still takes a newline-delimited body under the application/x-ndjson content type, still alternates an action line with a source line, and still requires a trailing newline. Only the action line is parsed on the receiving node, which is what makes the format fast to accept.

# OpenSearch _bulk - identical envelope to the Elasticsearch example above
curl -X POST "https://opensearch.example.com:9200/_bulk" \
  -H "Content-Type: application/x-ndjson" \
  --data-binary @logs.jsonl

# logs.jsonl - action line, then source line, then a trailing newline
{"index": {"_index": "logs-2026.07.01"}}
{"@timestamp": "2026-07-01T10:30:00Z", "level": "info", "message": "Server started", "service": "api"}
{"create": {"_index": "logs-2026.07.01", "_id": "evt-1002"}}
{"@timestamp": "2026-07-01T10:30:05Z", "level": "error", "message": "Database timeout", "service": "api"}

# Split a large file so one request stays a reasonable size.
# The split MUST fall on an action/source pair boundary, never between them.
split -l 2000 --additional-suffix=.jsonl logs.jsonl chunk-

for f in chunk-*.jsonl; do
  curl -s -X POST "https://opensearch.example.com:9200/_bulk" \
    -H "Content-Type: application/x-ndjson" \
    --data-binary "@$f"
done

Switching a shipper between the two clusters is a matter of endpoint and credentials, not format. The differences that bite are elsewhere: index lifecycle configuration, client libraries that refuse the other project's version banner, and query dialects that have begun to diverge.

Source: OpenSearch Documentation - Bulk

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
import time
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.

The Unbounded Line Hazard

Nothing in the format caps line length, but almost every reader has a buffer that does. Go is the canonical example: the standard library documents bufio.MaxScanTokenSize as 64 * 1024 bytes, and a bufio.Scanner meeting a longer line returns bufio.ErrTooLong and stops. Because the loop condition is just scanner.Scan(), code that never checks scanner.Err() treats that as a clean end of file: the program exits successfully, having silently discarded every line after the big one.

// Go: the default 64 KB scanner ceiling, and how to survive it.
scanner := bufio.NewScanner(f)

// Raise the ceiling deliberately: 64 KB initial buffer, 16 MB maximum.
scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024)

var parsed, malformed int
for scanner.Scan() {
    var entry LogEntry
    if err := json.Unmarshal(scanner.Bytes(), &entry); err != nil {
        // One bad line must never end the stream. Count it and move on.
        malformed++
        continue
    }
    parsed++
    process(entry)
}

// THIS is the check people forget. Without it, bufio.ErrTooLong looks
// exactly like a successful end of file and the tail of the file is lost.
if err := scanner.Err(); err != nil {
    log.Fatalf("log scan aborted after %d lines: %v", parsed, err)
}
log.Printf("parsed=%d malformed=%d", parsed, malformed)

Guarding Both Ends of the Pipe

Raising a buffer limit only moves the wall. Bound the line where it is written, and make every reader explicit about what it accepts - which is why the collector configs above set those knobs deliberately.

• At the producer: truncate long strings, cap stack traces at a fixed number of frames, never log a body in full
• For large payloads: write the blob to object storage and log a pointer plus a checksum
• At the collector: buffer_max_size and skip_long_lines in Fluent Bit, max_line_bytes in Vector, max_log_size in filelog
• At the reader: set the buffer ceiling explicitly, and inspect the terminal error after the loop ends
• Everywhere: emit a counter for skipped and malformed lines, so silent loss becomes visible

# Python: skip-and-count instead of crash-on-first-bad-line
import json

def read_logs(path, max_bytes=1_048_576):
    stats = {"parsed": 0, "malformed": 0, "oversized": 0}

    with open(path, "r", encoding="utf-8") as f:
        for lineno, line in enumerate(f, 1):
            if len(line) > max_bytes:
                stats["oversized"] += 1
                continue

            line = line.strip()
            if not line:
                continue

            try:
                yield json.loads(line)
                stats["parsed"] += 1
            except json.JSONDecodeError as exc:
                stats["malformed"] += 1
                # Log the position, never the whole offending line
                print(f"bad JSON at line {lineno}: {exc.msg}")

    print(stats)