comet-ml--opik/apps/opik-backend/docs/webhook-event-debouncing-implementation.md

Webhook Alert Debouncing Implementation

Overview

This document describes the implementation of webhook alert debouncing in the Opik backend. The system aggregates multiple alert events within a configurable time window and sends consolidated webhook notifications.

Architecture

Core Components

┌────────────────────────────────────────────────────────────────────────┐
│                          Alert Evaluation                               │
│  (Checks if events match alert config and adds to bucket)             │
└─────────────────────────────────┬──────────────────────────────────────┘
                                  │
                                  ↓
┌────────────────────────────────────────────────────────────────────────┐
│                        AlertBucketService                               │
│              (Manages alert event buckets in Redis)                    │
│                                                                         │
│  • addEventToBucket()     - Adds event to Redis bucket                │
│  • getBucketsReadyToProcess() - Returns buckets past debounce window  │
│  • getBucketEventIds()    - Retrieves aggregated event IDs            │
│  • deleteBucket()         - Removes processed bucket                   │
└─────────────────────────────────┬──────────────────────────────────────┘
                                  │
                                  ↓
┌────────────────────────────────────────────────────────────────────────┐
│                            AlertJob                                     │
│             (Scheduled job runs every 5 seconds)                       │
│                                                                         │
│  1. Check buckets ready to process                                     │
│  2. For each ready bucket:                                             │
│     • Retrieve bucket data (event IDs, workspace ID)                  │
│     • Fetch alert configuration from AlertService                     │
│     • Create payload with alert and event data                        │
│     • Publish webhook via WebhookPublisher                            │
│     • Delete bucket                                                     │
└─────────────────────────────────┬──────────────────────────────────────┘
                                  │
                                  ↓
┌────────────────────────────────────────────────────────────────────────┐
│                        WebhookPublisher                                 │
│              (Publishes webhooks to message queue)                     │
│                                                                         │
│  • publishWebhookEvent() - Publishes event to webhook queue           │
└────────────────────────────────────────────────────────────────────────┘

Separation of Concerns

Component	Responsibility
Alert Evaluation	Checks if event matches alert config, adds to bucket via AlertBucketService
AlertBucketService	Stores and retrieves alert event buckets in Redis with workspace context
AlertJob	Scheduled job that checks buckets, retrieves alert config from Alert Service, creates payload, publishes webhooks
WebhookPublisher	ONLY publishes webhook events - no knowledge of aggregation

Data Flow

1. Event Occurs

// Alert evaluation logic (to be implemented)
if (eventMatchesAlertConfig(event, alert)) {
    alertBucketService.addEventToBucket(
        alert.getId(),
        workspaceId,
        eventType,
        event.getId()
    ).subscribe();
}

2. Event Storage in Redis

Bucket Key Format: alert_bucket:{alertId}:{eventType}

Bucket Data Structure:

{
  "eventIds": "[\"event-1\", \"event-2\", \"event-3\"]",
  "firstSeen": "1704067200000",
  "windowSize": "60000",
  "workspaceId": "workspace-123"
}

Note: The bucket stores:

• windowSize: The debouncing window (in milliseconds) active when the bucket was created
• workspaceId: The workspace ID for retrieving alert configuration without RequestContext These fields ensure that configuration changes do not affect existing buckets and allow background jobs to access alert data.

3. Background Processing

AlertJob (runs every 5 seconds):

@Every("5s")
public class AlertJob extends Job {
    @Override
    public void doJob(JobExecutionContext context) {
        // Check buckets ready to process
        bucketService.getBucketsReadyToProcess()
            .flatMap(this::processBucket)
            .blockLast();
    }
}

4. Webhook Notification

When a bucket is ready:

private Mono<Void> processBucket(String bucketKey) {
    // 1. Parse alert ID and event type
    // 2. Retrieve alert configuration from AlertService
    // 3. Get aggregated event IDs from bucket
    // 4. Create consolidated webhook event
    // 5. Send via WebhookSubscriber
    // 6. Delete bucket
}

Configuration

Debouncing Settings

webhook:
  debouncing:
    enabled: true
    windowSize: 60 seconds    # Time to wait before sending consolidated notification
    bucketTtl: 3 minutes      # Bucket expiration time (safety cleanup)

Alert Configuration

Alert {
    id: UUID
    name: String
    enabled: Boolean
    webhook: {
        url: String
        headers: Map<String, String>
        secretToken: String
    }
    triggers: List<AlertTrigger> {
        eventType: AlertEventType
        triggerConfigs: List<TriggerConfig>
    }
}

