Centralized logging — structured log, pipeline আর খরচ

# ssh into every server and grep. Fine for 3 servers, painful for 50.
for host in web-1 web-2 web-3; do
  ssh "$host" "grep 'order 502' /var/log/app/app.log"
done

{
  "log-driver": "json-file",
  "log-opts": { "max-size": "10m", "max-file": "3" }
}

import json, logging, os, sys

SERVICE = os.environ["SERVICE"]

class JsonFormatter(logging.Formatter):
    def format(self, r):
        event = {"ts": self.formatTime(r, "%Y-%m-%dT%H:%M:%S"),
                 "level": r.levelname.lower(), "service": SERVICE,
                 "msg": r.getMessage()}
        event.update(getattr(r, "fields", {}))
        return json.dumps(event)

h = logging.StreamHandler(sys.stdout)     # stdout, never a file
h.setFormatter(JsonFormatter())
log = logging.getLogger(SERVICE)
log.addHandler(h)
log.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

# … six fake orders: (trace_id, order_id, amount, card)
for trace, order, amount, card in ORDERS:
    f = {"trace_id": trace, "order_id": order}
    if SERVICE == "orders":
        log.debug("cart loaded", extra={"fields": f})
        log.info("order received", extra={"fields": {**f, "amount": amount}})
    elif SERVICE == "payment":
        log.info(f"charging card {card}", extra={"fields": f})
        if trace in ("t-1002", "t-1005"):
            log.error("gateway timeout", extra={"fields": {**f, "ms": 3000}})
    # … notify logs "sms sent to rahim@example.com"

#!/usr/bin/env bash
set -euo pipefail
cat > app.log <<'EOF'
2026-10-30 10:00:01 INFO orders trace=t-1001 order received amount=450
2026-10-30 10:00:01 INFO payment trace=t-1001 card charged ms=210
2026-10-30 10:00:02 ERROR payment trace=t-1002 gateway timeout ms=3000
Traceback (most recent call last):
  File "pay.py", line 42, in charge
2026-10-30 10:00:03 ERROR payment trace=t-1002 retry failed ms=3000
2026-10-30 10:00:03 WARN notify trace=t-1003 sms queue slow ms=900
2026-10-30 10:00:04 ERROR notify trace=t-1003 sms provider down ms=120
2026-10-30 10:00:05 ERROR payment trace=t-1002 order cancelled ms=15
EOF

# 1. Free text -> named fields. Lines that do not fit go to a separate file.
awk '$1 ~ /^20[0-9][0-9]-/ && $5 ~ /^trace=/ {
  trace = substr($5, 7)
  msg = $6; for (i = 7; i <= NF; i++) msg = msg " " $i
  printf "ts=%sT%s level=%s service=%s trace_id=%s msg=\"%s\"\n",
         $1, $2, tolower($3), $4, trace, msg > "fields.log"
  next
}
{ print > "unparsed.log" }' app.log

echo "--- first 2 structured lines"
head -2 fields.log
echo "--- lines that did not fit: $(wc -l < unparsed.log)"

# 2. Count errors by trace id: a question free text answers badly.
echo "--- errors by trace id"
grep 'level=error' fields.log | grep -o 'trace_id=[^ ]*' | sort | uniq -c | sort -rn

# 3. One trace, every service it touched, in time order.
echo "--- trace t-1002"
grep 'trace_id=t-1002' fields.log | sed 's/ trace_id=[^ ]*//'

service:
  flush: 1
  log_level: info
  parsers_file: /fluent-bit/etc/parsers.conf

pipeline:
  inputs:
    - name: forward
      listen: 0.0.0.0
      port: 24224
      mem_buf_limit: 8MB

  filters:
    - name: parser          # the "log" string becomes real fields
      match: '*'
      key_name: log
      parser: json
    - name: lua             # redaction, before anything leaves
      match: '*'
      script: /cfg/redact.lua
      call: redact

  outputs:
    - name: loki
      match: '*'
      host: abcl-lg-loki
      port: 3100
      labels: job=shop
      label_keys: $service,$level
      line_format: json
      retry_limit: 5

function redact(tag, ts, record)
  local m = record["msg"]
  if m == nil then return 0, ts, record end
  m = m:gsub("%d%d%d%d[ -]?%d%d%d%d[ -]?%d%d%d%d[ -]?(%d%d%d%d)", "[CARD-%1]")
  m = m:gsub("[%w._-]+@[%w.-]+", "[EMAIL]")
  record["msg"] = m
  return 2, ts, record
