GitHub

sdk

PreviousNext

API reference for sdk.

One layer setting up OpenTelemetry for a pyfect program (Effect's NodeSdk).

layer(Configuration(...)) provides the OtelResource (from the environment and the resource given) and turns on each signal given something to send it to: tracing with span processors, metrics with metric readers, logging with log record processors::

otel_layer = otel.sdk.layer(
    otel.sdk.Configuration(
        resource=otel.sdk.ResourceConfig(service_name="api"),
        span_processor=BatchSpanProcessor(OTLPSpanExporter()),
    )
)
AttributeTypeDescription
layer_emptyLayer[OtelResource] = otel_resource.layer_empty

Only an empty OtelResource.

Classes

ResourceConfigConfiguration

Functions

layer_tracer_provider

layer_tracer_provider(processors, *, sampler=None, id_generator=None, span_limits=None, shutdown_timeout=DEFAULT_SHUTDOWN_TIMEOUT) -> Layer[otel_tracer.OtelTracerProvider, Never, OtelResource]

An SDK tracer provider with the OtelResource and processors, flushed (for at most shutdown_timeout) and shut down when the layer is released.

ParameterTypeDescription
processorsSpanProcessor | Sequence[SpanProcessor]
samplerSampler | None = None
id_generatorIdGenerator | None = None
span_limitsSpanLimits | None = None
shutdown_timeoutDuration = DEFAULT_SHUTDOWN_TIMEOUT

Returns Layer[otel_tracer.OtelTracerProvider, Never, OtelResource]

def layer_tracer_provider(
    processors: SpanProcessor | Sequence[SpanProcessor],
    *,
    sampler: Sampler | None = None,
    id_generator: IdGenerator | None = None,
    span_limits: SpanLimits | None = None,
    shutdown_timeout: Duration = DEFAULT_SHUTDOWN_TIMEOUT,
) -> Layer[otel_tracer.OtelTracerProvider, Never, OtelResource]:
    """An SDK tracer provider with the ``OtelResource`` and ``processors``,
    flushed (for at most ``shutdown_timeout``) and shut down when the layer
    is released."""
    given = _all(processors, SpanProcessor)

    def make(resource: Resource) -> TracerProvider:
        provider = TracerProvider(
            sampler=sampler,
            resource=resource,
            shutdown_on_exit=False,
            id_generator=id_generator,
            span_limits=span_limits,
        )
        for processor in given:
            provider.add_span_processor(processor)
        return provider

    return _layer.effect(
        otel_tracer.OtelTracerProvider,
        OtelResource().flat_map(
            lambda resource: acquire_release(
                sync(lambda: make(resource)),
                lambda provider, _exit: flush_and_shut_down(provider, shutdown_timeout),
            )
        ),
    )

layer

layer(config) -> Layer[Any, Any, Any]

Set up OpenTelemetry from config: a configuration, a function making one, or an effect making one when the layer builds.

ParameterTypeDescription
configConfiguration | Callable[[], Configuration] | Effect[Configuration, Any, Any]

Returns Layer[Any, Any, Any]

def layer(
    config: Configuration | Callable[[], Configuration] | Effect[Configuration, Any, Any], /
) -> Layer[Any, Any, Any]:
    """Set up OpenTelemetry from ``config``: a configuration, a function
    making one, or an effect making one when the layer builds."""
    match config:
        case Configuration():
            made: Effect[Configuration, Any, Any] = sync(lambda: config)
        case Effect():
            made = config
        case _:
            made = sync(config)
    return _layer.unwrap(made.map(_signals))