Rivulet.Diagnostics

Enterprise observability for Rivulet.Core with EventListener wrappers, metric aggregators, and health check integration.

Features

• EventListener Wrappers: Console, File, and Structured JSON logging
• Metrics Aggregation: Time-window based metric aggregation with statistics
• Prometheus Export: Export metrics in Prometheus text format
• Health Check Integration: Microsoft.Extensions.Diagnostics.HealthChecks support
• Fluent Builder API: Easy configuration with DiagnosticsBuilder

Installation

dotnet add package Rivulet.Diagnostics

Quick Start

Console Listener

using Rivulet.Diagnostics; using var listener = new RivuletConsoleListener(); await Enumerable.Range(1, 100) .ToAsyncEnumerable() .SelectParallelAsync(async x => { await ProcessAsync(x); return x; }, new ParallelOptionsRivulet { MaxDegreeOfParallelism = 10 });

File Listener with Rotation

using var listener = new RivuletFileListener( "metrics.log", maxFileSizeBytes: 10 * 1024 * 1024 // 10MB ); // Your parallel operations...

Structured JSON Logging

using var listener = new RivuletStructuredLogListener("metrics.json"); // Or with custom action using var listener = new RivuletStructuredLogListener(json => { // Send to your logging system logger.LogInformation(json); });

Metrics Aggregation

using var aggregator = new MetricsAggregator(TimeSpan.FromSeconds(10)); aggregator.OnAggregation += metrics => { foreach (var metric in metrics) { Console.WriteLine($"{metric.DisplayName}:"); Console.WriteLine($" Min: {metric.Min:F2}"); Console.WriteLine($" Max: {metric.Max:F2}"); Console.WriteLine($" Avg: {metric.Average:F2}"); Console.WriteLine($" Current: {metric.Current:F2}"); } }; // Your parallel operations...

Prometheus Export

using var exporter = new PrometheusExporter(); // Your parallel operations... // Export to Prometheus format var prometheusText = exporter.Export(); await File.WriteAllTextAsync("metrics.prom", prometheusText); // Or get as dictionary var metrics = exporter.ExportDictionary();

Health Check Integration

// In Program.cs or Startup.cs // Step 1: Register PrometheusExporter (required dependency for health check) builder.Services.AddSingleton<PrometheusExporter>(); // Step 2: Register health check builder.Services.AddHealthChecks() .AddCheck<RivuletHealthCheck>("rivulet", tags: new[] { "ready" }); // Step 3: Configure thresholds (optional) builder.Services.Configure<RivuletHealthCheckOptions>(options => { options.ErrorRateThreshold = 0.1; // 10% error rate triggers degraded status options.FailureCountThreshold = 100; // 100 failures triggers unhealthy status }); // Step 4: Add health check endpoint app.MapHealthChecks("/health"); // Health check will return: // - Healthy: Error rate below threshold and failures below threshold // - Degraded: Error rate exceeds threshold // - Unhealthy: Failure count exceeds threshold or unable to collect metrics

Fluent Builder API

using var diagnostics = new DiagnosticsBuilder() .AddConsoleListener() .AddFileListener("metrics.log") .AddStructuredLogListener("metrics.json") .AddMetricsAggregator(TimeSpan.FromSeconds(10), metrics => { // Handle aggregated metrics }) .AddPrometheusExporter(out var exporter) .Build(); // Your parallel operations... // Export Prometheus metrics var prometheusText = exporter.Export();

Available Metrics

Rivulet.Diagnostics exposes the following metrics from Rivulet.Core:

• items-started: Total number of items that have started processing
• items-completed: Total number of items that have completed processing
• total-retries: Total number of retry attempts
• total-failures: Total number of failed items
• throttle-events: Number of throttle events (backpressure)
• drain-events: Number of drain events

Advanced Usage

Custom Metric Thresholds

// When creating RivuletHealthCheck manually (not recommended, use DI instead) var exporter = new PrometheusExporter(); var healthCheck = new RivuletHealthCheck(exporter, new RivuletHealthCheckOptions { ErrorRateThreshold = 0.05, // 5% error rate triggers degraded FailureCountThreshold = 50 // 50 failures triggers unhealthy }); // Recommended: Use dependency injection (shown above in Health Check Integration)

Health Check Options

RivuletHealthCheckOptions properties:

• ErrorRateThreshold (double, default: 0.1)

  • Error rate (0.0 to 1.0) above which health check reports degraded status
  • Calculated as: total_failures / items_started
  • Example: 0.1 = 10% error rate

• FailureCountThreshold (long, default: 1000)

  • Absolute failure count above which health check reports unhealthy status
  • Useful for catching high-volume failure scenarios

Multiple Listeners

using var console = new RivuletConsoleListener(); using var file = new RivuletFileListener("metrics.log"); using var structured = new RivuletStructuredLogListener("metrics.json"); using var aggregator = new MetricsAggregator(TimeSpan.FromSeconds(5)); // All listeners will receive metrics simultaneously

Integration with Monitoring Systems

Application Insights

using var listener = new RivuletStructuredLogListener(json => { telemetryClient.TrackEvent("RivuletMetrics", new Dictionary<string, string> { ["metrics"] = json }); });

ELK Stack

using var listener = new RivuletStructuredLogListener("metrics.json"); // Configure Filebeat to ship metrics.json to Elasticsearch

Prometheus

using var exporter = new PrometheusExporter(); // Expose metrics endpoint app.MapGet("/metrics", () => exporter.Export());

Performance

• Zero-cost when not listening: EventCounters have zero overhead when no listeners are attached
• Minimal allocation: Uses polling counters to avoid allocations
• Thread-safe: All listeners are thread-safe and can be used concurrently

Requirements

• .NET 8.0 or .NET 9.0
• Rivulet.Core 1.2.0+
• Microsoft.Extensions.Diagnostics.HealthChecks 9.0.0+ (for health checks)

License

MIT License - see LICENSE file for details

---

Made with ❤️ by Jeffeek | NuGet | GitHub