starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 16:10:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: histogram stream sync (#6316)

Browse Source 
fixed #6315
This commit is contained in:
Hengfei Yang
2025-03-19 19:30:10 +08:00
committed by GitHub
parent d220d134ea
commit 9f33b16feb
7 changed files with 76 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/infra/ofga/migrations/alert_folders.rs
+12 -18
View File
@@ -78,8 +78,8 @@ pub async fn migrate_alert_folders<C: ConnectionTrait>(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<C: ConnectionTrait>(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<C: ConnectionTrait>(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);
src/common/infra/ofga/mod.rs
+31 -11
View File
@@ -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!(
src/job/stats.rs
+12 -1
View File
@@ -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 {
src/service/db/ofga.rs
+3 -3
View File
@@ -79,9 +79,9 @@ pub async fn set_ofga_model_to_db(mut meta: OFGAModel) -> Result<String, anyhow:
pub async fn get_ofga_model() -> Result<Option<OFGAModel>, 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> {
src/service/metrics/otlp.rs
+8 -3
View File
@@ -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: {}",
src/service/metrics/prom.rs
+8 -3
View File
@@ -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: {}",
src/super_cluster_queue/alerts.rs
+2 -1
View File
@@ -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,