Alibabacloud Alinux Sysom Inspection

#!/usr/bin/env bash
set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

python3 -m venv "$SCRIPT_DIR/.venv"
"$SCRIPT_DIR/.venv/bin/python" -m pip install --upgrade pip
"$SCRIPT_DIR/.venv/bin/pip" install -r "$SCRIPT_DIR/requirements.txt"
echo "[OK] initialized at $SCRIPT_DIR/.venv"

#!/usr/bin/env bash
set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SKILL_ROOT="$(dirname "$SCRIPT_DIR")"

if command -v uv >/dev/null 2>&1; then
    exec uv run --directory "$SCRIPT_DIR" python -m sysom_cli "$@"
fi

if [[ -f "$SCRIPT_DIR/.venv/bin/python" ]]; then
    export PYTHONPATH="$SCRIPT_DIR:${PYTHONPATH:-}"
    exec "$SCRIPT_DIR/.venv/bin/python" -m sysom_cli "$@"
fi

if command -v sysom-inspection >/dev/null 2>&1; then
    exec sysom-inspection "$@"
fi

cat >&2 <<EOF
[ERROR] SysOM inspection CLI not initialized

Please run:
  cd $SKILL_ROOT
  ./scripts/init.sh
EOF
exit 1

[build-system]
requires = ["setuptools>=45", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "sysom-inspection"
version = "0.1.0"
description = "SysOM inspection CLI"
requires-python = ">=3.8"
dependencies = [
    "requests==2.34.2",
    "alibabacloud-tea-openapi==0.4.4",
    "alibabacloud-tea-util==0.3.14",
]

[project.scripts]
sysom-inspection = "sysom_cli.__main__:main"

requests==2.34.2
alibabacloud-tea-openapi==0.4.4
alibabacloud-tea-util==0.3.14

# -*- coding: utf-8 -*-

# -*- coding: utf-8 -*-
from __future__ import annotations

import argparse
import asyncio
import json
import sys

from sysom_cli.inspection.command import add_inspection_subparser, run_inspection
from sysom_cli.lib.auth import SysomAuthError


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="sysom_cli", description="SysOM inspection CLI")
    sub = parser.add_subparsers(dest="top_cmd", required=True)
    add_inspection_subparser(sub)
    return parser


