perf: bound runtime memory and storage usage

back/services/cron.ts

@@ -32,6 +32,7 @@ import { t } from '../shared/i18n';
 import { ScheduleType } from '../interface/schedule';
 import { logStreamManager } from '../shared/logStreamManager';
 import { isEmpty } from 'lodash';
+import { LogReadOptions, readLogChunk } from '../shared/logReader';
 
 @Service()
 export default class CronService {
@@ -766,27 +767,63 @@ export default class CronService {
     await this.setCrontab();
   }
 
-  public async log(id: number): Promise<{ content: string; status: string }> {
+  public async log(
+    id: number,
+    options: LogReadOptions = {},
+  ): Promise<{
+    content: string;
+    status: string;
+    offset: number;
+    nextOffset: number;
+    total: number;
+    truncated: boolean;
+  }> {
     const doc = await this.getDb({ id });
     if (!doc) {
-      return { content: '', status: 'empty' };
+      return {
+        content: '',
+        status: 'empty',
+        offset: 0,
+        nextOffset: 0,
+        total: 0,
+        truncated: false,
+      };
     }
     if (doc.log_name === '/dev/null') {
-      return { content: t('日志设置为忽略'), status: 'ignored' };
+      return {
+        content: t('日志设置为忽略'),
+        status: 'ignored',
+        offset: 0,
+        nextOffset: 0,
+        total: 0,
+        truncated: false,
+      };
     }
     const absolutePath = path.resolve(config.logPath, `${doc.log_path}`);
     const logFileExist = doc.log_path && (await fileExist(absolutePath));
     if (logFileExist) {
-      const content = await getFileContentByName(`${absolutePath}`);
+      const chunk = await readLogChunk(`${absolutePath}`, options);
       const isRunning =
         typeof doc.status === 'number' &&
         [CrontabStatus.running, CrontabStatus.queued].includes(doc.status);
-      return { content, status: isRunning ? 'running' : 'completed' };
+      return {
+        ...chunk,
+        status: isRunning ? 'running' : 'completed',
+      };
     } else {
-      return typeof doc.status === 'number' &&
+      const status =
+        typeof doc.status === 'number' &&
         [CrontabStatus.queued, CrontabStatus.running].includes(doc.status)
-        ? { content: t('运行中...'), status: 'running' }
-        : { content: t('日志不存在...'), status: 'notFound' };
+          ? 'running'
+          : 'notFound';
+      return {
+        content: status === 'running' ? t('运行中...') : t('日志不存在...'),
+        status,
+        offset: 0,
+        nextOffset: 0,
+        total: 0,
+        truncated: false,
+      };
     }
   }
 

back/services/metrics.ts

@@ -8,16 +8,19 @@ interface Metric {
   tags?: Record<string, string>;
 }
 
-class MetricsService {
-  private metrics: Metric[] = [];
+export const MAX_METRIC_SAMPLES = 1000;
+const METRIC_RETENTION_MS = 60 * 60 * 1000;
+
+export class MetricsService {
+  private metrics: Array<Metric | undefined> = new Array(MAX_METRIC_SAMPLES);
+  private metricCount = 0;
+  private nextMetricIndex = 0;
   private static instance: MetricsService;
 
   private constructor() {
     // 定期清理旧数据
-    setInterval(() => {
-      const oneHourAgo = Date.now() - 3600000;
-      this.metrics = this.metrics.filter(m => m.timestamp > oneHourAgo);
-    }, 60000);
+    const cleanupTimer = setInterval(() => this.removeExpiredMetrics(), 60000);
+    cleanupTimer.unref();
   }
 
   static getInstance(): MetricsService {
@@ -28,12 +31,14 @@ class MetricsService {
   }
 
   record(name: string, value: number, tags?: Record<string, string>) {
-    this.metrics.push({
+    this.metrics[this.nextMetricIndex] = {
       name,
       value,
       timestamp: Date.now(),
       tags,
-    });
+    };
+    this.nextMetricIndex = (this.nextMetricIndex + 1) % MAX_METRIC_SAMPLES;
+    this.metricCount = Math.min(this.metricCount + 1, MAX_METRIC_SAMPLES);
   }
 
   measure(name: string, fn: () => void, tags?: Record<string, string>) {
@@ -46,7 +51,11 @@ class MetricsService {
     }
   }
 
-  async measureAsync(name: string, fn: () => Promise<void>, tags?: Record<string, string>) {
+  async measureAsync(
+    name: string,
+    fn: () => Promise<void>,
+    tags?: Record<string, string>,
+  ) {
     const start = performance.now();
     try {
       await fn();
@@ -57,28 +66,61 @@ class MetricsService {
   }
 
   getMetrics(name?: string, tags?: Record<string, string>) {
-    let filtered = this.metrics;
-    
+    this.removeExpiredMetrics();
+    let filtered = this.getMetricSnapshot();
+
     if (name) {
-      filtered = filtered.filter(m => m.name === name);
+      filtered = filtered.filter((m) => m.name === name);
     }
-    
+
     if (tags) {
-      filtered = filtered.filter(m => {
+      filtered = filtered.filter((m) => {
         if (!m.tags) return false;
-        return Object.entries(tags).every(([key, value]) => m.tags![key] === value);
+        return Object.entries(tags).every(
+          ([key, value]) => m.tags![key] === value,
+        );
       });
     }
 
+    const values = filtered.map((metric) => metric.value);
     return {
       count: filtered.length,
-      average: filtered.reduce((acc, curr) => acc + curr.value, 0) / filtered.length,
-      min: Math.min(...filtered.map(m => m.value)),
-      max: Math.max(...filtered.map(m => m.value)),
+      average: values.length
+        ? values.reduce((acc, value) => acc + value, 0) / values.length
+        : 0,
+      min: values.length ? Math.min(...values) : 0,
+      max: values.length ? Math.max(...values) : 0,
       metrics: filtered,
     };
   }
 
+  private getMetricSnapshot(): Metric[] {
+    if (this.metricCount < MAX_METRIC_SAMPLES) {
+      return this.metrics.slice(0, this.metricCount) as Metric[];
+    }
+    return [
+      ...this.metrics.slice(this.nextMetricIndex),
+      ...this.metrics.slice(0, this.nextMetricIndex),
+    ] as Metric[];
+  }
+
+  private removeExpiredMetrics() {
+    const oldestTimestamp = Date.now() - METRIC_RETENTION_MS;
+    const retained = this.getMetricSnapshot().filter(
+      (metric) => metric.timestamp > oldestTimestamp,
+    );
+    if (retained.length === this.metricCount) return;
+
+    this.metrics = new Array(MAX_METRIC_SAMPLES);
+    this.metricCount = 0;
+    this.nextMetricIndex = 0;
+    for (const metric of retained) {
+      this.metrics[this.nextMetricIndex] = metric;
+      this.nextMetricIndex = (this.nextMetricIndex + 1) % MAX_METRIC_SAMPLES;
+      this.metricCount++;
+    }
+  }
+
   report() {
     const report = {
       timestamp: Date.now(),
@@ -89,4 +131,4 @@ class MetricsService {
   }
 }
 
-export const metricsService = MetricsService.getInstance(); 
+export const metricsService = MetricsService.getInstance();

back/services/retention.ts

@@ -0,0 +1,155 @@
+import fs from 'fs/promises';
+import path from 'path';
+import dayjs from 'dayjs';
+import { Op } from 'sequelize';
+import { Service } from 'typedi';
+import config from '../config';
+import { CrontabStatModel } from '../data/cronStats';
+import { sequelize } from '../data';
+import { InstanceStatus, RunningInstanceModel } from '../data/runningInstance';
+import { AuthDataType, SystemModel } from '../data/system';
+import {
+  DependenceCacheType,
+  getDirectorySize,
+  isDependenceCacheType,
+  normalizeRetentionPolicy,
+  RetentionPolicy,
+} from '../shared/retention';
+
+export interface StorageCleanupRequest extends RetentionPolicy {
+  dependenceCacheTypes?: DependenceCacheType[];
+  compactDatabase?: boolean;
+}
+
+function runningInstanceWhere(days: number) {
+  const cutoff = dayjs().subtract(days, 'day').unix();
+  return {
+    status: { [Op.ne]: InstanceStatus.running },
+    [Op.or]: [
+      { finished_at: { [Op.lt]: cutoff } },
+      {
+        finished_at: { [Op.is]: null },
+        started_at: { [Op.lt]: cutoff },
+      },
+    ],
+  };
+}
+
+function cronStatWhere(days: number) {
+  return {
+    date: { [Op.lt]: dayjs().subtract(days, 'day').format('YYYY-MM-DD') },
+  };
+}
+
+@Service()
+export default class RetentionService {
+  public async updatePolicy(policy: Partial<RetentionPolicy>) {
+    const normalized = normalizeRetentionPolicy(policy);
+    const systemConfig = await SystemModel.findOne({
+      where: { type: AuthDataType.systemConfig },
+    });
+    if (!systemConfig) {
+      throw new Error('System config not found');
+    }
+    await SystemModel.update(
+      {
+        info: {
+          ...systemConfig.info,
+          ...normalized,
+        },
+      },
+      { where: { id: systemConfig.id } },
+    );
+    return normalized;
+  }
+
+  public async preview(request: StorageCleanupRequest) {
+    const policy = normalizeRetentionPolicy(request);
+    const dependenceCacheTypes = (request.dependenceCacheTypes || []).filter(
+      isDependenceCacheType,
+    );
+    const [runningInstances, cronStats, dependenceCaches] = await Promise.all([
+      policy.runningInstanceRetentionDays > 0
+        ? RunningInstanceModel.count({
+            where: runningInstanceWhere(policy.runningInstanceRetentionDays),
+          })
+        : 0,
+      policy.cronStatRetentionDays > 0
+        ? CrontabStatModel.count({
+            where: cronStatWhere(policy.cronStatRetentionDays),
+          })
+        : 0,
+      Promise.all(
+        dependenceCacheTypes.map(async (type) => ({
+          type,
+          bytes: await getDirectorySize(
+            path.join(config.dependenceCachePath, type),
+          ),
+        })),
+      ),
+    ]);
+
+    return {
+      policy,
+      runningInstances,
+      cronStats,
+      dependenceCaches,
+      dependenceCacheBytes: dependenceCaches.reduce(
+        (total, cache) => total + cache.bytes,
+        0,
+      ),
+      compactDatabase: Boolean(request.compactDatabase),
+    };
+  }
+
+  public async cleanup(request: StorageCleanupRequest) {
+    const preview = await this.preview(request);
+    const deleted = await sequelize.transaction(async (transaction) => {
+      const runningInstances =
+        preview.policy.runningInstanceRetentionDays > 0
+          ? await RunningInstanceModel.destroy({
+              where: runningInstanceWhere(
+                preview.policy.runningInstanceRetentionDays,
+              ),
+              transaction,
+            })
+          : 0;
+      const cronStats =
+        preview.policy.cronStatRetentionDays > 0
+          ? await CrontabStatModel.destroy({
+              where: cronStatWhere(preview.policy.cronStatRetentionDays),
+              transaction,
+            })
+          : 0;
+      return { runningInstances, cronStats };
+    });
+
+    const dependenceCaches = [];
+    for (const cache of preview.dependenceCaches) {
+      await fs.rm(path.join(config.dependenceCachePath, cache.type), {
+        recursive: true,
+        force: true,
+      });
+      dependenceCaches.push(cache);
+    }
+
+    if (
+      request.compactDatabase &&
+      (deleted.runningInstances || deleted.cronStats)
+    ) {
+      await sequelize.query('VACUUM');
+    }
+
+    return {
+      deleted: {
+        ...deleted,
+        dependenceCaches,
+        dependenceCacheBytes: preview.dependenceCacheBytes,
+      },
+      compactedDatabase: Boolean(
+        request.compactDatabase &&
+          (deleted.runningInstances || deleted.cronStats),
+      ),
+    };
+  }
+}

back/services/subscription.ts

@@ -33,6 +33,7 @@ import { CrontabModel } from '../data/cron';
 import CrontabService from './cron';
 import taskLimit from '../shared/pLimit';
 import { logStreamManager } from '../shared/logStreamManager';
+import { LogReadOptions, readLogChunk } from '../shared/logReader';
 
 @Service()
 export default class SubscriptionService {
@@ -366,14 +367,20 @@ export default class SubscriptionService {
     }
   }
 
-  public async log(id: number) {
+  public async log(id: number, options: LogReadOptions = {}) {
     const doc = await this.getDb({ id });
     if (!doc || !doc.log_path) {
-      return '';
+      return {
+        content: '',
+        offset: 0,
+        nextOffset: 0,
+        total: 0,
+        truncated: false,
+      };
     }
 
     const absolutePath = await handleLogPath(doc.log_path as string);
-    return await getFileContentByName(absolutePath);
+    return await readLogChunk(absolutePath, options);
   }
 
   public async logs(id: number) {

back/services/system.ts

@@ -2,7 +2,6 @@ import { spawn } from 'cross-spawn';
 import { Response } from 'express';
 import fs from 'fs';
 import { Agent, request } from 'undici';
-import sum from 'lodash/sum';
 import path from 'path';
 import { Inject, Service } from 'typedi';
 import winston from 'winston';
@@ -467,6 +466,7 @@ export default class SystemService {
     query: {
       startTime?: string;
       endTime?: string;
+      limit?: number;
     },
   ) {
     const startTime = dayjs(query.startTime || undefined)
@@ -481,8 +481,30 @@ export default class SystemService {
       .filter((x) => x.title.endsWith('.log'))
       .filter((x) => x.createTime >= startTime && x.createTime <= endTime);
 
+    const limit = Math.min(
+      Math.max(Number(query.limit) || 1024 * 1024, 1),
+      1024 * 1024,
+    );
+    const total = logs.reduce((size, log) => size + (log.size || 0), 0);
+    let remaining = limit;
+    const selected: Array<
+      (typeof logs)[number] & { start: number; length: number }
+    > = [];
+    for (let index = logs.length - 1; index >= 0 && remaining > 0; index--) {
+      const log = logs[index];
+      const size = log.size || 0;
+      const length = Math.min(size, remaining);
+      if (length > 0) {
+        selected.unshift({ ...log, start: size - length, length });
+        remaining -= length;
+      }
+    }
+
+    const contentLength = selected.reduce((size, log) => size + log.length, 0);
     res.set({
-      'Content-Length': sum(logs.map((x) => x.size)),
+      'Content-Length': contentLength,
+      'X-QL-Log-Total': total,
+      'X-QL-Log-Truncated': total > contentLength ? 'true' : 'false',
     });
     (function sendFiles(res, fileNames) {
       if (fileNames.length === 0) {
@@ -494,13 +516,17 @@ export default class SystemService {
       if (currentLog) {
         const currentFileStream = fs.createReadStream(
           path.join(config.systemLogPath, currentLog.title),
+          {
+            start: currentLog.start,
+            end: currentLog.start + currentLog.length - 1,
+          },
         );
         currentFileStream.on('end', () => {
           sendFiles(res, fileNames);
         });
         currentFileStream.pipe(res, { end: false });
       }
-    })(res, logs);
+    })(res, selected);
   }
 
   public async deleteSystemLog() {