Step-by-Step: Diagnosing Memory leaks in long-running Python Celery worker daemons on OVH Servers
Identifying the Problem: Unexplained Memory Growth in Celery Workers
You’ve deployed a Python application using Celery for asynchronous task processing on OVH cloud servers. Over time, you notice that the memory footprint of your Celery worker processes steadily increases, eventually leading to OOM (Out Of Memory) killer interventions or performance degradation. This is a classic symptom of a memory leak, and diagnosing it in long-running daemons requires a systematic approach.
OVH servers, like any other infrastructure, can present unique challenges. While the core Python memory management principles remain the same, understanding how to instrument and monitor your processes within this environment is key. We’ll focus on practical steps using standard Python tools and system utilities.
Phase 1: Initial Monitoring and Baseline Establishment
Before diving deep, establish a clear baseline of your worker’s memory usage under normal load. This helps differentiate between a true leak and expected memory fluctuations due to task processing.
1. System-Level Monitoring (OVH Control Panel & `htop`)
Your first port of call should be the OVH control panel for your instance. Look for metrics like RAM usage, CPU load, and disk I/O. While these are high-level, they can indicate if the entire server is under strain or if a specific process is the culprit.
On the server itself, `htop` is an invaluable tool. It provides a real-time, interactive view of running processes. Sort by memory usage (press ‘M’) to quickly identify your Celery worker processes (often named `celery worker -A …`). Note the `RES` (Resident Set Size) and `VIRT` (Virtual Memory Size) columns. A steadily increasing `RES` is a strong indicator of a leak.
To get historical data, consider setting up a simple monitoring agent or using a tool like `collectd` or Prometheus Node Exporter. For this guide, we’ll focus on on-demand diagnostics.
2. Celery Worker Logging
Ensure your Celery workers are configured with adequate logging. This includes logging task execution times, any exceptions, and potentially custom debug information. While not directly for memory, logs can correlate memory spikes with specific task types.
A typical Celery configuration snippet for logging might look like this (in your `celeryconfig.py` or similar):
import logging
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
CELERYD_LOG_FILE = '/var/log/celery/worker.log'
CELERYD_LOG_LEVEL = 'INFO'
Phase 2: In-Process Memory Profiling
Once you’ve confirmed a memory issue with system tools, you need to inspect the Python process itself. This involves using profiling tools to understand which objects are consuming memory and why they aren’t being garbage collected.
1. Using `objgraph` for Object Tracking
`objgraph` is a fantastic library for visualizing and debugging Python’s memory usage. It allows you to inspect the reference graph of objects, helping you find what’s keeping them alive.
First, install it:
pip install objgraph
To use `objgraph` effectively, you’ll typically attach it to your running worker process. This can be done by:
- Modifying your worker’s startup script to import `objgraph` and periodically dump object counts or references.
- Attaching a debugger (like `pdb` or `ipdb`) to a running worker and then importing `objgraph` interactively.
- Using `gcore` to get a core dump of the process and then analyzing it with `gdb` and Python extensions (more advanced).
Let’s illustrate with a simple approach: adding diagnostic code to your worker’s task execution. This requires modifying your application code, so ensure you have a way to deploy these changes to your workers.
In one of your frequently executed tasks, or a task that you suspect might be related to the leak, add the following:
import objgraph
import gc
import os
import time
# Assume this is part of your tasks.py or similar
from celery import Celery
app = Celery('my_app', broker='redis://localhost:6379/0')
@app.task
def my_leaky_task():
# ... your task logic ...
# Periodically check memory, e.g., every 100 calls or based on time
if getattr(my_leaky_task, 'call_count', 0) % 100 == 0:
pid = os.getpid()
print(f"PID: {pid}, Task Call Count: {my_leaky_task.call_count}")
# Force garbage collection to get a cleaner snapshot
gc.collect()
# Get top 10 most common objects
top_objects = objgraph.most_common_types(limit=10)
print(f"Top 10 object types: {top_objects}")
# If you suspect a specific type of object is leaking (e.g., custom classes)
# You can track them specifically:
# print(objgraph.by_type('MyLeakyClass'))
# To visualize the reference chain of a specific object (requires graphviz)
# Find a problematic object instance and then:
# objgraph.show_refs([obj_instance], filename='refs.png')
# objgraph.show_backrefs([obj_instance], filename='backrefs.png')
my_leaky_task.call_count = getattr(my_leaky_task, 'call_count', 0) + 1
time.sleep(1) # Simulate work
return "Task completed"
# In your worker startup script or a separate diagnostic script:
# You might want to run this periodically or attach it to a running process.
# For a running process, you could use signal handlers or a management command.
# Example: A simple script to attach and inspect a running worker PID
# import objgraph
# import gc
# import sys
#
# pid_to_inspect = int(sys.argv[1]) # Pass PID as argument
#
# try:
# objgraph.attach(pid_to_inspect)
# gc.collect()
# print(objgraph.count('MyLeakyClass')) # Count instances of a specific class
# print(objgraph.show_most_common_types(limit=20))
# except Exception as e:
# print(f"Error attaching or profiling: {e}")
# finally:
# objgraph.detach() # Important to detach if attached dynamically
Run your workers with this modified code. After some time, check your worker logs for the `objgraph` output. Look for object types whose counts are consistently increasing without bound. If you can identify a specific custom class or a standard library object that’s growing, you’re on the right track.
2. Using `memory_profiler`
`memory_profiler` is excellent for line-by-line memory usage analysis within a function. It’s less about object graphs and more about pinpointing which lines of code are allocating memory.
Install it:
pip install memory_profiler
Decorate your suspect task function with `@profile` and run the script using `mprof run`.
# In your tasks.py
from celery import Celery
from memory_profiler import profile
import time
import os
app = Celery('my_app', broker='redis://localhost:6379/0')
@profile # Add this decorator
def my_potentially_leaky_task():
# Simulate some work that might allocate memory
data = []
for i in range(10000):
data.append(f"Item {i} - " + "A" * 100) # Allocating strings
if i % 1000 == 0:
time.sleep(0.1)
print(f"Task finished. PID: {os.getpid()}")
return "Done"
# To run this with memory_profiler, you'd typically run it as a standalone script
# for testing, or integrate it carefully with Celery.
# A common pattern is to run a single task with mprof for debugging.
#
# Example of how you might run a single task for profiling:
#
# 1. Save the above code as tasks.py
# 2. Run from your terminal:
# mprof run tasks.py my_potentially_leaky_task
# (Note: This requires adapting Celery's execution model or running the task
# directly for profiling purposes, which might not perfectly replicate
# the daemon environment. A better approach is often to attach mprof to
# a running worker, but that's more complex.)
#
# A more practical approach for a running worker:
# You can use `mprof attach ` to attach to a running process.
#
# After running `mprof run` or `mprof attach`, generate the report:
# mprof plot
# mprof peak
The `mprof plot` command will generate a graph showing memory usage over time. Look for a consistently upward trend that doesn’t flatten out. `mprof peak` will show the lines of code that consumed the most memory during the run.
Phase 3: Analyzing the Root Cause
Once you’ve identified *what* is leaking (e.g., a list of objects, a cache, a specific class instance), you need to understand *why* it’s not being released.
1. Circular References and `gc`
Python’s garbage collector (GC) handles most memory management. However, circular references (object A refers to B, and B refers to A) can sometimes be tricky, especially if they involve objects with `__del__` methods. The `gc` module can help detect these.
import gc
# After collecting objects with objgraph or memory_profiler
# Force a collection and look for uncollectable objects
gc.collect()
uncollectable = gc.garbage
if uncollectable:
print(f"Found {len(uncollectable)} uncollectable objects:")
for obj in uncollectable:
print(f"- {type(obj)}: {obj}")
# You can use objgraph to inspect these further
# import objgraph
# objgraph.show_refs([obj], filename='uncollectable_refs.png')
else:
print("No uncollectable objects found.")
If `gc.garbage` contains objects, it means the GC couldn’t reclaim them. Inspecting these objects might reveal the circular references or other issues preventing their cleanup.
2. Caching Mechanisms
Many applications implement in-memory caches to speed up repeated computations or data retrieval. If these caches are not bounded (i.e., they grow indefinitely) or don’t have proper eviction policies, they are prime candidates for memory leaks.
Review your code for any dictionary, list, or custom cache objects that store results. Ensure they have a maximum size or an expiration mechanism. Libraries like `functools.lru_cache` can be helpful, but even they have limits.
from functools import lru_cache
import time
@lru_cache(maxsize=128) # Limit cache to 128 most recent calls
def expensive_computation(arg1, arg2):
print(f"Computing for {arg1}, {arg2}...")
time.sleep(1) # Simulate work
return arg1 + arg2
# In your task:
# result = expensive_computation(10, 20)
If you’re using a custom cache, ensure it has a mechanism to remove old entries. For example:
class BoundedCache:
def __init__(self, max_size=1000):
self.cache = {}
self.max_size = max_size
self.order = [] # To track insertion order for LRU
def get(self, key):
if key in self.cache:
# Move to end of order (most recently used)
self.order.remove(key)
self.order.append(key)
return self.cache[key]
return None
def set(self, key, value):
if key not in self.cache and len(self.cache) >= self.max_size:
# Remove least recently used item
lru_key = self.order.pop(0)
del self.cache[lru_key]
self.cache[key] = value
self.order.append(key)
# Use this cache within your tasks.
3. External Resource Handles
Sometimes, memory leaks aren’t directly in Python objects but in unclosed file handles, network sockets, database connections, or other resources managed by underlying C libraries that Python interacts with. Ensure all such resources are properly closed, ideally using `with` statements (context managers).
# Example with file handling
try:
with open('large_data.txt', 'r') as f:
content = f.read()
# Process content
except FileNotFoundError:
pass # Handle error
# Example with database connections (using a hypothetical library)
# Ensure your DB connection pool or individual connections are managed.
# Many ORMs handle this, but custom queries might require manual closing.
#
# with db_connection.cursor() as cursor:
# cursor.execute("SELECT ...")
# results = cursor.fetchall()
4. Third-Party Libraries
The leak might originate from a third-party library you’re using. If `objgraph` points to objects from a specific library, check its issue tracker, documentation, and consider updating to the latest version. Sometimes, a specific version might have a known memory leak.
Phase 4: Mitigation and Prevention
Once the leak is identified and a fix is implemented, consider these strategies to prevent recurrence:
- Automated Profiling: Integrate memory profiling into your CI/CD pipeline. Tools like `memory_profiler` can be run in a test environment to catch regressions.
- Health Checks: Implement periodic health checks for your Celery workers that include memory usage thresholds. If a worker exceeds a certain memory limit, it can be automatically restarted.
- Code Reviews: Foster a culture of careful code review, specifically looking for potential memory management issues, especially around caching and resource handling.
- Worker Restart Strategy: Even with fixes, long-running processes can accumulate fragmentation. Implement a rolling restart strategy for your Celery workers (e.g., restart one worker at a time every 24 hours) to periodically reset their memory state. This can be managed by your process supervisor (like `systemd` or `supervisor`).
Example `systemd` Service for Rolling Restarts
To manage worker restarts, `systemd` is commonly used on modern Linux systems. Here’s a basic service file that could be extended with restart logic (though true rolling restarts often require external orchestration or custom scripts).
[Unit] Description=Celery Worker Daemon After=network.target [Service] Type=simple User=your_user Group=your_group WorkingDirectory=/path/to/your/app ExecStart=/usr/bin/python3 -m celery worker -A your_app --loglevel=info --concurrency=4 --pidfile=/var/run/celery/%n.pid --logfile=/var/log/celery/%n%I.log Restart=on-failure RestartSec=5 # For rolling restarts, you'd typically have a separate script that # signals workers to gracefully shut down and then restarts them one by one. # This service file itself doesn't implement rolling restarts directly. [Install] WantedBy=multi-user.target
Remember to replace placeholders like `your_user`, `your_app`, and paths with your actual configuration. For sophisticated rolling restarts, consider tools like Kubernetes, Docker Swarm, or custom orchestration scripts that manage worker lifecycle events.