Skip to content

Commit 161f087

Browse files
lalitblalitb
committed
set_resource status, and call synchronously
1 parent 4df74e8 commit 161f087

File tree

16 files changed

+116
-50
lines changed

16 files changed

+116
-50
lines changed

‎opentelemetry-otlp/src/exporter/http/logs.rs

+2-1
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,8 @@ impl LogExporter for OtlpHttpClient {
5757
let _ = self.client.lock().map(|mut c| c.take());
5858
}
5959

60-
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
60+
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) -> LogResult<()> {
6161
self.resource = resource.into();
62+
Ok(())
6263
}
6364
}

‎opentelemetry-otlp/src/exporter/http/trace.rs

+2-1
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,8 @@ impl SpanExporter for OtlpHttpClient {
6767
let _ = self.client.lock().map(|mut c| c.take());
6868
}
6969

70-
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
70+
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) -> ExportResult {
7171
self.resource = resource.into();
72+
Ok(())
7273
}
7374
}

‎opentelemetry-otlp/src/exporter/tonic/logs.rs

+2-1
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,8 @@ impl LogExporter for TonicLogsClient {
9191
let _ = self.inner.take();
9292
}
9393

94-
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
94+
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) -> LogResult<()> {
9595
self.resource = resource.into();
96+
Ok(())
9697
}
9798
}

‎opentelemetry-otlp/src/exporter/tonic/trace.rs

+2-1
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,8 @@ impl SpanExporter for TonicTracesClient {
9292
let _ = self.inner.take();
9393
}
9494

95-
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
95+
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) -> ExportResult {
9696
self.resource = resource.into();
97+
Ok(())
9798
}
9899
}

‎opentelemetry-otlp/src/logs.rs

+5-2
Original file line numberDiff line numberDiff line change
@@ -105,8 +105,11 @@ impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
105105
self.client.export(batch).await
106106
}
107107

108-
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
109-
self.client.set_resource(resource);
108+
fn set_resource(
109+
&mut self,
110+
resource: &opentelemetry_sdk::Resource,
111+
) -> opentelemetry::logs::LogResult<()> {
112+
self.client.set_resource(resource)
110113
}
111114
}
112115

‎opentelemetry-otlp/src/span.rs

+2-2
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,7 @@ impl opentelemetry_sdk::export::trace::SpanExporter for SpanExporter {
214214
self.0.export(batch)
215215
}
216216

217-
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
218-
self.0.set_resource(resource);
217+
fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) -> ExportResult {
218+
self.0.set_resource(resource)
219219
}
220220
}

‎opentelemetry-sdk/src/export/logs/mod.rs

+18-2
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,18 @@ use std::fmt::Debug;
1414
/// `LogExporter` defines the interface that log exporters should implement.
1515
#[async_trait]
1616
pub trait LogExporter: Send + Sync + Debug {
17-
/// Exports a batch of [`LogData`].
17+
/// Exports a batch of readable logs. Protocol exporters that will
18+
/// implement this function are typically expected to serialize and transmit
19+
/// the data to the destination.
20+
///
21+
/// This function will never be called concurrently for the same exporter
22+
/// instance. It can be called again only after the current call returns.
23+
///
24+
/// This function must not block indefinitely, there must be a reasonable
25+
/// upper limit after which the call must time out with an error result.
26+
///
27+
/// Any retry logic that is required by the exporter is the responsibility
28+
/// of the exporter.
1829
async fn export<'a>(&mut self, batch: Vec<Cow<'a, LogData>>) -> LogResult<()>;
1930
/// Shuts down the exporter.
2031
fn shutdown(&mut self) {}
@@ -24,7 +35,12 @@ pub trait LogExporter: Send + Sync + Debug {
2435
true
2536
}
2637
/// Set the resource for the exporter.
27-
fn set_resource(&mut self, _resource: &Resource) {}
38+
/// This function SHOULD only be called once during the initialization of the exporter.
39+
/// This function SHOULD complete or abort within some timeout. This function SHOULD be
40+
/// implemented as a blocking API
41+
fn set_resource(&mut self, _resource: &Resource) -> LogResult<()> {
42+
Ok(())
43+
}
2844
}
2945

