Skip to content

Commit

Permalink
chore: code review comments.
Browse files Browse the repository at this point in the history
  • Loading branch information
@hessjcg
hessjcg committed Jul 18, 2023
1 parent c2d5224 commit 2c3d6cc
Show file tree
Hide file tree
Showing 2 changed files with 41 additions and 59 deletions.
22 changes: 5 additions & 17 deletions core/src/main/java/com/google/cloud/sql/core/CoreSocketFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,6 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.net.ssl.SSLSocket;
import jnr.unixsocket.UnixSocketAddress;
Expand Down Expand Up @@ -80,7 +78,6 @@ public final class CoreSocketFactory {
private final CredentialFactory credentialFactory;
private final int serverProxyPort;
private final SqlAdminApiFetcher adminApiService;
private final AtomicInteger atomicInteger;

@VisibleForTesting
CoreSocketFactory(
Expand All @@ -94,7 +91,6 @@ public final class CoreSocketFactory {
this.serverProxyPort = serverProxyPort;
this.executor = executor;
this.localKeyPair = localKeyPair;
this.atomicInteger = new AtomicInteger();
}

/** Returns the {@link CoreSocketFactory} singleton. */
Expand Down Expand Up @@ -221,6 +217,11 @@ public static String getHostIp(String csqlInstanceName, String ipTypes) throws I
return getInstance().getHostIp(csqlInstanceName, listIpTypes(ipTypes));
}

private String getHostIp(String instanceName, List<String> ipTypes) {
CloudSqlInstance instance = getCloudSqlInstance(instanceName, AuthType.PASSWORD);
return instance.getPreferredIp(ipTypes);
}

/**
* Converts the string property of IP types to a list by splitting by commas, and upper-casing.
*/
Expand Down Expand Up @@ -304,11 +305,6 @@ public static void setApplicationName(String applicationName) {
System.setProperty(USER_TOKEN_PROPERTY_NAME, applicationName);
}

private String getHostIp(String instanceName, List<String> ipTypes) {
CloudSqlInstance instance = getCloudSqlInstance(instanceName, AuthType.PASSWORD);
return instance.getPreferredIp(ipTypes);
}

/**
* Creates a secure socket representing a connection to a Cloud SQL instance.
*
Expand All @@ -324,10 +320,6 @@ Socket createSslSocket(String instanceName, List<String> ipTypes, AuthType authT
CloudSqlInstance instance = getCloudSqlInstance(instanceName, authType);

try {
int i = atomicInteger.incrementAndGet();
if (i > 10 && i % 2 == 0) {
throw new IOException("Failed to connect WOOO!");
}
SSLSocket socket = instance.createSslSocket();

// TODO(kvg): Support all socket related options listed here:
Expand All @@ -342,10 +334,6 @@ Socket createSslSocket(String instanceName, List<String> ipTypes, AuthType authT

return socket;
} catch (Exception ex) {
logger.log(
Level.WARNING,
String.format("[%s] Failed to Connect, force refreshing", instanceName),
ex);
// TODO(kvg): Let user know about the rate limit
instance.forceRefresh();
throw ex;
Expand Down
78 changes: 36 additions & 42 deletions core/src/test/java/com/google/cloud/sql/core/CloudSqlInstanceConcurrencyTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import dev.failsafe.RateLimiter;
import java.io.IOException;
import java.security.KeyPair;
Expand All @@ -22,8 +21,6 @@
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.ConsoleHandler;
import java.util.logging.Formatter;
Expand All @@ -35,6 +32,7 @@
import org.junit.Test;

public class CloudSqlInstanceConcurrencyTest {

private static final long start = System.currentTimeMillis();

private static final Logger logger =
Expand Down Expand Up @@ -138,16 +136,16 @@ public void initialize(HttpRequest var1) throws IOException {
}
}

@Test
public void testCloudSqlInstanceConcurrency() throws Exception {
@Test(timeout = 45000)
public void testCloudSqlInstanceRefreshesConsistentlyWithoutRaceConditions() throws Exception {
// Run the test 10 times to ensure we don't have race conditions
for (int i = 0; i < 10; i++) {
runConcurrencyTest();
}
}

@Test
public void testBasicHappyPath() throws Exception {
public void testCloudSqlInstanceCorrectlyRefreshesInstanceData() throws Exception {
MockAdminApi mockAdminApi = new MockAdminApi();
ListenableFuture<KeyPair> keyPairFuture =
Futures.immediateFuture(mockAdminApi.getClientKeyPair());
Expand All @@ -170,7 +168,7 @@ public void testBasicHappyPath() throws Exception {
}

@Test
public void testPermanentFailure() throws Exception {
public void testRefreshWhenRefreshRequestAlwaysFails() throws Exception {
MockAdminApi mockAdminApi = new MockAdminApi();
ListenableFuture<KeyPair> keyPairFuture =
Futures.immediateFuture(mockAdminApi.getClientKeyPair());
Expand Down Expand Up @@ -263,8 +261,8 @@ public void runConcurrencyTest() throws Exception {
}
}

@Test
public void testForceRefreshDeadlock() throws Exception {
@Test(timeout = 10000)
public void testForceRefreshDoesNotCauseADeadlock() throws Exception {
MockAdminApi mockAdminApi = new MockAdminApi();
ListenableFuture<KeyPair> keyPairFuture =
Futures.immediateFuture(mockAdminApi.getClientKeyPair());
Expand Down Expand Up @@ -293,35 +291,13 @@ public void testForceRefreshDeadlock() throws Exception {

assertThat(supplier.counter.get()).isEqualTo(instanceCount);

// Start a thread for each instance that will force refresh and get InstanceData
// 50 times.
List<Thread> threads =
instances.stream()
.map(
(inst) -> {
Thread t =
new Thread(
() -> {
for (int i = 0; i < 50; i++) {
try {
inst.forceRefresh();
inst.forceRefresh();
Thread.yield();
inst.forceRefresh();
inst.getSslData();
Thread.sleep(10);
} catch (Exception e) {
logger.info("Exception in force refresh loop.");
}
}
logger.info("Done spamming");
});
t.setName("test-" + inst.getInstanceName());
t.start();
return t;
})
.collect(Collectors.toList());
instances.stream().map(this::startForceRefreshThread).collect(Collectors.toList());

for (Thread t : threads) {
t.join();
t.join(10000);
}

// Check if there is a scheduled future
Expand All @@ -335,14 +311,32 @@ public void testForceRefreshDeadlock() throws Exception {
assertThat(brokenLoop).isEqualTo(0);
}

private ListeningScheduledExecutorService newTestExecutor() {
ScheduledThreadPoolExecutor executor =
(ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(8);
executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
private Thread startForceRefreshThread(CloudSqlInstance inst) {
Runnable forceRefreshRepeat =
() -> {
for (int i = 0; i < 50; i++) {
try {
Thread.sleep(100);
inst.forceRefresh();
inst.forceRefresh();
Thread.yield();
inst.forceRefresh();
inst.getSslData();
} catch (Exception e) {
logger.info("Exception in force refresh loop.");
}
}
logger.info("Done spamming");
};

Thread t = new Thread(forceRefreshRepeat);
t.setName("test-" + inst.getInstanceName());
t.start();
return t;
}

//noinspection UnstableApiUsage
return MoreExecutors.listeningDecorator(
MoreExecutors.getExitingScheduledExecutorService(executor));
private ListeningScheduledExecutorService newTestExecutor() {
return CoreSocketFactory.getDefaultExecutor();
}

private RateLimiter<Object> newRateLimiter() {
Expand Down

0 comments on commit 2c3d6cc

Please sign in to comment.