[Core] ocean core next resync #835

shalev007 · 2024-07-23T13:04:05Z

Description

What:

Integrated functionality to send the state of the integration on each sync.
Implemented prediction of the next sync date.

Why:

To ensure that the integration state is consistently updated and monitored.
To enhance the accuracy and reliability of the next sync date prediction by utilizing server data for SaaS applications, rather than relying solely on environment variables.

How:

Updated the sync logic to include the current integration state in the payload sent to our monitoring system.
Modified the sync prediction mechanism for SaaS applications to use data from our servers, providing more accurate and context-aware predictions.

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)

Tankilevitch

CI doesn't and left some more comments

.python-version

port_ocean/ocean.py

Tankilevitch · 2024-07-25T15:24:44Z

port_ocean/ocean.py

+            interval_str = (
+                integration.get("spec", {})
+                .get("appSpec", {})
+                .get("scheduledResyncInterval")
+            )


I think we already have a code that turns the integration to an instance of integration, that way we don't need to actually use dictionary to get that key

I think your referring to ocean.config.integration but it only contains data from the env variables config and integration identifier

Tankilevitch · 2024-07-25T15:25:17Z

port_ocean/ocean.py

+        next_resync_date = now + datetime.timedelta(minutes=float(interval or 0))
+        next_resync = next_resync_date.now(datetime.timezone.utc).timestamp()


lets leave it in seconds and make the UI show it in minutes if we want

WDYM?
I changed it to timestamp to have better control over how we show it, and changed it to UTC so we won't have any timezone issues

port_ocean/ocean.py

yairsimantov20

Lets talk about the implementation

yairsimantov20 · 2024-07-29T07:36:57Z

port_ocean/core/event_listener/once.py

+    def should_update_resync_state(self) -> bool:
+        return ocean.config.runtime == RuntimeType.Saas.value
+
+    async def before_resync(self) -> None:
+        if not self.should_update_resync_state():
+            return None
+
+        now = datetime.datetime.now()
+        try:
+            integration = await ocean.port_client.get_current_integration()
+            interval_str = (
+                integration.get("spec", {})
+                .get("appSpec", {})
+                .get("scheduledResyncInterval")
+            )
+            interval = convert_time_to_minutes(interval_str)
+            self.resync_state["next_resync"] = calculate_next_resync(now, interval)
+        except Exception:
+            logger.exception("Error occurred while calculating next resync")
+            return None
+
+    async def after_resync(self) -> None:
+        if not self.should_update_resync_state():
+            return None
+
+        await ocean.port_client.update_resync_state(self.resync_state)


I want this implemented a bit different
With a lifecycle context manager

Seems like a great idea!
but I think it's out of this PR's scope and it will inflate this PR more than it should, WDYT?

yairsimantov20 · 2024-07-29T07:38:28Z

port_ocean/utils/misc.py

+def convert_time_to_minutes(time_str: str) -> int:
+    """
+    Convert a string representing time to minutes.
+    :param time_str: a string representing time in the format "1h" or "1m"
+    """
+    if time_str.endswith("h"):
+        hours = int(time_str[:-1])
+        return hours * 60
+    elif time_str.endswith("m"):
+        minutes = int(time_str[:-1])
+        return minutes
+    else:
+        raise ValueError("Invalid format. Expected a string ending with 'h' or 'm'.")


Why? Just send the time to port and let it handle it

it's inconsistent with the way we send out time in non-saas integrations

port_ocean/utils/misc.py

yairsimantov20 · 2024-07-29T07:40:29Z

port_ocean/ocean.py

+            )(
+                lambda: threading.Thread(
+                    target=asyncio.run(execute_resync_all())
+                ).start()
+            )


Try to implement with asyncio task instead

But tasks needs to be awaited which is not compatible with the target type Callable

FYI its incorrect task is a background task and we are using it in multiple places in ocean

for now its not important enough

yairsimantov20 · 2024-07-29T07:41:10Z

port_ocean/core/models.py

+    Saas = "Saas"
+    OnPrem = "OnPrem"
+
+
 Runtime = Literal["OnPrem", "Saas"]


Why is that needed anymore?

great point

port_ocean/core/event_listener/polling.py

port_ocean/clients/port/client.py

port_ocean/utils/misc.py

Tankilevitch · 2024-08-09T10:23:13Z

port_ocean/clients/port/client.py

