feat: 添加流式查看进程日志的 API (#2255)

Author: xogas

apiserver/paasng/paas_wl/bk_app/processes/processes.py

@@ -297,6 +297,7 @@ def get_instance_logs(
         process_type: str,
         instance_name: str,
         previous: bool,
+        timestamps: bool = False,
         container_name: str | None = None,
         tail_lines: Optional[int] = None,
     ):
@@ -305,6 +306,7 @@ def get_instance_logs(
         :param process_type: 进程类型
         :param instance_name: 进程实例名称
         :param previous: 是否获取上一次运行的日志
+        :param timestamps: 是否在日志前添加时间戳
         :param container_name: 容器名称
         :param tail_lines: 获取日志末尾的行数
         :return: str
@@ -320,6 +322,7 @@ def get_instance_logs(
                 name=instance_name,
                 namespace=self.wl_app.namespace,
                 previous=previous,
+                timestamps=timestamps,
                 container=container_name,
                 tail_lines=tail_lines,
             )
@@ -333,6 +336,50 @@ def get_instance_logs(
 
         return ensure_text(rsp.data)
 
+    def get_instance_logs_stream(
+        self,
+        process_type: str,
+        instance_name: str,
+        container_name: str | None = None,
+        since_seconds: Optional[int] = None,
+        timestamps: bool = True,
+    ):
+        """获取进程实例日志流
+
+        :param process_type: 进程类型
+        :param instance_name: 进程实例名称
+        :param container_name: 容器名称
+        :param since_seconds: 获取日志的起始时间
+        :param timestamps: 是否在日志前添加时间戳
+        :return: Generator yielding log lines
+        :raise: InstanceNotFound when instance not found
+        """
+        if not container_name:
+            container_name = process_kmodel.get_by_type(self.wl_app, type=process_type).main_container_name
+
+        params = {
+            "name": instance_name,
+            "namespace": self.wl_app.namespace,
+            "container": container_name,
+            "follow": True,
+            "timestamps": timestamps,
+        }
+        if since_seconds:
+            params["since_seconds"] = since_seconds
+
+        k8s_client = get_client_by_app(self.wl_app)
+
+        try:
+            for line in KPod(k8s_client).get_log(**params):
+                yield ensure_text(line)
+        except ApiException as e:
+            if e.status == 400 and "previous terminated container" in json.loads(e.body)["message"]:
+                raise InstanceNotFound("Terminated container not found")
+            elif e.status == 404:
+                raise InstanceNotFound("Instance not found")
+            else:
+                raise
+
     def _list_default_specs(self, target_status: Optional[str] = None) -> list[dict]:
         """查询普通应用的进程 specs"""
         results = []

apiserver/paasng/paas_wl/bk_app/processes/serializers.py

@@ -16,12 +16,15 @@
 # to the current version of the project delivered to anyone in the future.
 
 import copy
+import re
 from dataclasses import dataclass, field
 from typing import Any, Dict, Generic, List, Optional, TypeVar
 from uuid import UUID
 
 import arrow
 import cattr
+from dateutil import parser
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from kubernetes.utils.quantity import parse_quantity
 from rest_framework import serializers
@@ -429,3 +432,27 @@ class InstanceLogQueryInputSLZ(serializers.Serializer):
     tail_lines = serializers.IntegerField(
         required=False, default=400, min_value=1, max_value=10000, help_text="获取日志的行数"
     )
+
+
+class InstanceLogStreamInputSLZ(serializers.Serializer):
+    """Serializer for instance log stream API"""
+
+    since_time = serializers.CharField(required=False, help_text="查询日志的起始时间 (UTC格式)")
+
+    def validate(self, attrs):
+        since_time = attrs.get("since_time")
+        # 如果没有传递 since_time, 则使用当前时间, 并转化为 RFC3339Nano 字符串
+        if not since_time:
+            attrs["since_time"] = timezone.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+        # 验证是否是合法的时间格式
+        try:
+            parser.parse(since_time)
+        except Exception:
+            raise ValidationError({"since_time": "format must be RFC3339Nano"})
+        # 验证 since_time 是否是 RFC3339Nano 格式
+        rfc3339nano_pattern = re.compile(r"^(\d{4}-\d{2}-\d{2})T(\d{2}:\d{2}:\d{2})\.(\d{9})Z$")
+        if not re.match(rfc3339nano_pattern, since_time):
+            raise ValidationError({"since_time": "format must be RFC3339Nano"})
+
+        return attrs

apiserver/paasng/paas_wl/bk_app/processes/urls.py

@@ -68,6 +68,13 @@
         views.InstanceManageViewSet.as_view({"get": "download_logs"}),
         name="api.instances.logs_download",
     ),
+    re_path(
+        make_app_pattern(
+            r"/processes/(?P<process_type>[\w-]+)/instances/(?P<process_instance_name>[.\w-]+)/logs/stream/$"
+        ),
+        views.InstanceManageViewSet.as_view({"get": "logs_stream"}),
+        name="api.instances.logs_stream",
+    ),
     re_path(
         make_app_pattern(r"/instances/(?P<process_instance_name>[.\w-]+)/restart/$"),
         views.InstanceManageViewSet.as_view({"put": "restart"}),

apiserver/paasng/paas_wl/bk_app/processes/views.py

@@ -21,14 +21,17 @@
 from operator import attrgetter
 from typing import Dict, Optional
 
+from dateutil import parser
 from django.http import HttpResponse, StreamingHttpResponse
 from django.shortcuts import get_object_or_404
+from django.utils import timezone
 from django.utils.translation import gettext as _
 from drf_yasg.utils import swagger_auto_schema
 from rest_framework import status
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet
+from urllib3.exceptions import ReadTimeoutError
 
 from paas_wl.bk_app.applications.constants import WlAppType
 from paas_wl.bk_app.applications.models import WlApp
@@ -46,6 +49,7 @@
     EventSerializer,
     InstanceLogDownloadInputSLZ,
     InstanceLogQueryInputSLZ,
+    InstanceLogStreamInputSLZ,
     ListProcessesQuerySLZ,
     ListWatcherRespSLZ,
     ModuleScopedData,
@@ -447,6 +451,9 @@ class InstanceManageViewSet(GenericViewSet, ApplicationCodeInPathMixin):
 
     permission_classes = [IsAuthenticated, application_perm_class(AppAction.BASIC_DEVELOP)]
 
+    # Use special negotiation class to accept "text/event-stream" content type
+    content_negotiation_class = IgnoreClientContentNegotiation
+
     # 下载日志时，默认最大下载的行数一万
     download_log_lines_limit = 10000
 
@@ -462,14 +469,21 @@ def retrieve_logs(self, request, code, module_name, environment, process_type, p
         try:
             logs = manager.get_instance_logs(
                 process_type=process_type,
+                timestamps=True,
                 instance_name=process_instance_name,
                 previous=data["previous"],
                 tail_lines=data["tail_lines"],
             )
         except InstanceNotFound:
             raise error_codes.PROCESS_INSTANCE_NOT_FOUND
 
-        return Response(status=status.HTTP_200_OK, data=logs.splitlines())
+        # 拆分时间戳和具体日志, 用于开启流式日志时流畅过渡
+        result_data = []
+        for log in logs.splitlines():
+            rfc_timestamp, log_message = log.split(" ", 1)
+            result_data.append({"timestamp": rfc_timestamp, "message": log_message})
+
+        return Response(status=status.HTTP_200_OK, data=result_data)
 
     def download_logs(self, request, code, module_name, environment, process_type, process_instance_name):
         """下载进程实例的日志"""
@@ -495,6 +509,50 @@ def download_logs(self, request, code, module_name, environment, process_type, p
         response["Content-Disposition"] = f'attachment; filename="{code}-{process_instance_name}-{log_type}-logs.txt"'
         return response
 
+    def logs_stream(self, request, code, module_name, environment, process_type, process_instance_name):
+        """实时获取进程实例的日志"""
+        env = self.get_env_via_path()
+        manager = ProcessManager(env)
+
+        slz = InstanceLogStreamInputSLZ(data=request.query_params)
+        slz.is_valid(raise_exception=True)
+        data = slz.validated_data
+
+        # 计算时间差(秒)，并加1确保不会因为精度问题漏掉日志
+        start_rfc_timestamp = data["since_time"]
+        seconds_elapsed = int((timezone.now() - parser.parse(start_rfc_timestamp)).total_seconds()) + 1
+
+        def resp():
+            yield "event: ping\n"
+            yield "data: \n\n"
+
+            try:
+                for log_line in manager.get_instance_logs_stream(
+                    process_type=process_type,
+                    instance_name=process_instance_name,
+                    since_seconds=seconds_elapsed,
+                ):
+                    # log format: "2023-01-01T01:01:01.123456789Z ..."
+                    rfc_timestamp, log_message = log_line.split(" ", 1)
+                    if rfc_timestamp <= start_rfc_timestamp:
+                        # 跳过早于请求时间的日志
+                        continue
+
+                    data = json.dumps(
+                        {
+                            "timestamp": rfc_timestamp,
+                            "message": log_message.rstrip("\n"),
+                        }
+                    )
+                    yield "event: message\ndata: {}\n\n".format(data)
+            except ReadTimeoutError as e:
+                logger.warning("Log stream read timeout: %s", e)
+            finally:
+                # 发送结束事件
+                yield "event: EOF\ndata: \n\n"
+
+        return StreamingHttpResponse(resp(), content_type="text/event-stream")
+
     def restart(self, request, code, module_name, environment, process_instance_name):
         """重启进程实例"""
         env = self.get_env_via_path()

apiserver/paasng/tests/paas_wl/bk_app/processes/test_processes.py

@@ -196,9 +196,9 @@ def test_get_instance_logs(
 
         if expected_exception:
             with pytest.raises(expected_exception):
-                ProcessManager(bk_stag_env).get_instance_logs("", instance_name, previous, "main")
+                ProcessManager(bk_stag_env).get_instance_logs("", instance_name, previous, True, "main")
         else:
-            logs = ProcessManager(bk_stag_env).get_instance_logs("", instance_name, previous, "main")
+            logs = ProcessManager(bk_stag_env).get_instance_logs("", instance_name, previous, True, "main")
             assert logs is not None
             assert isinstance(logs, str)