Better behaviour in the presence of 429s (#786)

Author: j-baker

http-clients/src/main/java/com/palantir/remoting3/clients/ClientConfigurations.java

@@ -44,7 +44,7 @@ public final class ClientConfigurations {
     private static final Duration DEFAULT_FAILED_URL_COOLDOWN = Duration.ZERO;
     private static final boolean DEFAULT_ENABLE_GCM_CIPHERS = false;
     private static final NodeSelectionStrategy DEFAULT_NODE_SELECTION_STRATEGY = NodeSelectionStrategy.PIN_UNTIL_ERROR;
-    private static final int DEFAULT_MAX_NUM_RETRIES = 3;
+    private static final int DEFAULT_MAX_NUM_RETRIES = 4;
 
     private ClientConfigurations() {}
 
@@ -87,7 +87,7 @@ public static ClientConfiguration of(
                 .enableGcmCipherSuites(DEFAULT_ENABLE_GCM_CIPHERS)
                 .proxy(ProxySelector.getDefault())
                 .proxyCredentials(Optional.empty())
-                .maxNumRetries(uris.size())
+                .maxNumRetries(DEFAULT_MAX_NUM_RETRIES)
                 .backoffSlotSize(DEFAULT_BACKOFF_SLOT_SIZE)
                 .nodeSelectionStrategy(DEFAULT_NODE_SELECTION_STRATEGY)
                 .failedUrlCooldown(DEFAULT_FAILED_URL_COOLDOWN)

jaxrs-clients/versions.lock

@@ -99,6 +99,12 @@
                 "com.palantir.tracing:tracing"
             ]
         },
+        "com.netflix.concurrency-limits:concurrency-limits-core": {
+            "locked": "0.1.1",
+            "transitive": [
+                "com.palantir.remoting3:okhttp-clients"
+            ]
+        },
         "com.netflix.feign:feign-core": {
             "locked": "8.17.0",
             "transitive": [
@@ -273,6 +279,7 @@
         "org.slf4j:slf4j-api": {
             "locked": "1.7.12",
             "transitive": [
+                "com.netflix.concurrency-limits:concurrency-limits-core",
                 "com.netflix.feign:feign-slf4j",
                 "com.palantir.remoting3:error-handling",
                 "com.palantir.remoting3:okhttp-clients",
@@ -383,6 +390,12 @@
                 "com.palantir.tracing:tracing"
             ]
         },
+        "com.netflix.concurrency-limits:concurrency-limits-core": {
+            "locked": "0.1.1",
+            "transitive": [
+                "com.palantir.remoting3:okhttp-clients"
+            ]
+        },
         "com.netflix.feign:feign-core": {
             "locked": "8.17.0",
             "transitive": [
@@ -557,6 +570,7 @@
         "org.slf4j:slf4j-api": {
             "locked": "1.7.12",
             "transitive": [
+                "com.netflix.concurrency-limits:concurrency-limits-core",
                 "com.netflix.feign:feign-slf4j",
                 "com.palantir.remoting3:error-handling",
                 "com.palantir.remoting3:okhttp-clients",

jaxrs-scala-clients/versions.lock

@@ -114,6 +114,12 @@
                 "com.palantir.tracing:tracing"
             ]
         },
+        "com.netflix.concurrency-limits:concurrency-limits-core": {
+            "locked": "0.1.1",
+            "transitive": [
+                "com.palantir.remoting3:okhttp-clients"
+            ]
+        },
         "com.netflix.feign:feign-core": {
             "locked": "8.17.0",
             "transitive": [
@@ -334,6 +340,7 @@
         "org.slf4j:slf4j-api": {
             "locked": "1.7.12",
             "transitive": [
+                "com.netflix.concurrency-limits:concurrency-limits-core",
                 "com.netflix.feign:feign-slf4j",
                 "com.palantir.remoting3:error-handling",
                 "com.palantir.remoting3:jaxrs-clients",
@@ -460,6 +467,12 @@
                 "com.palantir.tracing:tracing"
             ]
         },
+        "com.netflix.concurrency-limits:concurrency-limits-core": {
+            "locked": "0.1.1",
+            "transitive": [
+                "com.palantir.remoting3:okhttp-clients"
+            ]
+        },
         "com.netflix.feign:feign-core": {
             "locked": "8.17.0",
             "transitive": [
@@ -680,6 +693,7 @@
         "org.slf4j:slf4j-api": {
             "locked": "1.7.12",
             "transitive": [
+                "com.netflix.concurrency-limits:concurrency-limits-core",
                 "com.netflix.feign:feign-slf4j",
                 "com.palantir.remoting3:error-handling",
                 "com.palantir.remoting3:jaxrs-clients",

okhttp-clients/build.gradle

@@ -7,6 +7,7 @@ dependencies {
     compile project(':http-clients')
     compile project(':tracing-okhttp3')
     compile 'com.google.guava:guava'
+    compile 'com.netflix.concurrency-limits:concurrency-limits-core'
     compile 'com.palantir.safe-logging:preconditions'
     compile 'com.palantir.tritium:tritium-registry'
     compile 'com.squareup.okhttp3:logging-interceptor'

okhttp-clients/src/main/java/com/palantir/remoting3/okhttp/ConcurrencyLimiters.java

@@ -0,0 +1,124 @@
+/*
+ * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.palantir.remoting3.okhttp;
+
+import com.codahale.metrics.Timer;
+import com.google.common.annotations.VisibleForTesting;
+import com.netflix.concurrency.limits.Limiter;
+import com.netflix.concurrency.limits.limit.VegasLimit;
+import com.netflix.concurrency.limits.limiter.SimpleLimiter;
+import com.palantir.logsafe.SafeArg;
+import com.palantir.remoting3.tracing.okhttp3.OkhttpTraceInterceptor;
+import com.palantir.tritium.metrics.registry.MetricName;
+import com.palantir.tritium.metrics.registry.TaggedMetricRegistry;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import okhttp3.Request;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@SuppressWarnings("DesignForExtension")
+class ConcurrencyLimiters {
+    private static final Logger log = LoggerFactory.getLogger(ConcurrencyLimiters.class);
+    private static final Duration DEFAULT_TIMEOUT = Duration.ofMinutes(1);
+    private static final Void NO_CONTEXT = null;
+    private static final String FALLBACK = "";
+    private static final MetricName SLOW_ACQUIRE =
+            MetricName.builder().safeName("conjure-java-client.qos.request-permit.slow-acquire").build();
+
+    private final Timer slowAcquire;
+    private final ConcurrentMap<String, Limiter<Void>> limiters = new ConcurrentHashMap<>();
+    private final Duration timeout;
+
+    @VisibleForTesting
+    ConcurrencyLimiters(TaggedMetricRegistry taggedMetricRegistry, Duration timeout) {
+        this.slowAcquire = taggedMetricRegistry.timer(SLOW_ACQUIRE);
+        this.timeout = timeout;
+    }
+
+    ConcurrencyLimiters(TaggedMetricRegistry taggedMetricRegistry) {
+        this(taggedMetricRegistry, DEFAULT_TIMEOUT);
+    }
+
+    /**
+     * Blocks until the request should be allowed to proceed.
+     * Caller must notify the listener to release the permit.
+     */
+    Limiter.Listener acquireLimiter(Request request) {
+        return acquireLimiter(limiterKey(request));
+    }
+
+    @VisibleForTesting
+    Limiter.Listener acquireLimiter(String name) {
+        Limiter<Void> limiter = limiters.computeIfAbsent(name, key -> newLimiter());
+        Optional<Limiter.Listener> listener = limiter.acquire(NO_CONTEXT);
+        return listener.orElseGet(() -> {
+            if (Thread.currentThread().isInterrupted()) {
+                throw new RuntimeException("Thread was interrupted");
+            }
+            log.warn("Timed out waiting to get permits for concurrency. In most cases this would indicate "
+                            + "some kind of deadlock. We expect that either this is caused by not closing response "
+                            + "bodies (there should be OkHttp log lines indicating this), or service overloading.",
+                    SafeArg.of("timeout", timeout));
+            limiters.replace(name, limiter, newLimiter());
+            return acquireLimiter(name);
+        });
+    }
+
+    private Limiter<Void> newLimiter() {
+        Limiter<Void> limiter = new InstrumentedLimiter(SimpleLimiter.newBuilder()
+                .limit(new RemotingWindowedLimit(VegasLimit.newDefault()))
+                .build(), slowAcquire);
+        return RemotingBlockingLimiter.wrap(limiter, timeout);
+    }
+
+    private static String limiterKey(Request request) {
+        String pathTemplate = request.header(OkhttpTraceInterceptor.PATH_TEMPLATE_HEADER);
+        if (pathTemplate == null) {
+            return FALLBACK;
+        } else {
+            return request.method() + " " + pathTemplate;
+        }
+    }
+
+    private static final class InstrumentedLimiter implements Limiter<Void> {
+        private final Limiter<Void> delegate;
+        private final Timer timer;
+
+        private InstrumentedLimiter(Limiter<Void> delegate, Timer timer) {
+            this.delegate = delegate;
+            this.timer = timer;
+        }
+
+        @Override
+        public Optional<Listener> acquire(Void context) {
+            long start = System.nanoTime();
+            try {
+                return delegate.acquire(context);
+            } finally {
+                long end = System.nanoTime();
+                long durationNanos = end - start;
+                if (TimeUnit.NANOSECONDS.toMillis(durationNanos) > 1) {
+                    timer.update(durationNanos, TimeUnit.NANOSECONDS);
+                }
+            }
+        }
+    }
+}

okhttp-clients/src/main/java/com/palantir/remoting3/okhttp/ConcurrencyLimitingInterceptor.java

@@ -0,0 +1,139 @@
+/*
+ * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.palantir.remoting3.okhttp;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableSet;
+import com.netflix.concurrency.limits.Limiter;
+import com.palantir.tritium.metrics.registry.TaggedMetricRegistry;
+import java.io.IOException;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import okhttp3.Interceptor;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+
+/**
+ * Flow control in Conjure is a collaborative effort between servers and clients. Servers advertise an overloaded state
+ * via 429/503 responses, and clients throttle the number of requests that they send concurrently as a response to this.
+ * The latter is implemented as a combination of two techniques, yielding a mechanism similar to flow control in TCP/IP.
+ * <ol>
+ *     <li>
+ *         Clients use the frequency of 429/503 responses (as well as the request latency) to determine an estimate
+ *         for the number of permissible concurrent requests
+ *    </li>
+ *     <li>
+ *         Each such request gets scheduled according to an exponential backoff algorithm.
+ *     </li>
+ * </ol>
+ * <p>
+ * This class utilises Netflix's
+ * <a href="https://github.com/Netflix/concurrency-limits/">concurrency-limits</a> library for determining the
+ * above mentioned concurrency estimates.
+ * <p>
+ * 429 and 503 response codes are used for backpressure, whilst 200 -> 399 request codes are used for determining
+ * new limits and all other codes are not factored in to timings.
+ * <p>
+ * Concurrency permits are only released when the response body is closed.
+ */
+final class ConcurrencyLimitingInterceptor implements Interceptor {
+    private static final ImmutableSet<Integer> DROPPED_CODES = ImmutableSet.of(429, 503);
+
+    private final ConcurrencyLimiters limiters;
+
+    @VisibleForTesting
+    ConcurrencyLimitingInterceptor(ConcurrencyLimiters limiters) {
+        this.limiters = limiters;
+    }
+
+    ConcurrencyLimitingInterceptor(TaggedMetricRegistry taggedMetricRegistry) {
+        this(new ConcurrencyLimiters(taggedMetricRegistry));
+    }
+
+    @Override
+    public Response intercept(Chain chain) throws IOException {
+        Limiter.Listener listener = limiters.acquireLimiter(chain.request());
+        try {
+            Response response = chain.proceed(chain.request());
+            if (DROPPED_CODES.contains(response.code())) {
+                listener.onDropped();
+                return response;
+            } else if (!response.isSuccessful() || response.isRedirect()) {
+                listener.onIgnore();
+                return response;
+            } else {
+                return wrapResponse(listener, response);
+            }
+        } catch (IOException e) {
+            listener.onIgnore();
+            throw e;
+        }
+    }
+
+    private static Response wrapResponse(Limiter.Listener listener, Response response) {
+        if (response.body() == null) {
+            return response;
+        }
+        ResponseBody currentBody = response.body();
+        ResponseBody newResponseBody =
+                ResponseBody.create(
+                        currentBody.contentType(),
+                        currentBody.contentLength(),
+                        wrapSource(currentBody.source(), listener));
+        return response.newBuilder()
+                .body(newResponseBody)
+                .build();
+    }
+
+    private static BufferedSource wrapSource(BufferedSource currentSource, Limiter.Listener listener) {
+        return (BufferedSource) Proxy.newProxyInstance(
+                BufferedSource.class.getClassLoader(),
+                new Class<?>[] { BufferedSource.class },
+                new ReleaseConcurrencyLimitProxy(currentSource, listener));
+    }
+
+    /**
+     * This proxy enables e.g. Okio to make additive additions to their API without breaking us.
+     */
+    private static final class ReleaseConcurrencyLimitProxy implements InvocationHandler {
+        private final BufferedSource delegate;
+        private final Limiter.Listener listener;
+        private boolean closed = false;
+
+        private ReleaseConcurrencyLimitProxy(BufferedSource delegate, Limiter.Listener listener) {
+            this.delegate = delegate;
+            this.listener = listener;
+        }
+
+        @Override
+        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
+            if (method.getName().equals("close") && !closed) {
+                closed = true;
+                listener.onSuccess();
+            }
+
+            try {
+                return method.invoke(delegate, args);
+            } catch (InvocationTargetException e) {
+                throw e.getCause();
+            }
+        }
+    }
+}