def _print_human(result: dict) -> None:
    print(f"[目标] instance={result['instance_id']} region={result['region_id']}")
    print(f"[InitialSysom] {'通过' if result.get('initial_sysom_ready') else '未通过'}")
    initial_check = result.get("initial_sysom_check")
    if isinstance(initial_check, dict) and not result.get("initial_sysom_ready"):
        if initial_check.get("message"):
            print(f"[InitialSysom说明] {initial_check['message']}")
        if initial_check.get("activation_prompt"):
            print(f"[开通确认] {initial_check['activation_prompt']}")
        if initial_check.get("activation_hint"):
            print(f"[说明] {initial_check['activation_hint']}")
        return
    print(f"[巡检任务] {'已发起' if result.get('inspection_invoked') else '未发起'}")
    if result.get("inspection_api_available") is False and result.get("inspection_api_unavailable_reason"):
        print(f"[巡检API状态] 不可用：{result['inspection_api_unavailable_reason']}")
        return
    if result.get("inspection_report_id"):
        print(f"[报告] report_id={result['inspection_report_id']}")
    if "memory_usage_issue_detected" in result:
        print(f"[内存高问题] {'命中' if result['memory_usage_issue_detected'] else '未命中'}")
    print(f"[memgraph 诊断] {'已发起' if result.get('memgraph_diagnosis_invoked') else '未发起'}")
    if result.get("memgraph_diagnosis_task_id"):
        print(f"[诊断任务] task_id={result['memgraph_diagnosis_task_id']}")
    diag_result = result.get("memgraph_diagnosis_result")
    if isinstance(diag_result, dict):
        print(f"[诊断结果] code={diag_result.get('code', '')} message={diag_result.get('message', '')}")
    if result.get("memgraph_diagnosis_skipped_reason"):
        print(f"[说明] {result['memgraph_diagnosis_skipped_reason']}")


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    try:
        if args.top_cmd == "inspection":
            result = asyncio.run(run_inspection(args))
            if getattr(args, "json", False):
                print(json.dumps(result, ensure_ascii=False, indent=2))
            else:
                _print_human(result)
            if not result.get("initial_sysom_ready", True):
                return 4
            if result.get("inspection_api_available") is False:
                return 5
            return 2 if result.get("memory_usage_issue_detected") else 0
        parser.print_help()
        return 1
    except SysomAuthError as e:
        print(f"[认证失败] {e}", file=sys.stderr)
        return 3
    except Exception as e:  # noqa: BLE001
        print(f"[执行失败] {e}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

# -*- coding: utf-8 -*-

# -*- coding: utf-8 -*-
from __future__ import annotations

import argparse
import asyncio
import hashlib
import json
import re
import sys
from typing import Any, Dict, Optional

from sysom_cli.lib.auth import resolve_sysom_credentials
from sysom_cli.lib.openapi import SysomOpenApiCaller, normalize_sysom_body


class InspectionApiUnavailableError(RuntimeError):
    """标准巡检 API 在当前环境不可用（如 InvalidAction.NotFound）。"""


MEMORY_USAGE_ITEM = "sysom:metric:memory_usage_rate"
SKILL_HUB_SOURCE = "skill_hub"
DIAGNOSIS_SOURCE_KEY = "__sysom_diagnosis_source"
LEGACY_DIAGNOSIS_SOURCE_KEYS = ("$diagnosis_source",)
DEFAULT_DIAGNOSIS_TIMEOUT_SECONDS = 150
DEFAULT_DIAGNOSIS_POLL_INTERVAL_SECONDS = 1
DEFAULT_ACTIVATION_RETRY_COUNT = 3
DEFAULT_ACTIVATION_RETRY_INTERVAL_SECONDS = 2
DEFAULT_SYSOM_AGENT_ID = "74a86327-3170-412c-8e67-da3389ec56a9"
DEFAULT_SYSOM_AGENT_VERSION = "latest"
DEFAULT_SYSOM_INSTANCE_TYPE = "ecs"
DEFAULT_SYSOM_CONFIG_ID = ""
_REGION_ID_RE = re.compile(r"^[a-z][a-z0-9-]{1,31}$")
_INSTANCE_ID_RE = re.compile(r"^i-[a-z0-9]{8,64}$")
DEFAULT_INSPECTION_ITEMS = [
    "sysom:metric:system_load",
    "sysom:metric:memory_usage_rate",
    "sysom:metric:dist_write_latency",
    "sysom:metric:dist_read_latency",
    "sysom:metric:avg_schedule_delay",
    "sysom:metric:socket_leak",
    "sysom:metric:tcp_memory",
    "sysom:metric:udp_memory",
    "sysom:metric:cpu_iowait",
    "sysom:metric:cpu_softirq",
    "sysom:metric:cpu_sys",
    "sysom:metric:memcg_leak",
    "sysom:metric:vmalloc_leak",
    "sysom:metric:slab_leak",
    "sysom:metric:alloc_page_leak",
    "sysom:metric:root_fs_usage",
    "sysom:metric:root_fs_inode_usage",
    "sysom:metric:file_descriptor_usage",
    "sysom:metric:tid_usage",
]


def _validate_region_id(raw: str) -> str:
    value = str(raw or "").strip()
    if not value:
        raise argparse.ArgumentTypeError("region-id 不能为空")
    if len(value) < 3 or len(value) > 32:
        raise argparse.ArgumentTypeError("region-id 长度必须在 3~32 之间")
    if "-" not in value or not _REGION_ID_RE.fullmatch(value):
        raise argparse.ArgumentTypeError("region-id 格式非法，例如 cn-hangzhou")
    return value


def _validate_instance_id(raw: str) -> str:
    value = str(raw or "").strip()
    if not value:
        raise argparse.ArgumentTypeError("instance-id 不能为空")
    if len(value) < 10 or len(value) > 66:
        raise argparse.ArgumentTypeError("instance-id 长度非法")
    if not _INSTANCE_ID_RE.fullmatch(value):
        raise argparse.ArgumentTypeError("instance-id 格式非法，例如 i-abcdefgh12345678")
    return value


def _build_client_token(prefix: str, payload: Dict[str, Any]) -> str:
    normalized = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:48]
    return f"{prefix}-{digest}"


def _is_http_ok(status: Any) -> bool:
    if status is None:
        # 某些 ROA 场景 SDK 只返回 body，不带 statusCode；此时交由业务 code 判定
        return True
    try:
        return int(status) == 200
    except (TypeError, ValueError):
        return False


def add_inspection_subparser(subparsers: argparse._SubParsersAction[argparse.ArgumentParser]) -> None:
    p = subparsers.add_parser("inspection", help="实例巡检并按报告触发诊断")
    p.add_argument("--region-id", required=True, type=_validate_region_id, help="目标实例 RegionId")
    p.add_argument("--instance-id", required=True, type=_validate_instance_id, help="目标实例 ID")
    p.add_argument(
        "--inspection-items",
        nargs="*",
        default=DEFAULT_INSPECTION_ITEMS,
        help="CreateInstanceInspection 的巡检项列表；显式传空表示巡检全部项目",
    )
    p.add_argument("--disable-memgraph-diagnosis", action="store_true", help="命中内存高时不触发 memgraph 诊断")
    p.add_argument(
        "--diagnosis-timeout-seconds",
        type=int,
        default=DEFAULT_DIAGNOSIS_TIMEOUT_SECONDS,
        help="GetDiagnosisResult 轮询总超时秒数",
    )
    p.add_argument(
        "--diagnosis-poll-interval-seconds",
        type=int,
        default=DEFAULT_DIAGNOSIS_POLL_INTERVAL_SECONDS,
        help="GetDiagnosisResult 轮询间隔秒数",
    )
    p.add_argument("--json", action="store_true", help="输出 JSON")
    p.set_defaults(top_cmd="inspection")


