diff --git a/.cargo/config.toml b/.cargo/config.toml index 493077312e1..d637b80719e 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -2,6 +2,10 @@ rustflags = ["--cfg", "tokio_unstable"] rustdocflags = ["--cfg", "tokio_unstable"] +# Keeps tests and `cargo run` from sending telemetry. A value already set in the environment wins. +[env] +QW_DISABLE_TELEMETRY = "1" + [target.x86_64-unknown-linux-gnu] # x86-64-v3 requires AVX2, BMI1/BMI2, FMA — available on Intel Haswell (2013)+ # and AMD Zen 1 (2017)+. Covers all current-gen AWS and GCP instance families. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7a43b963e44..583e44ae69d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,6 +18,7 @@ permissions: env: CARGO_INCREMENTAL: 0 + QW_DISABLE_TELEMETRY: 1 QW_TEST_DATABASE_URL: postgres://quickwit-dev:quickwit-dev@localhost:5432/quickwit-metastore-dev RUST_BACKTRACE: 1 RUSTDOCFLAGS: -Dwarnings -Arustdoc::private_intra_doc_links diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index 6026dcc409f..e4ab7939c59 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -16,6 +16,7 @@ env: AWS_SECRET_ACCESS_KEY: "placeholder" CARGO_INCREMENTAL: 0 PUBSUB_EMULATOR_HOST: "localhost:8681" + QW_DISABLE_TELEMETRY: 1 QW_S3_ENDPOINT: "http://localhost:4566" # Services are exposed as localhost because we are not running coverage in a container. QW_S3_FORCE_PATH_STYLE_ACCESS: 1 QW_TEST_DATABASE_URL: postgres://quickwit-dev:quickwit-dev@localhost:5432/quickwit-metastore-dev diff --git a/.github/workflows/full-tests.yml b/.github/workflows/full-tests.yml index e891b225488..ca17a4028fa 100644 --- a/.github/workflows/full-tests.yml +++ b/.github/workflows/full-tests.yml @@ -55,6 +55,8 @@ jobs: timeout-minutes: 60 permissions: contents: read + env: + QW_DISABLE_TELEMETRY: 1 steps: - name: Free disk space run: | diff --git a/.github/workflows/ui-ci.yml b/.github/workflows/ui-ci.yml index 916ab4955cd..83edd65f54b 100644 --- a/.github/workflows/ui-ci.yml +++ b/.github/workflows/ui-ci.yml @@ -59,6 +59,7 @@ jobs: --health-retries 5 env: CARGO_INCREMENTAL: 0 + QW_DISABLE_TELEMETRY: 1 RUST_BACKTRACE: 1 RUSTFLAGS: -Dwarnings --cfg tokio_unstable RUSTDOCFLAGS: -Dwarnings -Arustdoc::private_intra_doc_links diff --git a/CHANGELOG.md b/CHANGELOG.md index d0675117ebc..b9f2b1cfcdb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Azure Blob Storage: support custom endpoints via `endpoint` and `endpoint_suffix` configuration options for sovereign clouds (#6624) +- Restore anonymous usage telemetry, removed in 0.9.0. It is on by default: set `QW_DISABLE_TELEMETRY=1` to turn it off. `docs/telemetry.md` lists what is sent and where it goes (#6927) ### Fixed - (Jaeger) Query resource attributes when Jaeger request carries tags diff --git a/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv index d1099b1a52f..4252326571f 100644 --- a/LICENSE-3rdparty.csv +++ b/LICENSE-3rdparty.csv @@ -448,6 +448,7 @@ nom,https://github.com/rust-bakery/nom,MIT,contact@geoffroycouprie.com nom-language,https://github.com/rust-bakery/nom,MIT,contact@geoffroycouprie.com nonzero_ext,https://github.com/antifuchs/nonzero_ext,Apache-2.0,Andreas Fuchs normalize-line-endings,https://github.com/derekdreery/normalize-line-endings,Apache-2.0,Richard Dodd +ntapi,https://github.com/MSxDOS/ntapi,Apache-2.0 OR MIT,MSxDOS nu-ansi-term,https://github.com/nushell/nu-ansi-term,MIT,"ogham@bsago.me, Ryan Scheel (Havvy) , Josh Triplett , The Nushell Project Developers" num,https://github.com/rust-num/num,MIT OR Apache-2.0,The Rust Project Developers num-bigint,https://github.com/rust-num/num-bigint,MIT OR Apache-2.0,The Rust Project Developers diff --git a/docs/operating/data-directory.md b/docs/operating/data-directory.md index aca5b4c8baa..b3bbaae8452 100644 --- a/docs/operating/data-directory.md +++ b/docs/operating/data-directory.md @@ -22,6 +22,7 @@ qwdata ├── indexing │ ├── wikipedia%01H13SVKDS03P%_ingest-api-source%RbaOAI │ └── wikipedia%01H13SVKDS03P%kafka-source%cNqQtI +├── telemetry_install_id ├── wal │ ├── wal-00000000000000000056 │ └── wal-00000000000000000057 @@ -58,6 +59,10 @@ You can [configure](../configuration/node-config#indexer-configuration) the numb This directory is used by searcher nodes to cache entire splits and reduce calls to the object store. It won't be created unless you set the `split_cache` fields in the [searcher configuration](../configuration/node-config.md#searcher-configuration). +### `/telemetry_install_id` file + +This file stores the install ID that [telemetry](../telemetry.md) reports, and is only created while telemetry is on. Nodes started from copies of the same data directory report the same install ID, and deleting the file makes the node report a new `install` event. + ## Setting the right splits cache limits diff --git a/docs/reference/cli.md b/docs/reference/cli.md index e4cbaa5e1e3..cc212dfe17f 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -855,6 +855,22 @@ Specifies the path to the [quickwit config](../configuration/node-config.md). Co `export QW_CONFIG=config/quickwit.yaml` +### QW_DISABLE_TELEMETRY + +Disables [telemetry](../telemetry.md) when set to any non-empty value other than `false`, `no` or `0`. + +*Example* + +`QW_DISABLE_TELEMETRY=1 quickwit --help` + +### TELEMETRY_INGEST_API + +Sends [telemetry](../telemetry.md) to this URL instead of `https://telemetry.quickwit.io/`. The URL is used as is, so the data is not encrypted with an `http://` URL. It does not turn telemetry off. + +*Example* + +`TELEMETRY_INGEST_API=http://127.0.0.1:8080/ quickwit run` + ### QW_POSTGRES_SKIP_MIGRATIONS Don't run database migrations (but verify that migrations were run successfully before, and no that unknown migration was run). diff --git a/docs/telemetry.md b/docs/telemetry.md new file mode 100644 index 00000000000..b6b2761461e --- /dev/null +++ b/docs/telemetry.md @@ -0,0 +1,87 @@ +--- +title: Telemetry +sidebar_position: 12 +--- + +Quickwit collects anonymous usage data to learn which versions are running, on which platforms, and which features are in use. Quickwit is maintained by Datadog, Inc., which receives and stores this data. + +## Disabling telemetry + +Telemetry is on by default. To turn it off, set `QW_DISABLE_TELEMETRY` before starting Quickwit: + +```bash +export QW_DISABLE_TELEMETRY=1 +``` + +Any non-empty value other than `false`, `no` or `0` disables telemetry. + +To check whether telemetry is on, run `quickwit --help`. While it is on, the output includes a `Telemetry: enabled` line: + +```bash +quickwit --help +Sub-second search & analytics engine on cloud storage. + Find more information at https://quickwit.io/docs + +Telemetry: enabled +``` + +The line disappears when telemetry is off. If the node can't set up HTTPS, for example because no CA certificates are installed, `quickwit run` logs a warning and sends nothing even though the line is shown. + +Only `quickwit run` sends telemetry. Other CLI commands never do. + +## What is sent + +Each time a node starts, it sends a request right away. After that, it sends at most one request per minute, plus one last request with the pending events when it shuts down. Each request holds up to 10 events. + +Events, each with the time it happened, in seconds since the Unix epoch: + +- `install`: the node found no install ID in its data directory and created one. This happens with a new data directory, but also on the first start after upgrading from a release without install IDs, and on the first start after telemetry is turned on. +- `run_command`: the node started. +- `end_command`: the node stopped, with its exit code. +- `running`: the node is still up. Sent every 12 hours. +- `ui_index_page_load`: the node served the Quickwit UI page. Browsers request it when someone opens or reloads the UI, and so does any other client that requests a UI URL, such as an uptime check. + +Each request also includes: + +- The version of the request format. +- The Quickwit version, and how the binary was built: the commit, the target platform, and whether it was built from a release tag. +- The services the node runs, such as `indexer`, `searcher` or `metastore`. +- The features in use: + - the metastore type, file-backed or PostgreSQL, if the node runs the metastore or a metastore read replica; + - where the default index root URI points: S3, Azure Blob Storage, Google Cloud Storage or the local file system; + - whether the S3 storage config targets an S3-compatible store, through a custom endpoint or a flavor such as MinIO, Garage, DigitalOcean Spaces or Google Cloud Storage; + - whether the OTLP endpoints are enabled on an indexer, and the Jaeger endpoint on a searcher; + - whether ingest API v1 and v2 are enabled; + - whether the searcher uses AWS Lambda for leaf searches or a local split cache; + - whether document clustering is configured; + - whether the REST or gRPC server uses TLS. +- The OS and the CPU architecture. +- The number of CPUs and the amount of memory available to the node, as ranges such as `5-8` CPUs or `8-16GiB`. +- Whether the node runs on Kubernetes, based on whether `KUBERNETES_SERVICE_HOST` is set. +- A session ID, generated at random each time the node starts. +- An install ID, generated at random when the node starts with telemetry on and finds none in its data directory, then stored there in the `telemetry_install_id` file. It is never created while telemetry is off. +- An MD5 hash of the host name and the user name, used to count distinct installations from older releases, which have no install ID. The relay replaces it with a keyed hash before storage. See below. +- The number of events dropped because too many were queued. + +`running` and `end_command` events also include how many requests each API received since the previous report, as ranges such as `10-99`: search and bulk ingestion on the Elasticsearch-compatible API, other Elasticsearch-compatible requests, search and ingestion on the native API, OTLP, and Jaeger. Searches from the Quickwit UI count as native searches. + +Quickwit never sends index or field names, documents, queries, configuration values or credentials. + +## Where the data goes + +By default, Quickwit sends the data over HTTPS to `telemetry.quickwit.io`, which Datadog operates on AWS. + +- The request goes through Amazon CloudFront, which looks up the country and city of your IP address and passes them on to the relay. CloudFront also passes on the IP address itself, in the `X-Forwarded-For` header that it adds to every request, but the relay ignores it. +- A small relay checks the request against the format above, rejects anything else, and keeps only the fields listed on this page. +- The relay replaces the host hash with a keyed hash, computed with a secret that only the relay knows, so the stored value can't be matched against guessed host names and user names. +- The relay forwards the events, with the country and city, to Datadog, where they are stored as logs. Your IP address is not stored. + +For nodes running in the cloud, the country and city are usually those of the data center. + +To send the data somewhere else instead, for example to a local HTTP server to see exactly what is sent, set `TELEMETRY_INGEST_API` to a URL before starting Quickwit. Nothing then goes to `telemetry.quickwit.io`, and the relay is bypassed: the receiver gets the requests as Quickwit sends them, unencrypted with an `http://` URL. This does not turn telemetry off. + +Releases 0.6.0 to 0.8.x send an older version of this data to the same address: the events other than `install`, each with its time, without the API counts, plus the version, services, OS, architecture, Kubernetes flag, session ID, host hash, number of dropped events, and the metastore, OTLP and Jaeger features. Their AWS Lambda indexer and searcher send `run_command` with only the `aws_lambda` feature. + +Releases before 0.6.0 also send data to the same address, from every CLI command: an event when a command ends, with its exit code, and an event when an index is created, ingested into, deleted or garbage collected, or when services start, with their names (in 0.1.x, with whether peer addresses were given). Each event has its time, and each request includes the number of dropped events, the version, OS, architecture, Kubernetes flag, session ID and host hash. + +To turn telemetry off in releases before 0.9.0, set `QW_DISABLE_TELEMETRY` to any value, or `DISABLE_QUICKWIT_TELEMETRY` in 0.1.x. diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 8fad29951e6..d9646f52a00 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -5547,6 +5547,15 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61807f77802ff30975e01f4f071c8ba10c022052f98b3294119f3e615d13e5be" +[[package]] +name = "ntapi" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3b335231dfd352ffb0f8017f3b6027a4917f7df785ea2143d8af2adc66980ae" +dependencies = [ + "winapi", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -7476,6 +7485,7 @@ dependencies = [ "quickwit-search", "quickwit-serve", "quickwit-storage", + "quickwit-telemetry", "quickwit-telemetry-exporters", "quickwit-transport", "reqwest 0.13.5", @@ -8340,6 +8350,7 @@ dependencies = [ "quickwit-query", "quickwit-search", "quickwit-storage", + "quickwit-telemetry", "quickwit-telemetry-exporters", "quickwit-transport", "regex", @@ -8427,6 +8438,25 @@ dependencies = [ "ulid 3.0.0", ] +[[package]] +name = "quickwit-telemetry" +version = "0.9.3" +dependencies = [ + "async-trait", + "enum-iterator", + "hostname", + "md5", + "quickwit-common", + "reqwest 0.13.5", + "serde", + "serde_json", + "sysinfo", + "tempfile", + "tokio", + "tracing", + "uuid", +] + [[package]] name = "quickwit-telemetry-exporters" version = "0.9.3" @@ -10665,6 +10695,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2071df9448915b71c4fe6d25deaf1c22f12bd234f01540b77312bb8e41361e6" dependencies = [ "libc", + "memchr", + "ntapi", "objc2-core-foundation", "objc2-io-kit", "windows", diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml index 10c9c222bf2..275449f3f39 100644 --- a/quickwit/Cargo.toml +++ b/quickwit/Cargo.toml @@ -37,6 +37,7 @@ members = [ "quickwit-search", "quickwit-serve", "quickwit-storage", + "quickwit-telemetry", "quickwit-telemetry-exporters", "quickwit-transport", ] @@ -375,6 +376,7 @@ quickwit-rest-client = { path = "quickwit-rest-client" } quickwit-search = { path = "quickwit-search" } quickwit-serve = { path = "quickwit-serve" } quickwit-storage = { path = "quickwit-storage" } +quickwit-telemetry = { path = "quickwit-telemetry" } quickwit-telemetry-exporters = { path = "quickwit-telemetry-exporters" } quickwit-transport = { path = "quickwit-transport" } diff --git a/quickwit/quickwit-cli/Cargo.toml b/quickwit/quickwit-cli/Cargo.toml index 5257d6f5c85..41733db7b6b 100644 --- a/quickwit/quickwit-cli/Cargo.toml +++ b/quickwit/quickwit-cli/Cargo.toml @@ -65,6 +65,7 @@ quickwit-rest-client = { workspace = true } quickwit-search = { workspace = true } quickwit-serve = { workspace = true } quickwit-storage = { workspace = true } +quickwit-telemetry = { workspace = true } quickwit-transport = { workspace = true } [target.'cfg(target_os = "linux")'.dependencies] diff --git a/quickwit/quickwit-cli/src/lib.rs b/quickwit/quickwit-cli/src/lib.rs index 4cae684979c..30264cacde3 100644 --- a/quickwit/quickwit-cli/src/lib.rs +++ b/quickwit/quickwit-cli/src/lib.rs @@ -56,6 +56,7 @@ pub mod service; pub mod source; pub mod split; pub mod stats; +mod telemetry; pub mod tool; /// Throughput calculation window size. diff --git a/quickwit/quickwit-cli/src/main.rs b/quickwit/quickwit-cli/src/main.rs index 0013aab4314..4d8b406cd7a 100644 --- a/quickwit/quickwit-cli/src/main.rs +++ b/quickwit/quickwit-cli/src/main.rs @@ -64,8 +64,7 @@ fn main() -> anyhow::Result<()> { } fn parse_cli_command() -> (CliCommand, bool) { - let about_text = "Sub-second search & analytics engine on cloud storage.\n Find more \ - information at https://quickwit.io/docs\n\n"; + let about_text = about_text(); let version_text = BuildInfo::get_version_text(); let app = build_cli().about(about_text).version(version_text); @@ -151,6 +150,17 @@ async fn main_impl() -> anyhow::Result<()> { std::process::exit(return_code) } +/// Return the about text with telemetry info. +fn about_text() -> String { + let mut about_text = String::from( + "Sub-second search & analytics engine on cloud storage.\n Find more information at https://quickwit.io/docs\n\n", + ); + if !quickwit_telemetry::is_telemetry_disabled() { + about_text += "Telemetry: enabled"; + } + about_text +} + #[cfg(test)] mod tests { use std::str::FromStr; diff --git a/quickwit/quickwit-cli/src/service.rs b/quickwit/quickwit-cli/src/service.rs index 04b5762db54..52e06edbca8 100644 --- a/quickwit/quickwit-cli/src/service.rs +++ b/quickwit/quickwit-cli/src/service.rs @@ -25,10 +25,12 @@ use quickwit_common::uri::Uri; use quickwit_config::service::QuickwitService; use quickwit_serve::tcp_listener::DefaultTcpListenerResolver; use quickwit_serve::{BuildInfo, EnvFilterReloadFn, reload_tls_cert, serve_quickwit}; +use quickwit_telemetry::payload::TelemetryEvent; use tokio::signal; use tracing::{debug, info}; use crate::checklist::{BLUE_COLOR, RED_COLOR}; +use crate::telemetry::quickwit_telemetry_info; use crate::{config_cli_arg, get_resolvers, load_node_config, start_actor_runtimes}; pub fn build_run_command() -> Command { @@ -124,6 +126,11 @@ impl RunCliCommand { let (storage_resolver, metastore_resolver) = get_resolvers(&node_config.storage_configs, &node_config.metastore_configs); crate::busy_detector::set_enabled(true); + let telemetry_info = quickwit_telemetry_info(&node_config); + let telemetry_handle_opt = + quickwit_telemetry::start_telemetry_loop(telemetry_info, &node_config.data_dir_path) + .await; + quickwit_telemetry::send_telemetry_event(TelemetryEvent::RunCommand).await; // TODO move in serve quickwit? let runtimes_config = RuntimesConfig::default(); start_actor_runtimes(runtimes_config, &node_config.enabled_services)?; @@ -132,7 +139,7 @@ impl RunCliCommand { }); // Reload TLS certificates on SIGHUP for the lifetime of the process. tokio::spawn(listen_sighup()); - serve_quickwit( + let serve_result = serve_quickwit( node_config, runtimes_config, metastore_resolver, @@ -141,7 +148,16 @@ impl RunCliCommand { shutdown_signal, env_filter_reload_fn, ) - .await?; + .await; + let return_code = match serve_result { + Ok(_) => 0, + Err(_) => 1, + }; + quickwit_telemetry::send_telemetry_event(TelemetryEvent::end_command(return_code)).await; + if let Some(telemetry_handle) = telemetry_handle_opt { + telemetry_handle.terminate_telemetry().await; + } + serve_result?; info!("quickwit successfully terminated"); Ok(()) } diff --git a/quickwit/quickwit-cli/src/telemetry.rs b/quickwit/quickwit-cli/src/telemetry.rs new file mode 100644 index 00000000000..a60c55cf0c6 --- /dev/null +++ b/quickwit/quickwit-cli/src/telemetry.rs @@ -0,0 +1,185 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashSet; + +use quickwit_common::uri::Protocol; +use quickwit_config::service::QuickwitService; +use quickwit_config::{NodeConfig, StorageBackendFlavor, disable_ingest_v1, enable_ingest_v2}; +use quickwit_serve::BuildInfo; +use quickwit_telemetry::payload::{BuildTelemetryInfo, QuickwitFeature, QuickwitTelemetryInfo}; + +pub(crate) fn quickwit_telemetry_info(config: &NodeConfig) -> QuickwitTelemetryInfo { + let services = config + .enabled_services + .iter() + .map(|service| service.to_string()) + .collect(); + let build_info = BuildInfo::get(); + QuickwitTelemetryInfo { + build: BuildTelemetryInfo { + commit: build_info.commit_short_hash.to_string(), + target: build_info.build_target.to_string(), + release: !build_info.version.ends_with("-nightly"), + }, + ..QuickwitTelemetryInfo::new(services, telemetry_features(config)) + } +} + +fn telemetry_features(config: &NodeConfig) -> HashSet { + let mut features = HashSet::new(); + if config.is_service_enabled(QuickwitService::Indexer) + && config.indexer_config.enable_otlp_endpoint + { + features.insert(QuickwitFeature::Otlp); + } + // The metastore URI is only relevant on the nodes that serve the metastore. + let served_metastore_uri_opt = if config.is_service_enabled(QuickwitService::Metastore) { + Some(&config.metastore_uri) + } else if config.is_service_enabled(QuickwitService::MetastoreReadReplica) { + config.metastore_read_replica_uri.as_ref() + } else { + None + }; + if let Some(served_metastore_uri) = served_metastore_uri_opt { + let feature = if served_metastore_uri.protocol() == Protocol::PostgreSQL { + QuickwitFeature::PostgresqMetastore + } else { + QuickwitFeature::FileBackedMetastore + }; + features.insert(feature); + } + let storage_feature_opt = match config.default_index_root_uri.protocol() { + Protocol::Azure => Some(QuickwitFeature::AzureStorage), + Protocol::File => Some(QuickwitFeature::FileStorage), + Protocol::Google => Some(QuickwitFeature::GoogleCloudStorage), + Protocol::S3 => Some(QuickwitFeature::S3Storage), + Protocol::Actor | Protocol::Grpc | Protocol::PostgreSQL | Protocol::Ram => None, + }; + if let Some(storage_feature) = storage_feature_opt { + features.insert(storage_feature); + } + // Same fallback as the storage resolver, so that `QW_S3_ENDPOINT` alone is reported. + let s3_storage_config = config + .storage_configs + .find_s3() + .cloned() + .unwrap_or_default(); + if s3_storage_config.endpoint().is_some() { + features.insert(QuickwitFeature::S3CustomEndpoint); + } + let flavor_feature_opt = match s3_storage_config.flavor { + Some(StorageBackendFlavor::DigitalOcean) => Some(QuickwitFeature::S3FlavorDigitalOcean), + Some(StorageBackendFlavor::Garage) => Some(QuickwitFeature::S3FlavorGarage), + Some(StorageBackendFlavor::Gcs) => Some(QuickwitFeature::S3FlavorGcs), + Some(StorageBackendFlavor::MinIO) => Some(QuickwitFeature::S3FlavorMinio), + None => None, + }; + if let Some(flavor_feature) = flavor_feature_opt { + features.insert(flavor_feature); + } + if config.is_service_enabled(QuickwitService::Searcher) { + if config.jaeger_config.enable_endpoint { + features.insert(QuickwitFeature::Jaeger); + } + if config.searcher_config.lambda.is_some() { + features.insert(QuickwitFeature::LambdaLeafSearch); + } + if config.searcher_config.split_cache.is_some() { + features.insert(QuickwitFeature::SplitCache); + } + } + if config.rest_config.tls_config.is_some() || config.grpc_config.tls_config.is_some() { + features.insert(QuickwitFeature::Tls); + } + if !disable_ingest_v1() { + features.insert(QuickwitFeature::IngestV1); + } + if enable_ingest_v2() { + features.insert(QuickwitFeature::IngestV2); + } + if config.docs_clustering_config.is_some() { + features.insert(QuickwitFeature::DocsClustering); + } + features +} + +#[cfg(test)] +mod tests { + use quickwit_common::uri::Uri; + use quickwit_config::{ + DocsClusteringConfig, LambdaConfig, S3StorageConfig, StorageConfig, StorageConfigs, + }; + + use super::*; + + #[test] + fn test_telemetry_features() { + let mut node_config = NodeConfig::for_test(); + node_config.enabled_services = HashSet::from([QuickwitService::Metastore]); + node_config.indexer_config.enable_otlp_endpoint = true; + node_config.jaeger_config.enable_endpoint = true; + let features = telemetry_features(&node_config); + assert!(!features.contains(&QuickwitFeature::Otlp)); + assert!(!features.contains(&QuickwitFeature::Jaeger)); + + node_config.enabled_services = HashSet::from([QuickwitService::MetastoreReadReplica]); + node_config.metastore_read_replica_uri = + Some(Uri::for_test("postgresql://replica/metastore")); + let features = telemetry_features(&node_config); + assert!(features.contains(&QuickwitFeature::PostgresqMetastore)); + + node_config.enabled_services = HashSet::from([ + QuickwitService::Indexer, + QuickwitService::Metastore, + QuickwitService::Searcher, + ]); + node_config.metastore_uri = Uri::for_test("postgresql://localhost/metastore"); + node_config.default_index_root_uri = Uri::for_test("s3://bucket/indexes"); + node_config.storage_configs = + StorageConfigs::new(vec![StorageConfig::S3(S3StorageConfig { + flavor: Some(StorageBackendFlavor::MinIO), + endpoint: Some("http://minio:9000".to_string()), + ..Default::default() + })]); + node_config.searcher_config.lambda = Some(LambdaConfig::for_test()); + node_config.searcher_config.split_cache = + Some(serde_json::from_value(serde_json::json!({"max_num_bytes": "1GB"})).unwrap()); + node_config.grpc_config.tls_config = Some( + serde_json::from_value(serde_json::json!({ + "cert_path": "/certs/cert.pem", + "key_path": "/certs/key.pem", + })) + .unwrap(), + ); + node_config.docs_clustering_config = Some(DocsClusteringConfig { + policies: Vec::new(), + }); + let features = telemetry_features(&node_config); + for feature in [ + QuickwitFeature::DocsClustering, + QuickwitFeature::Jaeger, + QuickwitFeature::LambdaLeafSearch, + QuickwitFeature::Otlp, + QuickwitFeature::PostgresqMetastore, + QuickwitFeature::S3CustomEndpoint, + QuickwitFeature::S3FlavorMinio, + QuickwitFeature::S3Storage, + QuickwitFeature::SplitCache, + QuickwitFeature::Tls, + ] { + assert!(features.contains(&feature), "{feature:?}"); + } + } +} diff --git a/quickwit/quickwit-serve/Cargo.toml b/quickwit/quickwit-serve/Cargo.toml index 0d554ffc2ee..a09676ac03b 100644 --- a/quickwit/quickwit-serve/Cargo.toml +++ b/quickwit/quickwit-serve/Cargo.toml @@ -76,6 +76,7 @@ quickwit-proto = { workspace = true } quickwit-query = { workspace = true } quickwit-search = { workspace = true } quickwit-storage = { workspace = true } +quickwit-telemetry = { workspace = true } quickwit-telemetry-exporters = { workspace = true } quickwit-transport = { workspace = true } diff --git a/quickwit/quickwit-serve/src/api_usage.rs b/quickwit/quickwit-serve/src/api_usage.rs new file mode 100644 index 00000000000..4574d945b4c --- /dev/null +++ b/quickwit/quickwit-serve/src/api_usage.rs @@ -0,0 +1,152 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use quickwit_telemetry::ApiKind; +use warp::http::Method; + +pub(crate) fn classify_rest_request(method: &Method, path: &str) -> Option { + let api_path = path.strip_prefix("/api/v1/")?; + let api_path = api_path.strip_suffix('/').unwrap_or(api_path); + let (first_segment, rest) = api_path.split_once('/').unwrap_or((api_path, "")); + if first_segment == "_elastic" { + return Some(classify_elasticsearch_request(method, rest)); + } + // Management routes such as `indexes/{index_id}/...` and `templates/{template_id}` put a + // user-chosen ID after the first segment, so only the exact shapes of the counted routes match. + match (method, rest) { + (&Method::GET | &Method::POST, "search") => Some(ApiKind::NativeSearch), + (&Method::POST, "ingest") => Some(ApiKind::NativeIngest), + (&Method::POST, "v1/logs" | "v1/traces") if first_segment == "otlp" => Some(ApiKind::Otlp), + (&Method::POST, "otlp/v1/logs" | "otlp/v1/traces") => Some(ApiKind::Otlp), + (&Method::GET, _) if rest.starts_with("jaeger/api/") => Some(ApiKind::Jaeger), + _ => None, + } +} + +/// Matches the shapes and methods of the search and bulk routes in `elasticsearch_api/filter.rs`. +/// Any other request, including `DELETE _search/scroll` and requests rejected for their method, +/// counts as other. +fn classify_elasticsearch_request(method: &Method, elasticsearch_path: &str) -> ApiKind { + let (first_segment, rest_opt) = match elasticsearch_path.split_once('/') { + Some((first_segment, rest)) => (first_segment, Some(rest)), + None => (elasticsearch_path, None), + }; + match (method, first_segment, rest_opt) { + (&Method::POST | &Method::PUT, "_bulk", None) + | (&Method::POST | &Method::PUT, _, Some("_bulk")) => ApiKind::ElasticsearchIngest, + (&Method::GET | &Method::POST, "_search", None | Some("scroll")) + | (&Method::GET | &Method::POST, _, Some("_search" | "_count")) + | (&Method::POST, "_msearch", None) => ApiKind::ElasticsearchSearch, + _ => ApiKind::ElasticsearchOther, + } +} + +pub(crate) fn count_otlp_grpc_request( + request: tonic::Request<()>, +) -> Result, tonic::Status> { + quickwit_telemetry::record_api_request(ApiKind::Otlp); + Ok(request) +} + +pub(crate) fn count_jaeger_grpc_request( + request: tonic::Request<()>, +) -> Result, tonic::Status> { + quickwit_telemetry::record_api_request(ApiKind::Jaeger); + Ok(request) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_classify_rest_request() { + let cases = [ + ( + Method::POST, + "/api/v1/_elastic/logs/_bulk", + Some(ApiKind::ElasticsearchIngest), + ), + ( + Method::POST, + "/api/v1/_elastic/_msearch", + Some(ApiKind::ElasticsearchSearch), + ), + ( + Method::GET, + "/api/v1/_elastic/_field_caps", + Some(ApiKind::ElasticsearchOther), + ), + ( + Method::GET, + "/api/v1/_elastic/_bulk", + Some(ApiKind::ElasticsearchOther), + ), + ( + Method::GET, + "/api/v1/_elastic/_search/scroll", + Some(ApiKind::ElasticsearchSearch), + ), + ( + Method::DELETE, + "/api/v1/_elastic/_search/scroll", + Some(ApiKind::ElasticsearchOther), + ), + ( + Method::POST, + "/api/v1/_elastic/logs/_msearch", + Some(ApiKind::ElasticsearchOther), + ), + ( + Method::GET, + "/api/v1/logs/search", + Some(ApiKind::NativeSearch), + ), + ( + Method::POST, + "/api/v1/logs/ingest", + Some(ApiKind::NativeIngest), + ), + (Method::POST, "/api/v1/otlp/v1/logs", Some(ApiKind::Otlp)), + ( + Method::POST, + "/api/v1/logs/otlp/v1/traces", + Some(ApiKind::Otlp), + ), + ( + Method::GET, + "/api/v1/otlp/search", + Some(ApiKind::NativeSearch), + ), + ( + Method::GET, + "/api/v1/traces/jaeger/api/services", + Some(ApiKind::Jaeger), + ), + (Method::GET, "/api/v1/indexes", None), + (Method::GET, "/api/v1/indexes/otlp/splits", None), + (Method::DELETE, "/api/v1/indexes/jaeger", None), + (Method::PUT, "/api/v1/templates/ingest", None), + (Method::GET, "/api/v1/indexes/search/describe", None), + (Method::GET, "/ui/search", None), + ]; + for (method, path, expected_api_kind) in cases { + assert_eq!( + classify_rest_request(&method, path), + expected_api_kind, + "{method} {path}" + ); + } + } +} diff --git a/quickwit/quickwit-serve/src/grpc.rs b/quickwit/quickwit-serve/src/grpc.rs index ae33f21cde0..8435d0576f9 100644 --- a/quickwit/quickwit-serve/src/grpc.rs +++ b/quickwit/quickwit-serve/src/grpc.rs @@ -32,12 +32,14 @@ use quickwit_proto::tonic::transport::Server; use quickwit_proto::tonic::transport::server::TcpIncoming; use tokio::net::TcpListener; use tokio_rustls::TlsAcceptor; +use tonic::service::interceptor::InterceptedService; use tonic_health::pb::FILE_DESCRIPTOR_SET as HEALTH_FILE_DESCRIPTOR_SET; use tonic_health::pb::health_server::{Health, HealthServer}; use tonic_reflection::pb::v1::FILE_DESCRIPTOR_SET as REFLECTION_FILE_DESCRIPTOR_SET; use tonic_reflection::server::v1::{ServerReflection, ServerReflectionServer}; use tracing::*; +use crate::api_usage::{count_jaeger_grpc_request, count_otlp_grpc_request}; use crate::developer_api::DeveloperApiServer; use crate::search_api::GrpcSearchAdapter; use crate::{INDEXING_GRPC_SERVER_METRICS_LAYER, QuickwitServices}; @@ -175,7 +177,10 @@ pub(crate) async fn start_grpc_server( .accept_compressed(CompressionEncoding::Zstd) .max_decoding_message_size(grpc_config.max_message_size.0 as usize) .max_encoding_message_size(grpc_config.max_message_size.0 as usize); - Some(trace_service) + Some(InterceptedService::new( + trace_service, + count_otlp_grpc_request, + )) } else { None }; @@ -187,7 +192,10 @@ pub(crate) async fn start_grpc_server( .accept_compressed(CompressionEncoding::Zstd) .max_decoding_message_size(grpc_config.max_message_size.0 as usize) .max_encoding_message_size(grpc_config.max_message_size.0 as usize); - Some(logs_service) + Some(InterceptedService::new( + logs_service, + count_otlp_grpc_request, + )) } else { None }; @@ -213,7 +221,10 @@ pub(crate) async fn start_grpc_server( // Mount gRPC jaeger service if present. let jaeger_grpc_service = if let Some(jaeger_service) = services.jaeger_service_opt.clone() { enabled_grpc_services.insert("jaeger"); - Some(SpanReaderPluginServer::new(jaeger_service)) + Some(InterceptedService::new( + SpanReaderPluginServer::new(jaeger_service), + count_jaeger_grpc_request, + )) } else { None }; @@ -221,7 +232,10 @@ pub(crate) async fn start_grpc_server( // Mount gRPC jaeger v2 service (TraceReader) if present. let jaeger_v2_grpc_service = if let Some(jaeger_service) = services.jaeger_service_opt.clone() { enabled_grpc_services.insert("jaeger-v2"); - Some(TraceReaderServer::new(jaeger_service)) + Some(InterceptedService::new( + TraceReaderServer::new(jaeger_service), + count_jaeger_grpc_request, + )) } else { None }; diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index 3fcbb2d5593..a2437cbc102 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -14,6 +14,7 @@ #![recursion_limit = "256"] +mod api_usage; mod build_info; mod cluster_api; mod decompression; diff --git a/quickwit/quickwit-serve/src/rest.rs b/quickwit/quickwit-serve/src/rest.rs index 2ddcfc6c73a..e9976c8ff66 100644 --- a/quickwit/quickwit-serve/src/rest.rs +++ b/quickwit/quickwit-serve/src/rest.rs @@ -27,6 +27,7 @@ use quickwit_common::tower::BoxFutureInfaillible; use quickwit_config::{disable_ingest_v1, enable_ingest_v2}; use quickwit_metrics::{counter, histogram, labels}; use quickwit_search::SearchService; +use quickwit_telemetry::ApiKind; use tokio::net::{TcpListener, TcpStream}; use tokio::task::JoinSet; use tokio_rustls::TlsAcceptor; @@ -44,6 +45,7 @@ use warp::hyper::http::HeaderValue; use warp::hyper::{Method, StatusCode, http}; use warp::{Filter, Rejection, Reply, redirect}; +use crate::api_usage::classify_rest_request; use crate::cluster_api::cluster_handler; use crate::decompression::{CorruptedData, UnsupportedEncoding}; use crate::delete_task_api::delete_task_api_handlers; @@ -147,7 +149,9 @@ pub(crate) async fn start_rest_server( readiness_trigger: BoxFutureInfaillible<()>, shutdown_signal: BoxFutureInfaillible<()>, ) -> anyhow::Result<()> { - let request_counter = warp::log::custom(|info: Info| { + let is_otlp_enabled = quickwit_services.otlp_logs_service_opt.is_some(); + let is_jaeger_enabled = quickwit_services.jaeger_service_opt.is_some(); + let request_counter = warp::log::custom(move |info: Info| { let elapsed = info.elapsed(); let labels = labels!( "method" => info.method().as_str().to_string(), @@ -163,6 +167,15 @@ pub(crate) async fn start_rest_server( labels: [labels], ) .inc(); + if !quickwit_telemetry::is_telemetry_enabled() { + return; + } + match classify_rest_request(info.method(), info.path()) { + Some(ApiKind::Otlp) if !is_otlp_enabled => {} + Some(ApiKind::Jaeger) if !is_jaeger_enabled => {} + Some(api_kind) => quickwit_telemetry::record_api_request(api_kind), + None => {} + } }); // Docs routes let api_doc = warp::path("openapi.json") diff --git a/quickwit/quickwit-serve/src/ui_handler.rs b/quickwit/quickwit-serve/src/ui_handler.rs index f00e95238c2..0c9ded6b887 100644 --- a/quickwit/quickwit-serve/src/ui_handler.rs +++ b/quickwit/quickwit-serve/src/ui_handler.rs @@ -14,6 +14,7 @@ use std::sync::LazyLock; +use quickwit_telemetry::payload::TelemetryEvent; use regex::Regex; use rust_embed::RustEmbed; use warp::hyper::header::HeaderValue; @@ -47,14 +48,20 @@ async fn serve_file(path: Tail) -> Result { async fn serve_impl(path: &str) -> Result, Rejection> { static PATH_PTN: LazyLock = LazyLock::new(|| Regex::new(PATH_PATTERN).unwrap()); - let path_to_file = if PATH_PTN.is_match(path) { - path - } else { - // Quickwit UI is a single page application. - // Any path request that is not an asset should serve the `index.html` file. + // Quickwit UI is a single page application. + // Any path request that is not an asset should serve the `index.html` file. + let is_index_page = !PATH_PTN.is_match(path); + let path_to_file = if is_index_page { UI_INDEX_FILE_NAME + } else { + path }; let asset = Asset::get(path_to_file).ok_or_else(warp::reject::not_found)?; + if is_index_page { + // The client (browser) usually request `index.html` once unless the user refreshes the + // page. + quickwit_telemetry::send_telemetry_event(TelemetryEvent::UiIndexPageLoad).await; + } let mime = mime_guess::from_path(path_to_file).first_or_octet_stream(); let mut res = Response::new(asset.data.into_owned().into()); diff --git a/quickwit/quickwit-telemetry/Cargo.toml b/quickwit/quickwit-telemetry/Cargo.toml new file mode 100644 index 00000000000..96788f76ad5 --- /dev/null +++ b/quickwit/quickwit-telemetry/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "quickwit-telemetry" +description = "Anonymous usage telemetry" + +version.workspace = true +edition.workspace = true +homepage.workspace = true +documentation.workspace = true +repository.workspace = true +authors.workspace = true +license.workspace = true + +[dependencies] +async-trait = { workspace = true } +enum-iterator = { workspace = true } +hostname = { workspace = true } +md5 = { workspace = true } +reqwest = { workspace = true } +serde = { workspace = true } +sysinfo = { workspace = true, features = ["system"] } +tokio = { workspace = true } +tracing = { workspace = true } +uuid = { workspace = true } + +quickwit-common = { workspace = true } + +[dev-dependencies] +serde_json = { workspace = true } +tempfile = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } diff --git a/quickwit/quickwit-telemetry/src/clock.rs b/quickwit/quickwit-telemetry/src/clock.rs new file mode 100644 index 00000000000..7ce60618eb5 --- /dev/null +++ b/quickwit/quickwit-telemetry/src/clock.rs @@ -0,0 +1,56 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::time::Duration; + +#[cfg(test)] +use tokio::sync::Mutex; +#[cfg(test)] +use tokio::sync::mpsc::{Receiver, Sender}; + +#[cfg(test)] +pub(crate) struct ClockButton(Sender<()>); + +#[cfg(test)] +impl ClockButton { + pub(crate) async fn tick(&self) { + let _ = self.0.send(()).await; + } +} + +pub(crate) enum Clock { + /// Each tick waits for the cooldown. The telemetry loop ticks right after each request, so + /// requests are at least a cooldown apart. + Cooldown(Duration), + #[cfg(test)] + Manual(Mutex>), +} + +impl Clock { + #[cfg(test)] + pub(crate) fn manual() -> (ClockButton, Clock) { + let (tx, rx) = tokio::sync::mpsc::channel(1); + (ClockButton(tx), Clock::Manual(Mutex::new(rx))) + } + + pub(crate) async fn tick(&self) { + match self { + Clock::Cooldown(cooldown) => tokio::time::sleep(*cooldown).await, + #[cfg(test)] + Clock::Manual(channel) => { + channel.lock().await.recv().await; + } + } + } +} diff --git a/quickwit/quickwit-telemetry/src/install.rs b/quickwit/quickwit-telemetry/src/install.rs new file mode 100644 index 00000000000..460199b5a99 --- /dev/null +++ b/quickwit/quickwit-telemetry/src/install.rs @@ -0,0 +1,84 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::fs::File; +use std::io::{self, Write}; +use std::path::Path; + +use uuid::Uuid; + +const INSTALL_ID_FILE_NAME: &str = "telemetry_install_id"; +const INSTALL_ID_TEMP_FILE_NAME: &str = "telemetry_install_id.tmp"; + +/// Returns the install ID stored in `data_dir`, creating it if needed, and whether it was just +/// created. Returns an error if the stored install ID is invalid. +pub(crate) fn load_or_create_install_id(data_dir: &Path) -> io::Result<(Uuid, bool)> { + let install_id_path = data_dir.join(INSTALL_ID_FILE_NAME); + match std::fs::read_to_string(&install_id_path) { + Ok(install_id_str) => { + let install_id = Uuid::parse_str(install_id_str.trim()).map_err(|error| { + io::Error::new( + io::ErrorKind::InvalidData, + format!( + "invalid install ID in `{}`: {error}", + install_id_path.display() + ), + ) + })?; + return Ok((install_id, false)); + } + Err(error) if error.kind() == io::ErrorKind::NotFound => {} + Err(error) => return Err(error), + } + let install_id = Uuid::new_v4(); + std::fs::create_dir_all(data_dir)?; + let temp_install_id_path = data_dir.join(INSTALL_ID_TEMP_FILE_NAME); + let mut temp_install_id_file = File::create(&temp_install_id_path)?; + temp_install_id_file.write_all(install_id.to_string().as_bytes())?; + temp_install_id_file.sync_all()?; + std::fs::rename(&temp_install_id_path, &install_id_path)?; + Ok((install_id, true)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_load_or_create_install_id() { + let temp_dir = tempfile::tempdir().unwrap(); + let data_dir = temp_dir.path().join("qwdata"); + + let (install_id, created) = load_or_create_install_id(&data_dir).unwrap(); + assert!(created); + + let (reloaded_install_id, created) = load_or_create_install_id(&data_dir).unwrap(); + assert!(!created); + assert_eq!(reloaded_install_id, install_id); + } + + #[test] + fn test_load_or_create_install_id_rejects_invalid_install_id() { + let temp_dir = tempfile::tempdir().unwrap(); + let install_id_path = temp_dir.path().join(INSTALL_ID_FILE_NAME); + std::fs::write(&install_id_path, "not-a-uuid").unwrap(); + + let error = load_or_create_install_id(temp_dir.path()).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); + assert_eq!( + std::fs::read_to_string(&install_id_path).unwrap(), + "not-a-uuid" + ); + } +} diff --git a/quickwit/quickwit-telemetry/src/lib.rs b/quickwit/quickwit-telemetry/src/lib.rs new file mode 100644 index 00000000000..dd0d3421529 --- /dev/null +++ b/quickwit/quickwit-telemetry/src/lib.rs @@ -0,0 +1,161 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! This crate contains the code responsible for sending anonymous usage data to the telemetry +//! relay. See `docs/telemetry.md` for the data that is sent. + +#![deny(clippy::disallowed_methods)] + +mod clock; +mod install; +mod machine; +pub mod payload; +mod sender; +pub(crate) mod sink; +mod usage; + +use std::ffi::OsStr; +use std::path::Path; +use std::sync::OnceLock; + +use payload::QuickwitTelemetryInfo; +use tracing::{info, warn}; + +use crate::install::load_or_create_install_id; +use crate::payload::TelemetryEvent; +use crate::sender::{TelemetryLoopHandle, TelemetrySender}; +use crate::sink::HttpClient; +pub use crate::usage::{ApiKind, record_api_request}; + +static TELEMETRY_SENDER: OnceLock = OnceLock::new(); + +/// Starts the telemetry loop, and queues the `install` event if the install ID stored in +/// `data_dir` was just created. Returns `None` if telemetry is disabled, if the HTTP client can't +/// be created, or if the loop is already started. +pub async fn start_telemetry_loop( + mut quickwit_info: QuickwitTelemetryInfo, + data_dir: &Path, +) -> Option { + if is_telemetry_disabled() { + info!("telemetry is disabled"); + return None; + } + let http_client = match HttpClient::try_new() { + Ok(http_client) => http_client, + Err(error) => { + warn!( + ?error, + "failed to create the telemetry HTTP client, telemetry is disabled" + ); + return None; + } + }; + let mut is_new_install_opt = None; + let telemetry_sender = TELEMETRY_SENDER.get_or_init(|| { + // Only created once telemetry can be sent, so that a node with telemetry off never writes + // the install ID. + let is_new_install = match load_or_create_install_id(data_dir) { + Ok((install_uuid, is_new_install)) => { + quickwit_info.install_uuid = Some(install_uuid); + is_new_install + } + Err(error) => { + warn!(%error, "failed to load or create the telemetry install ID"); + false + } + }; + is_new_install_opt = Some(is_new_install); + info!("telemetry to {} is enabled", http_client.endpoint()); + TelemetrySender::with_http_client(quickwit_info, http_client) + }); + let Some(is_new_install) = is_new_install_opt else { + // This should not happen... unless telemetry is enabled and you are running tests in + // parallel in the same process. + info!("telemetry loop already started. please disable telemetry during tests"); + return None; + }; + let telemetry_loop_handle = telemetry_sender.start_loop(); + if is_new_install { + telemetry_sender.send(TelemetryEvent::Install).await; + } + Some(telemetry_loop_handle) +} + +/// Queues a telemetry event for the telemetry loop to send to the telemetry relay. Does nothing +/// unless `start_telemetry_loop` started the loop. +/// +/// Telemetry sends at most 1 request per minute, except for one last request with the pending +/// events when the node shuts down. Each request holds at most 10 events, so some events are +/// dropped if this function is called too often. +/// +/// If a request fails, the error is only logged at the debug level. +/// +/// The event is an enum so that readers can audit what information is sent. +pub async fn send_telemetry_event(event: TelemetryEvent) { + if let Some(telemetry_sender) = TELEMETRY_SENDER.get() { + telemetry_sender.send(event).await; + } +} + +/// Returns true once `start_telemetry_loop` has started the telemetry loop. It is false before +/// that, so check it on each use instead of caching it. +pub fn is_telemetry_enabled() -> bool { + TELEMETRY_SENDER.get().is_some() +} + +/// This environment variable can be set to disable sending telemetry events. +pub const DISABLE_TELEMETRY_ENV_KEY: &str = "QW_DISABLE_TELEMETRY"; + +/// Returns true if `QW_DISABLE_TELEMETRY` disables telemetry. +pub fn is_telemetry_disabled() -> bool { + let env_value_opt = std::env::var_os(DISABLE_TELEMETRY_ENV_KEY); + is_telemetry_disabled_by_env_value(env_value_opt.as_deref()) +} + +/// Telemetry is disabled when `QW_DISABLE_TELEMETRY` is set to any non-empty value other than +/// `false`, `no` or `0`. Unrecognized values disable telemetry so that a mistyped opt-out never +/// leaves telemetry on. +fn is_telemetry_disabled_by_env_value(env_value_opt: Option<&OsStr>) -> bool { + let Some(env_value) = env_value_opt else { + return false; + }; + let Some(env_value_str) = env_value.to_str() else { + return true; + }; + if env_value_str.is_empty() { + return false; + } + quickwit_common::parse_bool_lenient(env_value_str) != Some(false) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_is_telemetry_disabled_by_env_value() { + assert!(!is_telemetry_disabled_by_env_value(None)); + assert!(!is_telemetry_disabled_by_env_value(Some(OsStr::new("")))); + for enabled_value in ["false", "FALSE", "no", "0"] { + assert!(!is_telemetry_disabled_by_env_value(Some(OsStr::new( + enabled_value + )))); + } + for disabled_value in [" ", "1", "true", "yes", "on", "disabled", "anything"] { + assert!(is_telemetry_disabled_by_env_value(Some(OsStr::new( + disabled_value + )))); + } + } +} diff --git a/quickwit/quickwit-telemetry/src/machine.rs b/quickwit/quickwit-telemetry/src/machine.rs new file mode 100644 index 00000000000..37e926ba320 --- /dev/null +++ b/quickwit/quickwit-telemetry/src/machine.rs @@ -0,0 +1,67 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use sysinfo::{MemoryRefreshKind, RefreshKind, System}; + +const GIB: u64 = 1 << 30; + +pub(crate) fn cpu_count_bucket(num_cpus: usize) -> &'static str { + match num_cpus { + 0 | 1 => "1", + 2 => "2", + 3..=4 => "3-4", + 5..=8 => "5-8", + 9..=16 => "9-16", + 17..=32 => "17-32", + 33..=64 => "33-64", + _ => "65+", + } +} + +pub(crate) fn memory_bucket(num_bytes: u64) -> &'static str { + match num_bytes { + num_bytes if num_bytes < 2 * GIB => "<2GiB", + num_bytes if num_bytes < 4 * GIB => "2-4GiB", + num_bytes if num_bytes < 8 * GIB => "4-8GiB", + num_bytes if num_bytes < 16 * GIB => "8-16GiB", + num_bytes if num_bytes < 32 * GIB => "16-32GiB", + num_bytes if num_bytes < 64 * GIB => "32-64GiB", + num_bytes if num_bytes < 128 * GIB => "64-128GiB", + _ => "128GiB+", + } +} + +pub(crate) fn available_memory_num_bytes() -> u64 { + let system = System::new_with_specifics( + RefreshKind::nothing().with_memory(MemoryRefreshKind::nothing().with_ram()), + ); + match system.cgroup_limits() { + Some(cgroup_limits) => cgroup_limits.total_memory, + None => system.total_memory(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_buckets() { + assert_eq!(cpu_count_bucket(4), "3-4"); + assert_eq!(cpu_count_bucket(96), "65+"); + assert_eq!(memory_bucket(GIB), "<2GiB"); + assert_eq!(memory_bucket(31 * GIB), "16-32GiB"); + assert_eq!(memory_bucket(512 * GIB), "128GiB+"); + } +} diff --git a/quickwit/quickwit-telemetry/src/payload.rs b/quickwit/quickwit-telemetry/src/payload.rs new file mode 100644 index 00000000000..551c00ad122 --- /dev/null +++ b/quickwit/quickwit-telemetry/src/payload.rs @@ -0,0 +1,266 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::{BTreeMap, HashSet}; +use std::env; +use std::time::UNIX_EPOCH; + +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::machine::{available_memory_num_bytes, cpu_count_bucket, memory_bucket}; +use crate::usage::ApiKind; + +/// Version of the payload format. Payloads without one are version 1. +pub(crate) const TELEMETRY_SCHEMA_VERSION: u32 = 2; + +/// Represents the payload of the request sent with telemetry requests. +#[derive(Debug, Serialize, Deserialize)] +pub struct TelemetryPayload { + pub schema_version: u32, + pub client_info: ClientInfo, + pub quickwit_info: QuickwitTelemetryInfo, + pub events: Vec, + /// Number of events dropped since the previous request because the queue was full. + pub num_dropped_events: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct EventWithTimestamp { + /// Unix time in seconds. + pub unixtime: u64, + #[serde(flatten)] + pub event: TelemetryEvent, +} + +/// Returns the number of seconds elapsed since UNIX_EPOCH. +/// +/// If the system clock is set before 1970, returns 0. +fn unixtime() -> u64 { + match UNIX_EPOCH.elapsed() { + Ok(duration) => duration.as_secs(), + Err(_) => 0u64, + } +} + +impl From for EventWithTimestamp { + fn from(event: TelemetryEvent) -> Self { + EventWithTimestamp { + unixtime: unixtime(), + event, + } + } +} + +/// Represents a telemetry event sent to the telemetry relay for usage information. +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "type")] +#[serde(rename_all = "snake_case")] +pub enum TelemetryEvent { + RunCommand, + /// Event sent when the node stops, with its exit code. + EndCommand { + return_code: i32, + #[serde(default, skip_serializing_if = "Option::is_none")] + api_usage: Option>, + }, + /// Event sent every 12h to signal the server is running. + Running { + #[serde(default, skip_serializing_if = "Option::is_none")] + api_usage: Option>, + }, + /// UI index.html was requested. + UiIndexPageLoad, + /// The install ID was just created. + Install, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ClientInfo { + session_uuid: uuid::Uuid, + os: String, + arch: String, + hashed_host_username: String, + kubernetes: bool, + cpus: String, + memory: String, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct BuildTelemetryInfo { + pub commit: String, + pub target: String, + pub release: bool, +} + +impl Default for BuildTelemetryInfo { + fn default() -> Self { + Self { + commit: "unknown".to_string(), + target: "unknown".to_string(), + release: false, + } + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct QuickwitTelemetryInfo { + pub version: String, + pub services: HashSet, + pub features: HashSet, + pub build: BuildTelemetryInfo, + #[serde(skip_serializing_if = "Option::is_none")] + pub install_uuid: Option, +} + +impl QuickwitTelemetryInfo { + pub fn new(services: HashSet, features: HashSet) -> Self { + Self { + features, + services, + ..Default::default() + } + } +} + +impl Default for QuickwitTelemetryInfo { + fn default() -> Self { + Self { + features: HashSet::new(), + version: env!("CARGO_PKG_VERSION").to_string(), + services: HashSet::new(), + build: BuildTelemetryInfo::default(), + install_uuid: None, + } + } +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, Hash)] +#[serde(rename_all = "snake_case")] +pub enum QuickwitFeature { + AzureStorage, + DocsClustering, + FileBackedMetastore, + FileStorage, + GoogleCloudStorage, + IngestV1, + IngestV2, + Jaeger, + LambdaLeafSearch, + Otlp, + PostgresqMetastore, + S3CustomEndpoint, + S3FlavorDigitalOcean, + S3FlavorGarage, + S3FlavorGcs, + S3FlavorMinio, + S3Storage, + SplitCache, + Tls, +} + +fn hashed_host_username() -> String { + let hostname = hostname::get() + .map(|hostname| hostname.to_string_lossy().to_string()) + .unwrap_or_default(); + // What the `username` crate used by releases 0.2 to 0.8 returns on Linux and macOS, so that the + // hash stays the same for nodes that upgrade. + let username = env::var("LOGNAME").unwrap_or_default(); + let hashed_value = format!("{hostname}:{username}"); + let digest = md5::compute(hashed_value.as_bytes()); + format!("{digest:x}") +} + +impl Default for ClientInfo { + fn default() -> ClientInfo { + ClientInfo { + session_uuid: Uuid::new_v4(), + os: env::consts::OS.to_string(), + arch: env::consts::ARCH.to_string(), + hashed_host_username: hashed_host_username(), + kubernetes: std::env::var_os("KUBERNETES_SERVICE_HOST").is_some(), + cpus: cpu_count_bucket(quickwit_common::num_cpus()).to_string(), + memory: memory_bucket(available_memory_num_bytes()).to_string(), + } + } +} + +#[cfg(test)] +mod tests { + use serde_json; + + use super::*; + + #[test] + fn test_serialize_features_as_expected() { + // These names are part of the telemetry payload documented in `docs/telemetry.md`. + let features = [ + QuickwitFeature::AzureStorage, + QuickwitFeature::DocsClustering, + QuickwitFeature::FileBackedMetastore, + QuickwitFeature::FileStorage, + QuickwitFeature::GoogleCloudStorage, + QuickwitFeature::IngestV1, + QuickwitFeature::IngestV2, + QuickwitFeature::Jaeger, + QuickwitFeature::LambdaLeafSearch, + QuickwitFeature::Otlp, + QuickwitFeature::PostgresqMetastore, + QuickwitFeature::S3CustomEndpoint, + QuickwitFeature::S3FlavorDigitalOcean, + QuickwitFeature::S3FlavorGarage, + QuickwitFeature::S3FlavorGcs, + QuickwitFeature::S3FlavorMinio, + QuickwitFeature::S3Storage, + QuickwitFeature::SplitCache, + QuickwitFeature::Tls, + ]; + let json = serde_json::to_string(&features).unwrap(); + assert_eq!( + json, + r#"["azure_storage","docs_clustering","file_backed_metastore","file_storage","google_cloud_storage","ingest_v1","ingest_v2","jaeger","lambda_leaf_search","otlp","postgresq_metastore","s3_custom_endpoint","s3_flavor_digital_ocean","s3_flavor_garage","s3_flavor_gcs","s3_flavor_minio","s3_storage","split_cache","tls"]"# + ); + } + + #[test] + fn test_serialize_payload_as_expected() { + let event = EventWithTimestamp { + unixtime: 0, + event: TelemetryEvent::EndCommand { + return_code: 0, + api_usage: None, + }, + }; + let json = serde_json::to_string(&event).unwrap(); + assert_eq!( + json, + r#"{"unixtime":0,"type":"end_command","return_code":0}"# + ); + } + + #[test] + fn test_serialize_activity_as_expected() { + let event = EventWithTimestamp { + unixtime: 0, + event: TelemetryEvent::Running { + api_usage: Some(BTreeMap::from([(ApiKind::NativeSearch, "1-9".to_string())])), + }, + }; + let json = serde_json::to_string(&event).unwrap(); + assert_eq!( + json, + r#"{"unixtime":0,"type":"running","api_usage":{"native_search":"1-9"}}"# + ); + } +} diff --git a/quickwit/quickwit-telemetry/src/sender.rs b/quickwit/quickwit-telemetry/src/sender.rs new file mode 100644 index 00000000000..ae1a33a2599 --- /dev/null +++ b/quickwit/quickwit-telemetry/src/sender.rs @@ -0,0 +1,361 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::mem; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::mpsc::{Receiver, Sender}; +use tokio::sync::{RwLock, oneshot}; +use tokio::task::JoinHandle; + +use crate::clock::Clock; +use crate::payload::{ + ClientInfo, EventWithTimestamp, QuickwitTelemetryInfo, TELEMETRY_SCHEMA_VERSION, + TelemetryEvent, TelemetryPayload, +}; +use crate::sink::{HttpClient, Sink}; + +/// At most 1 request per minute, except for the last request on termination. +const TELEMETRY_PUSH_COOLDOWN: Duration = Duration::from_secs(60); + +/// Interval at which to send telemetry `Running` event. +const TELEMETRY_RUNNING_EVENT_INTERVAL: Duration = + Duration::from_secs(if cfg!(test) { 3 } else { 60 * 60 * 12 }); + +/// Upon termination of the program, we send one last telemetry request with pending events. +/// This duration is the amount of time we wait for at most to send that last telemetry request. +const LAST_REQUEST_TIMEOUT: Duration = Duration::from_secs(1); + +const MAX_NUM_EVENTS_IN_QUEUE: usize = 10; + +#[derive(Default)] +struct EventsState { + events: Vec, + num_dropped_events: usize, +} + +impl EventsState { + /// Returns true if the event was queued into an empty queue. + fn push_event(&mut self, mut event: TelemetryEvent) -> bool { + if self.events.len() >= MAX_NUM_EVENTS_IN_QUEUE { + self.num_dropped_events += 1; + // UI page loads are the only events that can fill the queue. Any other event, such as + // a summary carrying the API counts, replaces the latest one instead of being dropped. + if matches!(event, TelemetryEvent::UiIndexPageLoad) { + return false; + } + let Some(ui_page_load_idx) = self + .events + .iter() + .rposition(|queued| matches!(queued.event, TelemetryEvent::UiIndexPageLoad)) + else { + return false; + }; + self.events.remove(ui_page_load_idx); + } + event.add_usage(); + let events_was_empty = self.events.is_empty(); + self.events.push(EventWithTimestamp::from(event)); + events_was_empty + } +} + +struct Events { + state: RwLock, + /// Wakes the telemetry loop up when an event is queued into an empty queue. + items_available_tx: Sender<()>, + items_available_rx: RwLock>, +} + +impl Default for Events { + fn default() -> Self { + let (items_available_tx, items_available_rx) = tokio::sync::mpsc::channel(1); + Events { + state: RwLock::new(EventsState::default()), + items_available_tx, + items_available_rx: RwLock::new(items_available_rx), + } + } +} + +impl Events { + async fn wait_for_events(&self) { + self.items_available_rx.write().await.recv().await; + } + + async fn take_events(&self) -> EventsState { + mem::take(&mut *self.state.write().await) + } + + async fn push_event(&self, event: TelemetryEvent) { + let is_first_event = self.state.write().await.push_event(event); + if is_first_event { + // A full channel means that a wake-up is already pending. + let _ = self.items_available_tx.try_send(()); + } + } +} + +struct Inner { + sink: Box, + client_info: ClientInfo, + quickwit_info: QuickwitTelemetryInfo, + events: Events, + clock: Clock, +} + +impl Inner { + /// Sends the queued events, if any. + async fn send_queued_events(&self) { + let events_state = self.events.take_events().await; + if events_state.events.is_empty() { + return; + } + let payload = TelemetryPayload { + schema_version: TELEMETRY_SCHEMA_VERSION, + client_info: self.client_info.clone(), + quickwit_info: self.quickwit_info.clone(), + events: events_state.events, + num_dropped_events: events_state.num_dropped_events, + }; + self.sink.send_payload(payload).await; + } +} + +pub struct TelemetrySender { + inner: Arc, +} + +pub struct TelemetryLoopHandle { + join_handle: JoinHandle<()>, + terminate_command_tx: oneshot::Sender<()>, +} + +impl TelemetryLoopHandle { + /// Stops the telemetry loop, which sends the queued events without waiting for the cooldown, + /// and waits for it at most `LAST_REQUEST_TIMEOUT`. + pub async fn terminate_telemetry(self) { + let _ = self.terminate_command_tx.send(()); + let _ = tokio::time::timeout(LAST_REQUEST_TIMEOUT, self.join_handle).await; + } +} + +impl TelemetrySender { + pub(crate) fn with_http_client( + quickwit_info: QuickwitTelemetryInfo, + http_client: HttpClient, + ) -> Self { + Self::new( + quickwit_info, + http_client, + Clock::Cooldown(TELEMETRY_PUSH_COOLDOWN), + ) + } + + fn new(quickwit_info: QuickwitTelemetryInfo, sink: impl Sink, clock: Clock) -> Self { + Self { + inner: Arc::new(Inner { + sink: Box::new(sink), + client_info: ClientInfo::default(), + quickwit_info, + events: Events::default(), + clock, + }), + } + } + + /// Starts the loop that sends the queued events. Must be called at most once. + pub(crate) fn start_loop(&self) -> TelemetryLoopHandle { + let (terminate_command_tx, mut terminate_command_rx) = oneshot::channel(); + let inner = self.inner.clone(); + start_monitor_if_server_running_task(inner.clone()); + let join_handle = tokio::spawn(async move { + loop { + let quit_loop = tokio::select! { + _ = &mut terminate_command_rx => true, + _ = inner.events.wait_for_events() => false, + }; + inner.send_queued_events().await; + if quit_loop { + break; + } + let quit_loop = tokio::select! { + _ = &mut terminate_command_rx => true, + _ = inner.clock.tick() => false, + }; + if quit_loop { + inner.send_queued_events().await; + break; + } + } + }); + TelemetryLoopHandle { + join_handle, + terminate_command_tx, + } + } + + pub async fn send(&self, event: TelemetryEvent) { + self.inner.events.push_event(event).await; + } +} + +fn start_monitor_if_server_running_task(inner: Arc) { + let mut clock = tokio::time::interval(TELEMETRY_RUNNING_EVENT_INTERVAL); + tokio::spawn(async move { + // Drop the first immediate tick. + clock.tick().await; + loop { + clock.tick().await; + inner.events.push_event(TelemetryEvent::running()).await; + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_push_event_replaces_ui_page_load_when_queue_is_full() { + let mut events_state = EventsState::default(); + for _ in 0..=MAX_NUM_EVENTS_IN_QUEUE { + events_state.push_event(TelemetryEvent::UiIndexPageLoad); + } + events_state.push_event(TelemetryEvent::end_command(0)); + + assert_eq!(events_state.events.len(), MAX_NUM_EVENTS_IN_QUEUE); + assert_eq!(events_state.num_dropped_events, 2); + assert!(matches!( + events_state.events.last().unwrap().event, + TelemetryEvent::EndCommand { .. } + )); + } + + #[tokio::test] + async fn test_telemetry_two_events() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (clock_btn, clock) = Clock::manual(); + let telemetry_sender = TelemetrySender::new(QuickwitTelemetryInfo::default(), tx, clock); + let loop_handler = telemetry_sender.start_loop(); + telemetry_sender.send(TelemetryEvent::UiIndexPageLoad).await; + { + let payload = rx.recv().await.unwrap(); + assert_eq!(payload.events.len(), 1); + } + clock_btn.tick().await; + telemetry_sender.send(TelemetryEvent::UiIndexPageLoad).await; + { + let payload = rx.recv().await.unwrap(); + assert_eq!(payload.events.len(), 1); + } + loop_handler.terminate_telemetry().await; + } + + #[tokio::test] + async fn test_telemetry_uptime_events() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (clock_btn, clock) = Clock::manual(); + let telemetry_sender = TelemetrySender::new(QuickwitTelemetryInfo::default(), tx, clock); + let loop_handler = telemetry_sender.start_loop(); + telemetry_sender.send(TelemetryEvent::UiIndexPageLoad).await; + { + let payload = rx.recv().await.unwrap(); + assert_eq!(payload.events.len(), 1); + } + clock_btn.tick().await; + tokio::time::sleep(TELEMETRY_RUNNING_EVENT_INTERVAL + Duration::from_secs(1)).await; + { + let payload = rx.recv().await.unwrap(); + assert_eq!(payload.events.len(), 1); + assert!(matches!( + payload.events[0].event, + TelemetryEvent::Running { .. } + )); + } + loop_handler.terminate_telemetry().await; + } + + #[tokio::test] + async fn test_telemetry_cooldown_observed() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (clock_btn, clock) = Clock::manual(); + let telemetry_sender = TelemetrySender::new(QuickwitTelemetryInfo::default(), tx, clock); + let loop_handler = telemetry_sender.start_loop(); + telemetry_sender.send(TelemetryEvent::UiIndexPageLoad).await; + { + let payload = rx.recv().await.unwrap(); + assert_eq!(payload.events.len(), 1); + } + tokio::task::yield_now().await; + telemetry_sender.send(TelemetryEvent::UiIndexPageLoad).await; + + let timeout_res = tokio::time::timeout(Duration::from_millis(1), rx.recv()).await; + assert!(timeout_res.is_err()); + + telemetry_sender.send(TelemetryEvent::UiIndexPageLoad).await; + clock_btn.tick().await; + { + let payload = rx.recv().await.unwrap(); + assert_eq!(payload.events.len(), 2); + } + loop_handler.terminate_telemetry().await; + } + + #[tokio::test] + async fn test_terminate_telemetry_sends_pending_events() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (_clock_btn, clock) = Clock::manual(); + let telemetry_sender = TelemetrySender::new(QuickwitTelemetryInfo::default(), tx, clock); + let loop_handler = telemetry_sender.start_loop(); + telemetry_sender.send(TelemetryEvent::UiIndexPageLoad).await; + let payload = rx.recv().await.unwrap(); + assert_eq!(payload.events.len(), 1); + telemetry_sender + .send(TelemetryEvent::end_command(2i32)) + .await; + loop_handler.terminate_telemetry().await; + let payload = rx.recv().await.unwrap(); + assert_eq!(payload.events.len(), 1); + assert!(matches!( + &payload.events[0].event, + &TelemetryEvent::EndCommand { .. } + )); + tokio::time::timeout( + Duration::from_secs(1), + telemetry_sender.send(TelemetryEvent::UiIndexPageLoad), + ) + .await + .unwrap(); + } + + #[tokio::test(start_paused = true)] + async fn test_terminate_telemetry_does_not_wait_without_pending_events() { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let (clock_btn, clock) = Clock::manual(); + let telemetry_sender = TelemetrySender::new(QuickwitTelemetryInfo::default(), tx, clock); + let loop_handler = telemetry_sender.start_loop(); + telemetry_sender + .send(TelemetryEvent::end_command(0i32)) + .await; + rx.recv().await.unwrap(); + clock_btn.tick().await; + tokio::task::yield_now().await; + + let terminate_start = tokio::time::Instant::now(); + loop_handler.terminate_telemetry().await; + assert!(terminate_start.elapsed() < LAST_REQUEST_TIMEOUT); + } +} diff --git a/quickwit/quickwit-telemetry/src/sink.rs b/quickwit/quickwit-telemetry/src/sink.rs new file mode 100644 index 00000000000..efaabbe2457 --- /dev/null +++ b/quickwit/quickwit-telemetry/src/sink.rs @@ -0,0 +1,84 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::time::Duration; + +use async_trait::async_trait; +use reqwest::redirect::Policy; +use reqwest::{Client, Response}; +use tokio::sync::mpsc::UnboundedSender; +use tracing::debug; + +use crate::payload::TelemetryPayload; + +const DEFAULT_TELEMETRY_INGEST_API_URL: &str = "https://telemetry.quickwit.io/"; + +fn telemetry_ingest_api_url() -> String { + if let Some(ingest_api_url) = std::env::var_os("TELEMETRY_INGEST_API") { + ingest_api_url.to_string_lossy().to_string() + } else { + DEFAULT_TELEMETRY_INGEST_API_URL.to_string() + } +} + +#[async_trait] +pub trait Sink: Send + Sync + 'static { + async fn send_payload(&self, payload: TelemetryPayload); +} +pub struct HttpClient { + client: Client, + endpoint: String, +} + +impl HttpClient { + pub fn try_new() -> reqwest::Result { + let client = Client::builder() + .redirect(Policy::limited(3)) + .timeout(Duration::from_secs(10)) + .build()?; + Ok(HttpClient { + client, + endpoint: telemetry_ingest_api_url(), + }) + } + + pub fn endpoint(&self) -> &str { + &self.endpoint + } +} + +#[async_trait] +impl Sink for UnboundedSender { + async fn send_payload(&self, payload: TelemetryPayload) { + let _ = self.send(payload); + } +} + +#[async_trait] +impl Sink for HttpClient { + async fn send_payload(&self, payload: TelemetryPayload) { + // Telemetry is best effort, so failures, including payloads that the relay rejects, are + // only logged at the debug level. + let send_result = self + .client + .post(&self.endpoint) + .json(&payload) + .send() + .await + .and_then(Response::error_for_status); + if let Err(error) = send_result { + debug!(?error, "failed to send telemetry request"); + } + } +} diff --git a/quickwit/quickwit-telemetry/src/usage.rs b/quickwit/quickwit-telemetry/src/usage.rs new file mode 100644 index 00000000000..10852a5d67e --- /dev/null +++ b/quickwit/quickwit-telemetry/src/usage.rs @@ -0,0 +1,144 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicU64, Ordering}; + +use enum_iterator::{Sequence, all, cardinality}; +use serde::{Deserialize, Serialize}; + +use crate::payload::TelemetryEvent; + +#[derive( + Clone, Copy, Debug, Eq, PartialEq, Hash, Ord, PartialOrd, Serialize, Deserialize, Sequence, +)] +#[serde(rename_all = "snake_case")] +pub enum ApiKind { + ElasticsearchSearch, + ElasticsearchIngest, + ElasticsearchOther, + NativeSearch, + NativeIngest, + Otlp, + Jaeger, +} + +/// Indexed by `api_kind as usize`. +struct ApiRequestCounts([AtomicU64; cardinality::()]); + +impl ApiRequestCounts { + const fn new() -> Self { + Self([const { AtomicU64::new(0) }; cardinality::()]) + } + + fn record(&self, api_kind: ApiKind) { + self.0[api_kind as usize].fetch_add(1, Ordering::Relaxed); + } + + /// Returns the bucketed counts of the APIs that received requests, and resets the counts. + fn take(&self) -> Option> { + let mut api_usage = BTreeMap::new(); + for api_kind in all::() { + let num_requests = self.0[api_kind as usize].swap(0, Ordering::Relaxed); + if num_requests > 0 { + api_usage.insert(api_kind, request_count_bucket(num_requests).to_string()); + } + } + if api_usage.is_empty() { + return None; + } + Some(api_usage) + } +} + +static API_REQUEST_COUNTS: ApiRequestCounts = ApiRequestCounts::new(); + +pub fn record_api_request(api_kind: ApiKind) { + if crate::is_telemetry_enabled() { + API_REQUEST_COUNTS.record(api_kind); + } +} + +impl TelemetryEvent { + pub fn running() -> Self { + TelemetryEvent::Running { api_usage: None } + } + + pub fn end_command(return_code: i32) -> Self { + TelemetryEvent::EndCommand { + return_code, + api_usage: None, + } + } + + /// Adds the API request counts since the previous report. Only called once the event is + /// queued, so that a dropped event never loses the counts. + pub(crate) fn add_usage(&mut self) { + if let TelemetryEvent::Running { api_usage } + | TelemetryEvent::EndCommand { api_usage, .. } = self + { + *api_usage = API_REQUEST_COUNTS.take(); + } + } +} + +fn request_count_bucket(num_requests: u64) -> &'static str { + match num_requests { + 0 => "0", + 1..=9 => "1-9", + 10..=99 => "10-99", + 100..=999 => "100-999", + 1_000..=9_999 => "1000-9999", + 10_000..=99_999 => "10000-99999", + 100_000..=999_999 => "100000-999999", + _ => "1000000+", + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_api_kind_discriminants_follow_declaration_order() { + for (index, api_kind) in all::().enumerate() { + assert_eq!(api_kind as usize, index); + } + } + + #[test] + fn test_api_request_counts_reports_buckets_and_resets() { + let api_request_counts = ApiRequestCounts::new(); + api_request_counts.record(ApiKind::NativeSearch); + for _ in 0..12 { + api_request_counts.record(ApiKind::ElasticsearchIngest); + } + let api_usage = api_request_counts.take().unwrap(); + assert_eq!( + api_usage, + BTreeMap::from([ + (ApiKind::ElasticsearchIngest, "10-99".to_string()), + (ApiKind::NativeSearch, "1-9".to_string()), + ]) + ); + assert_eq!(api_request_counts.take(), None); + } + + #[test] + fn test_buckets() { + assert_eq!(request_count_bucket(9), "1-9"); + assert_eq!(request_count_bucket(10), "10-99"); + assert_eq!(request_count_bucket(1_000_000), "1000000+"); + } +} diff --git a/quickwit/rest-api-tests/run_tests.py b/quickwit/rest-api-tests/run_tests.py index c1ad6c3b950..e2644163602 100755 --- a/quickwit/rest-api-tests/run_tests.py +++ b/quickwit/rest-api-tests/run_tests.py @@ -363,7 +363,7 @@ def __init__(self, quickwit_bin_path): mkdir(config) shutil.copy("../../config/quickwit.yaml", config) shutil.copy(quickwit_bin_path, self.quickwit_dir.name) - self.proc = subprocess.Popen(["./quickwit", "run"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=self.quickwit_dir.name) + self.proc = subprocess.Popen(["./quickwit", "run"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=self.quickwit_dir.name, env={"QW_DISABLE_TELEMETRY": "1", **os.environ}) for i in range(100): try: print("Checking on quickwit")