[Core] Enable metrics delivery to Port (#1686)

# Description

What - removed comments from the end points of the metrics and fixed
url's to the metrics endpoints

Why - feature is ready to use

How - deleted the comments

## Type of change

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

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] 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,10 @@ 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.23.4 (2025-05-29)
+
+### Improvements
+- Fixed metrics urls and added reconciliation kind to report
 
 ## 0.23.3 (2025-05-28)
 
@@ -25,7 +29,6 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ## 0.23.1 (2025-05-27)
 
 ### Bug Fixes
-
 - Event loop is blocked by waiting for a process.
 
 ## 0.23.0 (2025-05-27)

port_ocean/clients/port/mixins/integrations.py

@@ -214,8 +214,9 @@ async def post_integration_sync_metrics(
         logger.debug("starting POST metrics request", metrics=metrics)
         metrics_attributes = await self.get_metrics_attributes()
         headers = await self.auth.headers()
+        url = metrics_attributes["ingestUrl"] + "/syncMetrics"
         response = await self.client.post(
-            metrics_attributes["ingestUrl"],
+            url,
             headers=headers,
             json={
                 "syncKindsMetrics": metrics,
@@ -229,7 +230,7 @@ async def put_integration_sync_metrics(self, kind_metrics: dict[str, Any]) -> No
         metrics_attributes = await self.get_metrics_attributes()
         url = (
             metrics_attributes["ingestUrl"]
-            + f"/resync/{kind_metrics['eventId']}/kind/{kind_metrics['kindIdentifier']}"
+            + f"/syncMetrics/resync/{kind_metrics['eventId']}/kind/{kind_metrics['kindIdentifier']}"
         )
         headers = await self.auth.headers()
         response = await self.client.put(

port_ocean/context/metric_resource.py

@@ -0,0 +1,63 @@
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from typing import AsyncIterator, TYPE_CHECKING
+
+from loguru import logger
+from werkzeug.local import LocalStack, LocalProxy
+
+from port_ocean.exceptions.context import (
+    ResourceContextNotFoundError,
+)
+
+if TYPE_CHECKING:
+    pass
+
+
+@dataclass
+class MetricResourceContext:
+    """
+    The metric resource context is a context manager that allows you to access the current metric resource if there is one.
+    This is useful for getting the metric resource kind
+    """
+
+    metric_resource_kind: str
+    index: int
+
+    @property
+    def kind(self) -> str:
+        return self.metric_resource_kind
+
+
+_resource_context_stack: LocalStack[MetricResourceContext] = LocalStack()
+
+
+def _get_metric_resource_context() -> MetricResourceContext:
+    """
+    Get the context from the current thread.
+    """
+    top_resource_context = _resource_context_stack.top
+    if top_resource_context is None:
+        raise ResourceContextNotFoundError(
+            "You must be inside an metric resource context in order to use it"
+        )
+
+    return top_resource_context
+
+
+metric_resource: MetricResourceContext = LocalProxy(lambda: _get_metric_resource_context())  # type: ignore
+
+
+@asynccontextmanager
+async def metric_resource_context(
+    metric_resource_kind: str, index: int = 0
+) -> AsyncIterator[MetricResourceContext]:
+    _resource_context_stack.push(
+        MetricResourceContext(metric_resource_kind=metric_resource_kind, index=index)
+    )
+
+    with logger.contextualize(
+        metric_resource_kind=metric_resource.metric_resource_kind
+    ):
+        yield metric_resource
+
+    _resource_context_stack.pop()

port_ocean/core/handlers/resync_state_updater/updater.py

@@ -94,6 +94,6 @@ async def update_after_resync(
         await ocean.metrics.send_metrics_to_webhook(
             kind=ocean.metrics.current_resource_kind()
         )
-        # await ocean.metrics.report_sync_metrics(
-        #     kinds=[ocean.metrics.current_resource_kind()]
-        # ) # TODO: uncomment this when end points are ready
+        await ocean.metrics.report_sync_metrics(
+            kinds=[ocean.metrics.current_resource_kind()]
+        )

port_ocean/core/integrations/mixins/sync_raw.py

@@ -9,6 +9,7 @@
 from loguru import logger
 from port_ocean.clients.port.types import UserAgentType
 from port_ocean.context.event import TriggerType, event_context, EventType, event
+from port_ocean.context.metric_resource import metric_resource_context
 from port_ocean.context.ocean import ocean
 from port_ocean.context.resource import resource_context
 from port_ocean.context import resource
@@ -33,8 +34,8 @@
 )
 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 SyncState, MetricType, MetricPhase
-from port_ocean.helpers.metric.utils import TimeMetric
+from port_ocean.helpers.metric.metric import MetricResourceKind, SyncState, MetricType, MetricPhase
+from port_ocean.helpers.metric.utils import TimeMetric, TimeMetricWithResourceKind
 from port_ocean.utils.ipc import FileIPC
 
 SEND_RAW_DATA_EXAMPLES_AMOUNT = 5
@@ -249,9 +250,16 @@ async def _register_resource_raw(
                             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(
+                    upserted_entities = await self.entities_state_applier.upsert(
                         changed_entities, user_agent_type
                     )
+
+                    ocean.metrics.set_metric(
+                        name=MetricType.OBJECT_COUNT_NAME,
+                        labels=[ocean.metrics.current_resource_kind(), MetricPhase.LOAD, MetricPhase.LoadResult.LOADED],
+                        value=len(upserted_entities)
+                    )
+
                 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(
@@ -265,6 +273,11 @@ async def _register_resource_raw(
                 modified_objects = await self.entities_state_applier.upsert(
                     objects_diff[0].entity_selector_diff.passed, user_agent_type
                     )
+                ocean.metrics.set_metric(
+                        name=MetricType.OBJECT_COUNT_NAME,
+                        labels=[ocean.metrics.current_resource_kind(), MetricPhase.LOAD, MetricPhase.LoadResult.LOADED],
+                        value=len(upserted_entities)
+                    )
         else:
            modified_objects = await self.entities_state_applier.upsert(
                     objects_diff[0].entity_selector_diff.passed, user_agent_type
@@ -633,27 +646,101 @@ async def _process_resource(self,resource: ResourceConfig, index: int, user_agen
         async with resource_context(resource,index):
             resource_kind_id = f"{resource.kind}-{index}"
             ocean.metrics.sync_state = SyncState.SYNCING
+
             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, MetricPhase.LoadResult.LOADED],
-                value=len(kind_results[0])
-            )
 
             if ocean.metrics.sync_state != SyncState.FAILED:
                 ocean.metrics.sync_state = SyncState.COMPLETED
 
             await ocean.metrics.send_metrics_to_webhook(
                 kind=resource_kind_id
             )
-            # await ocean.metrics.report_kind_sync_metrics(kind=resource_kind_id) # TODO: uncomment this when end points are ready
+            await ocean.metrics.report_kind_sync_metrics(kind=resource_kind_id, blueprint=resource.port.entity.mappings.blueprint)
 
             return kind_results
 
+    @TimeMetricWithResourceKind(MetricPhase.RESYNC)
+    async def resync_reconciliation(
+        self,
+        creation_results: list[tuple[list[Entity], list[Exception]]],
+        did_fetched_current_state: bool,
+        user_agent_type: UserAgentType,
+        app_config: Any,
+        silent: bool = True,
+    ) -> None:
+        """Handle the reconciliation phase of the resync process.
+
+        This method handles:
+        1. Sorting and upserting failed entities
+        2. Checking if current state was fetched
+        3. Calculating resync diff
+        4. Handling errors
+        5. Deleting entities that are no longer needed
+        6. Executing resync complete hooks
+
+        Args:
+            creation_results (list[tuple[list[Entity], list[Exception]]]): Results from entity creation
+            did_fetched_current_state (bool): Whether the current state was successfully fetched
+            user_agent_type (UserAgentType): The type of user agent
+            app_config (Any): The application configuration
+            silent (bool): Whether to raise exceptions or handle them silently
+
+        """
+        await self.sort_and_upsert_failed_entities(user_agent_type)
+
+        if not did_fetched_current_state:
+            logger.warning(
+                "Due to an error before the resync, the previous state of entities at Port is unknown."
+                " Skipping delete phase due to unknown initial state."
+            )
+            return False
+
+        logger.info("Starting resync diff calculation")
+        generated_entities, errors = zip_and_sum(creation_results) or [
+            [],
+            [],
+        ]
+
+        if errors:
+            message = f"Resync failed with {len(errors)} errors, skipping delete phase due to incomplete state"
+            error_group = ExceptionGroup(
+                message,
+                errors,
+            )
+            if not silent:
+                raise error_group
+
+            logger.error(message, exc_info=error_group)
+            return False
+
+        logger.info(
+            f"Running resync diff calculation, number of entities created during sync: {len(generated_entities)}"
+        )
+        entities_at_port = await ocean.port_client.search_entities(
+            user_agent_type
+        )
+
+        await self.entities_state_applier.delete_diff(
+            {"before": entities_at_port, "after": generated_entities},
+            user_agent_type, app_config.get_entity_deletion_threshold()
+        )
+
+        logger.info("Resync finished successfully")
+
+        # Execute resync_complete hooks
+        if "resync_complete" in self.event_strategy:
+            logger.info("Executing resync_complete hooks")
+
+            for resync_complete_fn in self.event_strategy["resync_complete"]:
+                await resync_complete_fn()
+
+            logger.info("Finished executing resync_complete hooks")
+
+
     @TimeMetric(MetricPhase.RESYNC)
     async def sync_raw_all(
         self,
@@ -689,8 +776,9 @@ async def sync_raw_all(
             logger.info(f"Resync will use the following mappings: {app_config.dict()}")
 
             kinds = [f"{resource.kind}-{index}" for index, resource in enumerate(app_config.resources)]
+            blueprints = [resource.port.entity.mappings.blueprint for resource in app_config.resources]
             ocean.metrics.initialize_metrics(kinds)
-            # await ocean.metrics.report_sync_metrics(kinds=kinds) # TODO: uncomment this when end points are ready
+            await ocean.metrics.report_sync_metrics(kinds=kinds, blueprints=blueprints)
 
             # Clear cache
             await ocean.app.cache_provider.clear()
@@ -716,65 +804,20 @@ async def sync_raw_all(
             multiprocessing.set_start_method('fork', True)
             try:
                 for index,resource in enumerate(app_config.resources):
-
                     logger.info(f"Starting processing resource {resource.kind} with index {index}")
-
                     creation_results.append(await self.process_resource(resource,index,user_agent_type))
-
-                await self.sort_and_upsert_failed_entities(user_agent_type)
-
             except asyncio.CancelledError as e:
                 logger.warning("Resync aborted successfully, skipping delete phase. This leads to an incomplete state")
                 raise
             else:
-                if not did_fetched_current_state:
-                    logger.warning(
-                        "Due to an error before the resync, the previous state of entities at Port is unknown."
-                        " Skipping delete phase due to unknown initial state."
-                    )
-                    return
-
-                logger.info("Starting resync diff calculation")
-                generated_entities, errors = zip_and_sum(creation_results) or [
-                    [],
-                    [],
-                ]
-
-                if errors:
-                    message = f"Resync failed with {len(errors)} errors, skipping delete phase due to incomplete state"
-                    error_group = ExceptionGroup(
-                        message,
-                        errors,
-                    )
-                    if not silent:
-                        raise error_group
-
-                    logger.error(message, exc_info=error_group)
-                    return False
-                else:
-                    logger.info(
-                        f"Running resync diff calculation, number of entities created during sync: {len(generated_entities)}"
-                    )
-                    entities_at_port = await ocean.port_client.search_entities(
-                        user_agent_type
-                    )
-                    await self.entities_state_applier.delete_diff(
-                        {"before": entities_at_port, "after": generated_entities},
-                        user_agent_type, app_config.get_entity_deletion_threshold()
-                    )
-
-                    logger.info("Resync finished successfully")
-
-                    # Execute resync_complete hooks
-                    if "resync_complete" in self.event_strategy:
-                        logger.info("Executing resync_complete hooks")
-
-                        for resync_complete_fn in self.event_strategy["resync_complete"]:
-                            await resync_complete_fn()
-
-                        logger.info("Finished executing resync_complete hooks")
-
-                    return True
+                await self.resync_reconciliation(
+                    creation_results,
+                    did_fetched_current_state,
+                    user_agent_type,
+                    app_config,
+                    silent
+                )
+                await ocean.metrics.report_sync_metrics(kinds=[MetricResourceKind.RECONCILIATION])
             finally:
                 await ocean.app.cache_provider.clear()
                 if ocean.app.process_execution_mode == ProcessExecutionMode.multi_process: