common_greptimedb_telemetry/
lib.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::env;
16use std::io::ErrorKind;
17use std::path::{Path, PathBuf};
18use std::sync::atomic::{AtomicBool, Ordering};
19use std::sync::{Arc, LazyLock};
20use std::time::{Duration, SystemTime};
21
22use common_runtime::error::{Error, Result};
23use common_runtime::{BoxedTaskFunction, RepeatedTask, TaskFunction};
24use common_telemetry::{debug, info};
25use common_version::build_info;
26use reqwest::{Client, Response};
27use serde::{Deserialize, Serialize};
28
29/// The URL to report telemetry data.
30pub const TELEMETRY_URL: &str = "https://telemetry.greptimestats.com/db/otel/statistics";
31/// The local installation uuid cache file
32const UUID_FILE_NAME: &str = ".greptimedb-telemetry-uuid";
33
34/// System start time for uptime calculation
35static START_TIME: LazyLock<SystemTime> = LazyLock::new(SystemTime::now);
36
37/// The default interval of reporting telemetry data to greptime cloud
38pub static TELEMETRY_INTERVAL: Duration = Duration::from_secs(60 * 30);
39/// The default connect timeout to greptime cloud.
40const GREPTIMEDB_TELEMETRY_CLIENT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
41/// The default request timeout to greptime cloud.
42const GREPTIMEDB_TELEMETRY_CLIENT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
43
44pub enum GreptimeDBTelemetryTask {
45    Enable((RepeatedTask<Error>, Arc<AtomicBool>)),
46    Disable,
47}
48
49impl GreptimeDBTelemetryTask {
50    pub fn should_report(&self, value: bool) {
51        match self {
52            GreptimeDBTelemetryTask::Enable((_, should_report)) => {
53                should_report.store(value, Ordering::Relaxed);
54            }
55            GreptimeDBTelemetryTask::Disable => {}
56        }
57    }
58
59    pub fn enable(
60        interval: Duration,
61        task_fn: BoxedTaskFunction<Error>,
62        should_report: Arc<AtomicBool>,
63    ) -> Self {
64        GreptimeDBTelemetryTask::Enable((
65            RepeatedTask::new(interval, task_fn).with_initial_delay(Some(Duration::ZERO)),
66            should_report,
67        ))
68    }
69
70    pub fn disable() -> Self {
71        GreptimeDBTelemetryTask::Disable
72    }
73
74    pub fn start(&self) -> Result<()> {
75        match self {
76            GreptimeDBTelemetryTask::Enable((task, _)) => {
77                print_anonymous_usage_data_disclaimer();
78                task.start(common_runtime::global_runtime())
79            }
80            GreptimeDBTelemetryTask::Disable => Ok(()),
81        }
82    }
83
84    pub async fn stop(&self) -> Result<()> {
85        match self {
86            GreptimeDBTelemetryTask::Enable((task, _)) => task.stop().await,
87            GreptimeDBTelemetryTask::Disable => Ok(()),
88        }
89    }
90}
91
92/// Telemetry data to report
93#[derive(Serialize, Deserialize, Debug)]
94struct StatisticData {
95    /// Operating system name, such as `linux`, `windows` etc.
96    pub os: String,
97    /// The greptimedb version
98    pub version: String,
99    /// The architecture of the CPU, such as `x86`, `x86_64` etc.
100    pub arch: String,
101    /// The running mode, `standalone` or `distributed`.
102    pub mode: Mode,
103    /// The git commit revision of greptimedb
104    pub git_commit: String,
105    /// The node number
106    pub nodes: Option<i32>,
107    /// The local installation uuid
108    pub uuid: String,
109    /// System uptime range (e.g., "hours", "days", "weeks")
110    pub uptime: String,
111}
112
113#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
114#[serde(rename_all = "lowercase")]
115pub enum Mode {
116    Distributed,
117    Standalone,
118}
119
120#[async_trait::async_trait]
121pub trait Collector {
122    fn get_version(&self) -> String {
123        build_info().version.to_string()
124    }
125
126    fn get_git_hash(&self) -> String {
127        build_info().commit.to_string()
128    }
129
130    fn get_os(&self) -> String {
131        env::consts::OS.to_string()
132    }
133
134    fn get_arch(&self) -> String {
135        env::consts::ARCH.to_string()
136    }
137
138    fn get_mode(&self) -> Mode;
139
140    fn get_retry(&self) -> i32;
141
142    fn inc_retry(&mut self);
143
144    fn set_uuid_cache(&mut self, uuid: String);
145
146    fn get_uuid_cache(&self) -> Option<String>;
147
148    async fn get_nodes(&self) -> Option<i32>;
149
150    fn get_uuid(&mut self, working_home: &Option<String>) -> Option<String> {
151        match self.get_uuid_cache() {
152            Some(uuid) => Some(uuid),
153            None => {
154                if self.get_retry() > 3 {
155                    return None;
156                }
157                match default_get_uuid(working_home) {
158                    Some(uuid) => {
159                        self.set_uuid_cache(uuid.clone());
160                        Some(uuid)
161                    }
162                    None => {
163                        self.inc_retry();
164                        None
165                    }
166                }
167            }
168        }
169    }
170}
171
172fn print_anonymous_usage_data_disclaimer() {
173    info!("Attention: GreptimeDB now collects anonymous usage data to help improve its roadmap and prioritize features.");
174    info!(
175        "To learn more about this anonymous program and how to deactivate it if you don't want to participate, please visit the following URL: ");
176    info!("https://docs.greptime.com/reference/telemetry");
177}
178
179/// Format uptime duration into a general time range string
180/// Returns privacy-friendly descriptions like "hours", "days", etc.
181fn format_uptime() -> String {
182    let uptime_duration = START_TIME.elapsed().unwrap_or(Duration::ZERO);
183    let total_seconds = uptime_duration.as_secs();
184
185    if total_seconds < 86400 {
186        "hours".to_string()
187    } else if total_seconds < 604800 {
188        "days".to_string()
189    } else if total_seconds < 2629746 {
190        "weeks".to_string()
191    } else if total_seconds < 31556952 {
192        "months".to_string()
193    } else {
194        "years".to_string()
195    }
196}
197
198pub fn default_get_uuid(working_home: &Option<String>) -> Option<String> {
199    let temp_dir = env::temp_dir();
200
201    let mut path = PathBuf::new();
202    path.push(
203        working_home
204            .as_ref()
205            .map(Path::new)
206            .unwrap_or_else(|| temp_dir.as_path()),
207    );
208    path.push(UUID_FILE_NAME);
209
210    let path = path.as_path();
211    match std::fs::read(path) {
212        Ok(bytes) => Some(String::from_utf8_lossy(&bytes).to_string()),
213        Err(e) => {
214            if e.kind() == ErrorKind::NotFound {
215                let uuid = uuid::Uuid::new_v4().to_string();
216                let _ = std::fs::write(path, uuid.as_bytes());
217                Some(uuid)
218            } else {
219                None
220            }
221        }
222    }
223}
224
225/// Report version info to GreptimeDB.
226///
227/// We do not collect any identity-sensitive information.
228/// This task is scheduled to run every 30 minutes.
229/// The task will be disabled default. It can be enabled by setting the build feature `greptimedb-telemetry`
230/// Collector is used to collect the version info. It can be implemented by different components.
231/// client is used to send the HTTP request to GreptimeDB.
232/// telemetry_url is the GreptimeDB url.
233pub struct GreptimeDBTelemetry {
234    statistics: Box<dyn Collector + Send + Sync>,
235    client: Option<Client>,
236    working_home: Option<String>,
237    telemetry_url: &'static str,
238    should_report: Arc<AtomicBool>,
239    report_times: usize,
240}
241
242#[async_trait::async_trait]
243impl TaskFunction<Error> for GreptimeDBTelemetry {
244    fn name(&self) -> &str {
245        "Greptimedb-telemetry-task"
246    }
247
248    async fn call(&mut self) -> Result<()> {
249        if self.should_report.load(Ordering::Relaxed) {
250            self.report_telemetry_info().await;
251        }
252        Ok(())
253    }
254}
255
256impl GreptimeDBTelemetry {
257    pub fn new(
258        working_home: Option<String>,
259        statistics: Box<dyn Collector + Send + Sync>,
260        should_report: Arc<AtomicBool>,
261    ) -> Self {
262        let client = Client::builder()
263            .connect_timeout(GREPTIMEDB_TELEMETRY_CLIENT_CONNECT_TIMEOUT)
264            .timeout(GREPTIMEDB_TELEMETRY_CLIENT_REQUEST_TIMEOUT)
265            .build();
266        Self {
267            working_home,
268            statistics,
269            client: client.ok(),
270            telemetry_url: TELEMETRY_URL,
271            should_report,
272            report_times: 0,
273        }
274    }
275
276    pub async fn report_telemetry_info(&mut self) -> Option<Response> {
277        match self.statistics.get_uuid(&self.working_home) {
278            Some(uuid) => {
279                let data = StatisticData {
280                    os: self.statistics.get_os(),
281                    version: self.statistics.get_version(),
282                    git_commit: self.statistics.get_git_hash(),
283                    arch: self.statistics.get_arch(),
284                    mode: self.statistics.get_mode(),
285                    nodes: self.statistics.get_nodes().await,
286                    uuid,
287                    uptime: format_uptime(),
288                };
289
290                if let Some(client) = self.client.as_ref() {
291                    if self.report_times == 0 {
292                        info!("reporting greptimedb version: {:?}", data);
293                    }
294                    let result = client.post(self.telemetry_url).json(&data).send().await;
295                    self.report_times += 1;
296                    debug!("report version result: {:?}", result);
297                    result.ok()
298                } else {
299                    None
300                }
301            }
302            None => None,
303        }
304    }
305}
306
307#[cfg(test)]
308mod tests {
309    use std::convert::Infallible;
310    use std::env;
311    use std::sync::atomic::{AtomicBool, AtomicUsize};
312    use std::sync::Arc;
313    use std::time::Duration;
314
315    use common_test_util::ports;
316    use common_version::build_info;
317    use hyper::service::{make_service_fn, service_fn};
318    use hyper::Server;
319    use reqwest::{Client, Response};
320    use tokio::spawn;
321
322    use crate::{
323        default_get_uuid, format_uptime, Collector, GreptimeDBTelemetry, Mode, StatisticData,
324    };
325
326    static COUNT: AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
327
328    async fn echo(req: hyper::Request<hyper::Body>) -> hyper::Result<hyper::Response<hyper::Body>> {
329        let path = req.uri().path();
330        if path == "/req-cnt" {
331            let body = hyper::Body::from(format!(
332                "{}",
333                COUNT.load(std::sync::atomic::Ordering::SeqCst)
334            ));
335            Ok(hyper::Response::new(body))
336        } else {
337            COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
338            Ok(hyper::Response::new(req.into_body()))
339        }
340    }
341
342    #[tokio::test]
343    async fn test_gretimedb_telemetry() {
344        let (tx, rx) = tokio::sync::oneshot::channel::<()>();
345        let port: u16 = ports::get_port() as u16;
346        spawn(async move {
347            let make_svc = make_service_fn(|_conn| {
348                // This is the `Service` that will handle the connection.
349                // `service_fn` is a helper to convert a function that
350                // returns a Response into a `Service`.
351                async { Ok::<_, Infallible>(service_fn(echo)) }
352            });
353            let addr = ([127, 0, 0, 1], port).into();
354
355            let server = Server::try_bind(&addr).unwrap().serve(make_svc);
356            let graceful = server.with_graceful_shutdown(async {
357                rx.await.ok();
358            });
359            let _ = graceful.await;
360            Ok::<_, Infallible>(())
361        });
362        struct TestStatistic;
363
364        struct FailedStatistic;
365
366        #[async_trait::async_trait]
367        impl Collector for TestStatistic {
368            fn get_mode(&self) -> Mode {
369                Mode::Standalone
370            }
371
372            async fn get_nodes(&self) -> Option<i32> {
373                Some(1)
374            }
375
376            fn get_retry(&self) -> i32 {
377                unimplemented!()
378            }
379
380            fn inc_retry(&mut self) {
381                unimplemented!()
382            }
383
384            fn set_uuid_cache(&mut self, _: String) {
385                unimplemented!()
386            }
387
388            fn get_uuid_cache(&self) -> Option<String> {
389                unimplemented!()
390            }
391
392            fn get_uuid(&mut self, _working_home: &Option<String>) -> Option<String> {
393                Some("test".to_string())
394            }
395        }
396
397        #[async_trait::async_trait]
398        impl Collector for FailedStatistic {
399            fn get_mode(&self) -> Mode {
400                Mode::Standalone
401            }
402
403            async fn get_nodes(&self) -> Option<i32> {
404                None
405            }
406
407            fn get_retry(&self) -> i32 {
408                unimplemented!()
409            }
410
411            fn inc_retry(&mut self) {
412                unimplemented!()
413            }
414
415            fn set_uuid_cache(&mut self, _: String) {
416                unimplemented!()
417            }
418
419            fn get_uuid_cache(&self) -> Option<String> {
420                unimplemented!()
421            }
422
423            fn get_uuid(&mut self, _working_home: &Option<String>) -> Option<String> {
424                None
425            }
426        }
427
428        async fn get_telemetry_report(
429            mut report: GreptimeDBTelemetry,
430            url: &'static str,
431        ) -> Option<Response> {
432            report.telemetry_url = url;
433            report.report_telemetry_info().await
434        }
435
436        fn contravariance<'a>(x: &'a str) -> &'static str
437        where
438            'static: 'a,
439        {
440            unsafe { std::mem::transmute(x) }
441        }
442
443        let working_home_temp = tempfile::Builder::new()
444            .prefix("greptimedb_telemetry")
445            .tempdir()
446            .unwrap();
447        let working_home = working_home_temp.path().to_str().unwrap().to_string();
448
449        let test_statistic = Box::new(TestStatistic);
450        let test_report = GreptimeDBTelemetry::new(
451            Some(working_home.clone()),
452            test_statistic,
453            Arc::new(AtomicBool::new(true)),
454        );
455        let url = format!("http://localhost:{}", port);
456        let response = {
457            let url = contravariance(url.as_str());
458            get_telemetry_report(test_report, url).await.unwrap()
459        };
460
461        let body = response.json::<StatisticData>().await.unwrap();
462        assert_eq!(env::consts::ARCH, body.arch);
463        assert_eq!(env::consts::OS, body.os);
464        assert_eq!(build_info().version, body.version);
465        assert_eq!(build_info().commit, body.git_commit);
466        assert_eq!(Mode::Standalone, body.mode);
467        assert_eq!(1, body.nodes.unwrap());
468        assert!(!body.uptime.is_empty());
469
470        let failed_statistic = Box::new(FailedStatistic);
471        let failed_report = GreptimeDBTelemetry::new(
472            Some(working_home),
473            failed_statistic,
474            Arc::new(AtomicBool::new(true)),
475        );
476        let response = {
477            let url = contravariance(url.as_str());
478            get_telemetry_report(failed_report, url).await
479        };
480        assert!(response.is_none());
481
482        let client = Client::builder()
483            .connect_timeout(Duration::from_secs(3))
484            .timeout(Duration::from_secs(3))
485            .build()
486            .unwrap();
487
488        let cnt_url = format!("{}/req-cnt", url);
489        let response = client.get(cnt_url).send().await.unwrap();
490        let body = response.text().await.unwrap();
491        assert_eq!("1", body);
492        tx.send(()).unwrap();
493    }
494
495    #[test]
496    fn test_get_uuid() {
497        let working_home_temp = tempfile::Builder::new()
498            .prefix("greptimedb_telemetry")
499            .tempdir()
500            .unwrap();
501        let working_home = working_home_temp.path().to_str().unwrap().to_string();
502
503        let uuid = default_get_uuid(&Some(working_home.clone()));
504        assert!(uuid.is_some());
505        assert_eq!(uuid, default_get_uuid(&Some(working_home.clone())));
506        assert_eq!(uuid, default_get_uuid(&Some(working_home)));
507    }
508
509    #[test]
510    fn test_format_uptime() {
511        let uptime = format_uptime();
512        assert!(!uptime.is_empty());
513        // Should be a valid general time range (no specific numbers)
514        assert!(
515            uptime == "hours"
516                || uptime == "days"
517                || uptime == "weeks"
518                || uptime == "months"
519                || uptime == "years"
520        );
521    }
522}
common_greptimedb_telemetry/lib.rs

common_greptimedb_telemetry/
lib.rs
