datadog_agent_apm/semantics/
provider.rs

1//! Provides access to either the embedded or remotely updated semantic registry mappings.
2//!
3//! With Remote Configuration disabled, the default provider always uses the embedded registry. A subscribed provider
4//! also starts with the embedded registry, then returns the latest accepted `APM_SEMANTIC_CORE_DD` update when
5//! available.
6
7use std::fmt;
8use std::sync::Arc;
9
10use datadog_agent_remote_config::{ApplyError, ConfigId, RemoteConfigurationClient, Subscription};
11
12use super::registry::{Registry, EMBEDDED_REGISTRY};
13use super::remote_config::SemanticCoreDecoder;
14
15/// Identifies whether a [`SemanticCore`] registry is embedded or supplied by a remote configuration.
16#[derive(Clone, Debug, PartialEq, Eq)]
17pub(crate) enum Origin {
18    /// The registry embedded in the binary, used when no configuration is assigned.
19    Embedded,
20
21    /// The ID of the selected remote configuration.
22    Remote(ConfigId),
23}
24
25impl fmt::Display for Origin {
26    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
27        // Match upstream registry source labels (`pkg/trace/semantics/registry.go:22-25`).
28        match self {
29            Self::Embedded => f.write_str("embedded"),
30            Self::Remote(id) => write!(f, "remote-config ({id})"),
31        }
32    }
33}
34
35/// The registry selected from a set of remote semantic-mapping configurations.
36///
37/// `APM_SEMANTIC_CORE_DD` is the Remote Configuration product that distributes semantic registries. An assignment is
38/// the set of configurations currently supplied for that product. This snapshot holds the selected registry and its
39/// origin, or the embedded registry when the assignment is empty.
40pub(crate) struct SemanticCore {
41    /// The complete registry to use, not a set of changes to the previous registry.
42    pub(crate) registry: Arc<Registry>,
43
44    /// Where the registry came from.
45    pub(crate) origin: Origin,
46
47    /// How many configurations were assigned, including ones that failed to decode.
48    pub(crate) assigned: usize,
49}
50
51#[cfg(any(test, feature = "test-util"))]
52impl SemanticCore {
53    /// Creates a snapshot holding `registry`, as if selected from configuration `id`.
54    pub(crate) fn remote(registry: Registry, id: &str) -> Self {
55        Self {
56            registry: Arc::new(registry),
57            origin: Origin::Remote(ConfigId::new(id)),
58            assigned: 1,
59        }
60    }
61}
62
63impl fmt::Debug for SemanticCore {
64    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
65        f.debug_struct("SemanticCore")
66            .field("content_hash", &self.registry.content_hash())
67            .field("version", &self.registry.version())
68            .field("origin", &self.origin)
69            .field("assigned", &self.assigned)
70            .finish()
71    }
72}
73
74/// Explains why a remote semantic registry could not be parsed or selected.
75///
76/// The Remote Configuration client reports this error to the server. Error text must not include payload contents.
77/// A rejected update leaves the provider's last accepted registry unchanged.
78#[derive(Debug)]
79pub(crate) enum SemanticCoreError {
80    /// A configuration's payload is not a valid registry.
81    InvalidRegistry {
82        /// Rejection reason safe to send to the server, without quoting payload contents.
83        reason: String,
84    },
85
86    /// Configurations were assigned but none of them is a valid registry.
87    NoValidConfiguration {
88        /// How many configurations were assigned.
89        assigned: usize,
90    },
91}
92
93impl fmt::Display for SemanticCoreError {
94    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
95        match self {
96            Self::InvalidRegistry { reason } => f.write_str(reason),
97            Self::NoValidConfiguration { assigned } => write!(
98                f,
99                "None of the {assigned} assigned APM_SEMANTIC_CORE_DD configurations is a valid registry."
100            ),
101        }
102    }
103}
104
105impl std::error::Error for SemanticCoreError {}
106
107impl ApplyError for SemanticCoreError {
108    fn apply_error(&self) -> String {
109        self.to_string()
110    }
111}
112
113/// Supplies semantic mappings from either the embedded registry or the latest accepted remote update.
114///
115/// Consumers find attribute names for concepts such as HTTP status code without choosing a registry source.
116/// The default provider always returns the embedded registry: it never subscribes or waits for updates. Use it when
117/// Remote Configuration is disabled by configuration or deployment policy.
118///
119/// With a live subscription to `APM_SEMANTIC_CORE_DD`, the provider uses the embedded registry until an update is
120/// accepted, then returns the latest accepted registry. Rejected updates leave that registry unchanged.
121///
122/// Clones share the subscription. Each component calls [`Self::snapshot`] at its batch boundary to keep one set of
123/// mappings throughout processing.
124#[derive(Clone, Debug)]
125pub struct SemanticRegistryProvider {
126    subscription: Subscription<SemanticCore, SemanticCoreError>,
127}
128
129impl SemanticRegistryProvider {
130    /// Subscribes to `APM_SEMANTIC_CORE_DD` on `client`.
131    ///
132    /// Clone the returned provider for every component that reads the registry; a second subscription to the product
133    /// on the same client fails.
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if `client` already has a live subscription to `APM_SEMANTIC_CORE_DD`.
138    pub fn subscribe(client: &RemoteConfigurationClient) -> Result<Self, datadog_agent_remote_config::Error> {
139        Ok(Self::from_subscription(client.subscribe::<SemanticCoreDecoder>()?))
140    }
141
142    /// Wraps a registry subscription. An inert subscription keeps using the embedded registry.
143    pub(crate) fn from_subscription(subscription: Subscription<SemanticCore, SemanticCoreError>) -> Self {
144        Self { subscription }
145    }
146
147    /// Returns the latest accepted registry, or the embedded registry if the subscription has no accepted value.
148    ///
149    /// Returns immediately, without waiting for an update. The registry is immutable and remains valid after later
150    /// updates. Call once per batch and use the result throughout; calling again can return a newer registry.
151    pub fn snapshot(&self) -> Arc<Registry> {
152        match self.subscription.current() {
153            Some(core) => Arc::clone(&core.registry),
154            None => Arc::clone(&EMBEDDED_REGISTRY),
155        }
156    }
157}
158
159impl Default for SemanticRegistryProvider {
160    fn default() -> Self {
161        Self::from_subscription(Subscription::inert())
162    }
163}
164
165#[cfg(test)]
166mod tests {
167    use datadog_agent_remote_config::TestPublisher;
168
169    use super::*;
170    use crate::semantics::registry::registry_json;
171
172    fn remote(concepts: &str, id: &str) -> SemanticCore {
173        let registry = Registry::from_json(&registry_json(concepts)).expect("registry should parse");
174        SemanticCore::remote(registry, id)
175    }
176
177    const DB_STATEMENT: &str =
178        r#"{"db.statement":{"fallbacks":[{"name":"db.statement","provider":"datadog","type":"string"}]}}"#;
179
180    #[test]
181    fn default_provider_returns_the_embedded_registry() {
182        let provider = SemanticRegistryProvider::default();
183        assert!(Arc::ptr_eq(&provider.snapshot(), &EMBEDDED_REGISTRY));
184    }
185
186    #[test]
187    fn provider_returns_the_embedded_registry_before_a_snapshot_is_accepted() {
188        let (_publisher, subscription) = TestPublisher::new();
189        let provider = SemanticRegistryProvider::from_subscription(subscription);
190        assert!(Arc::ptr_eq(&provider.snapshot(), &EMBEDDED_REGISTRY));
191    }
192
193    #[test]
194    fn clones_observe_an_accepted_snapshot() {
195        let (publisher, subscription) = TestPublisher::new();
196        let provider = SemanticRegistryProvider::from_subscription(subscription);
197        let clone = provider.clone();
198
199        let core = remote(DB_STATEMENT, "a");
200        let registry = Arc::clone(&core.registry);
201        publisher.accept(core);
202
203        assert!(Arc::ptr_eq(&provider.snapshot(), &registry));
204        assert!(Arc::ptr_eq(&clone.snapshot(), &registry));
205    }
206
207    #[test]
208    fn a_snapshot_is_unchanged_by_a_later_publication() {
209        let (publisher, subscription) = TestPublisher::new();
210        let provider = SemanticRegistryProvider::from_subscription(subscription);
211
212        let pinned = provider.snapshot();
213        publisher.accept(remote(DB_STATEMENT, "a"));
214
215        assert!(Arc::ptr_eq(&pinned, &EMBEDDED_REGISTRY));
216        assert!(!Arc::ptr_eq(&provider.snapshot(), &EMBEDDED_REGISTRY));
217    }
218
219    #[test]
220    fn a_rejection_keeps_the_accepted_registry() {
221        let (publisher, subscription) = TestPublisher::new();
222        let provider = SemanticRegistryProvider::from_subscription(subscription);
223
224        let core = remote(DB_STATEMENT, "a");
225        let registry = Arc::clone(&core.registry);
226        publisher.accept(core);
227        publisher.reject(SemanticCoreError::NoValidConfiguration { assigned: 1 });
228
229        assert!(Arc::ptr_eq(&provider.snapshot(), &registry));
230    }
231
232    #[test]
233    fn independent_providers_do_not_share_publications() {
234        let (publisher, subscription) = TestPublisher::new();
235        let (_other_publisher, other_subscription) = TestPublisher::new();
236        let provider = SemanticRegistryProvider::from_subscription(subscription);
237        let other = SemanticRegistryProvider::from_subscription(other_subscription);
238
239        publisher.accept(remote(DB_STATEMENT, "a"));
240
241        assert!(!Arc::ptr_eq(&provider.snapshot(), &EMBEDDED_REGISTRY));
242        assert!(Arc::ptr_eq(&other.snapshot(), &EMBEDDED_REGISTRY));
243    }
244}
datadog_agent_apm/semantics/provider.rs

datadog_agent_apm/semantics/
provider.rs
