Chapter Pipeline Logging di section Data Quality dan Observability menuntaskan logging di dalam satu proses: baris JSON terstruktur, run id di setiap baris, dan event baku per stage. Pipeline yang dideploy jarang semudah itu — satu run biasanya melewati beberapa layanan yang berjalan terpisah: layanan ekstraksi menulis berkas staging, layanan pemuatan membacanya satu jam kemudian, dan saat hasil pagi salah, kronologinya tersebar di banyak berkas log yang saling asing. Chapter ini menyambungnya: correlation id yang diteruskan lintas layanan, trace yang merekam bentuk pohon dari seluruh pekerjaan, dan pola agregasi log terpusat yang membuat pertanyaan “apa yang terjadi pada run ini” dapat dijawab dari satu tempat.
Korelasi Lintas Layanan #
Kuncinya satu nilai yang sama di mana-mana. Pada layanan HTTP, peran ini dipegang header X-Request-Id yang diteruskan dari panggilan ke panggilan; pada pipeline batch, pembawanya bisa berupa apa pun yang ikut mengalir bersama data — di demo ini, berkas manifest.json di staging. Layanan pertama membuat correlation id dan menuliskannya ke manifest; layanan kedua membaca manifest sehingga log keduanya membawa run_id yang identik. Tanpa penanganan ini, layanan kedua cenderung membuat id barunya sendiri dan kronologi run terpotong dua.
Demo: Dua Layanan dengan Correlation Id #
Demo berikut memakai run_id turunan tanggal batch agar keluaran deterministik; durasi memakai JamPalsu — jam simulasi yang maju eksplisit, bukan hasil pengukuran nyata — sementara penulisan log, span, staging, dan manifest semuanya berkas sungguhan. Tiga komponen pertama:
class JamPalsu:
"""Jam simulasi: waktu maju eksplisit, bukan hasil pengukuran nyata."""
def __init__(self):
self.detik = 0.0
def maju(self, durasi):
self.detik += durasi
def now(self):
return self.detik
def catat_log(layanan, trace_id, pesan, jam, **kolom):
"""Satu baris log JSON terstruktur, mirip chapter Pipeline Logging."""
baris = {"ts": round(jam.now(), 3), "layanan": layanan,
"run_id": trace_id, "pesan": pesan, **kolom}
with open(AKAR / "logs" / f"{layanan}.log", "a", encoding="utf-8") as f:
f.write(json.dumps(baris) + "\n")
@contextmanager
def span(nama, trace_id, parent, jam):
"""Catat satu span: mulai saat masuk blok, selesai saat keluar."""
sid = hashlib.sha256(f"{trace_id}:{nama}".encode()).hexdigest()[:8]
mulai = jam.now()
try:
yield sid
finally:
rekaman = {"trace_id": trace_id, "span_id": sid, "parent": parent,
"nama": nama, "mulai": round(mulai, 3),
"selesai": round(jam.now(), 3)}
with open(AKAR / "spans" / "spans.jsonl", "a", encoding="utf-8") as f:
f.write(json.dumps(rekaman) + "\n")Kedua layanan mengalir lewat komponen tersebut:
def layanan_ekstraksi(trace_id, jam):
catat_log("ekstraksi", trace_id, "run diterima", jam, sumber="api-pesanan")
with span("ekstraksi", trace_id, None, jam) as sp:
with span("parse", trace_id, sp, jam):
baris = [b.split(",") for b in BARIS_MENTAH]
jam.maju(0.4)
catat_log("ekstraksi", trace_id, "parse selesai", jam,
baris_masuk=len(baris))
with span("validate", trace_id, sp, jam):
valid = []
for id_trx, qty, harga in baris:
if int(qty) <= 0:
catat_log("ekstraksi", trace_id, "baris ditolak", jam,
id_pesanan=id_trx, alasan="qty <= 0")
else:
valid.append({"id": id_trx, "nilai": int(qty) * int(harga)})
jam.maju(0.2)
jam.maju(0.6) # tulis berkas staging
with open(AKAR / "staging" / "pesanan.jsonl", "w") as f:
f.writelines(json.dumps(b) + "\n" for b in valid)
manifest = {"run_id": trace_id, "jumlah_baris": len(valid)}
(AKAR / "staging" / "manifest.json").write_text(json.dumps(manifest))
catat_log("ekstraksi", trace_id, "staging tertulis", jam,
baris_valid=len(valid))
def layanan_pemuatan(jam):
manifest = json.loads((AKAR / "staging" / "manifest.json").read_text())
trace_id = manifest["run_id"] # correlation id diteruskan lewat manifest
catat_log("pemuatan", trace_id, "manifest dibaca", jam,
baris=manifest["jumlah_baris"])
with span("pemuatan", trace_id, None, jam):
jam.maju(0.5) # baca staging, tulis tabel warehouse
catat_log("pemuatan", trace_id, "pemuatan selesai", jam,
tabel="warehouse.pesanan")Driver menjalankan kedua layanan, mencetak isi kedua berkas log, memeriksa run_id di keduanya, lalu menyusun span menjadi pohon berdasarkan kolom parent. Keluarannya:
Output yang Dihasilkan:
== logs/ekstraksi.log ==
{"ts": 0.0, "layanan": "ekstraksi", "run_id": "e85c5195", "pesan": "run diterima", "sumber": "api-pesanan"}
{"ts": 0.4, "layanan": "ekstraksi", "run_id": "e85c5195", "pesan": "parse selesai", "baris_masuk": 4}
{"ts": 0.4, "layanan": "ekstraksi", "run_id": "e85c5195", "pesan": "baris ditolak", "id_pesanan": "TRX-903", "alasan": "qty <= 0"}
{"ts": 1.2, "layanan": "ekstraksi", "run_id": "e85c5195", "pesan": "staging tertulis", "baris_valid": 3}
== logs/pemuatan.log ==
{"ts": 1.2, "layanan": "pemuatan", "run_id": "e85c5195", "pesan": "manifest dibaca", "baris": 3}
{"ts": 1.7, "layanan": "pemuatan", "run_id": "e85c5195", "pesan": "pemuatan selesai", "tabel": "warehouse.pesanan"}
== Korelasi lintas layanan ==
run_id ekstraksi: ['e85c5195']
run_id pemuatan : ['e85c5195']
-> sama; satu run dapat direkonstruksi dari dua berkas log
== Trace (span tree) ==
ekstraksi [0.0s-1.2s] 1.2s
├─ parse [0.0s-0.4s] 0.4s
└─ validate [0.4s-0.6s] 0.2s
pemuatan [1.2s-1.7s] 0.5sMari kita telaah bagian-bagian kuncinya:
span()sebagai context manager:mulaidicatat saat masuk blok danselesaisaat keluar —finallymemastikan span tetap tercatat walau blok mengangkat exception;span_iddihasilkan dari hashtrace_iddan nama sehingga deterministik.parentdi rekaman span: spanparsedanvalidatemenyimpan id spanekstraksisebagai induk; itulah satu-satunya informasi yang dibutuhkan untuk menyusun pohon.- Baris
manifest = ...dilayanan_pemuatan: korelasi terjadi tepat di sini — layanan pemuatan tidak membuat id sendiri, melainkan membacanya dari manifest hasil layanan ekstraksi. - Nilai
e85c5195sama di dua berkas: pemeriksaan korelasi membaca kedua berkas dan membandingkan kumpulanrun_id; hasil sama berarti seluruh kronologi satu run bisa direkonstruksi dengan satu filter.
Trace, Sampling, dan OpenTelemetry #
Pohon span menjawab pertanyaan yang tidak bisa dijawab log: pekerjaan apa yang paling mahal. Pada keluaran di atas, ekstraksi memakan 1,2 detik dengan parse 0,4 detik dan validate 0,2 detik — sisa 0,6 detiknya ada di penulisan staging; pemuatan hanya 0,5 detik. Ketika run mendadak memburuk, lonjakan durasi di satu cabang pohon langsung menunjuk tersangkanya. Volume adalah kendalanya: mencatat span untuk setiap baris data mustahil, jadi span dibuat per tahap yang berarti, dan pada lalu lintas besar dipakai sampling — run yang sukses dicatat sebagian, run yang gagal dicatat seluruhnya. Standar industri untuk hal ini adalah OpenTelemetry
: satu pustaka dengan konsep trace, span, dan context propagation lintas bahasa, menggantikan format buatan sendiri seperti demo di atas.
Agregasi Log Terpusat #
Dari mesin produksi, log JSONL dikirim oleh agent (Promtail untuk Loki, Filebeat untuk Elasticsearch) ke satu penyimpanan yang bisa dicari; pola pengiriman dan pencarian ini tidak memerlukan eksekusi lokal untuk dipahami. Setelah terkumpul, query “semua baris dengan run_id e85c5195” menampilkan kronologi lintas layanan dalam satu tampilan terurut — persis hasil pemeriksaan korelasi di demo, tanpa membuka dua berkas. Atur retensi per jenis: log error lebih lama dari log info, dan pastikan label yang dipakai filter (layanan, pipeline, run_id) memang selalu terisi.
Kronologi yang bisa disusun kembali dengan cepat adalah bekal utama saat alert P1 berbunyi tengah malam. Bab berikutnya menyusun proses lengkapnya: bagaimana tim merespons insiden, mengoordinasikan perbaikan, dan belajar dari pasca-insiden — materi chapter Incident Response.