Async custom metrics

[gpkc]

Question

It seems that async custom metrics are not supported, as this is the signature for the callback:

    Callable[[CallbackOptions], Iterable[Observation]],
    Generator[Iterable[Observation], CallbackOptions, None],
]

Does it mean that collecting each metric is blocking, or for example do they run in separate threads and then aggregated before the interval ends?

[alexmojaki]

The PeriodicExportingMetricReader runs in a single separate thread. In that thread, the callbacks run synchronously.

[gpkc]

For anyone reading in the future, my solution for this was to create a function that creates my metrics and returns some closured periodic exporter. For example, in my case I needed something that grabbed values from SQLAlchemy:

async def get_metric_value(session: AsyncSession) -> Iterable[Observation]:
    counts = await # select from DB
    return [Observation(count, {"status": status}) for status, count in counts.items()]


def init_custom_metrics():
    sample_metric = logfire.metric_gauge(
        "sample_metric", unit="count"
    )

    async def export():
        while True:
            await asyncio.sleep(60)

            async with sessionmanager.session() as session:
                observations = await get_metric_value(session)

                for observation in observations:
                    sample_metric.set(
                        amount=observation.value, attributes=observation.attributes
                    )

    return export

Then, in my case, I wanted this to run alongside my FastAPI app. So I simply added it as a startup callback that creates an asyncio task that runs forever:

@server.on_event("startup")
async def setup_custom_metrics():
    export = init_custom_metrics()
    asyncio.ensure_future(export())

Not sure if it's ideal or not but it seems to work.