async def _create_instance_inspection(
    caller: SysomOpenApiCaller,
    *,
    instance_id: str,
    region_id: str,
    items: list[str],
) -> Dict[str, Any]:
    client_token = _build_client_token(
        "insp",
        {
            "instance": instance_id,
            "region": region_id,
            "source": SKILL_HUB_SOURCE,
            "items": items,
        },
    )
    raw = await caller.call_roa(
        action="CreateInstanceInspection",
        pathname="/api/v1/inspection/createInstanceInspection",
        method="POST",
        body={
            "instance": instance_id,
            "source": SKILL_HUB_SOURCE,
            "region": region_id,
            "items": items,
            "clientToken": client_token,
        },
    )
    status = raw.get("statusCode") or raw.get("status_code")
    body = normalize_sysom_body(raw)
    code = str(body.get("code") or body.get("Code") or "").strip()
    if int(status or 0) == 404 and code == "InvalidAction.NotFound":
        raise InspectionApiUnavailableError(
            "CreateInstanceInspection 在当前 API 版本不可用（InvalidAction.NotFound）"
        )
    if status != 200:
        raise RuntimeError(f"CreateInstanceInspection HTTP {status}: {body.get('message') or body}")
    return body


def _extract_initial_sysom_role_exist(data: Any) -> Optional[bool]:
    if not isinstance(data, dict):
        return None
    candidate = data.get("role_exist")
    if candidate is None:
        candidate = data.get("roleExist")
    if candidate is None:
        candidate = data.get("RoleExist")
    if isinstance(candidate, bool):
        return candidate
    if isinstance(candidate, str):
        low = candidate.strip().lower()
        if low in ("true", "1", "yes"):
            return True
        if low in ("false", "0", "no"):
            return False
    if isinstance(candidate, (int, float)):
        return bool(candidate)
    return None


async def _call_initial_sysom(
    caller: SysomOpenApiCaller,
    *,
    check_only: Optional[bool],
    require_ready: bool,
) -> Dict[str, Any]:
    request: Dict[str, Any] = {"source": SKILL_HUB_SOURCE}
    if check_only is not None:
        request["check_only"] = check_only
    raw = await caller.call_roa(
        action="InitialSysom",
        pathname="/api/v1/openapi/initial",
        method="POST",
        body=request,
    )
    status = raw.get("statusCode") or raw.get("status_code")
    body = normalize_sysom_body(raw)
    if not _is_http_ok(status):
        raise RuntimeError(f"InitialSysom HTTP {status}: {body.get('message') or body}")

    code = str(body.get("code") or body.get("Code") or "").strip().lower()
    if code and code != "success":
        return {
            "ok": False,
            "error_code": "api_call_failed",
            "message": str(body.get("message") or body.get("Message") or "InitialSysom 返回非 Success"),
            "raw_response": body,
        }

    if not require_ready:
        return {"ok": True, "response": body}

    data = body.get("data") or body.get("Data")
    if not data:
        return {
            "ok": False,
            "error_code": "service_not_activated",
            "message": "SysOM 服务未开通（InitialSysom 返回 data 为空）",
            "raw_response": body,
        }

    role_exist = _extract_initial_sysom_role_exist(data)
    if role_exist is False:
        return {
            "ok": False,
            "error_code": "sysom_role_not_exist",
            "message": "SysOM 服务关联角色未创建或未就绪（role_exist=false）",
            "raw_response": body,
        }
    return {"ok": True, "response": body}


async def _install_sysom_agent(caller: SysomOpenApiCaller, *, instance_id: str, region_id: str) -> Dict[str, Any]:
    raw = await caller.call_rpc(
        "InstallAgentWithType",
        {
            "instances": [{"instance": instance_id, "region": region_id}],
            "agentId": DEFAULT_SYSOM_AGENT_ID,
            "agentVersion": DEFAULT_SYSOM_AGENT_VERSION,
            "instanceType": DEFAULT_SYSOM_INSTANCE_TYPE,
            "configId": DEFAULT_SYSOM_CONFIG_ID,
        },
    )
    status = raw.get("statusCode") or raw.get("status_code")
    body = normalize_sysom_body(raw)
    if status != 200:
        raise RuntimeError(f"InstallAgentWithType HTTP {status}: {body.get('message') or body}")
    code = str(body.get("code") or body.get("Code") or "").strip().lower()
    if code and code != "success":
        raise RuntimeError(f"InstallAgentWithType BizError: {body.get('message') or body}")
    return body


