Traffic Concurrency Refresh Service

Overview

The Traffic Concurrency Refresh Service is a background process that computes the measured peak concurrent API calls for each hour. It provides accurate, sub-hour peak concurrency data for capacity dashboards, which is more precise than statistical estimates.

This service functions as an offline curator. It reads from a raw request log within a bounded, trailing time window, calculates the peak concurrency for each hour in that window, and stores the result in an aggregate table. The capacity dashboard and other observability tools then read from this pre-computed aggregate table, ensuring they do not need to process the full raw log.

The service is designed to be robust and fail-soft. Any error encountered while processing a specific time window or tenant is logged without interrupting the overall platform or subsequent processing tasks.

How it works

The service's operation is driven by an internal scheduler and is executed on a per-tenant basis to ensure data isolation.

flowchart TD
    subgraph Scheduled Job
        A[Scheduler] -- Triggers every `delay-ms` --> B{Service Enabled?};
    end

    B -- Yes --> C[For each tenant];
    B -- No --> D[Stop];

    subgraph "Tenant-Specific Refresh (refresh)"
        C --> E[Iterate `window-hours`];
        E --> F[Query raw request log for one hour];
        F --> G[Calculate peak concurrency];
        G --> H[Upsert aggregate to `s2r_traffic_concurrency_hourly`];
        H --> E;
        E -- All hours processed --> I[Prune old aggregates];
    end

    subgraph "Error Handling"
        style G fill:#fff,stroke:#c00,stroke-width:2px
        style I fill:#fff,stroke:#c00,stroke-width:2px
        G -- Fails --> J[Log warning and continue];
        I -- Fails --> K[Log warning and finish];
    end

The process follows these steps:

1. Trigger: A scheduler initiates the scheduledRefresh job after an initial delay and then at a fixed interval. The job only proceeds if the service is enabled via the s2r.capacity.concurrency-refresh.enabled property.
2. Per-Tenant Execution: The job runs independently for each tenant. This scopes all database reads and writes to that tenant's data. If the refresh for one tenant fails, the failure is logged, and the process continues to the next tenant.
3. Time Window Processing: The refresh method iterates through a trailing window of hours, defined by s2r.capacity.concurrency-refresh.window-hours. For each hour in this window, it performs the following calculation: a. Data Fetch: The refreshHour method queries the s2r_f5_request_log table to retrieve all requests that started within that specific hour. It selects the observed_at (start time) and backend_latency_ms (duration) for each request. b. Peak Calculation: Each request is treated as a time interval. The service uses a sweep-line algorithm (PeakConcurrency.peak) to find the maximum number of overlapping intervals, which represents the true peak number of concurrent calls during that hour. c. Store Aggregate: The calculated peak value and the total number of requests (sample_count) are written to the s2r_traffic_concurrency_hourly table. An INSERT ... ON CONFLICT statement is used to either create a new row for the hour or update the existing one. This ensures that data for recent hours remains fresh as more requests are logged.
4. Data Pruning: After processing all hours in the window, the service deletes records from s2r_traffic_concurrency_hourly that are older than the configured retention period (s2r.capacity.concurrency-refresh.retention-days).

Because the service re-computes every hour within the window-hours on each run, the concurrency data for the current (partial) hour and recently completed hours is kept up-to-date. Once an hour ages out of the window, its calculated peak is considered final.

Configuration

The service is configured through the following properties.

Database Operations

The service interacts with the database using several key SQL queries. These examples illustrate the core logic of fetching raw data and storing aggregates.

Fetching Request Intervals

To calculate concurrency for a given hour, the service queries the raw request log. The ? placeholders correspond to the start and end Timestamp of the hour being processed.

select observed_at, backend_latency_ms from s2r_f5_request_log
where observed_at >= ? and observed_at < ? and backend_latency_ms is not null

Storing Hourly Aggregates

After calculating the peak concurrency, the service inserts or updates the corresponding record in the aggregate table. This operation is atomic.

insert into s2r_traffic_concurrency_hourly
    (hour_bucket, source, peak_concurrent_calls, sample_count, updated_at)
values (?, ?, ?, ?, now())
on conflict (hour_bucket, source) do update set
    peak_concurrent_calls = excluded.peak_concurrent_calls,
    sample_count = excluded.sample_count,
    updated_at = now()

The placeholders correspond to:

1. hour_bucket: The start time of the hour.
2. source: The source of the traffic data (hardcoded to "f5").
3. peak_concurrent_calls: The calculated peak concurrency.
4. sample_count: The number of requests used in the calculation.

Pruning Old Data

To manage the size of the aggregate table, the service periodically deletes records older than the configured retention period.

delete from s2r_traffic_concurrency_hourly where hour_bucket < now() - make_interval(days => ?)

The ? placeholder corresponds to the retentionDays configuration value.

Troubleshooting

The service is designed to be resilient to transient errors.

All SOAP-to-REST docs