Port 10639 add memray to jira integration

integrations/_infra/Dockerfile

@@ -5,14 +5,14 @@
 
 LABEL INTEGRATION_VERSION=${INTEGRATION_VERSION}
 # Used to ensure that new integrations will be public, see https://docs.github.com/en/packages/learn-github-packages/configuring-a-packages-access-control-and-visibility
 LABEL org.opencontainers.image.source https://github.com/port-labs/ocean
 
 ENV LIBRDKAFKA_VERSION 1.9.2
 
 WORKDIR /app
 
 RUN apt update && \
-    apt install -y wget make g++ libssl-dev autoconf automake libtool curl librdkafka-dev && \
+    apt install -y wget make g++ libssl-dev autoconf automake libtool curl librdkafka-dev google-perftools libjemalloc-dev && \
     apt-get clean
 
 COPY ./integrations/_infra/init.sh /app/init.sh
@@ -25,4 +25,4 @@
 
 RUN export POETRY_VIRTUALENVS_CREATE=false && make install/prod && pip cache purge
 
 ENTRYPOINT ./init.sh

integrations/_infra/init.sh

@@ -1,4 +1,5 @@
 if test -e /usr/local/share/ca-certificates/cert.crt; then
   update-ca-certificates
 fi
-ocean sail
+
+python ./debug.py

integrations/jira/pyproject.toml

@@ -1,12 +1,12 @@
 [tool.poetry]
 name = "jira"
-version = "0.1.90"
+version = "0.1.89-dev017"
 description = "Integration to bring information from Jira into Port"
 authors = ["Mor Paz <[email protected]>"]
 
 [tool.poetry.dependencies]
 python = "^3.11"
-port_ocean = {version = "^0.12.1", extras = ["cli"]}
+port_ocean = {version = "0.12.2-dev10", extras = ["cli"]}
 httpx = "^0.27.0"
 
 [tool.poetry.group.dev.dependencies]

port_ocean/core/event_listener/base.py

@@ -1,12 +1,14 @@
+import asyncio
 from abc import abstractmethod
+from concurrent.futures import ProcessPoolExecutor
 from typing import TypedDict, Callable, Any, Awaitable
 
 from pydantic import Extra
 
 from port_ocean.config.base import BaseOceanModel
-from port_ocean.utils.signal import signal_handler
 from port_ocean.context.ocean import ocean
 from port_ocean.utils.misc import IntegrationStateStatus
+from port_ocean.utils.signal import signal_handler
 
 
 class EventListenerEvents(TypedDict):
@@ -67,7 +69,14 @@ async def _resync(
         """
         await self._before_resync()
         try:
-            await self.events["on_resync"](resync_args)
+            loop = asyncio.get_event_loop()
+            with ProcessPoolExecutor() as executor:
+                e = executor.submit(
+                    lambda: asyncio.run_coroutine_threadsafe(
+                        self.events["on_resync"](resync_args), loop
+                    )
+                )
+                signal_handler.register(e.cancel)
             await self._after_resync()
         except Exception as e:
             await self._on_resync_failure(e)

port_ocean/core/integrations/mixins/sync_raw.py

@@ -140,11 +140,11 @@ async def _register_resource_raw(
         objects_diff = await self._calculate_raw(
             [(resource, results)], parse_all, send_raw_data_examples_amount
         )
-        modified_objects = await self.entities_state_applier.upsert(
+        await self.entities_state_applier.upsert(
             objects_diff[0].entity_selector_diff.passed, user_agent_type
         )
         return CalculationResult(
-            objects_diff[0].entity_selector_diff._replace(passed=modified_objects),
+            objects_diff[0].entity_selector_diff._replace(passed=objects_diff[0].entity_selector_diff.passed),
             errors=objects_diff[0].errors,
         )
 

port_ocean/log/handlers.py

@@ -1,8 +1,8 @@
 import asyncio
 import logging
 import sys
-import threading
 import time
+from concurrent.futures import ProcessPoolExecutor
 from datetime import datetime
 from logging.handlers import MemoryHandler
 from typing import Any
@@ -65,18 +65,19 @@ def flush(self) -> None:
         if self.ocean is None or not self.buffer:
             return
 
-        def _wrap_event_loop(_ocean: Ocean, logs_to_send: list[dict[str, Any]]) -> None:
-            loop = asyncio.new_event_loop()
-            loop.run_until_complete(self.send_logs(_ocean, logs_to_send))
-            loop.close()
-
         self.acquire()
         logs = list(self._serialized_buffer)
         if logs:
             self.buffer.clear()
             self._serialized_buffer.clear()
             self.last_flush_time = time.time()
-            threading.Thread(target=_wrap_event_loop, args=(self.ocean, logs)).start()
+            loop = asyncio.new_event_loop()
+            with ProcessPoolExecutor() as executor:
+                executor.submit(
+                    lambda: asyncio.run_coroutine_threadsafe(
+                        self.send_logs(self.ocean, logs), loop
+                    )
+                )
         self.release()
 
     async def send_logs(

port_ocean/ocean.py

@@ -1,6 +1,6 @@
 import asyncio
 import sys
-import threading
+from concurrent.futures import ProcessPoolExecutor
 from contextlib import asynccontextmanager
 from typing import Callable, Any, Dict, AsyncIterator, Type
 
@@ -9,8 +9,6 @@
 from pydantic import BaseModel
 from starlette.types import Scope, Receive, Send
 
-from port_ocean.core.handlers.resync_state_updater import ResyncStateUpdater
-from port_ocean.core.models import Runtime
 from port_ocean.clients.port.client import PortClient
 from port_ocean.config.settings import (
     IntegrationConfiguration,
@@ -20,13 +18,15 @@
     ocean,
     initialize_port_ocean_context,
 )
+from port_ocean.core.handlers.resync_state_updater import ResyncStateUpdater
 from port_ocean.core.integrations.base import BaseIntegration
+from port_ocean.core.models import Runtime
 from port_ocean.log.sensetive import sensitive_log_filter
 from port_ocean.middlewares import request_handler
+from port_ocean.utils.misc import IntegrationStateStatus
 from port_ocean.utils.repeat import repeat_every
 from port_ocean.utils.signal import signal_handler
 from port_ocean.version import __integration_version__
-from port_ocean.utils.misc import IntegrationStateStatus
 
 
 class Ocean:
@@ -92,6 +92,13 @@ async def execute_resync_all() -> None:
                 )
                 raise e
 
+        def pool_executor_wrapper(event_loop) -> None:
+            with ProcessPoolExecutor() as executor:
+                e = executor.submit(lambda: asyncio.run_coroutine_threadsafe(
+                    execute_resync_all(), event_loop
+                ))
+                signal_handler.register(e.cancel)
+
         interval = self.config.scheduled_resync_interval
         loop = asyncio.get_event_loop()
         if interval is not None:
@@ -103,13 +110,7 @@ async def execute_resync_all() -> None:
                 seconds=interval * 60,
                 # Not running the resync immediately because the event listener should run resync on startup
                 wait_first=True,
-            )(
-                lambda: threading.Thread(
-                    target=lambda: asyncio.run_coroutine_threadsafe(
-                        execute_resync_all(), loop
-                    )
-                ).start()
-            )
+            )(lambda _: pool_executor_wrapper(loop))
             await repeated_function()
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "port-ocean"
-version = "0.12.1"
+version = "0.12.2-dev06"
 description = "Port Ocean is a CLI tool for managing your Port projects."
 readme = "README.md"
 homepage = "https://app.getport.io"