From dc21a54c01cfa884bd5182cb0d45b692f924d63c Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Thu, 16 Jul 2026 14:39:43 -0700 Subject: [PATCH 1/3] Prevent Event Hub stalls from exhausting memory Bound telemetry buffers by bytes, cancel stalled transmissions, and only count events after successful sends. Add coverage for timeout, requeue, drop, and concurrent producer behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3694b841-cb8d-41c6-80d2-9023a312675c --- .../Logging/EventHubTelemetryChannelTests.cs | 121 ++++++++++++++++ .../Logging/EventHubTelemetryChannel.cs | 134 ++++++++++++++---- .../Logging/EventHubTelemetryLogger.cs | 1 + 3 files changed, 229 insertions(+), 27 deletions(-) create mode 100644 src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs diff --git a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs new file mode 100644 index 0000000000..6700213a76 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs @@ -0,0 +1,121 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Logging +{ + using System; + using System.Collections.Generic; + using System.Net.Http; + using System.Threading; + using System.Threading.Tasks; + using Azure.Messaging.EventHubs; + using NUnit.Framework; + + [TestFixture] + [Category("Unit")] + public class EventHubTelemetryChannelTests + { + [Test] + public void EventHubTelemetryChannelLimitsTheBufferByBytes() + { + using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + { + channel.AutoFlushInterval = TimeSpan.FromHours(1); + channel.MaxBufferSizeBytes = 10; + + channel.Add(new EventData(new byte[6])); + channel.Add(new EventData(new byte[6])); + + Assert.AreEqual(1, channel.BufferCount); + Assert.AreEqual(6, channel.BufferSizeBytes); + Assert.AreEqual(1, channel.Diagnostics.EventsDropped()); + } + } + + [Test] + public void EventHubTelemetryChannelTimesOutAndRequeuesFailedTransmissions() + { + using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + { + channel.AutoFlushInterval = TimeSpan.FromHours(1); + channel.TransmissionTimeout = TimeSpan.FromMilliseconds(25); + channel.TransmissionBehavior = (events, cancellationToken) => + Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + + channel.Add(new EventData(new byte[10])); + channel.Flush(TimeSpan.FromMilliseconds(50)); + + Assert.AreEqual(1, channel.BufferCount); + Assert.AreEqual(10, channel.BufferSizeBytes); + Assert.GreaterOrEqual(channel.Diagnostics.EventsTransmissionFailed(), 1); + Assert.AreEqual(0, channel.Diagnostics.EventsTransmitted()); + } + } + + [Test] + public void EventHubTelemetryChannelCountsEventsAsTransmittedAfterTheSendCompletes() + { + using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + { + channel.AutoFlushInterval = TimeSpan.FromHours(1); + + channel.Add(new EventData(new byte[10])); + channel.Flush(TimeSpan.FromSeconds(1)); + + Assert.AreEqual(0, channel.BufferCount); + Assert.AreEqual(0, channel.BufferSizeBytes); + Assert.AreEqual(1, channel.Diagnostics.EventsTransmitted()); + } + } + + [Test] + public async Task EventHubTelemetryChannelMaintainsTheByteLimitWhileATransmissionIsBlocked() + { + using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + { + TaskCompletionSource transmissionStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + channel.AutoFlushInterval = TimeSpan.FromHours(1); + channel.MaxBufferSizeBytes = 20; + channel.TransmissionTimeout = TimeSpan.FromMilliseconds(100); + channel.TransmissionBehavior = (events, cancellationToken) => + { + transmissionStarted.TrySetResult(); + return Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + }; + + channel.Add(new EventData(new byte[10])); + Task flushTask = Task.Run(() => channel.Flush(TimeSpan.FromMilliseconds(150))); + + await transmissionStarted.Task.WaitAsync(TimeSpan.FromSeconds(1)); + channel.Add(new EventData(new byte[10])); + channel.Add(new EventData(new byte[10])); + await flushTask; + + Assert.AreEqual(2, channel.BufferCount); + Assert.AreEqual(20, channel.BufferSizeBytes); + Assert.GreaterOrEqual(channel.Diagnostics.EventsTransmissionFailed(), 1); + Assert.GreaterOrEqual(channel.Diagnostics.EventsDropped(), 1); + Assert.AreEqual(0, channel.Diagnostics.EventsTransmitted()); + } + } + + private class TestEventHubTelemetryChannel : EventHubTelemetryChannel + { + public TestEventHubTelemetryChannel() + : base(new HttpClient + { + BaseAddress = new Uri("https://localhost") + }, enableDiagnostics: true) + { + } + + public Func, CancellationToken, Task> TransmissionBehavior { get; set; } = + (events, cancellationToken) => Task.CompletedTask; + + protected override Task TransmitBatchAsync(IEnumerable eventDataBatch) + { + return this.TransmissionBehavior.Invoke(eventDataBatch, this.TransmissionCancellationToken); + } + } + } +} diff --git a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs index f743623cba..4794481797 100644 --- a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs +++ b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs @@ -33,14 +33,18 @@ public class EventHubTelemetryChannel : IEnumerable, IFlushableChanne internal const int MaxEventDataBytes = 700000; private const int DefaultMaxCapacity = 1000000; + private const long DefaultMaxBufferSizeBytes = 268435456; private const int DefaultMinCapacity = 1001; private readonly object transmissionLock = new object(); private readonly object bufferLock = new object(); private int maxCapacity; + private long maxBufferSizeBytes; + private long bufferSizeBytes; private int minCapacity; private AutoResetEvent autoFlushWaitHandle; private CancellationTokenSource cancellationTokenSource; + private CancellationToken transmissionCancellationToken; private Task transmissionAutoSendTask; private SendEventOptions sendEventOptions; @@ -76,6 +80,7 @@ private EventHubTelemetryChannel(bool enableDiagnostics = false) this.minCapacity = EventHubTelemetryChannel.DefaultMinCapacity; this.maxCapacity = EventHubTelemetryChannel.DefaultMaxCapacity; + this.maxBufferSizeBytes = EventHubTelemetryChannel.DefaultMaxBufferSizeBytes; this.cancellationTokenSource = new CancellationTokenSource(); this.autoFlushWaitHandle = new AutoResetEvent(false); this.DiagnosticsEnabled = enableDiagnostics; @@ -112,7 +117,24 @@ public int BufferCount { get { - return this.Buffer.Count; + lock (this.bufferLock) + { + return this.Buffer.Count; + } + } + } + + /// + /// Gets the size in bytes of the events in the telemetry channel buffer. + /// + public long BufferSizeBytes + { + get + { + lock (this.bufferLock) + { + return this.bufferSizeBytes; + } } } @@ -144,6 +166,24 @@ public int MaxCapacity } } + /// + /// Gets or sets the maximum number of bytes that can be buffered for transmission. + /// + public long MaxBufferSizeBytes + { + get + { + return this.maxBufferSizeBytes; + } + + set + { + this.maxBufferSizeBytes = value > 0 + ? value + : EventHubTelemetryChannel.DefaultMaxBufferSizeBytes; + } + } + /// /// The client to use for publishing events to the Event Hub. /// @@ -155,6 +195,11 @@ public int MaxCapacity /// public TimeSpan AutoFlushInterval { get; set; } = TimeSpan.FromMilliseconds(5000); + /// + /// Gets or sets the maximum amount of time allowed for a single transmission. + /// + public TimeSpan TransmissionTimeout { get; set; } = TimeSpan.FromSeconds(30); + /// /// The client to use for publishing events to the Event Hub. /// @@ -170,6 +215,17 @@ public int MaxCapacity /// protected Queue Buffer { get; private set; } + /// + /// Gets the cancellation token for the current transmission. + /// + protected CancellationToken TransmissionCancellationToken + { + get + { + return this.transmissionCancellationToken; + } + } + /// /// Add a telemetry item to the buffer. /// @@ -195,7 +251,7 @@ public void Flush(TimeSpan? timeout = null) { DateTime flushTimeout = DateTime.Now.Add(timeout ?? TimeSpan.FromSeconds(60)); - while (this.Buffer.Count > 0) + while (this.BufferCount > 0) { this.TransmitEvents(); if (DateTime.Now >= flushTimeout) @@ -246,7 +302,9 @@ protected virtual void AddToBuffer(EventData item) } } - if (this.Buffer.Count >= this.MaxCapacity || item.Body.Length >= EventHubTelemetryChannel.MaxEventDataBytes) + if (this.Buffer.Count >= this.MaxCapacity + || this.bufferSizeBytes + item.Body.Length > this.MaxBufferSizeBytes + || item.Body.Length >= EventHubTelemetryChannel.MaxEventDataBytes) { this.Diagnostics?.EventsDropped(1); this.OnEventsDropped(new List { item }); @@ -255,6 +313,7 @@ protected virtual void AddToBuffer(EventData item) } this.Buffer.Enqueue(item); + this.bufferSizeBytes += item.Body.Length; this.Diagnostics?.EventsExpected(1); } } @@ -320,11 +379,7 @@ protected virtual async Task TransmitBatchAsync(IEnumerable eventData { if (this.AmqpClient != null) { - // Context on CancellationToken.None Here: - // We purposefully DO NOT honor the channel CancellationToken here. We do not want the - // transmission logic to exit on cancellation but to keep trying to get the telemetry through. - // We prefer a delayed exit of the application to losing telemetry. - await this.AmqpClient.SendAsync(eventDataBatch, this.sendEventOptions, CancellationToken.None); + await this.AmqpClient.SendAsync(eventDataBatch, this.sendEventOptions, this.TransmissionCancellationToken); } else { @@ -337,11 +392,7 @@ protected virtual async Task TransmitBatchAsync(IEnumerable eventData { request.Content = content; - // Context on CancellationToken.None Here: - // We purposefully DO NOT honor the channel CancellationToken here. We do not want the - // transmission logic to exit on cancellation but to keep trying to get the telemetry through. - // We prefer a delayed exit of the application to losing telemetry. - HttpResponseMessage response = await this.RestClient.SendAsync(request, CancellationToken.None); + using HttpResponseMessage response = await this.RestClient.SendAsync(request, this.TransmissionCancellationToken); response.EnsureSuccessStatusCode(); } } @@ -355,7 +406,7 @@ protected virtual async Task TransmitBatchAsync(IEnumerable eventData /// private void TransmitEvents() { - if (this.Buffer.Count > 0) + if (this.BufferCount > 0) { lock (this.transmissionLock) { @@ -373,13 +424,13 @@ private void TransmitEvents() /// private void TransmitEvents(int? maxBatchSize = null) { - if (this.Buffer.Count > 0) + if (this.BufferCount > 0) { List currentBatch = new List(); try { - while (this.Buffer.Count > 0) + while (this.BufferCount > 0) { lock (this.bufferLock) { @@ -388,7 +439,7 @@ private void TransmitEvents(int? maxBatchSize = null) for (int currentEventIndex = 0; currentEventIndex < batchSize; currentEventIndex++) { EventData nextEventItem = this.Buffer.Dequeue(); - this.Diagnostics?.EventsTransmitted(1); + this.bufferSizeBytes -= nextEventItem.Body.Length; if (nextEventItem != null) { @@ -403,7 +454,20 @@ private void TransmitEvents(int? maxBatchSize = null) } } - this.TransmitBatchAsync(currentBatch).GetAwaiter().GetResult(); + using (CancellationTokenSource transmissionTimeoutSource = new CancellationTokenSource(this.TransmissionTimeout)) + { + try + { + this.transmissionCancellationToken = transmissionTimeoutSource.Token; + this.TransmitBatchAsync(currentBatch).GetAwaiter().GetResult(); + } + finally + { + this.transmissionCancellationToken = CancellationToken.None; + } + } + + this.Diagnostics?.EventsTransmitted(currentBatch.Count); this.OnEventsTransmitted(currentBatch); currentBatch.Clear(); } @@ -412,6 +476,7 @@ private void TransmitEvents(int? maxBatchSize = null) { if (currentBatch.Any()) { + List droppedEvents = new List(); lock (this.bufferLock) { // If we failed to transmit the events, we need to add them back to the @@ -419,9 +484,24 @@ private void TransmitEvents(int? maxBatchSize = null) this.Diagnostics?.EventsTransmissionFailed(currentBatch.Count); currentBatch.ForEach(eventItem => { - this.Buffer.Enqueue(eventItem); + if (this.Buffer.Count < this.MaxCapacity + && this.bufferSizeBytes + eventItem.Body.Length <= this.MaxBufferSizeBytes) + { + this.Buffer.Enqueue(eventItem); + this.bufferSizeBytes += eventItem.Body.Length; + } + else + { + this.Diagnostics?.EventsDropped(1); + droppedEvents.Add(eventItem); + } }); } + + if (droppedEvents.Any()) + { + this.OnEventsDropped(droppedEvents); + } } // Telemetry transmission is a best-effort process. We do not want to crash @@ -476,40 +556,40 @@ public long EventsDropped(long? count = null) { if (count != null) { - Interlocked.Exchange(ref this.eventsDroppedCount, this.eventsDroppedCount + count.Value); + Interlocked.Add(ref this.eventsDroppedCount, count.Value); } - return this.eventsDroppedCount; + return Interlocked.Read(ref this.eventsDroppedCount); } public long EventsExpected(long? count = null) { if (count != null) { - Interlocked.Exchange(ref this.eventsExpectedCount, this.eventsExpectedCount + count.Value); + Interlocked.Add(ref this.eventsExpectedCount, count.Value); } - return this.eventsExpectedCount; + return Interlocked.Read(ref this.eventsExpectedCount); } public long EventsTransmitted(long? count = null) { if (count != null) { - Interlocked.Exchange(ref this.eventsTransmittedCount, this.eventsTransmittedCount + count.Value); + Interlocked.Add(ref this.eventsTransmittedCount, count.Value); } - return this.eventsTransmittedCount; + return Interlocked.Read(ref this.eventsTransmittedCount); } public long EventsTransmissionFailed(long? count = null) { if (count != null) { - Interlocked.Exchange(ref this.eventsTransmissionFailureCount, this.eventsTransmissionFailureCount + count.Value); + Interlocked.Add(ref this.eventsTransmissionFailureCount, count.Value); } - return this.eventsTransmissionFailureCount; + return Interlocked.Read(ref this.eventsTransmissionFailureCount); } } } diff --git a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs index 94dbd886cf..95a778a698 100644 --- a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs +++ b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs @@ -238,6 +238,7 @@ private EventData CreateEvent(LogLevel logLevel, EventId eventId, TState bufferInfo = new { bufferedEvents = this.underlyingTelemetryChannel.BufferCount, + bufferedBytes = this.underlyingTelemetryChannel.BufferSizeBytes, eventsExpected = diagnostics?.EventsExpected(), eventsTransmitted = diagnostics?.EventsTransmitted(), eventTransmissionFailures = diagnostics?.EventsTransmissionFailed(), From c65667a011725c39bd1aa408c4a3a94915fec831 Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Wed, 29 Jul 2026 10:05:02 -0700 Subject: [PATCH 2/3] Handle Event Hub logs without event context Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: b81a86bc-b161-4c2b-9f94-33116b68869e --- .../Logging/EventHubTelemetryLoggerTests.cs | 20 +++++++++++++++++++ .../Logging/EventHubTelemetryLogger.cs | 11 ++++++---- 2 files changed, 27 insertions(+), 4 deletions(-) diff --git a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryLoggerTests.cs b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryLoggerTests.cs index 6ad45c875a..41a727399e 100644 --- a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryLoggerTests.cs +++ b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryLoggerTests.cs @@ -140,6 +140,19 @@ public void EventHubTelemetryLoggerEnsuresTimestampsAreInUtcFormat() Assert.AreEqual(expectedTimestamp, eventBody["timestamp"].Value()); } + [Test] + public void EventHubTelemetryLoggerSupportsNamedEventsWithoutAnEventContext() + { + TestEventHubTelemetryLogger logger = new TestEventHubTelemetryLogger(this.mockChannel, LogLevel.Information); + + logger.Log(LogLevel.Information, new EventId(1, "TestMessage"), "state", null, null); + + Assert.AreEqual(1, logger.Events.Count); + JObject eventBody = JObject.Parse(logger.Events[0].EventBody.ToString()); + Assert.AreEqual("TestMessage", eventBody["message"].ToString()); + Assert.IsFalse(eventBody.ContainsKey("customDimensions")); + } + private class TestEventHubTelemetryLogger : EventHubTelemetryLogger { public TestEventHubTelemetryLogger(EventHubTelemetryChannel channel, LogLevel level) @@ -147,10 +160,17 @@ public TestEventHubTelemetryLogger(EventHubTelemetryChannel channel, LogLevel le { } + public IList Events { get; } = new List(); + public new EventData CreateEventObject(string eventMessage, LogLevel logLevel, DateTime eventTimestamp, EventContext eventContext, object bufferInfo = null) { return base.CreateEventObject(eventMessage, logLevel, eventTimestamp, eventContext, bufferInfo); } + + protected override void AddEventDataToChannel(EventData eventData) + { + this.Events.Add(eventData); + } } } } diff --git a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs index 95a778a698..1dc0e2ecf9 100644 --- a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs +++ b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs @@ -142,12 +142,15 @@ protected virtual void AddEventDataToChannel(EventData eventData) protected virtual EventData CreateEventObject(string eventMessage, LogLevel logLevel, DateTime eventTimestamp, EventContext eventContext, object bufferInfo = null) { // Allow for specific property overrides. - eventContext.Properties.TryGetValue(MetadataContract.AppHost, out object appHost); - eventContext.Properties.TryGetValue(MetadataContract.AppName, out object appName); - eventContext.Properties.TryGetValue(MetadataContract.AppVersion, out object appVersion); + object appHost = null; + object appName = null; + object appVersion = null; + eventContext?.Properties.TryGetValue(MetadataContract.AppHost, out appHost); + eventContext?.Properties.TryGetValue(MetadataContract.AppName, out appName); + eventContext?.Properties.TryGetValue(MetadataContract.AppVersion, out appVersion); DateTime timestamp = eventTimestamp; - if (eventContext.Properties.TryGetValue(MetadataContract.Timestamp, out object datetime)) + if (eventContext?.Properties.TryGetValue(MetadataContract.Timestamp, out object datetime) == true) { DateTime.TryParse(datetime?.ToString(), out timestamp); } From 13333b4745184b2b1e72e5d0cb13b56cb159696b Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Wed, 29 Jul 2026 10:38:00 -0700 Subject: [PATCH 3/3] Bump version to 3.3.29 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: b81a86bc-b161-4c2b-9f94-33116b68869e --- VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VERSION b/VERSION index 7efa37f800..081395c658 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -3.3.28 +3.3.29