Syncing Kubernetes Events to Elasticsearch for Troubleshooting

Hey there, Darian Vance here. As a Senior DevOps Engineer at TechResolve, I’ve spent more hours than I’d like to admit tailing logs and running kubectl get events -w in a dozen different terminals. I used to think of it as just part of the job, until I realized how much time I was losing. Chasing down a pod failure by manually correlating events across namespaces is a nightmare. That’s when I decided to build a better system: a simple, robust way to stream all Kubernetes events directly into Elasticsearch. Now, instead of hunting, I have a searchable, historical record. This setup has probably saved me a few hours a week and made our incident response way faster. Let me walk you through how I do it.

Prerequisites

Before we dive in, make sure you have the following ready to go. We’re keeping it simple, no complex frameworks needed.

• Python 3.8+ installed. Most modern systems have it, but it’s good to check.
• Access to a Kubernetes cluster. You’ll need a working kubeconfig file that your script can use to authenticate.
• An Elasticsearch instance. This can be on Elastic Cloud or a self-hosted cluster. You’ll need the host URL and an API Key or other credentials.
• A few Python libraries. We’ll need kubernetes, elasticsearch, and python-dotenv. I’ll assume you have a standard Python project workflow. You’d typically create a virtual environment and then run a pip command to install these packages, something like pip install kubernetes elasticsearch python-dotenv.

The Guide: Step-by-Step

Alright, let’s get this pipeline built. The core idea is to run a Python script that watches the Kubernetes API for new events and pushes them to an Elasticsearch index as they happen.

Step 1: Project Setup and Configuration

First, get your project folder organized. I usually have a simple structure: a main script file and a configuration file. I’ll skip the standard virtualenv setup since you likely have your own workflow for that. Let’s jump straight to the logic.

Create a file named config.env to hold our secrets and configuration. This keeps them out of the code, which is always a good practice.

# Elasticsearch Connection Details
ES_HOST="https://your-elastic-cloud-id.kb.us-central1.gcp.cloud.es.io:9243"
ES_API_KEY="YourBase64EncodedAPIKey"

# The name of the index where events will be stored
K8S_INDEX_NAME="kubernetes-events"

Step 2: The Python Script

Now for the main event. Create a file, let’s call it event_syncer.py. I’ll break down the code section by section so you understand what each part does.

First, we import the necessary libraries and load our configuration from the config.env file.

import os
import json
from dotenv import load_dotenv
from kubernetes import client, config, watch
from elasticsearch import Elasticsearch

# Load environment variables from config.env
load_dotenv('config.env')

# --- Configuration ---
ES_HOST = os.getenv("ES_HOST")
ES_API_KEY = os.getenv("ES_API_KEY")
INDEX_NAME = os.getenv("K8S_INDEX_NAME")

Next, we initialize the clients for both Kubernetes and Elasticsearch. The Kubernetes client will automatically try to load configuration from your kubeconfig file, which is perfect for local testing. The Elasticsearch client uses the API key for secure authentication.

def get_k8s_client():
    """Initializes and returns the Kubernetes API client."""
    try:
        config.load_kube_config()
        print("Successfully loaded kubeconfig.")
        return client.CoreV1Api()
    except Exception as e:
        print(f"Error loading Kubernetes configuration: {e}")
        return None

def get_es_client():
    """Initializes and returns the Elasticsearch client."""
    try:
        es_client = Elasticsearch(
            ES_HOST,
            api_key=ES_API_KEY
        )
        if es_client.ping():
            print("Successfully connected to Elasticsearch.")
            return es_client
        else:
            print("Could not connect to Elasticsearch.")
            return None
    except Exception as e:
        print(f"Error connecting to Elasticsearch: {e}")
        return None

Now, the core logic. We’ll create a function that watches for events and sends them to Elasticsearch. The watch.Watch() object gives us a continuous stream of events from the Kubernetes API. We loop through this stream, format each event into a clean dictionary, and then index it.

Pro Tip: Before you run this script for the first time, I strongly recommend setting up an Index Template in Kibana for your kubernetes-events-* index pattern. This lets you define mappings, like ensuring event.firstTimestamp is treated as a date field. It makes searching and visualizing the data much, much easier down the road.

