From 6c81baed480c9f9ded789d8134a6193cb13d8bb7 Mon Sep 17 00:00:00 2001 From: evanlowe <62918515+evanlowe@users.noreply.github.com> Date: Thu, 17 Sep 2026 18:26:30 +0800 Subject: [PATCH 1/3] feat(studio): show code upload progress and transfer speed --- tests/test_cloud.py | 159 +++++++++++++++++- tests/test_upload_progress.py | 114 +++++++++++++ veadk/integrations/ve_faas/upload_progress.py | 121 +++++++++++++ veadk/integrations/ve_faas/ve_faas.py | 30 ++-- 4 files changed, 405 insertions(+), 19 deletions(-) create mode 100644 tests/test_upload_progress.py create mode 100644 veadk/integrations/ve_faas/upload_progress.py diff --git a/tests/test_cloud.py b/tests/test_cloud.py index ac81186bc..ffbf90671 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -14,10 +14,12 @@ import os import tempfile +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path +from threading import Thread from types import SimpleNamespace from typing import Any, cast -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import ANY, AsyncMock, Mock, patch import pytest import requests @@ -416,7 +418,7 @@ def test_vefaas_code_upload_callback_uses_configured_region() -> None: upload.assert_called_once_with( url="https://example.com/upload", - data=b"archive", + data=ANY, headers={"Content-Type": "application/zip"}, timeout=(300, 300), ) @@ -456,7 +458,7 @@ def test_vefaas_large_code_upload_uses_bounded_extended_timeout() -> None: upload.assert_called_once_with( url="https://example.com/upload", - data=b"archive", + data=ANY, headers={"Content-Type": "application/zip"}, timeout=(300, 1800), ) @@ -473,6 +475,15 @@ def test_vefaas_large_code_upload_retries_one_connection_interruption() -> None: upload_address="https://example.com/upload" ) archive_size = 64 * 1024 * 1024 + 1 + sent = [] + + def send_archive(**kwargs): + body = kwargs["data"] + assert body.uploaded == 0 + sent.append(b"".join(body)) + if len(sent) == 1: + raise requests.ConnectionError() + return Mock(status_code=200) with ( patch( @@ -481,7 +492,7 @@ def test_vefaas_large_code_upload_retries_one_connection_interruption() -> None: ), patch( "veadk.integrations.ve_faas.ve_faas.requests.put", - side_effect=[requests.ConnectionError(), Mock(status_code=200)], + side_effect=send_archive, ) as upload, patch("veadk.integrations.ve_faas.ve_faas.signed_request") as callback, patch("veadk.integrations.ve_faas.ve_faas.time.sleep") as sleep, @@ -489,11 +500,84 @@ def test_vefaas_large_code_upload_retries_one_connection_interruption() -> None: service._upload_and_mount_code("function-id", ".") assert upload.call_count == 2 - assert upload.call_args_list[0] == upload.call_args_list[1] + assert sent == [b"archive", b"archive"] + assert ( + upload.call_args_list[0].kwargs["data"] is not upload.call_args.kwargs["data"] + ) sleep.assert_called_once_with(1) callback.assert_called_once() +@pytest.mark.parametrize( + "provider,region", [("volcengine", "cn-shanghai"), ("byteplus", "ap-southeast-1")] +) +@pytest.mark.parametrize("status_code", [200, 403, 503]) +def test_vefaas_progress_upload_preserves_http_body_and_checks_response( + provider: str, region: str, status_code: int, capsys: pytest.CaptureFixture[str] +) -> None: + archive = bytes(range(256)) * 1024 + b"zip trailer" + received = [] + + class UploadHandler(BaseHTTPRequestHandler): + def do_PUT(self) -> None: + received.append( + ( + dict(self.headers), + self.rfile.read(int(self.headers["Content-Length"])), + ) + ) + self.send_response(status_code) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *_args) -> None: + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), UploadHandler) + worker = Thread(target=server.serve_forever, daemon=True) + worker.start() + service = VeFaaS( + "test_access_key", "test_secret_key", region=region, provider=provider + ) + service.client = Mock() + service.client.get_code_upload_address.return_value = Mock( + upload_address=f"http://127.0.0.1:{server.server_port}/upload" + ) + try: + with ( + patch( + "veadk.integrations.ve_faas.ve_faas.zip_and_encode_folder", + return_value=(archive, len(archive), None), + ), + patch("veadk.integrations.ve_faas.ve_faas.signed_request") as callback, + ): + if status_code == 200: + service._upload_and_mount_code("function-id", ".") + callback.assert_called_once() + assert callback.call_args.kwargs["region"] == region + else: + with pytest.raises(ValueError, match=f"status code {status_code}"): + service._upload_and_mount_code("function-id", ".") + callback.assert_not_called() + finally: + server.shutdown() + server.server_close() + worker.join(timeout=5) + + assert len(received) == 1 + headers, payload = received[0] + assert payload == archive + assert headers["Content-Length"] == str(len(archive)) + assert headers["Content-Type"] == "application/zip" + assert "Transfer-Encoding" not in headers + output = capsys.readouterr().err + assert "100%" in output + assert "MB/s" in output + assert "Waiting for response" in output + assert ("Uploaded code" in output) == (status_code == 200) + assert ("Upload failed" in output) == (status_code != 200) + + def test_vefaas_code_upload_callback_uses_byteplus_host() -> None: service = VeFaaS( access_key="test_access_key", @@ -528,6 +612,71 @@ def test_vefaas_code_upload_callback_uses_byteplus_host() -> None: ) +@pytest.mark.parametrize( + "provider,region", [("volcengine", "cn-shanghai"), ("byteplus", "ap-southeast-1")] +) +@pytest.mark.parametrize("retry_succeeds", [True, False]) +def test_vefaas_partial_upload_retry_restarts_body_and_progress( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + provider: str, + region: str, + retry_succeeds: bool, +) -> None: + archive = b"x" * 150_000 + service = VeFaaS( + "test_access_key", "test_secret_key", region=region, provider=provider + ) + service.client = Mock() + service.client.get_code_upload_address.return_value = Mock( + upload_address="https://example.com/upload" + ) + monkeypatch.setattr( + "veadk.integrations.ve_faas.ve_faas._LARGE_CODE_BUNDLE_BYTES", 100_000 + ) + attempts = [] + + def send_archive(**kwargs): + body = kwargs["data"] + assert body.uploaded == 0 + attempts.append(body) + if len(attempts) == 1 or not retry_succeeds: + chunks = iter(body) + assert next(chunks) == archive[: 64 * 1024] + next(chunks) + assert body.uploaded == 64 * 1024 + raise requests.Timeout("upload timed out") + assert b"".join(body) == archive + return Mock(status_code=200) + + with ( + patch( + "veadk.integrations.ve_faas.ve_faas.zip_and_encode_folder", + return_value=(archive, len(archive), None), + ), + patch( + "veadk.integrations.ve_faas.ve_faas.requests.put", side_effect=send_archive + ), + patch("veadk.integrations.ve_faas.ve_faas.signed_request") as callback, + patch("veadk.integrations.ve_faas.ve_faas.time.sleep"), + ): + if retry_succeeds: + service._upload_and_mount_code("function-id", ".") + callback.assert_called_once() + else: + with pytest.raises(ValueError, match="upload request failed"): + service._upload_and_mount_code("function-id", ".") + callback.assert_not_called() + assert len(attempts) == 2 + output = capsys.readouterr().err + assert "(1/2)" in output + assert "(2/2)" in output + retry_line = next(line for line in output.splitlines() if "(2/2)" in line) + assert " 0%" in retry_line + assert "0.00 / 0.14 MB" in retry_line + assert ("Uploaded code" in output) == retry_succeeds + + def test_vefaas_byteplus_application_uses_configured_template() -> None: service = VeFaaS( access_key="test_access_key", diff --git a/tests/test_upload_progress.py b/tests/test_upload_progress.py new file mode 100644 index 000000000..2ff043f38 --- /dev/null +++ b/tests/test_upload_progress.py @@ -0,0 +1,114 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + +from io import StringIO + +import pytest + +from veadk.integrations.ve_faas.upload_progress import CodeUploadProgress + + +@pytest.mark.parametrize("terminal", [True, False]) +def test_upload_progress_displays_rate_size_eta_and_waits_for_response( + monkeypatch: pytest.MonkeyPatch, terminal: bool +) -> None: + output = StringIO() + monkeypatch.setattr(output, "isatty", lambda: terminal) + monkeypatch.setattr("sys.stderr", output) + clock = [100.0] + monkeypatch.setattr( + "veadk.integrations.ve_faas.upload_progress.time.monotonic", lambda: clock[0] + ) + + with CodeUploadProgress(b"x" * 1024**2) as body: + chunks = iter(body) + assert len(next(chunks)) == 64 * 1024 + assert body.uploaded == 0 # A yielded chunk has not been sent yet + clock[0] += 5 + next(chunks) + assert body.uploaded == 64 * 1024 + line = output.getvalue().split("\r" if terminal else "\n")[ + -1 if terminal else -2 + ] + assert "6%" in line + assert "0.06 / 1.00 MB" in line + assert "0.01 MB/s" in line + assert "ETA 01:15" in line + for _chunk in chunks: + clock[0] += 0.1 + assert body.uploaded == len(body) + assert "Waiting for response" in output.getvalue() + assert "Uploaded code" not in output.getvalue() + assert "Uploaded code" in output.getvalue() + assert "100%" in output.getvalue() + assert output.getvalue().endswith("\n") + assert ("\r" in output.getvalue()) == terminal + + +def test_upload_progress_throttles_redirected_logs( + monkeypatch: pytest.MonkeyPatch, +) -> None: + output = StringIO() + monkeypatch.setattr("sys.stderr", output) + monkeypatch.setattr( + "veadk.integrations.ve_faas.upload_progress.time.monotonic", lambda: 100.0 + ) + with CodeUploadProgress(b"x" * 1024**2) as body: + assert b"".join(body) == body.data + lines = output.getvalue().splitlines() + assert len(lines) == 3 + assert "Uploading code" in lines[0] + assert "Waiting for response" in lines[1] + assert "Uploaded code" in lines[2] + + +def test_upload_progress_can_replay_body_after_http_redirect( + monkeypatch: pytest.MonkeyPatch, +) -> None: + output = StringIO() + monkeypatch.setattr("sys.stderr", output) + with CodeUploadProgress(b"x" * 150_000) as body: + assert b"".join(body) == body.data + assert b"".join(body) == body.data + assert body.uploaded == len(body) + assert "200%" not in output.getvalue() + + +def test_upload_progress_output_failure_does_not_interrupt_body( + monkeypatch: pytest.MonkeyPatch, +) -> None: + output = StringIO() + monkeypatch.setattr("sys.stderr", output) + body = CodeUploadProgress(b"archive") + output.close() + with body: + assert b"".join(body) == b"archive" + + +@pytest.mark.parametrize( + "error", [RuntimeError("connection lost"), KeyboardInterrupt()] +) +def test_upload_progress_finishes_failed_line_and_preserves_error( + monkeypatch: pytest.MonkeyPatch, error: BaseException +) -> None: + output = StringIO() + monkeypatch.setattr(output, "isatty", lambda: True) + monkeypatch.setattr("sys.stderr", output) + with pytest.raises(type(error)): + with CodeUploadProgress(b"archive", 2, 2): + raise error + assert "(2/2)" in output.getvalue() + assert "Upload failed" in output.getvalue() + assert "Uploaded code" not in output.getvalue() + assert output.getvalue().endswith("\n") diff --git a/veadk/integrations/ve_faas/upload_progress.py b/veadk/integrations/ve_faas/upload_progress.py new file mode 100644 index 000000000..aaee5bb08 --- /dev/null +++ b/veadk/integrations/ve_faas/upload_progress.py @@ -0,0 +1,121 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + +import math +import shutil +import sys +import time +from collections import deque +from collections.abc import Iterator +from types import TracebackType + + +class CodeUploadProgress: + """A sized HTTP body that reports bytes sent without copying the archive. + + Advance after each yielded chunk has been sent by the HTTP client. Completion + is reported only when the caller exits successfully after checking the response. + """ + + def __init__(self, data: bytes, attempt: int = 1, attempts: int = 1) -> None: + self.data = data + self.uploaded = 0 + self.output = sys.stderr + self.terminal = self.output.isatty() + self.label = "Uploading code" + if attempts > 1: + self.label += f" ({attempt}/{attempts})" + self.samples: deque[tuple[float, int]] = deque() + self.last_render = 0.0 + self.line_width = 0 + + def __len__(self) -> int: + # requests uses this to retain Content-Length instead of chunked encoding. + return len(self.data) + + def __enter__(self) -> "CodeUploadProgress": + self.samples.append((time.monotonic(), 0)) + self._render(force=True) + return self + + def __iter__(self) -> Iterator[bytes]: + # requests can replay a PUT body when following a 307/308 redirect. + if self.uploaded: + self.uploaded = 0 + self.samples.clear() + self.samples.append((time.monotonic(), 0)) + self._render(force=True) + for offset in range(0, len(self), 64 * 1024): + chunk = self.data[offset : offset + 64 * 1024] + yield chunk + self.uploaded += len(chunk) + self._render() + self._render(status="Waiting for response", force=True) + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self._render( + status="Upload failed" if exc_type else "Uploaded code", + force=True, + final=True, + ) + + def _render( + self, *, status: str = "", force: bool = False, final: bool = False + ) -> None: + now = time.monotonic() + interval = 0.2 if self.terminal else 5.0 + if not force and now - self.last_render < interval: + return + self.last_render = now + self.samples.append((now, self.uploaded)) + while len(self.samples) > 2 and self.samples[1][0] < now - 5: + self.samples.popleft() + started, previous = self.samples[0] + speed = (self.uploaded - previous) / max(now - started, 0.001) + remaining = max(0, len(self) - self.uploaded) + eta = "--:--" + if speed > 0: + minutes, seconds = divmod(math.ceil(remaining / speed), 60) + eta = f"{minutes:02d}:{seconds:02d}" + fraction = self.uploaded / len(self) if len(self) else 0.0 + details = ( + f" {fraction:4.0%} {self.uploaded / 1024**2:.2f} /" + f" {len(self) / 1024**2:.2f} MB {speed / 1024**2:.2f} MB/s" + f" ETA {eta}" + ) + label = status or self.label + width = shutil.get_terminal_size().columns + bar_width = max(1, min(20, width - len(label) - len(details) - 4)) + filled = int(bar_width * fraction) + bar = "#" * filled + "-" * (bar_width - filled) + line = f"{label} [{bar}]{details}" + try: + if self.terminal: + print( + "\r" + line.ljust(self.line_width), + end="\n" if final else "", + file=self.output, + flush=True, + ) + self.line_width = len(line) + else: + print(line, file=self.output, flush=True) + except (OSError, ValueError): + # A closed output stream must not interrupt an otherwise valid upload. + pass diff --git a/veadk/integrations/ve_faas/ve_faas.py b/veadk/integrations/ve_faas/ve_faas.py index 58951ea3a..9aae289d4 100644 --- a/veadk/integrations/ve_faas/ve_faas.py +++ b/veadk/integrations/ve_faas/ve_faas.py @@ -36,6 +36,7 @@ import veadk.config import veadk.integrations.ve_faas as vefaas from veadk.integrations.ve_apig.ve_apig import APIGateway +from veadk.integrations.ve_faas.upload_progress import CodeUploadProgress from veadk.integrations.ve_faas.ve_faas_utils import ( signed_request, zip_and_encode_folder, @@ -313,15 +314,21 @@ def _upload_and_mount_code(self, function_id: str, path: str): response = None for attempt in range(1, attempts + 1): try: - response = requests.put( - url=upload_url, - data=code_zip_data, - headers=headers, - timeout=( - _STANDARD_CODE_UPLOAD_TIMEOUT_SECONDS, - _code_upload_timeout_seconds(code_zip_size), - ), - ) + with CodeUploadProgress(code_zip_data, attempt, attempts) as body: + response = requests.put( + url=upload_url, + data=body, + headers=headers, + timeout=( + _STANDARD_CODE_UPLOAD_TIMEOUT_SECONDS, + _code_upload_timeout_seconds(code_zip_size), + ), + ) + if not (200 <= response.status_code < 300): + raise ValueError( + "Function code upload failed with status code " + f"{response.status_code}." + ) break except (requests.ConnectionError, requests.Timeout) as upload_error: if attempt == attempts: @@ -336,11 +343,6 @@ def _upload_and_mount_code(self, function_id: str, path: str): raise ValueError("Function code upload request failed.") from None if response is None: raise ValueError("Function code upload request failed.") - if not (200 <= response.status_code < 300): - raise ValueError( - f"Function code upload failed with status code {response.status_code}." - ) - # Mount the TOS bucket to function instance res = signed_request( ak=self.ak, From 9855ea4881fe6494f58327381041919bd6295f1d Mon Sep 17 00:00:00 2001 From: evanlowe <62918515+evanlowe@users.noreply.github.com> Date: Thu, 17 Sep 2026 22:01:50 +0800 Subject: [PATCH 2/3] feat(studio): stream cloud build and deployment logs --- tests/test_release_progress.py | 290 ++++++++++++++++++ .../integrations/ve_faas/release_progress.py | 235 ++++++++++++++ veadk/integrations/ve_faas/ve_faas.py | 77 +++-- 3 files changed, 582 insertions(+), 20 deletions(-) create mode 100644 tests/test_release_progress.py create mode 100644 veadk/integrations/ve_faas/release_progress.py diff --git a/tests/test_release_progress.py b/tests/test_release_progress.py new file mode 100644 index 000000000..8d02f5021 --- /dev/null +++ b/tests/test_release_progress.py @@ -0,0 +1,290 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + +import json +from io import BytesIO + +import pytest + +from veadk.integrations.ve_faas import release_progress +from veadk.integrations.ve_faas.release_progress import ReleaseProgress +from veadk.integrations.ve_faas.ve_faas import VeFaaS + + +@pytest.fixture +def progress(monkeypatch: pytest.MonkeyPatch): + clock = [0.0] + messages = [] + monkeypatch.setattr(release_progress.time, "monotonic", lambda: clock[0]) + reporter = ReleaseProgress( + provider="volcengine", + region="cn-beijing", + app_id="app-id", + secrets=("private-access-key", "private-secret-key"), + emit=messages.append, + ) + return reporter, messages, clock + + +def test_progress_shows_stage_and_elapsed_when_logs_are_quiet(progress) -> None: + reporter, messages, clock = progress + reporter.status("deploying", 9) + reporter.logs(["[function][fn-id][install][Info] installing dependencies"]) + count = len(messages) + clock[0] = 14 + reporter.waiting() + assert len(messages) == count + clock[0] = 15 + reporter.waiting() + assert "安装依赖" in messages[-1] + assert "00:15" in messages[-1] + assert "暂无新日志" in messages[-1] + reporter.status("deploying", 9) + assert len(messages) == count + 1 + + +def test_progress_only_prints_new_lines_in_growing_and_rolling_logs(progress) -> None: + reporter, messages, _ = progress + reporter.logs(["first", "second"]) + reporter.logs(["first", "second", "third"]) + reporter.logs(["first", "second"]) # Ignore a stale shorter snapshot + reporter.logs([]) # A temporarily empty response must not erase the cursor + reporter.logs(["second", "third", "fourth"]) + reporter.logs(["second", "third", "fourth"]) + assert [line for line in messages if line.startswith("[发布日志]")] == [ + "[发布日志] first", + "[发布日志] second", + "[发布日志] third", + "[发布日志] fourth", + ] + + +def test_build_logs_retry_grow_and_refresh_signed_url_without_duplicate_output( + progress, + monkeypatch: pytest.MonkeyPatch, +) -> None: + reporter, messages, _ = progress + contents = iter( + [ + OSError("not ready"), + ("step one\npartial", False), + ("step one\npartial line\n", False), + ] + ) + urls = [] + + def read(url): + urls.append(url) + result = next(contents) + if isinstance(result, Exception): + raise result + return result + + monkeypatch.setattr(release_progress, "_read_build_log", read) + old_url = "https://build.example/step.log?signature=old-secret" + new_url = "https://build.example/step.log?signature=new-secret" + reporter.logs([old_url]) + reporter.logs([old_url]) + reporter.logs([new_url], final=True) + assert urls == [old_url, old_url, new_url] + assert messages.count("[构建日志] step one") == 1 + assert messages.count("[构建日志] partial line") == 1 + assert "[构建日志] partial" not in messages + assert "new-secret" not in "\n".join(messages) + assert "old-secret" not in "\n".join(messages) + assert sum(message.startswith("[发布日志]") for message in messages) == 1 + + +def test_progress_redacts_credentials_and_signed_queries(progress) -> None: + reporter, messages, _ = progress + reporter.logs( + [ + "private-access-key private-secret-key token=dynamic-secret", + '"api_key": "json-secret" Authorization: Bearer bearer-secret', + "https://example.com/resource?X-Tos-Signature=signed-secret", + ] + ) + output = "\n".join(messages) + for secret in ( + "private-access-key", + "private-secret-key", + "dynamic-secret", + "json-secret", + "bearer-secret", + "signed-secret", + ): + assert secret not in output + + +@pytest.mark.parametrize("content_range", ["bytes 100-125/126", ""]) +def test_build_log_reader_bounds_download_and_handles_partial_utf8( + monkeypatch: pytest.MonkeyPatch, + content_range: str, +) -> None: + requests = [] + + class Response(BytesIO): + headers = {"Content-Range": content_range} + + def open_url(request, timeout): + requests.append((request, timeout)) + return Response(b"\x80partial\ncomplete line\n") + + monkeypatch.setattr(release_progress.urllib.request, "urlopen", open_url) + text, truncated = release_progress._read_build_log("https://build.example/step.log") + if content_range: + assert text == "complete line\n" + assert not truncated + assert requests[0][0].get_header("Range") == "bytes=-262144" + assert requests[0][1] == 3 + + +def test_build_log_reader_handles_servers_ignoring_range( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class Response(BytesIO): + headers = {} + + monkeypatch.setattr( + release_progress.urllib.request, + "urlopen", + lambda *_args, **_kwargs: Response(b"x" * 300_000), + ) + text, truncated = release_progress._read_build_log("https://build.example/step.log") + assert len(text) == 256 * 1024 + assert truncated + + +@pytest.mark.parametrize("provider", ["volcengine", "byteplus"]) +def test_release_streams_current_revision_logs_and_final_lines( + monkeypatch: pytest.MonkeyPatch, + provider: str, +) -> None: + service = object.__new__(VeFaaS) + service.provider = provider + service.region = "cn-beijing" if provider == "volcengine" else "ap-southeast-1" + service.ak = "private-ak" + service.sk = "private-sk" + service.session_token = "private-sts" + monkeypatch.setattr( + service, + "_start_application_release", + lambda _: {"Result": {"RevisionNumber": 9}}, + ) + states = iter(["deploying", "deploying", "deploy_success"]) + monkeypatch.setattr( + service, + "_get_application_status", + lambda _: ( + next(states), + { + "Result": { + "CloudResource": json.dumps( + {"framework": {"url": {"system_url": "https://studio.example"}}} + ) + } + }, + ), + ) + snapshots = iter( + [["building"], ["building", "starting"], ["building", "starting", "ready"]] + ) + revisions = [] + + def logs(**kwargs): + revisions.append(kwargs["revision_number"]) + assert kwargs["timeout"] == 5 + return next(snapshots) + + monkeypatch.setattr(service, "_get_application_logs", logs) + messages = [] + delays = [] + monkeypatch.setattr( + "veadk.integrations.ve_faas.ve_faas.logger.info", + lambda message, *_: messages.append(message), + ) + monkeypatch.setattr("veadk.integrations.ve_faas.ve_faas.time.sleep", delays.append) + assert service._release_application("app-id") == "https://studio.example" + assert revisions == [9, 9, 9] + assert delays == [3, 3] + for line in ("building", "starting", "ready"): + assert sum(message.endswith("] " + line) for message in messages) == 1 + assert "https://studio.example" in messages[-1] + + +def test_release_keeps_waiting_when_logs_unavailable_and_never_reads_old_revision( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = object.__new__(VeFaaS) + monkeypatch.setattr(service, "_start_application_release", lambda _: {}) + results = iter( + [ + ("deploying", {"StableRevisionNumber": 7}), + ("deploying", {"NewRevisionNumber": 8}), + ("deploying", {"NewRevisionNumber": 8}), + ( + "deploy_success", + { + "NewRevisionNumber": 8, + "CloudResource": json.dumps( + {"framework": {"url": {"system_url": "https://studio.example"}}} + ), + }, + ), + ] + ) + monkeypatch.setattr( + service, + "_get_application_status", + lambda _: (lambda state: (state[0], {"Result": state[1]}))(next(results)), + ) + revisions = [] + + def logs(**kwargs): + revisions.append(kwargs["revision_number"]) + raise RuntimeError("permission denied") + + monkeypatch.setattr(service, "_get_application_logs", logs) + monkeypatch.setattr("veadk.integrations.ve_faas.ve_faas.time.sleep", lambda _: None) + messages = [] + monkeypatch.setattr( + "veadk.integrations.ve_faas.ve_faas.logger.info", messages.append + ) + assert service._release_application("app-id") == "https://studio.example" + assert revisions == [8, 8, 8] + assert sum("暂时无法读取云端日志" in message for message in messages) == 1 + + +def test_failed_release_preserves_cloud_failure_when_log_request_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = object.__new__(VeFaaS) + monkeypatch.setattr( + service, + "_start_application_release", + lambda _: {"Result": {"RevisionNumber": 9}}, + ) + monkeypatch.setattr( + service, + "_get_application_status", + lambda _: ("deploy_fail", {"Result": {"Message": "runtime failed to start"}}), + ) + monkeypatch.setattr( + service, + "_get_application_logs", + lambda **_: (_ for _ in ()).throw(RuntimeError("logs unavailable")), + ) + with pytest.raises(Exception, match="runtime failed to start") as error: + service._release_application("app-id") + assert "未能读取最终发布日志" in str(error.value) diff --git a/veadk/integrations/ve_faas/release_progress.py b/veadk/integrations/ve_faas/release_progress.py new file mode 100644 index 000000000..4980ef10b --- /dev/null +++ b/veadk/integrations/ve_faas/release_progress.py @@ -0,0 +1,235 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + +"""Customer-facing progress while VeFaaS builds and releases a code bundle.""" + +import re +import time +import urllib.request +from collections.abc import Callable + +_BUILD_LOG_BYTES = 256 * 1024 +_LOG_URL_PATTERN = re.compile(r"https://[^\s<>\]\"']+") + + +def extract_release_log_urls(text: str) -> list[str]: + urls = [] + for match in _LOG_URL_PATTERN.finditer(text): + url = match.group(0).rstrip(").,;") + if ".log" in url and url not in urls: + urls.append(url) + return urls + + +def redact_release_log(text: str, secrets: tuple[str, ...]) -> str: + for secret in secrets: + if secret: + text = text.replace(secret, "***") + text = re.sub(r"\x1b\[[0-?]*[ -/]*[@-~]", "", text) + text = re.sub(r"(?i)(\bbearer\s+)[a-z0-9._~+/=-]+", r"\1***", text) + text = re.sub( + r"(?i)((?:api[_-]?key|access[_-]?key|secret[_-]?key|client[_-]?secret|" + r"session[_-]?token|security[_-]?token|token|password)[\"']?\s*[:=]\s*)" + r"(?:[\"'][^\"']*[\"']|[^\s,;]+)", + r"\1***", + text, + ) + return re.sub(r"(https?://[^\s?]+)\?[^\s]+", r"\1?[REDACTED]", text) + + +def _read_build_log(url: str) -> tuple[str, bool]: + """Request a log tail and bound the response if the server ignores Range.""" + request = urllib.request.Request( + url, headers={"Range": f"bytes=-{_BUILD_LOG_BYTES}"} + ) + with urllib.request.urlopen(request, timeout=3) as response: + data = response.read(_BUILD_LOG_BYTES + 1) + content_range = response.headers.get("Content-Range", "") + partial_start = content_range.startswith( + "bytes " + ) and not content_range.startswith("bytes 0-") + truncated = len(data) > _BUILD_LOG_BYTES + data = data[:_BUILD_LOG_BYTES] + if partial_start: + # The byte range may start inside a UTF-8 character or a log line. + data = data.partition(b"\n")[2] + return data.decode("utf-8", "replace"), truncated + + +class ReleaseProgress: + def __init__( + self, + *, + provider: str, + region: str, + app_id: str, + secrets: tuple[str, ...], + emit: Callable[[str], None], + ) -> None: + self.english = provider == "byteplus" + self.secrets = secrets + self.emit = emit + self.started = time.monotonic() + self.last_output = self.started + self.stage = self.text("提交云端发布", "Submitting cloud release") + self.last_status = "" + self.revision: int | None = None + self.snapshots: dict[str, list[str]] = {} + self.links: dict[str, str] = {} + self.link_cursor = 0 + self.warnings: set[str] = set() + self.message( + self.text("开始云端构建与部署", "Starting cloud build and deployment") + + f" | {provider} / {region} | Application: {app_id}" + ) + + def text(self, chinese: str, english: str) -> str: + return english if self.english else chinese + + def elapsed(self) -> str: + minutes, seconds = divmod(int(time.monotonic() - self.started), 60) + return f"{minutes:02d}:{seconds:02d}" + + def message(self, text: str) -> None: + self.emit(redact_release_log(text, self.secrets)) + self.last_output = time.monotonic() + + def status(self, status: str, revision: int | None) -> None: + if revision is not None and revision != self.revision: + self.revision = revision + self.message(f"Revision: {revision}") + if status == self.last_status: + return + self.last_status = status + self.stage = { + "create_success": self.text( + "等待云端构建与部署", "Waiting for cloud build and deployment" + ), + "deploying": self.text("云端构建与部署", "Cloud build and deployment"), + "deploy_success": self.text("发布完成", "Deployment complete"), + "deploy_fail": self.text("发布失败", "Deployment failed"), + }.get(status, status) + self.message( + f"{self.stage} | {self.text('已用时', 'Elapsed')} {self.elapsed()}" + ) + + def waiting(self) -> None: + if time.monotonic() - self.last_output >= 15: + self.message( + f"{self.stage} | {self.text('已用时', 'Elapsed')} {self.elapsed()} | " + + self.text( + "暂无新日志，仍在等待云端处理", + "No new logs; waiting for the cloud service", + ) + ) + + def warning(self, key: str, message: str) -> None: + if key not in self.warnings: + self.warnings.add(key) + self.message(message) + + def log_error(self) -> None: + self.warning( + "control", + self.text( + "暂时无法读取云端日志，将重试读取；部署状态仍会持续更新", + "Cloud logs are temporarily unavailable; retrying while continuing to check deployment status", + ), + ) + + def _new_lines(self, key: str, lines: list[str], label: str) -> list[str]: + if not lines: + return [] + previous = [ + redact_release_log(line, self.secrets) + for line in self.snapshots.get(key, []) + ] + comparable = [redact_release_log(line, self.secrets) for line in lines] + if ( + len(comparable) < len(previous) + and previous[: len(comparable)] == comparable + ): + return [] + overlap = min(len(previous), len(lines)) + while overlap and previous[-overlap:] != comparable[:overlap]: + overlap -= 1 + self.snapshots[key] = lines + for line in lines[overlap:]: + self.message(f"[{label}] {line}") + return lines[overlap:] + + def logs(self, lines: list[str], *, final: bool = False) -> None: + self.warnings.discard("control") + flattened = [line for item in lines for line in str(item).splitlines()] + for url in extract_release_log_urls("\n".join(flattened)): + # Signed URLs may be refreshed while referring to the same log object. + key = url.partition("?")[0] + self.links[key] = url + while len(self.links) > 8: + key = next(iter(self.links)) + del self.links[key] + self.snapshots.pop(key, None) + new_lines = self._new_lines( + "control", flattened, self.text("发布日志", "Release log") + ) + for line in new_lines: + # Use the stage reported by VeFaaS, never estimate build percentages. + match = re.search( + r"\[function\]\[[^\]]+\]\[(install|build|deploy|start)\]", line + ) + if match and not final: + self.stage = { + "install": self.text("安装依赖", "Installing dependencies"), + "build": self.text("构建镜像", "Building image"), + "deploy": self.text("发布服务", "Deploying service"), + "start": self.text("启动服务", "Starting service"), + }[match.group(1)] + if not self.links: + return + keys = list(self.links) + # Fetch one build log per poll to keep deployment status checks responsive. + selected = keys if final else [keys[self.link_cursor % len(keys)]] + self.link_cursor += 1 + for key in selected: + try: + content, truncated = _read_build_log(self.links[key]) + except Exception: + # Build logs can appear later than the control-plane link. + self.warning( + key, + self.text( + "构建日志暂不可用，将继续尝试读取", + "Build log is not available yet; it will be retried", + ), + ) + continue + self.warnings.discard(key) + lines = content.splitlines() + if lines and (truncated or (not final and not content.endswith("\n"))): + lines.pop() + self._new_lines(key, lines, self.text("构建日志", "Build log")) + if truncated: + self.warning( + "truncated:" + key, + self.text( + "构建日志超出单次读取范围，完整日志可在云控制台查看", + "Build log exceeds the read limit; view the full log in the cloud console", + ), + ) + + def complete(self, url: str) -> None: + self.message( + f"{self.text('云端部署成功', 'Cloud deployment succeeded')} | " + f"{self.text('总耗时', 'Total elapsed')} {self.elapsed()} | {url}" + ) diff --git a/veadk/integrations/ve_faas/ve_faas.py b/veadk/integrations/ve_faas/ve_faas.py index 9aae289d4..e509e31be 100644 --- a/veadk/integrations/ve_faas/ve_faas.py +++ b/veadk/integrations/ve_faas/ve_faas.py @@ -14,7 +14,6 @@ import json import os -import re import shutil import tempfile import time @@ -36,6 +35,10 @@ import veadk.config import veadk.integrations.ve_faas as vefaas from veadk.integrations.ve_apig.ve_apig import APIGateway +from veadk.integrations.ve_faas.release_progress import ( + ReleaseProgress, + extract_release_log_urls as _extract_release_log_urls, +) from veadk.integrations.ve_faas.upload_progress import CodeUploadProgress from veadk.integrations.ve_faas.ve_faas_utils import ( signed_request, @@ -70,7 +73,6 @@ def _code_upload_timeout_seconds(code_zip_size: int) -> int: _APPLICATION_REVISION_LOG_MAX_BYTES = 50_000 -_RELEASE_LOG_URL_PATTERN = re.compile(r"https://[^\s<>\]\"']+") _TRANSIENT_VEFAAS_ERROR_MARKERS = ( "connection aborted", "connection error", @@ -104,18 +106,6 @@ def _is_transient_vefaas_error(error: BaseException) -> bool: return False -def _extract_release_log_urls(text: str) -> list[str]: - urls: list[str] = [] - seen: set[str] = set() - for match in _RELEASE_LOG_URL_PATTERN.finditer(text): - url = match.group(0).rstrip(").,;") - if ".log" not in url or url in seen: - continue - seen.add(url) - urls.append(url) - return urls - - def _download_release_log_url(url: str) -> str: with urllib.request.urlopen(url, timeout=30) as log_stream: return log_stream.read().decode("utf-8", "replace") @@ -291,12 +281,14 @@ def _upload_and_mount_code(self, function_id: str, path: str): path (str): Local project path. """ # Get zipped code data + logger.info("Packaging project for upload") code_zip_data, code_zip_size, error = zip_and_encode_folder(path) logger.info( f"Zipped project size: {code_zip_size / 1024 / 1024:.2f} MB", ) # Upload code to VeFaaS temp bucket + logger.info("Preparing code upload address for function %s", function_id) req = volcenginesdkvefaas.GetCodeUploadAddressRequest( function_id=function_id, content_length=code_zip_size ) @@ -344,6 +336,7 @@ def _upload_and_mount_code(self, function_id: str, path: str): if response is None: raise ValueError("Function code upload request failed.") # Mount the TOS bucket to function instance + logger.info("Code uploaded; attaching the bundle to function %s", function_id) res = signed_request( ak=self.ak, sk=self.sk, @@ -354,6 +347,7 @@ def _upload_and_mount_code(self, function_id: str, path: str): host=self._openapi_host(), ) + logger.info("Code bundle attached; ready for cloud build and deployment") return res def _create_function(self, function_name: str, path: str): @@ -463,29 +457,70 @@ def _start_application_release(self, app_id: str) -> dict[str, Any]: ) def _release_application(self, app_id: str): + progress = ReleaseProgress( + provider=getattr(self, "provider", DEFAULT_CLOUD_PROVIDER), + region=getattr(self, "region", ""), + app_id=app_id, + secrets=tuple( + getattr(self, key, "") for key in ("ak", "sk", "session_token") + ), + emit=logger.info, + ) release_response = self._start_application_release(app_id) release_revision_number = _release_revision_number(release_response) - status, full_response = self._get_application_status(app_id) - while status not in ["deploy_success", "deploy_fail"]: - time.sleep(10) + while True: status, full_response = self._get_application_status(app_id) + if release_revision_number is None: + # Do not attach an older stable revision's logs to this release. + revision = full_response.get("Result", {}).get("NewRevisionNumber") + if revision: + release_revision_number = _release_revision_number( + {"NewRevisionNumber": revision} + ) + progress.status(status, release_revision_number) + if status == "deploy_fail": + break + if release_revision_number is not None: + try: + lines = self._get_application_logs( + app_id=app_id, + revision_number=release_revision_number, + timeout=5, + ) + except Exception: + # Optional diagnostics must not abort a running deployment. + progress.log_error() + else: + progress.logs(lines, final=status == "deploy_success") + if status == "deploy_success": + break + progress.waiting() + time.sleep(3) if status == "deploy_success": cloud_resource = full_response["Result"]["CloudResource"] cloud_resource = json.loads(cloud_resource) url = cloud_resource["framework"]["url"]["system_url"] + progress.complete(url) return url else: logger.error( f"Release application failed. Application ID: {app_id}, Status: {status}" ) - raw_logs = "\n".join( - self._get_application_logs( + try: + failure_logs = self._get_application_logs( app_id=app_id, revision_number=release_revision_number, ) - ) + except Exception: + failure_logs = progress.snapshots.get("control", []) + [ + progress.text( + "未能读取最终发布日志，请检查日志权限或网络；下方保留云端失败状态", + "Final release logs could not be read; check log permissions or connectivity. Cloud failure status follows", + ) + ] + raw_logs = "\n".join(failure_logs) provider = getattr(self, "provider", DEFAULT_CLOUD_PROVIDER) log_text = _format_release_failure_text( raw_logs=raw_logs, @@ -1429,6 +1464,7 @@ def _get_application_logs( *, revision_number: int | None = None, limit: int = _APPLICATION_REVISION_LOG_MAX_BYTES, + timeout: float = 5, ) -> list[str]: if revision_number is None: _, application = self._get_application_status(app_id) @@ -1461,6 +1497,7 @@ def request_page(offset: int | None = None) -> dict[str, Any]: region=self.region, host=self._openapi_host(), session_token=self.session_token, + timeout=timeout, ) response = request_page() From 06971c8d14284922cdba1d4ade089135e0d04ede Mon Sep 17 00:00:00 2001 From: evanlowe <62918515+evanlowe@users.noreply.github.com> Date: Thu, 17 Sep 2026 22:34:07 +0800 Subject: [PATCH 3/3] fix(studio): report scheduler function deployment progress --- frontend/README.md | 11 + frontend/service/studio_scheduler/deploy.py | 198 ++++++++++------- .../studio_scheduler/deploy_progress.py | 131 ++++++++++++ .../studio_scheduler/test_deploy_progress.py | 200 ++++++++++++++++++ .../studio_scheduler/test_scheduler_deploy.py | 110 ++++++++++ .../integrations/ve_faas/release_progress.py | 66 +++--- 6 files changed, 613 insertions(+), 103 deletions(-) create mode 100644 frontend/service/studio_scheduler/deploy_progress.py create mode 100644 tests/frontend/service/studio_scheduler/test_deploy_progress.py diff --git a/frontend/README.md b/frontend/README.md index 28f28c9ea..29be374f6 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -1304,6 +1304,17 @@ schedule without waiting for Runtime execution. A separate asynchronous worker drains ready entries, invokes Runtime, and writes terminal results. The scanner, worker, and Studio BFF can therefore restart independently without losing work. +Deployment output identifies the scanner and worker separately, including each +Function name, ID, console link, dependency installation status, cloud build log, +release status, revision, and minute timer ID. Build logs are printed as they +become available, with repeated lines suppressed. Release failures include the +status message and failed-instance logs returned by VeFaaS. During quiet periods, +a progress message reports the current stage and elapsed time about every 15 +seconds, including while an SDK request is pending. Optional log retrieval uses +short timeouts and does not interrupt deployment; credentials and signed URL +queries are redacted. Volcengine uses Chinese progress messages and BytePlus uses +English messages + Duplicate timer deliveries are deduplicated with immutable run IDs and TOS conditional writes; an ETag lock prevents concurrent executions of the same task across Studio replicas or worker instances. Ready entries are deleted only diff --git a/frontend/service/studio_scheduler/deploy.py b/frontend/service/studio_scheduler/deploy.py index 2476f43da..2cb48507b 100644 --- a/frontend/service/studio_scheduler/deploy.py +++ b/frontend/service/studio_scheduler/deploy.py @@ -21,7 +21,6 @@ import shutil import tempfile import time -import urllib.request from collections.abc import Iterator from contextlib import contextmanager from functools import partial @@ -35,7 +34,7 @@ build_studio_offline_requirements, ) -from .diagnostics import sanitize_diagnostic +from .deploy_progress import FunctionDeploymentProgress _SCAN_TIMER_NAME = "veadk-studio-cronjobs-minute" _WORKER_TIMER_NAME = "veadk-studio-cronjobs-worker-minute" @@ -141,57 +140,81 @@ def _deploy_scheduler( worker_name = scheduler_worker_function_name(studio_application_name) with tempfile.TemporaryDirectory(prefix="studio_cronjob_scheduler_") as tmp: deployment_root = Path(tmp) - _stage_package(package_root, deployment_root) - function_id = _find_function_id(service, function_name) - if function_id: - service._replace_application_code_bundle( - function_id=function_id, - path=str(deployment_root), - environment_overrides=environment, - ) - else: - function_id = _create_function( + with FunctionDeploymentProgress(service, function_name) as progress: + progress.step("准备部署文件", "Preparing deployment files") + _stage_package(package_root, deployment_root) + progress.step("查找扫描器 Function", "Looking up scanner Function") + function_id = _find_function_id(service, function_name) + if function_id: + progress.function(function_id) + progress.step("更新代码并上传", "Updating and uploading code") + service._replace_application_code_bundle( + function_id=function_id, + path=str(deployment_root), + environment_overrides=environment, + ) + else: + progress.step( + "创建 Function 并上传代码", "Creating Function and uploading code" + ) + function_id = _create_function( + service, + function_name=function_name, + deployment_root=deployment_root, + role_trn=role_trn, + environment=environment, + ) + progress.function(function_id) + _install_dependencies(service, function_id, progress) + _release_function(service, function_id, progress) + progress.step("配置每分钟触发器", "Configuring minute timer") + timer_id = _ensure_minute_timer( service, - function_name=function_name, - deployment_root=deployment_root, - role_trn=role_trn, - environment=environment, + function_id, + name=_SCAN_TIMER_NAME, + phase="scan", + enable_concurrency=False, ) - _install_dependencies(service, function_id) - _release_function(service, function_id) - timer_id = _ensure_minute_timer( - service, - function_id, - name=_SCAN_TIMER_NAME, - phase="scan", - enable_concurrency=False, - ) - - worker_function_id = _find_function_id(service, worker_name) - if worker_function_id: - _require_async_worker(service, worker_function_id, worker_name) - service._replace_application_code_bundle( - function_id=worker_function_id, - path=str(deployment_root), - environment_overrides=environment, - ) - else: - worker_function_id = _create_async_worker_function( + progress.ready(timer_id) + + with FunctionDeploymentProgress(service, worker_name, worker=True) as progress: + progress.step("查找执行器 Function", "Looking up worker Function") + worker_function_id = _find_function_id(service, worker_name) + if worker_function_id: + progress.function(worker_function_id) + progress.step( + "检查执行器配置并更新代码", + "Checking worker configuration and updating code", + ) + _require_async_worker(service, worker_function_id, worker_name) + service._replace_application_code_bundle( + function_id=worker_function_id, + path=str(deployment_root), + environment_overrides=environment, + ) + else: + progress.step( + "创建 Function 并上传代码", "Creating Function and uploading code" + ) + worker_function_id = _create_async_worker_function( + service, + function_name=worker_name, + deployment_root=deployment_root, + role_trn=role_trn, + environment=environment, + ) + progress.function(worker_function_id) + _install_dependencies(service, worker_function_id, progress) + _release_function(service, worker_function_id, progress) + progress.step("配置每分钟触发器", "Configuring minute timer") + worker_timer_id = _ensure_minute_timer( service, - function_name=worker_name, - deployment_root=deployment_root, - role_trn=role_trn, - environment=environment, + worker_function_id, + name=_WORKER_TIMER_NAME, + phase="execute", + enable_concurrency=True, ) - _install_dependencies(service, worker_function_id) - _release_function(service, worker_function_id) - worker_timer_id = _ensure_minute_timer( - service, - worker_function_id, - name=_WORKER_TIMER_NAME, - phase="execute", - enable_concurrency=True, - ) + progress.ready(worker_timer_id) return function_id, timer_id, worker_function_id, worker_timer_id @@ -409,28 +432,33 @@ def _find_function_id(service: Any, function_name: str) -> str: return str(getattr(matches[0], "id", "") or "") if matches else "" -def _release_function(service: Any, function_id: str) -> None: +def _release_function( + service: Any, function_id: str, progress: FunctionDeploymentProgress +) -> None: from volcenginesdkvefaas import GetReleaseStatusRequest, ReleaseRequest + progress.step("提交 Function 发布", "Submitting Function release") service.client.release(ReleaseRequest(function_id=function_id, revision_number=0)) + progress.step("等待云端发布", "Waiting for cloud deployment") for _ in range(120): response = service.client.get_release_status( GetReleaseStatusRequest(function_id=function_id) ) state = str(getattr(response, "status", "") or "").lower() + progress.release_status(response) if "succ" in state or state == "done": return if "fail" in state or "error" in state: - detail = sanitize_diagnostic( + detail = progress.detail( " ".join( str(value or "").strip() for value in ( getattr(response, "error_code", ""), getattr(response, "status_message", ""), + getattr(response, "failed_instance_logs", ""), ) if value ), - limit=2_000, ) suffix = f". {detail}" if detail else "" raise RuntimeError(f"Scheduler function release failed: {state}{suffix}") @@ -438,44 +466,62 @@ def _release_function(service: Any, function_id: str) -> None: raise RuntimeError("Scheduler function release did not finish in 10 minutes") -def _install_dependencies(service: Any, function_id: str) -> None: +def _dependency_install_logs( + service: Any, function_id: str, progress: FunctionDeploymentProgress, *, final: bool +) -> str: + from volcenginesdkvefaas import GetDependencyInstallTaskLogDownloadURIRequest + + try: + method = service.client.get_dependency_install_task_log_download_uri + kwargs = {"_request_timeout": 5} if _accepts_request_timeout(method) else {} + response = method( + GetDependencyInstallTaskLogDownloadURIRequest(function_id=function_id), + **kwargs, + ) + url = str(getattr(response, "download_url", "") or "").strip() + if not url: + progress.log_error(final=final) + return "" + progress.warnings.discard("control") + return progress.build_log(url, final=final) + except Exception: # noqa: BLE001 - optional diagnostics must not fail deployment + progress.log_error(final=final) + return "" + + +def _install_dependencies( + service: Any, function_id: str, progress: FunctionDeploymentProgress +) -> None: from volcenginesdkvefaas import ( CreateDependencyInstallTaskRequest, - GetDependencyInstallTaskLogDownloadURIRequest, GetDependencyInstallTaskStatusRequest, ) + progress.step("提交依赖安装任务", "Submitting dependency installation") service.client.create_dependency_install_task( CreateDependencyInstallTaskRequest(function_id=function_id) ) + progress.step("等待云端安装依赖", "Waiting for cloud dependency installation") + detail = "" for _ in range(120): response = service.client.get_dependency_install_task_status( GetDependencyInstallTaskStatusRequest(function_id=function_id) ) state = str(getattr(response, "status", "") or "").lower() - if "succ" in state or state == "done": + progress.step( + f"依赖安装状态: {state or 'unknown'}", + f"Dependency installation status: {state or 'unknown'}", + ) + success = "succ" in state or state == "done" + failed = "fail" in state or "error" in state + log = _dependency_install_logs( + service, function_id, progress, final=success or failed + ) + if log: + detail = progress.detail(log[-2_000:]) + if success: return - if "fail" in state or "error" in state: - detail = "" - try: - log_response = ( - service.client.get_dependency_install_task_log_download_uri( - GetDependencyInstallTaskLogDownloadURIRequest( - function_id=function_id - ) - ) - ) - download_url = str( - getattr(log_response, "download_url", "") or "" - ).strip() - if download_url: - with urllib.request.urlopen(download_url, timeout=30) as log_stream: - detail = sanitize_diagnostic( - log_stream.read().decode("utf-8", "replace"), - limit=2_000, - ) - except Exception: # noqa: BLE001 - diagnostics must not mask failure - detail = "" + if failed: suffix = f". {detail}" if detail else "" raise RuntimeError( f"Scheduler dependency installation failed: {state}{suffix}" diff --git a/frontend/service/studio_scheduler/deploy_progress.py b/frontend/service/studio_scheduler/deploy_progress.py new file mode 100644 index 000000000..e8edeb51f --- /dev/null +++ b/frontend/service/studio_scheduler/deploy_progress.py @@ -0,0 +1,131 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + +"""Progress for the scheduler's standalone Function deployments.""" + +from __future__ import annotations + +from threading import Event, Thread +from types import TracebackType +from typing import Any + +from veadk.integrations.ve_faas.release_progress import ( + ReleaseProgress, + redact_release_log, +) +from veadk.utils.logger import get_logger + +from .diagnostics import sanitize_diagnostic + +logger = get_logger(__name__) +_HEARTBEAT_INTERVAL_SECONDS = 15 + + +class FunctionDeploymentProgress(ReleaseProgress): + def __init__(self, service: Any, name: str, *, worker: bool = False) -> None: + provider = getattr(service, "provider", "volcengine") + self.region = getattr(service, "region", "") + label = ( + ("Worker" if worker else "Scanner") + if provider == "byteplus" + else ("任务执行器" if worker else "定时扫描器") + ) + secrets = tuple( + str(getattr(service, key, "") or "") + for key in ("ak", "sk", "session_token") + ) + super().__init__( + provider=provider, + region=self.region, + app_id=name, + resource_label="Function", + secrets=secrets, + emit=lambda message: logger.info( + "[%s] %s", label, sanitize_diagnostic(message, secrets=secrets) + ), + ) + self._stop = Event() + self._heartbeat = Thread( + target=self._keep_waiting, name="studio-function-progress", daemon=True + ) + + def __enter__(self) -> FunctionDeploymentProgress: + self._heartbeat.start() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self._stop.set() + self._heartbeat.join() + if exc_type is not None: + self.message( + self.text("部署未完成", "Deployment did not complete") + + f" | {self.stage} | {self.text('已用时', 'Elapsed')} {self.elapsed()}" + ) + + def _keep_waiting(self) -> None: + # No network calls here: even a blocking SDK mutation needs a heartbeat. + while not self._stop.wait(min(1, _HEARTBEAT_INTERVAL_SECONDS)): + self.waiting(_HEARTBEAT_INTERVAL_SECONDS) + + def step(self, chinese: str, english: str) -> None: + self.status(self.text(chinese, english), None) + + def function(self, function_id: str) -> None: + host = "console.byteplus.com" if self.english else "console.volcengine.com" + self.message( + f"Function ID: {function_id} | " + f"https://{host}/vefaas/region:vefaas+{self.region}/function/detail/{function_id}" + ) + + def release_status(self, response: Any) -> None: + state = str(getattr(response, "status", "") or "Unknown") + revision = getattr(response, "new_revision_number", None) + self.status(self.text("发布状态", "Release status") + f": {state}", revision) + lines = [ + self.detail(getattr(response, field, "")) + for field in ("status_message", "error_code", "failed_instance_logs") + if getattr(response, field, None) + ] + # Function status responses contain snapshots, not Application log URLs. + self._new_lines("release-status", lines, self.text("发布日志", "Release log")) + + def detail(self, value: Any) -> str: + return sanitize_diagnostic( + redact_release_log(str(value or ""), self.secrets), + secrets=self.secrets, + limit=2_000, + ) + + def log_error(self, *, final: bool = False) -> None: + if not final: + super().log_error() + return + self.warning( + "final-dependency-log", + self.text( + "未能读取最终依赖安装日志，可在 Function 控制台查看", + "Final dependency installation logs could not be read; check the Function console", + ), + ) + + def ready(self, timer_id: str) -> None: + self.message( + self.text("部署完成，每分钟触发", "Deployment complete; runs every minute") + + f" | Timer ID: {timer_id} | {self.text('总耗时', 'Total elapsed')} {self.elapsed()}" + ) diff --git a/tests/frontend/service/studio_scheduler/test_deploy_progress.py b/tests/frontend/service/studio_scheduler/test_deploy_progress.py new file mode 100644 index 000000000..8a27b1187 --- /dev/null +++ b/tests/frontend/service/studio_scheduler/test_deploy_progress.py @@ -0,0 +1,200 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed 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. + +from threading import Event, enumerate as enumerate_threads +from types import SimpleNamespace +from typing import Any + +import pytest + +from frontend.service.studio_scheduler.deploy import ( + _extended_vefaas_request_timeout, + _install_dependencies, + _release_function, +) +from frontend.service.studio_scheduler.deploy_progress import FunctionDeploymentProgress + + +@pytest.fixture +def messages(monkeypatch: pytest.MonkeyPatch) -> list[str]: + lines: list[str] = [] + monkeypatch.setattr( + "frontend.service.studio_scheduler.deploy_progress.logger.info", + lambda message, *args: lines.append(message % args if args else message), + ) + return lines + + +@pytest.fixture +def service() -> Any: + return SimpleNamespace( + provider="byteplus", + region="ap-southeast-1", + ak="private-access-key", + sk="private-secret-key", + session_token="private-session-token", + client=SimpleNamespace(), + ) + + +@pytest.mark.parametrize("operation", ["install", "release"]) +@pytest.mark.parametrize("error", [None, RuntimeError, KeyboardInterrupt]) +def test_blocking_sdk_call_keeps_printing_and_stops_heartbeat( + service: Any, + monkeypatch: pytest.MonkeyPatch, + operation: str, + error: type[BaseException] | None, +) -> None: + messages: list[str] = [] + waiting = Event() + + def emit(message: str, *args: Any) -> None: + text = message % args if args else message + messages.append(text) + if "No new logs" in text: + waiting.set() + + def blocking_request(_: Any) -> None: + # The heartbeat must arrive before this synchronous SDK request returns. + assert waiting.wait(2), "No progress while the SDK request is blocked" + if error: + raise error("interrupted request") + + monkeypatch.setattr( + "frontend.service.studio_scheduler.deploy_progress.logger.info", emit + ) + monkeypatch.setattr( + "frontend.service.studio_scheduler.deploy_progress._HEARTBEAT_INTERVAL_SECONDS", + 0.01, + ) + service.client.release = blocking_request + service.client.create_dependency_install_task = blocking_request + service.client.get_release_status = lambda _: SimpleNamespace(status="Success") + service.client.get_dependency_install_task_status = lambda _: SimpleNamespace( + status="Success" + ) + before = set(enumerate_threads()) + + def deploy() -> None: + with FunctionDeploymentProgress( + service, "test-worker", worker=True + ) as progress: + progress.function("fn-1") + action = ( + _install_dependencies if operation == "install" else _release_function + ) + action(service, "fn-1", progress) + + if error: + with pytest.raises(error): + deploy() + else: + deploy() + assert waiting.is_set() + assert not any( + thread not in before and thread.name == "studio-function-progress" + for thread in enumerate_threads() + ) + heartbeat = next(message for message in messages if "No new logs" in message) + assert "[Worker]" in heartbeat + assert "Submitting" in heartbeat and "Elapsed" in heartbeat + assert any("console.byteplus.com" in message for message in messages) + + +def test_release_failure_prints_status_and_sanitized_instance_logs( + service: Any, + messages: list[str], +) -> None: + service.client.release = lambda _: None + service.client.get_release_status = lambda _: SimpleNamespace( + status="Failed", + new_revision_number=3, + error_code="StartFailed", + status_message="Health check failed", + failed_instance_logs="ImportError: missing_module private-secret-key token=private-token https://logs.example/x?signature=private-signature", + ) + with pytest.raises(RuntimeError, match="ImportError: missing_module") as error: + with FunctionDeploymentProgress(service, "worker", worker=True) as progress: + _release_function(service, "fn-1", progress) + output = "\n".join(messages) + str(error.value) + for text in ( + "Health check failed", + "StartFailed", + "Revision: 3", + "Deployment did not complete", + ): + assert text in output + for secret in ("private-secret-key", "private-token", "private-signature"): + assert secret not in output + + +def test_dependency_failure_retains_build_tail_and_redacts_secrets( + service: Any, + monkeypatch: pytest.MonkeyPatch, + messages: list[str], +) -> None: + service.client.create_dependency_install_task = lambda _: None + service.client.get_dependency_install_task_status = lambda _: SimpleNamespace( + status="Failed" + ) + service.client.get_dependency_install_task_log_download_uri = lambda _: ( + SimpleNamespace( + download_url="https://logs.example/download?signature=private-signature" + ) + ) + monkeypatch.setattr( + "veadk.integrations.ve_faas.release_progress._read_build_log", + lambda _: ( + "No matching distribution found\nprivate-access-key password=private-password", + False, + ), + ) + with pytest.raises(RuntimeError, match="No matching distribution found") as error: + with FunctionDeploymentProgress(service, "scanner") as progress: + _install_dependencies(service, "fn-1", progress) + output = "\n".join(messages) + str(error.value) + assert "[Build log] No matching distribution found" in output + for secret in ("private-access-key", "private-password", "private-signature"): + assert secret not in output + + +def test_optional_log_requests_use_short_timeout_and_do_not_abort_installation( + service: Any, + monkeypatch: pytest.MonkeyPatch, + messages: list[str], +) -> None: + states = iter(["Running", "Running", "Success"]) + timeouts: list[int] = [] + + def unavailable(_: Any, **kwargs: Any) -> Any: + timeouts.append(kwargs["_request_timeout"]) + raise TimeoutError("private-secret-key") + + service.client.create_dependency_install_task = lambda _: None + service.client.get_dependency_install_task_status = lambda _: SimpleNamespace( + status=next(states) + ) + service.client.get_dependency_install_task_log_download_uri = unavailable + monkeypatch.setattr( + "frontend.service.studio_scheduler.deploy.time.sleep", lambda _: None + ) + with _extended_vefaas_request_timeout(service): + with FunctionDeploymentProgress(service, "scanner") as progress: + _install_dependencies(service, "fn-1", progress) + output = "\n".join(messages) + assert timeouts == [5, 5, 5] + assert output.count("Cloud logs are temporarily unavailable") == 1 + assert "success" in output + assert "private-secret-key" not in output + assert service.client.get_dependency_install_task_log_download_uri is unavailable diff --git a/tests/frontend/service/studio_scheduler/test_scheduler_deploy.py b/tests/frontend/service/studio_scheduler/test_scheduler_deploy.py index 4572b9d7c..21a7c0a3d 100644 --- a/tests/frontend/service/studio_scheduler/test_scheduler_deploy.py +++ b/tests/frontend/service/studio_scheduler/test_scheduler_deploy.py @@ -15,6 +15,7 @@ from __future__ import annotations from hashlib import sha256 +from io import BytesIO from pathlib import Path from types import SimpleNamespace from typing import Any @@ -65,6 +66,8 @@ def create_function(self, request: Any) -> Any: class _Service: def __init__(self) -> None: + self.provider = "volcengine" + self.region = "cn-beijing" self.client = _Client() self.created_bundle: Path | None = None @@ -185,6 +188,113 @@ def test_deploy_extends_vefaas_sdk_request_timeout(tmp_path: Path) -> None: assert service.client.list_functions == original_list_functions +@pytest.mark.parametrize("provider", ["volcengine", "byteplus"]) +@pytest.mark.parametrize("existing", [False, True]) +def test_scheduler_deploy_prints_both_functions_and_cloud_logs( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + provider: str, + existing: bool, +) -> None: + (tmp_path / "requirements.txt").write_text("veadk-python\n", encoding="utf-8") + service = _Service() + service.provider = provider + service.region = "cn-beijing" if provider == "volcengine" else "ap-southeast-1" + counters: dict[str, int] = {} + messages: list[str] = [] + monkeypatch.setattr( + "frontend.service.studio_scheduler.deploy_progress.logger.info", + lambda message, *args: messages.append(message % args if args else message), + ) + if existing: + monkeypatch.setattr( + service.client, + "list_functions", + lambda _: SimpleNamespace( + total=2, + items=[ + SimpleNamespace(name="studio-test-cronjobs", id="function-1"), + SimpleNamespace( + name="studio-test-cronjobs-worker", id="worker-function-1" + ), + ], + ), + ) + monkeypatch.setattr( + service.client, + "get_function", + lambda _: SimpleNamespace( + async_task_config=SimpleNamespace(enable_async_task=True), + ), + raising=False, + ) + monkeypatch.setattr( + service, "_replace_application_code_bundle", lambda **_: None, raising=False + ) + + def install_status(request: Any) -> Any: + key = request.function_id + counters[key] = counters.get(key, 0) + 1 + return SimpleNamespace(status="Running" if counters[key] == 1 else "Success") + + monkeypatch.setattr( + service.client, "get_dependency_install_task_status", install_status + ) + monkeypatch.setattr( + service.client, + "get_dependency_install_task_log_download_uri", + lambda request: SimpleNamespace( + download_url=f"https://logs.example/{request.function_id}.log?signature=private-signature" + ), + raising=False, + ) + monkeypatch.setattr( + service.client, + "get_release_status", + lambda _: SimpleNamespace( + status="Success", + status_message="Health check passed", + new_revision_number=2, + ), + ) + + class LogResponse(BytesIO): + headers: dict[str, str] = {} + + monkeypatch.setattr( + "urllib.request.urlopen", + lambda *_args, **_kwargs: LogResponse( + b"Collecting scheduler-dependency\nInstalling collected packages\n" + ), + ) + monkeypatch.setattr( + "frontend.service.studio_scheduler.deploy.time.sleep", lambda _: None + ) + deploy_scheduler( + service, + studio_application_name="studio_test", + package_root=tmp_path, + role_trn="trn:iam::role/studio", + environment={"VEADK_STUDIO_TOS_BUCKET": "studio"}, + ) + + output = "\n".join(messages) + assert "studio-test-cronjobs" in output + assert "studio-test-cronjobs-worker" in output + assert "function-1" in output + assert "worker-function-1" in output + assert output.count("Collecting scheduler-dependency") == 2 + assert output.count("Installing collected packages") == 2 + assert "Health check passed" in output + assert "timer-1" in output and "timer-2" in output + assert "private-signature" not in output + assert ( + ("Scanner" in output and "Worker" in output) + if provider == "byteplus" + else ("定时扫描器" in output and "任务执行器" in output) + ) + + def test_stage_package_preserves_offline_runtime_dependencies( tmp_path: Path, ) -> None: diff --git a/veadk/integrations/ve_faas/release_progress.py b/veadk/integrations/ve_faas/release_progress.py index 4980ef10b..6c0554785 100644 --- a/veadk/integrations/ve_faas/release_progress.py +++ b/veadk/integrations/ve_faas/release_progress.py @@ -76,6 +76,7 @@ def __init__( app_id: str, secrets: tuple[str, ...], emit: Callable[[str], None], + resource_label: str = "Application", ) -> None: self.english = provider == "byteplus" self.secrets = secrets @@ -91,7 +92,7 @@ def __init__( self.warnings: set[str] = set() self.message( self.text("开始云端构建与部署", "Starting cloud build and deployment") - + f" | {provider} / {region} | Application: {app_id}" + + f" | {provider} / {region} | {resource_label}: {app_id}" ) def text(self, chinese: str, english: str) -> str: @@ -124,8 +125,8 @@ def status(self, status: str, revision: int | None) -> None: f"{self.stage} | {self.text('已用时', 'Elapsed')} {self.elapsed()}" ) - def waiting(self) -> None: - if time.monotonic() - self.last_output >= 15: + def waiting(self, interval: float = 15) -> None: + if time.monotonic() - self.last_output >= interval: self.message( f"{self.stage} | {self.text('已用时', 'Elapsed')} {self.elapsed()} | " + self.text( @@ -202,31 +203,42 @@ def logs(self, lines: list[str], *, final: bool = False) -> None: selected = keys if final else [keys[self.link_cursor % len(keys)]] self.link_cursor += 1 for key in selected: - try: - content, truncated = _read_build_log(self.links[key]) - except Exception: - # Build logs can appear later than the control-plane link. - self.warning( - key, - self.text( - "构建日志暂不可用，将继续尝试读取", - "Build log is not available yet; it will be retried", - ), - ) - continue - self.warnings.discard(key) - lines = content.splitlines() - if lines and (truncated or (not final and not content.endswith("\n"))): - lines.pop() - self._new_lines(key, lines, self.text("构建日志", "Build log")) - if truncated: - self.warning( - "truncated:" + key, - self.text( - "构建日志超出单次读取范围，完整日志可在云控制台查看", - "Build log exceeds the read limit; view the full log in the cloud console", - ), + self.build_log(self.links[key], final=final) + + def build_log(self, url: str, *, final: bool = False) -> str: + """Read a known build log URL, including URLs without a .log suffix.""" + key = url.partition("?")[0] + try: + content, truncated = _read_build_log(url) + except Exception: + # Build logs can appear later than the control-plane link. + self.warning( + "final:" + key if final else key, + self.text( + "未能读取最终构建日志，可在云控制台查看", + "Final build log could not be read; check the cloud console", ) + if final + else self.text( + "构建日志暂不可用，将继续尝试读取", + "Build log is not available yet; it will be retried", + ), + ) + return "" + self.warnings.discard(key) + lines = content.splitlines() + if lines and (truncated or (not final and not content.endswith("\n"))): + lines.pop() + self._new_lines(key, lines, self.text("构建日志", "Build log")) + if truncated: + self.warning( + "truncated:" + key, + self.text( + "构建日志超出单次读取范围，完整日志可在云控制台查看", + "Build log exceeds the read limit; view the full log in the cloud console", + ), + ) + return redact_release_log("\n".join(lines), self.secrets) def complete(self, url: str) -> None: self.message(