Control Document Quotas for QuickPulse

WEB/Src/PerformanceCollector/Perf.Tests/Filtering/CollectionConfigurationTests.cs

@@ -8,6 +8,7 @@ namespace Microsoft.ApplicationInsights.Tests
     using Microsoft.ApplicationInsights.Extensibility;
     using Microsoft.ApplicationInsights.Extensibility.Filtering;
     using Microsoft.ApplicationInsights.Extensibility.PerfCounterCollector.Implementation.QuickPulse;
+    using Microsoft.ApplicationInsights.Extensibility.PerfCounterCollector.Implementation.Service_contract;
     using Microsoft.ApplicationInsights.Extensibility.PerfCounterCollector.QuickPulse;
     using Microsoft.ApplicationInsights.Web.Helpers;
     using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -653,6 +654,223 @@ public void CollectionConfigurationCarriesOverQuotaWhenCreatingDocumentStreams()
             Assert.AreEqual(3, documentStreams[3].EventQuotaTracker.CurrentQuota);
         }
 
+        [TestMethod]
+        public void CollectionConfigurationWithMaxAndInitialQuotas()
+        {
+            // ARRANGE
+            var timeProvider = new ClockMock();
+
+            CollectionConfigurationError[] errors;
+            var oldDocumentStreamInfos = new[]
+            {
+                new DocumentStreamInfo()
+                {
+                    Id = "Stream1",
+                    DocumentFilterGroups =
+                        new[]
+                        {
+                            new DocumentFilterConjunctionGroupInfo()
+                            {
+                                TelemetryType = TelemetryType.Request,
+                                Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] }
+                            }
+                        }
+                },
+                new DocumentStreamInfo()
+                {
+                    Id = "Stream2",
+                    DocumentFilterGroups =
+                        new[]
+                        {
+                            new DocumentFilterConjunctionGroupInfo()
+                            {
+                                TelemetryType = TelemetryType.Dependency,
+                                Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] }
+                            }
+                        }
+                },
+                new DocumentStreamInfo()
+                {
+                    Id = "Stream3",
+                    DocumentFilterGroups =
+                        new[]
+                        {
+                            new DocumentFilterConjunctionGroupInfo()
+                            {
+                                TelemetryType = TelemetryType.Exception,
+                                Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] }
+                            }
+                        }
+                }
+            };
+
+            var newDocumentStreamInfos = new[]
+            {
+                new DocumentStreamInfo()
+                {
+                    Id = "Stream1",
+                    DocumentFilterGroups =
+                        new[]
+                        {
+                            new DocumentFilterConjunctionGroupInfo()
+                            {
+                                TelemetryType = TelemetryType.Request,
+                                Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] }
+                            }
+                        }
+                },
+                new DocumentStreamInfo()
+                {
+                    Id = "Stream2",
+                    DocumentFilterGroups =
+                        new[]
+                        {
+                            new DocumentFilterConjunctionGroupInfo()
+                            {
+                                TelemetryType = TelemetryType.Dependency,
+                                Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] }
+                            }
+                        }
+                },
+                new DocumentStreamInfo()
+                {
+                    Id = "Stream3",
+                    DocumentFilterGroups =
+                        new[]
+                        {
+                            new DocumentFilterConjunctionGroupInfo()
+                            {
+                                TelemetryType = TelemetryType.Exception,
+                                Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] }
+                            }
+                        }
+                },
+                new DocumentStreamInfo()
+                {
+                    Id = "Stream4",
+                    DocumentFilterGroups =
+                        new[]
+                        {
+                            new DocumentFilterConjunctionGroupInfo()
+                            {
+                                TelemetryType = TelemetryType.Event,
+                                Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] }
+                            }
+                        }
+                },
+                new DocumentStreamInfo()
+                {
+                    Id = "Stream5",
+                    DocumentFilterGroups =
+                        new[]
+                        {
+                            new DocumentFilterConjunctionGroupInfo()
+                            {
+                                TelemetryType = TelemetryType.Trace,
+                                Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] }
+                            }
+                        }
+                }
+            };
+
+            var oldCollectionConfigurationInfo = new CollectionConfigurationInfo()
+            {
+                DocumentStreams = oldDocumentStreamInfos,
+                ETag = "ETag1",
+                QuotaInfo = new QuotaConfigurationInfo()
+                {
+                    InitialQuota = 10,
+                    MaxQuota = 30,
+                    QuotaAccrualRatePerSec = null
+                }
+            };
+            var oldCollectionConfiguration = new CollectionConfiguration(oldCollectionConfigurationInfo, out errors, timeProvider);
+
+            // spend some quota on the old configuration
+            var accumulatorManager = new QuickPulseDataAccumulatorManager(oldCollectionConfiguration);
+            var telemetryProcessor = new QuickPulseTelemetryProcessor(new SimpleTelemetryProcessorSpy());
+            ((IQuickPulseTelemetryProcessor)telemetryProcessor).StartCollection(
+                accumulatorManager,
+                new Uri("http://microsoft.com"),
+                new TelemetryConfiguration() { InstrumentationKey = "some ikey" });
+
+            // ACT
+            // the initial quota is 3
+            telemetryProcessor.Process(new RequestTelemetry() { Context = { InstrumentationKey = "some ikey" } });
+
+            telemetryProcessor.Process(new DependencyTelemetry() { Context = { InstrumentationKey = "some ikey" } });
+            telemetryProcessor.Process(new DependencyTelemetry() { Context = { InstrumentationKey = "some ikey" } });
+
+            telemetryProcessor.Process(new ExceptionTelemetry() { Context = { InstrumentationKey = "some ikey" } });
+            telemetryProcessor.Process(new ExceptionTelemetry() { Context = { InstrumentationKey = "some ikey" } });
+            telemetryProcessor.Process(new ExceptionTelemetry() { Context = { InstrumentationKey = "some ikey" } });
+
+            // ACT
+            // the new configuration must carry the quotas over from the old one (only for those document streams that already existed)
+            var newCollectionConfigurationInfo = new CollectionConfigurationInfo() { DocumentStreams = newDocumentStreamInfos, ETag = "ETag1" };
+            var newCollectionConfiguration = new CollectionConfiguration(
+                newCollectionConfigurationInfo,
+                out errors,
+                timeProvider,
+                oldCollectionConfiguration.DocumentStreams);
+
+            // ASSERT
+            DocumentStream[] oldDocumentStreams = oldCollectionConfiguration.DocumentStreams.ToArray();
+            Assert.AreEqual(3, oldDocumentStreams.Length);
+
+            Assert.AreEqual("Stream1", oldDocumentStreams[0].Id);
+            Assert.AreEqual(9, oldDocumentStreams[0].RequestQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[0].DependencyQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[0].ExceptionQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[0].EventQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[0].EventQuotaTracker.CurrentQuota);
+
+            Assert.AreEqual("Stream2", oldDocumentStreams[1].Id);
+            Assert.AreEqual(10, oldDocumentStreams[1].RequestQuotaTracker.CurrentQuota);
+            Assert.AreEqual(8, oldDocumentStreams[1].DependencyQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[1].ExceptionQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[1].EventQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[1].EventQuotaTracker.CurrentQuota);
+
+            Assert.AreEqual("Stream3", oldDocumentStreams[2].Id);
+            Assert.AreEqual(10, oldDocumentStreams[2].RequestQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[2].DependencyQuotaTracker.CurrentQuota);
+            Assert.AreEqual(7, oldDocumentStreams[2].ExceptionQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[2].EventQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, oldDocumentStreams[2].EventQuotaTracker.CurrentQuota);
+
+            DocumentStream[] newDocumentStreams = newCollectionConfiguration.DocumentStreams.ToArray();
+            Assert.AreEqual(5, newDocumentStreams.Length);
+
+            Assert.AreEqual("Stream1", newDocumentStreams[0].Id);
+            Assert.AreEqual(9, newDocumentStreams[0].RequestQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[0].DependencyQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[0].ExceptionQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[0].EventQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[0].EventQuotaTracker.CurrentQuota);
+
+            Assert.AreEqual("Stream2", newDocumentStreams[1].Id);
+            Assert.AreEqual(10, newDocumentStreams[1].RequestQuotaTracker.CurrentQuota);
+            Assert.AreEqual(8, newDocumentStreams[1].DependencyQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[1].ExceptionQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[1].EventQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[1].EventQuotaTracker.CurrentQuota);
+
+            Assert.AreEqual("Stream3", newDocumentStreams[2].Id);
+            Assert.AreEqual(10, newDocumentStreams[2].RequestQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[2].DependencyQuotaTracker.CurrentQuota);
+            Assert.AreEqual(7, newDocumentStreams[2].ExceptionQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[2].EventQuotaTracker.CurrentQuota);
+            Assert.AreEqual(10, newDocumentStreams[2].EventQuotaTracker.CurrentQuota);
+
+            Assert.AreEqual("Stream4", newDocumentStreams[3].Id);
+            Assert.AreEqual(3, newDocumentStreams[3].RequestQuotaTracker.CurrentQuota);
+            Assert.AreEqual(3, newDocumentStreams[3].DependencyQuotaTracker.CurrentQuota);
+            Assert.AreEqual(3, newDocumentStreams[3].ExceptionQuotaTracker.CurrentQuota);
+            Assert.AreEqual(3, newDocumentStreams[3].EventQuotaTracker.CurrentQuota);
+            Assert.AreEqual(3, newDocumentStreams[3].EventQuotaTracker.CurrentQuota);
+        }
+
         [TestMethod]
         public void CollectionConfigurationReportsDocumentStreamsWithDuplicateIds()
         {

WEB/Src/PerformanceCollector/Perf.Tests/QuickPulse/QuickPulseQuotaTrackerTests.cs

@@ -163,5 +163,28 @@ public void QuickPulseQuotaTrackerIsThreadSafe()
             Assert.AreEqual(correctResult, passedQuotaCount);
             Assert.IsFalse(quotaTracker.ApplyQuota());
         }
