特斯拉级系统设计面试题（十六）：PB 级自动驾驶数据备份系统 — 增量备份、异地容灾与快速恢复

1
┌─────────────────────────────────────────────────────────────┐
2
│                  生产存储 (Production)                         │
3
│  - 高速 NVMe  - 实时写入  - 多副本                            │
4
└────────────────────────┬────────────────────────────────────┘
5
                         │
6
┌────────────────────────▼────────────────────────────────────┐
7
│                  备份层 (Backup)                              │
8
│  - 增量备份  - 块级去重  - 压缩                                │
9
└────────────────────────┬────────────────────────────────────┘
10
                         │
11
┌────────────────────────▼────────────────────────────────────┐
12
│                  跨区域复制 (Replication)                       │
13
│  - 同城双活  - 异地灾备                                       │
14
└────────────────────────┬────────────────────────────────────┘
15
                         │
16
┌────────────────────────▼────────────────────────────────────┐
17
│                  冷存储 (Cold Archive)                        │
18
│  - S3 Glacier  - 长期保存                                     │
19
└──────────────────────────────────────────────────────────────┘

1
/**
2
 * 数据生命周期管理
3
 */
4
@Service
5
class DataLifecycleService {
6
    /**
7
     * 数据分层规则
8
     */
9
    fun tierRules() = listOf(
10
        // 0-30 天：热存储（高频访问）
11
        TierRule(age = 0..30, storage = "HOT",
12
                 pricePerGB = 0.023,  // 美元
13
                 accessLatency = "ms"),
14
        // 30-180 天：温存储（中等访问）
15
        TierRule(age = 30..180, storage = "WARM",
16
                 pricePerGB = 0.0125,
17
                 accessLatency = "s"),
18
        // 180 天-3 年：冷存储（低频访问）
19
        TierRule(age = 180..1095, storage = "COLD",
20
                 pricePerGB = 0.004,
21
                 accessLatency = "min"),
22
        // 3 年+：归档存储（几乎不访问）
23
        TierRule(age = 1095..Int.MAX_VALUE, storage = "ARCHIVE",
24
                 pricePerGB = 0.00099,
25
                 accessLatency = "hour")
26
    )
27

28
    /**
29
     * 自动迁移
30
     */
31
    @Scheduled(cron = "0 0 2 * * *")  // 每天 2 点
32
    fun migrateData() {
33
        for (rule in tierRules()) {
34
            val candidates = dataRepo.findByAgeAndTier(
35
                ageMin = rule.age.first,
36
                ageMax = rule.age.last,
37
                currentTier = rule.storage.downgrade()
38
            )
39
            for (data in candidates) {
40
                storageService.moveTo(data, rule.storage)
41
            }
42
        }
43
    }
44
}

1
/**
2
 * 块级增量备份
3
 */
4
@Service
5
class IncrementalBackupService(
6
    private val chunkService: ChunkService,
7
    private val deduplicationService: DeduplicationService
8
) {
9
    /**
10
     * 块级差分备份
11
     */
12
    fun incrementalBackup(dataId: String, newData: ByteArray): BackupResult {
13
        // 1. 块级切分（4MB 一个块）
14
        val newChunks = chunkService.split(newData, chunkSize = 4 * 1024 * 1024)
15

16
        // 2. 去重（Content-Defined Chunking）
17
        val uniqueChunks = deduplicationService.deduplicate(dataId, newChunks)
18

19
        // 3. 计算差分（只备份新增的块）
20
        val existingChunks = metadataRepo.getChunks(dataId).toSet()
21
        val toBackup = newChunks.filter { it.hash !in existingChunks }
22

23
        // 4. 上传到备份存储
24
        for (chunk in toBackup) {
25
            backupStorage.put(chunk.hash, chunk.data)
26
        }
27

28
        // 5. 更新元数据
29
        metadataRepo.updateChunks(dataId, newChunks.map { it.hash })
30

31
        return BackupResult(
32
            originalSize = newData.size,
33
            backupSize = toBackup.sumOf { it.data.size },
34
            dedupRatio = 1.0 - toBackup.size.toDouble() / newChunks.size
35
        )
36
    }
37
}

1
/**
2
 * 跨区域异步复制
3
 */
