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

refactor: read_org_setting and a shared TtlCache own the cached reads (#15840)

Browse Source
This commit is contained in:
uddhavdave
2026-10-26 17:18:36 +00:00
committed by GitHub
parent dcbb49370d
commit e7563d2eb7
17 changed files with 260 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/management/src/request/alerts/incidents.rs
+5 -2
View File
@@ -960,7 +960,8 @@ pub async fn get_incident_rca_thread(
};
let show_tool_io = match &session {
RunSession::Present(_) => {
match db::organization::get_org_setting_ai_rca_thread_tool_output(&org_id).await {
match db::organization::read_org_setting(&org_id, |s| s.ai_rca_thread_tool_output).await
{
Ok(show) => show,
Err(e) => return MetaHttpResponse::internal_error(e),
}
@@ -1566,7 +1567,9 @@ pub async fn get_incident_rca_live_stream(
return response;
}
let show_tool_io =
match db::organization::get_org_setting_ai_rca_thread_tool_output(&org_id).await {
match db::organization::read_org_setting(&org_id, |s| s.ai_rca_thread_tool_output)
.await
{
Ok(show) => show,
Err(e) => return MetaHttpResponse::internal_error(e),
};
src/core/src/logs/mod.rs
+1 -1
View File
@@ -657,7 +657,7 @@ async fn ingestion_log_enabled() -> bool {
return false;
}
// the logging will be enabled through meta only
db::organization::get_org_setting_toggle_ingestion_logs(META_ORG_ID)
db::organization::read_org_setting(META_ORG_ID, |s| s.toggle_ingestion_logs)
.await
.unwrap_or(false)
}
src/core/src/self_reporting/ingestion.rs
+2 -1
View File
@@ -241,7 +241,8 @@ pub(super) async fn ingest_usages(mut curr_usages: Vec<UsageData>) {
// do not skip self-reporting even if super org reporting fails
// Check if org has usage stream enabled
match crate::db::organization::get_org_setting_usage_stream_enabled(&org).await {
match crate::db::organization::read_org_setting(&org, |s| s.usage_stream_enabled).await
{
Ok(true) => {}
Ok(false) => continue, // self-report not enabled, so skip
Err(e) => {
src/db/src/organization.rs
+15 -112
View File
@@ -87,29 +87,31 @@ pub async fn get_org_setting(org_id: &str) -> Result<OrganizationSetting, Error>
}
};
let mut settings = read_org_setting(org_id, OrganizationSetting::clone).await?;
settings.free_trial_expiry = trial_period_expiry;
Ok(settings)
}
/// Picks from the org's cached settings, caching the stored or default settings on a miss.
pub async fn read_org_setting<T>(
org_id: &str,
pick: impl FnOnce(&OrganizationSetting) -> T,
) -> Result<T, Error> {
let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}");
if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) {
let mut ret = v.clone();
ret.free_trial_expiry = trial_period_expiry;
return Ok(ret);
return Ok(pick(v));
}
// Try to get settings from DB, but use default if not found
let mut settings: OrganizationSetting = match db::get(&key).await {
let settings: OrganizationSetting = match db::get(&key).await {
Ok(settings) => json::from_slice(&settings)?,
Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => {
OrganizationSetting::default()
}
Err(e) => return Err(e),
};
// Cache the org setting (even if it's default)
ORGANIZATION_SETTING
.write()
.await
.insert(key.to_string(), settings.clone());
settings.free_trial_expiry = trial_period_expiry;
Ok(settings)
let picked = pick(&settings);
ORGANIZATION_SETTING.write().await.insert(key, settings);
Ok(picked)
}
/// Applies `apply` to the settings in the store's `get_for_update`; true if it wrote them.
@@ -120,105 +122,6 @@ where
update_org_setting_in(infra::db::get_db().await.as_ref(), org_id, apply).await
}
/// Get the toggle ingestion logs setting for an org
/// If the setting is not found, return false
/// we add a separate function for avoid clone the setting
pub async fn get_org_setting_toggle_ingestion_logs(org_id: &str) -> Result<bool, Error> {
let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}");
if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) {
return Ok(v.toggle_ingestion_logs);
}
// Try to get settings from DB, but use default if not found
let settings: OrganizationSetting = match db::get(&key).await {
Ok(settings) => json::from_slice(&settings)?,
Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => {
OrganizationSetting::default()
}
Err(e) => return Err(e),
};
let toggle_ingestion_logs = settings.toggle_ingestion_logs;
// Cache the org setting (even if it's default)
ORGANIZATION_SETTING
.write()
.await
.insert(key.to_string(), settings);
Ok(toggle_ingestion_logs)
}
/// Get the usage stream enabled setting for an org.
/// Without a stored setting this falls back to `default_usage_stream_enabled`, which is off.
pub async fn get_org_setting_usage_stream_enabled(org_id: &str) -> Result<bool, Error> {
let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}");
if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) {
return Ok(v.usage_stream_enabled);
}
// Try to get settings from DB, but use default if not found
let settings: OrganizationSetting = match db::get(&key).await {
Ok(settings) => json::from_slice(&settings)?,
Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => {
OrganizationSetting::default()
}
Err(e) => return Err(e),
};
let usage_stream_enabled = settings.usage_stream_enabled;
// Cache the org setting (even if it's default)
ORGANIZATION_SETTING
.write()
.await
.insert(key.to_string(), settings);
Ok(usage_stream_enabled)
}
/// Whether automatic RED anomaly insights are on for an org; absent settings mean off.
pub async fn get_org_setting_red_insights_enabled(org_id: &str) -> Result<bool, Error> {
let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}");
if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) {
return Ok(v.red_insights_enabled);
}
let settings: OrganizationSetting = match db::get(&key).await {
Ok(settings) => json::from_slice(&settings)?,
Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => {
OrganizationSetting::default()
}
Err(e) => return Err(e),
};
let red_insights_enabled = settings.red_insights_enabled;
ORGANIZATION_SETTING
.write()
.await
.insert(key.to_string(), settings);
Ok(red_insights_enabled)
}
/// Whether investigation threads show raw tool input and output; absent settings mean off.
pub async fn get_org_setting_ai_rca_thread_tool_output(org_id: &str) -> Result<bool, Error> {
let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}");
if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) {
return Ok(v.ai_rca_thread_tool_output);
}
let settings: OrganizationSetting = match db::get(&key).await {
Ok(settings) => json::from_slice(&settings)?,
Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => {
OrganizationSetting::default()
}
Err(e) => return Err(e),
};
let ai_rca_thread_tool_output = settings.ai_rca_thread_tool_output;
ORGANIZATION_SETTING
.write()
.await
.insert(key.to_string(), settings);
Ok(ai_rca_thread_tool_output)
}
/// The coordinator delete evicts `ORGANIZATION_SETTING` on every node.
pub async fn delete_org_setting(org_id: &str) -> errors::Result<()> {
let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}");
src/infra/src/table/cipher.rs
+16 -26
View File
@@ -13,19 +13,13 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{
sync::{LazyLock, OnceLock},
time::{Duration, Instant},
};
use std::{sync::OnceLock, time::Duration};
use base64::{Engine, prelude::BASE64_STANDARD};
use config::{
RwHashMap,
utils::{
encryption::{Algorithm, SECRET_PREFIX, decode_encryption_key},
rand::random_bytes,
time::now_micros,
},
use config::utils::{
encryption::{Algorithm, SECRET_PREFIX, decode_encryption_key},
rand::random_bytes,
time::now_micros,
};
use sea_orm::{
ColumnTrait, ConnectionTrait, EntityTrait, Order, QueryFilter, QueryOrder, QuerySelect, Set,
@@ -33,7 +27,7 @@ use sea_orm::{
};
use serde::{Deserialize, Serialize};
use super::entity::cipher_keys::*;
use super::{entity::cipher_keys::*, ttl_cache::TtlCache};
use crate::{
db::{get_orm_client_ro, get_orm_client_rw},
errors,
@@ -52,8 +46,8 @@ pub const CIPHER_KEY_PREFIX: &str = "/cipher_keys/";
/// public API.
pub const DEFAULT_DEK_NAME: &str = "__default__";
/// In-memory DEK cache: org → (raw key bytes, time of last load).
static DEK_CACHE: LazyLock<RwHashMap<String, (Vec<u8>, Instant)>> = LazyLock::new(Default::default);
/// In-memory DEK cache: org → raw key bytes.
static DEK_CACHE: TtlCache<Vec<u8>> = TtlCache::new(DEK_CACHE_TTL);
const DEK_CACHE_TTL: Duration = Duration::from_secs(300);
/// Key state, installed once by [`boot`] from `src/main.rs` and `tests/integration_test.rs`.
@@ -406,14 +400,12 @@ pub async fn get_or_create_dek(org: &str) -> Result<Vec<u8>, errors::Error> {
/// Returns the per-org DEK, using an in-memory cache with a 5-minute TTL to
/// avoid a database round-trip on every encryption/decryption operation.
pub async fn get_dek(org: &str) -> Result<Vec<u8>, errors::Error> {
if let Some(entry) = DEK_CACHE.get(org)
&& entry.1.elapsed() < DEK_CACHE_TTL
{
return Ok(entry.0.clone());
if let Some(cached) = DEK_CACHE.get(org) {
return Ok(cached);
}
let dek = get_or_create_dek(org).await?;
DEK_CACHE.insert(org.to_string(), (dek.clone(), Instant::now()));
DEK_CACHE.insert(org.to_string(), dek.clone());
Ok(dek)
}
@@ -488,10 +480,8 @@ pub async fn boot(key_b64: Option<&str>) -> Result<(), errors::Error> {
}
async fn find_dek(org: &str) -> Result<Option<Vec<u8>>, errors::Error> {
if let Some(entry) = DEK_CACHE.get(org)
&& entry.1.elapsed() < DEK_CACHE_TTL
{
return Ok(Some(entry.0.clone()));
if let Some(dek) = DEK_CACHE.get(org) {
return Ok(Some(dek));
}
let Some(b64) = get_data(org, EntryKind::CipherKey, DEFAULT_DEK_NAME).await? else {
return Ok(None);
@@ -499,7 +489,7 @@ async fn find_dek(org: &str) -> Result<Option<Vec<u8>>, errors::Error> {
let dek = BASE64_STANDARD
.decode(b64)
.map_err(|e| errors::Error::Message(format!("failed to decode DEK for org {org}: {e}")))?;
DEK_CACHE.insert(org.to_string(), (dek.clone(), Instant::now()));
DEK_CACHE.insert(org.to_string(), dek.clone());
Ok(Some(dek))
}
@@ -1014,7 +1004,7 @@ mod tests {
],
)
.await;
DEK_CACHE.insert(org.to_string(), (vec![9u8], Instant::now()));
DEK_CACHE.insert(org.to_string(), vec![9u8]);
let (first, installed) = boot_db(&db, aes(1)).await;
assert_eq!(installed, Some(Ok("AES")));
@@ -1039,7 +1029,7 @@ mod tests {
.iter()
.all(|r| r.2.starts_with(SECRET_PREFIX))
);
assert_eq!(DEK_CACHE.get(org).unwrap().0, vec![9u8]);
assert_eq!(DEK_CACHE.get(org).unwrap(), vec![9u8]);
let before = stored(&db).await;
let second = boot_db(&db, aes(1)).await.0.unwrap().unwrap();
src/infra/src/table/mod.rs
+1
View File
@@ -133,6 +133,7 @@ pub mod timed_annotation_panels;
pub mod timed_annotations;
#[cfg(feature = "cloud")]
pub mod trial_quota_usage;
mod ttl_cache;
pub mod user_auth_state;
pub mod user_password_history;
pub mod users;
src/infra/src/table/oncall_ownership.rs
+10 -14
View File
@@ -15,14 +15,10 @@
//! Ownership rules, and the queue of signals that fell through them — one question, twice.
use std::{
collections::HashMap,
sync::LazyLock,
time::{Duration, Instant},
};
use std::{collections::HashMap, time::Duration};
use config::{
RwHashMap, ider,
ider,
meta::oncall::{OwnershipRule, SubjectType, UnroutedSignal, canonical_path},
utils::time::now_micros,
};
@@ -31,7 +27,10 @@ use sea_orm::{
QuerySelect, Set,
};
use super::entity::{oncall_ownership_rules, oncall_unrouted_signals};
use super::{
entity::{oncall_ownership_rules, oncall_unrouted_signals},
ttl_cache::TtlCache,
};
use crate::{db::get_orm_client_rw, errors};
/// An org's ownership rules, for the paging path (`06` §3).
@@ -50,8 +49,7 @@ use crate::{db::get_orm_client_rw, errors};
///
/// Keyed by org, not by rule: the value is the whole set, which is what
/// resolution needs. Backs [`list_cached`] only; the screens read [`list`].
static RULES_CACHE: LazyLock<RwHashMap<String, (Vec<OwnershipRule>, Instant)>> =
LazyLock::new(Default::default);
static RULES_CACHE: TtlCache<Vec<OwnershipRule>> = TtlCache::new(RULES_CACHE_TTL);
const RULES_CACHE_TTL: Duration = Duration::from_secs(60);
@@ -187,13 +185,11 @@ pub async fn list(org_id: &str) -> Result<Vec<OwnershipRule>, errors::Error> {
/// anything here" is a stable answer and the common one for a young org, and it
/// is exactly the read that would otherwise run on every firing.
pub async fn list_cached(org_id: &str) -> Result<Vec<OwnershipRule>, errors::Error> {
if let Some(entry) = RULES_CACHE.get(org_id)
&& entry.1.elapsed() < RULES_CACHE_TTL
{
return Ok(entry.0.clone());
if let Some(cached) = RULES_CACHE.get(org_id) {
return Ok(cached);
}
let rules = list(org_id).await?;
RULES_CACHE.insert(org_id.to_string(), (rules.clone(), Instant::now()));
RULES_CACHE.insert(org_id.to_string(), rules.clone());
Ok(rules)
}
src/infra/src/table/oncall_policies.rs
+19 -20
View File
@@ -15,19 +15,16 @@
//! A team's escalation policy, stored as JSON.
use std::{
sync::LazyLock,
time::{Duration, Instant},
};
use std::time::Duration;
use config::{
RwHashMap, ider,
ider,
meta::oncall::{EscalationPolicy, PriorityRung},
utils::time::now_micros,
};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set};
use super::entity::oncall_policies;
use super::{entity::oncall_policies, ttl_cache::TtlCache};
use crate::{db::get_orm_client_rw, errors};
/// The team's escalation policy, for the paging path (`06` §3).
@@ -43,8 +40,7 @@ use crate::{db::get_orm_client_rw, errors};
///
/// Backs [`get_or_create_cached`] only. The policy screen reads
/// [`get_or_create`].
static POLICY_CACHE: LazyLock<RwHashMap<String, (EscalationPolicy, Instant)>> =
LazyLock::new(Default::default);
static POLICY_CACHE: TtlCache<EscalationPolicy> = TtlCache::new(POLICY_CACHE_TTL);
const POLICY_CACHE_TTL: Duration = Duration::from_secs(60);
@@ -61,7 +57,7 @@ pub fn invalidate_cache(org_id: &str, team_id: &str) {
/// Drops every policy an org holds, for when the org itself is deleted.
pub fn invalidate_org(org_id: &str) {
let prefix = format!("{org_id}/");
POLICY_CACHE.retain(|key, _| !key.starts_with(&prefix));
POLICY_CACHE.retain(|key| !key.starts_with(&prefix));
}
pub(super) async fn invalidate_and_publish(org_id: &str, team_id: &str) {
@@ -176,13 +172,11 @@ pub async fn get_or_create_cached(
team_id: &str,
) -> Result<EscalationPolicy, errors::Error> {
let key = policy_cache_key(org_id, team_id);
if let Some(entry) = POLICY_CACHE.get(&key)
&& entry.1.elapsed() < POLICY_CACHE_TTL
{
return Ok(entry.0.clone());
if let Some(cached) = POLICY_CACHE.get(&key) {
return Ok(cached);
}
let policy = get_or_create(org_id, team_id).await?;
POLICY_CACHE.insert(key, (policy.clone(), Instant::now()));
POLICY_CACHE.insert(key, policy.clone());
Ok(policy)
}
@@ -428,14 +422,19 @@ mod tests {
EscalationPolicy::default_for_team("pol_1", org, "team_1", "rot_primary", None)
};
for org in ["inv_pol_a", "inv_pol_ab"] {
POLICY_CACHE.insert(
policy_cache_key(org, "team_1"),
(policy(org), Instant::now()),
);
POLICY_CACHE.insert(policy_cache_key(org, "team_1"), policy(org));
}
invalidate_org("inv_pol_a");
assert!(!POLICY_CACHE.contains_key(&policy_cache_key("inv_pol_a", "team_1")));
assert!(POLICY_CACHE.contains_key(&policy_cache_key("inv_pol_ab", "team_1")));
assert!(
POLICY_CACHE
.get(&policy_cache_key("inv_pol_a", "team_1"))
.is_none()
);
assert!(
POLICY_CACHE
.get(&policy_cache_key("inv_pol_ab", "team_1"))
.is_some()
);
invalidate_org("inv_pol_ab");
}
}
src/infra/src/table/oncall_schedules.rs
+8 -14
View File
@@ -15,19 +15,16 @@
//! A team's schedule: rotations in one JSON column, because they are read and written as a set.
use std::{
sync::LazyLock,
time::{Duration, Instant},
};
use std::time::Duration;
use config::{
RwHashMap, ider,
ider,
meta::oncall::{Rotation, Schedule, Unavailability},
utils::time::now_micros,
};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set};
use super::entity::oncall_schedules;
use super::{entity::oncall_schedules, ttl_cache::TtlCache};
use crate::{db::get_orm_client_rw, errors};
/// The team's schedule with its covers, for the paging path (`06` §3).
@@ -48,8 +45,7 @@ use crate::{db::get_orm_client_rw, errors};
/// exists for, and `oncall_overrides` emits one on every write.
///
/// Backs [`get_by_team_cached`] only. Every screen still reads [`get_by_team`].
static SCHEDULE_CACHE: LazyLock<RwHashMap<String, (Schedule, Instant)>> =
LazyLock::new(Default::default);
static SCHEDULE_CACHE: TtlCache<Schedule> = TtlCache::new(SCHEDULE_CACHE_TTL);
/// One minute. See [`SCHEDULE_CACHE`] — the backstop for a coordinator event
/// that never arrived, not the invalidation mechanism, and what it backstops is
@@ -73,7 +69,7 @@ pub fn invalidate_cache(org_id: &str, team_id: &str) {
/// single team to name. Bounded by the org's team count.
pub fn invalidate_org(org_id: &str) {
let prefix = format!("{org_id}/");
SCHEDULE_CACHE.retain(|key, _| !key.starts_with(&prefix));
SCHEDULE_CACHE.retain(|key| !key.starts_with(&prefix));
}
/// Invalidates every schedule in the org locally, and tells every other node.
@@ -257,14 +253,12 @@ pub async fn get_by_team_cached(
team_id: &str,
) -> Result<Option<Schedule>, errors::Error> {
let key = schedule_cache_key(org_id, team_id);
if let Some(entry) = SCHEDULE_CACHE.get(&key)
&& entry.1.elapsed() < SCHEDULE_CACHE_TTL
{
return Ok(Some(entry.0.clone()));
if let Some(cached) = SCHEDULE_CACHE.get(&key) {
return Ok(Some(cached));
}
let found = get_by_team(org_id, team_id).await?;
if let Some(schedule) = &found {
SCHEDULE_CACHE.insert(key, (schedule.clone(), Instant::now()));
SCHEDULE_CACHE.insert(key, schedule.clone());
}
Ok(found)
}
src/infra/src/table/oncall_teams.rs
+24 -18
View File
@@ -30,7 +30,10 @@ use sea_orm::{
TransactionTrait,
};
use super::entity::{oncall_team_members, oncall_teams};
use super::{
entity::{oncall_team_members, oncall_teams},
ttl_cache::TtlCache,
};
use crate::{
db::{get_orm_client_ro, get_orm_client_rw},
errors,
@@ -43,7 +46,7 @@ use crate::{
///
/// Backs [`get_cached`] only, never [`get`]: the screens that edit a team read
/// through `get`, so an admin never sees what they just replaced.
static TEAM_CACHE: LazyLock<RwHashMap<String, (Team, Instant)>> = LazyLock::new(Default::default);
static TEAM_CACHE: TtlCache<Team> = TtlCache::new(TEAM_CACHE_TTL);
/// The roster, for the paging path.
///
@@ -51,8 +54,7 @@ static TEAM_CACHE: LazyLock<RwHashMap<String, (Team, Instant)>> = LazyLock::new(
/// past the third rung. `06` §6 budgets five minutes of staleness because the
/// consequence is bounded and one-directional: a member removed moments ago may
/// receive one more page. Nothing here can suppress a page.
static MEMBERS_CACHE: LazyLock<RwHashMap<String, (Vec<TeamMember>, Instant)>> =
LazyLock::new(Default::default);
static MEMBERS_CACHE: TtlCache<Vec<TeamMember>> = TtlCache::new(TEAM_CACHE_TTL);
/// Whether an org has any team at all, for the firing path.
///
@@ -95,7 +97,7 @@ pub fn invalidate_members_cache(team_id: &str) {
/// Drops every org-keyed on-call cache of a deleted org; rosters are keyed by team alone.
pub fn invalidate_org(org_id: &str) {
let prefix = format!("{org_id}/");
TEAM_CACHE.retain(|key, _| !key.starts_with(&prefix));
TEAM_CACHE.retain(|key| !key.starts_with(&prefix));
ORG_HAS_TEAMS_CACHE.remove(org_id);
super::oncall_schedules::invalidate_org(org_id);
super::oncall_policies::invalidate_org(org_id);
@@ -376,14 +378,12 @@ pub async fn list_members(team_id: &str) -> Result<Vec<TeamMember>, errors::Erro
/// keep saying so.
pub async fn get_cached(org_id: &str, id: &str) -> Result<Option<Team>, errors::Error> {
let key = team_cache_key(org_id, id);
if let Some(entry) = TEAM_CACHE.get(&key)
&& entry.1.elapsed() < TEAM_CACHE_TTL
{
return Ok(Some(entry.0.clone()));
if let Some(cached) = TEAM_CACHE.get(&key) {
return Ok(Some(cached));
}
let found = get(org_id, id).await?;
if let Some(team) = &found {
TEAM_CACHE.insert(key, (team.clone(), Instant::now()));
TEAM_CACHE.insert(key, team.clone());
}
Ok(found)
}
@@ -423,13 +423,11 @@ pub async fn has_any_cached(org_id: &str) -> Result<bool, errors::Error> {
/// roster IS cached, unlike a missing team: "this team has nobody on it" is a
/// real, stable answer, and it is what a coverage gap is made of.
pub async fn list_members_cached(team_id: &str) -> Result<Vec<TeamMember>, errors::Error> {
if let Some(entry) = MEMBERS_CACHE.get(team_id)
&& entry.1.elapsed() < TEAM_CACHE_TTL
{
return Ok(entry.0.clone());
if let Some(cached) = MEMBERS_CACHE.get(team_id) {
return Ok(cached);
}
let members = list_members(team_id).await?;
MEMBERS_CACHE.insert(team_id.to_string(), (members.clone(), Instant::now()));
MEMBERS_CACHE.insert(team_id.to_string(), members.clone());
Ok(members)
}
@@ -711,13 +709,21 @@ mod tests {
})
};
for org in ["inv_team_a", "inv_team_ab"] {
TEAM_CACHE.insert(team_cache_key(org, "team_1"), (team(org), Instant::now()));
TEAM_CACHE.insert(team_cache_key(org, "team_1"), team(org));
ORG_HAS_TEAMS_CACHE.insert(org.to_string(), (true, Instant::now()));
}
invalidate_org("inv_team_a");
assert!(!TEAM_CACHE.contains_key(&team_cache_key("inv_team_a", "team_1")));
assert!(
TEAM_CACHE
.get(&team_cache_key("inv_team_a", "team_1"))
.is_none()
);
assert!(!ORG_HAS_TEAMS_CACHE.contains_key("inv_team_a"));
assert!(TEAM_CACHE.contains_key(&team_cache_key("inv_team_ab", "team_1")));
assert!(
TEAM_CACHE
.get(&team_cache_key("inv_team_ab", "team_1"))
.is_some()
);
assert!(ORG_HAS_TEAMS_CACHE.contains_key("inv_team_ab"));
invalidate_org("inv_team_ab");
}
src/infra/src/table/org_ingestion_tokens.rs
+10 -14
View File
@@ -13,19 +13,18 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{
sync::LazyLock,
time::{Duration, Instant},
};
use std::time::Duration;
use config::RwHashMap;
use sea_orm::{
ColumnTrait, EntityTrait, FromQueryResult, NotSet, Order, PaginatorTrait, QueryFilter,
QueryOrder, QuerySelect, Set, entity::prelude::*, sea_query::OnConflict,
};
use serde::{Deserialize, Deserializer, Serialize};
use super::entity::org_ingestion_tokens::{ActiveModel, Column, Entity, Model};
use super::{
entity::org_ingestion_tokens::{ActiveModel, Column, Entity, Model},
ttl_cache::TtlCache,
};
use crate::{
db::{get_orm_client_ro, get_orm_client_rw},
errors::{self, DbError, Error},
@@ -48,9 +47,8 @@ pub const ORG_INGESTION_TOKEN_PREFIX: &str = "o2oi_";
///
/// `None` is cached too, so an org with no enabled token does not re-query on
/// every job.
static DEFAULT_TOKEN_CACHE: LazyLock<
RwHashMap<String, (Option<OrgIngestionTokenRecord>, Instant)>,
> = LazyLock::new(Default::default);
static DEFAULT_TOKEN_CACHE: TtlCache<Option<OrgIngestionTokenRecord>> =
TtlCache::new(DEFAULT_TOKEN_CACHE_TTL);
/// Backstop only. Cross-node invalidation arrives through the existing
/// `/org_ingestion_tokens/` coordinator watch (see `db::org_ingestion_tokens::watch`),
@@ -73,10 +71,8 @@ pub fn invalidate_default_cache(org_id: &str) {
pub async fn find_default_enabled(
org_id: &str,
) -> Result<Option<OrgIngestionTokenRecord>, errors::Error> {
if let Some(entry) = DEFAULT_TOKEN_CACHE.get(org_id)
&& entry.1.elapsed() < DEFAULT_TOKEN_CACHE_TTL
{
return Ok(entry.0.clone());
if let Some(cached) = DEFAULT_TOKEN_CACHE.get(org_id) {
return Ok(cached);
}
let client = get_orm_client_rw().await;
@@ -90,7 +86,7 @@ pub async fn find_default_enabled(
.map_err(|e| Error::DbError(DbError::SeaORMError(e.to_string())))?
.map(OrgIngestionTokenRecord::from);
DEFAULT_TOKEN_CACHE.insert(org_id.to_string(), (record.clone(), Instant::now()));
DEFAULT_TOKEN_CACHE.insert(org_id.to_string(), record.clone());
Ok(record)
}
src/infra/src/table/synthetics_agents.rs
+9 -13
View File
@@ -19,17 +19,16 @@
//! `/synthetics/agent/register`; `last_seen_at` refreshed by register and by
//! every job lease. A location whose agents are all stale is reported "down".
use std::{
sync::LazyLock,
time::{Duration, Instant},
};
use std::time::Duration;
use config::RwHashMap;
use sea_orm::{
ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set, sea_query::Expr,
};
use super::entity::synthetics_agents::{ActiveModel, Column, Entity, Model};
use super::{
entity::synthetics_agents::{ActiveModel, Column, Entity, Model},
ttl_cache::TtlCache,
};
use crate::{
db::{get_orm_client_ro, get_orm_client_rw},
errors::{self, DbError, Error},
@@ -78,8 +77,7 @@ impl From<Model> for SyntheticsAgentRecord {
/// not invalidate — invalidating there would defeat the cache entirely, since it
/// fires at exactly the rate of the reads being served. Anything that judges
/// agent liveness (the staleness watcher, location health) must use [`get`].
static AGENT_CACHE: LazyLock<RwHashMap<String, (SyntheticsAgentRecord, Instant)>> =
LazyLock::new(Default::default);
static AGENT_CACHE: TtlCache<SyntheticsAgentRecord> = TtlCache::new(AGENT_CACHE_TTL);
const AGENT_CACHE_TTL: Duration = Duration::from_secs(15);
@@ -255,14 +253,12 @@ pub async fn count_by_token(
/// Reads an agent from cache when fresh — see [`AGENT_CACHE`] for the staleness
/// contract. Use [`get`] when `last_seen_at` matters.
pub async fn get_cached(agent_id: &str) -> Result<Option<SyntheticsAgentRecord>, errors::Error> {
if let Some(entry) = AGENT_CACHE.get(agent_id)
&& entry.1.elapsed() < AGENT_CACHE_TTL
{
return Ok(Some(entry.0.clone()));
if let Some(cached) = AGENT_CACHE.get(agent_id) {
return Ok(Some(cached));
}
let found = get(agent_id).await?;
if let Some(rec) = &found {
AGENT_CACHE.insert(agent_id.to_string(), (rec.clone(), Instant::now()));
AGENT_CACHE.insert(agent_id.to_string(), rec.clone());
}
Ok(found)
}
src/infra/src/table/synthetics_checks.rs
+12 -18
View File
@@ -13,17 +13,11 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{
sync::LazyLock,
time::{Duration, Instant},
};
use std::time::Duration;
use config::{
RwHashMap,
meta::synthetics::{
BrowserConfig, ListSyntheticsParams, Synthetic, SyntheticAuth, SyntheticCookie,
SyntheticFrequency, SyntheticSettings, SyntheticStatus, SyntheticType, SyntheticVariable,
},
use config::meta::synthetics::{
BrowserConfig, ListSyntheticsParams, Synthetic, SyntheticAuth, SyntheticCookie,
SyntheticFrequency, SyntheticSettings, SyntheticStatus, SyntheticType, SyntheticVariable,
};
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter,
@@ -31,7 +25,10 @@ use sea_orm::{
};
use serde::{Deserialize, Serialize};
use super::entity::synthetics_checks::{self, ActiveModel, Column, Entity};
use super::{
entity::synthetics_checks::{self, ActiveModel, Column, Entity},
ttl_cache::TtlCache,
};
use crate::errors;
/// `org_id/synthetics_id` → the synthetic check **definition**, with load time.
@@ -48,8 +45,7 @@ use crate::errors;
/// make the cache useless (they fire at the same rate as the reads it serves),
/// so instead they deliberately do **not** invalidate — and callers that need
/// scheduling or status state must use [`get`], `fetch_due` or `get_alert_state`.
static SYNTHETIC_CACHE: LazyLock<RwHashMap<String, (Synthetic, Instant)>> =
LazyLock::new(Default::default);
static SYNTHETIC_CACHE: TtlCache<Synthetic> = TtlCache::new(SYNTHETIC_CACHE_TTL);
const SYNTHETIC_CACHE_TTL: Duration = Duration::from_secs(15);
@@ -207,15 +203,13 @@ pub async fn get_cached<C: ConnectionTrait>(
id: &str,
) -> Result<Option<Synthetic>, errors::Error> {
let key = synthetic_cache_key(org_id, id);
if let Some(entry) = SYNTHETIC_CACHE.get(&key)
&& entry.1.elapsed() < SYNTHETIC_CACHE_TTL
{
return Ok(Some(entry.0.clone()));
if let Some(cached) = SYNTHETIC_CACHE.get(&key) {
return Ok(Some(cached));
}
let found = get(conn, org_id, id).await?;
if let Some(synthetic) = &found {
SYNTHETIC_CACHE.insert(key, (synthetic.clone(), Instant::now()));
SYNTHETIC_CACHE.insert(key, synthetic.clone());
}
Ok(found)
}
src/infra/src/table/synthetics_probe_tokens.rs
+14 -21
View File
@@ -19,17 +19,16 @@
//! (`/synthetics/jobs/resolve`, `/ack`, `/lease`). Separate from
//! `org_ingestion_tokens` (`o2oi_`) which is write-only ingest.
use std::{
sync::LazyLock,
time::{Duration, Instant},
};
use std::time::Duration;
use config::RwHashMap;
use sea_orm::{
ColumnTrait, EntityTrait, QueryFilter, Set, SqlErr, TransactionTrait, sea_query::Expr,
};
use super::entity::synthetics_probe_tokens::{ActiveModel, Column, Entity, Model};
use super::{
entity::synthetics_probe_tokens::{ActiveModel, Column, Entity, Model},
ttl_cache::TtlCache,
};
use crate::{
db::{get_orm_client_ro, get_orm_client_rw},
errors::{self, DbError, Error},
@@ -53,13 +52,11 @@ pub const SYNTHETICS_PROBE_TOKEN_PREFIX: &str = "o2syn_";
/// effectively immediate for an operator and long enough to remove ~99 % of the
/// queries. Writes in this module invalidate eagerly, so the TTL only covers a
/// disable performed on a *different* node.
static TOKEN_CACHE: LazyLock<RwHashMap<String, (Option<SyntheticsProbeTokenRecord>, Instant)>> =
LazyLock::new(Default::default);
static TOKEN_CACHE: TtlCache<Option<SyntheticsProbeTokenRecord>> = TtlCache::new(TOKEN_CACHE_TTL);
/// Org → its default enabled token, same TTL rules as [`TOKEN_CACHE`].
static DEFAULT_TOKEN_CACHE: LazyLock<
RwHashMap<String, (Option<SyntheticsProbeTokenRecord>, Instant)>,
> = LazyLock::new(Default::default);
static DEFAULT_TOKEN_CACHE: TtlCache<Option<SyntheticsProbeTokenRecord>> =
TtlCache::new(TOKEN_CACHE_TTL);
const TOKEN_CACHE_TTL: Duration = Duration::from_secs(10);
@@ -167,10 +164,8 @@ pub async fn add(record: &SyntheticsProbeTokenRecord) -> Result<(), errors::Erro
/// path (the tenant boundary). Matches ANY enabled token, so old + new tokens
/// coexist during a rotation overlap window.
pub async fn find_global(token: &str) -> Result<Option<SyntheticsProbeTokenRecord>, errors::Error> {
if let Some(entry) = TOKEN_CACHE.get(token)
&& entry.1.elapsed() < TOKEN_CACHE_TTL
{
return Ok(entry.0.clone());
if let Some(cached) = TOKEN_CACHE.get(token) {
return Ok(cached);
}
let client = get_orm_client_rw().await;
@@ -182,7 +177,7 @@ pub async fn find_global(token: &str) -> Result<Option<SyntheticsProbeTokenRecor
.map_err(|e| Error::DbError(DbError::SeaORMError(e.to_string())))?
.map(SyntheticsProbeTokenRecord::from);
TOKEN_CACHE.insert(token.to_string(), (record.clone(), Instant::now()));
TOKEN_CACHE.insert(token.to_string(), record.clone());
Ok(record)
}
@@ -191,10 +186,8 @@ pub async fn find_global(token: &str) -> Result<Option<SyntheticsProbeTokenRecor
pub async fn find_default(
org_id: &str,
) -> Result<Option<SyntheticsProbeTokenRecord>, errors::Error> {
if let Some(entry) = DEFAULT_TOKEN_CACHE.get(org_id)
&& entry.1.elapsed() < TOKEN_CACHE_TTL
{
return Ok(entry.0.clone());
if let Some(cached) = DEFAULT_TOKEN_CACHE.get(org_id) {
return Ok(cached);
}
let client = get_orm_client_rw().await;
@@ -207,7 +200,7 @@ pub async fn find_default(
.map_err(|e| Error::DbError(DbError::SeaORMError(e.to_string())))?
.map(SyntheticsProbeTokenRecord::from);
DEFAULT_TOKEN_CACHE.insert(org_id.to_string(), (record.clone(), Instant::now()));
DEFAULT_TOKEN_CACHE.insert(org_id.to_string(), record.clone());
Ok(record)
}
src/infra/src/table/synthetics_variables.rs
+10 -16
View File
@@ -13,18 +13,15 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{
sync::LazyLock,
time::{Duration, Instant},
};
use std::time::Duration;
use config::{
RwHashMap,
meta::synthetics_variables::{SyntheticsVariableView, VariableValueView},
};
use config::meta::synthetics_variables::{SyntheticsVariableView, VariableValueView};
use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QueryOrder, Set, SqlErr};
use super::entity::synthetics_variables::{ActiveModel, Column, Entity, Model};
use super::{
entity::synthetics_variables::{ActiveModel, Column, Entity, Model},
ttl_cache::TtlCache,
};
use crate::errors::{self, DbError, Error};
/// The stored `kind` for a write-only variable.
@@ -33,8 +30,7 @@ pub const KIND_SECRET: &str = "secret";
pub const KIND_PLAIN: &str = "plain";
/// `org_id` → every shared variable in the org, with the time it was loaded.
static VARIABLE_CACHE: LazyLock<RwHashMap<String, (Vec<SyntheticsVariableRecord>, Instant)>> =
LazyLock::new(Default::default);
static VARIABLE_CACHE: TtlCache<Vec<SyntheticsVariableRecord>> = TtlCache::new(VARIABLE_CACHE_TTL);
const VARIABLE_CACHE_TTL: Duration = Duration::from_secs(15);
@@ -131,13 +127,11 @@ pub async fn list_cached<C: ConnectionTrait>(
conn: &C,
org_id: &str,
) -> Result<Vec<SyntheticsVariableRecord>, errors::Error> {
if let Some(entry) = VARIABLE_CACHE.get(org_id)
&& entry.1.elapsed() < VARIABLE_CACHE_TTL
{
return Ok(entry.0.clone());
if let Some(cached) = VARIABLE_CACHE.get(org_id) {
return Ok(cached);
}
let rows = list(conn, org_id).await?;
VARIABLE_CACHE.insert(org_id.to_string(), (rows.clone(), Instant::now()));
VARIABLE_CACHE.insert(org_id.to_string(), rows.clone());
Ok(rows)
}
src/infra/src/table/ttl_cache.rs
+93
View File
@@ -0,0 +1,93 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{
sync::LazyLock,
time::{Duration, Instant},
};
use config::RwHashMap;
/// Entries read as missing past the TTL; stale ones are never swept, only overwritten.
pub(crate) struct TtlCache<V> {
ttl: Duration,
entries: LazyLock<RwHashMap<String, (V, Instant)>>,
}
impl<V: Clone> TtlCache<V> {
pub(crate) const fn new(ttl: Duration) -> Self {
Self {
ttl,
entries: LazyLock::new(RwHashMap::default),
}
}
pub(crate) fn get(&self, key: &str) -> Option<V> {
self.entries
.get(key)
.filter(|entry| entry.1.elapsed() < self.ttl)
.map(|entry| entry.0.clone())
}
pub(crate) fn insert(&self, key: String, value: V) {
self.entries.insert(key, (value, Instant::now()));
}
pub(crate) fn remove(&self, key: &str) {
self.entries.remove(key);
}
pub(crate) fn retain(&self, mut keep: impl FnMut(&str) -> bool) {
self.entries.retain(|key, _| keep(key));
}
pub(crate) fn clear(&self) {
self.entries.clear();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_fresh_entry_is_returned() {
let cache = TtlCache::new(Duration::from_secs(60));
cache.insert("k".to_string(), 7);
assert_eq!(cache.get("k"), Some(7));
}
#[test]
fn an_entry_past_the_ttl_reads_as_missing() {
let cache = TtlCache::new(Duration::ZERO);
cache.insert("k".to_string(), 7);
assert_eq!(cache.get("k"), None);
}
#[test]
fn remove_retain_and_clear_drop_entries() {
let cache = TtlCache::new(Duration::from_secs(60));
for key in ["org/a", "org/b", "other/c", "x"] {
cache.insert(key.to_string(), 1);
}
cache.remove("x");
assert_eq!(cache.get("x"), None);
cache.retain(|key| !key.starts_with("org/"));
assert_eq!(cache.get("org/a"), None);
assert_eq!(cache.get("other/c"), Some(1));
cache.clear();
assert_eq!(cache.get("other/c"), None);
}
}
src/jobs/src/job/red_insights.rs
+11 -10
View File
@@ -63,15 +63,16 @@ pub async fn run() {
}
async fn reconcile_if_needed(org_id: &str) -> anyhow::Result<()> {
let prefetched = if db::organization::get_org_setting_red_insights_enabled(org_id).await? {
None
} else {
let configs = managed_configs(org_id).await?;
if configs.is_empty() {
return Ok(());
}
Some(configs)
};
let prefetched =
if db::organization::read_org_setting(org_id, |s| s.red_insights_enabled).await? {
None
} else {
let configs = managed_configs(org_id).await?;
if configs.is_empty() {
return Ok(());
}
Some(configs)
};
let locker = infra::dist_lock::lock(&format!("red_insights/{org_id}"), 0).await?;
let result = reconcile(org_id, prefetched).await;
if let Err(e) = infra::dist_lock::unlock(&locker).await {
@@ -85,7 +86,7 @@ async fn reconcile(org_id: &str, prefetched: Option<Vec<serde_json::Value>>) ->
Some(configs) => configs,
None => managed_configs(org_id).await?,
};
if !db::organization::get_org_setting_red_insights_enabled(org_id).await? {
if !db::organization::read_org_setting(org_id, |s| s.red_insights_enabled).await? {
for config in &configs {
delete(org_id, config_str(config, "anomaly_id")).await;
}