end

$ docker network create abcl-lg-net
$ docker run -d --name abcl-lg-loki --hostname demo --network abcl-lg-net \
    -p 127.0.0.1:18641:3100 grafana/loki:3.5.5 \
    -config.file=/etc/loki/local-config.yaml
$ docker run -d --name abcl-lg-fluent --hostname demo --network abcl-lg-net \
    -p 127.0.0.1:18642:24224 -v "$PWD:/cfg:ro" \
    fluent/fluent-bit:4.0 -c /cfg/fluent-bit.yaml
$ docker logs abcl-lg-fluent 2>&1 | tail -3
… [ info] [input:forward:forward.0] storage_strategy='memory' (memory only)
… [ info] [input:forward:forward.0] listening on 0.0.0.0:24224
… [ info] [output:loki:loki.0] configured, hostname=abcl-lg-loki:3100

$ for s in orders payment notify; do
    docker run --name abcl-lg-$s --hostname demo --network abcl-lg-net \
      -e SERVICE=$s -v "$PWD:/app:ro" \
      --log-driver fluentd \
      --log-opt fluentd-address=127.0.0.1:18642 \
      --log-opt tag=shop.$s \
      python:3.12-alpine python /app/app.py >/dev/null
  done

$ B=http://127.0.0.1:18641/loki/api/v1
$ curl -s $B/labels
{"status":"success","data":["job","level","service","service_name"]}

# every line of one trace, oldest first
$ curl -sG $B/query_range --data-urlencode 'direction=forward' \
    --data-urlencode 'query={job="shop"} | json | trace_id="t-1002"'
orders  info  {"trace_id":"t-1002","order_id":502,"ts":"2026-10-30T21:09:42","amount":1200,"msg":"order received"}
payment error {"trace_id":"t-1002","order_id":502,"ts":"2026-10-30T21:09:43","ms":3000,"msg":"gateway timeout"}
payment info  {"trace_id":"t-1002","order_id":502,"ts":"2026-10-30T21:09:43","msg":"charging card [CARD-1111]"}

