Add SlowConsumingPressureMonitor #2873

xiazen · 2017-03-22T01:13:37Z

mostly finished coding, still need to test it. submit for early feedback

xiazen · 2017-03-22T18:49:24Z

Discussed with @jason-bragg offline, decided to take a simpler but less absolute approach. will shelve this commit. But still update the new approach to the same PR.

xiazen · 2017-03-23T01:17:54Z

switched to a simpler approach, added one programmatic option to configure EventHubQueueCache with the new monitor.

another option can be through EventHubStreamProviderSettings

ReubenBond · 2017-03-23T01:55:57Z

test/ServiceBus.Tests/TestStreamProviders/EHStreamProviderWithSlowConsumingPressureDetecting.cs

+
+            private IEventHubQueueCache CreateQueueCache(string partition, IStreamQueueCheckpointer<string> checkpointer, Logger log)
+            {
+                var bufferPool = new FixedSizeObjectPool<FixedSizeBuffer>(adapterSettings.CacheSizeMb, () => new FixedSizeBuffer(1 << 20));


Just to make it blatantly obvious, maybe have a const for this value called Megabyte or BlockSize with a comment saying it's 1MB

good idea. will do it !

jason-bragg · 2017-03-23T20:31:33Z

src/OrleansServiceBus/Providers/Streams/EventHub/EventHubQueueCache.cs

@@ -162,7 +228,7 @@ internal class AveragingCachePressureMonitor
        public AveragingCachePressureMonitor(double flowControlThreshold, Logger logger)
        {
            this.flowControlThreshold = flowControlThreshold;
-            this.logger = logger.GetSubLogger("flowcontrol", "-");
+            this.logger = logger.GetSubLogger("flowcontrol-averaging-cache-pressure", "-");


Suggestion:
As we're moving more towards an asp.net logging model, I suggest using the typename here rather than custom name.
Also, delilimiter argument is optional, and "-" is default, so it should not be needed.
so:
this.logger = logger.GetSubLogger(GetType().Name);

jason-bragg · 2017-03-23T23:36:14Z

src/OrleansServiceBus/Providers/Streams/EventHub/ICachePressureMonitor.cs

+        bool IsUnderPressure(DateTime utcNow);
+    }
+
+    internal class AggregatedCachePressureMonitor : List<ICachePressureMonitor>, ICachePressureMonitor


Suggest AggregatedCachePressureMonitor be public and placed in it's own file.

jason-bragg · 2017-03-23T23:37:49Z

test/ServiceBus.Tests/TestStreamProviders/EHStreamProviderWithSlowConsumingPressureDetecting.cs

+
+namespace ServiceBus.Tests.TestStreamProviders
+{
+    public class EHStreamProviderWithSlowConsumingPressureDetecting : PersistentStreamProvider<EHStreamProviderWithSlowConsumingPressureDetecting.AdapterFactory>


Suggest test run against default stream provider, configured to use SlowConsumingPressureMonitor, rather than custom adapter. That is the preferred user scenario, so the one we should ensure works and is maintained going forward.

jason-bragg · 2017-03-23T23:39:58Z

src/OrleansServiceBus/Providers/Streams/EventHub/EventHubAdapterFactory.cs

@@ -140,7 +140,20 @@ public virtual void Init(IProviderConfiguration providerCfg, string providerName
            {
                var bufferPool = new FixedSizeObjectPool<FixedSizeBuffer>(adapterSettings.CacheSizeMb, () => new FixedSizeBuffer(1 << 20));
                var timePurge = new TimePurgePredicate(adapterSettings.DataMinTimeInCache, adapterSettings.DataMaxAgeInCache);
-                CacheFactory = (partition,checkpointer,cacheLogger) => new EventHubQueueCache(checkpointer, bufferPool, timePurge, cacheLogger, this.SerializationManager);
+                if (adapterSettings.SlowConsumingMonitorThreshold > 0)


We tend to use nullable values for optional settings. Suggest SlowConsumingMonitorThreshold be nullable rather than using it's value to detect if it's set or not.

jason-bragg · 2017-03-23T23:49:16Z

src/OrleansServiceBus/Providers/Streams/EventHub/EventHubAdapterFactory.cs

+                if (adapterSettings.SlowConsumingMonitorThreshold > 0)
+                {
+                    CacheFactory = (partition, checkpointer, cacheLogger) =>
+                    {   var cache = new EventHubQueueCache(checkpointer, bufferPool, timePurge, cacheLogger, this.SerializationManager);


suggest minimizing code in conditional statement. Only thing that depend on the condition is if the SlowConsumingPressureMonitor is added. so

var cache = new ..... if(..) { add monitor } return cache;

jason-bragg · 2017-03-23T23:56:04Z

src/OrleansServiceBus/Providers/Streams/EventHub/EventHubQueueCache.cs

@@ -148,7 +159,62 @@ public bool TryGetNextMessage(object cursorObj, out IBatchContainer message)

    }

-    internal class AveragingCachePressureMonitor
+    public class SlowConsumingPressureMonitor : ICachePressureMonitor


Suggest AveragingCachePressureMonitor be broke into it's own file, and added in the default cache creation logic, like other ICachePressureMonitors.

xiazen · 2017-03-24T00:30:17Z

all good suggestions. will address them

xiazen · 2017-03-24T00:31:09Z

don't feel pressure that you need to finish reviewing in one pass. I'm totally fine if you finish the review in less than 4 pass. ;) jk

…PressureMonitor

xiazen · 2017-03-30T17:29:24Z

test/ServiceBus.Tests/SlowConsumingTests/EHSlowConsumingTests.cs

+        }
+
+        [Fact, TestCategory("EventHub"), TestCategory("Streaming"), TestCategory("Functional")]
+        public async Task EHSlowConsuming_ShouldFavorSlowConsumer()


@jdom I think I need to skip this test if azure connection string is not set up correctly in Jenkins?

I got this when running in Jenkins:
MESSAGE:
System.AggregateException : One or more errors occurred.\r\n---- System.TypeInitializationException : The type initializer for 'ServiceBus.Tests.SlowConsumingTests.EHSlowConsumingTests' threw an exception.\r\n-------- System.ArgumentNullException : Value cannot be null.\r\nParameter name: dataConnectionString\r\n---- The following constructor parameters did not have matching fixture data: Fixture fixture
+++++++++++++++++++
STACK TRACE:
----- Inner Stack Trace #1 (System.TypeInitializationException) ----- at ServiceBus.Tests.SlowConsumingTests.EHSlowConsumingTests.Fixture.CreateTestCluster() in D:\j\workspace\netfx-functio---224c30ba\test\ServiceBus.Tests\SlowConsumingTests\EHSlowConsumingTests.cs:line 57 at TestExtensions.BaseTestClusterFixture..ctor() in D:\j\workspace\netfx-functio---224c30ba\test\TestExtensions\BaseClusterFixture.cs:line 30 ----- Inner Stack Trace ----- at Orleans.ServiceBus.Providers.EventHubCheckpointerSettings..ctor(String dataConnectionString, String table, String checkpointNamespace, Nullable`1 persistInterval) in D:\j\workspace\netfx-functio---224c30ba\src\OrleansServiceBus\Providers\Streams\EventHub\ICheckpointerSettings.cs:line 65 at ServiceBus.Tests.SlowConsumingTests.EHSlowConsumingTests..cctor() in D:\j\workspace\netfx-functio---224c30ba\test\ServiceBus.Tests\SlowConsumingTests\EHSlowConsumingTests.cs:line 45 ----- Inner Stack Trace #2 (Xunit.Sdk.TestClassException) -----

This test pass when I have connection string set up locally

It's likely the Eventhub connection string. We don't run eh tests in Jenkins yet (to my knowledge).

Correct, although we should start running EH tests in functionals in VSO, but skip the tests when the connection string is not set. I added some info on how to do it

yeah I agree . We should set up the EH related config in VSO and start to run EH tests.

Although this test do run successfully in VSO. so I think some EH related config is set up in VSO. But I'm not sure how much of the EH related config is set up in VSO. where can I check?

yes, configuration is already set up in VSO, it's just that currently none of the EH tests are marked as Functional, so they just don't run because of that. I'd start adding these tests using the Skippable approach, and then we can look at adding more of the other EH tests using the same approach

will do. I can add skippable to this test. A lot of EH tests run successfully locally. I think we can star add them to VSO and skip them in Jenkins.

jason-bragg · 2017-03-30T18:00:14Z

...erviceBus/Providers/Streams/EventHub/CachePressureMonitors/AggregatedCachePressureMonitor.cs

+        {
+            bool isUnderPressure = false;
+
+            this.ForEach(monitor =>


nit: Link statement would be much cleaner.

return this.Any(monitor => monitor.IsUnderPressure(utcNow);

jason-bragg · 2017-03-30T18:03:29Z

src/OrleansServiceBus/Providers/Streams/EventHub/EventHubAdapterFactory.cs

-                CacheFactory = (partition,checkpointer,cacheLogger) => new EventHubQueueCache(checkpointer, bufferPool, timePurge, cacheLogger, this.SerializationManager);
+                CacheFactory = (partition, checkpointer, cacheLogger) =>
+                {
+                    var cache = new EventHubQueueCache(checkpointer, bufferPool, timePurge, cacheLogger, this.SerializationManager);


nit: We generally try to keep lambdas small, when they get large, we tend to put them into their own call. This may be better as a CreateCacheFactory(..) function.

jason-bragg · 2017-03-30T18:10:01Z

test/ServiceBus.Tests/SlowConsumingTests/EHSlowConsumingTests.cs

+namespace ServiceBus.Tests.SlowConsumingTests
+{
+    [Collection(TestEnvironmentFixture.DefaultCollection)]
+    public class EHSlowConsumingTests : OrleansTestingBase, IClassFixture<EHSlowConsumingTests.Fixture>


For targeted tests of the persistent stream provider infrastructure (kinda like this), we tend to use the generator stream provider, as it allows better control of the data flow as well as eliminates external dependencies like EventHub. We can test this using EH, I'm just suggesting that the stream generator may be better in the long run.

The test is also testing if the monitor config can be set correctly through EventHubProviderSettings class, so I'd prefer use real EventHubStreamProvider

jdom · 2017-03-31T00:10:46Z

@xiazen this failure is a real one (not timing related, which we are fixing): https://ci.dot.net/job/dotnet_orleans/job/master/job/netfx-functional_prtest/10/testReport/junit/ServiceBus.Tests.SlowConsumingTests/EHSlowConsumingTests/EHSlowConsuming_ShouldFavorSlowConsumer/
Did you miss a SkippableFact or something of the sort?

jdom · 2017-03-31T00:17:10Z

...ServiceBus/Providers/Streams/EventHub/CachePressureMonitors/AveragingCachePressureMonitor.cs

+        /// <summary>
+        /// Default flow control threshold
+        /// </summary>
+        public static readonly double DefaultThreashold = 1 / 3;


typo in threshold

jdom · 2017-03-31T00:21:15Z

src/OrleansServiceBus/Providers/Streams/EventHub/EventHubStreamProviderSettings.cs

@@ -17,6 +17,37 @@ public class EventHubStreamProviderSettings
        public string StreamProviderName { get; }

        /// <summary>
+        /// SlowConsumingMonitorFlowControlThresholdName
+        /// </summary>
+        public const string SlowConsumingMonitorFlowControlThresholdName = "SlowConsumingMonitorFlowControlThreshold";


nameof(SlowConsumingMonitorFlowControlThreshold)
Avoid hardcoding strings in several places in this class. We were probably doing this before we upgraded to c# 6, but there's no reason to do more hardcoding now

jdom · 2017-03-31T00:33:41Z

...sServiceBus/Providers/Streams/EventHub/CachePressureMonitors/SlowConsumingPressureMonitor.cs

+        /// </summary>
+        /// <param name="checkPeriod"></param>
+        /// <param name="logger"></param>
+        public SlowConsumingPressureMonitor(TimeSpan checkPeriod, Logger logger)


checkPeriod is a confusing name. It took me some time by reading and re-reading the code to understand what it does, because it's not the time period between checks.
Maybe it should be pressure window size? Or something else not related to check time?

I will think of a better name. PressureWindowPeriod? size is too general for me. let me think

jdom · 2017-03-31T00:35:46Z

test/ServiceBus.Tests/Streaming/EHSlowConsumingTests.cs

+{
+    [TestCategory("EventHub"), TestCategory("Streaming")]
+    [Collection(TestEnvironmentFixture.DefaultCollection)]
+    class EHSlowConsumingTests : OrleansTestingBase, IClassFixture<EHSlowConsumingTests.Fixture>


is this class dead code? I don't see any tests here, and it's named the same as the other tests

Wow I didn't notice this ! there's another EHSlowConsumingTests.cs on the top... That's the real one. I will delete this one. Thanks for noticing

jdom · 2017-03-31T00:39:42Z

test/ServiceBus.Tests/SlowConsumingTests/EHSlowConsumingTests.cs

+
+        private readonly Fixture fixture;
+
+        public class Fixture : BaseTestClusterFixture


You should override CheckPreconditionsOrThrow and throw SkipException if the connection string is null.
Then in the test class' constructor you must call fixture.EnsurePreconditionsMet()
Also mark all of the tests with SkippableFact.

See usages of EnsurePreconditionsMet if you want some examples.

jdom · 2017-04-03T17:17:00Z

LGTM

xiazen · 2017-04-03T17:29:44Z

@jdom the Dispose() method was called. That's why I added that Boolean check. maybe there's a bug somewhere or I used OrleansTestBase in an incorrectly way?

jason-bragg · 2017-04-04T16:56:04Z

...ServiceBus/Providers/Streams/EventHub/CachePressureMonitors/AveragingCachePressureMonitor.cs

+            // If we changed state, log
+            if (isUnderPressure != wasUnderPressure)
+            {
+                logger.Info(isUnderPressure


I think the info level version of this log message should be in the aggregate pressure monitor, and the logging in the specific monitors should be verbose.
The reason for this is that a specific monitor may become healthy, but reading will still be off until all monitors are healthy. So having per monitor logging will make it very hard for users to decipher, from logs, if a queue should be reading or not. These details may be useful in verbose logging, as they would indicate what monitor is stopping reading, but this level of detail is less important, imo, than the overall flow control state.

will do . valid concern.

jdom · 2017-04-04T20:52:28Z

...sServiceBus/Providers/Streams/EventHub/CachePressureMonitors/SlowConsumingPressureMonitor.cs

+            if (this.isUnderPressure != underPressure)
+            {
+                this.isUnderPressure = underPressure;
+                logger.Verbose(this.isUnderPressure


how about logging as Info when under pressure, and as Verbose when not? Otherwise the aggregating monitor might not log the necessary information, right?

It logs only when there is a state change.
under pressure => not under pressure or not under pressure => under pressure.
In both cases, this is not necessarily authoritative of the behavior, so users shouldn't care about this under normal logging conditions.

ouch. I submitted a commit to address this. but under a second thought, I think Jason might be right. So I reverted it

Sure, on a state change makes more sense. But still it would be nice to know the details for when an individual monitor starts throttling. Maybe in the individual monitor, log as Info only when it transitions to under pressure when it wasn't (but not necessarily log as Info when it goes back to not under pressure, at that point the aggregated monitor logger can be enough).

xiazen · 2017-04-05T00:24:12Z

...sServiceBus/Providers/Streams/EventHub/CachePressureMonitors/SlowConsumingPressureMonitor.cs

+                this.wasUnderPressure = underPressure;
+            }
+
+            return this.wasUnderPressure;


in this logic, monitor will stay underPressure for a whole window if a slow consumer appears, while biggestPressureInCurrentWindow will be reset to 0. So at the end of this window, if still underPressure, then underPressure stays for another whole window. Otherwise, monitor become notUnderPressure.

run through real life case,
monitor will be notUnderPressure at the beginning-> a slow consumer appears ->monitor become underPressure, and stay at underPressure for a whole window -> window ends, if slow consumer cleared out in this window, monitor become notUnderPressure.

welcome to run through more scenario with me, and let me know if this algorithm is conservative enough

or if you want something even more conservative, such as :

public bool IsUnderPressure(DateTime utcNow) { //if any pressure contribution in current period is bigger than flowControlThreshold //we see the cache is under pressure bool underPressure = this.biggestPressureInCurrentPeriod > this.FlowControlThreshold; if (underPressure) { //if under pressure, extend the nextCheckedTime, make sure wasUnderPressure is true for a whole window this.wasUnderPressure = underPressure; this.nextCheckedTime = utcNow + this.PressureWindowSize; logger.Verbose($"Ingesting messages too fast. Throttling message reading. BiggestPressureInCurrentPeriod: {biggestPressureInCurrentPeriod}, Threshold: {FlowControlThreshold}"); this.biggestPressureInCurrentPeriod = 0; } else { if (this.nextCheckedTime < utcNow) { //at the end of each check period, reset biggestPressureInCurrentPeriod this.nextCheckedTime = utcNow + this.PressureWindowSize; this.biggestPressureInCurrentPeriod = 0; this.wasUnderPressure = underPressure; if(!this.wasUnderPressure) logger.Verbose($"Message ingestion is healthy. BiggestPressureInCurrentPeriod: {biggestPressureInCurrentPeriod}, Threshold: {FlowControlThreshold}"); } } return this.wasUnderPressure; }

isUnderPressure will be true until slow consumer catch up to above threshold, and be healthy for a whole window

I think the logic works. It immediately stops reading if there is a cursor past the threshold, but if currently under pressure or no cursor is past the threshold, it only checks again every window.

xiazen · 2017-04-05T17:07:30Z

@dotnet-bot test this please

xiazen · 2017-04-05T19:27:41Z

functionals do succeed for me in vso .

jdom · 2017-04-05T19:54:57Z

@dotnet-bot test netfx-functional
It might be a non-deterministic test failing

jason-bragg · 2017-04-06T14:51:14Z

...sServiceBus/Providers/Streams/EventHub/CachePressureMonitors/SlowConsumingPressureMonitor.cs

+    /// </summary>
+    public class SlowConsumingPressureMonitor : ICachePressureMonitor
+    {
+        private static TimeSpan DefaultPressureWindowSize = TimeSpan.FromMinutes(1);


Defaults should probably be public, so users can specify them in configurations if they want to be explicit.

dnfclas added the cla-already-signed label Mar 22, 2017

xiazen added the Status: work-in-progress label Mar 22, 2017

xiazen force-pushed the cachemiss branch from 4f2fa2d to 64aac82 Compare March 23, 2017 00:15

xiazen changed the title ~~adding CursorUsageBasedFlowControl to EventHubQueueCache~~ add SlowConsumingPressureMonitor Mar 23, 2017

xiazen force-pushed the cachemiss branch 2 times, most recently from 65587d8 to 93e2deb Compare March 23, 2017 00:18

ReubenBond reviewed Mar 23, 2017

View reviewed changes

xiazen force-pushed the cachemiss branch from a16db73 to 03bbd1d Compare March 23, 2017 19:49

jason-bragg reviewed Mar 23, 2017

View reviewed changes

xiazen added 3 commits March 28, 2017 22:53

add SlowConsumingPressureMonitor

378d76b

add one option to configure EventHubQeueuCache with the SlowConsuming…

300db7c

…PressureMonitor

add configure option through EventHubProviderSettings

9372367

xiazen force-pushed the cachemiss branch 4 times, most recently from 06db2fa to 68d77e0 Compare March 30, 2017 02:31

xiazen removed the Status: work-in-progress label Mar 30, 2017

PR feedback and add testing

6908027

xiazen force-pushed the cachemiss branch from 68d77e0 to 6908027 Compare March 30, 2017 17:27

xiazen commented Mar 30, 2017

View reviewed changes

jason-bragg reviewed Mar 30, 2017

View reviewed changes

jdom reviewed Mar 31, 2017

View reviewed changes

PR feedback

6562a19

jason-bragg reviewed Apr 4, 2017

View reviewed changes

jason-bragg self-assigned this Apr 4, 2017

PR feedback

96f7de0

xiazen force-pushed the cachemiss branch from 3761247 to 96f7de0 Compare April 4, 2017 20:37

jdom reviewed Apr 4, 2017

View reviewed changes

xiazen force-pushed the cachemiss branch from ac14c58 to 96f7de0 Compare April 4, 2017 21:09

make slowpressuremonitor more conservative

a728bca

xiazen force-pushed the cachemiss branch from f11c1ac to a728bca Compare April 5, 2017 00:18

xiazen commented Apr 5, 2017

View reviewed changes

jason-bragg reviewed Apr 6, 2017

View reviewed changes

jason-bragg approved these changes Apr 6, 2017

View reviewed changes

make SlowConsumerMonitor.DefaultWindowSize public

bfcc1fb

jason-bragg merged commit bb1db69 into dotnet:master Apr 7, 2017

jdom changed the title ~~add SlowConsumingPressureMonitor~~ Add SlowConsumingPressureMonitor Apr 25, 2017

xiazen mentioned this pull request Jun 8, 2017

Subtle but important behavior change in orleans streaming coming with 1.5 #3104

Closed

xiazen deleted the cachemiss branch November 1, 2017 20:31

github-actions bot locked and limited conversation to collaborators Dec 9, 2023


		private readonly Fixture fixture;

		public class Fixture : BaseTestClusterFixture

Add SlowConsumingPressureMonitor #2873

Add SlowConsumingPressureMonitor #2873

Conversation

xiazen commented Mar 22, 2017

xiazen commented Mar 22, 2017 • edited Loading

xiazen commented Mar 23, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiazen commented Mar 24, 2017

xiazen commented Mar 24, 2017

xiazen Mar 30, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jdom commented Mar 31, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jdom commented Apr 3, 2017

xiazen commented Apr 3, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiazen Apr 4, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiazen commented Apr 5, 2017

xiazen commented Apr 5, 2017

jdom commented Apr 5, 2017

Choose a reason for hiding this comment

xiazen commented Mar 22, 2017 •

edited

Loading

xiazen commented Mar 23, 2017 •

edited

Loading

xiazen Mar 30, 2017 •

edited

Loading

xiazen Apr 4, 2017 •

edited

Loading