database-migrations-migration-observability

name	database-migrations-migration-observability
description	Migration monitoring, CDC, and observability infrastructure
category	Document Processing
source	antigravity
tags	["python","javascript","api","ai","automation","workflow","document","cro"]
url	https://github.com/sickn33/antigravity-awesome-skills/tree/main/skills/database-migrations-migration-observability

Migration Observability and Real-time Monitoring

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', ],
                : [, , , , , ],
                : [register]
            }),
            :  prometheus.({
                : ,
                : ,
                : [, ],
                : [register]
            }),
            :  prometheus.({
                : ,
                : ,
                : [, ],
                : [register]
            }),
            register
        };
    }

     () {
         ..();
         db = ..();

         ( [version, migration]  .) {
             .(db, version, migration);
        }
    }

     () {
         timer = ...({ version });
         session = ..();

         {
            ..();

             session.( () => {
                 migration.(db, session,  {
                    ...({
                        version,
                        collection
                    }, count);
                });
            });

            ({ :  });
            ..();

        }  (error) {
            ...({
                version,
                : error.
            });
            ({ :  });
             error;
        }  {
             session.();
        }
    }
}

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=,