# a metric from logs: errors per service in the last hour
$ curl -sG $B/query --data-urlencode \
    'query=sum by (service) (count_over_time({job="shop",level="error"}[1h]))'
{"status":"success","data":{"resultType":"vector","result":[{"metric":{"service":"payment"},"value":[…,"2"]}],…

# clean up: exact names only
$ docker rm -f abcl-lg-orders abcl-lg-payment abcl-lg-notify abcl-lg-fluent abcl-lg-loki
$ docker network rm abcl-lg-net
$ docker rmi grafana/loki:3.5.5 fluent/fluent-bit:4.0

// Monthly log volume and cost. Prices are ASSUMED round numbers, not any provider's.
const RPS = 200, SECONDS = 30 * 86400;     // requests per second, 30-day month
const PER_GB = 0.50, STORE_GB_MONTH = 0.03; // ingest price, stored price
const HOT_DAYS = 30, COMPRESS = 5;          // keep 30 days; stored size = raw / 5
const LEVELS = {                            // lines per request, bytes per line
  debug: { lines: 12, bytes: 300 },
  info:  { lines: 4, bytes: 350 },
  warn:  { lines: 0.1, bytes: 400 },
  error: { lines: 0.02, bytes: 900 },
};

function plan(name, keep) {                 // keep = share kept per level, 0 to 1
  let total = 0;
  const parts = [];
  for (const [lvl, v] of Object.entries(LEVELS)) {
    const gb = RPS * v.lines * v.bytes * SECONDS / 1e9 * keep[lvl];
    total += gb;
    parts.push(`${lvl} ${gb.toFixed(1)}`);
  }
  const ingest = total * PER_GB;
  const stored = total / COMPRESS * (HOT_DAYS / 30);
  const store = stored * STORE_GB_MONTH;
  console.log(`${name}\n  GB/month: ${parts.join(", ")} = ${total.toFixed(1)}`);
  console.log(`  ingest $${ingest.toFixed(2)} + store $${store.toFixed(2)}` +
              ` = $${(ingest + store).toFixed(2)} per month`);
}

plan("A. everything", { debug: 1, info: 1, warn: 1, error: 1 });
plan("B. debug off", { debug: 0, info: 1, warn: 1, error: 1 });
plan("C. debug off, info 10%", { debug: 0, info: 0.1, warn: 1, error: 1 });
plan("D. debug 1%, info 10%", { debug: 0.01, info: 0.1, warn: 1, error: 1 });

// A redaction filter with FAKE data. Change RULES or LINES and press Run again.
const RULES = [
  ["card",   /\b(?:\d{4}[ -]?){3}(\d{4})\b/g, "[CARD-$1]"],
  ["email",  /[\w.+-]+@[\w-]+(?:\.[\w-]+)+/g, "[EMAIL]"],
  ["phone",  /\b01[3-9]\d{8}\b/g,             "[PHONE]"],
  ["bearer", /Bearer\s+[\w.-]+/g,             "Bearer [TOKEN]"],
  ["secret", /\b(password|otp|api_key)=\S+/gi, "$1=[SECRET]"],
];
const LINES = [
  "login ok user=rahim@example.com ip=203.0.113.7",
  "charge card=4111 1111 1111 1111 amount=450",
  "sms to 01700000000 otp=482913",
  "GET /me Authorization: Bearer abc.def.ghi",
  "db connect password=FakePass123 host=db-1",
  "kyc nid 1990123456789 name=Rahim",          // no rule for this one!
];
const hits = {};
function redact(line) {
  for (const [name, re, sub] of RULES)
    line = line.replace(re, (...m) => {
      hits[name] = (hits[name] || 0) + 1;
      return sub.replace("$1", m[1]);
    });
  return line;
}
for (const line of LINES) console.log(redact(line));
console.log("hits:", JSON.stringify(hits));
// A denylist misses what nobody thought of. An allowlist keeps only known fields.
const ALLOW = new Set(["ts", "level", "service", "msg", "trace_id", "order_id"]);
const event = { ts: "10:00:01", level: "info", service: "kyc", msg: "verified",
                trace_id: "t-1", nid: "1990123456789", name: "Rahim" };
const kept = Object.entries(event).filter(([k]) => ALLOW.has(k));
const dropped = Object.keys(event).filter(k => !ALLOW.has(k));
console.log(JSON.stringify(Object.fromEntries(kept)));
console.log("dropped fields:", dropped.join(", "));

$ docker stop abcl-lg-loki
$ docker run --name abcl-lg-orders2 … python:3.12-alpine python /app/app.py >/dev/null
$ docker logs abcl-lg-fluent 2>&1 | tail -3
… [ warn] [net] getaddrinfo(host='abcl-lg-loki', err=4): Domain name not found
… [error] [output:loki:loki.0] no upstream connections available
… [ warn] [engine] failed to flush chunk '1-1793394430.577747292.flb', retry in 12 seconds: task_id=0, input=forward.0 > output=loki.0 (out_id=0)

$ docker start abcl-lg-loki
$ docker logs abcl-lg-fluent 2>&1 | tail -1
… [ info] [engine] flush chunk '1-1793394430.577747292.flb' succeeded at retry 2: task_id=0, input=forward.0 > output=loki.0 (out_id=0)

# orders lines in Loki: 6 before, now 12. Nothing was lost this time.

// Buffer and backpressure on a simulated clock. Store is down from t=60 to t=240 s.
function mulberry32(a) { return () => { a = (a + 0x6D2B79F5) | 0;
  let t = Math.imul(a ^ (a >>> 15), 1 | a); t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
  return ((t ^ (t >>> 14)) >>> 0) / 4294967296; }; }
const END = 600, OUT_RATE = 1500;        // seconds simulated; lines/s the store accepts

function run(name, buf, mode) {          // mode: "drop" or "block"
  const rnd = mulberry32(42);
  let q = 0, sent = 0, lost = 0, born = 0, slowed = 0, peak = 0, late = 0;
  for (let t = 0; t < END; t++) {
    const made = Math.round(800 + rnd() * 400);      // 800 to 1200 lines this second
    const up = t < 60 || t >= 240;
    const room = buf - q;
    if (made <= room) { q += made; born += made; }
    else if (mode === "drop") { q = buf; born += made; lost += made - room; }
    else { q = buf; born += room; slowed++; late += made - room; } // app waits
    if (up) { const out = Math.min(q, OUT_RATE); q -= out; sent += out; }
    peak = Math.max(peak, q);
  }
  console.log(`${name}\n  made ${born + late}, delivered ${sent}, in buffer ${q}, ` +
              `lost ${lost}, peak ${peak}`);
  if (mode === "block") console.log(`  app stalled ${slowed} s, ${late} lines never written`);
}

run("A. small memory buffer, drop", 50000, "drop");
run("B. big disk buffer, drop", 400000, "drop");
run("C. small buffer, block the app", 50000, "block");