Merge pull request #781 from zalando/aruha-1322

aruha-1322: set max per route open connections to 20

src/main/java/org/zalando/nakadi/config/AuthenticationConfig.java

@@ -2,43 +2,133 @@
 
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.Timer;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Profile;
+import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
 import org.springframework.security.core.AuthenticationException;
+import org.springframework.security.oauth2.common.OAuth2AccessToken;
 import org.springframework.security.oauth2.provider.OAuth2Authentication;
 import org.springframework.security.oauth2.provider.token.ResourceServerTokenServices;
+import org.springframework.web.client.RestTemplate;
 import org.zalando.nakadi.metrics.MetricUtils;
+import org.zalando.nakadi.util.FeatureToggleService;
+import org.zalando.stups.oauth2.spring.authorization.DefaultUserRolesProvider;
+import org.zalando.stups.oauth2.spring.server.DefaultAuthenticationExtractor;
 import org.zalando.stups.oauth2.spring.server.TokenInfoResourceServerTokenServices;
+import org.zalando.stups.oauth2.spring.server.TokenResponseErrorHandler;
 
 @Configuration
 @Profile("!test")
 public class AuthenticationConfig {
 
     @Bean
     public ResourceServerTokenServices zalandoResourceTokenServices(final SecuritySettings settings,
-                                                                    final MetricRegistry metricRegistry) {
-        return new MeasuringTokenInfoResourceServerTokenServices(
-                settings.getTokenInfoUrl(), settings.getClientId(), metricRegistry);
+                                                                    final MetricRegistry metricRegistry,
+                                                                    final RestTemplate restTemplate,
+                                                                    final FeatureToggleService featureToggleService) {
+        return new MeasureAndDispatchResourceServerTokenServices(
+                metricRegistry, settings, restTemplate, featureToggleService);
     }
 
-    public static class MeasuringTokenInfoResourceServerTokenServices extends TokenInfoResourceServerTokenServices {
+    @Bean
+    public PoolingHttpClientConnectionManager poolingHttpClientConnectionManager(
+            @Value("${nakadi.http.pool.connection.max.total}") final int maxTotal,
+            @Value("${nakadi.http.pool.connection.max.per.route}") final int maxPerRoute) {
+        final PoolingHttpClientConnectionManager result = new PoolingHttpClientConnectionManager();
+        result.setMaxTotal(maxTotal);
+        result.setDefaultMaxPerRoute(maxPerRoute);
+        return result;
+    }
+
+    @Bean
+    public RequestConfig requestConfig(
+            @Value("${nakadi.http.pool.connection.request.timeout}") final int requestTimeout,
+            @Value("${nakadi.http.pool.connection.connect.timeout}") final int connectTimeout,
+            @Value("${nakadi.http.pool.connection.socket.timeout}") final int socketTimeout) {
+        final RequestConfig result = RequestConfig.custom()
+                .setConnectionRequestTimeout(requestTimeout)
+                .setConnectTimeout(connectTimeout)
+                .setSocketTimeout(socketTimeout)
+                .build();
+        return result;
+    }
+
+    @Bean
+    public CloseableHttpClient httpClient(final PoolingHttpClientConnectionManager poolingHttpClientConnectionManager,
+                                          final RequestConfig requestConfig) {
+        final CloseableHttpClient result = HttpClientBuilder
+                .create()
+                .setConnectionManager(poolingHttpClientConnectionManager)
+                .setDefaultRequestConfig(requestConfig)
+                .build();
+        return result;
+    }
+
+    @Bean
+    public RestTemplate restTemplate(final HttpClient httpClient) {
+        final HttpComponentsClientHttpRequestFactory requestFactory = new HttpComponentsClientHttpRequestFactory();
+        requestFactory.setHttpClient(httpClient);
+        final RestTemplate restTemplate = new RestTemplate(requestFactory);
+        restTemplate.setErrorHandler(TokenResponseErrorHandler.getDefault());
+        return restTemplate;
+    }
+
+    public static class MeasureAndDispatchResourceServerTokenServices implements ResourceServerTokenServices {
+
         private final Timer timer;
+        private final TokenInfoResourceServerTokenServices remoteService;
+        private final TokenInfoResourceServerTokenServices localService;
+        private final FeatureToggleService featureToggleService;
 
-        public MeasuringTokenInfoResourceServerTokenServices(final String tokenInfoEndpointUrl, final String clientId,
-                                                             final MetricRegistry metricRegistry) {
-            super(tokenInfoEndpointUrl, clientId);
+        public MeasureAndDispatchResourceServerTokenServices(final MetricRegistry metricRegistry,
+                                                             final SecuritySettings securitySettings,
+                                                             final RestTemplate restTemplate,
+                                                             final FeatureToggleService featureToggleService) {
+            remoteService = new TokenInfoResourceServerTokenServices(
+                    securitySettings.getTokenInfoUrl(),
+                    securitySettings.getClientId(),
+                    new DefaultAuthenticationExtractor(),
+                    new DefaultUserRolesProvider(),
+                    restTemplate);
+            localService = new TokenInfoResourceServerTokenServices(
+                    securitySettings.getLocalTokenInfoUrl(),
+                    securitySettings.getClientId(),
+                    new DefaultAuthenticationExtractor(),
+                    new DefaultUserRolesProvider(),
+                    restTemplate);
             timer = metricRegistry.timer(MetricUtils.NAKADI_PREFIX + "general.accessTokenValidation");
+            this.featureToggleService = featureToggleService;
         }
 
         @Override
         public OAuth2Authentication loadAuthentication(final String accessToken) throws AuthenticationException {
             final Timer.Context context = timer.time();
             try {
-                return super.loadAuthentication(accessToken);
+                if (featureToggleService.isFeatureEnabled(FeatureToggleService.Feature.REMOTE_TOKENINFO)) {
+                    return remoteService.loadAuthentication(accessToken);
+                } else {
+                    return localService.loadAuthentication(accessToken);
+                }
             } finally {
                 context.stop();
             }
         }
+
+        @Override
+        public OAuth2AccessToken readAccessToken(final String accessToken) {
+            if (featureToggleService.isFeatureEnabled(FeatureToggleService.Feature.REMOTE_TOKENINFO)) {
+                return remoteService.readAccessToken(accessToken);
+            } else {
+                return localService.readAccessToken(accessToken);
+            }
+        }
     }
+
 }

src/main/java/org/zalando/nakadi/config/SecuritySettings.java

@@ -17,16 +17,19 @@ public enum AuthMode {
     }
 
     private final String tokenInfoUrl;
+    private final String localTokenInfoUrl;
     private final String clientId;
     private final AuthMode authMode;
     private final String adminClientId;
 
     @Autowired
     public SecuritySettings(@Value("${nakadi.oauth2.tokenInfoUrl}") final String tokenInfoUrl,
+                            @Value("${nakadi.oauth2.localTokenInfoUrl}") final String localTokenInfoUrl,
                             @Value("${nakadi.oauth2.clientId}") final String clientId,
                             @Value("${nakadi.oauth2.mode:BASIC}") final AuthMode authMode,
                             @Value("${nakadi.oauth2.adminClientId}") final String adminClientId) {
         this.tokenInfoUrl = tokenInfoUrl;
+        this.localTokenInfoUrl = localTokenInfoUrl;
         this.clientId = clientId;
         this.authMode = authMode;
         this.adminClientId = adminClientId;
@@ -47,4 +50,8 @@ public AuthMode getAuthMode() {
     public String getAdminClientId() {
         return adminClientId;
     }
+
+    public String getLocalTokenInfoUrl() {
+        return localTokenInfoUrl;
+    }
 }

src/main/java/org/zalando/nakadi/util/FeatureToggleService.java

@@ -36,7 +36,8 @@ enum Feature {
         CHECK_PARTITIONS_KEYS("check_partitions_keys"),
         CHECK_OWNING_APPLICATION("check_owning_application"),
         LIMIT_CONSUMERS_NUMBER("limit_consumers_number"),
-        SEND_BATCH_VIA_OUTPUT_STREAM("send_batch_via_output_stream");
+        SEND_BATCH_VIA_OUTPUT_STREAM("send_batch_via_output_stream"),
+        REMOTE_TOKENINFO("remote_tokeninfo");
 
         private final String id;
 

src/main/resources/application.yml

@@ -69,6 +69,7 @@ nakadi:
     mode: BASIC
     adminClientId: adminClientId
     tokenInfoUrl: https://example.com/tokeninfo
+    localTokenInfoUrl: "http://localhost:9021/oauth2/tokeninfo"
     clientId: stups_aruha-event-store-poc
     realms: '/arealm, /anotherone'
     scopes:
@@ -92,6 +93,12 @@ nakadi:
       runPeriodMs: 3600000 # 1 hour
       deletionDelayMs: 2000 # 2 seconds, to be on the safe side
     consumerNodesCleanup.runPeriodMs: 21600000 # 6 hours
+  http.pool.connection:
+      max.total: 20
+      max.per.route: 10
+      request.timeout: 2000
+      connect.timeout: 1000
+      socket.timeout: 2000
   timelines.storage.default: "default"
 
 twintip: