Skip to content

Instantly share code, notes, and snippets.

Last active December 6, 2018 11:20
What would you like to do?
Flamegraph django middleware
Django middleware for generating request flame graphs.
Requires the perl script:
1. Create a directory for flame graphs
2. Copy the script to it
3. Add the FLAMES_DIR django setting
4. Add the flames.FlamesMiddleware to MIDDLEWARE_CLASSES
To generate a flame graph just append ?flames to the requested url.
Middleware will create an svg in the FLAMES_DIR with the current timestamp.
Uncomment line 88 to automatically open the svg in a new google chrome tab.
import os
import subprocess
import sys
import threading
import time
import traceback
from datetime import datetime
from xml.dom.minidom import Text
from django.conf import settings
FLAMES_DIR = os.path.abspath(settings.FLAMES_DIR)
def get_module_name(module_path):
for path in sys.path:
path = path or os.getcwd()
if module_path.startswith(path):
rel_path = module_path[len(path) + 1:]
return (
.replace(u'/', u'')
.replace(u'/', '.')
.replace(u'.py', u'')
return module_path
def write_samples(file, samples):
for _, frame in samples:
stack = traceback.extract_stack(frame)
frame_strings = (
func_name + u'@' + get_module_name(module_path)
for module_path, _, func_name, _ in stack
stack_string = u';'.join(frame_strings)
file.write(u'{} {}\n'.format(stack_string, 1))
def write_flames(logger, title='flames'):
base_path = os.path.join(
out_txt_path = base_path + '.txt'
out_svg_path = base_path + '.svg'
title_element = Text() = title
title_xml = title_element.toxml()
with open(out_txt_path, 'w') as out_txt:
write_samples(out_txt, logger.samples)
with open(out_svg_path, 'w') as out_svg:
'perl', FLAMEGRAPH_SCRIPT_PATH, out_txt_path,
'--title', title_xml,
#['google-chrome', 'file://' + out_svg_path])
class StackLogger(object):
def __init__(self, thread_id, interval=0.001):
super(StackLogger, self).__init__()
self.thread_id = thread_id
self.interval = interval
self.should_stop = False
self.samples = []
self.started = threading.Event()
self.thread = threading.Thread(
def run(self):
start_time = time.time()
self.samples = []
while not self.should_stop:
frame_start = time.clock()
timestamp = time.time() - start_time
frame = sys._current_frames()[self.thread_id]
self.samples.append((timestamp, frame))
frame_dt = time.clock() - frame_start
time.sleep(max(0, self.interval - frame_dt))
def start(self):
self.should_stop = False
def stop(self):
self.should_stop = True
class FlamesMiddleware(object):
def process_request(self, request):
if settings.DEBUG and 'flames' in request.GET:
logger = StackLogger(threading.current_thread().ident)
request._stack_logger = logger
def process_response(self, request, response):
if settings.DEBUG and hasattr(request, '_stack_logger'):
logger = request._stack_logger
url = request.build_absolute_uri()
write_flames(logger, title=url)
return response
Copy link

Use this fork for >= django 1.10.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment