feat: add flags to control telemetry behaviors

influxdb3/src/commands/serve.rs

@@ -60,7 +60,7 @@ pub const DEFAULT_DATA_DIRECTORY_NAME: &str = ".influxdb3";
 /// The default bind address for the HTTP API.
 pub const DEFAULT_HTTP_BIND_ADDR: &str = "0.0.0.0:8181";
 
-pub const DEFAULT_TELMETRY_ENDPOINT: &str = "https://telemetry.v3.influxdata.com";
+pub const DEFAULT_TELEMETRY_ENDPOINT: &str = "https://telemetry.v3.influxdata.com";
 
 #[derive(Debug, Error)]
 pub enum Error {
@@ -325,6 +325,26 @@ pub struct Config {
         action
     )]
     pub force_snapshot_mem_threshold: MemorySize,
+
+    /// Disable sending telemetry data to telemetry.v3.influxdata.com.
+    #[clap(
+        long = "disable-telemetry-upload",
+        env = "INFLUXDB3_TELEMETRY_DISABLE_UPLOAD",
+        default_value_t = false,
+        hide = true,
+        action
+    )]
+    pub disable_telemetry_upload: bool,
+
+    /// Send telemetry data to the specified endpoint.
+    #[clap(
+        long = "telemetry-endpoint",
+        env = "INFLUXDB3_TELEMETRY_ENDPOINT",
+        default_value = DEFAULT_TELEMETRY_ENDPOINT,
+        hide = true,
+        action
+    )]
+    pub telemetry_endpoint: String,
 }
 
 /// Specified size of the Parquet cache in megabytes (MB)
@@ -538,7 +558,8 @@ pub async fn command(config: Config) -> Result<()> {
         catalog.instance_id(),
         num_cpus,
         Some(Arc::clone(&write_buffer_impl.persisted_files())),
-        DEFAULT_TELMETRY_ENDPOINT,
+        config.telemetry_endpoint.as_str(),
+        config.disable_telemetry_upload,
     )
     .await;
 
@@ -593,7 +614,8 @@ async fn setup_telemetry_store(
     instance_id: Arc<str>,
     num_cpus: usize,
     persisted_files: Option<Arc<PersistedFiles>>,
-    telemetry_endpoint: &'static str,
+    telemetry_endpoint: &str,
+    disable_upload: bool,
 ) -> Arc<TelemetryStore> {
     let os = std::env::consts::OS;
     let influxdb_pkg_version = env!("CARGO_PKG_VERSION");
@@ -605,16 +627,22 @@ async fn setup_telemetry_store(
         .unwrap_or(ObjectStoreType::Memory);
     let storage_type = obj_store_type.as_str();
 
-    TelemetryStore::new(
-        instance_id,
-        Arc::from(os),
-        Arc::from(influx_version),
-        Arc::from(storage_type),
-        num_cpus,
-        persisted_files.map(|p| p as _),
-        telemetry_endpoint,
-    )
-    .await
+    if disable_upload {
+        debug!("Initializing TelemetryStore with upload disabled.");
+        TelemetryStore::new_without_background_runners(persisted_files.map(|p| p as _))
+    } else {
+        debug!("Initializing TelemetryStore with upload enabled for {telemetry_endpoint}.");
+        TelemetryStore::new(
+            instance_id,
+            Arc::from(os),
+            Arc::from(influx_version),
+            Arc::from(storage_type),
+            num_cpus,
+            persisted_files.map(|p| p as _),
+            telemetry_endpoint.to_string(),
+        )
+        .await
+    }
 }
 
 async fn background_buffer_checker(

influxdb3/tests/server/cli.rs

@@ -6,13 +6,14 @@ use std::{
 
 use crate::{ConfigProvider, TestServer};
 use assert_cmd::cargo::CommandCargoExt;
+use assert_cmd::Command as AssertCmd;
 use observability_deps::tracing::debug;
 use pretty_assertions::assert_eq;
 use serde_json::{json, Value};
-use test_helpers::assert_contains;
 use test_helpers::tempfile::NamedTempFile;
 #[cfg(feature = "system-py")]
 use test_helpers::tempfile::TempDir;
+use test_helpers::{assert_contains, assert_not_contains};
 
 const WRITE_REPORTS_PLUGIN_CODE: &str = r#"
 def process_writes(influxdb3_local, table_batches, args=None):
@@ -107,6 +108,121 @@ fn create_plugin_file(code: &str) -> NamedTempFile {
     file
 }
 
+#[test_log::test(tokio::test)]
+async fn test_telemetry_disabled_with_debug_msg() {
+    let serve_args = &[
+        "serve",
+        "--writer-id",
+        "the-best-writer",
+        "--object-store",
+        "memory",
+    ];
+
+    let expected_disabled: &str = "Initializing TelemetryStore with upload disabled.";
+
+    // validate we get a debug message indicating upload disabled
+    let output = AssertCmd::cargo_bin("influxdb3")
+        .unwrap()
+        .args(serve_args)
+        .arg("-vv")
+        .arg("--disable-telemetry-upload")
+        .timeout(std::time::Duration::from_millis(500))
+        .assert()
+        .failure()
+        .get_output()
+        .stdout
+        .clone();
+    let output = String::from_utf8(output).expect("must be able to convert output to String");
+    assert_contains!(output, expected_disabled);
+}
+
+#[test_log::test(tokio::test)]
+async fn test_telemetry_disabled() {
+    let serve_args = &[
+        "serve",
+        "--writer-id",
+        "the-best-writer",
+        "--object-store",
+        "memory",
+    ];
+
+    let expected_disabled: &str = "Initializing TelemetryStore with upload disabled.";
+    // validate no message when debug output disabled
+    let output = AssertCmd::cargo_bin("influxdb3")
+        .unwrap()
+        .args(serve_args)
+        .arg("-v")
+        .arg("--disable-telemetry-upload")
+        .timeout(std::time::Duration::from_millis(500))
+        .assert()
+        .failure()
+        .get_output()
+        .stdout
+        .clone();
+    let output = String::from_utf8(output).expect("must be able to convert output to String");
+    assert_not_contains!(output, expected_disabled);
+}
+
+#[test_log::test(tokio::test)]
+async fn test_telemetry_enabled_with_debug_msg() {
+    let serve_args = &[
+        "serve",
+        "--writer-id",
+        "the-best-writer",
+        "--object-store",
+        "memory",
+    ];
+
+    let expected_enabled: &str =
+        "Initializing TelemetryStore with upload enabled for http://localhost:9999.";
+
+    // validate debug output shows which endpoint we are hitting when telemetry enabled
+    let output = AssertCmd::cargo_bin("influxdb3")
+        .unwrap()
+        .args(serve_args)
+        .arg("-vv")
+        .arg("--telemetry-endpoint")
+        .arg("http://localhost:9999")
+        .timeout(std::time::Duration::from_millis(500))
+        .assert()
+        .failure()
+        .get_output()
+        .stdout
+        .clone();
+    let output = String::from_utf8(output).expect("must be able to convert output to String");
+    assert_contains!(output, expected_enabled);
+}
+
+#[test_log::test(tokio::test)]
+async fn test_telementry_enabled() {
+    let serve_args = &[
+        "serve",
+        "--writer-id",
+        "the-best-writer",
+        "--object-store",
+        "memory",
+    ];
+
+    let expected_enabled: &str =
+        "Initializing TelemetryStore with upload enabled for http://localhost:9999.";
+
+    // validate no telemetry endpoint reported when debug output not enabled
+    let output = AssertCmd::cargo_bin("influxdb3")
+        .unwrap()
+        .args(serve_args)
+        .arg("-v")
+        .arg("--telemetry-endpoint")
+        .arg("http://localhost:9999")
+        .timeout(std::time::Duration::from_millis(500))
+        .assert()
+        .failure()
+        .get_output()
+        .stdout
+        .clone();
+    let output = String::from_utf8(output).expect("must be able to convert output to String");
+    assert_not_contains!(output, expected_enabled);
+}
+
 #[test_log::test(tokio::test)]
 async fn test_show_databases() {
     let server = TestServer::spawn().await;

influxdb3/tests/server/main.rs

@@ -155,6 +155,7 @@ impl TestServer {
         let mut command = Command::cargo_bin("influxdb3").expect("create the influxdb3 command");
         let command = command
             .arg("serve")
+            .arg("--disable-telemetry-upload")
             // bind to port 0 to get a random port assigned:
             .args(["--http-bind", "0.0.0.0:0"])
             .args(["--wal-flush-interval", "10ms"])

influxdb3_telemetry/src/sender.rs

@@ -87,7 +87,7 @@ pub(crate) struct TelemetryPayload {
 /// This function runs in the background and if any call fails
 /// there is no retrying mechanism and it is ok to lose a few samples
 pub(crate) async fn send_telemetry_in_background(
-    full_url: &'static str,
+    full_url: String,
     store: Arc<TelemetryStore>,
     duration_secs: Duration,
 ) -> tokio::task::JoinHandle<()> {

influxdb3_telemetry/src/store.rs

@@ -43,7 +43,7 @@ impl TelemetryStore {
         storage_type: Arc<str>,
         cores: usize,
         persisted_files: Option<Arc<dyn ParquetMetrics>>,
-        telemetry_endpoint: &'static str,
+        telemetry_endpoint: String,
     ) -> Arc<Self> {
         debug!(
             instance_id = ?instance_id,
@@ -322,7 +322,7 @@ mod tests {
             Arc::from("Memory"),
             10,
             Some(parqet_file_metrics),
-            "http://localhost/telemetry",
+            "http://localhost/telemetry".to_string(),
         )
         .await;
         tokio::time::sleep(Duration::from_secs(1)).await;