When maintaining production web servers, we often face a dilemma: how to test configuration changes without impacting live customer traffic. The ideal solution would be real-time traffic mirroring that sends identical requests to both production and test servers simultaneously.
After extensive testing, I found most common approaches had limitations:
- IPTables TEE only works for local network destinations
- Proxy solutions typically don't maintain dual forwarding
- Simple port forwarding tools lose connection state
Here's a robust Python implementation that handles multiple ports and maintains proper socket connections:
import socket
import sys
import thread
import time
class TrafficMirror:
def __init__(self, config_file):
self.config = self._parse_config(config_file)
def _parse_config(self, config_file):
"""Parse config file with format: local_port local_target_port remote_ip remote_port"""
configs = []
with open(config_file) as f:
for line in f:
if line.strip() and not line.startswith('#'):
parts = line.split()
configs.append({
'listen_port': int(parts[0]),
'local_port': int(parts[1]),
'remote_ip': parts[2],
'remote_port': int(parts[3])
})
return configs
def _handle_connection(self, client_sock, config):
try:
# Receive client data
data = client_sock.recv(4096)
if not data:
return
# Forward to local server
local_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
local_sock.connect(('127.0.0.1', config['local_port']))
local_sock.sendall(data)
# Get response from local
local_response = local_sock.recv(4096)
local_sock.close()
# Send response to client
client_sock.sendall(local_response)
client_sock.close()
# Mirror to remote (async to avoid latency impact)
thread.start_new_thread(
self._mirror_to_remote,
(data, config['remote_ip'], config['remote_port'])
)
except Exception as e:
sys.stderr.write(f"Error handling connection: {str(e)}\n")
def _mirror_to_remote(self, data, remote_ip, remote_port):
try:
remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_sock.connect((remote_ip, remote_port))
remote_sock.sendall(data)
remote_sock.close()
except Exception as e:
sys.stderr.write(f"Mirroring failed: {str(e)}\n")
def start(self):
for config in self.config:
thread.start_new_thread(self._create_listener, (config,))
# Keep main thread alive
while True:
time.sleep(60)
def _create_listener(self, config):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('0.0.0.0', config['listen_port']))
sock.listen(5)
while True:
client_sock, addr = sock.accept()
thread.start_new_thread(
self._handle_connection,
(client_sock, config)
)
if __name__ == '__main__':
mirror = TrafficMirror('mirror_config.cfg')
mirror.start()
Create a mirror_config.cfg file:
# Format: listen_port local_port remote_ip remote_port
80 8080 192.168.1.100 80
443 8443 192.168.1.100 443
For enterprise deployments, consider these enhancements:
- Add connection pooling for remote mirrors
- Implement request buffering for high traffic volumes
- Add health checks for mirror destinations
- Include detailed logging and metrics
For those preferring non-Python solutions:
- HAProxy with mirroring (version 1.8+):
frontend http-in bind *:80 mode http use_backend primary mirror backend test - Nginx with mirror module:
location / { mirror /mirror; proxy_pass http://primary; } location = /mirror { internal; proxy_pass http://test$request_uri; }
When working with production web servers handling customer traffic, testing configuration changes or new code versions becomes particularly challenging. The need arises to duplicate incoming HTTP requests to both:
- The existing production server (Apache 2.2.16 in this case)
- One or more test servers for benchmarking purposes
This becomes especially complex when dealing with multiple ports (60+ in this scenario) beyond the standard 80/443.
Several approaches have been attempted without success:
- iptables TEE: Limited to local network only
- agnoster duplicator: Requires one session per port (not scalable)
- kklis proxy: Forwards only (no local processing)
- socat: TEE function writes to filesystem only
The following Python script provides a working solution that:
- Listens on specified ports
- Forwards traffic to local Apache instance
- Mirrors traffic to remote servers
- Handles responses from local server
import socket
import SimpleHTTPServer
import SocketServer
import sys, thread, time
def main(config, errorlog):
sys.stderr = file(errorlog, 'a')
for settings in parse(config):
thread.start_new_thread(server, settings)
while True:
time.sleep(60)
def parse(configline):
settings = list()
for line in file(configline):
parts = line.split()
settings.append((int(parts[0]), int(parts[1]), parts[2], int(parts[3])))
return settings
def server(*settings):
try:
dock_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
dock_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
dock_socket.bind(('', settings[0]))
dock_socket.listen(5)
while True:
client_socket = dock_socket.accept()[0]
client_data = client_socket.recv(1024)
# Forward to local
local_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
local_socket.connect(('', settings[1]))
local_socket.sendall(client_data)
# Get response
client_response = local_socket.recv(1024)
local_socket.close()
# Respond to client
client_socket.sendall(client_response)
client_socket.close()
# Mirror to remote
remote_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_socket.connect((settings[2], settings[3]))
remote_socket.sendall(client_data)
remote_socket.close()
except:
print "[ERROR]: ", sys.exc_info()
raise
if __name__ == '__main__':
main('multiforwarder.config', 'error.log')
Create a configuration file multiforwarder.config with entries in the format:
8080 80 remote1.example.com 80
8443 443 remote2.example.com 443
Where each line specifies:
listen_port local_forward_port remote_ip remote_port
For cases where real-time mirroring isn't strictly necessary, consider:
# Capture traffic
tcpdump -i eth0 -w capture.pcap port 80 or port 443
# Replay to test servers
tcpreplay -i eth1 -K --loop=5 capture.pcap
When implementing this solution:
- Monitor CPU and memory usage
- Consider implementing queueing for high traffic volumes
- Ensure proper error handling for network interruptions
- Test with increasing load to identify bottlenecks