cyberyan_linkedin_search/elk/logstash/pipeline/logstash.conf

# =============================================================================
# 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 }
  }
}