16 KiB

Raw Permalink Blame History

自定义诊断功能指南

1. 诊断流程

2. 新增一个诊断

2.1 新增诊断菜单

首先需要在指定路径下配置新诊断功能的菜单名称

（未部署）源码路径：sysom/sysom_web/public/resource/diagnose/v1/locales.json
（已部署）部署路径：/usr/local/sysom/web/resource/diagnose/v1/locales.json

修改配置文件后，刷新网页前端即可看到菜单栏已更新（如果没有更新可以清一下浏览器缓存）

2.2 诊断页面配置

在对应目录下新增诊断页面配置，定义诊断所需的参数、诊断结果的数据格式以及诊断结果该如何渲染：

注意：诊断页面配置定义了诊断结果的渲染方式和数据格式，后处理脚本返回的结果要和此处匹配

每个面板（pannel）都有对应的数据格式要求，详情请看 docs/diagnosis_center.md

（未部署）源码路径：sysom/sysom_web/public/resource/diagnose/v1/custom/demo.json
（已部署）部署路径：/usr/local/sysom/web/resource/diagnose/v1/custom/demo.json

修改配置文件后，刷新网页前端即可看到刚才定义的表单配置已经生效了（如果没有更新可以清一下浏览器缓存）

2.3 定义前处理脚本

在指定目录下定义诊断的前处理脚本（前处理脚本主要负责将输入的参数转换成实际要执行的诊断命令）：

（未部署）源码路径：sysom/sysom_server/sysom_diagnosis/service_scripts/demo_pre.py
（已部署）部署路径：/usr/local/sysom/server/sysom_diagnosis/service_scripts/demo_pre.py

2.3.1 前处理脚本的作用

前处理脚本主要的作用是将前端输入的参数转化成一个或多个在 Node 节点执行的命令，比如：

大多数诊断的前处理脚本只需要返回一个在指定Node执行的命令即可，部分诊断可以返回对一组节点执行命令

{
  "service_name": "loadtask",
  "instance": "127.0.0.1"
}
           ↓
127.0.0.1 => sysak loadtask -s -g >> /dev/null && cat /var/log/sysak/loadtask/.tmplog

2.3.2 基于Python编写前处理脚本（建议）

所有的前处理脚本都应该继承 DiagnosisPreProcessor 类，并实现对应的抽象方法：

class DiagnosisPreProcessor(DiagnosisProcessorBase):
    """Pre-processor used to perform: <parms> -> <diagnosis cmd>"""

    ...

    @abstractmethod
    def get_diagnosis_cmds(self, params: dict) -> DiagnosisTask:
        """Convert params to diagnosis cmds

        params => { "instance": "127.0.0.1", "service_name": "xxx", "time": 15 }

        cmds => [
            DiagnosisJob(instance="127.0.0.1", command="sysak memleak"),
            DiagnosisJob(instance="192.168.0.1", command="sysak memleak")
        ]

        Args:
            params (dict): Diagnosis parameters

        Returns:
            DiagnosisTask: Diagnosis task
        """

诊断工具开发者可以在前处理脚本中复写 get_diagnosis_cmds 来实现从前端参数 params 到诊断任务 DiagnosisTask 的转换，其中 DiagnosisTask 的结构如下：

class FileItem:
    def __init__(self, name: str, remote_path: str, local_path: str = "") -> None:
        self.name = name
        self.remote_path = remote_path
        self.local_path = local_path

class DiagnosisJob:
  	"""每个 DiagnosisJob 代表在一个指定的节点上执行一条命令"""
    def __init__(self, instance: str, cmd: str, fetch_file_list: List[FileItem] = []) -> None:
      	# 目标节点实例
        self.instance = instance
        # 要在目标节点执行的命令
        self.cmd = cmd
        # 指定诊断结束后，需要从目标 instance 下载那些文件回来
        self.fetch_file_list = fetch_file_list

class DiagnosisTask:
    """Diagnosis Task
		每个 DiagnosisTask 代表一组需要执行的 DiagnosisJob，并且可以使用
		in_order 参数决定是同步还是异步执行多个 DiagnosisJob
    Args:
        jobs([DiagnosisJob]): Diagnosis jobs
        in_order(bool): Whether to execute all jobs in order
                        False => Concurrent execution of all Jobs
                        True => Execute each job in turn
        offline_mode(bool): Whether to execute in offline mode
        offline_results([str]): offline mode diagnosis result
    """

    def __init__(
        self,
        jobs: List[DiagnosisJob] = [],
        in_order: bool = False,
        offline_mode: bool = False,
        offline_results: List[str] = [],
    ) -> None:
        self.jobs = jobs
        self.in_order = in_order
        self.offline_mode = offline_mode
        self.offline_results = offline_results

本指南的 demo 诊断的前处理脚本实现如下：

注意：后处理脚本中后处理类的名字必须为 PreProcessor

from .base import DiagnosisJob, DiagnosisPreProcessor, DiagnosisTask


