Unify trace and logs runtime extensions traits. (#1067)

Merge the TraceRuntime and LogRuntime traits in the trace and logs
module respectively into MessageRuntime, an extension trait to Runtime
generic over the type of Message/Signal type to be sent and
received. All references to the older traits now become
MessageRuntime<trace::BatchMessage> and
MessageRuntime<logs::BatchMessage> respectively.

Author: vibhavp

opentelemetry-contrib/src/trace/exporter/jaeger_json.rs

@@ -3,8 +3,9 @@
 
 use async_trait::async_trait;
 use futures::{future::BoxFuture, FutureExt};
+use opentelemetry::runtime::RuntimeChannel;
 use opentelemetry::sdk::export::trace::{ExportResult, SpanData, SpanExporter};
-use opentelemetry::sdk::trace::{TraceRuntime, Tracer};
+use opentelemetry::sdk::trace::{BatchMessage, Tracer};
 use opentelemetry::trace::{SpanId, TraceError};
 use opentelemetry_semantic_conventions::SCHEMA_URL;
 use std::collections::HashMap;
@@ -211,11 +212,11 @@ fn opentelemetry_value_to_json(value: &opentelemetry::Value) -> (&str, serde_jso
     }
 }
 
-/// Jaeger Json Runtime is an extension to [`TraceRuntime`].
+/// Jaeger Json Runtime is an extension to [`RuntimeChannel`].
 ///
-/// [`TraceRuntime`]: opentelemetry::sdk::trace::TraceRuntime
+/// [`RuntimeChannel`]: opentelemetry::sdk::runtime::RuntimeChannel
 #[async_trait]
