Distributed Systems Hub

# Simple Raft-like leader election
import asyncio
import random
import time

class Node:
    def __init__(self, node_id, peers):
        self.id = node_id
        self.peers = peers
        self.state = "follower"
        self.term = 0
        self.voted_for = None
        self.leader = None
        
    async def start_election(self):
        self.state = "candidate"
        self.term += 1
        self.voted_for = self.id
        
        votes = 1  # Vote for self
        for peer in self.peers:
            if await self.request_vote(peer):
                votes += 1
                
        if votes > len(self.peers) / 2:
            self.become_leader()
            
    def become_leader(self):
        self.state = "leader"
        self.leader = self.id
        print(f"Node {self.id} became leader for term {self.term}")

# Producer
import asyncio
from aiokafka import AIOKafkaProducer

async def send_order_event(order_data):
    producer = AIOKafkaProducer(
        bootstrap_servers='localhost:9092',
        value_serializer=lambda v: json.dumps(v).encode()
    )
    await producer.start()
    try:
        await producer.send_and_wait(
            "order-events",
            {"order_id": order_data["id"], "status": "created"}
        )
    finally:
        await producer.stop()

# Consumer
from aiokafka import AIOKafkaConsumer

async def process_orders():
    consumer = AIOKafkaConsumer(
        'order-events',
        bootstrap_servers='localhost:9092',
        group_id="order-processor"
    )
    await consumer.start()
    try:
        async for msg in consumer:
            order = json.loads(msg.value)
            await handle_order(order)
    finally:
        await consumer.stop()

from circuit_breaker import CircuitBreaker
import aiohttp
import asyncio

class ResilientService:
    def __init__(self):
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60,
            expected_exception=aiohttp.ClientError
        )
        
    @circuit_breaker
    async def call_external_service(self, endpoint):
        async with aiohttp.ClientSession() as session:
            async with session.get(endpoint) as response:
                return await response.json()
                
    async def health_check(self):
        checks = {
            "database": await self.check_database(),
            "cache": await self.check_cache(),
            "dependencies": await self.check_dependencies()
        }
        
        status = "healthy" if all(checks.values()) else "unhealthy"
        return {"status": status, "checks": checks}

from opentelemetry import trace
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Setup tracing
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

