ELK Stack — Structured Logging

What Is Structured Logging?

Structured logging means emitting log events as machine-readable data structures (typically JSON) instead of free-form text strings. Each log entry is a set of key-value pairs with consistent field names.

# Unstructured log (hard to parse)
2026-04-01 10:15:30 ERROR UserService - Failed to fetch user id=42 reason=timeout duration=5023ms

# Structured log (JSON — easy to parse)
{"timestamp":"2026-04-01T10:15:30.456Z","level":"ERROR","service":"UserService","message":"Failed to fetch user","user_id":42,"reason":"timeout","duration_ms":5023}

Benefits of Structured Logging

Benefit	Description
No parsing needed	Fields are already extracted — no Grok or Dissect required
No multiline issues	Each JSON object is a single line
Type preservation	Numbers stay numbers, booleans stay booleans
Consistent schema	Every event has the same field structure
Easier correlation	Use request_id, trace_id fields across services
Better aggregation	Numeric fields enable avg(), sum(), percentile() queries

---

JSON Logging in Common Frameworks

Java — Log4j2 JSON Layout

log4j2.xml:

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN">
  <Appenders>
    <Console name="Console" target="SYSTEM_OUT">
      <JsonLayout compact="true" eventEol="true" stacktraceAsString="true">
        <KeyValuePair key="service" value="user-service" />
        <KeyValuePair key="environment" value="production" />
      </JsonLayout>
    </Console>
  </Appenders>
  <Loggers>
    <Root level="info">
      <AppenderRef ref="Console" />
    </Root>
  </Loggers>
</Configuration>

Output:

{"instant":{"epochSecond":1775178930,"nanoOfSecond":456000000},"thread":"main","level":"ERROR","loggerName":"com.example.UserService","message":"Failed to fetch user","thrown":{"commonElementCount":0,"localizedMessage":"Connection refused","message":"Connection refused","name":"java.sql.SQLException","extendedStackTrace":"java.sql.SQLException: Connection refused\n\tat com.mysql.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:800)"},"service":"user-service","environment":"production"}

Java — Logback with logstash-logback-encoder

logback.xml:

<configuration>
  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
    <encoder class="net.logstash.logback.encoder.LogstashEncoder">
      <customFields>{"service":"user-service","env":"production"}</customFields>
    </encoder>
  </appender>

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

Output:

{"@timestamp":"2026-04-01T10:15:30.456Z","@version":"1","message":"Failed to fetch user","logger_name":"com.example.UserService","thread_name":"main","level":"ERROR","level_value":40000,"stack_trace":"java.sql.SQLException: Connection refused\n\tat ...","service":"user-service","env":"production"}

Python — structlog

import structlog

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer()
    ]
)

logger = structlog.get_logger()
logger.error("Failed to fetch user", user_id=42, reason="timeout", duration_ms=5023)

Output:

{"timestamp":"2026-04-01T10:15:30.456Z","level":"error","event":"Failed to fetch user","user_id":42,"reason":"timeout","duration_ms":5023}

Python — python-json-logger

import logging
from pythonjsonlogger import jsonlogger

logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
    rename_fields={"asctime": "timestamp", "levelname": "level"}
)
handler.setFormatter(formatter)
logger.addHandler(handler)

logger.error("Failed to fetch user", extra={"user_id": 42, "reason": "timeout"})

Output:

{"timestamp":"2026-04-01 10:15:30,456","level":"ERROR","name":"root","message":"Failed to fetch user","user_id":42,"reason":"timeout"}

Node.js — Winston

const winston = require('winston');

const logger = winston.createLogger({
  level: 'info',
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  defaultMeta: { service: 'user-service' },
  transports: [new winston.transports.Console()]
});

logger.error('Failed to fetch user', { user_id: 42, reason: 'timeout', duration_ms: 5023 });

Output:

{"level":"error","message":"Failed to fetch user","user_id":42,"reason":"timeout","duration_ms":5023,"service":"user-service","timestamp":"2026-04-01T10:15:30.456Z"}

Node.js — Pino

const pino = require('pino');
const logger = pino({ name: 'user-service' });

logger.error({ user_id: 42, reason: 'timeout', duration_ms: 5023 }, 'Failed to fetch user');

Output:

{"level":50,"time":1775178930456,"pid":12345,"hostname":"web01","name":"user-service","user_id":42,"reason":"timeout","duration_ms":5023,"msg":"Failed to fetch user"}

---

Elastic Common Schema (ECS)

ECS is a standard field naming convention published by Elastic. It ensures that data from different sources uses the same field names, enabling cross-source correlation and reusable dashboards.

Core ECS Fields

ECS Field	Type	Description
@timestamp	date	Event timestamp
message	text	Log message
log.level	keyword	Log level (info, error, etc.)
log.logger	keyword	Logger name
service.name	keyword	Service name
service.version	keyword	Service version
service.environment	keyword	Environment (production, staging)
host.name	keyword	Hostname
host.ip	ip	Host IP address
event.kind	keyword	Event kind (event, alert, metric)
event.category	keyword	Event category (authentication, web)
event.outcome	keyword	Outcome (success, failure)

HTTP ECS Fields

ECS Field	Type	Description
http.request.method	keyword	HTTP method
http.response.status_code	integer	Response status code
http.response.body.bytes	long	Response size in bytes
url.path	keyword	URL path
url.query	keyword	Query string
user_agent.original	text	User agent string

Source/Client ECS Fields

ECS Field	Type	Description
source.ip	ip	Source IP address
source.port	integer	Source port
client.ip	ip	Client IP address
user.name	keyword	Username

