Use a metrics dashboard for recurring admin analytics, keep structured logs searchable for investigation, and derive both from the same durable pipeline events. For a Node.js SaaS marketplace, that is the easiest backend choice that still gives each team a defensible nightly cost attribution.
The three useful signals are processed items, consumed units, and failed items. They answer different questions: how much work finished, what drove spend, and which records need inspection. A dashboard can make the first two cheap to scan; log search preserves the dimensions and error context behind the third. Neither store should be the only record of what the pipeline did.
Small distinction. Big consequences.
How should Node.js SaaS admin analytics combine metrics and logs search?
Treat the nightly marketplace run as a data product with three outputs. The first is an append-only event record for each completed work unit. The second is a small set of aggregates keyed by billing day, tenant, pipeline stage, and workload class. The third is a searchable log document that retains identifiers and diagnostic context but excludes secrets and unrestricted payloads. The Node.js worker can emit the source event; the example below uses Python because a notebook-friendly reducer is handy for validating the contract before the production job consumes it.
This separation is less convenient than pointing the admin dashboard straight at a log index. The catch is that direct log queries quietly turn presentation choices into accounting rules: a renamed field, a changed retention window, sampling, or duplicate delivery can alter a chart. It is also less flexible than allowing every admin to search every attribute. That constraint is deliberate. Cost attribution needs a narrow, reviewed schema; debugging needs richer context and tighter access control.
A useful mental model is “count, cost, context.” processed_items is the count. input_units and output_units are cost drivers rather than currency, because rates can change independently of workload. The log document supplies context such as run_id, item_id, stage, duration, and a bounded error code. Currency can be calculated later by joining versioned rates to the recorded units. Don't bake a mutable price into every operational log line and then expect an admin total to remain reproducible.
Metrics alone are not suitable when an operator must find the exact listing that failed validation, compare retries, or inspect a rare attribute combination. Stick with log search for those investigations. Logs alone are a poor dashboard backend when totals must be repeatable across retention changes, access scopes, and evolving document shapes. If the output becomes invoice-grade, use a dedicated usage ledger and reconciliation process; an observability backend is evidence, not the financial system of record.
Build the dual path before choosing a backend
Start with a contract that can be produced by the Node.js service and reduced anywhere. This runnable Python example accepts newline-delimited JSON from a nightly catalog-enrichment pipeline, rejects malformed records, deduplicates completed work by event ID, emits metric rows, and prepares a bounded document for log search. It uses only the standard library, so it can move from a notebook check into a batch test without an SDK decision getting in the way.
from __future__ import annotations
import json
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
ALLOWED_STAGES = {"extract", "enrich", "publish"}
ALLOWED_OUTCOMES = {"ok", "error"}
@dataclass(frozen=True)
class PipelineEvent:
event_id: str
occurred_at: datetime
tenant_id: str
run_id: str
item_id: str
stage: str
outcome: str
input_units: int
output_units: int
duration_ms: int
error_code: str | None
@classmethod
def from_dict(cls, raw: dict[str, Any]) -> "PipelineEvent":
event = cls(
event_id=str(raw["event_id"]),
occurred_at=datetime.fromisoformat(
str(raw["occurred_at"]).replace("Z", "+00:00")
).astimezone(timezone.utc),
tenant_id=str(raw["tenant_id"]),
run_id=str(raw["run_id"]),
item_id=str(raw["item_id"]),
stage=str(raw["stage"]),
outcome=str(raw["outcome"]),
input_units=int(raw.get("input_units", 0)),
output_units=int(raw.get("output_units", 0)),
duration_ms=int(raw["duration_ms"]),
error_code=str(raw["error_code"]) if raw.get("error_code") else None,
)
if event.stage not in ALLOWED_STAGES:
raise ValueError(f"unknown stage: {event.stage}")
if event.outcome not in ALLOWED_OUTCOMES:
raise ValueError(f"unknown outcome: {event.outcome}")
if min(event.input_units, event.output_units, event.duration_ms) < 0:
raise ValueError("unit and duration values must be non-negative")
return event
def read_events(path: Path) -> Iterable[PipelineEvent]:
seen: set[str] = set()
with path.open(encoding="utf-8") as source:
for line_number, line in enumerate(source, start=1):
if not line.strip():
continue
try:
event = PipelineEvent.from_dict(json.loads(line))
except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
raise ValueError(f"invalid event on line {line_number}: {exc}") from exc
if event.event_id in seen:
continue
seen.add(event.event_id)
yield event
def aggregate(events: Iterable[PipelineEvent]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
totals: dict[tuple[str, str, str, str], dict[str, int]] = defaultdict(
lambda: {
"processed_items": 0,
"failed_items": 0,
"input_units": 0,
"output_units": 0,
"duration_ms": 0,
}
)
searchable: list[dict[str, Any]] = []
for event in events:
billing_day = event.occurred_at.date().isoformat()
key = (billing_day, event.tenant_id, event.stage, event.outcome)
row = totals[key]
row["processed_items"] += 1
row["failed_items"] += int(event.outcome == "error")
row["input_units"] += event.input_units
row["output_units"] += event.output_units
row["duration_ms"] += event.duration_ms
searchable.append(
{
"event_id": event.event_id,
"occurred_at": event.occurred_at.isoformat(),
"tenant_id": event.tenant_id,
"run_id": event.run_id,
"item_id": event.item_id,
"stage": event.stage,
"outcome": event.outcome,
"duration_ms": event.duration_ms,
"error_code": event.error_code,
}
)
metric_rows = []
for (day, tenant, stage, outcome), values in sorted(totals.items()):
metric_rows.append(
{
"billing_day": day,
"tenant_id": tenant,
"stage": stage,
"outcome": outcome,
**values,
}
)
return metric_rows, searchable
if __name__ == "__main__":
metrics, log_documents = aggregate(read_events(Path("pipeline-events.ndjson")))
print(json.dumps({"metrics": metrics, "logs": log_documents}, indent=2))
The strict parse is intentional. An unknown stage should stop the reduction rather than create a nearly identical dashboard series that nobody notices. Unit values remain integers, timestamps become UTC, and raw listing text never enters the search document. Those choices make the eval harness straightforward: feed fixtures through the same reducer, then compare exact rows instead of taking screenshots of charts.
Duplicates count once.
I would test at least four fixtures: one valid event, a duplicate event, an unknown stage, and two tenants with the same item_id. The last case catches an easy isolation mistake. Property-based tests can go further by asserting that shuffling input does not change aggregate rows and that every accepted event increments exactly one processed_items bucket. The production writer and notebook reducer need the same contract tests even if they are implemented in different languages.
Make cost attribution boring
Admin analytics often begins with a request such as “show nightly AI cost by seller.” The tempting implementation groups logs by tenant_id and sums a field called cost. That works until retries create two completion lines, a prompt revision changes the unit mix, a rate changes halfway through a reporting period, or sampling removes ordinary successes while preserving errors. Then the chart still looks plausible. That is the dangerous state.
Prefer stable dimensions with bounded cardinality for the dashboard: tenant, pipeline stage, outcome, workload class, model class if it is part of the approved cost policy, and billing day. Keep item IDs, run IDs, prompt hashes, and error details in searchable records. A metric label per marketplace listing may feel powerful in a demo, but it pushes an unbounded identifier into a system optimized for aggregation. Search is the right interaction for “find listing L-18427”; a dashboard is the right interaction for “compare enrichment units across tenants last night.”
Sampling requires special care. OpenTelemetry distinguishes head sampling, where the decision is made near the start of a trace, from tail sampling, where the decision can use information from the completed trace. Either approach changes which trace records remain available. That can be appropriate for diagnostic telemetry, but a sampled stream cannot silently become the source for complete usage totals. Keep the durable usage event outside that sampling decision, then correlate retained telemetry with run_id or another approved identifier.
No sampled totals.
I'm not sure there is one retention period that fits every marketplace; legal, support, and seasonality needs vary too much. The decision can still be explicit. Retain aggregate usage long enough for the reporting and reconciliation window, retain searchable context for the shortest operationally useful period, and document what an admin query can no longer answer after expiry. Prompt-cost awareness belongs here too: store measured input and output units under a versioned event contract, but don't store full prompts merely because search makes that convenient.
Backfill policy matters more than dashboard polish. When the schema changes, write a new version and rerun the reducer from durable events into a separate aggregate partition. Compare old and new totals before switching reads. If historical source events are unavailable, label the series boundary rather than manufacturing continuity from partial logs.
Operate it without turning telemetry into a ledger
Before deployment, define the owner of the event schema, the deduplication key, UTC day boundaries, late-arrival handling, and the exact rate table version used to convert units into money. Run contract fixtures in the Node.js producer and the Python evaluator. In staging, compare the count of accepted source events with the sum of processed_items; separately verify that every error row can be found through the permitted log-search scope. Alert on reconciliation drift, parser rejection counts, and missing nightly partitions, not on a chart's appearance.
Access deserves its own review. A marketplace operator may need cross-tenant totals while a support engineer needs record-level context for one tenant. Those are different permissions. Redact before indexing, avoid prompt and listing bodies by default, and make the dashboard query a curated aggregate rather than forwarding arbitrary filters from an admin browser. The easiest backend is the one that keeps these boundaries visible, even if it requires two read paths.
Permissions are architecture.
During an incident, start with the aggregate discontinuity, pivot using the bounded dimensions, and then search by run_id for exact records. After a replay, reconcile event IDs before refreshing the aggregate partition. This is where the split pays for itself: operators get context without asking a high-cardinality dashboard to behave like a search engine, while finance-facing totals do not depend on whichever diagnostic records happened to be retained.
The final limitation is organizational. A dual path asks a team to maintain a schema and a small reducer. For a tiny internal tool with no tenant billing, no replay, and a handful of nightly records, that may be needless machinery; a structured file plus a basic search interface can be enough. Once admins compare tenants, attribute variable AI work, or dispute a total, move the repeated questions into reviewed aggregates and preserve the source events that can reproduce them.
Ship the contract first. Backend selection gets much easier after the correctness boundary is visible.
Further reading
- OpenTelemetry, “Sampling”: https://opentelemetry.io/docs/concepts/sampling/
Top comments (0)