update

Author: JqRrt

cozeloop/internal/trace/exporter.py

@@ -18,10 +18,10 @@
 logger = logging.getLogger(__name__)
 
 class Exporter:
-    def export_spans(self, ctx: dict, spans: List['UploadSpan']) -> (bool, str):
+    def export_spans(self, ctx: dict, spans: List['UploadSpan']):
         raise NotImplementedError
 
-    def export_files(self, ctx: dict, files: List['UploadFile']) -> (bool, str):
+    def export_files(self, ctx: dict, files: List['UploadFile']):
         raise NotImplementedError
 
 
@@ -54,7 +54,7 @@ def __init__(
         self.client = client
         self.upload_path = upload_path
 
-    def export_files(self, ctx: dict, files: List['UploadFile']) -> (bool, str):
+    def export_files(self, ctx: dict, files: List['UploadFile']):
         for file in files:
             if not file:
                 continue
@@ -69,16 +69,16 @@ def export_files(self, ctx: dict, files: List['UploadFile']) -> (bool, str):
                     {"workspace_id": file.space_id},
                 )
                 if resp.code != 0: # todo: some err code do not need retry
-                    return False, f"export files[{file.tos_key}] fail, code:[{resp.code}], msg:[{resp.msg}]"
+                    raise Exception(f"code:[{resp.code}], msg:[{resp.msg}]")
             except Exception as e:
-                return False, f"export files[{file.tos_key}] fail, err:[{e}], file.name:[{file.name}]"
+                raise Exception(f"export files[{file.tos_key}] fail, err:[{e}], file.name:[{file.name}]")
 
             logger.debug(f"uploadFile end, file name: {file.name}")
-        return True, ''
 
-    def export_spans(self, ctx: dict, spans: List['UploadSpan']) -> (bool, str):
-        if not spans:
-            return True, ''
+
+    def export_spans(self, ctx: dict, spans: List['UploadSpan']):
+        if not spans or len(spans) == 0:
+            return
 
         try:
             resp = self.client.post(
@@ -87,11 +87,9 @@ def export_spans(self, ctx: dict, spans: List['UploadSpan']) -> (bool, str):
                 UploadSpanData(spans=spans),
             )
             if resp.code != 0: # todo: some err code do not need retry
-                return False, f"export spans fail, code:[{resp.code}], msg:[{resp.msg}]"
+                raise Exception(f"code:[{resp.code}], msg:[{resp.msg}]")
         except Exception as e:
-            return False, f"export spans fail, err:[{e}]"
-
-        return True, ""
+            raise Exception(f"export spans fail, err:[{e}]")
 
 
 class UploadSpanData(BaseModel):

cozeloop/internal/trace/span_processor.py

@@ -153,34 +153,43 @@ def _new_export_spans_func(
     ):
         def export_func(ctx: dict, items: List[Any]):
             spans = [s for s in items if isinstance(s, Span)]
+            if not spans or len(spans) == 0:
+                return
             try:
                 upload_spans, upload_files = transfer_to_upload_span_and_file(spans)
             except Exception as e:
                 logger.warning(f"transfer_to_upload_span_and_file fail")
+                return
 
-            is_fail = False
-            err_msg = ""
+            event_err_msg = ""
             before = time.perf_counter()
-            is_pass, msg = exporter.export_spans(ctx, upload_spans)
+
+            is_export_pass = True
+            export_msg = ""
+            try:
+                exporter.export_spans(ctx, upload_spans)
+            except Exception as e:
+                is_export_pass = False
+                export_msg = f"{e}"
+
             elapsed_time_ms = (time.perf_counter() - before) * 1000
-            if not is_pass:
+            if not is_export_pass:
                 if span_retry_queue:
                     for span in spans:
                         span_retry_queue.enqueue(span, span.bytes_size)
-                    err_msg = f'{msg}, retry later'
+                    event_err_msg = f'{export_msg}, retry later'
                 else:
-                    err_msg = f'{msg}, retry second time failed'
-                is_fail = True
+                    event_err_msg = f'{export_msg}, retry second time failed'
             else:
                 for file in upload_files:
                     if file and file_queue:
                         file_queue.enqueue(file, len(file.data))
             if finish_event_processor:
                 finish_event_processor(FinishEventInfo(
                     event_type="exporter.span_flush.rate",
-                    is_event_fail=is_fail,
+                    is_event_fail=not is_export_pass,
                     item_num=len(spans),
-                    detail_msg=err_msg,
+                    detail_msg=event_err_msg,
                     extra_params=FinishEventInfoExtra(
                         latency_ms=int(elapsed_time_ms)
                     )
@@ -196,26 +205,34 @@ def _new_export_files_func(
     ):
         def export_func(ctx: dict, items: List[Any]):
             files = [f for f in items if isinstance(f, UploadFile)]
+            if not files or len(files) == 0:
+                return
 
-            is_fail = False
-            err_msg = ""
+            event_err_msg = ""
             before = time.perf_counter()
-            is_pass, msg = exporter.export_files(ctx, files)
+
+            is_export_pass = True
+            export_msg = ""
+            try:
+                exporter.export_files(ctx, files)
+            except Exception as e:
+                is_export_pass = True
+                export_msg = f"{e}"
+
             elapsed_time_ms = (time.perf_counter() - before) * 1000
-            if not is_pass:
+            if not is_export_pass:
                 if file_retry_queue:
                     for file in files:
                         file_retry_queue.enqueue(file, len(file.data))
-                    err_msg = f'{msg}, retry later'
+                    event_err_msg = f'{export_msg}, retry later'
                 else:
-                    err_msg = f'{msg}, retry second time failed'
-                is_fail = True
+                    event_err_msg = f'{export_msg}, retry second time failed'
             if finish_event_processor:
                 finish_event_processor(FinishEventInfo(
                     event_type="exporter.file_flush.rate",
-                    is_event_fail=is_fail,
+                    is_event_fail=not is_export_pass,
                     item_num=len(files),
-                    detail_msg=err_msg,
+                    detail_msg=event_err_msg,
                     extra_params=FinishEventInfoExtra(
                         latency_ms=int(elapsed_time_ms)
                     )