class PreProcessor(DiagnosisPreProcessor):
    """Command diagnosis

    在目标节点执行 sar 命令采集CPU占用率

    Args:
        DiagnosisPreProcessor (_type_): _description_
    """

    def get_diagnosis_cmds(self, params: dict) -> DiagnosisTask:
        command = params.get("command", "")
        instance = params.get("instance", None)
        if instance is None:
            raise Exception("Missing required params: instance")
        interval = params.get("interval", 1)
        samples = params.get("samples", 10)
        command = f"sar -u -t {interval} {samples} | grep all"
        command += f' | grep "Average" > /tmp/demo.log && cat /tmp/demo.log'
        return DiagnosisTask(
            jobs=[
                DiagnosisJob(instance=instance, cmd=command)
            ],
            in_order=False,
        )

2.3.3 使用任意语言编写（不建议）

SysOM的诊断框架同样兼容旧版的语法，可以使用任意语言编写前处理脚本，只需要保证和诊断名称同名即可。

2.4 定义后处理脚本

2.4.1 后处理脚本的作用

后处理脚本主要对命令执行的结果进行汇总处理，并且转换成前端可渲染的数据格式

03:32:17 PM     all      0.09      0.00      0.16      0.00      0.00     99.75
03:32:18 PM     all      0.41      0.00      0.22      0.03      0.00     99.34
03:32:19 PM     all      0.06      0.00      0.16      0.00      0.00     99.78
03:32:20 PM     all      0.09      0.00      0.25      0.03      0.00     99.62
03:32:21 PM     all      0.16      0.00      0.03      0.00      0.00     99.81
03:32:22 PM     all      0.03      0.00      0.03      0.00      0.00     99.94
03:32:23 PM     all      0.16      0.00      0.06      0.00      0.00     99.78
03:32:24 PM     all      0.16      0.00      0.03      0.00      0.00     99.81
03:32:25 PM     all      0.06      0.00      0.06      0.03      0.00     99.84
03:32:26 PM     all      0.09      0.00      0.03      0.00      0.00     99.87

                                         ↓


{
  "code": 0,
  "err_msg": "",
  "result": {
    "cpu_usage_set": {
      "data": [
        {
          "time": "03:32:17 PM",
          "user": 0.09,
          "idle": 99.75
        },
        ...
      ]
    }
  }
}

2.4.2 基于Python编写后处理脚本

所有的前处理脚本都应该继承 DiagnosisPreProcessor 类，并实现对应的抽象方法：

class PostProcessResult:
    def __init__(self, code: int, result: str, err_msg: str = "") -> None:
      	# 后处理脚本执行状态码，0表示成功，其他值表示后处理脚本出错
        self.code = code
        # 后处理脚本转换后的结果，用于前端动态渲染
        self.result = result
        # 如果后处理脚本执行错误，可以在本字段放置错误的原因
        self.err_msg = err_msg

class DiagnosisPostProcessor(DiagnosisProcessorBase):
    """Post-processor used to perform: <diagnosis result> -> <Front-end formatted data>

    Args:
        DiagnosisProcessorBase (_type_): _description_
    """

    def __init__(self, service_name: str, **kwargs):
        self.service_name = service_name

    @abstractmethod
    def parse_diagnosis_result(self, results: List[DiagnosisJobResult]) -> dict:
        """Parse diagnosis results to front-end formmated data

				return:
					code
					{
						"code": 0,				=> 后处理脚本执行状态码，0表示成功，其他值表示后处理脚本出错
						"err_msg": "",		=> 如果后处理脚本执行错误，可以在本字段放置错误的原因
						"result": "xxxx"	=> 后处理脚本转换后的结果，用于前端动态渲染
					}
        Args:
            results (dict): Diagnosis result
        """

本指南的 demo 诊断的后处理脚本实现如下：

**注意：后处理脚本中后处理类的名字必须为 PostProcessor **

from typing import List
from .base import DiagnosisJobResult, DiagnosisPostProcessor, PostProcessResult


class PostProcessor(DiagnosisPostProcessor):
    def parse_diagnosis_result(self, results: List[DiagnosisJobResult]) -> PostProcessResult:
        postprocess_result = PostProcessResult(
            code=0,
            err_msg="",
            result={}
        )

        log = results[0].stdout
        datas = []
        for line in log.strip().split("\n"):
            values = line.strip().split("     ")
            datas.append({
                "time": values[0].strip(),
                "user": float(values[2].strip()),
                "idle": float(values[7].strip())
            })
        postprocess_result.result = {
            "cpu_usage_set": {"data": datas}
        }
        return postprocess_result

3. 使用案例

3.1 使用离线模式

离线模式支持诊断的前处理脚本直接返回诊断的结果，而不需要再某个纳管节点上执行命令进行采集，典型的比如可以去读取prometheus中的监控数据，并形成离线诊断结果，本节展示基于离线模式实现的一个使用案例。

3.1.1 echo 诊断功能说明

