+18

Cargo.lock

reviewed

··· 322 322 "serde_json", 323 323 "thiserror", 324 324 "tokio", 325 325 + "tonic", 325 326 "tracing", 326 327 "tracing-opentelemetry", 327 328 "tracing-subscriber", ··· 744 745 ] 745 746 746 747 [[package]] 748 748 + name = "opentelemetry-http" 749 749 + version = "0.27.0" 750 750 + source = "registry+https://github.com/rust-lang/crates.io-index" 751 751 + checksum = "10a8a7f5f6ba7c1b286c2fbca0454eaba116f63bbe69ed250b642d36fbb04d80" 752 752 + dependencies = [ 753 753 + "async-trait", 754 754 + "bytes", 755 755 + "http", 756 756 + "http-body-util", 757 757 + "hyper", 758 758 + "hyper-util", 759 759 + "opentelemetry", 760 760 + "tokio", 761 761 + ] 762 762 + 763 763 + [[package]] 747 764 name = "opentelemetry-otlp" 748 765 version = "0.27.0" 749 766 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 753 770 "futures-core", 754 771 "http", 755 772 "opentelemetry", 773 773 + "opentelemetry-http", 756 774 "opentelemetry-proto", 757 775 "opentelemetry_sdk", 758 776 "prost",

+2 -1

Cargo.toml

reviewed

··· 6 6 [dependencies] 7 7 opentelemetry = "0.27" 8 8 opentelemetry_sdk = { version = "0.27", features = ["rt-tokio"] } 9 9 - opentelemetry-otlp = { version = "0.27", features = ["tokio", "trace"] } 9 9 + opentelemetry-otlp = { version = "0.27", features = ["tokio", "trace", "http-proto", "grpc-tonic", "hyper-client"] } 10 10 tracing = "0.1" 11 11 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } 12 12 tracing-opentelemetry = "0.28" 13 13 tokio = { version = "1", features = ["full"] } 14 14 + tonic = { version = "0.12" } 14 15 serde = { version = "1.0", features = ["derive"] } 15 16 serde_json = "1.0" 16 17 regex = "1.11"

+60

src/cli.rs

reviewed

··· 50 50 Table, 51 51 } 52 52 53 53 + #[derive(ValueEnum, Clone, Copy, Debug, PartialEq)] 54 54 + pub enum OtlpProtocol { 55 55 + /// Use HTTP/JSON for OTLP export 56 56 + Http, 57 57 + /// Use gRPC for OTLP export 58 58 + Grpc, 59 59 + } 60 60 + 61 61 + impl std::fmt::Display for OtlpProtocol { 62 62 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 63 63 + match self { 64 64 + OtlpProtocol::Http => write!(f, "http"), 65 65 + OtlpProtocol::Grpc => write!(f, "grpc"), 66 66 + } 67 67 + } 68 68 + } 69 69 + 53 70 #[derive(Subcommand)] 54 71 pub enum Commands { 55 72 /// Export log to OpenTelemetry spans ··· 75 92 /// Filter by role (user, assistant) 76 93 #[arg(short, long)] 77 94 pub filter_role: Option<String>, 95 95 + 96 96 + /// OTLP endpoint (e.g., http://localhost:4318 for HTTP, http://localhost:4317 for gRPC) 97 97 + #[arg(long, default_value = "http://localhost:4318")] 98 98 + pub otlp_endpoint: String, 99 99 + 100 100 + /// OTLP protocol to use 101 101 + #[arg(long, default_value_t = OtlpProtocol::Http)] 102 102 + pub otlp_protocol: OtlpProtocol, 103 103 + 104 104 + /// Custom headers for OTLP export (e.g., "Authorization=Bearer $TOKEN") 105 105 + #[arg(long, value_parser = parse_key_value)] 106 106 + pub otlp_header: Vec<(String, String)>, 107 107 + 108 108 + /// Service name for the traces 109 109 + #[arg(long, default_value = "exp2span")] 110 110 + pub otlp_service_name: String, 111 111 + 112 112 + /// Batch size for span export 113 113 + #[arg(long, default_value_t = 100)] 114 114 + pub batch_size: usize, 115 115 + 116 116 + /// Timeout for OTLP export in seconds 117 117 + #[arg(long, default_value_t = 30)] 118 118 + pub otlp_timeout_secs: u64, 119 119 + 120 120 + /// Maximum number of retry attempts for failed exports 121 121 + #[arg(long, default_value_t = 3)] 122 122 + pub otlp_max_retries: u32, 123 123 + 124 124 + /// Initial delay between retries in milliseconds 125 125 + #[arg(long, default_value_t = 100)] 126 126 + pub otlp_retry_delay_ms: u64, 127 127 + } 128 128 + 129 129 + fn parse_key_value(s: &str) -> Result<(String, String)> { 130 130 + let parts: Vec<&str> = s.splitn(2, '=').collect(); 131 131 + if parts.len() != 2 { 132 132 + return Err(anyhow::anyhow!( 133 133 + "Invalid key-value pair: '{}'. Expected format: KEY=VALUE", 134 134 + s 135 135 + )); 136 136 + } 137 137 + Ok((parts[0].to_string(), parts[1].to_string())) 78 138 } 79 139 80 140 #[derive(Parser, Debug)]

