From 26d3d15e2b3f75bb3624ed11700109b8777b21af Mon Sep 17 00:00:00 2001 From: Shuran Zhang Date: Mon, 11 Aug 2025 17:57:30 +0000 Subject: [PATCH] feat: add configurable health check point --- README.md | 2 + .../google/cloud/spanner/adapter/Adapter.java | 25 +++--- .../cloud/spanner/adapter/HealthServer.java | 77 ++++++++++++++++ .../cloud/spanner/adapter/Launcher.java | 71 ++++++++++++--- .../cloud/spanner/adapter/AdapterTest.java | 52 +++++++---- .../cloud/spanner/adapter/LauncherTest.java | 87 +++++++++++++++++++ 6 files changed, 269 insertions(+), 45 deletions(-) create mode 100644 google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/HealthServer.java create mode 100644 google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherTest.java diff --git a/README.md b/README.md index c5bf6cb5..4d842d8f 100644 --- a/README.md +++ b/README.md @@ -155,11 +155,13 @@ For non-Java applications or tools like `cqlsh`, you can run the Spanner Cassand -Dhost=127.0.0.1 \ -Dport=9042 \ -DnumGrpcChannels=4 \ + -DhealthCheckPort=8080 \ -jar path/to/your/spanner-cassandra-launcher.jar ``` * Replace the value of `-DdatabaseUri` with your Spanner database URI. * You can omit `-Dhost` to use the default `0.0.0.0`, omit `-Dport` to use the default `9042`, and omit `-DnumGrpcChannels` to use the default `4`. + * You can optionally specify a `-DhealthCheckPort`. If you specify a port number, a health check endpoint will be started on that port. The IP address of the health check endpoint will be the same as the IP address that is used for the client. The health check endpoint will return an HTTP status 200 OK if the client is up and running, and a 503 Service Unavailable if it is not. The health check endpoint is not enabled by default. ## View and manage client-side metrics diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java index 7b9edd46..15cec753 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java @@ -15,7 +15,18 @@ */ package com.google.cloud.spanner.adapter; -import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadPerTaskExecutor; +import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketException; +import java.util.Collections; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import javax.annotation.concurrent.NotThreadSafe; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import com.google.api.gax.core.CredentialsProvider; import com.google.api.gax.core.FixedCredentialsProvider; @@ -28,20 +39,11 @@ import com.google.auth.Credentials; import com.google.auth.oauth2.GoogleCredentials; import com.google.cloud.NoCredentials; +import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadPerTaskExecutor; import com.google.common.base.MoreObjects; import com.google.common.collect.ImmutableSet; import com.google.spanner.adapter.v1.AdapterClient; import com.google.spanner.adapter.v1.AdapterSettings; -import java.io.IOException; -import java.net.ServerSocket; -import java.net.Socket; -import java.net.SocketException; -import java.util.Collections; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import javax.annotation.concurrent.NotThreadSafe; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** Manages client connections, acting as an intermediary for communication with Spanner. */ @NotThreadSafe @@ -157,7 +159,6 @@ void start() { started = true; LOG.info("Adapter started for database '{}'.", options.getDatabaseUri()); - } catch (IOException | RuntimeException e) { throw new AdapterStartException(e); } diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/HealthServer.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/HealthServer.java new file mode 100644 index 00000000..9d8348b3 --- /dev/null +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/HealthServer.java @@ -0,0 +1,77 @@ +/* +Copyright 2025 Google LLC + +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.google.cloud.spanner.adapter; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpHandler; +import com.sun.net.httpserver.HttpServer; + +class HealthServer { + private static final Logger LOG = LoggerFactory.getLogger(HealthServer.class); + private static final int HTTP_OK_STATUS = 200; + private static final int HTTP_UNAVAILABLE_STATUS = 503; + + private final HttpServer server; + private final AtomicBoolean isReady = new AtomicBoolean(false); + + HealthServer(InetAddress address, int port) throws IOException { + this.server = HttpServer.create(new InetSocketAddress(address, port), 0); + this.server.createContext("/debug/health", new HealthHandler()); + this.server.setExecutor(null); // creates a default executor + } + + void start() { + server.start(); + LOG.info("Health server started on {}", server.getAddress()); + } + + void stop() { + server.stop(0); + LOG.info("Health server stopped."); + } + + void setReady(boolean ready) { + isReady.set(ready); + } + + private class HealthHandler implements HttpHandler { + @Override + public void handle(HttpExchange exchange) throws IOException { + if (isReady.get()) { + sendResponse(exchange, HTTP_OK_STATUS, "All listeners are up and running"); + } else { + sendResponse(exchange, HTTP_UNAVAILABLE_STATUS, "Service Unavailable"); + } + } + + private void sendResponse(HttpExchange exchange, int statusCode, String response) + throws IOException { + exchange.sendResponseHeaders(statusCode, response.length()); + try (OutputStream os = exchange.getResponseBody()) { + os.write(response.getBytes()); + } + } + } +} diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java index 4d761746..3a7e6d3e 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java @@ -13,18 +13,21 @@ See the License for the specific language governing permissions and limitations under the License. */ - package com.google.cloud.spanner.adapter; -import com.google.cloud.spanner.adapter.metrics.BuiltInMetricsProvider; -import com.google.cloud.spanner.adapter.metrics.BuiltInMetricsRecorder; -import com.google.spanner.adapter.v1.DatabaseName; -import io.opentelemetry.api.OpenTelemetry; import java.net.InetAddress; import java.time.Duration; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.cloud.spanner.adapter.metrics.BuiltInMetricsProvider; +import com.google.cloud.spanner.adapter.metrics.BuiltInMetricsRecorder; +import com.google.common.annotations.VisibleForTesting; +import com.google.spanner.adapter.v1.DatabaseName; + +import io.opentelemetry.api.OpenTelemetry; + /** * Main entry point for running a Spanner Cassandra Adapter as a stand-alone application. * @@ -43,6 +46,8 @@ * with Spanner. Defaults to 4. *
  • {@code maxCommitDelayMillis}: (Optional) The max commit delay to set in requests to * optimize write throughput, in milliseconds. Defaults to none. + *
  • {@code healthCheckPort}: (Optional) The port number for the health check server. Defaults + * to none. * * * Example usage: @@ -53,6 +58,7 @@ * -Dport=9042 \ * -DnumGrpcChannels=4 \ * -DmaxCommitDelayMillis=5 \ + * -DhealthCheckPort=8080 \ * -cp path/to/your/spanner-cassandra-launcher.jar com.google.cloud.spanner.adapter.SpannerCassandraLauncher * * @@ -72,6 +78,48 @@ public class Launcher { private static final String DEFAULT_NUM_GRPC_CHANNELS = "4"; private static final String MAX_COMMIT_DELAY_PROP_KEY = "maxCommitDelayMillis"; private static final String ENABLE_BUILTIN_METRICS_PROP_KEY = "enableBuiltInMetrics"; + private static final String HEALTH_CHECK_PORT_PROP_KEY = "healthCheckPort"; + + private final Adapter adapter; + + @VisibleForTesting + Launcher(Adapter adapter) { + this.adapter = adapter; + } + + void start(InetAddress address, int healthCheckPort) throws Exception { + HealthServer healthServer = null; + if (healthCheckPort > 0) { + healthServer = new HealthServer(address, healthCheckPort); + healthServer.start(); + } + + adapter.start(); + if (healthServer != null) { + healthServer.setReady(true); + } + + final HealthServer finalHealthServer = healthServer; + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + if (finalHealthServer != null) { + finalHealthServer.stop(); + } + try { + adapter.stop(); + } catch (Exception e) { + LOG.error("Failed to stop adapter", e); + } + })); + + try { + Thread.currentThread().join(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } public static void main(String[] args) throws Exception { final String databaseUri = System.getProperty(DATABASE_URI_PROP_KEY); @@ -83,6 +131,8 @@ public static void main(String[] args) throws Exception { final String maxCommitDelayProperty = System.getProperty(MAX_COMMIT_DELAY_PROP_KEY); final boolean enableBuiltInMetrics = Boolean.parseBoolean(System.getProperty(ENABLE_BUILTIN_METRICS_PROP_KEY, "false")); + final int healthCheckPort = + Integer.parseInt(System.getProperty(HEALTH_CHECK_PORT_PROP_KEY, "0")); if (databaseUri == null) { throw new IllegalArgumentException( @@ -112,10 +162,9 @@ public static void main(String[] args) throws Exception { opBuilder.maxCommitDelay(Duration.ofMillis(Integer.parseInt(maxCommitDelayProperty))); } - Adapter adapter = new Adapter(opBuilder.build()); LOG.info( "Starting Adapter for Spanner database {} on {}:{} with {} gRPC channels, max commit" - + " delay of {} and built-in metrics enabled: {}", + + " delay of {} and built-in metrics enabled: {},", databaseUri, inetAddress, port, @@ -123,12 +172,6 @@ public static void main(String[] args) throws Exception { maxCommitDelayProperty, enableBuiltInMetrics); - adapter.start(); - - try { - Thread.currentThread().join(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } + new Launcher(new Adapter(opBuilder.build())).start(inetAddress, healthCheckPort); } } diff --git a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/AdapterTest.java b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/AdapterTest.java index 14e909eb..0700bbf2 100644 --- a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/AdapterTest.java +++ b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/AdapterTest.java @@ -16,9 +16,23 @@ package com.google.cloud.spanner.adapter; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.URL; +import java.net.UnknownHostException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThrows; +import org.junit.Before; +import org.junit.Test; import static org.mockito.ArgumentMatchers.any; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockConstruction; import static org.mockito.Mockito.mockStatic; @@ -26,23 +40,13 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.google.api.gax.core.NoCredentialsProvider; import com.google.auth.oauth2.GoogleCredentials; import com.google.spanner.adapter.v1.AdapterClient; import com.google.spanner.adapter.v1.AdapterSettings; -import com.google.spanner.adapter.v1.CreateSessionRequest; import com.google.spanner.adapter.v1.Session; -import java.net.InetAddress; -import java.net.ServerSocket; -import java.net.UnknownHostException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import org.junit.Before; -import org.junit.Test; -import org.mockito.MockedConstruction; -import org.mockito.MockedStatic; public final class AdapterTest { - private static final String TEST_HOST = "localhost"; private static final String TEST_DATABASE_URI = "projects/test-project/instances/test-instance/databases/test-db"; private static final int TEST_PORT = 12345; @@ -50,45 +54,55 @@ public final class AdapterTest { private Adapter adapter; public AdapterTest() throws UnknownHostException { - inetAddress = InetAddress.getByName("0.0.0.0"); + inetAddress = InetAddress.getByName("localhost"); } @Before - public void setUp() { + public void setUp() throws IOException { AdapterOptions options = new AdapterOptions.Builder() - .spannerEndpoint(TEST_HOST) + .spannerEndpoint("localhost:1234") .tcpPort(TEST_PORT) .databaseUri(TEST_DATABASE_URI) .inetAddress(inetAddress) + .credentials(NoCredentialsProvider.create().getCredentials()) .build(); adapter = new Adapter(options); } + private int getAvailablePort() throws IOException { + try (ServerSocket serverSocket = new ServerSocket(0)) { + return serverSocket.getLocalPort(); + } + } + @Test public void successfulStartStopFlow() throws Exception { - try (MockedConstruction mockedServerSocketConstruction = mockConstruction(ServerSocket.class); MockedStatic mockedExecutors = mockStatic(Executors.class); MockedStatic mockedStaticAdapterClient = mockStatic(AdapterClient.class); MockedStatic mockedGoogleCredentials = - mockStatic(GoogleCredentials.class)) { + mockStatic(GoogleCredentials.class); + MockedConstruction mockedSessionManager = + mockConstruction( + SessionManager.class, + (mock, context) -> { + when(mock.getSession()).thenReturn(mock(Session.class)); + })) { AdapterClient mockAdapterClient = mock(AdapterClient.class); - Session mockSession = mock(Session.class); mockedGoogleCredentials.when(GoogleCredentials::getApplicationDefault).thenReturn(null); mockedStaticAdapterClient .when(() -> AdapterClient.create(any(AdapterSettings.class))) .thenReturn(mockAdapterClient); - when(mockAdapterClient.createSession(any())).thenReturn(mockSession); ExecutorService mockExecutor = mock(ExecutorService.class); mockedExecutors.when(Executors::newCachedThreadPool).thenReturn(mockExecutor); adapter.start(); adapter.stop(); - verify(mockAdapterClient, times(1)).createSession(any(CreateSessionRequest.class)); + verify(mockedSessionManager.constructed().get(0), times(1)).getSession(); verify(mockExecutor).execute(any(Runnable.class)); // Verify ServerSocket was constructed assertEquals(1, mockedServerSocketConstruction.constructed().size()); diff --git a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherTest.java b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherTest.java new file mode 100644 index 00000000..21384419 --- /dev/null +++ b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/LauncherTest.java @@ -0,0 +1,87 @@ +/* +Copyright 2025 Google LLC + +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.google.cloud.spanner.adapter; + +import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.URL; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.Mock; +import static org.mockito.Mockito.doNothing; +import org.mockito.MockitoAnnotations; + +import static com.google.common.truth.Truth.assertThat; + +@RunWith(JUnit4.class) +public class LauncherTest { + + @Mock private Adapter adapter; + + @Before + public void setUp() { + MockitoAnnotations.initMocks(this); + } + + @Test + public void testHealthServerReturnsOk() throws Exception { + doNothing().when(adapter).start(); + final int port = getFreePort(); + final Launcher launcher = new Launcher(adapter); + Thread launcherThread = + new Thread( + () -> { + try { + launcher.start(InetAddress.getLoopbackAddress(), port); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + launcherThread.start(); + + // Wait for the health server to start + HttpURLConnection connection = null; + for (int i = 0; i < 100; i++) { + try { + URL url = new URL("http://localhost:" + port + "/debug/health"); + connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod("GET"); + if (connection.getResponseCode() == 200) { + break; + } + } catch (IOException e) { + // Ignore and retry. + } + Thread.sleep(100); + } + assertThat(connection).isNotNull(); + assertThat(connection.getResponseCode()).isEqualTo(200); + + launcherThread.interrupt(); + launcherThread.join(); + } + + private static int getFreePort() throws IOException { + try (ServerSocket serverSocket = new ServerSocket(0)) { + return serverSocket.getLocalPort(); + } + } +}