DiAL/src/server/checker/runner/icmp/execute.ts

import { isError } from "es-toolkit";

import type { CheckResult, RawTargetConfig } from "../../types";
import type { CheckerContext, CheckerDefinition, CheckerValidationInput, ResolveContext } from "../types";
import type { PingExpectConfig, PingStats, PingTargetConfig, ResolvedPingTarget } from "./types";

import { errorFailure } from "../../expect/failure";
import { checkValueMatcher } from "../../expect/matcher";
import { buildPingCommand } from "./command";
import { checkAlive, checkAvgLatency, checkMaxLatency, checkPacketLoss } from "./expect";
import { parsePingOutput } from "./parse";
import { icmpCheckerSchemas } from "./schema";
import { validatePingConfig } from "./validate";

const DEFAULT_COUNT = 3;
const DEFAULT_PACKET_SIZE = 56;

export class IcmpChecker implements CheckerDefinition<ResolvedPingTarget> {
  readonly configKey = "ping";

  readonly schemas = icmpCheckerSchemas;

  readonly type = "ping";

  async execute(t: ResolvedPingTarget, ctx: CheckerContext): Promise<CheckResult> {
    const timestamp = new Date().toISOString();
    const start = performance.now();
    let proc: ReturnType<typeof Bun.spawn>;

    try {
      proc = Bun.spawn(buildPingCommand(t), {
        stderr: "pipe",
        stdin: "ignore",
        stdout: "pipe",
      });
    } catch (error) {
      const durationMs = Math.round(performance.now() - start);
      return {
        durationMs,
        failure: errorFailure("ping", "spawn", `ping 命令不可用: ${isError(error) ? error.message : String(error)}`),
        matched: false,
        statusDetail: "ping command not found",
        targetId: t.id,
        timestamp,
      };
    }

    ctx.signal.addEventListener(
      "abort",
      () => {
        try {
          proc.kill();
        } catch {
          /* best-effort kill */
        }
      },
      { once: true },
    );

    const stdout = await readStream(proc.stdout as ReadableStream<Uint8Array>);
    await proc.exited;

    const durationMs = Math.round(performance.now() - start);
    if (ctx.signal.aborted) {
      return {
        durationMs,
        failure: errorFailure("ping", "timeout", `ping 执行超时 (${t.timeoutMs}ms)`),
        matched: false,
        statusDetail: null,
        targetId: t.id,
        timestamp,
      };
    }

    const stats = parsePingOutput(stdout, process.platform);
    if (!stats) {
      return {
        durationMs,
        failure: errorFailure("ping", "parse", "无法解析 ping 输出"),
        matched: false,
        statusDetail: truncateOutput(stdout),
        targetId: t.id,
        timestamp,
      };
    }

    const result = checkStats(stats, t.expect, durationMs);
    return {
      durationMs,
      failure: result.failure,
      matched: result.matched,
      statusDetail: buildStatusDetail(stats),
      targetId: t.id,
      timestamp,
    };
  }

  resolve(target: RawTargetConfig, context: ResolveContext): ResolvedPingTarget {
    const t = target as RawTargetConfig & { ping: PingTargetConfig; type: "ping" };
    return {
      description: null,
      expect: target.expect as PingExpectConfig | undefined,
      group: target.group ?? "default",
      id: t.id,
      intervalMs: context.defaultIntervalMs,
      name: t.name ?? null,
      ping: {
        count: t.ping.count ?? DEFAULT_COUNT,
        host: t.ping.host,
        packetSize: t.ping.packetSize ?? DEFAULT_PACKET_SIZE,
      },
      timeoutMs: context.defaultTimeoutMs,
      type: "ping",
    } satisfies ResolvedPingTarget;
  }

  serialize(t: ResolvedPingTarget): { config: string; target: string } {
    return {
      config: JSON.stringify(t.ping),
      target: `ping ${t.ping.host}`,
    };
  }

  validate(input: CheckerValidationInput) {
    return validatePingConfig(input);
  }
}

function buildStatusDetail(stats: PingStats): string {
  if (!stats.alive) return `unreachable (${stats.received}/${stats.transmitted} received)`;
  const avg = stats.avgLatencyMs === null ? "n/a" : formatNumber(stats.avgLatencyMs);
  const loss = formatNumber(stats.packetLoss);
  let detail = `alive, avg ${avg}ms, loss ${loss}% (${stats.received}/${stats.transmitted})`;
  if (stats.packetLoss > 0 && stats.maxLatencyMs !== null) {
    detail = `${detail}, max ${formatNumber(stats.maxLatencyMs)}ms`;
  }
  return detail;
}

function checkStats(stats: PingStats, expect: PingExpectConfig | undefined, durationMs: number) {
  const aliveResult = checkAlive(stats.alive, expect?.alive ?? true);
  if (!aliveResult.matched) return aliveResult;
  const packetLossResult = checkPacketLoss(stats.packetLoss, expect?.packetLossPercent);
  if (!packetLossResult.matched) return packetLossResult;
  const avgLatencyResult = checkAvgLatency(stats.avgLatencyMs, expect?.avgLatencyMs);
  if (!avgLatencyResult.matched) return avgLatencyResult;
  const maxLatencyResult = checkMaxLatency(stats.maxLatencyMs, expect?.maxLatencyMs);
  if (!maxLatencyResult.matched) return maxLatencyResult;
  return checkValueMatcher(durationMs, expect?.durationMs, {
    message: "durationMs mismatch",
    path: "durationMs",
    phase: "duration",
  });
}

function formatNumber(value: number): string {
  return Number.isInteger(value) ? String(value) : String(Number(value.toFixed(3)));
}

async function readStream(stream: ReadableStream<Uint8Array>): Promise<string> {
  const reader = stream.getReader();
  const decoder = new TextDecoder();
  let text = "";
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      text += decoder.decode(value, { stream: true });
    }
    text += decoder.decode();
  } catch {
    /* stream already closed */
  } finally {
    try {
      reader.releaseLock();
    } catch {
      /* already released */
    }
  }
  return text;
}

function truncateOutput(output: string, maxLen = 80): string {
  if (output.length <= maxLen) return output;
  return `${output.slice(0, maxLen)}…`;
}