3046
/// `LogData` represents a single log event without resource context.

‎opentelemetry-sdk/src/export/trace.rs

+6-1
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,12 @@ pub trait SpanExporter: Send + Sync + Debug {
6565
}
6666

6767
/// Set the resource for the exporter.
68-
fn set_resource(&mut self, _resource: &Resource) {}
68+
/// This function SHOULD only be called once during the initialization of the exporter.
69+
/// This function SHOULD complete or abort within some timeout. This function SHOULD be
70+
/// implemented as a blocking API
71+
fn set_resource(&mut self, _resource: &Resource) -> ExportResult {
72+
Ok(())
73+
}
6974
}
7075

7176
/// `SpanData` contains all the information collected by a `Span` and can be used

‎opentelemetry-sdk/src/logs/log_emitter.rs

+3-1
Original file line numberDiff line numberDiff line change
@@ -205,7 +205,9 @@ impl Builder {
205205

206206
// invoke set_resource on all the processors
207207
for processor in logger_provider.log_processors() {
208-
processor.set_resource(logger_provider.resource());
208+
if let Err(err) = processor.set_resource(logger_provider.resource()) {
209+
global::handle_error(err);
210+
}
209211
}
210212
logger_provider
211213
}

‎opentelemetry-sdk/src/logs/log_processor.rs

+35-17
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,9 @@ pub trait LogProcessor: Send + Sync + Debug {
6767
fn event_enabled(&self, level: Severity, target: &str, name: &str) -> bool;
6868

6969
/// Set the resource for the log processor.
70-
fn set_resource(&self, _resource: &Resource) {}
70+
fn set_resource(&self, _resource: &Resource) -> LogResult<()> {
71+
Ok(())
72+
}
7173
}
7274

7375
/// A [LogProcessor] that passes logs to the configured `LogExporter`, as soon
@@ -125,9 +127,13 @@ impl LogProcessor for SimpleLogProcessor {
125127
}
126128
}
127129

128-
fn set_resource(&self, resource: &Resource) {
130+
fn set_resource(&self, resource: &Resource) -> LogResult<()> {
129131
if let Ok(mut exporter) = self.exporter.lock() {
130-
exporter.set_resource(resource);
132+
exporter.set_resource(resource)
133+
} else {
134+
Err(LogError::Other(
135+
"simple logprocessor mutex poison during set_resource".into(),
136+
))
131137
}
132138
}
133139

@@ -189,11 +195,16 @@ impl<R: RuntimeChannel> LogProcessor for BatchLogProcessor<R> {
189195
.and_then(std::convert::identity)
190196
}
191197

192-
fn set_resource(&self, resource: &Resource) {
198+
fn set_resource(&self, resource: &Resource) -> LogResult<()> {
199+
let (res_sender, res_receiver) = oneshot::channel();
193200
let resource = Arc::new(resource.clone());
194-
let _ = self
195-
.message_sender
196-
.try_send(BatchMessage::SetResource(resource));
201+
self.message_sender
202+
.try_send(BatchMessage::SetResource(resource, res_sender))
203+
.map_err(|err| LogError::Other(err.into()))?;
204+
205+
futures_executor::block_on(res_receiver)
206+
.map_err(|err| LogError::Other(err.into()))
207+
.and_then(std::convert::identity)
197208
}
198209
}
199210

@@ -275,8 +286,14 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
275286
}
276287

