Echo Chamber #

# Tanda-tanda echo chamber di trace:

# Trace yang sehat:
# Client → A (50ms) → B (20ms) → [done]

# Trace yang menunjukkan echo chamber:
# Client → A → B → A → B → A → B → ... (ratusan span dari dua service)

# Di Jaeger/Zipkin: cari trace dengan:
# - Span count yang sangat tinggi dari dua service yang sama
# - Depth yang sangat dalam (nested calls yang tidak berakhir)
# - Duration yang jauh melebihi ekspektasi

# Dengan OpenTelemetry, trace menunjukkan ini secara visual

import re
from collections import Counter

def detect_echo_chamber_from_logs(log_file: str, window_seconds: int = 5):
    """
    Deteksi pola echo chamber dari log:
    Jika dua service saling memanggil lebih dari N kali dalam window waktu,
    kemungkinan terjadi echo chamber.
    """
    call_pairs = Counter()
    time_windows = {}

    with open(log_file) as f:
        for line in f:
            # Parse log: timestamp, caller_service, callee_service, request_id
            match = re.search(
                r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}) (\w+) -> (\w+) req=(\w+)',
                line
            )
            if not match:
                continue

            timestamp_str, caller, callee, request_id = match.groups()

            pair = (caller, callee)
            reverse_pair = (callee, caller)

            # Cek apakah pasangan yang berlawanan ada dalam window waktu
            if reverse_pair in time_windows:
                last_time = time_windows[reverse_pair]
                # Jika pasangan saling memanggil dalam window
                call_pairs[(caller, callee, reverse_pair)] += 1

            time_windows[pair] = timestamp_str

    # Report pasangan yang mencurigakan
    for (caller, callee, _), count in call_pairs.most_common():
        if count > 10:  # threshold
            print(f"⚠️  Possible echo chamber: {caller} ↔ {callee} ({count} mutual calls)")

# Prometheus metrics untuk mendeteksi echo chamber

from prometheus_client import Counter, Histogram

# Track setiap outbound request dengan sumber dan tujuan
outbound_requests = Counter(
    'service_outbound_requests_total',
    'Total outbound requests ke service lain',
    ['caller_service', 'callee_service', 'endpoint']
)

# Jika service A dan B saling memanggil dengan frekuensi tinggi:
# Query Prometheus:
# rate(service_outbound_requests_total{caller_service="service-a",callee_service="service-b"}[1m])
# DAN
# rate(service_outbound_requests_total{caller_service="service-b",callee_service="service-a"}[1m])
# Keduanya tinggi secara bersamaan → sinyal echo chamber

# Alert rule:
# alert: PossibleEchoChamber
# expr: |
#   (
#     rate(service_outbound_requests_total{caller_service="service-a", callee_service="service-b"}[2m])
#     > 10
#   ) and (
#     rate(service_outbound_requests_total{caller_service="service-b", callee_service="service-a"}[2m])
#     > 10
#   )
# for: 1m
# annotations:
#   summary: "Possible echo chamber between service-a and service-b"

import uuid
from flask import Flask, request, jsonify, g

app = Flask(__name__)

MAX_REQUEST_DEPTH = 5  # maksimum 5 level request chaining

@app.before_request
def check_request_depth():
    """
    Cek dan enforce request depth limit.
    Depth di-propagate via header X-Request-Depth.
    """
    # Ambil depth dari header (dikirim oleh caller)
    depth_str = request.headers.get('X-Request-Depth', '0')
    try:
        depth = int(depth_str)
    except ValueError:
        depth = 0

    # Simpan depth di request context
    g.request_depth = depth

    # Tolak jika sudah terlalu dalam
    if depth >= MAX_REQUEST_DEPTH:
        return jsonify({
            'error': 'Request depth limit exceeded',
            'depth': depth,
            'max_depth': MAX_REQUEST_DEPTH,
            'message': 'Possible circular dependency detected'
        }), 508  # HTTP 508 Loop Detected

def call_service(url: str, **kwargs):
    """
    Wrapper untuk HTTP call ke service lain yang meng-increment depth.
    Gunakan ini di semua inter-service call, bukan httpx/requests langsung.
    """
    import httpx

    current_depth = getattr(g, 'request_depth', 0)
    request_id = getattr(g, 'request_id', str(uuid.uuid4()))

    headers = kwargs.pop('headers', {})
    headers.update({
        'X-Request-Depth': str(current_depth + 1),  # increment depth
        'X-Request-ID': request_id,                  # propagate ID untuk tracing
        'X-Caller-Service': 'service-a',             # identify caller
    })

    response = httpx.get(url, headers=headers, timeout=10.0, **kwargs)
    response.raise_for_status()
    return response.json()

# Penggunaan:
@app.route('/orders', methods=['POST'])
def create_order():
    order_data = request.json

    # Call ke service lain dengan automatic depth tracking
    user = call_service(f"http://user-service/users/{order_data['user_id']}")

    return jsonify(create_order_record(order_data, user))

import hashlib
import redis
from datetime import timedelta

redis_client = redis.Redis(host='redis', port=6379, decode_responses=True)

