Observability: Logging, Metrics, Tracing

# Error Rate — শেষ ৫ মিনিটে error percentage
rate(http_requests_total{status=~"5.."}[5m])
  /
rate(http_requests_total[5m]) * 100

# P99 Latency — ৯৯% requests এর latency
histogram_quantile(0.99,
  rate(http_request_duration_seconds_bucket[5m])
)

# Request Rate by Service — প্রতিটা service-এর RPS
sum(rate(http_requests_total[1m])) by (service)

# Active Connections per Pod
sum(tcp_connections_active) by (pod)

# Memory Usage Percentage
(node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes)
  / node_memory_MemTotal_bytes * 100

# ───────────── Alert Rule Example ─────────────
# prometheus_rules.yml
groups:
  - name: payment-alerts
    rules:
      - alert: HighErrorRate
        expr: |
          rate(http_requests_total{service="payment",status=~"5.."}[5m])
          /
          rate(http_requests_total{service="payment"}[5m]) > 0.05
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Payment service error rate > 5%"
          description: "Error rate: {{ $value | humanizePercentage }}"

{
  "title": "Payment Service — SLO Dashboard",
  "panels": [
    {
      "title": "Error Rate (5m)",
      "type": "stat",
      "datasource": "Prometheus",
      "targets": [
        {
          "expr": "sum(rate(http_requests_total{service='payment',status=~'5..'}[5m])) / sum(rate(http_requests_total{service='payment'}[5m])) * 100",
          "legendFormat": "Error %"
        }
      ],
      "thresholds": {
        "steps": [
          { "color": "green", "value": null },
          { "color": "yellow", "value": 1 },
          { "color": "red", "value": 5 }
        ]
      }
    },
    {
      "title": "P99 Latency",
      "type": "timeseries",
      "datasource": "Prometheus",
      "targets": [
        {
          "expr": "histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{service='payment'}[5m]))",
          "legendFormat": "P99"
        },
        {
          "expr": "histogram_quantile(0.95, rate(http_request_duration_seconds_bucket{service='payment'}[5m]))",
          "legendFormat": "P95"
        }
      ]
    },
    {
      "title": "Request Rate (RPS)",
      "type": "timeseries",
      "targets": [
        {
          "expr": "sum(rate(http_requests_total{service='payment'}[1m])) by (status_code)",
          "legendFormat": "{{ status_code }}"
        }
      ]
    }
  ]
}

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from fastapi import FastAPI, Request
import httpx

# ──── Setup: Jaeger Exporter ────
jaeger_exporter = JaegerExporter(
    agent_host_name="jaeger",  # Docker service name
    agent_port=6831,
)

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("order-service")

app = FastAPI()
FastAPIInstrumentor.instrument_app(app)  # Auto-instrument all routes

# ──── Manual Span: Custom business logic ────
@app.post("/order")
async def create_order(request: Request, item_id: str, quantity: int):
    # trace_id is automatically injected by FastAPIInstrumentor
    with tracer.start_as_current_span("validate-inventory") as span:
        span.set_attribute("item_id", item_id)
        span.set_attribute("quantity", quantity)

        # Check inventory — nested span
        with tracer.start_as_current_span("db-query-inventory"):
            available = await check_inventory(item_id)
            if available < quantity:
                span.set_status(trace.StatusCode.ERROR, "insufficient stock")
                return {"error": "out of stock"}

    # Call Payment Service — trace_id propagates via HTTP header
    with tracer.start_as_current_span("call-payment-service") as pay_span:
        pay_span.set_attribute("amount", quantity * 100)
        async with httpx.AsyncClient() as client:
            # OpenTelemetry automatically injects traceparent header
            response = await client.post(
                "http://payment-svc/charge",
                json={"item_id": item_id, "quantity": quantity},
                headers=get_trace_headers()  # Propagate context
            )

    return {"order_id": "ord-123", "status": "confirmed"}


# ──── Trace Context Propagation ────
def get_trace_headers() -> dict:
    """Extract current trace context as HTTP headers for downstream services"""
    from opentelemetry.propagate import inject
    headers = {}
    inject(headers)  # Adds 'traceparent' header: 00-{trace_id}-{span_id}-01
    return headers

# traceparent example:
# traceparent: 00-abc123def456-789abc-01
#              ver  trace_id    span_id flags

import logging
import json
import time
import uuid
from contextvars import ContextVar

# Correlation ID — request-এর পুরো journey track করতে
_correlation_id: ContextVar[str] = ContextVar('correlation_id', default='')

class StructuredLogger:
    def __init__(self, service_name: str):
        self.service = service_name

    def _log(self, level: str, message: str, **extra):
        record = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime()),
            "level": level,
            "service": self.service,
            "correlation_id": _correlation_id.get(),  # Track across services
            "message": message,
            **extra  # Additional context fields
        }
        print(json.dumps(record))  # ELK collects stdout

    def info(self, msg: str, **kw): self._log("INFO", msg, **kw)
    def error(self, msg: str, **kw): self._log("ERROR", msg, **kw)
    def warn(self, msg: str, **kw): self._log("WARN", msg, **kw)


