1use crate::carrier;
3use crate::convert::MaybeAsRef;
4use crate::log::{Log, LogBuilder, StdErrorLogFieldsBuilder};
5use crate::sampler::{AllSampler, Sampler};
6use crate::tag::{StdTag, Tag, TagValue};
7use crate::Result;
8use std::borrow::Cow;
9use std::fmt;
10use std::io::{Read, Write};
11use std::sync::Arc;
12use std::time::SystemTime;
13use tokio::sync::mpsc;
14
15pub trait SpanConsumer<T>: Send + Sync {
17 fn consume_span(&self, span: FinishedSpan<T>);
22}
23
24impl<T: Send> SpanConsumer<T> for mpsc::UnboundedSender<FinishedSpan<T>> {
25 fn consume_span(&self, span: FinishedSpan<T>) {
26 let _ = self.send(span);
27 }
28}
29
30impl<T: Send> SpanConsumer<T> for mpsc::Sender<FinishedSpan<T>> {
31 fn consume_span(&self, span: FinishedSpan<T>) {
32 let _ = self.try_send(span);
33 }
34}
35
36impl<T: Send> SpanConsumer<T> for std::sync::mpsc::Sender<FinishedSpan<T>> {
37 fn consume_span(&self, span: FinishedSpan<T>) {
38 let _ = self.send(span);
39 }
40}
41
42impl<T: Send> SpanConsumer<T> for std::sync::mpsc::SyncSender<FinishedSpan<T>> {
43 fn consume_span(&self, span: FinishedSpan<T>) {
44 let _ = self.try_send(span);
45 }
46}
47
48pub type SpanReceiver<T> = mpsc::UnboundedReceiver<FinishedSpan<T>>;
50#[deprecated = "SpanSender is an implementation detail of rustracing. It should not be public."]
52pub type SpanSender<T> = mpsc::UnboundedSender<FinishedSpan<T>>;
53
54pub(crate) struct SharedSpanConsumer<T>(Arc<dyn SpanConsumer<T>>);
56
57impl<T> SharedSpanConsumer<T> {
58 pub(crate) fn new(consumer: impl SpanConsumer<T> + 'static) -> Self {
59 Self(Arc::new(consumer))
60 }
61}
62
63impl<T> fmt::Debug for SharedSpanConsumer<T> {
64 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
65 f.write_str("SharedSpanConsumer")
66 }
67}
68
69impl<T> Clone for SharedSpanConsumer<T> {
70 fn clone(&self) -> Self {
71 Self(Arc::clone(&self.0))
72 }
73}
74
75pub struct FinishSpanCallback<T>(FinishCallbackInner<T>);
77type FinishCallbackInner<T> = Arc<dyn Fn(&mut Span<T>) + Send + Sync>;
78
79impl<T> fmt::Debug for FinishSpanCallback<T> {
80 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
81 f.write_str("FinishSpanCallback")
82 }
83}
84
85impl<T> Clone for FinishSpanCallback<T> {
86 fn clone(&self) -> Self {
87 Self(self.0.clone())
88 }
89}
90
91impl<T> From<FinishSpanCallback<T>> for FinishCallbackInner<T> {
92 fn from(v: FinishSpanCallback<T>) -> Self {
93 v.0
94 }
95}
96
97impl<T> From<FinishCallbackInner<T>> for FinishSpanCallback<T> {
98 fn from(v: FinishCallbackInner<T>) -> Self {
99 Self(v)
100 }
101}
102
103impl<T, F: Fn(&mut Span<T>) + Send + Sync + 'static> From<F> for FinishSpanCallback<T> {
104 fn from(v: F) -> Self {
105 Self(Arc::new(v))
106 }
107}
108
109pub trait RoutingMetadata: fmt::Debug + Send + Sync {
112 fn group_key(&self) -> String;
115
116 fn encode(&self) -> String;
118}
119
120#[derive(Debug)]
125pub struct Span<T>(Option<SpanInner<T>>);
126impl<T> Span<T> {
127 pub const fn inactive() -> Self {
140 Span(None)
141 }
142
143 pub fn handle(&self) -> SpanHandle<T>
145 where
146 T: Clone,
147 {
148 SpanHandle(
149 self.0
150 .as_ref()
151 .map(|inner| (inner.context.clone(), inner.span_tx.clone())),
152 )
153 }
154
155 pub fn is_sampled(&self) -> bool {
157 self.0.is_some()
158 }
159
160 pub fn context(&self) -> Option<&SpanContext<T>> {
162 self.0.as_ref().map(|x| &x.context)
163 }
164
165 pub fn set_operation_name<F, N>(&mut self, f: F)
167 where
168 F: FnOnce() -> N,
169 N: Into<Cow<'static, str>>,
170 {
171 if let Some(inner) = self.0.as_mut() {
172 inner.operation_name = f().into();
173 }
174 }
175
176 pub fn set_start_time<F>(&mut self, f: F)
178 where
179 F: FnOnce() -> SystemTime,
180 {
181 if let Some(inner) = self.0.as_mut() {
182 inner.start_time = f();
183 }
184 }
185
186 pub fn set_finish_time<F>(&mut self, f: F)
188 where
189 F: FnOnce() -> SystemTime,
190 {
191 if let Some(inner) = self.0.as_mut() {
192 inner.finish_time = Some(f());
193 }
194 }
195
196 pub fn set_finish_callback<C>(&mut self, cb: C)
198 where
199 C: Into<FinishSpanCallback<T>>,
200 {
201 if let Some(inner) = &mut self.0 {
202 inner.finish_cb = Some(cb.into());
203 }
204 }
205
206 #[doc(alias = "remove_finish_callback")]
211 pub fn take_finish_callback(&mut self) -> Option<FinishSpanCallback<T>> {
212 self.0.as_mut().and_then(|s| s.finish_cb.take())
213 }
214
215 pub fn set_tag<F>(&mut self, f: F)
217 where
218 F: FnOnce() -> Tag,
219 {
220 use std::iter::once;
221 self.set_tags(|| once(f()));
222 }
223
224 pub fn set_tags<F, I>(&mut self, f: F)
226 where
227 F: FnOnce() -> I,
228 I: IntoIterator<Item = Tag>,
229 {
230 if let Some(inner) = self.0.as_mut() {
231 for tag in f() {
232 inner.tags.retain(|x| x.name() != tag.name());
233 inner.tags.push(tag);
234 }
235 }
236 }
237
238 pub fn set_baggage_item<F>(&mut self, f: F)
240 where
241 F: FnOnce() -> BaggageItem,
242 {
243 if let Some(inner) = self.0.as_mut() {
244 let item = f();
245 inner.context.baggage_items.retain(|x| x.name != item.name);
246 inner.context.baggage_items.push(item);
247 }
248 }
249
250 pub fn get_baggage_item(&self, name: &str) -> Option<&BaggageItem> {
252 if let Some(inner) = self.0.as_ref() {
253 inner.context.baggage_items.iter().find(|x| x.name == name)
254 } else {
255 None
256 }
257 }
258
259 pub fn log<F>(&mut self, f: F)
261 where
262 F: FnOnce(&mut LogBuilder),
263 {
264 if let Some(inner) = self.0.as_mut() {
265 let mut builder = LogBuilder::new();
266 f(&mut builder);
267 if let Some(log) = builder.finish() {
268 inner.logs.push(log);
269 }
270 }
271 }
272
273 pub fn error_log<F>(&mut self, f: F)
278 where
279 F: FnOnce(&mut StdErrorLogFieldsBuilder),
280 {
281 if let Some(inner) = self.0.as_mut() {
282 let mut builder = LogBuilder::new();
283 f(&mut builder.error());
284 if let Some(log) = builder.finish() {
285 inner.logs.push(log);
286 }
287 if !inner.tags.iter().any(|x| x.name() == "error") {
288 inner.tags.push(StdTag::error());
289 }
290 }
291 }
292
293 pub fn child<N, F>(&self, operation_name: N, f: F) -> Span<T>
299 where
300 N: Into<Cow<'static, str>>,
301 T: Clone,
302 F: FnOnce(StartSpanOptions<AllSampler, T>) -> Span<T>,
303 {
304 self.handle().child(operation_name, move |mut opts| {
305 if let Some(finish_cb) = self.0.as_ref().and_then(|s| s.finish_cb.clone()) {
306 opts = opts.finish_callback(finish_cb);
307 }
308 if let Some(routing) = self.0.as_ref().and_then(|s| s.routing.clone()) {
309 opts = opts.routing(routing);
310 }
311 f(opts)
312 })
313 }
314
315 pub fn follower<N, F>(&self, operation_name: N, f: F) -> Span<T>
317 where
318 N: Into<Cow<'static, str>>,
319 T: Clone,
320 F: FnOnce(StartSpanOptions<AllSampler, T>) -> Span<T>,
321 {
322 self.handle().follower(operation_name, f)
323 }
324
325 pub(crate) fn new<S>(state: T, opts: StartSpanOptions<S, T>) -> Self {
326 let context = SpanContext::new(state, opts.baggage_items);
327 let inner = SpanInner {
328 operation_name: opts.operation_name,
329 start_time: opts.start_time.unwrap_or_else(SystemTime::now),
330 finish_time: None,
331 references: opts.references,
332 tags: opts.tags,
333 logs: Vec::new(),
334 context,
335 finish_cb: opts.finish_cb,
336 span_tx: opts.span_tx.clone(),
337 routing: opts.routing,
338 };
339 Span(Some(inner))
340 }
341}
342impl<T> Drop for Span<T> {
343 fn drop(&mut self) {
344 if let Some(finish_cb) = self.take_finish_callback() {
345 finish_cb.0(self);
346 }
347
348 if let Some(inner) = self.0.take() {
349 let finished = FinishedSpan {
350 operation_name: inner.operation_name,
351 start_time: inner.start_time,
352 finish_time: inner.finish_time.unwrap_or_else(SystemTime::now),
353 references: inner.references,
354 tags: inner.tags,
355 logs: inner.logs,
356 context: inner.context,
357 routing: inner.routing,
358 };
359 inner.span_tx.0.consume_span(finished);
360 }
361 }
362}
363impl<T> MaybeAsRef<SpanContext<T>> for Span<T> {
364 fn maybe_as_ref(&self) -> Option<&SpanContext<T>> {
365 self.context()
366 }
367}
368
369#[derive(Debug)]
370struct SpanInner<T> {
371 operation_name: Cow<'static, str>,
372 start_time: SystemTime,
373 finish_time: Option<SystemTime>,
374 references: Vec<SpanReference<T>>,
375 tags: Vec<Tag>,
376 logs: Vec<Log>,
377 context: SpanContext<T>,
378 finish_cb: Option<FinishSpanCallback<T>>,
379 span_tx: SharedSpanConsumer<T>,
380 routing: Option<Arc<dyn RoutingMetadata>>,
381}
382
383#[derive(Debug)]
385pub struct FinishedSpan<T> {
386 operation_name: Cow<'static, str>,
387 start_time: SystemTime,
388 finish_time: SystemTime,
389 references: Vec<SpanReference<T>>,
390 tags: Vec<Tag>,
391 logs: Vec<Log>,
392 context: SpanContext<T>,
393 routing: Option<Arc<dyn RoutingMetadata>>,
394}
395impl<T> FinishedSpan<T> {
396 pub fn operation_name(&self) -> &str {
398 self.operation_name.as_ref()
399 }
400
401 pub fn start_time(&self) -> SystemTime {
403 self.start_time
404 }
405
406 pub fn finish_time(&self) -> SystemTime {
408 self.finish_time
409 }
410
411 pub fn logs(&self) -> &[Log] {
413 &self.logs
414 }
415
416 pub fn tags(&self) -> &[Tag] {
418 &self.tags
419 }
420
421 pub fn references(&self) -> &[SpanReference<T>] {
423 &self.references
424 }
425
426 pub fn context(&self) -> &SpanContext<T> {
428 &self.context
429 }
430
431 pub fn routing(&self) -> Option<&dyn RoutingMetadata> {
433 self.routing.as_deref()
434 }
435}
436
437#[derive(Debug, Clone)]
444pub struct SpanContext<T> {
445 state: T,
446 baggage_items: Vec<BaggageItem>,
447}
448impl<T> SpanContext<T> {
449 pub fn new(state: T, mut baggage_items: Vec<BaggageItem>) -> Self {
451 baggage_items.reverse();
452 baggage_items.sort_by(|a, b| a.name().cmp(b.name()));
453 baggage_items.dedup_by(|a, b| a.name() == b.name());
454 SpanContext {
455 state,
456 baggage_items,
457 }
458 }
459
460 pub fn state(&self) -> &T {
462 &self.state
463 }
464
465 pub fn baggage_items(&self) -> &[BaggageItem] {
467 &self.baggage_items
468 }
469
470 pub fn inject_to_text_map<C>(&self, carrier: &mut C) -> Result<()>
472 where
473 C: carrier::TextMap,
474 T: carrier::InjectToTextMap<C>,
475 {
476 track!(T::inject_to_text_map(self, carrier))
477 }
478
479 pub fn inject_to_http_header<C>(&self, carrier: &mut C) -> Result<()>
481 where
482 C: carrier::SetHttpHeaderField,
483 T: carrier::InjectToHttpHeader<C>,
484 {
485 track!(T::inject_to_http_header(self, carrier))
486 }
487
488 pub fn inject_to_binary<C>(&self, carrier: &mut C) -> Result<()>
490 where
491 C: Write,
492 T: carrier::InjectToBinary<C>,
493 {
494 track!(T::inject_to_binary(self, carrier))
495 }
496
497 pub fn extract_from_text_map<C>(carrier: &C) -> Result<Option<Self>>
499 where
500 C: carrier::TextMap,
501 T: carrier::ExtractFromTextMap<C>,
502 {
503 track!(T::extract_from_text_map(carrier))
504 }
505
506 pub fn extract_from_http_header<'a, C>(carrier: &'a C) -> Result<Option<Self>>
508 where
509 C: carrier::IterHttpHeaderFields<'a>,
510 T: carrier::ExtractFromHttpHeader<'a, C>,
511 {
512 track!(T::extract_from_http_header(carrier))
513 }
514
515 pub fn extract_from_binary<C>(carrier: &mut C) -> Result<Option<Self>>
517 where
518 C: Read,
519 T: carrier::ExtractFromBinary<C>,
520 {
521 track!(T::extract_from_binary(carrier))
522 }
523}
524impl<T> MaybeAsRef<SpanContext<T>> for SpanContext<T> {
525 fn maybe_as_ref(&self) -> Option<&Self> {
526 Some(self)
527 }
528}
529
530#[derive(Debug, Clone)]
545pub struct BaggageItem {
546 name: String,
547 value: String,
548}
549impl BaggageItem {
550 pub fn new(name: &str, value: &str) -> Self {
552 BaggageItem {
553 name: name.to_owned(),
554 value: value.to_owned(),
555 }
556 }
557
558 pub fn name(&self) -> &str {
560 &self.name
561 }
562
563 pub fn value(&self) -> &str {
565 &self.value
566 }
567}
568
569#[derive(Debug, Clone)]
571#[allow(missing_docs)]
572pub enum SpanReference<T> {
573 ChildOf(T),
574 FollowsFrom(T),
575}
576impl<T> SpanReference<T> {
577 pub fn span(&self) -> &T {
579 match *self {
580 SpanReference::ChildOf(ref x) | SpanReference::FollowsFrom(ref x) => x,
581 }
582 }
583
584 pub fn is_child_of(&self) -> bool {
586 matches!(*self, SpanReference::ChildOf(_))
587 }
588
589 pub fn is_follows_from(&self) -> bool {
591 matches!(*self, SpanReference::FollowsFrom(_))
592 }
593}
594
595#[derive(Debug)]
597pub struct CandidateSpan<'a, T: 'a> {
598 tags: &'a [Tag],
599 references: &'a [SpanReference<T>],
600 baggage_items: &'a [BaggageItem],
601}
602impl<'a, T: 'a> CandidateSpan<'a, T> {
603 pub fn tags(&self) -> &[Tag] {
605 self.tags
606 }
607
608 pub fn references(&self) -> &[SpanReference<T>] {
610 self.references
611 }
612
613 pub fn baggage_items(&self) -> &[BaggageItem] {
615 self.baggage_items
616 }
617}
618
619#[derive(Debug)]
621pub struct StartSpanOptions<'a, S: 'a, T: 'a> {
622 operation_name: Cow<'static, str>,
623 start_time: Option<SystemTime>,
624 tags: Vec<Tag>,
625 references: Vec<SpanReference<T>>,
626 baggage_items: Vec<BaggageItem>,
627 finish_cb: Option<FinishSpanCallback<T>>,
628 routing: Option<Arc<dyn RoutingMetadata>>,
629 span_tx: &'a SharedSpanConsumer<T>,
630 sampler: &'a S,
631}
632impl<'a, S: 'a, T: 'a> StartSpanOptions<'a, S, T>
633where
634 S: Sampler<T>,
635{
636 pub fn start_time(mut self, time: SystemTime) -> Self {
638 self.start_time = Some(time);
639 self
640 }
641
642 pub fn tag(mut self, tag: Tag) -> Self {
644 self.tags.push(tag);
645 self
646 }
647
648 pub fn finish_callback<C>(mut self, cb: C) -> Self
650 where
651 C: Into<FinishSpanCallback<T>>,
652 {
653 self.finish_cb = Some(cb.into());
654 self
655 }
656
657 pub fn routing(mut self, routing: Arc<dyn RoutingMetadata>) -> Self {
664 self.routing = Some(routing);
665 self
666 }
667
668 pub fn child_of<C>(mut self, context: &C) -> Self
670 where
671 C: MaybeAsRef<SpanContext<T>>,
672 T: Clone,
673 {
674 if let Some(context) = context.maybe_as_ref() {
675 let reference = SpanReference::ChildOf(context.state().clone());
676 self.references.push(reference);
677 self.baggage_items
678 .extend(context.baggage_items().iter().cloned());
679 }
680 self
681 }
682
683 pub fn follows_from<C>(mut self, context: &C) -> Self
685 where
686 C: MaybeAsRef<SpanContext<T>>,
687 T: Clone,
688 {
689 if let Some(context) = context.maybe_as_ref() {
690 let reference = SpanReference::FollowsFrom(context.state().clone());
691 self.references.push(reference);
692 self.baggage_items
693 .extend(context.baggage_items().iter().cloned());
694 }
695 self
696 }
697
698 pub fn start(mut self) -> Span<T>
700 where
701 T: for<'b> From<CandidateSpan<'b, T>>,
702 {
703 self.normalize();
704 if !self.is_sampled() {
705 return Span(None);
706 }
707 let state = T::from(self.span());
708 Span::new(state, self)
709 }
710
711 pub fn start_with_state(mut self, state: T) -> Span<T> {
713 self.normalize();
714 if !self.is_sampled() {
715 return Span(None);
716 }
717 Span::new(state, self)
718 }
719
720 pub(crate) fn new<N>(
721 operation_name: N,
722 span_tx: &'a SharedSpanConsumer<T>,
723 sampler: &'a S,
724 ) -> Self
725 where
726 N: Into<Cow<'static, str>>,
727 {
728 StartSpanOptions {
729 operation_name: operation_name.into(),
730 start_time: None,
731 tags: Vec::new(),
732 references: Vec::new(),
733 baggage_items: Vec::new(),
734 finish_cb: None,
735 routing: None,
736 span_tx,
737 sampler,
738 }
739 }
740
741 fn normalize(&mut self) {
742 self.tags.reverse();
743 self.tags.sort_by(|a, b| a.name().cmp(b.name()));
744 self.tags.dedup_by(|a, b| a.name() == b.name());
745
746 self.baggage_items.reverse();
747 self.baggage_items.sort_by(|a, b| a.name().cmp(b.name()));
748 self.baggage_items.dedup_by(|a, b| a.name() == b.name());
749 }
750
751 fn span(&self) -> CandidateSpan<'_, T> {
752 CandidateSpan {
753 references: &self.references,
754 tags: &self.tags,
755 baggage_items: &self.baggage_items,
756 }
757 }
758
759 fn is_sampled(&self) -> bool {
760 if let Some(&TagValue::Integer(n)) = self
761 .tags
762 .iter()
763 .find(|t| t.name() == "sampling.priority")
764 .map(|t| t.value())
765 {
766 n > 0
767 } else {
768 self.sampler.is_sampled(&self.span())
769 }
770 }
771}
772
773#[derive(Debug, Clone)]
775pub struct SpanHandle<T>(Option<(SpanContext<T>, SharedSpanConsumer<T>)>);
776impl<T> SpanHandle<T> {
777 pub fn is_sampled(&self) -> bool {
779 self.0.is_some()
780 }
781
782 pub fn context(&self) -> Option<&SpanContext<T>> {
784 self.0.as_ref().map(|(context, _)| context)
785 }
786
787 pub fn get_baggage_item(&self, name: &str) -> Option<&BaggageItem> {
789 if let Some(context) = self.context() {
790 context.baggage_items.iter().find(|x| x.name == name)
791 } else {
792 None
793 }
794 }
795
796 pub fn child<N, F>(&self, operation_name: N, f: F) -> Span<T>
798 where
799 N: Into<Cow<'static, str>>,
800 T: Clone,
801 F: FnOnce(StartSpanOptions<AllSampler, T>) -> Span<T>,
802 {
803 if let Some((context, span_tx)) = self.0.as_ref() {
804 let options =
805 StartSpanOptions::new(operation_name, span_tx, &AllSampler).child_of(context);
806 f(options)
807 } else {
808 Span::inactive()
809 }
810 }
811
812 pub fn follower<N, F>(&self, operation_name: N, f: F) -> Span<T>
814 where
815 N: Into<Cow<'static, str>>,
816 T: Clone,
817 F: FnOnce(StartSpanOptions<AllSampler, T>) -> Span<T>,
818 {
819 if let Some((context, span_tx)) = self.0.as_ref() {
820 let options =
821 StartSpanOptions::new(operation_name, span_tx, &AllSampler).follows_from(context);
822 f(options)
823 } else {
824 Span::inactive()
825 }
826 }
827}
828
829pub trait InspectableSpan<T> {
831 fn operation_name(&self) -> &str;
833
834 fn start_time(&self) -> SystemTime;
836
837 fn finish_time(&self) -> Option<SystemTime>;
839
840 fn logs(&self) -> &[Log];
842
843 fn tags(&self) -> &[Tag];
845
846 fn references(&self) -> &[SpanReference<T>];
848}
849
850impl<T> InspectableSpan<T> for Span<T> {
851 fn operation_name(&self) -> &str {
853 self.0
854 .as_ref()
855 .map(|inner| inner.operation_name.as_ref())
856 .unwrap_or("")
857 }
858
859 fn start_time(&self) -> SystemTime {
861 self.0
862 .as_ref()
863 .map(|inner| inner.start_time)
864 .unwrap_or_else(SystemTime::now)
865 }
866
867 fn finish_time(&self) -> Option<SystemTime> {
869 self.0.as_ref().and_then(|inner| inner.finish_time)
870 }
871
872 fn logs(&self) -> &[Log] {
874 self.0
875 .as_ref()
876 .map(|inner| inner.logs.as_ref())
877 .unwrap_or(&[])
878 }
879
880 fn tags(&self) -> &[Tag] {
882 self.0
883 .as_ref()
884 .map(|inner| inner.tags.as_ref())
885 .unwrap_or(&[])
886 }
887
888 fn references(&self) -> &[SpanReference<T>] {
890 self.0
891 .as_ref()
892 .map(|inner| inner.references.as_ref())
893 .unwrap_or(&[])
894 }
895}