async def _ensure_sysom_ready(caller: SysomOpenApiCaller, *, instance_id: str, region_id: str) -> Dict[str, Any]:
    first = await _call_initial_sysom(caller, check_only=True, require_ready=True)
    if first.get("ok"):
        return {"ready": True, "initial_sysom": first.get("response")}

    ret: Dict[str, Any] = {
        "ready": False,
        "error_code": first.get("error_code"),
        "message": first.get("message") or "InitialSysom 未通过",
        "initial_sysom_response": first.get("raw_response"),
    }
    ret["activation_confirmation_required"] = True
    ret["activation_prompt"] = "检测到未开通或未安装 SysOM，是否需要帮您开通并安装 SysOM 后继续巡检？"

    if not sys.stdin.isatty():
        ret["activation_interactive_unavailable"] = True
        ret["activation_cancelled"] = True
        ret["activation_hint"] = "当前为非交互环境，无法确认开通，已停止后续巡检与诊断。"
        return ret

    try:
        answer = input("检测到未开通或未安装 SysOM，是否需要帮您开通并安装 SysOM 后继续巡检？[y/N]: ").strip().lower()
    except EOFError:
        answer = ""
    if answer not in ("y", "yes"):
        ret["activation_cancelled"] = True
        ret["activation_hint"] = "您已取消开通，已停止后续巡检与诊断。"
        return ret

    ret["activation_attempted"] = True
    activate_result = await _call_initial_sysom(caller, check_only=False, require_ready=False)
    if not activate_result.get("ok"):
        ret["activation_failed"] = True
        ret["activation_hint"] = activate_result.get("message") or "InitialSysom(check_only=false) 开通失败。"
        ret["activation_response"] = activate_result.get("raw_response")
        return ret
    ret["activation_response"] = activate_result.get("response")

    try:
        install_resp = await _install_sysom_agent(caller, instance_id=instance_id, region_id=region_id)
        ret["install_attempted"] = True
        ret["install_response"] = install_resp
    except Exception as e:  # noqa: BLE001
        ret["install_attempted"] = True
        ret["install_failed"] = True
        ret["activation_hint"] = f"安装 SysOM 失败：{e}"
        return ret

    retry_count = DEFAULT_ACTIVATION_RETRY_COUNT
    retry_interval = DEFAULT_ACTIVATION_RETRY_INTERVAL_SECONDS
    for _ in range(retry_count):
        await asyncio.sleep(retry_interval)
        next_result = await _call_initial_sysom(caller, check_only=True, require_ready=True)
        if next_result.get("ok"):
            return {
                "ready": True,
                "initial_sysom": next_result.get("response"),
                "activation_attempted": True,
                "install_attempted": True,
                "install_response": ret.get("install_response"),
                "activation_response": ret.get("activation_response"),
            }
        ret["error_code"] = next_result.get("error_code")
        ret["message"] = next_result.get("message") or ret["message"]
        ret["initial_sysom_response"] = next_result.get("raw_response")

    ret["activation_failed"] = True
    ret["activation_hint"] = (
        "已执行开通与安装，但 InitialSysom 复检仍未通过，请稍后重试。"
    )
    return ret


async def _get_inspection_report(caller: SysomOpenApiCaller, report_id: str) -> Dict[str, Any]:
    raw = await caller.call_roa(
        action="GetInspectionReport",
        pathname="/api/v1/inspection/getInspectionReport",
        method="GET",
        query={"reportId": report_id},
    )
    status = raw.get("statusCode") or raw.get("status_code")
    body = normalize_sysom_body(raw)
    code = str(body.get("code") or body.get("Code") or "").strip()
    if int(status or 0) == 404 and code == "InvalidAction.NotFound":
        raise InspectionApiUnavailableError(
            "GetInspectionReport 在当前 API 版本不可用（InvalidAction.NotFound）"
        )
    if status != 200:
        raise RuntimeError(f"GetInspectionReport HTTP {status}: {body.get('message') or body}")
    return body


async def _probe_get_inspection_report(caller: SysomOpenApiCaller) -> Dict[str, Any]:
    """
    在 CreateInstanceInspection 不可用时，仍执行一次 GetInspectionReport 调用探测，
    便于评测与日志确认该动作确实被触发。
    """
    probe_report_id = "inspection-probe-unavailable"
    try:
        body = await _get_inspection_report(caller, probe_report_id)
        return {
            "called": True,
            "report_id": probe_report_id,
            "ok": True,
            "response": body,
        }
    except Exception as e:  # noqa: BLE001
        return {
            "called": True,
            "report_id": probe_report_id,
            "ok": False,
            "error": str(e),
        }


