86 lines
3.0 KiB
Plaintext
86 lines
3.0 KiB
Plaintext
|
# =============================================================================
|
||
|
|
# Application log pipeline.
|
||
|
|
#
|
||
|
|
# Django emits one JSON object per log record and ships it over TCP with
|
||
|
|
# `python-logstash-async` (asynchronous, buffered - a Logstash outage never
|
||
|
|
# blocks a request thread). Because the payload is already structured there is
|
||
|
|
# no grok parsing here: the pipeline only enriches, classifies and routes.
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
input {
|
||
|
|
tcp {
|
||
|
|
port => 5959
|
||
|
|
codec => json_lines
|
||
|
|
tags => ["django"]
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
filter {
|
||
|
|
# Fields Logstash adds itself and that carry no information for us.
|
||
|
|
mutate {
|
||
|
|
remove_field => ["port", "@version"]
|
||
|
|
}
|
||
|
|
|
||
|
|
# The application already stamps an ISO-8601 timestamp; trust it so the log
|
||
|
|
# line is filed under the moment it happened, not the moment it arrived.
|
||
|
|
date {
|
||
|
|
match => ["@timestamp", "ISO8601"]
|
||
|
|
target => "@timestamp"
|
||
|
|
tag_on_failure => ["_timestamp_parse_failure"]
|
||
|
|
}
|
||
|
|
|
||
|
|
# Route by event type. `event` is set by the application through `extra=`.
|
||
|
|
if [event] == "search_executed" {
|
||
|
|
mutate { add_field => { "[@metadata][stream]" => "search" } }
|
||
|
|
|
||
|
|
# A slow search is worth alerting on; pre-computing the flag keeps the
|
||
|
|
# Kibana query simple.
|
||
|
|
if [took_ms] and [took_ms] > 500 {
|
||
|
|
mutate { add_tag => ["slow_search"] }
|
||
|
|
}
|
||
|
|
if [result_count] == 0 {
|
||
|
|
mutate { add_tag => ["zero_results"] }
|
||
|
|
}
|
||
|
|
} else if [event] in ["login_succeeded", "login_failed", "logout", "user_registered"] {
|
||
|
|
mutate { add_field => { "[@metadata][stream]" => "audit" } }
|
||
|
|
} else if [level] in ["ERROR", "CRITICAL"] {
|
||
|
|
mutate { add_field => { "[@metadata][stream]" => "error" } }
|
||
|
|
} else {
|
||
|
|
mutate { add_field => { "[@metadata][stream]" => "app" } }
|
||
|
|
}
|
||
|
|
|
||
|
|
# Never let a stray contact detail reach the log store.
|
||
|
|
mutate {
|
||
|
|
remove_field => ["password", "authorization", "emails", "phone_numbers"]
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
output {
|
||
|
|
elasticsearch {
|
||
|
|
# Credentials, when the cluster has security enabled, travel inside the
|
||
|
|
# URL (see docker-compose.prod.yml) so this single pipeline file works
|
||
|
|
# both with and without authentication.
|
||
|
|
hosts => ["${ELASTICSEARCH_HOSTS:http://elasticsearch:9200}"]
|
||
|
|
|
||
|
|
# One index per stream per day keeps retention policies simple and makes
|
||
|
|
# "delete everything older than 30 days" a single index deletion.
|
||
|
|
index => "app-logs-%{[@metadata][stream]}-%{+YYYY.MM.dd}"
|
||
|
|
|
||
|
|
# Composable index template (Elasticsearch 8 style): pins the types of the
|
||
|
|
# fields Kibana filters and aggregates on instead of letting dynamic
|
||
|
|
# mapping guess them.
|
||
|
|
manage_template => true
|
||
|
|
template_api => "composable"
|
||
|
|
template_name => "app-logs"
|
||
|
|
template => "/usr/share/logstash/templates/app-logs.json"
|
||
|
|
template_overwrite => true
|
||
|
|
|
||
|
|
action => "index"
|
||
|
|
}
|
||
|
|
|
||
|
|
# Anything that fails to parse is still visible in `docker compose logs`.
|
||
|
|
if "_jsonparsefailure" in [tags] or "_timestamp_parse_failure" in [tags] {
|
||
|
|
stdout { codec => rubydebug }
|
||
|
|
}
|
||
|
|
}
|