[Gitlab] Fix handling of real time events (#1051)

integrations/gitlab/CHANGELOG.md

@@ -7,6 +7,14 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 <!-- towncrier release notes start -->
 
+0.1.128 (2024-10-02)
+====================
+
+### Improvements
+
+- Improved real time event handling and added more verbosity on event handling
+
+
 0.1.127 (2024-10-01)
 ====================
 

integrations/gitlab/gitlab_integration/events/event_handler.py

@@ -1,4 +1,5 @@
 import asyncio
+import inspect
 from abc import abstractmethod, ABC
 from asyncio import Queue
 from collections import defaultdict
@@ -39,6 +40,11 @@ async def _start_event_processor(self) -> None:
                     f"Error notifying observers for event: {event_id}, error: {e}"
                 )
             finally:
+                logger.info(
+                    f"Processed event {event_id}",
+                    event_id=event_id,
+                    event_context=event_ctx.id,
+                )
                 self.webhook_tasks_queue.task_done()
 
     async def start_event_processor(self) -> None:
@@ -49,7 +55,10 @@ async def _notify(self, event_id: str, body: dict[str, Any]) -> None:
         pass
 
     async def notify(self, event_id: str, body: dict[str, Any]) -> None:
-        logger.debug(f"Received event: {event_id}, putting it in Queue for processing")
+        logger.debug(
+            f"Received event: {event_id}, putting it in Queue for processing",
+            event_context=current_event_context.id,
+        )
         await self.webhook_tasks_queue.put(
             (
                 deepcopy(current_event_context),
@@ -75,8 +84,16 @@ async def _notify(self, event_id: str, body: dict[str, Any]) -> None:
                 f"event: {event_id} has no matching handler. the handlers available are for events: {self._observers.keys()}"
             )
             return
-
-        await asyncio.gather(*(observer(event_id, body) for observer in observers_list))
+        for observer in observers_list:
+            if asyncio.iscoroutinefunction(observer):
+                if inspect.ismethod(observer):
+                    handler = observer.__self__.__class__.__name__
+                    logger.debug(
+                        f"Notifying observer: {handler}, for event: {event_id}",
+                        event_id=event_id,
+                        handler=handler,
+                    )
+                asyncio.create_task(observer(event_id, body))  # type: ignore
 
 
 class SystemEventHandler(BaseEventHandler):

integrations/gitlab/gitlab_integration/events/hooks/issues.py

@@ -1,5 +1,5 @@
 from typing import Any
-
+from loguru import logger
 from gitlab.v4.objects import Project
 
 from gitlab_integration.core.async_fetcher import AsyncFetcher
@@ -13,6 +13,10 @@ class Issues(ProjectHandler):
     system_events = ["issue"]
 
     async def _on_hook(self, body: dict[str, Any], gitlab_project: Project) -> None:
+        logger.debug(
+            f"Handling issue hook for project {gitlab_project.path_with_namespace}, issue_id: {body.get('object_attributes', {}).get('iid')},"
+            f" issue_title: {body.get('object_attributes', {}).get('title')}, status: {body.get('object_attributes', {}).get('state')}"
+        )
         issue = await AsyncFetcher.fetch_single(
             gitlab_project.issues.get, body["object_attributes"]["iid"]
         )

integrations/gitlab/gitlab_integration/events/hooks/jobs.py

@@ -1,5 +1,6 @@
 from typing import Any
 
+from loguru import logger
 from gitlab.v4.objects import Project
 
 from gitlab_integration.core.async_fetcher import AsyncFetcher
@@ -13,5 +14,9 @@ class Job(ProjectHandler):
     system_events = ["job"]
 
     async def _on_hook(self, body: dict[str, Any], gitlab_project: Project) -> None:
+        logger.info(
+            f"Handling job hook for project {gitlab_project.path_with_namespace}, job_id: {body.get('build_id')},"
+            f" job_name: {body.get('build_name')}, status: {body.get('build_status')}"
+        )
         job = await AsyncFetcher.fetch_single(gitlab_project.jobs.get, body["build_id"])
         await ocean.register_raw(ObjectKind.JOB, [job.asdict()])

integrations/gitlab/gitlab_integration/events/hooks/merge_request.py

@@ -1,4 +1,5 @@
 from typing import Any
+from loguru import logger
 
 from gitlab.v4.objects import Project
 
@@ -13,6 +14,10 @@ class MergeRequest(ProjectHandler):
     system_events = ["merge_request"]
 
     async def _on_hook(self, body: dict[str, Any], gitlab_project: Project) -> None:
+        logger.debug(
+            f"Handling merge request hook for project {gitlab_project.path_with_namespace}, merge_request_id: {body.get('object_attributes', {}).get('iid')},"
+            f" merge_request_title: {body.get('object_attributes', {}).get('title')}, status: {body.get('object_attributes', {}).get('state')}"
+        )
         merge_requests = await AsyncFetcher.fetch_single(
             gitlab_project.mergerequests.get,
             body["object_attributes"]["iid"],

integrations/gitlab/gitlab_integration/events/hooks/project_files.py

@@ -20,6 +20,9 @@ class ProjectFiles(ProjectHandler):
     system_events = ["push"]
 
     async def _on_hook(self, body: dict[str, Any], gitlab_project: Project) -> None:
+        logger.debug(
+            f"Handling push hook for project {gitlab_project.path_with_namespace}, ref: {body.get('ref')}, commit_id: {body.get('after')}"
+        )
         added_files = [
             added_file
             for commit in body.get("commits", [])
@@ -50,7 +53,9 @@ async def _on_hook(self, body: dict[str, Any], gitlab_project: Project) -> None:
             )
         ]
         if not matching_resource_configs:
-            logger.debug("Could not find file kind to handle the push event")
+            logger.debug(
+                f"Could not find file kind to handle the push event for project {gitlab_project.path_with_namespace}"
+            )
             return
 
         for resource_config in matching_resource_configs:

integrations/gitlab/gitlab_integration/events/hooks/push.py

@@ -26,6 +26,9 @@ class PushHook(ProjectHandler):
     system_events = ["push"]
 
     async def _on_hook(self, body: dict[str, Any], gitlab_project: Project) -> None:
+        logger.debug(
+            f"Handling push hook for project {gitlab_project.path_with_namespace}, ref: {body.get('ref')}, commit_id: {body.get('after')}"
+        )
         commit_before, commit_after, ref = (
             body.get("before"),
             body.get("after"),

integrations/gitlab/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gitlab"
-version = "0.1.127"
+version = "0.1.128"
 description = "Gitlab integration for Port using Port-Ocean Framework"
 authors = ["Yair Siman-Tov <[email protected]>"]