otlp-arrow-library 0.6.4

Cross-platform Rust library for receiving OTLP messages via gRPC and writing to Arrow IPC files
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

"""Unit test for trace export"""

import tempfile
import os
import pytest
import time


def test_export_single_trace():
    """Test exporting a single trace span"""
    import otlp_arrow_library
    
    with tempfile.TemporaryDirectory() as tmpdir:
        library = otlp_arrow_library.PyOtlpLibrary(
            output_dir=tmpdir,
            write_interval_secs=1
        )
        
        # Create a test span
        trace_id = bytes([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16])
        span_id = bytes([1, 2, 3, 4, 5, 6, 7, 8])
        
        span_dict = {
            "trace_id": trace_id,
            "span_id": span_id,
            "name": "test-span",
            "kind": "server",
            "attributes": {
                "service.name": "test-service",
                "http.method": "GET"
            }
        }
        
        # Export the span
        library.export_trace(span_dict)
        
        # Flush to ensure it's written
        library.flush()
        
        # Verify file was created
        traces_dir = os.path.join(tmpdir, "otlp", "traces")
        files = os.listdir(traces_dir)
        assert len(files) > 0, "Expected at least one trace file to be created"
        
        # Cleanup
        library.shutdown()


def test_export_multiple_traces():
    """Test exporting multiple trace spans"""
    import otlp_arrow_library
    
    with tempfile.TemporaryDirectory() as tmpdir:
        library = otlp_arrow_library.PyOtlpLibrary(
            output_dir=tmpdir,
            write_interval_secs=1
        )
        
        # Create multiple test spans
        spans = []
        for i in range(3):
            trace_id = bytes([i] * 16)
            span_id = bytes([i] * 8)
            span_dict = {
                "trace_id": trace_id,
                "span_id": span_id,
                "name": f"test-span-{i}",
                "kind": "internal",
            }
            spans.append(span_dict)
        
        # Export the spans
        library.export_traces(spans)
        
        # Flush to ensure they're written
        library.flush()
        
        # Verify file was created
        traces_dir = os.path.join(tmpdir, "otlp", "traces")
        files = os.listdir(traces_dir)
        assert len(files) > 0, "Expected at least one trace file to be created"
        
        # Cleanup
        library.shutdown()


if __name__ == "__main__":
    pytest.main([__file__])