[Core] [Logging] | Integration logs not being ingested

integrations/gitlab/CHANGELOG.md

@@ -7,6 +7,16 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 <!-- towncrier release notes start -->
 
+0.1.136 (2024-11-05)
+===================
+
+### Bug Fixes
+
+- Await logger writing exception on exit (Integration logs not being ingested)
+- Await logger thread on exit (Integration logs not being ingested)
+- Sirealize exception (Integration logs not being ingested)
+
+
 0.1.135 (2024-10-31)
 ====================
 

port_ocean/clients/port/mixins/integrations.py

@@ -3,7 +3,8 @@
 
 import httpx
 from loguru import logger
-
+import copy
+from traceback import format_exception
 from port_ocean.clients.port.authentication import PortAuthentication
 from port_ocean.clients.port.utils import handle_status_code
 from port_ocean.log.sensetive import sensitive_log_filter
@@ -99,15 +100,24 @@ async def patch_integration(
         handle_status_code(response)
         return response.json()["integration"]
 
+    def sirealize_logs(self,logs: list[dict[str, Any]]):
+        _logs = copy.deepcopy(logs)
+        for log in _logs:
+            if 'extra' in log.keys() and 'exc_info' in log['extra'].keys() and isinstance(log['extra']['exc_info'],Exception):
+                sirealized_exception = ''.join(format_exception(log['extra']['exc_info']))
+                log['extra']['exc_info'] = sirealized_exception
+        return _logs
+
     async def ingest_integration_logs(self, logs: list[dict[str, Any]]) -> None:
         logger.debug("Ingesting logs")
+        sirealized_logs=self.sirealize_logs(logs)
         log_attributes = await self.get_log_attributes()
         headers = await self.auth.headers()
         response = await self.client.post(
             log_attributes["ingestUrl"],
             headers=headers,
             json={
-                "logs": logs,
+                "logs": sirealized_logs,
             },
         )
         handle_status_code(response, should_log=False)

port_ocean/log/handlers.py

@@ -37,6 +37,7 @@ def __init__(
         self.flush_size = flush_size
         self.last_flush_time = time.time()
         self._serialized_buffer: list[dict[str, Any]] = []
+        self._thread_pool: list[threading.Thread] = []
 
     @property
     def ocean(self) -> Ocean | None:
@@ -60,6 +61,11 @@ def shouldFlush(self, record: logging.LogRecord) -> bool:
         ):
             return True
         return False
+
+    def wait_for_lingering_threads(self)->None:
+        for thread in self._thread_pool:
+            if thread.is_alive():
+                thread.join()
 
     def flush(self) -> None:
         if self.ocean is None or not self.buffer:
@@ -70,15 +76,24 @@ def _wrap_event_loop(_ocean: Ocean, logs_to_send: list[dict[str, Any]]) -> None:
             loop.run_until_complete(self.send_logs(_ocean, logs_to_send))
             loop.close()
 
+        def clear_thread_pool()->None:
+            for thread in self._thread_pool:
+                if not thread.is_alive():
+                    self._thread_pool.remove(thread)
+
         self.acquire()
         logs = list(self._serialized_buffer)
         if logs:
             self.buffer.clear()
             self._serialized_buffer.clear()
             self.last_flush_time = time.time()
-            threading.Thread(target=_wrap_event_loop, args=(self.ocean, logs)).start()
+            clear_thread_pool()
+            thread=threading.Thread(target=_wrap_event_loop, args=(self.ocean, logs))
+            thread.start()
+            self._thread_pool.append(thread)
         self.release()
 
+
     async def send_logs(
         self, _ocean: Ocean, logs_to_send: list[dict[str, Any]]
     ) -> None:

port_ocean/log/logger_setup.py

@@ -55,6 +55,7 @@ def _http_loguru_handler(level: LogLevelType) -> None:
     logger.configure(patcher=exception_deserializer)
 
     http_memory_handler = HTTPMemoryHandler()
+    signal_handler.register(http_memory_handler.wait_for_lingering_threads)
     signal_handler.register(http_memory_handler.flush)
 
     queue_listener = QueueListener(queue, http_memory_handler)

port_ocean/ocean.py

@@ -123,6 +123,7 @@ async def lifecycle(_: FastAPI) -> AsyncIterator[None]:
                 yield None
             except Exception:
                 logger.exception("Integration had a fatal error. Shutting down.")
+                logger.complete()
                 sys.exit("Server stopped")
             finally:
                 signal_handler.exit()