datadog_agent_remote_config/decoder.rs
1//! The subscriber's decoding contract.
2
3use std::panic::{catch_unwind, AssertUnwindSafe};
4
5use crate::{ApplyError, ConfigId};
6
7/// Decodes one product's assigned configurations into a snapshot.
8///
9/// A decoder names the product it decodes, so subscribing with a decoder cannot attach it to the wrong product. Define
10/// a decoder in the crate that owns its snapshot type. A decoder for a product whose type has no other owner belongs in
11/// this crate instead; otherwise this crate knows no product's payload.
12///
13/// To evaluate an assignment, the client constructs a decoder with [`Default`], calls [`decode`](Self::decode) once
14/// per assigned configuration in ascending [`ConfigId`] order, and then calls [`build`](Self::build). On success,
15/// [`keeps_previous`](Self::keeps_previous) decides whether to replace or retain the current snapshot. Ascending order
16/// is part of the contract, so a reduction that keeps the last valid configuration is well defined.
17///
18/// A product is decoded again only when its assigned configurations or their contents change, so a rejected
19/// assignment is not retried until it changes. When the Agent reports its configuration expired, every product that
20/// had configurations is decoded as an empty assignment; a product that already had none is not decoded again. The
21/// client then asks the Agent for the full assignment until a response is not expired, so configurations that return
22/// are decoded as new.
23///
24/// A rejected snapshot leaves subscribers with the last accepted one, so an implementation must build a snapshot
25/// from an empty assignment; see [`build`](Self::build).
26///
27/// Decoding runs on the client's worker task and delays polling for every product while it runs, so implementations
28/// **MUST NOT** block. A panic in [`decode`](Self::decode) or [`build`](Self::build) is caught: the client discards the
29/// decoder, rejects every configuration in the assignment, and publishes nothing, so subscribers keep the last accepted
30/// snapshot and are not notified.
31///
32/// # Design
33///
34/// A product's assignment is several configurations, each with its own payload, and the protocol carries an apply
35/// status for each one separately. Decoding therefore accumulates configuration by configuration rather than consuming
36/// the assignment as a whole: a configuration that [`decode`](Self::decode) rejects is attributed to itself and
37/// skipped, and the decoder still builds from the rest. Because the client drives the loop, that attribution cannot be
38/// forgotten or misdirected. Whether a rejected configuration invalidates the whole snapshot remains the decoder's
39/// choice: [`build`](Self::build) may reject the snapshot, for example when a configuration it requires was rejected.
40///
41/// Accumulating into the decoder, rather than returning one decoded value per configuration, is what lets a product
42/// whose configurations have different shapes hold each of them in a field of its own type instead of funneling them
43/// through a shared enum.
44///
45/// # Examples
46///
47/// A product that accepts at most one configuration records every configuration it is given and rejects more than one
48/// in [`build`](Self::build), since only `build` sees the whole assignment:
49///
50/// ```
51/// use datadog_agent_remote_config::{ConfigId, ProductDecoder};
52///
53/// /// Holds the product's single configuration, if it has one.
54/// #[derive(Default)]
55/// struct SingleDecoder {
56/// payloads: Vec<(ConfigId, Vec<u8>)>,
57/// }
58///
59/// impl ProductDecoder for SingleDecoder {
60/// const PRODUCT: &'static str = "EXAMPLE_SINGLE";
61///
62/// type Snapshot = Option<Vec<u8>>;
63/// type Error = String;
64///
65/// fn decode(&mut self, id: &ConfigId, payload: &[u8]) -> Result<(), Self::Error> {
66/// self.payloads.push((id.clone(), payload.to_vec()));
67/// Ok(())
68/// }
69///
70/// fn build(mut self) -> Result<Self::Snapshot, Self::Error> {
71/// if self.payloads.len() > 1 {
72/// return Err(format!("Expected at most one configuration; got {}.", self.payloads.len()));
73/// }
74/// Ok(self.payloads.pop().map(|(_, payload)| payload))
75/// }
76/// }
77///
78/// let mut decoder = SingleDecoder::default();
79/// decoder.decode(&ConfigId::new("a"), b"one").unwrap();
80/// decoder.decode(&ConfigId::new("b"), b"two").unwrap();
81/// assert!(decoder.build().is_err());
82/// ```
83pub trait ProductDecoder: Default + Send + 'static {
84 /// The product this decoder decodes, by its protocol name, such as `APM_SEMANTIC_CORE_DD`.
85 ///
86 /// The client requests this product from the Agent and keys the product's subscription by this name. The client
87 /// can only subscribe once per product: attempting a second subscription to the same product while the first is
88 /// live returns [`Error::AlreadySubscribed`](crate::Error::AlreadySubscribed), whichever decoder it uses.
89 const PRODUCT: &'static str;
90
91 /// The type that holds the product's decoded configuration.
92 ///
93 /// [`build`](Self::build) produces it from the product's configurations. [`keeps_previous`](Self::keeps_previous)
94 /// decides whether subscribers receive this snapshot or retain the previous one.
95 type Snapshot: Send + Sync + 'static;
96
97 /// The error this product's decoding and validation produces.
98 ///
99 /// Use [`String`] when there is nothing richer to report; a product that wants to describe a failure more
100 /// precisely for its own diagnostics defines its own type and implements [`ApplyError`] for it.
101 type Error: ApplyError + Send + Sync + 'static;
102
103 /// Accumulates one of the product's assigned configurations.
104 ///
105 /// The client calls this at most once per distinct `id` within a single snapshot, then calls
106 /// [`build`](Self::build) once. A decoder may therefore hold one slot per configuration ID without a second
107 /// configuration silently displacing the first.
108 ///
109 /// # Errors
110 ///
111 /// Returns [`Self::Error`] to reject this configuration alone. The client attributes the rejection to it and skips
112 /// it, and still decodes the product's remaining configurations.
113 fn decode(&mut self, id: &ConfigId, payload: &[u8]) -> Result<(), Self::Error>;
114
115 /// Validates the accumulated configurations and produces a snapshot.
116 ///
117 /// Check requirements across configurations here, including whether a required configuration is present when the
118 /// decoder receives a non-empty assignment. The client also calls this without calling [`decode`](Self::decode)
119 /// when no configurations are assigned (after the backend removes the last one or the Agent reports its
120 /// configuration expired), or when every assigned configuration was rejected for sharing an ID.
121 ///
122 /// An empty assignment to the decoder means "no configuration," even if files were assigned but all collided.
123 /// This method **MUST** build a snapshot representing that state. If the product should retain its previous
124 /// configuration on withdrawal, use [`keeps_previous`](Self::keeps_previous) rather than rejecting the assignment.
125 ///
126 /// The client acknowledges successfully decoded configurations only when this method succeeds.
127 ///
128 /// # Errors
129 ///
130 /// Returns [`Self::Error`] to reject the snapshot. The client rejects every successfully decoded configuration
131 /// with this error's apply reason; configurations rejected by [`decode`](Self::decode) keep their own errors.
132 /// No new snapshot is published, and subscribers retain the last accepted snapshot.
133 ///
134 /// This method cannot reject selected configurations while publishing the rest. Selective rejection belongs in
135 /// [`decode`](Self::decode).
136 fn build(self) -> Result<Self::Snapshot, Self::Error>;
137
138 /// Controls whether an unassigned-marker update preserves the last accepted configuration.
139 ///
140 /// Remote Configuration assigns configuration files to each product. When the backend withdraws the last file, or
141 /// the Agent reports that its cached configuration has expired, the product's assignment becomes empty: there is
142 /// no longer a remote configuration assigned to it.
143 ///
144 /// The decoder must still represent this valid state. Its [`build`](Self::build) can return a marker such as an
145 /// `Unassigned` variant in its snapshot type, meaning "no remote configuration is assigned." This is a value the
146 /// decoder creates, not an error or a special payload in a file. Another decoder might instead build a snapshot
147 /// containing its local defaults. Whether either result should replace the previous configuration depends on the
148 /// product's rules for withdrawal.
149 ///
150 /// Use this hook to retain the last accepted configuration when the product becomes unassigned. The client calls
151 /// it for every successfully built snapshot; your decoder identifies its product's unassigned marker. Return
152 /// `true` **only for that marker** when withdrawal should preserve the previous configuration, and `false` for
153 /// ordinary configurations.
154 ///
155 /// - `false` (the default): publish this snapshot, replacing the previous one.
156 /// - `true`: accept this unassigned-marker update without replacing the previous snapshot. Later configuration
157 /// updates still return `false` and replace it normally. If no snapshot exists yet, publish the marker.
158 ///
159 /// Trace sampling (`APM_SAMPLING`) keeps its last sampling settings on withdrawal. Its decoder returns `true` for
160 /// `Unassigned` and `false` for every actual configuration. Even an assigned `{}` or JSON `null` returns `false`:
161 /// it is a configuration with no overrides, which restores local settings, not a withdrawal.
162 ///
163 /// The semantic registry (`APM_SEMANTIC_CORE_DD`) must instead revert to embedded mappings on withdrawal. Its
164 /// decoder builds an embedded-registry snapshot and uses the default `false`, so that snapshot replaces the remote
165 /// registry. Consumers, including APM stats, then use the embedded mappings.
166 ///
167 /// Retention happens here because subscriptions hold only the latest snapshot, not a queue. If configuration A is
168 /// followed by an `Unassigned` marker before a sampler reads either, publishing the marker would erase A. Retaining
169 /// A lets the sampler apply it even after withdrawal. New concrete configurations can still replace one another.
170 ///
171 /// Both return values accept the update, clear any earlier rejection, notify subscribers, and acknowledge
172 /// configuration verdicts normally. Retaining a snapshot does not mean its configuration remains remotely assigned.
173 /// The call runs inside the same panic guard as `build`.
174 fn keeps_previous(snapshot: &Self::Snapshot) -> bool {
175 let _ = snapshot;
176 false
177 }
178}
179
180/// The fixed reason reported for every configuration when a decoder panics.
181///
182/// A panic is a bug in the decoder rather than a verdict about any one configuration's bytes, so it overwrites any
183/// rejection a configuration had already earned, including from an earlier `decode` call.
184pub(crate) const PANICKED: &str = "Product decoder panicked.";
185
186/// What one run of a decoder over a product's assignment produced.
187pub(crate) struct Evaluation<T, E> {
188 pub(crate) outcome: Outcome<T, E>,
189 /// Each assigned configuration's verdict, in ascending ID order.
190 pub(crate) verdicts: Vec<(ConfigId, Verdict)>,
191}
192
193/// The verdict one configuration earned from a run of a decoder.
194///
195/// The reason is the string the client reports to the Agent; the variant is which rule produced it, which is what
196/// the client counts.
197#[derive(Debug, PartialEq, Eq)]
198pub(crate) enum Verdict {
199 /// `decode` accepted this configuration and the snapshot it was part of was built.
200 Acknowledged,
201
202 /// `decode` rejected this configuration, with the reason it gave.
203 DecodeRejected(String),
204
205 /// `decode` accepted this configuration, but `build` rejected the snapshot, with the reason it gave.
206 BuildRejected(String),
207
208 /// The decoder panicked, so this configuration is rejected without a reason of its own.
209 Panicked,
210
211 /// This configuration shares its ID with another, so decode did not run.
212 ///
213 /// Set by the client rather than a decoder; see [`Repository`](crate::repository::Repository).
214 Collided(String),
215}
216
217pub(crate) enum Outcome<T, E> {
218 /// `build` succeeded; the snapshot is published.
219 Accepted(T),
220
221 /// `build` succeeded and [`ProductDecoder::keeps_previous`] returned `true`; the snapshot is published only if no
222 /// snapshot has been accepted yet.
223 KeepsPrevious(T),
224
225 /// `build` failed; the error is published as a rejection and `current` keeps the last accepted snapshot.
226 Rejected {
227 error: E,
228
229 /// The error's apply reason, computed inside the decoder's `catch_unwind` so that a panic in
230 /// [`ApplyError::apply_error`] is caught like any other decoder panic.
231 reason: String,
232 },
233
234 /// `decode` or `build` panicked; nothing is published and subscribers are not notified.
235 Panicked,
236}
237
238/// Runs a fresh decoder over one product's assignment exactly as the client does.
239///
240/// This is the only implementation of the decoding rules, shared by the worker and by
241/// [`TestPublisher::assign`](crate::TestPublisher::assign), so that what a subscriber tests is what production runs:
242/// configurations in ascending [`ConfigId`] order, a rejected configuration skipped while the rest are still decoded,
243/// then `build`, with a panic in either caught.
244pub(crate) fn evaluate<P: ProductDecoder>(mut assignment: Vec<(ConfigId, &[u8])>) -> Evaluation<P::Snapshot, P::Error> {
245 assignment.sort_by(|(left, _), (right, _)| left.cmp(right));
246
247 let mut verdicts = Vec::with_capacity(assignment.len());
248 let result = catch_unwind(AssertUnwindSafe(|| {
249 let mut decoder = P::default();
250 for (id, payload) in &assignment {
251 let verdict = match decoder.decode(id, payload) {
252 Ok(()) => Verdict::Acknowledged,
253 Err(error) => Verdict::DecodeRejected(error.apply_error()),
254 };
255 verdicts.push((id.clone(), verdict));
256 }
257 match decoder.build() {
258 Ok(snapshot) if P::keeps_previous(&snapshot) => Outcome::KeepsPrevious(snapshot),
259 Ok(snapshot) => Outcome::Accepted(snapshot),
260 Err(error) => {
261 let reason = error.apply_error();
262 for (_, verdict) in &mut verdicts {
263 if matches!(verdict, Verdict::Acknowledged) {
264 *verdict = Verdict::BuildRejected(reason.clone());
265 }
266 }
267 Outcome::Rejected { error, reason }
268 }
269 }
270 }));
271
272 let outcome = match result {
273 Ok(outcome) => outcome,
274 Err(_) => {
275 verdicts = assignment.into_iter().map(|(id, _)| (id, Verdict::Panicked)).collect();
276 Outcome::Panicked
277 }
278 };
279
280 Evaluation { outcome, verdicts }
281}