Merge pull request #1695 from bluewave-labs/feat/be/distributed-uptime
feat: be/distributed uptime
This commit is contained in:
8 files changed
+571
-52
No files matched your search
@@ -1,16 +1,213 @@
|
||||
import { handleError } from "./controllerUtils.js";
|
||||
import { successMessages } from "../utils/messages.js";
|
||||
import Monitor from "../db/models/Monitor.js";
|
||||
import DistributedUptimeCheck from "../db/models/DistributedUptimeCheck.js";
|
||||
|
||||
const SERVICE_NAME = "DistributedUptimeQueueController";
|
||||
|
||||
class DistributedUptimeController {
|
||||
constructor() {}
|
||||
constructor(db, http, statusService) {
|
||||
this.db = db;
|
||||
this.http = http;
|
||||
this.statusService = statusService;
|
||||
this.resultsCallback = this.resultsCallback.bind(this);
|
||||
this.getDistributedUptimeMonitors = this.getDistributedUptimeMonitors.bind(this);
|
||||
this.getDistributedUptimeMonitorDetails =
|
||||
this.getDistributedUptimeMonitorDetails.bind(this);
|
||||
}
|
||||
|
||||
async resultsCallback(req, res, next) {
|
||||
try {
|
||||
const { id, result } = req.body;
|
||||
// Calculate response time
|
||||
const {
|
||||
first_byte_took,
|
||||
body_read_took,
|
||||
dns_took,
|
||||
conn_took,
|
||||
connect_took,
|
||||
tls_took,
|
||||
status_code,
|
||||
error,
|
||||
} = result;
|
||||
|
||||
// Calculate response time
|
||||
const responseTime =
|
||||
(first_byte_took +
|
||||
body_read_took +
|
||||
dns_took +
|
||||
conn_took +
|
||||
connect_took +
|
||||
tls_took) /
|
||||
1_000_000;
|
||||
|
||||
// Calculate if server is up or down
|
||||
const isErrorStatus = status_code >= 400;
|
||||
const hasError = error !== "";
|
||||
|
||||
const status = isErrorStatus || hasError ? false : true;
|
||||
|
||||
// Build response
|
||||
const distributedUptimeResponse = {
|
||||
monitorId: id,
|
||||
type: "distributed_http",
|
||||
payload: result,
|
||||
status,
|
||||
code: status_code,
|
||||
responseTime,
|
||||
};
|
||||
|
||||
if (error) {
|
||||
const code = status_code || this.NETWORK_ERROR;
|
||||
distributedUptimeResponse.code = code;
|
||||
distributedUptimeResponse.message =
|
||||
this.http.STATUS_CODES[code] || "Network Error";
|
||||
} else {
|
||||
distributedUptimeResponse.message = this.http.STATUS_CODES[status_code];
|
||||
}
|
||||
|
||||
await this.statusService.updateStatus(distributedUptimeResponse);
|
||||
|
||||
res.status(200).json({ message: "OK" });
|
||||
} catch (error) {
|
||||
throw handleError(error, SERVICE_NAME, "resultsCallback");
|
||||
next(handleError(error, SERVICE_NAME, "resultsCallback"));
|
||||
}
|
||||
}
|
||||
|
||||
async getDistributedUptimeMonitors(req, res, next) {
|
||||
try {
|
||||
res.setHeader("Content-Type", "text/event-stream");
|
||||
res.setHeader("Cache-Control", "no-cache");
|
||||
res.setHeader("Connection", "keep-alive");
|
||||
res.setHeader("Access-Control-Allow-Origin", "*");
|
||||
|
||||
const BATCH_DELAY = 1000;
|
||||
let batchTimeout = null;
|
||||
let opInProgress = false;
|
||||
|
||||
// Do things here
|
||||
const notifyChange = async () => {
|
||||
if (opInProgress) {
|
||||
// Get data
|
||||
const monitors = await this.db.getMonitorsByTeamId(req);
|
||||
res.write(`data: ${JSON.stringify({ monitors })}\n\n`);
|
||||
opInProgress = false;
|
||||
}
|
||||
batchTimeout = null;
|
||||
};
|
||||
|
||||
const handleChange = () => {
|
||||
opInProgress = true;
|
||||
if (batchTimeout) clearTimeout(batchTimeout);
|
||||
batchTimeout = setTimeout(notifyChange, BATCH_DELAY);
|
||||
};
|
||||
|
||||
const monitorStream = Monitor.watch(
|
||||
[{ $match: { operationType: { $in: ["insert", "update", "delete"] } } }],
|
||||
{ fullDocument: "updateLookup" }
|
||||
);
|
||||
|
||||
const checksStream = DistributedUptimeCheck.watch(
|
||||
[{ $match: { operationType: { $in: ["insert", "update", "delete"] } } }],
|
||||
{ fullDocument: "updateLookup" }
|
||||
);
|
||||
|
||||
monitorStream.on("change", handleChange);
|
||||
checksStream.on("change", handleChange);
|
||||
|
||||
// Send initial data
|
||||
const monitors = await this.db.getMonitorsByTeamId(req);
|
||||
res.write(`data: ${JSON.stringify({ monitors })}\n\n`);
|
||||
|
||||
// Handle client disconnect
|
||||
req.on("close", () => {
|
||||
if (batchTimeout) {
|
||||
clearTimeout(batchTimeout);
|
||||
}
|
||||
monitorStream.close();
|
||||
checksStream.close();
|
||||
clearInterval(keepAlive);
|
||||
});
|
||||
|
||||
// Keep connection alive
|
||||
const keepAlive = setInterval(() => {
|
||||
res.write(": keepalive\n\n");
|
||||
}, 30000);
|
||||
|
||||
// Clean up on close
|
||||
req.on("close", () => {
|
||||
clearInterval(keepAlive);
|
||||
});
|
||||
} catch (error) {
|
||||
next(handleError(error, SERVICE_NAME, "getDistributedUptimeMonitors"));
|
||||
}
|
||||
}
|
||||
|
||||
async getDistributedUptimeMonitorDetails(req, res, next) {
|
||||
try {
|
||||
res.setHeader("Content-Type", "text/event-stream");
|
||||
res.setHeader("Cache-Control", "no-cache");
|
||||
res.setHeader("Connection", "keep-alive");
|
||||
res.setHeader("Access-Control-Allow-Origin", "*");
|
||||
|
||||
const BATCH_DELAY = 1000;
|
||||
let batchTimeout = null;
|
||||
let opInProgress = false;
|
||||
|
||||
// Do things here
|
||||
const notifyChange = async () => {
|
||||
try {
|
||||
if (opInProgress) {
|
||||
// Get data
|
||||
const monitor = await this.db.getDistributedUptimeDetailsById(req);
|
||||
res.write(`data: ${JSON.stringify({ monitor })}\n\n`);
|
||||
opInProgress = false;
|
||||
}
|
||||
batchTimeout = null;
|
||||
} catch (error) {
|
||||
console.error("Error in notifyChange:", error);
|
||||
opInProgress = false;
|
||||
batchTimeout = null;
|
||||
next(handleError(error, SERVICE_NAME, "getDistributedUptimeMonitorDetails"));
|
||||
}
|
||||
};
|
||||
|
||||
const handleChange = () => {
|
||||
opInProgress = true;
|
||||
if (batchTimeout) clearTimeout(batchTimeout);
|
||||
batchTimeout = setTimeout(notifyChange, BATCH_DELAY);
|
||||
};
|
||||
|
||||
const checksStream = DistributedUptimeCheck.watch(
|
||||
[{ $match: { operationType: { $in: ["insert", "update", "delete"] } } }],
|
||||
{ fullDocument: "updateLookup" }
|
||||
);
|
||||
|
||||
checksStream.on("change", handleChange);
|
||||
|
||||
// Send initial data
|
||||
const monitor = await this.db.getDistributedUptimeDetailsById(req);
|
||||
res.write(`data: ${JSON.stringify({ monitor })}\n\n`);
|
||||
|
||||
// Handle client disconnect
|
||||
req.on("close", () => {
|
||||
if (batchTimeout) {
|
||||
clearTimeout(batchTimeout);
|
||||
}
|
||||
checksStream.close();
|
||||
clearInterval(keepAlive);
|
||||
});
|
||||
|
||||
// Keep connection alive
|
||||
const keepAlive = setInterval(() => {
|
||||
res.write(": keepalive\n\n");
|
||||
}, 30000);
|
||||
|
||||
// Clean up on close
|
||||
req.on("close", () => {
|
||||
clearInterval(keepAlive);
|
||||
});
|
||||
} catch (error) {
|
||||
next(handleError(error, SERVICE_NAME, "getDistributedUptimeMonitorDetails"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,13 +1,97 @@
|
||||
import mongoose from "mongoose";
|
||||
|
||||
import { BaseCheckSchema } from "./Check.js";
|
||||
|
||||
// {
|
||||
// "id": "12123",
|
||||
// "result": {
|
||||
// "task_arrived": "2025-01-13T19:21:37.463466602Z",
|
||||
// "dns_start": "2025-01-14T00:21:33.1801319+05:00",
|
||||
// "dns_end": "2025-01-14T00:21:33.4582552+05:00",
|
||||
// "conn_start": "2025-01-14T00:21:33.1801319+05:00",
|
||||
// "conn_end": "2025-01-14T00:21:33.7076318+05:00",
|
||||
// "connect_start": "2025-01-14T00:21:33.4582552+05:00",
|
||||
// "connect_end": "2025-01-14T00:21:33.541899+05:00",
|
||||
// "tls_hand_shake_start": "2025-01-14T00:21:33.541899+05:00",
|
||||
// "tls_hand_shake_end": "2025-01-14T00:21:33.7076318+05:00",
|
||||
// "body_read_start": "2025-01-14T00:21:34.1894707+05:00",
|
||||
// "body_read_end": "2025-01-14T00:21:34.1894707+05:00",
|
||||
// "wrote_request": "2025-01-14T00:21:33.7076318+05:00",
|
||||
// "got_first_response_byte": "2025-01-14T00:21:34.1327652+05:00",
|
||||
// "first_byte_took": 425133400,
|
||||
// "body_read_took": 56030000,
|
||||
// "dns_took": 278123300,
|
||||
// "conn_took": 527499900,
|
||||
// "connect_took": 83643800,
|
||||
// "tls_took": 165732800,
|
||||
// "sni_name": "uprock.com",
|
||||
// "status_code": 200,
|
||||
// "body_size": 19320,
|
||||
// "request_header_size": 95,
|
||||
// "response_header_size": 246,
|
||||
// "response_headers": "X-Vercel-Id: bom1::iad1::sm87v-1736796096856-aec270c01f23\nDate: Mon, 13 Jan 2025 19:21:37 GMT\nServer: Vercel\nStrict-Transport-Security: max-age=63072000\nVary: RSC, Next-Router-State-Tree, Next-Router-Prefetch, Next-Url\nX-Matched-Path: /\nX-Powered-By: Next.js\nX-Vercel-Cache: MISS\nAge: 0\nCache-Control: private, no-cache, no-store, max-age=0, must-revalidate\nContent-Type: text/html; charset=utf-8",
|
||||
// "error": "",
|
||||
// "device_id": "d5f578e143a2cd603dd6bf5f846a86a538bde4a8fbe2ad1fca284ad9f033daf8",
|
||||
// "ip_address": "223.123.19.0",
|
||||
// "proof": "",
|
||||
// "created_at": "2025-01-13T19:21:37.463466912Z",
|
||||
// "continent": "AS",
|
||||
// "country_code": "PK",
|
||||
// "city": "Muzaffargarh",
|
||||
// "upt_burnt" : "0.01",
|
||||
// "location": {
|
||||
// "lat": 71.0968,
|
||||
// "lng": 30.0208
|
||||
// },
|
||||
// "payload": {
|
||||
// "callback": "https://webhook.site/2a15b0af-545a-4ac2-b913-153b97592d7a",
|
||||
// "x": "y"
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
const LocationSchema = new mongoose.Schema(
|
||||
{
|
||||
lat: { type: Number, required: true },
|
||||
lng: { type: Number, required: true },
|
||||
},
|
||||
{ _id: false }
|
||||
);
|
||||
|
||||
const DistributedUptimeCheckSchema = mongoose.Schema(
|
||||
{
|
||||
...BaseCheckSchema.obj,
|
||||
location: {
|
||||
type: LocationSchema,
|
||||
required: false,
|
||||
},
|
||||
continent: {
|
||||
type: String,
|
||||
required: false,
|
||||
},
|
||||
countryCode: {
|
||||
type: String,
|
||||
required: false,
|
||||
},
|
||||
city: {
|
||||
type: String,
|
||||
required: false,
|
||||
},
|
||||
uptBurnt: {
|
||||
type: mongoose.Schema.Types.Decimal128,
|
||||
required: false,
|
||||
},
|
||||
},
|
||||
{ timestamps: true }
|
||||
);
|
||||
|
||||
DistributedUptimeCheckSchema.pre("save", function (next) {
|
||||
if (this.isModified("uptBurnt") && typeof this.uptBurnt === "string") {
|
||||
this.uptBurnt = mongoose.Types.Decimal128.fromString(this.uptBurnt);
|
||||
}
|
||||
next();
|
||||
});
|
||||
|
||||
DistributedUptimeCheckSchema.index({ createdAt: 1 });
|
||||
DistributedUptimeCheckSchema.index({ monitorId: 1, createdAt: 1 });
|
||||
DistributedUptimeCheckSchema.index({ monitorId: 1, createdAt: -1 });
|
||||
|
||||
@@ -2,6 +2,7 @@ import Monitor from "../../models/Monitor.js";
|
||||
import Check from "../../models/Check.js";
|
||||
import PageSpeedCheck from "../../models/PageSpeedCheck.js";
|
||||
import HardwareCheck from "../../models/HardwareCheck.js";
|
||||
import DistributedUptimeCheck from "../../models/DistributedUptimeCheck.js";
|
||||
import { errorMessages } from "../../../utils/messages.js";
|
||||
import Notification from "../../models/Notification.js";
|
||||
import { NormalizeData, NormalizeDataUptimeDetails } from "../../../utils/dataUtils.js";
|
||||
@@ -11,6 +12,7 @@ import { fileURLToPath } from "url";
|
||||
import {
|
||||
buildUptimeDetailsPipeline,
|
||||
buildHardwareDetailsPipeline,
|
||||
buildDistributedUptimeDetailsPipeline,
|
||||
} from "./monitorModuleQueries.js";
|
||||
import { ObjectId } from "mongodb";
|
||||
const __filename = fileURLToPath(import.meta.url);
|
||||
@@ -366,6 +368,48 @@ const getUptimeDetailsById = async (req) => {
|
||||
}
|
||||
};
|
||||
|
||||
const getDistributedUptimeDetailsById = async (req) => {
|
||||
try {
|
||||
const { monitorId } = req.params;
|
||||
const monitor = await Monitor.findById(monitorId);
|
||||
if (monitor === null || monitor === undefined) {
|
||||
throw new Error(errorMessages.DB_FIND_MONITOR_BY_ID(monitorId));
|
||||
}
|
||||
|
||||
const { dateRange, normalize } = req.query;
|
||||
const dates = getDateRange(dateRange);
|
||||
const formatLookup = {
|
||||
day: "%Y-%m-%dT%H:%M:00Z",
|
||||
week: "%Y-%m-%dT%H:00:00Z",
|
||||
month: "%Y-%m-%dT00:00:00Z",
|
||||
};
|
||||
|
||||
const dateString = formatLookup[dateRange];
|
||||
const results = await DistributedUptimeCheck.aggregate(
|
||||
buildDistributedUptimeDetailsPipeline(monitor, dates, dateString)
|
||||
);
|
||||
|
||||
const monitorData = results[0];
|
||||
const normalizedGroupChecks = NormalizeDataUptimeDetails(
|
||||
monitorData.groupedChecks,
|
||||
10,
|
||||
100
|
||||
);
|
||||
|
||||
const monitorStats = {
|
||||
...monitor.toObject(),
|
||||
...monitorData,
|
||||
groupedChecks: normalizedGroupChecks,
|
||||
};
|
||||
|
||||
return monitorStats;
|
||||
} catch (error) {
|
||||
error.service = SERVICE_NAME;
|
||||
error.method = "getUptimeDetailsById";
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Get stats by monitor ID
|
||||
* @async
|
||||
@@ -815,6 +859,7 @@ export {
|
||||
getMonitorById,
|
||||
getMonitorsByTeamId,
|
||||
getUptimeDetailsById,
|
||||
getDistributedUptimeDetailsById,
|
||||
createMonitor,
|
||||
deleteMonitor,
|
||||
deleteAllMonitors,
|
||||
|
||||
@@ -534,4 +534,188 @@ const buildHardwareDetailsPipeline = (monitor, dates, dateString) => {
|
||||
];
|
||||
};
|
||||
|
||||
export { buildUptimeDetailsPipeline, buildHardwareDetailsPipeline };
|
||||
const buildDistributedUptimeDetailsPipeline = (monitor, dates, dateString) => {
|
||||
return [
|
||||
{
|
||||
$match: {
|
||||
monitorId: monitor._id,
|
||||
},
|
||||
},
|
||||
{
|
||||
$sort: {
|
||||
createdAt: 1,
|
||||
},
|
||||
},
|
||||
{
|
||||
$facet: {
|
||||
aggregateData: [
|
||||
{
|
||||
$group: {
|
||||
_id: null,
|
||||
avgResponseTime: {
|
||||
$avg: "$responseTime",
|
||||
},
|
||||
lastCheck: {
|
||||
$last: "$$ROOT",
|
||||
},
|
||||
totalChecks: {
|
||||
$sum: 1,
|
||||
},
|
||||
uptBurnt: {
|
||||
$sum: "$uptBurnt",
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
// For the response time chart, should return checks for date window
|
||||
// Grouped by: {day: hour}, {week: day}, {month: day}
|
||||
groupedMapChecks: [
|
||||
{
|
||||
$match: {
|
||||
createdAt: { $gte: dates.start, $lte: dates.end },
|
||||
},
|
||||
},
|
||||
{
|
||||
$group: {
|
||||
_id: {
|
||||
date: {
|
||||
$dateToString: {
|
||||
format: dateString,
|
||||
date: "$createdAt",
|
||||
},
|
||||
},
|
||||
city: "$city",
|
||||
lat: "$location.lat",
|
||||
lng: "$location.lng",
|
||||
},
|
||||
city: { $first: "$city" }, // Add this line to include city in output
|
||||
lat: { $first: "$location.lat" },
|
||||
lng: { $first: "$location.lng" },
|
||||
avgResponseTime: {
|
||||
$avg: "$responseTime",
|
||||
},
|
||||
totalChecks: {
|
||||
$sum: 1,
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
$sort: {
|
||||
"_id.date": 1,
|
||||
},
|
||||
},
|
||||
],
|
||||
groupedChecks: [
|
||||
{
|
||||
$match: {
|
||||
createdAt: { $gte: dates.start, $lte: dates.end },
|
||||
},
|
||||
},
|
||||
{
|
||||
$group: {
|
||||
_id: {
|
||||
$dateToString: {
|
||||
format: dateString,
|
||||
date: "$createdAt",
|
||||
},
|
||||
},
|
||||
avgResponseTime: {
|
||||
$avg: "$responseTime",
|
||||
},
|
||||
totalChecks: {
|
||||
$sum: 1,
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
$sort: {
|
||||
_id: 1,
|
||||
},
|
||||
},
|
||||
],
|
||||
// Average response time for the date window
|
||||
groupAvgResponseTime: [
|
||||
{
|
||||
$match: {
|
||||
createdAt: { $gte: dates.start, $lte: dates.end },
|
||||
},
|
||||
},
|
||||
{
|
||||
$group: {
|
||||
_id: null,
|
||||
avgResponseTime: {
|
||||
$avg: "$responseTime",
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
latestChecks: [
|
||||
{
|
||||
$sort: { createdAt: -1 }, // Sort by newest first
|
||||
},
|
||||
{
|
||||
$limit: 5, // Get only the first 5 documents
|
||||
},
|
||||
{
|
||||
$project: {
|
||||
responseTime: 1,
|
||||
city: 1,
|
||||
countryCode: 1,
|
||||
uptBurnt: { $toString: "$uptBurnt" },
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
{
|
||||
$project: {
|
||||
totalUptBurnt: {
|
||||
$toString: {
|
||||
$arrayElemAt: ["$aggregateData.uptBurnt", 0],
|
||||
},
|
||||
},
|
||||
avgResponseTime: {
|
||||
$arrayElemAt: ["$aggregateData.avgResponseTime", 0],
|
||||
},
|
||||
totalChecks: {
|
||||
$arrayElemAt: ["$aggregateData.totalChecks", 0],
|
||||
},
|
||||
latestResponseTime: {
|
||||
$arrayElemAt: ["$aggregateData.lastCheck.responseTime", 0],
|
||||
},
|
||||
timeSinceLastCheck: {
|
||||
$let: {
|
||||
vars: {
|
||||
lastCheck: {
|
||||
$arrayElemAt: ["$aggregateData.lastCheck", 0],
|
||||
},
|
||||
},
|
||||
in: {
|
||||
$cond: [
|
||||
{
|
||||
$ifNull: ["$$lastCheck", false],
|
||||
},
|
||||
{
|
||||
$subtract: [new Date(), "$$lastCheck.createdAt"],
|
||||
},
|
||||
0,
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
groupedMapChecks: "$groupedMapChecks",
|
||||
groupedChecks: "$groupedChecks",
|
||||
groupedAvgResponseTime: {
|
||||
$arrayElemAt: ["$groupAvgResponseTime", 0],
|
||||
},
|
||||
latestChecks: "$latestChecks",
|
||||
},
|
||||
},
|
||||
];
|
||||
};
|
||||
|
||||
export {
|
||||
buildUptimeDetailsPipeline,
|
||||
buildHardwareDetailsPipeline,
|
||||
buildDistributedUptimeDetailsPipeline,
|
||||
};
|
||||
@@ -8,6 +8,14 @@ class DistributedUptimeRoutes {
|
||||
}
|
||||
initRoutes() {
|
||||
this.router.post("/callback", this.distributedUptimeController.resultsCallback);
|
||||
this.router.get(
|
||||
"/monitors/:teamId",
|
||||
this.distributedUptimeController.getDistributedUptimeMonitors
|
||||
);
|
||||
this.router.get(
|
||||
"/monitors/details/:monitorId",
|
||||
this.distributedUptimeController.getDistributedUptimeMonitorDetails
|
||||
);
|
||||
}
|
||||
|
||||
getRouter() {
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
const QUEUE_NAMES = ["uptime", "pagespeed", "hardware"];
|
||||
const QUEUE_NAMES = ["uptime", "pagespeed", "hardware", "distributed"];
|
||||
const SERVICE_NAME = "JobQueue";
|
||||
const JOBS_PER_WORKER = 5;
|
||||
const QUEUE_LOOKUP = {
|
||||
@@ -8,6 +8,7 @@ const QUEUE_LOOKUP = {
|
||||
port: "uptime",
|
||||
docker: "uptime",
|
||||
pagespeed: "pagespeed",
|
||||
distributed_http: "distributed",
|
||||
};
|
||||
const getSchedulerId = (monitor) => `scheduler:${monitor.type}:${monitor._id}`;
|
||||
|
||||
@@ -163,6 +164,9 @@ class NewJobQueue {
|
||||
// Get the current status
|
||||
await job.updateProgress(30);
|
||||
const networkResponse = await this.networkService.getStatus(job);
|
||||
if (job.data.type === "distributed_http") {
|
||||
return;
|
||||
}
|
||||
// Handle status change
|
||||
await job.updateProgress(60);
|
||||
const { monitor, statusChanged, prevStatus } =
|
||||
@@ -219,51 +223,6 @@ class NewJobQueue {
|
||||
connection: this.connection,
|
||||
concurrency: 5,
|
||||
});
|
||||
|
||||
// worker.on("active", (job) => {
|
||||
// this.logger.info({
|
||||
// message: `Worker started processing job: ${job.id}`,
|
||||
// service: SERVICE_NAME,
|
||||
// method: "createWorker",
|
||||
// });
|
||||
// });
|
||||
|
||||
// worker.on("completed", (job) => {
|
||||
// this.logger.info({
|
||||
// message: `Worker completed job: ${job.id}`,
|
||||
// service: SERVICE_NAME,
|
||||
// method: "createWorker",
|
||||
// });
|
||||
// });
|
||||
|
||||
// // Log job progress updates
|
||||
// worker.on("progress", (job, progress) => {
|
||||
// this.logger.info({
|
||||
// message: `Job progress: ${job.id}`,
|
||||
// service: SERVICE_NAME,
|
||||
// method: "createWorker",
|
||||
// details: `Progress: ${progress}%`,
|
||||
// });
|
||||
// });
|
||||
|
||||
// // Log when a job fails
|
||||
// worker.on("failed", (job, err) => {
|
||||
// this.logger.error({
|
||||
// message: `Worker failed job: ${job.id}`,
|
||||
// service: SERVICE_NAME,
|
||||
// method: "createWorker",
|
||||
// details: `Error: ${err.message}`,
|
||||
// stack: err.stack,
|
||||
// });
|
||||
// });
|
||||
|
||||
// worker.on("stalled", (jobId) => {
|
||||
// this.logger.warn({
|
||||
// message: `Worker stalled job: ${jobId}`,
|
||||
// service: SERVICE_NAME,
|
||||
// method: "createWorker",
|
||||
// });
|
||||
// });
|
||||
return worker;
|
||||
}
|
||||
|
||||
@@ -392,7 +351,6 @@ class NewJobQueue {
|
||||
const jobs = await queue.getJobs();
|
||||
const ret = await Promise.all(
|
||||
jobs.map(async (job) => {
|
||||
console.log(job);
|
||||
const state = await job.getState();
|
||||
return { url: job.data.url, state, progress: job.progress };
|
||||
})
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { errorMessages, successMessages } from "../utils/messages.js";
|
||||
const SERVICE_NAME = "NetworkService";
|
||||
const UPROCK_ENDPOINT = "https://api.uprock.com/checkmate/push";
|
||||
|
||||
/**
|
||||
* Constructs a new NetworkService instance.
|
||||
@@ -19,6 +20,7 @@ class NetworkService {
|
||||
this.TYPE_HARDWARE = "hardware";
|
||||
this.TYPE_DOCKER = "docker";
|
||||
this.TYPE_PORT = "port";
|
||||
this.TYPE_DISTRIBUTED_HTTP = "distributed_http";
|
||||
this.SERVICE_NAME = SERVICE_NAME;
|
||||
this.NETWORK_ERROR = 5000;
|
||||
this.PING_ERROR = 5001;
|
||||
@@ -323,6 +325,35 @@ class NetworkService {
|
||||
}
|
||||
}
|
||||
|
||||
async requestDistributedHttp(job) {
|
||||
try {
|
||||
const monitor = job.data;
|
||||
const CALLBACK_URL = process.env.NGROK_URL
|
||||
? process.env.NGROK_URL
|
||||
: process.env.CALLBACK_URL;
|
||||
|
||||
const response = await this.axios.post(
|
||||
UPROCK_ENDPOINT,
|
||||
{
|
||||
id: monitor._id,
|
||||
url: monitor.url,
|
||||
callback: `${CALLBACK_URL}/api/v1/distributed-uptime/callback`,
|
||||
},
|
||||
{
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
"x-checkmate-key": process.env.UPROCK_API_KEY,
|
||||
},
|
||||
}
|
||||
);
|
||||
} catch (error) {
|
||||
console.log(error.message);
|
||||
error.service = this.SERVICE_NAME;
|
||||
error.method = "requestDistributedHttp";
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles unsupported job types by throwing an error with details.
|
||||
*
|
||||
@@ -360,6 +391,8 @@ class NetworkService {
|
||||
return await this.requestDocker(job);
|
||||
case this.TYPE_PORT:
|
||||
return await this.requestPort(job);
|
||||
case this.TYPE_DISTRIBUTED_HTTP:
|
||||
return await this.requestDistributedHttp(job);
|
||||
|
||||
default:
|
||||
return this.handleUnsupportedType(type);
|
||||
|
||||
@@ -25,7 +25,7 @@ class StatusService {
|
||||
* @param {Object} networkResponse - The network response containing monitorId and status.
|
||||
* @param {string} networkResponse.monitorId - The ID of the monitor.
|
||||
* @param {string} networkResponse.status - The new status of the monitor.
|
||||
* @returns {Promise<Object>} - A promise that resolves to an object containing the monitor, statusChanged flag, and previous status if the status changed, or false if an error occurred.
|
||||
* @returns {Promise<Object>} - A promise that resolves to an object containinfg the monitor, statusChanged flag, and previous status if the status changed, or false if an error occurred.
|
||||
* @returns {Promise<Object>} returnObject - The object returned by the function.
|
||||
* @returns {Object} returnObject.monitor - The monitor object.
|
||||
* @returns {boolean} returnObject.statusChanged - Flag indicating if the status has changed.
|
||||
@@ -85,6 +85,7 @@ class StatusService {
|
||||
buildCheck = (networkResponse) => {
|
||||
const { monitorId, teamId, type, status, responseTime, code, message, payload } =
|
||||
networkResponse;
|
||||
|
||||
const check = {
|
||||
monitorId,
|
||||
teamId,
|
||||
@@ -94,6 +95,14 @@ class StatusService {
|
||||
message,
|
||||
};
|
||||
|
||||
if (type === "distributed_http") {
|
||||
check.continent = payload.continent;
|
||||
check.countryCode = payload.country_code;
|
||||
check.city = payload.city;
|
||||
check.location = payload.location;
|
||||
check.uptBurnt = payload.upt_burnt;
|
||||
}
|
||||
|
||||
if (type === "pagespeed") {
|
||||
const categories = payload.lighthouseResult?.categories;
|
||||
const audits = payload.lighthouseResult?.audits;
|
||||
@@ -145,6 +154,7 @@ class StatusService {
|
||||
hardware: this.db.createHardwareCheck,
|
||||
docker: this.db.createCheck,
|
||||
port: this.db.createCheck,
|
||||
distributed_http: this.db.createDistributedCheck,
|
||||
};
|
||||
const operation = operationMap[networkResponse.type];
|
||||
|
||||
|
||||
Reference in new issue
Block a user