Log management pipeline in Elasticsearch - Mini Project: Build & Apply

Log Management Pipeline

📖 Scenario: You work as a system administrator managing server logs. You want to organize logs in Elasticsearch to quickly find errors and monitor system health.

🎯 Goal: Build a simple Elasticsearch index and pipeline to store logs, filter error logs, and add a timestamp field.

📋 What You'll Learn

Create an Elasticsearch index called server_logs with fields message and level

Define a pipeline that adds a timestamp field with the current time

Filter logs to only include those with level equal to error

Ingest sample logs using the pipeline

💡 Why This Matters

🌍 Real World

System administrators and DevOps engineers use Elasticsearch pipelines to organize and filter logs for monitoring and troubleshooting.

💼 Career

Understanding how to create indices and pipelines in Elasticsearch is essential for roles involving log management, monitoring, and data analysis.

1

Create the server_logs index

Create an Elasticsearch index called server_logs with two fields: message of type text and level of type keyword. Write the JSON mapping for this index.

Elasticsearch

{
  "mappings": {
    "properties": {
      // Define fields here
    }
  }
}

Need a hint?

Use mappings to define fields. message should be text for full-text search. level should be keyword for exact matching.

2

Define an ingest pipeline to add a timestamp

Create an ingest pipeline called add_timestamp that adds a timestamp field with the current date and time using the set processor.

Elasticsearch

{
  "mappings": {
    "properties": {
      "message": { "type": "text" },
      "level": { "type": "keyword" }
    }
  }
}

{
  "description": "Add current timestamp",
  "processors": [
    // Your code here
  ]
}

Need a hint?

Use the set processor to add a field. The value {{_ingest.timestamp}} inserts the current time.

3

Filter logs to only include errors

Add a pipeline processor to filter logs so only documents with level equal to error are processed further. Use the drop processor inside a conditional processor to drop non-error logs.

Elasticsearch

{
  "mappings": {
    "properties": {
      "message": { "type": "text" },
      "level": { "type": "keyword" }
    }
  }
}

{
  "description": "Add current timestamp and filter errors",
  "processors": [
    {
      "set": {
        "field": "timestamp",
        "value": "{{_ingest.timestamp}}"
      }
    },
    // Your code here
  ]
}

Need a hint?

Use the drop processor with an if condition to remove logs where level is not error.

4

Ingest sample logs using the pipeline

Index two sample log documents into the server_logs index using the add_timestamp pipeline. The first log has message "Disk full" and level "error". The second log has message "User login" and level "info".

Elasticsearch

{
  "mappings": {
    "properties": {
      "message": { "type": "text" },
      "level": { "type": "keyword" }
    }
  }
}

{
  "description": "Add current timestamp and filter errors",
  "processors": [
    {
      "set": {
        "field": "timestamp",
        "value": "{{_ingest.timestamp}}"
      }
    },
    {
      "drop": {
        "if": "ctx.level != 'error'"
      }
    }
  ]
}

// Your code here to index documents

Need a hint?

Use the POST method to index documents with the pipeline parameter set to add_timestamp.