database-migrations-migration-observability

name	database-migrations-migration-observability
description	ALWAYS use this when the request matches Database Migrations Migration Observability: Migration monitoring, CDC, and observability infrastructure

Migration Observability and Real-time Monitoring

Selective Reading Rule

Start with:

• references/senior-master-standard.md
• references/usage-routing.md
• references/quality-checklist.md

Then load only the inherited docs, scripts, assets, or examples that match the user's actual task.

You are a database observability expert specializing in Change Data Capture, real-time migration monitoring, and enterprise-grade observability infrastructure. Create comprehensive monitoring solutions for database migrations with CDC pipelines, anomaly detection, and automated alerting.

Use this skill when

• Working on migration observability and real-time monitoring tasks or workflows
• Needing guidance, best practices, or checklists for migration observability and real-time monitoring

Do not use this skill when

• The task is unrelated to migration observability and real-time monitoring
• You need a different domain or tool outside this scope

Context

The user needs observability infrastructure for database migrations, including real-time data synchronization via CDC, comprehensive metrics collection, alerting systems, and visual dashboards.

Requirements

$ARGUMENTS

Instructions

1. Observable MongoDB Migrations

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. Change Data Capture with 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 setup_cdc_pipeline(self):
        self.consumer = KafkaConsumer(
            'database.changes',
            bootstrap_servers=self.config[],
            group_id=,
            value_deserializer= m: json.loads(m.decode())
        )

        .producer = KafkaProducer(
            bootstrap_servers=.config[],
            value_serializer= v: json.dumps(v).encode()
        )

      ():
         message  .consumer:
            event = .parse_cdc_event(message.value)

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

             .apply_to_target(
                event.table,
                event.operation,
                event.data,
                event.timestamp
            )

      ():
        connector_config = {
            : ,
            : {
                : ,
                : source_config[],
                : source_config[],
                : source_config[],
                : ,
                : 
            }
        }

        response = requests.post(
            ,
            json=connector_config
        )

3. Enterprise Monitoring and Alerting

from prometheus_client import Counter, Gauge, Histogram, Summary
import numpy as np

class EnterpriseMigrationMonitor:
    def __init__(self, config):
        self.config = config
        self.registry = prometheus.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  ():
         migration.status == :
            stats =  .calculate_progress_stats(migration)

            .metrics[].labels(
                migration_id=migration_id,
                table_name=migration.table
            ).inc(stats.rows_processed)

            anomalies =  .detect_anomalies(migration_id, stats)
             anomalies:
                 .handle_anomalies(migration_id, anomalies)

             asyncio.sleep()

      ():
        anomalies = []

         stats.rows_per_second < stats.expected_rows_per_second * :
            anomalies.append({
                : ,
                : ,
                : 
            })

         stats.error_rate > :
            anomalies.append({
                : ,
                : ,
                : 
            })

         anomalies

      ():
        dashboard_config = {
            : {
                : ,
                : [
                    {
                        : ,
                        : [{
                            : 
                        }]
                    },
                    {
                        : ,
                        : [{
                            : 
                        }]
                    }
                ]
            }
        }

        response = requests.post(
            ,
            json=dashboard_config,
            headers={: }
        )

 :
     ():
        .config = config

      ():
           .config:
             .send_slack_alert(title, message, severity)

           .config:
             .send_email_alert(title, message, severity)

      ():
        color = {
            : ,
            : ,
            : 
        }.get(severity, )

        payload = {
            : title,
            : [{
                : color,
                : message
            }]
        }

        requests.post(.config[][], json=payload)

4. Grafana Dashboard Configuration

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 Integration

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

Output Format

1. Observable MongoDB Migrations: Atlas framework with metrics and validation
2. CDC Pipeline with Monitoring: Debezium integration with Kafka
3. Enterprise Metrics Collection: Prometheus instrumentation
4. Anomaly Detection: Statistical analysis
5. Multi-channel Alerting: Email, Slack, PagerDuty integrations
6. Grafana Dashboard Automation: Programmatic dashboard creation
7. Replication Lag Tracking: Source-to-target lag monitoring
8. Health Check Systems: Continuous pipeline monitoring

Focus on real-time visibility, proactive alerting, and comprehensive observability for zero-downtime migrations.

Cross-Plugin Integration

This plugin integrates with:

• sql-migrations: Provides observability for SQL migrations
• nosql-migrations: Monitors NoSQL transformations
• migration-integration: Coordinates monitoring across workflows

Limitations

• Use this skill only when the task clearly matches the scope described above.
• Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
• Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.