Benzene

Observing the mesh (the collector + the Mesh UI)

You have one or more Benzene services already reporting into a mesh — or at least exposing /benzene/spec and /benzene/health. This cookbook stands up the receiving side: a MeshCollector that ingests the fleet's feeds, projects them into the cross-language mesh-ui artifacts, and serves the canonical Benzene Mesh UI so you can see your estate in one dashboard.

What the mesh gives you

A collector is an ordinary Benzene service. It builds a live catalog of your fleet from two feeds:

The two feeds compose in one collector. From the catalog it derives the fleet view and publishes the mesh-ui artifacts — a fixed set of static JSON documents that the canonical mesh-ui.html (the same page every Benzene port vendors) fetches by relative path and renders. No UI configuration is needed.

Prerequisites

1. Feed the mesh

Pull: poll a fleet's well-known surfaces

MeshPoller folds each source's spec + health into the collector. HttpServiceSource GET-polls {prefix}/spec and {prefix}/health (default prefix /benzene); a source that is down is a failed PollResult, never a broken sweep. /benzene/spec is every port's Contract Document — the same document a push registers — so the poller ingests it as the registration: its requests are what a service consumes and its events and sends what it produces, and a polled fleet gets provider edges as well as consumer edges from the pull, for every topic the services declare with register_event/register_send. The poller's heartbeat carries each document's served registrationHash as the drift signal. A CallableServiceSource can reach the same document over any transport by invoking the benzene:spec topic (as the AWS Lambda mesh example does).

# poll.py
import asyncio
from benzene.mesh import MeshCollector, MeshPoller, HttpServiceSource

collector = MeshCollector()
sources = [
    HttpServiceSource("orders", "https://orders.svc"),
    HttpServiceSource("inventory", "https://inventory.svc"),
]
poller = MeshPoller(collector, sources)

async def main():
    results = await poller.poll_once()           # one sweep; call on a timer for a live fleet
    for r in results:
        print(r.service, "ok" if r.ok else f"failed: {r.error}")
    fleet = collector.query_fleet({})            # the catalog now reflects both services
    print([s["service"] for s in fleet["services"]])

asyncio.run(main())

Push: register the declared graph, then feed it real stats

A service's register push is its Contract Document — requests and events/sends — so it declares its provider and consumer edges the instant it registers, with zero traffic (mesh.md §2, §4). A collector requires only service; a member the body leaves out (or names in degraded) is unknown, not empty. The collector is itself a Benzene service — collector_registry wires the ingest topics onto a registry — so you can drive it in-process:

from benzene.core import BenzeneMessageApplication
from benzene.mesh import MeshCollector, collector_registry

collector = MeshCollector()
app = BenzeneMessageApplication(collector_registry(collector))

# What orders' MeshFeedSender.register(document) sends over the wire (trimmed to the members that
# matter here) — orders declares it sends to inventory:reserve, so the edge exists before a single
# call is made:
await app.handle({
    "topic": "benzene:mesh:register",
    "headers": {},
    "body": '{"service": "orders", "requests": [], "events": [],'
            ' "sends": [{"topic": "inventory:reserve", "request": {}}]}',
})
# The collector already knows orders is a provider of inventory:reserve.

# The trace feed then adds invocation/error stats for that same declared edge — it never adds or
# removes the edge itself:
await app.handle({
    "topic": "benzene:mesh:traces",
    "headers": {},
    "body": '{"events": [{"traceId": "t1", "spanId": "s2", "parentSpanId": "s1",'
            ' "service": "inventory", "topic": "inventory:reserve", "status": "ok"}]}',
})

In a deployment the same collector polls the fleet and receives pushed traces — see step 4 for the host that exposes both over HTTP.

2. Build the artifacts

build_artifacts projects the catalog into the mesh-ui read-model contract as plain dicts; write_artifacts lays them out on disk (atomically) for the UI to fetch. Both take sources= (any objects with name / spec_url / health_url, e.g. HttpServiceSource) to supply the manifest's per-service links, and generated_at= (inject it for a deterministic result — it is stamped as each artifact's generatedAtUtc / fetchedAtUtc).

from datetime import datetime, timezone
from benzene.mesh import build_artifacts, write_artifacts

now = datetime.now(timezone.utc).isoformat()

artifacts = build_artifacts(collector, sources=sources, generated_at=now)   # dicts, in memory
write_artifacts("/data/mesh-ui", collector, sources=sources, generated_at=now)   # on disk