def is_event_processed(event_id: str, ttl_seconds: int = 3600) -> bool:
    """
    Cek apakah event dengan ID ini sudah pernah diproses.
    Menggunakan Redis untuk menyimpan event ID yang sudah diproses.
    """
    key = f"processed_event:{event_id}"

    # SET NX: set hanya jika belum ada (atomic)
    # Return True jika berhasil di-set (event belum pernah diproses)
    was_set = redis_client.set(key, "1", ex=ttl_seconds, nx=True)
    return not was_set  # True = sudah pernah diproses (tidak perlu proses lagi)

def process_event_idempotent(event: dict):
    """
    Proses event dengan idempotency check.
    Jika event sudah pernah diproses, skip tanpa error.
    """
    event_id = event.get('event_id')
    if not event_id:
        # Generate event ID dari konten jika tidak ada
        event_id = hashlib.sha256(
            str(sorted(event.items())).encode()
        ).hexdigest()[:16]

    if is_event_processed(event_id):
        logger.info(
            "Event already processed, skipping",
            extra={'event_id': event_id, 'event_type': event.get('type')}
        )
        return {'status': 'already_processed', 'event_id': event_id}

    # Proses event
    result = handle_event(event)
    logger.info(
        "Event processed successfully",
        extra={'event_id': event_id, 'event_type': event.get('type')}
    )
    return result

# Untuk webhook: selalu gunakan webhook ID dari provider
@app.route('/webhooks/payment', methods=['POST'])
def payment_webhook():
    webhook_id = request.headers.get('X-Webhook-ID') or \
                 request.json.get('id')

    if not webhook_id:
        return jsonify({'error': 'Missing webhook ID'}), 400

    # Idempotency check
    if is_event_processed(f"webhook:{webhook_id}"):
        return jsonify({'status': 'already_processed'}), 200

    # Proses webhook
    process_payment_event(request.json)
    return jsonify({'status': 'processed'}), 200

# Event envelope yang menyimpan chain of causality

from dataclasses import dataclass, field
from typing import List
import uuid, time

@dataclass
class EventEnvelope:
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    event_type: str = ""
    payload: dict = field(default_factory=dict)

    # Causality tracking
    correlation_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    causation_id: str = ""      # ID dari event yang menyebabkan event ini
    cause_chain: List[str] = field(default_factory=list)  # chain of causation
    source_service: str = ""
    timestamp: float = field(default_factory=time.time)

    MAX_CAUSE_CHAIN_DEPTH = 10

    def create_child_event(self, event_type: str, payload: dict, source: str) -> 'EventEnvelope':
        """Buat event baru yang merupakan akibat dari event ini."""
        if len(self.cause_chain) >= self.MAX_CAUSE_CHAIN_DEPTH:
            raise EchoChamberDetected(
                f"Cause chain too deep ({len(self.cause_chain)}). "
                f"Possible echo chamber. Chain: {self.cause_chain}"
            )

        return EventEnvelope(
            event_type=event_type,
            payload=payload,
            correlation_id=self.correlation_id,  # sama untuk semua event terkait
            causation_id=self.event_id,           # ID event penyebab
            cause_chain=self.cause_chain + [self.event_id],  # tambah ke chain
            source_service=source,
        )

class EchoChamberDetected(Exception):
    pass

# Consumer yang aware terhadap echo chamber
def consume_user_updated_event(envelope: EventEnvelope):
    user_id = envelope.payload['user_id']

    # Cek apakah event ini sudah dalam chain yang terlalu dalam
    if 'user.profile.sync' in [get_event_type(e) for e in envelope.cause_chain]:
        logger.warning(
            "Skipping user profile sync — already in cause chain",
            extra={'cause_chain': envelope.cause_chain}
        )
        return

    # Update profil lokal
    update_local_profile(user_id, envelope.payload)

    # Kalau perlu trigger event lain, buat sebagai child event
    try:
        child = envelope.create_child_event(
            event_type='user.profile.sync',
            payload={'user_id': user_id, 'synced_at': time.time()},
            source='profile-service'
        )
        publish_event(child)
    except EchoChamberDetected as e:
        logger.error(f"Echo chamber prevented: {e}")
        # Stop chain — jangan publish event yang akan membuat loop

# Contoh refactor: dari circular dependency ke event-driven

# SEBELUM (circular):
# Order Service
def create_order(order_data):
    # Panggil User Service untuk validasi
    user = user_service.get_user(order_data['user_id'])  # → User Service
    # User Service juga panggil Order Service untuk cek limit order
    # → CIRCULAR!

# User Service
def get_user(user_id):
    user = db.get_user(user_id)
    # Cek order limit
    orders = order_service.get_orders(user_id)  # → Order Service → CIRCULAR!
    user['can_order'] = len(orders) < user['order_limit']
    return user

