otlp_stdout_span_exporter/
lib.rs

1//! A span exporter that writes OpenTelemetry spans to stdout in OTLP format.
2//!
3//! This crate provides an implementation of OpenTelemetry's [`SpanExporter`] that writes spans to stdout
4//! in OTLP (OpenTelemetry Protocol) format. It is particularly useful in serverless environments like
5//! AWS Lambda where writing to stdout is a common pattern for exporting telemetry data.
6//!
7//! # Features
8//!
9//! - Uses OTLP Protobuf serialization for efficient encoding
10//! - Applies GZIP compression with configurable levels
11//! - Detects service name from environment variables
12//! - Supports custom headers via environment variables
13//! - Consistent JSON output format
14//!
15//! # Example
16//!
17//! ```rust,no_run
18//! use opentelemetry::trace::{Tracer, TracerProvider};
19//! use opentelemetry_sdk::{trace::SdkTracerProvider, Resource};
20//! use otlp_stdout_span_exporter::OtlpStdoutSpanExporter;
21//!
22//! #[tokio::main]
23//! async fn main() {
24//!     // Create a new stdout exporter
25//!     let exporter = OtlpStdoutSpanExporter::new();
26//!
27//!     // Create a new tracer provider with batch export
28//!     let provider = SdkTracerProvider::builder()
29//!         .with_batch_exporter(exporter)
30//!         .with_resource(Resource::builder().build())
31//!         .build();
32//!
33//!     // Create a tracer
34//!     let tracer = provider.tracer("my-service");
35//!
36//!     // Create spans
37//!     tracer.in_span("parent-operation", |_cx| {
38//!         println!("Doing work...");
39//!         
40//!         // Create nested spans
41//!         tracer.in_span("child-operation", |_cx| {
42//!             println!("Doing more work...");
43//!         });
44//!     });
45//!     
46//!     // Shut down the provider
47//!     let _ = provider.shutdown();
48//! }
49//! ```
50//!
51//! # Environment Variables
52//!
53//! The exporter respects the following environment variables:
54//!
55//! - `OTEL_SERVICE_NAME`: Service name to use in output
56//! - `AWS_LAMBDA_FUNCTION_NAME`: Fallback service name (if `OTEL_SERVICE_NAME` not set)
57//! - `OTEL_EXPORTER_OTLP_HEADERS`: Global headers for OTLP export
58//! - `OTEL_EXPORTER_OTLP_TRACES_HEADERS`: Trace-specific headers (takes precedence)
59//! - `OTLP_STDOUT_SPAN_EXPORTER_COMPRESSION_LEVEL`: GZIP compression level (0-9, default: 6)
60//!
61//! # Output Format
62//!
63//! The exporter writes each batch of spans as a JSON object to stdout:
64//!
65//! ```json
66//! {
67//!   "__otel_otlp_stdout": "0.1.0",
68//!   "source": "my-service",
69//!   "endpoint": "http://localhost:4318/v1/traces",
70//!   "method": "POST",
71//!   "content-type": "application/x-protobuf",
72//!   "content-encoding": "gzip",
73//!   "headers": {
74//!     "api-key": "secret123",
75//!     "custom-header": "value"
76//!   },
77//!   "payload": "<base64-encoded-gzipped-protobuf>",
78//!   "base64": true
79//! }
80//! ```
81
82use async_trait::async_trait;
83use base64::{engine::general_purpose::STANDARD as base64_engine, Engine};
84use flate2::{write::GzEncoder, Compression};
85use futures_util::future::BoxFuture;
86use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
87use opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema;
88use opentelemetry_proto::transform::trace::tonic::group_spans_by_resource_and_scope;
89use opentelemetry_sdk::resource::Resource;
90use opentelemetry_sdk::{
91    error::OTelSdkError,
92    trace::{SpanData, SpanExporter},
93};
94use prost::Message;
95use serde::Serialize;
96#[cfg(test)]
97use std::sync::Mutex;
98use std::{collections::HashMap, env, io::Write, result::Result, sync::Arc};
99
100const VERSION: &str = env!("CARGO_PKG_VERSION");
101const DEFAULT_ENDPOINT: &str = "http://localhost:4318/v1/traces";
102const DEFAULT_COMPRESSION_LEVEL: u8 = 6;
103const COMPRESSION_LEVEL_ENV_VAR: &str = "OTLP_STDOUT_SPAN_EXPORTER_COMPRESSION_LEVEL";
104
105/// Trait for output handling
106///
107/// This trait defines the interface for writing output lines. It is implemented
108/// by both the standard output handler and test output handler.
109trait Output: Send + Sync + std::fmt::Debug {
110    /// Writes a single line of output
111    ///
112    /// # Arguments
113    ///
114    /// * `line` - The line to write
115    ///
116    /// # Returns
117    ///
118    /// Returns `Ok(())` if the write was successful, or a `TraceError` if it failed
119    fn write_line(&self, line: &str) -> Result<(), OTelSdkError>;
120}
121
122/// Standard output implementation that writes to stdout
123#[derive(Debug, Default)]
124struct StdOutput;
125
126impl Output for StdOutput {
127    fn write_line(&self, line: &str) -> Result<(), OTelSdkError> {
128        println!("{}", line);
129        Ok(())
130    }
131}
132
133/// Test output implementation that captures to a buffer
134#[cfg(test)]
135#[derive(Debug, Default)]
136struct TestOutput {
137    buffer: Arc<Mutex<Vec<String>>>,
138}
139
140#[cfg(test)]
141impl TestOutput {
142    fn new() -> Self {
143        Self {
144            buffer: Arc::new(Mutex::new(Vec::new())),
145        }
146    }
147
148    fn get_output(&self) -> Vec<String> {
149        self.buffer.lock().unwrap().clone()
150    }
151}
152
153#[cfg(test)]
154impl Output for TestOutput {
155    fn write_line(&self, line: &str) -> Result<(), OTelSdkError> {
156        self.buffer.lock().unwrap().push(line.to_string());
157        Ok(())
158    }
159}
160
161/// Output format for the OTLP stdout exporter
162///
163/// This struct defines the JSON structure that will be written to stdout
164/// for each batch of spans.
165#[derive(Debug, Serialize)]
166struct ExporterOutput<'a> {
167    /// Version identifier for the output format
168    #[serde(rename = "__otel_otlp_stdout")]
169    version: &'a str,
170    /// Service name that generated the spans
171    source: String,
172    /// OTLP endpoint (always http://localhost:4318/v1/traces)
173    endpoint: &'a str,
174    /// HTTP method (always POST)
175    method: &'a str,
176    /// Content type (always application/x-protobuf)
177    #[serde(rename = "content-type")]
178    content_type: &'a str,
179    /// Content encoding (always gzip)
180    #[serde(rename = "content-encoding")]
181    content_encoding: &'a str,
182    /// Custom headers from environment variables
183    #[serde(skip_serializing_if = "HashMap::is_empty")]
184    headers: HashMap<String, String>,
185    /// Base64-encoded, gzipped, protobuf-serialized span data
186    payload: String,
187    /// Whether the payload is base64 encoded (always true)
188    base64: bool,
189}
190
191/// A span exporter that writes spans to stdout in OTLP format
192///
193/// This exporter implements the OpenTelemetry [`SpanExporter`] trait and writes spans
194/// to stdout in OTLP format with Protobuf serialization and GZIP compression.
195///
196/// # Features
197///
198/// - Configurable GZIP compression level (0-9)
199/// - Environment variable support for service name and headers
200/// - Efficient batching of spans
201/// - Base64 encoding of compressed data
202///
203/// # Example
204///
205/// ```rust,no_run
206/// use opentelemetry_sdk::runtime;
207/// use otlp_stdout_span_exporter::OtlpStdoutSpanExporter;
208///
209/// // Create an exporter with maximum compression
210/// let exporter = OtlpStdoutSpanExporter::with_gzip_level(9);
211/// ```
212#[derive(Debug)]
213pub struct OtlpStdoutSpanExporter {
214    /// GZIP compression level (0-9)
215    gzip_level: u8,
216    /// Optional resource to be included with all spans
217    resource: Option<Resource>,
218    /// Output implementation (stdout or test buffer)
219    output: Arc<dyn Output>,
220}
221
222impl Default for OtlpStdoutSpanExporter {
223    fn default() -> Self {
224        Self::new()
225    }
226}
227
228impl OtlpStdoutSpanExporter {
229    /// Creates a new exporter with default configuration
230    ///
231    /// The default GZIP compression level is determined by:
232    /// 1. The `OTLP_STDOUT_SPAN_EXPORTER_COMPRESSION_LEVEL` environment variable if set
233    /// 2. Otherwise, defaults to level 6
234    ///
235    /// # Example
236    ///
237    /// ```rust
238    /// use otlp_stdout_span_exporter::OtlpStdoutSpanExporter;
239    ///
240    /// let exporter = OtlpStdoutSpanExporter::new();
241    /// ```
242    pub fn new() -> Self {
243        let gzip_level =
244            Self::get_compression_level_from_env().unwrap_or(DEFAULT_COMPRESSION_LEVEL);
245        Self {
246            gzip_level,
247            resource: None,
248            output: Arc::new(StdOutput),
249        }
250    }
251
252    /// Creates a new exporter with custom GZIP compression level
253    ///
254    /// This method explicitly sets the compression level, overriding any value
255    /// set in the `OTLP_STDOUT_SPAN_EXPORTER_COMPRESSION_LEVEL` environment variable.
256    ///
257    /// # Arguments
258    ///
259    /// * `gzip_level` - GZIP compression level (0-9, where 0 is no compression and 9 is maximum compression)
260    ///
261    /// # Example
262    ///
263    /// ```rust
264    /// use otlp_stdout_span_exporter::OtlpStdoutSpanExporter;
265    ///
266    /// // Create an exporter with maximum compression
267    /// let exporter = OtlpStdoutSpanExporter::with_gzip_level(9);
268    /// ```
269    pub fn with_gzip_level(gzip_level: u8) -> Self {
270        Self {
271            gzip_level,
272            resource: None,
273            output: Arc::new(StdOutput),
274        }
275    }
276
277    /// Get the compression level from environment variable
278    ///
279    /// This function tries to read the compression level from the
280    /// `OTLP_STDOUT_SPAN_EXPORTER_COMPRESSION_LEVEL` environment variable.
281    /// It returns None if the variable is not set or cannot be parsed as a u8.
282    fn get_compression_level_from_env() -> Option<u8> {
283        env::var(COMPRESSION_LEVEL_ENV_VAR)
284            .ok()
285            .and_then(|val| val.parse::<u8>().ok())
286            .and_then(|level| if level <= 9 { Some(level) } else { None })
287    }
288
289    #[cfg(test)]
290    fn with_test_output() -> (Self, Arc<TestOutput>) {
291        let output = Arc::new(TestOutput::new());
292        let gzip_level =
293            Self::get_compression_level_from_env().unwrap_or(DEFAULT_COMPRESSION_LEVEL);
294        let exporter = Self {
295            gzip_level,
296            resource: None,
297            output: output.clone() as Arc<dyn Output>,
298        };
299        (exporter, output)
300    }
301
302    /// Parse headers from environment variables
303    ///
304    /// This function reads headers from both global and trace-specific
305    /// environment variables, with trace-specific headers taking precedence.
306    fn parse_headers() -> HashMap<String, String> {
307        let mut headers = HashMap::new();
308
309        // Parse global headers first
310        if let Ok(global_headers) = env::var("OTEL_EXPORTER_OTLP_HEADERS") {
311            Self::parse_header_string(&global_headers, &mut headers);
312        }
313
314        // Parse trace-specific headers (these take precedence)
315        if let Ok(trace_headers) = env::var("OTEL_EXPORTER_OTLP_TRACES_HEADERS") {
316            Self::parse_header_string(&trace_headers, &mut headers);
317        }
318
319        headers
320    }
321
322    /// Parse a header string in the format key1=value1,key2=value2
323    ///
324    /// # Arguments
325    ///
326    /// * `header_str` - The header string to parse
327    /// * `headers` - The map to store parsed headers in
328    fn parse_header_string(header_str: &str, headers: &mut HashMap<String, String>) {
329        for pair in header_str.split(',') {
330            if let Some((key, value)) = pair.split_once('=') {
331                let key = key.trim().to_lowercase();
332                // Skip content-type and content-encoding as they are fixed
333                if key != "content-type" && key != "content-encoding" {
334                    headers.insert(key, value.trim().to_string());
335                }
336            }
337        }
338    }
339
340    /// Get the service name from environment variables
341    ///
342    /// This function tries to get the service name from:
343    /// 1. OTEL_SERVICE_NAME
344    /// 2. AWS_LAMBDA_FUNCTION_NAME
345    /// 3. Falls back to "unknown-service" if neither is set
346    fn get_service_name() -> String {
347        env::var("OTEL_SERVICE_NAME")
348            .or_else(|_| env::var("AWS_LAMBDA_FUNCTION_NAME"))
349            .unwrap_or_else(|_| "unknown-service".to_string())
350    }
351}
352
353#[async_trait]
354impl SpanExporter for OtlpStdoutSpanExporter {
355    /// Export spans to stdout in OTLP format
356    ///
357    /// This function:
358    /// 1. Converts spans to OTLP format
359    /// 2. Serializes them to protobuf
360    /// 3. Compresses the data with GZIP
361    /// 4. Base64 encodes the result
362    /// 5. Writes a JSON object to stdout
363    ///
364    /// # Arguments
365    ///
366    /// * `batch` - A vector of spans to export
367    ///
368    /// # Returns
369    ///
370    /// Returns a resolved future with `Ok(())` if the export was successful, or a `TraceError` if it failed
371    fn export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, Result<(), OTelSdkError>> {
372        // Do all work synchronously
373        let result = (|| {
374            // Convert spans to OTLP format
375            let resource = self
376                .resource
377                .clone()
378                .unwrap_or_else(|| opentelemetry_sdk::Resource::builder_empty().build());
379            let resource_attrs = ResourceAttributesWithSchema::from(&resource);
380            let resource_spans = group_spans_by_resource_and_scope(batch, &resource_attrs);
381            let request = ExportTraceServiceRequest { resource_spans };
382
383            // Serialize to protobuf
384            let proto_bytes = request.encode_to_vec();
385
386            // Compress with GZIP
387            let mut encoder = GzEncoder::new(Vec::new(), Compression::new(self.gzip_level as u32));
388            encoder
389                .write_all(&proto_bytes)
390                .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
391            let compressed_bytes = encoder
392                .finish()
393                .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
394
395            // Base64 encode
396            let payload = base64_engine.encode(compressed_bytes);
397
398            // Prepare the output
399            let output_data = ExporterOutput {
400                version: VERSION,
401                source: Self::get_service_name(),
402                endpoint: DEFAULT_ENDPOINT,
403                method: "POST",
404                content_type: "application/x-protobuf",
405                content_encoding: "gzip",
406                headers: Self::parse_headers(),
407                payload,
408                base64: true,
409            };
410
411            // Write using the output implementation
412            self.output.write_line(
413                &serde_json::to_string(&output_data)
414                    .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?,
415            )?;
416
417            Ok(())
418        })();
419
420        // Return a resolved future with the result
421        Box::pin(std::future::ready(result))
422    }
423
424    /// Shuts down the exporter
425    ///
426    /// This is a no-op for stdout export as no cleanup is needed.
427    ///
428    /// # Returns
429    ///
430    /// Returns `Ok(())` as there is nothing to clean up.
431    fn shutdown(&mut self) -> Result<(), OTelSdkError> {
432        Ok(())
433    }
434
435    /// Force flushes any pending spans
436    ///
437    /// This is a no-op for stdout export as spans are written immediately.
438    ///
439    /// # Returns
440    ///
441    /// Returns `Ok(())` as there is nothing to flush.
442    fn force_flush(&mut self) -> Result<(), OTelSdkError> {
443        Ok(())
444    }
445
446    /// Sets the resource for this exporter.
447    ///
448    /// This method stores a clone of the provided resource to be used when exporting spans.
449    /// The resource represents the entity producing telemetry and will be included in the
450    /// exported trace data.
451    ///
452    /// # Arguments
453    ///
454    /// * `resource` - The resource to associate with this exporter
455    fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
456        self.resource = Some(<opentelemetry_sdk::Resource as Into<Resource>>::into(
457            resource.clone(),
458        ));
459    }
460}
461
462#[cfg(test)]
463mod tests {
464    use super::*;
465    use opentelemetry::{
466        trace::{SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState},
467        InstrumentationScope, KeyValue,
468    };
469    use opentelemetry_sdk::trace::{SpanData, SpanEvents, SpanLinks};
470    use serde_json::Value;
471    use std::time::SystemTime;
472
473    fn create_test_span() -> SpanData {
474        let trace_id_bytes = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 42];
475        let span_id_bytes = [0, 0, 0, 0, 0, 0, 0, 123];
476        let parent_id_bytes = [0, 0, 0, 0, 0, 0, 0, 42];
477
478        let span_context = SpanContext::new(
479            TraceId::from_bytes(trace_id_bytes),
480            SpanId::from_bytes(span_id_bytes),
481            TraceFlags::default(),
482            false,
483            TraceState::default(),
484        );
485
486        SpanData {
487            span_context,
488            parent_span_id: SpanId::from_bytes(parent_id_bytes),
489            span_kind: SpanKind::Client,
490            name: "test-span".into(),
491            start_time: SystemTime::UNIX_EPOCH,
492            end_time: SystemTime::UNIX_EPOCH,
493            attributes: vec![KeyValue::new("test.key", "test-value")],
494            dropped_attributes_count: 0,
495            events: SpanEvents::default(),
496            links: SpanLinks::default(),
497            status: Status::Ok,
498            instrumentation_scope: InstrumentationScope::builder("test-library")
499                .with_version("1.0.0")
500                .with_schema_url("https://opentelemetry.io/schema/1.0.0")
501                .build(),
502        }
503    }
504
505    #[test]
506    fn test_parse_headers() {
507        std::env::set_var("OTEL_EXPORTER_OTLP_HEADERS", "key1=value1,key2=value2");
508        std::env::set_var(
509            "OTEL_EXPORTER_OTLP_TRACES_HEADERS",
510            "key2=override,key3=value3",
511        );
512
513        let headers = OtlpStdoutSpanExporter::parse_headers();
514
515        assert_eq!(headers.get("key1").unwrap(), "value1");
516        assert_eq!(headers.get("key2").unwrap(), "override");
517        assert_eq!(headers.get("key3").unwrap(), "value3");
518
519        // Clean up
520        std::env::remove_var("OTEL_EXPORTER_OTLP_HEADERS");
521        std::env::remove_var("OTEL_EXPORTER_OTLP_TRACES_HEADERS");
522    }
523
524    #[test]
525    fn test_service_name_resolution() {
526        // Test OTEL_SERVICE_NAME priority
527        std::env::set_var("OTEL_SERVICE_NAME", "otel-service");
528        std::env::set_var("AWS_LAMBDA_FUNCTION_NAME", "lambda-function");
529        assert_eq!(OtlpStdoutSpanExporter::get_service_name(), "otel-service");
530
531        // Test AWS_LAMBDA_FUNCTION_NAME fallback
532        std::env::remove_var("OTEL_SERVICE_NAME");
533        assert_eq!(
534            OtlpStdoutSpanExporter::get_service_name(),
535            "lambda-function"
536        );
537
538        // Test default value
539        std::env::remove_var("AWS_LAMBDA_FUNCTION_NAME");
540        assert_eq!(
541            OtlpStdoutSpanExporter::get_service_name(),
542            "unknown-service"
543        );
544    }
545
546    #[test]
547    fn test_compression_level_from_env() {
548        // Test with valid compression level
549        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "3");
550        assert_eq!(
551            OtlpStdoutSpanExporter::get_compression_level_from_env(),
552            Some(3)
553        );
554
555        // Test with invalid compression level (>9)
556        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "10");
557        assert_eq!(
558            OtlpStdoutSpanExporter::get_compression_level_from_env(),
559            None
560        );
561
562        // Test with non-numeric value
563        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "invalid");
564        assert_eq!(
565            OtlpStdoutSpanExporter::get_compression_level_from_env(),
566            None
567        );
568
569        // Test with unset variable
570        std::env::remove_var(COMPRESSION_LEVEL_ENV_VAR);
571        assert_eq!(
572            OtlpStdoutSpanExporter::get_compression_level_from_env(),
573            None
574        );
575    }
576
577    #[test]
578    fn test_new_uses_env_compression_level() {
579        // Set environment variable
580        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "3");
581        let exporter = OtlpStdoutSpanExporter::new();
582        assert_eq!(exporter.gzip_level, 3);
583
584        // Test with unset variable (should use default)
585        std::env::remove_var(COMPRESSION_LEVEL_ENV_VAR);
586        let exporter = OtlpStdoutSpanExporter::new();
587        assert_eq!(exporter.gzip_level, DEFAULT_COMPRESSION_LEVEL);
588    }
589
590    #[test]
591    fn test_with_gzip_level_overrides_env() {
592        // Set environment variable
593        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "3");
594
595        // Explicit level should override environment
596        let exporter = OtlpStdoutSpanExporter::with_gzip_level(8);
597        assert_eq!(exporter.gzip_level, 8);
598
599        // Clean up
600        std::env::remove_var(COMPRESSION_LEVEL_ENV_VAR);
601    }
602
603    #[tokio::test]
604    async fn test_compression_level_affects_output_size() {
605        // Create a large span batch to make compression differences more noticeable
606        let mut spans = Vec::new();
607        for i in 0..100 {
608            let mut span = create_test_span();
609            // Add unique attributes to each span to increase data size
610            span.attributes.push(KeyValue::new("index", i));
611            span.attributes.push(KeyValue::new("data", "a".repeat(100)));
612            spans.push(span);
613        }
614
615        // Test with no compression (level 0)
616        let (mut no_compression_exporter, no_compression_output) =
617            OtlpStdoutSpanExporter::with_test_output();
618        no_compression_exporter.gzip_level = 0;
619        let _ = no_compression_exporter.export(spans.clone()).await;
620        let no_compression_size = extract_payload_size(&no_compression_output.get_output()[0]);
621
622        // Test with medium compression (level 5)
623        let (mut medium_compression_exporter, medium_compression_output) =
624            OtlpStdoutSpanExporter::with_test_output();
625        medium_compression_exporter.gzip_level = 5;
626        let _ = medium_compression_exporter.export(spans.clone()).await;
627        let medium_compression_size =
628            extract_payload_size(&medium_compression_output.get_output()[0]);
629
630        // Test with maximum compression (level 9)
631        let (mut max_compression_exporter, max_compression_output) =
632            OtlpStdoutSpanExporter::with_test_output();
633        max_compression_exporter.gzip_level = 9;
634        let _ = max_compression_exporter.export(spans.clone()).await;
635        let max_compression_size = extract_payload_size(&max_compression_output.get_output()[0]);
636
637        // Verify that higher compression levels result in smaller payloads
638        assert!(no_compression_size > medium_compression_size,
639            "Medium compression (level 5) should produce smaller output than no compression (level 0). Got {} vs {}",
640            medium_compression_size, no_compression_size);
641
642        assert!(medium_compression_size >= max_compression_size,
643            "Maximum compression (level 9) should produce output no larger than medium compression (level 5). Got {} vs {}",
644            max_compression_size, medium_compression_size);
645
646        // Verify that all outputs can be properly decoded and contain the same data
647        let no_compression_spans = decode_and_count_spans(&no_compression_output.get_output()[0]);
648        let medium_compression_spans =
649            decode_and_count_spans(&medium_compression_output.get_output()[0]);
650        let max_compression_spans = decode_and_count_spans(&max_compression_output.get_output()[0]);
651
652        assert_eq!(
653            no_compression_spans,
654            spans.len(),
655            "No compression output should contain all spans"
656        );
657        assert_eq!(
658            medium_compression_spans,
659            spans.len(),
660            "Medium compression output should contain all spans"
661        );
662        assert_eq!(
663            max_compression_spans,
664            spans.len(),
665            "Maximum compression output should contain all spans"
666        );
667    }
668
669    // Helper function to extract the size of the base64-decoded payload
670    fn extract_payload_size(json_str: &str) -> usize {
671        let json: Value = serde_json::from_str(json_str).unwrap();
672        let payload = json["payload"].as_str().unwrap();
673        base64_engine.decode(payload).unwrap().len()
674    }
675
676    // Helper function to decode the payload and count the number of spans
677    fn decode_and_count_spans(json_str: &str) -> usize {
678        let json: Value = serde_json::from_str(json_str).unwrap();
679        let payload = json["payload"].as_str().unwrap();
680        let decoded = base64_engine.decode(payload).unwrap();
681
682        let mut decoder = flate2::read::GzDecoder::new(&decoded[..]);
683        let mut decompressed = Vec::new();
684        std::io::Read::read_to_end(&mut decoder, &mut decompressed).unwrap();
685
686        let request = ExportTraceServiceRequest::decode(&*decompressed).unwrap();
687
688        // Count total spans across all resource spans
689        let mut span_count = 0;
690        for resource_span in &request.resource_spans {
691            for scope_span in &resource_span.scope_spans {
692                span_count += scope_span.spans.len();
693            }
694        }
695
696        span_count
697    }
698
699    #[tokio::test]
700    async fn test_export_single_span() {
701        let (mut exporter, output) = OtlpStdoutSpanExporter::with_test_output();
702        let span = create_test_span();
703
704        let result = exporter.export(vec![span]).await;
705        assert!(result.is_ok());
706
707        let output = output.get_output();
708        assert_eq!(output.len(), 1);
709
710        // Parse and verify the output
711        let json: Value = serde_json::from_str(&output[0]).unwrap();
712        assert_eq!(json["__otel_otlp_stdout"], VERSION);
713        assert_eq!(json["method"], "POST");
714        assert_eq!(json["content-type"], "application/x-protobuf");
715        assert_eq!(json["content-encoding"], "gzip");
716        assert_eq!(json["base64"], true);
717
718        // Verify payload is valid base64 and can be decoded
719        let payload = json["payload"].as_str().unwrap();
720        let decoded = base64_engine.decode(payload).unwrap();
721
722        // Verify it can be decompressed
723        let mut decoder = flate2::read::GzDecoder::new(&decoded[..]);
724        let mut decompressed = Vec::new();
725        std::io::Read::read_to_end(&mut decoder, &mut decompressed).unwrap();
726
727        // Verify it's valid OTLP protobuf
728        let request = ExportTraceServiceRequest::decode(&*decompressed).unwrap();
729        assert_eq!(request.resource_spans.len(), 1);
730    }
731
732    #[tokio::test]
733    async fn test_export_empty_batch() {
734        let mut exporter = OtlpStdoutSpanExporter::new();
735        let result = exporter.export(vec![]).await;
736        assert!(result.is_ok());
737    }
738
739    #[test]
740    fn test_gzip_level_configuration() {
741        let exporter = OtlpStdoutSpanExporter::with_gzip_level(9);
742        assert_eq!(exporter.gzip_level, 9);
743    }
744
745    #[tokio::test]
746    async fn test_env_var_affects_export_compression() {
747        // Create test data
748        let span = create_test_span();
749        let spans = vec![span];
750
751        // Test with environment variable set to level 0 (no compression)
752        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "0");
753        let (mut env_exporter_0, env_output_0) = OtlpStdoutSpanExporter::with_test_output();
754        let _ = env_exporter_0.export(spans.clone()).await;
755        let env_size_0 = extract_payload_size(&env_output_0.get_output()[0]);
756
757        // Test with environment variable set to level 9 (max compression)
758        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "9");
759        let (mut env_exporter_9, env_output_9) = OtlpStdoutSpanExporter::with_test_output();
760        let _ = env_exporter_9.export(spans.clone()).await;
761        let env_size_9 = extract_payload_size(&env_output_9.get_output()[0]);
762
763        // Verify that the environment variable affected the compression level
764        assert!(env_size_0 > env_size_9,
765            "Environment variable COMPRESSION_LEVEL=9 should produce smaller output than COMPRESSION_LEVEL=0. Got {} vs {}",
766            env_size_9, env_size_0);
767
768        // Test with invalid environment variable (should use default)
769        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "invalid");
770        let (mut env_exporter_invalid, _env_output_invalid) =
771            OtlpStdoutSpanExporter::with_test_output();
772        let _ = env_exporter_invalid.export(spans.clone()).await;
773
774        // Test with explicit level (should override environment variable)
775        std::env::set_var(COMPRESSION_LEVEL_ENV_VAR, "0");
776        let (mut explicit_exporter, explicit_output) = OtlpStdoutSpanExporter::with_test_output();
777        explicit_exporter.gzip_level = 9;
778        let _ = explicit_exporter.export(spans.clone()).await;
779        let explicit_size = extract_payload_size(&explicit_output.get_output()[0]);
780
781        // Verify that explicit level overrides environment variable
782        assert!(env_size_0 > explicit_size,
783            "Explicit level 9 should produce smaller output than environment variable level 0. Got {} vs {}",
784            explicit_size, env_size_0);
785
786        // Clean up
787        std::env::remove_var(COMPRESSION_LEVEL_ENV_VAR);
788    }
789}
otlp_stdout_span_exporter/lib.rs

otlp_stdout_span_exporter/
lib.rs
