[Core] Add retry for failed upserts and handle circular dependencies #1241

ivankalinovski · 2024-12-15T13:51:11Z

Description

What -

Register callbacks of failed entities.
When done with upserts, try topological sort on failed entities.
On fail of retry because of topological sort - try unsorted upsert.
Update topological's sort tree creation so an entity cannot be it's own dependency.
Test upsert with dependencies, with self circular dependency and external entity dependency.

Why -
Users having circular dependency errors and entities are not upserted.

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. Register callbacks of failed entities. 2. When done with upserts, try topological sort on failed entities. 3. On fail of retry because of topological sort - try unsorted upsert. 4. Update topological's sort tree creation so an entity cannot be it's own dependency. 5. Test upsert with dependencies, with self circular dependency and external entity dependency.

Tankilevitch

nice job!! left a few comments

Tankilevitch · 2024-12-15T16:52:44Z

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

+                if upsertedEntity is False:
+                    event.register_failed_upsert_call_arguments(


maybe worth adding a log message that we will try it at the end of the sync

Tankilevitch · 2024-12-17T15:59:57Z

CHANGELOG.md

+- Register callbacks of failed entities.
+- Test upsert with dependencies, with self circular dependency and external entity dependency.
+- Update topologicals sort tree creation so an entity cannot be its own dependency.
+- When done with upserts, try topological sort on failed entities.


Suggested change

- When done with upserts, try topological sort on failed entities.

- When `createMissingRelatedEntities` is set to `false` and upserting entity failed on not existing entity, the entity will be gathered to the end of the resync and will try sorting all the failed entities through a topological sort and upsert them as well

Tankilevitch · 2024-12-17T16:00:15Z

CHANGELOG.md

+- On fail of retry because of topological sort - try unsorted upsert.
+- Register callbacks of failed entities.
+- Test upsert with dependencies, with self circular dependency and external entity dependency.
+- Update topologicals sort tree creation so an entity cannot be its own dependency.


Suggested change

- Update topologicals sort tree creation so an entity cannot be its own dependency.

- Fix topologicals sort tree creation so an entity cannot be its own dependency

Tankilevitch · 2024-12-17T16:01:27Z

CHANGELOG.md

+
+### Bug Fixes
+
+- On fail of retry because of topological sort - try unsorted upsert.


Suggested change

- On fail of retry because of topological sort - try unsorted upsert.

- When experiencing cyclic error on topological sort try unsorted upsert of the entities

Tankilevitch · 2024-12-17T16:01:37Z

CHANGELOG.md

+### Bug Fixes
+
+- On fail of retry because of topological sort - try unsorted upsert.
+- Register callbacks of failed entities.


can be removed

Tankilevitch · 2024-12-17T16:04:35Z

port_ocean/clients/port/mixins/entities.py

+            result = response.json()
+            if (
+                response.status_code == 404
+                and result.get("ok") is False


Suggested change

and result.get("ok") is False

and not result.get("ok")

port_ocean/clients/port/mixins/entities.py

port_ocean/context/event.py

Tankilevitch · 2024-12-17T16:09:34Z

port_ocean/context/event.py

+    async def handle_failed(self) -> None:
+        entity_map: dict[
+            str, Callable[[], Coroutine[Any, Any, Entity | Literal[False] | None]]
+        ] = {
+            f"{obj.identifier}-{obj.blueprint}": func
+            for obj, func in self._failed_entity_callback_list
+        }
+        entity_list: list[Entity] = [
+            obj for obj, func in self._failed_entity_callback_list
+        ]


I wonder if this is actually a method that should be in the event(and maybe the whole handling stuff), I am leaning towards having a class for managing all of this and moving the instance for it to the event instance.

e.g.

EntityTopologicalHandler: def register_entities async def sort_and_upsert(): EventContext: entity_topological_handler = EntityTopologicalHandler()

Tankilevitch · 2024-12-17T16:18:50Z

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

            ordered_created_entities = reversed(
-                entities_with_search_identifier
-                + order_by_entities_dependencies(entities_without_search_identifier)
+                entities_with_search_identifier + entities_without_search_identifier


what happens with search relation and identifier? do we want to register them as well?

1. Add robust tests. 2. Change the implementation of registering failed entities.

Tankilevitch · 2024-12-19T15:51:18Z

port_ocean/clients/port/mixins/entities.py

@@ -57,6 +57,13 @@ async def upsert_entity(
                f"entity: {entity.identifier} of "
                f"blueprint: {entity.blueprint}"
            )
+            result = response.json()
+            if (
+                response.status_code == 404


lets use const of httpstatus.code

Tankilevitch · 2024-12-19T15:53:58Z

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

@@ -127,6 +126,13 @@ async def upsert(
                )
                if upsertedEntity:
                    modified_entities.append(upsertedEntity)
+                if upsertedEntity is False:


lets add a comment why is False and not if not upsertedEntity

Tankilevitch · 2024-12-19T15:59:38Z

port_ocean/core/integrations/mixins/sync_raw.py

+                except:
+                    await event.failed_entity_handler.handle_failed_no_sort()


lets except specific error type

Tankilevitch · 2024-12-19T16:08:42Z

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

+                    event.failed_entity_handler.register_failed_upsert_call_arguments(
+                        entity,
+                        event.port_app_config.get_port_request_options(),
+                        user_agent_type,
+                        self.context.port_client.upsert_entity,


lets only pass entity to it

Tankilevitch · 2024-12-19T16:09:14Z

port_ocean/utils/failed_entity_handler.py

+
+
+@dataclass
+class FailedEntityHandler:


I think this can be called
EntityTopologicalSorter

Tankilevitch · 2024-12-19T16:13:02Z

port_ocean/core/integrations/mixins/sync_raw.py

+                try:
+                    await event.failed_entity_handler.handle_failed()
+                except:
+                    await event.failed_entity_handler.handle_failed_no_sort()


maybe something like this

Suggested change

try:

await event.failed_entity_handler.handle_failed()

except:

await event.failed_entity_handler.handle_failed_no_sort()

try:

for entity in event.failed_entity_handler.get_sorted()

await self.upsert_entity(

entity,

request_options,

user_agent_type,

should_raise=should_raise,

)

for entity in entities

),

except:

await event.failed_entity_handler.handle_failed_no_sort()

1. Extrac logic of topological sort into a class. 2. Modify tests.

Tankilevitch · 2024-12-22T12:16:12Z

port_ocean/clients/port/mixins/entities.py

+
+            if (
+                response.status_code == status.HTTP_404_NOT_FOUND
+                and not result.get("ok")
+                and result.get("error") == PortApiStatus.NOT_FOUND.value
+            ):
+                return False


lets add a comment why we return False and what it means, add it to the method description as well

Tankilevitch · 2024-12-22T12:18:16Z

port_ocean/core/models.py

@@ -27,6 +27,10 @@ def is_installation_type_compatible(self, installation_type: str) -> bool:
        ) or installation_type == self.value


+class PortApiStatus(Enum):


Suggested change

class PortApiStatus(Enum):

class PortAPIErrorMessage(Enum):

Tankilevitch · 2024-12-22T12:19:18Z

port_ocean/utils/entity_topological_sorter.py

I think this should be in the core/utils rather than in the utils exposed to the integrations clients

Tankilevitch · 2024-12-22T12:21:41Z

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

I guess we can squeeze the entity_topological_sorter together with this?

Tankilevitch · 2024-12-22T12:22:11Z

port_ocean/core/integrations/mixins/sync_raw.py

@@ -29,7 +30,7 @@
 )
 from port_ocean.core.utils import zip_and_sum, gather_and_split_errors_from_results
 from port_ocean.exceptions.core import OceanAbortException
-
+import json


Tankilevitch · 2024-12-22T12:23:54Z

port_ocean/utils/entity_topological_sorter.py

+    def get_entities(self) -> Generator[Entity, Any, None]:
+        entity_map: dict[str, Entity] = {
+            f"{entity.identifier}-{entity.blueprint}": entity
+            for entity in self.entities
+        }
+        sorted_and_mapped = order_by_entities_dependencies(self.entities)
+        for obj in sorted_and_mapped:
+            entity = entity_map.get(f"{obj.identifier}-{obj.blueprint}")
+            if entity is not None:
+                yield entity


Suggested change

def get_entities(self) -> Generator[Entity, Any, None]:

entity_map: dict[str, Entity] = {

f"{entity.identifier}-{entity.blueprint}": entity

for entity in self.entities

}

sorted_and_mapped = order_by_entities_dependencies(self.entities)

for obj in sorted_and_mapped:

entity = entity_map.get(f"{obj.identifier}-{obj.blueprint}")

if entity is not None:

yield entity

def get_entities(self, sorted: bool = True) -> Generator[Entity, Any, None]:

if not sorted:

for entity in self.entities:

yield entity

entity_map: dict[str, Entity] = {

f"{entity.identifier}-{entity.blueprint}": entity

for entity in self.entities

}

sorted_and_mapped = order_by_entities_dependencies(self.entities)

for obj in sorted_and_mapped:

entity = entity_map.get(f"{obj.identifier}-{obj.blueprint}")

if entity is not None:

yield entity

Tankilevitch · 2024-12-22T12:25:51Z

port_ocean/core/integrations/mixins/sync_raw.py

+                try:
+                    for entity in event.entity_topological_sorter.get_entities():
+                        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)
+
+                except OceanAbortException as ocean_abort:
+                    if isinstance(ocean_abort.__cause__,CycleError):
+                        for entity in event.entity_topological_sorter.entities:
+                            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)