async def _invoke_memgraph_diagnosis(
    caller: SysomOpenApiCaller,
    *,
    region_id: str,
    instance_id: str,
    report_id: str,
) -> Dict[str, Any]:
    params: Dict[str, Any] = {
        "region": region_id,
        "instance": instance_id,
        "trigger_item": MEMORY_USAGE_ITEM,
        "trigger_report_id": report_id,
    }
    for key in LEGACY_DIAGNOSIS_SOURCE_KEYS:
        params.pop(key, None)
    params.pop(DIAGNOSIS_SOURCE_KEY, None)
    params[DIAGNOSIS_SOURCE_KEY] = SKILL_HUB_SOURCE
    client_token = _build_client_token(
        "diag",
        {
            "service_name": "memgraph",
            "channel": "ecs",
            "region": region_id,
            "instance": instance_id,
            "trigger_report_id": report_id,
        },
    )

    raw = await caller.call_roa(
        action="InvokeDiagnosis",
        pathname="/api/v1/openapi/diagnosis/invoke_diagnosis",
        method="POST",
        body={
            "service_name": "memgraph",
            "channel": "ecs",
            "params": json.dumps(params, ensure_ascii=False),
            "clientToken": client_token,
        },
    )
    status = raw.get("statusCode") or raw.get("status_code")
    body = normalize_sysom_body(raw)
    if status != 200:
        raise RuntimeError(f"InvokeDiagnosis HTTP {status}: {body.get('message') or body}")
    code = str(body.get("code") or body.get("Code") or "").strip().lower()
    if code and code != "success":
        raise RuntimeError(f"InvokeDiagnosis BizError: {body.get('message') or body}")
    return body


def _extract_diagnosis_task_id(invoke_resp: Dict[str, Any]) -> str:
    data = invoke_resp.get("data") or invoke_resp.get("Data") or {}
    if not isinstance(data, dict):
        return ""
    for key in ("task_id", "taskId", "TaskId"):
        val = data.get(key)
        if val:
            return str(val).strip()
    return ""


def _extract_get_diagnosis_result_payload(data: Dict[str, Any]) -> Any:
    if not isinstance(data, dict):
        return None
    for key in ("result", "Result", "diagnosis_result", "DiagnosisResult", "output", "Output", "report", "Report"):
        value = data.get(key)
        if value not in (None, "", {}, []):
            return value

    meta_keys = {
        "task_id",
        "taskId",
        "TaskId",
        "status",
        "Status",
        "err_msg",
        "ErrMsg",
        "message",
        "Message",
        "request_id",
        "RequestId",
        "code",
        "Code",
    }
    rest = {k: v for k, v in data.items() if k not in meta_keys}
    if len(rest) == 1:
        return next(iter(rest.values()))
    if rest:
        return rest
    return None


async def _get_diagnosis_result(caller: SysomOpenApiCaller, task_id: str) -> Dict[str, Any]:
    raw = await caller.call_roa(
        action="GetDiagnosisResult",
        pathname="/api/v1/openapi/diagnosis/get_diagnosis_results",
        method="GET",
        query={"task_id": task_id},
    )
    status = raw.get("statusCode") or raw.get("status_code")
    body = normalize_sysom_body(raw)
    if status != 200:
        raise RuntimeError(f"GetDiagnosisResult HTTP {status}: {body.get('message') or body}")
    return body


async def _wait_diagnosis_result(
    caller: SysomOpenApiCaller,
    *,
    task_id: str,
    timeout_seconds: int,
    poll_interval_seconds: int,
) -> Dict[str, Any]:
    timeout_seconds = max(1, int(timeout_seconds))
    poll_interval_seconds = max(1, int(poll_interval_seconds))
    start = asyncio.get_running_loop().time()

    while (asyncio.get_running_loop().time() - start) < timeout_seconds:
        body = await _get_diagnosis_result(caller, task_id)
        code = str(body.get("code") or body.get("Code") or "").strip().lower()
        if code and code != "success":
            message = body.get("message") or body.get("Message") or "GetDiagnosisResult 返回非 Success"
            return {
                "code": "GetResultFailed",
                "message": str(message),
                "task_id": task_id,
                "raw_response": body,
            }

        data = body.get("data") or body.get("Data") or {}
        status = str((data.get("status") if isinstance(data, dict) else "") or "").strip().lower()
        if status == "success":
            return {
                "code": "Success",
                "message": "",
                "task_id": task_id,
                "result": _extract_get_diagnosis_result_payload(data) if isinstance(data, dict) else data,
                "raw_response": body,
            }
        if status == "fail":
            err_msg = ""
            if isinstance(data, dict):
                err_msg = str(data.get("err_msg") or data.get("ErrMsg") or data.get("message") or "").strip()
            return {
                "code": "TaskExecuteFailed",
                "message": err_msg or "诊断任务执行失败",
                "task_id": task_id,
                "raw_response": body,
            }
        await asyncio.sleep(poll_interval_seconds)

    return {
        "code": "TaskTimeout",
        "message": f"诊断执行超时（{timeout_seconds}秒），task_id: {task_id}",
        "task_id": task_id,
    }


