Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/openstatus. 🫖 Status page with uptime monitoring & API monitoring as code 🫖 openstatus.dev
bun drizzle-orm monitoring monitoring-as-code nextjs observability on-call open-source shadcn-ui status-page statuspage synthetic-monitoring tinybird turso uptime uptime-checker uptime-monitor
Something went wrong. Try again.
13 kB · 460 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461import { CloudTasksClient } from "@google-cloud/tasks";import type { google } from "@google-cloud/tasks/build/protos";import { getLogger } from "@logtape/logtape";import { and, eq, gte, inArray, isNotNull, lte, notInArray,} from "@openstatus/db";import { type MonitorStatus, maintenance, monitor, monitorStatusTable, selectMonitorSchema, selectMonitorStatusSchema,} from "@openstatus/db/src/schema";import type { Region } from "@openstatus/db/src/schema/constants";import type { monitorPeriodicitySchema } from "@openstatus/db/src/schema/constants";import { maintenancesToPageComponents, pageComponent,} from "@openstatus/db/src/schema/page_components";import { regionDict } from "@openstatus/regions";import { type DNSPayloadSchema, type httpPayloadSchema, type grpcPayloadSchema, type icmpPayloadSchema, type tpcPayloadSchema, transformHeaders,} from "@openstatus/utils";import { Effect, Result, Schedule } from "effect";import { z } from "zod";
import { env } from "../env";import { db } from "../lib/db";
type TaskInput = { row: z.infer<typeof selectMonitorSchema>; timestamp: number; status: MonitorStatus; region: Region;};
export const isAuthorizedDomain = (url: string) => { return url.includes(env().SITE_URL);};
const logger = getLogger("workflow");
/** * Check if GCP Cloud Tasks is properly configured. * Returns false if credentials are missing or set to placeholder values. */function isGcpConfigured(): boolean { const projectId = env().GCP_PROJECT_ID; const location = env().GCP_LOCATION; const clientEmail = env().GCP_CLIENT_EMAIL; const privateKey = env().GCP_PRIVATE_KEY;
return ( projectId !== "" && projectId !== "your-value" && location !== "" && location !== "your-value" && clientEmail !== "" && clientEmail !== "your-value" && privateKey !== "" && privateKey !== "your-value" );}
// Only initialize CloudTasksClient if GCP credentials are properly configured// This allows self-hosted deployments to run without GCP Cloud Taskslet client: CloudTasksClient | null = null;
if (isGcpConfigured()) { client = new CloudTasksClient({ // fallback: true, projectId: env().GCP_PROJECT_ID, credentials: { client_email: env().GCP_CLIENT_EMAIL, private_key: env().GCP_PRIVATE_KEY.replaceAll("\\n", "\n"), }, });}
export async function sendCheckerTasks( periodicity: z.infer<typeof monitorPeriodicitySchema>,): Promise<{ success: number; failed: number }> { // If GCP Cloud Tasks is not configured, skip task creation if (!client) { logger.warn("GCP Cloud Tasks not configured - skipping checker tasks", { periodicity, }); return { success: 0, failed: 0 }; }
const parent = client.queuePath( env().GCP_PROJECT_ID, env().GCP_LOCATION, periodicity, );
const timestamp = Date.now();
const currentMaintenance = db .select({ id: maintenance.id }) .from(maintenance) .where( and(lte(maintenance.from, new Date()), gte(maintenance.to, new Date())), ) .as("currentMaintenance");
const currentMaintenanceMonitors = db .select({ id: pageComponent.monitorId }) .from(maintenancesToPageComponents) .innerJoin( currentMaintenance, eq(maintenancesToPageComponents.maintenanceId, currentMaintenance.id), ) .innerJoin( pageComponent, eq(maintenancesToPageComponents.pageComponentId, pageComponent.id), ) .where(isNotNull(pageComponent.monitorId));
const result = await db .select() .from(monitor) .where( and( eq(monitor.periodicity, periodicity), eq(monitor.active, true), notInArray(monitor.id, currentMaintenanceMonitors), ), ) .all();
logger.info("Starting cron job", { periodicity, monitor_count: result.length, });
const monitors = z.array(selectMonitorSchema).safeParse(result); const taskInputs: TaskInput[] = []; if (!monitors.success) { logger.error(`Error while fetching the monitors ${monitors.error}`); throw new Error("Error while fetching the monitors"); }
if (monitors.data.length === 0) { logger.info("No monitors to check", { periodicity }); return { success: 0, failed: 0 }; }
// Batch fetch all monitor statuses in a single query (N+1 fix) const monitorIds = monitors.data.map((m) => m.id); const rawStatuses = await db .select() .from(monitorStatusTable) .where(inArray(monitorStatusTable.monitorId, monitorIds)) .all();
const statusMap = new Map< number, z.infer<typeof selectMonitorStatusSchema>[] >(); for (const raw of rawStatuses) { const parsed = selectMonitorStatusSchema.safeParse(raw); if (!parsed.success) { logger.error("Failed to parse monitor status row", { monitor_id: raw.monitorId, error_message: parsed.error.message, }); continue; } const list = statusMap.get(raw.monitorId) ?? []; list.push(parsed.data); statusMap.set(raw.monitorId, list); }
for (const row of monitors.data) { const monitorStatuses = statusMap.get(row.id) ?? [];
for (const region of row.regions) { const status = monitorStatuses.find((m) => region === m.region)?.status || "active";
const r = regionDict[region as keyof typeof regionDict];
if (!r) { logger.error(`Invalid region ${region}`); continue; } if (r.deprecated) { // Let's uncomment this when we are ready to remove deprecated regions // We should not use deprecated regions anymore logger.error(`Deprecated region ${region}`); continue; } taskInputs.push({ row, timestamp, status, region }); if (periodicity === "30s") { const scheduledAt = timestamp + 30 * 1000; taskInputs.push({ row, timestamp: scheduledAt, status, region, }); } } }
const results = await Effect.runPromise( Effect.forEach( taskInputs, (input) => Effect.tryPromise({ try: () => createCronTask(input, parent), catch: (err) => { if (err instanceof Error && "code" in err && err.code === 6) { return "ALREADY_EXISTS" as const; } return new Error( `Failed creating task for monitor ${input.row.id} in region ${input.region}`, ); }, }).pipe( Effect.catchIf( (err): err is "ALREADY_EXISTS" => err === "ALREADY_EXISTS", () => Effect.void, ), Effect.retry({ times: 3, schedule: Schedule.exponential("1000 millis"), }), Effect.result, ), { concurrency: 100 }, ), );
for (const result of results) { if (Result.isFailure(result)) { logger.error("Task creation failed after retries", { error_message: result.failure.message, }); } }
const success = results.filter(Result.isSuccess).length; const failed = results.filter(Result.isFailure).length;
logger.info("Completed cron job", { periodicity, total_tasks: taskInputs.length, success_count: success, failed_count: failed, duration_ms: Date.now() - timestamp, }); if (failed > 0) { logger.error("Cron job had failures", { periodicity, failed_count: failed, success_count: success, }); }
return { success, failed };}// timestamp needs to be in msconst createCronTask = async ( { row, timestamp, status, region }: TaskInput, parent: string,) => { // This should never happen due to the guard in sendCheckerTasks, but TypeScript needs this if (!client) { throw new Error("GCP Cloud Tasks client is not initialized"); }
let payload: | z.infer<typeof httpPayloadSchema> | z.infer<typeof tpcPayloadSchema> | z.infer<typeof DNSPayloadSchema> | z.infer<typeof icmpPayloadSchema> | z.infer<typeof grpcPayloadSchema> | null = null;
// if (row.jobType === "http") { payload = { workspaceId: String(row.workspaceId), monitorId: String(row.id), url: row.url, method: row.method || "GET", cronTimestamp: timestamp, body: row.body, headers: row.headers, status: status, assertions: row.assertions ? JSON.parse(row.assertions) : null, degradedAfter: row.degradedAfter, timeout: row.timeout, trigger: "cron", otelConfig: row.otelEndpoint ? { endpoint: row.otelEndpoint, headers: transformHeaders(row.otelHeaders), } : undefined, retry: row.retry || 3, followRedirects: row.followRedirects === null ? true : row.followRedirects, }; } if (row.jobType === "tcp") { payload = { workspaceId: String(row.workspaceId), monitorId: String(row.id), uri: row.url, status: status, assertions: row.assertions ? JSON.parse(row.assertions) : null, cronTimestamp: timestamp, degradedAfter: row.degradedAfter, timeout: row.timeout, trigger: "cron", retry: row.retry || 3, otelConfig: row.otelEndpoint ? { endpoint: row.otelEndpoint, headers: transformHeaders(row.otelHeaders), } : undefined, }; } if (row.jobType === "dns") { payload = { workspaceId: String(row.workspaceId), monitorId: String(row.id), uri: row.url, cronTimestamp: timestamp, status: status, assertions: row.assertions ? JSON.parse(row.assertions) : null, degradedAfter: row.degradedAfter, timeout: row.timeout, trigger: "cron", otelConfig: row.otelEndpoint ? { endpoint: row.otelEndpoint, headers: transformHeaders(row.otelHeaders), } : undefined, retry: row.retry || 3, }; }
if (row.jobType === "icmp") { payload = { workspaceId: String(row.workspaceId), monitorId: String(row.id), uri: row.url, cronTimestamp: timestamp, status: status, degradedAfter: row.degradedAfter, timeout: row.timeout, trigger: "cron", otelConfig: row.otelEndpoint ? { endpoint: row.otelEndpoint, headers: transformHeaders(row.otelHeaders), } : undefined, retry: row.retry || 3, }; }
if (row.jobType === "grpc") { payload = { workspaceId: String(row.workspaceId), monitorId: String(row.id), uri: row.url, service: row.grpcService ?? undefined, tls: row.grpcTls ?? "tls", headers: transformHeaders(row.headers), cronTimestamp: timestamp, status: status, degradedAfter: row.degradedAfter, timeout: row.timeout, trigger: "cron", otelConfig: row.otelEndpoint ? { endpoint: row.otelEndpoint, headers: transformHeaders(row.otelHeaders), } : undefined, retry: row.retry || 3, }; }
if (!payload) { throw new Error("Invalid jobType"); } const regionInfo = regionDict[region]; let regionHeader = {}; if (regionInfo.provider === "fly") { regionHeader = { "fly-prefer-region": region }; } if (regionInfo.provider === "koyeb") { regionHeader = { "X-KOYEB-REGION-OVERRIDE": region.replace("koyeb_", "") }; } if (regionInfo.provider === "railway") { regionHeader = { "railway-region": region.replace("railway_", "") }; } const taskName = `${parent}/tasks/monitor-${row.id}-${region}-${timestamp}`; const newTask: google.cloud.tasks.v2beta3.ITask = { name: taskName, httpRequest: { headers: { "Content-Type": "application/json", // Set content type to ensure compatibility your application's request parsing ...regionHeader, Authorization: `Basic ${env().CRON_SECRET}`, }, httpMethod: "POST", url: generateUrl({ row, region }), body: Buffer.from(JSON.stringify(payload)).toString("base64"), }, scheduleTime: { seconds: timestamp / 1000, }, };
const request = { parent: parent, task: newTask }; return client.createTask(request);};
function generateUrl({ row, region,}: { row: z.infer<typeof selectMonitorSchema>; region: Region;}) { const regionInfo = regionDict[region];
switch (regionInfo.provider) { case "fly": return `https://openstatus-checker.fly.dev/checker/${row.jobType}?monitor_id=${row.id}`; case "koyeb": return `https://openstatus-checker.koyeb.app/checker/${row.jobType}?monitor_id=${row.id}`; case "railway": return `https://railway-proxy-production-9cb1.up.railway.app/checker/${row.jobType}?monitor_id=${row.id}`;
default: throw new Error("Invalid jobType"); }}