---

name: database-migrations-migration-observability description: "迁移监控、CDC 和可观测性基础设施" risk: unknown source: community tags: "database, cdc, debezium, kafka, prometheus, grafana, monitoring" date_added: "2026-02-27"

迁移可观测性与实时监控

你是一位数据库可观测性专家，专注于变更数据捕获（CDC）、实时迁移监控和企业级可观测性基础设施。为数据库迁移创建包含 CDC 管道、异常检测和自动告警的综合监控方案。

适用场景

• 处理迁移可观测性和实时监控相关的任务或工作流
• 需要迁移可观测性和实时监控的指导、最佳实践或检查清单

不适用场景

• 任务与迁移可观测性和实时监控无关
• 需要此范围之外的其他领域或工具

上下文

用户需要数据库迁移的可观测性基础设施，包括通过 CDC 实现实时数据同步、全面的指标采集、告警系统和可视化仪表板。

需求

$ARGUMENTS

操作指南

1. 可观测的 MongoDB 迁移

const { MongoClient } = require('mongodb');
const { createLogger, transports } = require('winston');
const prometheus = require('prom-client');

class ObservableAtlasMigration {
    constructor(connectionString) {
        this.client = new MongoClient(connectionString);
        this.logger = createLogger({
            transports: [
                new transports.File({ filename: 'migrations.log' }),
                new transports.Console()
            ]
        });
        this.metrics = this.setupMetrics();
    }

    setupMetrics() {
        const register = new prometheus.Registry();

        return {
            migrationDuration: new prometheus.Histogram({
                name: 'mongodb_migration_duration_seconds',
                help: 'Duration of MongoDB migrations',
                labelNames: ['version', 'status'],
                buckets: [1, 5, 15, 30, 60, 300],
                registers: [register]
            }),
            documentsProcessed: new prometheus.Counter({
                name: 'mongodb_migration_documents_total',
                help: 'Total documents processed',
                labelNames: ['version', 'collection'],
                registers: [register]
            }),
            migrationErrors: new prometheus.Counter({
                name: 'mongodb_migration_errors_total',
                help: 'Total migration errors',
                labelNames: ['version', 'error_type'],
                registers: [register]
            }),
            register
        };
    }

    async migrate() {
        await this.client.connect();
        const db = this.client.db();

        for (const [version, migration] of this.migrations) {
            await this.executeMigrationWithObservability(db, version, migration);
        }
    }

    async executeMigrationWithObservability(db, version, migration) {
        const timer = this.metrics.migrationDuration.startTimer({ version });
        const session = this.client.startSession();

        try {
            this.logger.info(`Starting migration ${version}`);

            await session.withTransaction(async () => {
                await migration.up(db, session, (collection, count) => {
                    this.metrics.documentsProcessed.inc({
                        version,
                        collection
                    }, count);
                });
            });

            timer({ status: 'success' });
            this.logger.info(`Migration ${version} completed`);

        } catch (error) {
            this.metrics.migrationErrors.inc({
                version,
                error_type: error.name
            });
            timer({ status: 'failed' });
            throw error;
        } finally {
            await session.endSession();
        }
    }
}

2. 使用 Debezium 进行变更数据捕获

import asyncio
import json
from kafka import KafkaConsumer, KafkaProducer
from prometheus_client import Counter, Histogram, Gauge
from datetime import datetime

class CDCObservabilityManager:
    def __init__(self, config):
        self.config = config
        self.metrics = self.setup_metrics()

    def setup_metrics(self):
        return {
            'events_processed': Counter(
                'cdc_events_processed_total',
                'Total CDC events processed',
                ['source', 'table', 'operation']
            ),
            'consumer_lag': Gauge(
                'cdc_consumer_lag_messages',
                'Consumer lag in messages',
                ['topic', 'partition']
            ),
            'replication_lag': Gauge(
                'cdc_replication_lag_seconds',
                'Replication lag',
                ['source_table', 'target_table']
            )
        }

    async def consume_cdc_events(self):
        consumer = KafkaConsumer(
            self.config['topic'],
            bootstrap_servers=self.config['kafka_brokers'],
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )

        for message in consumer:
            event = message.value
            self.metrics['events_processed'].labels(
                source=event.get('source', 'unknown'),
                table=event.get('table', 'unknown'),
                operation=event.get('op', 'unknown')
            ).inc()

            lag = (datetime.now() - datetime.fromtimestamp(
                event.get('ts_ms', 0) / 1000
            )).total_seconds()
            self.metrics['replication_lag'].labels(
                source_table=event.get('table', 'unknown'),
                target_table=event.get('target', 'unknown')
            ).set(lag)