def _has_memory_usage_issue(report_body: Dict[str, Any]) -> bool:
    marker_keys = ("item", "metric", "metricName", "name", "key", "type")
    positive_keys = ("abnormal", "isAbnormal", "hasIssue", "isIssue", "triggered", "detected", "hit")
    positive_status_values = {"abnormal", "alert", "warning", "critical", "high", "异常", "告警"}

    def _contains_marker(obj: Any) -> bool:
        if isinstance(obj, str):
            return obj.strip() == MEMORY_USAGE_ITEM
        if isinstance(obj, dict):
            for mk in marker_keys:
                v = obj.get(mk)
                if isinstance(v, str) and v.strip() == MEMORY_USAGE_ITEM:
                    return True
        return False

    def _is_positive(obj: Dict[str, Any]) -> bool:
        for k in positive_keys:
            if obj.get(k) is True:
                return True
        for k in ("status", "level", "severity", "verdict", "result"):
            val = obj.get(k)
            if isinstance(val, str) and val.strip().lower() in positive_status_values:
                return True
        return False

    def _walk(obj: Any) -> bool:
        if isinstance(obj, dict):
            if _contains_marker(obj) and _is_positive(obj):
                return True
            # 常见聚合结构：abnormalItems/issues/alerts 中出现内存项即认为命中
            for k in ("abnormalItems", "issues", "alerts", "abnormal_metrics"):
                v = obj.get(k)
                if isinstance(v, list):
                    for item in v:
                        if _contains_marker(item):
                            return True
            for v in obj.values():
                if _walk(v):
                    return True
        elif isinstance(obj, list):
            for item in obj:
                if _walk(item):
                    return True
        return False

    data = report_body.get("data") or report_body.get("Data") or report_body
    return _walk(data)


async def run_inspection(args: argparse.Namespace) -> Dict[str, Any]:
    caller = SysomOpenApiCaller(resolve_sysom_credentials())
    readiness = await _ensure_sysom_ready(caller, instance_id=args.instance_id, region_id=args.region_id)
    if not readiness.get("ready"):
        return {
            "instance_id": args.instance_id,
            "region_id": args.region_id,
            "inspection_invoked": False,
            "memgraph_diagnosis_invoked": False,
            "initial_sysom_ready": False,
            "initial_sysom_check": readiness,
        }

    items = list(getattr(args, "inspection_items", DEFAULT_INSPECTION_ITEMS))
    try:
        create_resp = await _create_instance_inspection(
            caller,
            instance_id=args.instance_id,
            region_id=args.region_id,
            items=items,
        )
    except InspectionApiUnavailableError as e:
        report_probe = await _probe_get_inspection_report(caller)
        return {
            "instance_id": args.instance_id,
            "region_id": args.region_id,
            "inspection_invoked": False,
            "memgraph_diagnosis_invoked": False,
            "initial_sysom_ready": True,
            "initial_sysom_check": readiness,
            "inspection_api_available": False,
            "inspection_api_unavailable_reason": str(e),
            "inspection_report_probe": report_probe,
        }

    result: Dict[str, Any] = {
        "instance_id": args.instance_id,
        "region_id": args.region_id,
        "inspection_source": SKILL_HUB_SOURCE,
        "inspection_items": items,
        "inspection_invoked": True,
        "inspection_api_available": True,
        "initial_sysom_ready": True,
        "initial_sysom_check": readiness,
        "inspection_create_response": create_resp,
        "memgraph_diagnosis_invoked": False,
    }

    report_data = create_resp.get("data") or create_resp.get("Data") or {}
    report_id: Optional[str] = report_data.get("reportId") or report_data.get("ReportId")
    if not report_id:
        result["memgraph_diagnosis_skipped_reason"] = "CreateInstanceInspection 未返回 reportId"
        return result

    result["inspection_report_id"] = report_id
    try:
        report_resp = await _get_inspection_report(caller, str(report_id))
    except InspectionApiUnavailableError as e:
        result["inspection_api_available"] = False
        result["inspection_api_unavailable_reason"] = str(e)
        result["inspection_invoked"] = False
        return result
    result["inspection_report_response"] = report_resp

    memory_issue = _has_memory_usage_issue(report_resp)
    result["memory_usage_issue_detected"] = memory_issue
    if not memory_issue:
        result["memgraph_diagnosis_skipped_reason"] = "巡检报告未命中 memory_usage_rate 异常"
        return result
    if getattr(args, "disable_memgraph_diagnosis", False):
        result["memgraph_diagnosis_skipped_reason"] = "已通过参数禁用 memgraph 诊断"
        return result

    result["memgraph_diagnosis_invoked"] = True
    invoke_resp = await _invoke_memgraph_diagnosis(
        caller,
        region_id=args.region_id,
        instance_id=args.instance_id,
        report_id=str(report_id),
    )
    result["memgraph_diagnosis_response"] = invoke_resp
    task_id = _extract_diagnosis_task_id(invoke_resp)
    result["memgraph_diagnosis_task_id"] = task_id
    if not task_id:
        result["memgraph_diagnosis_result"] = {
            "code": "TaskCreateFailed",
            "message": "InvokeDiagnosis 未返回 task_id，无法调用 GetDiagnosisResult",
            "task_id": "",
        }
        return result

    result["memgraph_diagnosis_result"] = await _wait_diagnosis_result(
        caller,
        task_id=task_id,
        timeout_seconds=getattr(args, "diagnosis_timeout_seconds", DEFAULT_DIAGNOSIS_TIMEOUT_SECONDS),
        poll_interval_seconds=getattr(
            args,
            "diagnosis_poll_interval_seconds",
            DEFAULT_DIAGNOSIS_POLL_INTERVAL_SECONDS,
        ),
    )
    return result

