[query] log

Author: agentscope-ai-java

dashscope/finetune/agentic_rl.py

@@ -43,6 +43,7 @@
 from dashscope.finetune.reinforcement.common.errors import (
     RegistrationError,
     ValidationError,
+    InstanceQueryError,
     RuntimeErrorWithCode,
     ValueErrorWithCode,
     DatasetsError,
@@ -426,7 +427,22 @@ async def test_functions(
         functype: FunctionType,
         input_data: Dict[str, Any],
         api_key: str = None,
+        pull_logs: bool = False,
+        log_page_size: int = 100,
     ):
+        """Test a deployed function instance with custom input data.
+
+        Args:
+            instance_id: Target function instance ID.
+            functype: Function type (ROLLOUT/REWARD/GROUP_REWARD).
+            input_data: Test input payload.
+            api_key: DashScope API key (uses DASHSCOPE_API_KEY env var
+                if omitted).
+            pull_logs: If True, pull all logs of the function instance
+                (with pagination) after verification and print them
+                between separator markers.
+            log_page_size: Page size used when pulling logs.
+        """
         try:
             set_api_key(api_key)
 
@@ -450,13 +466,146 @@ async def test_functions(
                 },
             )
 
-            return await AgenticRLFunctionComponent.verify_function(
+            result = await AgenticRLFunctionComponent.verify_function(
                 value,
                 instance_id,
             )
 
+            if pull_logs:
+                await cls._pull_function_instance_logs(
+                    function_instance_id=instance_id,
+                    page_size=log_page_size,
+                )
+
+            return result
+
         except Exception as e:
             raise ValidationError(
                 "Function test failed",
                 error_code=3008,
             ) from e
+
+    @classmethod
+    async def query_function_instance_logs(
+        cls,
+        function_instance_id: str,
+        page_number: int = 1,
+        page_size: int = 100,
+        start_time: Optional[int] = None,
+        end_time: Optional[int] = None,
+        keywords: Optional[List[str]] = None,
+        api_key: str = None,
+    ) -> Dict[str, Any]:
+        """Query one page of logs for a function (faas) instance.
+
+        Args:
+            function_instance_id: Target function instance ID.
+            page_number: Page number, starting from 1.
+            page_size: Number of log entries per page.
+            start_time: Optional start time filter (in seconds).
+            end_time: Optional end time filter (in seconds).
+            keywords: Optional keyword filters for log messages.
+            api_key: DashScope API key (uses DASHSCOPE_API_KEY env var
+                if omitted).
+
+        Returns:
+            Raw response dict of the log query API.
+        """
+        try:
+            set_api_key(api_key)
+
+            fc_component = AgenticRLFunctionComponent
+            return await fc_component.query_function_instance_logs(
+                function_instance_id=function_instance_id,
+                page_number=page_number,
+                page_size=page_size,
+                start_time=start_time,
+                end_time=end_time,
+                keywords=keywords,
+            )
+
+        except Exception as e:
+            if hasattr(e, "error_code"):
+                raise
+            raise InstanceQueryError(
+                "Function instance log query failed",
+                error_code=3009,
+            ) from e
+
+    @classmethod
+    async def query_all_function_instance_logs(
+        cls,
+        function_instance_id: str,
+        page_size: int = 100,
+        start_time: Optional[int] = None,
+        end_time: Optional[int] = None,
+        keywords: Optional[List[str]] = None,
+        api_key: str = None,
+    ) -> List[str]:
+        """Fetch all logs of a function (faas) instance.
+
+        Pages are fetched one by one via the paginated log query until
+        all entries are collected.
+
+        Args:
+            function_instance_id: Target function instance ID.
+            page_size: Number of log entries per page.
+            start_time: Optional start time filter (in seconds).
+            end_time: Optional end time filter (in seconds).
+            keywords: Optional keyword filters for log messages.
+            api_key: DashScope API key (uses DASHSCOPE_API_KEY env var
+                if omitted).
+
+        Returns:
+            Aggregated list of log messages across all pages.
+        """
+        try:
+            set_api_key(api_key)
+
+            fc_component = AgenticRLFunctionComponent
+            return await fc_component.query_all_function_instance_logs(
+                function_instance_id=function_instance_id,
+                page_size=page_size,
+                start_time=start_time,
+                end_time=end_time,
+                keywords=keywords,
+            )
+
+        except Exception as e:
+            if hasattr(e, "error_code"):
+                raise
+            raise InstanceQueryError(
+                "Function instance log query failed",
+                error_code=3010,
+            ) from e
+
+    @classmethod
+    async def _pull_function_instance_logs(
+        cls,
+        function_instance_id: str,
+        page_size: int = 100,
+    ) -> None:
+        """Pull all logs of a function instance and print them between
+        separator markers (best effort, never raises)."""
+        logger.info(
+            "************start query log "
+            f"(function_instance_id={function_instance_id})*******",
+        )
+        try:
+            logs = await cls.query_all_function_instance_logs(
+                function_instance_id=function_instance_id,
+                page_size=page_size,
+            )
+            for entry in logs:
+                logger.info(f"[function instance log] {entry}")
+            logger.info(
+                f"Pulled {len(logs)} log entries for function instance "
+                f"{function_instance_id}",
+            )
+        except Exception as e:
+            logger.warning(
+                f"Failed to pull logs for function instance "
+                f"{function_instance_id}: {e}",
+            )
+        finally:
+            logger.info("************end query log *******")

dashscope/finetune/reinforcement/__init__.py

@@ -21,6 +21,7 @@
     FC_PYPI_LIB,
     FC_PYPI_REPO,
     FC_QUERY_API,
+    FC_QUERY_LOG_API,
     FC_REGISTER_REWARD_API,
     FC_REGISTER_ROLLOUT_API,
     FC_REGISTER_GROUP_REWARD_API,
@@ -159,6 +160,7 @@ def __getattr__(name):
     "FC_PYPI_LIB",
     "FC_PYPI_REPO",
     "FC_QUERY_API",
+    "FC_QUERY_LOG_API",
     "FC_REGISTER_REWARD_API",
     "FC_REGISTER_ROLLOUT_API",
     "FC_REGISTER_GROUP_REWARD_API",

dashscope/finetune/reinforcement/common/constants.py

@@ -83,6 +83,10 @@ def get_int_env(env_var: str, default: int) -> int:
 )
 FC_LOAD_API = os.environ.get("FC_LOAD_API", f"{FC_BASE_ENDPOINT}/online/faas")
 FC_QUERY_API = os.environ.get("FC_QUERY_API", f"{FC_BASE_ENDPOINT}/query/faas")
