wenlanai/scripts/host-agent.ts

import os from 'os';
import { execFile } from 'child_process';
import { promisify } from 'util';

const execFileAsync = promisify(execFile);

type AgentJob = {
  runId: string;
  traceId: string;
  targetId: string;
  targetName: string;
  serviceKey: string;
  type: string;
  endpoint: string;
  config: Record<string, unknown>;
};

type HeartbeatResponse = {
  ok: boolean;
  control: {
    enabled: boolean;
    monitoringEnabled: boolean;
    logIngestEnabled: boolean;
    proactivePushEnabled: boolean;
    heartbeatIntervalMs: number;
  };
};

const API_BASE =
  (process.env.MONITORING_API_BASE_URL ?? 'http://localhost:3001').replace(
    /\/$/,
    '',
  );
const AGENT_KEY = process.env.MONITORING_AGENT_KEY ?? 'local-windows-agent';
const AGENT_TOKEN =
  process.env.MONITORING_AGENT_TOKEN ?? 'sentai-agent-token';
const POLL_INTERVAL_MS = Number(process.env.MONITORING_AGENT_POLL_MS ?? 5000);
const RUN_ONCE = process.argv.includes('--once');

async function request<T>(path: string, init?: RequestInit) {
  const headers = new Headers(init?.headers);
  headers.set('Content-Type', 'application/json');
  headers.set('x-monitoring-agent-token', AGENT_TOKEN);

  const response = await fetch(`${API_BASE}${path}`, {
    ...init,
    headers,
  });

  if (!response.ok) {
    const text = await response.text();
    throw new Error(`HTTP_${response.status}: ${text || response.statusText}`);
  }

  return (await response.json()) as T;
}

async function powershellJson<T>(script: string) {
  const { stdout } = await execFileAsync('powershell', [
    '-NoProfile',
    '-Command',
    script,
  ]);

  const raw = stdout.trim();
  if (!raw) {
    return [] as T[];
  }

  const parsed = JSON.parse(raw) as T | T[];
  return Array.isArray(parsed) ? parsed : [parsed];
}

