hourly stats and daily rollup
This commit is contained in:
Alex Holliday committed
1 parent
e1db8ea513
commit
59e85f7613
10 files changed
+1132
-19
No files matched your search
@@ -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";
|
||||
@@ -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();
|
||||
|
||||
@@ -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<IInfraCpu>(
|
||||
{
|
||||
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<IInfraMemory>(
|
||||
{
|
||||
total_bytes: { type: Number },
|
||||
available_bytes: { type: Number },
|
||||
used_bytes: { type: Number },
|
||||
usage_percent: { type: Number },
|
||||
},
|
||||
{ _id: false }
|
||||
);
|
||||
|
||||
const InfraDiskEntrySchema = new Schema<IInfraDiskEntry>(
|
||||
{
|
||||
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<IInfraHost>(
|
||||
{
|
||||
os: { type: String },
|
||||
platform: { type: String },
|
||||
kernel_version: { type: String },
|
||||
pretty_name: { type: String },
|
||||
},
|
||||
{ _id: false }
|
||||
);
|
||||
|
||||
const InfraNetEntrySchema = new Schema<IInfraNetEntry>(
|
||||
{
|
||||
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<IStatsDaily>(
|
||||
{
|
||||
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<IStatsDaily>(
|
||||
"StatsDaily",
|
||||
StatsDailySchema
|
||||
);
|
||||
|
||||
@@ -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<IInfraCpu>(
|
||||
{
|
||||
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<IInfraMemory>(
|
||||
{
|
||||
total_bytes: { type: Number },
|
||||
available_bytes: { type: Number },
|
||||
used_bytes: { type: Number },
|
||||
usage_percent: { type: Number },
|
||||
},
|
||||
{ _id: false }
|
||||
);
|
||||
|
||||
const InfraDiskEntrySchema = new Schema<IInfraDiskEntry>(
|
||||
{
|
||||
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<IInfraHost>(
|
||||
{
|
||||
os: { type: String },
|
||||
platform: { type: String },
|
||||
kernel_version: { type: String },
|
||||
pretty_name: { type: String },
|
||||
},
|
||||
{ _id: false }
|
||||
);
|
||||
|
||||
const InfraNetEntrySchema = new Schema<IInfraNetEntry>(
|
||||
{
|
||||
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<IStatsHourly>(
|
||||
{
|
||||
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<IStatsHourly>(
|
||||
"StatsHourly",
|
||||
StatsHourlySchema
|
||||
);
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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<string, any> => {
|
||||
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") {
|
||||
|
||||
@@ -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<number>;
|
||||
rollupDaily: (
|
||||
dayStart: Date,
|
||||
dayEnd: Date,
|
||||
finalize?: boolean
|
||||
) => Promise<number>;
|
||||
}
|
||||
|
||||
const BASE_TYPES: MonitorType[] = ["http", "https", "port", "ping"];
|
||||
|
||||
class StatsAggregationService implements IStatsAggregationService {
|
||||
upsertHourly = async (windowStart: Date, windowEnd: Date): Promise<number> => {
|
||||
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<any>([
|
||||
{ $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<any>([
|
||||
{ $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<number> => {
|
||||
// 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<string, Weighted>;
|
||||
}
|
||||
|
||||
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<string, DeviceAgg>;
|
||||
host: { os?: string; platform?: string; kernel_version?: string; pretty_name?: string };
|
||||
net: Map<string, DeviceAgg>;
|
||||
lastTs: number; // for choosing "last" fields
|
||||
}
|
||||
|
||||
const groups = new Map<string, GroupAgg>();
|
||||
|
||||
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;
|
||||
@@ -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";
|
||||
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
|
||||
Reference in new issue
Block a user