Throttle agent usage writes and cache inventory per request
This commit is contained in:
+3
-2
@@ -6,6 +6,7 @@ import { authenticationHandle } from '$lib/server/auth/guard.js';
|
||||
import { invalidateFleet } from '$lib/server/pipeline/bus.js';
|
||||
import { startPerformanceMonitor } from '$lib/server/performance.js';
|
||||
import { withOperationSource } from '$lib/server/hosts/operation-context.js';
|
||||
import { withHostRequestContext } from '$lib/server/hosts/request-context.js';
|
||||
|
||||
let booted = false;
|
||||
function boot(): void {
|
||||
@@ -23,8 +24,8 @@ function boot(): void {
|
||||
export const handle: Handle = async (input) => {
|
||||
boot();
|
||||
const started = performance.now();
|
||||
const response = await withOperationSource(`${input.event.request.method} ${input.event.route.id ?? 'Web request'}`,
|
||||
() => authenticationHandle(input));
|
||||
const response = await withHostRequestContext(() => withOperationSource(`${input.event.request.method} ${input.event.route.id ?? 'Web request'}`,
|
||||
() => authenticationHandle(input)));
|
||||
const elapsed = performance.now() - started;
|
||||
response.headers.append('Server-Timing', `app;dur=${elapsed.toFixed(1)}`);
|
||||
if (process.env.EC_PERF_LOG === 'true' && elapsed > 250 && !input.event.route.id?.startsWith('/api/agent/')) {
|
||||
|
||||
@@ -6,6 +6,7 @@ import { db } from '../db/index.js';
|
||||
import { ensureHostSchema } from './schema.js';
|
||||
import { successfulObservation, routineObservationSql, importantOperationSql, operationTargetSql, safeOperationTarget } from './observations.js';
|
||||
import { currentOperationSource } from './operation-context.js';
|
||||
import { parseHostInventory } from './request-context.js';
|
||||
import { fleetSettings } from '../settings.js';
|
||||
import { recordInstanceResources, validateInstanceResources } from './calibration.js';
|
||||
import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js';
|
||||
@@ -38,7 +39,7 @@ function presentDecommission(row: DecommissionRow | undefined): HostDecommission
|
||||
}
|
||||
function hostFromRow(row: HostRow): Host {
|
||||
const enabling = database().prepare('SELECT requested_at FROM ec_host_enable_requests WHERE host_id=?').get(row.id) as { requested_at: string } | undefined;
|
||||
const inventory: HostInventory = JSON.parse(row.inventory);
|
||||
const inventory = parseHostInventory(row.id, row.inventory);
|
||||
const connected = !!row.credential_active && !!row.last_heartbeat && Date.now() - Date.parse(row.last_heartbeat) <= HEARTBEAT_MAX_AGE_MS;
|
||||
const reasons: string[] = [];
|
||||
if (!row.enrolled_at) reasons.push('Agent has not enrolled.');
|
||||
@@ -304,9 +305,15 @@ export function enrollHost(token: unknown, rawInventory: unknown): { hostId: str
|
||||
}
|
||||
export function authenticateCredential(token: string): Host | null {
|
||||
if (!token || token.length > 200) return null;
|
||||
const row = database().prepare('SELECT id,host_id FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL').get(digest(token)) as { id: string; host_id: string } | undefined;
|
||||
const row = database().prepare('SELECT id,host_id,last_used_at FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL')
|
||||
.get(digest(token)) as { id: string; host_id: string; last_used_at: string | null } | undefined;
|
||||
if (!row) return null;
|
||||
database().prepare('UPDATE ec_host_credentials SET last_used_at=? WHERE id=?').run(now(), row.id);
|
||||
const timestamp = Date.now();
|
||||
if (row.last_used_at === null || timestamp - Date.parse(row.last_used_at) >= 60_000) {
|
||||
database().prepare(`UPDATE ec_host_credentials SET last_used_at=? WHERE id=? AND revoked_at IS NULL
|
||||
AND (last_used_at IS NULL OR last_used_at<=?)`)
|
||||
.run(new Date(timestamp).toISOString(), row.id, new Date(timestamp - 60_000).toISOString());
|
||||
}
|
||||
return getHost(row.host_id);
|
||||
}
|
||||
export function revokeCredentials(hostId: string): void {
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
import { AsyncLocalStorage } from 'node:async_hooks';
|
||||
import type { HostInventory } from '../../hosts.js';
|
||||
|
||||
type InventoryCache = Map<string, { json: string; inventory: HostInventory }>;
|
||||
const requests = new AsyncLocalStorage<{ inventories: InventoryCache | null }>();
|
||||
|
||||
export function withHostRequestContext<T>(work: () => T | Promise<T>): Promise<T> {
|
||||
const context = { inventories: new Map() as InventoryCache | null };
|
||||
return requests.run(context, async () => {
|
||||
try { return await work(); }
|
||||
finally {
|
||||
// Detached work can inherit this context. Release request data and
|
||||
// stop caching when the response is ready, even if that work continues.
|
||||
context.inventories = null;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export function parseHostInventory(hostId: string, json: string): HostInventory {
|
||||
const cache = requests.getStore()?.inventories;
|
||||
const saved = cache?.get(hostId);
|
||||
if (saved?.json === json) return saved.inventory;
|
||||
const inventory: HostInventory = JSON.parse(json);
|
||||
// Compare the stored JSON so heartbeats and transaction rollbacks within
|
||||
// a request remain visible. Other host state is always read afresh.
|
||||
cache?.set(hostId, { json, inventory });
|
||||
return inventory;
|
||||
}
|
||||
Reference in New Issue
Block a user