[Core] Fix stop iterator bug in jqpy (#1059)

CHANGELOG.md

@@ -7,6 +7,23 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 <!-- towncrier release notes start -->
 
+## 0.12.1 (2024-10-02)
+
+### Bug Fixes
+
+- Fixed a bug when running jq with iterator that caused the integration to crash
+- Reverted image to `python:3.11-slim-buster` to fix the issue with the alpine image
+
+## 0.12.0 (2024-10-01)
+
+### Improvements
+
+- Replace `python:3.11-slim-bookworm` with `python:3.11-alpine` to reduce dependencies and fix vulnerabilities
+
+### Bug Fixes
+
+- Fixed smoke tests to run concurrently and clean up after themselves
+
 ## 0.11.0 (2024-09-29)
 
 ### Improvements

integrations/_infra/Dockerfile

@@ -1,72 +1,28 @@
-FROM python:3.11-alpine AS base
+FROM python:3.11-slim-bookworm
 
 ARG BUILD_CONTEXT
-
-ENV LIBRDKAFKA_VERSION=1.9.2
-
-# Install system dependencies and libraries
-RUN apk add --no-cache \
-    gcc \
-    musl-dev \
-    build-base \
-    bash \
-    oniguruma-dev \
-    make \
-    autoconf \
-    automake \
-    libtool \
-    curl \
-    # librdkafka-dev \
-    libffi-dev \
-        # Install community librdkafka-dev since the default in alpine is older
-        && echo "@edge http://dl-cdn.alpinelinux.org/alpine/edge/main" >> /etc/apk/repositories \
-        && echo "@edgecommunity http://dl-cdn.alpinelinux.org/alpine/edge/community" >> /etc/apk/repositories \
-        && apk add --no-cache alpine-sdk "librdkafka@edgecommunity>=${LIBRDKAFKA_VERSION}" "librdkafka-dev@edgecommunity>=${LIBRDKAFKA_VERSION}" \
-        && curl -sSL https://install.python-poetry.org | python3 - \
-        && /root/.local/bin/poetry config virtualenvs.in-project true
-
-
-WORKDIR /app
-
-COPY ./${BUILD_CONTEXT}/pyproject.toml ./${BUILD_CONTEXT}/poetry.lock /app/
-
-RUN /root/.local/bin/poetry install --without dev --no-root --no-interaction --no-ansi --no-cache && pip cache purge
-
-FROM python:3.11-alpine AS prod
-
 ARG INTEGRATION_VERSION
-ARG BUILD_CONTEXT
 
 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
+LABEL org.opencontainers.image.source https://github.com/port-labs/ocean
 
-# Install only runtime dependencies
-RUN apk add --no-cache \
-        librdkafka-dev \
-        bash \
-        oniguruma-dev \
-    # Install community librdkafka-dev since the default in alpine is older
-    && echo "@edge http://dl-cdn.alpinelinux.org/alpine/edge/main" >> /etc/apk/repositories \
-    && echo "@edgecommunity http://dl-cdn.alpinelinux.org/alpine/edge/community" >> /etc/apk/repositories \
-    && apk add --no-cache alpine-sdk "librdkafka@edgecommunity>=${LIBRDKAFKA_VERSION}" "librdkafka-dev@edgecommunity>=${LIBRDKAFKA_VERSION}" \
-    && test -e /usr/local/share/ca-certificates/cert.crt && update-ca-certificates || true
+ENV LIBRDKAFKA_VERSION 1.9.2
 
 WORKDIR /app
 
-# Copy dependencies from the build stage
-COPY --from=base /app /app
+RUN apt update && \
+    apt install -y wget make g++ libssl-dev autoconf automake libtool curl librdkafka-dev && \
+    apt-get clean
+
+COPY ./integrations/_infra/init.sh /app/init.sh
+
+RUN chmod +x /app/init.sh
 
-# Copy the application code
 COPY ./${BUILD_CONTEXT} /app
 
-# Ensure that ocean is available for all in path
-RUN chmod a+x /app/.venv/bin/ocean \
-    && ln -s /app/.venv/bin/ocean /usr/bin/ocean \
-    #    # Fix security issues
-    && apk upgrade busybox --repository=http://dl-cdn.alpinelinux.org/alpine/edge/main \
-    # Clean up old setuptools
-    && pip uninstall -y setuptools py3-setuptools
+COPY ./integrations/_infra/Makefile /app/Makefile
+
+RUN export POETRY_VIRTUALENVS_CREATE=false && make install/prod && pip cache purge
 
-# Run the application
-CMD ["ocean", "sail"]
+ENTRYPOINT ./init.sh

integrations/_infra/init.sh

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

port_ocean/core/handlers/entity_processor/jq_entity_processor.py

@@ -47,37 +47,48 @@ def _compile(self, pattern: str) -> Any:
             pattern = "def env: {}; {} as $ENV | " + pattern
         return jq.compile(pattern)
 
+    @staticmethod
+    def _stop_iterator_handler(func: Any) -> Any:
+        """
+        Wrap the function to handle StopIteration exceptions.
+        Prevents StopIteration from stopping the thread and skipping further queue processing.
+        """
+
+        def inner() -> Any:
+            try:
+                return func()
+            except StopIteration:
+                return None
+
+        return inner
+
     async def _search(self, data: dict[str, Any], pattern: str) -> Any:
         try:
             loop = asyncio.get_event_loop()
             compiled_pattern = self._compile(pattern)
             func = compiled_pattern.input_value(data)
-            return await loop.run_in_executor(None, func.first)
+            return await loop.run_in_executor(
+                None, self._stop_iterator_handler(func.first)
+            )
         except Exception as exc:
             logger.debug(
-                f"Failed to search for pattern {pattern} in data {data}, {exc}"
+                f"Search failed for pattern '{pattern}' in data: {data}, Error: {exc}"
             )
             return None
 
     async def _search_as_bool(self, data: dict[str, Any], pattern: str) -> bool:
         loop = asyncio.get_event_loop()
-        start_time = loop.time()
+
         compiled_pattern = self._compile(pattern)
         func = compiled_pattern.input_value(data)
-        compile_time = loop.time() - start_time
-        value = await loop.run_in_executor(None, func.first)
-        execute_time = loop.time() - start_time - compile_time
-        logger.debug(
-            f"Search for pattern {execute_time:.2f} seconds, compile time {compile_time:.2f} seconds",
-            pattern=pattern,
-            compile_time=compile_time,
-            execute_time=execute_time,
+
+        value = await loop.run_in_executor(
+            None, self._stop_iterator_handler(func.first)
         )
         if isinstance(value, bool):
             return value
-
         raise EntityProcessorException(
-            f"Expected boolean value, got {type(value)} instead"
+            f"Expected boolean value, got value:{value} of type: {type(value)} instead"
         )
 
     async def _search_as_object(

port_ocean/tests/core/handlers/entity_processor/test_jq_entity_processor.py

@@ -195,10 +195,45 @@ async def test_search_as_bool_failure(
         pattern = ".foo"
         with pytest.raises(
             EntityProcessorException,
-            match="Expected boolean value, got <class 'str'> instead",
+            match="Expected boolean value, got value:bar of type: <class 'str'> instead",
         ):
             await mocked_processor._search_as_bool(data, pattern)
 
+    @pytest.mark.parametrize(
+        "pattern, expected",
+        [
+            ('.parameters[] | select(.name == "not_exists") | .value', None),
+            (
+                '.parameters[] | select(.name == "parameter_name") | .value',
+                "parameter_value",
+            ),
+            (
+                '.parameters[] | select(.name == "another_parameter") | .value',
+                "another_value",
+            ),
+        ],
+    )
+    async def test_search_fails_on_stop_iteration(
+        self, mocked_processor: JQEntityProcessor, pattern: str, expected: Any
+    ) -> None:
+        data = {
+            "parameters": [
+                {"name": "parameter_name", "value": "parameter_value"},
+                {"name": "another_parameter", "value": "another_value"},
+                {"name": "another_parameter", "value": "another_value2"},
+            ]
+        }
+        result = await mocked_processor._search(data, pattern)
+        assert result == expected
+
+    async def test_return_a_list_of_values(
+        self, mocked_processor: JQEntityProcessor
+    ) -> None:
+        data = {"parameters": ["parameter_value", "another_value", "another_value2"]}
+        pattern = ".parameters"
+        result = await mocked_processor._search(data, pattern)
+        assert result == ["parameter_value", "another_value", "another_value2"]
+
     @pytest.mark.timeout(3)
     async def test_search_performance_10000(
         self, mocked_processor: JQEntityProcessor

pyproject.toml

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