llm_optimizer_processor/telemetry/
resource.rs

1//! Resource attributes and detection for OpenTelemetry.
2//!
3//! This module provides utilities for creating and managing resource attributes
4//! that describe the service and environment.
5
6use opentelemetry::KeyValue;
7use opentelemetry_sdk::resource::{
8    EnvResourceDetector, Resource,
9    SdkProvidedResourceDetector,
10};
11use opentelemetry_semantic_conventions::resource::{
12    DEPLOYMENT_ENVIRONMENT, HOST_ARCH, HOST_NAME, OS_TYPE, PROCESS_COMMAND_ARGS,
13    PROCESS_EXECUTABLE_NAME, PROCESS_EXECUTABLE_PATH, PROCESS_OWNER, PROCESS_PID,
14    PROCESS_RUNTIME_DESCRIPTION, PROCESS_RUNTIME_NAME, PROCESS_RUNTIME_VERSION,
15    SERVICE_INSTANCE_ID, SERVICE_NAME, SERVICE_VERSION,
16};
17use std::collections::HashMap;
18use uuid::Uuid;
19
20/// Resource builder for creating OpenTelemetry resources with semantic conventions.
21#[derive(Debug, Clone)]
22pub struct ResourceBuilder {
23    service_name: Option<String>,
24    service_version: Option<String>,
25    service_instance_id: Option<String>,
26    deployment_environment: Option<String>,
27    additional_attributes: HashMap<String, String>,
28}
29
30impl ResourceBuilder {
31    /// Creates a new resource builder.
32    pub fn new() -> Self {
33        Self {
34            service_name: None,
35            service_version: None,
36            service_instance_id: None,
37            deployment_environment: None,
38            additional_attributes: HashMap::new(),
39        }
40    }
41
42    /// Sets the service name.
43    pub fn with_service_name(mut self, name: impl Into<String>) -> Self {
44        self.service_name = Some(name.into());
45        self
46    }
47
48    /// Sets the service version.
49    pub fn with_service_version(mut self, version: impl Into<String>) -> Self {
50        self.service_version = Some(version.into());
51        self
52    }
53
54    /// Sets the service instance ID.
55    pub fn with_service_instance_id(mut self, instance_id: impl Into<String>) -> Self {
56        self.service_instance_id = Some(instance_id.into());
57        self
58    }
59
60    /// Sets the deployment environment (e.g., "production", "staging", "development").
61    pub fn with_deployment_environment(mut self, env: impl Into<String>) -> Self {
62        self.deployment_environment = Some(env.into());
63        self
64    }
65
66    /// Adds a custom attribute.
67    pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
68        self.additional_attributes.insert(key.into(), value.into());
69        self
70    }
71
72    /// Adds multiple custom attributes.
73    pub fn with_attributes(mut self, attributes: HashMap<String, String>) -> Self {
74        self.additional_attributes.extend(attributes);
75        self
76    }
77
78    /// Builds the resource with automatic detection of system resources.
79    pub fn build(self) -> Resource {
80        let mut key_values = Vec::new();
81
82        // Service attributes
83        if let Some(name) = self.service_name {
84            key_values.push(KeyValue::new(SERVICE_NAME, name));
85        }
86
87        if let Some(version) = self.service_version {
88            key_values.push(KeyValue::new(SERVICE_VERSION, version));
89        }
90
91        // Generate instance ID if not provided
92        let instance_id = self
93            .service_instance_id
94            .unwrap_or_else(|| Uuid::new_v4().to_string());
95        key_values.push(KeyValue::new(SERVICE_INSTANCE_ID, instance_id));
96
97        if let Some(env) = self.deployment_environment {
98            key_values.push(KeyValue::new(DEPLOYMENT_ENVIRONMENT, env));
99        }
100
101        // Add custom attributes
102        for (key, value) in self.additional_attributes {
103            key_values.push(KeyValue::new(key, value));
104        }
105
106        // Create base resource
107        let base_resource = Resource::new(key_values);
108
109        // Merge with detected resources (Note: ResourceDetector API changed in 0.24)
110        // For now, just return base resource without auto-detection
111        // TODO: Update to use new opentelemetry_sdk 0.24 resource detection API
112        base_resource
113    }
114
115    /// Builds the resource without automatic detection (manual attributes only).
116    pub fn build_without_detection(self) -> Resource {
117        let mut key_values = Vec::new();
118
119        if let Some(name) = self.service_name {
120            key_values.push(KeyValue::new(SERVICE_NAME, name));
121        }
122
123        if let Some(version) = self.service_version {
124            key_values.push(KeyValue::new(SERVICE_VERSION, version));
125        }
126
127        let instance_id = self
128            .service_instance_id
129            .unwrap_or_else(|| Uuid::new_v4().to_string());
130        key_values.push(KeyValue::new(SERVICE_INSTANCE_ID, instance_id));
131
132        if let Some(env) = self.deployment_environment {
133            key_values.push(KeyValue::new(DEPLOYMENT_ENVIRONMENT, env));
134        }
135
136        for (key, value) in self.additional_attributes {
137            key_values.push(KeyValue::new(key, value));
138        }
139
140        Resource::new(key_values)
141    }
142}
143
144impl Default for ResourceBuilder {
145    fn default() -> Self {
146        Self::new()
147    }
148}
149
150/// Creates a default resource for the stream processor service.
151pub fn create_processor_resource(
152    service_name: Option<String>,
153    service_version: Option<String>,
154    environment: Option<String>,
155) -> Resource {
156    let mut builder = ResourceBuilder::new();
157
158    if let Some(name) = service_name {
159        builder = builder.with_service_name(name);
160    } else {
161        builder = builder.with_service_name("llm-optimizer-processor");
162    }
163
164    if let Some(version) = service_version {
165        builder = builder.with_service_version(version);
166    } else {
167        builder = builder.with_service_version(env!("CARGO_PKG_VERSION"));
168    }
169
170    if let Some(env) = environment {
171        builder = builder.with_deployment_environment(env);
172    }
173
174    // Add processor-specific attributes
175    builder = builder.with_attribute("service.component", "stream-processor");
176
177    builder.build()
178}
179
180#[cfg(test)]
181mod tests {
182    use super::*;
183
184    #[test]
185    fn test_resource_builder_basic() {
186        let resource = ResourceBuilder::new()
187            .with_service_name("test-service")
188            .with_service_version("1.0.0")
189            .with_deployment_environment("test")
190            .build_without_detection();
191
192        // Verify service name is set
193        let attributes: Vec<_> = resource.iter().collect();
194        assert!(attributes
195            .iter()
196            .any(|(k, v)| k.as_str() == SERVICE_NAME && v.as_str() == "test-service"));
197        assert!(attributes
198            .iter()
199            .any(|(k, v)| k.as_str() == SERVICE_VERSION && v.as_str() == "1.0.0"));
200        assert!(attributes
201            .iter()
202            .any(|(k, v)| k.as_str() == DEPLOYMENT_ENVIRONMENT && v.as_str() == "test"));
203    }
204
205    #[test]
206    fn test_resource_builder_custom_attributes() {
207        let resource = ResourceBuilder::new()
208            .with_service_name("test")
209            .with_attribute("custom.key", "custom.value")
210            .build_without_detection();
211
212        let attributes: Vec<_> = resource.iter().collect();
213        assert!(attributes
214            .iter()
215            .any(|(k, v)| k.as_str() == "custom.key" && v.as_str() == "custom.value"));
216    }
217
218    #[test]
219    fn test_resource_builder_auto_instance_id() {
220        let resource = ResourceBuilder::new()
221            .with_service_name("test")
222            .build_without_detection();
223
224        let attributes: Vec<_> = resource.iter().collect();
225        // Instance ID should be auto-generated
226        assert!(attributes
227            .iter()
228            .any(|(k, _v)| k.as_str() == SERVICE_INSTANCE_ID));
229    }
230
231    #[test]
232    fn test_create_processor_resource_defaults() {
233        let resource = create_processor_resource(None, None, None);
234
235        let attributes: Vec<_> = resource.iter().collect();
236        assert!(attributes
237            .iter()
238            .any(|(k, v)| k.as_str() == SERVICE_NAME && v.as_str() == "llm-optimizer-processor"));
239        assert!(attributes
240            .iter()
241            .any(|(k, v)| k.as_str() == "service.component"
242                && v.as_str() == "stream-processor"));
243    }
244
245    #[test]
246    fn test_create_processor_resource_custom() {
247        let resource = create_processor_resource(
248            Some("custom-processor".to_string()),
249            Some("2.0.0".to_string()),
250            Some("production".to_string()),
251        );
252
253        let attributes: Vec<_> = resource.iter().collect();
254        assert!(attributes
255            .iter()
256            .any(|(k, v)| k.as_str() == SERVICE_NAME && v.as_str() == "custom-processor"));
257        assert!(attributes
258            .iter()
259            .any(|(k, v)| k.as_str() == SERVICE_VERSION && v.as_str() == "2.0.0"));
260        assert!(attributes
261            .iter()
262            .any(|(k, v)| k.as_str() == DEPLOYMENT_ENVIRONMENT && v.as_str() == "production"));
263    }
264}
llm_optimizer_processor/telemetry/resource.rs

llm_optimizer_processor/telemetry/
resource.rs
