Real-Time ML Model Monitoring at Scale

The Silent Killer: Model Drift

Our recommendation model was degrading. Click-through rates dropped 15% over 3 weeks. We only noticed when revenue metrics plummeted.

The problem? No one was watching the model’s health in real-time.

What We Built: MLWatch

A production monitoring platform that:

• Processes 5M prediction events/day from 50+ models
• Detects drift 48 hours before business impact
• Tracks performance, data quality, and bias
• Sends real-time alerts to Slack/PagerDuty

Architecture

Reactive Streams with Spring WebFlux + Kafka

@Service
public class PredictionStreamProcessor {
    
    @Autowired
    private ReactiveKafkaConsumerTemplate<String, PredictionEvent> consumer;
    
    public Flux<ModelMetrics> processPredictionStream() {
        return consumer
            .receiveAutoAck()
            .map(record -> record.value())
            // Window by model_id and time
            .windowTimeout(1000, Duration.ofMinutes(1))
            .flatMap(window -> window
                .collectList()
                .map(this::calculateMetrics)
            )
            // Detect anomalies
            .filter(this::isAnomalous)
            // Save to InfluxDB
            .flatMap(this::persistMetrics)
            // Send alerts
            .doOnNext(this::sendAlertIfNeeded);
    }
    
    private ModelMetrics calculateMetrics(List<PredictionEvent> events) {
        return ModelMetrics.builder()
            .modelId(events.get(0).getModelId())
            .predictionCount(events.size())
            .avgConfidence(calculateAvgConfidence(events))
            .distributionShift(detectDistributionShift(events))
            .build();
    }
}

Drift Detection Algorithm

public class DriftDetector {
    
    // Kullback-Leibler Divergence for distribution comparison
    public double calculateKLDivergence(
        Distribution baseline,
        Distribution current
    ) {
        double kl = 0.0;
        
        for (int i = 0; i < baseline.getBins().length; i++) {
            double p = baseline.getBins()[i];
            double q = current.getBins()[i];
            
            if (p > 0 && q > 0) {
                kl += p * Math.log(p / q);
            }
        }
        
        return kl;
    }
    
    // Population Stability Index (PSI)
    public double calculatePSI(
        Distribution baseline,
        Distribution current
    ) {
        double psi = 0.0;
        
        for (int i = 0; i < baseline.getBins().length; i++) {
            double expected = baseline.getBins()[i];
            double actual = current.getBins()[i];
            
            if (expected > 0 && actual > 0) {
                psi += (actual - expected) * Math.log(actual / expected);
            }
        }
        
        return psi;
    }
    
    // Alert thresholds
    public DriftSeverity assessDrift(double psi) {
        if (psi < 0.1) return DriftSeverity.NONE;
        if (psi < 0.2) return DriftSeverity.WARNING;
        return DriftSeverity.CRITICAL;
    }
}

Key Features

1. Real-Time Dashboard

WebSocket updates for live metrics:

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic");
        config.setApplicationDestinationPrefixes("/app");
    }
}

@Service
public class MetricsWebSocketPublisher {
    
    @Autowired
    private SimpMessagingTemplate messagingTemplate;
    
    public void publishMetrics(ModelMetrics metrics) {
        messagingTemplate.convertAndSend(
            "/topic/model/" + metrics.getModelId(),
            metrics
        );
    }
}

Frontend receives updates:

import SockJS from 'sockjs-client';
import { Client } from '@stomp/stompjs';

const client = new Client({
  webSocketFactory: () => new SockJS('/ws'),
  onConnect: () => {
    client.subscribe('/topic/model/recommender-v2', (message) => {
      const metrics = JSON.parse(message.body);
      updateDashboard(metrics);
    });
  }
});

2. Automated Alerting

@Service
public class AlertService {
    
    public void checkAndAlert(ModelMetrics metrics) {
        List<Alert> alerts = new ArrayList<>();
        
        // Check accuracy drop
        if (metrics.getAccuracy() < metrics.getBaselineAccuracy() * 0.95) {
            alerts.add(Alert.builder()
                .severity(Severity.HIGH)
                .title("Accuracy dropped 5%")
                .message(formatAccuracyDrop(metrics))
                .build());
        }
        
        // Check drift
        double psi = calculatePSI(metrics);
        if (psi > 0.2) {
            alerts.add(Alert.builder()
                .severity(Severity.CRITICAL)
                .title("Critical data drift detected")
                .message(formatDriftAlert(psi))
                .build());
        }
        
        // Send alerts
        alerts.forEach(alert -> {
            slackService.sendAlert(alert);
            pagerDutyService.createIncident(alert);
        });
    }
}

3. Time-Series Optimization

Using InfluxDB for fast queries:

@Repository
public class MetricsRepository {
    
    @Autowired
    private InfluxDBClient influxDB;
    
    public Flux<ModelMetrics> queryMetrics(
        String modelId,
        Duration timeRange
    ) {
        String query = String.format("""
            from(bucket: "ml_metrics")
              |> range(start: -%s)
              |> filter(fn: (r) => r["model_id"] == "%s")
              |> filter(fn: (r) => r["_measurement"] == "predictions")
              |> aggregateWindow(every: 5m, fn: mean)
            """, timeRange.toString(), modelId);
        
        return influxDB.getQueryReactiveApi()
            .query(query)
            .map(this::toModelMetrics);
    }
}

Performance Numbers

Metric	Value
Events/Day	5,000,000+
Models Monitored	50+
Query Latency (P95)	<100ms
Dashboard Update	Real-time (WebSocket)
Storage	30-day retention
Uptime	99.99%

Real Impact: Case Study

The Recommendation Model Incident

Timeline:

• Day 1: MLWatch detects PSI = 0.15 (WARNING)
• Day 2: PSI = 0.23 (CRITICAL) → Alert sent
• Day 3: Team investigates, finds new user segment
• Day 4: Model retrained with new data
• Day 5: Deployed, metrics back to normal

Without MLWatch: Would have taken 2-3 weeks to notice via business metrics

Prevented: Estimated $50K revenue loss

Lessons Learned

1. Monitor the Right Metrics: Accuracy alone isn’t enough. Track:

   • Prediction distribution
   • Feature distributions
   • Confidence scores
   • Latency
   • Data quality

2. Baseline Matters: Store production distributions as baselines for comparison

3. Alert Fatigue is Real: Tune thresholds carefully. We went from 50 alerts/day → 3/day

4. Reactive Streams Scale: WebFlux handled 5M events/day on 3 pods

Code Architecture

┌──────────────────┐
│  ML Model API    │
│  (predictions)   │
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  Kafka Topic     │
│  ml.predictions  │
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  Spring WebFlux  │
│  Consumer        │
└────────┬─────────┘
         │
    ┌────┴─────┐
    ▼          ▼
┌────────┐  ┌──────────┐
│InfluxDB│  │ Alerting │
│Metrics │  │ Service  │
└────┬───┘  └────┬─────┘
     │           │
     ▼           ▼
┌────────┐  ┌──────────┐
│Dashboard│  │  Slack   │
│WebSocket│  │PagerDuty │
└─────────┘  └──────────┘

Tech Stack

• Backend: Spring Boot 3, WebFlux
• Streaming: Apache Kafka, Kafka Streams
• Storage: InfluxDB (time-series), PostgreSQL
• Frontend: React, TypeScript, Recharts
• Real-time: WebSocket (STOMP)
• Deployment: Kubernetes, Helm
• Monitoring: Prometheus, Grafana

Try It

GitHub: github.com/vaibhav7k/mlwatch

---

Questions? DM me on LinkedIn