+    async def update_integration_state(self, state: dict[str, Any]) -> dict[str, Any]:
+        logger.debug(f"Updating integration state with: {state}")
+        response = await self.client.patch(
+            f"{self.api_url}/integration/{self.integration_identifier}",
+            headers=await self.auth.headers(),
+            json={"state": state},
+        )
+        handle_status_code(response)
+        return response.json().get("integration", {})


lets add it an endpoint of its own, that way modifying to implementation in the future will be easier.
As well as passing the patch to state = null won't override the last state ( As it is a computed thing by the integration )

is this a blocker?
is it mandatory for this PR, because if we do have time to invest in this infra then I'd rather use a whole other infra set to save this data in

port_ocean/core/event_listener/once.py

port_ocean/utils/misc.py

port_ocean/core/event_listener/polling.py

Tankilevitch · 2024-08-09T10:59:34Z

port_ocean/core/event_listener/polling.py

                running_task: Task[Any] = get_event_loop().create_task(
                    self.events["on_resync"]({})  # type: ignore
                )
                signal_handler.register(running_task.cancel)

                await running_task
+                await self._after_resync()


what happens when it fails? will it stay running?

Yes it will not stay up and running, I think whatever error that will come up to this it will not exit the app.
But we need to take into account that if this request fails in polling event-listener we might run an un called for resync since the latest updatedAt will not inline with the current one

port_ocean/core/event_listener/kafka.py

port_ocean/core/event_listener/once.py

port_ocean/ocean.py

port_ocean/core/event_listener/base.py

port_ocean/core/event_listener/once.py

The calculation is running in a non blocking way before the start of the resync to avoid the calculating the resync time

…n-service

Tankilevitch · 2024-08-18T06:51:42Z

port_ocean/ocean.py

+        # TODO: remove this once we separate the state from the integration
+        self.last_integration_updated_at: str = ""


explain why this is needed

Added a comment, but this purely technical I do think the code is self explanatory once you look it up

Tankilevitch · 2024-08-18T06:52:31Z

port_ocean/core/event_listener/once.py

+        # we use the last updated time of the integration config as the start time since in saas application the interval is configured by the user from the portal
+        if not last_updated_saas_integration_config_str:
+            logger.error("updatedAt not found for integration")
+            return (None, None)


what is the effect and whether there are any action items?

for example, more detailed log for whether this was expected or not

Tankilevitch · 2024-08-18T14:04:20Z

port_ocean/clients/port/client.py

+
+    async def update_integration_state(self, state: dict[str, Any]) -> dict[str, Any]:
+        logger.debug(f"Updating integration state with: {state}")


make sure that should_log and should_raise are parameters that can be passed

Why?
We don't need this to be changed, these parameters stay constant across event-listener

this is code in the client class, therefor it should be an interface for other uses as well. and therefor each call to the client should decide whether it wants to raise the error or not

Tankilevitch · 2024-08-18T14:05:11Z

port_ocean/clients/port/client.py

+        )
+        handle_status_code(response, should_raise=False, should_log=True)
+        if not response.is_error:
+            logger.info("Integration state updated successfully")


and use the should_log here as well

once you add the should_raise and should_log then decide if to log based on the parameter passed

Tankilevitch

LGTM

…n-service

yairsimantov20 · 2024-08-20T08:13:39Z

port_ocean/clients/port/client.py

+        if response.is_success and should_log:
+            logger.info("Integration state updated successfully")
+
+        return response.json().get("integration", {})


Suggested change

return response.json().get("integration", {})

return response.json()["integration"]

yairsimantov20 · 2024-08-20T08:25:07Z

port_ocean/core/event_listener/base.py

+    async def _before_resync(self) -> None:
+        """
+        Can be used for event listeners that need to perform some action before resync.
+        """
+        await ocean.app.update_state_before_scheduled_sync()
+
+    async def _after_resync(self) -> None:
+        """
+        Can be used for event listeners that need to perform some action after resync.
+        """
+        await ocean.app.update_state_after_scheduled_sync()
+
+    async def _on_resync_failure(self, e: Exception) -> None:
+        """
+        Can be used for event listeners that need to handle resync failures.
+        """
+        await ocean.app.update_state_after_scheduled_sync(IntegrationStateStatus.Failed)
+
+    async def _resync(
+        self,
+        resync_args: dict[Any, Any],
+    ) -> None:
+        """
+        Triggers the "on_resync" event.
+        """
+        await self._before_resync()
+        try:
+            await self.events["on_resync"](resync_args)
+            await self._after_resync()
+        except Exception as e:
+            await self._on_resync_failure(e)
+            raise e


can we instead use a context manager syntax?

