Logfireを使用したテスト

自分のAPIが期待するデータを記録しているか、それらがどのような働きでパッケージ化されているかなどを確認する必要があるかもしれません。これはPythonを使ったビルトインのロギングやOpenTelemometryを使ったSDKも含めて難しいことが多いです。

Logfireを使用すると、logfire.testing モジュール内のユーティリティを使用して、送信されたログとスパンをテストすることが非常に簡単になります。これはLogfireが自身をテストするために内部で使用する方法でもあります。

capfire ランプ

これには2つの属性exporter とmetrics_reader があります。

exporter

これは、TestExporter のインスタンスであり、エクスポートされたスパンをメモリに保持するOpenTelemometry SDKと互換性のあるスパンエクスポートです。

exporter.exported_spans_as_dict() メソッドを使用すると、エクスポートされたスパンの簡単な辞書表現を取得できます。これにより、範囲を簡単にアサートし、そこから優れた違いを得ることができます。 このメソッドは、行番号を 123 に置き換えたり、ファイルパスをファイル名に置き換えたりするなど、出力を読みやすく確実にするためにいくつかのデータ処理を行います。

test.py

import pytest

import logfire
from logfire.testing import  CaptureLogfire


def test_observability(capfire: CaptureLogfire) -> None:
    with pytest.raises(Exception):
        with logfire.span('a span!'):
            logfire.info('a log!')
            raise Exception('an exception!')

    exporter = capfire.exporter

    # insert_assert(exporter.exported_spans_as_dict()) (1)
    assert exporter.exported_spans_as_dict() == [
        {
            'name': 'a log!',
            'context': {'trace_id': 1, 'span_id': 3, 'is_remote': False},
            'parent': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
            'start_time': 2000000000,
            'end_time': 2000000000,
            'attributes': {
                'logfire.span_type': 'log',
                'logfire.level_num': 9,
                'logfire.msg_template': 'a log!',
                'logfire.msg': 'a log!',
                'code.filepath': 'test.py',
                'code.lineno': 123,
                'code.function': 'test_observability',
            },
        },
        {
            'name': 'a span!',
            'context': {'trace_id': 1, 'span_id': 1, 'is_remote': False},
            'parent': None,
            'start_time': 1000000000,
            'end_time': 4000000000,
            'attributes': {
                'code.filepath': 'test.py',
                'code.lineno': 123,
                'code.function': 'test_observability',
                'logfire.msg_template': 'a span!',
                'logfire.span_type': 'span',
                'logfire.msg': 'a span!',
            },
            'events': [
                {
                    'name': 'exception',
                    'timestamp': 3000000000,
                    'attributes': {
                        'exception.type': 'Exception',
                        'exception.message': 'an exception!',
                        'exception.stacktrace': 'Exception: an exception!',
                        'exception.escaped': 'True',
                    },
                }
            ],
        },
    ]

1. insert_assert はい DevTools 実用的な関数を提供します。

   それについての詳細は以下をご覧ください

エクスポートされたスパンに 按exporter.exported_spans でアクセスできます。

import logfire
from logfire.testing import CaptureLogfire


def test_exported_spans(capfire: CaptureLogfire) -> None:
    with logfire.span('a span!'):
        logfire.info('a log!')

    exporter = capfire.exporter

    expected_span_names = ['a span! (pending)', 'a log!', 'a span!']
    span_names = [span.name for span in exporter.exported_spans]

    assert span_names == expected_span_names

exporter.clear() を呼び出して、テストでキャプチャされたスパンをリセットできます。

import logfire
from logfire.testing import CaptureLogfire


def test_reset_exported_spans(capfire: CaptureLogfire) -> None:
    exporter = capfire.exporter

    assert len(exporter.exported_spans) == 0

    logfire.info('First log!')
    assert len(exporter.exported_spans) == 1
    assert exporter.exported_spans[0].name == 'First log!'

    logfire.info('Second log!')
    assert len(exporter.exported_spans) == 2
    assert exporter.exported_spans[1].name == 'Second log!'

    exporter.clear()
    assert len(exporter.exported_spans) == 0

    logfire.info('Third log!')
    assert len(exporter.exported_spans) == 1
    assert exporter.exported_spans[0].name == 'Third log!'

metrics_reader

これは、メトリックをメモリに読み込むInMemoryMetricReader の例です。

import json
from typing import cast

from opentelemetry.sdk.metrics.export import MetricsData

from logfire.testing import CaptureLogfire


def test_system_metrics_collection(capfire: CaptureLogfire) -> None:
    exported_metrics = json.loads(cast(MetricsData, capfire.metrics_reader.get_metrics_data()).to_json())  # type: ignore

    metrics_collected = {
        metric['name']
        for resource_metric in exported_metrics['resource_metrics']
        for scope_metric in resource_metric['scope_metrics']
        for metric in scope_metric['metrics']
    }

    # collected metrics vary by platform, etc.
    # assert that we at least collected _some_ of the metrics we expect
    assert metrics_collected.issuperset(
        {
            'system.swap.usage',
            'system.disk.operations',
            'system.memory.usage',
            'system.cpu.utilization',
        }
    ), metrics_collected

使用しているユーティリティを見てみましょう。

IncrementalId生成器

ログ出力を期待される結果と比較するときに最も複雑なことの1つは、不確実性の原因です。OpenTelemometryスパンの場合、最大の2つのスパンは、スパンとトレースIDとタイムスタンプです。

IncrementalIdGenerator テスト出力が常に同じになるように、スパンとトレースIDを順番に増やします。

时间生成器

このクラスはナノ秒タイムスタンプを生成し、タイムスタンプが生成されるたびに1秒増加します。

logfire.配置

これは、本番環境で使用する構成関数と同じで、すべてがまとめられています。

特別に構成されていることに注意してください。

• send_to_logfire=False 実際の生産サービスに触れたくないからです

• id_generator=IncrementalIdGenerator() スパンIDを決定的にする

• ns_timestamp_generator=TimeGenerator() タイムスタンプを確実にする

• processors=[SimpleSpanProcessor(exporter)] は、私たちの TestExporter を使用してスパンをキャプチャします。私たちは SimpleSpanProcessor を使用して、遅延なくスパンをエクスポートします。

insert_assert

これはdevtoolsが提供するユーティリティ関数で、pytestで実行すると、呼び出し元のコード出力をテストファイルに自動的に挿入します。 つまり、その行をコメントアウトすると、断言 capfire.exported_spans_as_dict() == [...] 行が capfire.exported_spans_as_dict() の現在の出力に置き換えられていることがわかります。これは、テストが確実であることを考えると、まったく同じである必要があります。

---

本文总阅读量次