Remove old queues
This commit is contained in:
Alex Holliday committed
1 parent
f476710c29
commit
cd91e83817
2 files changed
-832
No files matched your search
@@ -1,395 +0,0 @@
|
||||
import { IMonitorsRepository } from "@/repositories/index.js";
|
||||
import { ILogger } from "@/utils/logger.js";
|
||||
import Scheduler from "super-simple-scheduler";
|
||||
import { ISuperSimpleQueueHelper } from "@/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.js";
|
||||
import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js";
|
||||
const SERVICE_NAME = "JobQueue";
|
||||
|
||||
type QueueJobFailure = {
|
||||
monitorId: string | number;
|
||||
monitorUrl: string | null;
|
||||
monitorType: string | null;
|
||||
failedAt: number | null;
|
||||
failCount: number;
|
||||
failReason: string | null;
|
||||
};
|
||||
|
||||
type QueueMetrics = {
|
||||
jobs: number;
|
||||
activeJobs: number;
|
||||
failingJobs: number;
|
||||
jobsWithFailures: QueueJobFailure[];
|
||||
totalRuns: number;
|
||||
totalFailures: number;
|
||||
};
|
||||
|
||||
type QueueJobSummary = {
|
||||
monitorId: string | number;
|
||||
monitorUrl: string | null;
|
||||
monitorType: string | null;
|
||||
monitorInterval: number | null;
|
||||
monitorGeoInterval: number | null;
|
||||
monitorActive: boolean | null;
|
||||
active: boolean;
|
||||
lockedAt: number | null;
|
||||
runCount: number;
|
||||
failCount: number;
|
||||
failReason: string | null;
|
||||
lastRunAt: number | null;
|
||||
lastFinishedAt: number | null;
|
||||
lastRunTook: number | null;
|
||||
lastFailedAt: number | null;
|
||||
repeat: number | null;
|
||||
};
|
||||
|
||||
export interface ISuperSimpleQueue {
|
||||
readonly serviceName: string;
|
||||
init(): Promise<boolean>;
|
||||
addJob(monitorId: string, monitor: Monitor): Promise<void>;
|
||||
deleteJob(monitor: Monitor): Promise<void>;
|
||||
pauseJob(monitor: Monitor): Promise<void>;
|
||||
resumeJob(monitor: Monitor): Promise<void>;
|
||||
updateJob(monitor: Monitor): Promise<void>;
|
||||
shutdown(): Promise<void>;
|
||||
getMetrics(): Promise<QueueMetrics>;
|
||||
getJobs(): Promise<QueueJobSummary[]>;
|
||||
flushQueues(): Promise<{ success: boolean }>;
|
||||
}
|
||||
|
||||
export class SuperSimpleQueue implements ISuperSimpleQueue {
|
||||
static SERVICE_NAME = SERVICE_NAME;
|
||||
|
||||
private logger: ILogger;
|
||||
private helper: ISuperSimpleQueueHelper;
|
||||
private monitorsRepository: IMonitorsRepository;
|
||||
private readonly scheduler: Scheduler;
|
||||
|
||||
constructor(logger: ILogger, helper: ISuperSimpleQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) {
|
||||
this.logger = logger;
|
||||
this.helper = helper;
|
||||
this.monitorsRepository = monitorsRepository;
|
||||
this.scheduler = scheduler;
|
||||
}
|
||||
|
||||
get serviceName() {
|
||||
return SuperSimpleQueue.SERVICE_NAME;
|
||||
}
|
||||
|
||||
private registerListeners = () => {
|
||||
this.scheduler.on("scheduler:start", () => {
|
||||
this.logger.info({
|
||||
message: "Scheduler started",
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("scheduler:stop", () => {
|
||||
this.logger.info({
|
||||
message: "Scheduler stopped",
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("scheduler:error", (error) => {
|
||||
this.logger.error({
|
||||
message: `Scheduler error: ${error instanceof Error ? error.message : String(error)}`,
|
||||
service: SERVICE_NAME,
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:abort", (job, reason) => {
|
||||
this.logger.warn({
|
||||
message: `${job.id} aborted: ${reason}`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:attempt", (job, attempt) => {
|
||||
this.logger.debug({
|
||||
message: `${job.id} attempt ${attempt}`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:complete", (job) => {
|
||||
this.logger.debug({
|
||||
message: `${job.id} completed successfully`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:exhausted", (job, error) => {
|
||||
this.logger.error({
|
||||
message: `${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(error)}`,
|
||||
service: SERVICE_NAME,
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:fail", (job, error, attempt) => {
|
||||
this.logger.warn({
|
||||
message: `${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`,
|
||||
service: SERVICE_NAME,
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:start", (job) => {
|
||||
this.logger.debug({
|
||||
message: `${job.id} started`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
static async create(logger: ILogger, helper: ISuperSimpleQueueHelper, monitorsRepository: IMonitorsRepository) {
|
||||
const scheduler = new Scheduler({
|
||||
concurrency: 50,
|
||||
// storeType: "mongo",
|
||||
// storeType: "redis",
|
||||
// dbUri: envSettings.dbConnectionString,
|
||||
});
|
||||
const instance = new SuperSimpleQueue(logger, helper, monitorsRepository, scheduler);
|
||||
await instance.init();
|
||||
|
||||
return instance;
|
||||
}
|
||||
|
||||
init = async () => {
|
||||
try {
|
||||
this.registerListeners();
|
||||
|
||||
this.scheduler.start();
|
||||
this.scheduler.addTemplate("monitor-job", this.helper.getHeartbeatJob());
|
||||
this.scheduler.addTemplate("geo-check-job", this.helper.getHeartbeatGeoJob());
|
||||
this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob());
|
||||
this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob());
|
||||
const monitors = await this.monitorsRepository.findAll();
|
||||
if (!monitors) {
|
||||
return true;
|
||||
}
|
||||
for (const monitor of monitors) {
|
||||
const randomOffset = Math.floor(Math.random() * 100);
|
||||
setTimeout(() => {
|
||||
this.addJob(monitor.id, monitor);
|
||||
}, randomOffset);
|
||||
}
|
||||
|
||||
this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true });
|
||||
this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000 });
|
||||
|
||||
return true;
|
||||
} catch (error: unknown) {
|
||||
this.logger.error({
|
||||
message: `Failed to initialize SuperSimpleQueue: ${error instanceof Error ? error.message : String(error)}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "init",
|
||||
});
|
||||
return false;
|
||||
}
|
||||
};
|
||||
|
||||
addJob = async (monitorId: string, monitor: Monitor) => {
|
||||
this.scheduler.addJob({
|
||||
id: monitorId,
|
||||
template: "monitor-job",
|
||||
repeat: monitor.interval,
|
||||
active: monitor.isActive,
|
||||
data: monitor,
|
||||
});
|
||||
|
||||
// Return early if we don't need geo checks
|
||||
if (!supportsGeoCheck(monitor.type)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Add geo check job if enabled for HTTP monitors
|
||||
if (monitor.geoCheckEnabled) {
|
||||
this.scheduler.addJob({
|
||||
id: `${monitorId}-geo`,
|
||||
template: "geo-check-job",
|
||||
repeat: monitor.geoCheckInterval,
|
||||
active: monitor.isActive,
|
||||
data: monitor,
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
deleteJob = async (monitor: Monitor) => {
|
||||
this.scheduler.removeJob(monitor.id);
|
||||
const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
|
||||
if (geoJob) await this.scheduler.removeJob(`${monitor.id}-geo`);
|
||||
};
|
||||
|
||||
pauseJob = async (monitor: Monitor) => {
|
||||
const result = await this.scheduler.pauseJob(monitor.id);
|
||||
if (result === false) {
|
||||
throw new Error("Failed to pause monitor");
|
||||
}
|
||||
this.scheduler.updateJob(monitor.id, { data: monitor });
|
||||
|
||||
const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
|
||||
if (geoJob) {
|
||||
const geoRes = await this.scheduler.pauseJob(`${monitor.id}-geo`);
|
||||
if (geoRes === false) {
|
||||
this.logger.error({
|
||||
message: `Failed to pause geo check job for monitor ${monitor.id}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "pauseJob",
|
||||
});
|
||||
}
|
||||
this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor });
|
||||
}
|
||||
|
||||
this.logger.debug({
|
||||
message: `Paused monitor ${monitor.id}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "pauseJob",
|
||||
});
|
||||
};
|
||||
|
||||
resumeJob = async (monitor: Monitor) => {
|
||||
const result = await this.scheduler.resumeJob(monitor.id);
|
||||
if (result === false) {
|
||||
throw new Error("Failed to resume monitor");
|
||||
}
|
||||
this.scheduler.updateJob(monitor.id, { data: monitor });
|
||||
|
||||
const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
|
||||
if (geoJob) {
|
||||
const geoRes = await this.scheduler.resumeJob(`${monitor.id}-geo`);
|
||||
if (geoRes === false) {
|
||||
this.logger.error({
|
||||
message: `Failed to resume geo check job for monitor ${monitor.id}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "resumeJob",
|
||||
});
|
||||
}
|
||||
this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor });
|
||||
}
|
||||
this.logger.debug({
|
||||
message: `Resumed monitor ${monitor.id}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "resumeJob",
|
||||
});
|
||||
};
|
||||
|
||||
private syncGeoJob = async (monitor: Monitor) => {
|
||||
const geoJobId = `${monitor.id}-geo`;
|
||||
const existingGeoJob = await this.scheduler.getJob(geoJobId);
|
||||
|
||||
// If geoChecks have been disabled, or the monitor type doesn't support them, remove
|
||||
if (!monitor.geoCheckEnabled || !supportsGeoCheck(monitor.type)) {
|
||||
if (existingGeoJob) this.scheduler.removeJob(geoJobId);
|
||||
return;
|
||||
}
|
||||
|
||||
// If the job exists, update it
|
||||
if (existingGeoJob) {
|
||||
this.scheduler.updateJob(geoJobId, { repeat: monitor.geoCheckInterval, active: monitor.isActive, data: monitor });
|
||||
return;
|
||||
}
|
||||
|
||||
// Otherwise, create it
|
||||
this.scheduler.addJob({
|
||||
id: geoJobId,
|
||||
template: "geo-check-job",
|
||||
repeat: monitor.geoCheckInterval,
|
||||
active: monitor.isActive,
|
||||
data: monitor,
|
||||
});
|
||||
};
|
||||
|
||||
updateJob = async (monitor: Monitor) => {
|
||||
this.scheduler.updateJob(monitor.id, { repeat: monitor.interval, data: monitor });
|
||||
await this.syncGeoJob(monitor);
|
||||
};
|
||||
|
||||
shutdown = async () => {
|
||||
this.scheduler.stop();
|
||||
};
|
||||
|
||||
getMetrics = async () => {
|
||||
const jobs = await this.scheduler.getJobs();
|
||||
const metrics = jobs.reduce(
|
||||
(acc, job) => {
|
||||
const runCount = job.runCount ?? 0;
|
||||
const failCount = job.failCount ?? 0;
|
||||
const lastFailedAt = job.lastFailedAt ?? 0;
|
||||
const lastRunAt = job.lastRunAt ?? 0;
|
||||
acc.totalRuns += runCount;
|
||||
acc.totalFailures += failCount;
|
||||
acc.jobs++;
|
||||
if (failCount > 0 && lastFailedAt >= lastRunAt) {
|
||||
acc.failingJobs++;
|
||||
}
|
||||
|
||||
if (job.lockedAt) {
|
||||
acc.activeJobs++;
|
||||
}
|
||||
|
||||
if (failCount > 0) {
|
||||
acc.jobsWithFailures.push({
|
||||
monitorId: job.id,
|
||||
monitorUrl: job.data?.url || null,
|
||||
monitorType: job.data?.type || null,
|
||||
failedAt: job.lastFailedAt ?? null,
|
||||
failCount,
|
||||
failReason: job.lastFailReason ?? null,
|
||||
});
|
||||
}
|
||||
return acc;
|
||||
},
|
||||
{
|
||||
jobs: 0,
|
||||
activeJobs: 0,
|
||||
failingJobs: 0,
|
||||
jobsWithFailures: [] as Array<{
|
||||
monitorId: string | number;
|
||||
monitorUrl: string;
|
||||
monitorType: MonitorType;
|
||||
failedAt: number | null;
|
||||
failCount: number;
|
||||
failReason: string | null;
|
||||
}>,
|
||||
totalRuns: 0,
|
||||
totalFailures: 0,
|
||||
}
|
||||
);
|
||||
return metrics;
|
||||
};
|
||||
|
||||
getJobs = async () => {
|
||||
const jobs = await this.scheduler.getJobs();
|
||||
return jobs.map((job) => {
|
||||
return {
|
||||
monitorId: job.id,
|
||||
monitorUrl: job.data?.url || null,
|
||||
monitorType: job.data?.type || null,
|
||||
monitorInterval: job.data?.interval || null,
|
||||
monitorGeoInterval: job.data?.geoCheckInterval || null,
|
||||
monitorActive: job.data?.isActive ?? null,
|
||||
active: job.active,
|
||||
repeat: job.repeat ?? null,
|
||||
lockedAt: job.lockedAt ?? null,
|
||||
runCount: job.runCount ?? 0,
|
||||
failCount: job.failCount ?? 0,
|
||||
failReason: job.lastFailReason ?? null,
|
||||
lastRunAt: job.lastRunAt ?? null,
|
||||
lastFinishedAt: job.lastFinishedAt ?? null,
|
||||
lastRunTook: job.lockedAt ? null : (job.lastFinishedAt ?? 0) - (job.lastRunAt ?? 0),
|
||||
lastFailedAt: job.lastFailedAt ?? null,
|
||||
};
|
||||
});
|
||||
};
|
||||
|
||||
flushQueues = async () => {
|
||||
const stopRes = await this.scheduler.stop();
|
||||
const flushRes = await this.scheduler.flushJobs();
|
||||
const initRes = await this.init();
|
||||
return {
|
||||
success: Boolean(stopRes && flushRes && initRes),
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -1,437 +0,0 @@
|
||||
const SERVICE_NAME = "JobQueueHelper";
|
||||
import type { Monitor } from "@/types/monitor.js";
|
||||
import { supportsGeoCheck } from "@/types/monitor.js";
|
||||
import { AppError } from "@/utils/AppError.js";
|
||||
import {
|
||||
ICheckService,
|
||||
INetworkService,
|
||||
INotificationsService,
|
||||
ISettingsService,
|
||||
IStatusService,
|
||||
IncidentService,
|
||||
type IGeoChecksService,
|
||||
} from "@/service/index.js";
|
||||
import { CHECK_TTL_SENTINEL, type StatusChangeResult } from "@/types/index.js";
|
||||
import { isWindowActive } from "@/utils/maintenanceWindow.js";
|
||||
import {
|
||||
IMaintenanceWindowsRepository,
|
||||
IMonitorsRepository,
|
||||
ITeamsRepository,
|
||||
IMonitorStatsRepository,
|
||||
IChecksRepository,
|
||||
IIncidentsRepository,
|
||||
IGeoChecksRepository,
|
||||
} from "@/repositories/index.js";
|
||||
import { ILogger } from "@/utils/logger.js";
|
||||
import { IBufferService } from "@/service/index.js";
|
||||
|
||||
export interface ISuperSimpleQueueHelper {
|
||||
readonly serviceName: string;
|
||||
getHeartbeatJob(): (monitor: Monitor) => Promise<void>;
|
||||
getHeartbeatGeoJob(): (monitor: Monitor) => Promise<void>;
|
||||
getCleanupOrphanedJob(): () => Promise<void>;
|
||||
getCleanupRetentionJob(): () => Promise<void>;
|
||||
isInMaintenanceWindow(monitorId: string, teamId: string): Promise<boolean>;
|
||||
}
|
||||
|
||||
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 SuperSimpleQueueHelper implements ISuperSimpleQueueHelper {
|
||||
static SERVICE_NAME = SERVICE_NAME;
|
||||
|
||||
private logger: ILogger;
|
||||
private networkService: INetworkService;
|
||||
private statusService: IStatusService;
|
||||
private notificationsService: INotificationsService;
|
||||
private checkService: ICheckService;
|
||||
private settingsService: ISettingsService;
|
||||
private buffer: IBufferService;
|
||||
private incidentService: IncidentService;
|
||||
private maintenanceWindowsRepository: IMaintenanceWindowsRepository;
|
||||
private monitorsRepository: IMonitorsRepository;
|
||||
private teamsRepository: ITeamsRepository;
|
||||
private monitorStatsRepository: IMonitorStatsRepository;
|
||||
private checksRepository: IChecksRepository;
|
||||
private incidentsRepository: IIncidentsRepository;
|
||||
private geoChecksService: IGeoChecksService;
|
||||
private geoChecksRepository: IGeoChecksRepository;
|
||||
|
||||
constructor(
|
||||
logger: ILogger,
|
||||
networkService: INetworkService,
|
||||
statusService: IStatusService,
|
||||
notificationsService: INotificationsService,
|
||||
checkService: ICheckService,
|
||||
settingsService: ISettingsService,
|
||||
buffer: IBufferService,
|
||||
incidentService: IncidentService,
|
||||
maintenanceWindowsRepository: IMaintenanceWindowsRepository,
|
||||
monitorsRepository: IMonitorsRepository,
|
||||
teamsRepository: ITeamsRepository,
|
||||
monitorStatsRepository: IMonitorStatsRepository,
|
||||
checksRepository: IChecksRepository,
|
||||
incidentsRepository: IIncidentsRepository,
|
||||
geoChecksService: IGeoChecksService,
|
||||
geoChecksRepository: IGeoChecksRepository
|
||||
) {
|
||||
this.logger = logger;
|
||||
this.networkService = networkService;
|
||||
this.statusService = statusService;
|
||||
this.checkService = checkService;
|
||||
this.settingsService = settingsService;
|
||||
this.buffer = buffer;
|
||||
this.notificationsService = notificationsService;
|
||||
this.incidentService = incidentService;
|
||||
this.maintenanceWindowsRepository = maintenanceWindowsRepository;
|
||||
this.monitorsRepository = monitorsRepository;
|
||||
this.teamsRepository = teamsRepository;
|
||||
this.monitorStatsRepository = monitorStatsRepository;
|
||||
this.checksRepository = checksRepository;
|
||||
this.incidentsRepository = incidentsRepository;
|
||||
this.geoChecksService = geoChecksService;
|
||||
this.geoChecksRepository = geoChecksRepository;
|
||||
}
|
||||
|
||||
get serviceName() {
|
||||
return SuperSimpleQueueHelper.SERVICE_NAME;
|
||||
}
|
||||
|
||||
getHeartbeatJob = () => {
|
||||
return async (monitor: Monitor) => {
|
||||
try {
|
||||
const monitorId = monitor.id;
|
||||
const teamId = monitor.teamId;
|
||||
if (!monitorId) {
|
||||
throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getMonitorJob" });
|
||||
}
|
||||
|
||||
// Step 1. Check for maintenance window, if found, skip the check
|
||||
|
||||
const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId, teamId);
|
||||
if (maintenanceWindowActive) {
|
||||
this.logger.debug({
|
||||
message: `Monitor ${monitorId} is in maintenance window`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getMonitorJob",
|
||||
});
|
||||
if (monitor.status !== "maintenance") {
|
||||
await this.monitorsRepository.updateById(monitorId, teamId, { status: "maintenance" });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 2. Request monitor status
|
||||
const status = await this.networkService.requestStatus(monitor);
|
||||
if (!status) {
|
||||
throw new Error("No network response");
|
||||
}
|
||||
|
||||
// Step 3. Build check
|
||||
const check = this.checkService.buildCheck(status);
|
||||
if (!check) {
|
||||
this.logger.warn({
|
||||
message: `No check could be built for monitor ${monitorId}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getMonitorJob",
|
||||
details: { code: status.code, message: status.message },
|
||||
});
|
||||
return;
|
||||
}
|
||||
// Step 4 Add check to buffer
|
||||
this.buffer.addToBuffer(check);
|
||||
// Step 4. Update monitor status
|
||||
const statusChangeResult = await this.statusService.updateMonitorStatus(status, check);
|
||||
|
||||
// Step 5. Get decisions
|
||||
const decision = this.evaluateMonitorAction(statusChangeResult);
|
||||
|
||||
// Step 6. Handle notifications (best effort, continue even in event of failure, don't wait)
|
||||
if (decision.shouldSendNotification) {
|
||||
this.notificationsService.handleNotifications(statusChangeResult.monitor, status, decision).catch((error: unknown) => {
|
||||
this.logger.error({
|
||||
message: `Error sending notifications for job ${statusChangeResult.monitor.id}: ${error instanceof Error ? error.message : "Unknown error"}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getMonitorJob",
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// Step 7. Handle incidents
|
||||
try {
|
||||
await this.incidentService.handleIncident(statusChangeResult.monitor, statusChangeResult.code, decision, status);
|
||||
} catch (error: unknown) {
|
||||
this.logger.warn({
|
||||
message: `Error handling incident for job ${monitor.id}: ${error instanceof Error ? error.message : "Unknown error"}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getMonitorJob",
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
this.logger.warn({
|
||||
message: error instanceof Error ? error.message : "Unknown error",
|
||||
service: SERVICE_NAME,
|
||||
method: "getMonitorJob",
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
getCleanupOrphanedJob = () => {
|
||||
return async () => {
|
||||
try {
|
||||
this.logger.info({
|
||||
message: "Starting cleanup of orphaned data",
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
|
||||
// Get all valid team IDs
|
||||
const validTeamIds = await this.teamsRepository.findAllTeamIds();
|
||||
this.logger.debug({
|
||||
message: `Found ${validTeamIds.length} valid teams`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
|
||||
// Remove orphaned monitors (monitors without a valid team)
|
||||
const deletedMonitorCount = await this.monitorsRepository.deleteByTeamIdsNotIn(validTeamIds);
|
||||
if (deletedMonitorCount > 0) {
|
||||
this.logger.info({
|
||||
message: `Deleted ${deletedMonitorCount} orphaned monitors`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
}
|
||||
|
||||
// Remove orphaned monitorStats (stats without a valid monitor)
|
||||
const allMonitorIds = await this.monitorsRepository.findAllMonitorIds();
|
||||
this.logger.debug({
|
||||
message: `Found ${allMonitorIds.length} valid monitors`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
|
||||
const deletedStatsCount = await this.monitorStatsRepository.deleteByMonitorIdsNotIn(allMonitorIds);
|
||||
if (deletedStatsCount > 0) {
|
||||
this.logger.info({
|
||||
message: `Deleted ${deletedStatsCount} orphaned monitor stats`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
}
|
||||
|
||||
// Remove orphaned checks
|
||||
const deletedChecksCount = await this.checksRepository.deleteByMonitorIdsNotIn(allMonitorIds);
|
||||
if (deletedChecksCount > 0) {
|
||||
this.logger.info({
|
||||
message: `Deleted ${deletedChecksCount} orphaned checks`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
}
|
||||
|
||||
// Remove orphaned incidents
|
||||
const deletedIncidentsCount = await this.incidentsRepository.deleteByMonitorIdsNotIn(allMonitorIds);
|
||||
if (deletedIncidentsCount > 0) {
|
||||
this.logger.info({
|
||||
message: `Deleted ${deletedIncidentsCount} orphaned incidents`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
}
|
||||
|
||||
// Remove orphaned geo checks
|
||||
const deletedGeoChecksCount = await this.geoChecksRepository.deleteByMonitorIdsNotIn(allMonitorIds);
|
||||
if (deletedGeoChecksCount > 0) {
|
||||
this.logger.info({
|
||||
message: `Deleted ${deletedGeoChecksCount} orphaned geo checks`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
}
|
||||
|
||||
this.logger.info({
|
||||
message: "Cleanup of orphaned data completed",
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
});
|
||||
} catch (error: unknown) {
|
||||
this.logger.warn({
|
||||
message: error instanceof Error ? error.message : "Unknown error",
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupOrphanedJob",
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
getHeartbeatGeoJob = () => {
|
||||
return async (monitor: Monitor) => {
|
||||
try {
|
||||
const monitorId = monitor.id;
|
||||
const teamId = monitor.teamId;
|
||||
|
||||
// Step 1: Validate monitor eligibility
|
||||
if (!monitorId) {
|
||||
throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getHeartbeatGeoJob" });
|
||||
}
|
||||
|
||||
if (!monitor.geoCheckEnabled) {
|
||||
return;
|
||||
}
|
||||
if (!supportsGeoCheck(monitor.type)) {
|
||||
this.logger.debug({
|
||||
message: `Monitor ${monitorId} type does not support geo checks, skipping`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getHeartbeatGeoJob",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (!monitor.geoCheckLocations || monitor.geoCheckLocations.length === 0) {
|
||||
this.logger.warn({
|
||||
message: `No geo check locations configured for monitor ${monitorId}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getHeartbeatGeoJob",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 2: Check for maintenance window
|
||||
const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId, teamId);
|
||||
if (maintenanceWindowActive) {
|
||||
this.logger.debug({
|
||||
message: `Monitor ${monitorId} is in maintenance window, skipping geo check`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getHeartbeatGeoJob",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 3: Build geo check (handles API calls and polling)
|
||||
const geoCheck = await this.geoChecksService.buildGeoCheck(monitor);
|
||||
if (!geoCheck) {
|
||||
this.logger.warn({
|
||||
message: `No geo check could be built for monitor ${monitorId}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getHeartbeatGeoJob",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 4: Add geo check to buffer
|
||||
this.buffer.addGeoCheckToBuffer(geoCheck);
|
||||
|
||||
this.logger.debug({
|
||||
message: `Geo check job executed for monitor ${monitorId}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getHeartbeatGeoJob",
|
||||
});
|
||||
} catch (error: unknown) {
|
||||
this.logger.error({
|
||||
message: error instanceof Error ? error.message : "Unknown error",
|
||||
service: SERVICE_NAME,
|
||||
method: "getHeartbeatGeoJob",
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
// Don't throw - geo check failures shouldn't crash the job scheduler
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
async isInMaintenanceWindow(monitorId: string, teamId: string) {
|
||||
const maintenanceWindows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId);
|
||||
const now = new Date();
|
||||
return maintenanceWindows.some((window) => isWindowActive(window, now));
|
||||
}
|
||||
|
||||
getCleanupRetentionJob = () => {
|
||||
return async () => {
|
||||
try {
|
||||
const settings = await this.settingsService.getDBSettings();
|
||||
|
||||
const checkTTL = settings.checkTTL; // Check TTL is in DAYS, not MS
|
||||
|
||||
if (checkTTL === CHECK_TTL_SENTINEL) {
|
||||
this.logger.info({
|
||||
message: `Check TTL is set to unlimited, skipping cleanup`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupRetentionJob",
|
||||
});
|
||||
return;
|
||||
}
|
||||
const checkTTLInMs = checkTTL * 24 * 60 * 60 * 1000;
|
||||
const cutoffDate = new Date(Date.now() - checkTTLInMs);
|
||||
const deleteCount = await this.checkService.deleteOlderThan(cutoffDate);
|
||||
this.logger.info({
|
||||
message: `Deleted ${deleteCount} checks older than ${cutoffDate.toISOString()}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupRetentionJob",
|
||||
});
|
||||
} catch (error: unknown) {
|
||||
this.logger.error({
|
||||
message: error instanceof Error ? error.message : "Unknown error",
|
||||
service: SERVICE_NAME,
|
||||
method: "getCleanupRetentionJob",
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
private evaluateMonitorAction(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;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user