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 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..370fd021af --- /dev/null +++ b/src/VirtualClient/VirtualClient.Core.UnitTests/Logging/EventHubTelemetryChannelTests.cs @@ -0,0 +1,231 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Logging +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Net.Http; + using System.Threading; + using System.Threading.Tasks; + using Azure.Messaging.EventHubs; + using Azure.Messaging.EventHubs.Producer; + using NUnit.Framework; + + [TestFixture] + [Category("Unit")] + public class EventHubTelemetryChannelTests + { + [Test] + public void EventHubTelemetryChannelLimitsTheBufferByBytes() + { + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) + { + 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 EventHubTelemetryChannelRequeuesFailedTransmissions() + { + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) + { + int transmissionAttempts = 0; + channel.AutoFlushInterval = TimeSpan.FromHours(1); + 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.FromSeconds(1)); + + 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 (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) + { + 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 void EventHubTelemetryChannelDoesNotAddAnEventThatExceedsTheBatchByteLimit() + { + using (TestAmqpEventHubTelemetryChannel channel = new TestAmqpEventHubTelemetryChannel()) + { + List transmittedBatchSizes = new List(); + channel.AutoFlushInterval = TimeSpan.FromHours(1); + channel.TransmissionBehavior = events => + { + 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 EventHubTelemetryChannelMaintainsTheByteLimitWhileATransmissionIsInProgress() + { + 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.TransmissionBehavior = async events => + { + transmissionStarted.TrySetResult(); + await releaseTransmission.Task; + }; + + 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])); + 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.EventsDropped()); + Assert.AreEqual(3, channel.Diagnostics.EventsTransmitted()); + } + } + + [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() + { + 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 TestRestEventHubTelemetryChannel : EventHubTelemetryChannel + { + public TestRestEventHubTelemetryChannel() + : base(new HttpClient + { + BaseAddress = new Uri("https://localhost") + }, enableDiagnostics: true) + { + } + + public Func, Task> TransmissionBehavior { get; set; } = + events => Task.CompletedTask; + + protected override Task TransmitBatchAsync(IEnumerable eventDataBatch) + { + return this.TransmissionBehavior.Invoke(eventDataBatch); + } + } + } +} diff --git a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs index f743623cba..2ef16358a4 100644 --- a/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs +++ b/src/VirtualClient/VirtualClient.Core/Logging/EventHubTelemetryChannel.cs @@ -33,11 +33,14 @@ 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; @@ -76,6 +79,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 +116,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 +165,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. /// @@ -195,7 +234,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 +285,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 +296,7 @@ protected virtual void AddToBuffer(EventData item) } this.Buffer.Enqueue(item); + this.bufferSizeBytes += item.Body.Length; this.Diagnostics?.EventsExpected(1); } } @@ -341,7 +383,7 @@ protected virtual async Task TransmitBatchAsync(IEnumerable eventData // 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, CancellationToken.None); response.EnsureSuccessStatusCode(); } } @@ -355,7 +397,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 +415,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) { @@ -387,23 +429,27 @@ 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.Diagnostics?.EventsTransmitted(1); - - if (nextEventItem != null) + EventData nextEventItem = this.Buffer.Peek(); + if (currentBatch.Count > 0 + && currentBatchSize + nextEventItem.Body.Length > EventHubTelemetryChannel.MaxEventDataBytes) { - currentBatch.Add(nextEventItem); - currentBatchSize += nextEventItem.Body.Length; + break; + } - if (currentBatchSize > EventHubTelemetryChannel.MaxEventDataBytes || maxBatchSize == 1) - { - break; - } + 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(); } @@ -412,6 +458,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 +466,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 +538,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(),