metricsSync.js

Go to the documentation of this file.

/**
 * Background Metrics Sync Worker
 * 
 * Runs every 30 minutes to sync DigitalOcean metrics to database for all tenants
 * Prevents rate limiting and provides instant page loads directly from cached data
 */
 
const cron = require('node-cron');
const { Pool } = require('pg');
const DigitalOceanService = require('../services/digitalOceanService');
 
const SYNC_INTERVAL = process.env.SYNC_INTERVAL || '*/30 * * * *'; // Every 30 minutes
 
// Database connection
const pool = new Pool({
    host: process.env.DB_HOST,
    port: process.env.DB_PORT,
    user: process.env.DB_USER,
    password: process.env.DB_PASSWORD,
    database:process.env.DB_NAME || 'defaultdb',
    ssl: { rejectUnauthorized: false }
});
 
console.log('[MetricsSync] Worker starting...');
console.log(`[MetricsSync] Database: ${process.env.DB_HOST}:${process.env.DB_PORT}`);
console.log(`[MetricsSync] Sync Interval: ${SYNC_INTERVAL} (every 30 minutes)`);
 
/**
 * Sync metrics for a single tenant from DigitalOcean.
 * Fetches database and droplet resources, retrieves metrics, and stores in database.
 * @function syncTenantMetrics
 * @param {number} tenantId - Tenant ID to sync metrics for
 * @returns {Promise<void>}
 */
async function syncTenantMetrics(tenantId) {
    try {
        const doService = new DigitalOceanService(tenantId);
        const resources = await doService.getAllResources();
 
        let syncStats = {
            databases: 0,
            droplets: 0,
            metrics: 0
        };
 
        // =====================
        // Sync Databases + Metrics
        // =====================
        const databases = resources.databases || [];
        const activeDatabaseIds = databases.map(db => db.id);
 
        for (const db of databases) {
            let metrics = null;
            try {
                metrics = await doService.getDatabaseMetrics(db.id);
                syncStats.metrics++;
            } catch (err) {
                console.error(`[MetricsSync] Failed to fetch metrics for database ${db.id}:`, err.message);
            }
 
            try {
                // Prepare metadata - use null if stringify fails
                let metadata = null;
                try {
                    metadata = JSON.stringify(db);
                } catch (jsonErr) {
                    console.error(`[MetricsSync] Failed to stringify metadata for database ${db.id}:`, jsonErr.message);
                    metadata = JSON.stringify({ id: db.id, name: db.name, engine: db.engine });
                }
 
                await pool.query(
                    `INSERT INTO hosting_databases (
                      tenant_id, do_database_id, database_name, engine, version, status,
                      region, size, num_nodes, connection_host, connection_port,
                      connection_database, connection_user, tags,
                      cpu_count, memory_mb, disk_gb, metrics_last_updated,
                      metadata, created_at, updated_at
                    )
                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
                    ON CONFLICT (do_database_id)
                    DO UPDATE SET
                      database_name = EXCLUDED.database_name,
                      status = EXCLUDED.status,
                      version = EXCLUDED.version,
                      size = EXCLUDED.size,
                      num_nodes = EXCLUDED.num_nodes,
                      connection_host = EXCLUDED.connection_host,
                      connection_port = EXCLUDED.connection_port,
                      cpu_count = EXCLUDED.cpu_count,
                      memory_mb = EXCLUDED.memory_mb,
                      disk_gb = EXCLUDED.disk_gb,
                      metrics_last_updated = EXCLUDED.metrics_last_updated,
                      metadata = EXCLUDED.metadata,
                      updated_at = CURRENT_TIMESTAMP`,
                    [
                        tenantId, db.id, db.name, db.engine, db.version, db.status,
                        db.region, db.size, db.num_nodes || 1,
                        db.connection?.host || null, db.connection?.port || null,
                        db.connection?.database || null, db.connection?.user || null,
                        db.tags || [],
                        metrics?.cpu_count || null, metrics?.memory_mb || null, metrics?.disk_gb || null,
                        metrics ? new Date() : null,
                        metadata
                    ]
                );
 
                if (metrics) {
                    await pool.query(
                        `INSERT INTO hosting_metrics_history (
                          resource_type, resource_id, tenant_id, cpu_value, memory_value, disk_value
                        ) VALUES ($1, $2, $3, $4, $5, $6)`,
                        ['database', db.id, tenantId, metrics.cpu_count, metrics.memory_mb, metrics.disk_gb]
                    );
                }
 
                syncStats.databases++;
            } catch (err) {
                console.error(`[MetricsSync] Failed to sync database ${db.id} (${db.name}):`, err.message);
            }
        }
 
        // Mark databases as deleted if they no longer exist
        if (activeDatabaseIds.length > 0) {
            await pool.query(
                `UPDATE hosting_databases 
                 SET status = 'deleted', updated_at = CURRENT_TIMESTAMP
                 WHERE tenant_id = $1 
                 AND do_database_id NOT IN (${activeDatabaseIds.map((_, i) => `$${i + 2}`).join(', ')})
                 AND status != 'deleted'`,
                [tenantId, ...activeDatabaseIds]
            );
        }
 
        // =====================
        // Sync Droplets + Metrics
        // =====================
        const droplets = resources.droplets || [];
        const activeDropletIds = droplets.map(d => d.id.toString());
 
        for (const droplet of droplets) {
            let cpuMetrics = null;
            let avgCpu = null;
            
            try {
                cpuMetrics = await doService.getDropletMetrics(droplet.id, 'cpu');
                syncStats.metrics++;
 
                if (cpuMetrics?.data?.result?.[0]?.values) {
                    const values = cpuMetrics.data.result[0].values;
                    const sum = values.reduce((acc, [, value]) => acc + parseFloat(value), 0);
                    avgCpu = values.length > 0 ? (sum / values.length).toFixed(2) : null;
                }
            } catch (err) {
                console.error(`[MetricsSync] Failed to fetch metrics for droplet ${droplet.id}:`, err.message);
            }
 
            try {
                // Prepare metadata - use null if stringify fails
                let metadata = null;
                try {
                    metadata = JSON.stringify(droplet);
                } catch (jsonErr) {
                    console.error(`[MetricsSync] Failed to stringify metadata for droplet ${droplet.id}:`, jsonErr.message);
                    metadata = JSON.stringify({ id: droplet.id, name: droplet.name, status: droplet.status });
                }
 
                await pool.query(
                    `INSERT INTO hosting_droplets (
                      tenant_id, do_droplet_id, droplet_name, status, region, size,
                      ip_address, ipv6_address, vcpus, memory, disk, image, tags,
                      cpu_usage_percent, metrics_last_updated,
                      metadata, created_at, updated_at
                    )
                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
                    ON CONFLICT (do_droplet_id)
                    DO UPDATE SET
                      droplet_name = EXCLUDED.droplet_name,
                      status = EXCLUDED.status,
                      size = EXCLUDED.size,
                      ip_address = EXCLUDED.ip_address,
                      vcpus = EXCLUDED.vcpus,
                      memory = EXCLUDED.memory,
                      disk = EXCLUDED.disk,
                      cpu_usage_percent = EXCLUDED.cpu_usage_percent,
                      metrics_last_updated = EXCLUDED.metrics_last_updated,
                      metadata = EXCLUDED.metadata,
                      updated_at = CURRENT_TIMESTAMP`,
                    [
                        tenantId, droplet.id.toString(), droplet.name, droplet.status,
                        droplet.region.slug, droplet.size_slug,
                        droplet.networks?.v4?.[0]?.ip_address || null,
                        droplet.networks?.v6?.[0]?.ip_address || null,
                        droplet.vcpus, droplet.memory, droplet.disk,
                        droplet.image?.slug || droplet.image?.name || null,
                        droplet.tags || [], avgCpu, avgCpu ? new Date() : null,
                        metadata
                    ]
                );
 
                if (avgCpu) {
                    await pool.query(
                        `INSERT INTO hosting_metrics_history (
                          resource_type, resource_id, tenant_id, cpu_value
                        ) VALUES ($1, $2, $3, $4)`,
                        ['droplet', droplet.id.toString(), tenantId, avgCpu]
                    );
                }
 
                syncStats.droplets++;
            } catch (err) {
                console.error(`[MetricsSync] Failed to sync droplet ${droplet.id} (${droplet.name}):`, err.message);
            }
        }
 
        // Mark droplets as deleted if they no longer exist
        if (activeDropletIds.length > 0) {
            await pool.query(
                `UPDATE hosting_droplets 
                 SET status = 'deleted', updated_at = CURRENT_TIMESTAMP
                 WHERE tenant_id = $1 
                 AND do_droplet_id NOT IN (${activeDropletIds.map((_, i) => `$${i + 2}`).join(', ')})
                 AND status != 'deleted'`,
                [tenantId, ...activeDropletIds]
            );
        }
 
        return syncStats;
    } catch (error) {
        console.error(`[MetricsSync] Error syncing tenant ${tenantId}:`, error.message);
        throw error;
    }
}
 