span_processor = BatchSpanProcessor(jaeger_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

# Use in application
async def process_request(request_id):
    with tracer.start_as_current_span("process_request") as span:
        span.set_attribute("request.id", request_id)
        
        # Process stages
        with tracer.start_as_current_span("validate"):
            await validate_request(request_id)
            
        with tracer.start_as_current_span("process"):
            result = await process_data(request_id)
            
        with tracer.start_as_current_span("persist"):
            await save_result(result)
            
        return result

import hashlib

class ShardedDatabase:
    def __init__(self, shard_configs):
        self.shards = {}
        self.num_shards = len(shard_configs)
        
        for i, config in enumerate(shard_configs):
            self.shards[i] = create_connection(config)
            
    def get_shard(self, key):
        """Consistent hashing for shard selection"""
        hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
        shard_id = hash_value % self.num_shards
        return self.shards[shard_id]
        
    async def get(self, key):
        shard = self.get_shard(key)
        return await shard.get(key)
        
    async def set(self, key, value):
        shard = self.get_shard(key)
        return await shard.set(key, value)

import etcd3
import threading
import time

class LeaderElection:
    def __init__(self, node_id, etcd_host='localhost'):
        self.node_id = node_id
        self.etcd = etcd3.client(host=etcd_host)
        self.lease = None
        self.is_leader = False
        
    def campaign(self):
        """Run for leader position"""
        # Create lease with TTL
        self.lease = self.etcd.lease(5)  # 5 second TTL
        
        # Try to create leader key
        try:
            self.etcd.put(
                '/election/leader',
                self.node_id,
                lease=self.lease
            )
            self.is_leader = True
            print(f"Node {self.node_id} became leader")
            
            # Keep lease alive
            self.lease.refresh()
            
        except etcd3.exceptions.PreconditionFailedError:
            # Another node is leader
            self.is_leader = False
            self.watch_leader()
            
    def watch_leader(self):
        """Watch for leader changes"""
        events_iterator, cancel = self.etcd.watch('/election/leader')
        for event in events_iterator:
            if isinstance(event, etcd3.events.DeleteEvent):
                # Leader stepped down, campaign again
                self.campaign()
                break

import redis
import uuid
import time

class RedisLock:
    def __init__(self, redis_client, key, timeout=10):
        self.redis = redis_client
        self.key = key
        self.timeout = timeout
        self.identifier = str(uuid.uuid4())
        
    def acquire(self):
        """Acquire lock with timeout"""
        end = time.time() + self.timeout
        while time.time() < end:
            if self.redis.set(self.key, self.identifier, nx=True, ex=self.timeout):
                return True
            time.sleep(0.001)
        return False
        
    def release(self):
        """Release lock if we own it"""
        pipe = self.redis.pipeline(True)
        while True:
            try:
                pipe.watch(self.key)
                if pipe.get(self.key) == self.identifier:
                    pipe.multi()
                    pipe.delete(self.key)
                    pipe.execute()
                    return True
                pipe.unwatch()
                break
            except redis.WatchError:
                pass
        return False
        
    def __enter__(self):
        if not self.acquire():
            raise Exception("Could not acquire lock")
        return self
        
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

class SagaOrchestrator:
    def __init__(self):
        self.steps = []
        self.compensations = []
        
    def add_step(self, action, compensation):
        self.steps.append(action)
        self.compensations.append(compensation)
        
    async def execute(self):
        completed_steps = []
        
        try:
            # Execute all steps
            for i, step in enumerate(self.steps):
                result = await step()
                completed_steps.append(i)
                
        except Exception as e:
            # Compensate in reverse order
            for i in reversed(completed_steps):
                try:
                    await self.compensations[i]()
                except Exception as comp_error:
                    # Log compensation failure
                    print(f"Compensation {i} failed: {comp_error}")
            raise e
            
# Usage example
saga = SagaOrchestrator()

saga.add_step(
    lambda: create_order(order_data),
    lambda: cancel_order(order_id)
)

saga.add_step(
    lambda: charge_payment(payment_data),
    lambda: refund_payment(payment_id)
)

saga.add_step(
    lambda: update_inventory(items),
    lambda: restore_inventory(items)
)

await saga.execute()

import random
import asyncio

class ChaosMonkey:
    def __init__(self, failure_rate=0.1):
        self.failure_rate = failure_rate
        self.failures = {
            "network_delay": self.inject_network_delay,
            "service_crash": self.inject_service_crash,
            "disk_full": self.inject_disk_full,
            "cpu_spike": self.inject_cpu_spike
        }
        
    async def inject_failure(self):
        if random.random() < self.failure_rate:
            failure_type = random.choice(list(self.failures.keys()))
            await self.failures[failure_type]()
            
    async def inject_network_delay(self):
        delay = random.uniform(0.1, 2.0)
        await asyncio.sleep(delay)
        
    async def inject_service_crash(self):
        if random.random() < 0.5:
            raise Exception("Service crashed!")

from hypothesis import given, strategies as st
import asyncio

class DistributedCounter:
    def __init__(self, nodes=3):
        self.counters = [0] * nodes
        self.nodes = nodes
        
    async def increment(self, node_id):
        # Simulate network delay
        await asyncio.sleep(random.uniform(0, 0.1))
        self.counters[node_id] += 1
        
    async def get_total(self):
        # Eventually consistent read
        await asyncio.sleep(0.2)
        return sum(self.counters)

# Property: Total equals number of increments
@given(st.lists(st.integers(min_value=0, max_value=2), min_size=1, max_size=100))
async def test_counter_consistency(operations):
    counter = DistributedCounter()
    
    # Perform operations concurrently
    tasks = [counter.increment(node_id) for node_id in operations]
    await asyncio.gather(*tasks)
    
    # Check eventual consistency
    total = await counter.get_total()
    assert total == len(operations)

import asyncio
from functools import wraps
import hashlib
import json

class DistributedCache:
    def __init__(self, redis_client, default_ttl=3600):
        self.redis = redis_client
        self.default_ttl = default_ttl
        
    def cache_key(self, func_name, *args, **kwargs):
        """Generate cache key from function and arguments"""
        key_data = {
            'func': func_name,
            'args': args,
            'kwargs': kwargs
        }
        key_str = json.dumps(key_data, sort_keys=True)
        return f"cache:{hashlib.md5(key_str.encode()).hexdigest()}"
        
    def cached(self, ttl=None):
        """Decorator for caching function results"""
        def decorator(func):
            @wraps(func)
            async def wrapper(*args, **kwargs):
                cache_key = self.cache_key(func.__name__, *args, **kwargs)
                
                # Try to get from cache
                cached_value = await self.redis.get(cache_key)
                if cached_value:
                    return json.loads(cached_value)
                    
                # Compute and cache result
                result = await func(*args, **kwargs)
                await self.redis.set(
                    cache_key,
                    json.dumps(result),
                    ex=ttl or self.default_ttl
                )
                return result
            return wrapper
        return decorator

# Usage
cache = DistributedCache(redis_client)

@cache.cached(ttl=300)
async def expensive_computation(user_id):
    # Simulate expensive operation
    await asyncio.sleep(2)
    return {"user_id": user_id, "score": random.random()}

import asyncio
from collections import defaultdict

class BatchProcessor:
    def __init__(self, batch_size=100, batch_timeout=1.0):
        self.batch_size = batch_size
        self.batch_timeout = batch_timeout
        self.pending = []
        self.results = {}
        self.lock = asyncio.Lock()
        self.batch_event = asyncio.Event()
        
    async def add_item(self, item):
        """Add item to batch and wait for result"""
        future = asyncio.Future()
        
        async with self.lock:
            self.pending.append((item, future))
            
            if len(self.pending) >= self.batch_size:
                self.batch_event.set()
                
        # Start batch timer if this is first item
        if len(self.pending) == 1:
            asyncio.create_task(self._batch_timer())
            
        return await future
        
    async def _batch_timer(self):
        """Trigger batch processing after timeout"""
        await asyncio.sleep(self.batch_timeout)
        self.batch_event.set()
        
    async def process_batch(self):
        """Process accumulated batch"""
        while True:
            await self.batch_event.wait()
            self.batch_event.clear()
            
            async with self.lock:
                if not self.pending:
                    continue
                    
                batch = self.pending
                self.pending = []
                
            # Process batch
            items = [item for item, _ in batch]
            results = await self._do_batch_processing(items)
            
            # Deliver results
            for (item, future), result in zip(batch, results):
                future.set_result(result)

from prometheus_client import Counter, Histogram, Gauge
import time

# Define metrics
request_count = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint'])
request_duration = Histogram('http_request_duration_seconds', 'HTTP request duration', ['method', 'endpoint'])
active_connections = Gauge('active_connections', 'Number of active connections')

# Middleware for metrics collection
async def metrics_middleware(request, handler):
    start_time = time.time()
    active_connections.inc()
    
    try:
        # Process request
        response = await handler(request)
        
        # Record metrics
        request_count.labels(
            method=request.method,
            endpoint=request.path
        ).inc()
        
        return response
        
    finally:
        # Record duration
        duration = time.time() - start_time
        request_duration.labels(
            method=request.method,
            endpoint=request.path
        ).observe(duration)
        
        active_connections.dec()

import logging
import json
from pythonjsonlogger import jsonlogger

# Configure structured logging
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)

logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

class DistributedLogger:
    def __init__(self, service_name, node_id):
        self.service_name = service_name
        self.node_id = node_id
        
    def log(self, level, message, **kwargs):
        """Log with distributed context"""
        logger.log(
            level,
            message,
            extra={
                "service": self.service_name,
                "node_id": self.node_id,
                "timestamp": time.time(),
                "trace_id": kwargs.get("trace_id"),
                "span_id": kwargs.get("span_id"),
                **kwargs
            }
        )
        
# Usage
dist_logger = DistributedLogger("order-service", "node-1")
dist_logger.log(
    logging.INFO,
    "Order processed",
    order_id="12345",
    user_id="user-789",
    trace_id="abc-def-ghi"
)

import jwt
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding

class ZeroTrustAuth:
    def __init__(self, private_key, public_keys):
        self.private_key = private_key
        self.public_keys = public_keys  # Dict of service_name -> public_key
        
    def create_service_token(self, source_service, target_service, ttl=300):
        """Create JWT for service-to-service auth"""
        payload = {
            "iss": source_service,
            "aud": target_service,
            "exp": time.time() + ttl,
            "iat": time.time(),
            "jti": str(uuid.uuid4())
        }
        
        return jwt.encode(payload, self.private_key, algorithm="RS256")
        
    def verify_service_token(self, token, expected_audience):
        """Verify incoming service token"""
        try:
            # Decode without verification first to get issuer
            unverified = jwt.decode(token, options={"verify_signature": False})
            issuer = unverified.get("iss")
            
            if issuer not in self.public_keys:
                raise Exception(f"Unknown issuer: {issuer}")
                
            # Verify with issuer's public key
            payload = jwt.decode(
                token,
                self.public_keys[issuer],
                algorithms=["RS256"],
                audience=expected_audience
            )
            
            return payload
            
        except jwt.ExpiredSignatureError:
            raise Exception("Token expired")
        except jwt.InvalidTokenError:
            raise Exception("Invalid token")

import ssl
import aiohttp

class SecureClient:
    def __init__(self, ca_cert_path, client_cert_path, client_key_path):
        # Create SSL context with mutual TLS
        self.ssl_context = ssl.create_default_context(
            ssl.Purpose.SERVER_AUTH,
            cafile=ca_cert_path
        )
        self.ssl_context.load_cert_chain(
            certfile=client_cert_path,
            keyfile=client_key_path
        )
        
    async def secure_request(self, url, method="GET", **kwargs):
        """Make secure HTTPS request with mTLS"""
        connector = aiohttp.TCPConnector(ssl=self.ssl_context)
        
        async with aiohttp.ClientSession(connector=connector) as session:
            async with session.request(method, url, **kwargs) as response:
                return await response.json()