-pub trait JaegerJsonRuntime: TraceRuntime + std::fmt::Debug {
+pub trait JaegerJsonRuntime: RuntimeChannel<BatchMessage> + std::fmt::Debug {
     /// Create a new directory if the given path does not exist yet
     async fn create_dir(&self, path: &Path) -> ExportResult;
     /// Write the provided content to a new file at the given path

opentelemetry-datadog/src/exporter/mod.rs

@@ -4,6 +4,7 @@ mod model;
 pub use model::ApiVersion;
 pub use model::Error;
 pub use model::FieldMappingFn;
+use opentelemetry::runtime::RuntimeChannel;
 
 use std::borrow::Cow;
 use std::fmt::{Debug, Formatter};
@@ -16,7 +17,7 @@ use opentelemetry::sdk::export::trace;
 use opentelemetry::sdk::export::trace::SpanData;
 use opentelemetry::sdk::resource::ResourceDetector;
 use opentelemetry::sdk::resource::SdkProvidedResourceDetector;
-use opentelemetry::sdk::trace::{Config, TraceRuntime};
+use opentelemetry::sdk::trace::{BatchMessage, Config};
 use opentelemetry::sdk::Resource;
 use opentelemetry::trace::TraceError;
 use opentelemetry::{global, sdk, trace::TracerProvider, KeyValue};
@@ -301,7 +302,7 @@ impl DatadogPipelineBuilder {
 
     /// Install the Datadog trace exporter pipeline using a batch span processor with the specified
     /// runtime.
-    pub fn install_batch<R: TraceRuntime>(
+    pub fn install_batch<R: RuntimeChannel<BatchMessage>>(
         mut self,
         runtime: R,
     ) -> Result<sdk::trace::Tracer, TraceError> {

opentelemetry-jaeger/src/exporter/runtime.rs

@@ -5,14 +5,14 @@
 ))]
 use crate::exporter::addrs_and_family;
 use async_trait::async_trait;
-use opentelemetry::sdk::trace::TraceRuntime;
+use opentelemetry::{runtime::RuntimeChannel, sdk::trace::BatchMessage};
 use std::net::ToSocketAddrs;
 
-/// Jaeger Trace Runtime is an extension to [`TraceRuntime`].
+/// Jaeger Trace Runtime is an extension to [`RuntimeChannel`].
 ///
-/// [`TraceRuntime`]: opentelemetry::sdk::trace::TraceRuntime
+/// [`RuntimeChannel`]: opentelemetry::sdk::runtime::RuntimeChannel
 #[async_trait]
-pub trait JaegerTraceRuntime: TraceRuntime + std::fmt::Debug {
+pub trait JaegerTraceRuntime: RuntimeChannel<BatchMessage> + std::fmt::Debug {
     /// A communication socket between Jaeger client and agent.
     type Socket: std::fmt::Debug + Send + Sync;
 

opentelemetry-otlp/src/logs.rs

@@ -55,7 +55,7 @@ use std::{
 };
 
 use opentelemetry_api::logs::{LogError, LoggerProvider};
-use opentelemetry_sdk::{self, export::logs::LogData, logs::LogRuntime};
+use opentelemetry_sdk::{self, export::logs::LogData, logs::BatchMessage, runtime::RuntimeChannel};
 
 impl OtlpPipeline {
     /// Create a OTLP logging pipeline.
@@ -435,7 +435,7 @@ impl OtlpLogPipeline {
     /// batch log processor.
     ///
     /// [`Logger`]: opentelemetry_sdk::logs::Logger
-    pub fn batch<R: LogRuntime>(
+    pub fn batch<R: RuntimeChannel<BatchMessage>>(
         self,
         runtime: R,
         include_trace_context: bool,
@@ -471,7 +471,7 @@ fn build_simple_with_exporter(
     )
 }
 
-fn build_batch_with_exporter<R: LogRuntime>(
+fn build_batch_with_exporter<R: RuntimeChannel<BatchMessage>>(
     exporter: LogExporter,
     log_config: Option<opentelemetry_sdk::logs::Config>,
     runtime: R,

opentelemetry-otlp/src/span.rs

@@ -60,11 +60,12 @@ use opentelemetry_api::{
 use opentelemetry_sdk::{
     self as sdk,
     export::trace::{ExportResult, SpanData},
-    trace::TraceRuntime,
+    trace::BatchMessage,
 };
 use opentelemetry_semantic_conventions::SCHEMA_URL;
 
 use async_trait::async_trait;
+use sdk::runtime::RuntimeChannel;
 
 /// Target to which the exporter is going to send spans, defaults to https://localhost:4317/v1/traces.
 /// Learn about the relationship between this constant and default/metrics/logs at
@@ -141,7 +142,7 @@ impl OtlpTracePipeline {
     /// `install_batch` will panic if not called within a tokio runtime
     ///
     /// [`Tracer`]: opentelemetry_api::trace::Tracer
-    pub fn install_batch<R: TraceRuntime>(
+    pub fn install_batch<R: RuntimeChannel<BatchMessage>>(
         self,
         runtime: R,
     ) -> Result<sdk::trace::Tracer, TraceError> {
@@ -175,7 +176,7 @@ fn build_simple_with_exporter(
     tracer
 }
 
-fn build_batch_with_exporter<R: TraceRuntime>(
+fn build_batch_with_exporter<R: RuntimeChannel<BatchMessage>>(
     exporter: SpanExporter,
     trace_config: Option<sdk::trace::Config>,
     runtime: R,

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -1,5 +1,8 @@
-use super::{BatchLogProcessor, Config, LogProcessor, LogRuntime, SimpleLogProcessor};
-use crate::export::logs::{LogData, LogExporter};
+use super::{BatchLogProcessor, BatchMessage, Config, LogProcessor, SimpleLogProcessor};
+use crate::{
+    export::logs::{LogData, LogExporter},
+    runtime::RuntimeChannel,
+};
 use opentelemetry_api::{
     global::{self},
     logs::{LogRecord, LogResult},
@@ -129,7 +132,7 @@ impl Builder {
     }
 
     /// The `LogExporter` setup using a default `BatchLogProcessor` that this provider should use.
-    pub fn with_batch_exporter<T: LogExporter + 'static, R: LogRuntime>(
+    pub fn with_batch_exporter<T: LogExporter + 'static, R: RuntimeChannel<BatchMessage>>(
         self,
         exporter: T,
         runtime: R,

opentelemetry-sdk/src/logs/log_processor.rs

@@ -1,7 +1,6 @@
-use super::LogRuntime;
 use crate::{
     export::logs::{ExportResult, LogData, LogExporter},
-    logs::TrySend,
+    runtime::{RuntimeChannel, TrySend},
 };
 use futures_channel::oneshot;
 use futures_util::{
@@ -101,31 +100,32 @@ impl LogProcessor for SimpleLogProcessor {
 
 /// A [`LogProcessor`] that asynchronously buffers log records and reports
 /// them at a preconfigured interval.
-pub struct BatchLogProcessor<R: LogRuntime> {
+pub struct BatchLogProcessor<R: RuntimeChannel<BatchMessage>> {
     message_sender: R::Sender,
 }
 
-impl<R: LogRuntime> Debug for BatchLogProcessor<R> {
+impl<R: RuntimeChannel<BatchMessage>> Debug for BatchLogProcessor<R> {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         f.debug_struct("BatchLogProcessor")
             .field("message_sender", &self.message_sender)
             .finish()
     }
 }
 
-impl<R: LogRuntime> LogProcessor for BatchLogProcessor<R> {
+impl<R: RuntimeChannel<BatchMessage>> LogProcessor for BatchLogProcessor<R> {
     fn emit(&self, data: LogData) {
         let result = self.message_sender.try_send(BatchMessage::ExportLog(data));
 
         if let Err(err) = result {
-            global::handle_error(err);
+            global::handle_error(LogError::Other(err.into()));
         }
     }
 
     fn force_flush(&self) -> LogResult<()> {
         let (res_sender, res_receiver) = oneshot::channel();
         self.message_sender
-            .try_send(BatchMessage::Flush(Some(res_sender)))?;
+            .try_send(BatchMessage::Flush(Some(res_sender)))
+            .map_err(|err| LogError::Other(err.into()))?;
 
         futures_executor::block_on(res_receiver)
             .map_err(|err| LogError::Other(err.into()))
@@ -135,15 +135,16 @@ impl<R: LogRuntime> LogProcessor for BatchLogProcessor<R> {
     fn shutdown(&mut self) -> LogResult<()> {
         let (res_sender, res_receiver) = oneshot::channel();
         self.message_sender
-            .try_send(BatchMessage::Shutdown(res_sender))?;
+            .try_send(BatchMessage::Shutdown(res_sender))
+            .map_err(|err| LogError::Other(err.into()))?;
 
         futures_executor::block_on(res_receiver)
             .map_err(|err| LogError::Other(err.into()))
             .and_then(std::convert::identity)
     }
 }
 
-impl<R: LogRuntime> BatchLogProcessor<R> {
+impl<R: RuntimeChannel<BatchMessage>> BatchLogProcessor<R> {
     pub(crate) fn new(mut exporter: Box<dyn LogExporter>, config: BatchConfig, runtime: R) -> Self {
         let (message_sender, message_receiver) =
             runtime.batch_message_channel(config.max_queue_size);
@@ -247,7 +248,7 @@ async fn export_with_timeout<R, E>(
     batch: Vec<LogData>,
 ) -> ExportResult
 where
-    R: LogRuntime,
+    R: RuntimeChannel<BatchMessage>,
     E: LogExporter + ?Sized,
 {
     if batch.is_empty() {
@@ -308,7 +309,7 @@ pub struct BatchLogProcessorBuilder<E, R> {
 impl<E, R> BatchLogProcessorBuilder<E, R>
 where
     E: LogExporter + 'static,
-    R: LogRuntime,
+    R: RuntimeChannel<BatchMessage>,
 {
     /// Set max queue size for batches
     pub fn with_max_queue_size(self, size: usize) -> Self {

opentelemetry-sdk/src/logs/mod.rs

@@ -3,12 +3,10 @@
 mod config;
 mod log_emitter;
 mod log_processor;
-mod runtime;
 
 pub use config::Config;
 pub use log_emitter::{Builder, Logger, LoggerProvider};
 pub use log_processor::{
     BatchConfig, BatchLogProcessor, BatchLogProcessorBuilder, BatchMessage, LogProcessor,
     SimpleLogProcessor,
 };
-pub use runtime::{LogRuntime, TrySend};