Change from global total storage limit to epic based storage and date limit

Signed-off-by: Argus Li <contactme@chunkeili.com>

Author: ArgusLi

apps/metrics/config.yml

@@ -9,43 +9,54 @@ collector:
   batch_ms: 60000
   batch_max: 500
 
-storage:
-  retention_days: 5
-
 epics:
   - name: "memory"
     type: "memory_stats"
     poll_ms: 5000
     file_prefix: "memory"
+    data_retention_mb: 15
+    data_retention_days: 5
 
   - name: "cpu"
     type: "info_cpu"
     poll_ms: 5000
     file_prefix: "cpu"
+    data_retention_mb: 5
+    data_retention_days: 5
 
   - name: "commandlog_large_reply"
     type: "commandlog_large_reply"
     poll_ms: 10000
     file_prefix: "commandlog_large_reply"
+    data_retention_mb: 5
+    data_retention_days: 5
 
   - name: "commandlog_large_request"
     type: "commandlog_large_request"
     poll_ms: 10000
     file_prefix: "commandlog_large_request"
+    data_retention_mb: 5
+    data_retention_days: 5
 
   - name: "commandlog_slow"
     type: "commandlog_slow"
     poll_ms: 10000
     file_prefix: "commandlog_slow"
+    data_retention_mb: 5
+    data_retention_days: 5
 
   - name: "slowlog_len"
     type: "slowlog_len"
     poll_ms: 60000
     file_prefix: "slowlog_len"
-  
+    data_retention_mb: 3
+    data_retention_days: 5
+
   - name: "monitor"
     type: "monitor"
     monitoringDuration: 10000
     monitoringInterval: 10000
     maxCommandsPerRun: 1000000
     file_prefix: "monitor"
+    data_retention_mb: 12
+    data_retention_days: 5

apps/metrics/src/__tests__/test-helpers.js

