feat: add orchestrator shadow observability

2026-07-24 21:44:21 +08:00
parent 24336d0178
commit 3e57f85d3a
18 changed files with 992 additions and 9 deletions
@@ -0,0 +1,254 @@
+import { createRemoteWorkflowEngine } from './engine-registry.mjs';
+
+const SHADOW_EVENT_TYPES = Object.freeze([
+  'workflow_shadow_completed',
+  'workflow_shadow_failed',
+]);
+const MAX_METRIC_EVENTS = 5000;
+
+function clampInteger(value, fallback, min, max) {
+  const parsed = Number(value);
+  if (!Number.isFinite(parsed)) return fallback;
+  return Math.min(max, Math.max(min, Math.floor(parsed)));
+}
+
+function parseJsonColumn(value, fallback = null) {
+  if (value == null || value === '') return fallback;
+  if (typeof value === 'object') return value;
+  try {
+    return JSON.parse(String(value));
+  } catch {
+    return fallback;
+  }
+}
+
+function percentile(values, ratio) {
+  if (!values.length) return null;
+  const sorted = [...values].sort((left, right) => left - right);
+  const index = Math.max(0, Math.ceil(sorted.length * ratio) - 1);
+  return sorted[index];
+}
+
+function normalizeRunId(value) {
+  const runId = String(value ?? '').trim();
+  return /^[a-zA-Z0-9_-]{1,128}$/.test(runId) ? runId : '';
+}
+
+function projectShadowEventRow(row) {
+  const data = parseJsonColumn(row.data_json, {}) ?? {};
+  const succeeded = row.event_type === 'workflow_shadow_completed';
+  const latency = Number(data.latencyMs);
+  return {
+    eventId: row.event_id,
+    runId: row.run_id,
+    requestId: row.request_id,
+    userId: row.user_id,
+    sessionId: row.agent_session_id ?? null,
+    nativeStatus: row.native_status,
+    nativeAttempts: Number(row.native_attempts ?? 0),
+    shadowStatus: succeeded ? 'succeeded' : 'failed',
+    engine: data.engine ?? 'langgraph',
+    configVersion: data.configVersion == null ? null : Number(data.configVersion),
+    phase: data.phase ?? null,
+    taskType: data.taskType ?? null,
+    executorAdapter: data.executorAdapter ?? null,
+    latencyMs: Number.isFinite(latency) && latency >= 0 ? latency : null,
+    error: succeeded ? null : {
+      code: data.code ?? 'WORKFLOW_SHADOW_FAILED',
+      message: data.message ?? 'Shadow observation failed',
+    },
+    observedAt: Number(row.event_created_at ?? 0),
+    nativeCompletedAt: row.native_completed_at == null
+      ? null
+      : Number(row.native_completed_at),
+  };
+}
+
+function summarizeRuns(runs, { capped = false } = {}) {
+  const successes = runs.filter((run) => run.shadowStatus === 'succeeded').length;
+  const failures = runs.length - successes;
+  const latencies = runs
+    .map((run) => run.latencyMs)
+    .filter((value) => Number.isFinite(value));
+  return {
+    observations: runs.length,
+    successes,
+    failures,
+    successRate: runs.length ? successes / runs.length : null,
+    failureRate: runs.length ? failures / runs.length : null,
+    latencyP50Ms: percentile(latencies, 0.5),
+    latencyP95Ms: percentile(latencies, 0.95),
+    nativeSucceeded: runs.filter((run) => run.nativeStatus === 'succeeded').length,
+    nativeFailed: runs.filter((run) => run.nativeStatus === 'failed').length,
+    lastObservedAt: runs[0]?.observedAt ?? null,
+    sampled: capped,
+  };
+}
+
+function safeRemoteError(error) {
+  return {
+    code: String(error?.code ?? 'ORCHESTRATOR_UNAVAILABLE').slice(0, 128),
+    message: String(error instanceof Error ? error.message : error).slice(0, 1000),
+  };
+}
+
+export function createOrchestratorObservabilityService({
+  pool,
+  configService,
+  serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN,
+  fetchImpl = globalThis.fetch,
+} = {}) {
+  if (!pool?.query) throw new Error('Orchestrator observability requires a database pool');
+  if (!configService?.getRuntimeState) {
+    throw new Error('Orchestrator observability requires config service');
+  }
+
+  async function loadShadowEvents({ hours = 24 } = {}) {
+    const normalizedHours = clampInteger(hours, 24, 1, 24 * 30);
+    const from = Date.now() - normalizedHours * 60 * 60 * 1000;
+    const [rows] = await pool.query(
+      `SELECT
+         e.id AS event_id,
+         e.run_id,
+         e.event_type,
+         e.data_json,
+         e.created_at AS event_created_at,
+         r.request_id,
+         r.user_id,
+         r.agent_session_id,
+         r.status AS native_status,
+         r.attempts AS native_attempts,
+         r.completed_at AS native_completed_at
+       FROM h5_agent_run_events e
+       INNER JOIN h5_agent_runs r ON r.id = e.run_id
+       WHERE e.event_type IN (?, ?)
+         AND e.created_at >= ?
+       ORDER BY e.created_at DESC, e.id DESC
+       LIMIT ?`,
+      [...SHADOW_EVENT_TYPES, from, MAX_METRIC_EVENTS],
+    );
+    return {
+      from,
+      hours: normalizedHours,
+      capped: rows.length >= MAX_METRIC_EVENTS,
+      runs: rows.map(projectShadowEventRow),
+    };
+  }
+
+  return {
+    async listShadowRuns({
+      hours = 24,
+      limit = 50,
+      status = 'all',
+    } = {}) {
+      const loaded = await loadShadowEvents({ hours });
+      const normalizedLimit = clampInteger(limit, 50, 1, 200);
+      const normalizedStatus = ['succeeded', 'failed'].includes(status) ? status : 'all';
+      const filtered = normalizedStatus === 'all'
+        ? loaded.runs
+        : loaded.runs.filter((run) => run.shadowStatus === normalizedStatus);
+      return {
+        generatedAt: Date.now(),
+        window: {
+          hours: loaded.hours,
+          from: loaded.from,
+        },
+        metrics: summarizeRuns(loaded.runs, { capped: loaded.capped }),
+        runs: filtered.slice(0, normalizedLimit),
+      };
+    },
+
+    async getShadowRun(runId) {
+      const normalizedRunId = normalizeRunId(runId);
+      if (!normalizedRunId) return null;
+      const [runRows] = await pool.query(
+        `SELECT id, request_id, user_id, agent_session_id, status, attempts,
+                error_message, created_at, updated_at, started_at, completed_at
+           FROM h5_agent_runs
+          WHERE id = ?
+          LIMIT 1`,
+        [normalizedRunId],
+      );
+      if (!runRows[0]) return null;
+      const [eventRows] = await pool.query(
+        `SELECT id AS event_id, event_type, data_json, created_at
+           FROM h5_agent_run_events
+          WHERE run_id = ? AND event_type IN (?, ?)
+          ORDER BY created_at ASC, id ASC`,
+        [normalizedRunId, ...SHADOW_EVENT_TYPES],
+      );
+      const native = runRows[0];
+      const localEvents = eventRows.map((row) => ({
+        eventId: row.event_id,
+        type: row.event_type,
+        data: parseJsonColumn(row.data_json, null),
+        createdAt: Number(row.created_at),
+      }));
+
+      let remote = {
+        available: false,
+        state: null,
+        events: [],
+        error: null,
+      };
+      try {
+        const runtimeState = await configService.getRuntimeState();
+        if (runtimeState.config.serviceUrl) {
+          const engine = createRemoteWorkflowEngine({
+            baseUrl: runtimeState.config.serviceUrl,
+            serviceToken,
+            timeoutMs: runtimeState.config.requestTimeoutMs,
+            fetchImpl,
+          });
+          const [state, events] = await Promise.all([
+            engine.getState(normalizedRunId),
+            (async () => {
+              const collected = [];
+              for await (const event of engine.streamEvents(normalizedRunId)) {
+                collected.push(event);
+                if (collected.length >= 500) break;
+              }
+              return collected;
+            })(),
+          ]);
+          remote = {
+            available: true,
+            state,
+            events,
+            error: null,
+          };
+        }
+      } catch (error) {
+        remote.error = safeRemoteError(error);
+      }
+
+      return {
+        native: {
+          runId: native.id,
+          requestId: native.request_id,
+          userId: native.user_id,
+          sessionId: native.agent_session_id ?? null,
+          status: native.status,
+          attempts: Number(native.attempts ?? 0),
+          error: native.error_message ?? null,
+          createdAt: Number(native.created_at ?? 0),
+          updatedAt: Number(native.updated_at ?? 0),
+          startedAt: native.started_at == null ? null : Number(native.started_at),
+          completedAt: native.completed_at == null ? null : Number(native.completed_at),
+        },
+        localEvents,
+        remote,
+      };
+    },
+  };
+}
+
+export const orchestratorObservabilityInternals = {
+  MAX_METRIC_EVENTS,
+  SHADOW_EVENT_TYPES,
+  normalizeRunId,
+  parseJsonColumn,
+  percentile,
+  projectShadowEventRow,
+  summarizeRuns,
+};