mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 08:09:27 +00:00
fix: read user session from db if the cache is missing (#8846)
This commit is contained in:
+574
-522
File diff suppressed because it is too large
Load Diff
@@ -1198,10 +1198,10 @@ pub struct Limit {
|
||||
#[env_config(name = "ZO_MAX_FILE_RETENTION_TIME", default = 600)] // seconds
|
||||
pub max_file_retention_time: u64,
|
||||
// MB, per log file size limit on disk
|
||||
#[env_config(name = "ZO_MAX_FILE_SIZE_ON_DISK", default = 256)]
|
||||
#[env_config(name = "ZO_MAX_FILE_SIZE_ON_DISK", default = 128)]
|
||||
pub max_file_size_on_disk: usize,
|
||||
// MB, per data file size limit in memory
|
||||
#[env_config(name = "ZO_MAX_FILE_SIZE_IN_MEMORY", default = 256)]
|
||||
#[env_config(name = "ZO_MAX_FILE_SIZE_IN_MEMORY", default = 128)]
|
||||
pub max_file_size_in_memory: usize,
|
||||
#[deprecated(
|
||||
since = "0.14.1",
|
||||
@@ -1305,7 +1305,7 @@ pub struct Limit {
|
||||
pub metrics_max_series_per_query: usize,
|
||||
#[env_config(name = "ZO_METRICS_MAX_POINTS_PER_SERIES", default = 30000)]
|
||||
pub metrics_max_points_per_series: usize,
|
||||
#[env_config(name = "ZO_METRICS_CACHE_MAX_ENTRIES", default = 100000)]
|
||||
#[env_config(name = "ZO_METRICS_CACHE_MAX_ENTRIES", default = 10000)]
|
||||
pub metrics_cache_max_entries: usize,
|
||||
#[env_config(name = "ZO_COLS_PER_RECORD_LIMIT", default = 1000)]
|
||||
pub req_cols_per_record_limit: usize,
|
||||
@@ -1455,13 +1455,13 @@ pub struct Limit {
|
||||
pub consistent_hash_vnodes: usize,
|
||||
#[env_config(
|
||||
name = "ZO_DATAFUSION_FILE_STAT_CACHE_MAX_ENTRIES",
|
||||
default = 100000,
|
||||
default = 10000,
|
||||
help = "Maximum number of entries in the file stat cache. Higher values increase memory usage but may improve query performance."
|
||||
)]
|
||||
pub datafusion_file_stat_cache_max_entries: usize,
|
||||
#[env_config(
|
||||
name = "ZO_DATAFUSION_STREAMING_AGGS_CACHE_MAX_ENTRIES",
|
||||
default = 100000,
|
||||
default = 10000,
|
||||
help = "Maximum number of entries in the streaming aggs cache. Higher values increase memory usage but may improve query performance."
|
||||
)]
|
||||
pub datafusion_streaming_aggs_cache_max_entries: usize,
|
||||
@@ -1631,7 +1631,7 @@ pub struct CacheLatestFiles {
|
||||
|
||||
#[derive(EnvConfig, Default)]
|
||||
pub struct MemoryCache {
|
||||
#[env_config(name = "ZO_MEMORY_CACHE_ENABLED", default = true)]
|
||||
#[env_config(name = "ZO_MEMORY_CACHE_ENABLED", default = false)]
|
||||
pub enabled: bool,
|
||||
// Memory data cache strategy, default is lru, other value is fifo, time_lru
|
||||
#[env_config(name = "ZO_MEMORY_CACHE_STRATEGY", default = "lru")]
|
||||
@@ -2337,7 +2337,7 @@ fn check_common_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
|
||||
cfg.limit.metrics_max_points_per_series = 30_000;
|
||||
}
|
||||
if cfg.limit.metrics_cache_max_entries == 0 {
|
||||
cfg.limit.metrics_cache_max_entries = 100_000;
|
||||
cfg.limit.metrics_cache_max_entries = 10_000;
|
||||
}
|
||||
|
||||
// check search job retention
|
||||
|
||||
@@ -232,7 +232,7 @@ pub async fn chat_stream(
|
||||
|
||||
return MetaHttpResponse::bad_request("AI is not enabled");
|
||||
}
|
||||
let auth_str = crate::common::utils::auth::extract_auth_str(&in_req);
|
||||
let auth_str = crate::common::utils::auth::extract_auth_str(&in_req).await;
|
||||
|
||||
let stream = match ai::service::chat_stream(
|
||||
ai::meta::AiServerRequest::new(req_body.messages, req_body.model, req_body.context),
|
||||
|
||||
@@ -847,7 +847,7 @@ async fn logout(req: actix_web::HttpRequest) -> HttpResponse {
|
||||
let conf = get_config();
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
let auth_str = extract_auth_str(&req);
|
||||
let auth_str = extract_auth_str(&req).await;
|
||||
// Only get the user email from the auth_str, no need to check for permissions and others
|
||||
#[cfg(feature = "enterprise")]
|
||||
let user_email = get_user_email_from_auth_str(&auth_str).await;
|
||||
|
||||
@@ -77,7 +77,7 @@ impl QueryRecommendationEngine for OptimizerContext {
|
||||
recommendations: Vec<OptimiserRecommendation>,
|
||||
) -> Pin<Box<dyn Future<Output = Result<IngestionResponse, anyhow::Error>> + Send>> {
|
||||
Box::pin(async move {
|
||||
let ingester = Ingester;
|
||||
let ingester = Ingester {};
|
||||
let request = IngestionRequest {
|
||||
org_id: META_ORG_ID.to_string(),
|
||||
stream_type: StreamType::Logs.to_string(),
|
||||
|
||||
@@ -33,7 +33,7 @@ use regex::Regex;
|
||||
use utoipa::OpenApi;
|
||||
|
||||
use crate::{
|
||||
common::utils::auth::extract_auth_str,
|
||||
common::utils::auth::extract_basic_auth_str,
|
||||
handler::http::{
|
||||
auth::validator::get_user_email_from_auth_str,
|
||||
request::ratelimit::QUOTA_PAGE_GLOBAL_RULES_ORG, router::openapi::ApiDoc,
|
||||
@@ -136,7 +136,7 @@ fn rule_extractor(
|
||||
}
|
||||
|
||||
pub fn default_extractor(req: &ServiceRequest) -> BoxFuture<'_, ExtractorRuleResult> {
|
||||
let auth_str = extract_auth_str(req.request());
|
||||
let auth_str = extract_basic_auth_str(req.request());
|
||||
let local_path = req.path().to_string();
|
||||
rule_extractor(auth_str, local_path, req.method())
|
||||
}
|
||||
|
||||
+16
-11
@@ -26,18 +26,23 @@ use crate::{
|
||||
pub const USER_SESSION_KEY: &str = "/user_sessions/";
|
||||
|
||||
pub async fn get(session_id: &str) -> Result<String, anyhow::Error> {
|
||||
match USER_SESSIONS.get(session_id) {
|
||||
Some(val) => Ok(val.to_string()),
|
||||
None => {
|
||||
let val = db::get(&format!("{USER_SESSION_KEY}{session_id}")).await?;
|
||||
// base64 format: convert bytes to string and trim any quotes
|
||||
let val = String::from_utf8(val.to_vec())
|
||||
.unwrap()
|
||||
.trim_matches('"')
|
||||
.to_string();
|
||||
Ok(val)
|
||||
}
|
||||
if let Some(val) = USER_SESSIONS.get(session_id) {
|
||||
return Ok(val.to_string());
|
||||
}
|
||||
|
||||
// get from db
|
||||
log::warn!("Cache miss for user session, read from db: {}", session_id);
|
||||
let val = db::get(&format!("{USER_SESSION_KEY}{session_id}")).await?;
|
||||
// json format: convert bytes to string and trim any quotes
|
||||
let val = String::from_utf8(val.to_vec())
|
||||
.unwrap()
|
||||
.trim_matches('"')
|
||||
.to_string();
|
||||
// cache it in memory
|
||||
if !val.is_empty() {
|
||||
USER_SESSIONS.insert(session_id.to_string(), val.to_string());
|
||||
}
|
||||
Ok(val)
|
||||
}
|
||||
|
||||
pub async fn set(session_id: &str, val: &str) -> Result<(), anyhow::Error> {
|
||||
|
||||
Reference in New Issue
Block a user