diff --git a/src/common/infra/ofga/migrations/alert_folders.rs b/src/common/infra/ofga/migrations/alert_folders.rs index 978cf89f76..a5b11d8fd4 100644 --- a/src/common/infra/ofga/migrations/alert_folders.rs +++ b/src/common/infra/ofga/migrations/alert_folders.rs @@ -78,8 +78,8 @@ pub async fn migrate_alert_folders(db: &C) -> Result<(), any let object = authorizer::authz::get_ownership_tuple(&org_id, "alerts", &alert_id, &mut tuples); authorizer::authz::get_parent_tuple( - &alert_folder_id, - &alert_folders_ofga_type, + alert_folder_id, + alert_folders_ofga_type, &object, &mut tuples, ); @@ -110,25 +110,20 @@ pub async fn migrate_alert_folders(db: &C) -> Result<(), any // 2. Get all the alerts assigned to the roles for role in roles.iter() { let mut add_roles = vec![]; - let alerts = match authorizer::roles::get_role_permissions( - org, - &role, - &alerts_ofga_type, - ) - .await - { - Ok(alerts) => alerts, - Err(e) => { - log::error!("Error openfga getting alerts for role: {}", e); - continue; - } - }; + let alerts = + match authorizer::roles::get_role_permissions(org, role, alerts_ofga_type).await { + Ok(alerts) => alerts, + Err(e) => { + log::error!("Error openfga getting alerts for role: {}", e); + continue; + } + }; // 3. Add the role assignment tuples for the alerts for alert in alerts.iter() { // Get the alert id from the alert name // TODO: Optimize this workflow - let alert_name = match alert.object.split(':').last() { + let alert_name = match alert.object.split(':').next_back() { Some(alert_name) => alert_name, None => { log::error!("Error openfga getting alert id from alert name"); @@ -161,8 +156,7 @@ pub async fn migrate_alert_folders(db: &C) -> Result<(), any } if !add_roles.is_empty() { let add_roles_len = add_roles.len(); - match authorizer::roles::update_role(&org, &role, add_roles, vec![], None, None) - .await + match authorizer::roles::update_role(org, role, add_roles, vec![], None, None).await { Ok(_) => { log::debug!("{} roles added to openfga", add_roles_len); diff --git a/src/common/infra/ofga/mod.rs b/src/common/infra/ofga/mod.rs index cbc0d60852..27f74b33dc 100644 --- a/src/common/infra/ofga/mod.rs +++ b/src/common/infra/ofga/mod.rs @@ -75,10 +75,20 @@ pub async fn init() -> Result<(), anyhow::Error> { (Some(model), Some(existing_model)) => match model.version.cmp(&existing_model.version) { Ordering::Less => { + log::info!( + "[OFGA:SuperCluster] model version changed: {} -> {}", + existing_model.version, + model.version + ); // update version in super cluster set_model(Some(existing_model.clone())).await?; } Ordering::Greater => { + log::info!( + "[OFGA:SuperCluster] model version changed: {} -> {}", + existing_model.version, + model.version + ); // update version in local existing_meta = Some(model.clone()); migrate_native_objects = false; @@ -94,11 +104,11 @@ pub async fn init() -> Result<(), anyhow::Error> { get_all_init_tuples(&mut init_tuples).await; if let Some(existing_model) = &existing_meta { if meta.version == existing_model.version { - log::info!("OFGA model already exists & no changes required"); + log::info!("[OFGA:Local] model already exists & no changes required"); if !init_tuples.is_empty() { match update_tuples(init_tuples, vec![]).await { Ok(_) => { - log::info!("Data migrated to openfga"); + log::info!("[OFGA:Local] Data migrated to openfga"); } Err(e) => { log::error!( @@ -110,6 +120,13 @@ pub async fn init() -> Result<(), anyhow::Error> { } return Ok(()); } + + log::info!( + "[OFGA:Local] model version changed: {} -> {}", + existing_model.version, + model.version + ); + // Check if ofga migration of index streams are needed let meta_version = version_compare::Version::from(&meta.version).unwrap(); let existing_model_version = @@ -135,7 +152,7 @@ pub async fn init() -> Result<(), anyhow::Error> { need_action_scripts_migration = true; } if meta_version > v0_0_12 && existing_model_version < v0_0_13 { - log::info!("Alert folders migration needed"); + log::info!("[OFGA:Local] Alert folders migration needed"); need_alert_folders_migration = true; } } @@ -147,7 +164,7 @@ pub async fn init() -> Result<(), anyhow::Error> { match db::ofga::set_ofga_model(existing_meta).await { Ok(store_id) => { if store_id.is_empty() { - log::error!("OFGA store id is empty"); + log::error!("[OFGA:Local] OFGA store id is empty"); } o2_openfga::config::OFGA_STORE_ID.insert("store_id".to_owned(), store_id); @@ -169,7 +186,7 @@ pub async fn init() -> Result<(), anyhow::Error> { get_tuple_for_new_index(org_name, split_key[2], &mut tuples); } } - log::info!("Migrating native objects"); + log::info!("[OFGA:Local] Migrating native objects"); if migrate_native_objects { for org_name in orgs { get_org_creation_tuples( @@ -213,7 +230,7 @@ pub async fn init() -> Result<(), anyhow::Error> { } } } else { - log::info!("Migrating index streams"); + log::info!("[OFGA:Local] Migrating index streams"); for org_name in orgs { if need_migrate_index_streams { get_index_creation_tuples(org_name, &mut tuples).await; @@ -248,10 +265,13 @@ pub async fn init() -> Result<(), anyhow::Error> { if need_alert_folders_migration { match migrations::migrate_alert_folders().await { Ok(_) => { - log::info!("Alert folders migrated to openfga"); + log::info!("[OFGA:Local] Alert folders migrated to openfga"); } Err(e) => { - log::error!("Error migrating alert folders to openfga: {}", e); + log::error!( + "[OFGA:Local] Error migrating alert folders to openfga: {}", + e + ); } } } @@ -263,12 +283,12 @@ pub async fn init() -> Result<(), anyhow::Error> { } if tuples.is_empty() { - log::info!("No orgs to update to the openfga"); + log::info!("[OFGA:Local] No orgs to update to the openfga"); } else { - log::debug!("tuples not empty: {:#?}", tuples); + log::debug!("[OFGA:Local] tuples not empty: {:#?}", tuples); match update_tuples(tuples, vec![]).await { Ok(_) => { - log::info!("Data migrated to openfga"); + log::info!("[OFGA:Local] Data migrated to openfga"); } Err(e) => { log::error!( diff --git a/src/job/stats.rs b/src/job/stats.rs index 690034bf58..30c2c0f318 100644 --- a/src/job/stats.rs +++ b/src/job/stats.rs @@ -67,7 +67,18 @@ async fn cache_stream_stats() -> Result<(), anyhow::Error> { 300, get_config().limit.calculate_stats_interval, ))); - interval.tick().await; // trigger the first run + + #[cfg(feature = "enterprise")] + let need_wait_one_around = o2_enterprise::enterprise::common::infra::config::get_config() + .super_cluster + .enabled; + #[cfg(not(feature = "enterprise"))] + let need_wait_one_around = false; + if need_wait_one_around { + // wait one around to make sure the dependent models are ready + interval.tick().await; + } + loop { interval.tick().await; if let Err(e) = db::file_list::cache_stats().await { diff --git a/src/service/db/ofga.rs b/src/service/db/ofga.rs index dc70565b39..2a6c3bce70 100644 --- a/src/service/db/ofga.rs +++ b/src/service/db/ofga.rs @@ -79,9 +79,9 @@ pub async fn set_ofga_model_to_db(mut meta: OFGAModel) -> Result Result, anyhow::Error> { let key = "/ofga/model"; let ret = db::get(key).await?; - let loc_value = json::from_slice(&ret).unwrap(); - let value = Some(loc_value); - Ok(value) + let mut loc_value: OFGAModel = json::from_slice(&ret).unwrap(); + loc_value.version = loc_value.version.trim_matches('"').to_string(); + Ok(Some(loc_value)) } pub async fn watch() -> Result<(), anyhow::Error> { diff --git a/src/service/metrics/otlp.rs b/src/service/metrics/otlp.rs index 4e9ed32c68..1a463416eb 100644 --- a/src/service/metrics/otlp.rs +++ b/src/service/metrics/otlp.rs @@ -36,7 +36,7 @@ use config::{ utils::{flatten, json, schema_ext::SchemaExt}, }; use hashbrown::HashSet; -use infra::schema::{SchemaCache, unwrap_partition_time_level, update_setting}; +use infra::schema::{SchemaCache, unwrap_partition_time_level}; use opentelemetry::trace::{SpanId, TraceId}; use opentelemetry_proto::tonic::{ collector::metrics::v1::{ @@ -268,8 +268,13 @@ pub async fn handle_otlp_request( // update schema metadata if !schema_exists.has_metadata { - if let Err(e) = - update_setting(org_id, metric_name, StreamType::Metrics, prom_meta).await + if let Err(e) = db::schema::update_setting( + org_id, + metric_name, + StreamType::Metrics, + prom_meta, + ) + .await { log::error!( "Failed to set metadata for metric: {} with error: {}", diff --git a/src/service/metrics/prom.rs b/src/service/metrics/prom.rs index d941f3527d..89904294a1 100644 --- a/src/service/metrics/prom.rs +++ b/src/service/metrics/prom.rs @@ -35,7 +35,7 @@ use hashbrown::HashSet; use infra::{ cache::stats, errors::{Error, Result}, - schema::{SchemaCache, unwrap_partition_time_level, update_setting}, + schema::{SchemaCache, unwrap_partition_time_level}, }; use promql_parser::{label::MatchOp, parser}; use prost::Message; @@ -140,8 +140,13 @@ pub async fn remote_write( break; } if need_update { - if let Err(e) = - update_setting(org_id, &metric_name, StreamType::Metrics, extra_metadata).await + if let Err(e) = db::schema::update_setting( + org_id, + &metric_name, + StreamType::Metrics, + extra_metadata, + ) + .await { log::error!( "Error updating metadata for stream: {}, err: {}", diff --git a/src/super_cluster_queue/alerts.rs b/src/super_cluster_queue/alerts.rs index a4d8d2bc9b..646a178608 100644 --- a/src/super_cluster_queue/alerts.rs +++ b/src/super_cluster_queue/alerts.rs @@ -53,7 +53,8 @@ pub(crate) async fn process_msg(msg: AlertMessage) -> Result<()> { return Ok(()); } table::alerts::create(conn, &org_id, &folder_id, alert.clone(), true).await?; - infra::cluster_coordinator::alerts::emit_put_event(&org_id, &alert, Some(folder_id)).await?; + infra::cluster_coordinator::alerts::emit_put_event(&org_id, &alert, Some(folder_id)) + .await?; } AlertMessage::Update { org_id,