基于 SysOM 诊断框架提供的离线模式，开发一个简单的示例诊断，前处理脚本将传入的参数 value 作为离线诊断结果返回，并且跳过诊断执行节点，直接在后处理脚本中将前处理脚本返回的结果ECHO展示。

3.1.2 前处理脚本

from .base import DiagnosisPreProcessor, DiagnosisTask, DiagnosisJob


class PreProcessor(DiagnosisPreProcessor):
    """Command diagnosis

    Just invoke command in target instance and get stdout result

    Args:
        DiagnosisPreProcessor (_type_): _description_
    """

    def get_diagnosis_cmds(self, params: dict) -> DiagnosisTask:
        value = params.get("value", "")
        return DiagnosisTask(
            # 离线模式不需要在节点上执行，所以此处 instance 和 cmd 指定为空即可
            jobs=[DiagnosisJob(instance="", cmd="")],
            # 开启离线模式
            offline_mode=True,
            # 返回离线诊断的结果（数组中的每一项，最后都会传递到 DiagnosisJobResult.stdout）
            offline_results=[
                value
            ]
        )

3.1.3 后处理脚本

from typing import List
from .base import DiagnosisJobResult, DiagnosisPostProcessor, PostProcessResult


class PostProcessor(DiagnosisPostProcessor):
    def parse_diagnosis_result(self, results: List[DiagnosisJobResult]) -> PostProcessResult:
        postprocess_result = PostProcessResult(
            code=0,
            err_msg="",
            result={}
        )
        # 提取出上述返回的离线诊断结果，并整理成前端可识别的格式进行echo展示
        postprocess_result.result = {
            "EchoResult": {"data": [{"key": "", "value": results[0].stdout}]}
        }
        return postprocess_result

3.2 诊断命令执行结束后需要拉取节点端的文件

部分场景下，在节点端执行完诊断命令后，可能诊断的结果较大，或者分布在多个文件，或者诊断的结果是二进制一类的文件，不是字符类型的结果，因此，在这种场景下需要支持诊断命令执行结束后，将一个或者多个文件拉取回来供后处理脚本分析处理，本节展示诊断命令执行结束后需要拉取节点端的文件的一个使用案例。

3.2.1 get_release 诊断说明

get_release 诊断，在目标节点上执行 uname -r 获取内核版本，并且在执行结束后拉取节点端的 /etc/os-release 文件，后处理脚本根据终端输出的结果和拉取到的文件进行分析展示

3.2.2 前处理脚本

from .base import DiagnosisJob, DiagnosisPreProcessor, DiagnosisTask, FileItem


class PreProcessor(DiagnosisPreProcessor):
    """Get release info diagnosis

    Just invoke command in target instance and get stdout result

    Args:
        DiagnosisPreProcessor (_type_): _description_
    """

    def get_diagnosis_cmds(self, params: dict) -> DiagnosisTask:
        # 从前端传递的参数中读取目标实例IP
        instance = params.get("instance", "")
        return DiagnosisTask(
            jobs=[
                # 在 instance 上执行 uname -r 获取内核版本
                # 并在命令执行结束后，拉取节点端的 /etc/os-release 文件到本地
                DiagnosisJob(instance=instance, cmd="uname -r", fetch_file_list=[
                    FileItem("os-release", remote_path="/etc/os-release")
                ])
            ]
        )

3.2.3 后处理脚本

from typing import List
import json
from .base import DiagnosisJobResult, DiagnosisPostProcessor, PostProcessResult


class PostProcessor(DiagnosisPostProcessor):
    def parse_diagnosis_result(self, results: List[DiagnosisJobResult]) -> PostProcessResult:
        postprocess_result = PostProcessResult(
            code=0,
            err_msg="",
            result={}
        )
        with open(results[0].file_list[0].local_path, "r") as f:
            # 从本地读取 `/etc/os-release` 文件，提取 release 信息
            release_info = f.read()
            postprocess_result.result = {
                "KernelVersion": {"data": [{"key": "", "value": results[0].stdout}]},
                "ReleaseInfo": {"data": [{"key": "", "value": release_info}]}
            }
        print(json.dumps(postprocess_result.result, indent=4))
        return postprocess_result

后处理脚本打印的信息如下

{
    "KernelVersion": {
        "data": [
            {
                "key": "",
                "value": "5.10.134-14.an8.x86_64\n"
            }
        ]
    },
    "ReleaseInfo": {
        "data": [
            {
                "key": "",
                "value": "NAME=\"Anolis OS\"\nVERSION=\"8.8\"\nID=\"anolis\"\nID_LIKE=\"rhel fedora centos\"\nVERSION_ID=\"8.8\"\nPLATFORM_ID=\"platform:an8\"\nPRETTY_NAME=\"Anolis OS 8.8\"\nANSI_COLOR=\"0;31\"\nHOME_URL=\"https://openanolis.cn/\"\n\n"
            }
        ]
    }
}

16 KiB Raw Permalink Blame History Unescape Escape