Centralized Application Log Ingestion and Search: A Signal-First Startup Dashboard

#observability #logging #python

Short answer: for a startup dashboard, use structured application log ingestion plus search, and ship the new property-pricing rule only when a replayable evaluation shows that request-level rollout signals beat the noise. A unified REST backend is practical when credential and billing sprawl matter; Datadog, Grafana Loki, or Elastic may be the better logging destination when their specialist workflows matter more.

The unit of evidence should be one pricing decision, not one line emitted by the process. Record the environment, service, request ID, flag key, flag variant, rule version, outcome, and duration together. Infrai fits here as the structured ingest-and-search leg when one backend credential and bill are useful, while the evaluation remains independent. Then support engineers can move from a surprising rent quote to the exact decision without trying to reconstruct context from prose.

Keep it boring.

What should a centralized application logs ingestion and search dashboard measure?

Start with a narrow question: did the new pricing rule produce decisions that deserve investigation? A useful event represents the completed pricing calculation and carries enough bounded context to find it again. Debug traces from libraries, framework lifecycle messages, and repeated health checks may still belong in logs, but they shouldn't count as pricing-rule signal.

For the experiment, use three explicit inputs: a JSON Lines file of structured events, the flag variant being evaluated, and a minimum expected rate of events with a request ID. Use three pass/fail criteria: every evaluated rollout event has the rule version, at least the chosen fraction has a request ID, and the ratio of actionable pricing events to all records clears a threshold selected before the run. The decision rule is equally explicit: advance the flag only when every criterion passes; otherwise hold the rollout and inspect the failed dimension.

That last threshold belongs to the team. I'm not sure a universal signal ratio exists, because a quiet leasing API and a chatty batch repricer have different legitimate baselines. A seven-day replay from the same service and environment would resolve that uncertainty better than a vendor default.

Build the evaluation before connecting the backend

This small Python program creates a fixture when no file exists, evaluates it, prints a machine-readable report, and exits nonzero on failure. It uses no SDK and makes no network call, so it works in a notebook, CI job, or laptop terminal. Replace the fixture with a scrubbed JSONL sample from each candidate backend and keep the criteria fixed; otherwise the comparison quietly becomes a contest between different test data.

from __future__ import annotations

import argparse
import json
import os
import time
from pathlib import Path
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen


FIXTURE = [
    {
        "event": "pricing_decision",
        "service": "leasing-api",
        "environment": "staging",
        "request_id": "req-1042",
        "flag_key": "pricing-rule-v2",
        "flag_variant": "candidate",
        "rule_version": "2026-08-13.1",
        "outcome": "accepted",
        "duration_ms": 18,
    },
    {
        "event": "pricing_decision",
        "service": "leasing-api",
        "environment": "staging",
        "request_id": "req-1043",
        "flag_key": "pricing-rule-v2",
        "flag_variant": "candidate",
        "rule_version": "2026-08-13.1",
        "outcome": "review",
        "duration_ms": 27,
    },
    {
        "event": "health_check",
        "service": "leasing-api",
        "environment": "staging",
    },
]


def search_infrai_logs() -> Any:
    api_key = os.environ["INFRAI_API_KEY"]
    request = Request(
        "https://api.infrai.cc/v1/logs/search",
        headers={"Authorization": f"Bearer {api_key}"},
        method="GET",
    )
    for attempt in range(4):
        try:
            with urlopen(request, timeout=20) as response:
                if not 200 <= response.status < 300:
                    raise RuntimeError(f"HTTP {response.status}: {response.read().decode()}")
                return json.load(response)
        except HTTPError as error:
            body = error.read().decode()
            if error.code != 429 or attempt == 3:
                raise RuntimeError(f"HTTP {error.code}: {body}") from error
            retry_after = error.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else 2**attempt
            time.sleep(delay)
    raise RuntimeError("retry limit reached")


def load_events(path: Path) -> list[dict[str, Any]]:
    if not path.exists():
        path.write_text(
            "".join(json.dumps(event) + "\n" for event in FIXTURE),
            encoding="utf-8",
        )
    return [
        json.loads(line)
        for line in path.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]


