fix: stream stats with sqlite (#1477)

src/common/infra/file_list/dynamo.rs

@@ -465,6 +465,10 @@ impl super::FileList for DynamoFileList {
         Ok(())
     }
 
+    async fn reset_stream_stats(&self) -> Result<()> {
+        Ok(()) // TODO
+    }
+
     async fn reset_stream_stats_min_ts(
         &self,
         org_id: &str,

src/common/infra/file_list/mod.rs

@@ -79,6 +79,7 @@ pub trait FileList: Sync + Send + 'static {
     ) -> Result<Vec<(String, StreamStats)>>;
     async fn set_stream_stats(&self, org_id: &str, streams: &[(String, StreamStats)])
         -> Result<()>;
+    async fn reset_stream_stats(&self) -> Result<()>;
     async fn reset_stream_stats_min_ts(
         &self,
         org_id: &str,
@@ -191,6 +192,11 @@ pub async fn set_stream_stats(org_id: &str, streams: &[(String, StreamStats)]) -
     CLIENT.set_stream_stats(org_id, streams).await
 }
 
+#[inline]
+pub async fn reset_stream_stats() -> Result<()> {
+    CLIENT.reset_stream_stats().await
+}
+
 #[inline]
 pub async fn reset_stream_stats_min_ts(org_id: &str, stream: &str, min_ts: i64) -> Result<()> {
     CLIENT

src/common/infra/file_list/postgres.rs

@@ -391,6 +391,14 @@ UPDATE stream_stats
         Ok(())
     }
 
+    async fn reset_stream_stats(&self) -> Result<()> {
+        let pool = CLIENT.clone();
+        sqlx::query(r#"UPDATE stream_stats SET file_num = 0, min_ts = 0, max_ts = 0, records = 0, original_size = 0, compressed_size = 0;"#)
+             .execute(&pool)
+            .await?;
+        Ok(())
+    }
+
     async fn reset_stream_stats_min_ts(
         &self,
         _org_id: &str,

src/common/infra/file_list/sqlite.rs

@@ -394,6 +394,14 @@ UPDATE stream_stats
         Ok(())
     }
 
+    async fn reset_stream_stats(&self) -> Result<()> {
+        let pool = CLIENT.clone();
+        sqlx::query(r#"UPDATE stream_stats SET file_num = 0, min_ts = 0, max_ts = 0, records = 0, original_size = 0, compressed_size = 0;"#)
+             .execute(&pool)
+            .await?;
+        Ok(())
+    }
+
     async fn reset_stream_stats_min_ts(
         &self,
         _org_id: &str,
@@ -512,12 +520,10 @@ CREATE UNIQUE INDEX IF NOT EXISTS stream_stats_stream_idx on stream_stats (strea
     // create trigger
     sqlx::query(
     r#"
-CREATE TRIGGER IF NOT EXISTS update_stream_stats_delete
-    AFTER DELETE ON file_list
-BEGIN
-    UPDATE stream_stats SET file_num = file_num - 1, records = records - OLD.records, original_size = original_size - OLD.original_size, compressed_size = compressed_size - OLD.compressed_size
-        WHERE stream = OLD.stream;
-END;
+CREATE TRIGGER IF NOT EXISTS update_stream_stats_delete AFTER DELETE ON file_list
+    BEGIN
+        UPDATE stream_stats SET file_num = file_num - 1, records = records - OLD.records, original_size = original_size - OLD.original_size, compressed_size = compressed_size - OLD.compressed_size WHERE stream = OLD.stream;
+    END;
         "#,
     )
     .execute(&pool)

src/job/mod.rs

@@ -23,7 +23,7 @@ use crate::common::{
     meta::{organization::DEFAULT_ORG, user::UserRequest},
     utils::file::clean_empty_dirs,
 };
-use crate::service::{db, users};
+use crate::service::{compact::stats::update_stats_from_file_list, db, users};
 
 mod alert_manager;
 mod compact;
@@ -138,11 +138,14 @@ pub async fn init() -> Result<(), anyhow::Error> {
         db::file_list::remote::cache("", false)
             .await
             .expect("file list remote cache failed");
-        db::file_list::remote::cache_stats()
+        update_stats_from_file_list()
             .await
-            .expect("file list remote cache stats failed");
+            .expect("file list remote calculate stats failed");
     }
     infra_file_list::create_table_index().await?;
+    db::file_list::remote::cache_stats()
+        .await
+        .expect("Load stream stats failed");
 
     // check wal directory
     if cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) {

src/job/stats.rs

@@ -15,10 +15,8 @@
 use tokio::time;
 
 use crate::common::infra::{
-    cache::stats,
     cluster::{is_compactor, is_querier},
     config::CONFIG,
-    file_list as infra_file_list,
 };
 use crate::service::{compact::stats::update_stats_from_file_list, db, usage};
 
@@ -80,24 +78,8 @@ async fn cache_stream_stats() -> Result<(), anyhow::Error> {
     interval.tick().await; // trigger the first run
     loop {
         interval.tick().await;
-        let orgs = db::schema::list_organizations_from_cache();
-        for org_id in orgs {
-            let ret = infra_file_list::get_stream_stats(&org_id, None, None).await;
-            if ret.is_err() {
-                log::error!(
-                    "[STATS] run cache stream stats error: {}",
-                    ret.err().unwrap()
-                );
-                continue;
-            }
-            for (stream, stats) in ret.unwrap() {
-                let columns = stream.split('/').collect::<Vec<&str>>();
-                let org_id = columns[0];
-                let stream_type = columns[1];
-                let stream_name = columns[2];
-                stats::set_stream_stats(org_id, stream_name, stream_type.into(), stats);
-            }
-            time::sleep(time::Duration::from_millis(100)).await;
+        if let Err(e) = db::file_list::remote::cache_stats().await {
+            log::error!("[STATS] run cache stream stats error: {}", e);
         }
     }
 }

src/main.rs

@@ -41,7 +41,6 @@ use openobserve::{
     common::infra::{
         self, cluster,
         config::{CONFIG, USERS, VERSION},
-        metrics, wal,
     },
     common::meta,
     common::migration,
@@ -65,7 +64,7 @@ use openobserve::{
         http::router::*,
     },
     job,
-    service::{db, file_list, router, users},
+    service::{compact, db, file_list, router, users},
 };
 
 #[cfg(feature = "profiling")]
@@ -159,7 +158,7 @@ async fn main() -> Result<(), anyhow::Error> {
         .expect("EnrichmentTables cache failed");
 
     // metrics
-    let prometheus = metrics::create_prometheus_handler();
+    let prometheus = infra::metrics::create_prometheus_handler();
 
     // HTTP server
     let thread_id = Arc::new(AtomicU16::new(0));
@@ -230,7 +229,7 @@ async fn main() -> Result<(), anyhow::Error> {
     // leave the cluster
     let _ = cluster::leave().await;
     // flush WAL cache to disk
-    wal::flush_all_to_disk();
+    infra::wal::flush_all_to_disk();
     // flush compact offset cache to disk disk
     let _ = db::compact::files::sync_cache_to_db().await;
 
@@ -329,7 +328,7 @@ async fn cli() -> Result<bool, anyhow::Error> {
                         .short('c')
                         .long("component")
                         .help(
-                            "reset data of the component: root, user, alert, dashboard, function",
+                            "reset data of the component: root, user, alert, dashboard, function, stream-stats",
                         ),
                 ),
             clap::Command::new("view")
@@ -350,6 +349,7 @@ async fn cli() -> Result<bool, anyhow::Error> {
                         .required(false)
                         .help("only migrate specified prefix, default is all"),
                 ),
+            clap::Command::new("migrate-meta").about("migrate meta"),
             clap::Command::new("delete-parquet")
                 .about("delete parquet files from s3 and file_list")
                 .arg(
@@ -359,7 +359,6 @@ async fn cli() -> Result<bool, anyhow::Error> {
                         .value_name("file")
                         .help("the parquet file name"),
                 ),
-            clap::Command::new("migrate-meta").about("migrate meta"),
         ])
         .get_matches();
 
@@ -399,6 +398,18 @@ async fn cli() -> Result<bool, anyhow::Error> {
                 "function" => {
                     db::functions::reset().await?;
                 }
+                "stream-stats" => {
+                    // reset stream stats update offset
+                    db::compact::stats::set_offset(0, None).await?;
+                    // reset stream stats table data
+                    infra::file_list::reset_stream_stats().await?;
+                    // load stream list
+                    db::schema::cache().await?;
+                    // update stats from file list
+                    compact::stats::update_stats_from_file_list()
+                        .await
+                        .expect("file list remote calculate stats failed");
+                }
                 _ => {
                     return Err(anyhow::anyhow!("unsupport reset component: {}", component));
                 }

src/service/compact/mod.rs

@@ -26,8 +26,8 @@ use crate::service::db;
 
 mod file_list;
 mod merge;
-pub(crate) mod retention;
-pub(crate) mod stats;
+pub mod retention;
+pub mod stats;
 
 pub(crate) static QUEUE_LOCKER: Lazy<Arc<Mutex<bool>>> =
     Lazy::new(|| Arc::new(Mutex::const_new(false)));

src/service/db/file_list/remote.rs

@@ -138,20 +138,18 @@ pub async fn cache_stats() -> Result<(), anyhow::Error> {
     let start = std::time::Instant::now();
     let orgs = db::schema::list_organizations_from_cache();
     for org_id in orgs {
-        let ret = infra_file_list::stats(&org_id, None, None, None).await;
+        let ret = infra_file_list::get_stream_stats(&org_id, None, None).await;
         if ret.is_err() {
             log::error!("Load stream stats error: {}", ret.err().unwrap());
             continue;
         }
-        let stream_stats = ret.unwrap();
-        for (stream, stats) in stream_stats.iter() {
+        for (stream, stats) in ret.unwrap() {
             let columns = stream.split('/').collect::<Vec<&str>>();
             let org_id = columns[0];
             let stream_type = columns[1];
             let stream_name = columns[2];
-            stats::set_stream_stats(org_id, stream_name, stream_type.into(), *stats);
+            stats::set_stream_stats(org_id, stream_name, stream_type.into(), stats);
         }
-        infra_file_list::set_stream_stats(&org_id, &stream_stats).await?;
         time::sleep(time::Duration::from_millis(100)).await;
     }
     log::info!("Load stream stats in {}s", start.elapsed().as_secs());

src/service/traces/mod.rs

@@ -197,7 +197,7 @@ pub async fn handle_trace_request(
                     operation_name: span.name.clone(),
                     start_time,
                     end_time,
-                    duration: end_time - start_time,
+                    duration: (end_time - start_time) / 1000, // microseconds
                     reference: span_ref,
                     service_name: service_name.clone(),
                     attributes: span_att_map,

src/service/traces/otlp_http.rs

@@ -269,7 +269,7 @@ pub async fn traces_json(
                         operation_name: span.get("name").unwrap().as_str().unwrap().to_string(),
                         start_time,
                         end_time,
-                        duration: end_time - start_time,
+                        duration: (end_time - start_time) / 1000, // microseconds
                         reference: span_ref,
                         service_name: service_name.clone(),
                         attributes: span_att_map,

web/src/plugins/traces/Index.vue

@@ -235,12 +235,12 @@ export default defineComponent({
       if (
         this.searchObj.meta.refreshInterval == 0 &&
         this.searchObj.data.queryResults.total >
-          this.searchObj.data.queryResults.from &&
+        this.searchObj.data.queryResults.from &&
         this.searchObj.data.queryResults.total >
-          this.searchObj.data.queryResults.size &&
+        this.searchObj.data.queryResults.size &&
         this.searchObj.data.queryResults.total >
-          this.searchObj.data.queryResults.size +
-            this.searchObj.data.queryResults.from
+        this.searchObj.data.queryResults.size +
+        this.searchObj.data.queryResults.from
       ) {
         this.searchObj.loading = true;
         this.getQueryData();
@@ -478,8 +478,8 @@ export default defineComponent({
           } else {
             start = new Date(
               searchObj.data.datetime.absolute.date.from +
-                " " +
-                searchObj.data.datetime.absolute.startTime
+              " " +
+              searchObj.data.datetime.absolute.startTime
             );
           }
           if (
@@ -490,8 +490,8 @@ export default defineComponent({
           } else {
             end = new Date(
               searchObj.data.datetime.absolute.date.to +
-                " " +
-                searchObj.data.datetime.absolute.endTime
+              " " +
+              searchObj.data.datetime.absolute.endTime
             );
           }
           const rVal = {
@@ -583,8 +583,8 @@ export default defineComponent({
         let timestamps: any =
           searchObj.data.datetime.type === "relative"
             ? getConsumableRelativeTime(
-                searchObj.data.datetime.relativeTimePeriod
-              )
+              searchObj.data.datetime.relativeTimePeriod
+            )
             : cloneDeep(searchObj.data.datetime);
 
         req.query.start_time = timestamps.startTime;
@@ -940,12 +940,11 @@ export default defineComponent({
       }
     }
 
-    function formatTimeWithSuffix(ms) {
-      if (ms < 1000) {
-        return `${ms}ms`;
-      } else {
-        return `${(ms / 1000).toFixed(2)}s`;
+    function formatTimeWithSuffix(us: number) {
+      if (us >= 1000 * 1000) {
+        return `${(us / 1000 / 1000).toFixed(2)}s`;
       }
+      return `${(us / 1000).toFixed(2)}ms`;
     }
 
     function generateHistogramData() {
@@ -1000,7 +999,7 @@ export default defineComponent({
             unparsed_x_data.push(bucket.zo_sql_timestamp);
             let histDate = new Date(Math.floor(bucket.zo_sql_timestamp / 1000));
             xData.push(Math.floor(histDate.getTime()));
-            yData.push(Number(bucket.duration.toFixed(2)));
+            yData.push(Number((bucket.duration / 1000).toFixed(2)));
           }
         );
       }
@@ -1064,7 +1063,7 @@ export default defineComponent({
       if (searchObj.loading == false) {
         // eslint-disable-next-line no-prototype-builtins
         if (!router.currentRoute.value.query.hasOwnProperty("query")) {
-          searchObj.data.editorValue = "duration>10";
+          searchObj.data.editorValue = "duration>1000";
         }
         loadPageData();
         restoreUrlQueryParams();

web/src/plugins/traces/SpanBlock.vue

@@ -110,12 +110,11 @@ export default defineComponent({
     const spanBlock: any = ref(null);
     const spanBlockWidth = ref(0);
     const onePixelPercent = ref(0);
-    function formatTimeWithSuffix(ns: number) {
-      if (ns < 10000) {
-        return `${ns}ms`;
-      } else {
-        return `${(ns / 1000).toFixed(2)}s`;
+    function formatTimeWithSuffix(us: number) {
+      if (us >= 1000 * 1000) {
+        return `${(us / 1000 / 1000).toFixed(2)}s`;
       }
+      return `${(us / 1000).toFixed(2)}ms`;
     }
     const isSpanSelected = computed(() => {
       if (!searchObj.data.traceDetails.selectedSpanId) return true;

web/src/plugins/traces/SpanRenderer.vue

@@ -70,12 +70,11 @@ export default defineComponent({
   },
   emits: ["toggleCollapse"],
   setup(props, { emit }) {
-    function formatTimeWithSuffix(ns: number) {
-      if (ns < 10000) {
-        return `${ns}ms`;
-      } else {
-        return `${(ns / 1000).toFixed(2)}s`;
+    function formatTimeWithSuffix(us: number) {
+      if (us >= 1000 * 1000) {
+        return `${(us / 1000 / 1000).toFixed(2)}s`;
       }
+      return `${(us / 1000).toFixed(2)}ms`;
     }
     function toggleSpanCollapse(spanId: number | string) {
       emit("toggleCollapse", spanId);

web/src/plugins/traces/TraceTree.vue

@@ -101,12 +101,11 @@ export default defineComponent({
   },
   emits: ["toggleCollapse"],
   setup(props, { emit }) {
-    function formatTimeWithSuffix(ns: number) {
-      if (ns < 10000) {
-        return `${ns} ms`;
-      } else {
-        return `${(ns / 1000).toFixed(2)} s`;
+    function formatTimeWithSuffix(us: number) {
+      if (us >= 1000 * 1000) {
+        return `${(us / 1000 / 1000).toFixed(2)}s`;
       }
+      return `${(us / 1000).toFixed(2)}ms`;
     }
     function toggleSpanCollapse(spanId: number | string) {
       emit("toggleCollapse", spanId);