How to use Logstash filters for data enrichment

Data enrichment in Logstash adds context to incoming events so searches, dashboards, and alert rules can group and filter activity without expensive query-time parsing. Consistent enrichment improves correlation across data sources and reduces the amount of custom logic needed at query time.

Enrichment happens in the pipeline filter stage, where plugins such as geoip and mutate can add, normalize, and reshape fields before events are sent to outputs like Elasticsearch. Using structured field references (for example [client][ip]) produces nested JSON that aligns cleanly with common schemas and mappings.

Enrichment increases processing overhead and can create indexing failures when field types change (for example, switching a field from a string to an object). Keep enrichment deterministic, scope filters with conditionals, validate configuration syntax before applying changes, and verify results in the destination index.

1. Add enrichment filters to the pipeline configuration in /etc/logstash/conf.d/43-enrich.conf.

   input {
     file {
       path => "/var/lib/logstash/examples/enrich.log"
       start_position => "beginning"
       sincedb_path => "/var/lib/logstash/sincedb-enrich"
       codec => json
     }
   }
   
   filter {
     if [log][file][path] == "/var/lib/logstash/examples/enrich.log" {
       if [client][ip] {
         geoip {
           source => "[client][ip]"
           target => "[client]"
         }
       }
   
       mutate {
         add_field => { "[event][category]" => "web" }
       }
     }
   }
   
   output {
     if [log][file][path] == "/var/lib/logstash/examples/enrich.log" {
       elasticsearch {
         hosts => ["http://elasticsearch.example.net:9200"]
         index => "logstash-enrich-%{+YYYY.MM.dd}"
         manage_template => false
       }
     }
   }

   The default pipeline reads /etc/logstash/conf.d in lexical order; adjust the source field reference to match the IP field present in events (for example [client][ip], [source][ip], or client_ip).

   Changing a field from a scalar to an object (for example client as a string in older indices, then client.geo as an object) can trigger mapping conflicts and cause documents to be rejected in Elasticsearch.

2. Test the pipeline configuration for syntax errors.

   $ sudo -u logstash /usr/share/logstash/bin/logstash --path.settings /etc/logstash --path.data /tmp/logstash-configtest --config.test_and_exit
   Configuration OK

   Configuration OK confirms valid syntax, not successful runtime enrichment.

   Related: How to test a Logstash pipeline configuration

3. Restart the Logstash service to apply the enrichment filters.

   $ sudo systemctl restart logstash

   Restarting logstash briefly pauses ingestion and may delay in-flight events depending on input buffering and queue settings.

   Related: How to manage the Logstash service with systemctl in Linux

4. Check that the logstash service is active.

   $ sudo systemctl status logstash --no-pager -l
   ● logstash.service - logstash
        Loaded: loaded (/usr/lib/systemd/system/logstash.service; enabled; preset: enabled)
        Active: active (running) since Wed 2026-01-07 22:52:19 UTC; 4s ago
      Main PID: 41476 (java)
         Tasks: 31 (limit: 28486)
        Memory: 525.8M (peak: 525.8M)
           CPU: 16.642s
   ##### snipped #####

   Related: How to manage the Logstash service with systemctl in Linux

5. Verify that indexed events contain the enriched fields in Elasticsearch.

   $ curl -s -H "Content-Type: application/json" -X POST "http://elasticsearch.example.net:9200/logstash-enrich-*/_search?pretty" -d '{
     "size": 1,
     "sort": [
       { "@timestamp": "desc" }
     ],
     "_source": [
       "client.ip",
       "client.geo.country_name",
       "client.geo.region_name",
       "client.geo.city_name",
       "client.geo.location",
       "event.category"
     ],
     "query": { "exists": { "field": "client.geo.location" } }
   }'
   {
     "took" : 8,
     "timed_out" : false,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : {
         "value" : 1,
         "relation" : "eq"
       },
       "max_score" : null,
       "hits" : [
         {
           "_index" : "logstash-enrich-2026.01.07",
           "_id" : "SP2pmpsBMfcBipKWYBE2",
           "_score" : null,
           "_source" : {
             "client" : {
               "geo" : {
                 "location" : {
                   "lat" : 37.751,
                   "lon" : -97.822
                 },
                 "country_name" : "United States"
               },
               "ip" : "8.8.8.8"
             },
             "event" : {
               "category" : "web"
             }
           },
           "sort" : [
             1767826349739
           ]
         }
       ]
     }
   }

   Adjust the URL and index pattern (logstash-enrich-*) to match the target cluster and indices.