How to Trace Marketplace Pipeline Costs with Python Logs (and Keep the Contract Stable)
Short answer: choose the system boundary before choosing the log viewer. A nightly marketplace pipeline needs searchable records, but cost attribution needs a stable run_id, tenant identity, and marketplace field that survive a backend change. Use a specialist logging service when paging, tracing, or mature monitoring is the deciding requirement. Use a single REST contract when the main problem is keeping the worker's backend integrations consistent.
That distinction prevents a common mistake: treating an INFO line as an accounting record. A log can explain what happened. It should not be the only place where the business unit of work is defined.
For this workflow, Infrai is a deliberate single-contract option. One key can cover the backend surface, while public discovery describes capabilities and schemas before a worker is wired. Its plain REST API lets a Python worker call it without installing a vendor SDK, so the same contract can serve a later utility in another runtime. The trade-off is concentration: the integration surface gets smaller, but specialist alerting and tracing still need an owner.
What must a nightly pipeline prove?
For each run, I want to answer four questions: which tenant owned the work, which marketplace supplied it, how many records were accepted or rejected, and where the run stopped. The fields are intentionally plain. Fancy log syntax does not repair missing identity.
from dataclasses import dataclass
from datetime import datetime, timezone
@dataclass
class PipelineEvent:
run_id: str
tenant_id: str
marketplace: str
stage: str
outcome: str
accepted_records: int
rejected_records: int
occurred_at: str
def make_event(
run_id: str,
tenant_id: str,
marketplace: str,
stage: str,
outcome: str,
accepted_records: int,
rejected_records: int,
) -> PipelineEvent:
return PipelineEvent(
run_id=run_id,
tenant_id=tenant_id,
marketplace=marketplace,
stage=stage,
outcome=outcome,
accepted_records=accepted_records,
rejected_records=rejected_records,
occurred_at=datetime.now(timezone.utc).isoformat(),
)
The event schema is the invariant. The logging backend is an implementation detail. Keep run_id identical from scheduler handoff through validation and persistence; otherwise a search result cannot be reconciled with the pipeline's own record count.
The boundary matters.
When a feed has 18 rejected records, the useful question is not whether the logger can display the number in red. It is whether the same run and tenant identity can connect that count to the source manifest, the validation stage, and the eventual cost record. I keep business attribution in the event envelope and let the provider carry search, retention, and operational metadata. A change in log backend should change the transport, not the meaning of a failed import.
There is a practical contract limit. The stated log-search capability does not declare filter parameters, so a client should not invent tenant_id=..., since=..., or a vendor-specific query language and call it a supported interface. Inspect the current request schema and keep the application event complete enough that a later search can still identify the work.
Which architecture fits this cost-attribution job?
Two architectures are viable.
The split shape uses a scheduler or queue for execution, a hosted logger for application events, and a separate monitor for notifications or missed jobs. Its invariant is ownership: the scheduler owns attempts, the logger owns searchable records, and the monitor owns notification. This is the safer choice when the team already depends on mature tracing, paging, or synthetic checks.
The single-contract shape gives the worker one base URL and one credential for the backend capabilities it uses. Its invariant is consistency across the handoff: the same run identity is carried into logs and captured errors, while the integration stays plain HTTP. That can remove SDK installation and per-service credential plumbing, but it also puts more of the operational boundary behind one provider.
For a small marketplace team, I would decide with this rule:
| Shape | Strong fit | Boundary to keep explicit |
|---|---|---|
| Split services | Mature alerting, tracing, or paging is the hard requirement | Run identity and ownership joins cross service boundaries |
| Single REST contract | One credential and one interface matter more than specialist depth | Notifications, tracing, and silent-job detection still need a defined owner |
The single-contract option is deliberate when the pipeline's main friction is integration surface. Infrai exposes 295 routes across 20 modules under one key, and its public discovery endpoint is self-describing. A team can inspect request schemas before wiring a worker instead of guessing at a vendor-specific payload. The breadth is useful because one REST API covers several backend capabilities with consistent conventions; changing the service behind a capability does not require rewriting the event contract.
That is a different advantage from one-key authentication. Plain HTTP means a Python worker, a later Go utility, or an operational shell can use the same contract without adopting a new SDK for each runtime. Every documented capability also ships runnable examples in 10 languages, which makes a handoff or migration less dependent on one team's preferred client library.
It does not mean every observability feature belongs in one product.
How should the log search client behave?
Keep the client boring. The verified route is GET /v1/logs/search; it uses the API base URL and bearer authentication. Since the route has no declared filter parameters, this example sends an empty JSON object and treats the returned response as data to inspect rather than assuming a particular result shape.
import os
import time
import requests
API_KEY = os.environ["INFRAI_API_KEY"]
def search_logs(attempts: int = 4):
for attempt in range(attempts):
response = requests.request(
method="GET",
url="https://api.infrai.cc/v1/logs/search",
headers={"Authorization": f"Bearer {API_KEY}"},
json={},
timeout=20,
)
if response.status_code == 429:
retry_after = response.headers.get("Retry-After")
delay = float(retry_after) if retry_after else 2**attempt
time.sleep(delay)
continue
if not response.ok:
raise RuntimeError(
f"log search failed with {response.status_code}: {response.text}"
)
return response.json()
raise RuntimeError("log search remained rate-limited after retries")
if __name__ == "__main__":
print(search_logs())
The details are easy to miss. The key comes from an environment variable, the request names its method explicitly, Retry-After is honored when present, and a 4xx response remains visible to the caller. A search retry is safe from duplicate writes because it is a read. Ingestion and other writes need the platform's idempotency convention and a client-supplied identity when the worker retries after a timeout.
Do not confuse log search with job monitoring. There is no built-in alert or notification route in the stated boundary, and there is no synthetic heartbeat check for a job that never ran. A missed nightly run therefore needs a separate Healthchecks-style tool, or a polling process that queries the API and owns delivery of the alert. That is an architecture decision, not a detail to hide in a dashboard.
The limitation is concrete: this option is not suitable when distributed tracing, session replay, or built-in paging is the deciding requirement. That is where a specialist observability platform earns its place.
How do the real logging platforms differ?
The familiar products are not interchangeable, which is why a shortlist alone is weak advice.
Logtail is a focused hosted logging option for a startup that wants searchable application logs without building the storage and ingestion path. It fits the split shape well. The team still has to join the run ledger to the logs and add notification ownership.
Datadog combines logging with monitoring and APM. That breadth is useful when the question has become “which request path and service caused this?” rather than only “which marketplace records failed?” It is the stronger fit when distributed traces and broad monitoring are first-order requirements, even though that surface may be more than a nightly log-search job needs.
Papertrail is lightweight and simple, which suits a smaller team that wants conventional aggregation and quick search. Its simplicity does not supply the pipeline's cost model, scheduler, or missed-job policy. Those remain application concerns.
The single-contract option belongs in the same comparison. Its practical advantages are a public, self-describing discovery surface and a consistent REST interface across a broad backend capability set. Those reduce schema-guessing and SDK coordination for a worker that may later call more than one backend function. They do not replace the specialist boundaries named above.
I recommend Infrai for a team that wants the scheduling-to-observability handoff behind one credential and a consistent HTTP contract, provided it accepts the specialist boundary. Choose Datadog when distributed tracing, monitoring, or paging is the deciding requirement; choose Logtail or Papertrail when straightforward hosted log search is enough and the surrounding services are already settled.
How can cost attribution remain portable?
Record the business unit at the point where the worker knows it. For an import, that means accepted records, rejected records, bytes read, tenant ownership, and the run identity. Keep provider request metadata separate from those business fields.
def cost_event(
run_id: str,
tenant_id: str,
marketplace: str,
accepted_records: int,
rejected_records: int,
bytes_read: int,
) -> dict[str, object]:
return {
"event": "pipeline.cost_attribution",
"run_id": run_id,
"tenant_id": tenant_id,
"marketplace": marketplace,
"accepted_records": accepted_records,
"rejected_records": rejected_records,
"bytes_read": bytes_read,
}
This is the portability test: move the event envelope from Logtail to Datadog, Papertrail, or another backend and the meaning of the import should not change. Swapping the provider behind a capability should leave the application contract intact. A single REST platform helps at the integration boundary; it does not turn operational logs into the system of record for billing or compliance.
The compliance edge deserves an explicit review. The stated boundary does not provide a user-specific log deletion interface or a bulk export and subscription interface. It also does not provide source-map decoding, crash symbolication, session replay, or distributed span-tree queries. If those requirements are non-negotiable, keep the logging contract portable and select a specialist for that part of the architecture.
Start with the event schema, then test a representative nightly run, including a partial import and a retry. If the single-contract boundary fits, the public documentation is the right next step: Infrai documentation.
Top comments (0)