xiaoxia-saas/scripts/ci/ci_trace_report.py

#!/usr/bin/env python3
"""
CI Trace 涓婃姤鑴氭湰 - 鐢ㄤ簬 Gitea Actions workflow 涓笂鎶?Trace 鏁版嵁鍒?AgentLoop

鍦?CI workflow 鐨勬瘡涓?job 涓皟鐢細
  - 寮€濮嬫椂锛歱ython3 .gitea/scripts/ci_trace_report.py --status running
  - 缁撴潫鏃讹細python3 .gitea/scripts/ci_trace_report.py --status ok --start-time $CI_TRACE_START_TIME

鐜鍙橀噺锛圙itea Actions 鍐呯疆锛夛細
  GITEA_REPOSITORY / GITHUB_REPOSITORY - 浠撳簱鍚?(owner/repo)
  GITEA_WORKFLOW / GITHUB_WORKFLOW     - workflow 鍚嶇О
  GITEA_JOB / GITHUB_JOB               - job ID
  GITEA_SHA / GITHUB_SHA               - commit SHA
  GITEA_REF_NAME / GITHUB_REF_NAME     - 鍒嗘敮鍚?  GITEA_RUN_ID / GITHUB_RUN_ID         - run ID
  GITEA_ACTOR / GITHUB_ACTOR           - 瑙﹀彂鑰?  GITEA_EVENT_NAME / GITHUB_EVENT_NAME - 浜嬩欢绫诲瀷
  PR_NUMBER / GITEA_PR_NUMBER          - PR 鍙凤紙濡傛灉鏄?PR 瑙﹀彂锛?
AgentLoop 閰嶇疆锛堥€氳繃 Secrets 娉ㄥ叆锛夛細
  AGENTLOOP_LICENSE_KEY - LicenseKey锛堝繀濉級
  AGENTLOOP_ENDPOINT    - Trace 涓婃姤鍦板潃锛堝彲閫夛紝鏈夐粯璁ゅ€硷級
  AGENTLOOP_PROJECT     - SLS Project 鍚嶏紙鍙€夛級
  AGENTLOOP_WORKSPACE   - CMS Workspace 鍚嶏紙鍙€夛級
"""

import os
import sys
import json
import time
import uuid
import argparse
import urllib.request
import urllib.error


# ========== 榛樿閰嶇疆 ==========
DEFAULT_ENDPOINT = "https://proj-xtrace-495e81719a1fd9a2c5fd671eefafbe-cn-hangzhou.cn-hangzhou.log.aliyuncs.com/apm/trace/opentelemetry/v1/traces"
DEFAULT_PROJECT = "proj-xtrace-495e81719a1fd9a2c5fd671eefafbe-cn-hangzhou"
DEFAULT_WORKSPACE = "agentloop-13b8d6efb7fde6e9b193eb982ade68e2"


# ========== OTLP Protobuf 鎵嬪姩缂栫爜 ==========

def _encode_varint(value):
    result = bytearray()
    while value > 0x7F:
        result.append((value & 0x7F) | 0x80)
        value >>= 7
    result.append(value & 0x7F)
    return bytes(result)

def _encode_tag(field_number, wire_type):
    return _encode_varint((field_number << 3) | wire_type)

def _encode_string_field(field_number, value):
    value_bytes = value.encode('utf-8')
    return _encode_tag(field_number, 2) + _encode_varint(len(value_bytes)) + value_bytes

def _encode_bytes_field(field_number, value_bytes):
    return _encode_tag(field_number, 2) + _encode_varint(len(value_bytes)) + value_bytes

def _encode_int_field(field_number, value):
    return _encode_tag(field_number, 0) + _encode_varint(value & 0xFFFFFFFFFFFFFFFF)

def _encode_message_field(field_number, message_bytes):
    return _encode_tag(field_number, 2) + _encode_varint(len(message_bytes)) + message_bytes

