Stream Python Logs to Elasticsearch for Better Monitoring

Logging is one of those things every Python developer knows they should do properly, but most of us end up with a messy pile of text files that we only look at when something breaks. I've been there too, scrolling through hundreds of lines in a terminal window, trying to find the one error that happened three hours ago.

But there's a better way. Instead of treating your logs like digital clutter, you can stream them directly to Elasticsearch and turn them into something you can actually search, visualize, and get alerts from. Let me show you how.

Why Elasticsearch for Logs?

Before we jump into the code, let me give you a real scenario. At PythonSkillset, we had a Django application running in production. Every day, it generated about 200MB of log data. When something went wrong, we had to SSH into servers, grep through files, and pray we found the right line. It was slow and painful.

Elasticsearch changes that completely. You get: - Instant search across all your logs, no matter how many there are - Visual dashboards in Kibana that show you trends and patterns - Real-time alerts when error rates spike - No more lost logs when servers crash or rotate files

The Setup

First, you'll need Elasticsearch running. For testing, Docker makes this easy:

docker run -d --name elasticsearch \
  -p 9200:9200 \
  -e "discovery.type=single-node" \
  docker.elastic.co/elasticsearch/elasticsearch:8.11.0

Make sure it's responding:

curl http://localhost:9200

Python Logging with Elasticsearch

Now here's where the magic happens. Instead of writing logs to files, we'll send them straight to Elasticsearch using the python-json-logger and elasticsearch libraries.

Install what you need:

pip install python-json-logger elasticsearch

Here's a complete example you can adapt for your own projects:

import logging
from pythonjsonlogger import jsonlogger
from elasticsearch import Elasticsearch

# Connect to Elasticsearch
es = Elasticsearch(['http://localhost:9200'])

# Create a custom handler
class ElasticsearchHandler(logging.Handler):
    def __init__(self, es_client, index_prefix="logs"):
        super().__init__()
        self.es = es_client
        self.index_prefix = index_prefix

    def emit(self, record):
        try:
            # Format the log entry as JSON
            log_entry = self.format(record)

            # Send to Elasticsearch with a daily index pattern
            index_name = f"{self.index_prefix}-{record.created:.0f}"
            self.es.index(
                index=index_name,
                document=log_entry
            )
        except Exception as e:
            # Fallback to stderr so we don't lose errors
            print(f"Failed to send log to ES: {e}", file=sys.stderr)

# Set up the logger
logger = logging.getLogger("my_app")
logger.setLevel(logging.INFO)

# Use JSON formatter for structured logs
formatter = jsonlogger.JsonFormatter(
    fmt='%(asctime)s %(name)s %(levelname)s %(message)s'
)

# Add our Elasticsearch handler
es_handler = ElasticsearchHandler(es)
es_handler.setFormatter(formatter)
logger.addHandler(es_handler)

# Test it
logger.info("Application started", extra={"version": "1.0.0"})
logger.error("Something went wrong", extra={"error_code": 500})

Making It Production-Ready

The example above works, but in production you'll want to handle a few things better:

1. Index management: Don't let indexes grow forever. Use Elasticsearch's Index Lifecycle Management (ILM) to roll over indexes daily and delete old ones.

2. Batching: Sending one request per log entry is slow. Use a buffered approach:

from collections import deque
import threading
import time

class BufferedElasticsearchHandler(logging.Handler):
    def __init__(self, es_client, buffer_size=100, flush_interval=5):
        super().__init__()
        self.es = es_client
        self.buffer = deque()
        self.buffer_size = buffer_size
        self.flush_interval = flush_interval
        self.lock = threading.Lock()

        # Start background flusher
        self.timer = threading.Thread(target=self._flush_periodically, daemon=True)
        self.timer.start()

    def emit(self, record):
        log_entry = self.format(record)
        with self.lock:
            self.buffer.append(log_entry)
            if len(self.buffer) >= self.buffer_size:
                self._flush()

    def _flush(self):
        if not self.buffer:
            return
        # Bulk index all buffered entries
        body = []
        for entry in self.buffer:
            body.append({"index": {"_index": "logs"}})
            body.append(entry)
        try:
            self.es.bulk(body=body)
            self.buffer.clear()
        except Exception as e:
            print(f"Bulk flush failed: {e}", file=sys.stderr)

    def _flush_periodically(self):
        while True:
            time.sleep(self.flush_interval)
            with self.lock:
                self._flush()

3. Structured logging: Always include context. Instead of "User logged in", log {"action": "login", "user_id": 123, "ip": "192.168.1.1"}. This makes searching incredibly powerful later.

What You'll See in Kibana

Once logs start flowing, head to Kibana (usually port 5601). Create an index pattern for your logs, and you'll get:

• A search bar where you can type level: ERROR to see only errors
• Time-based histograms showing when errors spike
• Tables with every field searchable and sortable
• Ability to create dashboards that combine multiple log sources

A Few Tips I've Learned

• Don't log sensitive data - no passwords, tokens, or personal info. Elasticsearch is searchable by anyone with access.
• Use log levels properly - INFO for normal operations, WARNING for recoverable issues, ERROR for failures, CRITICAL for system-breaking problems.
• Rate limit if needed - If you have a sudden burst of errors, you don't want to overwhelm Elasticsearch. Add a rate limiter to your handler.
• Test locally first - Run Elasticsearch in Docker and test your logging before touching production.

Wrapping Up

Streaming logs to Elasticsearch transformed how we debug at PythonSkillset. What used to take hours of digging through files now takes seconds of searching. And when something breaks at 2 AM, you can actually find the problem without waking up your whole team.

The code here gives you a solid foundation, but don't be afraid to customize it for your specific needs. Logging is deeply personal - what works for one application might not work for another. Start simple, monitor how it performs, and iterate.

Your future self, debugging at 3 AM, will thank you.