The Log Pipeline: Fluent Bit, Logstash, Kibana, and Friends

1. 1

   An ingest pipeline that parses and masks

   Parse a JSON message, set the timestamp, and mask anything that looks like a card number before it's stored (DevSecOps course, secrets and sensitive data).

   terminalcopy cmd

   $ curl -s -XPUT localhost:9200/_ingest/pipeline/shop-logs -H 'content-type: application/json' -d '{

   "processors": [

   { "json": { "field": "log", "target_field": "app", "ignore_failure": true } },

   { "date": { "field": "app.ts", "formats": ["ISO8601"], "ignore_failure": true } },

   { "gsub": { "field": "app.msg", "pattern": "\\b\\d{13,16}\\b", "replacement": "[CARD]", "ignore_missing": true } },

   { "remove": { "field": "log", "ignore_missing": true } } ] }'

   curl -s -XPOST 'localhost:9200/_ingest/pipeline/shop-logs/_simulate' -H 'content-type: application/json' -d '{"docs":[{"_source":{"log":"{\"ts\":\"2026-09-27T10:00:00Z\",\"level\":\"error\",\"msg\":\"charge failed for 4111111111111111\"}"}}]}' | jq '.docs[0].doc._source'

   ── expected output ──

   {

   "app": { "ts": "2026-09-27T10:00:00Z", "level": "error", "msg": "charge failed for [CARD]" },

   "@timestamp": "2026-09-27T10:00:00.000Z"

   }

2. 2

   An index template that prevents mapping explosions

   Known fields get explicit types; everything else under app.extra is stored as one flat_object field; unknown top-level fields aren't indexed.

   {
     "index_patterns": ["logs-shop-*"],
     "template": {
       "settings": { "number_of_shards": 1, "number_of_replicas": 1, "refresh_interval": "30s",
                     "default_pipeline": "shop-logs", "index.mapping.total_fields.limit": 500 },
       "mappings": {
         "dynamic": false,
         "properties": {
           "@timestamp": { "type": "date" },
           "kubernetes": { "properties": {
             "namespace_name": { "type": "keyword" }, "pod_name": { "type": "keyword" },
             "labels": { "type": "flat_object" } } },
           "app": { "properties": {
             "level": { "type": "keyword" }, "service": { "type": "keyword" },
             "trace_id": { "type": "keyword" }, "msg": { "type": "text" },
             "extra": { "type": "flat_object" } } }
         }
       }
     }
   }

   terminalcopy cmd

   $ curl -s -XPUT localhost:9200/_index_template/logs-shop -H 'content-type: application/json' -d @logs-template.json | jq .acknowledged

   ── expected output ──

   true

3. 3

   Fluent Bit as a DaemonSet

   Install with the Helm chart and give it this output config: filesystem buffering, retries, and the Kubernetes filter for pod metadata. The Logstash_Format-style daily indexes are avoided; we write to the rollover alias from Unit 4.2.

   config:
     service: |
       [SERVICE]
           Flush                     5
           storage.path              /var/fluent-bit/state/buffer
           storage.max_chunks_up     128
     inputs: |
       [INPUT]
           Name              tail
           Path              /var/log/containers/*.log
           multiline.parser  docker, cri
           Tag               kube.*
           storage.type      filesystem
           Mem_Buf_Limit     50MB
     filters: |
       [FILTER]
           Name        kubernetes
           Match       kube.*
           Labels      On
           Annotations Off
     outputs: |
       [OUTPUT]
           Name                      opensearch
           Match                     kube.*
           Host                      opensearch.logging.svc
           Port                      9200
           Index                     logs-shop
           Suppress_Type_Name        On
           Retry_Limit               False
           storage.total_limit_size  5G

   terminalcopy cmd

   $ helm repo add fluent https://fluent.github.io/helm-charts

   helm upgrade --install fluent-bit fluent/fluent-bit -n logging --create-namespace -f fluent-bit-values.yaml

   kubectl -n logging get ds fluent-bit

   ── expected output ──

   NAME DESIRED CURRENT READY UP-TO-DATE AVAILABLE

   fluent-bit 3 3 3 3 3

4. 4

   Query logs like an operator

   In Dashboards' Discover, or via the API: errors from checkout in the last 15 minutes, grouped by pod. The same query in PPL/SQL is available in OpenSearch (source=logs-shop | where app.level='error').

   terminalcopy cmd

   $ curl -s localhost:9200/logs-shop/_search -H 'content-type: application/json' -d '{"size":0,"query":{"bool":{"filter":[{"term":{"app.level":"error"}},{"term":{"kubernetes.labels.app":"checkout"}},{"range":{"@timestamp":{"gte":"now-15m"}}}]}},"aggs":{"pods":{"terms":{"field":"kubernetes.pod_name"}}}}' | jq -c '.aggregations.pods.buckets'

   ── expected output ──

   [{"key":"checkout-6d8f9c7b5-x4k2p","doc_count":412},{"key":"checkout-6d8f9c7b5-q9w7n","doc_count":3}]

Each drill is a real failure mode. Read the scenario and the output, decide what's wrong, then reveal the diagnosis.