+
+        [TestMethod]
+        public void QuickPulseQuotaTrackerParameters()
+        {
+            // ARRANGE
+            int maxQuota = 500;
+            int experimentLengthInSeconds = 499;
+            var mockTimeProvider = new ClockMock();
+            var quotaTracker = new QuickPulseQuotaTracker(mockTimeProvider, maxQuota, 900, 1);
+
+            // ACT
+            for (int i = 0; i < experimentLengthInSeconds; i++)
+            {
+                mockTimeProvider.FastForward(TimeSpan.FromSeconds(1));
+
+                quotaTracker.ApplyQuota();
+            }
+
+            // ASSERT
+            Assert.AreEqual(499, quotaTracker.CurrentQuota);
+            Assert.AreEqual(500, quotaTracker.MaxQuota);
+            Assert.AreEqual(false, quotaTracker.QuotaExhausted);
+        }
     }
 }

WEB/Src/PerformanceCollector/Perf.Tests/QuickPulse/QuickPulseTelemetryModuleTests.cs

@@ -13,6 +13,7 @@
     using Microsoft.ApplicationInsights.Extensibility.Implementation;
     using Microsoft.ApplicationInsights.Extensibility.PerfCounterCollector.Implementation.QuickPulse;
     using Microsoft.ApplicationInsights.Extensibility.PerfCounterCollector.Implementation.QuickPulse.Helpers;
