oak-team/oak-gpui
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Start work on a client-side telemetry system

Browse Source
This commit is contained in:
Max Brunsfeld
2022-09-23 12:23:12 -07:00
parent dac0ce10e5
commit 04baccbea6
14 changed files with 191 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/client/src/channel.rs
+1 -1
View File
@@ -601,7 +601,7 @@ mod tests {
let user_id = 5;
let http_client = FakeHttpClient::with_404_response();
let client = Client::new(http_client.clone());
let client = cx.update(|cx| Client::new(http_client.clone(), cx));
let server = FakeServer::for_client(user_id, &client, cx).await;
Channel::init(&client);
crates/client/src/client.rs
+23 -8
View File
@@ -3,6 +3,7 @@ pub mod test;
pub mod channel;
pub mod http;
pub mod telemetry;
pub mod user;
use anyhow::{anyhow, Context, Result};
@@ -13,8 +14,9 @@ use async_tungstenite::tungstenite::{
};
use futures::{future::LocalBoxFuture, FutureExt, SinkExt, StreamExt, TryStreamExt};
use gpui::{
actions, AnyModelHandle, AnyViewHandle, AnyWeakModelHandle, AnyWeakViewHandle, AsyncAppContext,
Entity, ModelContext, ModelHandle, MutableAppContext, Task, View, ViewContext, ViewHandle,
actions, serde_json::Value, AnyModelHandle, AnyViewHandle, AnyWeakModelHandle,
AnyWeakViewHandle, AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle,
MutableAppContext, Task, View, ViewContext, ViewHandle,
};
use http::HttpClient;
use lazy_static::lazy_static;
@@ -31,6 +33,7 @@ use std::{
sync::{Arc, Weak},
time::{Duration, Instant},
};
use telemetry::Telemetry;
use thiserror::Error;
use url::Url;
use util::{ResultExt, TryFutureExt};
@@ -63,6 +66,7 @@ pub struct Client {
id: usize,
peer: Arc<Peer>,
http: Arc<dyn HttpClient>,
telemetry: Arc<Telemetry>,
state: RwLock<ClientState>,
#[allow(clippy::type_complexity)]
@@ -232,10 +236,11 @@ impl Drop for Subscription {
}
impl Client {
pub fn new(http: Arc<dyn HttpClient>) -> Arc<Self> {
pub fn new(http: Arc<dyn HttpClient>, cx: &AppContext) -> Arc<Self> {
Arc::new(Self {
id: 0,
peer: Peer::new(),
telemetry: Telemetry::new(http.clone(), cx),
http,
state: Default::default(),
@@ -595,6 +600,9 @@ impl Client {
if credentials.is_none() && try_keychain {
credentials = read_credentials_from_keychain(cx);
read_from_keychain = credentials.is_some();
if read_from_keychain {
self.log_event("read_credentials_from_keychain", Default::default());
}
}
if credentials.is_none() {
let mut status_rx = self.status();
@@ -878,6 +886,7 @@ impl Client {
) -> Task<Result<Credentials>> {
let platform = cx.platform();
let executor = cx.background();
let telemetry = self.telemetry.clone();
executor.clone().spawn(async move {
// Generate a pair of asymmetric encryption keys. The public key will be used by the
// zed server to encrypt the user's access token, so that it can'be intercepted by
@@ -956,6 +965,8 @@ impl Client {
.context("failed to decrypt access token")?;
platform.activate(true);
telemetry.log_event("authenticate_with_browser", Default::default());
Ok(Credentials {
user_id: user_id.parse()?,
access_token,
@@ -1020,6 +1031,10 @@ impl Client {
log::debug!("rpc respond. client_id:{}. name:{}", self.id, T::NAME);
self.peer.respond_with_error(receipt, error)
}
pub fn log_event(&self, kind: &str, properties: Value) {
self.telemetry.log_event(kind, properties)
}
}
impl AnyWeakEntityHandle {
@@ -1085,7 +1100,7 @@ mod tests {
cx.foreground().forbid_parking();
let user_id = 5;
let client = Client::new(FakeHttpClient::with_404_response());
let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
let server = FakeServer::for_client(user_id, &client, cx).await;
let mut status = client.status();
assert!(matches!(
@@ -1124,7 +1139,7 @@ mod tests {
let auth_count = Arc::new(Mutex::new(0));
let dropped_auth_count = Arc::new(Mutex::new(0));
let client = Client::new(FakeHttpClient::with_404_response());
let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
client.override_authenticate({
let auth_count = auth_count.clone();
let dropped_auth_count = dropped_auth_count.clone();
@@ -1173,7 +1188,7 @@ mod tests {
cx.foreground().forbid_parking();
let user_id = 5;
let client = Client::new(FakeHttpClient::with_404_response());
let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
let server = FakeServer::for_client(user_id, &client, cx).await;
let (done_tx1, mut done_rx1) = smol::channel::unbounded();
@@ -1219,7 +1234,7 @@ mod tests {
cx.foreground().forbid_parking();
let user_id = 5;
let client = Client::new(FakeHttpClient::with_404_response());
let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
let server = FakeServer::for_client(user_id, &client, cx).await;
let model = cx.add_model(|_| Model::default());
@@ -1247,7 +1262,7 @@ mod tests {
cx.foreground().forbid_parking();
let user_id = 5;
let client = Client::new(FakeHttpClient::with_404_response());
let client = cx.update(|cx| Client::new(FakeHttpClient::with_404_response(), cx));
let server = FakeServer::for_client(user_id, &client, cx).await;
let model = cx.add_model(|_| Model::default());
crates/client/src/telemetry.rs
+128
View File
@@ -0,0 +1,128 @@
use crate::{http::HttpClient, ZED_SECRET_CLIENT_TOKEN};
use gpui::{
executor::Background,
serde_json::{self, value::Map, Value},
AppContext, AppVersion, Task,
};
use isahc::Request;
use parking_lot::Mutex;
use serde::Serialize;
use std::{
mem,
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use util::ResultExt;
pub struct Telemetry {
client: Arc<dyn HttpClient>,
executor: Arc<Background>,
state: Mutex<TelemetryState>,
}
#[derive(Default)]
struct TelemetryState {
metrics_id: Option<i32>,
device_id: Option<String>,
app_version: Option<AppVersion>,
os_version: Option<AppVersion>,
queue: Vec<Event>,
flush_task: Option<Task<()>>,
}
#[derive(Serialize)]
struct RecordEventParams {
token: &'static str,
metrics_id: Option<i32>,
device_id: Option<String>,
app_version: Option<String>,
os_version: Option<String>,
events: Vec<Event>,
}
#[derive(Serialize)]
struct Event {
#[serde(rename = "type")]
kind: String,
time: u128,
properties: Option<Map<String, Value>>,
}
const MAX_QUEUE_LEN: usize = 30;
const EVENTS_URI: &'static str = "https://zed.dev/api/telemetry";
const DEBOUNCE_INTERVAL: Duration = Duration::from_secs(30);
impl Telemetry {
pub fn new(client: Arc<dyn HttpClient>, cx: &AppContext) -> Arc<Self> {
let platform = cx.platform();
Arc::new(Self {
client,
executor: cx.background().clone(),
state: Mutex::new(TelemetryState {
os_version: platform.os_version().log_err(),
app_version: platform.app_version().log_err(),
metrics_id: None,
device_id: None,
queue: Default::default(),
flush_task: Default::default(),
}),
})
}
pub fn set_metrics_id(&self, metrics_id: Option<i32>) {
self.state.lock().metrics_id = metrics_id;
}
pub fn log_event(self: &Arc<Self>, kind: &str, properties: Value) {
let mut state = self.state.lock();
state.queue.push(Event {
kind: kind.to_string(),
time: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis(),
properties: if let Value::Object(properties) = properties {
Some(properties)
} else {
None
},
});
if state.queue.len() >= MAX_QUEUE_LEN {
self.flush();
} else {
let this = self.clone();
let executor = self.executor.clone();
state.flush_task = Some(self.executor.spawn(async move {
executor.timer(DEBOUNCE_INTERVAL).await;
this.flush();
}));
}
}
fn flush(&self) {
let mut state = self.state.lock();
let events = mem::take(&mut state.queue);
let client = self.client.clone();
let app_version = state.app_version;
let os_version = state.os_version;
let metrics_id = state.metrics_id;
let device_id = state.device_id.clone();
state.flush_task.take();
self.executor
.spawn(async move {
let body = serde_json::to_vec(&RecordEventParams {
token: ZED_SECRET_CLIENT_TOKEN,
events,
app_version: app_version.map(|v| v.to_string()),
os_version: os_version.map(|v| v.to_string()),
metrics_id,
device_id,
})
.log_err()?;
let request = Request::post(EVENTS_URI).body(body.into()).log_err()?;
client.send(request).await.log_err();
Some(())
})
.detach();
}
}