fix(cf): CF forkjoin threading improvements (#5071)

* feat(kubernetes): Send SIGKILL to kubectl * fix(cf): CF forkjoin threading improvements
spinnaker · Nov 5, 2020 · 1f7d7b7 · 1f7d7b7
1 parent 54b52cb
commit 1f7d7b7
Show file tree

Hide file tree

Showing 16 changed files with 157 additions and 33 deletions.
diff --git a/...dry/src/main/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/Applications.java b/...dry/src/main/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/Applications.java
@@ -75,15 +75,15 @@ public Applications(
       ApplicationService api,
       Spaces spaces,
       Integer resultsPerPage,
-      int maxConnections) {
+      ForkJoinPool forkJoinPool) {
     this.account = account;
     this.appsManagerUri = appsManagerUri;
     this.metricsUri = metricsUri;
     this.api = api;
     this.spaces = spaces;
     this.resultsPerPage = resultsPerPage;
 
-    this.forkJoinPool = new ForkJoinPool(maxConnections);
+    this.forkJoinPool = forkJoinPool;
     this.serverGroupCache =
         CacheBuilder.newBuilder()
             .build(

diff --git a/...in/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/HttpCloudFoundryClient.java b/...in/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/HttpCloudFoundryClient.java
@@ -51,6 +51,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Matcher;
@@ -181,7 +182,7 @@ public HttpCloudFoundryClient(
       String password,
       boolean skipSslValidation,
       Integer resultsPerPage,
-      int maxCapiConnectionsForCache) {
+      ForkJoinPool forkJoinPool) {
     this.apiHost = apiHost;
     this.user = user;
     this.password = password;
@@ -217,7 +218,7 @@ public HttpCloudFoundryClient(
             createService(ApplicationService.class),
             spaces,
             resultsPerPage,
-            maxCapiConnectionsForCache);
+            forkJoinPool);
     this.domains = new Domains(createService(DomainService.class), organizations);
     this.serviceInstances =
         new ServiceInstances(
@@ -233,7 +234,7 @@ public HttpCloudFoundryClient(
             domains,
             spaces,
             resultsPerPage,
-            maxCapiConnectionsForCache);
+            forkJoinPool);
     this.serviceKeys = new ServiceKeys(createService(ServiceKeyService.class), spaces);
     this.tasks = new Tasks(createService(TaskService.class));
 

diff --git a/...udfoundry/src/main/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/Routes.java b/...udfoundry/src/main/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/Routes.java
@@ -67,15 +67,15 @@ public Routes(
       Domains domains,
       Spaces spaces,
       Integer resultsPerPage,
-      int maxConnections) {
+      ForkJoinPool forkJoinPool) {
     this.account = account;
     this.api = api;
     this.applications = applications;
     this.domains = domains;
     this.spaces = spaces;
     this.resultsPerPage = resultsPerPage;
 
-    this.forkJoinPool = new ForkJoinPool(maxConnections);
+    this.forkJoinPool = forkJoinPool;
     this.routeMappings =
         CacheBuilder.newBuilder()
             .expireAfterWrite(3, TimeUnit.MINUTES)

diff --git a/...etflix/spinnaker/clouddriver/cloudfoundry/config/CloudFoundryConfigurationProperties.java b/...etflix/spinnaker/clouddriver/cloudfoundry/config/CloudFoundryConfigurationProperties.java
@@ -45,6 +45,8 @@ public class CloudFoundryConfigurationProperties implements DisposableBean {
 
   private List<ManagedAccount> accounts = new ArrayList<>();
 
+  private int apiRequestParallelism = 100;
+
   @Override
   public void destroy() {
     this.accounts = new ArrayList<>();
@@ -63,7 +65,11 @@ public static class ManagedAccount implements CredentialsDefinition {
     private String environment;
     private boolean skipSslValidation;
     private Integer resultsPerPage;
-    private Integer maxCapiConnectionsForCache;
+
+    @Deprecated
+    private Integer
+        maxCapiConnectionsForCache; // Deprecated in favor of cloudfoundry.apiRequestParallelism
+
     private Permissions.Builder permissions = new Permissions.Builder();
   }
 }
diff --git a/...etflix/spinnaker/clouddriver/cloudfoundry/provider/config/CloudFoundryProviderConfig.java b/...etflix/spinnaker/clouddriver/cloudfoundry/provider/config/CloudFoundryProviderConfig.java
@@ -28,6 +28,7 @@
 import com.netflix.spinnaker.credentials.definition.BasicCredentialsLoader;
 import com.netflix.spinnaker.credentials.definition.CredentialsDefinitionSource;
 import com.netflix.spinnaker.credentials.poller.Poller;
+import java.util.concurrent.ForkJoinPool;
 import javax.annotation.Nullable;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
@@ -41,6 +42,12 @@ public CloudFoundryProvider cloudFoundryProvider() {
     return new CloudFoundryProvider();
   }
 
+  @Bean
+  public ForkJoinPool cloudFoundryThreadPool(
+      CloudFoundryConfigurationProperties cloudFoundryConfigurationProperties) {
+    return new ForkJoinPool(cloudFoundryConfigurationProperties.getApiRequestParallelism());
+  }
+
   @Bean
   @ConditionalOnMissingBean(
       value = CloudFoundryCredentials.class,
@@ -51,7 +58,8 @@ public AbstractCredentialsLoader<CloudFoundryCredentials> cloudFoundryCredential
               cloudFoundryCredentialSource,
       CloudFoundryConfigurationProperties configurationProperties,
       CacheRepository cacheRepository,
-      CredentialsRepository<CloudFoundryCredentials> cloudFoundryCredentialsRepository) {
+      CredentialsRepository<CloudFoundryCredentials> cloudFoundryCredentialsRepository,
+      ForkJoinPool cloudFoundryThreadPool) {
 
     if (cloudFoundryCredentialSource == null) {
       cloudFoundryCredentialSource = configurationProperties::getAccounts;
@@ -69,9 +77,9 @@ public AbstractCredentialsLoader<CloudFoundryCredentials> cloudFoundryCredential
                 a.getEnvironment(),
                 a.isSkipSslValidation(),
                 a.getResultsPerPage(),
-                a.getMaxCapiConnectionsForCache(),
                 cacheRepository,
-                a.getPermissions().build()),
+                a.getPermissions().build(),
+                cloudFoundryThreadPool),
         cloudFoundryCredentialsRepository);
   }
 

diff --git a/...java/com/netflix/spinnaker/clouddriver/cloudfoundry/security/CloudFoundryCredentials.java b/...java/com/netflix/spinnaker/clouddriver/cloudfoundry/security/CloudFoundryCredentials.java
@@ -31,6 +31,7 @@
 import com.netflix.spinnaker.clouddriver.security.AbstractAccountCredentials;
 import com.netflix.spinnaker.fiat.model.resources.Permissions;
 import java.util.*;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -68,8 +69,6 @@ public class CloudFoundryCredentials extends AbstractAccountCredentials<CloudFou
 
   @Nullable private final Integer resultsPerPage;
 
-  private final int maxCapiConnectionsForCache;
-
   private final Supplier<List<CloudFoundrySpace>> spaceSupplier =
       Memoizer.memoizeWithExpiration(this::spaceSupplier, SPACE_EXPIRY_SECONDS, TimeUnit.SECONDS);
 
@@ -79,6 +78,8 @@ public class CloudFoundryCredentials extends AbstractAccountCredentials<CloudFou
 
   private Permissions permissions;
 
+  private final ForkJoinPool forkJoinPool;
+
   public CloudFoundryCredentials(
       String name,
       String appsManagerUri,
@@ -89,9 +90,9 @@ public CloudFoundryCredentials(
       String environment,
       boolean skipSslValidation,
       Integer resultsPerPage,
-      Integer maxCapiConnectionsForCache,
       CacheRepository cacheRepository,
-      Permissions permissions) {
+      Permissions permissions,
+      ForkJoinPool forkJoinPool) {
     this.name = name;
     this.appsManagerUri = appsManagerUri;
     this.metricsUri = metricsUri;
@@ -101,9 +102,9 @@ public CloudFoundryCredentials(
     this.environment = Optional.ofNullable(environment).orElse("dev");
     this.skipSslValidation = skipSslValidation;
     this.resultsPerPage = Optional.ofNullable(resultsPerPage).orElse(100);
-    this.maxCapiConnectionsForCache = Optional.ofNullable(maxCapiConnectionsForCache).orElse(16);
     this.cacheRepository = cacheRepository;
     this.permissions = permissions == null ? Permissions.EMPTY : permissions;
+    this.forkJoinPool = forkJoinPool;
   }
 
   public CloudFoundryClient getCredentials() {
@@ -118,7 +119,7 @@ public CloudFoundryClient getCredentials() {
               password,
               skipSslValidation,
               resultsPerPage,
-              maxCapiConnectionsForCache);
+              forkJoinPool);
     }
     return credentials;
   }

diff --git a/...c/test/java/com/netflix/spinnaker/clouddriver/cloudfoundry/cache/CacheRepositoryTest.java b/...c/test/java/com/netflix/spinnaker/clouddriver/cloudfoundry/cache/CacheRepositoryTest.java
@@ -41,6 +41,7 @@
 import com.netflix.spinnaker.clouddriver.cloudfoundry.security.CloudFoundryCredentials;
 import com.netflix.spinnaker.clouddriver.model.HealthState;
 import java.util.List;
+import java.util.concurrent.ForkJoinPool;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -121,9 +122,9 @@ private CloudFoundryCredentials createCredentials(String name) {
         null,
         false,
         null,
-        16,
         repo,
-        null);
+        null,
+        ForkJoinPool.commonPool());
   }
 
   @Test

diff --git a/...src/test/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/ApplicationsTest.java b/...src/test/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/ApplicationsTest.java
@@ -36,6 +36,7 @@
 import io.vavr.collection.HashMap;
 import java.time.ZonedDateTime;
 import java.util.*;
+import java.util.concurrent.ForkJoinPool;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -47,7 +48,13 @@ class ApplicationsTest {
   private Spaces spaces = mock(Spaces.class);
   private Applications apps =
       new Applications(
-          "pws", "some-apps-man-uri", "some-metrics-uri", applicationService, spaces, 500, 16);
+          "pws",
+          "some-apps-man-uri",
+          "some-metrics-uri",
+          applicationService,
+          spaces,
+          500,
+          ForkJoinPool.commonPool());
   private String spaceId = "space-guid";
   private CloudFoundrySpace cloudFoundrySpace =
       CloudFoundrySpace.builder()
@@ -68,7 +75,7 @@ void errorHandling() {
             "badpassword",
             false,
             500,
-            16);
+            ForkJoinPool.commonPool());
 
     assertThatThrownBy(() -> client.getApplications().all())
         .isInstanceOf(CloudFoundryApiException.class);

