clusterflux-public/scripts/scheduler-placement-smoke.js

#!/usr/bin/env node

const assert = require("assert");
const cp = require("child_process");
const net = require("net");
const path = require("path");

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(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 linuxCapabilities() {
  return {
    os: "Linux",
    arch: "x86_64",
    capabilities: [
      "Command",
      "Containers",
      "RootlessPodman",
      "VfsArtifacts"
    ],
    environment_backends: ["Container"],
    source_providers: ["filesystem"]
  };
}

async function attachNode(addr, node) {
  const attached = await send(addr, {
    type: "attach_node",
    tenant: "tenant",
    project: "project",
    node,
    public_key: `${node}-public-key`
  });
  assert.strictEqual(attached.type, "node_attached");
  assert.strictEqual(attached.node, node);
}

async function reportNode(addr, node, locality) {
  const recorded = await send(addr, {
    type: "report_node_capabilities",
    tenant: "tenant",
    project: "project",
    node,
    capabilities: linuxCapabilities(),
    cached_environment_digests: locality.cached_environment_digests,
    dependency_cache_digests: locality.dependency_cache_digests,
    source_snapshots: locality.source_snapshots,
    artifact_locations: locality.artifact_locations,
    direct_connectivity: locality.direct_connectivity !== false,
    online: true
  });
  assert.strictEqual(recorded.type, "node_capabilities_recorded");
  assert.strictEqual(recorded.node, node);
  return recorded;
}

(async () => {
  const coordinator = cp.spawn(
    "cargo",
    [
      "run",
      "-q",
      "-p",
      "disasmer-coordinator",
      "--bin",
      "disasmer-coordinator",
      "--",
      "--listen",
      "127.0.0.1:0"
    ],
    { 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");

    await attachNode(addr, "cold-node");
    await attachNode(addr, "warm-node");

    const cold = await reportNode(addr, "cold-node", {
      cached_environment_digests: [],
      dependency_cache_digests: [],
      source_snapshots: [],
      artifact_locations: [],
      direct_connectivity: false
    });
    assert.strictEqual(cold.node_descriptors, 1);

    const warm = await reportNode(addr, "warm-node", {
      cached_environment_digests: ["sha256:env-linux-container"],
      dependency_cache_digests: ["sha256:deps-toolchain"],
      source_snapshots: ["sha256:source-tree"],
      artifact_locations: ["toolchain-cache"]
    });
    assert.strictEqual(warm.node_descriptors, 2);
    const reportedNodes = new Set([cold.node, warm.node]);
    assert.strictEqual(reportedNodes.size, 2);
    assert(reportedNodes.has("cold-node"));
    assert(reportedNodes.has("warm-node"));

    const inspected = await send(addr, {
      type: "list_node_descriptors",
      tenant: "tenant",
      project: "project",
      actor_user: "operator"
    });
    assert.strictEqual(inspected.type, "node_descriptors");
    assert.strictEqual(inspected.actor, "operator");
    assert.strictEqual(inspected.descriptors.length, 2);
    const warmDescriptor = inspected.descriptors.find(
      (descriptor) => descriptor.id === "warm-node"
    );
    assert(warmDescriptor, "warm node descriptor must be visible to inspector state");
    assert(warmDescriptor.capabilities.capabilities.includes("Command"));
    assert(warmDescriptor.capabilities.capabilities.includes("RootlessPodman"));
    assert(warmDescriptor.cached_environments.includes("sha256:env-linux-container"));
    assert(warmDescriptor.dependency_caches.includes("sha256:deps-toolchain"));
    assert(warmDescriptor.source_snapshots.includes("sha256:source-tree"));
    assert(warmDescriptor.artifact_locations.includes("toolchain-cache"));

    const crossScopeInspection = await send(addr, {
      type: "list_node_descriptors",
      tenant: "other-tenant",
      project: "project",
      actor_user: "operator"
    });
    assert.strictEqual(crossScopeInspection.type, "node_descriptors");
    assert.strictEqual(crossScopeInspection.descriptors.length, 0);

    const crossTenantReport = await send(addr, {
      type: "report_node_capabilities",
      tenant: "other-tenant",
      project: "project",
      node: "warm-node",
      capabilities: linuxCapabilities(),
      cached_environment_digests: [],
      dependency_cache_digests: [],
      source_snapshots: [],
      artifact_locations: [],
      direct_connectivity: true,
      online: true
    });
    assert.strictEqual(crossTenantReport.type, "error");
    assert.match(crossTenantReport.message, /tenant\/project scope/);

    const placement = await send(addr, {
      type: "schedule_task",
      tenant: "tenant",
      project: "project",
      environment: {
        os: "Linux",
        arch: null,
        capabilities: ["Containers", "RootlessPodman"]
      },
      environment_digest: "sha256:env-linux-container",
      required_capabilities: ["Command"],
      dependency_cache: "sha256:deps-toolchain",
      source_snapshot: "sha256:source-tree",
      required_artifacts: ["toolchain-cache"],
      prefer_node: null
    });
    assert.strictEqual(placement.type, "task_placement");
    assert.strictEqual(placement.placement.node, "warm-node");
    assert.ok(placement.placement.score > 0);
    assert.ok(placement.placement.reasons.includes("warm environment cache"));
    assert.ok(placement.placement.reasons.includes("warm dependency cache"));
    assert.ok(placement.placement.reasons.includes("source snapshot already local"));
    assert.ok(
      placement.placement.reasons.includes("1 required artifact(s) already local")
    );

    const impossible = await send(addr, {
      type: "schedule_task",
      tenant: "tenant",
      project: "project",
      environment: null,
      environment_digest: null,
      required_capabilities: ["WindowsCommandDev"],
      dependency_cache: null,
      source_snapshot: null,
      required_artifacts: [],
      prefer_node: null
    });
    assert.strictEqual(impossible.type, "error");
    assert.match(impossible.message, /WindowsCommandDev/);

    const quotaDenied = await send(addr, {
      type: "schedule_task",
      tenant: "tenant",
      project: "project",
      environment: null,
      environment_digest: null,
      required_capabilities: ["Command"],
      dependency_cache: null,
      source_snapshot: null,
      required_artifacts: [],
      quota_available: false,
      policy_allowed: true,
      prefer_node: null
    });
    assert.strictEqual(quotaDenied.type, "error");
    assert.match(quotaDenied.message, /quota unavailable for placement/);

    const policyDenied = await send(addr, {
      type: "schedule_task",
      tenant: "tenant",
      project: "project",
      environment: null,
      environment_digest: null,
      required_capabilities: ["Command"],
      dependency_cache: null,
      source_snapshot: null,
      required_artifacts: [],
      quota_available: true,
      policy_allowed: false,
      prefer_node: null
    });
    assert.strictEqual(policyDenied.type, "error");
    assert.match(policyDenied.message, /policy denied placement/);

    await reportNode(addr, "warm-node", {
      cached_environment_digests: [],
      dependency_cache_digests: [],
      source_snapshots: [],
      artifact_locations: [],
      direct_connectivity: false
    });
    const disconnectedTransfer = await send(addr, {
      type: "schedule_task",
      tenant: "tenant",
      project: "project",
      environment: null,
      environment_digest: null,
      required_capabilities: ["Command"],
      dependency_cache: null,
      source_snapshot: "sha256:source-tree",
      required_artifacts: ["toolchain-cache"],
      prefer_node: null
    });
    assert.strictEqual(disconnectedTransfer.type, "error");
    assert.match(disconnectedTransfer.message, /source snapshot unavailable/);
    assert.match(disconnectedTransfer.message, /required artifact\(s\) unavailable/);
    assert.match(disconnectedTransfer.message, /direct connectivity unavailable/);
  } catch (error) {
    if (coordinatorStderr) {
      error.message = `${error.message}\ncoordinator stderr:\n${coordinatorStderr}`;
    }
    throw error;
  } finally {
    coordinator.kill("SIGTERM");
  }

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