Pino and Winston to an HTTP log ingest API, with request and user ids

Map Pino and Winston output onto Infrai's /v1/logs/ingest entry shape, keep request_id and user_id attached, batch the writes, and read the lines back.

Send POST /v1/logs/ingest on Infrai a body of {"entries": [...]} where every entry carries at least message and level, and read it back with GET /v1/logs/search. Request ids and user ids don’t get their own columns — they belong inside the entry’s attributes object. Pino and Winston both already produce JSON, so the work is a field mapping plus a batching writer.

The mapping matters more than it sounds. Infrai’s log entry has a fixed set of top-level keys, and anything else you put beside them is dropped without an error: we sent an entry with a top-level request_id and user_id on 26 July 2026, got {"accepted": 1} back, and the search result came back with neither field. No 400, no warning. That silent-drop behaviour is the single thing to get right before you wire a transport.

The entry shape, and the two required fields

Here’s a complete, concrete call. Nothing in the path is a placeholder, so you can run it as-is with your own key.

curl -s -X POST "https://api.infrai.cc/v1/logs/ingest" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "entries": [
      {
        "message": "checkout.charge failed: card_declined",
        "level": "error",
        "timestamp": "2026-07-26T01:30:00Z",
        "service": "checkout",
        "environment": "production",
        "trace_id": "trc_9f2c",
        "span_id": "spn_11a",
        "attributes": {
          "request_id": "req_8c41",
          "user_id": "usr_4410",
          "order_id": "ord_772",
          "status": 402
        }
      }
    ]
  }'

The response is small and worth parsing rather than ignoring:

{
  "ok": true,
  "data": { "accepted": 1 },
  "metadata": {
    "request_id": "req_9303d9ca222344e4a5e4da10",
    "latency_ms": 13,
    "vendor": "infrai",
    "cost_usd": 0.0
  }
}

accepted is a count, not a boolean. An entry missing message or missing level is skipped silently, so a batch of 50 that returns {"accepted": 49} is telling you one line never landed. Compare the number you sent against the number that came back and log the difference to stderr — that check costs nothing and catches an entire class of mapping bug.

One enum detail bites Pino users specifically: Pino’s level 40 serialises as warn, and the documented enum value is warning. The API accepts "warn" anyway — we sent it and got it back unchanged — so you end up with two spellings of the same severity in one account and level=warning quietly misses half your rows. Map it in the transport.

A Pino transport that batches

Per-line HTTP requests are the wrong shape here, because billing is per call and not per entry. Buffer and flush.

// logger.mjs — Pino writing NDJSON into a batching HTTP shipper. Node 22 ESM.
import { Writable } from "node:stream";
import pino from "pino";

const INGEST = "https://api.infrai.cc/v1/logs/ingest";
const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("INFRAI_API_KEY is not set");

const LEVELS = { 10: "debug", 20: "debug", 30: "info", 40: "warning", 50: "error", 60: "fatal" };
const MAX_BATCH = 200;
let buffer = [];
let timer = null;

async function flush() {
  if (timer) { clearTimeout(timer); timer = null; }
  if (buffer.length === 0) return;
  const entries = buffer.splice(0, buffer.length);
  try {
    const res = await fetch(INGEST, {
      method: "POST",
      headers: { authorization: `Bearer ${KEY}`, "content-type": "application/json" },
      body: JSON.stringify({ entries }),
    });
    const payload = await res.json();
    if (!res.ok || payload.ok !== true) {
      process.stderr.write(`log ship failed ${res.status}: ${JSON.stringify(payload)}\n`);
      return;
    }
    const dropped = entries.length - payload.data.accepted;
    if (dropped > 0) process.stderr.write(`log ship dropped ${dropped} malformed entries\n`);
  } catch (err) {
    process.stderr.write(`log ship error: ${err.message}\n`);
  }
}

function toEntry(line) {
  const { level, time, msg, name, ...rest } = JSON.parse(line);
  return {
    message: msg ?? "",
    level: LEVELS[level] ?? "info",
    timestamp: new Date(time ?? Date.now()).toISOString(),
    service: name ?? process.env.SERVICE_NAME ?? "api",
    environment: process.env.NODE_ENV ?? "production",
    trace_id: rest.trace_id ?? null,
    attributes: rest,
  };
}

const shipper = new Writable({
  write(chunk, _enc, done) {
    for (const line of chunk.toString().split("\n")) {
      if (!line.trim()) continue;
      try { buffer.push(toEntry(line)); } catch { /* not JSON — skip */ }
    }
    if (buffer.length >= MAX_BATCH) flush();
    else if (!timer) timer = setTimeout(flush, 2000).unref();
    done();
  },
});

export const logger = pino({ name: "checkout", base: null }, shipper);
process.on("beforeExit", flush);

A 200-entry batch took 185ms of server time in our testing and uploaded about 22 KB. That’s one billable call.

Winston, if that’s what the app already uses

Winston’s transport interface is a class, so the same buffer lives behind a log() method.

// infrai-transport.mjs — a Winston transport for /v1/logs/ingest. Node 22 ESM.
import Transport from "winston-transport";

const LEVELS = { silly: "debug", debug: "debug", verbose: "debug", http: "info", info: "info", warn: "warning", error: "error" };

