fix test bug

Author: ioki-smore

agent/Cargo.lock

@@ -6,7 +6,7 @@ use observ_config::{Configurator, trace_config};
 use observ_core::Module;
 use observ_runtime::handle;
 use observ_sender::{Sender, elastic::ElasticSender, file::FileSender};
-use observ_trace::TraceCollector;
+use observ_trace::{TraceCollector, span::SpanConstructor};
 use tokio::{
 	sync::{mpsc, watch},
 	task::JoinHandle,
@@ -104,16 +104,17 @@ async fn run(
 	info!("Start configurator");
 	configurator.start()?;
 	info!("Starting configurator");
-	let (span_sender, span_receiver) = mpsc::channel(1024);
-	let mut trace_sender = Sender::new(
-		"Trace sender",
-		span_receiver,
-		ElasticSender::new(trace_config().load().sender.clone())?,
-		JsonEncoderBuilder::new().build(),
-	);
+	let (message_sender, message_receiver) = crossbeam_channel::bounded(1024);
+	let (span_sender, span_receiver) = crossbeam_channel::bounded(1024);
+	let sender = ElasticSender::new(trace_config().load().sender.clone())?;
+	let mut trace_sender =
+		Sender::new("Trace sender", span_receiver, sender, JsonEncoderBuilder::new().build());
 	trace_sender.start()?;
 
-	let mut trace_collector = TraceCollector::new(span_sender)?;
+	let mut span_constructor = SpanConstructor::new(message_receiver, span_sender);
+	span_constructor.start()?;
+
+	let mut trace_collector = TraceCollector::new(message_sender)?;
 	trace_collector.start()?;
 
 	let _ = state_tx.send(State::Running);
@@ -128,6 +129,7 @@ async fn run(
 	}
 
 	trace_collector.stop().await?;
+	span_constructor.stop().await?;
 	trace_sender.stop().await?;
 	configurator.stop().await?;
 

agent/crates/deeptrace/src/agent.rs

@@ -13,4 +13,6 @@ pub enum AgentError {
 	Sender(#[from] observ_sender::SendError),
 	#[error("Elastic sender error: {0}")]
 	ElasticSender(#[from] observ_sender::elastic::ElasticError),
+	#[error("Span constructor error: {0}")]
+	SpanConstructor(#[from] observ_trace::span::SpanError),
 }

agent/crates/deeptrace/src/error.rs

@@ -5,7 +5,7 @@ use crate::{
 };
 use aya_ebpf::{
 	check_bounds_signed,
-	helpers::{bpf_probe_read_kernel_str_bytes, r#gen}
+	helpers::{bpf_probe_read_kernel_str_bytes, r#gen},
 };
 use core::cmp::min;
 

agent/crates/ebpf-common/src/buffer.rs

@@ -212,7 +212,7 @@ macro_rules! try_or_log {
 			Ok(val) => val,
 			Err(code) => {
 				// Remove aya_log_ebpf::error! to avoid __bpf_trap
-				aya_log_ebpf::error!(
+				aya_log_ebpf::debug!(
 					$ctx,
 					"ERROR: [{:X}:{:X}]",
 					$crate::error::extract_module(code),

agent/crates/ebpf-common/src/error.rs

@@ -1,6 +1,6 @@
 use crate::{
-	cassandra::Cassandra, dns::DNS, http1::HTTP1, memcached::Memcached, mongodb::MongoDB,
-	mysql::MySQL, redis::Redis, rocketmq::RocketMQ, thrift::Thrift, Classification,
+	Classification, cassandra::Cassandra, dns::DNS, http1::HTTP1, memcached::Memcached,
+	mongodb::MongoDB, mysql::MySQL, redis::Redis, rocketmq::RocketMQ, thrift::Thrift,
 };
 use aya_ebpf::programs::TracePointContext;
 use ebpf_common::{

agent/crates/l7-parser/src/infer.rs

@@ -23,6 +23,7 @@ observ-metric-utils.workspace = true
 observ-network.workspace = true
 observ-runtime.workspace = true
 
+crossbeam-channel.workspace = true
 log.workspace = true
 thiserror.workspace = true
 tokio.workspace = true

agent/crates/observ-metric/Cargo.toml

@@ -1,3 +1,4 @@
+use crossbeam_channel::Sender;
 pub use error::MetricError;
 use host::HostCollector;
 use log::{info, warn};
@@ -13,7 +14,7 @@ use std::{
 	},
 	time::Duration,
 };
-use tokio::{sync::mpsc::Sender, task::JoinHandle, time};
+use tokio::{task::JoinHandle, time};
 use tokio_stream::{StreamExt, wrappers::IntervalStream};
 mod error;
 mod host;
@@ -74,7 +75,7 @@ impl Module for MetricCollector {
 					}
 				}
 				let metrics = mem::take(&mut buffer);
-				output.send(metrics).await.map_err(|_| MetricError::Send)?;
+				output.send(metrics).map_err(|_| MetricError::Send)?;
 				// for metrics in buffer.drain(..) {
 				// 	if output.send(metrics).await.is_err() {
 				// 		warn!("Metric channel closed");

agent/crates/observ-metric/src/lib.rs

@@ -21,8 +21,10 @@ codec.workspace = true
 arc-swap.workspace = true
 bytes.workspace = true
 chrono.workspace = true
+crossbeam-channel.workspace = true
 elasticsearch.workspace = true
 log.workspace = true
 serde.workspace = true
+serde_json.workspace = true
 thiserror.workspace = true
 tokio.workspace = true

agent/crates/observ-sender/Cargo.toml

@@ -8,9 +8,11 @@ use elasticsearch::{
 	},
 };
 pub use error::ElasticError;
+use log::info;
 use observ_config::{ElasticSenderConfig, elastic_sender_config};
 use observ_core::{Sendable, Sender};
 use serde::Serialize;
+use serde_json::json;
 use std::time::Duration;
 
 mod error;
@@ -42,33 +44,25 @@ impl ElasticSender {
 impl<S: Sendable + Serialize> Sender<S> for ElasticSender {
 	type Error = ElasticError;
 	async fn send(&mut self, item: BytesMut) -> Result<(), Self::Error> {
-		self.buf.push(BytesMut::from(
-			format!(
-				r#"{{
-					"index": {{
-						"_index": "{}"
-					}}
-				}}"#,
-				self.config.index_name
-			)
-			.as_bytes(),
-		));
+		let index = json!({
+			"index": {
+				"_index": self.config.index_name,
+			}
+		})
+		.to_string();
+		self.buf.push(BytesMut::from(index.as_bytes()));
 		self.buf.push(item);
-		if self.buf.len() > self.config.bulk_size * 1024 {
+		if self.buf.len() > self.config.bulk_size * 2 {
 			<Self as Sender<S>>::flush(self).await?;
 		}
 		Ok(())
 	}
 
 	async fn flush(&mut self) -> Result<(), Self::Error> {
 		let bulk_body = self.buf.drain(..).collect();
-		let response = self
-			.client
-			.bulk(BulkParts::Index(self.config.index_name.as_str()))
-			.body(bulk_body)
-			.send()
-			.await?;
+		let response = self.client.bulk(BulkParts::None).body(bulk_body).send().await?;
 		let status = response.status_code();
+		info!("Elastic response: {}", status);
 		if !status.is_success() {
 			let err = response.text().await?;