mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-05 00:09:35 +00:00
fix: add some debug logs for query slow (#9072)
This commit is contained in:
@@ -214,6 +214,7 @@ impl FlightService for FlightServiceImpl {
|
||||
|
||||
let mut stream = FlightEncoderStreamBuilder::new(write_options, 33554432)
|
||||
.with_trace_id(trace_id.to_string())
|
||||
.with_is_super(is_super_cluster)
|
||||
.with_defer(defer)
|
||||
.with_start(start)
|
||||
.with_custom_message(PreCustomMessage::ScanStats(scan_stats))
|
||||
|
||||
@@ -33,6 +33,7 @@ pub struct FlightEncoderStreamBuilder {
|
||||
custom_messages: Vec<PreCustomMessage>,
|
||||
// query context
|
||||
trace_id: String,
|
||||
is_super: bool,
|
||||
defer: Option<AsyncDefer>,
|
||||
start: std::time::Instant,
|
||||
}
|
||||
@@ -44,6 +45,7 @@ impl FlightEncoderStreamBuilder {
|
||||
queue: VecDeque::new(),
|
||||
custom_messages: vec![],
|
||||
trace_id: String::new(),
|
||||
is_super: false,
|
||||
defer: None,
|
||||
start: std::time::Instant::now(),
|
||||
}
|
||||
@@ -59,6 +61,11 @@ impl FlightEncoderStreamBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_is_super(mut self, is_super: bool) -> Self {
|
||||
self.is_super = is_super;
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_defer(mut self, defer: Option<AsyncDefer>) -> Self {
|
||||
self.defer = defer;
|
||||
self
|
||||
@@ -83,11 +90,15 @@ impl FlightEncoderStreamBuilder {
|
||||
done: false,
|
||||
custom_messages: self.custom_messages,
|
||||
trace_id: self.trace_id,
|
||||
is_super: self.is_super,
|
||||
defer: self.defer,
|
||||
start: self.start,
|
||||
first_batch: true,
|
||||
span,
|
||||
child_span,
|
||||
req_id: 0,
|
||||
req_last_time: std::time::Instant::now(),
|
||||
print_key_event: config::get_config().common.print_key_event,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -101,10 +112,14 @@ pub struct FlightEncoderStream {
|
||||
custom_messages: Vec<PreCustomMessage>,
|
||||
// query context
|
||||
trace_id: String,
|
||||
is_super: bool,
|
||||
defer: Option<AsyncDefer>,
|
||||
start: std::time::Instant,
|
||||
span: tracing::Span,
|
||||
child_span: tracing::Span,
|
||||
req_id: u64,
|
||||
req_last_time: std::time::Instant,
|
||||
print_key_event: bool,
|
||||
}
|
||||
|
||||
impl FlightEncoderStream {
|
||||
@@ -155,10 +170,31 @@ impl Stream for FlightEncoderStream {
|
||||
) -> Poll<Option<Self::Item>> {
|
||||
loop {
|
||||
if self.done && self.queue.is_empty() {
|
||||
if self.print_key_event {
|
||||
log::info!(
|
||||
"[trace_id {}] flight->search: stream Poll::Ready(None) is_super: {}, took: {} ms",
|
||||
self.trace_id,
|
||||
self.is_super,
|
||||
self.req_last_time.elapsed().as_millis(),
|
||||
);
|
||||
}
|
||||
return Poll::Ready(None);
|
||||
}
|
||||
|
||||
if let Some(data) = self.queue.pop_front() {
|
||||
self.req_id += 1;
|
||||
let took = self.req_last_time.elapsed().as_millis();
|
||||
self.req_last_time = std::time::Instant::now();
|
||||
if self.print_key_event
|
||||
&& (took > 100 || config::utils::util::is_power_of_two(self.req_id))
|
||||
{
|
||||
log::info!(
|
||||
"[trace_id {}] flight->search: stream Poll::Ready(#{}) message, is_super: {}, took: {took} ms",
|
||||
self.trace_id,
|
||||
self.req_id,
|
||||
self.is_super,
|
||||
);
|
||||
}
|
||||
return Poll::Ready(Some(Ok(data)));
|
||||
}
|
||||
|
||||
@@ -181,8 +217,9 @@ impl Stream for FlightEncoderStream {
|
||||
self.done = true;
|
||||
self.queue.clear();
|
||||
log::error!(
|
||||
"[trace_id {}] flight->search: stream error: {e:?}, took: {} ms",
|
||||
"[trace_id {}] flight->search: stream error: {e:?}, is_super: {}, took: {} ms",
|
||||
self.trace_id,
|
||||
self.is_super,
|
||||
self.start.elapsed().as_millis()
|
||||
);
|
||||
return Poll::Ready(Some(Err(tonic::Status::internal(e.to_string()))));
|
||||
@@ -211,8 +248,11 @@ impl Stream for FlightEncoderStream {
|
||||
impl Drop for FlightEncoderStream {
|
||||
fn drop(&mut self) {
|
||||
let trace_id = &self.trace_id;
|
||||
let is_super = self.is_super;
|
||||
let took = self.start.elapsed().as_millis();
|
||||
log::info!("[trace_id {trace_id}] flight->search: stream end, took: {took} ms",);
|
||||
log::info!(
|
||||
"[trace_id {trace_id}] flight->search: stream end, is_super: {is_super}, took: {took} ms"
|
||||
);
|
||||
|
||||
let _child_enter = self.child_span.enter();
|
||||
let _enter = self.span.enter();
|
||||
@@ -232,7 +272,9 @@ impl Drop for FlightEncoderStream {
|
||||
if let Some(defer) = self.defer.take() {
|
||||
drop(defer);
|
||||
} else {
|
||||
log::info!("[trace_id {trace_id}] flight->search: drop FlightEncoderStream",);
|
||||
log::info!(
|
||||
"[trace_id {trace_id}] flight->search: drop FlightEncoderStream, is_super: {is_super}",
|
||||
);
|
||||
// clear session data
|
||||
clear_session_data(&self.trace_id);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user