# SESUDAH (event-driven, tanpa circular):
# Order Service — hanya tahu tentang order
def create_order(order_data):
    # Tidak memanggil User Service!
    # Order limit info sudah ada di payload (dikirim oleh client/API gateway)
    if order_data.get('user_order_count', 0) >= order_data.get('user_order_limit', 10):
        raise OrderLimitExceeded()

    order = Order(**order_data)
    db.save(order)

    # Publish event — tidak perlu tahu siapa yang consume
    publish_event('order.created', {'order_id': order.id, 'user_id': order.user_id})
    return order

# User Service — hanya tahu tentang user
def get_user(user_id):
    # Tidak memanggil Order Service!
    # Hanya return data user yang ada di domain-nya sendiri
    return db.get_user(user_id)

# API Gateway atau BFF — yang mengagregasi
def checkout(request):
    user = user_service.get_user(request.user_id)
    order_count = order_service.count_orders(request.user_id)

    # Combine data di layer atas, bukan di dalam service
    return order_service.create_order({
        **request.order_data,
        'user_order_count': order_count,
        'user_order_limit': user['order_limit'],
    })

from circuit_breaker import CircuitBreaker, CircuitOpenError

# Circuit breaker di setiap inter-service call
user_service_breaker = CircuitBreaker(
    name="user-service",
    failure_threshold=5,
    recovery_timeout=30,
)

order_service_breaker = CircuitBreaker(
    name="order-service",
    failure_threshold=5,
    recovery_timeout=30,
)

def get_user_safe(user_id: int) -> dict:
    try:
        return user_service_breaker.call(
            lambda: httpx.get(f"http://user-service/users/{user_id}", timeout=5).json()
        )
    except CircuitOpenError:
        # User Service tidak tersedia atau sedang dalam echo chamber
        # Kembalikan data minimal atau raise error yang jelas
        return {'id': user_id, 'status': 'unavailable'}

# Dengan circuit breaker:
# Jika echo chamber terjadi dan call mulai gagal,
# circuit terbuka setelah 5 kegagalan
# → Request langsung gagal (fast fail) alih-alih loop terus
# → Dependency punya waktu untuk recover

# Custom metrics untuk mendeteksi echo chamber lebih awal

from prometheus_client import Counter, Histogram, Gauge

# Track request depth distribution
request_depth_histogram = Histogram(
    'http_request_depth',
    'Distribution of request depth (how deep in the call chain)',
    buckets=[0, 1, 2, 3, 5, 10]
)

# Alert jika ada request dengan depth tinggi
request_depth_exceeded = Counter(
    'request_depth_limit_exceeded_total',
    'Number of requests rejected due to depth limit'
)

# Track mutual calls antar service
mutual_calls = Counter(
    'inter_service_mutual_calls_total',
    'Number of times service A called B while processing a call from B',
    ['service_a', 'service_b']
)

@app.before_request
def track_request_depth():
    depth = int(request.headers.get('X-Request-Depth', 0))
    request_depth_histogram.observe(depth)
    g.request_depth = depth

    caller = request.headers.get('X-Caller-Service', 'unknown')
    if caller != 'unknown' and caller != 'service-a':
        # Track bahwa service lain memanggil kita
        g.called_by = caller

# Alert rules untuk echo chamber

groups:
  - name: echo_chamber
    rules:
      # Alert jika ada request yang di-reject karena depth limit
      - alert: RequestDepthLimitExceeded
        expr: rate(request_depth_limit_exceeded_total[5m]) > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Request depth limit exceeded — possible echo chamber"
          description: "{{ $value }} requests/s rejected due to depth limit"
          runbook_url: "https://runbook.company.com/echo-chamber"

      # Alert untuk mutual call rate yang tinggi
      - alert: HighMutualCallRate
        expr: |
          rate(inter_service_mutual_calls_total[2m]) > 10          
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "High mutual call rate between {{ $labels.service_a }} and {{ $labels.service_b }}"

# ✗ Anti-pattern 1: Tidak ada request depth limit
# Service yang saling memanggil tanpa batas
# Satu request bisa menghasilkan ribuan call sebelum timeout
# ✓ Solusi: X-Request-Depth header + batas maksimum

# ✗ Anti-pattern 2: Webhook tanpa idempotency check
@app.route('/webhooks/payment', methods=['POST'])
def payment_webhook():
    process_payment(request.json)  # bisa diproses berkali-kali!
    return jsonify({'ok': True})
# ✓ Solusi: simpan webhook ID yang sudah diproses di Redis

# ✗ Anti-pattern 3: Event consumer yang publish event dengan type yang sama
def handle_user_updated(event):
    update_profile(event['user_id'])
    publish('user.updated', event)  # akan di-consume oleh dirinya sendiri lagi!
# ✓ Solusi: event consumer tidak publish event yang sama dengan yang di-consume
# atau: cek source service sebelum publish

# ✗ Anti-pattern 4: Saling bergantung tanpa abstraksi
# Service A import langsung client Service B
# Service B import langsung client Service A
# ✓ Solusi: redesain dengan shared dependency atau event-driven

# ✗ Anti-pattern 5: Tidak ada timeout pada inter-service call
response = requests.get("http://service-b/data")  # bisa hang selamanya
# ✓ Solusi: selalu set timeout
response = httpx.get("http://service-b/data", timeout=10.0)