Record batch transmission latency in BatchingSink via SelfMetrics

Delivery to a remote sink is the part of logging that's least under anyone's control. Even when the application and the sink are both behaving perfectly, how long a batch takes to land depends on a long list of things that live outside the process: network latency and packet loss, DNS or TLS handshakes when connections are re-established, a proxy or load balancer in the path, the ingestion service itself being slow, throttling or rate limiting on the receiving side, a cross-region hop, etc. Locally it can be just as indirect: CPU throttling in a container, thread pool starvation, or a GC pause stretching out an otherwise quick send.

From Serilog's side all of them look the same. Batches take longer to send. The queue in BatchingSink grows, and memory grows with it. In the end events are dropped, either at QueueLimit, or when RetryTimeLimit is over. Only this last step is visible today, through SelfLog or ILoggingFailureListener. The slow degradation before it is not visible at all.

It would be very useful to see the timing of batch delivery to be able to react in timely manner when latency is growing.

SelfMetrics already covers the pipeline, but not the delivery side of it. The same approach can be extended to BatchingSink. The main part would be a duration histogram around _targetSink.EmitBatchAsync(...) in BatchingSink.LoopAsync(), something like:

// SelfMetrics
public static readonly Histogram<double> BatchingEmitBatchDuration = Meter.CreateHistogram<double>(
    "serilog.batching.emit_batch.duration",
    unit: "ms",
    description: "The time taken by a batched sink to accept a batch of log events.");

Also I would suggest adding a tag for the target sink (batchedSink.GetType().FullName) for the case when there are multiple sinks. I would suggest to modify BatchingSink code like this:

var startTimestamp = 0L;
try
{
    if (_currentBatch.Count == 0)
    {
        await _targetSink.OnEmptyBatchAsync().ConfigureAwait(false);
    }
    else
    {
        isEagerBatch = false;
        startTimestamp = Stopwatch.GetTimestamp();
        await _targetSink.EmitBatchAsync(_currentBatch).ConfigureAwait(false);
        RecordEmitBatchDuration(startTimestamp, error: null);
        _currentBatch.Clear();
        _batchScheduler.MarkSuccess();
    }
}
catch (Exception ex)
{
    RecordEmitBatchDuration(startTimestamp, ex);
    // ... existing handling unchanged
}

The tags are per measurement, so they go into Record(), with the sink tag prepared once:

// this line goes to BatchingSink constructor
_metricTags = new TagList { { SelfMetrics.TagNames.BatchedSinkType, batchedSink.GetType().FullName } };

void RecordEmitBatchDuration(long startTimestamp, Exception? error)
{
    if (startTimestamp == 0) return;

    var elapsedMs = (Stopwatch.GetTimestamp() - startTimestamp) * 1000.0 / Stopwatch.Frequency;

    var tags = _metricTags;
    if (error != null)
        tags.Add(SelfMetrics.TagNames.ErrorType, error.GetType().FullName);

    SelfMetrics.BatchingEmitBatchDuration.Record(elapsedMs, tags);
}

If you're ok about this I will be happy to take care of implementing this functionality.