diff --git a/...ava/com/netflix/spinnaker/clouddriver/cloudfoundry/client/HttpCloudFoundryClientTest.java b/...ava/com/netflix/spinnaker/clouddriver/cloudfoundry/client/HttpCloudFoundryClientTest.java
@@ -39,6 +39,7 @@
 import java.io.InputStream;
 import java.net.URISyntaxException;
 import java.util.List;
+import java.util.concurrent.ForkJoinPool;
 import org.cloudfoundry.dropsonde.events.EventFactory.Envelope;
 import org.junit.jupiter.api.Test;
 import retrofit.converter.ConversionException;
@@ -68,7 +69,15 @@ void createRetryInterceptorShouldRetryOnInternalServerErrorsThenTimeOut() {
 
     HttpCloudFoundryClient cloudFoundryClient =
         new HttpCloudFoundryClient(
-            "account", "appsManUri", "metricsUri", "host", "user", "password", false, 500, 16);
+            "account",
+            "appsManUri",
+            "metricsUri",
+            "host",
+            "user",
+            "password",
+            false,
+            500,
+            ForkJoinPool.commonPool());
     Response response = cloudFoundryClient.createRetryInterceptor(chain);
 
     try {
@@ -97,7 +106,15 @@ void createRetryInterceptorShouldNotRefreshTokenOnBadCredentials() {
 
     HttpCloudFoundryClient cloudFoundryClient =
         new HttpCloudFoundryClient(
-            "account", "appsManUri", "metricsUri", "host", "user", "password", false, 500, 16);
+            "account",
+            "appsManUri",
+            "metricsUri",
+            "host",
+            "user",
+            "password",
+            false,
+            500,
+            ForkJoinPool.commonPool());
     Response response = cloudFoundryClient.createRetryInterceptor(chain);
 
     try {
@@ -126,7 +143,15 @@ void createRetryInterceptorShouldReturnOnEverythingElse() {
 
     HttpCloudFoundryClient cloudFoundryClient =
         new HttpCloudFoundryClient(
-            "account", "appsManUri", "metricsUri", "host", "user", "password", false, 500, 16);
+            "account",
+            "appsManUri",
+            "metricsUri",
+            "host",
+            "user",
+            "password",
+            false,
+            500,
+            ForkJoinPool.commonPool());
     Response response = cloudFoundryClient.createRetryInterceptor(chain);
 
     try {

diff --git a/...undry/src/test/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/RoutesTest.java b/...undry/src/test/java/com/netflix/spinnaker/clouddriver/cloudfoundry/client/RoutesTest.java
@@ -33,6 +33,7 @@
 import com.netflix.spinnaker.clouddriver.cloudfoundry.model.CloudFoundrySpace;
 import java.util.Arrays;
 import java.util.Optional;
+import java.util.concurrent.ForkJoinPool;
 import org.junit.jupiter.api.Test;
 
 class RoutesTest {
@@ -57,7 +58,8 @@ void toRouteId() {
     when(routeService.all(any(), any(), any())).thenReturn(Page.singleton(route, "abc123"));
     when(routeService.routeMappings(any(), any())).thenReturn(new Page<>());
 
-    Routes routes = new Routes("pws", routeService, null, domains, spaces, 500, 16);
+    Routes routes =
+        new Routes("pws", routeService, null, domains, spaces, 500, ForkJoinPool.commonPool());
     RouteId routeId = routes.toRouteId("demo1-prod.apps.calabasas.cf-app.com/path");
     assertThat(routeId).isNotNull();
     assertThat(routeId.getHost()).isEqualTo("demo1-prod");
@@ -67,7 +69,7 @@ void toRouteId() {
 
   @Test
   void toRouteIdReturnsNullForInvalidRoute() {
-    Routes routes = new Routes(null, null, null, null, null, 500, 16);
+    Routes routes = new Routes(null, null, null, null, null, 500, ForkJoinPool.commonPool());
     assertNull(routes.toRouteId("demo1-pro cf-app.com/path"));
   }
 
@@ -127,7 +129,8 @@ void findShouldFilterCorrectlyOnMultipleResults() {
     when(routeService.all(any(), any(), any())).thenReturn(routePage);
     when(routeService.routeMappings(any(), any())).thenReturn(routeMappingPage);
 
-    Routes routes = new Routes("pws", routeService, null, domains, spaces, 500, 16);
+    Routes routes =
+        new Routes("pws", routeService, null, domains, spaces, 500, ForkJoinPool.commonPool());
 
     CloudFoundryLoadBalancer loadBalancer =
         routes.find(new RouteId().setHost("somehost").setDomainGuid("domain-guid"), "space-guid");

diff --git a/...ver/cloudfoundry/deploy/converters/AbstractLoadBalancersAtomicOperationConverterTest.java b/...ver/cloudfoundry/deploy/converters/AbstractLoadBalancersAtomicOperationConverterTest.java
@@ -41,6 +41,7 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ForkJoinPool;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.stubbing.Answer;
@@ -87,7 +88,18 @@ class AbstractLoadBalancersAtomicOperationConverterTest {
 
   private final CloudFoundryCredentials cloudFoundryCredentials =
       new CloudFoundryCredentials(
-          "test", "", "", "", "", "", "", false, 500, 16, cacheRepository, null) {
+          "test",
+          "",
+          "",
+          "",
+          "",
+          "",
+          "",
+          false,
+          500,
+          cacheRepository,
+          null,
+          ForkJoinPool.commonPool()) {
         public CloudFoundryClient getClient() {
           return cloudFoundryClient;
         }

diff --git a/...udfoundry/deploy/converters/CreateCloudFoundryServiceKeyAtomicOperationConverterTest.java b/...udfoundry/deploy/converters/CreateCloudFoundryServiceKeyAtomicOperationConverterTest.java
@@ -34,6 +34,7 @@
 import io.vavr.collection.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ForkJoinPool;
 import org.junit.jupiter.api.Test;
 
 class CreateCloudFoundryServiceKeyAtomicOperationConverterTest {
@@ -54,7 +55,18 @@ class CreateCloudFoundryServiceKeyAtomicOperationConverterTest {
 
   private final CloudFoundryCredentials cloudFoundryCredentials =
       new CloudFoundryCredentials(
-          "my-account", "", "", "", "", "", "", false, 500, 16, cacheRepository, null) {
+          "my-account",
+          "",
+          "",
+          "",
+          "",
+          "",
+          "",
+          false,
+          500,
+          cacheRepository,
+          null,
+          ForkJoinPool.commonPool()) {
         public CloudFoundryClient getClient() {
           return cloudFoundryClient;
         }