From b4f3dc4c3c2bd95fa96e346b051332df1bb03368 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Sat, 9 Sep 2023 23:32:34 +0800 Subject: [PATCH] fix: stream stats with sqlite (#1477) --- src/common/infra/file_list/dynamo.rs | 4 +++ src/common/infra/file_list/mod.rs | 6 +++++ src/common/infra/file_list/postgres.rs | 8 ++++++ src/common/infra/file_list/sqlite.rs | 18 +++++++++----- src/job/mod.rs | 9 ++++--- src/job/stats.rs | 22 ++--------------- src/main.rs | 23 ++++++++++++----- src/service/compact/mod.rs | 4 +-- src/service/db/file_list/remote.rs | 8 +++--- src/service/traces/mod.rs | 2 +- src/service/traces/otlp_http.rs | 2 +- web/src/plugins/traces/Index.vue | 33 ++++++++++++------------- web/src/plugins/traces/SpanBlock.vue | 9 +++---- web/src/plugins/traces/SpanRenderer.vue | 9 +++---- web/src/plugins/traces/TraceTree.vue | 9 +++---- 15 files changed, 90 insertions(+), 76 deletions(-) diff --git a/src/common/infra/file_list/dynamo.rs b/src/common/infra/file_list/dynamo.rs index a9d3b75d0a..92b2453590 100644 --- a/src/common/infra/file_list/dynamo.rs +++ b/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, diff --git a/src/common/infra/file_list/mod.rs b/src/common/infra/file_list/mod.rs index d76d5bfb76..2bcb6f0c13 100644 --- a/src/common/infra/file_list/mod.rs +++ b/src/common/infra/file_list/mod.rs @@ -79,6 +79,7 @@ pub trait FileList: Sync + Send + 'static { ) -> Result>; 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 diff --git a/src/common/infra/file_list/postgres.rs b/src/common/infra/file_list/postgres.rs index e593534136..23d5f06478 100644 --- a/src/common/infra/file_list/postgres.rs +++ b/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, diff --git a/src/common/infra/file_list/sqlite.rs b/src/common/infra/file_list/sqlite.rs index 338b442237..ff473b9c84 100644 --- a/src/common/infra/file_list/sqlite.rs +++ b/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) diff --git a/src/job/mod.rs b/src/job/mod.rs index dbe065ab37..a25f2c1557 100644 --- a/src/job/mod.rs +++ b/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) { diff --git a/src/job/stats.rs b/src/job/stats.rs index c9706e7805..56c3bb4549 100644 --- a/src/job/stats.rs +++ b/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::>(); - 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); } } } diff --git a/src/main.rs b/src/main.rs index d80caf2d64..ec60a72730 100644 --- a/src/main.rs +++ b/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 { .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 { .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 { .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 { "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)); } diff --git a/src/service/compact/mod.rs b/src/service/compact/mod.rs index c6e716c057..c67edc1238 100644 --- a/src/service/compact/mod.rs +++ b/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>> = Lazy::new(|| Arc::new(Mutex::const_new(false))); diff --git a/src/service/db/file_list/remote.rs b/src/service/db/file_list/remote.rs index 9ed8400153..8b87ef61c4 100644 --- a/src/service/db/file_list/remote.rs +++ b/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::>(); 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()); diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index 1cf3024cee..52747eadc6 100644 --- a/src/service/traces/mod.rs +++ b/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, diff --git a/src/service/traces/otlp_http.rs b/src/service/traces/otlp_http.rs index 8ec0589249..b83be8eba1 100644 --- a/src/service/traces/otlp_http.rs +++ b/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, diff --git a/web/src/plugins/traces/Index.vue b/web/src/plugins/traces/Index.vue index ec1680a0bd..b451d6ac7b 100644 --- a/web/src/plugins/traces/Index.vue +++ b/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(); diff --git a/web/src/plugins/traces/SpanBlock.vue b/web/src/plugins/traces/SpanBlock.vue index e2980bf87a..d5de30c908 100644 --- a/web/src/plugins/traces/SpanBlock.vue +++ b/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; diff --git a/web/src/plugins/traces/SpanRenderer.vue b/web/src/plugins/traces/SpanRenderer.vue index 3ac413d23f..9b463648a8 100644 --- a/web/src/plugins/traces/SpanRenderer.vue +++ b/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); diff --git a/web/src/plugins/traces/TraceTree.vue b/web/src/plugins/traces/TraceTree.vue index 1333fdea78..5e7f8eded9 100644 --- a/web/src/plugins/traces/TraceTree.vue +++ b/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);