def _encode_key_value(key, value_str):
    any_value = _encode_string_field(1, value_str)
    return _encode_string_field(1, key) + _encode_message_field(2, any_value)

def _encode_status(status_code, status_msg=""):
    data = _encode_int_field(1, status_code)
    if status_msg:
        data += _encode_string_field(2, status_msg)
    return data

def _encode_span(trace_id_bytes, span_id_bytes, parent_span_id_bytes,
                 name, start_time_unix_nano, end_time_unix_nano,
                 span_kind, attributes, status_code, status_msg=""):
    data = b""
    data += _encode_bytes_field(1, trace_id_bytes)
    data += _encode_bytes_field(2, span_id_bytes)
    if parent_span_id_bytes:
        data += _encode_bytes_field(3, parent_span_id_bytes)
    data += _encode_string_field(4, name)
    data += _encode_int_field(5, span_kind)
    data += _encode_int_field(6, start_time_unix_nano)
    data += _encode_int_field(7, end_time_unix_nano)
    for key, value in attributes.items():
        kv = _encode_key_value(key, str(value))
        data += _encode_message_field(9, kv)
    status = _encode_status(status_code, status_msg)
    data += _encode_message_field(12, status)
    return data

def _encode_resource_spans(service_name, scope_spans_bytes):
    svc_kv = _encode_key_value("service.name", service_name)
    resource = _encode_message_field(1, svc_kv)
    data = _encode_message_field(1, resource)
    data += _encode_message_field(2, scope_spans_bytes)
    return data

def _encode_scope_spans(scope_name, spans_bytes_list):
    scope = _encode_string_field(1, scope_name)
    data = _encode_message_field(1, scope)
    for span_bytes in spans_bytes_list:
        data += _encode_message_field(2, span_bytes)
    return data

def _encode_traces_data(resource_spans_bytes_list):
    data = b""
    for rs_bytes in resource_spans_bytes_list:
        data += _encode_message_field(1, rs_bytes)
    return data


# ========== 杈呭姪鍑芥暟 ==========

def _gen_trace_id():
    return uuid.uuid4().bytes

def _gen_span_id():
    return uuid.uuid4().bytes[:8]

def _env(name, default=""):
    """鑾峰彇鐜鍙橀噺锛屾敮鎸?GITEA_ 鍜?GITHUB_ 鍓嶇紑"""
    val = os.getenv(name, "")
    if val:
        return val
    # 灏濊瘯鍙︿竴绉嶅墠缂€
    if name.startswith("GITEA_"):
        alt = "GITHUB_" + name[6:]
        return os.getenv(alt, default)
    if name.startswith("GITHUB_"):
        alt = "GITEA_" + name[7:]
        return os.getenv(alt, default)
    return default


def _get_pr_number():
    """浠庣幆澧冨彉閲忔垨浜嬩欢鏂囦欢涓幏鍙?PR 鍙?""
    # 鐩存帴浠庣幆澧冨彉閲?    pr = os.getenv("PR_NUMBER", "") or os.getenv("GITEA_PR_NUMBER", "")
    if pr:
        return pr

    # 浠?GITHUB_EVENT_PATH 璇诲彇
    event_path = os.getenv("GITHUB_EVENT_PATH", "") or os.getenv("GITEA_EVENT_PATH", "")
    if event_path and os.path.isfile(event_path):
        try:
            with open(event_path, "r") as f:
                event = json.load(f)
            if "pull_request" in event and "number" in event["pull_request"]:
                return str(event["pull_request"]["number"])
        except Exception:
            pass

    return ""


def _get_ci_attributes():
    """浠?CI 鐜鍙橀噺涓敹闆嗗睘鎬?""
    attrs = {
        "ci.repo": _env("GITEA_REPOSITORY") or _env("GITHUB_REPOSITORY") or "unknown",
        "ci.workflow": _env("GITEA_WORKFLOW") or _env("GITHUB_WORKFLOW") or "unknown",
        "ci.job": _env("GITEA_JOB") or _env("GITHUB_JOB") or "unknown",
        "ci.commit_sha": _env("GITEA_SHA") or _env("GITHUB_SHA") or "unknown",
        "ci.branch": _env("GITEA_REF_NAME") or _env("GITHUB_REF_NAME") or "unknown",
        "ci.run_id": _env("GITEA_RUN_ID") or _env("GITHUB_RUN_ID") or "unknown",
        "ci.actor": _env("GITEA_ACTOR") or _env("GITHUB_ACTOR") or "unknown",
        "ci.event": _env("GITEA_EVENT_NAME") or _env("GITHUB_EVENT_NAME") or "unknown",
    }
    pr = _get_pr_number()
    if pr:
        attrs["ci.pr_number"] = pr
    return attrs


