server: add traffic requests api

crates/kernel/src/prom/mod.rs

@@ -1,6 +1,7 @@
 use anyhow::{anyhow, Result};
 use once_cell::sync::OnceCell;
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 use tracing::info;
 
 #[derive(Serialize, Deserialize, Clone)]
@@ -35,3 +36,49 @@ pub fn init_prom_env() -> Result<()> {
         .map_err(|_| anyhow!("PromEnv is already set"))?;
     Ok(())
 }
+
+/// QueryRangeParams is the parameters for querying range
+#[derive(Serialize)]
+pub struct QueryRangeParams {
+    pub query: String,
+    pub start: i64,
+    pub end: i64,
+    pub step: i64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct QueryResponse {
+    pub status: String,
+    pub data: QueryResponseData,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct QueryResponseData {
+    #[serde(rename = "resultType")]
+    pub result_type: String,
+    pub result: Vec<QueryResponseDataItem>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct QueryResponseDataItem {
+    pub metric: HashMap<String, String>,
+    pub values: Vec<(i64, String)>,
+}
+
+/// query_range queries range from Prometheus
+pub async fn query_range(params: QueryRangeParams) -> Result<QueryResponse> {
+    let prom_env = PROM_ENV
+        .get()
+        .ok_or_else(|| anyhow!("PromEnv is not set"))?;
+    let client = reqwest::Client::new();
+    // use post to query
+    let resp = client
+        .post(&format!("{}/api/v1/query_range", prom_env.addr))
+        .basic_auth(prom_env.user.clone(), Some(prom_env.password.clone()))
+        .form(&params)
+        .send()
+        .await?;
+    let resp = resp.error_for_status()?;
+    let resp = resp.json::<QueryResponse>().await?;
+    Ok(resp)
+}

land-server/src/server/dash/mod.rs

@@ -17,6 +17,7 @@ mod overview;
 mod playground;
 mod projects;
 mod settings;
+mod traffic;
 
 /// index is a handler for GET /
 pub async fn index() -> impl IntoResponse {
@@ -64,6 +65,8 @@ pub fn router(assets_dir: &str) -> Result<Router> {
             get(playground::index).post(playground::save),
         )
         .route("/playground/:name/check", get(playground::check))
+        .route("/traffic/requests", get(traffic::requests))
+        .route("/traffic/flows", get(traffic::flows))
         .route("/settings", get(settings::index))
         .route("/settings/create-token", post(settings::create_token))
         .route("/settings/manage", get(settings::manage))

land-server/src/server/dash/traffic.rs

@@ -0,0 +1,70 @@
+use super::auth::SessionUser;
+use crate::server::ServerError;
+use axum::{extract::Query, response::IntoResponse, Extension, Json};
+use serde::{Deserialize, Serialize};
+use tracing::debug;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RequestsQuery {
+    pub account: Option<String>,
+    pub project: Option<String>,
+    pub period: Option<String>, //only for daily or weekly
+}
+
+impl RequestsQuery {
+    pub fn get_period(&self) -> (i64, i64, i64, String) {
+        if let Some(p) = self.period.as_ref() {
+            if p.eq("weekly") {
+                let end = chrono::Utc::now().timestamp() / 3600 * 3600;
+                let start = end - 604800; // 86400 * 7
+                return (start, end, 3600, "1h".to_string());
+            }
+        }
+        let end = chrono::Utc::now().timestamp() / 600 * 600;
+        let start = end - 86400;
+        (start, end, 600, "10m".to_string())
+    }
+}
+
+pub async fn flows(
+    Extension(user): Extension<SessionUser>,
+    Query(q): Query<RequestsQuery>,
+) -> Result<impl IntoResponse, ServerError> {
+    Ok("flows".to_string())
+}
+
+/// requests is a handler for GET /traffic/requests
+pub async fn requests(
+    Extension(user): Extension<SessionUser>,
+    Query(q): Query<RequestsQuery>,
+) -> Result<impl IntoResponse, ServerError> {
+    let (start, end, step, step_word) = q.get_period();
+    let acc = q.account.unwrap_or_default();
+    if acc != user.id.to_string() {
+        return Err(ServerError::forbidden("user id does not match"));
+    }
+    let query = if let Some(pid) = q.project {
+        format!(
+            "increase(req_fn_total{{project_id=\"{}\"}}[{}])",
+            pid, step_word
+        )
+    } else {
+        format!(
+            "sum(increase(req_fn_total{{user_id=\"{}\"}}[{}]))",
+            acc, step_word
+        )
+    };
+    // end time is now ts with latest 10 decade
+    debug!(
+        "query: {}, start:{}, end:{}, step:{}",
+        query, start, end, step
+    );
+    let params = land_kernel::prom::QueryRangeParams {
+        query,
+        step,
+        start,
+        end,
+    };
+    let res = land_kernel::prom::query_range(params).await?;
+    Ok(Json(res).into_response())
+}