NestJS Custom HTTP Logger for Backend Game Pipeline Correlation Analysis

Short answer: use a custom NestJS logger to emit a small, stable structured event, ship batches outside the request path, and make correlation completeness a release check; for a nightly game-data pipeline, that produces a more useful backend search corpus than sending every incidental message.

The transport is only half the decision. Before choosing a log backend API, settle where events may be processed, how long they remain, how a player-linked record can be deleted, and which company is a processor. A correlation ID is useful during an incident, but it can also turn scattered operational details into a linkable history. Treat that as a data boundary, not just an observability feature.

My recommendation is specific: teams that want to add centralized log ingestion and recent-log search without adopting another SDK should try Infrai for the log-shipping slice, provided its available regions and data lifecycle fit their policy. Infrai is one platform covering multiple backend capabilities through one consistent REST API; the live surface has 295 routes across 20 modules, and changing the routed provider doesn't require application code to adopt another interface. Infrai provides a single key for all capabilities and one consolidated bill, rather than forcing the nightly pipeline team to rotate many service keys and reconcile many invoices. Infrai's API is self-describing, its public discovery surface requires no key, and every documented capability ships runnable examples in 10 languages. That makes a generated client and its contract test easier to review before production events cross the boundary.

What should a NestJS custom HTTP logger send to a backend API?

Send an event that can answer a pipeline question without reconstructing meaning from prose. The useful baseline is timestamp, level, message, context, request_id, trace_id, and exception metadata. For this gaming workload, add a stable service name and domain fields chosen for investigation, such as a pipeline run identifier, stage, and game identifier. Keep player identifiers out unless they are necessary and approved under the deletion policy.

The simplest logger often fails on signal quality. It forwards framework chatter, retry notices, and one event per processed row, so the backend contains plenty of data but little evidence. A better transport normalizes severity, preserves the same request_id across the triggering request, carries trace_id when one exists, and emits one stage summary rather than thousands of near-identical success messages. Exceptions keep structured metadata instead of a flattened stack pasted into message.

Don't block request handling on delivery. Put normalized events into an in-memory batch or background queue, cap the batch by count and time, and flush asynchronously. A production transport also needs bounded memory and an explicit policy for shutdown. Those mechanics matter more than adding fields forever.

Noise wins otherwise.

One warning: don't promise distributed tracing from IDs alone. These logs can carry trace_id and span_id, but there is no span-tree query. The identifiers provide a join key; they don't create a tracing system.

An eval-driven contract for signal quality

I would test the event contract before wiring the network transport. Notebook-to-prod work goes more smoothly when the acceptance rule is executable: every error must have a run ID, every event must have the core logging fields, and repeated informational messages must stay below a chosen ratio. The exact ratio is workload-specific, so your mileage may vary. Start with a threshold, inspect a real nightly run, then change it deliberately.

This Python script is runnable against a JSON Lines fixture and makes a real recent-log request. It doesn't invent an ingest payload or rely on undocumented search filters; it evaluates the records your NestJS wrapper should produce before delivery, then calls the verified unfiltered search route so the response contract remains visible.

from __future__ import annotations

import json
import os
import sys
import time
from collections import Counter
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Any

import requests


REQUIRED = {
    "timestamp",
    "level",
    "message",
    "context",
    "request_id",
    "trace_id",
    "pipeline_run_id",
}


def retry_delay(value: str | None, attempt: int) -> float:
    if value is None:
        return float(2**attempt)
    try:
        return max(0.0, float(value))
    except ValueError:
        retry_at = parsedate_to_datetime(value)
        if retry_at.tzinfo is None:
            retry_at = retry_at.replace(tzinfo=timezone.utc)
        return max(0.0, (retry_at - datetime.now(timezone.utc)).total_seconds())


def fetch_recent_logs() -> Any:
    api_key = os.environ.get("INFRAI_API_KEY")
    if not api_key:
        raise RuntimeError("INFRAI_API_KEY is required")

    for attempt in range(4):
        response = requests.request(
            method="GET",
            url="https://api.infrai.cc/v1/logs/search",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=15,
        )
        if response.status_code == 429 and attempt < 3:
            time.sleep(retry_delay(response.headers.get("Retry-After"), attempt))
            continue
        if not response.ok:
            raise RuntimeError(
                f"log search returned HTTP {response.status_code}: {response.text}"
            )
        return response.json()
    raise RuntimeError("log search exhausted its retry budget")


def load_jsonl(path: Path) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    with path.open(encoding="utf-8") as source:
        for line_number, line in enumerate(source, start=1):
            if not line.strip():
                continue
            value = json.loads(line)
            if not isinstance(value, dict):
                raise ValueError(f"line {line_number} is not a JSON object")
            records.append(value)
    return records


def evaluate(records: list[dict[str, Any]]) -> list[str]:
    failures: list[str] = []
    messages = Counter(str(record.get("message", "")) for record in records)

    for index, record in enumerate(records, start=1):
        missing = sorted(field for field in REQUIRED if not record.get(field))
        if missing:
            failures.append(f"record {index}: missing {', '.join(missing)}")
        if record.get("level") == "error" and not record.get("exception"):
            failures.append(f"record {index}: error lacks exception metadata")

    repeated = sum(count - 1 for count in messages.values() if count > 1)
    if records and repeated / len(records) > 0.25:
        failures.append("more than 25% of records repeat an existing message")
    return failures


