Distributed Tracing — request-এর পুরো জীবন দেখা

"""trace_collector.py — ingestion, out-of-order assembly, waterfall,
critical-path, orphan/skew diagnostics (~110 line)"""
import json


class TraceCollector:
    """Backend-collector-স্বাদ: spans যেকোনো order-এ আসবে বলেই ধরে নাও"""

    def __init__(self):
        self.traces = {}                     # trace_id -> {span_id: span}

    def ingest(self, raw):                   # str-json বা dict, order-free
        sp = json.loads(raw) if isinstance(raw, str) else dict(raw)
        self.traces.setdefault(sp['trace_id'], {})[sp['span_id']] = sp

    def spans(self, tid):
        return list(self.traces.get(tid, {}).values())

    def by_id(self, tid):
        return self.traces.get(tid, {})

    def children_map(self, tid):
        ch = {}
        for s in self.spans(tid):
            ch.setdefault(s['parent_span_id'], []).append(s)
        for lst in ch.values():
            lst.sort(key=lambda s: s['start_ms'])
        return ch

    def roots(self, tid):
        ids = set(self.by_id(tid))
        return [s for s in self.spans(tid)
                if s['parent_span_id'] not in ids]

    def critical_path(self, tid):   # root→leaf max cumulative (simplified!)
        ch = self.children_map(tid)
        memo = {}

        def best(s):
            if s['span_id'] not in memo:
                kids = ch.get(s['span_id'], [])
                kids = ch.get(s['span_id'], [])
                if kids:
                    bk, bv = max((best(k) for k in kids),
                                 key=lambda t: t[1])
                    memo[s['span_id']] = ([s] + bk, s['duration_ms'] + bv)
                else:
                    memo[s['span_id']] = ([s], s['duration_ms'])
            return memo[s['span_id']]

        rs = self.roots(tid)
        return max((best(r) for r in rs), key=lambda t: t[1])[0] if rs else []


    def waterfall(self, tid, width=58):
        spans = sorted(self.spans(tid), key=lambda s: s['start_ms'])
        if not spans:
            print('(empty trace)')
            return
        cp_ids = {s['span_id'] for s in self.critical_path(tid)}
        origin = min(s['start_ms'] for s in spans)
        total = max(max(s['start_ms'] + s['duration_ms'] for s in spans)
                    - origin, 1e-9)
        for s in spans:
            off = round((s['start_ms'] - origin) / total * width)
            w = max(1, round(s['duration_ms'] / total * width))
            star = ' <-critical' if s['span_id'] in cp_ids else ''
            print(f"{s['name']:>12} │{'·'*off}{'█'*w}"
                  f" {s['duration_ms']:6.1f}ms{star}")

    def diagnostics(self, tid):
        byid = self.by_id(tid)
        issues = []
        for s in sorted(self.spans(tid), key=lambda x: x['start_ms']):
            pid = s['parent_span_id']
            if pid and pid not in byid:
                issues.append(f"ORPHAN  : '{s['name']}'-র parent {pid} "
                              f"এই trace-তে নেই (dropped export?)")
                continue
            p = byid.get(pid)
            if p:
                if s['start_ms'] < p['start_ms']:
                    issues.append(f"CLOCK-SKEW: '{s['name']}' parent-এর "
                                  f"{p['start_ms'] - s['start_ms']:.1f}ms আগে শুরু!")
                over = (s['start_ms'] + s['duration_ms']) - \
                       (p['start_ms'] + p['duration_ms'])
                if over > 1e-6:
                    issues.append(f"OVERRUN : '{s['name']}' parent-কে "
                                  f"{over:.1f}ms পেরিয়ে শেষ")
        return issues


def sp(trace_id, span_id, parent, name, start, dur):
    return dict(trace_id=trace_id, span_id=span_id, parent_span_id=parent,
                name=name, start_ms=start, duration_ms=dur)

c = TraceCollector()
T = 'tr-42'
SHUFFLED = [                                   # ইচ্ছা-করে এলোমেলো order
    sp(T, 'i1', 'c1', 'inventory', 15, 503),
    sp(T, 'o7', 'zz9', 'lost-span', 400, 10),      # orphan!
    sp(T, 'a9', 'g1', 'auth-check', -3, 33),       # clock-skew!
    sp(T, 'g1', '', 'gateway', 0, 562),
    sp(T, 'd4', 'c1', 'db-query', 40, 38),
    sp(T, 'c1', 'g1', 'cart', 8, 537),
]
for s in SHUFFLED:
    c.ingest(s)

print('waterfall (trace tr-42):')
c.waterfall(T)
chain = c.critical_path(T)
cum = sum(s['duration_ms'] for s in chain)
print('\ncritical path:', ' -> '.join(s['name'] for s in chain),
      f'(cumulative {cum:.0f}ms)')
print('\ndiagnostics:')
for line in c.diagnostics(T):
    print(' ', line)

নমুনা output:

waterfall (trace tr-42):
  auth-check │███   33.0ms
     gateway │██████████████████████████████████████████████████████████  562.0ms <-critical
        cart │·███████████████████████████████████████████████████████  537.0ms <-critical
   inventory │··████████████████████████████████████████████████████  503.0ms <-critical
    db-query │····████   38.0ms
   lost-span │·········································█   10.0ms

critical path: gateway -> cart -> inventory (cumulative 1602ms)

diagnostics:
  CLOCK-SKEW: 'auth-check' parent-এর 3.0ms আগে শুরু!
  ORPHAN  : 'lost-span'-র parent zz9 এই trace-তে নেই (dropped export?)

তিনটা পাঠ:

• Ingest-order এলোমেলো হলেও tree perfect — order-এর ভরসা timestamp না, parent_edge; production-collector-এর প্রথম design-rule।
• Critical-path cumulative 1602 > wall 565 — overlapping parent-duration double-count-এর সৎ স্বীকারোক্তি; real tools interval-arithmetic-এ refine করে, teaching-version intent-ই দেখায়।
• Diagnostics demo-data-তে দুই bug-class ধরল — dropped-export (orphan), unsynced-clock (skew); তৃতীয় detector (overrun — instrumentation-bug) trigger করতে inventory-র duration ৫৬০ করে চালান।

নিজে বাড়ান:

1. Interval-arithmetic critical-path — overlap বাদ দিয়ে true self-time chain
2. Per-service latency histogram — হাজার trace-ingest করে p50/p99 রিপোর্ট
3. Tail-sampling filter — keep-errors + keep-slowest policy, capture-ratio মাপুন