/**
 * Sync all tenants
 */
async function syncAllTenants() {
    const startTime = Date.now();
    console.log(`[MetricsSync] Starting metrics sync at ${new Date().toISOString()}`);
 
    try {
        // Get all tenants
        const tenantsResult = await pool.query(
            `SELECT tenant_id FROM tenants`
        );
        const tenants = tenantsResult.rows;
 
        console.log(`[MetricsSync] Syncing ${tenants.length} tenant(s)`);
 
        let totalStats = { databases: 0, droplets: 0, metrics: 0 };
 
        for (const tenant of tenants) {
            try {
                const stats = await syncTenantMetrics(tenant.tenant_id);
                totalStats.databases += stats.databases;
                totalStats.droplets += stats.droplets;
                totalStats.metrics += stats.metrics;
                console.log(`[MetricsSync] Tenant ${tenant.tenant_id}: ${JSON.stringify(stats)}`);
            } catch (err) {
                console.error(`[MetricsSync] Failed to sync tenant ${tenant.tenant_id}:`, err.message);
            }
        }
 
        const duration = ((Date.now() - startTime) / 1000).toFixed(2);
        console.log(`[MetricsSync] ✓ Sync completed in ${duration}s - Total:`, totalStats);
 
    } catch (error) {
        const duration = ((Date.now() - startTime) / 1000).toFixed(2);
        console.error(`[MetricsSync] ✗ Sync failed after ${duration}s:`, error.message);
    }
}
 
// Run sync every 30 minutes
cron.schedule(SYNC_INTERVAL, syncAllTenants);
 
// Run sync immediately on startup
(async () => {
    console.log('[MetricsSync] Running initial sync...');
    await syncAllTenants();
    console.log('[MetricsSync] Worker ready. Next sync in 30 minutes.');
})();
 
// Graceful shutdown
process.on('SIGINT', () => {
    console.log('[MetricsSync] Shutting down...');
    pool.end();
    process.exit(0);
});
 
process.on('SIGTERM', () => {
    console.log('[MetricsSync] Shutting down...');
    pool.end();
    process.exit(0);
});