I built an inventory system for warehouse operators. Non-technical people. When something breaks, the conversation goes like this:
"It's not working."
"What does the error say?"
"I don't know, there's no error on screen."
They don't have SSH access. They shouldn't need it. So I added a log console directly in the web interface.
This post walks through the implementation: a ring buffer on the backend, a simple REST endpoint, and browser polling. No WebSocket, no external services, just Python and vanilla JavaScript.
Alternatives I considered
Before settling on polling, I looked at a few options:
WebSocket — works great for bidirectional communication, but here I only need server-to-client. Felt like overkill, plus it adds connection state management.
Server-Sent Events (SSE) — actually a good fit for one-way streaming. But I've had SSE connections dropped by certain corporate proxies and load balancers. Debugging that remotely is not fun.
External logging service (Papertrail, Logtail, etc.) — solid products, but I wanted something self-contained. No external dependencies, no additional costs, no data leaving the server.
Polling — boring, but works everywhere. Every proxy, every browser, every network config. For operational logs where 2-second delay is acceptable, I went with boring.
How it fits together
┌---------------------------------------------------------┐
| FLASK APP |
| |
| logging.info() --> BufferHandler --> LogBuffer |
| (ring buffer) |
| | |
| GET /api/logs?since_id=N |
└---------------------------------------------------------┘
|
JSON response
|
▼
┌---------------------------------------------------------┐
| BROWSER |
| |
| setInterval(2s) --> fetch() --> append to DOM |
| |
| lastSeenId: 47 ----------------------------------> |
| (sent as since_id parameter) |
└---------------------------------------------------------┘
The key idea: client tracks the last log ID it received, asks only for newer entries. Server doesn't need to track client state.
Ring buffer with thread safety
Flask can handle requests in multiple threads, so the buffer needs locking. Python's deque with a max length handles the "ring" part automatically — when it's full, oldest entry gets dropped.
import logging
from collections import deque
from datetime import datetime
import threading
class LogBuffer:
def __init__(self, max_size=500):
self.buffer = deque(maxlen=max_size)
self.lock = threading.Lock()
self.last_id = 0
def add(self, level, message):
with self.lock:
self.last_id += 1
self.buffer.append({
'id': self.last_id,
'time': datetime.now().strftime('%H:%M:%S'),
'level': level,
'message': message
})
def get_logs(self, since_id=0):
with self.lock:
return [log for log in self.buffer if log['id'] > since_id]
def clear(self):
with self.lock:
self.buffer.clear()
log_buffer = LogBuffer(max_size=500)
One detail worth mentioning: last_id never resets, even when you clear the buffer. If the client remembers since_id=47 and you reset the counter to 0, things get weird — client keeps asking for logs newer than 47, server has nothing above 0. Keeping the counter monotonic avoids that edge case.
Hooking into Python's logging
A custom handler grabs every log message and pushes it to the buffer:
class BufferHandler(logging.Handler):
def __init__(self, log_buffer):
super().__init__()
self.log_buffer = log_buffer
def emit(self, record):
try:
message = self.format(record)
level = record.levelname.lower()
# mapping to CSS class names used in frontend
level_map = {
'debug': 'info',
'info': 'info',
'warning': 'warning',
'error': 'error',
'critical': 'error'
}
self.log_buffer.add(level_map.get(level, 'info'), message)
except Exception:
self.handleError(record)
Attaching it to the root logger in Flask app:
import logging
from log_buffer import log_buffer, BufferHandler
buffer_handler = BufferHandler(log_buffer)
buffer_handler.setFormatter(logging.Formatter('%(name)s - %(message)s'))
logging.getLogger().addHandler(buffer_handler)
After this, every logger.info("something") call anywhere in the app ends up in the buffer.
The API endpoint
Client passes the last ID it received. Server returns only newer entries.
from flask import Blueprint, request, jsonify
from log_buffer import log_buffer
api = Blueprint('api', __name__)
@api.route('/api/get-server-logs')
def get_server_logs():
since_id = request.args.get('since_id', 0, type=int)
logs = log_buffer.get_logs(since_id)
return jsonify({'success': True, 'logs': logs})
@api.route('/api/clear-server-logs', methods=['POST'])
def clear_server_logs():
log_buffer.clear()
return jsonify({'success': True})
A typical exchange looks like:
GET /api/get-server-logs?since_id=0 -> returns logs with id 1, 2, 3
GET /api/get-server-logs?since_id=3 -> returns logs with id 4, 5
GET /api/get-server-logs?since_id=5 -> returns empty list (nothing new)
No pagination tokens, no cursor management. Just an integer that only goes up.
Frontend polling
let lastServerLogId = 0;
let pollingInterval = null;
async function fetchServerLogs() {
try {
const response = await fetch(
`/api/get-server-logs?since_id=${lastServerLogId}`
);
const data = await response.json();
if (data.success && data.logs.length > 0) {
data.logs.forEach(log => {
appendLogEntry(log);
lastServerLogId = Math.max(lastServerLogId, log.id);
});
}
} catch (err) {
console.error('Log fetch failed:', err);
}
}
function appendLogEntry(log) {
const container = document.getElementById('logConsole');
const div = document.createElement('div');
div.className = 'log-entry ' + log.level;
div.textContent = '[' + log.time + '] ' + log.message;
container.insertBefore(div, container.firstChild);
// keep DOM size bounded, same as backend buffer
while (container.children.length > 500) {
container.removeChild(container.lastChild);
}
}
function startPolling() {
if (!pollingInterval) {
fetchServerLogs();
pollingInterval = setInterval(fetchServerLogs, 2000);
}
}
function stopPolling() {
if (pollingInterval) {
clearInterval(pollingInterval);
pollingInterval = null;
}
}
// pause when tab is hidden, resume when visible
document.addEventListener('visibilitychange', () => {
document.hidden ? stopPolling() : startPolling();
});
document.addEventListener('DOMContentLoaded', startPolling);
The HTML and CSS
<div class="log-console-wrapper">
<div class="log-header">
<span>Server Logs</span>
<button onclick="clearLogs()">Clear</button>
</div>
<div id="logConsole" class="log-console"></div>
</div>
.log-console {
background: #1e1e1e;
color: #d4d4d4;
font-family: Consolas, Monaco, monospace;
font-size: 12px;
padding: 10px;
height: 300px;
overflow-y: auto;
border-radius: 4px;
}
.log-entry {
padding: 2px 0;
border-bottom: 1px solid #333;
}
.log-entry.warning { color: #dcdcaa; }
.log-entry.error { color: #f48771; }
async function clearLogs() {
await fetch('/api/clear-server-logs', { method: 'POST' });
document.getElementById('logConsole').innerHTML = '';
}
Trade-offs worth knowing
Polling interval. I use 2 seconds. Shorter interval means more requests and slightly faster updates. Longer interval saves resources but feels less "live". For server logs, 2 seconds felt right — fast enough to watch things happen, slow enough to not hammer the server.
Buffer size. 500 entries is a reasonable default - order of magnitude small enough to keep memory usage low. Measure actual usage in your environment. That covers maybe 10-15 minutes of moderate activity in my case. Increase if you need longer history, but watch memory on long-running processes.
No persistence. When the Flask process restarts, logs are gone. If you need persistence, write to a file or database separately. This buffer is for live viewing only.
Single server only. If you run multiple Flask instances behind a load balancer, each has its own buffer. Client might hit different instances and see inconsistent logs. For multi-instance setups, you'd need a shared store (Redis, etc.) — which defeats the "no external dependencies" goal.
Gotchas I ran into
Thread safety on the counter. First version didn't have the lock. Under load, I occasionally saw duplicate IDs. self.last_id += 1 looks atomic but isn't — it's actually read-increment-write, and two threads can read the same value.
DOM bloat. Without the while (children.length > 500) cleanup, the browser slows down after a few hours of heavy logging. The ring buffer keeps the backend bounded; need to do the same on the frontend.
Clearing logs and monotonic IDs. Initially I reset last_id to 0 on clear. Then the client, still holding since_id=200, would get empty responses until the counter caught back up to 200. Took me a while to figure out why "clear" seemed to break log streaming.
Is this production-ready?
It's been running in my system for about six months, handling around 50 retail locations. Memory stays flat, no crashes, operators actually use it to troubleshoot before calling me.
For internal tools, single-user dashboards, or admin panels — this approach is solid. If you need sub-second latency, multiple concurrent viewers, or log aggregation across services, look into SSE, WebSocket, or a proper logging stack.
File structure
project/
├--log_buffer.py # LogBuffer class + BufferHandler
├-- routes.py # Flask API endpoints
├-- app.py # Flask app, attaches BufferHandler
├-- static/
│ └-- js/
│ └-- logs.js # Polling logic
└-- templates/
└-- index.html # Console markup + CSS
Tags: #python, #flask, #logging, #webdev, #tutorial
Top comments (0)