write_artifacts writes manifest.json, topology.json, topics.json, usage.json, asyncapi.json, annotations.json at the root and one services/{name}.json per service. What each artifact carries:

Artifact What it is
manifest.json The estate: one entry per service with name, status (healthy / unhealthy / unreachable), contractDrift, and its specUrl / healthUrl links (from sources).
topics.json The functional map: each topic's declared producers / consumers, version, request/response/message schemas, reserved flag for benzene:*, schemaMismatch (two consumers/handlers declaring different contracts — schema authority stays with the handler registration), and changes (a consumer re-registered a topic with a new schema). Plus removedTopics — topics once declared, now consumed by no one.
topology.json The declared call graph: client→server edges from each topic's declared providers (events/sends, the publisher or sender) to its consumers (requests, the handler), each with errorRate from the matching traces.
usage.json Exercise counts per (topic, service, status) derived from traces.
services/{name}.json Per service: specJson, specHash / previousSpecHash, contractDrift, and health (isHealthy + per-check healthChecks).
asyncapi.json An AsyncAPI 3.0 export of the domain (non-reserved) topics — the UI's download / Studio deep-link.
annotations.json An honest empty read-model (writing notes is a backend-gated live-plane feature, not on the static floor).

The projection never invents fields — it degrades to null where the pull + trace catalog can't derive a value. With no metrics feed, topology.json's requestsPerMinute and p50/p95/p99LatencyMs stay null (the UI renders those edges as reduced); usage.json's windowStartUtc / windowEndUtc, per-entry version, transport, and avgDurationMs stay null; a topic with no metrics carries a null status. The structural counts and the fleet/topology map are still fully populated.

3. Refresh on a timer

The artifacts are a snapshot. Poll and re-publish on an interval so the dashboard tracks the fleet:

import asyncio
from datetime import datetime, timezone
from benzene.mesh import MeshCollector, MeshPoller, HttpServiceSource, write_artifacts

collector = MeshCollector()
sources = [HttpServiceSource("orders", "https://orders.svc")]
poller = MeshPoller(collector, sources)

async def refresh_loop(interval_seconds: float = 30.0):
    while True:
        await poller.poll_once()                                  # pull the fleet
        now = datetime.now(timezone.utc).isoformat()
        write_artifacts("/data/mesh-ui", collector, sources=sources, generated_at=now)
        await asyncio.sleep(interval_seconds)

To keep the fleet view across a restart, give the collector a durable store: MeshCollector(store=JsonFileCollectorStore("/data/mesh-state.json")) — see Persistence.

4. Serve the dashboard

You don't have to wire the loop above yourself: deploy/mesh ships the Mesh Host — a container that runs the collector + poller, exposes the ingest/query API over HTTP, re-publishes the artifacts after each sweep, and serves the vendored mesh-ui.html. One terraform apply stands it up on Fargate behind an ALB, with the dashboard at /benzene/mesh-ui/.

Run it locally (no AWS)

The host reads its fleet from MESH_SERVICES (inline JSON) or MESH_CONFIG (a file), and serves the UI whenever MESH_ARTIFACTS_DIR is set:

# From the repo root: install the layers the host needs, plus uvicorn.
pip install -e packages/benzene-results -e packages/benzene-core \
            -e packages/benzene-http -e packages/benzene-mesh && pip install uvicorn

MESH_SERVICES='{"pollIntervalSeconds":15,"services":[{"name":"orders","baseUrl":"http://localhost:9000"}]}' \
MESH_ARTIFACTS_DIR=/tmp/mesh-ui \
PYTHONPATH=deploy/mesh python -m collector.main

Then read the mesh over HTTP and open the dashboard:

curl localhost:8080/benzene/health          # {"isHealthy": true, ...}
curl localhost:8080/mesh/fleet              # the polled fleet (after one sweep)
open  http://localhost:8080/benzene/mesh-ui/        # the Benzene Mesh UI

The host's HTTP surface: the push feeds are POSTs (/mesh/register, /mesh/heartbeat, /mesh/traces, /mesh/issues) — point each service's MeshFeedSender (over an HTTP MessageSender) at POST <host>/mesh/register to declare the call graph (its Contract Document's events/sends) and at POST <host>/mesh/traces to feed that graph's invocation/error stats — and the query read models are GETs (/mesh/fleet, /mesh/service/{service}, /mesh/topic/{topic}, /mesh/trace/{traceId}). Setting MESH_ARTIFACTS_DIR is what enables /benzene/mesh-ui/; leave it unset to run the API alone.

Troubleshooting

See also

Edit this page on GitHub