4
@Service
5
class CrossRegionReplicationService(
6
    private val primaryStorage: ObjectStorage,
7
    private val secondaryStorage: ObjectStorage
8
) {
9
    /**
10
     * 异步复制（异地灾备）
11
     */
12
    @Scheduled(fixedRate = 60000)  // 1 分钟
13
    fun replicate() {
14
        // 1. 找出主区域待复制的数据
15
        val pending = primaryStorage.listPendingReplication()
16

17
        // 2. 批量复制到异地
18
        for (batch in pending.chunked(1000)) {
19
            val tasks = batch.map { dataId ->
20
                async {
21
                    val data = primaryStorage.get(dataId)
22
                    secondaryStorage.put(dataId, data)
23
                    replicationLog.recordReplicated(dataId)
24
                }
25
            }
26
            tasks.awaitAll()
27
        }
28
    }
29
}

1
/**
2
 * 灾难恢复服务
3
 */
4
@Service
5
class DisasterRecoveryService(
6
    private val backupStorage: BackupStorage,
7
    private val primaryStorage: ObjectStorage
8
) {
9
    companion object {
10
        // RPO = 1 小时（数据丢失容忍 1 小时）
11
        // RTO = 4 小时（恢复时间不超过 4 小时）
12
        private const val RPO_HOURS = 1
13
        private const val RTO_HOURS = 4
14
    }
15

16
    /**
17
     * 灾难恢复（主区域故障）
18
     */
19
    fun failoverToSecondary(): FailoverResult {
20
        log.warn("Initiating failover to secondary region")
21

22
        // 1. 提升灾备为主
23
        val startTime = Instant.now()
24
        primaryStorage.promoteSecondary()
25

26
        // 2. 切换流量
27
        trafficService.redirectToSecondary()
28

29
        // 3. 校验数据完整性
30
        val integrityCheck = backupStorage.verifyIntegrity()
31
        if (!integrityCheck.passed) {
32
            log.error("Integrity check failed: {}", integrityCheck.errors)
33
        }
34

35
        val rto = Duration.between(startTime, Instant.now())
36
        log.info("Failover completed in {} (RTO target: {} hours)", rto, RTO_HOURS)
37

38
        return FailoverResult(rto = rto)
39
    }
40

41
    /**
42
     * 季度灾难恢复演练
43
     */
44
    @Scheduled(cron = "0 0 3 1 1,4,7,10 *")  // 季度首月
45
    fun quarterlyDrill() {
46
        log.info("Starting quarterly DR drill")
47

48
        // 1. 在隔离环境模拟恢复
49
        val drillResult = simulateRecovery()
50

51
        // 2. 报告 RTO / RPO
52
        val report = DrillReport(
53
            rto = drillResult.actualRto,
54
            rpo = drillResult.actualRpo,
55
            targetRto = Duration.ofHours(RTO_HOURS.toLong()),
56
            targetRpo = Duration.ofHours(RPO_HOURS.toLong()),
57
            passed = drillResult.passed
58
        )
59

60
        // 3. 通知结果
61
        reportService.publish(report)
62
    }
63
}

1
// 备份时先做快照
2
val snapshot = storageService.createSnapshot(dataId)
3
// 然后基于快照做增量
4
incrementalBackupService.backupFromSnapshot(snapshot.id)

1
// 优先级：核心训练数据 > 普通数据
2
fun restore(priority: Priority) {
3
    when (priority) {
4
        Priority.CRITICAL -> parallelRestore(concurrency = 100)
5
        Priority.HIGH -> parallelRestore(concurrency = 50)
6
        Priority.NORMAL -> parallelRestore(concurrency = 10)
7
    }
8
}

1
// 每周校验一次
2
@Scheduled(cron = "0 0 3 * * SUN")
3
fun verifyIntegrity() {
4
    val samples = backupStorage.randomSample(0.01)  // 1% 抽样
5
    for (sample in samples) {
6
        val original = primaryStorage.get(sample.id)
7
        val restored = backupStorage.get(sample.id)
8
        if (original.hash != restored.hash) {
9
            alertService.send("BACKUP_CORRUPTED", sample.id)
10
        }
11
    }
12
}