跳到主要内容

保留策略

数据保留策略允许您通过定义数据应保留多长时间来自动管理数据生命周期。

概述​

IEDB 中的保留策略可以帮助您：

  • 在数据库或测量表级别定义数据保留期限。
  • 通过手动执行/自动清理旧数据
  • 通过删除不必要的历史数据来降低存储成本
  • 遵守数据保留要求
  • 使用预运行模式安全地测试删除操作

工作原理​

IEDB 通过物理删除文件来实现数据保留策略：

  1. 扫描：检查测量目录中的 Parquet 文件
  2. 元数据分析：读取文件元数据以查找最大时间戳
  3. 识别：查找所有行都早于截止日期的文件
  4. 删除：从磁盘中物理移除整个文件。

截止日期计算：cutoff_date = today - retention_days - buffer_days

API 接口​

创建策略​

创建新的数据保留策略：

POST /api/v1/retention

body：

{
  "name": "delete_old_metrics",
  "database": "telegraf",
  "measurement": "cpu",
  "retention_days": 90,
  "buffer_days": 7,
  "is_active": true
}

参数：

  • name（字符串，必填）：唯一策略标识符
  • database（字符串，必填）：目标数据库名称
  • measurement（字符串，可选）：目标测量表名称（null 表示到数据库级别）
  • retention_days（整数，必填）：数据保留天数
  • buffer_days（整数，必填）：缓冲（天数）
  • is_active（布尔值，必填）：启用/禁用该策略

查询列表​

获取所有保留策略：

GET /api/v1/retention

响应：

[
  {
    "id": "550e8400-e29b-41d4-a716-446655440000",
    "name": "delete_old_metrics",
    "database": "telegraf",
    "measurement": "cpu",
    "retention_days": 90,
    "buffer_days": 7,
    "is_active": true,
    "created_at": "2024-01-15T10:30:00Z",
    "last_executed_at": "2024-01-20T02:00:00Z",
    "last_deleted_count": 1500
  }
]

策略详情​

获取具体的保留策略：

GET /api/v1/retention/{policy_id}

更新策略​

更新现有数据保留策略：

PUT /api/v1/retention/{policy_id}

body：与创建策略相同

删除策略​

移除保留策略：

DELETE /api/v1/retention/{policy_id}

执行策略​

手动触发保留策略：

POST /api/v1/retention/{policy_id}/execute

body：

{
  "dry_run": false,
  "confirm": true
}

预运行示例：

{
  "dry_run": true,
  "confirm": false
}

响应：

{
  "policy_id": "550e8400-e29b-41d4-a716-446655440000",
  "cutoff_date": "2023-10-22T00:00:00Z",
  "files_to_delete": [
    "/data/telegraf/cpu/2023-10-15.parquet",
    "/data/telegraf/cpu/2023-10-20.parquet"
  ],
  "total_files": 2,
  "dry_run": true
}

查看执行历史记录​

查看以往的保留策略执行情况：

GET /api/v1/retention/{policy_id}/executions?limit=50

响应：

[
  {
    "execution_id": "abc123",
    "executed_at": "2024-01-20T02:00:00Z",
    "deleted_count": 1500,
    "execution_time_ms": 2500,
    "status": "success"
  }
]

配置参数​

保留天数​

数据保留天数（超过天数后将被删除）。选择依据：

  • 业务需求
  • 合规法规
  • 存储容量
  • 查询模式

例如：retention_days: 90数据保留 90 天。

缓冲天数​

在保留期限上增加安全缓冲，以防止意外删除近期数据。

推荐值：

  • 开发周期：7天
  • 生产周期：14-30天

例如：使用 retention_days: 90和buffer_days: 7参数，将删除超过 97 天的数据。

保留级别​

数据库级别：

{
  "database": "telegraf",
  "measurement": null,
  "retention_days": 365
}

测量表级别：

{
  "database": "telegraf",
  "measurement": "cpu",
  "retention_days": 90
}

使用针对特定测量表的级别策略，对不同的数据级别进行精细控制。

使用示例​

示例 1：清理旧指标​

import requests

# 为旧的 CPU 指标创建保留策略
response = requests.post(
    "http://localhost:8000/api/v1/retention",
    headers={"Authorization": "Bearer YOUR_TOKEN"},
    json={
        "name": "cpu_cleanup",
        "database": "telegraf",
        "measurement": "cpu",
        "retention_days": 90,
        "buffer_days": 7,
        "is_active": True
    }
)

policy_id = response.json()["id"]