Error ECS Fields

ECS Field	Type	Description
error.type	keyword	Exception class name
error.message	text	Error message
error.stack_trace	text	Full stack trace

ECS-Formatted Log Example

{
  "@timestamp": "2026-04-01T10:15:30.456Z",
  "message": "Failed to fetch user",
  "log.level": "error",
  "log.logger": "com.example.UserService",
  "service.name": "user-service",
  "service.version": "2.1.0",
  "service.environment": "production",
  "host.name": "web01",
  "error.type": "java.sql.SQLException",
  "error.message": "Connection refused",
  "error.stack_trace": "java.sql.SQLException: Connection refused\n\tat ...",
  "event.kind": "event",
  "event.category": "database",
  "event.outcome": "failure",
  "user.name": "admin"
}

---

Processing JSON Logs in Logstash

JSON Filter

Parse a JSON string field into structured fields:

filter {
  json {
    source => "message"
    target => ""
  }
}

• source — the field containing the JSON string (default: message)
• target — where to place parsed fields. Empty string "" merges fields into the root

JSON Codec

If the entire input is JSON, use the json codec on the input:

input {
  beats {
    port => 5044
    codec => json
  }
}

Mapping to ECS Fields

filter {
  json {
    source => "message"
  }

  # Map application fields to ECS
  mutate {
    rename => {
      "level"     => "[log][level]"
      "logger"    => "[log][logger]"
      "service"   => "[service][name]"
      "host"      => "[host][name]"
      "client_ip" => "[source][ip]"
      "method"    => "[http][request][method]"
      "status"    => "[http][response][status_code]"
      "path"      => "[url][path]"
    }
  }

  # Parse the timestamp
  date {
    match => ["timestamp", "ISO8601"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }
}

---

Processing JSON Logs with Filebeat

decode_json_fields Processor

Filebeat can decode JSON fields without Logstash:

filebeat.inputs:
  - type: filestream
    id: json-app-logs
    paths:
      - /var/log/myapp/*.json
    parsers:
      - ndjson:
          target: ""
          add_error_key: true
          message_key: message

processors:
  - decode_json_fields:
      fields: ["message"]
      target: ""
      overwrite_keys: true
      add_error_key: true

Using ndjson Parser (Filebeat 8.x+)

For newline-delimited JSON (NDJSON) files, use the ndjson parser directly:

filebeat.inputs:
  - type: filestream
    id: json-app-logs
    paths:
      - /var/log/myapp/*.json
    parsers:
      - ndjson:
          target: ""
          overwrite_keys: true
          add_error_key: true
          message_key: message

---

Log Schema Design Best Practices

Do’s

Practice	Example
Use ECS field names	service.name, log.level, source.ip
Include request/trace IDs	"trace_id": "abc-123" for distributed tracing
Use consistent types	Always use integers for status codes, not strings
Include service metadata	service.name, service.version, service.environment
Timestamp in ISO 8601	"2026-04-01T10:15:30.456Z"
Keep messages human-readable	"message": "User login failed"
Log at the right level	ERROR for failures, INFO for business events

Don’ts

Anti-Pattern	Why
Logging sensitive data	Don’t log passwords, tokens, credit card numbers
Dynamic field names	"user_42": "data" causes mapping explosion
Nested beyond 3 levels	Deep nesting makes querying difficult
Inconsistent field types	"status": 200 in one log, "status": "OK" in another
Missing timestamps	Every log event must have a timestamp
Giant log messages	Keep under 10KB per event

---

Structured Logging Pipeline — Complete Example

Application (Node.js with Winston)

const winston = require('winston');

const logger = winston.createLogger({
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  defaultMeta: {
    service: { name: 'api-gateway', version: '1.2.0', environment: 'production' }
  },
  transports: [
    new winston.transports.File({ filename: '/var/log/api-gateway/app.json' })
  ]
});

Filebeat Config

filebeat.inputs:
  - type: filestream
    id: api-gateway-logs
    paths:
      - /var/log/api-gateway/*.json
    parsers:
      - ndjson:
          target: ""
          overwrite_keys: true
          message_key: message

output.logstash:
  hosts: ["logstash:5044"]

Logstash Pipeline

input {
  beats {
    port => 5044
  }
}

filter {
  # Timestamp is already in the JSON — parse it
  date {
    match => ["timestamp", "ISO8601"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }

  # Map to ECS if needed
  mutate {
    rename => {
      "[service][name]"    => "[service][name]"
      "[service][version]" => "[service][version]"
      "level"              => "[log][level]"
    }
  }
}

output {
  elasticsearch {
    hosts => ["https://es01:9200"]
    index => "logs-api-gateway-%{+YYYY.MM.dd}"
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
  }
}

---

Quick Reference

# Structured logging = JSON logs with consistent field names

# Key frameworks
Java:    Log4j2 JsonLayout, logstash-logback-encoder
Python:  structlog, python-json-logger
Node.js: Winston (json format), Pino

# ECS core fields
@timestamp, message, log.level, log.logger
service.name, service.version, service.environment
host.name, source.ip, error.type, error.message

# Logstash JSON processing
filter {
  json { source => "message" }
}

# Filebeat JSON processing
parsers:
  - ndjson:
      target: ""
      message_key: message

# Best practices
✅ Use ECS field names
✅ Include trace/request IDs
✅ ISO 8601 timestamps
✅ Consistent field types
❌ No sensitive data in logs
❌ No dynamic field names
❌ No inconsistent types

---

Previous: Multiline Logs ← · Next: Log Parsing Exercises →