Lab: Ingest pipeline and Data Prepper with correction

What you will learn in this TP

• Create and simulate a pipeline
• Deploy Data Prepper
• Type fields correctly

Introduction #

What you will learn in this section

• Create and simulate a pipeline
• Deploy Data Prepper
• Type fields correctly

Context. Your logs arrive poorly structured. You normalize them first inside the cluster, then with a dedicated collector.

Prerequisites: the previous lab  ·  Estimated time: 45 min

Clean, well-typed data.

Setup #

Guided steps, with their solution.

1. 1. Create and simulate a pipeline

   Why? _simulate avoids damaging data.

   Create the pipeline, then test it without indexing anything.
   See the correction

   curl -sk -u 'admin:Str0ng!Passw0rd-2026' -H 'Content-Type: application/json' -X PUT https://localhost:9200/_ingest/pipeline/logs_nettoyage -d '{
     "description": "Normalizes the level and extracts the user",
     "processors": [
       {"lowercase": {"field": "level"}},
       {"set": {"field": "environnement", "value": "prod"}},
       {"grok": {"field": "message", "patterns": ["utilisateur=%{WORD:utilisateur}"], "ignore_failure": true}}
     ]
   }'
   
   # Try without indexing anything
   curl -sk -u 'admin:Str0ng!Passw0rd-2026' -H 'Content-Type: application/json' -X POST https://localhost:9200/_ingest/pipeline/logs_nettoyage/_simulate \
     -d '{"docs":[{"_source":{"level":"ERROR","message":"refus utilisateur=alice"}}]}'

2. 2. Apply it

   Why? The pipeline is chosen at write time.

   Reindex the batch of the previous lab with the pipeline, and find the document of user alice.
   See the correction

   curl -sk -u 'admin:Str0ng!Passw0rd-2026' -H 'Content-Type: application/x-ndjson' \
     -X POST 'https://localhost:9200/_bulk?pipeline=logs_nettoyage&refresh=true' --data-binary @lot.ndjson
   curl -sk -u 'admin:Str0ng!Passw0rd-2026' -H 'Content-Type: application/json' -X POST 'https://localhost:9200/logs-app-*/_search' -d '{"query":{"term":{"utilisateur.keyword":"alice"}}}'

3. 3. Prepare the typing

   Why? Without a template, clientip would be text and network search would fail.

   Create the web-logs template.
   See the correction

   curl -sk -u 'admin:Str0ng!Passw0rd-2026' -H 'Content-Type: application/json' -X PUT https://localhost:9200/_index_template/web_logs -d '{
     "index_patterns": ["web-logs*"],
     "template": {
       "settings": {"number_of_shards": 1, "number_of_replicas": 0},
       "mappings": {"properties": {
         "@timestamp": {"type": "date"},
         "clientip": {"type": "ip"},
         "verb": {"type": "keyword"},
         "request": {"type": "keyword"},
         "response": {"type": "integer"},
         "bytes": {"type": "integer"}
       }}
     }
   }'

4. 4. Deploy Data Prepper

   Why? Processing leaves the cluster and relieves its nodes.

   Write pipelines.yaml, add the service and start it.
   See the correction

   log-pipeline:
     source:
       http:
         ssl: false
         port: 2021
         path: /log/ingest
     processor:
       - grok:
           match:
             log: ['%{COMMONAPACHELOG}']
       - date:
           match:
             - key: timestamp
               patterns: ["dd/MMM/yyyy:HH:mm:ss Z"]
           destination: "@timestamp"
       - convert_type:
           key: response
           type: integer
       - delete_entries:
           with_keys: ["log"]
     sink:
       - opensearch:
           hosts: ["https://opensearch:9200"]
           insecure: true
           username: admin
           password: "Str0ng!Passw0rd-2026"
           index: web-logs
           flush_timeout: 5000

   services:
     opensearch:
       image: opensearchproject/opensearch:3
       environment:
         discovery.type: single-node
         OPENSEARCH_INITIAL_ADMIN_PASSWORD: ${OPENSEARCH_INITIAL_ADMIN_PASSWORD}
         OPENSEARCH_JAVA_OPTS: -Xms1g -Xmx1g
         path.repo: /usr/share/opensearch/data/snapshots
       ports:
         - "9200:9200"
       volumes:
         - os-data:/usr/share/opensearch/data
   
     dashboards:
       image: opensearchproject/opensearch-dashboards:3
       environment:
         OPENSEARCH_HOSTS: '["https://opensearch:9200"]'
       ports:
         - "5601:5601"
       depends_on: [opensearch]
   
     data-prepper:
       image: opensearchproject/data-prepper:2
       ports:
         - "2021:2021"
       volumes:
         - ./pipelines.yaml:/usr/share/data-prepper/pipelines/pipelines.yaml:ro
       depends_on: [opensearch]
   
   volumes:
     os-data:

   docker compose up -d

5. 5. Send and check

   Why? Network search proves that the ip type is in place.

   Send two log lines, wait a few seconds, then search addresses in 10.0.0.0/8.
   See the correction

   curl -s -X POST -H 'Content-Type: application/json' http://localhost:2021/log/ingest -d '[
     {"log":"127.0.0.1 - frank [04/Nov/2026:10:00:00 +0100] \"GET /index.html HTTP/1.1\" 200 2326"},
     {"log":"10.0.0.8 - alice [04/Nov/2026:10:00:01 +0100] \"GET /admin HTTP/1.1\" 403 120"}
   ]'

   sleep 10
   curl -sk -u 'admin:Str0ng!Passw0rd-2026' -H 'Content-Type: application/json' -X POST https://localhost:9200/web-logs/_search -d '{"query":{"term":{"clientip":"10.0.0.0/8"}}}'
   curl -sk -u 'admin:Str0ng!Passw0rd-2026' -H 'Content-Type: application/json' https://localhost:9200/web-logs/_mapping

Difficulty level : ●●○○○ (2/5)