Implementation Details

Redis Bucket Management

AlertBucketService:

public Mono<Void> addEventToBucket(
    UUID alertId,
    String workspaceId,
    AlertEventType eventType,
    String eventId) {
    
    String bucketKey = "alert_bucket:" + alertId + ":" + eventType.getValue();
    RMapReactive<String, String> bucket = redissonClient.getMap(bucketKey);
    long currentWindowSizeMillis = webhookConfig.getDebouncing().getWindowSize().toMilliseconds();
    
    // Add event ID to set
    // If first event in bucket:
    //   - Store firstSeen timestamp
    //   - Store windowSize (preserves config at creation time)
    //   - Store workspaceId (enables background job to retrieve alert)
    //   - Set bucket TTL (ONLY on first event, NOT refreshed on subsequent events)
    // Subsequent events:
    //   - DO NOT update windowSize or workspaceId (preserves original config and context)
    //   - DO NOT refresh TTL (bucket expires based on original TTL)
}

public Flux<String> getBucketsReadyToProcess() {
    // Scan all alert_bucket:* keys
    // For each bucket:
    //   - Retrieve firstSeen timestamp AND windowSize from bucket
    //   - If (now - firstSeen) >= stored windowSize
    //   - Return bucket key
    // Note: Uses the windowSize stored in the bucket, not current config
}

public Mono<BucketData> getBucketData(String bucketKey) {
    // Retrieves complete bucket data including:
    //   - eventIds: Set of aggregated event IDs
    //   - firstSeen: Timestamp when first event was added
    //   - windowSize: Debouncing window size (milliseconds)
    //   - workspaceId: Workspace ID for accessing alert configuration
}

Timestamp-Based Debouncing

Event Timeline:

t0: First event arrives → Store firstSeen = t0, windowSize = 60000ms, Set TTL = 3 minutes
t1: Second event arrives → Keep firstSeen = t0, Keep windowSize = 60000ms, Keep TTL (not refreshed)
t2: Third event arrives → Keep firstSeen = t0, Keep windowSize = 60000ms, Keep TTL (not refreshed)
...
t60: AlertJob runs → (t60 - t0) >= 60s → Send consolidated webhook
t60+: Bucket deleted by AlertJob after processing
     (If not deleted by AlertJob, Redis automatically expires bucket after TTL)

TTL Behavior:

• TTL is set ONLY when the first event is added to a new bucket
• Subsequent events to the same bucket do NOT refresh the TTL
• This prevents buckets from living indefinitely if events keep arriving
• The bucketTtl configuration (default: 3 minutes) is the safety cleanup mechanism
• After processing, buckets are explicitly deleted by AlertJob

Handling Configuration Changes

When the debouncing windowSize configuration is changed, the system handles existing and new buckets correctly:

Scenario:

1. Alert has debouncing window = 60 seconds
2. Bucket A is created at t0 with events
3. Configuration is changed to debouncing window = 120 seconds
4. New events for same alert arrive

Behavior:

• Bucket A (created with 60s window):

  • Continues to use its stored windowSize = 60000ms
  • Will be processed when (now - firstSeen) >= 60s
  • Not affected by configuration change

• New Bucket B (created after config change):

  • Stores new windowSize = 120000ms
  • Will be processed when (now - firstSeen) >= 120s
  • Uses the new configuration

Key Implementation Details:

1. addEventToBucket() stores windowSize only for the first event in a bucket
2. Subsequent events to the same bucket preserve the original windowSize
3. getBucketsReadyToProcess() reads the windowSize from each bucket (not from current config)
4. This ensures existing buckets are processed with their original configuration

Test Coverage: See AlertBucketServiceTest.java for comprehensive tests covering:

• Window size storage with first event
• Window size preservation with subsequent events
• Multiple buckets with different window sizes being processed independently
• Configuration changes creating new buckets while existing ones continue with original settings

Consolidated Webhook Payload

{
  "id": "alert-{alertId}-{uuid}",
  "eventType": "alert.fired",
  "alertId": "{alertId}",
  "alertName": "High Error Rate",
  "workspaceId": "{workspaceId}",
  "userName": "system",
  "url": "{webhookUrl}",
  "payload": {
    "alertId": "{alertId}",
    "alertName": "High Error Rate",
    "eventType": "trace:errors",
    "eventIds": ["event-1", "event-2", "event-3"],
    "eventCount": 3,
    "aggregationType": "consolidated",
    "message": "Alert 'High Error Rate': 3 trace:errors events aggregated"
  },
  "headers": {
    "X-Custom-Header": "value"
  }
}