export class InfraiTransport extends Transport {
  constructor(opts = {}) {
    super(opts);
    this.key = process.env.INFRAI_API_KEY;
    if (!this.key) throw new Error("INFRAI_API_KEY is not set");
    this.service = opts.service ?? "api";
    this.buffer = [];
    this.timer = null;
  }

  log(info, callback) {
    const { level, message, request_id, user_id, ...rest } = info;
    this.buffer.push({
      message: String(message ?? ""),
      level: LEVELS[level] ?? "info",
      timestamp: new Date().toISOString(),
      service: this.service,
      environment: process.env.NODE_ENV ?? "production",
      attributes: { request_id, user_id, ...rest },
    });
    if (this.buffer.length >= 200) this.flush();
    else if (!this.timer) this.timer = setTimeout(() => this.flush(), 2000).unref();
    callback();
  }

  async flush() {
    clearTimeout(this.timer);
    this.timer = null;
    const entries = this.buffer.splice(0, this.buffer.length);
    if (!entries.length) return;
    const res = await fetch("https://api.infrai.cc/v1/logs/ingest", {
      method: "POST",
      headers: { authorization: `Bearer ${this.key}`, "content-type": "application/json" },
      body: JSON.stringify({ entries }),
    });
    if (!res.ok) process.stderr.write(`winston->infrai ${res.status}\n`);
  }
}

Carrying the request id without threading it through every function

AsyncLocalStorage is the part of this that has nothing to do with any logging vendor, and it’s the part that actually makes the ids useful.

// context.mjs — one request id and user id per HTTP request, Express + Node 22.
import { AsyncLocalStorage } from "node:async_hooks";
import { randomUUID } from "node:crypto";
import { logger } from "./logger.mjs";

export const store = new AsyncLocalStorage();

export function withRequestContext(req, res, next) {
  const request_id = req.header("x-request-id") ?? randomUUID();
  res.setHeader("x-request-id", request_id);
  store.run({ request_id, user_id: req.user?.id ?? null }, () => next());
}

export function log(level, message, extra = {}) {
  const ctx = store.getStore() ?? {};
  // The ids go in the message AND the attributes: `filter` reaches the
  // attribute, `q=` only ever matches the message text.
  logger[level]({ ...ctx, ...extra }, `[${ctx.request_id ?? "-"}] ${message}`);
}

That duplication is deliberate. One copy makes the id greppable, the other makes it filterable.

Reading it back

GET /v1/logs/search is free and takes q, filter, since, until, level, service, environment, limit and cursor.

curl -s "https://api.infrai.cc/v1/logs/search?q=card_declined&level=error&environment=production&since=2026-07-26T00:00:00Z&limit=3" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

{
  "ok": true,
  "data": {
    "items": [
      {
        "message": "checkout.charge failed: card_declined",
        "level": "error",
        "timestamp": "2026-07-26T01:30:00Z",
        "service": "checkout",
        "environment": "production",
        "trace_id": "trc_9f2c",
        "attributes": { "request_id": "req_8c41", "user_id": "usr_4410", "status": 402 }
      }
    ],
    "next_cursor": null,
    "total": 1
  }
}

q is a case-insensitive substring match against message only, which is why q=usr_4410 won’t find a user id that lives solely in attributes — hence the [${request_id}] prefix in the helper above. To go at the structured half, use filter, the Observation Filter DSL from the logs reference:

curl -sG -X GET "https://api.infrai.cc/v1/logs/search" \
  --data-urlencode 'filter={"attributes.user_id": "usr_4410"}' \
  --data-urlencode 'since=2026-07-26T00:00:00Z' \
  --data-urlencode 'limit=50' \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

Treat that predicate as illustrative and check the reference for the operator list before you wire a support tool to it. since and until bound the window on either side, so “everything this user saw during the outage” is one request. Pass next_cursor back exactly as you were given it.

Cost, and when something else is the better tool

Ingest is billable at $0.00003 per call, verified 26 July 2026, and search is free. Per call is the load-bearing word: a single request carrying 200 entries is charged once, which is why the transports above buffer. A million log lines shipped 200 at a time is 5,000 calls. New accounts start with $2 of credit, which covers roughly 66,666 ingest calls before you pay anything. Rates on this platform move down over time and discount campaigns run, so read today’s number rather than trusting this paragraph:

curl -s "https://api.infrai.cc/v1/discovery" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  | python3 -c "import sys,json; print([c['billing'] for c in json.load(sys.stdin)['capabilities'] if c['id']=='logs.ingest'])"

The honest limitations are about everything that happens after the query. Retention is a 30-day TTL, there’s no live tail, no saved search, no log-derived metric and no alert rule that fires when errors spike — the API answers questions you ask, and nothing watches on your behalf. If you need any of that, Better Stack and Datadog both do it today and Infrai doesn’t: buy Better Stack if log search is your product’s debugging surface and you want live tail with a query builder; Datadog once logs have to sit beside traces and infrastructure metrics in one incident view. Grafana Loki is the pick if you’re already running Grafana and would rather own the storage than pay per line. What Infrai gives you instead is that the same key also reaches error capture, queues, cron and object storage, so the follow-on step — attach the failing payload to storage, push a retry onto a queue — isn’t another vendor, another key and another invoice.

References

• https://docs.infrai.cc/en/api/logs/
• https://docs.infrai.cc/en/api/observability/
• https://betterstack.com/community/guides/logging/how-to-install-setup-and-use-pino-to-log-node-js-applications/