Suggested change

try:

for entity in event.entity_topological_sorter.get_entities():

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)

except OceanAbortException as ocean_abort:

if isinstance(ocean_abort.__cause__,CycleError):

for entity in event.entity_topological_sorter.entities:

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)

try:

for entity in event.entity_topological_sorter.get_entities():

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)

except OceanAbortException as ocean_abort:

if isinstance(ocean_abort.__cause__,CycleError):

for entity in event.entity_topological_sorter.get_entities(sorted=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)

Tankilevitch · 2024-12-22T12:26:47Z

port_ocean/core/integrations/mixins/sync_raw.py

+                try:
+                    for entity in event.entity_topological_sorter.get_entities():
+                        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)


lets add a log that x entities found that failed due to non existing relations, trying to re-ingest

and also a log if not

1. Change location of files. 2. Exctract logic of handle failed into a function. 3. Update get_entities.

…ecting-argo-cd-integration

Tankilevitch · 2024-12-23T08:24:06Z

port_ocean/clients/port/mixins/entities.py

+    ) -> Entity | None | Literal[False]:
+        """
+        [Entity] will be returned on happy flow
+        [None] will be returned if entity is using search identifier
+        [False] will be returned if upsert failed because of unmet dependency
+        """


lets follow the function description format like in process_in_queue