+58 -15

src/commands/export.rs

reviewed

··· 1 1 - use anyhow::Result; 2 2 - use crate::cli::ExportArgs; 1 1 + use anyhow::{Context, Result}; 2 2 + 3 3 + use crate::cli::{ExportArgs, OtlpProtocol}; 4 4 + use crate::config::{OtlpConfig, OtlpProtocol as ConfigOtlpProtocol}; 3 5 use crate::parser::LogParser; 4 6 use crate::exporter::OtelExporter; 5 7 6 8 pub async fn execute(args: ExportArgs) -> Result<()> { 7 9 let parser = LogParser::new(); 8 8 - let exporter = OtelExporter::new("exp2span")?; 9 9 - 10 10 - // Parse the log file 10 10 + 11 11 + let otlp_config = OtlpConfig { 12 12 + endpoint: args.otlp_endpoint.clone(), 13 13 + protocol: match args.otlp_protocol { 14 14 + OtlpProtocol::Http => ConfigOtlpProtocol::Http, 15 15 + OtlpProtocol::Grpc => ConfigOtlpProtocol::Grpc, 16 16 + }, 17 17 + headers: args.otlp_header.into_iter().collect(), 18 18 + service_name: args.otlp_service_name.clone(), 19 19 + batch_size: args.batch_size, 20 20 + timeout: std::time::Duration::from_secs(args.otlp_timeout_secs), 21 21 + max_retries: args.otlp_max_retries, 22 22 + retry_delay: std::time::Duration::from_millis(args.otlp_retry_delay_ms), 23 23 + }; 24 24 + 25 25 + tracing::info!( 26 26 + "OTLP configuration: {} -> {}", 27 27 + otlp_config.protocol as u8, 28 28 + otlp_config.endpoint 29 29 + ); 30 30 + 31 31 + if args.dry_run { 32 32 + let log = parser.parse_file(args.file.to_str().unwrap())?; 33 33 + let entries = parser.parse_entries(&log); 34 34 + 35 35 + let filtered_entries = if let Some(ref role) = args.filter_role { 36 36 + entries.iter().filter(|e| e.role == *role).cloned().collect() 37 37 + } else { 38 38 + entries 39 39 + }; 40 40 + 41 41 + tracing::info!("Dry run mode - would export {} spans", filtered_entries.len()); 42 42 + return Ok(()); 43 43 + } 44 44 + 45 45 + let exporter = OtelExporter::new(otlp_config.clone()) 46 46 + .await 47 47 + .context("Failed to create OTLP exporter")?; 48 48 + 49 49 + tracing::info!("Created OTLP exporter: {}", exporter.get_endpoint()); 50 50 + 11 51 let log = parser.parse_file(args.file.to_str().unwrap())?; 12 52 let entries = parser.parse_entries(&log); 13 13 - 14 14 - // Filter by role if specified 53 53 + 15 54 let filtered_entries = if let Some(ref role) = args.filter_role { 16 55 entries.iter().filter(|e| e.role == *role).cloned().collect() 17 56 } else { 18 57 entries 19 58 }; 20 20 - 21 21 - if args.dry_run { 22 22 - tracing::info!("Dry run mode - would export {} spans", filtered_entries.len()); 23 23 - return Ok(()); 24 24 - } 25 25 - 26 26 - // Export spans 59 59 + 60 60 + tracing::info!("Exporting {} spans...", filtered_entries.len()); 61 61 + 27 62 for entry in &filtered_entries { 28 63 exporter.export_log(entry); 29 64 } 30 30 - 65 65 + 66 66 + tracing::info!("Flushing pending spans..."); 67 67 + exporter 68 68 + .shutdown() 69 69 + .await 70 70 + .context("Failed to shutdown OTLP exporter")?; 71 71 + 72 72 + tracing::info!("Export complete"); 73 73 + 31 74 Ok(()) 32 75 }

+37 -6

src/config.rs

reviewed

··· 1 1 use anyhow::Result; 2 2 + use std::collections::HashMap; 3 3 + use std::time::Duration; 4 4 + 5 5 + #[derive(Debug, Clone)] 6 6 + pub struct OtlpConfig { 7 7 + pub endpoint: String, 8 8 + pub protocol: OtlpProtocol, 9 9 + pub headers: HashMap<String, String>, 10 10 + pub service_name: String, 11 11 + pub batch_size: usize, 12 12 + pub timeout: Duration, 13 13 + pub max_retries: u32, 14 14 + pub retry_delay: Duration, 15 15 + } 16 16 + 17 17 + #[derive(Debug, Clone, Copy, PartialEq)] 18 18 + pub enum OtlpProtocol { 19 19 + Http, 20 20 + Grpc, 21 21 + } 22 22 + 23 23 + impl Default for OtlpConfig { 24 24 + fn default() -> Self { 25 25 + Self { 26 26 + endpoint: "http://localhost:4318".to_string(), 27 27 + protocol: OtlpProtocol::Http, 28 28 + headers: HashMap::new(), 29 29 + service_name: "exp2span".to_string(), 30 30 + batch_size: 100, 31 31 + timeout: Duration::from_secs(30), 32 32 + max_retries: 3, 33 33 + retry_delay: Duration::from_millis(100), 34 34 + } 35 35 + } 36 36 + } 2 37 3 38 #[derive(Debug, Clone)] 4 39 pub struct Config { 5 5 - pub otl_endpoint: Option<String>, 6 6 - pub otl_headers: Vec<(String, String)>, 7 7 - pub batch_size: usize, 40 40 + pub otlp: OtlpConfig, 8 41 pub output_format: String, 9 42 pub include_thinking: bool, 10 43 } ··· 12 45 impl Default for Config { 13 46 fn default() -> Self { 14 47 Self { 15 15 - otl_endpoint: None, 16 16 - otl_headers: Vec::new(), 17 17 - batch_size: 100, 48 48 + otlp: OtlpConfig::default(), 18 49 output_format: "json".to_string(), 19 50 include_thinking: false, 20 51 }

+104 -23

src/exporter.rs

reviewed

··· 1 1 - use opentelemetry::trace::{Span, SpanKind, Tracer, TracerProvider as TracerProviderTrait}; 2 2 - use opentelemetry::{trace::TraceError, KeyValue}; 1 1 + use anyhow::{Context, Result}; 2 2 + use opentelemetry::trace::{Span, SpanBuilder, SpanKind, Tracer, TracerProvider as TracerProviderTrait}; 3 3 + use opentelemetry::KeyValue; 4 4 + use opentelemetry_otlp::{SpanExporter, WithExportConfig, WithHttpConfig, WithTonicConfig}; 5 5 + use opentelemetry_sdk::trace::Tracer as SdkTracer; 3 6 use opentelemetry_sdk::trace::TracerProvider; 4 7 use opentelemetry_sdk::Resource; 5 8 9 9 + use crate::config::{OtlpConfig, OtlpProtocol}; 6 10 use crate::parser::ParsedLogEntry; 7 11 8 12 pub struct OtelExporter { 9 9 - tracer: opentelemetry_sdk::trace::Tracer, 13 13 + tracer: SdkTracer, 14 14 + provider: Option<TracerProvider>, 15 15 + config: OtlpConfig, 10 16 } 11 17 12 18 impl OtelExporter { 13 13 - pub fn new(service_name: &str) -> Result<Self, TraceError> { 19 19 + pub async fn new(config: OtlpConfig) -> Result<Self> { 14 20 let resource = Resource::new(vec![ 15 15 - KeyValue::new("service.name", service_name.to_string()), 21 21 + KeyValue::new("service.name", config.service_name.clone()), 16 22 KeyValue::new("service.version", env!("CARGO_PKG_VERSION")), 17 23 KeyValue::new("mcp.protocol.version", "2025-06-18"), 18 24 ]); 19 25 20 20 - let provider = TracerProvider::builder().with_resource(resource).build(); 26 26 + let (tracer, provider) = match config.protocol { 27 27 + OtlpProtocol::Http => Self::create_http_exporter(&config, resource)?, 28 28 + OtlpProtocol::Grpc => Self::create_grpc_exporter(&config, resource)?, 29 29 + }; 30 30 + 31 31 + Ok(Self { 32 32 + tracer, 33 33 + provider: Some(provider), 34 34 + config, 35 35 + }) 36 36 + } 37 37 + 38 38 + fn create_http_exporter( 39 39 + config: &OtlpConfig, 40 40 + resource: Resource, 41 41 + ) -> Result<(SdkTracer, TracerProvider)> { 42 42 + let mut exporter_builder = SpanExporter::builder().with_http(); 43 43 + 44 44 + if !config.endpoint.is_empty() && config.endpoint != "http://localhost:4318" { 45 45 + exporter_builder = exporter_builder.with_endpoint(config.endpoint.clone()); 46 46 + } 47 47 + 48 48 + exporter_builder = exporter_builder.with_timeout(config.timeout); 49 49 + 50 50 + if !config.headers.is_empty() { 51 51 + let headers: std::collections::HashMap<String, String> = config.headers.clone(); 52 52 + exporter_builder = exporter_builder.with_headers(headers); 53 53 + } 54 54 + 55 55 + let exporter = exporter_builder 56 56 + .build() 57 57 + .context("Failed to create HTTP OTLP exporter")?; 58 58 + 59 59 + let provider = TracerProvider::builder() 60 60 + .with_simple_exporter(exporter) 61 61 + .with_resource(resource) 62 62 + .build(); 21 63 22 64 let tracer = provider.tracer("exp2span"); 65 65 + Ok((tracer, provider)) 66 66 + } 67 67 + 68 68 + fn create_grpc_exporter( 69 69 + config: &OtlpConfig, 70 70 + resource: Resource, 71 71 + ) -> Result<(SdkTracer, TracerProvider)> { 72 72 + use tonic::metadata::MetadataMap; 73 73 + use tonic::metadata::MetadataKey; 74 74 + use tonic::metadata::MetadataValue; 23 75 24 24 - Ok(Self { tracer }) 76 76 + let mut exporter_builder = SpanExporter::builder().with_tonic(); 77 77 + 78 78 + if !config.endpoint.is_empty() && config.endpoint != "http://localhost:4317" { 79 79 + exporter_builder = exporter_builder.with_endpoint(config.endpoint.clone()); 80 80 + } 81 81 + 82 82 + exporter_builder = exporter_builder.with_timeout(config.timeout); 83 83 + 84 84 + if !config.headers.is_empty() { 85 85 + let mut metadata_map = MetadataMap::new(); 86 86 + for (key, value) in &config.headers { 87 87 + let meta_key = MetadataKey::from_bytes(key.as_bytes()) 88 88 + .map_err(|e| anyhow::anyhow!("Invalid metadata key '{}': {}", key, e))?; 89 89 + let meta_value = MetadataValue::try_from(value.as_str()) 90 90 + .map_err(|e| anyhow::anyhow!("Invalid metadata value '{}': {}", value, e))?; 91 91 + metadata_map.insert(meta_key, meta_value); 92 92 + } 93 93 + exporter_builder = exporter_builder.with_metadata(metadata_map); 94 94 + } 95 95 + 96 96 + let exporter = exporter_builder 97 97 + .build() 98 98 + .context("Failed to create gRPC OTLP exporter")?; 99 99 + 100 100 + let provider = TracerProvider::builder() 101 101 + .with_simple_exporter(exporter) 102 102 + .with_resource(resource) 103 103 + .build(); 104 104 + 105 105 + let tracer = provider.tracer("exp2span"); 106 106 + Ok((tracer, provider)) 25 107 } 26 108 27 109 pub fn export_log(&self, entry: &ParsedLogEntry) { 28 28 - use opentelemetry::trace::SpanBuilder; 29 29 - 30 110 let mut attributes = vec![ 31 111 KeyValue::new("mcp.session.id", entry.session_id.clone()), 32 112 KeyValue::new("network.transport", "tcp"), ··· 68 148 let mut span = self.tracer.build(builder); 69 149 span.end(); 70 150 71 71 - tracing::info!( 151 151 + tracing::debug!( 72 152 "Created span: {} with {} attributes", 73 153 span_name_clone, 74 154 attr_count 75 155 ); 76 156 } 77 157 78 78 - pub fn export_log_file( 79 79 - &self, 80 80 - parser: &crate::parser::LogParser, 81 81 - path: &str, 82 82 - ) -> anyhow::Result<()> { 83 83 - use anyhow::Context; 158 158 + pub async fn shutdown(self) -> Result<()> { 159 159 + if let Some(provider) = self.provider { 160 160 + provider 161 161 + .shutdown() 162 162 + .context("Failed to shutdown tracer provider")?; 163 163 + tracing::info!("OTLP exporter shutdown complete"); 164 164 + } 165 165 + Ok(()) 166 166 + } 84 167 85 85 - let log = parser.parse_file(path)?; 86 86 - let entries = parser.parse_entries(&log); 87 87 - 88 88 - for entry in &entries { 89 89 - self.export_log(entry); 90 90 - } 168 168 + pub fn get_endpoint(&self) -> &str { 169 169 + &self.config.endpoint 170 170 + } 91 171 92 92 - Ok(()) 172 172 + pub fn get_protocol(&self) -> OtlpProtocol { 173 173 + self.config.protocol 93 174 } 94 175 }