-
Notifications
You must be signed in to change notification settings - Fork 454
/
mod.rs
389 lines (348 loc) · 14.2 KB
/
mod.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
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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
//! Representations of entities producing telemetry.
//!
//! A [Resource] is an immutable representation of the entity producing
//! telemetry as attributes. For example, a process producing telemetry that is
//! running in a container on Kubernetes has a Pod name, it is in a namespace
//! and possibly is part of a Deployment which also has a name. All three of
//! these attributes can be included in the `Resource`. Note that there are
//! certain ["standard attributes"] that have prescribed meanings.
//!
//! ["standard attributes"]: https://github.com/open-telemetry/opentelemetry-specification/blob/v1.9.0/specification/resource/semantic_conventions/README.md
//!
//! # Resource detectors
//!
//! [`ResourceDetector`]s are used to detect resource from runtime or
//! environmental variables. The following are provided by default with this
//! SDK.
//!
//! - [`EnvResourceDetector`] - detect resource from environmental variables.
//! - [`TelemetryResourceDetector`] - detect telemetry SDK's information.
//!
//! The OS and Process resource detectors are packaged separately in the
//! [`opentelemetry-resource-detector` crate](https://github.com/open-telemetry/opentelemetry-rust-contrib/tree/main/opentelemetry-resource-detectors).
mod env;
mod telemetry;
mod attributes;
pub(crate) use attributes::*;
pub use env::EnvResourceDetector;
pub use env::SdkProvidedResourceDetector;
pub use telemetry::TelemetryResourceDetector;
use opentelemetry::{Key, KeyValue, Value};
use std::borrow::Cow;
use std::collections::{hash_map, HashMap};
use std::ops::Deref;
use std::sync::Arc;
/// Inner structure of `Resource` holding the actual data.
/// This structure is designed to be shared among `Resource` instances via `Arc`.
#[derive(Debug, Clone, PartialEq)]
struct ResourceInner {
attrs: HashMap<Key, Value>,
schema_url: Option<Cow<'static, str>>,
}
/// An immutable representation of the entity producing telemetry as attributes.
/// Utilizes `Arc` for efficient sharing and cloning.
#[derive(Clone, Debug, PartialEq)]
pub struct Resource {
inner: Arc<ResourceInner>,
}
impl Default for Resource {
fn default() -> Self {
Self::from_detectors(vec![
Box::new(SdkProvidedResourceDetector),
Box::new(TelemetryResourceDetector),
Box::new(EnvResourceDetector::new()),
])
}
}
impl Resource {
/// Creates an empty resource.
/// This is the basic constructor that initializes a resource with no attributes and no schema URL.
pub fn empty() -> Self {
Resource {
inner: Arc::new(ResourceInner {
attrs: HashMap::new(),
schema_url: None,
}),
}
}
/// Create a new `Resource` from key value pairs.
///
/// Values are de-duplicated by key, and the first key-value pair with a non-empty string value
/// will be retained
pub fn new<T: IntoIterator<Item = KeyValue>>(kvs: T) -> Self {
let mut attrs = HashMap::new();
for kv in kvs {
attrs.insert(kv.key, kv.value);
}
Resource {
inner: Arc::new(ResourceInner {
attrs,
schema_url: None,
}),
}
}
/// Create a new `Resource::default()` and merge with provided key value pairs.
///
/// Values are de-duplicated by key, and the first key-value pair with a non-empty string value
/// will be retained
pub fn new_with_defaults<T: IntoIterator<Item = KeyValue>>(keys: T) -> Self {
Resource::default().merge(&mut Resource::new(keys))
}
/// Create a new `Resource` from a key value pairs and [schema url].
///
/// Values are de-duplicated by key, and the first key-value pair with a non-empty string value
/// will be retained.
///
/// schema_url must be a valid URL using HTTP or HTTPS protocol.
///
/// [schema url]: https://github.com/open-telemetry/opentelemetry-specification/blob/v1.9.0/specification/schemas/overview.md#schema-url
pub fn from_schema_url<KV, S>(kvs: KV, schema_url: S) -> Self
where
KV: IntoIterator<Item = KeyValue>,
S: Into<Cow<'static, str>>,
{
let schema_url_str = schema_url.into();
let normalized_schema_url = if schema_url_str.is_empty() {
None
} else {
Some(schema_url_str)
};
let mut attrs = HashMap::new();
for kv in kvs {
attrs.insert(kv.key, kv.value);
}
Resource {
inner: Arc::new(ResourceInner {
attrs,
schema_url: normalized_schema_url,
}),
}
}
/// Create a new `Resource` from resource detectors.
///
/// timeout will be applied to each detector.
pub fn from_detectors(detectors: Vec<Box<dyn ResourceDetector>>) -> Self {
let mut resource = Resource::empty();
for detector in detectors {
let detected_res = detector.detect();
// This call ensures that if the Arc is not uniquely owned,
// the data is cloned before modification, preserving safety.
// If the Arc is uniquely owned, it simply returns a mutable reference to the data.
let inner = Arc::make_mut(&mut resource.inner);
for (key, value) in detected_res.into_iter() {
inner.attrs.insert(Key::new(key.clone()), value.clone());
}
}
resource
}
/// Create a new `Resource` by combining two resources.
///
/// ### Key value pairs
/// Keys from the `other` resource have priority over keys from this resource, even if the
/// updated value is empty.
///
/// ### [Schema url]
/// If both of the resource are not empty. Schema url is determined by the following rules, in order:
/// 1. If this resource has a schema url, it will be used.
/// 2. If this resource does not have a schema url, and the other resource has a schema url, it will be used.
/// 3. If both resources have a schema url and it's the same, it will be used.
/// 4. If both resources have a schema url and it's different, the schema url will be empty.
/// 5. If both resources do not have a schema url, the schema url will be empty.
///
/// [Schema url]: https://github.com/open-telemetry/opentelemetry-specification/blob/v1.9.0/specification/schemas/overview.md#schema-url
pub fn merge<T: Deref<Target = Self>>(&self, other: T) -> Self {
if self.is_empty() {
return other.clone();
}
if other.is_empty() {
return self.clone();
}
let mut combined_attrs = self.inner.attrs.clone();
for (k, v) in other.inner.attrs.iter() {
combined_attrs.insert(k.clone(), v.clone());
}
// Resolve the schema URL according to the precedence rules
let combined_schema_url = match (&self.inner.schema_url, &other.inner.schema_url) {
// If both resources have a schema URL and it's the same, use it
(Some(url1), Some(url2)) if url1 == url2 => Some(url1.clone()),
// If both resources have a schema URL but they are not the same, the schema URL will be empty
(Some(_), Some(_)) => None,
// If this resource does not have a schema URL, and the other resource has a schema URL, it will be used
(None, Some(url)) => Some(url.clone()),
// If this resource has a schema URL, it will be used (covers case 1 and any other cases where `self` has a schema URL)
(Some(url), _) => Some(url.clone()),
// If both resources do not have a schema URL, the schema URL will be empty
(None, None) => None,
};
Resource {
inner: Arc::new(ResourceInner {
attrs: combined_attrs,
schema_url: combined_schema_url,
}),
}
}
/// Return the [schema url] of the resource. If the resource does not have a schema url, return `None`.
///
/// [schema url]: https://github.com/open-telemetry/opentelemetry-specification/blob/v1.9.0/specification/schemas/overview.md#schema-url
pub fn schema_url(&self) -> Option<&str> {
self.inner.schema_url.as_ref().map(|s| s.as_ref())
}
/// Returns the number of attributes for this resource
pub fn len(&self) -> usize {
self.inner.attrs.len()
}
/// Returns `true` if the resource contains no attributes.
pub fn is_empty(&self) -> bool {
self.inner.attrs.is_empty()
}
/// Gets an iterator over the attributes of this resource.
pub fn iter(&self) -> Iter<'_> {
Iter(self.inner.attrs.iter())
}
/// Retrieve the value from resource associate with given key.
pub fn get(&self, key: Key) -> Option<Value> {
self.inner.attrs.get(&key).cloned()
}
}
/// An iterator over the entries of a `Resource`.
#[derive(Debug)]
pub struct Iter<'a>(hash_map::Iter<'a, Key, Value>);
impl<'a> Iterator for Iter<'a> {
type Item = (&'a Key, &'a Value);
fn next(&mut self) -> Option<Self::Item> {
self.0.next()
}
}
impl<'a> IntoIterator for &'a Resource {
type Item = (&'a Key, &'a Value);
type IntoIter = Iter<'a>;
fn into_iter(self) -> Self::IntoIter {
Iter(self.inner.attrs.iter())
}
}
/// ResourceDetector detects OpenTelemetry resource information
///
/// Implementations of this trait can be passed to
/// the [`Resource::from_detectors`] function to generate a Resource from the merged information.
pub trait ResourceDetector {
/// detect returns an initialized Resource based on gathered information.
///
/// timeout is used in case the detection operation takes too much time.
///
/// If source information to construct a Resource is inaccessible, an empty Resource should be returned
///
/// If source information to construct a Resource is invalid, for example,
/// missing required values. an empty Resource should be returned.
fn detect(&self) -> Resource;
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use super::*;
#[test]
fn new_resource() {
let args_with_dupe_keys = vec![KeyValue::new("a", ""), KeyValue::new("a", "final")];
let mut expected_attrs = HashMap::new();
expected_attrs.insert(Key::new("a"), Value::from("final"));
let resource = Resource::new(args_with_dupe_keys);
let resource_inner = Arc::try_unwrap(resource.inner).expect("Failed to unwrap Arc");
assert_eq!(resource_inner.attrs, expected_attrs);
assert_eq!(resource_inner.schema_url, None);
}
#[test]
fn merge_resource_key_value_pairs() {
let resource_a = Resource::new(vec![
KeyValue::new("a", ""),
KeyValue::new("b", "b-value"),
KeyValue::new("d", "d-value"),
]);
let resource_b = Resource::new(vec![
KeyValue::new("a", "a-value"),
KeyValue::new("c", "c-value"),
KeyValue::new("d", ""),
]);
let mut expected_attrs = HashMap::new();
expected_attrs.insert(Key::new("a"), Value::from("a-value"));
expected_attrs.insert(Key::new("b"), Value::from("b-value"));
expected_attrs.insert(Key::new("c"), Value::from("c-value"));
expected_attrs.insert(Key::new("d"), Value::from(""));
let expected_resource = Resource {
inner: Arc::new(ResourceInner {
attrs: expected_attrs,
schema_url: None, // Assuming schema_url handling if needed
}),
};
assert_eq!(resource_a.merge(&resource_b), expected_resource);
}
#[rstest]
#[case(Some("http://schema/a"), None, Some("http://schema/a"))]
#[case(Some("http://schema/a"), Some("http://schema/b"), None)]
#[case(None, Some("http://schema/b"), Some("http://schema/b"))]
#[case(
Some("http://schema/a"),
Some("http://schema/a"),
Some("http://schema/a")
)]
#[case(None, None, None)]
fn merge_resource_schema_url(
#[case] schema_url_a: Option<&'static str>,
#[case] schema_url_b: Option<&'static str>,
#[case] expected_schema_url: Option<&'static str>,
) {
let resource_a =
Resource::from_schema_url(vec![KeyValue::new("key", "")], schema_url_a.unwrap_or(""));
let resource_b =
Resource::from_schema_url(vec![KeyValue::new("key", "")], schema_url_b.unwrap_or(""));
let merged_resource = resource_a.merge(&resource_b);
let result_schema_url = merged_resource.schema_url();
assert_eq!(
result_schema_url.map(|s| s as &str),
expected_schema_url,
"Merging schema_url_a {:?} with schema_url_b {:?} did not yield expected result {:?}",
schema_url_a,
schema_url_b,
expected_schema_url
);
}
#[rstest]
#[case(vec![], vec![KeyValue::new("key", "b")], "http://schema/a", None)]
#[case(vec![KeyValue::new("key", "a")], vec![KeyValue::new("key", "b")], "http://schema/a", Some("http://schema/a"))]
fn merge_resource_with_missing_attribtes(
#[case] key_values_a: Vec<KeyValue>,
#[case] key_values_b: Vec<KeyValue>,
#[case] schema_url: &'static str,
#[case] expected_schema_url: Option<&'static str>,
) {
let resource = Resource::from_schema_url(key_values_a, schema_url);
let other_resource = Resource::new(key_values_b);
assert_eq!(
resource.merge(&other_resource).schema_url(),
expected_schema_url
);
}
#[test]
fn detect_resource() {
temp_env::with_vars(
[
(
"OTEL_RESOURCE_ATTRIBUTES",
Some("key=value, k = v , a= x, a=z"),
),
("IRRELEVANT", Some("20200810")),
],
|| {
let detector = EnvResourceDetector::new();
let resource = Resource::from_detectors(vec![Box::new(detector)]);
assert_eq!(
resource,
Resource::new(vec![
KeyValue::new("key", "value"),
KeyValue::new("k", "v"),
KeyValue::new("a", "x"),
KeyValue::new("a", "z"),
])
)
},
)
}
}