feat: add some debug log

src/config/src/metrics.rs

@@ -198,6 +198,32 @@ pub static INGEST_MEMTABLE_FILES: Lazy<IntGaugeVec> = Lazy::new(|| {
  .expect("Metric created")
 });
 
+pub static INGEST_MEMTABLE_LOCK_TIME: Lazy<HistogramVec> = Lazy::new(|| {
+ HistogramVec::new(
+ HistogramOpts::new("ingest_memtable_lock_time", "ingest memtable lock time")
+ .namespace(NAMESPACE)
+ .buckets(vec![
+ 0.2, 0.5, 1.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0, 2000.0,
+ ])
+ .const_labels(create_const_labels()),
+ &["organization"],
+ )
+ .expect("Metric created")
+});
+
+pub static INGEST_WAL_LOCK_TIME: Lazy<HistogramVec> = Lazy::new(|| {
+ HistogramVec::new(
+ HistogramOpts::new("ingest_wal_lock_time", "ingest wal lock time")
+ .namespace(NAMESPACE)
+ .buckets(vec![
+ 0.2, 0.5, 1.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0, 2000.0,
+ ])
+ .const_labels(create_const_labels()),
+ &["organization"],
+ )
+ .expect("Metric created")
+});
+
 // querier memory cache stats
 pub static QUERY_MEMORY_CACHE_LIMIT_BYTES: Lazy<IntGaugeVec> = Lazy::new(|| {
  IntGaugeVec::new(
@@ -540,6 +566,44 @@ pub static MEMORY_USAGE: Lazy<IntGaugeVec> = Lazy::new(|| {
  .expect("Metric created")
 });
 
+pub static SPAN_CALLS: Lazy<IntCounterVec> = Lazy::new(|| {
+ IntCounterVec::new(
+ Opts::new("span_calls", "span calls total")
+ .namespace(NAMESPACE)
+ .const_labels(create_const_labels()),
+ &[
+ "organization",
+ "stream",
+ "service_name",
+ "operation_name",
+ "status_code",
+ "span_kind",
+ ],
+ )
+ .expect("Metric created")
+});
+
+pub static SPAN_DURATION_MILLISECONDS: Lazy<HistogramVec> = Lazy::new(|| {
+ HistogramVec::new(
+ HistogramOpts::new("span_duration_milliseconds", "span duration milliseconds")
+ .namespace(NAMESPACE)
+ .buckets(vec![
+ 0.5, 1.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0, 2000.0, 5000.0,
+ 10000.0, 60000.0,
+ ])
+ .const_labels(create_const_labels()),
+ &[
+ "organization",
+ "stream",
+ "service_name",
+ "operation_name",
+ "status_code",
+ "span_kind",
+ ],
+ )
+ .expect("Metric created")
+});
+
 fn register_metrics(registry: &Registry) {
  // http latency
  registry
@@ -582,7 +646,12 @@ fn register_metrics(registry: &Registry) {
  registry
  .register(Box::new(INGEST_MEMTABLE_FILES.clone()))
  .expect("Metric registered");
-
+ registry
+ .register(Box::new(INGEST_MEMTABLE_LOCK_TIME.clone()))
+ .expect("Metric registered");
+ registry
+ .register(Box::new(INGEST_WAL_LOCK_TIME.clone()))
+ .expect("Metric registered");
  // querier stats
  registry
  .register(Box::new(QUERY_MEMORY_CACHE_LIMIT_BYTES.clone()))
@@ -679,6 +748,14 @@ fn register_metrics(registry: &Registry) {
  registry
  .register(Box::new(MEMORY_USAGE.clone()))
  .expect("Metric registered");
+
+ // span metrics
+ registry
+ .register(Box::new(SPAN_CALLS.clone()))
+ .expect("Metric registered");
+ registry
+ .register(Box::new(SPAN_DURATION_MILLISECONDS.clone()))
+ .expect("Metric registered");
 }
 
 fn create_const_labels() -> HashMap<String, String> {

src/handler/grpc/request/traces.rs

@@ -13,7 +13,7 @@
 // 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 config::CONFIG;
+use config::{ider, CONFIG};
 use opentelemetry_proto::tonic::collector::trace::v1::{
  trace_service_server::TraceService, ExportTraceServiceRequest, ExportTraceServiceResponse,
 };
@@ -51,12 +51,19 @@ impl TraceService for TraceServer {
  in_stream_name = Some(stream_name.to_str().unwrap());
  };
 
+ let default_session_id = tonic::metadata::MetadataValue::try_from(ider::uuid()).unwrap();
+ let session_id = metadata
+ .get("session_id")
+ .unwrap_or(&default_session_id)
+ .to_str()
+ .unwrap();
  let resp = handle_trace_request(
  org_id.unwrap().to_str().unwrap(),
  0,
  in_req,
  true,
  in_stream_name,
+ session_id,
  )
  .await;
  if resp.is_ok() {

src/ingester/Cargo.toml

@@ -25,7 +25,7 @@ serde_json.workspace = true
 snafu.workspace = true
 tokio.workspace = true
 async-walkdir.workspace = true
-
+futures-locks = "0.7.1"
 [[bin]]
 name = "wal-reader"
 path = "bin/wal-reader.rs"

src/ingester/src/entry.rs

@@ -35,6 +35,7 @@ pub struct Entry {
  pub partition_key: Arc<str>, // 2023/12/18/00/country=US/state=CA
  pub data: Vec<Arc<serde_json::Value>>,
  pub data_size: usize,
+ pub session_id: Arc<str>,
 }
 
 impl Entry {
@@ -45,6 +46,7 @@ impl Entry {
  partition_key: "".into(),
  data: Vec::new(),
  data_size: 0,
+ session_id: "".into(),
  }
  }
  pub fn into_bytes(&mut self) -> Result<Vec<u8>> {
@@ -96,6 +98,7 @@ impl Entry {
  partition_key: partition_key.into(),
  data,
  data_size: data_len as usize,
+ session_id: "".into(),
  })
  }
 

src/ingester/src/writer.rs

@@ -24,9 +24,10 @@ use std::{
 use arrow_schema::Schema;
 use chrono::{Duration, Utc};
 use config::{metrics, CONFIG};
+use futures::lock::Mutex;
 use once_cell::sync::Lazy;
 use snafu::ResultExt;
-use tokio::sync::{Mutex, RwLock};
+use tokio::sync::RwLock;
 use wal::Writer as WalWriter;
 
 use crate::{
@@ -178,6 +179,7 @@ impl Writer {
  if entry.data.is_empty() && !check_ttl {
  return Ok(());
  }
+ let session_id = entry.session_id.to_string();
  let entry_bytes = if !check_ttl {
  entry.into_bytes()?
  } else {
@@ -223,21 +225,36 @@ impl Writer {
  let key = self.key.clone();
  let path = old_wal.path().clone();
  let path_str = path.display().to_string();
+ let sid = session_id.clone();
  tokio::task::spawn(async move {
- log::info!("[INGESTER:WAL] start add to IMMUTABLES, file: {}", path_str,);
+ log::info!(
+ "[INGESTER:WAL] [{sid}] start add to IMMUTABLES, file: {}",
+ path_str,
+ );
  IMMUTABLES.write().await.insert(
  path,
  Arc::new(immutable::Immutable::new(thread_id, key.clone(), old_mem)),
  );
- log::info!("[INGESTER:WAL] dones add to IMMUTABLES, file: {}", path_str);
+ log::info!(
+ "[INGESTER:WAL] [{sid}] dones add to IMMUTABLES, file: {}",
+ path_str
+ );
  });
  }
 
  if !check_ttl {
  // write into wal
+ let start = std::time::Instant::now();
  wal.write(&entry_bytes, false).context(WalSnafu)?;
+ metrics::INGEST_WAL_LOCK_TIME
+ .with_label_values(&[&self.key.org_id])
+ .observe(start.elapsed().as_millis() as f64);
  // write into memtable
+ let start = std::time::Instant::now();
  mem.write(schema, entry).await?;
+ metrics::INGEST_MEMTABLE_LOCK_TIME
+ .with_label_values(&[&self.key.org_id])
+ .observe(start.elapsed().as_millis() as f64);
  }
 
  Ok(())

src/main.rs

@@ -357,10 +357,14 @@ fn init_router_grpc_server(
  .max_encoding_message_size(CONFIG.grpc.max_message_size * 1024 * 1024);
  let metrics_svc = MetricsServiceServer::new(router::grpc::ingest::metrics::MetricsServer)
  .send_compressed(CompressionEncoding::Gzip)
- .accept_compressed(CompressionEncoding::Gzip);
+ .accept_compressed(CompressionEncoding::Gzip)
+ .max_decoding_message_size(CONFIG.grpc.max_message_size * 1024 * 1024)
+ .max_encoding_message_size(CONFIG.grpc.max_message_size * 1024 * 1024);
  let traces_svc = TraceServiceServer::new(router::grpc::ingest::traces::TraceServer)
  .send_compressed(CompressionEncoding::Gzip)
- .accept_compressed(CompressionEncoding::Gzip);
+ .accept_compressed(CompressionEncoding::Gzip)
+ .max_decoding_message_size(CONFIG.grpc.max_message_size * 1024 * 1024)
+ .max_encoding_message_size(CONFIG.grpc.max_message_size * 1024 * 1024);
 
  tokio::task::spawn(async move {
  log::info!("starting gRPC server at {}", gaddr);

src/router/grpc/ingest/traces.rs

@@ -14,7 +14,7 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
 
 use async_trait::async_trait;
-use config::CONFIG;
+use config::{ider, CONFIG};
 use opentelemetry_proto::tonic::collector::trace::v1::{
  trace_service_client::TraceServiceClient, trace_service_server::TraceService,
  ExportTraceServiceRequest, ExportTraceServiceResponse,
@@ -33,7 +33,7 @@ impl TraceService for TraceServer {
  &self,
  request: Request<ExportTraceServiceRequest>,
  ) -> Result<Response<ExportTraceServiceResponse>, Status> {
- let (metadata, extensions, message) = request.into_parts();
+ let (mut metadata, extensions, message) = request.into_parts();
 
  // basic validation
  if !metadata.contains_key(&CONFIG.grpc.org_header_key) {
@@ -42,6 +42,8 @@ impl TraceService for TraceServer {
  &CONFIG.grpc.org_header_key
  )));
  }
+ let session_id = ider::uuid();
+ metadata.insert("session_id", session_id.parse().unwrap());
 
  // call ingester
  let mut request = Request::from_parts(metadata, extensions, message);
@@ -60,10 +62,27 @@ impl TraceService for TraceServer {
  req.metadata_mut().insert("authorization", token.clone());
  Ok(req)
  });
- client
+
+ let start = std::time::Instant::now();
+ match client
  .send_compressed(CompressionEncoding::Gzip)
  .accept_compressed(CompressionEncoding::Gzip)
+ .max_decoding_message_size(CONFIG.grpc.max_message_size * 1024 * 1024)
+ .max_encoding_message_size(CONFIG.grpc.max_message_size * 1024 * 1024)
  .export(request)
  .await
+ {
+ Ok(res) => {
+ if res.get_ref().partial_success.is_some() {
+ log::error!("export trace partial_success response:{:?}", res.get_ref());
+ }
+ Ok(res)
+ }
+ Err(e) => {
+ let time = start.elapsed().as_secs_f64();
+ log::error!("[{session_id}]export trace status: {e}, elapsed: {time}");
+ Err(e)
+ }
+ }
  }
 }

src/service/ingestion/mod.rs

@@ -364,6 +364,13 @@ pub async fn write_file(
  buf: HashMap<String, SchemaRecords>,
 ) -> RequestStats {
  let mut req_stats = RequestStats::default();
+ let (stream_name, seesion_id) =
+ if let [stream_name, seesion_id] = stream_name.split('^').collect::<Vec<_>>().as_slice() {
+ (*stream_name, *seesion_id)
+ } else {
+ (stream_name, "")
+ };
+
  for (hour_key, entry) in buf {
  if entry.records.is_empty() {
  continue;
@@ -378,6 +385,7 @@ pub async fn write_file(
  partition_key: Arc::from(hour_key.as_str()),
  data: entry.records,
  data_size: entry.records_size,
+ session_id: Arc::from(seesion_id),
  },
  false,
  )
@@ -625,4 +633,29 @@ mod tests {
  );
  assert!(result.is_err())
  }
+
+ #[tokio::test]
+ async fn test_stream_name_with_sid() {
+ let stream_name = "default^123456";
+ let (stream_name, seesion_id) = if let [stream_name, seesion_id] =
+ stream_name.split('^').collect::<Vec<_>>().as_slice()
+ {
+ (*stream_name, *seesion_id)
+ } else {
+ (stream_name, "")
+ };
+ assert_eq!(stream_name, "default");
+ assert_eq!(seesion_id, "123456");
+
+ let stream_name = "default";
+ let (stream_name, seesion_id) = if let [stream_name, seesion_id] =
+ stream_name.split('^').collect::<Vec<_>>().as_slice()
+ {
+ (*stream_name, *seesion_id)
+ } else {
+ (stream_name, "")
+ };
+ assert_eq!(stream_name, "default");
+ assert_eq!(seesion_id, "");
+ }
 }