import { BadRequestException, Injectable, Logger, NotFoundException, ServiceUnavailableException } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { Prisma } from '@prisma/client'; import { createHash } from 'node:crypto'; import { PrismaService } from '../prisma/prisma.service'; import { compareMountpoints, FILESYSTEM_LABELS, FILESYSTEM_SELECTOR, FILESYSTEM_USAGE_PERCENT, filesystemIdentity } from './filesystem-metrics'; import type { InfrastructureAlert, InfrastructureMetricPoint, InfrastructureMonitoringOverview, InfrastructureMonitoringRange, InfrastructureServiceStatus, InfrastructureServiceMetricGroup, } from './infrastructure-monitoring.contracts'; type PrometheusSample = [number, string]; type PrometheusSeries = { metric: Record; value?: PrometheusSample; values?: PrometheusSample[]; }; type PrometheusQueryResponse = { status: 'success' | 'error'; data?: { result?: PrometheusSeries[] }; error?: string; }; type PrometheusAlertResponse = { status: 'success' | 'error'; data?: { alerts?: Array<{ labels?: Record; annotations?: Record; state?: string; activeAt?: string; value?: string; }>; }; }; const RANGE_CONFIG: Record = { '1h': { seconds: 60 * 60, step: 60 }, '24h': { seconds: 24 * 60 * 60, step: 300 }, '7d': { seconds: 7 * 24 * 60 * 60, step: 1800 }, }; const QUERIES = { cpuUsagePercent: '100 - (avg by (instance) (rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)', memoryUsagePercent: '(1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100', memoryTotalBytes: 'node_memory_MemTotal_bytes', memoryAvailableBytes: 'node_memory_MemAvailable_bytes', diskUsagePercent: FILESYSTEM_USAGE_PERCENT, diskTotalBytes: `node_filesystem_size_bytes${FILESYSTEM_SELECTOR}`, diskAvailableBytes: `min by (${FILESYSTEM_LABELS}) (node_filesystem_avail_bytes${FILESYSTEM_SELECTOR})`, networkReceiveBytesPerSecond: 'sum(rate(node_network_receive_bytes_total{device!~"lo"}[5m]))', networkTransmitBytesPerSecond: 'sum(rate(node_network_transmit_bytes_total{device!~"lo"}[5m]))', load1: 'node_load1', uptimeSeconds: 'time() - node_boot_time_seconds', lastSampleAt: 'max(timestamp(node_uname_info))', // PromQL字符串本身需要两个反斜杠才能把正则的“\.”传给RE2;TypeScript字面量因此需要写四个。 services: 'max by (name) (node_systemd_unit_state{name=~"cmpp-api\\\\.service|cmpp-gateway\\\\.service|postgresql\\\\.service|redis(-server)?\\\\.service|cmpp-minio\\\\.service|nginx\\\\.service",state="active"})', } as const; const SERVICE_DEFINITIONS = [ { key: 'api', name: 'API服务', units: ['cmpp-api.service'] }, { key: 'gateway', name: 'Gateway服务', units: ['cmpp-gateway.service'] }, { key: 'postgresql', name: 'PostgreSQL', units: ['postgresql.service'] }, { key: 'redis', name: 'Redis', units: ['redis.service', 'redis-server.service'] }, { key: 'minio', name: 'MinIO', units: ['cmpp-minio.service'] }, { key: 'nginx', name: 'Nginx', units: ['nginx.service'] }, ] as const; const SERVICE_METRIC_DEFINITIONS = [ { key: 'api', name: 'API服务', metrics: [ ['requestsPerSecond', '请求速率', 'cmpp:service_api:requests_per_second', 'per_second'], ['errorPercent', '5xx错误率', 'cmpp:service_api:error_percent', 'percent'], ['latencyP95', 'P95响应', 'cmpp:service_api:latency_p95_seconds', 'seconds'], ['eventLoopP99', '事件循环P99', 'cmpp:service_api:event_loop_p99_seconds', 'seconds'], ] }, { key: 'gateway', name: 'Gateway服务', metrics: [ ['submitsPerSecond', '提交速率', 'cmpp:service_gateway:submits_per_second', 'per_second'], ['failurePercent', '提交失败率', 'cmpp:service_gateway:failure_percent', 'percent'], ['queuePending', 'Stream pending', 'cmpp:service_gateway:queue_pending', 'count'], ['queueOldestSeconds', '最旧pending', 'cmpp:service_gateway:queue_oldest_seconds', 'seconds'], ] }, { key: 'postgresql', name: 'PostgreSQL', metrics: [ ['connectionPercent', '连接使用率', 'cmpp:service_postgresql:connection_percent', 'percent'], ['deadlocks15m', '15分钟死锁', 'cmpp:service_postgresql:deadlocks_15m', 'count'], ] }, { key: 'redis', name: 'Redis', metrics: [ ['memoryPercent', '内存使用率', 'cmpp:service_redis:memory_percent', 'percent'], ['memoryUsedBytes', '已用内存', 'cmpp:service_redis:memory_used_bytes', 'bytes'], ['connectedClients', '客户端连接', 'cmpp:service_redis:connected_clients', 'count'], ['evictions5m', '5分钟淘汰', 'cmpp:service_redis:evictions_5m', 'count'], ] }, { key: 'minio', name: 'MinIO', metrics: [ ['capacityPercent', '存储容量使用率', 'cmpp:service_minio:capacity_percent', 'percent'], ['usageBytes', '对象数据量', 'cmpp:service_minio:usage_bytes', 'bytes'], ['objects', '对象数', 'cmpp:service_minio:objects', 'count'], ['drivesOffline', '离线存储盘', 'cmpp:service_minio:drives_offline', 'count'], ] }, { key: 'nginx', name: 'Nginx', metrics: [ ['connectionsActive', '活跃连接', 'cmpp:service_nginx:connections_active', 'count'], ['requestsPerSecond', '请求速率', 'cmpp:service_nginx:requests_per_second', 'per_second'], ] }, ] as const; const SERVICE_METRICS_QUERY = '{__name__=~"cmpp:service_.*"}'; function finiteNumber(value: string | number | undefined): number | null { const parsed = Number(value); return Number.isFinite(parsed) ? parsed : null; } function normalizePrometheusUrl(rawValue: unknown) { const url = new URL(String(rawValue ?? 'http://127.0.0.1:9090')); if (url.protocol !== 'http:' && url.protocol !== 'https:') throw new Error('PROMETHEUS_URL must use HTTP or HTTPS'); if (url.username || url.password) throw new Error('PROMETHEUS_URL must not contain credentials'); const privateIpv4 = /^(10\.|192\.168\.|172\.(1[6-9]|2\d|3[01])\.)/.test(url.hostname); const loopback = url.hostname === '127.0.0.1' || url.hostname === 'localhost' || url.hostname === '[::1]'; // Plain HTTP is only safe on loopback or an explicit RFC1918 address; named remote endpoints must use HTTPS. if (url.protocol === 'http:' && !loopback && !privateIpv4) throw new Error('Remote PROMETHEUS_URL must use HTTPS'); return url.toString().replace(/\/$/, ''); } function vectorValue(response: PrometheusQueryResponse): number | null { return finiteNumber(response.data?.result?.[0]?.value?.[1]); } function matrixValues(response: PrometheusQueryResponse): InfrastructureMetricPoint[] { return (response.data?.result?.[0]?.values ?? []).flatMap(([timestamp, value]) => { const parsed = finiteNumber(value); return parsed === null ? [] : [{ timestamp: new Date(timestamp * 1000).toISOString(), value: parsed }]; }); } function emptyMetrics(): InfrastructureMonitoringOverview['metrics'] { return { cpuUsagePercent: null, memoryUsagePercent: null, memoryTotalBytes: null, memoryAvailableBytes: null, diskUsagePercent: null, diskTotalBytes: null, diskAvailableBytes: null, networkReceiveBytesPerSecond: null, networkTransmitBytesPerSecond: null, load1: null, uptimeSeconds: null, }; } function emptyTrends(): InfrastructureMonitoringOverview['trends'] { return { cpuUsagePercent: [], memoryUsagePercent: [], diskUsagePercent: [], networkReceiveBytesPerSecond: [], networkTransmitBytesPerSecond: [], }; } @Injectable() export class InfrastructureMonitoringService { private readonly logger = new Logger(InfrastructureMonitoringService.name); private readonly prometheusUrl: string; private readonly queryTimeoutMs: number; constructor(config: ConfigService, private readonly prisma: PrismaService) { this.prometheusUrl = normalizePrometheusUrl(config.get('PROMETHEUS_URL')); this.queryTimeoutMs = Math.min(15_000, Math.max(1_000, Number(config.get('PROMETHEUS_QUERY_TIMEOUT_MS') ?? 5_000))); } async overview(rawRange?: string, userId?: string): Promise { const range = this.parseRange(rawRange); const collectedAt = new Date().toISOString(); try { const [instant, trends, serviceResponse, serviceMetricResponse, alertResponse] = await Promise.all([ this.loadInstantMetrics(), this.loadTrends(range), this.query(QUERIES.services), this.query(SERVICE_METRICS_QUERY), this.getJson('/api/v1/alerts'), ]); const services = this.parseServices(serviceResponse); const serviceMetrics = this.parseServiceMetrics(serviceMetricResponse); const alerts = await this.attachReadState(this.parseAlerts(alertResponse), userId); const warningAlerts = alerts.filter((item) => item.severity === 'warning').length; const criticalAlerts = alerts.filter((item) => item.severity === 'critical').length; const overallStatus = criticalAlerts > 0 ? 'critical' : warningAlerts > 0 ? 'warning' : 'healthy'; const rootDisk = instant.disks.find((disk) => disk.mountpoints.includes('/')); return { available: true, range, collectedAt, lastSampleAt: instant.lastSampleAt === null ? null : new Date(instant.lastSampleAt * 1000).toISOString(), summary: { overallStatus, serviceTotal: services.length, serviceHealthy: services.filter((item) => item.status === 'healthy').length, warningAlerts, criticalAlerts, activeAlerts: alerts.length, }, metrics: instant.metrics, trends: { ...trends.metrics, diskUsagePercent: rootDisk ? trends.disks.get(rootDisk.id) ?? [] : [] }, disks: instant.disks.map((disk) => ({ ...disk, trend: trends.disks.get(disk.id) ?? [] })), services, serviceMetrics, alerts, }; } catch (error) { // 页面必须整体清空陈旧指标,但服务端仍需留下不含PromQL/地址/凭据的根因摘要便于运维诊断。 this.logger.warn(`Prometheus monitoring overview unavailable: ${error instanceof Error ? error.message : 'unknown error'}`); return this.unavailable(range, collectedAt); } } async notificationSummary(userId?: string) { try { const alerts = await this.attachReadState(this.parseAlerts(await this.getJson('/api/v1/alerts')), userId); const unreadAlerts = alerts.filter((item) => !item.acknowledged); return { count: unreadAlerts.length, criticalCount: unreadAlerts.filter((item) => item.severity === 'critical').length }; } catch (error) { this.logger.warn(`Prometheus notification summary unavailable: ${error instanceof Error ? error.message : 'unknown error'}`); throw new ServiceUnavailableException('Prometheus活动告警当前不可用'); } } async markAlertRead(fingerprint: string, rawActiveAt: unknown, userId: string) { if (!/^[a-f0-9]{24}$/.test(fingerprint)) throw new BadRequestException('告警指纹无效'); const activeAt = new Date(String(rawActiveAt ?? '')); if (!Number.isFinite(activeAt.getTime())) throw new BadRequestException('告警开始时间无效'); const activeAlerts = this.parseAlerts(await this.getJson('/api/v1/alerts')); const current = activeAlerts.find((item) => item.fingerprint === fingerprint && Date.parse(item.startedAt) === activeAt.getTime()); if (!current) throw new NotFoundException('该次活动告警已结束或已重新触发,请刷新后重试'); const readAt = new Date(); const log = () => this.prisma.operationLog.create({ data: { userId, action: 'monitoring.alert_marked_read', resource: 'infrastructure_alert', resourceId: fingerprint, detail: { activeAt: activeAt.toISOString(), alertName: current.name, severity: current.severity } }, }); let read; try { [read] = await this.prisma.$transaction([ this.prisma.infrastructureAlertRead.create({ data: { fingerprint, activeAt, userId, readAt } }), log(), ]); } catch (error) { if (!(error instanceof Prisma.PrismaClientKnownRequestError) || error.code !== 'P2002') throw error; const existing = await this.prisma.infrastructureAlertRead.findUniqueOrThrow({ where: { fingerprint_userId: { fingerprint, userId } } }); // 同一次触发重复点击不更新readAt也不重复写日志;activeAt变化才代表同指纹的新触发周期。 if (existing.activeAt.getTime() === activeAt.getTime()) read = existing; else [read] = await this.prisma.$transaction([ this.prisma.infrastructureAlertRead.update({ where: { fingerprint_userId: { fingerprint, userId } }, data: { activeAt, readAt } }), log(), ]); } return { fingerprint, activeAt: read.activeAt.toISOString(), acknowledged: true, acknowledgedAt: read.readAt.toISOString() }; } private parseRange(value?: string): InfrastructureMonitoringRange { const range = value || '24h'; if (!(range in RANGE_CONFIG)) throw new BadRequestException('监控时间范围只支持1h、24h或7d'); return range as InfrastructureMonitoringRange; } private async loadInstantMetrics() { const keys = Object.keys(emptyMetrics()) as Array; const responses = await Promise.all([...keys.map((key) => this.query(QUERIES[key])), this.query(QUERIES.lastSampleAt)]); const metrics = emptyMetrics(); keys.forEach((key, index) => { if (!key.startsWith('disk')) metrics[key] = vectorValue(responses[index]); }); const diskSamples = (key: keyof typeof metrics) => responses[keys.indexOf(key)].data?.result ?? []; const usage = new Map(diskSamples('diskUsagePercent').map((item) => [filesystemIdentity(item.metric), finiteNumber(item.value?.[1])])); const available = new Map(diskSamples('diskAvailableBytes').map((item) => [filesystemIdentity(item.metric), finiteNumber(item.value?.[1])])); const groups = new Map(); for (const item of diskSamples('diskTotalBytes')) { if (!item.metric.device || !item.metric.mountpoint || (finiteNumber(item.value?.[1]) ?? 0) <= 0) continue; const id = filesystemIdentity(item.metric); const group = groups.get(id) ?? []; group.push(item); groups.set(id, group); } const disks = [...groups].map(([id, items]) => { const mountpoints = [...new Set(items.map((item) => item.metric.mountpoint))].sort(compareMountpoints); const metric = items[0].metric; return { id, instance: metric.instance ?? '', device: metric.device, filesystem: metric.fstype ?? '', mountpoint: mountpoints[0], mountpoints, // Never sum aliases. Max/min also tolerate slight sampling differences. totalBytes: Math.max(...items.map((item) => finiteNumber(item.value?.[1])!)), availableBytes: available.get(id) ?? null, usagePercent: usage.get(id) ?? null, }; }) .sort((left, right) => left.instance.localeCompare(right.instance) || (left.mountpoint === '/' ? -1 : right.mountpoint === '/' ? 1 : left.mountpoint.localeCompare(right.mountpoint))); const rootDisk = disks.find((disk) => disk.mountpoints.includes('/')); metrics.diskUsagePercent = rootDisk?.usagePercent ?? null; metrics.diskTotalBytes = rootDisk?.totalBytes ?? null; metrics.diskAvailableBytes = rootDisk?.availableBytes ?? null; return { metrics, disks, lastSampleAt: vectorValue(responses[responses.length - 1]) }; } private async loadTrends(range: InfrastructureMonitoringRange) { const config = RANGE_CONFIG[range]; const end = Math.floor(Date.now() / 1000); const start = end - config.seconds; const keys = Object.keys(emptyTrends()) as Array; const responses = await Promise.all(keys.map((key) => this.queryRange(QUERIES[key], start, end, config.step))); return { metrics: Object.fromEntries(keys.map((key, index) => [key, key === 'diskUsagePercent' ? [] : matrixValues(responses[index])])) as InfrastructureMonitoringOverview['trends'], disks: new Map((responses[keys.indexOf('diskUsagePercent')].data?.result ?? []).map((item) => [ filesystemIdentity(item.metric), matrixValues({ status: 'success', data: { result: [item] } }), ])), }; } private parseServices(response: PrometheusQueryResponse): InfrastructureServiceStatus[] { const values = new Map(); for (const item of response.data?.result ?? []) { if (item.metric.name) values.set(item.metric.name, vectorValue({ status: 'success', data: { result: [item] } }) ?? 0); } return SERVICE_DEFINITIONS.map((definition) => { const present = definition.units.filter((unit) => values.has(unit)); const status = present.length === 0 ? 'unknown' : present.some((unit) => (values.get(unit) ?? 0) >= 1) ? 'healthy' : 'unhealthy'; return { key: definition.key, name: definition.name, unit: present[0] ?? definition.units[0], status }; }); } private parseAlerts(response: PrometheusAlertResponse): InfrastructureAlert[] { return (response.data?.alerts ?? []) .filter((item) => item.state === 'firing' || item.state === 'pending') .map((item) => { const labels = item.labels ?? {}; const annotations = item.annotations ?? {}; const severity: InfrastructureAlert['severity'] = labels.severity === 'critical' ? 'critical' : labels.severity === 'warning' ? 'warning' : 'info'; const identity = JSON.stringify(Object.entries(labels).sort(([left], [right]) => left.localeCompare(right))); return { fingerprint: createHash('sha256').update(identity).digest('hex').slice(0, 24), name: labels.alertname || '未命名告警', severity, status: item.state || 'unknown', startedAt: item.activeAt || new Date().toISOString(), summary: annotations.summary || annotations.description || labels.alertname || '监控告警', description: annotations.description, currentValue: annotations.currentValue || item.value, threshold: annotations.threshold, service: labels.service, instance: labels.instance, acknowledged: false, }; }) .sort((left, right) => { const priority: Record = { critical: 0, warning: 1, info: 2 }; return priority[left.severity] - priority[right.severity] || Date.parse(left.startedAt) - Date.parse(right.startedAt); }); } private async attachReadState(alerts: InfrastructureAlert[], userId?: string) { if (!userId || alerts.length === 0) return alerts; const reads = await this.prisma.infrastructureAlertRead.findMany({ where: { userId, fingerprint: { in: alerts.map((item) => item.fingerprint) } }, select: { fingerprint: true, activeAt: true, readAt: true }, }); const byFingerprint = new Map(reads.map((item) => [item.fingerprint, item])); return alerts.map((alert) => { const read = byFingerprint.get(alert.fingerprint); const acknowledged = Boolean(read && read.activeAt.getTime() === Date.parse(alert.startedAt)); return { ...alert, acknowledged, acknowledgedAt: acknowledged ? read?.readAt.toISOString() : undefined }; }); } private parseServiceMetrics(response: PrometheusQueryResponse): InfrastructureServiceMetricGroup[] { const values = new Map(); for (const item of response.data?.result ?? []) { const metricName = item.metric.__name__; const value = vectorValue({ status: 'success', data: { result: [item] } }); if (metricName && value !== null) values.set(metricName, value); } return SERVICE_METRIC_DEFINITIONS.map((group) => ({ key: group.key, name: group.name, available: group.metrics.some((metric) => values.has(metric[2])), metrics: group.metrics.map(([key, label, metricName, unit]) => ({ key, label, value: values.get(metricName) ?? null, unit })), })); } private unavailable(range: InfrastructureMonitoringRange, collectedAt: string): InfrastructureMonitoringOverview { const services = SERVICE_DEFINITIONS.map((item) => ({ key: item.key, name: item.name, unit: item.units[0], status: 'unknown' as const })); return { available: false, range, collectedAt, lastSampleAt: null, error: 'Prometheus监控数据当前不可用,请检查采集与服务状态', summary: { overallStatus: 'unknown', serviceTotal: services.length, serviceHealthy: 0, warningAlerts: 0, criticalAlerts: 0, activeAlerts: 0 }, metrics: emptyMetrics(), disks: [], trends: emptyTrends(), services, serviceMetrics: SERVICE_METRIC_DEFINITIONS.map((group) => ({ key: group.key, name: group.name, available: false, metrics: [] })), alerts: [], }; } private query(query: string) { return this.getJson('/api/v1/query', { query }); } private queryRange(query: string, start: number, end: number, step: number) { return this.getJson('/api/v1/query_range', { query, start: String(start), end: String(end), step: String(step) }); } private async getJson(path: string, params: Record = {}): Promise { const url = new URL(`${this.prometheusUrl}${path}`); Object.entries(params).forEach(([key, value]) => url.searchParams.set(key, value)); const response = await fetch(url, { headers: { Accept: 'application/json' }, signal: AbortSignal.timeout(this.queryTimeoutMs) }); if (!response.ok) throw new Error(`Prometheus HTTP ${response.status}`); const result = await response.json() as T; if (result.status !== 'success') throw new Error('Prometheus query failed'); return result; } }