@asynccontextmanager async resync_life_cycle(): await ocean.app.update_state_before_scheduled_sync() try: yield except Exception as e: await ocean.app.update_state_after_scheduled_sync(IntegrationStateStatus.Failed) raise e else: await ocean.app.update_state_after_scheduled_sync()

Suggested change

async def _before_resync(self) -> None:

"""

Can be used for event listeners that need to perform some action before resync.

"""

await ocean.app.update_state_before_scheduled_sync()

async def _after_resync(self) -> None:

"""

Can be used for event listeners that need to perform some action after resync.

"""

await ocean.app.update_state_after_scheduled_sync()

async def _on_resync_failure(self, e: Exception) -> None:

"""

Can be used for event listeners that need to handle resync failures.

"""

await ocean.app.update_state_after_scheduled_sync(IntegrationStateStatus.Failed)

async def _resync(

self,

resync_args: dict[Any, Any],

) -> None:

"""

Triggers the "on_resync" event.

"""

await self._before_resync()

try:

await self.events["on_resync"](resync_args)

await self._after_resync()

except Exception as e:

await self._on_resync_failure(e)

raise e

async def _resync(

self,

resync_args: dict[Any, Any],

) -> None:

"""

Triggers the "on_resync" event.

"""

async with resync_life_cycle():

await self.events["on_resync"](resync_args)

yairsimantov20

A lot of code here is irrelevant to ocean core and can be calculated on the port side

yairsimantov20 · 2024-08-20T08:47:59Z

port_ocean/ocean.py

+            )(
+                lambda: threading.Thread(
+                    target=asyncio.run(execute_resync_all())
+                ).start()
+            )


FYI its incorrect task is a background task and we are using it in multiple places in ocean

for now its not important enough

yairsimantov20 · 2024-08-20T09:17:19Z

port_ocean/utils/time.py

+    aware_date = datetime.datetime.fromisoformat(time_str)
+    if time_str.endswith("Z"):
+        aware_date = datetime.datetime.fromisoformat(time_str.replace("Z", "+00:00"))
+    return aware_date.astimezone(datetime.timezone.utc)


Suggested change

aware_date = datetime.datetime.fromisoformat(time_str)

if time_str.endswith("Z"):

aware_date = datetime.datetime.fromisoformat(time_str.replace("Z", "+00:00"))

return aware_date.astimezone(datetime.timezone.utc)

return datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%SZ").astimezone(datetime.timezone.utc)

yairsimantov20 · 2024-08-20T09:21:39Z

port_ocean/utils/misc.py

+    Running = "running"
+    Failed = "failed"
+    Completed = "completed"


any reason those are lower case?

yairsimantov20 · 2024-08-20T09:23:43Z

port_ocean/ocean.py

+        self.initiated_at = datetime.datetime.now(tz=datetime.timezone.utc)

