saluki_components/forwarders/cluster_agent/
mod.rs

1//! Cluster Agent forwarder.
2
3use agent_data_plane_config::shared::SharedConfiguration;
4use async_trait::async_trait;
5use http::{
6    header::AUTHORIZATION,
7    uri::{Authority, Scheme},
8    HeaderName, HeaderValue, Request, Uri,
9};
10use saluki_common::buf::FrozenChunkedBytesBuffer;
11use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder, UsageExpr};
12use saluki_core::{
13    components::{forwarders::*, BuildContext},
14    data_model::payload::PayloadType,
15    observability::ComponentMetricsExt as _,
16};
17use saluki_error::{generic_error, GenericError};
18use saluki_metrics::MetricsBuilder;
19use stringtheory::MetaString;
20use tokio::select;
21use tracing::debug;
22
23use crate::common::datadog::{
24    config::ForwarderConfiguration,
25    endpoints::{ResolvedEndpoint, SingleDestination},
26    io::{EndpointRequestMapper, EndpointRequestMapperFactory, LiveForwarderConfiguration, TransactionForwarder},
27    telemetry::ComponentTelemetry,
28    transaction::{Metadata, Transaction, TransactionBody},
29    DEFAULT_INTAKE_COMPRESSED_SIZE_LIMIT,
30};
31
32const CLUSTER_AGENT_SERIES_PATH: &str = "/series";
33
34static DD_API_KEY_HEADER: HeaderName = HeaderName::from_static("dd-api-key");
35
36/// Cluster Agent forwarder configuration.
37///
38/// This forwarder sends encoded metric series payloads to the Cluster Agent `/series` endpoint using bearer-token
39/// authorization. It intentionally does not use Datadog intake endpoint configuration or `DD-Api-Key` auth.
40pub struct ClusterAgentForwarderConfiguration {
41    forwarder_config: ForwarderConfiguration,
42    auth_header_value: HeaderValue,
43}
44
45impl ClusterAgentForwarderConfiguration {
46    /// Creates a new `ClusterAgentForwarderConfiguration` from the given Cluster Agent endpoint and bearer token.
47    ///
48    /// The Cluster Agent is the only destination this forwarder sends to: the configured Datadog
49    /// intake endpoints, the alternate metrics intakes, and dual shipping do not apply to it, and it
50    /// accepts only V2 series payloads. Because the destination is part of construction, no later
51    /// step can overwrite it.
52    ///
53    /// # Errors
54    ///
55    /// Returns an error if the bearer token cannot be represented in an HTTP header.
56    pub fn from_configuration(
57        shared: &SharedConfiguration, endpoint_url: String, auth_token: String,
58    ) -> Result<Self, GenericError> {
59        let auth_header_value = bearer_auth_header_value(&auth_token)?;
60        let destination = SingleDestination {
61            url: endpoint_url,
62            api_key: auth_token,
63            accepts_v3_series: false,
64        };
65        let forwarder_config =
66            ForwarderConfiguration::for_single_destination(shared, &destination).with_allow_arbitrary_tags(false);
67
68        Ok(Self {
69            forwarder_config,
70            auth_header_value,
71        })
72    }
73}
74
75#[async_trait]
76impl ForwarderBuilder for ClusterAgentForwarderConfiguration {
77    fn input_payload_type(&self) -> PayloadType {
78        PayloadType::Http
79    }
80
81    async fn build(&self, context: BuildContext) -> Result<Box<dyn Forwarder + Send>, GenericError> {
82        let metrics_builder = MetricsBuilder::from_component_context(context.component_context());
83        let telemetry = ComponentTelemetry::from_builder(&metrics_builder);
84        let endpoint_request_mapper_factory = cluster_agent_request_mapper_factory(self.auth_header_value.clone());
85        let forwarder = TransactionForwarder::from_config_with_endpoint_request_mapper(
86            context.component_context().clone(),
87            self.forwarder_config.clone(),
88            // Nothing about this forwarder changes at runtime: the bearer token is not a configured API key, so
89            // nothing refreshes it, and a rejected token is not worth retrying on the chance that the Agent
90            // re-resolves a secret.
91            LiveForwarderConfiguration::default(),
92            get_cluster_agent_endpoint_name,
93            telemetry.clone(),
94            metrics_builder,
95            endpoint_request_mapper_factory,
96        )?;
97
98        Ok(Box::new(ClusterAgentForwarder { forwarder }))
99    }
100}
101
102impl MemoryBounds for ClusterAgentForwarderConfiguration {
103    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
104        builder
105            .minimum()
106            .with_single_value::<ClusterAgentForwarder>("component struct")
107            .with_array::<Transaction<FrozenChunkedBytesBuffer>>("requests channel", 8);
108
109        builder.firm().with_expr(UsageExpr::sum(
110            "in-flight requests",
111            UsageExpr::config(
112                "forwarder_retry_queue_payloads_max_size",
113                self.forwarder_config.retry().queue_max_size_bytes() as usize,
114            ),
115            UsageExpr::product(
116                "high priority queue",
117                UsageExpr::config(
118                    "forwarder_high_prio_buffer_size",
119                    self.forwarder_config.endpoint_buffer_size(),
120                ),
121                UsageExpr::constant("maximum compressed payload size", DEFAULT_INTAKE_COMPRESSED_SIZE_LIMIT),
122            ),
123        ));
124    }
125}
126
127/// Cluster Agent forwarder.
128pub struct ClusterAgentForwarder {
129    forwarder: TransactionForwarder<FrozenChunkedBytesBuffer>,
130}
131
132#[async_trait]
133impl Forwarder for ClusterAgentForwarder {
134    async fn run(mut self: Box<Self>, mut context: ForwarderContext) -> Result<(), GenericError> {
135        let Self { forwarder } = *self;
136
137        let mut health = context.take_health_handle();
138        // This call spawns supervised children of this component. When this run-future returns, the children stop, and
139        // their drain has a time limit.
140        let forwarder = forwarder.spawn();
141
142        health.mark_ready();
143        debug!("Cluster Agent forwarder started.");
144
145        loop {
146            select! {
147                _ = health.live() => continue,
148                maybe_payload = context.payloads().next() => match maybe_payload {
149                    Some(payload) => if let Some(http_payload) = payload.try_into_http_payload() {
150                        let (payload_meta, request) = http_payload.into_parts();
151                        let transaction_meta = Metadata::from_event_and_data_point_count(
152                            payload_meta.event_count(),
153                            payload_meta.data_point_count(),
154                        );
155                        let transaction = Transaction::from_original(transaction_meta, request);
156
157                        forwarder.send_transaction(transaction).await?;
158                    }
159                    None => break,
160                },
161            }
162        }
163
164        // Starts the drain. This component does not finish until the drain is done. The shutdown timeout of the
165        // component bounds this wait.
166        forwarder.shutdown();
167
168        debug!("Cluster Agent forwarder stopped.");
169
170        Ok(())
171    }
172}
173
174fn bearer_auth_header_value(auth_token: &str) -> Result<HeaderValue, GenericError> {
175    let raw_value = format!("Bearer {auth_token}");
176    HeaderValue::from_str(&raw_value)
177        .map_err(|_| generic_error!("cluster_agent.auth_token contains characters that are invalid in HTTP headers."))
178}
179
180fn cluster_agent_request_mapper_factory<B>(auth_header_value: HeaderValue) -> EndpointRequestMapperFactory<B>
181where
182    B: 'static,
183{
184    std::sync::Arc::new(move |endpoint| cluster_agent_request_mapper(endpoint, auth_header_value.clone()))
185}
186
187fn cluster_agent_request_mapper<B>(
188    endpoint: ResolvedEndpoint, auth_header_value: HeaderValue,
189) -> EndpointRequestMapper<B> {
190    let new_uri_authority = Authority::try_from(endpoint.endpoint().authority())
191        .expect("should not fail to construct new endpoint authority");
192    let new_uri_scheme =
193        Scheme::try_from(endpoint.endpoint().scheme()).expect("should not fail to construct new endpoint scheme");
194
195    Box::new(move |mut request: Request<TransactionBody<B>>| {
196        let new_uri = Uri::builder()
197            .scheme(new_uri_scheme.clone())
198            .authority(new_uri_authority.clone())
199            .path_and_query(CLUSTER_AGENT_SERIES_PATH)
200            .build()
201            .expect("should not fail to construct Cluster Agent URI");
202        *request.uri_mut() = new_uri;
203        request.headers_mut().remove(&DD_API_KEY_HEADER);
204        request.headers_mut().insert(AUTHORIZATION, auth_header_value.clone());
205
206        request
207    })
208}
209
210fn get_cluster_agent_endpoint_name(_uri: &Uri) -> Option<MetaString> {
211    Some(MetaString::from_static("cluster_agent_series"))
212}
213
214#[cfg(test)]
215mod tests {
216    use std::collections::HashMap;
217
218    use agent_data_plane_config::{
219        shared::{AltMetricsIntake, V3SeriesMode},
220        ConfigValue,
221    };
222    use http::Method;
223
224    use super::*;
225    use crate::common::datadog::{endpoints::EndpointRoute, test_util::shared_configuration};
226
227    #[test]
228    fn request_mapper_preserves_cluster_agent_series_identity_and_sets_bearer_auth() {
229        let auth_header_value = bearer_auth_header_value("secret-token").expect("auth header should be valid");
230        let endpoint = ResolvedEndpoint::from_raw_endpoint("https://cluster-agent.example.com:5005", "secret-token")
231            .expect("endpoint should resolve");
232        let mut mapper = cluster_agent_request_mapper::<()>(endpoint, auth_header_value);
233        let request = Request::builder()
234            .method(Method::POST)
235            .uri("/api/v2/series")
236            .header("dd-api-key", "primary-api-key")
237            .body(TransactionBody::<()>::Rehydrated(None))
238            .expect("request should build");
239
240        let input_endpoint_name = get_cluster_agent_endpoint_name(request.uri());
241        let request = mapper(request);
242        let mapped_endpoint_name = get_cluster_agent_endpoint_name(request.uri());
243
244        assert_eq!(input_endpoint_name.as_deref(), Some("cluster_agent_series"));
245        assert_eq!(mapped_endpoint_name.as_deref(), Some("cluster_agent_series"));
246        assert_eq!(
247            request.uri().to_string(),
248            "https://cluster-agent.example.com:5005/series"
249        );
250        assert_eq!(request.headers().get(AUTHORIZATION).unwrap(), "Bearer secret-token");
251        assert!(request.headers().get("dd-api-key").is_none());
252    }
253
254    #[test]
255    fn bearer_auth_header_rejects_invalid_token() {
256        assert!(bearer_auth_header_value("bad\ntoken").is_err());
257    }
258
259    #[tokio::test]
260    async fn configuration_uses_only_cluster_agent_endpoint() {
261        // Every configured Datadog intake setting here conflicts with the Cluster Agent destination:
262        // a different API key, an explicit `dd_url`, a `site`, an additional endpoint, an alternate
263        // metrics intake, and V3 series routing. None of them may reach the built forwarder.
264        let mut shared = shared_configuration();
265        shared.endpoints.api_key = "primary-api-key".to_string();
266        shared.endpoints.site = ConfigValue::explicit("datadoghq.eu".to_string());
267        shared.endpoints.dd_url = ConfigValue::explicit("https://app.datadoghq.com".to_string());
268        shared.endpoints.additional_endpoints = HashMap::from([(
269            "https://additional.example.com".to_string(),
270            vec!["additional-api-key".to_string()],
271        )]);
272        shared.endpoints.opw_intake = AltMetricsIntake {
273            enabled: true,
274            url: "https://opw.example.com".to_string(),
275            use_v3_series: true,
276        };
277        shared.metrics_encoding.v3_series_mode = V3SeriesMode::Enabled;
278        shared.metrics_encoding.v3_series_endpoint_modes =
279            HashMap::from([("https://app.datadoghq.com".to_string(), V3SeriesMode::Enabled)]);
280
281        // The same configuration drives a plain Datadog forwarder, which does honor all of it.
282        let datadog_forwarder = ForwarderConfiguration::from_configuration(&shared);
283        assert_eq!(V3SeriesMode::Enabled, datadog_forwarder.use_v3_api_series().enabled);
284
285        let config = ClusterAgentForwarderConfiguration::from_configuration(
286            &shared,
287            "https://cluster-agent.example.com".to_string(),
288            "secret-token".to_string(),
289        )
290        .expect("Cluster Agent forwarder configuration should parse");
291        let endpoints = config
292            .forwarder_config
293            .build_routable_endpoints()
294            .expect("endpoint should resolve");
295
296        assert_eq!(endpoints.len(), 1);
297        assert_eq!(endpoints[0].route(), EndpointRoute::Primary);
298        assert_eq!(
299            endpoints[0].endpoint().endpoint().as_str(),
300            "https://cluster-agent.example.com/"
301        );
302        assert_eq!(&*endpoints[0].endpoint().api_key(), "secret-token");
303        assert_eq!(
304            V3SeriesMode::Disabled,
305            config.forwarder_config.use_v3_api_series().enabled
306        );
307        assert!(config.forwarder_config.use_v3_api_series().endpoints.is_empty());
308        assert!(!config.forwarder_config.allow_arbitrary_tags());
309    }
310}
saluki_components/forwarders/cluster_agent/mod.rs

saluki_components/forwarders/cluster_agent/
mod.rs
