GitHub

Observability

PreviousNext

Logging, tracing and metrics, built into the runtime.

Logging

fx.log_info(...), log_debug, log_warning, log_error write to the loggers in CurrentLoggers, skipping entries below MinimumLogLevel (INFO by default). Annotations and log spans add context to every entry inside an effect.

log_entries.py
import pyfect as fx
from pyfect import logger

lines: list[str] = []

# A logger that formats entries as logfmt and keeps them (a real program
# would use logger.console_logfmt, console_json or logger.to_logging("app")).
Collect = logger.layer([logger.format_logfmt.map(lines.append)])


@fx.fn
def handle(user_id: str):
    yield from fx.log_debug("dropped: below the minimum level (INFO)")
    yield from fx.log_info("signed in")
    yield from fx.log_warning("password expires soon").with_log_span("check")


program = handle("42").annotate_logs("user_id", "42").pipe(fx.provide(Collect))
fx.run_sync(program)

assert len(lines) == 2
assert "level=INFO fiber=#" in lines[0] and 'message="signed in" user_id=42' in lines[0]
assert "level=WARN" in lines[1] and "check=" in lines[1]

logger.layer([...]) replaces the loggers: console_pretty, console_logfmt, console_json, batched(...), your own fx.Logger.make(...), or logger.to_logging("app") to hand entries to the standard library's logging.

Tracing

eff.with_span(name) runs an effect inside a span; @fx.fn("name") does the same for every call of a function. Spans nest across forked fibers, and logs inside a span become its events.

tracing.py
import pyfect as fx
from pyfect import tracer


@fx.fn("get_user")  # every call runs inside a span named get_user
def get_user(user_id: str):
    yield from fx.annotate_current_span("user.id", user_id)
    yield from fx.sleep(0.01).with_span("db.query")
    return (yield from fx.current_span())


span = fx.run_sync(get_user("42").with_span("GET /users/42").or_die())

assert span.name == "get_user"
assert span.attributes["user.id"] == "42"
assert isinstance(span.parent, tracer.NativeSpan) and span.parent.name == "GET /users/42"

The native tracer keeps spans in memory. To export them, install the OpenTelemetry bridge (pyfect[otel]) with otel.sdk.layer(...): spans, logs and metrics go to OpenTelemetry's SDK exporters.

Metrics

Metrics are descriptions: counter, gauge, histogram, summary, frequency, timer. track, track_errors and track_duration update one as an effect ends.

metrics.py
import pyfect as fx
from pyfect import metric

requests = metric.counter("http_requests", attributes={"route": "/users"})
latency = metric.timer("http_latency")


def handler() -> fx.Effect[str]:
    return fx.sleep(0.01).as_("ok")


# track adds 1 per success; track_duration records how long each call took.
tracked = handler().track_duration(latency).track(requests.with_constant_input(1))

fx.run_sync(fx.for_each(range(3), lambda _: tracked))

text = fx.run_sync(metric.prometheus_format())
assert 'http_requests{route="/users"} 3' in text
assert "http_latency_count 3" in text

metric.prometheus_format() writes the registry in the Prometheus text format for any HTTP framework to serve.

Reference: pyfect.logger, pyfect.tracer, pyfect.metric, pyfect.otel.