clusterflux-public/scripts/source-preparation-smoke.js

#!/usr/bin/env node

const assert = require("assert");
const cp = require("child_process");
const net = require("net");
const path = require("path");
const { coordinatorWireRequest } = require("./coordinator-wire");
const { nodeIdentity, signedNodeRequest } = require("./node-signing");

const repo = path.resolve(__dirname, "..");

function waitForJsonLine(child) {
  return new Promise((resolve, reject) => {
    let buffer = "";
    child.stdout.on("data", (chunk) => {
      buffer += chunk.toString();
      const newline = buffer.indexOf("\n");
      if (newline < 0) return;
      try {
        resolve(JSON.parse(buffer.slice(0, newline).trim()));
      } catch (error) {
        reject(error);
      }
    });
    child.once("exit", (code) => {
      reject(new Error(`process exited before JSON line with code ${code}`));
    });
  });
}

function send(addr, message) {
  return new Promise((resolve, reject) => {
    const socket = net.connect(addr.port, addr.host, () => {
      socket.write(`${JSON.stringify(coordinatorWireRequest(message))}\n`);
    });
    let buffer = "";
    socket.on("data", (chunk) => {
      buffer += chunk.toString();
      const newline = buffer.indexOf("\n");
      if (newline < 0) return;
      socket.end();
      try {
        resolve(JSON.parse(buffer.slice(0, newline)));
      } catch (error) {
        reject(error);
      }
    });
    socket.on("error", reject);
  });
}

function sourceCapableNode(sourceProviders = ["git"]) {
  return {
    os: "Linux",
    arch: "x86_64",
    capabilities: ["Command", "SourceFilesystem", "SourceGit"],
    environment_backends: [],
    source_providers: sourceProviders
  };
}

(async () => {
  const coordinator = cp.spawn(
    "cargo",
    [
      "run",
      "-q",
      "-p",
      "clusterflux-coordinator",
      "--bin",
      "clusterflux-coordinator",
      "--",
      "--listen",
      "127.0.0.1:0",
      "--allow-local-trusted-loopback"
    ],
    { cwd: repo }
  );
  let coordinatorStderr = "";
  coordinator.stderr.on("data", (chunk) => {
    coordinatorStderr += chunk.toString();
  });

  try {
    const ready = await waitForJsonLine(coordinator);
    const [host, portText] = ready.listen.split(":");
    const addr = { host, port: Number(portText) };
    assert.strictEqual((await send(addr, { type: "ping" })).type, "pong");

    const pending = await send(addr, {
      type: "request_source_preparation",
      tenant: "tenant",
      project: "project",
      provider: "Git"
    });
    assert.strictEqual(pending.type, "source_preparation");
    assert.strictEqual(pending.status.preparation.tenant, "tenant");
    assert.strictEqual(pending.status.preparation.project, "project");
    assert.strictEqual(pending.status.preparation.provider, "Git");
    assert.strictEqual(
      pending.status.preparation.coordinator_requires_checkout_access,
      false
    );
    assert.deepStrictEqual(pending.status.preparation.required_capabilities, [
      "SourceGit"
    ]);
    assert.match(pending.status.disposition.Pending.reason, /waiting|node/i);

    const nodeIdentities = new Map();
    for (const node of ["source-cold", "source-ready"]) {
      const identity = nodeIdentity("source-preparation-smoke", node);
      nodeIdentities.set(node, identity);
      const attached = await send(addr, {
        type: "attach_node",
        tenant: "tenant",
        project: "project",
        node,
        public_key: identity.publicKey
      });
      assert.strictEqual(attached.type, "node_attached");
    }

    const cold = await send(addr, signedNodeRequest("source-cold", nodeIdentities.get("source-cold"), "report_node_capabilities", {
      type: "report_node_capabilities",
      tenant: "tenant",
      project: "project",
      node: "source-cold",
      capabilities: sourceCapableNode(),
      cached_environment_digests: [],
      source_snapshots: [],
      artifact_locations: [],
      direct_connectivity: true,
      online: true
    }));
    assert.strictEqual(cold.type, "node_capabilities_recorded");

    const readyReport = await send(addr, signedNodeRequest("source-ready", nodeIdentities.get("source-ready"), "report_node_capabilities", {
      type: "report_node_capabilities",
      tenant: "tenant",
      project: "project",
      node: "source-ready",
      capabilities: sourceCapableNode(),
      cached_environment_digests: [],
      source_snapshots: [],
      artifact_locations: [],
      direct_connectivity: true,
      online: true
    }));
    assert.strictEqual(readyReport.type, "node_capabilities_recorded");

    const assigned = await send(addr, {
      type: "request_source_preparation",
      tenant: "tenant",
      project: "project",
      provider: "Git"
    });
    assert.strictEqual(assigned.type, "source_preparation");
    assert(["source-cold", "source-ready"].includes(assigned.status.disposition.Assigned.node));
    assert.strictEqual(
      assigned.status.preparation.coordinator_requires_checkout_access,
      false
    );

    const crossTenantCompletion = await send(addr, signedNodeRequest("source-ready", nodeIdentities.get("source-ready"), "complete_source_preparation", {
      type: "complete_source_preparation",
      tenant: "other",
      project: "project",
      node: "source-ready",
      provider: "Git",
      source_snapshot: "sha256:source-prepared"
    }));
    assert.strictEqual(crossTenantCompletion.type, "error");
    assert.match(crossTenantCompletion.message, /tenant\/project scope/i);

    const completed = await send(addr, signedNodeRequest("source-ready", nodeIdentities.get("source-ready"), "complete_source_preparation", {
      type: "complete_source_preparation",
      tenant: "tenant",
      project: "project",
      node: "source-ready",
      provider: "Git",
      source_snapshot: "sha256:source-prepared"
    }));
    assert.strictEqual(completed.type, "source_preparation_completed");
    assert.strictEqual(completed.node, "source-ready");
    assert.strictEqual(completed.provider, "Git");
    assert.strictEqual(completed.source_snapshot, "sha256:source-prepared");

    const placement = await send(addr, {
      type: "schedule_task",
      tenant: "tenant",
      project: "project",
      environment: null,
      environment_digest: null,
      required_capabilities: ["SourceGit"],
      source_snapshot: "sha256:source-prepared",
      required_artifacts: [],
      prefer_node: null
    });
    assert.strictEqual(placement.type, "task_placement");
    assert.strictEqual(placement.placement.node, "source-ready");
    assert(
      placement.placement.reasons.includes("source snapshot already local"),
      "completed source preparation must update node source locality"
    );
  } catch (error) {
    if (coordinatorStderr) {
      error.message = `${error.message}\ncoordinator stderr:\n${coordinatorStderr}`;
    }
    throw error;
  } finally {
    coordinator.kill("SIGTERM");
  }

  console.log("Source preparation smoke passed");
})().catch((error) => {
  console.error(error.stack || error.message);
  process.exit(1);
});