Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,11 @@ HUB_DATA_DIR=./tmp/hub-data
# The origin sidecars reach the hub REST API at, for workflows that call it. Unset = the origin of BETTER_AUTH_BASE_URL
# HUB_API_ORIGIN=

# Milliseconds the local sidecar gets to shut down after SIGTERM before SIGKILL. Default 25000.
# Milliseconds a local sidecar gets to shut down after SIGTERM before SIGKILL. Default 25000.
# HUB_SIDECAR_STOP_TIMEOUT_MS=

# Deployments the hub's local sidecar runs at once: the listener, one manager per pull request and the judge. Default 16, at most 1792.
# TRIAGE_SIDECAR_MAX_DEPLOYMENTS=
# Sidecar processes the hub runs on its own machine at once, one per deployment or probe: the listener, each manager and the judge. Default 8.
# TRIAGE_MAX_LOCAL_SIDECARS=

# Docker daemon for workflows that require isolation:workload, registered when it answers
# and holds the image from: docker build --target sidecar -t corbits-triage-sidecar .
Expand Down
14 changes: 5 additions & 9 deletions apps/hub/src/env.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ import { homedir } from "node:os";
import { type, type ArkError } from "arktype";
import type { DockerTls, FindDockerHostOpts } from "@corbits/provisioner-docker";
import { LifecycleDuration, lifecycleDurationMs, type ResolvedWorkflowLifecyclePolicy } from "@intx/types";
import { DEFAULT_SIDECAR_RETAINED_DEPLOYMENTS, MAX_SIDECAR_INCARNATIONS } from "@intx/types/sidecar";
import { DEFAULT_GITHUB_API_ORIGIN } from "./github/github-app-credential-adapter.js";
import type { CoalesceWindow } from "./github/coalescer.js";
import { DEFAULT_RECONCILE_POLICY, type ReconcilePolicy } from "./github/reconcile-plan.js";
Expand All @@ -14,9 +13,7 @@ const secret = type("string").narrow((value, ctx) =>
const positiveInteger = type("string.integer").narrow((value, ctx) =>
Number(value) > 0 || ctx.mustBe("a positive integer"));

/** The sidecar holds its active and kept deployments within Interchange's combined limit; the hub leaves it the default kept ones. */
const MAX_LOCAL_SIDECAR_DEPLOYMENTS = MAX_SIDECAR_INCARNATIONS - DEFAULT_SIDECAR_RETAINED_DEPLOYMENTS;
const DEFAULT_LOCAL_SIDECAR_DEPLOYMENTS = 16;
const DEFAULT_MAX_LOCAL_SIDECARS = 8;

const lifecycleDuration = type("string.trim").pipe(LifecycleDuration);

Expand Down Expand Up @@ -80,8 +77,7 @@ const HubEnvSchema = type({
"TRIAGE_MAX_IN_FLIGHT?": positiveInteger,
"TRIAGE_CAPPED_RETRY_MS?": positiveInteger,
"TRIAGE_COALESCE_QUIET_MS?": positiveInteger,
"TRIAGE_SIDECAR_MAX_DEPLOYMENTS?": positiveInteger.narrow((value, ctx) =>
Number(value) <= MAX_LOCAL_SIDECAR_DEPLOYMENTS || ctx.mustBe(`at most ${String(MAX_LOCAL_SIDECAR_DEPLOYMENTS)}`)),
"TRIAGE_MAX_LOCAL_SIDECARS?": positiveInteger,
"HUB_AGENT_GC_PACK_THRESHOLD?": positiveInteger,
"HUB_AGENT_GC_LOOSE_THRESHOLD?": positiveInteger,
"HUB_AGENT_GC_WARN_BYTES?": positiveInteger,
Expand Down Expand Up @@ -283,9 +279,9 @@ export function triageReconcilePolicy(env: Pick<HubEnv, "TRIAGE_MAX_IN_FLIGHT" |
};
}

/** Deployments the hub's local sidecar runs at once. */
export function localSidecarMaxDeployments(env: Pick<HubEnv, "TRIAGE_SIDECAR_MAX_DEPLOYMENTS">): number {
return numberOr(env.TRIAGE_SIDECAR_MAX_DEPLOYMENTS, DEFAULT_LOCAL_SIDECAR_DEPLOYMENTS);
/** Sidecar processes the hub runs on its own machine at once, one per deployment or probe. */
export function maxLocalSidecars(env: Pick<HubEnv, "TRIAGE_MAX_LOCAL_SIDECARS">): number {
return numberOr(env.TRIAGE_MAX_LOCAL_SIDECARS, DEFAULT_MAX_LOCAL_SIDECARS);
}

const COALESCE_BURST_MS = 3_000;
Expand Down
20 changes: 8 additions & 12 deletions apps/hub/src/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ import { connectRailwaySidecars, sweepRailwaySidecarsOf, type RailwaySidecarProv
import { createHostCapacity } from "./sidecar/host-capacity.js";
import { createLocalSidecarManifestStore } from "./sidecar/manifest-store.js";
import { createPlacementChooser, localSidecarUsage } from "./sidecar/placement.js";
import { restoreLocalSidecars } from "./sidecar/restore.js";
import { bootLocalSidecars } from "./sidecar/boot.js";
import { createSpawnLocalSidecar } from "./sidecar/spawn.js";
import { SIDECAR_ADAPTER_MANIFEST } from "./sidecar-config.js";
import { createPortalHandler, isPortalRequest, withPortalCors } from "./portal.js";
Expand All @@ -71,7 +71,7 @@ import { createListenerWatch, releaseWriteOf } from "./workflow-release/triggers
import { setUpPublisher, TRIAGE_PUBLISHER } from "./workflow-release/publisher.js";
import { catalogModels, createWorkspaceLock, deployedAllocation, deploymentState, liveDeployments } from "./workflow-release/reads.js";
import { createRolloutQueue, createWorkflowRollout, stopGivenUp, type DeployRequest } from "./workflow-release/rollout.js";
import { bitbucketEgressAllowlist, childProcessEnv, clientIpHeaders, databaseConfig, dockerSettings, e2bSettings, emailSettings, interchangeSettings, githubApiOrigin, loadHubEnv, inviteLinkOrigin, localSidecarMaxDeployments, railwaySandboxSettings, signInSettings, triageCoalesceWindow, triageReconcileIntervalMs, triageReconcilePolicy } from "./env.js";
import { bitbucketEgressAllowlist, childProcessEnv, clientIpHeaders, databaseConfig, dockerSettings, e2bSettings, emailSettings, interchangeSettings, githubApiOrigin, loadHubEnv, inviteLinkOrigin, maxLocalSidecars, railwaySandboxSettings, signInSettings, triageCoalesceWindow, triageReconcileIntervalMs, triageReconcilePolicy } from "./env.js";
import { HOOK_MOUNT_PATH } from "./hooks.js";
import { assertMigrated } from "./migrations.js";
import { createBridgeHandler, logJson, withAppLogin } from "./github/bridge.js";
Expand Down Expand Up @@ -159,24 +159,19 @@ await assertMigrated(database);

const credentialCipher = createEnvKeyCredentialCipher(hexDecode(env.CREDENTIAL_ENCRYPTION_KEY));
const localSidecarDataRoot = `${env.HUB_DATA_DIR}/local-sidecars`;
const localSidecarCap = localSidecarMaxDeployments(env);
const localSidecarCap = maxLocalSidecars(env);
const local = createLocalProcessSidecarProvisioner({
dataRoot: localSidecarDataRoot,
...(env.HUB_SIDECAR_STOP_TIMEOUT_MS !== undefined && { stopTimeoutMs: Number(env.HUB_SIDECAR_STOP_TIMEOUT_MS) }),
spawnSidecar: createSpawnLocalSidecar({
interchangeDir: V,
credentialEncryptionKey: env.SIDECAR_CREDENTIAL_ENCRYPTION_KEY,
maxActiveDeployments: localSidecarCap,
processEnv: childProcessEnv(env),
}),
manifests: createLocalSidecarManifestStore({
dataRoot: localSidecarDataRoot,
cipher: credentialCipher,
log: logJson,
}),
manifests: createLocalSidecarManifestStore({ dataRoot: localSidecarDataRoot, log: logJson }),
capacity: createHostCapacity({ cgroupRoot: "/sys/fs/cgroup", meminfo: "/proc/meminfo" }),
adapterManifest: SIDECAR_ADAPTER_MANIFEST,
onAccepted: notePlaced,
maxSidecars: localSidecarCap,
log: logJson,
});
const docker = await connectDockerSidecars({ docker: dockerSettings(env), database, adapterManifest: SIDECAR_ADAPTER_MANIFEST });
Expand All @@ -195,7 +190,7 @@ const codeReviewProbes = [...dockerPlaces(docker), { guide: E2B_PLACE, found: e2
const noIsolatedPlace = isolatedPlaceBootWarning(workloadProvisioners.length > 0, codeReviewProbes.map((probe) => probe.guide));
if (noIsolatedPlace) logJson(noIsolatedPlace);
function placeDeployments(db: DB["db"]) {
return createPlacementChooser({ db, local: local.provisioner, maxDeployments: localSidecarCap });
return createPlacementChooser({ db, local: local.provisioner, maxSidecars: localSidecarCap });
}

let shuttingDown = false;
Expand Down Expand Up @@ -250,6 +245,7 @@ const composition = await createInterchangeHub({
});
const stock = composition.server;
composition.sidecarRouter.events.on("sidecar.allocated.connected", notePlaced);
composition.sidecarRouter.events.on("sidecar.allocated.connected", local.connected);

/** The hub's credentials, and the push that takes back from sidecars an App key a read moved to the hub. */
const credentialDeps = {
Expand All @@ -264,7 +260,7 @@ const movedAppKeys: string[] = [];
for (const tenantId of await githubTenants(composition.db)) {
if (await moveAppKey(credentialDeps, tenantId)) movedAppKeys.push(tenantId);
}
await restoreLocalSidecars(composition.db, local);
await bootLocalSidecars(composition.db, local);
if (dockerUp) logJson({ level: "info", msg: "docker_sidecars_swept", removed: await sweepDockerSidecarsOf(composition.db, dockerUp) });
async function tendE2b(sidecars: E2bSidecarProvisioner): Promise<void> {
try {
Expand Down
20 changes: 20 additions & 0 deletions apps/hub/src/sidecar/boot.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
import { schema, type DB } from "@intx/db";
import { and, inArray, notInArray } from "drizzle-orm";
import type { LocalProcessSidecarProvisioner } from "./local-provisioner.js";

export async function bootLocalSidecars(db: DB["db"], local: LocalProcessSidecarProvisioner): Promise<void> {
async function stillOpen(allocationIds: readonly string[]): Promise<ReadonlySet<string>> {
if (allocationIds.length === 0) return new Set();
const ids = [...allocationIds];
const allocations = await db
.select({ id: schema.sidecarAllocation.id })
.from(schema.sidecarAllocation)
.where(and(inArray(schema.sidecarAllocation.id, ids), notInArray(schema.sidecarAllocation.status, ["released", "failed"])));
const probes = await db
.select({ id: schema.workflowProbe.id })
.from(schema.workflowProbe)
.where(and(inArray(schema.workflowProbe.id, ids), notInArray(schema.workflowProbe.status, ["succeeded", "failed"])));
return new Set([...allocations, ...probes].map((row) => row.id));
}
await local.boot({ stillOpen });
}
8 changes: 4 additions & 4 deletions apps/hub/src/sidecar/host-capacity.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -52,8 +52,8 @@ test("without a cgroup limit MemAvailable decides, and a box reporting neither h
expect(await createHostCapacity(await box({})).freeBytes()).toBeUndefined();
});

test("the sidecar needs one footprint free, and no figure never declines", () => {
expect(shortfall(SIDECAR_BYTES)).toBeUndefined();
expect(shortfall(SIDECAR_BYTES / 2)).toBe(`${SIDECAR_BYTES / 2 / MiB} MiB free, ${SIDECAR_BYTES / MiB} MiB needed`);
expect(shortfall(undefined)).toBeUndefined();
test("each sidecar still starting holds one footprint, and no figure never declines", () => {
expect(shortfall(2 * SIDECAR_BYTES, 1)).toBeUndefined();
expect(shortfall(2 * SIDECAR_BYTES, 2)).toBe(`${(2 * SIDECAR_BYTES) / MiB} MiB free, ${(3 * SIDECAR_BYTES) / MiB} MiB needed`);
expect(shortfall(undefined, 100)).toBeUndefined();
});
17 changes: 10 additions & 7 deletions apps/hub/src/sidecar/host-capacity.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
// Whether this box has the memory to start the local sidecar: free memory in
// Whether this box has the memory for one more local sidecar: free memory in
// the container's cgroup (v2, else v1) where a limit is set, else Linux's
// MemAvailable. A box with neither, such as a macOS dev machine, never
// declines: os.freemem there leaves out memory the system would reclaim.
// MemAvailable, less one footprint for each sidecar started that has not
// connected yet, as its memory is not in use yet. A box with neither, such as
// a macOS dev machine, never declines: os.freemem there leaves out memory the
// system would reclaim.
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
Expand Down Expand Up @@ -67,8 +69,9 @@ export function createHostCapacity({ cgroupRoot, meminfo }: HostCapacityOpts): H
};
}

/** Why there is no room to start the sidecar, or undefined when there is. */
export function shortfall(free: number | undefined): string | undefined {
if (free === undefined || free >= SIDECAR_BYTES) return undefined;
return `${Math.floor(free / MiB)} MiB free, ${SIDECAR_BYTES / MiB} MiB needed`;
/** Why there is no room for one more sidecar beside `starting` ones not connected yet, or undefined when there is. */
export function shortfall(free: number | undefined, starting: number): string | undefined {
const needed = SIDECAR_BYTES * (starting + 1);
if (free === undefined || free >= needed) return undefined;
return `${Math.floor(free / MiB)} MiB free, ${needed / MiB} MiB needed`;
}
61 changes: 61 additions & 0 deletions apps/hub/src/sidecar/local-provisioner.conformance.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { describeProvisionerConformance } from "@corbits/provisioner-test";
import { createLocalProcessSidecarProvisioner } from "./local-provisioner.js";
import { createLocalSidecarManifestStore } from "./manifest-store.js";

type FakeProcess = {
readonly sidecarId: string;
running: boolean;
};

describeProvisionerConformance("local-process", async () => {
const dataRoot = await fs.mkdtemp(path.join(os.tmpdir(), "local-provisioner-conformance-"));
const processes: FakeProcess[] = [];
const manifests = createLocalSidecarManifestStore({ dataRoot, log: () => {} });

const local = createLocalProcessSidecarProvisioner({
dataRoot,
capacity: { freeBytes: async () => undefined },
adapterManifest: [],
maxSidecars: 4,
spawnSidecar: function spawnSidecar({ sidecarId }) {
const process: FakeProcess = { sidecarId, running: true };
processes.push(process);
const exit = Promise.withResolvers<{ code: number | null; signal: string | null }>();
return {
pid: processes.length,
exited: exit.promise,
kill(signal) {
process.running = false;
exit.resolve({ code: null, signal });
},
};
},
manifests,
log: () => {},
});
local.provisioner.attach?.({ sidecarStopped: async () => ({ kind: "retired" }) });
await local.boot({ stillOpen: async (ids) => new Set(ids) });

return {
provisioner: local.provisioner,
async live(allocationId) {
let manifest: { generation: number; sidecarId: string };
try {
manifest = JSON.parse(await fs.readFile(path.join(dataRoot, "manifests", `${allocationId}.json`), "utf8"));
} catch {
return [];
}
return processes
.filter((process) => process.running && process.sidecarId === manifest.sidecarId)
.map(({ sidecarId }) => ({ generation: manifest.generation, sidecarId }));
},
capabilityChecks: {},
async dispose() {
await local.shutdown();
await fs.rm(dataRoot, { recursive: true, force: true });
},
};
});
Loading
Loading