Job Management

AlertJob Schedule

• Frequency: Every 5 seconds
• Concurrency: Disabled (@DisallowConcurrentExecution)
• Purpose: Check buckets and trigger webhooks for events past debouncing window

Job Execution Flow

1. getBucketsReadyToProcess() → Flux<String> of bucket keys
2. For each bucketKey:
   a. Parse alertId and eventType
   b. Retrieve Alert configuration
   c. Get aggregated event IDs
   d. Create consolidated webhook event
   e. Send via WebhookSubscriber
   f. Delete bucket
3. Error handling: Continue on error, log failures

Error Handling

Bucket Processing Errors

bucketService.getBucketsReadyToProcess()
    .flatMap(this::processBucket)
    .onErrorContinue((throwable, bucketKey) -> {
        log.error("Failed to process bucket '{}': {}",
            bucketKey, throwable.getMessage(), throwable);
    })
    .blockLast();

Alert Configuration Errors

// If alert is disabled
if (!alert.enabled()) {
    log.warn("Alert '{}' is disabled, skipping webhook", alert.id());
    return Mono.empty();
}

// If webhook configuration is missing
if (alert.webhook() == null || alert.webhook().url() == null) {
    log.error("Alert '{}' has no webhook configuration, skipping", alert.id());
    return Mono.empty();
}

Webhook Sending Errors

return webhookSubscriber.sendWebhook(webhookEvent)
    .doOnSuccess(__ -> log.info("Successfully sent webhook for alert '{}'", alert.id()))
    .doOnError(error -> log.error("Failed to send webhook for alert '{}': {}",
        alert.id(), error.getMessage(), error));

Testing

Unit Tests

@Test
void shouldAddEventToBucket() {
    // Test event addition to Redis bucket
}

@Test
void shouldReturnBucketsReadyToProcess() {
    // Test bucket readiness based on firstSeen timestamp
}

@Test
void shouldProcessBucketAndSendWebhook() {
    // Test AlertJob processing and webhook sending
}

@Test
void shouldHandleDisabledAlert() {
    // Test skipping webhooks for disabled alerts
}

Integration Tests

• Test complete flow from event to webhook
• Test bucket TTL and cleanup
• Test concurrent event additions to same bucket
• Test AlertJob execution with multiple ready buckets

Future Enhancements

1. Alert Configuration Management: Integration with Alert CRUD operations
2. Metrics: Track aggregation stats, publish latency, success/failure rates
3. Rate Limiting: Implement per-alert webhook rate limits
4. Batch Size Limits: Configurable maximum events per consolidated webhook

Key Files

Core Implementation

• AlertBucketService.java - Redis bucket management
• AlertJob.java - Scheduled job for processing buckets
• WebhookPublisher.java - Webhook event publisher
• WebhookConfig.java - Debouncing configuration
• AlertService.java - Alert configuration service (includes getByIdAndWorkspace)

Data Models

• Alert.java - Alert configuration
• AlertTrigger.java - Alert trigger configuration
• WebhookEvent.java - Webhook event structure
• WebhookEventTypes.java - Event type enum (includes ALERT_FIRED)

Services

• AlertService.java - Alert CRUD operations
• AlertDAO.java - Alert database access

Version History

Version 1.2 (Current)

• Workspace Context: Added workspaceId storage in buckets for background job access
  • AlertJob can now retrieve alert configuration without RequestContext
  • Added getByIdAndWorkspace() method to AlertService
• AlertJob Updates: Uses getBucketData() to retrieve complete bucket information
  • Fetches workspace ID from bucket data
  • Retrieves alert configuration using workspace context
• Documentation: Updated to use WebhookPublisher instead of WebhookSubscriber

Version 1.1

• Configuration Change Handling: Buckets now store their creation-time windowSize
  • Existing buckets continue to use their original debouncing window
  • New buckets created after configuration changes use the updated window
  • Added comprehensive test coverage in AlertBucketServiceTest.java
• Enhanced bucket data structure to include windowSize field
• Modified getBucketsReadyToProcess() to use stored window size instead of current config

Version 1.0 (2024-01-01)

• Initial implementation
• Timestamp-based debouncing
• 5-second AlertJob schedule
• WebhookPublisher only publishes events to queue
• AlertBucketService manages Redis storage with workspace context
• AlertJob orchestrates bucket processing, fetches alert data, and publishes webhooks
• Alert data is retrieved using workspaceId from bucket (no RequestContext needed)