# 先执行试运行
dry_run = requests.post(
    f"http://localhost:8000/api/v1/retention/{policy_id}/execute",
    headers={"Authorization": "Bearer YOUR_TOKEN"},
    json={"dry_run": True, "confirm": False}
)

print(f"将删除 {dry_run.json()['total_files']} 个文件")

# 实际执行
if input("是否继续？(yes/no): ") == "yes":
    result = requests.post(
        f"http://localhost:8000/api/v1/retention/{policy_id}/execute",
        headers={"Authorization": "Bearer YOUR_TOKEN"},
        json={"dry_run": False, "confirm": True}
    )
    print(f"已删除 {result.json()['total_files']} 个文件")

示例 2：数据库级别的保留​

# Apply retention to all measurements in a database
response = requests.post(
    "http://localhost:8000/api/v1/retention",
    headers={"Authorization": "Bearer YOUR_TOKEN"},
    json={
        "name": "database_cleanup",
        "database": "telegraf",
        "measurement": None,  # Apply to all measurements
        "retention_days": 180,
        "buffer_days": 14,
        "is_active": True
    }
)

示例 3：列出和监控策略​

# List all policies
policies = requests.get(
    "http://localhost:8000/api/v1/retention",
    headers={"Authorization": "Bearer YOUR_TOKEN"}
)

for policy in policies.json():
    print(f"Policy: {policy['name']}")
    print(f"  Last executed: {policy['last_executed_at']}")
    print(f"  Last deleted: {policy['last_deleted_count']} rows")

    # Get execution history
    history = requests.get(
        f"http://localhost:8000/api/v1/retention/{policy['id']}/executions?limit=10",
        headers={"Authorization": "Bearer YOUR_TOKEN"}
    )
    print(f"  Recent executions: {len(history.json())}")

最佳实践​

1. 务必先测试。​

在执行保留策略之前，请使用试运行模式：

result = requests.post(
    f"http://localhost:8000/api/v1/retention/{policy_id}/execute",
    json={"dry_run": True, "confirm": False}
)

# Review what will be deleted
print(f"Files to delete: {result.json()['files_to_delete']}")

2. 预留缓冲天数​

设置安全缓冲时间以防止意外删除：

{
  "retention_days": 90,
  "buffer_days": 14  // 14-day
}

3. 保守设置​

先延长保存时间，然后逐渐缩短：

// Start here
{"retention_days": 365, "buffer_days": 30}

// After monitoring, reduce if needed
{"retention_days": 180, "buffer_days": 14}

4. 在非生产环境中进行测试​

在开发环境中创建并测试策略

5. 监控执行历史记录​

定期检查该last_deleted_count区域：

# 检查删除数量是否符合预期
policy = requests.get(f"/api/v1/retention/{policy_id}").json()
if policy['last_deleted_count'] > 10000:
    print("Warning: Large deletion detected!")

6. 使用针对特定指标的策略​

针对不同数据类型创建细粒度策略：

# 高频指标 - 保留时间较短
{"measurement": "cpu", "retention_days": 30}

# 业务指标 - 保留时间较长
{"measurement": "revenue", "retention_days": 730}

局限性​

仅限本地存储​

目前，数据保留策略仅适用于本地文件系统存储。云存储后端（S3、MinIO）尚未实现。

文件级粒度​

保留策略是在文件级别而非行级别上执行的。只有当所有行都早于截止日期时，才会删除该文件。

警告 为了确保数据保留策略的最佳效果，请确保您的数据已正确压缩。时间戳混杂的文件可能无法删除。

无回滚​

已删除的数据无法恢复。：

  1. 首先使用预运行模式。
  2. 维护关键数据的备份
  3. 在非生产环境中进行测试

顺序处理​

数据保留策略按顺序处理各项指标。大型数据库的处理可能需要一些时间。

压缩文件效果最佳​

当文件包含来自相似时间段的数据时，保留策略最为有效。启用自动压缩功能可获得更佳效果。

故障排除​

文件没被删除​

问题：试运行显示没有文件需要删除。

解决方案：

  • 检查数据是否确实存在，且早于retention_days + buffer_days
  • 确认该策略针对的是正确的数据库和衡量指标。
  • 确保文件完全早于截止日期（文件级别粒度）

策略未执行​

问题：手动执行时返回错误。

解决方案：

  • 确认策略is_active已设置为true
  • 检查是否confirm: true已设置为实际执行
  • 请确保您拥有数据目录的写入权限

文件数量异常​

问题：删除的文件数量与预期不符。

解决方案：

  • 请注意：只有所有行都早于截止时间的文件才会被删除。
  • ls -l使用测量目录检查文件时间戳
  • 查看近期可能导致文件合并的压缩活动

相关主题​