[Core] Add metrics for ocean core #1260

ivankalinovski · 2024-12-23T13:23:14Z

Decorator that can be used to "metricify" a function - accepts phase as input.
Auto record duration.
increment_field method for handling counters.
increment_status method for handling requests.
populate event with MetricAggregator to track metrics.

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)

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

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

Integration testing checklist

Integration able to create all default resources from scratch
Resync able to create entities
Resync able to update entities
Resync able to detect and delete entities
Resync finishes successfully
If new resource kind is added or updated in the integration, add example raw data, mapping and expected result to the examples folder in the integration directory.
If resource kind is updated, run the integration with the example data and check if the expected result is achieved
If new resource kind is added or updated, validate that live-events for that resource are working as expected
Docs PR link here

Preflight checklist

Handled rate limiting
Handled pagination
Implemented the code in async
Support Multi account

Screenshots

Include screenshots from your environment showing how the resources of the integration will look.

API Documentation

Provide links to the API documentation used for this integration.

1. Add classes to encapsulate metrics data. 2. Add class to encapsulate metric modification. 3. Add Metric event into `event.py`. 4. Add metric helper decorator to provide context with phase and measure time. 5. Add metric wrapper for async generators to measure time. 6. Add increment metric calls to measure stats. 7. Add tests to validate how many entities were extraced and processed. 8. Add test to compare performance of two branches.

Refactor increment methods.

Set metrics as feature flag.

Remove perf tests results

1. Fix `validate_metrics`. 2. Pass Metric env to tests.

Add metrics to topological sorter

Tankilevitch · 2025-01-04T12:05:53Z

port_ocean/clients/port/mixins/entities.py

@@ -72,6 +75,8 @@ async def upsert_entity(
                extensions={"retryable": True},
            )
        if response.is_error:
+            await event.event.increment_metric(MetricFieldType.ERROR_COUNT)


Suggested change

await event.event.increment_metric(MetricFieldType.ERROR_COUNT)

await event.event.increment_metric(MetricFieldType.PORT_API_ERROR_COUNT)

Tankilevitch · 2025-01-04T12:08:03Z

port_ocean/clients/port/mixins/entities.py


        return reduced_entity

+    @metric(MetricType.LOAD)


shouldn't it be on top of the upsert_entity method rather than batch_upsert_entities ?

Tankilevitch · 2025-01-04T12:08:36Z

port_ocean/clients/port/mixins/entities.py

@@ -167,6 +178,7 @@ async def delete_entity(
            )

            if response.is_error:
+                await event.event.increment_metric(MetricFieldType.ERROR_COUNT)


should this kind of error count be in the upsert_entity as well?

Tankilevitch · 2025-01-04T12:20:17Z

port_ocean/config/settings.py

@@ -79,6 +79,7 @@ class IntegrationConfiguration(BaseOceanSettings, extra=Extra.allow):
    )
    runtime: Runtime = Runtime.OnPrem
    resources_path: str = Field(default=".port/resources")
+    metrics: bool = Field(default=False)


I would suggest to have the type as some kind of enum for cases where it won't be only true or false but rather might want to only monitor extract metrics

Tankilevitch · 2025-01-04T19:41:24Z

port_ocean/helpers/metric.py

+                break
+
+
+def metric(phase: str | None = None, should_capture_time: bool = True) -> Any:


Tankilevitch · 2025-01-04T19:47:58Z

port_ocean/tests/test_metric.py

+    assert metrics, "No 'metrics' key found in the parsed JSON."
+
+    assert "fake-person" in metrics, "'fake-person' key missing in metrics data."
+    fake_person = metrics["fake-person"]


what fake-person mean in a metric? I understand that its contains the extract, load transform but I wonder how we can monitor that?

Tankilevitch · 2025-01-04T19:49:39Z

port_ocean/context/event.py

@@ -163,6 +203,7 @@ def _handle_event(triggering_event_id: int) -> None:
    dispatcher.connect(_handle_event, event_type)
    dispatcher.send(event_type, triggering_event_id=event.id)

+    is_silent = EventType.METRIC == event_type


why this is needed for?

Tankilevitch · 2025-01-04T19:52:06Z

port_ocean/helpers/metric.py

+            except Exception:
+                break


why break? how the exception is being propogated? its not equivalent to the way you are iterating over when if not port_ocean.context.event.event.should_record_metrics

Tankilevitch · 2025-01-04T19:53:18Z

port_ocean/helpers/metric.py

+    async def increment_status(self, status_code: str) -> None:
+        metric = await self.get_metric()
+        if metric is None or not isinstance(metric, ApiStats):
+            return None
+        async with self._lock:
+
+            status = metric.requests.get(status_code)
+            if not status:
+                metric.requests[status_code] = 0
+            metric.requests[status_code] = metric.requests.get(status_code, 0) + 1


I wonder if this is actually needed 🤔 its sounds too specific, you already implemented increment_field

Tankilevitch · 2025-01-04T19:53:45Z

port_ocean/helpers/metric.py

+    async def flush(self) -> None:
+        async with self._lock:
+            metric_dict = asdict(self.metrics)
+            logger.info(f"integration metrics {json.dumps(metric_dict)}")


please share in the pr description the flush output

convert metrics to use prometheus

ivankalinovski requested a review from a team as a code owner December 23, 2024 13:23

github-actions bot added the size/L label Dec 23, 2024

Ivan Kalinovski added 3 commits December 30, 2024 01:42

[Core] Add metrics to monitor integration performance

a1df9b8

Refactor increment methods.

[Core] Add metrics to monitor integration performance

daf9cf0

Set metrics as feature flag.

ivankalinovski force-pushed the PORT-11681-improve-ocean-logs-for-our-metrics-dashboard branch from 69bb09b to daf9cf0 Compare December 30, 2024 12:59

Ivan Kalinovski added 3 commits December 30, 2024 15:01

[Core] Add metrics to monitor integration performance

9a46af8

Remove perf tests results

[Core] Add metrics to monitor integration performance

1d79694

1. Fix `validate_metrics`. 2. Pass Metric env to tests.

[Core] Add metrics to monitor integration performance

9d9e6ab

Add metrics to topological sorter

Tankilevitch requested changes Jan 4, 2025

View reviewed changes

Ivan Kalinovski added 2 commits January 9, 2025 12:00

[Core] Add metrics with prometheus

2577d9a

convert metrics to use prometheus

Update tests and add mode to ocean client

075a6ca

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Core] Add metrics for ocean core #1260

[Core] Add metrics for ocean core #1260

ivankalinovski commented Dec 23, 2024 •

edited

Loading

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

Tankilevitch Jan 4, 2025

	await event.event.increment_metric(MetricFieldType.ERROR_COUNT)
	await event.event.increment_metric(MetricFieldType.PORT_API_ERROR_COUNT)

		break


		def metric(phase: str \| None = None, should_capture_time: bool = True) -> Any:

[Core] Add metrics for ocean core #1260

Are you sure you want to change the base?

[Core] Add metrics for ocean core #1260

Conversation

ivankalinovski commented Dec 23, 2024 • edited Loading

Type of change

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

Core testing checklist

Integration testing checklist

Preflight checklist

Screenshots

API Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ivankalinovski commented Dec 23, 2024 •

edited

Loading