klod68/analytics-concurrent-collections

---

name: analytics-concurrent-collections description: "Use when collecting real-time telemetry or analytics with concurrent access, tracking in-process events without external infrastructure, or building lightweight admin dashboards showing usage statistics. Implements thread-safe counters and bounded event streams using ConcurrentDictionary and ConcurrentQueue to prevent memory growth under concurrent writes. Domain: Telemetry, In-Process Analytics. Level: Intermediate. Tags: analytics, concurrent-collections, telemetry, in-memory, event-tracking."

Analytics with Concurrent Collections

Problem

Your application needs lightweight, in-process analytics — tracking events, counting occurrences, computing aggregates — without external infrastructure (Application Insights, database, message queue). The tracking must be thread-safe because multiple requests write concurrently.

Solution: Lock-Free Event Tracking with Bounded Storage

Use ConcurrentQueue<T> for ordered event streams and ConcurrentDictionary<TKey, TValue> for aggregated statistics. Bound the queue to prevent unbounded memory growth.

Implementation

1. Define event and stats models

public sealed record TrackingEvent
{
    public required string Action { get; init; }
    public required int ResultCount { get; init; }
    public string? Category { get; init; }
    public DateTime Timestamp { get; init; } = DateTime.UtcNow;
}

public sealed class ActionStats
{
    public required string Action { get; set; }
    public int Count { get; set; }
    public double AverageResultCount { get; set; }
    public DateTime LastOccurrence { get; set; }
}

2. Build the analytics service

using System.Collections.Concurrent;

/// <summary>
/// Thread-safe, in-memory analytics service with bounded event storage.
/// Suitable for demo/dev/small-scale production use.
/// </summary>
public sealed class AnalyticsService
{
    private readonly ConcurrentQueue<TrackingEvent> _events = new();
    private readonly ConcurrentDictionary<string, ActionStats> _stats = new(StringComparer.OrdinalIgnoreCase);

    private const int MaxEventsToKeep = 1000;

    /// <summary>
    /// Records an event. Thread-safe, lock-free.
    /// </summary>
    public void Track(string action, int resultCount, string? category = null)
    {
        if (string.IsNullOrWhiteSpace(action))
            return;

        var normalized = action.Trim().ToLowerInvariant();

        // Append to event stream
        _events.Enqueue(new TrackingEvent
        {
            Action = normalized,
            ResultCount = resultCount,
            Category = category
        });

        TrimEventsIfNeeded();

        // Update aggregate stats (atomic, lock-free)
        _stats.AddOrUpdate(
            normalized,
            _ => new ActionStats
            {
                Action = normalized,
                Count = 1,
                AverageResultCount = resultCount,
                LastOccurrence = DateTime.UtcNow
            },
            (_, existing) =>
            {
                existing.Count++;
                existing.AverageResultCount =
                    ((existing.AverageResultCount * (existing.Count - 1)) + resultCount) / existing.Count;
                existing.LastOccurrence = DateTime.UtcNow;
                return existing;
            });
    }

    /// <summary>
    /// Gets the top N most frequent actions.
    /// </summary>
    public IEnumerable<ActionStats> GetTopActions(int count = 10)
        => _stats.Values
            .OrderByDescending(s => s.Count)
            .Take(count);

    /// <summary>
    /// Gets recent events in reverse chronological order.
    /// </summary>
    public IEnumerable<TrackingEvent> GetRecentEvents(int count = 50)
        => _events.Reverse().Take(count);

    /// <summary>
    /// Gets actions that produced zero results (potential content gaps).
    /// </summary>
    public IEnumerable<ActionStats> GetZeroResultActions()
        => _stats.Values
            .Where(s => s.AverageResultCount == 0)
            .OrderByDescending(s => s.Count);

    /// <summary>
    /// Gets total event count.
    /// </summary>
    public int TotalEventCount => _events.Count;

    /// <summary>
    /// Gets total distinct actions tracked.
    /// </summary>
    public int DistinctActionCount => _stats.Count;

    private void TrimEventsIfNeeded()
    {
        while (_events.Count > MaxEventsToKeep)
        {
            _events.TryDequeue(out _);
        }
    }
}

3. Register as singleton

// In Program.cs or Startup
builder.Services.AddSingleton<AnalyticsService>();

4. Track from any component

public class SearchHandler
{
    private readonly AnalyticsService _analytics;

    public SearchHandler(AnalyticsService analytics)
    {
        _analytics = analytics;
    }

    public IEnumerable<Result> Search(string query)
    {
        var results = _searchEngine.Execute(query);
        _analytics.Track(query, results.Count());
        return results;
    }
}

5. Expose analytics (API or admin page)

// Minimal API endpoint
app.MapGet("/api/analytics", (AnalyticsService analytics) => new
{
    TotalEvents = analytics.TotalEventCount,
    DistinctActions = analytics.DistinctActionCount,
    TopActions = analytics.GetTopActions(10),
    ZeroResultActions = analytics.GetZeroResultActions().Take(5)
});

Why ConcurrentQueue + ConcurrentDictionary

| Collection | Purpose | Thread Safety | Performance | |-----------|---------|--------------|-------------| | ConcurrentQueue<T> | Ordered event stream (FIFO) | Lock-free enqueue/dequeue | O(1) enqueue | | ConcurrentDictionary<K,V> | Aggregated stats by key | Lock-free read, fine-grained lock on write | O(1) lookup | | List<T> + lock | ❌ Alternative | Manual locking | Contention under load |

Bounded Storage Strategy

MaxEventsToKeep = 1000
│
├── Events 1-1000: kept in queue
├── Event 1001 arrives → Event 1 dequeued (FIFO trim)
│
└── Stats dictionary: never trimmed (one entry per distinct action)
    └── For high-cardinality actions, add periodic cleanup:
        RemoveStale(olderThan: TimeSpan.FromDays(30))

When to Use

• Demo/dev environments where external telemetry isn't set up
• Small-to-medium scale applications (single instance)
• Quick analytics for search behavior, feature usage, error patterns
• Admin dashboards showing real-time usage statistics

When NOT to Use

• Multi-instance/distributed applications (use Application Insights, OpenTelemetry)
• Data that must survive restarts (use a database)
• High-volume event streams (use a message queue + time-series DB)
• Compliance-sensitive analytics (use proper audit logging)

Gotchas

1. Data is lost on restart: This is in-memory only. For persistence, periodically flush to a database or file.
2. AddOrUpdate lambda is not atomic: The update lambda may execute multiple times under contention. Keep it idempotent.
3. Reverse() on ConcurrentQueue: Creates a snapshot. Under very high throughput this allocates. For hot paths, consider a fixed-size ring buffer.
4. Memory growth: The stats dictionary grows with distinct actions. If actions are user-generated strings (search queries), this can grow unbounded. Add periodic cleanup for stale entries.
5. Singleton lifetime: Register as Singleton, not Scoped or Transient, or each request gets its own empty instance.

Related Skills

• In-Memory Search Registry (track search queries)
• Health Check Endpoint (expose analytics health)