TRX/Checkmate
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3978 from bluewave-labs/chore/readability-refactor

Browse Source 
chore: refactor worker services into a pipeline for readability
This commit is contained in:
Alexander Holliday authored and GitHub committed 2026-09-21 09:38:56 -07:00
2 parents 199d68e8a2 461a5e35dd
commit d51fdbd772
26 files changed
+1525 -1505
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/config/services.worker.ts
+25 -28
View File
@@ -17,12 +17,9 @@ import { INetworkService, NetworkService } from "@/service/networkService.js";
import { IBufferService, BufferService } from "@/service/bufferService.js";
import { IStatusService, StatusService } from "@/service/statusService.js";
import { SecretsRotationService } from "@/service/encryption/secretsRotationService.js";
import { IQueueWorker } from "@/worker/worker.interface.js";
import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js";
import { IQueueWorker, JobHandlers } from "@/worker/worker.interface.js";
import { WorkerHelper } from "@/worker/worker.helper.js";
import { CheckProducer } from "@/worker/worker.check-producer.js";
import { CheckEvaluator } from "@/worker/worker.check-evaluator.js";
import { GeoChecksPipeline } from "@/worker/worker.check-pipeline.js";
import { GeoChecksPipeline } from "@/worker/worker.geo-pipeline.js";
import { NotificationReactor } from "@/worker/reactors/reactor.notification.js";
import { IncidentReactor } from "@/worker/reactors/reactor.incident.js";
import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
@@ -42,6 +39,7 @@ import { GrpcProvider } from "@/service/network/GrpcProvider.js";
import { WebSocketProvider } from "@/service/network/WebSocketProvider.js";
import { DNSProvider } from "@/service/network/DNSProvider.js";
import { AppError } from "@/utils/AppError.js";
import { WorkerPipeline } from "@/worker/worker.pipeline.js";
export interface WorkerServices {
worker: IQueueWorker;
networkService: INetworkService;
@@ -103,37 +101,33 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig
]);
const proxyResolver = new ProxyResolver(proxiesRepository, settingsService, logger);
const bufferService = new BufferService(logger, checkService, geoChecksService, dockerLogsService, settingsService, jobsRepository);
const bufferService = new BufferService(logger, geoChecksService, dockerLogsService, settingsService, (checks): Promise<void> =>
pipeline.ingestChecks(checks)
);
const statusService = new StatusService(logger, monitorsRepository, monitorStatsRepository);
const monitorStatusPolicy = new MonitorStatusPolicy();
const egressService = new EgressService(settingsService, egressStateRepository, jobsRepository, networkService, proxyResolver, logger);
// ***********************
// Reactors and dispatcher
// Handles notifications and incidents
// ***********************
const notificationReactor = new NotificationReactor(notificationsService);
const incidentReactor = new IncidentReactor(incidentService);
const reactorDispatcher = new ReactorDispatcher(logger, [notificationReactor, incidentReactor]);
// ***********************
// Check producer/evaluator
// Handles creating and evaluatiog checks
// ***********************
const checkProducer = new CheckProducer(
const pipeline = new WorkerPipeline({
logger,
monitorsRepository,
maintenanceWindowsRepository,
checksRepository,
jobsRepository,
checkService,
networkService,
proxyResolver,
bufferService,
dockerLogsService,
egressService,
logger
);
const checkEvaluator = new CheckEvaluator(statusService, monitorStatusPolicy, logger);
const geoCheckPipeline = new GeoChecksPipeline(maintenanceWindowsRepository, geoChecksService, bufferService, logger);
statusService,
dispatcher: reactorDispatcher,
});
const geoCheckPipeline = new GeoChecksPipeline(monitorsRepository, maintenanceWindowsRepository, geoChecksService, bufferService, logger);
// ***********************
// Worker
@@ -154,19 +148,22 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig
egressService
);
const handlers: JobHandlers = {
check: pipeline.handleCheck,
evaluate: pipeline.handleEvaluate,
"geo-check": geoCheckPipeline.handle,
"cleanup-orphaned": workerHelper.getCleanupOrphanedJob(),
"cleanup-retention": workerHelper.getCleanupRetentionJob(),
egress: workerHelper.getEgressRecoveryJob(),
};
const worker = await DBQueueWorker.create({
logger,
isDbConnected: () => mongoose.connection.readyState === 1,
jobsRepository,
monitorsRepository,
checksRepository,
checkService,
bufferService,
checkProducer,
checkEvaluator,
geoCheckPipeline,
dispatcher: reactorDispatcher,
helper: workerHelper,
handlers,
queueWorkersRepository,
queueMode: envSettings.queueMode,
queuePrimaryProcesses: envSettings.queuePrimaryProcesses,
server/src/domain/incidents/incident.service.ts
+1 -1
View File
@@ -8,7 +8,7 @@ import type { IMonitorsRepository } from "@/domain/monitors/monitor.repository.i
import type { IUsersRepository } from "@/domain/users/user.repository.interface.js";
import type { Incident, IncidentSummary } from "@/domain/incidents/incident.type.js";
import type { User } from "@/domain/users/user.type.js";
import type { MonitorActionDecision } from "@/worker/worker.helper.js";
import type { MonitorActionDecision } from "@/worker/worker.interface.js";
import type { INotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js";
import type { ILogger } from "@/utils/logger.js";
import { DateRange } from "@/types/query.js";
server/src/domain/notifications/notification.message-builder.ts
+1 -1
View File
@@ -1,6 +1,6 @@
import type { Monitor } from "@/domain/monitors/monitor.type.js";
import type { HardwareStatusPayload, MonitorStatusResponse } from "@/types/network.js";
import type { MonitorActionDecision } from "@/worker/worker.helper.js";
import type { MonitorActionDecision } from "@/worker/worker.interface.js";
import type {
NotificationMessage,
NotificationType,
server/src/domain/notifications/notification.service.ts
+1 -1
View File
@@ -5,7 +5,7 @@ import type { NotificationMessage } from "@/domain/notifications/notification.ty
import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
import { INotificationsRepository } from "@/domain/notifications/notification.repository.interface.js";
import { INotificationProvider } from "@/domain/notifications/providers/INotificationProvider.js";
import type { MonitorActionDecision } from "@/worker/worker.helper.js";
import type { MonitorActionDecision } from "@/worker/worker.interface.js";
import type { ISettingsService } from "@/domain/app-settings/app-settings.service.js";
import { ILogger } from "@/utils/logger.js";
import type { INotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js";
server/src/service/bufferService.ts
+6 -54
View File
@@ -3,18 +3,15 @@ import type { GeoCheck } from "@/domain/geo-checks/geo-check.type.js";
import type { IGeoChecksService } from "../domain/geo-checks/geo-check.service.js";
import type { ILogger } from "@/utils/logger.js";
import type { ISettingsService } from "@/domain/app-settings/app-settings.service.js";
import type { IJobsRepository } from "@/domain/jobs/job.repository.interface.js";
import { ICheckService } from "../domain/checks/check.service.js";
import { DockerLog } from "@/domain/docker/docker-log.type.js";
import { IDockerLogsService } from "@/domain/docker/docker-log.service.js";
import { PendingCheck } from "@/domain/jobs/job.type.js";
const SERVICE_NAME = "BufferService";
export type CheckIngest = (checks: Check[]) => Promise<void>;
export interface IBufferService {
addToBuffer(check: Check): void;
addGeoCheckToBuffer(geoCheck: GeoCheck): void;
addDockerLogToBuffer(dockerLog: DockerLog): void;
ingestChecks(checks: Check[]): Promise<void>;
scheduleNextFlush(): void;
flushBuffer(): Promise<void>;
flushGeoBuffer(): Promise<void>;
@@ -31,28 +28,20 @@ export class BufferService implements IBufferService {
private geoBuffer: GeoCheck[];
private dockerLogBuffer: DockerLog[];
private bufferTimer: NodeJS.Timeout | null = null;
private checksService: ICheckService;
private geoChecksService: IGeoChecksService;
private dockerLogsService: IDockerLogsService;
private jobsRepository: IJobsRepository;
// Checks that are stored but whose evaluate row could not be armed. Retried on the next flush so they are never unevaluated.
private unarmed: Map<string, PendingCheck[]> = new Map();
constructor(
logger: ILogger,
checkService: ICheckService,
geoChecksService: IGeoChecksService,
dockerLogsService: IDockerLogsService,
settingsService: ISettingsService,
jobsRepository: IJobsRepository
private ingest: CheckIngest
) {
this.BUFFER_TIMEOUT = settingsService.getSettings().nodeEnv === "development" ? 1000 : 1000 * 60 * 1; // 1 minute
this.logger = logger;
this.checksService = checkService;
this.geoChecksService = geoChecksService;
this.dockerLogsService = dockerLogsService;
this.jobsRepository = jobsRepository;
this.SERVICE_NAME = SERVICE_NAME;
this.buffer = [];
this.geoBuffer = [];
@@ -104,38 +93,6 @@ export class BufferService implements IBufferService {
}
}
ingestChecks = async (checks: Check[]) => {
if (checks.length > 0) await this.checksService.createChecks(checks);
// Group by monitor, starting from any entries a previous flush failed to arm
const byMonitor = this.unarmed;
this.unarmed = new Map();
for (const check of checks) {
const pending = byMonitor.get(check.metadata.monitorId) ?? [];
pending.push({ checkId: check.id, createdAt: new Date(check.createdAt).getTime() });
byMonitor.set(check.metadata.monitorId, pending);
}
if (byMonitor.size === 0) return;
const now = Date.now();
const ops: Promise<boolean>[] = [];
for (const [monitorId, pendingChecks] of byMonitor) {
ops.push(
this.jobsRepository.upsertEvaluate(monitorId, pendingChecks, now).catch((error: unknown) => {
// The checks are already stored, so a failed arm is retried on the next flush rather than dropped
this.unarmed.set(monitorId, pendingChecks);
this.logger.error({
message: `Could not arm evaluation for monitor ${monitorId}, retrying on next flush: ${error instanceof Error ? error.message : String(error)}`,
service: this.SERVICE_NAME,
method: "ingestChecks",
});
return false;
})
);
}
await Promise.all(ops);
};
scheduleNextFlush() {
if (this.bufferTimer) {
clearTimeout(this.bufferTimer);
@@ -160,19 +117,14 @@ export class BufferService implements IBufferService {
}
async flushBuffer() {
if (this.buffer.length === 0 && this.unarmed.size === 0) {
return;
}
// Take the batch first so a write that drains it can't be appended to mid-flush
const batch = this.buffer;
this.buffer = [];
try {
this.logger.debug({
message: `Flushing ${batch.length} checks to database`,
service: this.SERVICE_NAME,
method: "flushBuffer",
});
await this.ingestChecks(batch);
if (batch.length > 0) {
this.logger.debug({ message: `Flushing ${batch.length} checks to database`, service: this.SERVICE_NAME, method: "flushBuffer" });
}
await this.ingest(batch);
} catch (error: unknown) {
this.logger.error({
message: error instanceof Error ? error.message : "Unknown error",
server/src/worker/worker.check-evaluator.ts
-69
View File
@@ -1,69 +0,0 @@
import { MonitorStatusResponse } from "@/types/network.js";
import { MonitorEvaluation } from "@/worker/worker.interface.js";
import { Check } from "@/domain/checks/check.type.js";
import { Monitor } from "@/domain/monitors/monitor.type.js";
import { IMonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js";
import { IStatusService } from "@/service/statusService.js";
import { MonitorActionDecision } from "@/worker/worker.helper.js";
import { ILogger } from "@/utils/logger.js";
export interface ICheckEvaluator {
evaluate(status: MonitorStatusResponse, check: Check, monitor: Monitor): Promise<MonitorEvaluation>;
}
const SERVICE_NAME = "CheckEvaluator";
export class CheckEvaluator implements ICheckEvaluator {
constructor(
private statusService: IStatusService,
private monitorStatusPolicy: IMonitorStatusPolicy,
private logger: ILogger
) {}
// A check taken while the instance had no outbound connectivity says nothing about the target.
// It is left out of the status window, running stats and monitor status entirely, so that no incident
// opens for it and no spurious "resolved"/"up" fires once egress returns.
private skipDegradedEgressCheck = (status: MonitorStatusResponse, check: Check, monitor: Monitor): MonitorEvaluation => {
this.logger.debug({
message: `Skipping evaluation of check ${check.id} for monitor ${monitor.id}: instance egress was degraded`,
service: SERVICE_NAME,
method: "evaluate",
});
const decision: MonitorActionDecision = {
shouldCreateIncident: false,
shouldResolveIncident: false,
shouldSendNotification: false,
incidentReason: null,
notificationReason: null,
};
return {
monitor,
status,
check,
statusChange: { monitor, statusChanged: false, prevStatus: monitor.status, code: status.code, timestamp: Date.now() },
decision,
};
};
evaluate = async (status: MonitorStatusResponse, check: Check, monitor: Monitor) => {
if (check.egressStatus === "degraded") {
return this.skipDegradedEgressCheck(status, check, monitor);
}
// ****************************
// Step 3: Evaluate and return result to reactors
// ****************************
const statusChangeResult = await this.statusService.updateMonitorStatus(status, check, monitor);
// Step 5. Get decisions and create an evaluation obj
const decision = this.monitorStatusPolicy.evaluate(statusChangeResult);
const evaluation: MonitorEvaluation = {
monitor: statusChangeResult.monitor,
status,
check,
statusChange: statusChangeResult,
decision,
};
return evaluation;
};
}
server/src/worker/worker.check-producer.ts
-157
View File
@@ -1,157 +0,0 @@
import { Monitor } from "@/domain/monitors/monitor.type.js";
import { Check } from "@/domain/checks/check.type.js";
import { DockerStatusPayload, MonitorStatusResponse } from "@/types/network.js";
import { AppError } from "@/utils/AppError.js";
import { ILogger } from "@/utils/logger.js";
import { IBufferService } from "@/service/bufferService.js";
import { INetworkService } from "@/service/networkService.js";
import { ICheckService } from "@/domain/checks/check.service.js";
import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js";
import { isWindowActive } from "@/utils/maintenanceWindow.js";
import { IProxyResolver } from "@/service/network/ProxyResolver.js";
import { IDockerLogsService } from "@/domain/docker/docker-log.service.js";
import { IEgressService, isHttpStatusCode } from "@/domain/egress/egress.service.js";
import type { EgressStatus } from "@/domain/egress/egress.type.js";
import { isEgressAttributable } from "@/domain/monitors/monitor.type.js";
export interface ICheckProducer {
produce(monitor: Monitor): Promise<{ status: MonitorStatusResponse; check: Check } | null>;
}
const SERVICE_NAME = "CheckProducer";
export class CheckProducer implements ICheckProducer {
constructor(
private monitorsRepository: IMonitorsRepository,
private maintenanceWindowsRepository: IMaintenanceWindowsRepository,
private checkService: ICheckService,
private networkService: INetworkService,
private proxyResolver: IProxyResolver,
private bufferService: IBufferService,
private dockerLogsService: IDockerLogsService,
private egressService: IEgressService,
private logger: ILogger
) {}
private async isInMaintenanceWindow(monitorId: string, teamId: string) {
const windows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId);
const now = new Date();
return windows.some((w) => isWindowActive(w, now));
}
private async resolveDockerTlsKey(monitor: Monitor): Promise<string | undefined> {
if (monitor.type !== "docker" || !monitor.dockerTlsKeySet || !monitor.id) return undefined;
const dockerTlsKey = await this.monitorsRepository.findDockerTlsKeyById(monitor.id);
return dockerTlsKey ?? undefined;
}
// Only a failure to reach the target at all can be the instance's own fault. A provider that can tell the
// two apart says so outright; for the rest, any HTTP status code proves the target answered (a 4xx, a 5xx,
// or a 200 with a content mismatch).
private peerAnswered(status: MonitorStatusResponse): boolean {
return status.peerResponded ?? isHttpStatusCode(status.code);
}
private isTransportFailure(monitor: Monitor, status: MonitorStatusResponse): boolean {
return status.status === false && isEgressAttributable(monitor) && !this.peerAnswered(status);
}
// The egress check must never stop a check being recorded, so a rejection is logged and treated as unknown.
private async assessEgress(monitorId: string): Promise<EgressStatus | null> {
try {
return await this.egressService.assessAfterFailure();
} catch (error: unknown) {
this.logger.warn({
message: `Egress assessment failed for monitor ${monitorId}: ${error instanceof Error ? error.message : String(error)}`,
service: SERVICE_NAME,
method: "assessEgress",
});
return null;
}
}
produce = async (monitor: Monitor) => {
if (!monitor.id) {
throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "produce" });
}
// ****************************
// Step 1: Acquire
// ****************************
// Step 1a: Maintenance window gate - skip if in maintenance
const maintenanceWindowActive = await this.isInMaintenanceWindow(monitor.id, monitor.teamId);
if (maintenanceWindowActive) {
this.logger.debug({
message: `Monitor ${monitor.id} is in maintenance window`,
service: SERVICE_NAME,
method: "produce",
});
if (monitor.status !== "maintenance") {
// Clear the status window to avoid incidents being created on next check
await this.monitorsRepository.updateById(monitor.id, monitor.teamId, { status: "maintenance", statusWindow: [] });
}
return null;
}
// Leaving the window is explicit rather than a side effect of evaluating the resulting check, which the
// degraded-egress short-circuit skips. "initializing" is resolved to up or down by the first check that is
// evaluated, so the monitor cannot sit in "maintenance" for the length of an egress outage.
if (monitor.status === "maintenance") {
monitor = await this.monitorsRepository.updateById(monitor.id, monitor.teamId, { status: "initializing" });
}
// Step 1b: Acquire status
const proxyUrl = await this.proxyResolver.resolve(monitor);
const dockerTlsKey = await this.resolveDockerTlsKey(monitor);
const status = await this.networkService.requestStatus(monitor, { proxyUrl, dockerTlsKey });
if (!status) {
throw new Error("No network response");
}
// ****************************
// Step 2: Record
// ****************************
// Step 2a: Create & record a check, return null if fail.
// Built before the egress assessment below so that the probe's duration stays out of `createdAt`.
const check = this.checkService.toCheck(status);
if (!check) {
this.logger.warn({
message: `No check could be built for monitor ${monitor.id}`,
service: SERVICE_NAME,
method: "produce",
details: { code: status.code, message: status.message },
});
return null;
}
// Step 2b: On a transport failure, ask whether the instance itself can reach anything before blaming the
// target. Null means the egress check is disabled (or failed internally) and the check is treated as usual.
const egressStatus = this.isTransportFailure(monitor, status) ? await this.assessEgress(monitor.id) : null;
if (egressStatus !== null) {
check.egressStatus = egressStatus;
}
// Step 2c: Add to buffer
this.bufferService.addToBuffer(check);
// Step 2d: As with the maintenance gate above, the window is cleared at the point the monitor stops
// being evaluated, so the results either side of a spell never end up adjacent and cross the threshold
// on stale data. The length guard keeps this to one write per spell rather than one per degraded check,
// and it runs after the check is buffered so that a failed write cannot cost us the check.
if (egressStatus === "degraded" && monitor.statusWindow?.length) {
await this.monitorsRepository.updateById(monitor.id, monitor.teamId, { statusWindow: [] });
}
// Step 2e: Handle docker logs
if (status.type === "docker") {
const dockerLogs = await this.dockerLogsService.buildDockerLogs(status as MonitorStatusResponse<DockerStatusPayload>);
for (const dockerLog of dockerLogs) {
this.bufferService.addDockerLogToBuffer(dockerLog);
}
}
return { status, check };
};
}
server/src/worker/worker.db-queue.ts
+5 -108
View File
@@ -1,15 +1,8 @@
import { IQueueWorker } from "@/worker/worker.interface.js";
import { IQueueWorker, JobHandlers } from "@/worker/worker.interface.js";
import { type Job, type JobType, LOCK_MS } from "@/domain/jobs/job.type.js";
import { ILogger } from "@/utils/logger.js";
import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js";
import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
import { IChecksRepository } from "@/domain/checks/check.repository.interface.js";
import { ICheckService } from "@/domain/checks/check.service.js";
import { ICheckProducer } from "@/worker/worker.check-producer.js";
import { ICheckEvaluator } from "@/worker/worker.check-evaluator.js";
import { ICheckPipeline } from "@/worker/worker.check-pipeline.js";
import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
import { IWorkerHelper } from "@/worker/worker.helper.js";
import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js";
import { WORKER_STALE_MS } from "@/domain/queue-workers/queue-worker.model.js";
import { QueueMode } from "@/domain/app-settings/app-settings.type.js";
@@ -36,14 +29,8 @@ export interface DBQueueWorkerDependencies {
isDbConnected: () => boolean;
jobsRepository: IJobsRepository; // forwarded
monitorsRepository: IMonitorsRepository; // forwarded
checksRepository: IChecksRepository;
checkService: ICheckService;
handlers: JobHandlers;
bufferService: IBufferService;
checkProducer: ICheckProducer;
checkEvaluator: ICheckEvaluator;
geoCheckPipeline: ICheckPipeline;
dispatcher: IReactorDispatcher;
helper: IWorkerHelper;
queueWorkersRepository: IQueueWorkersRepository; // forwarded
queueMode: QueueMode; // forwarded
queuePrimaryProcesses: boolean; // consumed immediately
@@ -65,15 +52,9 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
egress: 0,
};
private checksRepository: IChecksRepository;
private checkProducer: ICheckProducer;
private checkService: ICheckService;
private checkEvaluator: ICheckEvaluator;
private dispatcher: IReactorDispatcher;
private geoCheckPipeline: ICheckPipeline;
private helper: IWorkerHelper;
private bufferService: IBufferService;
private isDbConnected: () => boolean;
private handlers: JobHandlers;
constructor(dependencies: DBQueueWorkerDependencies) {
super(
@@ -85,13 +66,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
dependencies.workerId
);
this.processesJobs = dependencies.queuePrimaryProcesses === true || dependencies.queueMode === "worker";
this.checksRepository = dependencies.checksRepository;
this.checkProducer = dependencies.checkProducer;
this.checkService = dependencies.checkService;
this.checkEvaluator = dependencies.checkEvaluator;
this.dispatcher = dependencies.dispatcher;
this.geoCheckPipeline = dependencies.geoCheckPipeline;
this.helper = dependencies.helper;
this.handlers = dependencies.handlers;
this.bufferService = dependencies.bufferService;
this.isDbConnected = dependencies.isDbConnected;
}
@@ -104,64 +79,6 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
private getInFlightCount = () => Object.values(this.inFlight).reduce((sum, n) => sum + n, 0);
// ********************
// Stage 1: This can eventually be offloaded to a separate service
// ********************
private runCheck = async (job: Job) => {
if (!job.refId) return;
const monitor = await this.monitorsRepository.findByIdLean(job.refId); // job row has no teamId
if (!monitor) return;
await this.checkProducer.produce(monitor);
};
// ********************
// Stage 2: Evaluator loop stays here
// ********************
private runEvaluate = async (job: Job) => {
if (!job.refId || job.pendingChecks.length === 0) return;
const pendingCheckIds = job.pendingChecks.map((entry) => entry.checkId);
const monitor = await this.monitorsRepository.findByIdLean(job.refId); // job row has no teamId
if (!monitor) {
await this.jobsRepository.pullEvaluated(job.id, pendingCheckIds);
return;
}
const checks = await this.checksRepository.findUnevaluatedByMonitorId(job.refId, job.pendingChecks);
let current = monitor;
for (const check of checks) {
const status = this.checkService.toStatusResponse(check);
const evaluation = await this.checkEvaluator.evaluate(status, check, current);
await this.dispatcher.dispatch(evaluation); // Handle incidents and notifications
const leaseHeld = await this.jobsRepository.pullEvaluated(job.id, [check.id]);
if (!leaseHeld) {
// Another worker has claimed this row and is evaluating the remaining ids; stop here so nothing is applied twice
this.logger.warn({ message: `Lost lease on ${job.id} after check ${check.id}, stopping`, service: SERVICE_NAME, method: "runEvaluate" });
return;
}
current = evaluation.statusChange.monitor; // fresh statusWindow/status/counters for the next check
}
// Ids whose check was not returned (deleted by retention cleanup) would otherwise sit on the row forever
const found = new Set(checks.map((check) => check.id));
const missing = pendingCheckIds.filter((checkId) => !found.has(checkId));
if (missing.length > 0) {
this.logger.warn({
message: `Dropping ${missing.length} pending checks with no stored check: ${missing.join(", ")}`,
service: SERVICE_NAME,
method: "runEvaluate",
});
await this.jobsRepository.pullEvaluated(job.id, missing);
}
};
private runGeoCheck = async (job: Job) => {
const monitor = await this.monitorsRepository.findByIdLean(job.refId!);
if (monitor) await this.geoCheckPipeline.run(monitor); // returns null; no evaluate handoff
};
// Extend the lock while a job runs so a slow-but-alive job is never reclaimed.
private renewLock = async (job: Job): Promise<boolean> => {
try {
@@ -183,27 +100,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
if (!renewLock) clearInterval(renewTimer);
}, LOCK_RENEW_MS);
try {
switch (job.type) {
case "check":
await this.runCheck(job);
break;
case "evaluate":
await this.runEvaluate(job);
break;
case "geo-check":
await this.runGeoCheck(job);
break;
case "cleanup-orphaned":
await this.helper.getCleanupOrphanedJob()(); // Get job and execute
break;
case "cleanup-retention":
await this.helper.getCleanupRetentionJob()(); // Get job and execute
break;
case "egress":
await this.helper.getEgressRecoveryJob()(job); // Re-probe while degraded; removes its own row on recovery
break;
}
await this.handlers[job.type](job);
if (job.intervalMs === null) {
// One-shot job
await this.jobsRepository.recordOneShot(job.id, Date.now());
server/src/worker/worker.check-pipeline.ts → server/src/worker/worker.geo-pipeline.ts
+20 -10
View File
@@ -1,5 +1,5 @@
import { Monitor } from "@/domain/monitors/monitor.type.js";
import { MonitorEvaluation } from "@/worker/worker.interface.js";
import { JobHandler, MonitorEvaluation } from "@/worker/worker.interface.js";
import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js";
import { isWindowActive } from "@/utils/maintenanceWindow.js";
import { ILogger } from "@/utils/logger.js";
@@ -7,15 +7,19 @@ import { IBufferService } from "@/service/bufferService.js";
import { AppError } from "@/utils/AppError.js";
import { supportsGeoCheck } from "@/domain/monitors/monitor.type.js";
import { IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js";
import { Job } from "@/domain/jobs/job.type.js";
import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
const SERVICE_NAME = "CheckPipeline";
const SERVICE_NAME = "GeoChecksPipeline";
export interface ICheckPipeline {
export interface IGeoChecksPipeline {
handle: JobHandler;
run(monitor: Monitor): Promise<MonitorEvaluation | null>; // null = skipped
}
export class GeoChecksPipeline implements ICheckPipeline {
export class GeoChecksPipeline implements IGeoChecksPipeline {
constructor(
private monitorsRepository: IMonitorsRepository,
private maintenanceWindowsRepository: IMaintenanceWindowsRepository,
private geoChecksService: IGeoChecksService,
private bufferService: IBufferService,
@@ -28,6 +32,12 @@ export class GeoChecksPipeline implements ICheckPipeline {
return maintenanceWindows.some((window) => isWindowActive(window, now));
}
handle: JobHandler = async (job: Job) => {
if (!job.refId) return;
const monitor = await this.monitorsRepository.findByIdLean(job.refId);
if (monitor) await this.run(monitor);
};
run = async (monitor: Monitor): Promise<MonitorEvaluation | null> => {
// ****************************
// Step 1: Acquire
@@ -36,7 +46,7 @@ export class GeoChecksPipeline implements ICheckPipeline {
// Step 1a: Guards - skip if unsupported or not enabled
if (!monitor.id) {
throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getHeartbeatGeoJob" });
throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "run" });
}
if (!monitor.geoCheckEnabled) {
@@ -46,7 +56,7 @@ export class GeoChecksPipeline implements ICheckPipeline {
this.logger.debug({
message: `Monitor ${monitor.id} type does not support geo checks, skipping`,
service: SERVICE_NAME,
method: "runGeoChecksPipeline",
method: "run",
});
return null;
}
@@ -55,7 +65,7 @@ export class GeoChecksPipeline implements ICheckPipeline {
this.logger.warn({
message: `No geo check locations configured for monitor ${monitor.id}`,
service: SERVICE_NAME,
method: "runGeoChecksPipeline",
method: "run",
});
return null;
}
@@ -66,7 +76,7 @@ export class GeoChecksPipeline implements ICheckPipeline {
this.logger.debug({
message: `Monitor ${monitor.id} is in maintenance window, skipping geo check`,
service: SERVICE_NAME,
method: "runGeoChecksPipeline",
method: "run",
});
return null;
}
@@ -81,7 +91,7 @@ export class GeoChecksPipeline implements ICheckPipeline {
this.logger.warn({
message: `No geo check could be built for monitor ${monitor.id}`,
service: SERVICE_NAME,
method: "runGeoChecksPipeline",
method: "run",
});
return null;
}
@@ -92,7 +102,7 @@ export class GeoChecksPipeline implements ICheckPipeline {
this.logger.debug({
message: `Geo check job executed for monitor ${monitor.id}`,
service: SERVICE_NAME,
method: "runGeoChecksPipeline",
method: "run",
});
return null;
};
server/src/worker/worker.helper.ts
-14
View File
@@ -20,20 +20,6 @@ export interface IWorkerHelper {
getEgressRecoveryJob(): (job: Job) => Promise<void>;
}
export interface MonitorActionDecision {
shouldCreateIncident: boolean;
shouldResolveIncident: boolean;
shouldSendNotification: boolean;
incidentReason: "status_down" | "threshold_breach" | null;
notificationReason: "status_change" | "threshold_breach" | null;
thresholdBreaches?: {
cpu?: boolean;
memory?: boolean;
disk?: boolean;
temp?: boolean;
};
}
export class WorkerHelper implements IWorkerHelper {
static SERVICE_NAME = SERVICE_NAME;
server/src/worker/worker.interface.ts
+18 -3
View File
@@ -1,17 +1,32 @@
import { Monitor } from "@/domain/monitors/monitor.type.js";
import { Check } from "@/domain/checks/check.type.js";
import { JobType } from "@/domain/jobs/job.type.js";
import { Job, JobType } from "@/domain/jobs/job.type.js";
import { MonitorPayloadMap, MonitorStatusResponse, StatusChangeResult } from "@/types/network.js";
import { MonitorActionDecision } from "@/worker/worker.helper.js";
import { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js";
import type { QueueMode } from "@/domain/app-settings/app-settings.type.js";
export type JobHandler = (job: Job) => Promise<void>;
export type JobHandlers = Record<JobType, JobHandler>;
export interface MonitorActionDecision {
shouldCreateIncident: boolean;
shouldResolveIncident: boolean;
shouldSendNotification: boolean;
incidentReason: "status_down" | "threshold_breach" | null;
notificationReason: "status_change" | "threshold_breach" | null;
thresholdBreaches?: {
cpu?: boolean;
memory?: boolean;
disk?: boolean;
temp?: boolean;
};
}
export type MonitorEvaluation = {
monitor: Monitor;
status: MonitorStatusResponse<MonitorPayloadMap[keyof MonitorPayloadMap]>; // raw result from networkService.requestStatus
check: Check;
statusChange: StatusChangeResult; // from statusService.updateMonitorStatus
decision: MonitorActionDecision; // from MonitorStatusPolicy
decision: MonitorActionDecision;
};
export type WorkerJobFailure = {
server/src/worker/worker.monitor-status-policy.ts
-46
View File
@@ -1,46 +0,0 @@
import { StatusChangeResult } from "@/types/network.js";
import { MonitorActionDecision } from "@/worker/worker.helper.js";
export interface IMonitorStatusPolicy {
evaluate(statusChange: StatusChangeResult): MonitorActionDecision;
}
export class MonitorStatusPolicy implements IMonitorStatusPolicy {
evaluate = (statusChangeResult: StatusChangeResult): MonitorActionDecision => {
const { monitor, statusChanged, prevStatus } = statusChangeResult;
// Initialize result
const decision: MonitorActionDecision = {
shouldCreateIncident: false,
shouldResolveIncident: false,
shouldSendNotification: false,
incidentReason: null,
notificationReason: null,
};
if (!statusChanged) {
return decision;
}
if (monitor.status === "down") {
// Monitor went down (unreachable)
decision.shouldCreateIncident = true;
decision.shouldSendNotification = true;
decision.incidentReason = "status_down";
decision.notificationReason = "status_change";
} else if (monitor.status === "breached") {
// Hardware monitor exceeded thresholds
decision.shouldCreateIncident = true;
decision.shouldSendNotification = true;
decision.incidentReason = "threshold_breach";
decision.notificationReason = "threshold_breach";
} else if (monitor.status === "up" && (prevStatus === "down" || prevStatus === "breached")) {
// Monitor recovered from down or breached state
decision.shouldResolveIncident = true;
decision.shouldSendNotification = true;
decision.notificationReason = "status_change";
}
return decision;
};
}
server/src/worker/worker.pipeline.ts
+336
View File
@@ -0,0 +1,336 @@
import { IChecksRepository } from "@/domain/checks/check.repository.interface.js";
import { ICheckService } from "@/domain/checks/check.service.js";
import { Check } from "@/domain/checks/check.type.js";
import { IDockerLogsService } from "@/domain/docker/docker-log.service.js";
import { IEgressService, isHttpStatusCode } from "@/domain/egress/egress.service.js";
import type { EgressStatus } from "@/domain/egress/egress.type.js";
import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js";
import { PendingCheck } from "@/domain/jobs/job.type.js";
import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js";
import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
import { isEgressAttributable, Monitor } from "@/domain/monitors/monitor.type.js";
import { IBufferService } from "@/service/bufferService.js";
import { IProxyResolver } from "@/service/network/ProxyResolver.js";
import { INetworkService } from "@/service/networkService.js";
import { IStatusService } from "@/service/statusService.js";
import { DockerStatusPayload, MonitorStatusResponse, StatusChangeResult } from "@/types/network.js";
import { AppError } from "@/utils/AppError.js";
import { ILogger } from "@/utils/logger.js";
import { isWindowActive } from "@/utils/maintenanceWindow.js";
import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
import { JobHandler, MonitorActionDecision, MonitorEvaluation } from "@/worker/worker.interface.js";
const SERVICE_NAME = "WorkerPipeline";
// The three entry points: the queue calls the two handlers, the buffer calls ingestChecks.
// produce and evaluateCheck stay public on the class so tests can drive one stage without the queue or the buffer.
export interface IWorkerPipeline {
handleCheck: JobHandler;
ingestChecks(checks: Check[]): Promise<void>;
handleEvaluate: JobHandler;
}
export interface WorkerPipelineDependencies {
logger: ILogger;
monitorsRepository: IMonitorsRepository;
maintenanceWindowsRepository: IMaintenanceWindowsRepository;
checksRepository: IChecksRepository;
jobsRepository: IJobsRepository;
checkService: ICheckService;
networkService: INetworkService;
proxyResolver: IProxyResolver;
bufferService: IBufferService;
dockerLogsService: IDockerLogsService;
egressService: IEgressService;
statusService: IStatusService;
dispatcher: IReactorDispatcher;
}
export class WorkerPipeline implements IWorkerPipeline {
static SERVICE_NAME = SERVICE_NAME;
private unarmed: Map<string, PendingCheck[]> = new Map();
constructor(private deps: WorkerPipelineDependencies) {}
private isInMaintenanceWindow = async (monitorId: string, teamId: string) => {
const windows = await this.deps.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId);
const now = new Date();
return windows.some((window) => isWindowActive(window, now));
};
// ****************************************************************
// Stage 1: Check job. Probe the host and buffer the result.
// ****************************************************************
handleCheck: JobHandler = async (job) => {
if (!job.refId) return;
const monitor = await this.deps.monitorsRepository.findByIdLean(job.refId);
if (!monitor) return;
await this.produce(monitor);
};
produce = async (monitor: Monitor) => {
if (!monitor.id) {
throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "produce" });
}
// ****************************
// Step 1: Acquire
// ****************************
// Step 1a: Maintenance window gate - skip if in maintenance
const maintenanceWindowActive = await this.isInMaintenanceWindow(monitor.id, monitor.teamId);
if (maintenanceWindowActive) {
this.deps.logger.debug({
message: `Monitor ${monitor.id} is in maintenance window`,
service: SERVICE_NAME,
method: "produce",
});
if (monitor.status !== "maintenance") {
// Clear the status window to avoid incidents being created on next check
await this.deps.monitorsRepository.updateById(monitor.id, monitor.teamId, { status: "maintenance", statusWindow: [] });
}
return null;
}
// Leaving the window is explicit rather than a side effect of evaluating the resulting check, which the
// degraded-egress short-circuit skips. "initializing" is resolved to up or down by the first check that is
// evaluated, so the monitor cannot sit in "maintenance" for the length of an egress outage.
if (monitor.status === "maintenance") {
monitor = await this.deps.monitorsRepository.updateById(monitor.id, monitor.teamId, { status: "initializing" });
}
// Step 1b: Acquire status
const proxyUrl = await this.deps.proxyResolver.resolve(monitor);
const dockerTlsKey = await this.resolveDockerTlsKey(monitor);
const status = await this.deps.networkService.requestStatus(monitor, { proxyUrl, dockerTlsKey });
if (!status) {
throw new Error("No network response");
}
// ****************************
// Step 2: Record
// ****************************
// Step 2a: Create & record a check, return null if fail.
// Built before the egress assessment below so that the probe's duration stays out of `createdAt`.
const check = this.deps.checkService.toCheck(status);
if (!check) {
this.deps.logger.warn({
message: `No check could be built for monitor ${monitor.id}`,
service: SERVICE_NAME,
method: "produce",
details: { code: status.code, message: status.message },
});
return null;
}
// Step 2b: On a transport failure, ask whether the instance itself can reach anything before blaming the
// target. Null means the egress check is disabled (or failed internally) and the check is treated as usual.
const egressStatus = this.isTransportFailure(monitor, status) ? await this.assessEgress(monitor.id) : null;
if (egressStatus !== null) {
check.egressStatus = egressStatus;
}
// Step 2c: Add to buffer
this.deps.bufferService.addToBuffer(check);
// Step 2d: As with the maintenance gate above, the window is cleared at the point the monitor stops
// being evaluated, so the results either side of a spell never end up adjacent and cross the threshold
// on stale data. The length guard keeps this to one write per spell rather than one per degraded check,
// and it runs after the check is buffered so that a failed write cannot cost us the check.
if (egressStatus === "degraded" && monitor.statusWindow?.length) {
await this.deps.monitorsRepository.updateById(monitor.id, monitor.teamId, { statusWindow: [] });
}
// Step 2e: Handle docker logs
if (status.type === "docker") {
const dockerLogs = await this.deps.dockerLogsService.buildDockerLogs(status as MonitorStatusResponse<DockerStatusPayload>);
for (const dockerLog of dockerLogs) {
this.deps.bufferService.addDockerLogToBuffer(dockerLog);
}
}
return { status, check };
};
private async resolveDockerTlsKey(monitor: Monitor): Promise<string | undefined> {
if (monitor.type !== "docker" || !monitor.dockerTlsKeySet || !monitor.id) return undefined;
const dockerTlsKey = await this.deps.monitorsRepository.findDockerTlsKeyById(monitor.id);
return dockerTlsKey ?? undefined;
}
// Only a failure to reach the target at all can be the instance's own fault. A provider that can tell the
// two apart says so outright; for the rest, any HTTP status code proves the target answered (a 4xx, a 5xx,
// or a 200 with a content mismatch).
private peerAnswered(status: MonitorStatusResponse): boolean {
return status.peerResponded ?? isHttpStatusCode(status.code);
}
private isTransportFailure(monitor: Monitor, status: MonitorStatusResponse): boolean {
return status.status === false && isEgressAttributable(monitor) && !this.peerAnswered(status);
}
// The egress check must never stop a check being recorded, so a rejection is logged and treated as unknown.
private async assessEgress(monitorId: string): Promise<EgressStatus | null> {
try {
return await this.deps.egressService.assessAfterFailure();
} catch (error: unknown) {
this.deps.logger.warn({
message: `Egress assessment failed for monitor ${monitorId}: ${error instanceof Error ? error.message : String(error)}`,
service: SERVICE_NAME,
method: "assessEgress",
});
return null;
}
}
// ****************************************************************
// Stage 2: Ingestion
// Called by the buffer, checks are inserted and evaluation jobs created
// ****************************************************************
ingestChecks = async (checks: Check[]) => {
if (checks.length > 0) await this.deps.checkService.createChecks(checks);
// Group by monitor, starting from any entries a previous flush failed to arm
const byMonitor = this.unarmed;
this.unarmed = new Map();
for (const check of checks) {
const pending = byMonitor.get(check.metadata.monitorId) ?? [];
pending.push({ checkId: check.id, createdAt: new Date(check.createdAt).getTime() });
byMonitor.set(check.metadata.monitorId, pending);
}
if (byMonitor.size === 0) return;
const now = Date.now();
const ops: Promise<boolean>[] = [];
for (const [monitorId, pendingChecks] of byMonitor) {
ops.push(
this.deps.jobsRepository.upsertEvaluate(monitorId, pendingChecks, now).catch((error: unknown) => {
// The checks are already stored, so a failed arm is retried on the next ingest rather than dropped
this.unarmed.set(monitorId, pendingChecks);
this.deps.logger.error({
message: `Could not arm evaluation for monitor ${monitorId}, retrying on next ingest: ${error instanceof Error ? error.message : String(error)}`,
service: SERVICE_NAME,
method: "ingestChecks",
});
return false;
})
);
}
await Promise.all(ops);
};
// ****************************************************************
// Stage 3: Evaluation
// Apply each pending check to the monitor -> decide -> react.
// ****************************************************************
handleEvaluate: JobHandler = async (job) => {
if (!job.refId || job.pendingChecks.length === 0) return;
const pendingCheckIds = job.pendingChecks.map((entry) => entry.checkId);
const monitor = await this.deps.monitorsRepository.findByIdLean(job.refId); // job row has no teamId
if (!monitor) {
await this.deps.jobsRepository.pullEvaluated(job.id, pendingCheckIds);
return;
}
const checks = await this.deps.checksRepository.findUnevaluatedByMonitorId(job.refId, job.pendingChecks);
let current = monitor;
for (const check of checks) {
const status = this.deps.checkService.toStatusResponse(check);
const evaluation = await this.evaluateCheck(status, check, current);
await this.deps.dispatcher.dispatch(evaluation); // Handle incidents and notifications
const leaseHeld = await this.deps.jobsRepository.pullEvaluated(job.id, [check.id]);
if (!leaseHeld) {
// Another worker has claimed this row and is evaluating the remaining ids; stop here so nothing is applied twice
this.deps.logger.warn({
message: `Lost lease on ${job.id} after check ${check.id}, stopping`,
service: SERVICE_NAME,
method: "handleEvaluate",
});
return;
}
current = evaluation.statusChange.monitor; // fresh statusWindow/status/counters for the next check
}
// Ids whose check was not returned (deleted by retention cleanup) would otherwise sit on the row forever
const found = new Set(checks.map((check) => check.id));
const missing = pendingCheckIds.filter((checkId) => !found.has(checkId));
if (missing.length > 0) {
this.deps.logger.warn({
message: `Dropping ${missing.length} pending checks with no stored check: ${missing.join(", ")}`,
service: SERVICE_NAME,
method: "handleEvaluate",
});
await this.deps.jobsRepository.pullEvaluated(job.id, missing);
}
};
// A check taken while the instance had no outbound connectivity says nothing about the target.
// It is left out of the status window, running stats and monitor status entirely, so that no incident
// opens for it and no spurious "resolved"/"up" fires once egress returns.
private skipDegradedEgressCheck = (status: MonitorStatusResponse, check: Check, monitor: Monitor): MonitorEvaluation => {
this.deps.logger.debug({
message: `Skipping evaluation of check ${check.id} for monitor ${monitor.id}: instance egress was degraded`,
service: SERVICE_NAME,
method: "evaluateCheck",
});
const decision: MonitorActionDecision = {
shouldCreateIncident: false,
shouldResolveIncident: false,
shouldSendNotification: false,
incidentReason: null,
notificationReason: null,
};
return {
monitor,
status,
check,
statusChange: { monitor, statusChanged: false, prevStatus: monitor.status, code: status.code, timestamp: Date.now() },
decision,
};
};
evaluateCheck = async (status: MonitorStatusResponse, check: Check, monitor: Monitor): Promise<MonitorEvaluation> => {
if (check.egressStatus === "degraded") {
return this.skipDegradedEgressCheck(status, check, monitor);
}
const statusChange = await this.deps.statusService.updateMonitorStatus(status, check, monitor);
const decision = this.decide(statusChange);
return { monitor: statusChange.monitor, status, check, statusChange, decision };
};
private decide = (statusChange: StatusChangeResult): MonitorActionDecision => {
const { monitor, statusChanged, prevStatus } = statusChange;
const decision: MonitorActionDecision = {
shouldCreateIncident: false,
shouldResolveIncident: false,
shouldSendNotification: false,
incidentReason: null,
notificationReason: null,
};
if (!statusChanged) return decision;
if (monitor.status === "down") {
decision.shouldCreateIncident = true;
decision.shouldSendNotification = true;
decision.incidentReason = "status_down";
decision.notificationReason = "status_change";
} else if (monitor.status === "breached") {
decision.shouldCreateIncident = true;
decision.shouldSendNotification = true;
decision.incidentReason = "threshold_breach";
decision.notificationReason = "threshold_breach";
} else if (monitor.status === "up" && (prevStatus === "down" || prevStatus === "breached")) {
decision.shouldResolveIncident = true;
decision.shouldSendNotification = true;
decision.notificationReason = "status_change";
}
return decision;
};
}