Support custom buffer len for trace stats concentrator

Author: lym953

libdd-trace-stats/benches/span_concentrator_bench.rs

@@ -47,6 +47,7 @@ pub fn criterion_benchmark(c: &mut Criterion) {
         now,
         vec![],
         vec!["db_name".into(), "bucket_s3".into()],
+        2,
     );
     let mut spans = vec![];
     for trace_id in 1..100 {

libdd-trace-stats/src/span_concentrator/mod.rs

@@ -79,6 +79,7 @@ impl SpanConcentrator {
         now: SystemTime,
         span_kinds_stats_computed: Vec<String>,
         peer_tag_keys: Vec<String>,
+        buffer_len: usize,
     ) -> SpanConcentrator {
         SpanConcentrator {
             bucket_size: bucket_size.as_nanos() as u64,
@@ -87,7 +88,7 @@ impl SpanConcentrator {
                 system_time_to_unix_duration(now).as_nanos() as u64,
                 bucket_size.as_nanos() as u64,
             ),
-            buffer_len: 2,
+            buffer_len,
             span_kinds_stats_computed,
             peer_tag_keys,
         }

libdd-trace-stats/src/span_concentrator/tests.rs

@@ -100,7 +100,7 @@ fn assert_counts_equal(expected: Vec<pb::ClientGroupedStats>, actual: Vec<pb::Cl
 fn test_concentrator_oldest_timestamp_cold() {
     let now = SystemTime::now();
     let mut concentrator =
-        SpanConcentrator::new(Duration::from_nanos(BUCKET_SIZE), now, vec![], vec![]);
+        SpanConcentrator::new(Duration::from_nanos(BUCKET_SIZE), now, vec![], vec![], 2);
     let mut spans = vec![
         get_test_span(now, 1, 0, 50, 5, "A1", "resource1", 0),
         get_test_span(now, 1, 0, 40, 4, "A1", "resource1", 0),
@@ -150,7 +150,7 @@ fn test_concentrator_oldest_timestamp_cold() {
 fn test_concentrator_oldest_timestamp_hot() {
     let now = SystemTime::now();
     let mut concentrator =
-        SpanConcentrator::new(Duration::from_nanos(BUCKET_SIZE), now, vec![], vec![]);
+        SpanConcentrator::new(Duration::from_nanos(BUCKET_SIZE), now, vec![], vec![], 2);
     let mut spans = vec![
         get_test_span(now, 1, 0, 50, 5, "A1", "resource1", 0),
         get_test_span(now, 1, 0, 40, 4, "A1", "resource1", 0),
@@ -223,7 +223,7 @@ fn test_concentrator_oldest_timestamp_hot() {
 fn test_concentrator_stats_totals() {
     let now = SystemTime::now();
     let mut concentrator =
-        SpanConcentrator::new(Duration::from_nanos(BUCKET_SIZE), now, vec![], vec![]);
+        SpanConcentrator::new(Duration::from_nanos(BUCKET_SIZE), now, vec![], vec![], 2);
     let aligned_now = align_timestamp(
         system_time_to_unix_duration(now).as_nanos() as u64,
         concentrator.bucket_size,
@@ -283,7 +283,7 @@ fn test_concentrator_stats_totals() {
 fn test_concentrator_stats_counts() {
     let now = SystemTime::now();
     let mut concentrator =
-        SpanConcentrator::new(Duration::from_nanos(BUCKET_SIZE), now, vec![], vec![]);
+        SpanConcentrator::new(Duration::from_nanos(BUCKET_SIZE), now, vec![], vec![], 2);
     let aligned_now = align_timestamp(
         system_time_to_unix_duration(now).as_nanos() as u64,
         concentrator.bucket_size,
@@ -578,6 +578,7 @@ fn test_span_should_be_included_in_stats() {
         now,
         get_span_kinds(),
         vec![],
+        2,
     );
     for span in &spans {
         concentrator.add_span(span);
@@ -656,6 +657,7 @@ fn test_ignore_partial_spans() {
         now,
         get_span_kinds(),
         vec![],
+        2,
     );
     for span in &spans {
         concentrator.add_span(span);
@@ -679,6 +681,7 @@ fn test_force_flush() {
         now,
         get_span_kinds(),
         vec![],
+        2,
     );
     for span in &spans {
         concentrator.add_span(span);
@@ -760,12 +763,14 @@ fn test_peer_tags_aggregation() {
         now,
         get_span_kinds(),
         vec![],
+        2,
     );
     let mut concentrator_with_peer_tags = SpanConcentrator::new(
         Duration::from_nanos(BUCKET_SIZE),
         now,
         get_span_kinds(),
         vec!["db.instance".to_string(), "db.system".to_string()],
+        2,
     );
     for span in &spans {
         concentrator_without_peer_tags.add_span(span);
@@ -1014,6 +1019,7 @@ fn test_pb_span() {
         now,
         get_span_kinds(),
         vec!["db.instance".to_string(), "db.system".to_string()],
+        2,
     );
     let aligned_now = align_timestamp(
         system_time_to_unix_duration(now).as_nanos() as u64,