fix(wasm)!: finish the WASM port of remote-config (#2315)

Fix tokio in Cargo.toml and adds missing SleepCapability.

Author: bwoebi

datadog-sidecar/src/shm_remote_config.rs

@@ -10,7 +10,7 @@ use datadog_ipc::one_way_shared_memory::{open_named_shm, OneWayShmReader, OneWay
 use datadog_ipc::platform::{FileBackedHandle, NamedShmHandle};
 use datadog_ipc::rate_limiter::ShmLimiter;
 use datadog_live_debugger::LiveDebuggingData;
-use libdd_capabilities_impl::NativeHttpClient;
+use libdd_capabilities_impl::{HttpClientCapability, NativeCapabilities};
 use libdd_common::{tag::Tag, MutexExt};
 use libdd_remote_config::config::dynamic::{parse_json, Configs};
 use libdd_remote_config::fetch::{
@@ -246,12 +246,12 @@ fn dynamic_instrumentation_is_enabled(apm_config: Option<bool>, info: &TargetInf
     }
 }
 
-impl<N: NotifyTarget + 'static> MultiTargetHandlers<N, Self, NativeHttpClient>
+impl<N: NotifyTarget + 'static> MultiTargetHandlers<N, Self, NativeCapabilities>
     for ConfigFileStorage<N>
 {
     fn fetched(
         &self,
-        fetcher: &Arc<MultiTargetFetcher<N, Self, NativeHttpClient>>,
+        fetcher: &Arc<MultiTargetFetcher<N, Self, NativeCapabilities>>,
         runtime_id: &Arc<String>,
         target: &Arc<Target>,
         files: &[Arc<StoredShmFile>],
@@ -425,7 +425,7 @@ impl<N: NotifyTarget + 'static> Drop for ShmRemoteConfigsGuard<N> {
 
 #[derive(Clone)]
 pub struct ShmRemoteConfigs<N: NotifyTarget + 'static>(
-    Arc<MultiTargetFetcher<N, ConfigFileStorage<N>, NativeHttpClient>>,
+    Arc<MultiTargetFetcher<N, ConfigFileStorage<N>, NativeCapabilities>>,
 );
 
 // we collect services per env, so that we always query, for each runtime + env, all the services
@@ -454,7 +454,7 @@ impl<N: NotifyTarget + 'static> ShmRemoteConfigs<N> {
         let fetcher = MultiTargetFetcher::new(
             storage,
             invariants,
-            NativeHttpClient::new_without_connection_pooling(),
+            NativeCapabilities::new_without_connection_pooling(),
         );
         fetcher
             .remote_config_interval

libdd-remote-config/Cargo.toml

@@ -48,7 +48,7 @@ base64 = { version = "0.22.1", optional = true }
 sha2 = { version = "0.10", optional = true }
 uuid = { workspace = true, features = ["v4", "std"], optional = true }
 futures-util = { version = "0.3", optional = true }
-tokio = { workspace = true, optional = true }
+tokio = { workspace = true, optional = true, features = ["macros"] }
 tokio-util = { version = "0.7.10",  optional = true }
 manual_future = { version = "0.1.1", optional = true }
 time = { version = "0.3", features = ["parsing", "serde", "formatting"], optional = true }
@@ -67,6 +67,7 @@ hyper-util = { workspace = true, features = ["http1", "client", "client-legacy",
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 libdd-capabilities-impl = { path = "../libdd-capabilities-impl", version = "3.0.0", default-features = false }
+tokio = { workspace = true, optional = true, features = ["rt", "sync", "time"] }
 
 [dev-dependencies]
 futures.workspace = true

libdd-remote-config/examples/remote_config_fetch.rs

@@ -1,7 +1,8 @@
 // Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
-use libdd_capabilities_impl::NativeHttpClient;
+use libdd_capabilities::HttpClientCapability;
+use libdd_capabilities_impl::NativeCapabilities;
 use libdd_common::Endpoint;
 use libdd_remote_config::fetch::{ConfigInvariants, ConfigOptions, SingleChangesFetcher};
 use libdd_remote_config::file_change_tracker::{Change, FilePath};
@@ -51,7 +52,7 @@ async fn main() {
             products: vec![ApmTracing],
             capabilities: vec![],
         },
-        NativeHttpClient::new_without_connection_pooling(),
+        NativeCapabilities::new_without_connection_pooling(),
     );
 
     loop {

libdd-remote-config/src/fetch/fetcher.rs

@@ -7,7 +7,7 @@ use base64::Engine;
 use hashbrown::HashMap;
 use http::uri::PathAndQuery;
 use http::StatusCode;
-use libdd_capabilities::HttpClientCapability;
+use libdd_capabilities::{HttpClientCapability, SleepCapability};
 use libdd_common::{Endpoint, MutexExt};
 use libdd_trace_protobuf::remoteconfig::{
     ClientGetConfigsRequest, ClientGetConfigsResponse, ClientState, ClientTracer, ConfigState,
@@ -104,7 +104,7 @@ impl ConfigProductCapabilities {
     }
 }
 
-pub struct ConfigFetcherState<S, C: HttpClientCapability> {
+pub struct ConfigFetcherState<S, C: HttpClientCapability + SleepCapability> {
     target_files_by_path: Mutex<HashMap<Arc<RemoteConfigPath>, StoredTargetFile<S>>>,
     pub invariants: ConfigInvariants,
     endpoint: Endpoint,
@@ -153,7 +153,7 @@ impl<S> ConfigFetcherFilesLock<'_, S> {
     }
 }
 
-impl<S, C: HttpClientCapability> ConfigFetcherState<S, C> {
+impl<S, C: HttpClientCapability + SleepCapability> ConfigFetcherState<S, C> {
     pub fn with_client(invariants: ConfigInvariants, http_client: C) -> Self {
         ConfigFetcherState {
             target_files_by_path: Default::default(),
@@ -203,7 +203,7 @@ impl<S, C: HttpClientCapability> ConfigFetcherState<S, C> {
     }
 }
 
-pub struct ConfigFetcher<S: FileStorage, C: HttpClientCapability> {
+pub struct ConfigFetcher<S: FileStorage, C: HttpClientCapability + SleepCapability> {
     pub file_storage: S,
     state: Arc<ConfigFetcherState<S::StoredFile, C>>,
 }
@@ -238,7 +238,7 @@ impl ConfigClientState {
     }
 }
 
-impl<S: FileStorage, C: HttpClientCapability> ConfigFetcher<S, C> {
+impl<S: FileStorage, C: HttpClientCapability + SleepCapability> ConfigFetcher<S, C> {
     pub fn new(file_storage: S, state: Arc<ConfigFetcherState<S::StoredFile, C>>) -> Self {
         ConfigFetcher {
             file_storage,
@@ -364,19 +364,31 @@ impl<S: FileStorage, C: HttpClientCapability> ConfigFetcher<S, C> {
                 libdd_common::header::APPLICATION_JSON,
             )
             .body(bytes::Bytes::from(serde_json::to_string(&config_req)?))?;
-        let response = tokio::time::timeout(
-            Duration::from_millis(self.state.endpoint.timeout_ms),
-            self.state.http_client.request(req),
-        )
-        .await
-        .map_err(|e| anyhow::Error::msg(e).context(format!("Url: {:?}", self.state.endpoint)))?
-        .map_err(|e| anyhow::Error::msg(e).context(format!("Url: {:?}", self.state.endpoint)))?;
+        let sleeper = <C as SleepCapability>::new();
+        let response = tokio::select! {
+            biased;
+            result = self.state.http_client.request(req) => result
+                .map_err(|e| anyhow::Error::msg(e).context(format!("Url: {:?}", self.state.endpoint)))?,
+            _ = sleeper.sleep(Duration::from_millis(self.state.endpoint.timeout_ms)) => {
+                anyhow::bail!(
+                    "Remote config request timed out after {}ms. Url: {:?}",
+                    self.state.endpoint.timeout_ms,
+                    self.state.endpoint
+                )
+            }
+        };
         let status = response.status();
         let body_bytes = response.into_body();
         if status != StatusCode::OK {
             // Not active
             if status == StatusCode::NOT_FOUND {
                 trace!("Requested remote config and but remote config not active");
+                if self.state.expire_unused_files {
+                    self.state.target_files_by_path.lock_or_panic().clear();
+                }
+                client_state.last_config_paths.clear();
+                client_state.targets_version = 0;
+                client_state.opaque_backend_state.clear();
                 return Ok(Some(vec![]));
             }
 
@@ -592,7 +604,7 @@ pub mod tests {
     use crate::fetch::test_server::RemoteConfigServer;
     use crate::RemoteConfigSource;
     use http::Response;
-    use libdd_capabilities_impl::NativeHttpClient;
+    use libdd_capabilities_impl::NativeCapabilities;
     use libdd_common::http_common;
     use std::mem::transmute;
     use std::sync::LazyLock;
@@ -708,7 +720,7 @@ pub mod tests {
             storage.clone(),
             Arc::new(ConfigFetcherState::with_client(
                 server.dummy_options().invariants,
-                NativeHttpClient::new_without_connection_pooling(),
+                NativeCapabilities::new_without_connection_pooling(),
             )),
         );
         let mut opaque_state = ConfigClientState::default();
@@ -746,7 +758,7 @@ pub mod tests {
             storage.clone(),
             Arc::new(ConfigFetcherState::with_client(
                 server.dummy_options().invariants,
-                NativeHttpClient::new_without_connection_pooling(),
+                NativeCapabilities::new_without_connection_pooling(),
             )),
         );
         let mut opaque_state = ConfigClientState::default();
@@ -863,7 +875,7 @@ pub mod tests {
             storage.clone(),
             Arc::new(ConfigFetcherState::with_client(
                 invariants,
-                NativeHttpClient::new_without_connection_pooling(),
+                NativeCapabilities::new_without_connection_pooling(),
             )),
         );
         let mut opaque_state = ConfigClientState::default();
@@ -1050,7 +1062,7 @@ pub mod tests {
             storage,
             Arc::new(ConfigFetcherState::with_client(
                 server.dummy_options().invariants,
-                NativeHttpClient::new_without_connection_pooling(),
+                NativeCapabilities::new_without_connection_pooling(),
             )),
         );
         let mut opaque_state = ConfigClientState::default();
@@ -1151,7 +1163,7 @@ pub mod tests {
             storage,
             Arc::new(ConfigFetcherState::with_client(
                 server.dummy_options().invariants,
-                NativeHttpClient::new_without_connection_pooling(),
+                NativeCapabilities::new_without_connection_pooling(),
             )),
         );
         let mut opaque_state = ConfigClientState::default();

libdd-remote-config/src/fetch/mod.rs

@@ -2,7 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 mod fetcher;
+#[cfg(not(target_arch = "wasm32"))]
 mod multitarget;
+#[cfg(not(target_arch = "wasm32"))]
 mod shared;
 mod single;
 #[cfg(any(test, feature = "test"))]
@@ -11,6 +13,8 @@ pub mod test_server;
 #[allow(clippy::useless_attribute)] // different clippy versions are differently picky
 #[cfg_attr(test, allow(ambiguous_glob_reexports))] // ignore mod tests re-export
 pub use fetcher::*;
+#[cfg(not(target_arch = "wasm32"))]
 pub use multitarget::*;
+#[cfg(not(target_arch = "wasm32"))]
 pub use shared::*;
 pub use single::*;

libdd-remote-config/src/fetch/multitarget.rs

@@ -8,7 +8,7 @@ use crate::fetch::{
 use crate::{RemoteConfigCapabilities, RemoteConfigProduct, Target};
 use futures_util::future::Shared;
 use futures_util::FutureExt;
-use libdd_capabilities::HttpClientCapability;
+use libdd_capabilities::{HttpClientCapability, SleepCapability};
 use libdd_common::MutexExt;
 use manual_future::ManualFuture;
 use serde::{Deserialize, Serialize};
@@ -37,7 +37,7 @@ use tracing::{debug, error, trace};
 pub struct MultiTargetFetcher<
     N: NotifyTarget,
     S: FileStorage + Clone + Sync + Send,
-    C: HttpClientCapability + Send + Sync,
+    C: HttpClientCapability + SleepCapability + Send + Sync,
 > where
     S::StoredFile: RefcountedFile + Sync + Send,
     S: MultiTargetHandlers<N, S, C>,
@@ -156,7 +156,7 @@ pub trait NotifyTarget: Sync + Send + Sized + Hash + Eq + Clone + Debug {
 pub trait MultiTargetHandlers<
     N: NotifyTarget,
     S: FileStorage + Clone + Sync + Send + MultiTargetHandlers<N, S, C>,
-    C: HttpClientCapability + Send + Sync,
+    C: HttpClientCapability + SleepCapability + Send + Sync,
 > where
     S::StoredFile: RefcountedFile + Sync + Send,
 {
@@ -186,7 +186,7 @@ struct RuntimeInfo<N: NotifyTarget> {
 impl<
         N: NotifyTarget + 'static,
         S: FileStorage + Clone + Sync + Send + 'static,
-        C: HttpClientCapability + Send + Sync + 'static,
+        C: HttpClientCapability + SleepCapability + Send + Sync + 'static,
     > MultiTargetFetcher<N, S, C>
 where
     S::StoredFile: RefcountedFile + Sync + Send,
@@ -785,7 +785,7 @@ mod tests {
     use crate::fetch::shared::tests::*;
     use crate::fetch::test_server::RemoteConfigServer;
     use crate::{RemoteConfigPath, Target};
-    use libdd_capabilities_impl::NativeHttpClient;
+    use libdd_capabilities_impl::NativeCapabilities;
     use manual_future::ManualFutureCompleter;
     use std::hash::Hasher;
     use std::sync::atomic::AtomicU8;
@@ -822,10 +822,10 @@ mod tests {
         }
     }
 
-    impl MultiTargetHandlers<Notifier, MultiFileStorage, NativeHttpClient> for MultiFileStorage {
+    impl MultiTargetHandlers<Notifier, MultiFileStorage, NativeCapabilities> for MultiFileStorage {
         fn fetched(
             &self,
-            _fetcher: &Arc<MultiTargetFetcher<Notifier, MultiFileStorage, NativeHttpClient>>,
+            _fetcher: &Arc<MultiTargetFetcher<Notifier, MultiFileStorage, NativeCapabilities>>,
             _runtime_id: &Arc<String>,
             target: &Arc<Target>,
             files: &[Arc<RcPathStore>],
@@ -967,10 +967,10 @@ mod tests {
 
         let fut = storage.await_fetches(1);
 
-        let fetcher = MultiTargetFetcher::<Notifier, MultiFileStorage, NativeHttpClient>::new(
+        let fetcher = MultiTargetFetcher::<Notifier, MultiFileStorage, NativeCapabilities>::new(
             storage.clone(),
             server.dummy_options().invariants,
-            NativeHttpClient::new_without_connection_pooling(),
+            NativeCapabilities::new_without_connection_pooling(),
         );
         fetcher.remote_config_interval.store(1000, Ordering::SeqCst);
 

libdd-remote-config/src/fetch/shared.rs

@@ -6,7 +6,7 @@ use crate::fetch::{
     ConfigFetcherStateStats, ConfigInvariants, ConfigProductCapabilities, FileStorage,
 };
 use crate::{RemoteConfigPath, Target};
-use libdd_capabilities::HttpClientCapability;
+use libdd_capabilities::{HttpClientCapability, SleepCapability};
 use libdd_common::MutexExt;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -127,7 +127,7 @@ impl RunnersGeneration {
     }
 }
 
-pub struct RefcountingStorage<S: FileStorage + Clone, C: HttpClientCapability>
+pub struct RefcountingStorage<S: FileStorage + Clone, C: HttpClientCapability + SleepCapability>
 where
     S::StoredFile: RefcountedFile,
 {
@@ -161,7 +161,8 @@ impl Add for RefcountingStorageStats {
     }
 }
 
-impl<S: FileStorage + Clone, C: HttpClientCapability> Clone for RefcountingStorage<S, C>
+impl<S: FileStorage + Clone, C: HttpClientCapability + SleepCapability> Clone
+    for RefcountingStorage<S, C>
 where
     S::StoredFile: RefcountedFile,
 {
@@ -175,7 +176,7 @@ where
     }
 }
 
-impl<S: FileStorage + Clone, C: HttpClientCapability> RefcountingStorage<S, C>
+impl<S: FileStorage + Clone, C: HttpClientCapability + SleepCapability> RefcountingStorage<S, C>
 where
     S::StoredFile: RefcountedFile,
 {
@@ -223,7 +224,8 @@ where
     }
 }
 
-impl<S: FileStorage + Clone, C: HttpClientCapability> FileStorage for RefcountingStorage<S, C>
+impl<S: FileStorage + Clone, C: HttpClientCapability + SleepCapability> FileStorage
+    for RefcountingStorage<S, C>
 where
     S::StoredFile: RefcountedFile,
 {
@@ -268,7 +270,7 @@ impl SharedFetcher {
     /// On successful fetches on_fetch() is called with the new configuration.
     /// Should not be called more than once.
     #[allow(clippy::type_complexity)]
-    pub async fn run<S: FileStorage + Clone, C: HttpClientCapability>(
+    pub async fn run<S: FileStorage + Clone, C: HttpClientCapability + SleepCapability>(
         &self,
         storage: RefcountingStorage<S, C>,
         on_fetch: Box<dyn Send + Fn(&Vec<Arc<S::StoredFile>>)>,
@@ -378,7 +380,7 @@ pub mod tests {
     use crate::fetch::test_server::RemoteConfigServer;
     use crate::Target;
     use futures::future::join_all;
-    use libdd_capabilities_impl::NativeHttpClient;
+    use libdd_capabilities_impl::NativeCapabilities;
     use std::sync::{Arc, LazyLock};
 
     pub(crate) static OTHER_TARGET: LazyLock<Arc<Target>> = LazyLock::new(|| {
@@ -439,7 +441,7 @@ pub mod tests {
             storage.clone(),
             ConfigFetcherState::with_client(
                 server.dummy_options().invariants,
-                NativeHttpClient::new_without_connection_pooling(),
+                NativeCapabilities::new_without_connection_pooling(),
             ),
         );
 
@@ -504,7 +506,7 @@ pub mod tests {
             storage.clone(),
             ConfigFetcherState::with_client(
                 server.dummy_options().invariants,
-                NativeHttpClient::new_without_connection_pooling(),
+                NativeCapabilities::new_without_connection_pooling(),
             ),
         );