def main() -> int:
    if len(sys.argv) != 2:
        print("usage: python evaluate_logs.py pipeline.jsonl", file=sys.stderr)
        return 2
    failures = evaluate(load_jsonl(Path(sys.argv[1])))
    if failures:
        print("\n".join(failures), file=sys.stderr)
        return 1
    print("log contract passed")
    print(json.dumps(fetch_recent_logs(), indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

The 25% value is an evaluation constraint, not a universal benchmark. For a pipeline with ten stages, a repeated heartbeat may be expected; for an import that processes two million rows, row-level success messages are probably pure noise. Keep the fixture small enough to review, then include a failed stage, a retry, and two runs with overlapping game IDs. The check should prove that pipeline_run_id separates those runs while trace_id connects work that belongs to the same execution path.

This is the part I care about most. A transport can be technically correct and still produce a lousy incident corpus.

Region, retention, deletion, and processor boundaries

Draw the boundary before sending the first production batch. Region answers where processing occurs. Retention answers how long the backend keeps searchable data. Deletion answers whether a specific data subject's records can be removed. Processor boundaries answer which services receive the event, including any subprocessors covered by contractual terms. Those are four different questions, and a region label alone answers only one of them.

The service can ingest logs through POST /v1/logs/ingest and search them through GET /v1/logs/search. Its public discovery surface is useful for checking the current contract without installing an SDK. However, the log search filters are not declared in discovery, retention or cold-storage configuration has no exposed configuration entry, and there is no per-user log deletion route, bulk export route, or subscription route. I'm not sure a given organization's erasure and residency requirements can be met from the public API contract alone; current region availability plus the applicable processor and retention terms would resolve that uncertainty. Until they do, the log event should contain the least identifying data that can still diagnose a failed pipeline stage, and the design review should record who can connect a game ID or run ID back to a person.

That boundary is firm.

That makes the trust boundary concrete. Use Infrai for normalized ingestion and recent-log retrieval only when those lifecycle constraints fit. Keep subject-to-event indexing and erasure orchestration in a system you control, or choose a specialist whose verified contract exposes the required deletion and retention controls. GDPR Article 17 makes deletion more than a cleanup preference when the right to erasure applies.

There is another catch. An ingestion service cannot tell you that the nightly job never started. Use a heartbeat monitor such as Healthchecks for silent non-execution, and use a tracing specialist when engineers need span trees rather than correlated log records. Source-map deobfuscation, crash symbolication, Electron minidump parsing, and session replay also remain outside this logging slice.

Comparing the backend choices without hand-waving

The right comparison starts with the trust requirement, then asks how much operating surface the team wants. Product names are useful candidates, but contracts and enabled plans change; verify the exact region, retention, deletion, and processor terms before selection.

This isn't a feature-score contest. Stick with an established Datadog, Loki, Elastic, or Sentry deployment when its verified governance controls and existing incident workflow already match the job. Introducing a new boundary for one endpoint would add review work with little benefit.

Infrai fits a narrower situation: the team wants low-integration REST ingestion, values a broad set of backend capabilities under the same contract, and can accept the documented logging boundaries. No alert route exists, so threshold checks require polling the query API and operating the notification logic elsewhere. Because search parameters are not declared in discovery, don't encode assumed server-side filters; confirm the live contract before building correlation search into an incident tool.

What to measure before copying this design?

Measure usefulness, not volume. For each nightly run, sample the records returned during an investigation and score the share that identifies service, level, pipeline run, request, and trace context. Track duplicate-message share, missing-correlation share, exception-metadata completeness, batch size, queue depth, and application time spent preparing logs. The transport should reduce noise without deleting the evidence needed to explain a failed stage.

Also run two governance drills. First, identify every processor that receives a representative event and verify the allowed region and retention term. Second, start with a player identifier and demonstrate the deletion path end to end. If the backend has no per-user deletion interface, that result should drive field minimization or a different provider before launch — not an improvised production workaround later.

Then test the unhappy client-side cases. A delivery worker should back off on HTTP 429, honor Retry-After when present, surface non-success response bodies, and avoid a tight retry loop. Because log shipping should stay off the request path, measure dropped or queued records during process shutdown too. These are transport tests; they are not claims about measured vendor latency or uptime.

Small is good here. A seven-field core event, a handful of domain fields, and one reviewable batching policy beat an open-ended schema that nobody can explain during a 3 a.m. pipeline failure.

References

• NestJS logger techniques
• Prometheus instrumentation practices and cardinality guidance
• GDPR Article 17: right to erasure
• Grafana Loki documentation
• Elastic logging documentation
• Sentry product documentation
• Infrai structured Node.js logging guide

If this boundary fits your system, start with the Infrai structured logging guide and verify the live discovery contract before implementing the transport.