diff --git a/server/src/db/models/index.ts b/server/src/db/models/index.ts index 17c4a33bf..8d679edaa 100644 --- a/server/src/db/models/index.ts +++ b/server/src/db/models/index.ts @@ -59,3 +59,8 @@ export type { IIncident, ResolutionType } from "./incidents/Incident.js"; export { SystemSettings } from "./system/SystemSettings.js"; export type { ISystemSettings } from "./system/SystemSettings.js"; + +export { StatsHourly } from "./monitors/StatsHourly.js"; +export type { IStatsHourly } from "./monitors/StatsHourly.js"; +export { StatsDaily } from "./monitors/StatsDaily.js"; +export type { IStatsDaily } from "./monitors/StatsDaily.js"; diff --git a/server/src/db/models/monitors/Monitor.ts b/server/src/db/models/monitors/Monitor.ts index badbae6dc..f18de6d58 100644 --- a/server/src/db/models/monitors/Monitor.ts +++ b/server/src/db/models/monitors/Monitor.ts @@ -1,5 +1,11 @@ import mongoose, { Schema, Document, Types } from "mongoose"; -import { Check, Incident, MonitorStats } from "@/db/models/index.js"; +import { + Check, + Incident, + MonitorStats, + StatsDaily, + StatsHourly, +} from "@/db/models/index.js"; export const MonitorTypes = [ "http", @@ -120,6 +126,8 @@ MonitorSchema.pre( const monitorId = this._id; await Check.deleteMany({ "metadata.monitorId": monitorId }); await MonitorStats.deleteMany({ monitorId }); + await StatsHourly.deleteMany({ monitorId }); + await StatsDaily.deleteMany({ monitorId }); await Incident.deleteMany({ monitorId }); next(); } catch (error: any) { @@ -139,6 +147,8 @@ MonitorSchema.pre( if (monitorIds.length > 0) { await Check.deleteMany({ "metadata.monitorId": { $in: monitorIds } }); await MonitorStats.deleteMany({ monitorId: { $in: monitorIds } }); + await StatsHourly.deleteMany({ monitorId: { $in: monitorIds } }); + await StatsDaily.deleteMany({ monitorId: { $in: monitorIds } }); await Incident.deleteMany({ monitorId: { $in: monitorIds } }); } next(); diff --git a/server/src/db/models/monitors/StatsDaily.ts b/server/src/db/models/monitors/StatsDaily.ts new file mode 100644 index 000000000..6a382f076 --- /dev/null +++ b/server/src/db/models/monitors/StatsDaily.ts @@ -0,0 +1,132 @@ +import mongoose, { Schema, Types } from "mongoose"; +import type { MonitorType } from "@/db/models/monitors/Monitor.js"; +import type { + IInfraCpu, + IInfraDiskEntry, + IInfraHost, + IInfraMemory, + IInfraNetEntry, + IPageSpeedMetrics, + IStatsBase, +} from "./StatsHourly.js"; + +export interface IStatsDaily + extends mongoose.Document, + IStatsBase, + IPageSpeedMetrics { + cpu?: IInfraCpu; + memory?: IInfraMemory; + disk?: IInfraDiskEntry[]; + host?: IInfraHost; + net?: IInfraNetEntry[]; +} + +const InfraCpuSchema = new Schema( + { + physical_core: { type: Number }, + logical_core: { type: Number }, + frequency: { type: Number }, + current_frequency: { type: Number }, + temperature: { type: [Number], default: undefined }, + free_percent: { type: Number }, + used_percent: { type: Number }, + }, + { _id: false } +); + +const InfraMemorySchema = new Schema( + { + total_bytes: { type: Number }, + available_bytes: { type: Number }, + used_bytes: { type: Number }, + usage_percent: { type: Number }, + }, + { _id: false } +); + +const InfraDiskEntrySchema = new Schema( + { + device: { type: String }, + total_bytes: { type: Number }, + free_bytes: { type: Number }, + used_bytes: { type: Number }, + usage_percent: { type: Number }, + total_inodes: { type: Number }, + free_inodes: { type: Number }, + used_inodes: { type: Number }, + inodes_usage_percent: { type: Number }, + read_bytes: { type: Number }, + write_bytes: { type: Number }, + read_time: { type: Number }, + write_time: { type: Number }, + }, + { _id: false } +); + +const InfraHostSchema = new Schema( + { + os: { type: String }, + platform: { type: String }, + kernel_version: { type: String }, + pretty_name: { type: String }, + }, + { _id: false } +); + +const InfraNetEntrySchema = new Schema( + { + name: { type: String }, + bytes_sent: { type: Number }, + bytes_recv: { type: Number }, + packets_sent: { type: Number }, + packets_recv: { type: Number }, + err_in: { type: Number }, + err_out: { type: Number }, + drop_in: { type: Number }, + drop_out: { type: Number }, + fifo_in: { type: Number }, + fifo_out: { type: Number }, + }, + { _id: false } +); + +const StatsDailySchema = new Schema( + { + monitorId: { type: Schema.Types.ObjectId, ref: "Monitor", index: true, required: true }, + teamId: { type: Schema.Types.ObjectId, ref: "Team", index: true, required: true }, + type: { type: String, required: true }, + windowStart: { type: Date, required: true, index: true }, + windowEnd: { type: Date, required: true }, + finalized: { type: Boolean, default: false }, + count: { type: Number, required: true, default: 0 }, + avgResponseTime: { type: Number, required: true, default: 0 }, + + // Pagespeed + accessibility: { type: Number }, + bestPractices: { type: Number }, + seo: { type: Number }, + performance: { type: Number }, + cls: { type: Number }, + si: { type: Number }, + fcp: { type: Number }, + lcp: { type: Number }, + tbt: { type: Number }, + + // Infrastructure + cpu: { type: InfraCpuSchema, required: false }, + memory: { type: InfraMemorySchema, required: false }, + disk: { type: [InfraDiskEntrySchema], required: false, default: undefined }, + host: { type: InfraHostSchema, required: false }, + net: { type: [InfraNetEntrySchema], required: false, default: undefined }, + }, + { timestamps: true, collection: "stats_daily" } +); + +// One document per (monitorId, windowStart) +StatsDailySchema.index({ monitorId: 1, windowStart: 1 }, { unique: true }); + +export const StatsDaily = mongoose.model( + "StatsDaily", + StatsDailySchema +); + diff --git a/server/src/db/models/monitors/StatsHourly.ts b/server/src/db/models/monitors/StatsHourly.ts new file mode 100644 index 000000000..3ee6ca685 --- /dev/null +++ b/server/src/db/models/monitors/StatsHourly.ts @@ -0,0 +1,202 @@ +import mongoose, { Schema, Types } from "mongoose"; +import type { MonitorType } from "@/db/models/monitors/Monitor.js"; + +// Common/base metrics across monitor types +export interface IStatsBase { + monitorId: Types.ObjectId; + teamId: Types.ObjectId; + type: MonitorType; + windowStart: Date; // inclusive + windowEnd: Date; // exclusive + finalized: boolean; + count: number; // number of checks in window + avgResponseTime: number; // ms + createdAt: Date; + updatedAt: Date; +} + +// Pagespeed rollups (averages per window) +export interface IPageSpeedMetrics { + accessibility?: number; + bestPractices?: number; + seo?: number; + performance?: number; + cls?: number; + si?: number; + fcp?: number; + lcp?: number; + tbt?: number; +} + +// Infrastructure rollups +export interface IInfraCpu { + physical_core?: number; + logical_core?: number; + frequency?: number; + current_frequency?: number; + temperature?: number[]; // averaged per sensor index + free_percent?: number; + used_percent?: number; +} + +export interface IInfraMemory { + total_bytes?: number; + available_bytes?: number; + used_bytes?: number; + usage_percent?: number; +} + +export interface IInfraDiskEntry { + device?: string; + total_bytes?: number; + free_bytes?: number; + used_bytes?: number; + usage_percent?: number; + total_inodes?: number; + free_inodes?: number; + used_inodes?: number; + inodes_usage_percent?: number; + read_bytes?: number; + write_bytes?: number; + read_time?: number; + write_time?: number; +} + +export interface IInfraHost { + os?: string; + platform?: string; + kernel_version?: string; + pretty_name?: string; +} + +export interface IInfraNetEntry { + name?: string; + bytes_sent?: number; + bytes_recv?: number; + packets_sent?: number; + packets_recv?: number; + err_in?: number; + err_out?: number; + drop_in?: number; + drop_out?: number; + fifo_in?: number; + fifo_out?: number; +} + +export interface IStatsHourly extends mongoose.Document, IStatsBase, IPageSpeedMetrics { + cpu?: IInfraCpu; + memory?: IInfraMemory; + disk?: IInfraDiskEntry[]; + host?: IInfraHost; + net?: IInfraNetEntry[]; +} + +const InfraCpuSchema = new Schema( + { + physical_core: { type: Number }, + logical_core: { type: Number }, + frequency: { type: Number }, + current_frequency: { type: Number }, + temperature: { type: [Number], default: undefined }, + free_percent: { type: Number }, + used_percent: { type: Number }, + }, + { _id: false } +); + +const InfraMemorySchema = new Schema( + { + total_bytes: { type: Number }, + available_bytes: { type: Number }, + used_bytes: { type: Number }, + usage_percent: { type: Number }, + }, + { _id: false } +); + +const InfraDiskEntrySchema = new Schema( + { + device: { type: String }, + total_bytes: { type: Number }, + free_bytes: { type: Number }, + used_bytes: { type: Number }, + usage_percent: { type: Number }, + total_inodes: { type: Number }, + free_inodes: { type: Number }, + used_inodes: { type: Number }, + inodes_usage_percent: { type: Number }, + read_bytes: { type: Number }, + write_bytes: { type: Number }, + read_time: { type: Number }, + write_time: { type: Number }, + }, + { _id: false } +); + +const InfraHostSchema = new Schema( + { + os: { type: String }, + platform: { type: String }, + kernel_version: { type: String }, + pretty_name: { type: String }, + }, + { _id: false } +); + +const InfraNetEntrySchema = new Schema( + { + name: { type: String }, + bytes_sent: { type: Number }, + bytes_recv: { type: Number }, + packets_sent: { type: Number }, + packets_recv: { type: Number }, + err_in: { type: Number }, + err_out: { type: Number }, + drop_in: { type: Number }, + drop_out: { type: Number }, + fifo_in: { type: Number }, + fifo_out: { type: Number }, + }, + { _id: false } +); + +const StatsHourlySchema = new Schema( + { + monitorId: { type: Schema.Types.ObjectId, ref: "Monitor", index: true, required: true }, + teamId: { type: Schema.Types.ObjectId, ref: "Team", index: true, required: true }, + type: { type: String, required: true }, + windowStart: { type: Date, required: true, index: true }, + windowEnd: { type: Date, required: true }, + finalized: { type: Boolean, default: false }, + count: { type: Number, required: true, default: 0 }, + avgResponseTime: { type: Number, required: true, default: 0 }, + + // Pagespeed + accessibility: { type: Number }, + bestPractices: { type: Number }, + seo: { type: Number }, + performance: { type: Number }, + cls: { type: Number }, + si: { type: Number }, + fcp: { type: Number }, + lcp: { type: Number }, + tbt: { type: Number }, + + // Infrastructure + cpu: { type: InfraCpuSchema, required: false }, + memory: { type: InfraMemorySchema, required: false }, + disk: { type: [InfraDiskEntrySchema], required: false, default: undefined }, + host: { type: InfraHostSchema, required: false }, + net: { type: [InfraNetEntrySchema], required: false, default: undefined }, + }, + { timestamps: true, collection: "stats_hourly" } +); + +// One document per (monitorId, windowStart) +StatsHourlySchema.index({ monitorId: 1, windowStart: 1 }, { unique: true }); + +export const StatsHourly = mongoose.model( + "StatsHourly", + StatsHourlySchema +); + diff --git a/server/src/init/services.ts b/server/src/init/services.ts index 0d54a1eaa..866f08696 100644 --- a/server/src/init/services.ts +++ b/server/src/init/services.ts @@ -29,6 +29,7 @@ import { SettingsService, StripeService, BillingService, + StatsAggregationService, } from "@/services/index.js"; import { EntitlementsFactory } from "@/services/system/EntitlementsService.js"; import { MeService } from "@/services/index.js"; @@ -55,6 +56,7 @@ export const initServices = async () => { settingsService ); const incidentService = new IncidentService(); + const statsAggregationService = new StatsAggregationService(); const jobGenerator = new JobGenerator( networkService, checkService, @@ -62,7 +64,8 @@ export const initServices = async () => { statusService, notificationService, incidentService, - maintenanceService + maintenanceService, + statsAggregationService ); const jobQueue = await JobQueue.create(jobGenerator); const entitlementsProvider = EntitlementsFactory.create(); diff --git a/server/src/services/business/MonitorService.ts b/server/src/services/business/MonitorService.ts index c4a41c2c3..0ed063826 100644 --- a/server/src/services/business/MonitorService.ts +++ b/server/src/services/business/MonitorService.ts @@ -1,6 +1,7 @@ import mongoose from "mongoose"; import { IMonitor, Monitor, MonitorStats, Check } from "@/db/models/index.js"; +import { StatsHourly, StatsDaily } from "@/db/models/index.js"; import ApiError from "@/utils/ApiError.js"; import { IJobQueue } from "@/services/infrastructure/JobQueue.js"; import { MonitorWithChecksResponse } from "@/types/index.js"; @@ -255,19 +256,7 @@ class MonitorService implements IMonitorService { } } - private getDateFormat(range: string): string { - switch (range) { - case "2h": - return "%Y-%m-%dT%H:%M:00Z"; - case "24h": - case "7d": - return "%Y-%m-%dT%H:00:00Z"; - case "30d": - return "%Y-%m-%d"; - default: - throw new ApiError("Invalid range parameter", 400); - } - } + // getDateFormat removed — only 2h (raw checks) remains, using minute buckets inline private getBaseGroup = (dateFormat: string): Record => { return { @@ -527,9 +516,66 @@ class MonitorService implements IMonitorService { } const startDate = this.getStartDate(range); - const dateFormat = this.getDateFormat(range); - // Build match stage + // Route for aggregated stats (24h, 7d, 30d) + if (range === "24h" || range === "7d" || range === "30d") { + const monitorStats = await MonitorStats.findOne({ monitorId: monitor._id }); + if (!monitorStats) { + throw new ApiError("Monitor stats not found", 404); + } + + const useDaily = range === "30d"; + const Model = useDaily ? StatsDaily : StatsHourly; + const endDate = new Date(); + + const docs = await Model.find({ + monitorId: monitor._id, + windowStart: { $gte: startDate, $lt: endDate }, + }) + .sort({ windowStart: -1 }) + .lean(); + + const checks = (docs as any[]).map((d) => { + const base: any = { + _id: new Date(d.windowStart).toISOString(), + count: d.count ?? 0, + avgResponseTime: d.avgResponseTime ?? 0, + }; + if (monitor.type === "pagespeed") { + return { + ...base, + accessibility: d.accessibility, + seo: d.seo, + bestPractices: d.bestPractices, + performance: d.performance, + cls: d.cls, + si: d.si, + fcp: d.fcp, + lcp: d.lcp, + tbt: d.tbt, + }; + } + if (monitor.type === "infrastructure") { + return { + ...base, + cpu: d.cpu, + memory: d.memory, + disk: d.disk, + host: d.host, + net: d.net, + }; + } + return base; + }); + + return { + monitor: monitor.toObject(), + checks, + stats: monitorStats, + }; + } + + // Route for raw checks (2h) const matchStage: { "metadata.monitorId": mongoose.Types.ObjectId; createdAt: { $gte: Date }; @@ -543,8 +589,8 @@ class MonitorService implements IMonitorService { matchStage.status = status; } + const dateFormat = "%Y-%m-%dT%H:%M:00Z"; let groupClause; - if (monitor.type === "pagespeed") { groupClause = this.getPageSpeedGroup(dateFormat); } else if (monitor.type === "infrastructure") { diff --git a/server/src/services/business/StatsAggregationService.ts b/server/src/services/business/StatsAggregationService.ts new file mode 100644 index 000000000..cfa3a9e32 --- /dev/null +++ b/server/src/services/business/StatsAggregationService.ts @@ -0,0 +1,639 @@ +import mongoose from "mongoose"; +import { Check } from "@/db/models/index.js"; +import { StatsHourly } from "@/db/models/index.js"; +import { StatsDaily } from "@/db/models/index.js"; +import type { MonitorType } from "@/db/models/monitors/Monitor.js"; + +export interface IStatsAggregationService { + upsertHourly: (windowStart: Date, windowEnd: Date) => Promise; + rollupDaily: ( + dayStart: Date, + dayEnd: Date, + finalize?: boolean + ) => Promise; +} + +const BASE_TYPES: MonitorType[] = ["http", "https", "port", "ping"]; + +class StatsAggregationService implements IStatsAggregationService { + upsertHourly = async (windowStart: Date, windowEnd: Date): Promise => { + const matchWindow = { createdAt: { $gte: windowStart, $lt: windowEnd } }; + + // 1) Base types (http/https/port/ping) + const baseAgg = await Check.aggregate<{ + _id: mongoose.Types.ObjectId; + teamId: mongoose.Types.ObjectId; + type: MonitorType; + count: number; + avgResponseTime: number; + }>([ + { $match: { ...matchWindow, "metadata.type": { $in: BASE_TYPES } } }, + { + $group: { + _id: "$metadata.monitorId", + teamId: { $first: "$metadata.teamId" }, + type: { $first: "$metadata.type" }, + count: { $sum: 1 }, + avgResponseTime: { $avg: "$responseTime" }, + }, + }, + ]); + + // 2) Pagespeed + const pagespeedAgg = await Check.aggregate([ + { $match: { ...matchWindow, "metadata.type": "pagespeed" } }, + { + $project: { + responseTime: 1, + createdAt: 1, + "lighthouse.accessibility": 1, + "lighthouse.bestPractices": 1, + "lighthouse.seo": 1, + "lighthouse.performance": 1, + "lighthouse.audits.cls.score": 1, + "lighthouse.audits.si.score": 1, + "lighthouse.audits.fcp.score": 1, + "lighthouse.audits.lcp.score": 1, + "lighthouse.audits.tbt.score": 1, + "metadata.monitorId": 1, + "metadata.teamId": 1, + "metadata.type": 1, + }, + }, + { + $group: { + _id: "$metadata.monitorId", + teamId: { $first: "$metadata.teamId" }, + type: { $first: "$metadata.type" }, + count: { $sum: 1 }, + avgResponseTime: { $avg: "$responseTime" }, + accessibility: { $avg: "$lighthouse.accessibility" }, + bestPractices: { $avg: "$lighthouse.bestPractices" }, + seo: { $avg: "$lighthouse.seo" }, + performance: { $avg: "$lighthouse.performance" }, + cls: { $avg: "$lighthouse.audits.cls.score" }, + si: { $avg: "$lighthouse.audits.si.score" }, + fcp: { $avg: "$lighthouse.audits.fcp.score" }, + lcp: { $avg: "$lighthouse.audits.lcp.score" }, + tbt: { $avg: "$lighthouse.audits.tbt.score" }, + }, + }, + ]); + + // 3) Infrastructure + const infraAgg = await Check.aggregate([ + { $match: { ...matchWindow, "metadata.type": "infrastructure" } }, + { + $project: { + responseTime: 1, + createdAt: 1, + "system.cpu.physical_core": 1, + "system.cpu.logical_core": 1, + "system.cpu.frequency": 1, + "system.cpu.current_frequency": 1, + "system.cpu.temperature": 1, + "system.cpu.free_percent": 1, + "system.cpu.usage_percent": 1, + "system.memory.total_bytes": 1, + "system.memory.available_bytes": 1, + "system.memory.used_bytes": 1, + "system.memory.usage_percent": 1, + "system.disk": 1, + "system.host.os": 1, + "system.host.platform": 1, + "system.host.kernel_version": 1, + "system.host.pretty_name": 1, + "system.net": 1, + "metadata.monitorId": 1, + "metadata.teamId": 1, + "metadata.type": 1, + }, + }, + { + $group: { + _id: "$metadata.monitorId", + teamId: { $first: "$metadata.teamId" }, + type: { $first: "$metadata.type" }, + count: { $sum: 1 }, + avgResponseTime: { $avg: "$responseTime" }, + physicalCores: { $last: "$system.cpu.physical_core" }, + logicalCores: { $last: "$system.cpu.logical_core" }, + frequency: { $avg: "$system.cpu.frequency" }, + currentFrequency: { $last: "$system.cpu.current_frequency" }, + tempsArrays: { $push: "$system.cpu.temperature" }, + freePercent: { $avg: "$system.cpu.free_percent" }, + usedPercent: { $avg: "$system.cpu.usage_percent" }, + total_bytes: { $last: "$system.memory.total_bytes" }, + available_bytes: { $last: "$system.memory.available_bytes" }, + used_bytes: { $last: "$system.memory.used_bytes" }, + memory_usage_percent: { $avg: "$system.memory.usage_percent" }, + disksArray: { $push: "$system.disk" }, + os: { $last: "$system.host.os" }, + platform: { $last: "$system.host.platform" }, + kernel_version: { $last: "$system.host.kernel_version" }, + pretty_name: { $last: "$system.host.pretty_name" }, + netsArray: { $push: "$system.net" }, + }, + }, + { + $project: { + _id: 1, + teamId: 1, + type: 1, + count: 1, + avgResponseTime: 1, + cpu: { + physical_core: "$physicalCores", + logical_core: "$logicalCores", + frequency: "$frequency", + current_frequency: "$currentFrequency", + temperature: { + $map: { + input: { + $range: [ + 0, + { + $size: { + $ifNull: [{ $arrayElemAt: ["$tempsArrays", 0] }, []], + }, + }, + ], + }, + as: "idx", + in: { + $avg: { + $map: { + input: { $ifNull: ["$tempsArrays", []] }, + as: "arr", + in: { $arrayElemAt: ["$$arr", "$$idx"] }, + }, + }, + }, + }, + }, + free_percent: "$freePercent", + used_percent: "$usedPercent", + }, + memory: { + total_bytes: "$total_bytes", + available_bytes: "$available_bytes", + used_bytes: "$used_bytes", + usage_percent: "$memory_usage_percent", + }, + disk: { + $map: { + input: { + $range: [ + 0, + { + $size: { + $ifNull: [{ $arrayElemAt: ["$disksArray", 0] }, []], + }, + }, + ], + }, + as: "idx", + in: { + $let: { + vars: { + diskGroup: { + $map: { + input: { $ifNull: ["$disksArray", []] }, + as: "diskArr", + in: { $arrayElemAt: ["$$diskArr", "$$idx"] }, + }, + }, + }, + in: { + device: { $arrayElemAt: ["$$diskGroup.device", 0] }, + total_bytes: { $avg: "$$diskGroup.total_bytes" }, + free_bytes: { $avg: "$$diskGroup.free_bytes" }, + used_bytes: { $avg: "$$diskGroup.used_bytes" }, + usage_percent: { $avg: "$$diskGroup.usage_percent" }, + total_inodes: { $avg: "$$diskGroup.total_inodes" }, + free_inodes: { $avg: "$$diskGroup.free_inodes" }, + used_inodes: { $avg: "$$diskGroup.used_inodes" }, + inodes_usage_percent: { $avg: "$$diskGroup.inodes_usage_percent" }, + read_bytes: { $avg: "$$diskGroup.read_bytes" }, + write_bytes: { $avg: "$$diskGroup.write_bytes" }, + read_time: { $avg: "$$diskGroup.read_time" }, + write_time: { $avg: "$$diskGroup.write_time" }, + }, + }, + }, + }, + }, + host: { + os: "$os", + platform: "$platform", + kernel_version: "$kernel_version", + pretty_name: "$pretty_name", + }, + net: { + $map: { + input: { + $range: [ + 0, + { + $size: { $ifNull: [{ $arrayElemAt: ["$netsArray", 0] }, []] }, + }, + ], + }, + as: "idx", + in: { + $let: { + vars: { + netGroup: { + $map: { + input: { $ifNull: ["$netsArray", []] }, + as: "netArr", + in: { $arrayElemAt: ["$$netArr", "$$idx"] }, + }, + }, + }, + in: { + name: { $arrayElemAt: ["$$netGroup.name", 0] }, + bytes_sent: { $avg: "$$netGroup.bytes_sent" }, + bytes_recv: { $avg: "$$netGroup.bytes_recv" }, + packets_sent: { $avg: "$$netGroup.packets_sent" }, + packets_recv: { $avg: "$$netGroup.packets_recv" }, + err_in: { $avg: "$$netGroup.err_in" }, + err_out: { $avg: "$$netGroup.err_out" }, + drop_in: { $avg: "$$netGroup.drop_in" }, + drop_out: { $avg: "$$netGroup.drop_out" }, + fifo_in: { $avg: "$$netGroup.fifo_in" }, + fifo_out: { $avg: "$$netGroup.fifo_out" }, + }, + }, + }, + }, + }, + }, + }, + ]); + + let upserts = 0; + + const bulkOps: any[] = []; + const addUpsert = (d: any) => { + bulkOps.push({ + updateOne: { + filter: { monitorId: d._id, windowStart }, + update: { + $setOnInsert: { + monitorId: d._id, + teamId: d.teamId, + type: d.type, + windowStart, + windowEnd, + }, + $set: { + finalized: false, + count: d.count ?? 0, + avgResponseTime: d.avgResponseTime ?? 0, + accessibility: d.accessibility, + bestPractices: d.bestPractices, + seo: d.seo, + performance: d.performance, + cls: d.cls, + si: d.si, + fcp: d.fcp, + lcp: d.lcp, + tbt: d.tbt, + cpu: d.cpu, + memory: d.memory, + disk: d.disk, + host: d.host, + net: d.net, + }, + }, + upsert: true, + }, + }); + }; + + baseAgg.forEach(addUpsert); + pagespeedAgg.forEach(addUpsert); + infraAgg.forEach(addUpsert); + + if (bulkOps.length > 0) { + const res = await StatsHourly.bulkWrite(bulkOps, { ordered: false }); + upserts += res.upsertedCount + (res.modifiedCount ?? 0); + } + return upserts; + }; + + // Placeholder: daily rollup will aggregate from StatsHourly across a day + rollupDaily = async ( + dayStart: Date, + dayEnd: Date, + finalize: boolean = true + ): Promise => { + // Fetch hourly docs for the day + const hourly = await StatsHourly.find({ + windowStart: { $gte: dayStart, $lt: dayEnd }, + }).lean(); + + type Weighted = { sum: number; w: number }; + const addW = (acc: Weighted, v?: number, w = 0) => { + if (typeof v === "number" && !Number.isNaN(v) && w > 0) { + acc.sum += v * w; + acc.w += w; + } + }; + const avg = (acc: Weighted) => (acc.w > 0 ? acc.sum / acc.w : undefined); + + interface DeviceAgg { + key: string; // device or name + fields: Record; + } + + interface GroupAgg { + monitorId: mongoose.Types.ObjectId; + teamId: mongoose.Types.ObjectId; + type: string; + // totals + count: number; + // base + rt: Weighted; + // pagespeed + accessibility: Weighted; + bestPractices: Weighted; + seo: Weighted; + performance: Weighted; + cls: Weighted; + si: Weighted; + fcp: Weighted; + lcp: Weighted; + tbt: Weighted; + // infra + cpu: { + lastPhysical?: number; + lastLogical?: number; + freq: Weighted; + lastCurrentFreq?: number; + freePct: Weighted; + usedPct: Weighted; + tempSums: number[]; + tempWeights: number[]; + }; + memory: { + lastTotal?: number; + lastAvail?: number; + lastUsed?: number; + usagePct: Weighted; + }; + disk: Map; + host: { os?: string; platform?: string; kernel_version?: string; pretty_name?: string }; + net: Map; + lastTs: number; // for choosing "last" fields + } + + const groups = new Map(); + + for (const h of hourly) { + const key = (h.monitorId as any).toString(); + let g = groups.get(key); + if (!g) { + g = { + monitorId: h.monitorId as any, + teamId: h.teamId as any, + type: h.type, + count: 0, + rt: { sum: 0, w: 0 }, + accessibility: { sum: 0, w: 0 }, + bestPractices: { sum: 0, w: 0 }, + seo: { sum: 0, w: 0 }, + performance: { sum: 0, w: 0 }, + cls: { sum: 0, w: 0 }, + si: { sum: 0, w: 0 }, + fcp: { sum: 0, w: 0 }, + lcp: { sum: 0, w: 0 }, + tbt: { sum: 0, w: 0 }, + cpu: { + freq: { sum: 0, w: 0 }, + freePct: { sum: 0, w: 0 }, + usedPct: { sum: 0, w: 0 }, + tempSums: [], + tempWeights: [], + }, + memory: { usagePct: { sum: 0, w: 0 } }, + disk: new Map(), + host: {}, + net: new Map(), + lastTs: 0, + }; + groups.set(key, g); + } + + const w = h.count ?? 0; + g.count += w; + addW(g.rt, h.avgResponseTime, w); + + // pagespeed + addW(g.accessibility, (h as any).accessibility, w); + addW(g.bestPractices, (h as any).bestPractices, w); + addW(g.seo, (h as any).seo, w); + addW(g.performance, (h as any).performance, w); + addW(g.cls, (h as any).cls, w); + addW(g.si, (h as any).si, w); + addW(g.fcp, (h as any).fcp, w); + addW(g.lcp, (h as any).lcp, w); + addW(g.tbt, (h as any).tbt, w); + + const ts = new Date(h.windowStart).getTime(); + if (ts >= g.lastTs) { + g.lastTs = ts; + // capture last-known fields + const cpu = (h as any).cpu || {}; + if (typeof cpu.physical_core === "number") g.cpu.lastPhysical = cpu.physical_core; + if (typeof cpu.logical_core === "number") g.cpu.lastLogical = cpu.logical_core; + if (typeof cpu.current_frequency === "number") g.cpu.lastCurrentFreq = cpu.current_frequency; + + const mem = (h as any).memory || {}; + if (typeof mem.total_bytes === "number") g.memory.lastTotal = mem.total_bytes; + if (typeof mem.available_bytes === "number") g.memory.lastAvail = mem.available_bytes; + if (typeof mem.used_bytes === "number") g.memory.lastUsed = mem.used_bytes; + + const host = (h as any).host || {}; + g.host = { + os: host.os, + platform: host.platform, + kernel_version: host.kernel_version, + pretty_name: host.pretty_name, + }; + } + + // weighted CPU averages + const cpu = (h as any).cpu || {}; + addW(g.cpu.freq, cpu.frequency, w); + addW(g.cpu.freePct, cpu.free_percent, w); + addW(g.cpu.usedPct, cpu.used_percent, w); + const temps: number[] | undefined = cpu.temperature; + if (Array.isArray(temps)) { + for (let i = 0; i < temps.length; i++) { + const v = temps[i]; + if (typeof v === "number") { + g.cpu.tempSums[i] = (g.cpu.tempSums[i] ?? 0) + v * w; + g.cpu.tempWeights[i] = (g.cpu.tempWeights[i] ?? 0) + w; + } + } + } + + // memory usage percent weighted + const mem = (h as any).memory || {}; + addW(g.memory.usagePct, mem.usage_percent, w); + + // disk merge by device + const disks: any[] | undefined = (h as any).disk; + if (Array.isArray(disks)) { + for (const d of disks) { + const keyD = d.device ?? ""; + if (!g.disk.has(keyD)) { + g.disk.set(keyD, { + key: keyD, + fields: { + total_bytes: { sum: 0, w: 0 }, + free_bytes: { sum: 0, w: 0 }, + used_bytes: { sum: 0, w: 0 }, + usage_percent: { sum: 0, w: 0 }, + total_inodes: { sum: 0, w: 0 }, + free_inodes: { sum: 0, w: 0 }, + used_inodes: { sum: 0, w: 0 }, + inodes_usage_percent: { sum: 0, w: 0 }, + read_bytes: { sum: 0, w: 0 }, + write_bytes: { sum: 0, w: 0 }, + read_time: { sum: 0, w: 0 }, + write_time: { sum: 0, w: 0 }, + }, + }); + } + const agg = g.disk.get(keyD)!; + for (const [f, acc] of Object.entries(agg.fields)) { + const val = (d as any)[f]; + if (typeof val === "number") addW(acc as Weighted, val, w); + } + } + } + + // net merge by name + const nets: any[] | undefined = (h as any).net; + if (Array.isArray(nets)) { + for (const n of nets) { + const keyN = n.name ?? ""; + if (!g.net.has(keyN)) { + g.net.set(keyN, { + key: keyN, + fields: { + bytes_sent: { sum: 0, w: 0 }, + bytes_recv: { sum: 0, w: 0 }, + packets_sent: { sum: 0, w: 0 }, + packets_recv: { sum: 0, w: 0 }, + err_in: { sum: 0, w: 0 }, + err_out: { sum: 0, w: 0 }, + drop_in: { sum: 0, w: 0 }, + drop_out: { sum: 0, w: 0 }, + fifo_in: { sum: 0, w: 0 }, + fifo_out: { sum: 0, w: 0 }, + }, + }); + } + const agg = g.net.get(keyN)!; + for (const [f, acc] of Object.entries(agg.fields)) { + const val = (n as any)[f]; + if (typeof val === "number") addW(acc as Weighted, val, w); + } + } + } + } + + // Prepare bulk upserts for daily + const bulkOps: any[] = []; + for (const g of groups.values()) { + const cpuTemps: number[] | undefined = g.cpu.tempSums.map((s, i) => { + const w = g.cpu.tempWeights[i] ?? 0; + return w > 0 ? s / w : undefined; + }).filter((v) => typeof v === "number") as number[]; + + const diskArr = Array.from(g.disk.values()).map((d) => { + const o: any = { device: d.key }; + for (const [f, acc] of Object.entries(d.fields)) { + const v = avg(acc as Weighted); + if (typeof v === "number") o[f] = v; + } + return o; + }); + + const netArr = Array.from(g.net.values()).map((n) => { + const o: any = { name: n.key }; + for (const [f, acc] of Object.entries(n.fields)) { + const v = avg(acc as Weighted); + if (typeof v === "number") o[f] = v; + } + return o; + }); + + const cpu: any = { + physical_core: g.cpu.lastPhysical, + logical_core: g.cpu.lastLogical, + frequency: avg(g.cpu.freq), + current_frequency: g.cpu.lastCurrentFreq, + free_percent: avg(g.cpu.freePct), + used_percent: avg(g.cpu.usedPct), + }; + if (cpuTemps.length > 0) cpu.temperature = cpuTemps; + + const memory: any = { + total_bytes: g.memory.lastTotal, + available_bytes: g.memory.lastAvail, + used_bytes: g.memory.lastUsed, + usage_percent: avg(g.memory.usagePct), + }; + + const updateDoc: any = { + $setOnInsert: { + monitorId: g.monitorId, + teamId: g.teamId, + type: g.type, + windowStart: dayStart, + windowEnd: dayEnd, + }, + $set: { + finalized: finalize, + count: g.count, + avgResponseTime: avg(g.rt) ?? 0, + accessibility: avg(g.accessibility), + bestPractices: avg(g.bestPractices), + seo: avg(g.seo), + performance: avg(g.performance), + cls: avg(g.cls), + si: avg(g.si), + fcp: avg(g.fcp), + lcp: avg(g.lcp), + tbt: avg(g.tbt), + cpu, + memory, + disk: diskArr.length > 0 ? diskArr : undefined, + host: g.host, + net: netArr.length > 0 ? netArr : undefined, + }, + }; + + bulkOps.push({ + updateOne: { + filter: { monitorId: g.monitorId, windowStart: dayStart }, + update: updateDoc, + upsert: true, + }, + }); + } + + if (bulkOps.length > 0) { + const res = await StatsDaily.bulkWrite(bulkOps, { ordered: false }); + return res.upsertedCount + (res.modifiedCount ?? 0); + } + return 0; + }; +} + +export default StatsAggregationService; diff --git a/server/src/services/index.ts b/server/src/services/index.ts index dea05c770..b7af253d8 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -15,6 +15,7 @@ export { default as DiagnosticService } from "@/services/business/DiagnosticServ export { default as RecoveryService } from "@/services/business/RecoveryService.js"; export { default as IncidentService } from "@/services/business/IncidentService.js"; export { default as MeService } from "@/services/business/MeService.js"; +export { default as StatsAggregationService } from "@/services/business/StatsAggregationService.js"; export { default as NetworkService } from "@/services/infrastructure/NetworkService.js"; export { default as NotificationService } from "@/services/infrastructure/NotificationService.js"; diff --git a/server/src/services/infrastructure/JobGenerator.ts b/server/src/services/infrastructure/JobGenerator.ts index 160722b0d..807c34da4 100644 --- a/server/src/services/infrastructure/JobGenerator.ts +++ b/server/src/services/infrastructure/JobGenerator.ts @@ -6,6 +6,10 @@ import { IStatusService } from "./StatusService.js"; import { INotificationService } from "./NotificationService.js"; import { IMaintenanceService } from "../business/MaintenanceService.js"; import { IIncidentService } from "../business/IncidentService.js"; +import StatsAggregationService, { + IStatsAggregationService, +} from "@/services/business/StatsAggregationService.js"; +import { StatsHourly } from "@/db/models/index.js"; import ApiError from "@/utils/ApiError.js"; import { getChildLogger } from "@/logger/Logger.js"; @@ -25,6 +29,7 @@ class JobGenerator implements IJobGenerator { private notificationService: INotificationService; private maintenanceService: IMaintenanceService; private incidentService: IIncidentService; + private statsAggregationService: IStatsAggregationService; constructor( networkService: INetworkService, @@ -33,7 +38,8 @@ class JobGenerator implements IJobGenerator { statusService: IStatusService, notificationService: INotificationService, incidentService: IIncidentService, - maintenanceService: IMaintenanceService + maintenanceService: IMaintenanceService, + statsAggregationService: IStatsAggregationService ) { this.SERVICE_NAME = SERVICE_NAME; this.networkService = networkService; @@ -43,6 +49,7 @@ class JobGenerator implements IJobGenerator { this.notificationService = notificationService; this.incidentService = incidentService; this.maintenanceService = maintenanceService; + this.statsAggregationService = statsAggregationService; } generateJob = () => { @@ -134,6 +141,62 @@ class JobGenerator implements IJobGenerator { } }; }; + + // Aggregates stats for the previous UTC hour and (at UTC midnight) the previous UTC day + generateStatsAggregationJob = () => { + return async () => { + // Compute previous hour window in UTC + const now = new Date(); + const utcNow = new Date( + Date.UTC( + now.getUTCFullYear(), + now.getUTCMonth(), + now.getUTCDate(), + now.getUTCHours(), + now.getUTCMinutes(), + now.getUTCSeconds(), + 0 + ) + ); + const hourEnd = new Date( + Date.UTC( + now.getUTCFullYear(), + now.getUTCMonth(), + now.getUTCDate(), + now.getUTCHours(), + 0, + 0, + 0 + ) + ); + const hourStart = new Date(hourEnd.getTime() - 60 * 60 * 1000); + + // Upsert hourly stats for the last full hour + await this.statsAggregationService.upsertHourly(hourStart, hourEnd); + + // Roll up the current day-so-far (not finalized) so charts have partial daily data + const todayStart = new Date( + Date.UTC( + now.getUTCFullYear(), + now.getUTCMonth(), + now.getUTCDate(), + 0, + 0, + 0, + 0 + ) + ); + await this.statsAggregationService.rollupDaily(todayStart, utcNow, false); + + // If at UTC midnight hour (i.e., hour just rolled over), finalize the previous day + const hour = utcNow.getUTCHours(); + if (hour === 0) { + const dayEnd = todayStart; // just reached 00:00, previous day ends now + const dayStart = new Date(dayEnd.getTime() - 24 * 60 * 60 * 1000); + await this.statsAggregationService.rollupDaily(dayStart, dayEnd, true); + } + }; + }; } export default JobGenerator; diff --git a/server/src/services/infrastructure/JobQueue.ts b/server/src/services/infrastructure/JobQueue.ts index fd36b8b77..c411accab 100644 --- a/server/src/services/infrastructure/JobQueue.ts +++ b/server/src/services/infrastructure/JobQueue.ts @@ -86,6 +86,18 @@ export default class JobQueue implements IJobQueue { active: true, }); + // Stats aggregation job (hourly) + this.scheduler.addTemplate( + "stats-aggregation-job", + this.jobGenerator.generateStatsAggregationJob() + ); + await this.scheduler.addJob({ + id: "stats-aggregation", + template: "stats-aggregation-job", + repeat: 60 * 60 * 1000, // hourly + active: true, + }); + const monitors = await Monitor.find(); for (const monitor of monitors) { const randomOffset = 1000 + Math.random() * 1000;