flute 1.11.0

File Delivery over Unidirectional Transport (FLUTE)
Documentation 
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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

use std::collections::HashMap;

use opentelemetry::{
    global::{self, BoxedSpan},
    propagation::{Extractor, Injector},
    trace::{Span, SpanKind, Status, TraceContextExt, Tracer},
    Context, KeyValue,
};

use crate::common::udpendpoint::UDPEndpoint;

pub struct ObjectReceiverLogger {
    cx: Context,
}

impl std::fmt::Debug for ObjectReceiverLogger {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ObjectReceiverLogger").finish()
    }
}

struct HeaderExtractor<'a>(pub &'a HashMap<String, String>);
impl Extractor for HeaderExtractor<'_> {
    fn get(&self, key: &str) -> Option<&str> {
        self.0.get(key).map(|s| s.as_str())
    }

    fn keys(&self) -> Vec<&str> {
        self.0.keys().map(|s| s.as_str()).collect()
    }
}

struct HeaderInjector<'a>(pub &'a mut HashMap<String, String>);
impl Injector for HeaderInjector<'_> {
    fn set(&mut self, key: &str, value: String) {
        self.0.insert(key.to_string(), value);
    }
}

impl ObjectReceiverLogger {
    fn extract_context_from_propagator(req: &HashMap<String, String>) -> Context {
        global::get_text_map_propagator(|propagator| propagator.extract(&HeaderExtractor(req)))
    }

    pub fn new(
        endpoint: &UDPEndpoint,
        tsi: u64,
        toi: u128,
        propagator: Option<&HashMap<String, String>>,
    ) -> Self {
        let tracer = global::tracer("FluteLogger");
        let name = match toi {
            0 => "FDT",
            _ => "FLUTEObject",
        };

        let mut span;
        if let Some(propagator) = propagator {
            let parent_cx = Self::extract_context_from_propagator(propagator);
            span = tracer
                .span_builder(name)
                .with_kind(SpanKind::Consumer)
                .start_with_context(&tracer, &parent_cx)
        } else {
            span = tracer
                .span_builder(name)
                .with_kind(SpanKind::Consumer)
                .start(&tracer);
        }

        span.set_attribute(KeyValue::new(
            opentelemetry_semantic_conventions::attribute::NETWORK_TRANSPORT,
            "flute",
        ));

        span.set_attribute(KeyValue::new(
            opentelemetry_semantic_conventions::attribute::NETWORK_PEER_ADDRESS,
            endpoint.destination_group_address.clone(),
        ));

        span.set_attribute(KeyValue::new(
            opentelemetry_semantic_conventions::attribute::NETWORK_PEER_PORT,
            endpoint.port as i64,
        ));

        if let Some(source_address) = endpoint.source_address.as_ref() {
            span.set_attribute(KeyValue::new(
                opentelemetry_semantic_conventions::attribute::NETWORK_LOCAL_ADDRESS,
                source_address.clone(),
            ));
        }

        span.set_attribute(KeyValue::new("flute.toi", toi.to_string()));
        span.set_attribute(KeyValue::new("flute.tsi", tsi.to_string()));

        let cx = Context::default().with_span(span);
        Self { cx }
    }

    pub fn get_propagator(&self) -> HashMap<String, String> {
        let propagator = global::get_text_map_propagator(|propagator| {
            let mut headers = HashMap::new();
            propagator.inject_context(&self.cx, &mut HeaderInjector(&mut headers));
            headers
        });

        propagator
    }
    /*
    pub fn block_span(&mut self) -> BoxedSpan {
        let tracer = global::tracer("FluteLogger");
        tracer.start_with_context("block", &self.cx)
    }
    */

    pub fn fdt_attached(&mut self) -> BoxedSpan {
        let tracer = global::tracer("FluteLogger");
        tracer.start_with_context("fdt_attached", &self.cx)
    }

    pub fn complete(&mut self) -> BoxedSpan {
        let tracer = global::tracer("FluteLogger");

        let span = self.cx.span();
        span.set_status(Status::Ok);

        tracer.start_with_context("complete", &self.cx)
    }

    pub fn interrupted(&mut self, description: &str) -> BoxedSpan {
        let tracer = global::tracer("FluteLogger");

        let span = self.cx.span();
        span.set_status(Status::Ok);

        span.set_attribute(KeyValue::new("description", description.to_string()));
        tracer.start_with_context("interrupted", &self.cx)
    }

    pub fn error(&mut self, description: &str) -> BoxedSpan {
        let tracer = global::tracer("FluteLogger");

        let span = self.cx.span();
        span.set_status(Status::Error {
            description: std::borrow::Cow::Owned(description.to_string()),
        });

        span.set_attribute(KeyValue::new("error_description", description.to_string()));
        tracer.start_with_context("error", &self.cx)
    }
}