# Usage example
logger = StructuredLogger("payment-service")

def process_payment(user_id: str, amount: float, request_id: str):
    _correlation_id.set(request_id)  # Set for this request lifecycle

    logger.info("payment_started",
        user_id=user_id,
        amount=amount,
        currency="BDT"
    )

    try:
        result = charge_card(user_id, amount)
        logger.info("payment_success",
            user_id=user_id,
            amount=amount,
            transaction_id=result.txn_id,
            duration_ms=result.duration
        )
        return result
    except Exception as e:
        logger.error("payment_failed",
            user_id=user_id,
            amount=amount,
            error=str(e),
            error_type=type(e).__name__
        )
        raise

# Output (ELK ingests this JSON):
# {"timestamp":"2026-05-08T03:15:22.000Z","level":"ERROR",
#  "service":"payment-service","correlation_id":"req-abc-123",
#  "message":"payment_failed","user_id":"u-456","amount":1500.0,
#  "error":"DB connection timeout","error_type":"TimeoutError"}

const express = require('express');
const { MeterProvider } = require('@opentelemetry/sdk-metrics');
const { trace, context } = require('@opentelemetry/api');
const promClient = require('prom-client');

const app = express();

// ─── 1. METRICS — Prometheus ───
const register = new promClient.Registry();
promClient.collectDefaultMetrics({ register });

const httpRequestCounter = new promClient.Counter({
  name: 'http_requests_total',
  help: 'Total HTTP requests',
  labelNames: ['method', 'route', 'status_code'],
  registers: [register],
});

const httpLatency = new promClient.Histogram({
  name: 'http_request_duration_seconds',
  help: 'HTTP request latency in seconds',
  labelNames: ['method', 'route'],
  buckets: [0.01, 0.05, 0.1, 0.3, 0.5, 1, 2, 5],
  registers: [register],
});

// ─── 2. STRUCTURED LOGGING ───
function log(level, message, fields = {}) {
  const span = trace.getActiveSpan();
  const spanCtx = span?.spanContext();

  const entry = {
    timestamp: new Date().toISOString(),
    level,
    service: 'order-service',
    message,
    trace_id: spanCtx?.traceId || '',   // Link log to trace
    span_id: spanCtx?.spanId || '',
    ...fields,
  };
  console.log(JSON.stringify(entry));   // ELK / Loki ingests stdout
}

// ─── 3. TRACES — OpenTelemetry ───
const tracer = trace.getTracer('order-service', '1.0.0');

// ─── Request Handler ───
app.post('/orders', async (req, res) => {
  const startTime = Date.now();
  const { userId, items } = req.body;

  // Span: entire request
  return tracer.startActiveSpan('create-order', async (span) => {
    try {
      span.setAttributes({
        'user.id': userId,
        'order.item_count': items.length,
      });

      log('INFO', 'order_request_received', {
        user_id: userId,
        item_count: items.length,
      });

      // Nested span: validate
      const inventory = await tracer.startActiveSpan('validate-inventory',
        async (invSpan) => {
          const result = await checkInventory(items);
          invSpan.setAttributes({ 'inventory.available': result.available });
          invSpan.end();
          return result;
        }
      );

      if (!inventory.available) {
        span.setStatus({ code: 2, message: 'out_of_stock' });
        log('WARN', 'order_rejected', { reason: 'out_of_stock', user_id: userId });
        res.status(400).json({ error: 'out_of_stock' });
        return;
      }

      // Nested span: save to DB
      const order = await tracer.startActiveSpan('save-order-db',
        async (dbSpan) => {
          const saved = await db.save({ userId, items, status: 'pending' });
          dbSpan.setAttribute('db.order_id', saved.id);
          dbSpan.end();
          return saved;
        }
      );

      const durationMs = Date.now() - startTime;
      log('INFO', 'order_created', {
        user_id: userId,
        order_id: order.id,
        duration_ms: durationMs,
      });

      // Record metrics
      httpRequestCounter.inc({ method: 'POST', route: '/orders', status_code: 201 });
      httpLatency.observe({ method: 'POST', route: '/orders' }, durationMs / 1000);

      span.setStatus({ code: 1 });
      res.status(201).json({ orderId: order.id });

    } catch (err) {
      span.recordException(err);
      span.setStatus({ code: 2, message: err.message });

      log('ERROR', 'order_creation_failed', {
        user_id: userId,
        error: err.message,
        stack: err.stack,
      });

      httpRequestCounter.inc({ method: 'POST', route: '/orders', status_code: 500 });
      res.status(500).json({ error: 'internal_error' });
    } finally {
      span.end();
    }
  });
});

// ─── Prometheus scrape endpoint ───
app.get('/metrics', async (req, res) => {
  res.set('Content-Type', register.contentType);
  res.end(await register.metrics());
});

app.listen(3000);