@@ -26,7 +26,6 @@ export const createMockMonitor = () => ({
 export const createMockConfig = (overrides = {}) => ({
   valkey: { url: "valkey://localhost:6379" },
   server: { port: 3000, data_dir: "/test/data" },
-  storage: { retention_days: 30, retention_size_mb: 50 },
   collector: { batch_ms: 1000, batch_max: 100 },
   epics: [],
   ...overrides,

apps/metrics/src/config.js

@@ -10,12 +10,13 @@ const cfgPath = process.env.CONFIG_PATH || path.join(__dirname, "..", "config.ym
 
 let config = null
 
+const EPIC_DEFAULTS = { data_retention_mb: 10, data_retention_days: 30 }
+
 const DEFAULTS = {
   backend: { ping_interval: 10000 },
   valkey: {},
   server: { port: 3000, data_dir: "/app/data" },
   collector: { batch_ms: 60000, batch_max: 500 },
-  storage: { retention_days: 30, retention_size_mb: 50 },
   epics: [],
 }
 
@@ -26,19 +27,18 @@ const loadConfig = () => {
   const cfg = mergeDeepLeft(parsed, DEFAULTS)
 
   // Type guards
-  for (const key of ["backend", "valkey", "server", "collector", "storage"]) {
+  for (const key of ["backend", "valkey", "server", "collector"]) {
     if (typeof cfg[key] !== "object" || Array.isArray(cfg[key])) {
       cfg[key] = DEFAULTS[key]
     }
   }
   if (!Array.isArray(cfg.epics)) cfg.epics = []
+  cfg.epics = cfg.epics.map((e) => ({ ...EPIC_DEFAULTS, ...e }))
 
   if (process.env.PORT) cfg.server.port = Number(process.env.PORT)
   if (process.env.DATA_DIR) cfg.server.data_dir = process.env.DATA_DIR
   if (process.env.BATCH_MS) cfg.collector.batch_ms = Number(process.env.BATCH_MS)
   if (process.env.BATCH_MAX) cfg.collector.batch_max = Number(process.env.BATCH_MAX)
-  if (process.env.RETENTION_DAYS) cfg.storage.retention_days = Number(process.env.RETENTION_DAYS)
-  if (process.env.RETENTION_SIZE) cfg.storage.retention_size_mb = Number(process.env.RETENTION_SIZE)
 
   if (cfg.logging && typeof cfg.logging === "object") {
     if (!process.env.LOG_LEVEL && cfg.logging.level) process.env.LOG_LEVEL = String(cfg.logging.level)

apps/metrics/src/effects/ndjson-cleaner.js

@@ -9,7 +9,16 @@ const log = createLogger("ndjson-cleaner")
 
 let cleanerStopper
 
+const retentionForFile = (fileName, retentionByPrefix) => {
+  for (const [prefix, days] of retentionByPrefix) {
+    if (fileName.startsWith(`${prefix}_`)) return days
+  }
+  return -1
+}
+
 export const setupNdjsonCleaner = ( cfg ) => {
+  const retentionByPrefix = cfg.epics.map((e) => [e.file_prefix || e.name, e.data_retention_days])
+
   const pipeline$ = timer(0, METRICS_EVICTION_POLICY.INTERVAL).pipe(
     exhaustMap(() => 
       fs.promises.readdir(cfg.server.data_dir) // get file names from directory
@@ -18,9 +27,10 @@ export const setupNdjsonCleaner = ( cfg ) => {
         .then((fileNames) => Promise.all(fileNames.map(
           async (fileName) => ({ stats: await fs.promises.stat(path.join(cfg.server.data_dir, fileName)), fileName }),
         )))
-        // filter out files that are not expired
-        .then((filesWithStats) => filesWithStats.reduce((acc, { stats, fileName }) => { 
-          if (stats.birthtime < Date.now() - cfg.storage.retention_days * MILLISECONDS_IN_A_DAY) acc.push(fileName)
+        // filter out files that are expired based on their epic's retention
+        .then((filesWithStats) => filesWithStats.reduce((acc, { stats, fileName }) => {
+          const days = retentionForFile(fileName, retentionByPrefix)
+          if (days >= 0 && stats.birthtime < Date.now() - days * MILLISECONDS_IN_A_DAY) acc.push(fileName)
           return acc
         }, []))
         // delete expired files

apps/metrics/src/effects/ndjson-cleaner.test.js

@@ -33,7 +33,10 @@ describe("ndjson-cleaner", () => {
 
     cfg = {
       server: { data_dir: "/app/data" },
-      storage: { retention_days: 30 },
+      epics: [
+        { name: "memory", file_prefix: "memory", data_retention_days: 30 },
+        { name: "cpu", file_prefix: "cpu", data_retention_days: 30 },
+      ],
     }
   })
 
@@ -69,8 +72,8 @@ describe("ndjson-cleaner", () => {
 
     it("keeps .ndjson files with recent birthtime", async () => {
       fs.promises.readdir.mockResolvedValue([
-        "notes.ndjson",
-        "backup.ndjson",
+        "memory_20250614.ndjson",
+        "cpu_20250614.ndjson",
       ])
       fs.promises.stat.mockResolvedValue({ birthtime: new Date("2025-06-14") })
 
@@ -101,11 +104,11 @@ describe("ndjson-cleaner", () => {
       // now = 2025-06-15T12:00:00.000Z, retention_days = 30 → cutoff = 2025-05-16T12:00:00.000Z
       // File with birthtime at cutoff should be kept, file before cutoff should be deleted
       fs.promises.readdir.mockResolvedValue([
-        "at_cutoff.ndjson",
-        "before_cutoff.ndjson",
+        "memory_20250516.ndjson",
+        "memory_20250516_1.ndjson",
       ])
       fs.promises.stat.mockImplementation((filePath) => {
-        if (filePath.endsWith("at_cutoff.ndjson")) {
+        if (filePath.endsWith("memory_20250516.ndjson")) {
           return Promise.resolve({ birthtime: new Date("2025-05-16T12:00:00.000Z") })
         }
         return Promise.resolve({ birthtime: new Date("2025-05-16T11:59:59.999Z") })
@@ -116,9 +119,25 @@ describe("ndjson-cleaner", () => {
 
       await vi.advanceTimersByTimeAsync(0)
 
-      // Only before_cutoff should be deleted (< cutoff), at_cutoff should be kept (== cutoff)
+      // Only the file before cutoff should be deleted (< cutoff), the one at cutoff should be kept (== cutoff)
       expect(fs.promises.unlink).toHaveBeenCalledTimes(1)
-      expect(fs.promises.unlink).toHaveBeenCalledWith("/app/data/before_cutoff.ndjson")
+      expect(fs.promises.unlink).toHaveBeenCalledWith("/app/data/memory_20250516_1.ndjson")
+
+      stopNdjsonCleaner()
+    })
+
+    it("skips files with no matching epic prefix", async () => {
+      fs.promises.readdir.mockResolvedValue([
+        "unknown_20250101.ndjson",
+      ])
+      fs.promises.stat.mockResolvedValue({ birthtime: new Date("2025-01-01") })
+
+      const { setupNdjsonCleaner, stopNdjsonCleaner } = await import("./ndjson-cleaner.js")
+      setupNdjsonCleaner(cfg)
+
+      await vi.advanceTimersByTimeAsync(0)
+
+      expect(fs.promises.unlink).not.toHaveBeenCalled()
 
       stopNdjsonCleaner()
     })
@@ -227,8 +246,8 @@ describe("ndjson-cleaner", () => {
 
     it("logs error when individual unlink fails", async () => {
       fs.promises.readdir.mockResolvedValue([
-        "old_a.ndjson",
-        "old_b.ndjson",
+        "memory_20250101.ndjson",
+        "cpu_20250101.ndjson",
       ])
       fs.promises.stat.mockResolvedValue({ birthtime: new Date("2025-01-01") })
       fs.promises.unlink

apps/metrics/src/init-collectors.js

@@ -21,50 +21,17 @@ const updateCollectorMeta = (name, patch) => {
   return next
 }
 
-// Metric files capacity weights per metric
-const CAPACITY_WEIGHTS = {
-  memory: 0.30,
-  monitor: 0.25,
-  commandlog_slow: 0.10,
-  commandlog_large_reply: 0.10,
-  commandlog_large_request: 0.10,
-  cpu: 0.10,
-  slowlog_len: 0.05,
-}
-
 const MIN_FILE_SIZE = 256 * 1024         // 256 KB
 const MAX_FILE_SIZE = 10 * 1024 * 1024  // 10 MB
 const MIN_FILES = 4
 
-const computeCapacity = (weight, totalMb) => {
-  const capacityBytes = weight * totalMb * 1024 * 1024
+const computeCapacity = (retentionMb) => {
+  const capacityBytes = retentionMb * 1024 * 1024
   const maxFileSize = Math.min(MAX_FILE_SIZE, Math.max(MIN_FILE_SIZE, Math.floor(capacityBytes / MIN_FILES)))
   const maxFiles = Math.max(MIN_FILES, Math.floor(capacityBytes / maxFileSize))
   return { maxFiles, maxFileSize }
 }
 
-const warnIfStorageTooSmall = (retentionSizeMb, weights) => {
-  const smallestWeight = Math.min(...Object.values(weights))
-  const minTotalBytes = (MIN_FILE_SIZE * MIN_FILES) / smallestWeight
-  const minTotalMb = Math.ceil(minTotalBytes / (1024 * 1024))
-  if (retentionSizeMb < minTotalMb) {
-    console.warn(
-      `retention_size_mb ${retentionSizeMb} is below minimum ${minTotalMb} for active epics — ` +
-      `per-metric capacity will be clamped to at least ${MIN_FILES} files × ${MIN_FILE_SIZE / 1024} KB`,
-    )
-  }
-}
-
-const normalizeWeights = (epics) => {
-  const activeEpicPrefixes = epics.map((e) => e.file_prefix || e.name)
-  const rawTotal = activeEpicPrefixes.reduce((sum, epicPrefix) => sum + (CAPACITY_WEIGHTS[epicPrefix] ?? 0), 0)
-  if (rawTotal === 0) {
-    const equal = 1 / activeEpicPrefixes.length
-    return Object.fromEntries(activeEpicPrefixes.map((p) => [p, equal]))
-  }
-  return Object.fromEntries(activeEpicPrefixes.map((p) => [p, (CAPACITY_WEIGHTS[p] ?? 0) / rawTotal]))
-}
-
 // Use it in endpoints to return metadata to server then to UI
 //  to show when the data was collected and will be refreshed
 export const getCollectorMeta = (name) => collectorsState[name]
@@ -75,18 +42,15 @@ updateCollectorMeta(MONITOR, {
   isRunning: false,
 })
 const startMonitor = (cfg) => {
-  const weights = normalizeWeights(cfg.epics)
-  warnIfStorageTooSmall(cfg.storage.retention_size_mb, weights)
-  const { maxFiles, maxFileSize } = computeCapacity(weights[MONITOR], cfg.storage.retention_size_mb)
+  const monitorEpic = cfg.epics.find((e) => e.name === MONITOR)
+  const { maxFiles, maxFileSize } = computeCapacity(monitorEpic.data_retention_mb)
   const nd = makeNdjsonWriter({
     dataDir: cfg.server.data_dir,
-    filePrefix: MONITOR,
+    filePrefix: monitorEpic.file_prefix || MONITOR,
     maxFiles,
     maxFileSize,
   })
 
-  const monitorEpic = cfg.epics.find((e) => e.name === MONITOR)
-
   const sink = {
     appendRows: async (rows) => {
       await nd.appendRows(rows)
@@ -145,15 +109,13 @@ const startMonitor = (cfg) => {
 const stopMonitor = async () => await monitorStopper()
 
 const setupCollectors = async (client, cfg) => {
-  const weights = normalizeWeights(cfg.epics)
-  warnIfStorageTooSmall(cfg.storage.retention_size_mb, weights)
   const fetcher = makeFetcher(client)
   await Promise.all(cfg.epics
     .filter((f) => f.name !== MONITOR && fetcher[f.type])
     .map(async (f) => {
       const fn = fetcher[f.type]
       const prefix = f.file_prefix || f.name
-      const { maxFiles, maxFileSize } = computeCapacity(weights[prefix], cfg.storage.retention_size_mb)
+      const { maxFiles, maxFileSize } = computeCapacity(f.data_retention_mb)
       const nd = makeNdjsonWriter({
         dataDir: cfg.server.data_dir,
         filePrefix: prefix,