common_greptimedb_telemetry/
lib.rsuse std::env;
use std::io::ErrorKind;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use common_runtime::error::{Error, Result};
use common_runtime::{BoxedTaskFunction, RepeatedTask, TaskFunction};
use common_telemetry::{debug, info};
use common_version::build_info;
use reqwest::{Client, Response};
use serde::{Deserialize, Serialize};
pub const TELEMETRY_URL: &str = "https://telemetry.greptimestats.com/db/otel/statistics";
const UUID_FILE_NAME: &str = ".greptimedb-telemetry-uuid";
pub static TELEMETRY_INTERVAL: Duration = Duration::from_secs(60 * 30);
const GREPTIMEDB_TELEMETRY_CLIENT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const GREPTIMEDB_TELEMETRY_CLIENT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
pub enum GreptimeDBTelemetryTask {
Enable((RepeatedTask<Error>, Arc<AtomicBool>)),
Disable,
}
impl GreptimeDBTelemetryTask {
pub fn should_report(&self, value: bool) {
match self {
GreptimeDBTelemetryTask::Enable((_, should_report)) => {
should_report.store(value, Ordering::Relaxed);
}
GreptimeDBTelemetryTask::Disable => {}
}
}
pub fn enable(
interval: Duration,
task_fn: BoxedTaskFunction<Error>,
should_report: Arc<AtomicBool>,
) -> Self {
GreptimeDBTelemetryTask::Enable((
RepeatedTask::new(interval, task_fn).with_initial_delay(Some(Duration::ZERO)),
should_report,
))
}
pub fn disable() -> Self {
GreptimeDBTelemetryTask::Disable
}
pub fn start(&self) -> Result<()> {
match self {
GreptimeDBTelemetryTask::Enable((task, _)) => {
print_anonymous_usage_data_disclaimer();
task.start(common_runtime::global_runtime())
}
GreptimeDBTelemetryTask::Disable => Ok(()),
}
}
pub async fn stop(&self) -> Result<()> {
match self {
GreptimeDBTelemetryTask::Enable((task, _)) => task.stop().await,
GreptimeDBTelemetryTask::Disable => Ok(()),
}
}
}
#[derive(Serialize, Deserialize, Debug)]
struct StatisticData {
pub os: String,
pub version: String,
pub arch: String,
pub mode: Mode,
pub git_commit: String,
pub nodes: Option<i32>,
pub uuid: String,
}
#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum Mode {
Distributed,
Standalone,
}
#[async_trait::async_trait]
pub trait Collector {
fn get_version(&self) -> String {
build_info().version.to_string()
}
fn get_git_hash(&self) -> String {
build_info().commit.to_string()
}
fn get_os(&self) -> String {
env::consts::OS.to_string()
}
fn get_arch(&self) -> String {
env::consts::ARCH.to_string()
}
fn get_mode(&self) -> Mode;
fn get_retry(&self) -> i32;
fn inc_retry(&mut self);
fn set_uuid_cache(&mut self, uuid: String);
fn get_uuid_cache(&self) -> Option<String>;
async fn get_nodes(&self) -> Option<i32>;
fn get_uuid(&mut self, working_home: &Option<String>) -> Option<String> {
match self.get_uuid_cache() {
Some(uuid) => Some(uuid),
None => {
if self.get_retry() > 3 {
return None;
}
match default_get_uuid(working_home) {
Some(uuid) => {
self.set_uuid_cache(uuid.clone());
Some(uuid)
}
None => {
self.inc_retry();
None
}
}
}
}
}
}
fn print_anonymous_usage_data_disclaimer() {
info!("Attention: GreptimeDB now collects anonymous usage data to help improve its roadmap and prioritize features.");
info!(
"To learn more about this anonymous program and how to deactivate it if you don't want to participate, please visit the following URL: ");
info!("https://docs.greptime.com/reference/telemetry");
}
pub fn default_get_uuid(working_home: &Option<String>) -> Option<String> {
let temp_dir = env::temp_dir();
let mut path = PathBuf::new();
path.push(
working_home
.as_ref()
.map(Path::new)
.unwrap_or_else(|| temp_dir.as_path()),
);
path.push(UUID_FILE_NAME);
let path = path.as_path();
match std::fs::read(path) {
Ok(bytes) => Some(String::from_utf8_lossy(&bytes).to_string()),
Err(e) => {
if e.kind() == ErrorKind::NotFound {
let uuid = uuid::Uuid::new_v4().to_string();
let _ = std::fs::write(path, uuid.as_bytes());
Some(uuid)
} else {
None
}
}
}
}
pub struct GreptimeDBTelemetry {
statistics: Box<dyn Collector + Send + Sync>,
client: Option<Client>,
working_home: Option<String>,
telemetry_url: &'static str,
should_report: Arc<AtomicBool>,
report_times: usize,
}
#[async_trait::async_trait]
impl TaskFunction<Error> for GreptimeDBTelemetry {
fn name(&self) -> &str {
"Greptimedb-telemetry-task"
}
async fn call(&mut self) -> Result<()> {
if self.should_report.load(Ordering::Relaxed) {
self.report_telemetry_info().await;
}
Ok(())
}
}
impl GreptimeDBTelemetry {
pub fn new(
working_home: Option<String>,
statistics: Box<dyn Collector + Send + Sync>,
should_report: Arc<AtomicBool>,
) -> Self {
let client = Client::builder()
.connect_timeout(GREPTIMEDB_TELEMETRY_CLIENT_CONNECT_TIMEOUT)
.timeout(GREPTIMEDB_TELEMETRY_CLIENT_REQUEST_TIMEOUT)
.build();
Self {
working_home,
statistics,
client: client.ok(),
telemetry_url: TELEMETRY_URL,
should_report,
report_times: 0,
}
}
pub async fn report_telemetry_info(&mut self) -> Option<Response> {
match self.statistics.get_uuid(&self.working_home) {
Some(uuid) => {
let data = StatisticData {
os: self.statistics.get_os(),
version: self.statistics.get_version(),
git_commit: self.statistics.get_git_hash(),
arch: self.statistics.get_arch(),
mode: self.statistics.get_mode(),
nodes: self.statistics.get_nodes().await,
uuid,
};
if let Some(client) = self.client.as_ref() {
if self.report_times == 0 {
info!("reporting greptimedb version: {:?}", data);
}
let result = client.post(self.telemetry_url).json(&data).send().await;
self.report_times += 1;
debug!("report version result: {:?}", result);
result.ok()
} else {
None
}
}
None => None,
}
}
}
#[cfg(test)]
mod tests {
use std::convert::Infallible;
use std::env;
use std::sync::atomic::{AtomicBool, AtomicUsize};
use std::sync::Arc;
use std::time::Duration;
use common_test_util::ports;
use common_version::build_info;
use hyper::service::{make_service_fn, service_fn};
use hyper::Server;
use reqwest::{Client, Response};
use tokio::spawn;
use crate::{default_get_uuid, Collector, GreptimeDBTelemetry, Mode, StatisticData};
static COUNT: AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
async fn echo(req: hyper::Request<hyper::Body>) -> hyper::Result<hyper::Response<hyper::Body>> {
let path = req.uri().path();
if path == "/req-cnt" {
let body = hyper::Body::from(format!(
"{}",
COUNT.load(std::sync::atomic::Ordering::SeqCst)
));
Ok(hyper::Response::new(body))
} else {
COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(hyper::Response::new(req.into_body()))
}
}
#[tokio::test]
async fn test_gretimedb_telemetry() {
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
let port: u16 = ports::get_port() as u16;
spawn(async move {
let make_svc = make_service_fn(|_conn| {
async { Ok::<_, Infallible>(service_fn(echo)) }
});
let addr = ([127, 0, 0, 1], port).into();
let server = Server::try_bind(&addr).unwrap().serve(make_svc);
let graceful = server.with_graceful_shutdown(async {
rx.await.ok();
});
let _ = graceful.await;
Ok::<_, Infallible>(())
});
struct TestStatistic;
struct FailedStatistic;
#[async_trait::async_trait]
impl Collector for TestStatistic {
fn get_mode(&self) -> Mode {
Mode::Standalone
}
async fn get_nodes(&self) -> Option<i32> {
Some(1)
}
fn get_retry(&self) -> i32 {
unimplemented!()
}
fn inc_retry(&mut self) {
unimplemented!()
}
fn set_uuid_cache(&mut self, _: String) {
unimplemented!()
}
fn get_uuid_cache(&self) -> Option<String> {
unimplemented!()
}
fn get_uuid(&mut self, _working_home: &Option<String>) -> Option<String> {
Some("test".to_string())
}
}
#[async_trait::async_trait]
impl Collector for FailedStatistic {
fn get_mode(&self) -> Mode {
Mode::Standalone
}
async fn get_nodes(&self) -> Option<i32> {
None
}
fn get_retry(&self) -> i32 {
unimplemented!()
}
fn inc_retry(&mut self) {
unimplemented!()
}
fn set_uuid_cache(&mut self, _: String) {
unimplemented!()
}
fn get_uuid_cache(&self) -> Option<String> {
unimplemented!()
}
fn get_uuid(&mut self, _working_home: &Option<String>) -> Option<String> {
None
}
}
async fn get_telemetry_report(
mut report: GreptimeDBTelemetry,
url: &'static str,
) -> Option<Response> {
report.telemetry_url = url;
report.report_telemetry_info().await
}
fn contravariance<'a>(x: &'a str) -> &'static str
where
'static: 'a,
{
unsafe { std::mem::transmute(x) }
}
let working_home_temp = tempfile::Builder::new()
.prefix("greptimedb_telemetry")
.tempdir()
.unwrap();
let working_home = working_home_temp.path().to_str().unwrap().to_string();
let test_statistic = Box::new(TestStatistic);
let test_report = GreptimeDBTelemetry::new(
Some(working_home.clone()),
test_statistic,
Arc::new(AtomicBool::new(true)),
);
let url = format!("http://localhost:{}", port);
let response = {
let url = contravariance(url.as_str());
get_telemetry_report(test_report, url).await.unwrap()
};
let body = response.json::<StatisticData>().await.unwrap();
assert_eq!(env::consts::ARCH, body.arch);
assert_eq!(env::consts::OS, body.os);
assert_eq!(build_info().version, body.version);
assert_eq!(build_info().commit, body.git_commit);
assert_eq!(Mode::Standalone, body.mode);
assert_eq!(1, body.nodes.unwrap());
let failed_statistic = Box::new(FailedStatistic);
let failed_report = GreptimeDBTelemetry::new(
Some(working_home),
failed_statistic,
Arc::new(AtomicBool::new(true)),
);
let response = {
let url = contravariance(url.as_str());
get_telemetry_report(failed_report, url).await
};
assert!(response.is_none());
let client = Client::builder()
.connect_timeout(Duration::from_secs(3))
.timeout(Duration::from_secs(3))
.build()
.unwrap();
let cnt_url = format!("{}/req-cnt", url);
let response = client.get(cnt_url).send().await.unwrap();
let body = response.text().await.unwrap();
assert_eq!("1", body);
tx.send(()).unwrap();
}
#[test]
fn test_get_uuid() {
let working_home_temp = tempfile::Builder::new()
.prefix("greptimedb_telemetry")
.tempdir()
.unwrap();
let working_home = working_home_temp.path().to_str().unwrap().to_string();
let uuid = default_get_uuid(&Some(working_home.clone()));
assert!(uuid.is_some());
assert_eq!(uuid, default_get_uuid(&Some(working_home.clone())));
assert_eq!(uuid, default_get_uuid(&Some(working_home)));
}
}