diff --git a/datadog-sidecar/src/service/sidecar_server.rs b/datadog-sidecar/src/service/sidecar_server.rs index d129dc84f5..9f2f46b797 100644 --- a/datadog-sidecar/src/service/sidecar_server.rs +++ b/datadog-sidecar/src/service/sidecar_server.rs @@ -4,7 +4,9 @@ use crate::log::{TemporarilyRetainedMapStats, MULTI_LOG_FILTER, MULTI_LOG_WRITER}; use crate::service::{ sidecar_interface::serve_sidecar_interface_connection, - telemetry::{TelemetryCachedClient, TelemetryCachedClientSet}, + telemetry::{ + InitialTelemetryData, MetricsLogsClientSet, TelemetryCachedClient, TelemetryCachedClientSet, + }, tracing::TraceFlusher, DynamicInstrumentationConfigState, InstanceId, QueueId, RuntimeInfo, RuntimeMetadata, SerializedTracerHeaderTags, SessionConfig, SessionInfo, SidecarAction, SidecarFlushOptions, @@ -100,6 +102,8 @@ pub struct SidecarServer { session_counter: Arc>>, /// A `Mutex` guarded `HashMap` that stores the active telemetry clients. pub(crate) telemetry_clients: TelemetryCachedClientSet, + /// Telemetry clients for logs and metrics that are independent of application lifecycle. + pub(crate) metrics_logs_clients: MetricsLogsClientSet, /// A `Mutex` guarded optional `ManualFutureCompleter` for telemetry configuration. pub self_telemetry_config: Arc>>>, @@ -320,32 +324,31 @@ impl SidecarServer { } pub async fn compute_stats(&self) -> SidecarStats { - let (futures, metric_counts): (Vec<_>, Vec<_>) = { - let clients = self.telemetry_clients.inner.lock_or_panic(); - - let futures = clients - .values() - .filter_map(|client| { - client - .client - .lock_or_panic() - .as_ref() - .and_then(|c| c.worker.stats().ok()) - }) - .collect::>(); - - let metric_counts = clients - .values() - .map(|client| { - client - .client - .lock_or_panic() - .as_ref() - .map_or(0, |c| c.telemetry_metrics.len() as u32) - }) - .collect::>(); + let (futures, metric_count, active_telemetry_clients) = { + let application_clients = self.telemetry_clients.clients(); + let metrics_logs_clients = self.metrics_logs_clients.clients(); + let mut futures = + Vec::with_capacity(application_clients.len() + metrics_logs_clients.len()); + let mut metric_count = 0; + for client in application_clients.iter() { + if let Some(client) = client.lock_or_panic().as_ref() { + metric_count += client.telemetry_metrics.len() as u32; + if let Ok(stats) = client.worker.stats() { + futures.push(stats); + } + } + } + for client in metrics_logs_clients.iter() { + let client = client.lock_or_panic(); + metric_count += client.telemetry_metrics.len() as u32; + if let Ok(stats) = client.worker.stats() { + futures.push(stats); + } + } - (futures, metric_counts) + let active_telemetry_clients = + (application_clients.len() + metrics_logs_clients.len()) as u32; + (futures, metric_count, active_telemetry_clients) }; let telemetry_stats = futures::future::join_all(futures).await; @@ -360,12 +363,7 @@ impl SidecarServer { .values() .map(|s| s.lock_runtimes().len() as u32) .sum(), - active_telemetry_clients: self - .telemetry_clients - .inner - .lock_or_panic() - .values() - .count() as u32, + active_telemetry_clients, active_apps: sessions .values() .map(|s| { @@ -391,7 +389,7 @@ impl SidecarServer { .sum(), remote_configs: self.remote_configs.stats(), debugger_diagnostics_bookkeeping: self.debugger_diagnostics_bookkeeper.stats(), - telemetry_metrics_contexts: metric_counts.into_iter().sum(), + telemetry_metrics_contexts: metric_count, telemetry_worker_errors: telemetry_stats_errors + telemetry_stats.iter().filter(|v| v.is_err()).count() as u32, telemetry_worker: telemetry_stats.into_iter().filter_map(|v| v.ok()).sum(), @@ -497,7 +495,6 @@ impl SidecarInterface for ConnectionSidecarHandler { let env = entry.get().env.as_deref().unwrap_or("none"); let process_tags = session.process_tags_with_svc_source(); - // Pre-compute session config so both the primary and retry get_or_create calls // can use it without re-locking the session. let session_config = session @@ -513,25 +510,28 @@ impl SidecarInterface for ConnectionSidecarHandler { // Get or create the telemetry client. If we observe None under the lock it means // another thread called take() (Stop) in the narrow window between get_or_create // returning and us acquiring the lock — retry once to get a fresh client. - let telemetry_mutex = self.server.telemetry_clients.get_or_create( - service, - env, - &instance_id, - &runtime_metadata, - || session_config.clone(), - process_tags.clone(), - ); - let telemetry_mutex = if telemetry_mutex.lock_or_panic().is_none() { - self.server.telemetry_clients.get_or_create( + let (telemetry_mutex, created) = + self.server.telemetry_clients.get_or_create_with_initial( + service, + env, + &instance_id, + &runtime_metadata, + || session_config.clone(), + process_tags.clone(), + || InitialTelemetryData::from_actions(&actions), + ); + let (telemetry_mutex, created) = if telemetry_mutex.lock_or_panic().is_none() { + self.server.telemetry_clients.get_or_create_with_initial( service, env, &instance_id, &runtime_metadata, || session_config, process_tags, + || InitialTelemetryData::from_actions(&actions), ) } else { - telemetry_mutex + (telemetry_mutex, created) }; let mut telemetry_guard = telemetry_mutex.lock_or_panic(); let Some(telemetry) = telemetry_guard.as_mut() else { @@ -561,7 +561,9 @@ impl SidecarInterface for ConnectionSidecarHandler { match action { SidecarAction::Telemetry(TelemetryActions::AddIntegration(ref integration)) => { if telemetry.shared.integrations.insert(integration.clone()) { - actions_to_process.push(action); + if !created { + actions_to_process.push(action); + } buffered_info_changed = true; } } @@ -574,8 +576,11 @@ impl SidecarInterface for ConnectionSidecarHandler { SidecarAction::Telemetry(TelemetryActions::AddConfig(_)) => { telemetry.shared.config_sent = true; buffered_info_changed = true; - actions_to_process.push(action); + if !created { + actions_to_process.push(action); + } } + SidecarAction::Telemetry(TelemetryActions::AddDependency(_)) if created => {} SidecarAction::Telemetry(TelemetryActions::AddEndpoint(_)) => { telemetry.shared.last_endpoints_push = SystemTime::now(); buffered_info_changed = true; @@ -1084,7 +1089,7 @@ impl SidecarInterface for ConnectionSidecarHandler { // Lazily create the concentrator on first IPC span for this (env, version, service). if let Some(state) = get_or_create_concentrator( &self.server.span_concentrators, - &self.server.telemetry_clients, + &self.server.metrics_logs_clients, &env, &version, session_id, @@ -1106,19 +1111,19 @@ impl SidecarInterface for ConnectionSidecarHandler { debug!("Finished executing flush() for traces and stats") } if options.telemetry { - let workers: Vec<_> = { - let clients = self.server.telemetry_clients.inner.lock_or_panic(); - clients - .values() - .filter_map(|entry| { - entry - .client - .lock_or_panic() - .as_ref() - .map(|c| c.worker.clone()) - }) - .collect() - }; + let workers = self + .server + .telemetry_clients + .clients() + .into_iter() + .filter_map(|client| { + client + .lock_or_panic() + .as_ref() + .map(|client| client.worker.clone()) + }) + .chain(self.server.metrics_logs_clients.workers()) + .collect::>(); futures::future::join_all(workers.into_iter().map(|worker| async move { let _ = worker .send_msg(TelemetryActions::Lifecycle( @@ -1189,6 +1194,7 @@ mod tests { use super::*; use crate::service::{FfeEvaluationMetric, FfeExposure, FfeExposureBatch, FfeTelemetryContext}; use httpmock::{Method::POST, MockServer}; + use tokio::sync::Barrier; use tokio::time::{sleep, Duration as TokioDuration}; /// Build a handler backed by a throwaway socketpair connection. These tests exercise @@ -1405,6 +1411,266 @@ mod tests { assert_eq!(exposures_mock.calls_async().await, 0); assert_eq!(metrics_mock.calls_async().await, 0); } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[cfg_attr(miri, ignore)] + async fn buffered_initial_data_reaches_app_started_through_enqueue_actions() { + const CLIENTS: usize = 16; + const TELEMETRY_PATH: &str = "/telemetry/proxy/api/v2/apmtelemetry"; + + let http_server = MockServer::start_async().await; + let app_started_with_initial_data = http_server + .mock_async(|when, then| { + when.method(POST) + .path(TELEMETRY_PATH) + .body_includes("\"request_type\":\"app-started\"") + .body_includes("\"name\":\"race_config\"") + .body_includes("\"name\":\"race_config_second\"") + .body_includes("\"name\":\"startup-dependency\"") + .body_includes("\"name\":\"startup-integration\""); + then.status(202); + }) + .await; + let app_started_without_initial_data = http_server + .mock_async(|when, then| { + when.method(POST) + .path(TELEMETRY_PATH) + .body_includes("\"request_type\":\"app-started\""); + then.status(202); + }) + .await; + + let handler = test_handler(SidecarServer::default()); + let session = handler.server.get_session("session"); + let mut telemetry_config = Config::default(); + telemetry_config + .set_endpoint_uri(http_server.url("/").parse().unwrap()) + .unwrap(); + *session.session_config.lock_or_panic() = Some(telemetry_config.clone()); + + for index in 0..CLIENTS { + let service = format!("telemetry-enqueue-race-{index}"); + let instance_id = InstanceId::new("session", &format!("runtime-{index}")); + let queue_id = QueueId::from(index as u64 + 1); + handler + .server + .get_runtime(&instance_id) + .lock_applications() + .entry(queue_id) + .or_default() + .set_metadata(String::new(), String::new(), service, Vec::new()); + handler.server.metrics_logs_clients.get_or_create( + &format!("telemetry-enqueue-race-{index}"), + "", + &instance_id, + &RuntimeMetadata::new("php", "8.3", "test"), + || telemetry_config.clone(), + Vec::new(), + ); + + let configuration = |name: &str| { + SidecarAction::Telemetry(TelemetryActions::AddConfig( + libdd_telemetry::data::Configuration { + name: name.to_string(), + value: "present".to_string(), + origin: libdd_telemetry::data::ConfigurationOrigin::Default, + config_id: None, + seq_id: None, + }, + )) + }; + handler + .enqueue_actions( + instance_id, + queue_id, + vec![ + SidecarAction::Telemetry(TelemetryActions::AddDependency( + libdd_telemetry::data::Dependency { + name: "startup-dependency".to_string(), + version: None, + }, + )), + SidecarAction::Telemetry(TelemetryActions::AddIntegration( + libdd_telemetry::data::Integration { + name: "startup-integration".to_string(), + enabled: true, + ..Default::default() + }, + )), + configuration("race_config"), + configuration("race_config_second"), + ], + ) + .await; + } + + tokio::time::timeout(TokioDuration::from_secs(10), async { + while app_started_with_initial_data.calls_async().await + + app_started_without_initial_data.calls_async().await + != CLIENTS + { + sleep(TokioDuration::from_millis(10)).await; + } + }) + .await + .expect("all app-started requests should arrive"); + + assert_eq!( + app_started_with_initial_data.calls_async().await, + CLIENTS, + "every app-started payload should contain the complete initial data batch" + ); + assert_eq!(app_started_without_initial_data.calls_async().await, 0); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 8)] + #[cfg_attr(miri, ignore)] + async fn concurrent_same_key_creation_starts_one_worker() { + const CALLERS: usize = 32; + const SERVICE: &str = "concurrent-client-creation"; + const ENV: &str = "test"; + const TELEMETRY_PATH: &str = "/telemetry/proxy/api/v2/apmtelemetry"; + + let http_server = MockServer::start_async().await; + let app_started = http_server + .mock_async(|when, then| { + when.method(POST) + .path(TELEMETRY_PATH) + .body_includes("\"request_type\":\"app-started\""); + then.status(202); + }) + .await; + let mut config = Config::default(); + config + .set_endpoint_uri(http_server.url("/").parse().unwrap()) + .unwrap(); + + let clients = TelemetryCachedClientSet::default(); + let barrier = Arc::new(Barrier::new(CALLERS)); + let tasks = (0..CALLERS).map(|index| { + let clients = clients.clone(); + let barrier = barrier.clone(); + let config = config.clone(); + tokio::spawn(async move { + let instance_id = InstanceId::new("session", &format!("runtime-{index}")); + barrier.wait().await; + clients.get_or_create( + SERVICE, + ENV, + &instance_id, + &RuntimeMetadata::new("php", "8.3", "test"), + || config, + Vec::new(), + ) + }) + }); + let returned_clients = futures::future::join_all(tasks) + .await + .into_iter() + .map(Result::unwrap) + .collect::>(); + + let first = &returned_clients[0]; + assert!( + returned_clients + .iter() + .all(|client| Arc::ptr_eq(first, client)), + "all same-key callers should receive the same telemetry client" + ); + let worker = first + .lock_or_panic() + .as_ref() + .expect("telemetry client") + .worker + .clone(); + let (tx, rx) = futures::channel::oneshot::channel(); + worker + .send_msg(TelemetryActions::CollectStats(tx)) + .await + .unwrap(); + rx.await.unwrap(); + + assert_eq!(clients.inner.lock_or_panic().len(), 1); + assert_eq!(app_started.calls_async().await, 1); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[cfg_attr(miri, ignore)] + async fn initial_stop_follows_app_started() { + const TELEMETRY_PATH: &str = "/telemetry/proxy/api/v2/apmtelemetry"; + + let http_server = MockServer::start_async().await; + let app_started = http_server + .mock_async(|when, then| { + when.method(POST) + .path(TELEMETRY_PATH) + .body_includes("\"request_type\":\"app-started\""); + then.status(202).delay(TokioDuration::from_millis(200)); + }) + .await; + let app_closing = http_server + .mock_async(|when, then| { + when.method(POST) + .path(TELEMETRY_PATH) + .body_includes("\"request_type\":\"app-closing\""); + then.status(202); + }) + .await; + + let handler = test_handler(SidecarServer::default()); + let session = handler.server.get_session("session"); + let mut telemetry_config = Config::default(); + telemetry_config + .set_endpoint_uri(http_server.url("/").parse().unwrap()) + .unwrap(); + *session.session_config.lock_or_panic() = Some(telemetry_config); + + let instance_id = InstanceId::new("session", "stop-runtime"); + let queue_id = QueueId::from(1); + handler + .server + .get_runtime(&instance_id) + .lock_applications() + .entry(queue_id) + .or_default() + .set_metadata( + String::new(), + String::new(), + "stop-service".to_string(), + Vec::new(), + ); + + handler + .enqueue_actions( + instance_id, + queue_id, + vec![SidecarAction::Telemetry(TelemetryActions::Lifecycle( + LifecycleAction::Stop, + ))], + ) + .await; + + tokio::time::timeout(TokioDuration::from_secs(10), async { + while app_started.calls_async().await != 1 { + sleep(TokioDuration::from_millis(10)).await; + } + }) + .await + .expect("app-started request should arrive"); + assert_eq!( + app_closing.calls_async().await, + 0, + "app-closing arrived before the delayed app-started response completed" + ); + + tokio::time::timeout(TokioDuration::from_secs(10), async { + while app_closing.calls_async().await != 1 { + sleep(TokioDuration::from_millis(10)).await; + } + }) + .await + .expect("app-closing request should arrive after app-started"); + } } // TODO: APMSP-1079 - Unit tests are sparse for the sidecar server. We should add more. diff --git a/datadog-sidecar/src/service/stats_flusher.rs b/datadog-sidecar/src/service/stats_flusher.rs index 224354c85a..030122f09f 100644 --- a/datadog-sidecar/src/service/stats_flusher.rs +++ b/datadog-sidecar/src/service/stats_flusher.rs @@ -215,7 +215,7 @@ pub async fn run_stats_flush_loop( /// Returns `None` when stats config is not available (agentless or not yet configured). pub(crate) fn get_or_create_concentrator( concentrators: &Arc>>>, - telemetry_clients: &crate::service::telemetry::TelemetryCachedClientSet, + telemetry_clients: &crate::service::telemetry::MetricsLogsClientSet, env: &str, version: &str, runtime_id: &str, @@ -299,11 +299,8 @@ pub(crate) fn get_or_create_concentrator( session_config_closure, process_tags, ); - let worker = telemetry_mutex - .lock_or_panic() - .as_ref() - .map(|c| c.worker.clone()); - worker + let worker = telemetry_mutex.lock_or_panic().worker.clone(); + Some(worker) }; let state = Arc::new(SpanConcentratorState { diff --git a/datadog-sidecar/src/service/telemetry.rs b/datadog-sidecar/src/service/telemetry.rs index 02d6d5c8c0..b84be9bae6 100644 --- a/datadog-sidecar/src/service/telemetry.rs +++ b/datadog-sidecar/src/service/telemetry.rs @@ -13,6 +13,7 @@ use base64::prelude::BASE64_URL_SAFE_NO_PAD; use base64::Engine; use datadog_ipc::one_way_shared_memory::OneWayShmWriter; use datadog_ipc::platform::NamedShmHandle; +use std::collections::hash_map::Entry; use std::collections::{HashMap, HashSet, VecDeque}; use std::ffi::CString; use std::hash::{Hash, Hasher}; @@ -34,7 +35,7 @@ use std::time::SystemTime; use libdd_telemetry::config::Config; use libdd_telemetry::data::{self, Integration}; use libdd_telemetry::metrics::{ContextKey, MetricContext}; -use libdd_telemetry::worker::{LifecycleAction, TelemetryActions}; +use libdd_telemetry::worker::{TelemetryActions, TelemetryWorkerFlavor}; /// Sidecar's telemetry worker is native-only, so its handle is pinned to /// [`NativeCapabilities`]. @@ -43,6 +44,34 @@ use manual_future::ManualFuture; use serde_with::{serde_as, VecSkipError}; use tokio::time::{sleep, sleep_until, Instant as TokioInstant}; +fn register_metric( + worker: &TelemetryWorkerHandle, + telemetry_metrics: &mut HashMap, + metric: MetricContext, +) { + if !telemetry_metrics.contains_key(&metric.name) { + telemetry_metrics.insert( + metric.name.clone(), + worker.register_metric_context( + metric.name, + metric.tags, + metric.metric_type, + metric.common, + metric.namespace, + ), + ); + } +} + +fn to_telemetry_point( + telemetry_metrics: &HashMap, + (name, value, tags): (String, f64, Vec), +) -> Option { + telemetry_metrics + .get(&name) + .map(|context_key| TelemetryActions::AddPoint((value, *context_key, tags))) +} + #[derive(Debug)] pub struct InternalTelemetryActions { pub instance_id: InstanceId, @@ -71,19 +100,7 @@ pub(crate) async fn telemetry_action_receiver_task( continue; }; - let Some(client) = telemetry_client - .lock_or_panic() - .as_ref() - .map(|t| t.worker.clone()) - else { - warn!( - "Telemetry client stopped before delivery for {}/{}; dropping {} actions", - batch.service_name(), - batch.env_name(), - batch.action_count(), - ); - continue; - }; + let client = telemetry_client.lock_or_panic().worker.clone(); batch.deliver(&telemetry_client, &client).await; } @@ -140,7 +157,7 @@ async fn next_entry( async fn deliver_batch( actions: Vec, - telemetry_client: &Arc>>, + telemetry_client: &Arc>, client: &TelemetryWorkerHandle, ) { for it_action in actions { @@ -156,16 +173,13 @@ async fn deliver_batch( } InternalTelemetryAction::RegisterTelemetryMetric(metric) => { debug!("Registered telemetry metric: {metric:?}"); - if let Some(t) = telemetry_client.lock_or_panic().as_mut() { - t.register_metric(metric); - } + telemetry_client.lock_or_panic().register_metric(metric); } InternalTelemetryAction::AddMetricPoint((value, name, tags)) => { let metric_name = name.clone(); let point = telemetry_client .lock_or_panic() - .as_ref() - .and_then(|t| t.to_telemetry_point((name, value, tags))); + .to_telemetry_point((name, value, tags)); match point { Some(p) => { if let Err(e) = client.send_msg(p).await { @@ -187,31 +201,7 @@ enum TelemetryBatch { } impl TelemetryBatch { - fn service_name(&self) -> &str { - match self { - TelemetryBatch::Fresh(a) => &a.service_name, - TelemetryBatch::Deferred(d) => &d.key.0, - } - } - - fn env_name(&self) -> &str { - match self { - TelemetryBatch::Fresh(a) => &a.env_name, - TelemetryBatch::Deferred(d) => &d.key.1, - } - } - - fn action_count(&self) -> usize { - match self { - TelemetryBatch::Fresh(a) => a.actions.len(), - TelemetryBatch::Deferred(d) => d.actions.iter().map(|b| b.actions.len()).sum(), - } - } - - fn get_client( - &self, - sidecar: &SidecarServer, - ) -> Option>>> { + fn get_client(&self, sidecar: &SidecarServer) -> Option>> { match self { TelemetryBatch::Fresh(a) => { get_telemetry_client(sidecar, &a.instance_id, &a.service_name, &a.env_name) @@ -286,7 +276,7 @@ impl TelemetryBatch { async fn deliver( self, - telemetry_client: &Arc>>, + telemetry_client: &Arc>, client: &TelemetryWorkerHandle, ) { match self { @@ -331,6 +321,34 @@ pub struct TelemetryCachedEntry { pub client: Arc>>, } +#[derive(Default)] +pub(crate) struct InitialTelemetryData { + configurations: Vec, + dependencies: Vec, + integrations: Vec, +} + +impl InitialTelemetryData { + pub(crate) fn from_actions(actions: &[SidecarAction]) -> Self { + let mut initial = Self::default(); + for action in actions { + match action { + SidecarAction::Telemetry(TelemetryActions::AddConfig(value)) => { + initial.configurations.push(value.clone()); + } + SidecarAction::Telemetry(TelemetryActions::AddDependency(value)) => { + initial.dependencies.push(value.clone()); + } + SidecarAction::Telemetry(TelemetryActions::AddIntegration(value)) => { + initial.integrations.push(value.clone()); + } + _ => {} + } + } + initial + } +} + pub struct TelemetryCachedClient { pub worker: TelemetryWorkerHandle, pub shm_writer: OneWayShmWriter, @@ -359,14 +377,13 @@ impl Default for TelemetryCachedClientShmData { } impl TelemetryCachedClient { - fn new( + fn worker_builder( service: &str, env: &str, instance_id: &InstanceId, runtime_meta: &RuntimeMetadata, - get_config: impl FnOnce() -> Config, process_tags: Vec, - ) -> Self { + ) -> TelemetryWorkerBuilder { let mut builder = TelemetryWorkerBuilder::new_fetch_host( service.to_string(), runtime_meta.language_name.to_string(), @@ -384,19 +401,32 @@ impl TelemetryCachedClient { .collect::>() .join(",") }); + builder + } + + fn new( + service: &str, + env: &str, + instance_id: &InstanceId, + runtime_meta: &RuntimeMetadata, + get_config: impl FnOnce() -> Config, + process_tags: Vec, + initial: InitialTelemetryData, + ) -> Self { + let mut builder = + Self::worker_builder(service, env, instance_id, runtime_meta, process_tags); let config = get_config(); builder.config = config.clone(); + builder.configurations.extend(initial.configurations); + builder.dependencies.extend(initial.dependencies); + builder.integrations.extend(initial.integrations); let (handle, _join) = builder.spawn(); info!("spawned telemetry worker {config:?}"); - let worker = handle.clone(); - tokio::spawn(async move { - worker - .send_msg(TelemetryActions::Lifecycle(LifecycleAction::Start)) - .await - .ok(); - }); + if let Err(error) = handle.send_start() { + warn!("Failed to start telemetry worker: {error}"); + } Self { worker: handle, @@ -419,27 +449,11 @@ impl TelemetryCachedClient { } pub fn register_metric(&mut self, metric: MetricContext) { - if !self.telemetry_metrics.contains_key(&metric.name) { - self.telemetry_metrics.insert( - metric.name.clone(), - self.worker.register_metric_context( - metric.name, - metric.tags, - metric.metric_type, - metric.common, - metric.namespace, - ), - ); - } + register_metric(&self.worker, &mut self.telemetry_metrics, metric); } - pub fn to_telemetry_point( - &self, - (name, val, tags): (String, f64, Vec), - ) -> Option { - self.telemetry_metrics - .get(&name) - .map(|context_key| TelemetryActions::AddPoint((val, *context_key, tags))) + pub fn to_telemetry_point(&self, point: (String, f64, Vec)) -> Option { + to_telemetry_point(&self.telemetry_metrics, point) } pub fn process_actions( @@ -547,6 +561,166 @@ type ServiceString = String; type EnvString = String; type TelemetryCachedClientKey = (ServiceString, EnvString); +pub(crate) struct MetricsLogsCachedClient { + pub(crate) worker: TelemetryWorkerHandle, + pub(crate) telemetry_metrics: HashMap, +} + +impl MetricsLogsCachedClient { + fn new( + service: &str, + env: &str, + instance_id: &InstanceId, + runtime_meta: &RuntimeMetadata, + get_config: impl FnOnce() -> Config, + process_tags: Vec, + ) -> Self { + let mut builder = TelemetryCachedClient::worker_builder( + service, + env, + instance_id, + runtime_meta, + process_tags, + ); + builder.config = get_config(); + builder.flavor = TelemetryWorkerFlavor::MetricsLogs; + + let (handle, _join) = builder.spawn(); + info!("spawned metrics/logs telemetry worker"); + if let Err(error) = handle.send_start() { + warn!("Failed to start metrics/logs telemetry worker: {error}"); + } + + Self { + worker: handle, + telemetry_metrics: HashMap::new(), + } + } + + fn register_metric(&mut self, metric: MetricContext) { + register_metric(&self.worker, &mut self.telemetry_metrics, metric); + } + + fn to_telemetry_point(&self, point: (String, f64, Vec)) -> Option { + to_telemetry_point(&self.telemetry_metrics, point) + } +} + +struct MetricsLogsCachedEntry { + last_used: Instant, + client: Arc>, +} + +pub(crate) struct MetricsLogsClientSet { + inner: Arc>>, + cleanup_handle: Option>, +} + +impl Default for MetricsLogsClientSet { + fn default() -> Self { + let inner: Arc>> = + Arc::new(Default::default()); + let clients = inner.clone(); + let cleanup_handle = tokio::spawn(async move { + loop { + sleep(Duration::from_secs(60)).await; + clients + .lock_or_panic() + .retain(|_, entry| entry.last_used.elapsed() < Duration::from_secs(1800)); + } + }); + Self { + inner, + cleanup_handle: Some(cleanup_handle), + } + } +} + +impl Clone for MetricsLogsClientSet { + fn clone(&self) -> Self { + Self { + inner: Arc::clone(&self.inner), + cleanup_handle: None, + } + } +} + +impl Drop for MetricsLogsClientSet { + fn drop(&mut self) { + if let Some(handle) = self.cleanup_handle.take() { + handle.abort(); + } + } +} + +impl MetricsLogsClientSet { + fn get_existing_client( + &self, + service: &str, + env: &str, + ) -> Option>> { + let key = (service.to_string(), env.to_string()); + let mut clients = self.inner.lock_or_panic(); + clients.get_mut(&key).map(|entry| { + entry.last_used = Instant::now(); + entry.client.clone() + }) + } + + pub(crate) fn get_or_create( + &self, + service: &str, + env: &str, + instance_id: &InstanceId, + runtime_meta: &RuntimeMetadata, + get_config: F, + process_tags: Vec, + ) -> Arc> + where + F: FnOnce() -> Config, + { + let mut clients = self.inner.lock_or_panic(); + let key = (service.to_string(), env.to_string()); + match clients.entry(key) { + Entry::Occupied(mut entry) => { + entry.get_mut().last_used = Instant::now(); + entry.get().client.clone() + } + Entry::Vacant(entry) => { + let client = Arc::new(Mutex::new(MetricsLogsCachedClient::new( + service, + env, + instance_id, + runtime_meta, + get_config, + process_tags, + ))); + entry.insert(MetricsLogsCachedEntry { + last_used: Instant::now(), + client: client.clone(), + }); + info!("Created new metrics/logs telemetry client for {service:?}/{env:?}"); + client + } + } + } + + pub(crate) fn clients(&self) -> Vec>> { + self.inner + .lock_or_panic() + .values() + .map(|entry| entry.client.clone()) + .collect() + } + + pub(crate) fn workers(&self) -> Vec { + self.clients() + .into_iter() + .map(|client| client.lock_or_panic().worker.clone()) + .collect() + } +} + pub struct TelemetryCachedClientSet { pub inner: Arc>>, cleanup_handle: Option>, @@ -591,15 +765,12 @@ impl Clone for TelemetryCachedClientSet { } impl TelemetryCachedClientSet { - fn get_existing_client( - &self, - service: &str, - env: &str, - ) -> Option>>> { - let key = (service.to_string(), env.to_string()); - - let map = self.inner.lock_or_panic(); - map.get(&key).map(|e| e.client.clone()) + pub(crate) fn clients(&self) -> Vec>>> { + self.inner + .lock_or_panic() + .values() + .map(|entry| entry.client.clone()) + .collect() } pub fn get_or_create( @@ -614,32 +785,55 @@ impl TelemetryCachedClientSet { where F: FnOnce() -> Config, { - if let Some(existing) = self.get_existing_client(service, env) { - return existing; - } - - let new_client = Arc::new(Mutex::new(Some(TelemetryCachedClient::new( + self.get_or_create_with_initial( service, env, instance_id, runtime_meta, get_config, process_tags, - )))); + InitialTelemetryData::default, + ) + .0 + } + #[allow(clippy::too_many_arguments)] + pub(crate) fn get_or_create_with_initial( + &self, + service: &str, + env: &str, + instance_id: &InstanceId, + runtime_meta: &RuntimeMetadata, + get_config: F, + process_tags: Vec, + get_initial: I, + ) -> (Arc>>, bool) + where + F: FnOnce() -> Config, + I: FnOnce() -> InitialTelemetryData, + { let mut map = self.inner.lock_or_panic(); let key = (service.to_string(), env.to_string()); - map.insert( - key.clone(), - TelemetryCachedEntry { - last_used: Instant::now(), - client: new_client.clone(), - }, - ); - - info!("Created new telemetry client for {key:?}"); - - new_client + match map.entry(key) { + Entry::Occupied(entry) => (entry.get().client.clone(), false), + Entry::Vacant(entry) => { + let new_client = Arc::new(Mutex::new(Some(TelemetryCachedClient::new( + service, + env, + instance_id, + runtime_meta, + get_config, + process_tags, + get_initial(), + )))); + entry.insert(TelemetryCachedEntry { + last_used: Instant::now(), + client: new_client.clone(), + }); + info!("Created new telemetry client for {service:?}/{env:?}"); + (new_client, true) + } + } } pub fn remove_telemetry_client(&self, service: &str, env: &str) { @@ -652,8 +846,10 @@ pub fn path_for_telemetry(service: &str, env: &str) -> CString { let mut hasher = ZwoHasher::default(); service.hash(&mut hasher); env.hash(&mut hasher); - let hash = hasher.finish(); + telemetry_path_from_hash(hasher.finish()) +} +fn telemetry_path_from_hash(hash: u64) -> CString { let mut path = format!( "/ddtl{}-{}", primary_sidecar_identifier(), @@ -686,9 +882,9 @@ fn get_telemetry_client( instance_id: &InstanceId, service_name: &str, env_name: &str, -) -> Option>>> { +) -> Option>> { if let Some(existing) = sidecar - .telemetry_clients + .metrics_logs_clients .get_existing_client(service_name, env_name) { return Some(existing); @@ -710,7 +906,7 @@ fn get_telemetry_client( let process_tags = session.process_tags_with_svc_source(); - Some(sidecar.telemetry_clients.get_or_create( + Some(sidecar.metrics_logs_clients.get_or_create( service_name, env_name, instance_id,