277288
// propagate the resource
278-
BatchMessage::SetResource(resource) => {
279-
exporter.set_resource(&resource);
289+
BatchMessage::SetResource(resource, res_sender) => {
290+
let result = exporter.set_resource(&resource);
291+
if let Err(result) = res_sender.send(result) {
292+
global::handle_error(LogError::from(format!(
293+
"failed to send set resource result: {:?}",
294+
result
295+
)));
296+
}
280297
}
281298
}
282299
}
@@ -500,7 +517,7 @@ enum BatchMessage {
500517
/// Shut down the worker thread, push all logs in buffer to the backend.
501518
Shutdown(oneshot::Sender<ExportResult>),
502519
/// Set the resource for the exporter.
503-
SetResource(Arc<Resource>),
520+
SetResource(Arc<Resource>, oneshot::Sender<LogResult<()>>),
504521
}
505522

506523
#[cfg(all(test, feature = "testing", feature = "logs"))]
@@ -527,7 +544,7 @@ mod tests {
527544
use opentelemetry::logs::AnyValue;
528545
#[cfg(feature = "logs_level_enabled")]
529546
use opentelemetry::logs::Severity;
530-
use opentelemetry::logs::{Logger, LoggerProvider as _};
547+
use opentelemetry::logs::{LogError, Logger, LoggerProvider as _};
531548
use opentelemetry::Key;
532549
use opentelemetry::{logs::LogResult, KeyValue};
533550
use std::borrow::Cow;
@@ -547,13 +564,14 @@ mod tests {
547564

548565
fn shutdown(&mut self) {}
549566

550-
fn set_resource(&mut self, resource: &Resource) {
551-
self.resource
552-
.lock()
553-
.map(|mut res_opt| {
567+
fn set_resource(&mut self, resource: &Resource) -> LogResult<()> {
568+
match self.resource.lock() {
569+
Ok(mut res_opt) => {
554570
res_opt.replace(resource.clone());
555-
})
556-
.expect("mock log exporter shouldn't error when setting resource");
571+
Ok(())
572+
}
573+
Err(_) => Err(LogError::Other("mock log exporter mutex poison".into())),
574+
}
557575
}
558576
}
559577

‎opentelemetry-sdk/src/testing/logs/in_memory_exporter.rs

+3-2
Original file line numberDiff line numberDiff line change
@@ -189,8 +189,9 @@ impl LogExporter for InMemoryLogsExporter {
189189
}
190190
}
191191

192-
fn set_resource(&mut self, resource: &Resource) {
193-
let mut res_guard = self.resource.lock().expect("Resource lock poisoned");
192+
fn set_resource(&mut self, resource: &Resource) -> LogResult<()> {
193+
let mut res_guard = self.resource.lock().map_err(LogError::from)?;
194194
*res_guard = resource.clone();
195+
Ok(())
195196
}
196197
}

‎opentelemetry-sdk/src/testing/trace/in_memory_exporter.rs

+4-5
Original file line numberDiff line numberDiff line change
@@ -146,10 +146,9 @@ impl SpanExporter for InMemorySpanExporter {
146146
self.reset()
147147
}
148148

149-
fn set_resource(&mut self, resource: &Resource) {
150-
self.resource
151-
.lock()
152-
.map(|mut res_guard| *res_guard = resource.clone())
153-
.expect("Resource lock poisoned");
149+
fn set_resource(&mut self, resource: &Resource) -> ExportResult {
150+
let mut res_guard = self.resource.lock().map_err(TraceError::from)?;
151+
*res_guard = resource.clone();
152+
Ok(())
154153
}
155154
}

‎opentelemetry-sdk/src/trace/provider.rs

+3-1
Original file line numberDiff line numberDiff line change
@@ -289,7 +289,9 @@ impl Builder {
289289

290290
// Set the resource for each processor
291291
for p in &mut processors {
292-
p.set_resource(config.resource.as_ref());
292+
if let Err(err) = p.set_resource(config.resource.as_ref()) {
293+
global::handle_error(err);
294+
}
293295
}
294296

295297
TracerProvider::new(TracerProviderInner { processors, config })

‎opentelemetry-sdk/src/trace/span_processor.rs

+25-11
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,9 @@ pub trait SpanProcessor: Send + Sync + std::fmt::Debug {
9696
/// Implementation should make sure shutdown can be called multiple times.
9797
fn shutdown(&self) -> TraceResult<()>;
9898
/// Set the resource for the log processor.
99-
fn set_resource(&mut self, _resource: &Resource) {}
99+
fn set_resource(&mut self, _resource: &Resource) -> TraceResult<()> {
100+
Ok(())
101+
}
100102
}
101103

102104
/// A [SpanProcessor] that passes finished spans to the configured
@@ -153,9 +155,10 @@ impl SpanProcessor for SimpleSpanProcessor {
153155
}
154156
}
155157

156-
fn set_resource(&mut self, resource: &Resource) {
157-
if let Ok(mut exporter) = self.exporter.lock() {
158-
exporter.set_resource(resource);
158+
fn set_resource(&mut self, resource: &Resource) -> ExportResult {
159+
match self.exporter.lock() {
160+
Ok(mut exporter) => exporter.set_resource(resource),
161+
Err(_) => Err(TraceError::Other("SimpleSpanProcessor mutex poison".into())),
159162
}
160163
}
161164
}
@@ -271,11 +274,16 @@ impl<R: RuntimeChannel> SpanProcessor for BatchSpanProcessor<R> {
271274
.and_then(|identity| identity)
272275
}
273276

274-
fn set_resource(&mut self, resource: &Resource) {
277+
fn set_resource(&mut self, resource: &Resource) -> ExportResult {
278+
let (res_sender, res_receiver) = oneshot::channel();
275279
let resource = Arc::new(resource.clone());
276-
let _ = self
277-
.message_sender
278-
.try_send(BatchMessage::SetResource(resource));
280+
self.message_sender
281+
.try_send(BatchMessage::SetResource(resource, res_sender))
282+
.map_err(|err| TraceError::Other(err.into()))?;
283+
284+
futures_executor::block_on(res_receiver)
285+
.map_err(|err| TraceError::Other(err.into()))
286+
.and_then(|identity| identity)
279287
}
280288
}
281289

@@ -294,7 +302,7 @@ enum BatchMessage {
294302
/// Shut down the worker thread, push all spans in buffer to the backend.
295303
Shutdown(oneshot::Sender<ExportResult>),
296304
/// Set the resource for the exporter.
297-
SetResource(Arc<Resource>),
305+
SetResource(Arc<Resource>, oneshot::Sender<ExportResult>),
298306
}
299307

300308
struct BatchSpanProcessorInternal<R> {
@@ -396,8 +404,14 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
396404
return false;
397405
}
398406
// propagate the resource
399-
BatchMessage::SetResource(resource) => {
400-
self.exporter.set_resource(&resource);
407+
BatchMessage::SetResource(resource, res_sender) => {
408+
let result = self.exporter.set_resource(&resource);
409+
if let Err(result) = res_sender.send(result) {
410+
global::handle_error(TraceError::from(format!(
411+
"failed to send set resource result: {:?}",
412+
result
413+
)));
414+
}
401415
}
402416
}
403417
true

‎opentelemetry-stdout/src/logs/exporter.rs

+2-1
Original file line numberDiff line numberDiff line change
@@ -63,8 +63,9 @@ impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
6363
self.writer.take();
6464
}
6565

66-
fn set_resource(&mut self, res: &opentelemetry_sdk::Resource) {
66+
fn set_resource(&mut self, res: &opentelemetry_sdk::Resource) -> LogResult<()> {
6767
self.resource = res.clone();
68+
Ok(())
6869
}
6970
}
7071

‎opentelemetry-stdout/src/trace/exporter.rs

+2-1
Original file line numberDiff line numberDiff line change
@@ -56,8 +56,9 @@ impl opentelemetry_sdk::export::trace::SpanExporter for SpanExporter {
5656
self.writer.take();
5757
}
5858

59-
fn set_resource(&mut self, res: &opentelemetry_sdk::Resource) {
59+
fn set_resource(&mut self, res: &opentelemetry_sdk::Resource) -> ExportResult {
6060
self.resource = res.clone();
61+
Ok(())
6162
}
6263
}
6364

0 commit comments

Comments
 (0)