mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 16:09:05 +00:00
feat: include search type in search calls (#3648)
<!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit - **New Features** - Introduced `searchType` attribute across various components to enhance search functionality. - Added support for different search event types like `UI`, `Dashboards`, `Reports`, `Alerts`, `Values`, and `RUM`. - **Bug Fixes** - Improved search parameter handling in multiple components and API calls. - **Enhancements** - Updated search-related functions to include `searchType` for more precise search results. - Enhanced dashboard and RUM views with new `searchType` attributes for better data filtering. - **Refactor** - Standardized search functionality across different modules and components. <!-- end of auto-generated comment: release notes by coderabbit.ai --> --------- Co-authored-by: ktx-abhay <abhay.padamani@kiara.tech>
This commit is contained in:
co-authored by
ktx-abhay
parent
5dabd86ba9
commit
79fe002f60
+10
-2
@@ -17,11 +17,14 @@ use std::{collections::HashMap, fs, path::Path};
|
||||
|
||||
use actix_web::web::Query;
|
||||
use async_trait::async_trait;
|
||||
use config::meta::{search, stream::StreamType};
|
||||
use config::meta::{
|
||||
search::{self},
|
||||
stream::StreamType,
|
||||
};
|
||||
|
||||
use crate::{
|
||||
cli::data::{cli::Cli, Context},
|
||||
common::utils::http::get_stream_type_from_request,
|
||||
common::utils::http::{get_search_type_from_request, get_stream_type_from_request},
|
||||
service::search as SearchService,
|
||||
};
|
||||
|
||||
@@ -39,6 +42,10 @@ impl Context for Export {
|
||||
|
||||
let cfg = config::get_config();
|
||||
let table = c.stream_name;
|
||||
let search_type = match get_search_type_from_request(&Query(map.clone())) {
|
||||
Ok(v) => v,
|
||||
Err(_) => return Ok(false),
|
||||
};
|
||||
let query = search::Query {
|
||||
sql: format!("select * from {}", table),
|
||||
from: 0,
|
||||
@@ -63,6 +70,7 @@ impl Context for Export {
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type,
|
||||
};
|
||||
|
||||
match SearchService::search("", &c.org, stream_type, None, &req).await {
|
||||
|
||||
@@ -20,7 +20,7 @@ use std::{
|
||||
|
||||
use actix_web::web::Query;
|
||||
use awc::http::header::HeaderMap;
|
||||
use config::meta::stream::StreamType;
|
||||
use config::meta::{search::SearchEventType, stream::StreamType};
|
||||
use opentelemetry::propagation::Extractor;
|
||||
|
||||
#[inline(always)]
|
||||
@@ -48,6 +48,31 @@ pub(crate) fn get_stream_type_from_request(
|
||||
Ok(stream_type)
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
pub(crate) fn get_search_type_from_request(
|
||||
query: &Query<HashMap<String, String>>,
|
||||
) -> Result<Option<SearchEventType>, Error> {
|
||||
let event_type = match query.get("search_type") {
|
||||
Some(s) => match s.to_lowercase().as_str() {
|
||||
"ui" => Some(SearchEventType::UI),
|
||||
"dashboards" => Some(SearchEventType::Dashboards),
|
||||
"reports" => Some(SearchEventType::Reports),
|
||||
"alerts" => Some(SearchEventType::Alerts),
|
||||
"values" => Some(SearchEventType::Values),
|
||||
"rum" => Some(SearchEventType::RUM),
|
||||
_ => {
|
||||
return Err(Error::new(
|
||||
ErrorKind::Other,
|
||||
"'event_type' query param with value 'ui', 'dashboards', 'reports', 'alerts' , 'rum' or 'values' allowed",
|
||||
));
|
||||
}
|
||||
},
|
||||
None => None,
|
||||
};
|
||||
|
||||
Ok(event_type)
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
pub(crate) fn get_folder(query: &Query<HashMap<String, String>>) -> String {
|
||||
match query.get("folder") {
|
||||
|
||||
@@ -568,6 +568,9 @@ pub struct Common {
|
||||
pub usage_reporting_creds: String,
|
||||
#[env_config(name = "ZO_USAGE_BATCH_SIZE", default = 2000)]
|
||||
pub usage_batch_size: usize,
|
||||
#[env_config(name = "ZO_USAGE_PUBLISH_INTERVAL", help = "duration in seconds after last reporting usage will be published " default = 600)]
|
||||
// in seconds
|
||||
pub usage_publish_interval: i64,
|
||||
#[env_config(name = "ZO_MMDB_DATA_DIR")] // ./data/openobserve/mmdb/
|
||||
pub mmdb_data_dir: String,
|
||||
#[env_config(name = "ZO_MMDB_DISABLE_DOWNLOAD", default = "false")]
|
||||
|
||||
@@ -60,6 +60,8 @@ pub struct Request {
|
||||
pub clusters: Vec<String>, // default query all clusters, local: only query local cluster
|
||||
#[serde(default)]
|
||||
pub timeout: i64,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub search_type: Option<SearchEventType>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
|
||||
@@ -495,6 +497,31 @@ impl From<&cluster_rpc::ScanStats> for ScanStats {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Hash, Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
|
||||
pub enum SearchEventType {
|
||||
UI,
|
||||
Dashboards,
|
||||
Reports,
|
||||
Alerts,
|
||||
Values,
|
||||
Other,
|
||||
RUM,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for SearchEventType {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
|
||||
match self {
|
||||
SearchEventType::UI => write!(f, "UI"),
|
||||
SearchEventType::Dashboards => write!(f, "Dashboards"),
|
||||
SearchEventType::Reports => write!(f, "Reports"),
|
||||
SearchEventType::Alerts => write!(f, "Alerts"),
|
||||
SearchEventType::Other => write!(f, "Other"),
|
||||
SearchEventType::Values => write!(f, "_values"),
|
||||
SearchEventType::RUM => write!(f, "RUM"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -586,6 +613,7 @@ mod tests {
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
req.aggs
|
||||
.insert("test".to_string(), "SELECT * FROM test".to_string());
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use super::search::SearchEventType;
|
||||
use crate::{
|
||||
meta::stream::{FileMeta, StreamType},
|
||||
SIZE_IN_MB,
|
||||
@@ -82,6 +83,8 @@ pub struct UsageData {
|
||||
pub min_ts: Option<i64>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub max_ts: Option<i64>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub search_type: Option<SearchEventType>,
|
||||
}
|
||||
|
||||
#[derive(Hash, PartialEq, Eq)]
|
||||
@@ -220,6 +223,8 @@ pub struct RequestStats {
|
||||
pub max_ts: Option<i64>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub user_email: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub search_type: Option<SearchEventType>,
|
||||
}
|
||||
impl Default for RequestStats {
|
||||
fn default() -> Self {
|
||||
@@ -233,6 +238,7 @@ impl Default for RequestStats {
|
||||
min_ts: None,
|
||||
max_ts: None,
|
||||
user_email: None,
|
||||
search_type: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -249,6 +255,7 @@ impl From<FileMeta> for RequestStats {
|
||||
min_ts: Some(meta.min_ts),
|
||||
max_ts: Some(meta.max_ts),
|
||||
user_email: None,
|
||||
search_type: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ use chrono::Duration;
|
||||
use config::{
|
||||
get_config, ider,
|
||||
meta::{
|
||||
search::SearchEventType,
|
||||
stream::StreamType,
|
||||
usage::{RequestStats, UsageType},
|
||||
},
|
||||
@@ -37,7 +38,9 @@ use crate::{
|
||||
meta::{self, http::HttpResponse as MetaHttpResponse},
|
||||
utils::{
|
||||
functions,
|
||||
http::{get_stream_type_from_request, RequestHeaderExtractor},
|
||||
http::{
|
||||
get_search_type_from_request, get_stream_type_from_request, RequestHeaderExtractor,
|
||||
},
|
||||
},
|
||||
},
|
||||
service::{
|
||||
@@ -145,6 +148,11 @@ pub async fn search(
|
||||
Err(e) => return Ok(MetaHttpResponse::bad_request(e)),
|
||||
};
|
||||
|
||||
let search_type = match get_search_type_from_request(&query) {
|
||||
Ok(v) => v,
|
||||
Err(e) => return Ok(MetaHttpResponse::bad_request(e)),
|
||||
};
|
||||
|
||||
// handle encoding for query and aggs
|
||||
let mut req: config::meta::search::Request = match json::from_slice(&body) {
|
||||
Ok(v) => v,
|
||||
@@ -283,6 +291,7 @@ pub async fn search(
|
||||
min_ts: Some(req.query.start_time),
|
||||
max_ts: Some(req.query.end_time),
|
||||
cached_ratio: Some(res.cached_ratio),
|
||||
search_type,
|
||||
..Default::default()
|
||||
};
|
||||
let num_fn = req.query.query_fn.is_some() as u16;
|
||||
@@ -521,6 +530,7 @@ pub async fn around(
|
||||
regions: regions.clone(),
|
||||
clusters: clusters.clone(),
|
||||
timeout,
|
||||
search_type: Some(SearchEventType::UI),
|
||||
};
|
||||
let user_id = in_req
|
||||
.headers()
|
||||
@@ -600,6 +610,7 @@ pub async fn around(
|
||||
regions,
|
||||
clusters,
|
||||
timeout,
|
||||
search_type: Some(SearchEventType::UI),
|
||||
};
|
||||
let search_fut = SearchService::search(&trace_id, &org_id, stream_type, user_id, &req);
|
||||
let search_res = if !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled {
|
||||
@@ -1008,6 +1019,7 @@ async fn values_v1(
|
||||
regions,
|
||||
clusters,
|
||||
timeout,
|
||||
search_type: Some(SearchEventType::Values),
|
||||
};
|
||||
|
||||
// skip fields which aren't part of the schema
|
||||
@@ -1130,6 +1142,7 @@ async fn values_v1(
|
||||
min_ts: Some(start_time),
|
||||
max_ts: Some(end_time),
|
||||
cached_ratio: Some(resp.cached_ratio),
|
||||
search_type: Some(SearchEventType::Values),
|
||||
..Default::default()
|
||||
};
|
||||
let num_fn = req.query.query_fn.is_some() as u16;
|
||||
@@ -1256,6 +1269,7 @@ async fn values_v2(
|
||||
regions,
|
||||
clusters,
|
||||
timeout,
|
||||
search_type: Some(SearchEventType::Values),
|
||||
};
|
||||
let search_fut = SearchService::search(
|
||||
&trace_id,
|
||||
@@ -1354,6 +1368,7 @@ async fn values_v2(
|
||||
min_ts: Some(start_time),
|
||||
max_ts: Some(end_time),
|
||||
cached_ratio: Some(resp.cached_ratio),
|
||||
search_type: Some(SearchEventType::Values),
|
||||
..Default::default()
|
||||
};
|
||||
let num_fn = req.query.query_fn.is_some() as u16;
|
||||
|
||||
@@ -272,6 +272,7 @@ pub async fn get_latest_traces(
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout,
|
||||
search_type: None,
|
||||
};
|
||||
let stream_type = StreamType::Traces;
|
||||
let user_id = in_req
|
||||
|
||||
@@ -23,7 +23,7 @@ use arrow_schema::DataType;
|
||||
use chrono::{Duration, Local, TimeZone, Utc};
|
||||
use config::{
|
||||
get_config, ider,
|
||||
meta::stream::StreamType,
|
||||
meta::{search::SearchEventType, stream::StreamType},
|
||||
utils::{
|
||||
base64,
|
||||
json::{Map, Value},
|
||||
@@ -465,6 +465,7 @@ impl QueryCondition {
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: Some(SearchEventType::Alerts),
|
||||
};
|
||||
let trace_id = ider::uuid();
|
||||
let resp =
|
||||
|
||||
@@ -49,6 +49,7 @@ pub async fn get(org_id: &str, name: &str) -> Result<Vec<vrl::value::Value>, any
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
// do search
|
||||
match SearchService::search("", org_id, StreamType::EnrichmentTables, None, &req).await {
|
||||
|
||||
@@ -624,6 +624,7 @@ pub(crate) async fn get_series(
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
let series = match search_service::search("", org_id, StreamType::Metrics, None, &req).await {
|
||||
Err(err) => {
|
||||
@@ -769,6 +770,7 @@ pub(crate) async fn get_label_values(
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
let mut label_values = match search_service::search("", org_id, stream_type, None, &req).await {
|
||||
Ok(resp) => resp
|
||||
|
||||
+61
-11
@@ -20,7 +20,7 @@ use config::{
|
||||
get_config, ider,
|
||||
meta::{
|
||||
search,
|
||||
stream::{FileKey, StreamType},
|
||||
stream::{FileKey, StreamType}, usage::{RequestStats, UsageType},
|
||||
},
|
||||
utils::str::find,
|
||||
};
|
||||
@@ -33,6 +33,8 @@ use opentelemetry::trace::TraceContextExt;
|
||||
use proto::cluster_rpc;
|
||||
use regex::Regex;
|
||||
use tracing_opentelemetry::OpenTelemetrySpanExt;
|
||||
use super::usage::report_request_usage_stats;
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
use {
|
||||
hashbrown::HashSet,
|
||||
@@ -63,14 +65,15 @@ pub(crate) static QUEUE_LOCKER: Lazy<Arc<Mutex<bool>>> =
|
||||
static RE_SELECT_WILDCARD: Lazy<Regex> =
|
||||
Lazy::new(|| Regex::new(r"(?i)select\s+\*\s+from").unwrap());
|
||||
|
||||
#[tracing::instrument(name = "service:search:enter", skip(req))]
|
||||
#[tracing::instrument(name = "service:search:enter", skip(in_req))]
|
||||
pub async fn search(
|
||||
trace_id: &str,
|
||||
org_id: &str,
|
||||
stream_type: StreamType,
|
||||
user_id: Option<String>,
|
||||
req: &search::Request,
|
||||
in_req: &search::Request,
|
||||
) -> Result<search::Response, Error> {
|
||||
let start = std::time::Instant::now();
|
||||
let cfg = get_config();
|
||||
let trace_id = if trace_id.is_empty() {
|
||||
if cfg.common.tracing_enabled || cfg.common.tracing_search_enabled {
|
||||
@@ -85,9 +88,9 @@ pub async fn search(
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
{
|
||||
let sql = Some(req.query.sql.clone());
|
||||
let start_time = Some(req.query.start_time);
|
||||
let end_time = Some(req.query.end_time);
|
||||
let sql = Some(in_req.query.sql.clone());
|
||||
let start_time = Some(in_req.query.start_time);
|
||||
let end_time = Some(in_req.query.end_time);
|
||||
// set search task
|
||||
SEARCH_SERVER
|
||||
.insert(
|
||||
@@ -95,7 +98,7 @@ pub async fn search(
|
||||
TaskStatus::new(
|
||||
vec![],
|
||||
true,
|
||||
user_id,
|
||||
user_id.clone(),
|
||||
Some(org_id.to_string()),
|
||||
Some(stream_type.to_string()),
|
||||
sql,
|
||||
@@ -107,20 +110,22 @@ pub async fn search(
|
||||
}
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
let req_regions = req.regions.clone();
|
||||
let req_regions = in_req.regions.clone();
|
||||
#[cfg(feature = "enterprise")]
|
||||
let req_clusters = req.clusters.clone();
|
||||
let req_clusters = in_req.clusters.clone();
|
||||
#[cfg(feature = "enterprise")]
|
||||
let local_cluster_search = req_regions == vec!["local"]
|
||||
&& !req_clusters.is_empty()
|
||||
&& (req_clusters == vec!["local"] || req_clusters == vec![config::get_cluster_name()]);
|
||||
|
||||
let mut req: cluster_rpc::SearchRequest = req.to_owned().into();
|
||||
let mut req: cluster_rpc::SearchRequest = in_req.to_owned().into();
|
||||
req.job.as_mut().unwrap().trace_id = trace_id.clone();
|
||||
req.org_id = org_id.to_string();
|
||||
req.stype = cluster_rpc::SearchType::Cluster as _;
|
||||
req.stream_type = stream_type.to_string();
|
||||
|
||||
let req_query = req.clone().query.unwrap();
|
||||
|
||||
let res = {
|
||||
#[cfg(feature = "enterprise")]
|
||||
if O2_CONFIG.super_cluster.enabled && !local_cluster_search {
|
||||
@@ -140,7 +145,52 @@ pub async fn search(
|
||||
|
||||
// do this because of clippy warning
|
||||
match res {
|
||||
Ok(res) => Ok(res),
|
||||
Ok(res) => {
|
||||
let time = start.elapsed().as_secs_f64();
|
||||
let (report_usage, search_type) = match in_req.search_type {
|
||||
Some(search_type) => match search_type {
|
||||
search::SearchEventType::UI => (false, None),
|
||||
search::SearchEventType::Dashboards => (true, in_req.search_type),
|
||||
search::SearchEventType::Reports => (true, in_req.search_type),
|
||||
search::SearchEventType::Alerts => (true, in_req.search_type),
|
||||
search::SearchEventType::RUM => (true, in_req.search_type),
|
||||
search::SearchEventType::Values => (false, None),
|
||||
search::SearchEventType::Other => (false, None),
|
||||
},
|
||||
None => (false, None),
|
||||
};
|
||||
|
||||
if report_usage {
|
||||
let stream_name = match config::meta::sql::Sql::new(&req_query.sql) {
|
||||
Ok(v) => v.source.to_string(),
|
||||
Err(e) => {
|
||||
log::error!("parse sql error: {:?}", e);
|
||||
"".to_string()
|
||||
}
|
||||
};
|
||||
let req_stats = RequestStats {
|
||||
records: res.hits.len() as i64,
|
||||
response_time: time,
|
||||
size: res.scan_size as f64,
|
||||
request_body: Some(req_query.sql.clone()),
|
||||
user_email: user_id,
|
||||
min_ts: Some(req_query.start_time),
|
||||
max_ts: Some(req_query.end_time),
|
||||
cached_ratio: Some(res.cached_ratio),
|
||||
search_type,
|
||||
..Default::default()
|
||||
};
|
||||
report_request_usage_stats(
|
||||
req_stats,
|
||||
org_id,
|
||||
&stream_name,
|
||||
StreamType::Logs,
|
||||
UsageType::Search,
|
||||
0,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
Ok(res)},
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1048,6 +1048,7 @@ mod tests {
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
|
||||
let mut rpc_req: cluster_rpc::SearchRequest = req.to_owned().into();
|
||||
@@ -1158,6 +1159,7 @@ mod tests {
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
let mut rpc_req: cluster_rpc::SearchRequest = req.to_owned().into();
|
||||
rpc_req.org_id = org_id.to_string();
|
||||
@@ -1281,6 +1283,7 @@ mod tests {
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
let mut rpc_req: cluster_rpc::SearchRequest = req.to_owned().into();
|
||||
rpc_req.org_id = org_id.to_string();
|
||||
|
||||
@@ -45,6 +45,12 @@ pub static USAGE_DATA: Lazy<Arc<RwLock<Vec<UsageData>>>> =
|
||||
pub static TRIGGERS_USAGE_DATA: Lazy<Arc<RwLock<Vec<TriggerData>>>> =
|
||||
Lazy::new(|| Arc::new(RwLock::new(vec![])));
|
||||
|
||||
pub static LAST_PUBLISHED_USAGE: Lazy<Arc<RwLock<i64>>> =
|
||||
Lazy::new(|| Arc::new(RwLock::new(Utc::now().timestamp())));
|
||||
|
||||
pub static LAST_PUBLISHED_TRIGGERS: Lazy<Arc<RwLock<i64>>> =
|
||||
Lazy::new(|| Arc::new(RwLock::new(Utc::now().timestamp())));
|
||||
|
||||
pub async fn report_request_usage_stats(
|
||||
stats: RequestStats,
|
||||
org_id: &str,
|
||||
@@ -98,6 +104,7 @@ pub async fn report_request_usage_stats(
|
||||
max_ts: None,
|
||||
cached_ratio: None,
|
||||
compressed_size: None,
|
||||
search_type: stats.search_type,
|
||||
});
|
||||
};
|
||||
|
||||
@@ -127,6 +134,7 @@ pub async fn report_request_usage_stats(
|
||||
max_ts: stats.max_ts,
|
||||
cached_ratio: stats.cached_ratio,
|
||||
compressed_size: None,
|
||||
search_type: stats.search_type,
|
||||
});
|
||||
if !usage.is_empty() {
|
||||
publish_usage(usage).await;
|
||||
@@ -136,9 +144,16 @@ pub async fn report_request_usage_stats(
|
||||
pub async fn publish_usage(mut usage: Vec<UsageData>) {
|
||||
let mut usages = USAGE_DATA.write().await;
|
||||
usages.append(&mut usage);
|
||||
if usages.len() < get_config().common.usage_batch_size {
|
||||
|
||||
let current_ts = Utc::now().timestamp();
|
||||
let r = LAST_PUBLISHED_USAGE.read().await;
|
||||
let last_published = *r;
|
||||
if usages.len() < get_config().common.usage_batch_size
|
||||
&& current_ts - last_published < get_config().common.usage_publish_interval
|
||||
{
|
||||
return;
|
||||
}
|
||||
drop(r);
|
||||
|
||||
let curr_usages = std::mem::take(&mut *usages);
|
||||
// release the write lock
|
||||
@@ -155,9 +170,17 @@ pub async fn publish_triggers_usage(trigger: TriggerData) {
|
||||
|
||||
let mut usages = TRIGGERS_USAGE_DATA.write().await;
|
||||
usages.push(trigger);
|
||||
if usages.len() < cfg.common.usage_batch_size {
|
||||
|
||||
let current_ts = Utc::now().timestamp();
|
||||
let r = LAST_PUBLISHED_TRIGGERS.read().await;
|
||||
let last_published = *r;
|
||||
|
||||
if usages.len() < cfg.common.usage_batch_size
|
||||
&& current_ts - last_published < cfg.common.usage_publish_interval
|
||||
{
|
||||
return;
|
||||
}
|
||||
drop(r);
|
||||
|
||||
let curr_usages = std::mem::take(&mut *usages);
|
||||
// release the write lock
|
||||
@@ -212,6 +235,10 @@ async fn flush_triggers_usage() {
|
||||
}
|
||||
|
||||
async fn ingest_usages(curr_usages: Vec<UsageData>) {
|
||||
if curr_usages.is_empty() {
|
||||
log::info!(" Returning as no usages reported ");
|
||||
return;
|
||||
}
|
||||
let mut groups: HashMap<GroupKey, AggregatedData> = HashMap::new();
|
||||
let mut search_events = vec![];
|
||||
for usage_data in &curr_usages {
|
||||
@@ -307,9 +334,17 @@ async fn ingest_usages(curr_usages: Vec<UsageData>) {
|
||||
drop(usages);
|
||||
}
|
||||
}
|
||||
let mut w = LAST_PUBLISHED_USAGE.write().await;
|
||||
*w = Utc::now().timestamp();
|
||||
drop(w);
|
||||
}
|
||||
|
||||
async fn ingest_trigger_usages(curr_usages: Vec<TriggerData>) {
|
||||
if curr_usages.is_empty() {
|
||||
log::info!(" Returning as no triggers reported ");
|
||||
return;
|
||||
}
|
||||
|
||||
let mut json_triggers = vec![];
|
||||
for trigger_data in &curr_usages {
|
||||
json_triggers.push(json::to_value(trigger_data).unwrap());
|
||||
@@ -328,6 +363,9 @@ async fn ingest_trigger_usages(curr_usages: Vec<TriggerData>) {
|
||||
usages.append(&mut curr_usages);
|
||||
drop(usages);
|
||||
}
|
||||
let mut w = LAST_PUBLISHED_TRIGGERS.write().await;
|
||||
*w = Utc::now().timestamp();
|
||||
drop(w);
|
||||
}
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
|
||||
@@ -93,6 +93,7 @@ pub async fn publish_stats() -> Result<(), anyhow::Error> {
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
// do search
|
||||
match SearchService::search("", &cfg.common.usage_org, StreamType::Logs, None, &req).await {
|
||||
@@ -151,6 +152,7 @@ async fn get_last_stats(
|
||||
regions: vec![],
|
||||
clusters: vec![],
|
||||
timeout: 0,
|
||||
search_type: None,
|
||||
};
|
||||
match SearchService::search(
|
||||
"",
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
:panelSchema="chartData"
|
||||
:selectedTimeObj="dashboardPanelData.meta.dateTime"
|
||||
:variablesData="{}"
|
||||
searchType="UI"
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -176,6 +176,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:height="props.height"
|
||||
:variablesData="props.variablesData"
|
||||
:forceLoad="props.forceLoad"
|
||||
:searchType="searchType"
|
||||
@metadata-update="metaDataValue"
|
||||
@updated:data-zoom="$emit('updated:data-zoom', $event)"
|
||||
@update:initial-variable-values="
|
||||
@@ -242,6 +243,7 @@ export default defineComponent({
|
||||
"dashboardId",
|
||||
"metaData",
|
||||
"forceLoad",
|
||||
"searchType",
|
||||
],
|
||||
components: {
|
||||
PanelSchemaRenderer,
|
||||
|
||||
@@ -211,6 +211,10 @@ export default defineComponent({
|
||||
default: false,
|
||||
required: false,
|
||||
},
|
||||
searchType: {
|
||||
default: null,
|
||||
type: String || null,
|
||||
},
|
||||
},
|
||||
emits: [
|
||||
"updated:data-zoom",
|
||||
@@ -230,15 +234,21 @@ export default defineComponent({
|
||||
const drilldownPopUpRef: any = ref(null);
|
||||
|
||||
// get refs from props
|
||||
const { panelSchema, selectedTimeObj, variablesData, forceLoad } =
|
||||
toRefs(props);
|
||||
const {
|
||||
panelSchema,
|
||||
selectedTimeObj,
|
||||
variablesData,
|
||||
forceLoad,
|
||||
searchType,
|
||||
} = toRefs(props);
|
||||
// calls the apis to get the data based on the panel config
|
||||
let { data, loading, errorDetail, metadata } = usePanelDataLoader(
|
||||
panelSchema,
|
||||
selectedTimeObj,
|
||||
variablesData,
|
||||
chartPanelRef,
|
||||
forceLoad
|
||||
forceLoad,
|
||||
searchType
|
||||
);
|
||||
|
||||
// need tableRendererRef to access downloadTableAsCSV method
|
||||
|
||||
@@ -86,6 +86,7 @@
|
||||
:selectedTimeObj="dashboardPanelData.meta.dateTime"
|
||||
:variablesData="variablesData"
|
||||
:width="6"
|
||||
:searchType="searchType"
|
||||
@error="handleChartApiError"
|
||||
@updated:data-zoom="onDataZoom"
|
||||
@update:initialVariableValues="onUpdateInitialVariableValues"
|
||||
@@ -154,6 +155,10 @@ export default defineComponent({
|
||||
initialVariableValues: {
|
||||
type: Object,
|
||||
},
|
||||
searchType: {
|
||||
default: null,
|
||||
type: String || null,
|
||||
},
|
||||
},
|
||||
emits: ["closePanel", "update:initialVariableValues"],
|
||||
setup(props, { emit }) {
|
||||
|
||||
@@ -32,6 +32,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:dashboardData="currentDashboardData.data"
|
||||
:currentTimeObj="dateTime"
|
||||
@variablesData="variablesDataUpdated"
|
||||
searchType="RUM"
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
@@ -162,11 +163,14 @@ export default defineComponent({
|
||||
};
|
||||
|
||||
searchService
|
||||
.search({
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
.search(
|
||||
{
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
},
|
||||
"RUM"
|
||||
)
|
||||
.then((res) => {
|
||||
res.data.hits.slice(0, topCount).forEach((element: any) => {
|
||||
topSlowResources.value.push({
|
||||
@@ -198,11 +202,14 @@ export default defineComponent({
|
||||
};
|
||||
|
||||
searchService
|
||||
.search({
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
.search(
|
||||
{
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
},
|
||||
"RUM"
|
||||
)
|
||||
.then((res) => {
|
||||
res.data.hits.forEach((element: any) => {
|
||||
topHeavyResources.value.push({
|
||||
@@ -234,11 +241,14 @@ export default defineComponent({
|
||||
};
|
||||
|
||||
searchService
|
||||
.search({
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
.search(
|
||||
{
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
},
|
||||
"RUM"
|
||||
)
|
||||
.then((res) => {
|
||||
res.data.hits.forEach((element: any) => {
|
||||
topErrorResources.value.push({
|
||||
|
||||
@@ -28,6 +28,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:viewOnly="true"
|
||||
:dashboardData="currentDashboardData.data"
|
||||
:currentTimeObj="dateTime"
|
||||
searchType="RUM"
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -27,6 +27,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:viewOnly="true"
|
||||
:dashboardData="currentDashboardData.data"
|
||||
:currentTimeObj="dateTime"
|
||||
searchType="RUM"
|
||||
>
|
||||
<template v-slot:before_panels>
|
||||
<div class="flex items-center q-pb q-pt-md text-subtitle1 text-bold">
|
||||
|
||||
@@ -47,6 +47,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:viewOnly="true"
|
||||
:dashboardData="currentDashboardData.data"
|
||||
:currentTimeObj="dateTime"
|
||||
searchType="RUM"
|
||||
/>
|
||||
</div>
|
||||
<div
|
||||
|
||||
@@ -38,7 +38,8 @@ export const usePanelDataLoader = (
|
||||
selectedTimeObj: any,
|
||||
variablesData: any,
|
||||
chartPanelRef: any,
|
||||
forceLoad: any
|
||||
forceLoad: any,
|
||||
searchType: any
|
||||
) => {
|
||||
const log = (...args: any[]) => {
|
||||
// if (true) {
|
||||
@@ -298,21 +299,25 @@ export const usePanelDataLoader = (
|
||||
queryType: panelSchema.value.queryType,
|
||||
variables: [...(metadata1 || []), ...(metadata2 || [])],
|
||||
};
|
||||
|
||||
// console.log("Calling search API", query, metadata);
|
||||
return await queryService
|
||||
.search({
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: {
|
||||
.search(
|
||||
{
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: {
|
||||
sql: query,
|
||||
sql_mode: "full",
|
||||
start_time: startISOTimestamp,
|
||||
end_time: endISOTimestamp,
|
||||
size: -1,
|
||||
query: {
|
||||
sql: query,
|
||||
sql_mode: "full",
|
||||
start_time: startISOTimestamp,
|
||||
end_time: endISOTimestamp,
|
||||
size: -1,
|
||||
},
|
||||
},
|
||||
page_type: pageType,
|
||||
},
|
||||
page_type: pageType,
|
||||
})
|
||||
searchType.value ?? "Dashboards"
|
||||
)
|
||||
.then((res) => {
|
||||
// Set searchQueryData.data to the API response hits
|
||||
// state.data = res.data.hits;
|
||||
|
||||
@@ -1488,7 +1488,7 @@ const useLogs = () => {
|
||||
org_identifier: searchObj.organizationIdetifier,
|
||||
query: queryReq,
|
||||
page_type: searchObj.data.stream.streamType,
|
||||
})
|
||||
},"UI")
|
||||
.then(async (res) => {
|
||||
// check for total records update for the partition and update pagination accordingly
|
||||
// searchObj.data.queryResults.partitionDetail.partitions.forEach(
|
||||
@@ -1591,7 +1591,7 @@ const useLogs = () => {
|
||||
org_identifier: searchObj.organizationIdetifier,
|
||||
query: queryReq,
|
||||
page_type: searchObj.data.stream.streamType,
|
||||
})
|
||||
},"UI")
|
||||
.then(async (res) => {
|
||||
searchObjDebug["paginatedDataReceivedStartTime"] = performance.now();
|
||||
// check for total records update for the partition and update pagination accordingly
|
||||
@@ -1820,7 +1820,7 @@ const useLogs = () => {
|
||||
org_identifier: searchObj.organizationIdetifier,
|
||||
query: queryReq,
|
||||
page_type: searchObj.data.stream.streamType,
|
||||
})
|
||||
},"UI")
|
||||
.then(async (res) => {
|
||||
searchObjDebug["histogramProcessingStartTime"] = performance.now();
|
||||
searchObj.loading = false;
|
||||
|
||||
@@ -939,7 +939,7 @@ export default defineComponent({
|
||||
org_identifier: this.searchObj.organizationIdetifier,
|
||||
query: this.searchObj.data.customDownloadQueryObj,
|
||||
page_type: this.searchObj.data.stream.streamType,
|
||||
})
|
||||
}, "UI")
|
||||
.then((res) => {
|
||||
this.customDownloadDialog = false;
|
||||
if (res.data.hits.length > 0) {
|
||||
|
||||
@@ -225,6 +225,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:panelSchema="chartData"
|
||||
:selectedTimeObj="dashboardPanelData.meta.dateTime"
|
||||
:variablesData="{}"
|
||||
searchType="UI"
|
||||
/>
|
||||
</div>
|
||||
</template>
|
||||
|
||||
@@ -667,7 +667,7 @@ export default defineComponent({
|
||||
org_identifier: searchObj.organizationIdetifier,
|
||||
query: req,
|
||||
page_type: "traces",
|
||||
})
|
||||
}, "UI")
|
||||
.then((res) => {
|
||||
searchObj.data.traceDetails.spanList = res.data?.hits || [];
|
||||
})
|
||||
|
||||
+17
-14
@@ -16,16 +16,19 @@
|
||||
import http from "./http";
|
||||
|
||||
const search = {
|
||||
search: ({
|
||||
org_identifier,
|
||||
query,
|
||||
page_type = "logs",
|
||||
}: {
|
||||
org_identifier: string;
|
||||
query: any;
|
||||
page_type: string;
|
||||
}) => {
|
||||
const url = `/api/${org_identifier}/_search?type=${page_type}`;
|
||||
search: (
|
||||
{
|
||||
org_identifier,
|
||||
query,
|
||||
page_type = "logs",
|
||||
}: {
|
||||
org_identifier: string;
|
||||
query: any;
|
||||
page_type: string;
|
||||
},
|
||||
search_type: string = "UI"
|
||||
) => {
|
||||
const url = `/api/${org_identifier}/_search?type=${page_type}&search_type=${search_type}`;
|
||||
return http().post(url, query);
|
||||
},
|
||||
search_around: ({
|
||||
@@ -54,11 +57,11 @@ const search = {
|
||||
url = url + `&query_fn=${query_fn}`;
|
||||
}
|
||||
|
||||
if (regions.trim()!= "") {
|
||||
if (regions.trim() != "") {
|
||||
url = url + `®ions=${regions}`;
|
||||
}
|
||||
|
||||
if (clusters.trim()!= "") {
|
||||
if (clusters.trim() != "") {
|
||||
url = url + `&clusters=${clusters}`;
|
||||
}
|
||||
return http().get(url);
|
||||
@@ -140,11 +143,11 @@ const search = {
|
||||
const url = `/api/${org_identifier}/_search_partition?type=${page_type}`;
|
||||
return http().post(url, query);
|
||||
},
|
||||
get_running_queries: (org_identifier:string) => {
|
||||
get_running_queries: (org_identifier: string) => {
|
||||
const url = `/api/${org_identifier}/query_manager/status`;
|
||||
return http().get(url);
|
||||
},
|
||||
delete_running_query: (org_identifier:string,traceID: string) => {
|
||||
delete_running_query: (org_identifier: string, traceID: string) => {
|
||||
const url = `/api/${org_identifier}/query_manager/${traceID}`;
|
||||
return http().delete(url);
|
||||
},
|
||||
|
||||
@@ -84,6 +84,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:width="getPanelLayout(item, 'w')"
|
||||
:height="getPanelLayout(item, 'h')"
|
||||
:forceLoad="forceLoad"
|
||||
:searchType="searchType"
|
||||
@updated:data-zoom="$emit('updated:data-zoom', $event)"
|
||||
@onMovePanel="onMovePanel"
|
||||
@refresh="refreshDashboard"
|
||||
@@ -107,6 +108,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:panelId="viewPanelId"
|
||||
:selectedDateForViewPanel="selectedDateForViewPanel"
|
||||
:initialVariableValues="variablesData"
|
||||
:searchType="searchType"
|
||||
@close-panel="() => (showViewPanel = false)"
|
||||
:class="store.state.theme == 'dark' ? 'dark-mode' : 'bg-white'"
|
||||
@update:initial-variable-values="updateInitialVariableValues"
|
||||
@@ -175,6 +177,10 @@ export default defineComponent({
|
||||
default: false,
|
||||
required: false,
|
||||
},
|
||||
searchType: {
|
||||
default: null,
|
||||
type: String || null,
|
||||
},
|
||||
},
|
||||
|
||||
components: {
|
||||
|
||||
@@ -185,6 +185,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
@refresh="loadDashboard"
|
||||
:showTabs="true"
|
||||
:forceLoad="store.state.printMode"
|
||||
:searchType="searchType"
|
||||
/>
|
||||
|
||||
<q-dialog
|
||||
@@ -295,6 +296,9 @@ export default defineComponent({
|
||||
// initial timezone, which will come from the route query
|
||||
const initialTimezone = ref(route.query.timezone ?? null);
|
||||
|
||||
// search_type for search query
|
||||
const searchType = ref(route.query.searchtype ?? null);
|
||||
|
||||
// dispatch setPrintMode, to set print mode
|
||||
const setPrint = (printMode: any) => {
|
||||
store.dispatch("setPrintMode", printMode);
|
||||
@@ -355,6 +359,7 @@ export default defineComponent({
|
||||
...getQueryParamsForDuration(selectedDate.value),
|
||||
...variableObj,
|
||||
print: store.state.printMode,
|
||||
searchtype: route.query.searchtype,
|
||||
},
|
||||
});
|
||||
};
|
||||
@@ -586,6 +591,7 @@ export default defineComponent({
|
||||
refresh: generateDurationLabel(refreshInterval.value),
|
||||
...getQueryParamsForDuration(selectedDate.value),
|
||||
print: store.state.printMode,
|
||||
searchtype: route.query.searchtype,
|
||||
},
|
||||
});
|
||||
});
|
||||
@@ -747,6 +753,7 @@ export default defineComponent({
|
||||
printDashboard,
|
||||
initialTimezone,
|
||||
timeString,
|
||||
searchType,
|
||||
quasar,
|
||||
};
|
||||
},
|
||||
|
||||
@@ -221,6 +221,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
:width="6"
|
||||
@error="handleChartApiError"
|
||||
@updated:data-zoom="onDataZoom"
|
||||
searchType="Dashboards"
|
||||
/>
|
||||
<q-dialog v-model="showViewPanel">
|
||||
<QueryInspector
|
||||
|
||||
@@ -371,7 +371,7 @@ const getErrorLogs = () => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
errorTrackingState.data.errors = res.data.hits;
|
||||
totalErrorsCount.value = res.data.hits.reduce(
|
||||
|
||||
@@ -440,7 +440,7 @@ const getSessions = () => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
res.data.hits.forEach((hit: any) => {
|
||||
sessionState.data.sessions[hit.session_id] = hit;
|
||||
@@ -493,7 +493,7 @@ const getSessionLogs = (req: any) => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
const hits = res.data.hits;
|
||||
hits.forEach((hit: any) => {
|
||||
|
||||
@@ -99,7 +99,7 @@ const getErrorLogs = () => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
const errorIndex = res.data.hits.findIndex(
|
||||
(hit: any) => hit.error_id === errorDetails.value.error_id
|
||||
@@ -149,7 +149,7 @@ const getError = () => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
errorDetails.value = { ...res.data.hits[0] };
|
||||
errorDetails.value["category"] = [];
|
||||
|
||||
@@ -188,7 +188,7 @@ const getSession = () => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
if (res.data.hits.length === 0) {
|
||||
return;
|
||||
@@ -238,7 +238,7 @@ const getSessionSegments = () => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
// const segmentsCopy = [];
|
||||
// const viewIds = [];
|
||||
@@ -288,7 +288,7 @@ const getSessionEvents = () => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
const events = ["action", "view", "error"];
|
||||
|
||||
@@ -336,7 +336,7 @@ const getSessionErrorLogs = () => {
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: req,
|
||||
page_type: "logs",
|
||||
})
|
||||
}, "RUM")
|
||||
.then((res) => {
|
||||
const events = res.data.hits.filter((hit: any) => {
|
||||
return hit.date >= Number(sessionState.data.selectedSession.start_time);
|
||||
|
||||
@@ -175,7 +175,7 @@ export default defineComponent({
|
||||
org_identifier: store.state.selectedOrganization.identifier,
|
||||
query: queryReq,
|
||||
page_type: streamData.value?.stream_type || "logs",
|
||||
})
|
||||
}, "UI")
|
||||
.then((res) => {
|
||||
isLoading.value.pop();
|
||||
if (res.data.from > 0) {
|
||||
|
||||
Reference in New Issue
Block a user