Use live host usage and running estimates for collection admission
This commit is contained in:
+14
-12
@@ -180,19 +180,21 @@ heartbeat ingestion, at most once per minute per host/fleet phase; new peaks
|
|||||||
raise the cache immediately. Launch admission reads only the cached model.
|
raise the cache immediately. Launch admission reads only the cached model.
|
||||||
|
|
||||||
Scheduling still leaves **5% host CPU and 500 MB RAM (500,000,000 bytes)** for the OS and bursts.
|
Scheduling still leaves **5% host CPU and 500 MB RAM (500,000,000 bytes)** for the OS and bursts.
|
||||||
These allowances are scheduling reservations, not Docker limits. Queued and
|
These allowances are scheduling estimates, not Docker limits. Collection admission
|
||||||
unresolved creations consume the startup allowance; established instances keep
|
budgets new and never-submitted queued servers at their running allowance, rather
|
||||||
the running allowance even while idle or stopped. Current container measurements
|
than assuming the entire collection starts simultaneously. Current host CPU and
|
||||||
are credited against their reservations because host usage already includes them;
|
available memory account for existing containers and other processes. Observed
|
||||||
only the unused allowance is added. Missing or mismatched-generation measurements
|
containers are not topped up to historical startup or running peaks, even while
|
||||||
cannot release an unresolved launch's reservation. Observed host load also counts,
|
awaiting health verification. Established instances are covered by host usage
|
||||||
so other processes and previously unseen load increases reduce available capacity.
|
when individual container measurements are unavailable.
|
||||||
|
|
||||||
Admission counts all queued reservations. Dispatch evaluates one launch against
|
Dispatch checks the next launch's startup allowance against live host usage.
|
||||||
existing instances and submitted or uncertain work; other never-submitted queued
|
Other never-submitted queued launches wait their turn and retain their port
|
||||||
launches wait their turn. They retain their port reservations. This lets the queue
|
reservations. Submitted or uncertain launches missing a matching-generation
|
||||||
make progress even if a changed allowance means the whole backlog no longer fits
|
measurement still consume a full startup allowance until telemetry accounts for
|
||||||
at once. Capacity errors report projected usage, usable capacity and reservations.
|
them. This lets collections queue based on running demand and start as capacity
|
||||||
|
becomes available, without stacking every queued server's startup peak.
|
||||||
|
Capacity errors report projected usage, usable capacity and reservations.
|
||||||
|
|
||||||
When startup capacity is available, provisioning checks each eligible host’s
|
When startup capacity is available, provisioning checks each eligible host’s
|
||||||
current CPU and available memory against reservations and headroom.
|
current CPU and available memory against reservations and headroom.
|
||||||
|
|||||||
@@ -55,42 +55,42 @@ function projectedLoad(host: Host, options: PlacementOptions): number {
|
|||||||
{ id: string; generation: number; pending: number; unsubmitted: number }[];
|
{ id: string; generation: number; pending: number; unsubmitted: number }[];
|
||||||
const { running, startup } = capacityCalibration(host.id);
|
const { running, startup } = capacityCalibration(host.id);
|
||||||
const planned = options.planned?.filter(item => item.hostId === host.id).length ?? 0;
|
const planned = options.planned?.filter(item => item.hostId === host.id).length ?? 0;
|
||||||
let extraCpu = (planned + 1) * startup.cpuCores;
|
// Admission budgets the eventual running collection. Dispatch budgets only
|
||||||
let extraMemory = (planned + 1) * startup.memoryBytes;
|
// the next startup; the rest of the queue waits for live capacity to recover.
|
||||||
let reservedCpu = extraCpu;
|
const launch = options.dispatch ? startup : running;
|
||||||
let reservedMemory = extraMemory;
|
let extraCpu = planned * running.cpuCores + launch.cpuCores;
|
||||||
|
let extraMemory = planned * running.memoryBytes + launch.memoryBytes;
|
||||||
let reservations = planned + 1;
|
let reservations = planned + 1;
|
||||||
const measurements = new Map(host.inventory.instanceResources?.map(sample => [sample.instanceId, sample]));
|
const measurements = new Map(host.inventory.instanceResources?.map(sample => [sample.instanceId, sample]));
|
||||||
for (const instance of instances) {
|
for (const instance of instances) {
|
||||||
if (options.dispatch && options.excludeInstanceId && instance.unsubmitted) continue;
|
if (options.dispatch && options.excludeInstanceId && instance.unsubmitted) continue;
|
||||||
|
if (instance.unsubmitted) {
|
||||||
|
extraCpu += running.cpuCores;
|
||||||
|
extraMemory += running.memoryBytes;
|
||||||
reservations++;
|
reservations++;
|
||||||
const allowance = instance.pending ? startup : running;
|
continue;
|
||||||
reservedCpu += allowance.cpuCores;
|
}
|
||||||
reservedMemory += allowance.memoryBytes;
|
|
||||||
const sample = measurements.get(instance.id);
|
const sample = measurements.get(instance.id);
|
||||||
if (sample?.generation === instance.generation) {
|
// Current host usage already includes observed containers, including
|
||||||
// Host usage already includes this container. Reserve only the
|
// those still awaiting health verification. Do not top up to old peaks.
|
||||||
// unused part of its allowance, including while awaiting health.
|
if (sample?.generation === instance.generation || !instance.pending) continue;
|
||||||
extraCpu += Math.max(0, allowance.cpuCores - sample.cpuCores);
|
// A submitted launch absent from current telemetry may still start.
|
||||||
extraMemory += Math.max(0, allowance.memoryBytes - sample.memoryBytes);
|
extraCpu += startup.cpuCores;
|
||||||
} else if (instance.pending) {
|
extraMemory += startup.memoryBytes;
|
||||||
// Missing observations cannot release an uncertain remote launch.
|
reservations++;
|
||||||
extraCpu += allowance.cpuCores;
|
|
||||||
extraMemory += allowance.memoryBytes;
|
|
||||||
}
|
}
|
||||||
}
|
const cores = cpuCount * cpuUsagePercent / 100 + extraCpu;
|
||||||
const cores = Math.max(cpuCount * cpuUsagePercent / 100 + extraCpu, reservedCpu);
|
|
||||||
const usedMemory = memoryTotal - memoryAvailable;
|
const usedMemory = memoryTotal - memoryAvailable;
|
||||||
const memory = Math.max(usedMemory + extraMemory, reservedMemory);
|
const memory = usedMemory + extraMemory;
|
||||||
const usableMemory = memoryTotal - MEMORY_HEADROOM;
|
const usableMemory = memoryTotal - MEMORY_HEADROOM;
|
||||||
const gib = (bytes: number) => `${(bytes / 1024 ** 3).toFixed(2)} GiB`;
|
const gib = (bytes: number) => `${(bytes / 1024 ** 3).toFixed(2)} GiB`;
|
||||||
if (cores > cpuCount * CPU_USABLE_FRACTION) throw new Error(`Insufficient CPU capacity: ${cores.toFixed(2)} cores projected, ${(cpuCount * CPU_USABLE_FRACTION).toFixed(2)} usable; ${reservations} reservations including this launch.`);
|
if (cores > cpuCount * CPU_USABLE_FRACTION) throw new Error(`Insufficient CPU capacity: ${cores.toFixed(2)} cores projected, ${(cpuCount * CPU_USABLE_FRACTION).toFixed(2)} usable; ${reservations} reservations including this launch.`);
|
||||||
if (memory > usableMemory) throw new Error(`Insufficient memory capacity: ${gib(memory)} projected, ${gib(usableMemory)} usable after host headroom. `
|
if (memory > usableMemory) throw new Error(`Insufficient memory capacity: ${gib(memory)} projected, ${gib(usableMemory)} usable after host headroom. `
|
||||||
+ `${gib(usedMemory)} currently used + ${gib(extraMemory)} additional reservations; ${reservations} reservations including this launch (${gib(startup.memoryBytes)} startup allowance).`);
|
+ `${gib(usedMemory)} currently used + ${gib(extraMemory)} additional reservations; ${reservations} reservations including this launch (${gib(launch.memoryBytes)} ${options.dispatch ? 'startup' : 'running'} allowance).`);
|
||||||
// Recovered resources must not remain unavailable until the historical
|
// Recovered resources must not remain unavailable until the historical
|
||||||
// average catches up. History ranks hosts after current capacity passes.
|
// average catches up. History ranks hosts after current capacity passes.
|
||||||
const historicalMemory = Math.max(memoryTotal - historicalAvailable + extraMemory, reservedMemory);
|
const historicalMemory = memoryTotal - historicalAvailable + extraMemory;
|
||||||
const historicalCores = Math.max(cpuCount * historicalCpu / 100 + extraCpu, reservedCpu);
|
const historicalCores = cpuCount * historicalCpu / 100 + extraCpu;
|
||||||
return Math.max(historicalCores / cpuCount, historicalMemory / memoryTotal);
|
return Math.max(historicalCores / cpuCount, historicalMemory / memoryTotal);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -730,6 +730,37 @@ describe('automatic capacity admission and placement', () => {
|
|||||||
expect(capacityCalibration(hostId).startup.source).toBe('host');
|
expect(capacityCalibration(hostId).startup.source).toBe('host');
|
||||||
expect(queueNativeCreation(input)).toMatchObject({ ok: true });
|
expect(queueNativeCreation(input)).toMatchObject({ ok: true });
|
||||||
});
|
});
|
||||||
|
it('queues a collection at running demand and releases startup reservations as live usage arrives', async () => {
|
||||||
|
await prepareLibrary();
|
||||||
|
const { heartbeat } = await import('../../src/lib/server/hosts/index.js');
|
||||||
|
const { db } = await import('../../src/lib/server/db/index.js');
|
||||||
|
const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js');
|
||||||
|
const { saveBlueprint } = await import('../../src/lib/server/fleet/launch-recipes.js');
|
||||||
|
const { previewCollection, queueCollection } = await import('../../src/lib/server/fleet/collections.js');
|
||||||
|
const GiB = 1024 ** 3;
|
||||||
|
const measured = { ...inventory(), cpuCount: 4, cpuUsagePercent: 5, memoryTotal: 4 * GiB, memoryAvailable: 3.75 * GiB };
|
||||||
|
heartbeat(hostId, measured);
|
||||||
|
saveFleetSettings({ ...fleetSettings(), createConcurrency: 3 });
|
||||||
|
const insert = db().prepare(`INSERT INTO ec_resource_allowances
|
||||||
|
(scope,phase,cpu_cores,memory_bytes,samples,window_start,observed_at,computed_at) VALUES(?,?,?,?,30,?,?,?)`);
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
const start = new Date(Date.now() - 30 * 60_000).toISOString();
|
||||||
|
insert.run(hostId, 'startup', 2, 2 * GiB, start, now, now);
|
||||||
|
insert.run(hostId, 'running', 0.2, GiB / 4, start, now, now);
|
||||||
|
const blueprint = saveBlueprint({ name: 'Staggered lobbies', entries: [{ game: 't5', mapSlug: 'theater', count: 3 }] });
|
||||||
|
const preview = previewCollection(blueprint.id);
|
||||||
|
expect(preview.ready).toBe(true);
|
||||||
|
expect(queueCollection({ previewId: preview.id, keys: Object.fromEntries(preview.servers.map(server => [server.name!, 'key'])) }).ok).toBe(true);
|
||||||
|
await advance();
|
||||||
|
const creates = () => db().prepare("SELECT instance_id FROM ec_host_operations WHERE kind='instance.create'").all() as { instance_id: string }[];
|
||||||
|
expect(creates()).toHaveLength(1);
|
||||||
|
await advance();
|
||||||
|
expect(creates()).toHaveLength(1); // No heartbeat yet: the first launch still reserves startup resources.
|
||||||
|
heartbeat(hostId, { ...measured, cpuUsagePercent: 10, memoryAvailable: 3.5 * GiB,
|
||||||
|
instanceResources: [{ instanceId: creates()[0].instance_id, generation: 1, uptimeSeconds: 60, cpuCores: 0.2, memoryBytes: GiB / 4 }] });
|
||||||
|
await advance();
|
||||||
|
expect(creates()).toHaveLength(2); // Its actual usage replaces the peak even before health verification.
|
||||||
|
});
|
||||||
it('charges queued reservations and releases them on cancellation', async () => {
|
it('charges queued reservations and releases them on cancellation', async () => {
|
||||||
await prepareLibrary();
|
await prepareLibrary();
|
||||||
const { heartbeat } = await import('../../src/lib/server/hosts/index.js');
|
const { heartbeat } = await import('../../src/lib/server/hosts/index.js');
|
||||||
@@ -801,8 +832,8 @@ describe('automatic capacity admission and placement', () => {
|
|||||||
expect(first.ok).toBe(true); if (!first.ok) return;
|
expect(first.ok).toBe(true); if (!first.ok) return;
|
||||||
await advance();
|
await advance();
|
||||||
const instance = nativeInstance(first.name)!;
|
const instance = nativeInstance(first.name)!;
|
||||||
// 0.2 GiB OS + 0.8 GiB container. The remaining 0.2 GiB of its 1 GiB
|
// 0.2 GiB OS + 0.8 GiB container are already in host usage.
|
||||||
// reservation plus a new 1 GiB server fit below 2.53 GiB usable memory.
|
// Adding a new 1 GiB server fits below 2.53 GiB usable memory.
|
||||||
heartbeat(hostId, { ...host, memoryAvailable: 2 * GiB, instanceResources: [{ instanceId: instance.id,
|
heartbeat(hostId, { ...host, memoryAvailable: 2 * GiB, instanceResources: [{ instanceId: instance.id,
|
||||||
generation: instance.generation, uptimeSeconds: 60, cpuCores: 0.1, memoryBytes: Math.floor(0.8 * GiB) }] });
|
generation: instance.generation, uptimeSeconds: 60, cpuCores: 0.1, memoryBytes: Math.floor(0.8 * GiB) }] });
|
||||||
const second = queueNativeCreation(input);
|
const second = queueNativeCreation(input);
|
||||||
|
|||||||
Reference in New Issue
Block a user