681 throughput limit #707

wojtkiewicz · 2017-02-05T20:32:17Z

Added two implementations of throughput limiter:

unlimited
fixed (uses single threshold for all topics)
dynamic (calculates dynamically thresholds per topic)

Dynamic throughput limiting is configured by following factors:

max_throughout which is a hard limit on throughout that we accept across all topics
emergency_threshold after which we trigger actual limiting procedures (below this value we won't interfere)
desired_throughput which is a desired global value that we are trying to achieve after throughout limiting
idle_threshold which is used for detecting if topic is idle (if throughout on this topic is below this value then we clear all calculated limits and forget about it)

Related to #681

pbetkier · 2017-02-05T21:10:35Z

hermes-common/src/main/java/pl/allegro/tech/hermes/common/config/Configs.java

@@ -102,6 +102,12 @@
    FRONTEND_GRACEFUL_SHUTDOWN_INITIAL_WAIT_MS("frontend.graceful.shutdown.initial.wait.ms", 10000),
    FRONTEND_HTTP2_ENABLED("frontend.http2.enabled", false),
    FRONTEND_FORCE_TOPIC_MAX_MESSAGE_SIZE("frontend.force.topic.max.message.size", false),
+    FRONTEND_THROUGHPUT_TYPE("frontend.throughput.type", "unlimited"),
+    FRONTEND_THROUGHPUT_MAX("frontend.throughput.max", Long.MAX_VALUE),


If these 5 parameters apply to dynamic implementation only, then let's put them in a proper namespace, e.g. frontend.throughput.dynamic.max.

pbetkier · 2017-02-05T21:14:09Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/DynamicThroughputLimiter.java

+        users.entrySet().removeIf(entry -> entry.getValue().getOneMinuteRate() <= idleThreshold);
+        int userCount = users.size();
+        if (userCount > 0) {
+            long total = users.reduceValuesToLong(10, Throughput::getRoundedOneMinuteRate, 0, ((left, right) -> left + right));


Introduce a private method for the right hand side please, I don't know what we're doing here ;)

pbetkier · 2017-02-05T21:27:34Z

...c/test/groovy/pl/allegro/tech/hermes/frontend/publishing/DynamicThroughputLimiterTest.groovy

+    def limiter = new DynamicThroughputLimiter(max, threshold, desired, idleThreshold, 1, globalMeter, executor)
+
+
+    def "global allow"() {


All test names: how about we stick to our convention of test names specifying "how the system should behave in a particular situation" instead of a 2-word scenario identifier?

yes, please use should .... when ... convention

pbetkier · 2017-02-05T21:31:05Z

...c/test/groovy/pl/allegro/tech/hermes/frontend/publishing/DynamicThroughputLimiterTest.groovy

+        abuserRate = 2001
+
+        expect:
+        !checkThroughput(abuser, abuserMeter)


It's not clear to me what to expect from this method without looking into it's implementation. It checks whether the throughput check passes for this given topic. How about throughputCheckPasses or isThroughputCheckPassing?

adamdubiel · 2017-02-06T07:50:55Z

I noticed sth strange in version that is already on master. At /topics/{topic}/subscriptions/{sub}/metrics there is throughput field which holds path to metrics in Graphite:

"throughput": "sumSeries(stats.tech.hermes.consumer.*.throughput.pl_allegro_tech_hermes.monitor.pl_allegro_tech_hermes_monitor_consumer1.m1_rate)",

This should not be like this, paths to graphite are calculated in console. Please fix this.

adamdubiel · 2017-02-06T08:00:04Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/DynamicThroughputLimiter.java

+    }
+
+    @Override
+    public void check(TopicName topic, Metered rate) {


this method uses exceptions as flow control - i think it should return proper values instead.

adamdubiel · 2017-02-06T08:02:18Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/DynamicThroughputLimiter.java

+
+    private void calibrateLimits() {
+        users.entrySet().removeIf(entry -> entry.getValue().getOneMinuteRate() <= idleThreshold);
+        int userCount = users.size();


i see a lot of potential for race conditions here - you remove entries here while check (which as i see is called for each request) makes putIfAbsent, then you check for size.. will it not create strange results?

not really because removal is done only for idle users, it takes minutes to become idle in the current implementation

furthermore it is rather difficult to publish messages and be idle at the same time

adamdubiel · 2017-02-06T08:03:43Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/DynamicThroughputLimiter.java

+
+    @Override
+    public void check(TopicName topic, Metered rate) {
+        if (!users.containsKey(topic)) {


why do you make if contains followed by put if absent, which is the same but actually atomic?

adamdubiel · 2017-02-06T08:05:11Z

...nd/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/MessageReadHandler.java

@@ -130,6 +139,9 @@ private void messageRead(HttpServerExchange exchange, byte[] messageContent, Att
        try {
            contentLengthChecker.check(exchange, messageContent.length, attachment);
            attachment.getCachedTopic().reportMessageContentSize(messageContent.length);
+            throughputLimiter.check(


why do we check throughput in 2 places? here and handleRequest

main reason is not to read entire requests of abusive client

adamdubiel · 2017-02-06T08:06:19Z

...end/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ThroughputLimiter.java

+import static java.lang.String.format;
+
+public interface ThroughputLimiter {
+    void check(TopicName topic, Metered throughput);


this interfaces contract is hidden behind assumptions about throwing exceptions of proper kinds which are used as flow control - please return values

adamdubiel · 2017-02-06T08:10:39Z

...c/test/groovy/pl/allegro/tech/hermes/frontend/publishing/DynamicThroughputLimiterTest.groovy

+
+    def "global block"() {
+        given:
+        globalRate = 12_000


i don't like the idea of buliding tests based on a class state. please be explicit, i.e.

given: globalMeter = globalMeter(12_000)

mocking using spock makes this operation very verbose because I can't dynamically change this value until test is over which violates idea of throughput that changes in time

adamdubiel · 2017-02-06T08:11:04Z

...c/test/groovy/pl/allegro/tech/hermes/frontend/publishing/DynamicThroughputLimiterTest.groovy

+    def limiter = new DynamicThroughputLimiter(max, threshold, desired, idleThreshold, 1, globalMeter, executor)
+
+
+    def "global allow"() {


yes, please use should .... when ... convention

adamdubiel · 2017-02-06T08:12:08Z

integration/src/integration/java/pl/allegro/tech/hermes/integration/PublishingTest.java

+    public void shouldReturn429ForQuotaViolation() {
+        // given
+        Topic topic = operations.buildTopic("publishAndConsumeGroup", "topic");
+        TestMessage message = TestMessage.of("content", StringUtils.repeat("X", 60_000));


this 60_000 - where does it come from? if this is based on config, please add this as test config instead of relying on defaults and add comment that describes origin of this magic number

added comment

dankraw · 2017-02-06T08:57:07Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/DynamicThroughputLimiter.java

+    private static class Throughput {
+        Metered current;
+        volatile long max;
+        long creationTime;


creationTime - not needed?

yes, thanks

dankraw · 2017-02-06T08:58:50Z

hermes-api/src/main/java/pl/allegro/tech/hermes/api/ErrorCode.java

@@ -39,6 +39,7 @@
    OAUTH_PROVIDER_ALREADY_EXISTS(BAD_REQUEST),
    SUPPORT_TEAMS_COULD_NOT_BE_LOADED(INTERNAL_SERVER_ERROR),
    TOPIC_BLACKLISTED(FORBIDDEN),
+    QUOTA_VIOLATION(429),


maybe THROUGHPUT_QUOTA_VIOLATION to be clear and let's create a constant for 429 (TOO_MANY_REQUESTS) if there is no available already.

dankraw · 2017-02-06T09:32:40Z

...nd/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/MessageReadHandler.java

@@ -141,12 +153,25 @@ private void messageRead(HttpServerExchange exchange, byte[] messageContent, Att
            attachment.removeTimeout();
            messageErrorProcessor.sendAndLog(exchange, attachment.getTopic(),
                    attachment.getMessageId(), error(e.getMessage(), VALIDATION_ERROR));
+        } catch (FixedThroughputLimiter.QuotaViolationException e) {


Why FixedThroughputLimiter.* and notThroughputLimiter.*?
I would suggest refactoring QuotaViolation exceptions making them proper top level classes.

sputnikci · 2017-02-06T10:52:48Z

...end/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ThroughputLimiter.java

+        }
+
+        public String getReason() {
+            return reason != null? reason : DEFAULT_REASON;


[Checkstyle] ERROR: '?' is not preceded with whitespace.

sputnikci · 2017-02-06T10:52:48Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ThroughputLimiterFactory.java

+    private ConfigFactory configs;
+    private HermesMetrics hermesMetrics;
+
+    private enum ThroughputLimiterType {UNLIMITED, FIXED, DYNAMIC}


[Checkstyle] ERROR: '{' is not followed by whitespace.

sputnikci · 2017-02-06T10:52:48Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ThroughputLimiterFactory.java

+    private ConfigFactory configs;
+    private HermesMetrics hermesMetrics;
+
+    private enum ThroughputLimiterType {UNLIMITED, FIXED, DYNAMIC}


[Checkstyle] ERROR: '}' is not preceded with whitespace.

sputnikci · 2017-02-06T10:56:59Z

hermes-api/src/main/java/pl/allegro/tech/hermes/api/ErrorCode.java

@@ -47,6 +48,10 @@ private ErrorCode(Response.Status httpCode) {
        this.httpCode = httpCode.getStatusCode();
    }

+    private ErrorCode(int httpCode) {


[Checkstyle] ERROR: Redundant 'private' modifier.

sputnikci · 2017-02-06T10:56:59Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/DynamicThroughputLimiter.java

+        users.entrySet().removeIf(entry -> entry.getValue().getOneMinuteRate() <= idleThreshold);
+        int userCount = users.size();
+        if (userCount > 0) {
+            long total = users.reduceValuesToLong(10, Throughput::getRoundedOneMinuteRate, 0, ((left, right) -> left + right));


[Checkstyle] INFO: '10' is a magic number.

sputnikci · 2017-02-06T10:57:00Z

...rc/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/FixedThroughputLimiter.java

+    @Override
+    public QuotaInsight checkQuota(TopicName topic, Metered throughput) {
+        long rate = (long) Math.floor(throughput.getOneMinuteRate());
+        return rate > limit? quotaViolation(rate, limit) : quotaConfirmed();


[Checkstyle] ERROR: '?' is not preceded with whitespace.

sputnikci · 2017-02-06T10:57:00Z

...end/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ThroughputLimiter.java

+        }
+
+        public String getReason() {
+            return reason != null? reason : DEFAULT_REASON;


[Checkstyle] ERROR: '?' is not preceded with whitespace.

sputnikci · 2017-02-06T10:57:00Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ThroughputLimiterFactory.java

+    private ConfigFactory configs;
+    private HermesMetrics hermesMetrics;
+
+    private enum ThroughputLimiterType {UNLIMITED, FIXED, DYNAMIC}


[Checkstyle] ERROR: '{' is not followed by whitespace.

sputnikci · 2017-02-06T10:57:00Z

.../main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/ThroughputLimiterFactory.java

+    private ConfigFactory configs;
+    private HermesMetrics hermesMetrics;
+
+    private enum ThroughputLimiterType {UNLIMITED, FIXED, DYNAMIC}


[Checkstyle] ERROR: '}' is not preceded with whitespace.

wojtkiewicz added the in progress label Feb 5, 2017

wojtkiewicz changed the base branch from master to develop February 5, 2017 20:32

pbetkier reviewed Feb 5, 2017

View reviewed changes

adamdubiel requested changes Feb 6, 2017

View reviewed changes

dankraw reviewed Feb 6, 2017

View reviewed changes

wojtkiewicz force-pushed the 681_throughput_limit branch 3 times, most recently from 75601c9 to d6a277e Compare February 6, 2017 10:43

sputnikci reviewed Feb 6, 2017

View reviewed changes

wojtkiewicz force-pushed the 681_throughput_limit branch 2 times, most recently from 5d8d6e4 to a2755ee Compare February 6, 2017 14:43

adamdubiel approved these changes Feb 6, 2017

View reviewed changes

pbetkier approved these changes Feb 7, 2017

View reviewed changes

wojtkiewicz added 2 commits February 17, 2017 11:28

681 added simple throughput limiting

b30906a

681 added dynamic throughput limiting

9755cec

wojtkiewicz force-pushed the 681_throughput_limit branch from a2755ee to 9755cec Compare February 17, 2017 10:30

adamdubiel removed the in progress label Feb 17, 2017

wojtkiewicz merged commit e4d8081 into develop Feb 21, 2017

adamdubiel added this to the 0.11.1 milestone Feb 28, 2017

wojtkiewicz mentioned this pull request Mar 7, 2017

Preparing release 0.11.1 #736

Merged

adamdubiel deleted the 681_throughput_limit branch October 25, 2017 13:18

dswiecki mentioned this pull request Nov 4, 2022

Throughput properties documentation #1615

Open

		def limiter = new DynamicThroughputLimiter(max, threshold, desired, idleThreshold, 1, globalMeter, executor)


		def "global allow"() {

681 throughput limit #707

681 throughput limit #707

Conversation

wojtkiewicz commented Feb 5, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pbetkier Feb 5, 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

adamdubiel commented Feb 6, 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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wojtkiewicz commented Feb 5, 2017 •

edited

Loading

pbetkier Feb 5, 2017 •

edited

Loading