database-migrations-migration-observability

移行の可観測性とリアルタイム監視

あなたは、変更データキャプチャ、リアルタイムの移行監視、エンタープライズグレードの可観測性インフラストラクチャを専門とする、データベース可観測性のエキスパートです。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
import requests
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 setup_cdc_pipeline(self):
        self.consumer = KafkaConsumer(
            'database.changes',
            bootstrap_servers=self.config['kafka_brokers'],
            group_id='migration-consumer',
            enable_auto_commit=False,
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )

        self.producer = KafkaProducer(
            bootstrap_servers=self.config['kafka_brokers'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )

    async def process_cdc_events(self):
        for message in self.consumer:
            event = self.parse_cdc_event(message.value)

            self.metrics['events_processed'].labels(
                source=event.source_db,
                table=event.table,
                operation=event.operation
            ).inc()

            await self.apply_to_target(
                event.table,
                event.operation,
                event.data,
                event.timestamp
            )
            self.consumer.commit() # only after durable idempotent target application

    async def setup_debezium_connector(self, source_config):
        connector_config = {
            "name": f"migration-connector-{source_config['name']}",
            "config": {
                "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
                "database.hostname": source_config['host'],
                "database.port": source_config['port'],
                "database.dbname": source_config['database'],
                "plugin.name": "pgoutput",
                "heartbeat.interval.ms": "10000"
            }
        }

        response = requests.post(
            f"{self.config['kafka_connect_url']}/connectors",
            json=connector_config
        )

3. エンタープライズ監視とアラート

from prometheus_client import Counter, Gauge, Histogram, Summary, CollectorRegistry
import asyncio
import requests
import numpy as np

class EnterpriseMigrationMonitor:
    def __init__(self, config):
        self.config = config
        self.registry = CollectorRegistry()
        self.metrics = self.setup_metrics()
        self.alerting = AlertingSystem(config.get('alerts', {}))

    def setup_metrics(self):
        return {
            'migration_duration': Histogram(
                'migration_duration_seconds',
                'Migration duration',
                ['migration_id'],
                buckets=[60, 300, 600, 1800, 3600],
                registry=self.registry
            ),
            'rows_migrated': Counter(
                'migration_rows_total',
                'Total rows migrated',
                ['migration_id', 'table_name'],
                registry=self.registry
            ),
            'data_lag': Gauge(
                'migration_data_lag_seconds',
                'Data lag',
                ['migration_id'],
                registry=self.registry
            )
        }

    async def track_migration_progress(self, migration_id, migration):
        while migration.status == 'running':
            stats = await self.calculate_progress_stats(migration)

            self.metrics['rows_migrated'].labels(
                migration_id=migration_id,
                table_name=migration.table
            ).inc(stats.rows_processed_delta) # increment only the new rows since the last sample

            anomalies = await self.detect_anomalies(migration_id, stats)
            if anomalies:
                await self.handle_anomalies(migration_id, anomalies)

            await asyncio.sleep(30)

    async def detect_anomalies(self, migration_id, stats):
        anomalies = []

        if stats.rows_per_second < stats.expected_rows_per_second * 0.5:
            anomalies.append({
                'type': 'low_throughput',
                'severity': 'warning',
                'message': f'Throughput below expected'
            })

        if stats.error_rate > 0.01:
            anomalies.append({
                'type': 'high_error_rate',
                'severity': 'critical',
                'message': f'Error rate exceeds threshold'
            })

        return anomalies

    async def setup_migration_dashboard(self):
        dashboard_config = {
            "dashboard": {
                "title": "Database Migration Monitoring",
                "panels": [
                    {
                        "title": "Migration Progress",
                        "targets": [{
                            "expr": "rate(migration_rows_total[5m])"
                        }]
                    },
                    {
                        "title": "Data Lag",
                        "targets": [{
                            "expr": "migration_data_lag_seconds"
                        }]
                    }
                ]
            }
        }

        response = requests.post(
            f"{self.config['grafana_url']}/api/dashboards/db",
            json=dashboard_config,
            headers={'Authorization': f"Bearer {self.config['grafana_token']}"}
        )

class AlertingSystem:
    def __init__(self, config):
        self.config = config

    async def send_alert(self, title, message, severity, **kwargs):
        if 'slack' in self.config:
            await self.send_slack_alert(title, message, severity)

        if 'email' in self.config:
            await self.send_email_alert(title, message, severity)

    async def send_slack_alert(self, title, message, severity):
        color = {
            'critical': 'danger',
            'warning': 'warning',
            'info': 'good'
        }.get(severity, 'warning')

        payload = {
            'text': title,
            'attachments': [{
                'color': color,
                'text': message
            }]
        }

        requests.post(self.config['slack']['webhook_url'], json=payload)

4. Grafana ダッシュボードの設定

dashboard_panels = [
    {
        "id": 1,
        "title": "Migration Progress",
        "type": "graph",
        "targets": [{
            "expr": "rate(migration_rows_total[5m])",
            "legendFormat": "{{migration_id}} - {{table_name}}"
        }]
    },
    {
        "id": 2,
        "title": "Data Lag",
        "type": "stat",
        "targets": [{
            "expr": "migration_data_lag_seconds"
        }],
        "fieldConfig": {
            "thresholds": {
                "steps": [
                    {"value": 0, "color": "green"},
                    {"value": 60, "color": "yellow"},
                    {"value": 300, "color": "red"}
                ]
            }
        }
    },
    {
        "id": 3,
        "title": "Error Rate",
        "type": "graph",
        "targets": [{
            "expr": "rate(migration_errors_total[5m])"
        }]
    }
]

5. CI/CD との統合

name: Migration Monitoring

on:
  push:
    branches: [main]

jobs:
  monitor-migration:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Start Monitoring
        run: |
          python migration_monitor.py start \
            --migration-id ${{ github.sha }} \
            --prometheus-url ${{ secrets.PROMETHEUS_URL }}

      - name: Run Migration
        run: |
          python migrate.py --environment production

      - name: Check Migration Health
        run: |
          python migration_monitor.py check \
            --migration-id ${{ github.sha }} \
            --max-lag 300

出力形式

1. 可観測性を備えた MongoDB の移行：メトリクスと検証を含む Atlas フレームワーク
2. 監視機能を備えた CDC パイプライン：Kafka との Debezium 統合
3. エンタープライズ向けメトリクス収集：Prometheus による計測
4. 異常検知：統計分析
5. マルチチャネルのアラート：メール、Slack、PagerDuty との統合
6. Grafana ダッシュボードの自動化：プログラムによるダッシュボード作成
7. レプリケーション遅延の追跡：ソースからターゲットまでの遅延監視
8. ヘルスチェックシステム：パイプラインの継続的な監視

リアルタイムの可視性、プロアクティブなアラート、包括的な可観測性に重点を置き、移行の判断を支援してください。可観測性によってダウンタイムがゼロになる保証はありません。

プラグイン間の統合

このプラグインは、次のプラグインと統合します。

• sql-migrations：SQL の移行に可観測性を提供します
• nosql-migrations：NoSQL の変換を監視します
• migration-integration：ワークフロー全体の監視を調整します

制限事項

• タスクが上記の説明範囲に明確に該当する場合にのみ、このスキルを使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全性の境界、または成功基準が不足している場合は、作業を中断して確認を求めてください。

以下は、移行、パーサー、エクスポーター、アラートアダプターなどを省略した、不完全なアプリケーションの概略です。CI スクリプトは同梱されていません。MongoDB の withTransaction はコールバックを再試行することがあります。その内部の外部カウンターやログは、コミットされた行数ではなく試行回数を数えるため、取り消し不能な副作用にはアウトボックスや冪等性が必要です。ターゲットへの適用後に Kafka の手動コミットを行う方式は、少なくとも1回の配信を実現するものであり、厳密に1回の配信を実現するものではありません。トゥームストーン、スナップショット、順序、永続的な重複排除に対応してください。同期的な Kafka の反復処理やリクエストは、非同期ループをブロックします。対応済みの非同期クライアントまたはワーカースレッド、タイムアウト、ステータスチェックを使用してください。Debezium には、バージョン固有のトピックプレフィックス、認証情報、レプリケーションスロット／パブリケーション、保持設定が必要です。ダッシュボードのエラーカウンターは、実際に出力されている必要があります。rate は1秒あたりのエラー数であり、分母なしの割合ではありません。上限のあるラベル、保護されたメトリクスエンドポイント、機密情報を除去したログを設定してください。本番環境の移行、コネクターのプロビジョニング、アラートメッセージの送信には、許可されたターゲットが必要です。CI を実行するだけでは、デプロイ権限は得られません。