async function findJavaProcess(config: Record<string, unknown>) {
  const matchMode = String(config.processMatchMode ?? 'command');
  if (matchMode === 'pid') {
    const pid = Number(config.processPid ?? config.processMatchValue ?? 0);
    if (!pid) {
      throw new Error('host-agent 监测要求填写 processPid 或 processMatchValue');
    }
    const [process] = await powershellJson<{
      ProcessId: number;
      Name: string;
      CommandLine: string;
    }>(
      `Get-CimInstance Win32_Process -Filter "ProcessId = ${pid}" | Select-Object ProcessId,Name,CommandLine | ConvertTo-Json -Compress`,
    );
    if (!process) {
      throw new Error(`未找到 PID=${pid} 的 Java 进程`);
    }
    return process;
  }

  const keyword = String(config.processMatchValue ?? '').trim();
  if (!keyword) {
    throw new Error('host-agent 监测要求填写 processMatchValue');
  }

  const escaped = keyword.replace(/'/g, "''");
  const [process] = await powershellJson<{
    ProcessId: number;
    Name: string;
    CommandLine: string;
  }>(
    `$keyword='${escaped}'; Get-CimInstance Win32_Process | Where-Object { $_.Name -match '^java' -and $_.CommandLine -like "*$keyword*" } | Select-Object -First 1 ProcessId,Name,CommandLine | ConvertTo-Json -Compress`,
  );

  if (!process) {
    throw new Error(`未找到命令行包含 ${keyword} 的 Java 进程`);
  }

  return process;
}

async function tailLog(config: Record<string, unknown>) {
  const logFilePath = String(config.logFilePath ?? '').trim();
  if (!logFilePath) {
    return null;
  }

  const tailLines = Number(config.logTailLines ?? 80);
  const keyword = String(config.logKeyword ?? 'ERROR|WARN|Exception|Timeout');
  const script = `$path='${logFilePath.replace(/'/g, "''")}'; if (Test-Path $path) { Get-Content $path -Tail ${tailLines} | Select-String -Pattern '${keyword.replace(/'/g, "''")}' | Select-Object -Last 20 | ForEach-Object { $_.Line } }`;
  const { stdout } = await execFileAsync('powershell', ['-NoProfile', '-Command', script]);
  const output = stdout.trim();
  return output || null;
}

async function runJcmd(pid: number, command: string, jcmdPath: string) {
  const parts = command.split(' ').filter(Boolean);
  const { stdout, stderr } = await execFileAsync(jcmdPath, [
    String(pid),
    ...parts,
  ]);
  return (stdout || stderr).trim();
}

async function runJstat(pid: number, options: string[], jstatPath: string) {
  const { stdout, stderr } = await execFileAsync(jstatPath, [
    ...options,
    String(pid),
  ]);
  return (stdout || stderr).trim();
}

async function executeJavaJob(job: AgentJob) {
  const config = job.config;
  const process = await findJavaProcess(config);
  const jcmdPath = String(
    config.jcmdPath ?? 'D:\\Antigravity\\soft\\Java\\bin\\jcmd.exe',
  );
  const jstatPath = String(
    config.jstatPath ?? 'D:\\Antigravity\\soft\\Java\\bin\\jstat.exe',
  );
  const jcmdCommands = Array.isArray(config.jcmdCommands)
    ? (config.jcmdCommands as string[])
    : ['VM.command_line', 'GC.heap_info'];
  const jstatOptions = Array.isArray(config.jstatOptions)
    ? (config.jstatOptions as string[])
    : ['-gcutil'];

  const jcmdResults: Record<string, string> = {};
  for (const command of jcmdCommands) {
    jcmdResults[command] = await runJcmd(process.ProcessId, command, jcmdPath);
  }

  let jstatResult = '';
  try {
    jstatResult = await runJstat(process.ProcessId, jstatOptions, jstatPath);
  } catch (error) {
    jstatResult = error instanceof Error ? error.message : 'jstat failed';
  }

  const logExcerpt = await tailLog(config);

  return {
    status: 'success' as const,
    errorTag: null,
    summary: '宿主机 Agent 监测完成',
    failureReason: null,
    metrics: {
      pid: process.ProcessId,
      host: os.hostname(),
      processCommand: process.CommandLine,
      jstatResult,
    },
    details: {
      process,
      jcmdResults,
      agentKey: AGENT_KEY,
      nodeKey: `${job.serviceKey}-node-1`,
    },
    logExcerpt,
  };
}

async function processNextJob() {
  const claim = await request<{ job: AgentJob | null }>(
    '/api/monitoring/agent/jobs/claim',
    {
      method: 'POST',
      body: JSON.stringify({ agentKey: AGENT_KEY }),
    },
  );

  if (!claim.job) {
    return false;
  }

  try {
    const result =
      claim.job.type === 'java-app'
        ? await executeJavaJob(claim.job)
        : {
            status: 'failed' as const,
            errorTag: 'UNSUPPORTED_AGENT_TARGET',
            summary: '当前 Agent 暂未支持该目标类型',
            failureReason: `type=${claim.job.type} 暂未实现`,
            metrics: {},
            details: { agentKey: AGENT_KEY },
            logExcerpt: null,
          };

    await request(`/api/monitoring/agent/jobs/${claim.job.runId}/result`, {
      method: 'POST',
      body: JSON.stringify(result),
    });
  } catch (error) {
    await request(`/api/monitoring/agent/jobs/${claim.job.runId}/result`, {
      method: 'POST',
      body: JSON.stringify({
        status: 'failed',
        errorTag: 'AGENT_EXECUTION_ERROR',
        summary: '宿主机 Agent 执行失败',
        failureReason: error instanceof Error ? error.message : 'unknown error',
        metrics: {},
        details: { agentKey: AGENT_KEY },
        logExcerpt: null,
      }),
    });
  }

  return true;
}

async function main() {
  let heartbeatIntervalMs = POLL_INTERVAL_MS;

  do {
    const heartbeat = await request<HeartbeatResponse>('/api/monitoring/agent/heartbeat', {
      method: 'POST',
      body: JSON.stringify({
        agentKey: AGENT_KEY,
        host: os.hostname(),
        platform: process.platform,
        version: 'host-agent-tsx',
      }),
    });

    heartbeatIntervalMs = heartbeat.control.heartbeatIntervalMs ?? POLL_INTERVAL_MS;

    if (!heartbeat.control.enabled || !heartbeat.control.monitoringEnabled) {
      if (RUN_ONCE) {
        break;
      }
      await new Promise((resolve) => setTimeout(resolve, heartbeatIntervalMs));
      continue;
    }

    const handled = await processNextJob();
    if (RUN_ONCE) {
      break;
    }
    if (!handled) {
      await new Promise((resolve) => setTimeout(resolve, heartbeatIntervalMs));
    }
  } while (true);
}

void main().catch((error) => {
  console.error(error);
  process.exit(1);
});