From dc21a54c01cfa884bd5182cb0d45b692f924d63c Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Thu, 16 Jul 2026 14:39:43 -0700 Subject: [PATCH 1/8] 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/8] 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/8] 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 From a58997d03fb170147d4cdb2c659772454cb243b0 Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Mon, 3 Aug 2026 09:31:18 -0700 Subject: [PATCH 4/8] Revert "Handle Event Hub logs without event context" This reverts commit c65667a011725c39bd1aa408c4a3a94915fec831. --- .../Logging/EventHubTelemetryLoggerTests.cs | 20 ------------------- .../Logging/EventHubTelemetryLogger.cs | 11 ++++------ 2 files changed, 4 insertions(+), 27 deletions(-) diff --git a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryLoggerTests.cs b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryLoggerTests.cs index 41a727399e..6ad45c875a 100644 --- a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryLoggerTests.cs +++ b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryLoggerTests.cs @@ -140,19 +140,6 @@ 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) @@ -160,17 +147,10 @@ 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 1dc0e2ecf9..95a778a698 100644 --- a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs +++ b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryLogger.cs @@ -142,15 +142,12 @@ 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. - 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); + eventContext.Properties.TryGetValue(MetadataContract.AppHost, out object appHost); + eventContext.Properties.TryGetValue(MetadataContract.AppName, out object appName); + eventContext.Properties.TryGetValue(MetadataContract.AppVersion, out object appVersion); DateTime timestamp = eventTimestamp; - if (eventContext?.Properties.TryGetValue(MetadataContract.Timestamp, out object datetime) == true) + if (eventContext.Properties.TryGetValue(MetadataContract.Timestamp, out object datetime)) { DateTime.TryParse(datetime?.ToString(), out timestamp); } From 860112f15d30cca8a624101bcb9233e0ea286ff3 Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Wed, 5 Aug 2026 09:58:13 -0700 Subject: [PATCH 5/8] Prevent oversized Event Hub batches Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a6b522c1-8e99-4de0-9051-d06d901f37ae --- .../Logging/EventHubTelemetryChannelTests.cs | 24 +++++++++++++++++++ .../Logging/EventHubTelemetryChannel.cs | 14 +++++++---- 2 files changed, 34 insertions(+), 4 deletions(-) diff --git a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs index 6700213a76..8ee5e80635 100644 --- a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs +++ b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs @@ -5,6 +5,7 @@ namespace VirtualClient.Logging { using System; using System.Collections.Generic; + using System.Linq; using System.Net.Http; using System.Threading; using System.Threading.Tasks; @@ -68,6 +69,29 @@ public void EventHubTelemetryChannelCountsEventsAsTransmittedAfterTheSendComplet } } + [Test] + public void EventHubTelemetryChannelDoesNotAddAnEventThatExceedsTheBatchByteLimit() + { + using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + { + List transmittedBatchSizes = new List(); + channel.AutoFlushInterval = TimeSpan.FromHours(1); + channel.TransmissionBehavior = (events, cancellationToken) => + { + transmittedBatchSizes.Add(events.Sum(eventData => eventData.Body.Length)); + return Task.CompletedTask; + }; + + channel.Add(new EventData(new byte[400000])); + channel.Add(new EventData(new byte[400000])); + channel.Flush(TimeSpan.FromSeconds(1)); + + CollectionAssert.AreEqual(new[] { 400000, 400000 }, transmittedBatchSizes); + Assert.AreEqual(2, channel.Diagnostics.EventsTransmitted()); + Assert.AreEqual(0, channel.Diagnostics.EventsTransmissionFailed()); + } + } + [Test] public async Task EventHubTelemetryChannelMaintainsTheByteLimitWhileATransmissionIsBlocked() { diff --git a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs index 4794481797..a83ca5a469 100644 --- a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs +++ b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs @@ -438,15 +438,21 @@ private void TransmitEvents(int? maxBatchSize = null) int batchSize = maxBatchSize ?? this.Buffer.Count; for (int currentEventIndex = 0; currentEventIndex < batchSize; currentEventIndex++) { - EventData nextEventItem = this.Buffer.Dequeue(); - this.bufferSizeBytes -= nextEventItem.Body.Length; - + EventData nextEventItem = this.Buffer.Peek(); if (nextEventItem != null) { + if (currentBatch.Any() + && currentBatchSize + nextEventItem.Body.Length > EventHubTelemetryChannel.MaxEventDataBytes) + { + break; + } + + this.Buffer.Dequeue(); + this.bufferSizeBytes -= nextEventItem.Body.Length; currentBatch.Add(nextEventItem); currentBatchSize += nextEventItem.Body.Length; - if (currentBatchSize > EventHubTelemetryChannel.MaxEventDataBytes || maxBatchSize == 1) + if (maxBatchSize == 1) { break; } From ed6bfdeb303e36b0bea00ab767b227d2aa8e5324 Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Wed, 5 Aug 2026 11:22:24 -0700 Subject: [PATCH 6/8] Simplify Event Hub buffering safeguards Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a6b522c1-8e99-4de0-9051-d06d901f37ae --- VERSION | 2 +- .../Logging/EventHubTelemetryChannelTests.cs | 109 +++++++++++++----- .../Logging/EventHubTelemetryChannel.cs | 70 ++++------- 3 files changed, 103 insertions(+), 78 deletions(-) diff --git a/VERSION b/VERSION index 081395c658..11d784e893 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -3.3.29 +3.3.29 \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs index 8ee5e80635..f654b6e882 100644 --- a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs +++ b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs @@ -10,6 +10,7 @@ namespace VirtualClient.Logging using System.Threading; using System.Threading.Tasks; using Azure.Messaging.EventHubs; + using Azure.Messaging.EventHubs.Producer; using NUnit.Framework; [TestFixture] @@ -19,7 +20,7 @@ public class EventHubTelemetryChannelTests [Test] public void EventHubTelemetryChannelLimitsTheBufferByBytes() { - using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) { channel.AutoFlushInterval = TimeSpan.FromHours(1); channel.MaxBufferSizeBytes = 10; @@ -34,29 +35,36 @@ public void EventHubTelemetryChannelLimitsTheBufferByBytes() } [Test] - public void EventHubTelemetryChannelTimesOutAndRequeuesFailedTransmissions() + public void EventHubTelemetryChannelRequeuesFailedTransmissions() { - using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) { + int transmissionAttempts = 0; channel.AutoFlushInterval = TimeSpan.FromHours(1); - channel.TransmissionTimeout = TimeSpan.FromMilliseconds(25); - channel.TransmissionBehavior = (events, cancellationToken) => - Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + channel.TransmissionBehavior = events => + { + if (Interlocked.Increment(ref transmissionAttempts) == 1) + { + throw new InvalidOperationException("Expected test failure."); + } + + return Task.CompletedTask; + }; channel.Add(new EventData(new byte[10])); - channel.Flush(TimeSpan.FromMilliseconds(50)); + channel.Flush(TimeSpan.FromSeconds(1)); - Assert.AreEqual(1, channel.BufferCount); - Assert.AreEqual(10, channel.BufferSizeBytes); - Assert.GreaterOrEqual(channel.Diagnostics.EventsTransmissionFailed(), 1); - Assert.AreEqual(0, channel.Diagnostics.EventsTransmitted()); + Assert.AreEqual(0, channel.BufferCount); + Assert.AreEqual(0, channel.BufferSizeBytes); + Assert.AreEqual(1, channel.Diagnostics.EventsTransmissionFailed()); + Assert.AreEqual(1, channel.Diagnostics.EventsTransmitted()); } } [Test] public void EventHubTelemetryChannelCountsEventsAsTransmittedAfterTheSendCompletes() { - using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) { channel.AutoFlushInterval = TimeSpan.FromHours(1); @@ -72,11 +80,11 @@ public void EventHubTelemetryChannelCountsEventsAsTransmittedAfterTheSendComplet [Test] public void EventHubTelemetryChannelDoesNotAddAnEventThatExceedsTheBatchByteLimit() { - using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) { List transmittedBatchSizes = new List(); channel.AutoFlushInterval = TimeSpan.FromHours(1); - channel.TransmissionBehavior = (events, cancellationToken) => + channel.TransmissionBehavior = events => { transmittedBatchSizes.Add(events.Sum(eventData => eventData.Body.Length)); return Task.CompletedTask; @@ -93,39 +101,80 @@ public void EventHubTelemetryChannelDoesNotAddAnEventThatExceedsTheBatchByteLimi } [Test] - public async Task EventHubTelemetryChannelMaintainsTheByteLimitWhileATransmissionIsBlocked() + public async Task EventHubTelemetryChannelMaintainsTheByteLimitWhileATransmissionIsInProgress() { - using (TestEventHubTelemetryChannel channel = new TestEventHubTelemetryChannel()) + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) { TaskCompletionSource transmissionStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseTransmission = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); channel.AutoFlushInterval = TimeSpan.FromHours(1); channel.MaxBufferSizeBytes = 20; - channel.TransmissionTimeout = TimeSpan.FromMilliseconds(100); - channel.TransmissionBehavior = (events, cancellationToken) => + channel.TransmissionBehavior = async events => { transmissionStarted.TrySetResult(); - return Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + await releaseTransmission.Task; }; channel.Add(new EventData(new byte[10])); - Task flushTask = Task.Run(() => channel.Flush(TimeSpan.FromMilliseconds(150))); + Task flushTask = Task.Run(() => channel.Flush(TimeSpan.FromSeconds(1))); await transmissionStarted.Task.WaitAsync(TimeSpan.FromSeconds(1)); channel.Add(new EventData(new byte[10])); channel.Add(new EventData(new byte[10])); + channel.Add(new EventData(new byte[10])); + releaseTransmission.SetResult(); 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()); + Assert.AreEqual(0, channel.BufferCount); + Assert.AreEqual(0, channel.BufferSizeBytes); + Assert.AreEqual(1, channel.Diagnostics.EventsDropped()); + Assert.AreEqual(3, channel.Diagnostics.EventsTransmitted()); + } + } + + [Test] + public void EventHubTelemetryChannelSendsRestEventsIndividually() + { + using (TestRestEventHubTelemetryChannel channel = new TestRestEventHubTelemetryChannel()) + { + List transmittedBatchCounts = new List(); + channel.AutoFlushInterval = TimeSpan.FromHours(1); + channel.TransmissionBehavior = events => + { + transmittedBatchCounts.Add(events.Count()); + return Task.CompletedTask; + }; + + channel.Add(new EventData(new byte[10])); + channel.Add(new EventData(new byte[10])); + channel.Flush(TimeSpan.FromSeconds(1)); + + CollectionAssert.AreEqual(new[] { 1, 1 }, transmittedBatchCounts); + } + } + + private class TestAmqpEventHubTelemetryChannel : EventHubTelemetryChannel + { + public TestAmqpEventHubTelemetryChannel() + : base(new EventHubProducerClient( + "Endpoint=sb://anynamespace.servicebus.windows.net/;SharedAccessKeyName=AnyAccessPolicy;SharedAccessKey=AnYacCEssKey=", + "any-hub"), + enableDiagnostics: true) + { + } + + public Func, Task> TransmissionBehavior { get; set; } = + events => Task.CompletedTask; + + protected override Task TransmitBatchAsync(IEnumerable eventDataBatch) + { + return this.TransmissionBehavior.Invoke(eventDataBatch); } } - private class TestEventHubTelemetryChannel : EventHubTelemetryChannel + private class TestRestEventHubTelemetryChannel : EventHubTelemetryChannel { - public TestEventHubTelemetryChannel() + public TestRestEventHubTelemetryChannel() : base(new HttpClient { BaseAddress = new Uri("https://localhost") @@ -133,12 +182,12 @@ public TestEventHubTelemetryChannel() { } - public Func, CancellationToken, Task> TransmissionBehavior { get; set; } = - (events, cancellationToken) => Task.CompletedTask; + public Func, Task> TransmissionBehavior { get; set; } = + events => Task.CompletedTask; protected override Task TransmitBatchAsync(IEnumerable eventDataBatch) { - return this.TransmissionBehavior.Invoke(eventDataBatch, this.TransmissionCancellationToken); + return this.TransmissionBehavior.Invoke(eventDataBatch); } } } diff --git a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs index a83ca5a469..2ef16358a4 100644 --- a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs +++ b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs @@ -44,7 +44,6 @@ public class EventHubTelemetryChannel : IEnumerable, IFlushableChanne private int minCapacity; private AutoResetEvent autoFlushWaitHandle; private CancellationTokenSource cancellationTokenSource; - private CancellationToken transmissionCancellationToken; private Task transmissionAutoSendTask; private SendEventOptions sendEventOptions; @@ -195,11 +194,6 @@ public long MaxBufferSizeBytes /// 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. /// @@ -215,17 +209,6 @@ public long MaxBufferSizeBytes /// 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. /// @@ -379,7 +362,11 @@ protected virtual async Task TransmitBatchAsync(IEnumerable eventData { if (this.AmqpClient != null) { - await this.AmqpClient.SendAsync(eventDataBatch, this.sendEventOptions, this.TransmissionCancellationToken); + // 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); } else { @@ -392,7 +379,11 @@ protected virtual async Task TransmitBatchAsync(IEnumerable eventData { request.Content = content; - using HttpResponseMessage response = await this.RestClient.SendAsync(request, this.TransmissionCancellationToken); + // 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. + using HttpResponseMessage response = await this.RestClient.SendAsync(request, CancellationToken.None); response.EnsureSuccessStatusCode(); } } @@ -439,40 +430,25 @@ private void TransmitEvents(int? maxBatchSize = null) for (int currentEventIndex = 0; currentEventIndex < batchSize; currentEventIndex++) { EventData nextEventItem = this.Buffer.Peek(); - if (nextEventItem != null) + if (currentBatch.Count > 0 + && currentBatchSize + nextEventItem.Body.Length > EventHubTelemetryChannel.MaxEventDataBytes) { - if (currentBatch.Any() - && currentBatchSize + nextEventItem.Body.Length > EventHubTelemetryChannel.MaxEventDataBytes) - { - break; - } - - this.Buffer.Dequeue(); - this.bufferSizeBytes -= nextEventItem.Body.Length; - currentBatch.Add(nextEventItem); - currentBatchSize += nextEventItem.Body.Length; - - if (maxBatchSize == 1) - { - break; - } + break; } - } - } - using (CancellationTokenSource transmissionTimeoutSource = new CancellationTokenSource(this.TransmissionTimeout)) - { - try - { - this.transmissionCancellationToken = transmissionTimeoutSource.Token; - this.TransmitBatchAsync(currentBatch).GetAwaiter().GetResult(); - } - finally - { - this.transmissionCancellationToken = CancellationToken.None; + this.Buffer.Dequeue(); + this.bufferSizeBytes -= nextEventItem.Body.Length; + currentBatch.Add(nextEventItem); + currentBatchSize += nextEventItem.Body.Length; + + if (maxBatchSize == 1) + { + break; + } } } + this.TransmitBatchAsync(currentBatch).GetAwaiter().GetResult(); this.Diagnostics?.EventsTransmitted(currentBatch.Count); this.OnEventsTransmitted(currentBatch); currentBatch.Clear(); From 942db85a59b4f5859e69fc3619b5080d56367c2a Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Wed, 5 Aug 2026 11:42:14 -0700 Subject: [PATCH 7/8] Cover failed batch drops at capacity Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a6b522c1-8e99-4de0-9051-d06d901f37ae --- .../Logging/EventHubTelemetryChannelTests.cs | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs index f654b6e882..370fd021af 100644 --- a/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs +++ b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs @@ -132,6 +132,43 @@ public async Task EventHubTelemetryChannelMaintainsTheByteLimitWhileATransmissio } } + [Test] + public async Task EventHubTelemetryChannelDropsFailedTransmissionsWhenTheBufferIsFull() + { + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) + { + int transmissionAttempts = 0; + TaskCompletionSource transmissionStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseTransmission = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + channel.AutoFlushInterval = TimeSpan.FromHours(1); + channel.MaxBufferSizeBytes = 20; + channel.TransmissionBehavior = async events => + { + if (Interlocked.Increment(ref transmissionAttempts) == 1) + { + transmissionStarted.SetResult(); + await releaseTransmission.Task; + throw new InvalidOperationException("Expected test failure."); + } + }; + + channel.Add(new EventData(new byte[10])); + Task flushTask = Task.Run(() => channel.Flush(TimeSpan.FromSeconds(1))); + + await transmissionStarted.Task.WaitAsync(TimeSpan.FromSeconds(1)); + channel.Add(new EventData(new byte[10])); + channel.Add(new EventData(new byte[10])); + releaseTransmission.SetResult(); + await flushTask; + + Assert.AreEqual(0, channel.BufferCount); + Assert.AreEqual(0, channel.BufferSizeBytes); + Assert.AreEqual(1, channel.Diagnostics.EventsTransmissionFailed()); + Assert.AreEqual(1, channel.Diagnostics.EventsDropped()); + Assert.AreEqual(2, channel.Diagnostics.EventsTransmitted()); + } + } + [Test] public void EventHubTelemetryChannelSendsRestEventsIndividually() { From 9c4b8244c2f3473a3b25e53bdee7f02f20208053 Mon Sep 17 00:00:00 2001 From: Erica Vellanoweth Date: Wed, 5 Aug 2026 11:55:18 -0700 Subject: [PATCH 8/8] Bump version to 3.3.30 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a6b522c1-8e99-4de0-9051-d06d901f37ae --- VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VERSION b/VERSION index 11d784e893..e2d9d24a5a 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -3.3.29 \ No newline at end of file +3.3.30 \ No newline at end of file