[Core] Integrations monitoring data structure (#1624)

# Description

What - 
1. updated the current event we send with the metrics today.
2. updated the places we send metrics in

example json:
```
{
   {
   "integration_type":"gitlab",
   "integration_identifier":"gitlab-local",
   "integration_version":"0.2.62",
   "ocean_version":"0.22.5",
   "kind_identifier":"group-with-members-0",
   "kind":"group-with-members",
   "event_id":"bf1926ab-ba16-468c-850a-0380530a06a8",
   "sync_state":"data ingested",
   "_id":"integration_RFSJKG22HJKGDF",
   "metrics":{
      "phase":{
         "resync":{
            "duration_seconds":8.380515999999943,
            "success":1.0
         },
         "load":{
            "object_count_type":{
               "skipped":{
                  "object_count":0.0
               },
               "failed":{
                  "object_count":2.0
               },
               "loaded":{
                  "object_count":0.0
               }
            }
         },
         "extract":{
            "object_count_type":{
               "raw_extracted":{
                  "object_count":2.0
               }
            }
         },
         "transform":{
            "object_count_type":{
               "transformed":{
                  "object_count":2.0
               },
               "filtered_out":{
                  "object_count":0.0
               },
               "failed":{
                  "object_count":0.0
               }
            }
         }
      }
   }
}
```


Why - in order to support showing integration visibility in the UI of
port


## Type of change

Please leave one option from the following and delete the rest:

- [x] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] New Integration (non-breaking change which adds a new integration)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Non-breaking change (fix of existing functionality that will not
change current behavior)
- [ ] Documentation (added/updated documentation)

<h4> All tests should be run against the port production
environment(using a testing org). </h4>

### Core testing checklist

- [ ] Integration able to create all default resources from scratch
- [ ] Resync finishes successfully
- [ ] Resync able to create entities
- [ ] Resync able to update entities
- [ ] Resync able to detect and delete entities
- [ ] Scheduled resync able to abort existing resync and start a new one
- [ ] Tested with at least 2 integrations from scratch
- [ ] Tested with Kafka and Polling event listeners
- [ ] Tested deletion of entities that don't pass the selector

Author: yaelibarg

CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 <!-- towncrier release notes start -->
+## 0.22.9 (2025-05-18)
+
+### Improvements
+- Enhanced Ocean metrics event structure for better data organization and analysis
+- Expanded metrics collection points to provide more comprehensive monitoring capabilities
+
 ## 0.22.8 (2025-05-15)
 
 ### Improvements

port_ocean/clients/port/mixins/entities.py

@@ -5,7 +5,7 @@
 
 import httpx
 from loguru import logger