+    using Microsoft.ApplicationInsights.Extensibility.PerfCounterCollector.Implementation.Service_contract;
     using Microsoft.ApplicationInsights.Extensibility.PerfCounterCollector.QuickPulse;
     using Microsoft.ApplicationInsights.TestFramework;
     using Microsoft.ApplicationInsights.Web.Helpers;
@@ -802,6 +803,69 @@ public void QuickPulseTelemetryModuleUpdatesCollectionConfiguration()
             Assert.AreEqual("ETag2", serviceClient.SnappedSamples.Last().CollectionConfigurationAccumulator.CollectionConfiguration.ETag);
         }
 
+        [TestMethod]
+        public void QuickPulseTelemetryModuleUpdatesCollectionConfigurationWithCutomQuotas()
+        {
+            if (QuickPulseTelemetryModuleTests.Ignored)
+            {
+                return;
+            }
+
+            // ARRANGE
+            var pollingInterval = TimeSpan.FromSeconds(1);
+            var collectionInterval = TimeSpan.FromMilliseconds(400);
+            var timings = new QuickPulseTimings(pollingInterval, collectionInterval);
+            var collectionTimeSlotManager = new QuickPulseCollectionTimeSlotManagerMock(timings);
+            var serviceClient = new QuickPulseServiceClientMock { ReturnValueFromPing = true, ReturnValueFromSubmitSample = true };
+            var performanceCollector = new PerformanceCollectorMock();
+            var topCpuCollector = new QuickPulseTopCpuCollectorMock();
+            var module = new QuickPulseTelemetryModule(collectionTimeSlotManager, null, serviceClient, performanceCollector, topCpuCollector, timings);
+
+            // ACT & ASSERT
+            CollectionConfigurationInfo collectionConfigurationInfo = new CollectionConfigurationInfo()
+            {
+                ETag = "ETag1",
+                DocumentStreams = new[] { new DocumentStreamInfo() { Id = "wx3", DocumentFilterGroups = new[] { new DocumentFilterConjunctionGroupInfo() { TelemetryType = TelemetryType.Request, Filters = new FilterConjunctionGroupInfo() { Filters = new FilterInfo[0] } } } } },
+                QuotaInfo = new QuotaConfigurationInfo()
+                {
+                    InitialQuota = 50,
+                    QuotaAccrualRatePerSec = 10,
+                    MaxQuota = 500
+                }
+            };
+            serviceClient.CollectionConfigurationInfo = collectionConfigurationInfo;
+            QuickPulseTelemetryProcessor telemetryProcessor = new QuickPulseTelemetryProcessor(new SimpleTelemetryProcessorSpy());
+            module.RegisterTelemetryProcessor(telemetryProcessor);
+            module.Initialize(new TelemetryConfiguration() { InstrumentationKey = "some ikey" });
+            var collectionConfiguration = new CollectionConfiguration(collectionConfigurationInfo, out _, new Clock());
+            var accumulatorManager = new QuickPulseDataAccumulatorManager(collectionConfiguration);
+            ((IQuickPulseTelemetryProcessor)telemetryProcessor).StartCollection(
+                accumulatorManager,
+                new Uri("http://microsoft.com"),
+                new TelemetryConfiguration() { InstrumentationKey = "some ikey" });
+
+            for (int i=0;i<100;i++)
+            {
+                telemetryProcessor.Process(new RequestTelemetry() { Id = "1", Name = "Request1", Context = { InstrumentationKey = "some ikey" } });
+            }
+
+
+            Thread.Sleep(pollingInterval);
+            Assert.AreEqual(accumulatorManager.CurrentDataAccumulator.AIRequestCount, 50);
+            for (int i = 0; i < 100; i++)
+            {
+                telemetryProcessor.Process(new RequestTelemetry() { Id = "1", Name = "Request1", Context = { InstrumentationKey = "some ikey" } });
+            }
+
+            Thread.Sleep((int)(10 * collectionInterval.TotalMilliseconds));
+            Assert.AreEqual("ETag1", serviceClient.SnappedSamples.Last().CollectionConfigurationAccumulator.CollectionConfiguration.ETag);
+
+            Assert.AreEqual(accumulatorManager.CurrentDataAccumulator.AIRequestCount, 60);
+            serviceClient.CollectionConfigurationInfo = new CollectionConfigurationInfo() { ETag = "ETag2" };
+            Thread.Sleep((int)(10 * collectionInterval.TotalMilliseconds));
+            Assert.AreEqual("ETag2", serviceClient.SnappedSamples.Last().CollectionConfigurationAccumulator.CollectionConfiguration.ETag);
+        }
+
         [TestMethod]
         public void QuickPulseTelemetryModuleUpdatesPerformanceCollectorWhenUpdatingCollectionConfiguration()
         {