Test otlp exporter in integration scenario (#424)

Author: mraerino

opentelemetry-otlp/Cargo.toml

@@ -15,10 +15,16 @@ keywords = ["opentelemetry", "otlp", "logging", "tracing", "metrics"]
 license = "Apache-2.0"
 edition = "2018"
 build = "build.rs"
+autotests = false
 
 [lib]
 doctest = false
 
+[[test]]
+name = "smoke"
+path = "tests/smoke.rs"
+required-features = ["integration-testing"]
+
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
@@ -47,6 +53,7 @@ tls = ["tonic/tls"]
 tls-roots = ["tls", "tonic/tls-roots"]
 openssl = ["grpcio/openssl"]
 openssl-vendored = ["grpcio/openssl-vendored"]
+integration-testing = ["tonic", "tonic-build", "prost", "tokio/full", "opentelemetry/trace"]
 
 [build-dependencies]
 protobuf-codegen = { version = "2.16", optional = true }

opentelemetry-otlp/build.rs

@@ -10,7 +10,7 @@ use protoc_grpcio::compile_grpc_protos;
 fn main() {
     #[cfg(feature = "tonic")]
     tonic_build::configure()
-        .build_server(false)
+        .build_server(std::env::var_os("CARGO_FEATURE_INTEGRATION_TESTING").is_some())
         .build_client(true)
         .format(false)
         .compile(

opentelemetry-otlp/src/lib.rs

@@ -125,15 +125,23 @@ use std::collections::HashMap;
 
 use std::time::Duration;
 
-#[cfg(feature = "tonic")]
+#[cfg(all(feature = "tonic", not(feature = "integration-testing")))]
 #[rustfmt::skip]
 #[allow(clippy::all, unreachable_pub)]
 mod proto;
 
-#[cfg(all(feature = "grpc-sys", not(feature = "tonic")))]
+#[cfg(all(
+    feature = "grpc-sys",
+    not(feature = "tonic"),
+    not(feature = "integration-testing")
+))]
 #[allow(clippy::all, unreachable_pub, dead_code)]
 mod proto;
 
+#[cfg(feature = "integration-testing")]
+#[allow(missing_docs, unreachable_pub)]
+pub mod proto;
+
 #[cfg(feature = "metrics")]
 #[allow(warnings)]
 mod metric;

opentelemetry-otlp/src/proto.rs

@@ -1,5 +1,5 @@
 #[cfg(feature = "tonic")]
-pub(crate) mod collector {
+pub mod collector {
     pub mod metrics {
         pub mod v1 {
             tonic::include_proto!("opentelemetry.proto.collector.metrics.v1");
@@ -14,28 +14,28 @@ pub(crate) mod collector {
 }
 
 #[cfg(feature = "tonic")]
-pub(crate) mod common {
+pub mod common {
     pub mod v1 {
         tonic::include_proto!("opentelemetry.proto.common.v1");
     }
 }
 
 #[cfg(feature = "tonic")]
-pub(crate) mod metrics {
+pub mod metrics {
     pub mod v1 {
         tonic::include_proto!("opentelemetry.proto.metrics.v1");
     }
 }
 
 #[cfg(feature = "tonic")]
-pub(crate) mod resource {
+pub mod resource {
     pub mod v1 {
         tonic::include_proto!("opentelemetry.proto.resource.v1");
     }
 }
 
 #[cfg(feature = "tonic")]
-pub(crate) mod trace {
+pub mod trace {
     pub mod v1 {
         tonic::include_proto!("opentelemetry.proto.trace.v1");
     }

opentelemetry-otlp/tests/smoke.rs

@@ -0,0 +1,97 @@
+use futures::StreamExt;
+use opentelemetry::trace::{Span, SpanKind, Tracer};
+use opentelemetry_otlp::proto::collector::trace::v1::{
+    trace_service_server::{TraceService, TraceServiceServer},
+    ExportTraceServiceRequest, ExportTraceServiceResponse,
+};
+use std::{net::SocketAddr, sync::Mutex};
+use tokio::sync::mpsc;
+
+struct MockServer {
+    tx: Mutex<mpsc::Sender<ExportTraceServiceRequest>>,
+}
+
+impl MockServer {
+    pub fn new(tx: mpsc::Sender<ExportTraceServiceRequest>) -> Self {
+        Self { tx: Mutex::new(tx) }
+    }
+}
+
+#[tonic::async_trait]
+impl TraceService for MockServer {
+    async fn export(
+        &self,
+        request: tonic::Request<ExportTraceServiceRequest>,
+    ) -> Result<tonic::Response<ExportTraceServiceResponse>, tonic::Status> {
+        println!("Sending request into channel...");
+        self.tx
+            .lock()
+            .unwrap()
+            .try_send(request.into_inner())
+            .expect("Channel full");
+        Ok(tonic::Response::new(ExportTraceServiceResponse {}))
+    }
+}
+
+async fn setup() -> (SocketAddr, mpsc::Receiver<ExportTraceServiceRequest>) {
+    let addr: SocketAddr = "[::1]:0".parse().unwrap();
+    let listener = tokio::net::TcpListener::bind(addr)
+        .await
+        .expect("failed to bind");
+    let addr = listener.local_addr().unwrap();
+    let stream = listener.map(|s| {
+        if let Ok(ref s) = s {
+            println!("Got new conn at {}", s.peer_addr().unwrap());
+        }
+        s
+    });
+
+    let (req_tx, req_rx) = mpsc::channel(10);
+    let service = TraceServiceServer::new(MockServer::new(req_tx));
+    tokio::task::spawn(async move {
+        tonic::transport::Server::builder()
+            .add_service(service)
+            .serve_with_incoming(stream)
+            .await
+            .expect("Server failed");
+    });
+    (addr, req_rx)
+}
+
+#[tokio::test(threaded_scheduler)]
+async fn smoke_tracer() {
+    println!("Starting server setup...");
+    let (addr, mut req_rx) = setup().await;
+
+    {
+        println!("Installing tracer...");
+        let (tracer, _uninstall) = opentelemetry_otlp::new_pipeline()
+            .with_endpoint(format!("http://{}", addr))
+            .install()
+            .expect("failed to install");
+
+        println!("Sending span...");
+        let span = tracer
+            .span_builder("my-test-span")
+            .with_kind(SpanKind::Server)
+            .start(&tracer);
+        span.add_event("my-test-event".into(), vec![]);
+        span.end();
+    }
+
+    println!("Waiting for request...");
+    let req = req_rx.recv().await.expect("missing export request");
+    let first_span = req
+        .resource_spans
+        .get(0)
+        .unwrap()
+        .instrumentation_library_spans
+        .get(0)
+        .unwrap()
+        .spans
+        .get(0)
+        .unwrap();
+    assert_eq!("my-test-span", first_span.name);
+    let first_event = first_span.events.get(0).unwrap();
+    assert_eq!("my-test-event", first_event.name);
+}