-    async def _setup_scheduled_resync(
+        # This is used to differ between integration changes that require a full resync and state changes
+        # So that the polling event-listener can decide whether to perform a full resync or not
+        # TODO: remove this once we separate the state from the integration
+        self.last_integration_state_updated_at: str = ""
+
+    def is_saas(self) -> bool:
+        return self.config.runtime == Runtime.Saas
+
+    def _calculate_next_scheduled_resync(
+        self,
+        interval: int | None = None,
+        custom_start_time: datetime.datetime | None = None,
+    ) -> str | None:
+        if interval is None:
+            return None
+        return get_next_occurrence(
+            interval * 60, custom_start_time or self.initiated_at
+        ).isoformat()
+
+    async def update_state_before_scheduled_sync(
        self,
+        interval: int | None = None,
+        custom_start_time: datetime.datetime | None = None,
    ) -> None:
-        def execute_resync_all() -> None:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
+        _interval = interval or self.config.scheduled_resync_interval
+        nest_resync = self._calculate_next_scheduled_resync(
+            _interval, custom_start_time
+        )
+        state: dict[str, Any] = {
+            "status": IntegrationStateStatus.Running.value,
+            "lastResyncEnd": None,
+            "lastResyncStart": datetime.datetime.now(
+                tz=datetime.timezone.utc
+            ).isoformat(),
+            "nextResync": nest_resync,
+            "intervalInMinuets": _interval,
+        }
+
+        integration = await self.port_client.update_integration_state(
+            state, should_raise=False
+        )
+        if integration:
+            self.last_integration_state_updated_at = integration["state"]["updatedAt"]

+    async def update_state_after_scheduled_sync(
+        self,
+        status: Literal[
+            IntegrationStateStatus.Completed, IntegrationStateStatus.Failed
+        ] = IntegrationStateStatus.Completed,
+        interval: int | None = None,
+        custom_start_time: datetime.datetime | None = None,
+    ) -> None:
+        _interval = interval or self.config.scheduled_resync_interval
+        nest_resync = self._calculate_next_scheduled_resync(
+            _interval, custom_start_time
+        )
+        state: dict[str, Any] = {
+            "status": status.value,
+            "lastResyncEnd": datetime.datetime.now(
+                tz=datetime.timezone.utc
+            ).isoformat(),
+            "nextResync": nest_resync,
+            "intervalInMinuets": _interval,
+        }
+
+        integration = await self.port_client.update_integration_state(
+            state, should_raise=False
+        )
+        if integration:
+            self.last_integration_state_updated_at = integration["state"]["updatedAt"]
+
+    async def _setup_scheduled_resync(
+        self,
+    ) -> None:
+        async def execute_resync_all() -> None:
+            await self.update_state_before_scheduled_sync()
            logger.info("Starting a new scheduled resync")
-            loop.run_until_complete(self.integration.sync_raw_all())
-            loop.close()
+            try:
+                await self.integration.sync_raw_all()
+                await self.update_state_after_scheduled_sync()
+            except Exception as e:
+                await self.update_state_after_scheduled_sync(
+                    IntegrationStateStatus.Failed
+                )
+                raise e


This logic does not belong in this class
it is irrelevant for the core
this feature is only a side effect to the resync no the main core feature

All remaining changes can be added after this feature is deployed

…n-service

Tankilevitch · 2024-08-21T14:20:49Z

port_ocean/clients/port/client.py

+    async def update_integration_state(
+        self, state: dict[str, Any], should_raise: bool = True, should_log: bool = True
+    ) -> dict[str, Any]:
+        if should_log:
+            logger.debug(f"Updating integration state with: {state}")
+        response = await self.client.patch(
+            f"{self.api_url}/integration/{self.integration_identifier}/state",
+            headers=await self.auth.headers(),
+            json=state,
+        )
+        handle_status_code(response, should_raise, should_log)
+        if response.is_success and should_log:
+            logger.info("Integration state updated successfully")


Suggested change

async def update_integration_state(

self, state: dict[str, Any], should_raise: bool = True, should_log: bool = True

) -> dict[str, Any]:

if should_log:

logger.debug(f"Updating integration state with: {state}")

response = await self.client.patch(

f"{self.api_url}/integration/{self.integration_identifier}/state",

headers=await self.auth.headers(),

json=state,

)

handle_status_code(response, should_raise, should_log)

if response.is_success and should_log:

logger.info("Integration state updated successfully")

async def update_integration_resync_state(

self, state: dict[str, Any], should_raise: bool = True, should_log: bool = True

) -> dict[str, Any]:

if should_log:

logger.debug(f"Updating integration state with: {state}")

response = await self.client.patch(

f"{self.api_url}/integration/{self.integration_identifier}/resync-state",

headers=await self.auth.headers(),

json=state,

)

handle_status_code(response, should_raise, should_log)

if response.is_success and should_log:

logger.info("Integration resync state updated successfully")

…n-service

github-actions bot added size/S size/M and removed size/S labels Jul 23, 2024

shalev007 changed the title ~~PORT 9302 ocean core send data to integration service~~ [Core] ocean core send data to integration service Jul 23, 2024

Tankilevitch requested changes Jul 25, 2024

View reviewed changes

shalev007 force-pushed the PORT-9302-ocean-core-send-data-to-integration-service branch from c52cdde to 22146da Compare July 28, 2024 15:05

shalev007 marked this pull request as ready for review July 28, 2024 15:12

yairsimantov20 requested changes Jul 29, 2024

View reviewed changes

yairsimantov20 changed the title ~~[Core] ocean core send data to integration service~~ [Core] ocean core next resync Jul 29, 2024

Tankilevitch reviewed Jul 29, 2024

View reviewed changes

port_ocean/clients/port/client.py Outdated Show resolved Hide resolved

port_ocean/clients/port/client.py Outdated Show resolved Hide resolved

port_ocean/utils/misc.py Outdated Show resolved Hide resolved

github-actions bot added size/L and removed size/M labels Aug 6, 2024

shalev007 force-pushed the PORT-9302-ocean-core-send-data-to-integration-service branch from 9092c8b to 1434dcf Compare August 7, 2024 07:13

shalev007 requested a review from Tankilevitch August 7, 2024 07:21

Tankilevitch requested changes Aug 9, 2024

View reviewed changes

shalev007 force-pushed the PORT-9302-ocean-core-send-data-to-integration-service branch from 83e5342 to 373e9e5 Compare August 13, 2024 08:20

Tankilevitch requested changes Aug 13, 2024

View reviewed changes

shalev007 force-pushed the PORT-9302-ocean-core-send-data-to-integration-service branch 2 times, most recently from 997cae6 to 6958ad1 Compare August 14, 2024 14:00

Shalev Avhar added 11 commits August 15, 2024 18:46

feat: add update port-api method

f1c9d90

feat: update port-api when sync is done WIP

446dbb0

fix: remove dev file

329245c

feat: revert execute_resync_all and calculate next_resync

d73a3ec

The calculation is running in a non blocking way before the start of the resync to avoid the calculating the resync time

feat: move convert function to utils

076cbae

fix: use the once event listener

383d99b

fix: remove update from the initialization itself

b7b9c4a

feat: add enum

79fde73

fix: use enum value

3f82fc2

fix: don't kill integration on update state error

d208f84

refactor: move calculate_next_resync to misc

b33fbfb

feat: update state parameters

8f66686

shalev007 requested a review from a team as a code owner August 18, 2024 13:02

Shalev Avhar added 2 commits August 18, 2024 16:03

Merge branch 'main' into PORT-9302-ocean-core-send-data-to-integratio…

53a40c3

…n-service

feat: do silent updates

5595405

Tankilevitch requested changes Aug 18, 2024

View reviewed changes

Shalev Avhar added 4 commits August 19, 2024 10:04

feat: add status as constant

f614823

feat: add explanation comment

e442985

feat: add better logs

a8e022c

feat: add should_raise and should_log

6d1e81f

Tankilevitch approved these changes Aug 19, 2024

View reviewed changes

Shalev Avhar added 5 commits August 19, 2024 13:25

refactor: rename variable

35c5688

feat: change default update state behavior

0ef8b52

feat: bump version

23d4ee9

Merge branch 'main' into PORT-9302-ocean-core-send-data-to-integratio…

975f226

…n-service

feat: change changelog

adb7967

shalev007 requested a review from yairsimantov20 August 19, 2024 11:29

yairsimantov20 reviewed Aug 20, 2024

View reviewed changes

yairsimantov20 previously requested changes Aug 20, 2024

View reviewed changes

Shalev Avhar and others added 4 commits August 20, 2024 15:56

Merge branch 'main' into PORT-9302-ocean-core-send-data-to-integratio…

661bea8

…n-service

fix: scheduled resync blocked the main thread

7b229fd

Merge branch 'main' into PORT-9302-ocean-core-send-data-to-integratio…

0a50bff

…n-service

refactor: move all update state functionality to a new class

36774ea

Tankilevitch reviewed Aug 21, 2024

View reviewed changes

Shalev Avhar and others added 3 commits August 21, 2024 17:59

feat: update resync-state new endpoint

e793681

fix: lint checks

19f93df

Merge branch 'main' into PORT-9302-ocean-core-send-data-to-integratio…

21848be

…n-service

shalev007 merged commit d54b97b into main Aug 21, 2024
12 checks passed

shalev007 deleted the PORT-9302-ocean-core-send-data-to-integration-service branch August 21, 2024 15:37

		next_resync_date = now + datetime.timedelta(minutes=float(interval or 0))
		next_resync = next_resync_date.now(datetime.timezone.utc).timestamp()

		# TODO: remove this once we separate the state from the integration
		self.last_integration_updated_at: str = ""


		async def update_integration_state(self, state: dict[str, Any]) -> dict[str, Any]:
		logger.debug(f"Updating integration state with: {state}")

	return response.json().get("integration", {})
	return response.json()["integration"]

[Core] ocean core next resync #835

[Core] ocean core next resync #835

Conversation

shalev007 commented Jul 23, 2024 • edited by Tankilevitch Loading

Description

Type of change

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

yairsimantov20 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

shalev007 Aug 19, 2024 • edited Loading

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

yairsimantov20 Aug 20, 2024 • edited Loading

Choose a reason for hiding this comment

yairsimantov20 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

shalev007 commented Jul 23, 2024 •

edited by Tankilevitch

Loading

shalev007 Aug 19, 2024 •

edited

Loading

yairsimantov20 Aug 20, 2024 •

edited

Loading