fix

Author: JqRrt

cozeloop/internal/trace/queue_manager.py

@@ -55,6 +55,7 @@ def __init__(self, options: BatchQueueManagerOptions):
         self.batch = []
         self.batch_byte_size = 0
         self.batch_lock = threading.Lock()
+        self.size_lock = threading.Lock()
         self.export_func = options.export_func
         self.stop_event = threading.Event()
 
@@ -88,8 +89,9 @@ def worker(self):
     def is_should_export(self) -> bool:
         if len(self.batch) >= self.options.max_export_batch_length:
             return True
-        if self.batch_byte_size >= self.options.max_export_batch_byte_size:
-            return True
+        with self.size_lock:
+            if self.batch_byte_size >= self.options.max_export_batch_byte_size:
+                return True
         return False
 
     def _drain_queue(self):
@@ -125,7 +127,8 @@ def _do_export_batch(self):
                 if self.export_func:
                     self.export_func({}, self.batch)
                 self.batch = []
-                self.batch_byte_size = 0
+                with self.size_lock:
+                    self.batch_byte_size = 0
 
     def enqueue(self, item: Any, byte_size: int):
         if self.stop_event.is_set():
@@ -144,7 +147,7 @@ def enqueue(self, item: Any, byte_size: int):
             detail_msg = f"{self.options.queue_name} queue is full, dropped span"
             self.dropped += 1
         else:
-            with self.batch_lock:
+            with self.size_lock:
                 self.batch_byte_size += byte_size
 
         event_typ = "queue_manager.file_entry.rate"

cozeloop/internal/trace/span_processor.py

@@ -71,7 +71,7 @@ def __init__(
         if queue_conf:
             if queue_conf.span_queue_length > 0:
                 span_queue_length = queue_conf.span_queue_length
-            if queue_conf.span_max_export_batch_length > 0:
+            if queue_conf.span_max_export_batch_length > 0:  # todo: need max limit
                 span_export_batch_size = queue_conf.span_max_export_batch_length
 
         self.file_retry_qm = BatchQueueManager(
@@ -117,7 +117,8 @@ def __init__(
                 max_queue_length=span_queue_length,
                 max_export_batch_length=span_export_batch_size,
                 max_export_batch_byte_size=DEFAULT_MAX_EXPORT_BATCH_BYTE_SIZE,
-                export_func=self._new_export_spans_func(self.exporter, self.span_retry_qm, self.file_qm, finish_event_processor),
+                export_func=self._new_export_spans_func(self.exporter, self.span_retry_qm, self.file_qm,
+                                                        finish_event_processor),
                 finish_event_processor=finish_event_processor,
             )
         )