def stream_k8s_events(k8s_api, es_client):
    """Watches for K8s events and sends them to Elasticsearch."""
    w = watch.Watch()
    print("Starting to stream Kubernetes events...")
    # Using '0' for resource_version tells the API to stream events from the beginning
    # In a production setup, you would store and reuse the last seen resourceVersion
    for event in w.stream(k8s_api.list_event_for_all_namespaces, resource_version='0'):
        try:
            event_data = event['object']
            
            # Create a structured document for Elasticsearch
            doc = {
                "timestamp": event_data.last_timestamp or event_data.first_timestamp,
                "message": event_data.message,
                "reason": event_data.reason,
                "type": event_data.type,
                "component": event_data.source.component,
                "host": event_data.source.host,
                "involved_object": {
                    "kind": event_data.involved_object.kind,
                    "name": event_data.involved_object.name,
                    "namespace": event_data.involved_object.namespace,
                    "uid": event_data.involved_object.uid
                },
                "metadata": {
                    "name": event_data.metadata.name,
                    "namespace": event_data.metadata.namespace,
                    "uid": event_data.metadata.uid,
                    "resource_version": event_data.metadata.resource_version
                }
            }
            
            # Use the event UID as the document ID to prevent duplicates
            doc_id = event_data.metadata.uid

            es_client.index(index=INDEX_NAME, id=doc_id, document=doc)
            print(f"Indexed event for {doc['involved_object']['kind']}/{doc['involved_object']['name']} in namespace {doc['involved_object']['namespace']}")

        except Exception as e:
            print(f"An error occurred while processing an event: {e}")
            # Optional: Add more robust error handling, like a retry mechanism
            continue

Finally, a simple main block to tie it all together and start the process.

if __name__ == "__main__":
    k8s_api_client = get_k8s_client()
    elasticsearch_client = get_es_client()

    if k8s_api_client and elasticsearch_client:
        stream_k8s_events(k8s_api_client, elasticsearch_client)
    else:
        print("Could not initialize clients. Exiting.")

Step 3: Running the Syncer

You can run this script directly from your terminal with python3 event_syncer.py. For a real production setup, I’d package this into a Docker container and run it as a Deployment within the Kubernetes cluster itself. That way, it’s managed by Kubernetes and automatically has in-cluster access to the API server.

Common Pitfalls (Where I Usually Mess Up)

I’ve set this up a few times, and here are the traps I’ve fallen into so you don’t have to:

• RBAC Permissions: This is the number one issue. If you run this script as a pod in the cluster, its ServiceAccount needs permissions to watch events. You’ll need a Role or ClusterRole with rules for events resources and verbs like get, list, and watch. Forgetting this leads to a stream of 403 Forbidden errors.
• Elasticsearch Connectivity: Double-check your ES_HOST and ES_API_KEY. A simple typo can cost you 30 minutes of debugging. Also, if your Elasticsearch instance is behind a firewall, ensure the cluster’s egress IP is allowed.
• Handling Restarts: The simple script I provided starts streaming from the beginning every time. In a real-world scenario, you should save the resource_version of the last event you processed to a file or a ConfigMap. When the script restarts, it can read that value and tell the Kubernetes API to only send events that have occurred since, preventing duplicate data.

Conclusion

And that’s it. With a fairly simple Python script, you’ve now got a powerful, centralized logging pipeline for your Kubernetes events. No more frantic `kubectl` commands during an outage. You can build Kibana dashboards to visualize event types, set up alerts for specific event reasons (like `FailedCreate` or `BackOff`), and have a full historical audit trail. It’s a small investment of time that pays huge dividends in operational sanity. Give it a shot, and feel free to adapt the script to your specific needs.

Cheers,

Darian Vance

Darian Vance

Lead Cloud Architect & DevOps Strategist

With over 12 years in system architecture and automation, Darian specializes in simplifying complex cloud infrastructures. An advocate for open-source solutions, he founded TechResolve to provide engineers with actionable, battle-tested troubleshooting guides and robust software alternatives.

🤖 Frequently Asked Questions