-
+from port_ocean.context.ocean import ocean
 from port_ocean.clients.port.authentication import PortAuthentication
 from port_ocean.clients.port.types import RequestOptions, UserAgentType
 from port_ocean.clients.port.utils import (
@@ -15,6 +15,8 @@
 from port_ocean.core.models import Entity, PortAPIErrorMessage
 from starlette import status
 
+from port_ocean.helpers.metric.metric import MetricPhase, MetricType
+
 
 class EntityClientMixin:
     def __init__(self, auth: PortAuthentication, client: httpx.AsyncClient):
@@ -81,6 +83,15 @@ async def upsert_entity(
                 f"blueprint: {entity.blueprint}"
             )
             result = response.json()
+            ocean.metrics.inc_metric(
+                name=MetricType.OBJECT_COUNT_NAME,
+                labels=[
+                    ocean.metrics.current_resource_kind(),
+                    MetricPhase.LOAD,
+                    MetricPhase.LoadResult.FAILED,
+                ],
+                value=1,
+            )
 
             if (
                 response.status_code == status.HTTP_404_NOT_FOUND
@@ -89,6 +100,17 @@ async def upsert_entity(
             ):
                 # Return false to differentiate from `result_entity.is_using_search_identifier`
                 return False
+        else:
+            ocean.metrics.inc_metric(
+                name=MetricType.OBJECT_COUNT_NAME,
+                labels=[
+                    ocean.metrics.current_resource_kind(),
+                    MetricPhase.LOAD,
+                    MetricPhase.LoadResult.LOADED,
+                ],
+                value=1,
+            )
+
         handle_port_status_code(response, should_raise)
         result = response.json()
 

port_ocean/core/handlers/entities_state_applier/port/applier.py

@@ -103,7 +103,7 @@ async def delete_diff(
             and deletion_rate <= entity_deletion_threshold
         ):
             await self._safe_delete(diff.deleted, kept_entities, user_agent_type)
-            ocean.metrics.set_metric(
+            ocean.metrics.inc_metric(
                 name=MetricType.DELETION_COUNT_NAME,
                 labels=[ocean.metrics.current_resource_kind(), MetricPhase.DELETE],
                 value=len(diff.deleted),

port_ocean/core/integrations/mixins/sync_raw.py

@@ -31,7 +31,7 @@
 )
 from port_ocean.core.utils.utils import resolve_entities_diff, zip_and_sum, gather_and_split_errors_from_results
 from port_ocean.exceptions.core import OceanAbortException
-from port_ocean.helpers.metric.metric import MetricType, MetricPhase
+from port_ocean.helpers.metric.metric import SyncState, MetricType, MetricPhase
 from port_ocean.helpers.metric.utils import TimeMetric
 
 SEND_RAW_DATA_EXAMPLES_AMOUNT = 5
@@ -238,11 +238,21 @@ async def _register_resource_raw(
                 if changed_entities:
                     logger.info("Upserting changed entities", changed_entities=len(changed_entities),
                         total_entities=len(objects_diff[0].entity_selector_diff.passed))
+                    ocean.metrics.inc_metric(
+                            name=MetricType.OBJECT_COUNT_NAME,
+                            labels=[ocean.metrics.current_resource_kind(), MetricPhase.LOAD, MetricPhase.LoadResult.SKIPPED],
+                            value=len(objects_diff[0].entity_selector_diff.passed) - len(changed_entities)
+                        )
                     await self.entities_state_applier.upsert(
                         changed_entities, user_agent_type
                     )
                 else:
                     logger.info("Entities in batch didn't changed since last sync, skipping", total_entities=len(objects_diff[0].entity_selector_diff.passed))
+                    ocean.metrics.inc_metric(
+                            name=MetricType.OBJECT_COUNT_NAME,
+                            labels=[ocean.metrics.current_resource_kind(), MetricPhase.LOAD, MetricPhase.LoadResult.SKIPPED],
+                            value=len(objects_diff[0].entity_selector_diff.passed)
+                        )
                 modified_objects = [ocean.port_client._reduce_entity(entity) for entity in objects_diff[0].entity_selector_diff.passed]
             except Exception as e:
                 logger.warning(f"Failed to resolve batch entities with Port, falling back to upserting all entities: {str(e)}")
@@ -335,6 +345,7 @@ async def _register_in_batches(
                     passed_entities.extend(calculation_result.entity_selector_diff.passed)
                     number_of_transformed_entities += calculation_result.number_of_transformed_entities
             except* OceanAbortException as error:
+                ocean.metrics.sync_state = SyncState.FAILED
                 errors.append(error)
 
         logger.info(
@@ -347,18 +358,30 @@ async def _register_in_batches(
             value=int(not errors)
         )
 
-        ocean.metrics.set_metric(
+        ocean.metrics.inc_metric(
             name=MetricType.OBJECT_COUNT_NAME,
-            labels=[ocean.metrics.current_resource_kind(), MetricPhase.EXTRACT],
+            labels=[ocean.metrics.current_resource_kind(), MetricPhase.EXTRACT , MetricPhase.ExtractResult.EXTRACTED],
             value=number_of_raw_results
         )
 
-        ocean.metrics.set_metric(
+        ocean.metrics.inc_metric(
             name=MetricType.OBJECT_COUNT_NAME,
-            labels=[ocean.metrics.current_resource_kind(), MetricPhase.TRANSFORM],
+            labels=[ocean.metrics.current_resource_kind(), MetricPhase.TRANSFORM , MetricPhase.TransformResult.TRANSFORMED],
             value=number_of_transformed_entities
         )
 
+        ocean.metrics.inc_metric(
+            name=MetricType.OBJECT_COUNT_NAME,
+            labels=[ocean.metrics.current_resource_kind(), MetricPhase.TRANSFORM , MetricPhase.TransformResult.FILTERED_OUT],
+            value=number_of_raw_results -number_of_transformed_entities
+        )
+
+        ocean.metrics.inc_metric(
+            name=MetricType.OBJECT_COUNT_NAME,
+            labels=[ocean.metrics.current_resource_kind(), MetricPhase.TRANSFORM , MetricPhase.TransformResult.FAILED],
+            value=len(errors)
+        )
+
         return passed_entities, errors
 
     async def register_raw(
@@ -558,7 +581,6 @@ async def sort_and_upsert_failed_entities(self,user_agent_type: UserAgentType)->
                 for entity in event.entity_topological_sorter.get_entities(False):
                     await self.entities_state_applier.context.port_client.upsert_entity(entity,event.port_app_config.get_port_request_options(),user_agent_type,should_raise=False)
 
-
     @TimeMetric(MetricPhase.RESYNC)
     async def sync_raw_all(
         self,
@@ -584,12 +606,16 @@ async def sync_raw_all(
             EventType.RESYNC,
             trigger_type=trigger_type,
         ):
+            ocean.metrics.event_id = event.id
+
             # If a resync is triggered due to a mappings change, we want to make sure that we have the updated version
             # rather than the old cache
             app_config = await self.port_app_config_handler.get_port_app_config(
                 use_cache=False
             )
             logger.info(f"Resync will use the following mappings: {app_config.dict()}")
+            ocean.metrics.initialize_metrics([f"{resource.kind}-{index}" for index, resource in enumerate(app_config.resources)])
+            await ocean.metrics.flush()
 
             # Execute resync_start hooks
             for resync_start_fn in self.event_strategy["resync_start"]:
@@ -616,20 +642,19 @@ async def sync_raw_all(
                     # config as we might have multiple resources in the same event
                     async with resource_context(resource,index):
                         resource_kind_id = f"{resource.kind}-{index}"
+                        ocean.metrics.sync_state = SyncState.SYNCING
+                        await ocean.metrics.flush(kind=resource_kind_id)
+
                         task = asyncio.create_task(
                             self._register_in_batches(resource, user_agent_type)
                         )
 
                         event.on_abort(lambda: task.cancel())
                         kind_results: tuple[list[Entity], list[Exception]] = await task
-                        ocean.metrics.set_metric(
-                            name=MetricType.OBJECT_COUNT_NAME,
-                            labels=[ocean.metrics.current_resource_kind(), MetricPhase.LOAD],
-                            value=len(kind_results[0])
-                        )
 
                         creation_results.append(kind_results)
-
+                        if ocean.metrics.sync_state != SyncState.FAILED:
+                            ocean.metrics.sync_state = SyncState.COMPLETED
                         await ocean.metrics.flush(kind=resource_kind_id)
 
                 await self.sort_and_upsert_failed_entities(user_agent_type)

port_ocean/helpers/metric/metric.py

@@ -22,6 +22,19 @@ class MetricPhase:
     RESYNC = "resync"
     DELETE = "delete"
 
+    class TransformResult:
+        TRANSFORMED = "transformed"
+        FILTERED_OUT = "filtered_out"
+        FAILED = "failed"
+
+    class LoadResult:
+        LOADED = "loaded"
+        FAILED = "failed"
+        SKIPPED = "skipped"
+
+    class ExtractResult:
+        EXTRACTED = "raw_extracted"
+
 
 class MetricType:
     # Define metric names as constants
@@ -33,6 +46,13 @@ class MetricType:
     DELETION_COUNT_NAME = "deletion_count"
 
 
+class SyncState:
+    SYNCING = "syncing"
+    COMPLETED = "completed"
+    QUEUED = "queued"
+    FAILED = "failed"
+
+
 # Registry for core and custom metrics
 _metrics_registry: Dict[str, Tuple[str, str, List[str]]] = {
     MetricType.DURATION_NAME: (
@@ -43,7 +63,7 @@ class MetricType:
     MetricType.OBJECT_COUNT_NAME: (
         MetricType.OBJECT_COUNT_NAME,
         "object_count description",
-        ["kind", "phase"],
+        ["kind", "phase", "object_count_type"],
     ),
     MetricType.ERROR_COUNT_NAME: (
         MetricType.ERROR_COUNT_NAME,
@@ -86,6 +106,9 @@ def set(self, *args: Any) -> None:
     def labels(self, *args: Any) -> None:
         return None
 
+    def inc(self, *args: Any) -> None:
+        return None
+
 
 class Metrics:
     def __init__(
@@ -100,6 +123,24 @@ def __init__(
         self.load_metrics()
         self._integration_version: Optional[str] = None
         self._ocean_version: Optional[str] = None
+        self.event_id = ""
+        self.sync_state = SyncState.QUEUED
+
+    @property
+    def event_id(self) -> str:
+        return self._event_id
+
+    @event_id.setter
+    def event_id(self, value: str) -> None:
+        self._event_id = value
+
+    @property
+    def sync_state(self) -> str:
+        return self._sync_state
+
+    @sync_state.setter
+    def sync_state(self, value: str) -> None:
+        self._sync_state = value
 
     @property
     def integration_version(self) -> str:
@@ -139,6 +180,19 @@ def get_metric(self, name: str, labels: list[str]) -> Gauge | EmptyMetric:
             return EmptyMetric()
         return metrics.labels(*labels)
 
+    def inc_metric(self, name: str, labels: list[str], value: float) -> None:
+        """Increment a metric value in a single method call.
+
+        Args:
+            name (str): The metric name to inc.
+            labels (list[str]): The labels to apply to the metric.
+            value (float): The value to inc.
+        """
+        if not self.enabled:
+            return None
+
+        self.get_metric(name, labels).inc(value)
+
     def set_metric(self, name: str, labels: list[str], value: float) -> None:
         """Set a metric value in a single method call.
 
@@ -152,6 +206,49 @@ def set_metric(self, name: str, labels: list[str], value: float) -> None:
 
         self.get_metric(name, labels).set(value)
 
+    def initialize_metrics(self, kind_blockes: list[str]) -> None:
+        for kind in kind_blockes:
+            self.set_metric(MetricType.SUCCESS_NAME, [kind, MetricPhase.RESYNC], 0)
+            self.set_metric(MetricType.DURATION_NAME, [kind, MetricPhase.RESYNC], 0)
+
+            self.set_metric(
+                MetricType.OBJECT_COUNT_NAME,
+                [kind, MetricPhase.EXTRACT, MetricPhase.ExtractResult.EXTRACTED],
+                0,
+            )
+
+            self.set_metric(
+                MetricType.OBJECT_COUNT_NAME,
+                [kind, MetricPhase.TRANSFORM, MetricPhase.TransformResult.TRANSFORMED],
+                0,
+            )
+            self.set_metric(
+                MetricType.OBJECT_COUNT_NAME,
+                [kind, MetricPhase.TRANSFORM, MetricPhase.TransformResult.FILTERED_OUT],
+                0,
+            )
+            self.set_metric(
+                MetricType.OBJECT_COUNT_NAME,
+                [kind, MetricPhase.TRANSFORM, MetricPhase.TransformResult.FAILED],
+                0,
+            )
+
+            self.set_metric(
+                MetricType.OBJECT_COUNT_NAME,
+                [kind, MetricPhase.LOAD, MetricPhase.LoadResult.LOADED],
+                0,
+            )
+            self.set_metric(
+                MetricType.OBJECT_COUNT_NAME,
+                [kind, MetricPhase.LOAD, MetricPhase.LoadResult.FAILED],
+                0,
+            )
+            self.set_metric(
+                MetricType.OBJECT_COUNT_NAME,
+                [kind, MetricPhase.LOAD, MetricPhase.LoadResult.SKIPPED],
+                0,
+            )
+
     def create_mertic_router(self) -> APIRouter:
         if not self.enabled:
             return APIRouter()
@@ -201,14 +298,21 @@ async def flush(
                         if kind and sample.labels.get("kind") != kind:
                             continue
 
-                        # Create nested dictionary structure based on labels
-                        for key, value in sample.labels.items():
-                            if key not in current_level:
-                                current_level[key] = {}
-                            current_level = current_level[key]
-                            if value not in current_level:
-                                current_level[value] = {}
-                            current_level = current_level[value]
+                        # Get the ordered labels from the registry
+                        ordered_labels = _metrics_registry.get(
+                            sample.name, (None, None, [])
+                        )[2]
+
+                        # Create nested dictionary structure based on ordered labels
+                        for label_name in ordered_labels:
+                            if label_name in sample.labels:
+                                value = sample.labels[label_name]
+                                if label_name not in current_level:
+                                    current_level[label_name] = {}
+                                current_level = current_level[label_name]
+                                if value not in current_level:
+                                    current_level[value] = {}
+                                current_level = current_level[value]
 
                     current_level[sample.name] = sample.value
 
@@ -228,6 +332,8 @@ async def flush(
                     "ocean_version": self.ocean_version,
                     "kind_identifier": kind_key,
                     "kind": "-".join(kind_key.split("-")[:-1]),
+                    "event_id": self.event_id,
+                    "sync_state": self.sync_state,
                     "metrics": metrics,
                 }
                 logger.info(f"Sending metrics to webhook {kind_key}: {event}")

port_ocean/helpers/metric/utils.py

@@ -17,7 +17,7 @@ async def wrapper(*args: Any, **kwargs: dict[Any, Any]) -> Any:
             res = await func(*args, **kwargs)
             end = time.monotonic()
             duration = end - start
-            ocean.metrics.set_metric(
+            ocean.metrics.inc_metric(
                 name=MetricType.DURATION_NAME,
                 labels=[ocean.metrics.current_resource_kind(), phase],
                 value=duration,