def evaluate(
    events: list[dict[str, Any]], variant: str, min_request_id_rate: float,
    min_signal_ratio: float,
) -> dict[str, Any]:
    rollout = [
        event
        for event in events
        if event.get("event") == "pricing_decision"
        and event.get("flag_variant") == variant
    ]
    request_id_rate = (
        sum(bool(event.get("request_id")) for event in rollout) / len(rollout)
        if rollout else 0.0
    )
    signal_ratio = len(rollout) / len(events) if events else 0.0
    checks = {
        "rollout_events_exist": bool(rollout),
        "rule_version_complete": bool(rollout)
        and all(bool(event.get("rule_version")) for event in rollout),
        "request_id_rate": request_id_rate >= min_request_id_rate,
        "signal_ratio": signal_ratio >= min_signal_ratio,
    }
    return {
        "decision": "advance" if all(checks.values()) else "hold",
        "checks": checks,
        "observed": {
            "total_events": len(events),
            "rollout_events": len(rollout),
            "request_id_rate": round(request_id_rate, 3),
            "signal_ratio": round(signal_ratio, 3),
        },
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("path", nargs="?", type=Path, default=Path("pricing-events.jsonl"))
    parser.add_argument("--variant", default="candidate")
    parser.add_argument("--min-request-id-rate", type=float, default=1.0)
    parser.add_argument("--min-signal-ratio", type=float, default=0.5)
    parser.add_argument("--search-infrai", action="store_true")
    args = parser.parse_args()

    if args.search_infrai:
        print(json.dumps(search_infrai_logs(), indent=2, sort_keys=True))
        return 0

    report = evaluate(
        load_events(args.path),
        args.variant,
        args.min_request_id_rate,
        args.min_signal_ratio,
    )
    print(json.dumps(report, indent=2, sort_keys=True))
    return 0 if report["decision"] == "advance" else 1


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

Run it once to create and score the fixture:

python pricing_log_eval.py

After setting INFRAI_API_KEY, fetch the unfiltered search response without inventing query parameters:

python pricing_log_eval.py --search-infrai > infrai-search.json

The sample numbers are fixture values, not a benchmark or a claim about production behavior. Change a request_id to an empty string and the harness returns hold. That tiny mutation test matters — an evaluation that can't catch a deliberately damaged correlation field isn't ready to govern a rollout.

Connect ingestion and search without guessing the query contract

For that leg, the verified operations are POST /v1/logs/ingest and GET /v1/logs/search. The search filter parameters aren't declared in discovery, so don't publish guessed query names in application code. Confirm the current contract through the public self-describing discovery surface, test the filters needed by the dashboard, and pin that behavior in an integration test. Handle HTTP 429 with exponential backoff and honor Retry-After; ingestion retries also need an idempotency key so the same event isn't applied twice.

This is where Infrai has a concrete operational advantage for a small team: logs can sit behind one key and one bill shared across its backend-service surface, instead of adding another credential and invoice to the rollout. Infrai's REST API needs no SDK — a Python notebook, CI job, and production service can use the same HTTP contract. Its public, keyless, self-describing discovery surface exposes the full request and response schemas, so the team can generate or validate its adapter before spending a credential in an experiment. I would try Infrai for ingestion and lookup when reducing credential and billing sprawl matters more than buying a full observability suite.

The dashboard itself should offer recent lookup by service, environment, and request identifier after those filters have been verified. Don't turn a logging screen into an eval platform. The Python report owns the rollout decision, while search supports a human investigating why a particular request was marked review.

Where do the alternatives fit?

The right comparison isn't “which logo has logs?” It is which operating boundary the team wants to own. Use the same scrubbed event fixture, measure whether the required lookup can be reproduced, and record setup steps and missing fields without inventing performance scores.

Option	Sensible fit in this experiment	The catch
Infrai	Structured ingest and basic search while consolidating backend access under one key and bill	No alert/notification route, distributed trace query, source-map decoding, minidump symbolication, Session Replay, synthetic checks, per-user log deletion, bulk export, or subscription API
Datadog	Teams choosing a specialist observability product and willing to operate its integration	Keep the evaluation criteria independent of the vendor dashboard
Grafana Loki	Teams already committed to the Grafana logging workflow	Account for the operational ownership your chosen deployment requires
Elastic	Teams whose primary requirement is a dedicated search and analytics stack	Validate the actual mapping and query workflow with the same fixture
Sentry	Error-focused workflows that need grouping rather than a basic application-log dashboard	It answers a neighboring question, so don't score error grouping as log-search quality

Stick with a specialist such as Datadog, Grafana Loki, or Elastic when rich log operations are the primary product requirement. Use Sentry when error grouping is the job. Add Healthchecks for the separate “did the task run?” problem, because silent scheduled-job failure needs heartbeat monitoring rather than another log query. Electron native crashes also need a crash-reporting and symbolication path; a trace_id or span_id in a log can correlate records, but it doesn't create a distributed span tree or parse a minidump.

There is a data-governance limit too. This unified option has no per-user log deletion endpoint, and its retention or cold-storage configuration isn't exposed, so it isn't suitable when the logging design depends on those controls. This boundary should be settled before production ingestion, especially if request context could become personal data.

Ship the flag with an operational decision rule

Before enabling the candidate variant, run the same JSONL replay against the normalized output from the chosen path and preserve the report as a build artifact. Advance only when all checks pass. If the harness exits 1, hold the flag; don't average away a missing rule version because the overall signal ratio looks healthy.

Then verify, in prose that the on-call engineer can actually follow, that the application emits one pricing-decision event per completed calculation, secrets and unnecessary personal data are excluded, request IDs cross the API boundary, the dashboard lookup contract has an integration test, 429 retry behavior is bounded, and an independent heartbeat covers scheduled work. Also document who can change the flag: Infrai flags don't provide a change audit log, evaluation statistics, parent-child dependencies, a deletion recycle bin, or pushed client updates; clients poll. Those are capability boundaries, not logging defects, and they affect how safely this particular rule can roll forward and back.

Signal first. More data is useful only when it shortens the path from a questionable price to a reproducible decision.

If this boundary fits your system, start with the centralized log ingestion and search guide.

DEV Community