chore: add support for OpenTelemetry metrics to Connection API

Author: hengfengli

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionOptions.java

@@ -39,6 +39,8 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.Sets;
 import com.google.spanner.v1.ExecuteSqlRequest.QueryOptions;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
 import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
@@ -54,6 +56,7 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 /**
@@ -484,6 +487,7 @@ public static class Builder {
     private List<StatementExecutionInterceptor> statementExecutionInterceptors =
         Collections.emptyList();
     private SpannerOptionsConfigurator configurator;
+    private OpenTelemetry openTelemetry;
 
     private Builder() {}
 
@@ -633,6 +637,11 @@ Builder setCredentials(Credentials credentials) {
       return this;
     }
 
+    public Builder setOpenTelemetry(OpenTelemetry openTelemetry) {
+      this.openTelemetry = openTelemetry;
+      return this;
+    }
+
     /** @return the {@link ConnectionOptions} */
     public ConnectionOptions build() {
       Preconditions.checkState(this.uri != null, "Connection URI is required");
@@ -691,6 +700,7 @@ public static Builder newBuilder() {
   private final boolean retryAbortsInternally;
   private final boolean useVirtualThreads;
   private final boolean useVirtualGrpcTransportThreads;
+  private final OpenTelemetry openTelemetry;
   private final List<StatementExecutionInterceptor> statementExecutionInterceptors;
   private final SpannerOptionsConfigurator configurator;
 
@@ -792,6 +802,7 @@ private ConnectionOptions(Builder builder) {
     this.retryAbortsInternally = parseRetryAbortsInternally(this.uri);
     this.useVirtualThreads = parseUseVirtualThreads(this.uri);
     this.useVirtualGrpcTransportThreads = parseUseVirtualGrpcTransportThreads(this.uri);
+    this.openTelemetry = builder.openTelemetry;
     this.statementExecutionInterceptors =
         Collections.unmodifiableList(builder.statementExecutionInterceptors);
     this.configurator = builder.configurator;
@@ -856,6 +867,19 @@ private static Integer parseIntegerProperty(String propertyName, String value) {
     return null;
   }
 
+  /**
+   * Returns an instance of OpenTelemetry. If OpenTelemetry object is not set then
+   * GlobalOpenTelemetry will be used as fallback.
+   */
+  @Nonnull
+  OpenTelemetry getOpenTelemetry() {
+    if (this.openTelemetry != null) {
+      return this.openTelemetry;
+    } else {
+      return GlobalOpenTelemetry.get();
+    }
+  }
+
   SpannerOptionsConfigurator getConfigurator() {
     return configurator;
   }

google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/SpannerPool.java

@@ -29,6 +29,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Ticker;
 import io.grpc.ManagedChannelBuilder;
+import io.opentelemetry.api.OpenTelemetry;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -156,6 +157,7 @@ static class SpannerPoolKey {
     private final String databaseRole;
     private final boolean routeToLeader;
     private final boolean useVirtualGrpcTransportThreads;
+    private final OpenTelemetry openTelemetry;
 
     @VisibleForTesting
     static SpannerPoolKey of(ConnectionOptions options) {
@@ -183,6 +185,7 @@ private SpannerPoolKey(ConnectionOptions options) throws IOException {
       this.userAgent = options.getUserAgent();
       this.routeToLeader = options.isRouteToLeader();
       this.useVirtualGrpcTransportThreads = options.isUseVirtualGrpcTransportThreads();
+      this.openTelemetry = options.getOpenTelemetry();
     }
 
     @Override
@@ -201,7 +204,8 @@ public boolean equals(Object o) {
           && Objects.equals(this.userAgent, other.userAgent)
           && Objects.equals(this.routeToLeader, other.routeToLeader)
           && Objects.equals(
-              this.useVirtualGrpcTransportThreads, other.useVirtualGrpcTransportThreads);
+              this.useVirtualGrpcTransportThreads, other.useVirtualGrpcTransportThreads)
+          && Objects.equals(this.openTelemetry, other.openTelemetry);
     }
 
     @Override
@@ -216,7 +220,8 @@ public int hashCode() {
           this.databaseRole,
           this.userAgent,
           this.routeToLeader,
-          this.useVirtualGrpcTransportThreads);
+          this.useVirtualGrpcTransportThreads,
+          this.openTelemetry);
     }
   }
 
@@ -349,6 +354,9 @@ Spanner createSpanner(SpannerPoolKey key, ConnectionOptions options) {
         .setDatabaseRole(options.getDatabaseRole())
         .setCredentials(options.getCredentials());
     builder.setSessionPoolOption(key.sessionPoolOptions);
+    if (key.openTelemetry != null) {
+      builder.setOpenTelemetry(key.openTelemetry);
+    }
     if (key.numChannels != null) {
       builder.setNumChannels(key.numChannels);
     }

google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/SpannerPoolTest.java

@@ -39,6 +39,10 @@
 import com.google.cloud.spanner.connection.SpannerPool.SpannerPoolKey;
 import com.google.common.base.Ticker;
 import com.google.common.testing.FakeTicker;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+
 import java.io.ByteArrayOutputStream;
 import java.io.OutputStream;
 import java.util.concurrent.TimeUnit;
@@ -69,6 +73,10 @@ public class SpannerPoolTest {
   private ConnectionOptions options7 = mock(ConnectionOptions.class);
   private ConnectionOptions options8 = mock(ConnectionOptions.class);
 
+  private ConnectionOptions optionsOpenTelemetry1 = mock(ConnectionOptions.class);
+  private ConnectionOptions optionsOpenTelemetry2 = mock(ConnectionOptions.class);
+  private ConnectionOptions optionsOpenTelemetry3 = mock(ConnectionOptions.class);
+
   private SpannerPool createSubjectAndMocks() {
     return createSubjectAndMocks(0L, Ticker.systemTicker());
   }
@@ -83,6 +91,9 @@ Spanner createSpanner(SpannerPoolKey key, ConnectionOptions options) {
           }
         };
 
+    OpenTelemetry openTelemetry1 = OpenTelemetrySdk.builder().build();
+    OpenTelemetry openTelemetry2 = OpenTelemetrySdk.builder().build();
+
     when(options1.getCredentialsUrl()).thenReturn(credentials1);
     when(options1.getProjectId()).thenReturn("test-project-1");
     when(options2.getCredentialsUrl()).thenReturn(credentials2);
@@ -101,6 +112,13 @@ Spanner createSpanner(SpannerPoolKey key, ConnectionOptions options) {
     when(options8.getProjectId()).thenReturn("test-project-3");
     when(options8.isRouteToLeader()).thenReturn(false);
 
+    when(optionsOpenTelemetry1.getProjectId()).thenReturn("test-project-1");
+    when(optionsOpenTelemetry1.getOpenTelemetry()).thenReturn(openTelemetry1);
+    when(optionsOpenTelemetry2.getProjectId()).thenReturn("test-project-1");
+    when(optionsOpenTelemetry2.getOpenTelemetry()).thenReturn(openTelemetry1);
+    when(optionsOpenTelemetry3.getProjectId()).thenReturn("test-project-1");
+    when(optionsOpenTelemetry3.getOpenTelemetry()).thenReturn(openTelemetry2);
+
     return pool;
   }
 
@@ -498,4 +516,21 @@ public void testSpannerPoolKeyEquality() {
     assertEquals(key3, key4);
     assertNotEquals(key4, key5);
   }
+
+  @Test
+  public void testOpenTelemetry() {
+    SpannerPool pool = createSubjectAndMocks();
+    Spanner spanner1;
+    Spanner spanner2;
+
+    // assert equal
+    spanner1 = pool.getSpanner(optionsOpenTelemetry1, connection1);
+    spanner2 = pool.getSpanner(optionsOpenTelemetry2, connection2);
+    assertEquals(spanner1, spanner2);
+
+    // assert not equal
+    spanner1 = pool.getSpanner(optionsOpenTelemetry1, connection1);
+    spanner2 = pool.getSpanner(optionsOpenTelemetry3, connection2);
+    assertNotEquals(spanner1, spanner2);
+  }
 }