From 87d2bb862af7612a02b1a208ad06287cdeab7391 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Sat, 26 Oct 2024 09:08:09 +0800 Subject: [PATCH] fix: bugs of v0.13.0-rc1 (#4916) --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/common/meta/organization.rs | 4 +- src/common/meta/stream.rs | 2 +- src/common/meta/telemetry.rs | 16 ++++---- src/common/migration/schema.rs | 5 +++ src/config/src/config.rs | 12 +++++- src/config/src/meta/search.rs | 3 +- src/config/src/meta/sql.rs | 3 +- src/config/src/meta/stream.rs | 26 ++++++------- src/config/src/meta/usage.rs | 4 +- .../http/request/enrichment_table/mod.rs | 8 ++-- src/handler/http/request/search/mod.rs | 2 +- src/handler/http/router/openapi.rs | 8 ++++ src/infra/src/cache/file_data/memory.rs | 4 +- src/infra/src/cache/stats.rs | 15 +++++--- src/infra/src/file_list/mod.rs | 4 +- src/infra/src/file_list/mysql.rs | 4 +- src/infra/src/file_list/postgres.rs | 4 +- src/infra/src/file_list/sqlite.rs | 4 +- src/job/files/parquet.rs | 38 +++++++++---------- src/job/metrics.rs | 4 +- src/service/compact/merge.rs | 37 ++++++++++-------- src/service/enrichment_table/mod.rs | 2 +- src/service/ingestion/mod.rs | 2 +- src/service/organization.rs | 4 +- src/service/stream.rs | 4 +- src/service/traces/mod.rs | 37 ++++++++++++------ src/service/usage/mod.rs | 2 +- 29 files changed, 151 insertions(+), 111 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6e71276221..3f6bb07751 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5891,7 +5891,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openobserve" -version = "0.12.0" +version = "0.13.0" dependencies = [ "actix", "actix-cors", diff --git a/Cargo.toml b/Cargo.toml index 9774ee13a8..d7c3f839f6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,7 @@ keywords = [ license = "AGPL-3.0-only" name = "openobserve" repository = "https://github.com/openobserve/openobserve/" -version = "0.12.0" +version = "0.13.0" publish = false # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html diff --git a/src/common/meta/organization.rs b/src/common/meta/organization.rs index 5d5dd25867..98da0b5d78 100644 --- a/src/common/meta/organization.rs +++ b/src/common/meta/organization.rs @@ -64,8 +64,8 @@ pub struct OrgSummary { #[derive(Serialize, Deserialize, ToSchema)] pub struct StreamSummary { pub num_streams: i64, - pub total_storage_size: f64, - pub total_compressed_size: f64, + pub total_storage_size: i64, + pub total_compressed_size: i64, } /// A container for passcodes and rumtokens diff --git a/src/common/meta/stream.rs b/src/common/meta/stream.rs index 1ebecb9dc1..79811ac49c 100644 --- a/src/common/meta/stream.rs +++ b/src/common/meta/stream.rs @@ -88,7 +88,7 @@ mod tests { #[test] fn test_stats() { let stats = StreamStats::default(); - let stats_str: String = stats.into(); + let stats_str: String = stats.clone().into(); let stats_frm_str = StreamStats::from(stats_str.as_str()); assert_eq!(stats, stats_frm_str); } diff --git a/src/common/meta/telemetry.rs b/src/common/meta/telemetry.rs index 0f62b610b7..e56cc95808 100644 --- a/src/common/meta/telemetry.rs +++ b/src/common/meta/telemetry.rs @@ -186,14 +186,14 @@ pub async fn add_zo_info(mut data: HashMap) -> HashMap Result { /// Migrate alerts, reports, templates, destination names with ofga compatible format pub async fn migrate_resource_names() -> Result<(), anyhow::Error> { + // fast path + if let Ok(false) = need_meta_resource_name_migration().await { + return Ok(()); // Resource name migration already done + } + // slow path let locker = infra::dist_lock::lock(META_MIGRATION_VERSION_KEY, 0, None).await?; match need_meta_resource_name_migration().await { Ok(true) => { diff --git a/src/config/src/config.rs b/src/config/src/config.rs index d17fff2927..8c32e46ce1 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -49,8 +49,8 @@ pub const MMDB_ASN_FILE_NAME: &str = "GeoLite2-ASN.mmdb"; pub const GEO_IP_CITY_ENRICHMENT_TABLE: &str = "maxmind_city"; pub const GEO_IP_ASN_ENRICHMENT_TABLE: &str = "maxmind_asn"; -pub const SIZE_IN_MB: f64 = 1024.0 * 1024.0; -pub const SIZE_IN_GB: f64 = 1024.0 * 1024.0 * 1024.0; +pub const SIZE_IN_MB: i64 = 1024 * 1024; +pub const SIZE_IN_GB: i64 = 1024 * 1024 * 1024; pub const PARQUET_BATCH_SIZE: usize = 8 * 1024; pub const PARQUET_PAGE_SIZE: usize = 1024 * 1024; pub const PARQUET_MAX_ROW_GROUP_SIZE: usize = 1024 * 1024; // this can't be change, it will cause segment matching error @@ -1524,6 +1524,14 @@ fn check_common_config(cfg: &mut Config) -> Result<(), anyhow::Error> { { cfg.common.meta_store_external = true; } + if !cfg.common.local_mode + && !cfg.common.meta_store.starts_with("postgres") + && !cfg.common.meta_store.starts_with("mysql") + { + return Err(anyhow::anyhow!( + "Meta store only support mysql or postgres in cluster mode." + )); + } if cfg.common.meta_store.starts_with("postgres") && cfg.common.meta_postgres_dsn.is_empty() { return Err(anyhow::anyhow!( "Meta store is PostgreSQL, you must set ZO_META_POSTGRES_DSN" diff --git a/src/config/src/meta/search.rs b/src/config/src/meta/search.rs index d3a4a93ed0..866a0bba35 100644 --- a/src/config/src/meta/search.rs +++ b/src/config/src/meta/search.rs @@ -21,6 +21,7 @@ use utoipa::ToSchema; use crate::{ ider, + meta::sql::OrderBy, utils::{base64, json}, }; @@ -371,7 +372,7 @@ pub struct SearchPartitionResponse { pub histogram_interval: Option, // seconds, for histogram pub max_query_range: i64, // hours, for histogram pub partitions: Vec<[i64; 2]>, - pub order_by: super::sql::OrderBy, + pub order_by: OrderBy, } #[derive(Clone, Debug, Default, Deserialize, ToSchema)] diff --git a/src/config/src/meta/sql.rs b/src/config/src/meta/sql.rs index a4bb05a1e0..6ec6f27002 100644 --- a/src/config/src/meta/sql.rs +++ b/src/config/src/meta/sql.rs @@ -26,13 +26,14 @@ use sqlparser::{ dialect::PostgreSqlDialect, parser::Parser, }; +use utoipa::ToSchema; use crate::get_config; pub const MAX_LIMIT: i64 = 100000; pub const MAX_OFFSET: i64 = 100000; -#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, ToSchema, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub enum OrderBy { #[default] diff --git a/src/config/src/meta/stream.rs b/src/config/src/meta/stream.rs index d4ab68036c..9c27862cee 100644 --- a/src/config/src/meta/stream.rs +++ b/src/config/src/meta/stream.rs @@ -247,15 +247,15 @@ impl From<&String> for MergeStrategy { } } -#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize, Deserialize, ToSchema)] +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, ToSchema)] pub struct StreamStats { pub created_at: i64, pub doc_time_min: i64, pub doc_time_max: i64, pub doc_num: i64, pub file_num: i64, - pub storage_size: f64, - pub compressed_size: f64, + pub storage_size: i64, + pub compressed_size: i64, } impl StreamStats { @@ -292,16 +292,16 @@ impl StreamStats { self.doc_num = max(0, self.doc_num + meta.records); self.doc_time_min = self.doc_time_min.min(meta.min_ts); self.doc_time_max = self.doc_time_max.max(meta.max_ts); - self.storage_size += meta.original_size as f64; - self.compressed_size += meta.compressed_size as f64; + self.storage_size += meta.original_size; + self.compressed_size += meta.compressed_size; if self.doc_time_min == 0 { self.doc_time_min = meta.min_ts; } - if self.storage_size < 0.0 { - self.storage_size = 0.0; + if self.storage_size < 0 { + self.storage_size = 0; } - if self.compressed_size < 0.0 { - self.compressed_size = 0.0; + if self.compressed_size < 0 { + self.compressed_size = 0; } } @@ -344,8 +344,8 @@ impl From for StreamStats { doc_time_max: meta.max_ts, doc_num: meta.records, file_num: 0, - storage_size: meta.original_size, - compressed_size: meta.compressed_size.unwrap_or_default(), + storage_size: meta.original_size as i64, + compressed_size: meta.compressed_size.unwrap_or_default() as i64, } } } @@ -360,8 +360,8 @@ impl std::ops::Sub for StreamStats { doc_num: self.doc_num - rhs.records, doc_time_min: self.doc_time_min.min(rhs.min_ts), doc_time_max: self.doc_time_max.max(rhs.max_ts), - storage_size: self.storage_size - rhs.original_size as f64, - compressed_size: self.compressed_size - rhs.compressed_size as f64, + storage_size: self.storage_size - rhs.original_size, + compressed_size: self.compressed_size - rhs.compressed_size, }; if ret.doc_time_min == 0 { ret.doc_time_min = rhs.min_ts; diff --git a/src/config/src/meta/usage.rs b/src/config/src/meta/usage.rs index c84cda6360..ddb01668b2 100644 --- a/src/config/src/meta/usage.rs +++ b/src/config/src/meta/usage.rs @@ -302,13 +302,13 @@ impl Default for RequestStats { impl From for RequestStats { fn from(meta: FileMeta) -> RequestStats { RequestStats { - size: meta.original_size as f64 / SIZE_IN_MB, + size: (meta.original_size / SIZE_IN_MB) as f64, records: meta.records, response_time: 0.0, function: None, request_body: None, cached_ratio: None, - compressed_size: Some(meta.compressed_size as f64 / SIZE_IN_MB), + compressed_size: Some((meta.compressed_size / SIZE_IN_MB) as f64), min_ts: Some(meta.min_ts), max_ts: Some(meta.max_ts), user_email: None, diff --git a/src/handler/http/request/enrichment_table/mod.rs b/src/handler/http/request/enrichment_table/mod.rs index 006c397384..fa424acb0b 100644 --- a/src/handler/http/request/enrichment_table/mod.rs +++ b/src/handler/http/request/enrichment_table/mod.rs @@ -51,18 +51,18 @@ pub async fn save_enrichment_table( let (org_id, table_name) = path.into_inner(); let content_type = req.headers().get("content-type"); let content_length = match req.headers().get("content-length") { - None => 0.0, + None => 0, Some(content_length) => { content_length .to_str() .unwrap_or("0") - .parse::() - .unwrap_or(0.0) + .parse::() + .unwrap_or(0) / SIZE_IN_MB } }; let cfg = config::get_config(); - if content_length > cfg.limit.enrichment_table_limit as f64 { + if content_length > cfg.limit.enrichment_table_limit as i64 { return Ok(MetaHttpResponse::bad_request(format!( "exceeds allowed limit of {} mb", cfg.limit.enrichment_table_limit diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 98b701c5d6..3cc5f6f8b0 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -1420,7 +1420,7 @@ pub async fn search_partition( }) ), responses( - (status = 200, description = "Success", content_type = "application/json", body = SearchHistoryResponse, example = json ! ({ + (status = 200, description = "Success", content_type = "application/json", body = SearchResponse, example = json ! ({ "took": 40, "took_detail": { "total": 0, diff --git a/src/handler/http/router/openapi.rs b/src/handler/http/router/openapi.rs index b14dfc55ce..b3811688b0 100644 --- a/src/handler/http/router/openapi.rs +++ b/src/handler/http/router/openapi.rs @@ -131,6 +131,9 @@ use crate::{common::meta, handler::http::request}; config::meta::stream::StreamPartitionType, config::meta::stream::StreamStats, config::meta::stream::PartitionTimeLevel, + config::meta::stream::UpdateStreamPartition, + config::meta::stream::UpdateStreamSettings, + config::meta::stream::UpdateStringSettingsArray, meta::ingestion::RecordStatus, meta::ingestion::StreamStatus, meta::ingestion::IngestionResponse, @@ -151,6 +154,7 @@ use crate::{common::meta, handler::http::request}; meta::dashboards::Folder, meta::dashboards::MoveDashboard, meta::dashboards::FolderList, + config::meta::sql::OrderBy, config::meta::search::Query, config::meta::search::Request, config::meta::search::RequestEncoding, @@ -160,11 +164,14 @@ use crate::{common::meta, handler::http::request}; config::meta::search::SearchEventType, config::meta::search::SearchPartitionRequest, config::meta::search::SearchPartitionResponse, + config::meta::search::SearchHistoryRequest, config::meta::search::CancelQueryResponse, config::meta::search::QueryStatusResponse, config::meta::search::QueryStatus, config::meta::search::QueryInfo, config::meta::search::ScanStats, + config::meta::short_url::ShortenUrlRequest, + config::meta::short_url::ShortenUrlResponse, meta::saved_view::View, meta::saved_view::ViewWithoutData, meta::saved_view::ViewsWithoutData, @@ -181,6 +188,7 @@ use crate::{common::meta, handler::http::request}; meta::alerts::TriggerCondition, meta::alerts::FrequencyType, meta::alerts::QueryCondition, + meta::alerts::CompareHistoricData, meta::alerts::destinations::Destination, meta::alerts::destinations::DestinationWithTemplate, meta::alerts::destinations::HTTPType, diff --git a/src/infra/src/cache/file_data/memory.rs b/src/infra/src/cache/file_data/memory.rs index ec7a15eb10..0cb93fb801 100644 --- a/src/infra/src/cache/file_data/memory.rs +++ b/src/infra/src/cache/file_data/memory.rs @@ -99,7 +99,7 @@ impl FileData { ); // cache is full, need release some space let need_release_size = min( - self.max_size, + self.cur_size, max(get_config().memory_cache.release_size, data_size * 100), ); self.gc(trace_id, need_release_size).await?; @@ -134,7 +134,7 @@ impl FileData { loop { let item = self.data.remove(); if item.is_none() { - log::error!( + log::warn!( "[trace_id {trace_id}] File memory cache is corrupt, it shouldn't be none" ); break; diff --git a/src/infra/src/cache/stats.rs b/src/infra/src/cache/stats.rs index 31bad6d2a5..3a54859034 100644 --- a/src/infra/src/cache/stats.rs +++ b/src/infra/src/cache/stats.rs @@ -30,7 +30,10 @@ pub fn get_stats() -> RwHashMap { #[inline] pub fn get_stream_stats(org_id: &str, stream_name: &str, stream_type: StreamType) -> StreamStats { let key = format!("{org_id}/{stream_type}/{stream_name}"); - STATS.get(&key).map(|v| *v.value()).unwrap_or_default() + STATS + .get(&key) + .map(|v| v.value().clone()) + .unwrap_or_default() } #[inline] @@ -78,8 +81,8 @@ pub fn incr_stream_stats(key: &str, val: &FileMeta) -> Result<(), anyhow::Error> } stats.doc_num += val.records; stats.file_num += 1; - stats.storage_size += val.original_size as f64; - stats.compressed_size += val.compressed_size as f64; + stats.storage_size += val.original_size; + stats.compressed_size += val.compressed_size; Ok(()) } @@ -112,11 +115,11 @@ mod tests { doc_time_max: 1667978845374, doc_num: 5000, file_num: 1, - storage_size: 200.00, - compressed_size: 3.00, + storage_size: 200, + compressed_size: 3, }; - set_stream_stats("nexus", "default", StreamType::Logs, val); + set_stream_stats("nexus", "default", StreamType::Logs, val.clone()); let stats = get_stream_stats("nexus", "default", StreamType::Logs); assert_eq!(stats, val); diff --git a/src/infra/src/file_list/mod.rs b/src/infra/src/file_list/mod.rs index 8c5bf2805c..a2c2054a37 100644 --- a/src/infra/src/file_list/mod.rs +++ b/src/infra/src/file_list/mod.rs @@ -491,8 +491,8 @@ impl From<&StatsRecord> for StreamStats { doc_time_max: record.max_ts, doc_num: record.records, file_num: record.file_num, - storage_size: record.original_size as f64, - compressed_size: record.compressed_size as f64, + storage_size: record.original_size, + compressed_size: record.compressed_size, } } } diff --git a/src/infra/src/file_list/mysql.rs b/src/infra/src/file_list/mysql.rs index 538459e5b7..710a228aa3 100644 --- a/src/infra/src/file_list/mysql.rs +++ b/src/infra/src/file_list/mysql.rs @@ -782,8 +782,8 @@ UPDATE stream_stats .bind(stats.doc_time_min) .bind(stats.doc_time_max) .bind(stats.doc_num) - .bind(stats.storage_size as i64) - .bind(stats.compressed_size as i64) + .bind(stats.storage_size) + .bind(stats.compressed_size) .bind(stream_key) .execute(&mut *tx) .await diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index 2f07ca15c2..b2a277178e 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -799,8 +799,8 @@ UPDATE stream_stats .bind(stats.doc_time_min) .bind(stats.doc_time_max) .bind(stats.doc_num) - .bind(stats.storage_size as i64) - .bind(stats.compressed_size as i64) + .bind(stats.storage_size) + .bind(stats.compressed_size) .bind(stream_key) .execute(&mut *tx).await { diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs index a7d2347cba..3a031ac22f 100644 --- a/src/infra/src/file_list/sqlite.rs +++ b/src/infra/src/file_list/sqlite.rs @@ -706,8 +706,8 @@ UPDATE stream_stats .bind(stats.doc_time_min) .bind(stats.doc_time_max) .bind(stats.doc_num) - .bind(stats.storage_size as i64) - .bind(stats.compressed_size as i64) + .bind(stats.storage_size) + .bind(stats.compressed_size) .bind(stream_key) .execute(&mut *tx) .await diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index d79043d9a8..554e443a8f 100644 --- a/src/job/files/parquet.rs +++ b/src/job/files/parquet.rs @@ -38,9 +38,7 @@ use config::{ bitvec::BitVec, inverted_index::{writer::ColumnIndexer, IndexFileMetas, InvertedIndexFormat}, puffin::writer::PuffinBytesWriter, - stream::{ - FileKey, FileMeta, PartitionTimeLevel, StreamPartition, StreamSettings, StreamType, - }, + stream::{FileKey, FileMeta, PartitionTimeLevel, StreamSettings, StreamType}, }, metrics, utils::{ @@ -832,10 +830,9 @@ pub(crate) async fn generate_index_on_ingester( "generate_index_on_ingester create schema error: schema not found" )); }; - // update schema to enable bloomfilter for field: term, file_name + // update schema to enable bloomfilter for field: term let settings = StreamSettings { bloom_filter_fields: vec!["term".to_string()], - partition_keys: vec![StreamPartition::new_prefix("term")], ..Default::default() }; let mut metadata = schema.metadata().clone(); @@ -872,23 +869,24 @@ pub(crate) async fn generate_index_on_ingester( }; } + // TODO: disable it, because the prefix partition key will cause the file_list much bigger // add prefix partition for index <= v0.12.1 - if let Some(settings) = stream_setting.as_mut() { - let term_partition_exists = settings - .partition_keys - .iter() - .any(|partition| partition.field == "term"); - if !term_partition_exists { - settings - .partition_keys - .push(StreamPartition::new_prefix("term")); + // if let Some(settings) = stream_setting.as_mut() { + // let term_partition_exists = settings + // .partition_keys + // .iter() + // .any(|partition| partition.field == "term"); + // if !term_partition_exists { + // settings + // .partition_keys + // .push(StreamPartition::new_prefix("term")); - let mut metadata = schema.schema().metadata().clone(); - metadata.insert("settings".to_string(), json::to_string(&settings).unwrap()); - db::schema::update_setting(org_id, &index_stream_name, StreamType::Index, metadata) - .await?; - } - } + // let mut metadata = schema.schema().metadata().clone(); + // metadata.insert("settings".to_string(), json::to_string(&settings).unwrap()); + // db::schema::update_setting(org_id, &index_stream_name, StreamType::Index, + // metadata) .await?; + // } + // } } let schema_key = idx_schema.hash_key(); diff --git a/src/job/metrics.rs b/src/job/metrics.rs index b56f0f1949..d855424b3a 100644 --- a/src/job/metrics.rs +++ b/src/job/metrics.rs @@ -264,10 +264,10 @@ async fn update_storage_metrics() -> Result<(), anyhow::Error> { let columns = key.split('/').collect::>(); metrics::STORAGE_ORIGINAL_BYTES .with_label_values(&[columns[0], columns[2], columns[1]]) - .set(stat.storage_size as i64); + .set(stat.storage_size); metrics::STORAGE_COMPRESSED_BYTES .with_label_values(&[columns[0], columns[2], columns[1]]) - .set(stat.compressed_size as i64); + .set(stat.compressed_size); metrics::STORAGE_FILES .with_label_values(&[columns[0], columns[2], columns[1]]) .set(stat.file_num); diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs index 00e2a02049..2f7c306d6d 100644 --- a/src/service/compact/merge.rs +++ b/src/service/compact/merge.rs @@ -350,9 +350,6 @@ pub async fn merge_by_stream( partition.push(file.to_owned()); } - // collect stream stats - let mut stream_stats = StreamStats::default(); - // use multiple threads to merge let semaphore = std::sync::Arc::new(Semaphore::new(cfg.limit.file_merge_thread_num)); let mut tasks = Vec::with_capacity(partition_files_with_size.len()); @@ -473,6 +470,9 @@ pub async fn merge_by_stream( deleted: false, segment_ids: None, }); + + // collect stream stats + let mut stream_stats: StreamStats = StreamStats::default(); for file in new_file_list.iter() { stream_stats = stream_stats - file.meta.clone(); events.push(FileKey { @@ -486,7 +486,24 @@ pub async fn merge_by_stream( // write file list to storage match write_file_list(&org_id, &events).await { - Ok(_) => {} + Ok(_) => { + // update stream stats + if stream_stats.doc_num != 0 { + let stream_key = format!("{org_id}/{stream_type}/{stream_name}"); + if let Err(e) = infra_file_list::set_stream_stats( + &org_id, + &[(stream_key.clone(), stream_stats)], + ) + .await + { + log::error!( + "[COMPACT] set_stream_stats failed: {}, err: {}", + stream_key, + e + ); + } + } + } Err(e) => { log::error!("[COMPACT] write file list failed: {}", e); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; @@ -512,18 +529,6 @@ pub async fn merge_by_stream( log::error!("[COMPACT] set_job_done failed: {e}"); } - // update stream stats - if stream_stats.doc_num != 0 { - infra_file_list::set_stream_stats( - org_id, - &[( - format!("{org_id}/{stream_type}/{stream_name}"), - stream_stats, - )], - ) - .await?; - } - // metrics let time = start.elapsed().as_secs_f64(); metrics::COMPACT_USED_TIME diff --git a/src/service/enrichment_table/mod.rs b/src/service/enrichment_table/mod.rs index 88ce63a99b..5fca9daf04 100644 --- a/src/service/enrichment_table/mod.rs +++ b/src/service/enrichment_table/mod.rs @@ -99,7 +99,7 @@ pub async fn save_enrichment_data( stats, max_enrichment_table_size ); - if (stats.storage_size / SIZE_IN_MB) > max_enrichment_table_size as f64 { + if (stats.storage_size / SIZE_IN_MB) > max_enrichment_table_size as i64 { return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), diff --git a/src/service/ingestion/mod.rs b/src/service/ingestion/mod.rs index 0afd0c21fc..271f214635 100644 --- a/src/service/ingestion/mod.rs +++ b/src/service/ingestion/mod.rs @@ -475,7 +475,7 @@ pub async fn write_file( for task in task_results { match task { Ok((entry_records, entry_size)) => { - req_stats.size += entry_size as f64 / SIZE_IN_MB; + req_stats.size += (entry_size as i64 / SIZE_IN_MB) as f64; req_stats.records += entry_records as i64; } Err(e) => { diff --git a/src/service/organization.rs b/src/service/organization.rs index 5111a7dec3..adb50352fc 100644 --- a/src/service/organization.rs +++ b/src/service/organization.rs @@ -37,8 +37,8 @@ pub async fn get_summary(org_id: &str) -> OrgSummary { let functions = db::functions::list(org_id).await.unwrap(); let alerts = db::alerts::alert::list(org_id, None, None).await.unwrap(); let mut num_streams = 0; - let mut total_storage_size = 0.00; - let mut total_compressed_size = 0.00; + let mut total_storage_size = 0; + let mut total_compressed_size = 0; for stream in streams.iter() { if !stream.stream_type.eq(&StreamType::Index) && !stream.stream_type.eq(&StreamType::Metadata) diff --git a/src/service/stream.rs b/src/service/stream.rs index 71d3df796b..6c54cab4d4 100644 --- a/src/service/stream.rs +++ b/src/service/stream.rs @@ -475,8 +475,6 @@ pub async fn delete_stream( fn transform_stats(stats: &mut StreamStats) { stats.storage_size /= SIZE_IN_MB; stats.compressed_size /= SIZE_IN_MB; - stats.storage_size = (stats.storage_size * 100.0).round() / 100.0; - stats.compressed_size = (stats.compressed_size * 100.0).round() / 100.0; } pub fn stream_created(schema: &Schema) -> Option { @@ -520,7 +518,7 @@ mod tests { fn test_stream_res() { let stats = StreamStats::default(); let schema = Schema::new(vec![Field::new("f.c", DataType::Int32, false)]); - let res = stream_res("Test", StreamType::Logs, schema, Some(stats)); + let res = stream_res("Test", StreamType::Logs, schema, Some(stats.clone())); assert_eq!(res.stats, stats); } } diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index aef2350de8..e48d242cdb 100644 --- a/src/service/traces/mod.rs +++ b/src/service/traces/mod.rs @@ -212,13 +212,6 @@ pub async fn handle_trace_request( for inst_span in inst_resources { let spans = inst_span.spans; for span in spans { - if span.span_id.len() != SPAN_ID_BYTES_COUNT { - log::error!("[TRACE] skipping span with invalid span id"); - partial_success.rejected_spans += 1; - continue; - } - let span_id: String = - SpanId::from_bytes(span.span_id.try_into().unwrap()).to_string(); if span.trace_id.len() != TRACE_ID_BYTES_COUNT { log::error!("[TRACE] skipping span with invalid trace id"); partial_success.rejected_spans += 1; @@ -226,6 +219,16 @@ pub async fn handle_trace_request( } let trace_id: String = TraceId::from_bytes(span.trace_id.try_into().unwrap()).to_string(); + if span.span_id.len() != SPAN_ID_BYTES_COUNT { + log::error!( + "[TRACE] skipping span with invalid span id, trace_id: {}", + trace_id + ); + partial_success.rejected_spans += 1; + continue; + } + let span_id: String = + SpanId::from_bytes(span.span_id.try_into().unwrap()).to_string(); let mut span_ref = HashMap::new(); if !span.parent_span_id.is_empty() && span.parent_span_id.len() == SPAN_ID_BYTES_COUNT @@ -277,13 +280,19 @@ pub async fn handle_trace_request( link_att_map.insert(link_att.key, get_val(&link_att.value.as_ref())); } if link.span_id.len() != SPAN_ID_BYTES_COUNT { - log::error!("[TRACE] skipping link with invalid span id"); + log::error!( + "[TRACE] skipping link with invalid span id, trace_id: {}", + trace_id + ); continue; } let span_id: String = SpanId::from_bytes(link.span_id.try_into().unwrap()).to_string(); if link.trace_id.len() != TRACE_ID_BYTES_COUNT { - log::error!("[TRACE] skipping link with invalid trace id"); + log::error!( + "[TRACE] skipping link with invalid trace id, trace_id: {}", + trace_id + ); continue; } let trace_id: String = @@ -303,7 +312,8 @@ pub async fn handle_trace_request( let timestamp = (start_time / 1000) as i64; if timestamp < min_ts { log::error!( - "[TRACE] skipping span with timestamp older than allowed retention period" + "[TRACE] skipping span with timestamp older than allowed retention period, trace_id: {}", + trace_id ); partial_success.rejected_spans += 1; continue; @@ -373,7 +383,10 @@ pub async fn handle_trace_request( v } _ => { - log::error!("[TRACE] stream functions did not return valid json object"); + log::error!( + "[TRACE] stream functions did not return valid json object, trace_id: {}", + trace_id + ); return Ok(HttpResponse::InternalServerError().json( MetaHttpResponse::error( http::StatusCode::INTERNAL_SERVER_ERROR.into(), @@ -436,7 +449,7 @@ pub async fn handle_trace_request( // send to metrics job if let Err(e) = crate::job::metrics::TRACE_METRICS_CHAN.0.try_send(m) { - log::error!("traces metrics item send to job fail : {e}") + log::error!("traces metrics item send to job fail: {e}") } } diff --git a/src/service/usage/mod.rs b/src/service/usage/mod.rs index 519c144541..500fbb84e7 100644 --- a/src/service/usage/mod.rs +++ b/src/service/usage/mod.rs @@ -59,7 +59,7 @@ pub async fn report_request_usage_stats( .inc_by(stats.records as u64); metrics::INGEST_BYTES .with_label_values(&[org_id, stream_name, stream_type.to_string().as_str()]) - .inc_by((stats.size * SIZE_IN_MB) as u64); + .inc_by((stats.size as i64 * SIZE_IN_MB) as u64); let event: UsageEvent = usage_type.into(); let now = DateTime::from_timestamp_micros(timestamp).unwrap();