Tankilevitch · 2024-12-23T14:03:58Z

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

@@ -115,8 +114,7 @@ async def upsert(
                    entities_without_search_identifier.append(entity)

            ordered_created_entities = reversed(


there is no ordering needed any more, lets just use entities

1. Update `upsert_entity` function description. 2. Remove unused logic from `upsert`.

…ecting-argo-cd-integration

1. Fix lint 2. change order of assert in test

1. Update `entity_topological_sorter` to be created seperatly for each class instance. 2. Remove data class from EntityTopologicalSorter. 3. Update `entity_topological_sorter` to `_entity_topological_sorter`.

Rename _entity_topological_sorter to entity_topological_sorter

1. Modify logs and pass count as param. 2. Verify `create_missing_related_entities` is false before `sort_and_upsert_failed_entities` execution.

Update version

Tankilevitch

lgtm

Tankilevitch · 2024-12-24T14:32:55Z

port_ocean/core/integrations/mixins/sync_raw.py

-
+    async def sort_and_upsert_failed_entities(self,user_agent_type: UserAgentType)->None:
+        try:
+            if not event.entity_topological_sorter.is_to_execute():


Suggested change

if not event.entity_topological_sorter.is_to_execute():

if not event.entity_topological_sorter.should_execute():

update is_to_execute to should_execute

…ecting-argo-cd-integration

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

github-actions bot added the size/M label Dec 15, 2024

ivankalinovski force-pushed the PORT-11871-bug-bug-in-cycle-dependencies-affecting-argo-cd-integration branch from e152ca7 to 6ef411a Compare December 15, 2024 14:23

ivankalinovski force-pushed the PORT-11871-bug-bug-in-cycle-dependencies-affecting-argo-cd-integration branch from 5ab5b5b to c89b550 Compare December 15, 2024 15:11

Tankilevitch requested changes Dec 17, 2024

View reviewed changes

[Core] Add retry for failed upsertes and handle circular dependencies

b7b88ff

1. Add robust tests. 2. Change the implementation of registering failed entities.

ivankalinovski force-pushed the PORT-11871-bug-bug-in-cycle-dependencies-affecting-argo-cd-integration branch from eac0300 to b7b88ff Compare December 19, 2024 14:37

github-actions bot added size/L and removed size/M labels Dec 19, 2024

Tankilevitch reviewed Dec 19, 2024

View reviewed changes

[Core] Add retry for failed upsertes and handle circular dependencies

209e64b

1. Extrac logic of topological sort into a class. 2. Modify tests.

ivankalinovski force-pushed the PORT-11871-bug-bug-in-cycle-dependencies-affecting-argo-cd-integration branch from c63bd2a to 209e64b Compare December 22, 2024 08:44

Tankilevitch requested changes Dec 22, 2024

View reviewed changes

ivankalinovski force-pushed the PORT-11871-bug-bug-in-cycle-dependencies-affecting-argo-cd-integration branch from 7041bd7 to b9cf2c3 Compare December 22, 2024 14:51

[Core] Add retry for failed upserts and handle circular dependencies

3d49fa7

1. Change location of files. 2. Exctract logic of handle failed into a function. 3. Update get_entities.

ivankalinovski force-pushed the PORT-11871-bug-bug-in-cycle-dependencies-affecting-argo-cd-integration branch from b9cf2c3 to 3d49fa7 Compare December 22, 2024 15:12

Merge branch 'main' into PORT-11871-bug-bug-in-cycle-dependencies-aff…

650d316

…ecting-argo-cd-integration

Tankilevitch requested changes Dec 23, 2024

View reviewed changes

Tankilevitch reviewed Dec 23, 2024

View reviewed changes

Ivan Kalinovski and others added 4 commits December 23, 2024 16:36

[Core] Add retry for failed upserts and handle circular dependencies

ab817e5

1. Update `upsert_entity` function description. 2. Remove unused logic from `upsert`.

Merge branch 'main' into PORT-11871-bug-bug-in-cycle-dependencies-aff…

f029a8f

…ecting-argo-cd-integration

[Core] Add retry for failed upserts and handle circular dependencies

79054b2

1. Fix lint 2. change order of assert in test

[Core] Add retry for failed upserts and handle circular dependencies

5099a80

1. Update `entity_topological_sorter` to be created seperatly for each class instance. 2. Remove data class from EntityTopologicalSorter. 3. Update `entity_topological_sorter` to `_entity_topological_sorter`.

ivankalinovski force-pushed the PORT-11871-bug-bug-in-cycle-dependencies-affecting-argo-cd-integration branch from ffe2c27 to 5099a80 Compare December 23, 2024 19:17

Ivan Kalinovski added 3 commits December 24, 2024 10:25

[Core] Add retry for failed upserts and handle circular dependencies

7b9d2fd

Rename _entity_topological_sorter to entity_topological_sorter

[Core] Add retry for failed upserts and handle circular dependencies

ac90c24

1. Modify logs and pass count as param. 2. Verify `create_missing_related_entities` is false before `sort_and_upsert_failed_entities` execution.

[Core] Add retry for failed upserts and handle circular dependencies

5c639eb

Update version

Tankilevitch approved these changes Dec 24, 2024

View reviewed changes

[Core] Add retry for failed upserts and handle circular dependencies

538bd4e

update is_to_execute to should_execute

Merge branch 'main' into PORT-11871-bug-bug-in-cycle-dependencies-aff…

fe32e7a

…ecting-argo-cd-integration

ivankalinovski merged commit b5690b8 into main Dec 24, 2024
18 checks passed

ivankalinovski deleted the PORT-11871-bug-bug-in-cycle-dependencies-affecting-argo-cd-integration branch December 24, 2024 14:51

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Core] Add retry for failed upserts and handle circular dependencies #1241

[Core] Add retry for failed upserts and handle circular dependencies #1241

ivankalinovski commented Dec 15, 2024 •

edited

Loading

Tankilevitch left a comment

Tankilevitch Dec 15, 2024

Tankilevitch Dec 17, 2024

Tankilevitch Dec 17, 2024

Tankilevitch Dec 17, 2024

Tankilevitch Dec 17, 2024

Tankilevitch Dec 17, 2024

Tankilevitch Dec 17, 2024

Tankilevitch Dec 17, 2024

Tankilevitch Dec 17, 2024

Tankilevitch Dec 19, 2024

Tankilevitch Dec 19, 2024

Tankilevitch Dec 19, 2024

Tankilevitch Dec 19, 2024

Tankilevitch Dec 19, 2024

Tankilevitch Dec 19, 2024

Tankilevitch Dec 22, 2024

Tankilevitch Dec 22, 2024

Tankilevitch Dec 22, 2024

Tankilevitch Dec 22, 2024

Tankilevitch Dec 22, 2024

Tankilevitch Dec 22, 2024

Tankilevitch Dec 22, 2024

Tankilevitch Dec 22, 2024

Tankilevitch Dec 23, 2024

Tankilevitch Dec 23, 2024

Tankilevitch left a comment

Tankilevitch Dec 24, 2024

		if upsertedEntity is False:
		event.register_failed_upsert_call_arguments(

	- When done with upserts, try topological sort on failed entities.
	- When `createMissingRelatedEntities` is set to `false` and upserting entity failed on not existing entity, the entity will be gathered to the end of the resync and will try sorting all the failed entities through a topological sort and upsert them as well

	- Update topologicals sort tree creation so an entity cannot be its own dependency.
	- Fix topologicals sort tree creation so an entity cannot be its own dependency


		### Bug Fixes

		- On fail of retry because of topological sort - try unsorted upsert.

	- On fail of retry because of topological sort - try unsorted upsert.
	- When experiencing cyclic error on topological sort try unsorted upsert of the entities

		except:
		await event.failed_entity_handler.handle_failed_no_sort()

-                try:
-                    await event.failed_entity_handler.handle_failed()
-                except:
-                    await event.failed_entity_handler.handle_failed_no_sort()
+                try:
+                    for entity in event.failed_entity_handler.get_sorted()
+                         await self.upsert_entity(
+                    entity,
+                    request_options,
+                    user_agent_type,
+                    should_raise=should_raise,
+                )
+                for entity in entities
+            ),
+                except:
+                    await event.failed_entity_handler.handle_failed_no_sort()

		@@ -27,6 +27,10 @@ def is_installation_type_compatible(self, installation_type: str) -> bool:
		) or installation_type == self.value


		class PortApiStatus(Enum):

		@@ -115,8 +114,7 @@ async def upsert(
		entities_without_search_identifier.append(entity)

		ordered_created_entities = reversed(

	if not event.entity_topological_sorter.is_to_execute():
	if not event.entity_topological_sorter.should_execute():

[Core] Add retry for failed upserts and handle circular dependencies #1241

[Core] Add retry for failed upserts and handle circular dependencies #1241

Conversation

ivankalinovski commented Dec 15, 2024 • edited Loading

Description

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

Tankilevitch left a 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

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

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

Tankilevitch left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ivankalinovski commented Dec 15, 2024 •

edited

Loading