Add telemetry for used SDKs (#3643)

Author: slinkydeveloper

Cargo.lock

@@ -247,6 +247,7 @@ tracing-test = { version = "0.2.5" }
 typed-builder = "0.21.0"
 ulid = { version = "1.2.0" }
 url = { version = "2.5" }
+urlencoding = { version = "2.1" }
 uuid = { version = "1.3.0", features = ["v7", "serde"] }
 xxhash-rust = { version = "0.8", features = ["xxh3"] }
 

Cargo.toml

@@ -61,6 +61,7 @@ tokio = { workspace = true }
 tonic = { workspace = true, features = ["transport", "codegen", "prost", "gzip", "zstd"] }
 tower = { workspace = true, features = ["load-shed", "limit"] }
 tracing = { workspace = true }
+urlencoding = { workspace = true }
 
 [dev-dependencies]
 restate-bifrost = { workspace = true, features = ["test-util"] }

crates/admin/Cargo.toml

@@ -11,16 +11,19 @@
 pub mod error;
 mod updater;
 
+use anyhow::Context;
+use http::uri::PathAndQuery;
+use http::{HeaderMap, HeaderValue, Uri};
 use std::borrow::Borrow;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::sync::Arc;
 use std::time::Duration;
-
-use http::Uri;
 use tracing::subscriber::NoSubscriber;
+use tracing::trace;
 
-use restate_core::{Metadata, MetadataWriter};
+use restate_core::{Metadata, MetadataWriter, TaskCenter, TaskKind};
+use restate_service_client::HttpClient;
 use restate_service_protocol::discovery::{DiscoverEndpoint, DiscoveredEndpoint, ServiceDiscovery};
 use restate_types::identifiers::{DeploymentId, ServiceRevision, SubscriptionId};
 use restate_types::schema::Schema;
@@ -77,18 +80,21 @@ pub enum ModifyServiceChange {
 pub struct SchemaRegistry<V> {
     metadata_writer: MetadataWriter,
     service_discovery: ServiceDiscovery,
+    telemetry_http_client: Option<HttpClient>,
     subscription_validator: V,
 }
 
 impl<V> SchemaRegistry<V> {
     pub fn new(
         metadata_writer: MetadataWriter,
         service_discovery: ServiceDiscovery,
+        telemetry_http_client: Option<HttpClient>,
         subscription_validator: V,
     ) -> Self {
         Self {
             metadata_writer,
             service_discovery,
+            telemetry_http_client,
             subscription_validator,
         }
     }
@@ -123,7 +129,7 @@ impl<V> SchemaRegistry<V> {
             ),
         };
 
-        let (id, services) = if !apply_mode.should_apply() {
+        let (deployment, services) = if !apply_mode.should_apply() {
             let mut updater =
                 SchemaUpdater::new(Metadata::with_current(|m| m.schema()).deref().clone());
 
@@ -171,7 +177,64 @@ impl<V> SchemaRegistry<V> {
             (deployment, services)
         };
 
-        Ok((id, services))
+        if apply_mode.should_apply() {
+            self.send_register_deployment_telemetry(deployment.metadata.sdk_version.clone());
+        }
+
+        Ok((deployment, services))
+    }
+
+    fn send_register_deployment_telemetry(&self, sdk_version: Option<String>) {
+        if let Some(telemetry_http_client) = &self.telemetry_http_client {
+            let client = telemetry_http_client.clone();
+            let _ = TaskCenter::spawn(TaskKind::Disposable, "telemetry-operation", async move {
+                let (sdk_type, full_sdk_version_string) = if let Some(sdk_version) = &sdk_version {
+                    (
+                        sdk_version
+                            .split_once('/')
+                            .map(|(version, _)| version)
+                            .unwrap_or_else(|| "unknown"),
+                        sdk_version.as_str(),
+                    )
+                } else {
+                    ("unknown", "unknown")
+                };
+
+                let uri = format!(
+                    "{TELEMETRY_URI_PREFIX}?sdk={}&version={}",
+                    urlencoding::encode(sdk_type),
+                    urlencoding::encode(full_sdk_version_string)
+                )
+                .parse()
+                .with_context(|| "cannot create telemetry uri")?;
+
+                trace!(%uri, "Sending telemetry data");
+
+                match client
+                    .request(
+                        uri,
+                        None,
+                        http::Method::GET,
+                        http_body_util::Empty::new(),
+                        PathAndQuery::from_static("/"),
+                        HeaderMap::from_iter([(
+                            http::header::USER_AGENT,
+                            HeaderValue::from_static("restate-server"),
+                        )]),
+                    )
+                    .await
+                {
+                    Ok(resp) => {
+                        trace!(status = %resp.status(), "Sent telemetry data")
+                    }
+                    Err(err) => {
+                        trace!(error = %err, "Failed to send telemetry data")
+                    }
+                }
+
+                Ok(())
+            });
+        }
     }
 
     pub async fn update_deployment(
@@ -467,3 +530,5 @@ impl Borrow<String> for ServiceName {
         &self.0
     }
 }
+
+static TELEMETRY_URI_PREFIX: &str = "https://restate.gateway.scarf.sh/sdk-registration/";

crates/admin/src/schema_registry/mod.rs

@@ -20,6 +20,7 @@ use crate::schema_registry::SchemaRegistry;
 use crate::{rest_api, state};
 use restate_core::MetadataWriter;
 use restate_core::network::net_util;
+use restate_service_client::HttpClient;
 use restate_service_protocol::discovery::ServiceDiscovery;
 use restate_types::invocation::client::InvocationClient;
 use restate_types::net::BindAddress;
@@ -51,6 +52,7 @@ where
         invocation_client: IC,
         subscription_validator: V,
         service_discovery: ServiceDiscovery,
+        telemetry_http_client: Option<HttpClient>,
     ) -> Self {
         Self {
             bifrost,
@@ -59,6 +61,7 @@ where
             schema_registry: SchemaRegistry::new(
                 metadata_writer,
                 service_discovery,
+                telemetry_http_client,
                 subscription_validator,
             ),
             invocation_client,

crates/admin/src/service.rs

@@ -48,7 +48,7 @@ tokio = { workspace = true }
 tower = { workspace = true, features = ["util"] }
 tower-http = { workspace = true, features = ["cors", "normalize-path"] }
 url = "2.5.0"
-urlencoding = "2.1"
+urlencoding = { workspace = true }
 
 [dev-dependencies]
 restate-core = { workspace = true, features = ["test-util"] }

crates/ingress-http/Cargo.toml

@@ -20,7 +20,7 @@ use restate_core::network::TransportConnect;
 use restate_core::partitions::PartitionRouting;
 use restate_core::worker_api::PartitionProcessorInvocationClient;
 use restate_core::{Metadata, MetadataWriter, TaskCenter, TaskKind};
-use restate_service_client::{AssumeRoleCacheMode, ServiceClient};
+use restate_service_client::{AssumeRoleCacheMode, HttpClient, ServiceClient};
 use restate_service_protocol::discovery::ServiceDiscovery;
 use restate_storage_query_datafusion::context::{QueryContext, SelectPartitionsFromMetadata};
 use restate_storage_query_datafusion::empty_invoker_status_handle::EmptyInvokerStatusHandle;
@@ -84,6 +84,12 @@ impl<T: TransportConnect> AdminRole<T> {
             ServiceClient::from_options(&config.common.service_client, AssumeRoleCacheMode::None)?;
         let service_discovery = ServiceDiscovery::new(retry_policy, client);
 
+        let telemetry_http_client = if config.common.disable_telemetry {
+            None
+        } else {
+            Some(HttpClient::from_options(&config.common.service_client.http))
+        };
+
         let query_context = if let Some(query_context) = local_query_context {
             query_context
         } else {
@@ -114,6 +120,7 @@ impl<T: TransportConnect> AdminRole<T> {
             ),
             config.ingress.clone(),
             service_discovery,
+            telemetry_http_client,
         )
         .with_query_context(query_context);
 

crates/node/src/roles/admin.rs

@@ -209,6 +209,7 @@ async fn generate_rest_api_doc() -> anyhow::Result<()> {
             ServiceClient::from_options(&config.common.service_client, AssumeRoleCacheMode::None)
                 .unwrap(),
         ),
+        None,
     );
 
     TaskCenter::spawn(