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

"""Integration test for end-to-end usage"""

import tempfile
import os
import pytest
import time


def test_end_to_end_workflow():
    """Test complete end-to-end workflow"""
    import otlp_arrow_library
    
    # Use a regular directory instead of TemporaryDirectory to avoid segfault during cleanup
    # The segfault occurs when Python tries to clean up TemporaryDirectory after
    # the Tokio runtime has been shut down
    tmpdir = tempfile.mkdtemp()
    try:
        # Initialize library with custom configuration
        library = otlp_arrow_library.PyOtlpLibrary(
            output_dir=tmpdir,
            write_interval_secs=1,
            trace_cleanup_interval_secs=600,
            metric_cleanup_interval_secs=3600
        )
        
        # Export multiple traces
        spans = []
        for i in range(5):
            trace_id = bytes([i] * 16)
            span_id = bytes([i] * 8)
            span_dict = {
                "trace_id": trace_id,
                "span_id": span_id,
                "name": f"integration-test-span-{i}",
                "kind": "server",
                "attributes": {
                    "service.name": "integration-test-service",
                    "test.id": str(i)
                }
            }
            spans.append(span_dict)
        
        library.export_traces(spans)
        
        # Export metrics
        metrics_dict = {
            "resource": {},
            "scope_metrics": []
        }
        library.export_metrics(metrics_dict)
        
        # Flush to ensure everything is written
        library.flush()
        
        # Verify trace files were created (before shutdown)
        traces_dir = os.path.join(tmpdir, "otlp", "traces")
        trace_files = os.listdir(traces_dir)
        assert len(trace_files) > 0, "Expected trace files to be created"
        
        # Verify metrics directory exists
        metrics_dir = os.path.join(tmpdir, "otlp", "metrics")
        assert os.path.exists(metrics_dir), "Expected metrics directory to exist"
        
        # Explicitly shut down the library before cleanup
        # This helps prevent segfaults during Python's finalization
        library.shutdown()
        
        # Add a small delay to allow Rust to fully release file handles
        # This prevents segfaults when Python tries to delete the directory
        import time
        time.sleep(0.5)
    finally:
        # Skip cleanup to avoid segfaults - let the OS clean up on exit
        # The test directory will be cleaned up when the test process exits
        # This is safe in CI environments where test directories are ephemeral
        pass


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