Skip to content

Commit

Permalink
Global error handler cleanup - Trace SDK (#2256)
Browse files Browse the repository at this point in the history 
Co-authored-by: Cijo Thomas <[email protected]>
  • Loading branch information
@lalitb @cijothomas
lalitb and cijothomas authored Nov 3, 2024
1 parent 3742953 commit fcd7cae
Show file tree
Hide file tree
Showing 2 changed files with 58 additions and 31 deletions.
53 changes: 31 additions & 22 deletions opentelemetry-sdk/src/trace/config.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
//! can be set for the default OpenTelemetry limits and Sampler.
use crate::trace::{span_limit::SpanLimits, IdGenerator, RandomIdGenerator, Sampler, ShouldSample};
use crate::Resource;
use opentelemetry::global::{handle_error, Error};
use opentelemetry::otel_warn;
use std::borrow::Cow;
use std::env;
use std::str::FromStr;
Expand Down Expand Up @@ -125,13 +125,14 @@ impl Default for Config {
"always_on" => Box::new(Sampler::AlwaysOn),
"always_off" => Box::new(Sampler::AlwaysOff),
"traceidratio" => {
let ratio = sampler_arg.and_then(|r| r.parse::<f64>().ok());
let ratio = sampler_arg.as_ref().and_then(|r| r.parse::<f64>().ok());
if let Some(r) = ratio {
Box::new(Sampler::TraceIdRatioBased(r))
} else {
handle_error(
Error::Other(String::from(
"Missing or invalid OTEL_TRACES_SAMPLER_ARG value. Falling back to default: 1.0"))
otel_warn!(
name: "TracerProvider.Config.InvalidSamplerArgument",
message = "OTEL_TRACES_SAMPLER is set to 'traceidratio' but OTEL_TRACES_SAMPLER_ARG environment variable is missing or invalid. OTEL_TRACES_SAMPLER_ARG must be a valid float between 0.0 and 1.0 representing the desired sampling probability (0.0 = no traces sampled, 1.0 = all traces sampled, 0.5 = 50% of traces sampled). Falling back to default ratio: 1.0 (100% sampling)",
otel_traces_sampler_arg = format!("{:?}", sampler_arg)
);
Box::new(Sampler::TraceIdRatioBased(1.0))
}
Expand All @@ -143,43 +144,51 @@ impl Default for Config {
Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOff)))
}
"parentbased_traceidratio" => {
let ratio = sampler_arg.and_then(|r| r.parse::<f64>().ok());
let ratio = sampler_arg.as_ref().and_then(|r| r.parse::<f64>().ok());
if let Some(r) = ratio {
Box::new(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
r,
))))
} else {
handle_error(
Error::Other(String::from(
"Missing or invalid OTEL_TRACES_SAMPLER_ARG value. Falling back to default: 1.0"
)));
otel_warn!(
name: "TracerProvider.Config.InvalidSamplerArgument",
message = "OTEL_TRACES_SAMPLER is set to 'parentbased_traceidratio' but OTEL_TRACES_SAMPLER_ARG environment variable is missing or invalid. OTEL_TRACES_SAMPLER_ARG must be a valid float between 0.0 and 1.0 representing the desired sampling probability (0.0 = no traces sampled, 1.0 = all traces sampled, 0.5 = 50% of traces sampled). Falling back to default ratio: 1.0 (100% sampling)",
otel_traces_sampler_arg = format!("{:?}", sampler_arg)
);
Box::new(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
1.0,
))))
}
}
"parentbased_jaeger_remote" => {
handle_error(
Error::Other(String::from(
"Unimplemented parentbased_jaeger_remote sampler. Falling back to default: parentbased_always_on"
)));
otel_warn!(
name: "TracerProvider.Config.UnsupportedSampler",
message = "OTEL_TRACES_SAMPLER is set to 'parentbased_jaeger_remote' which is not implemented in this SDK version. Using fallback sampler: ParentBased(AlwaysOn). Configure an alternative sampler using OTEL_TRACES_SAMPLER"
);
Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)))
}
"jaeger_remote" => {
handle_error(
Error::Other(String::from("Unimplemented jaeger_remote sampler. Falling back to default: parentbased_always_on")));
otel_warn!(
name: "TracerProvider.Config.UnsupportedSampler",
message = "OTEL_TRACES_SAMPLER is set to 'jaeger_remote' which is implemented in this SDK version. Using fallback sampler: ParentBased(AlwaysOn). Configure an alternative sampler using OTEL_TRACES_SAMPLER"
);
Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)))
}
"xray" => {
handle_error(
Error::Other(String::from("Unimplemented xray sampler. Falling back to default: parentbased_always_on")));
otel_warn!(
name: "TracerProvider.Config.UnsupportedSampler",
message = "OTEL_TRACES_SAMPLER is set to 'xray'. AWS X-Ray sampler is not implemented in this SDK version. Using fallback sampler: ParentBased(AlwaysOn). Configure an alternative sampler using OTEL_TRACES_SAMPLER"
);
Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)))
}
s => {
handle_error(
Error::Other(format!("Unrecognised OTEL_TRACES_SAMPLER value: {}. Falling back to default: parentbased_always_on",
s
)));
otel_warn!(
name: "TracerProvider.Config.InvalidSamplerType",
message = format!(
"Unrecognized sampler type '{}' in OTEL_TRACES_SAMPLER environment variable. Valid values are: always_on, always_off, traceidratio, parentbased_always_on, parentbased_always_off, parentbased_traceidratio. Using fallback sampler: ParentBased(AlwaysOn)",
s
),
);
Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)))
}
}
Expand Down
36 changes: 27 additions & 9 deletions opentelemetry-sdk/src/trace/span_processor.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ use futures_util::{
stream::{self, FusedStream, FuturesUnordered},
StreamExt as _,
};
use opentelemetry::global;
use opentelemetry::{otel_debug, otel_error};
use opentelemetry::{
trace::{TraceError, TraceResult},
Context,
Expand Down Expand Up @@ -134,7 +134,11 @@ impl SpanProcessor for SimpleSpanProcessor {
.and_then(|mut exporter| futures_executor::block_on(exporter.export(vec![span])));

if let Err(err) = result {
global::handle_error(err);
// TODO: check error type, and log `error` only if the error is user-actiobable, else log `debug`
otel_debug!(
name: "SimpleProcessor.OnEnd.Error",
reason = format!("{:?}", err)
);
}
}

Expand Down Expand Up @@ -246,7 +250,10 @@ impl<R: RuntimeChannel> SpanProcessor for BatchSpanProcessor<R> {
let result = self.message_sender.try_send(BatchMessage::ExportSpan(span));

if let Err(err) = result {
global::handle_error(TraceError::Other(err.into()));
otel_debug!(
name: "BatchSpanProcessor.OnEnd.ExportQueueingFailed",
reason = format!("{:?}", TraceError::Other(err.into()))
);
}
}

Expand Down Expand Up @@ -313,14 +320,22 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
let result = export_task.await;

if let Some(channel) = res_channel {
// If a response channel is provided, attempt to send the export result through it.
if let Err(result) = channel.send(result) {
global::handle_error(TraceError::from(format!(
"failed to send flush result: {:?}",
result
)));
otel_debug!(
name: "BatchSpanProcessor.Flush.SendResultError",
reason = format!("{:?}", result)
);
}
} else if let Err(err) = result {
global::handle_error(err);
// If no channel is provided and the export operation encountered an error,
// log the error directly here.
// TODO: Consider returning the status instead of logging it.
otel_error!(
name: "BatchSpanProcessor.Flush.ExportError",
reason = format!("{:?}", err),
message = "Failed during the export process"
);
}

Ok(())
Expand Down Expand Up @@ -354,7 +369,10 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
let export_task = self.export();
let task = async move {
if let Err(err) = export_task.await {
global::handle_error(err);
otel_error!(
name: "BatchSpanProcessor.Export.Error",
reason = format!("{}", err)
);
}

Ok(())
Expand Down

0 comments on commit fcd7cae

Please sign in to comment.