zozo123 commented on code in PR #71672: URL: https://github.com/apache/airflow/pull/71672#discussion_r4006178448
########## providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""islo.dev microVM backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import math +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxFileTooLargeError, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) +from airflow.providers.common.compat.sdk import BaseHook + +if TYPE_CHECKING: + from collections.abc import Iterator + + from islo import Islo + from islo.errors import NotFoundError + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "timeout"}) +_POLL_INITIAL = 0.2 +_POLL_MAX = 2.0 +_POLL_BACKOFF = 1.5 +_FILE_OP_TIMEOUT = 120.0 +_HELPER_OUTPUT_CAP = 1024 * 1024 +_COMMAND_WRAPPER = """\ +umask 077 +dir="${TMPDIR:-/tmp}/airflow-sandbox-$$" +mkdir "$dir" || exit 70 +trap 'rm -rf "$dir"' EXIT HUP INT TERM +mkfifo "$dir/out" "$dir/err" "$dir/out-tail" "$dir/err-tail" || exit 70 +tail -c "$2" <"$dir/out-tail" >"$dir/out-result" & +out_tail_pid=$! +tail -c "$2" <"$dir/err-tail" >"$dir/err-result" & +err_tail_pid=$! +tee "$dir/out-tail" <"$dir/out" | wc -c >"$dir/out-count" & +out_drain_pid=$! +tee "$dir/err-tail" <"$dir/err" | wc -c >"$dir/err-count" & +err_drain_pid=$! +sh -lc "$1" >"$dir/out" 2>"$dir/err" +status=$? +wait "$out_drain_pid" "$err_drain_pid" "$out_tail_pid" "$err_tail_pid" || exit 70 Review Comment: Confirmed, and it reproduced in a real microVM too: `sleep 20 & echo started` took 20.1s, because the backgrounded child inherits the fifo write end so the drains never see end-of-input. Rather than bound the drains, the wrapper no longer has any. Each stream is redirected to a scratch file and the last `max_output_bytes` are emitted with `tail -c` afterwards, so only the foreground command is waited on — the fifos, the `tee`/`wc` drains and the `wait` are all gone. Same command is now 0.3s against a live sandbox, and `nohup sleep 300 & echo server-started` returns immediately. The trade is that total output lands on the sandbox’s own ephemeral disk instead of streaming through a fixed-size window; that is called out in the class docstring and the docs. Covered by `test_a_backgrounded_process_does_not_hold_the_command_open` and `test_a_long_lived_daemon_does_not_hold_the_command_open`, which run the wrapper through a local `sh`. ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""islo.dev microVM backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import math +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxFileTooLargeError, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) +from airflow.providers.common.compat.sdk import BaseHook + +if TYPE_CHECKING: + from collections.abc import Iterator + + from islo import Islo + from islo.errors import NotFoundError + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "timeout"}) +_POLL_INITIAL = 0.2 +_POLL_MAX = 2.0 +_POLL_BACKOFF = 1.5 +_FILE_OP_TIMEOUT = 120.0 +_HELPER_OUTPUT_CAP = 1024 * 1024 +_COMMAND_WRAPPER = """\ +umask 077 Review Comment: Right, and it was visible exactly as you describe — `touch f; mkdir d` came back `-rw-------`/`drwx------` through the wrapper versus `-rw-r--r--`/`drwxr-xr-x` without it. Replaced with `mkdir -m 700 "$dir"`, which is atomic and never puts a umask in front of the agent’s command. `test_does_not_change_the_permissions_of_what_the_agent_creates` asserts the normal modes, and `test_the_scratch_directory_is_private_while_in_use_and_gone_after` stats the scratch directory mid-run to confirm it is still `0700` and removed afterwards. ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""islo.dev microVM backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import math +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxFileTooLargeError, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) +from airflow.providers.common.compat.sdk import BaseHook + +if TYPE_CHECKING: + from collections.abc import Iterator + + from islo import Islo + from islo.errors import NotFoundError + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "timeout"}) +_POLL_INITIAL = 0.2 +_POLL_MAX = 2.0 +_POLL_BACKOFF = 1.5 +_FILE_OP_TIMEOUT = 120.0 +_HELPER_OUTPUT_CAP = 1024 * 1024 +_COMMAND_WRAPPER = """\ +umask 077 +dir="${TMPDIR:-/tmp}/airflow-sandbox-$$" +mkdir "$dir" || exit 70 +trap 'rm -rf "$dir"' EXIT HUP INT TERM Review Comment: Fixed as suggested: `trap 'rm -rf "$dir"; exit 143' HUP INT TERM` with the bare `EXIT` trap kept for cleanup. Worth noting the old behaviour was environment-dependent — signalling the process group locally gave exit 70 from the failed `wait` rather than your 0 — but either way it cleaned up and then kept running with its scratch files gone. It is now an unambiguous 143 with no output. `test_a_terminated_command_exits_nonzero_without_emitting_garbage` covers it by signalling the whole process group, which is what stopping the microVM looks like from inside. ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""islo.dev microVM backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import math +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxFileTooLargeError, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) +from airflow.providers.common.compat.sdk import BaseHook + +if TYPE_CHECKING: + from collections.abc import Iterator + + from islo import Islo + from islo.errors import NotFoundError + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "timeout"}) +_POLL_INITIAL = 0.2 +_POLL_MAX = 2.0 +_POLL_BACKOFF = 1.5 +_FILE_OP_TIMEOUT = 120.0 +_HELPER_OUTPUT_CAP = 1024 * 1024 +_COMMAND_WRAPPER = """\ +umask 077 +dir="${TMPDIR:-/tmp}/airflow-sandbox-$$" +mkdir "$dir" || exit 70 +trap 'rm -rf "$dir"' EXIT HUP INT TERM +mkfifo "$dir/out" "$dir/err" "$dir/out-tail" "$dir/err-tail" || exit 70 +tail -c "$2" <"$dir/out-tail" >"$dir/out-result" & Review Comment: Reproduced exactly: with a 100 byte cap the model got `88`, the tail of `line988`. Truncation now drops the leading partial line when it cuts, the same way `sbx.py`’s drain does, so `output.py`’s promise holds here too. Live check with a 100 byte cap now starts at `line1990` and ends at `line2000`. The `list_directory` path you flagged needed separate handling, since its records are NUL-separated and have no newline for the cut to align on: `_run_helper` now returns the `SandboxExecResult` and `list_directory` drops a leading partial record when the stream was truncated. `test_truncation_never_emits_a_partial_leading_line` and `test_keeps_the_tail_and_never_a_partial_leading_line` cover the command path. ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""islo.dev microVM backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import math +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxFileTooLargeError, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) +from airflow.providers.common.compat.sdk import BaseHook + +if TYPE_CHECKING: + from collections.abc import Iterator + + from islo import Islo + from islo.errors import NotFoundError + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "timeout"}) +_POLL_INITIAL = 0.2 +_POLL_MAX = 2.0 +_POLL_BACKOFF = 1.5 +_FILE_OP_TIMEOUT = 120.0 +_HELPER_OUTPUT_CAP = 1024 * 1024 +_COMMAND_WRAPPER = """\ +umask 077 +dir="${TMPDIR:-/tmp}/airflow-sandbox-$$" +mkdir "$dir" || exit 70 +trap 'rm -rf "$dir"' EXIT HUP INT TERM +mkfifo "$dir/out" "$dir/err" "$dir/out-tail" "$dir/err-tail" || exit 70 +tail -c "$2" <"$dir/out-tail" >"$dir/out-result" & +out_tail_pid=$! +tail -c "$2" <"$dir/err-tail" >"$dir/err-result" & +err_tail_pid=$! +tee "$dir/out-tail" <"$dir/out" | wc -c >"$dir/out-count" & +out_drain_pid=$! +tee "$dir/err-tail" <"$dir/err" | wc -c >"$dir/err-count" & +err_drain_pid=$! +sh -lc "$1" >"$dir/out" 2>"$dir/err" +status=$? +wait "$out_drain_pid" "$err_drain_pid" "$out_tail_pid" "$err_tail_pid" || exit 70 +if [ "$(cat "$dir/out-count")" -gt "$2" ]; then printf '1\\n'; else printf '0\\n'; fi +cat "$dir/out-result" +if [ "$(cat "$dir/err-count")" -gt "$2" ]; then printf '1\\n' >&2; else printf '0\\n' >&2; fi +cat "$dir/err-result" >&2 +exit "$status" +""" + + +@contextmanager +def _translate_islo_errors(operation: str) -> Iterator[None]: + try: + yield + except SandboxError: + raise + except Exception as e: + try: + from islo.core.api_error import ApiError + except ImportError: + raise SandboxTerminalError( + 'The Islo SDK is not installed. Install "apache-airflow-providers-common-ai[sandbox-islo]".' + ) from e + if isinstance(e, ApiError): + status = f" (HTTP {e.status_code})" if e.status_code is not None else "" + raise SandboxTerminalError(f"Islo could not {operation}{status}.") from e + raise SandboxTerminalError(f"Islo could not {operation}: {type(e).__name__}.") from e + + +def _bound_result_stream(text: str, max_bytes: int, *, server_truncated: bool) -> tuple[str, bool]: + flag, separator, payload = text.partition("\n") + if separator and flag in {"0", "1"}: + truncated = flag == "1" or server_truncated + else: + payload = text + truncated = True + + encoded = payload.encode("utf-8") + if len(encoded) > max_bytes: + payload = encoded[-max_bytes:].decode("utf-8", errors="ignore") + truncated = True + return payload, truncated + + +class IsloSandboxBackend(SandboxBackend): + """ + Sandbox backend that runs agent commands in an `islo.dev <https://islo.dev>`__ microVM. + + Islo is a hosted API with no local daemon or host-virtualization requirement, + so this backend works from an Airflow worker running in a container. + Credentials resolve lazily from an Airflow connection on first use. + + Connection fields: ``password`` is the Islo API key (required), ``host`` the + compute URL (optional), and the extra may set ``base_url`` and ``timeout`` + (request timeout in seconds). + + File reads and writes use Islo's native streaming APIs. Directory listings + and command-output bounding require common Unix command-line tools in the + sandbox image, including ``sh``, ``mkfifo``, ``tail``, ``tee`` and a ``find`` + implementation with ``-printf`` support. + + :param islo_conn_id: Airflow connection ID for Islo. ``None`` lets the SDK + resolve credentials from its own environment variables (``ISLO_API_KEY``). + :param image: Sandbox image. ``None`` (default) uses the server default. + :param vcpus: Number of virtual CPUs. ``None`` uses the server default. + :param memory_mb: Memory in MB. ``None`` uses the server default. + :param delete_after: Server-side TTL in seconds after which the sandbox is + deleted even if the worker never got to destroy it. Default ``3600``. + """ + + name = "islo" + + def __init__( + self, + islo_conn_id: str | None = "islo_default", + *, + image: str | None = None, + vcpus: int | None = None, + memory_mb: int | None = None, + delete_after: int = 3600, + ) -> None: + _validate_positive_finite(delete_after, "delete_after") + if vcpus is not None: + _validate_positive_finite(vcpus, "vcpus") + if memory_mb is not None: + _validate_positive_finite(memory_mb, "memory_mb") + if image == "": + raise ValueError("image must not be empty.") + self._islo_conn_id = islo_conn_id + self._image = image + self._vcpus = vcpus + self._memory_mb = memory_mb + self._delete_after = delete_after + self._client: Islo | None = None + + def _get_client(self) -> Islo: + if self._client is not None: + return self._client + with _translate_islo_errors("initialize its client"): + from islo import Islo + + if self._islo_conn_id is None: + self._client = Islo() + return self._client + conn = BaseHook.get_connection(self._islo_conn_id) + api_key = (conn.password or "").strip() + if not api_key: + raise SandboxTerminalError( + f"Connection {self._islo_conn_id!r} has no password; set it to the Islo API key." + ) + kwargs: dict[str, Any] = {"api_key": api_key} + if conn.host: + kwargs["compute_url"] = conn.host + extra = conn.extra_dejson + if extra.get("base_url"): + kwargs["base_url"] = extra["base_url"] + if extra.get("timeout") is not None: + try: + request_timeout = float(extra["timeout"]) + _validate_positive_finite(request_timeout, "connection extra timeout") + except (TypeError, ValueError) as e: + raise SandboxTerminalError( + "The Islo connection extra timeout must be a positive finite number." + ) from e + kwargs["timeout"] = request_timeout + self._client = Islo(**kwargs) + return self._client + + @staticmethod + def _request_options(*, timeout: float, chunk_size: int | None = None) -> dict[str, int]: + options = {"timeout_in_seconds": max(1, math.ceil(timeout)), "max_retries": 0} Review Comment: Agreed. `_request_options` takes a `max_retries` argument now and `destroy` passes `2`, so the one call that can strand a microVM is no longer the one call that never retries. `_destroy_after_timeout` also no longer converts a failed cleanup into a task failure — it warns and leaves `delete_after` to reclaim the sandbox, matching `sbx.py`’s handling of the same path. `test_delete_keeps_the_sdk_retries` and `test_timeout_cleanup_failure_warns_and_leaves_the_ttl_to_reclaim` cover both. ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""islo.dev microVM backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import math +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxFileTooLargeError, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) +from airflow.providers.common.compat.sdk import BaseHook + +if TYPE_CHECKING: + from collections.abc import Iterator + + from islo import Islo + from islo.errors import NotFoundError + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "timeout"}) +_POLL_INITIAL = 0.2 +_POLL_MAX = 2.0 +_POLL_BACKOFF = 1.5 +_FILE_OP_TIMEOUT = 120.0 +_HELPER_OUTPUT_CAP = 1024 * 1024 +_COMMAND_WRAPPER = """\ +umask 077 +dir="${TMPDIR:-/tmp}/airflow-sandbox-$$" +mkdir "$dir" || exit 70 +trap 'rm -rf "$dir"' EXIT HUP INT TERM +mkfifo "$dir/out" "$dir/err" "$dir/out-tail" "$dir/err-tail" || exit 70 +tail -c "$2" <"$dir/out-tail" >"$dir/out-result" & +out_tail_pid=$! +tail -c "$2" <"$dir/err-tail" >"$dir/err-result" & +err_tail_pid=$! +tee "$dir/out-tail" <"$dir/out" | wc -c >"$dir/out-count" & +out_drain_pid=$! +tee "$dir/err-tail" <"$dir/err" | wc -c >"$dir/err-count" & +err_drain_pid=$! +sh -lc "$1" >"$dir/out" 2>"$dir/err" +status=$? +wait "$out_drain_pid" "$err_drain_pid" "$out_tail_pid" "$err_tail_pid" || exit 70 +if [ "$(cat "$dir/out-count")" -gt "$2" ]; then printf '1\\n'; else printf '0\\n'; fi +cat "$dir/out-result" +if [ "$(cat "$dir/err-count")" -gt "$2" ]; then printf '1\\n' >&2; else printf '0\\n' >&2; fi +cat "$dir/err-result" >&2 +exit "$status" +""" + + +@contextmanager +def _translate_islo_errors(operation: str) -> Iterator[None]: + try: + yield + except SandboxError: + raise + except Exception as e: + try: + from islo.core.api_error import ApiError + except ImportError: + raise SandboxTerminalError( + 'The Islo SDK is not installed. Install "apache-airflow-providers-common-ai[sandbox-islo]".' + ) from e + if isinstance(e, ApiError): + status = f" (HTTP {e.status_code})" if e.status_code is not None else "" + raise SandboxTerminalError(f"Islo could not {operation}{status}.") from e + raise SandboxTerminalError(f"Islo could not {operation}: {type(e).__name__}.") from e + + +def _bound_result_stream(text: str, max_bytes: int, *, server_truncated: bool) -> tuple[str, bool]: + flag, separator, payload = text.partition("\n") + if separator and flag in {"0", "1"}: + truncated = flag == "1" or server_truncated + else: + payload = text + truncated = True + + encoded = payload.encode("utf-8") + if len(encoded) > max_bytes: + payload = encoded[-max_bytes:].decode("utf-8", errors="ignore") + truncated = True + return payload, truncated + + +class IsloSandboxBackend(SandboxBackend): + """ + Sandbox backend that runs agent commands in an `islo.dev <https://islo.dev>`__ microVM. + + Islo is a hosted API with no local daemon or host-virtualization requirement, + so this backend works from an Airflow worker running in a container. + Credentials resolve lazily from an Airflow connection on first use. + + Connection fields: ``password`` is the Islo API key (required), ``host`` the + compute URL (optional), and the extra may set ``base_url`` and ``timeout`` + (request timeout in seconds). + + File reads and writes use Islo's native streaming APIs. Directory listings + and command-output bounding require common Unix command-line tools in the + sandbox image, including ``sh``, ``mkfifo``, ``tail``, ``tee`` and a ``find`` + implementation with ``-printf`` support. + + :param islo_conn_id: Airflow connection ID for Islo. ``None`` lets the SDK + resolve credentials from its own environment variables (``ISLO_API_KEY``). + :param image: Sandbox image. ``None`` (default) uses the server default. + :param vcpus: Number of virtual CPUs. ``None`` uses the server default. + :param memory_mb: Memory in MB. ``None`` uses the server default. + :param delete_after: Server-side TTL in seconds after which the sandbox is + deleted even if the worker never got to destroy it. Default ``3600``. + """ + + name = "islo" + + def __init__( + self, + islo_conn_id: str | None = "islo_default", + *, + image: str | None = None, + vcpus: int | None = None, + memory_mb: int | None = None, + delete_after: int = 3600, + ) -> None: + _validate_positive_finite(delete_after, "delete_after") + if vcpus is not None: + _validate_positive_finite(vcpus, "vcpus") + if memory_mb is not None: + _validate_positive_finite(memory_mb, "memory_mb") + if image == "": + raise ValueError("image must not be empty.") + self._islo_conn_id = islo_conn_id + self._image = image + self._vcpus = vcpus + self._memory_mb = memory_mb + self._delete_after = delete_after + self._client: Islo | None = None + + def _get_client(self) -> Islo: + if self._client is not None: + return self._client + with _translate_islo_errors("initialize its client"): + from islo import Islo + + if self._islo_conn_id is None: + self._client = Islo() + return self._client + conn = BaseHook.get_connection(self._islo_conn_id) + api_key = (conn.password or "").strip() + if not api_key: + raise SandboxTerminalError( + f"Connection {self._islo_conn_id!r} has no password; set it to the Islo API key." + ) + kwargs: dict[str, Any] = {"api_key": api_key} + if conn.host: + kwargs["compute_url"] = conn.host + extra = conn.extra_dejson + if extra.get("base_url"): + kwargs["base_url"] = extra["base_url"] + if extra.get("timeout") is not None: + try: + request_timeout = float(extra["timeout"]) + _validate_positive_finite(request_timeout, "connection extra timeout") + except (TypeError, ValueError) as e: + raise SandboxTerminalError( + "The Islo connection extra timeout must be a positive finite number." + ) from e + kwargs["timeout"] = request_timeout + self._client = Islo(**kwargs) + return self._client + + @staticmethod + def _request_options(*, timeout: float, chunk_size: int | None = None) -> dict[str, int]: + options = {"timeout_in_seconds": max(1, math.ceil(timeout)), "max_retries": 0} + if chunk_size is not None: + options["chunk_size"] = chunk_size + return options + + def create(self, *, spec: SandboxSpec | None = None) -> str: + if spec is not None and spec.allow_egress_to: + raise SandboxTerminalError( + "The Islo backend cannot apply a per-domain egress allowlist; it can only turn " + "outbound access on or off. Drop allow_egress_to, or use a backend with " + "per-domain network rules." + ) + with _translate_islo_errors("create a sandbox"): + from islo.types import LifecyclePolicy + + kwargs: dict[str, Any] = { + "internet_enabled": False if spec is None else not spec.block_network, + "lifecycle": LifecyclePolicy(delete_after=self._delete_after), + } + if self._image is not None: + kwargs["image"] = self._image + if self._vcpus is not None: + kwargs["vcpus"] = self._vcpus + if self._memory_mb is not None: + kwargs["memory_mb"] = self._memory_mb + if spec is not None and spec.env: + kwargs["env"] = dict(spec.env) Review Comment: Pinned it down against a live microVM: environment set at creation **is** visible to later `exec_in_sandbox` calls, including through the wrapper’s `sh -lc`. So the spec is honored for the sandbox’s whole life and the create-time call is enough. Left it at creation rather than moving to per-exec, since one authoritative place beats repeating it on every call, and added a comment recording that this was verified rather than assumed. `test_spec_env_is_forwarded_so_it_is_never_silently_dropped` guards the `base.py` contract, and the E2E now asserts `$SPEC_MARKER` is readable from inside the sandbox. ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""islo.dev microVM backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import math +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxFileTooLargeError, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) +from airflow.providers.common.compat.sdk import BaseHook + +if TYPE_CHECKING: + from collections.abc import Iterator + + from islo import Islo + from islo.errors import NotFoundError + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "timeout"}) Review Comment: Good catch — the SDK types `status` as a plain `str` on a model with `extra="allow"`, so the vocabulary is genuinely open. Inverted the check: `_RUNNING_EXEC_STATUSES` lists the statuses that mean "not finished yet" (`pending`, `queued`, `starting`, `running`), and anything else is terminal. An unknown value now surfaces a failure instead of polling to the deadline and destroying the sandbox. `test_an_unrecognised_status_is_terminal_rather_than_still_running` parametrises `cancelled`, `dead` and an unseen value, and `test_in_flight_statuses_keep_polling` pins the other side so the two cannot drift. Related finding from the live run: the server does not appear to enforce `timeout_secs` at all — `sleep 30` with `timeout_secs=2` ran the full 30s and reported `completed` with exit 0. The client-side polling deadline is what actually bounds a command, and that path already destroys the sandbox and reports a timeout. ########## providers/common/ai/src/airflow/providers/common/ai/sandbox/islo.py: ########## @@ -0,0 +1,397 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""islo.dev microVM backend for :class:`~airflow.providers.common.ai.toolsets.sandbox.SandboxToolset`.""" + +from __future__ import annotations + +import math +import shlex +import time +from contextlib import contextmanager, suppress +from typing import TYPE_CHECKING, Any + +from airflow.providers.common.ai.sandbox.base import ( + SandboxBackend, + SandboxError, + SandboxExecResult, + SandboxFileTooLargeError, + SandboxTerminalError, + _new_sandbox_name, + _validate_positive_finite, +) +from airflow.providers.common.compat.sdk import BaseHook + +if TYPE_CHECKING: + from collections.abc import Iterator + + from islo import Islo + from islo.errors import NotFoundError + + from airflow.providers.common.ai.sandbox.base import SandboxSpec + +_TERMINAL_EXEC_STATUSES = frozenset({"completed", "failed", "timeout"}) +_POLL_INITIAL = 0.2 +_POLL_MAX = 2.0 +_POLL_BACKOFF = 1.5 +_FILE_OP_TIMEOUT = 120.0 +_HELPER_OUTPUT_CAP = 1024 * 1024 +_COMMAND_WRAPPER = """\ +umask 077 +dir="${TMPDIR:-/tmp}/airflow-sandbox-$$" +mkdir "$dir" || exit 70 +trap 'rm -rf "$dir"' EXIT HUP INT TERM +mkfifo "$dir/out" "$dir/err" "$dir/out-tail" "$dir/err-tail" || exit 70 +tail -c "$2" <"$dir/out-tail" >"$dir/out-result" & +out_tail_pid=$! +tail -c "$2" <"$dir/err-tail" >"$dir/err-result" & +err_tail_pid=$! +tee "$dir/out-tail" <"$dir/out" | wc -c >"$dir/out-count" & +out_drain_pid=$! +tee "$dir/err-tail" <"$dir/err" | wc -c >"$dir/err-count" & +err_drain_pid=$! +sh -lc "$1" >"$dir/out" 2>"$dir/err" +status=$? +wait "$out_drain_pid" "$err_drain_pid" "$out_tail_pid" "$err_tail_pid" || exit 70 +if [ "$(cat "$dir/out-count")" -gt "$2" ]; then printf '1\\n'; else printf '0\\n'; fi +cat "$dir/out-result" +if [ "$(cat "$dir/err-count")" -gt "$2" ]; then printf '1\\n' >&2; else printf '0\\n' >&2; fi +cat "$dir/err-result" >&2 +exit "$status" +""" + + +@contextmanager +def _translate_islo_errors(operation: str) -> Iterator[None]: + try: + yield + except SandboxError: + raise + except Exception as e: + try: + from islo.core.api_error import ApiError + except ImportError: + raise SandboxTerminalError( + 'The Islo SDK is not installed. Install "apache-airflow-providers-common-ai[sandbox-islo]".' + ) from e + if isinstance(e, ApiError): + status = f" (HTTP {e.status_code})" if e.status_code is not None else "" + raise SandboxTerminalError(f"Islo could not {operation}{status}.") from e + raise SandboxTerminalError(f"Islo could not {operation}: {type(e).__name__}.") from e + + +def _bound_result_stream(text: str, max_bytes: int, *, server_truncated: bool) -> tuple[str, bool]: + flag, separator, payload = text.partition("\n") + if separator and flag in {"0", "1"}: + truncated = flag == "1" or server_truncated + else: + payload = text + truncated = True + + encoded = payload.encode("utf-8") + if len(encoded) > max_bytes: + payload = encoded[-max_bytes:].decode("utf-8", errors="ignore") + truncated = True + return payload, truncated + + +class IsloSandboxBackend(SandboxBackend): + """ + Sandbox backend that runs agent commands in an `islo.dev <https://islo.dev>`__ microVM. + + Islo is a hosted API with no local daemon or host-virtualization requirement, + so this backend works from an Airflow worker running in a container. + Credentials resolve lazily from an Airflow connection on first use. + + Connection fields: ``password`` is the Islo API key (required), ``host`` the + compute URL (optional), and the extra may set ``base_url`` and ``timeout`` + (request timeout in seconds). + + File reads and writes use Islo's native streaming APIs. Directory listings + and command-output bounding require common Unix command-line tools in the + sandbox image, including ``sh``, ``mkfifo``, ``tail``, ``tee`` and a ``find`` + implementation with ``-printf`` support. + + :param islo_conn_id: Airflow connection ID for Islo. ``None`` lets the SDK + resolve credentials from its own environment variables (``ISLO_API_KEY``). + :param image: Sandbox image. ``None`` (default) uses the server default. + :param vcpus: Number of virtual CPUs. ``None`` uses the server default. + :param memory_mb: Memory in MB. ``None`` uses the server default. + :param delete_after: Server-side TTL in seconds after which the sandbox is + deleted even if the worker never got to destroy it. Default ``3600``. + """ + + name = "islo" + + def __init__( + self, + islo_conn_id: str | None = "islo_default", + *, + image: str | None = None, + vcpus: int | None = None, + memory_mb: int | None = None, + delete_after: int = 3600, + ) -> None: + _validate_positive_finite(delete_after, "delete_after") + if vcpus is not None: + _validate_positive_finite(vcpus, "vcpus") + if memory_mb is not None: + _validate_positive_finite(memory_mb, "memory_mb") + if image == "": + raise ValueError("image must not be empty.") + self._islo_conn_id = islo_conn_id + self._image = image + self._vcpus = vcpus + self._memory_mb = memory_mb + self._delete_after = delete_after + self._client: Islo | None = None + + def _get_client(self) -> Islo: + if self._client is not None: + return self._client + with _translate_islo_errors("initialize its client"): + from islo import Islo + + if self._islo_conn_id is None: + self._client = Islo() + return self._client + conn = BaseHook.get_connection(self._islo_conn_id) + api_key = (conn.password or "").strip() + if not api_key: + raise SandboxTerminalError( + f"Connection {self._islo_conn_id!r} has no password; set it to the Islo API key." + ) + kwargs: dict[str, Any] = {"api_key": api_key} + if conn.host: + kwargs["compute_url"] = conn.host + extra = conn.extra_dejson + if extra.get("base_url"): + kwargs["base_url"] = extra["base_url"] + if extra.get("timeout") is not None: + try: + request_timeout = float(extra["timeout"]) + _validate_positive_finite(request_timeout, "connection extra timeout") + except (TypeError, ValueError) as e: + raise SandboxTerminalError( + "The Islo connection extra timeout must be a positive finite number." + ) from e + kwargs["timeout"] = request_timeout + self._client = Islo(**kwargs) + return self._client + + @staticmethod + def _request_options(*, timeout: float, chunk_size: int | None = None) -> dict[str, int]: + options = {"timeout_in_seconds": max(1, math.ceil(timeout)), "max_retries": 0} + if chunk_size is not None: + options["chunk_size"] = chunk_size + return options + + def create(self, *, spec: SandboxSpec | None = None) -> str: + if spec is not None and spec.allow_egress_to: + raise SandboxTerminalError( + "The Islo backend cannot apply a per-domain egress allowlist; it can only turn " + "outbound access on or off. Drop allow_egress_to, or use a backend with " + "per-domain network rules." + ) + with _translate_islo_errors("create a sandbox"): + from islo.types import LifecyclePolicy + + kwargs: dict[str, Any] = { + "internet_enabled": False if spec is None else not spec.block_network, + "lifecycle": LifecyclePolicy(delete_after=self._delete_after), + } + if self._image is not None: + kwargs["image"] = self._image + if self._vcpus is not None: + kwargs["vcpus"] = self._vcpus + if self._memory_mb is not None: + kwargs["memory_mb"] = self._memory_mb + if spec is not None and spec.env: + kwargs["env"] = dict(spec.env) + sandbox = self._get_client().sandboxes.create_sandbox( + name=_new_sandbox_name(), Review Comment: Fixed. The name is generated before the call and a `BaseException` handler best-effort deletes it, so a 5xx or reset after the server provisioned the microVM leaves a handle that can still reclaim it — the `sbx.py` pattern. `test_a_failed_create_deletes_the_name_it_had_already_bound` asserts the delete targets the same name that was requested, and `test_a_cleanup_failure_does_not_mask_the_original_create_error` makes sure the original error is what the caller sees. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