+FC_QUERY_LOG_API = os.environ.get(
+    "FC_QUERY_LOG_API",
+    f"{FC_BASE_ENDPOINT}/query/faas/log",
+)
 FC_LAYER_CREATE_API = os.environ.get(  # POST
     "FC_LAYER_CREATE_API",
     f"{FC_BASE_ENDPOINT}/create/faas/layer",

dashscope/finetune/reinforcement/common/model.py

@@ -27,6 +27,7 @@
     FC_API_KEY,
     FC_LOAD_API,
     FC_QUERY_API,
+    FC_QUERY_LOG_API,
     FC_REGISTER_REWARD_API,
     FC_REGISTER_ROLLOUT_API,
     FC_REGISTER_GROUP_REWARD_API,
@@ -1056,6 +1057,165 @@ async def verify_function(
                 error_code=2047,
             ) from e
 
+    @staticmethod
+    def _extract_log_page(
+        result: Dict[str, Any],
+    ) -> Tuple[List[str], Optional[int]]:
+        """Extract log messages and total count from a log query response.
+
+        Expected response format::
+
+            {
+              "success": true,
+              "code": null,
+              "message": null,
+              "data": {
+                "pageNumber": 1,
+                "pageSize": 50,
+                "totalCount": 128,
+                "logEntries": [
+                  {"instanceId": "...", "message": "...",
+                   "timestamp": 1699051200}
+                ]
+              }
+            }
+
+        Only the "message" field of each log entry is kept; other fields
+        are ignored.
+        """
+        data = result.get("data", {})
+        if not isinstance(data, Dict):
+            return [], None
+
+        entries = data.get("logEntries") or []
+        logs = [
+            entry.get("message", "") if isinstance(entry, Dict) else str(entry)
+            for entry in entries
+        ]
+
+        total = data.get("totalCount")
+        if not isinstance(total, int):
+            total = None
+
+        return logs, total
+
+    @classmethod
+    async def query_function_instance_logs(
+        cls,
+        function_instance_id: str,
+        page_number: int = 1,
+        page_size: int = 100,
+        start_time: Optional[int] = None,
+        end_time: Optional[int] = None,
+        keywords: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        """Query logs of a function (faas) instance (single page).
+
+        Args:
+            function_instance_id: Target function instance ID.
+            page_number: Page number, starting from 1.
+            page_size: Number of log entries per page.
+            start_time: Optional start time filter (in seconds).
+            end_time: Optional end time filter (in seconds).
+            keywords: Optional keyword filters for log messages.
+
+        Returns:
+            Raw response dict of the log query API.
+        """
+        if not function_instance_id:
+            raise InputError(
+                "No function instance ID available for log query",
+                error_code=2071,
+            )
+
+        url = f"{FC_QUERY_LOG_API}/{function_instance_id}"
+        request_data: Dict[str, Any] = {
+            "pageNumber": page_number,
+            "pageSize": page_size,
+        }
+        if start_time is not None:
+            request_data["startTime"] = start_time
+        if end_time is not None:
+            request_data["endTime"] = end_time
+        if keywords:
+            request_data["keywords"] = keywords
+
+        result = await client_fc(FC_API_KEY, url, request_data)
+        status = result.get("status", {})
+        if isinstance(status, Dict) and status.get("code", 200) != 200:
+            raise InstanceQueryError(
+                f"Log query failed: {result}",
+                error_code=2072,
+                instance_id=function_instance_id,
+            )
+        if result.get("success") is False:
+            raise InstanceQueryError(
+                f"Log query failed: {result.get('message') or result}",
+                error_code=2073,
+                instance_id=function_instance_id,
+            )
+
+        logger.debug(
+            f"Log query completed | FunctionInstanceID: "
+            f"{function_instance_id} | "
+            f"Page: {page_number} | Size: {page_size}",
+        )
+        return result
+
+    @classmethod
+    async def query_all_function_instance_logs(
+        cls,
+        function_instance_id: str,
+        page_size: int = 100,
+        start_time: Optional[int] = None,
+        end_time: Optional[int] = None,
+        keywords: Optional[List[str]] = None,
+        max_pages: int = 100,
+    ) -> List[str]:
+        """Fetch all logs of a function (faas) instance with pagination.
+
+        Iterates pages starting from page 1 until all entries are
+        collected (based on the reported total count, an empty page, or
+        a short page).
+
+        Args:
+            function_instance_id: Target function instance ID.
+            page_size: Number of log entries per page.
+            start_time: Optional start time filter (in seconds).
+            end_time: Optional end time filter (in seconds).
+            keywords: Optional keyword filters for log messages.
+            max_pages: Safety limit on the number of pages to fetch.
+
+        Returns:
+            Aggregated list of log messages across all pages.
+        """
+        all_logs: List[str] = []
+        page_number = 1
+        while page_number <= max_pages:
+            result = await cls.query_function_instance_logs(
+                function_instance_id=function_instance_id,
+                page_number=page_number,
+                page_size=page_size,
+                start_time=start_time,
+                end_time=end_time,
+                keywords=keywords,
+            )
+            logs, total = cls._extract_log_page(result)
+            all_logs.extend(logs)
+
+            if total is not None and len(all_logs) >= total:
+                break
+            if len(logs) < page_size:
+                break
+            page_number += 1
+
+        logger.debug(
+            f"All logs fetched | FunctionInstanceID: "
+            f"{function_instance_id} | "
+            f"Pages: {page_number} | Entries: {len(all_logs)}",
+        )
+        return all_logs
+
 
 class RolloutFunctionComponent(AgenticRLFunctionComponent):
     """Rollout function component with type fixed as ROLLOUT."""

dashscope/finetune/reinforcement/common/utils.py

@@ -679,6 +679,7 @@ async def to_bailian_data(files: List[FileSpec]) -> List[str]:
             timeout=BAILIAN_FILE_TIMEOUT,
             retry_times=1,
         )
+        logger.info(f"File upload result: {result}")
 
         # Handle errors
         if result.get("status", {}).get("code", 200) != 200:
@@ -690,7 +691,7 @@ async def to_bailian_data(files: List[FileSpec]) -> List[str]:
         data = result.get("data", {})
         if "failed_uploads" in data and data["failed_uploads"]:
             failed_files = ", ".join(
-                [f["name"] for f in data["failed_uploads"]],
+                [f"{f.get('name')}: {f}" for f in data["failed_uploads"]],
             )
             raise OutputError(
                 f"Partial upload failed: {failed_files}",