-
Notifications
You must be signed in to change notification settings - Fork 385
/
uploader.rs
80 lines (74 loc) · 2.48 KB
/
uploader.rs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
//! # Jaeger Span Uploader
#[cfg(any(feature = "collector_client", feature = "wasm_collector_client"))]
use crate::exporter::collector;
use crate::exporter::{agent, jaeger};
use async_trait::async_trait;
use opentelemetry::sdk::export::trace;
use opentelemetry::sdk::export::trace::ExportResult;
use std::fmt::Debug;
use crate::exporter::thrift::jaeger::Batch;
use crate::exporter::JaegerTraceRuntime;
#[async_trait]
pub(crate) trait Uploader: Debug + Send + Sync {
async fn upload(&self, batch: jaeger::Batch) -> trace::ExportResult;
}
#[derive(Debug)]
pub(crate) enum SyncUploader {
Agent(std::sync::Mutex<agent::AgentSyncClientUdp>),
}
#[async_trait]
impl Uploader for SyncUploader {
async fn upload(&self, batch: jaeger::Batch) -> trace::ExportResult {
match self {
SyncUploader::Agent(client) => {
// TODO Implement retry behaviour
client
.lock()
.expect("Failed to lock agent client")
.emit_batch(batch)
.map_err::<crate::Error, _>(Into::into)?;
}
}
Ok(())
}
}
/// Uploads a batch of spans to Jaeger
#[derive(Debug)]
pub(crate) enum AsyncUploader<R: JaegerTraceRuntime> {
/// Agent async client
Agent(futures::lock::Mutex<agent::AgentAsyncClientUdp<R>>),
/// Collector sync client
#[cfg(feature = "collector_client")]
Collector(collector::AsyncHttpClient),
#[cfg(feature = "wasm_collector_client")]
WasmCollector(collector::WasmCollector),
}
#[async_trait]
impl<R: JaegerTraceRuntime> Uploader for AsyncUploader<R> {
async fn upload(&self, batch: Batch) -> ExportResult {
match self {
Self::Agent(client) => {
// TODO Implement retry behaviour
client
.lock()
.await
.emit_batch(batch)
.await
.map_err::<crate::Error, _>(Into::into)?;
}
#[cfg(feature = "collector_client")]
Self::Collector(collector) => {
// TODO Implement retry behaviour
collector.submit_batch(batch).await?;
}
#[cfg(feature = "wasm_collector_client")]
Self::WasmCollector(collector) => {
collector
.submit_batch(batch)
.await
.map_err::<crate::Error, _>(Into::into)?;
}
}
Ok(())
}
}