# ========== 鏍稿績涓婃姤閫昏緫 ==========

def build_trace(service_name, trace_name, status, duration_ms, attributes=None):
    """鏋勫缓涓€鏉″畬鏁寸殑 Trace 鏁版嵁锛岃繑鍥?protobuf bytes銆?""
    trace_id = _gen_trace_id()
    end_time = int(time.time() * 1e9)
    start_time = end_time - int(duration_ms * 1e6)
    status_code = 1 if status in ("ok", "running") else 2
    status_msg = "" if status in ("ok", "running") else "Job failed"

    main_attrs = {
        "agent.trace_name": trace_name,
        "agent.service": service_name,
        "ci.trace_status": status,
    }
    if attributes:
        main_attrs.update(attributes)

    main_span = _encode_span(
        trace_id_bytes=trace_id,
        span_id_bytes=_gen_span_id(),
        parent_span_id_bytes=b"",
        name=trace_name,
        start_time_unix_nano=start_time,
        end_time_unix_nano=end_time,
        span_kind=1,
        attributes=main_attrs,
        status_code=status_code,
        status_msg=status_msg,
    )

    scope_spans = _encode_scope_spans("ci-trace", [main_span])
    resource_spans = _encode_resource_spans(service_name, scope_spans)
    return _encode_traces_data([resource_spans])


def report_ci_trace(service_name, trace_name, status="ok", duration_ms=1000,
                    endpoint=None, license_key=None, project=None, workspace=None,
                    extra_attributes=None):
    """
    涓婃姤 CI Trace 鏁版嵁銆傝繑鍥?(success: bool, message: str)
    姘歌繙涓嶄細鎶涘嚭寮傚父锛屽け璐ヤ篃杩斿洖 False銆?    """
    try:
        endpoint = endpoint or os.getenv("AGENTLOOP_ENDPOINT", DEFAULT_ENDPOINT)
        license_key = license_key or os.getenv("AGENTLOOP_LICENSE_KEY", "")
        project = project or os.getenv("AGENTLOOP_PROJECT", DEFAULT_PROJECT)
        workspace = workspace or os.getenv("AGENTLOOP_WORKSPACE", DEFAULT_WORKSPACE)

        if not license_key:
            return False, "[Trace] 璺宠繃涓婃姤锛氭湭閰嶇疆 AGENTLOOP_LICENSE_KEY"

        # 鏀堕泦 CI 灞炴€?        attrs = _get_ci_attributes()
        if extra_attributes:
            attrs.update(extra_attributes)

        payload = build_trace(
            service_name=service_name,
            trace_name=trace_name,
            status=status,
            duration_ms=duration_ms,
            attributes=attrs,
        )

        headers = {
            "Content-Type": "application/x-protobuf",
            "x-arms-license-key": license_key,
            "x-arms-project": project,
            "x-cms-workspace": workspace,
        }

        req = urllib.request.Request(endpoint, data=payload, headers=headers, method="POST")
        try:
            with urllib.request.urlopen(req, timeout=10) as resp:
                status_code = resp.status
                resp_body = resp.read().decode("utf-8", errors="replace")
        except urllib.error.HTTPError as e:
            status_code = e.code
            resp_body = e.read().decode("utf-8", errors="replace")

        if status_code in (200, 202):
            return True, f"[Trace] 鉁?涓婃姤鎴愬姛: {service_name} / {trace_name} ({status}, {duration_ms}ms)"
        else:
            return False, f"[Trace] 鈿?涓婃姤澶辫触: HTTP {status_code} - {resp_body[:200]}"
    except Exception as e:
        return False, f"[Trace] 鈿?涓婃姤寮傚父: {type(e).__name__}: {str(e)}"


def main():
    parser = argparse.ArgumentParser(description="CI AgentLoop Trace 涓婃姤")
    parser.add_argument("--service", dest="service_name",
                        default=os.getenv("TRACE_SERVICE", ""),
                        help="鏈嶅姟鍚嶇О锛堜篃鍙€氳繃 TRACE_SERVICE 鐜鍙橀噺璁剧疆锛?)
    parser.add_argument("--name", dest="trace_name",
                        default=os.getenv("TRACE_NAME", ""),
                        help="Trace 鍚嶇О锛堜篃鍙€氳繃 TRACE_NAME 鐜鍙橀噺璁剧疆锛?)
    parser.add_argument("--status", default=os.getenv("TRACE_STATUS", "ok"),
                        choices=["ok", "error", "running"],
                        help="鐘舵€? ok / error / running锛堥粯璁?ok锛?)
    parser.add_argument("--start-time", dest="start_time",
                        default=os.getenv("TRACE_START_TIME", ""),
                        help="寮€濮嬫椂闂存埑锛堢锛夛紝鐢ㄤ簬璁＄畻 duration锛涗笉濉垯鐢ㄩ粯璁?1s")
    parser.add_argument("--duration-ms", dest="duration_ms", type=int, default=0,
                        help="鐩存帴鎸囧畾鑰楁椂锛堟绉掞級锛屼紭鍏堢骇楂樹簬 --start-time")
    parser.add_argument("--attrs", default="",
                        help="闄勫姞灞炴€э紙JSON 瀛楃涓诧級")

    args = parser.parse_args()

    # 蹇呴』鍙傛暟妫€鏌?    if not args.service_name:
        print("[Trace] 鈿?璺宠繃涓婃姤锛氭湭鎸囧畾 service锛?-service 鎴?TRACE_SERVICE锛?)
        sys.exit(0)

    # 璁＄畻鑰楁椂
    duration_ms = args.duration_ms
    if duration_ms <= 0 and args.start_time:
        try:
            start_ts = float(args.start_time)
            duration_ms = int((time.time() - start_ts) * 1000)
        except (ValueError, TypeError):
            duration_ms = 1000
    if duration_ms <= 0:
        duration_ms = 1000  # 榛樿 1 绉?
    # 瑙ｆ瀽闄勫姞灞炴€?    extra_attrs = {}
    if args.attrs:
        try:
            extra_attrs = json.loads(args.attrs)
        except json.JSONDecodeError:
            pass

    # 鐢熸垚 trace_name锛氬鏋滄病鎸囧畾锛岀敤 workflow+job
    trace_name = args.trace_name
    if not trace_name:
        wf = _env("GITEA_WORKFLOW") or _env("GITHUB_WORKFLOW") or "CI"
        job = _env("GITEA_JOB") or _env("GITHUB_JOB") or "job"
        trace_name = f"{wf} / {job}"

    success, msg = report_ci_trace(
        service_name=args.service_name,
        trace_name=trace_name,
        status=args.status,
        duration_ms=duration_ms,
        extra_attributes=extra_attrs,
    )

    print(msg)
    # 姘歌繙 exit 0锛屼笉褰卞搷 CI
    sys.exit(0)


if __name__ == "__main__":
    main()