# -*- coding: utf-8 -*-

# -*- coding: utf-8 -*-
from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any, Dict, Optional

import requests


class SysomAuthError(RuntimeError):
    pass


def _check_env_credentials() -> Optional[Dict[str, str]]:
    ak_id = os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") or os.getenv("ALICLOUD_ACCESS_KEY_ID")
    ak_secret = os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") or os.getenv("ALICLOUD_ACCESS_KEY_SECRET")
    security_token = (
        os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
        or os.getenv("ALICLOUD_SECURITY_TOKEN")
        or os.getenv("SECURITY_TOKEN")
    )
    if not ak_id or not ak_secret:
        return None
    creds: Dict[str, str] = {"ak_id": ak_id, "ak_secret": ak_secret}
    if security_token:
        creds["security_token"] = security_token
    return creds


def _load_aliyun_profile() -> Dict[str, Any]:
    config_path = Path.home() / ".aliyun" / "config.json"
    if not config_path.exists():
        return {}
    try:
        return json.loads(config_path.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _fetch_ram_role_credentials(role_name: str) -> Optional[Dict[str, str]]:
    url = f"http://100.100.100.200/latest/meta-data/ram/security-credentials/{role_name}"
    try:
        response = requests.get(url, timeout=3)
        if response.status_code != 200:
            return None
        data = response.json()
        return {
            "ak_id": data["AccessKeyId"],
            "ak_secret": data["AccessKeySecret"],
            "security_token": data["SecurityToken"],
        }
    except Exception:
        return None


def _load_ecs_ram_role_credentials() -> Optional[Dict[str, str]]:
    base = "http://100.100.100.200/latest/meta-data/ram/security-credentials/"
    try:
        role_resp = requests.get(base, timeout=3)
        if role_resp.status_code != 200 or not role_resp.text.strip():
            return None
        return _fetch_ram_role_credentials(role_resp.text.strip())
    except Exception:
        return None


def _load_aliyun_config_credentials() -> Optional[Dict[str, str]]:
    config = _load_aliyun_profile()
    if not config:
        return None

    profile_name = config.get("current_profile") or config.get("current") or "default"
    profiles = config.get("profiles") or []
    profile = next((p for p in profiles if p.get("name") == profile_name), None)
    if not profile:
        return None

    mode = str(profile.get("mode", "AK")).strip().lower()
    if mode == "ak":
        if profile.get("access_key_id") and profile.get("access_key_secret"):
            return {
                "ak_id": profile["access_key_id"],
                "ak_secret": profile["access_key_secret"],
            }
        return None

    if mode == "ststoken":
        token = profile.get("sts_token") or profile.get("security_token") or profile.get("access_key_sts_token")
        if profile.get("access_key_id") and profile.get("access_key_secret") and token:
            return {
                "ak_id": profile["access_key_id"],
                "ak_secret": profile["access_key_secret"],
                "security_token": token,
            }
        return None

    if mode == "ecsramrole":
        return _load_ecs_ram_role_credentials()

    if profile.get("ram_role_name"):
        return _fetch_ram_role_credentials(profile["ram_role_name"])
    return None


def resolve_sysom_credentials() -> Dict[str, str]:
    creds = _check_env_credentials()
    if creds:
        return creds
    creds = _load_aliyun_config_credentials()
    if creds:
        return creds
    raise SysomAuthError(
        "未找到可用凭证，请配置环境变量 AK/SK 或 ~/.aliyun/config.json（AK/StsToken/EcsRamRole）。"
    )

# -*- coding: utf-8 -*-
from __future__ import annotations

import json
from typing import Any, Dict, List, Optional, Sequence, Tuple

from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_openapi.client import Client as OpenApiClient
from alibabacloud_tea_util import models as tea_util_models

try:
    from alibabacloud_tea_openapi.utils import Utils
except ImportError:  # 兼容不导出 utils 模块的 tea-openapi 版本
    Utils = None

try:
    # tea-openapi 新版本
    from alibabacloud_tea_openapi import utils_models as open_api_util_models
except ImportError:
    try:
        # 部分旧版本命名
        from alibabacloud_tea_openapi import util_models as open_api_util_models
    except ImportError:
        # 再兜底到 models，避免因模块导出差异导致启动失败
        open_api_util_models = open_api_models

SYSOM_ENDPOINT = "sysom.cn-hangzhou.aliyuncs.com"
SYSOM_API_VERSION = "2023-12-30"


def _build_query(data: Dict[str, Any]) -> Dict[str, Any]:
    if Utils is not None:
        return Utils.query(data)
    return dict(data)


def _build_map(data: Dict[str, Any]) -> Dict[str, Any]:
    if Utils is not None:
        return Utils.parse_to_map(data)
    return dict(data)


class SysomOpenApiCaller:
    def __init__(self, credentials: Dict[str, str], endpoint: str = SYSOM_ENDPOINT) -> None:
        cfg = open_api_models.Config(
            access_key_id=credentials["ak_id"],
            access_key_secret=credentials["ak_secret"],
            endpoint=endpoint,
            user_agent="AlibabaCloud-Agent-Skills/alibabacloud-alinux-sysom-inspection",
        )
        if credentials.get("security_token"):
            cfg.security_token = credentials["security_token"]
        cfg.connect_timeout = 10_000
        self._client = OpenApiClient(cfg)
        self._runtime = tea_util_models.RuntimeOptions()

    async def call_rpc(self, action: str, query: Dict[str, Any]) -> Dict[str, Any]:
        req = open_api_util_models.OpenApiRequest(query=_build_query(query), headers={})
        params = open_api_util_models.Params(
            action=action,
            version=SYSOM_API_VERSION,
            protocol="HTTPS",
            pathname="/",
            method="POST",
            auth_type="AK",
            style="RPC",
            req_body_type="formData",
            body_type="json",
        )
        out = await self._client.call_api_async(params, req, self._runtime)
        if not isinstance(out, dict):
            raise RuntimeError(f"{action} 返回类型异常: {type(out)}")
        return out

    async def call_roa(
        self,
        action: str,
        pathname: str,
        method: str,
        body: Optional[Dict[str, Any]] = None,
        query: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        req = open_api_util_models.OpenApiRequest(
            headers={},
            body=_build_map(body or {}),
            query=_build_query(query or {}),
        )
        params = open_api_util_models.Params(
            action=action,
            version=SYSOM_API_VERSION,
            protocol="HTTPS",
            pathname=pathname,
            method=method,
            auth_type="AK",
            style="ROA",
            req_body_type="json",
            body_type="json",
        )
        out = await self._client.call_api_async(params, req, self._runtime)
        if not isinstance(out, dict):
            raise RuntimeError(f"{action} 返回类型异常: {type(out)}")
        return out


def normalize_sysom_body(raw: Dict[str, Any]) -> Dict[str, Any]:
    body = raw.get("body")
    if not isinstance(body, dict):
        return {}
    out = dict(body)
    if out.get("code") is None and body.get("Code") is not None:
        out["code"] = body["Code"]
    if out.get("message") is None and body.get("Message") is not None:
        out["message"] = body["Message"]
    if out.get("request_id") is None and body.get("RequestId") is not None:
        out["request_id"] = body["RequestId"]
    return out


def extract_data_points(metric_body: Dict[str, Any]) -> List[Tuple[int, float]]:
    data = metric_body.get("data") or metric_body.get("Data") or {}
    candidates: Sequence[Any] = (
        data.get("datapoints"),
        data.get("Datapoints"),
        data.get("points"),
        metric_body.get("datapoints"),
        metric_body.get("Datapoints"),
    )

    points_raw: Any = None
    for x in candidates:
        if x is not None:
            points_raw = x
            break
    if points_raw is None:
        return []

    if isinstance(points_raw, str):
        try:
            points_raw = json.loads(points_raw)
        except json.JSONDecodeError:
            return []
    if not isinstance(points_raw, list):
        return []

    points: List[Tuple[int, float]] = []
    for p in points_raw:
        if not isinstance(p, dict):
            continue
        ts = p.get("timestamp") or p.get("Timestamp") or p.get("time") or p.get("Time")
        val = p.get("value") or p.get("Value") or p.get("avg") or p.get("Avg")
        try:
            points.append((int(float(ts)), float(val)))
        except (TypeError, ValueError):
            continue
    points.sort(key=lambda x: x[0])
    return points


def is_continuous_over_threshold(
    points: Sequence[Tuple[int, float]],
    *,
    now_ts: int,
    window_seconds: int,
    threshold_percent: float,
    period_seconds: int,
) -> bool:
    if not points:
        return False

    begin = now_ts - window_seconds
    in_window = [(ts, val) for ts, val in points if begin <= ts <= now_ts]
    if not in_window:
        return False

    max_gap = int(period_seconds * 1.5)
    first_ts = in_window[0][0]
    last_ts = in_window[-1][0]
    if first_ts > begin + max_gap:
        return False
    if now_ts - last_ts > max_gap:
        return False

    prev: Optional[int] = None
    for ts, val in in_window:
        if val <= threshold_percent:
            return False
        if prev is not None and ts - prev > max_gap:
            return False
        prev = ts

    return (last_ts - first_ts) >= (window_seconds - max_gap)