Geographic Load Distribution
Geographic Load Distribution
High-traffic websites often serve global audiences, requiring geographic distribution strategies that optimize both performance and security.
Anycast Firewall Deployment: Implement anycast routing for distributed firewall presence:
# Anycast health check and announcement manager
import subprocess
import requests
import time
class AnycastManager:
def __init__(self, anycast_ip, health_check_url):
self.anycast_ip = anycast_ip
self.health_check_url = health_check_url
self.bird_config = "/etc/bird/bird.conf"
def check_health(self):
"""Verify local services are healthy"""
try:
# Check firewall service
fw_status = subprocess.run(['systemctl', 'is-active', 'firewall'],
capture_output=True, text=True)
if fw_status.stdout.strip() != 'active':
return False
# Check web service
response = requests.get(self.health_check_url, timeout=5)
if response.status_code != 200:
return False
# Check system resources
load_avg = os.getloadavg()[0]
cpu_count = os.cpu_count()
if load_avg > cpu_count * 2: # System overloaded
return False
return True
except Exception as e:
print(f"Health check failed: {e}")
return False
def update_bgp_announcement(self, announce=True):
"""Update BGP announcements based on health"""
if announce:
# Announce anycast IP
bird_command = f"""
protocol static anycast_routes {{
route {self.anycast_ip}/32 blackhole;
}}
"""
else:
# Withdraw anycast IP
bird_command = f"""
protocol static anycast_routes {{
# route {self.anycast_ip}/32 blackhole;
}}
"""
# Update BIRD configuration
with open(self.bird_config, 'w') as f:
f.write(bird_command)
# Reload BIRD
subprocess.run(['birdc', 'configure'])
def run(self):
"""Main monitoring loop"""
consecutive_failures = 0
announced = True
while True:
if self.check_health():
consecutive_failures = 0
if not announced:
print("Health restored, announcing anycast IP")
self.update_bgp_announcement(True)
announced = True
else:
consecutive_failures += 1
if consecutive_failures >= 3 and announced:
print("Health check failed, withdrawing anycast IP")
self.update_bgp_announcement(False)
announced = False
time.sleep(10) # Check every 10 seconds