chore: support OpenTelemetry metrics

docs/open_telemetry.md

@@ -1,7 +1,8 @@
-# Open Telemetry in PGAdapter
+# OpenTelemetry in PGAdapter
 
-PGAdapter supports Open Telemetry tracing. You can enable this by adding the `-enable_otel` command
-line argument when starting PGAdapter.
+PGAdapter supports OpenTelemetry tracing and metrics. You can enable the tracing or the metrics
+by adding the `-enable_otel` or `-enable_otel_metrics` command line argument when starting
+PGAdapter. Tracing and metrics can be enabled at the same time.
 
 Optionally, you can also set a trace sample ratio to limit the number of traces that will be
 collected and set with the `-otel_trace_ratio=<ratio>` command line argument. If you omit this
@@ -16,7 +17,8 @@ docker run \
   gcr.io/cloud-spanner-pg-adapter/pgadapter \
   -p my-project -i my-instance \
   -c /credentials.json -x \
-  -enable_otel -otel_trace_ratio=0.1
+  -enable_otel -otel_trace_ratio=0.1 \
+  -enable_otel_metrics
 ```
 
 ## Exporter
@@ -144,6 +146,15 @@ statements consists of the following spans:
 
 ![PGAdapter Cloud Trace - Batch DML example](img/dml_batch_trace_sample.png?raw=true "PGAdapter Cloud Trace - Batch DML example")
 
+## Metrics
+
+The available metrics in PGAdapter are:
+
+* `spanner/pgadapter/roundtrip_latencies`: Latency between PGAdapter receiving a statement from
+   the client and PGAdapter returning the last row of the response to the client.
+* `spanner/pgadapter/client_lib_latencies`: Latency when the Spanner's client library receives
+   a call and returns a response.
+
 ## Frequently Asked Questions
 
 #### How can I find all the statements that were executed on the same connection as the trace I'm looking at?

pom.xml

@@ -37,6 +37,7 @@
     <excludedIntegrationTests/>
 
     <junixsocket.version>2.9.0</junixsocket.version>
+    <opentelemetry.exporter.version>0.23.0</opentelemetry.exporter.version>
   </properties>
 
   <modelVersion>4.0.0</modelVersion>
@@ -141,7 +142,12 @@
     <dependency>
       <groupId>com.google.cloud.opentelemetry</groupId>
       <artifactId>exporter-trace</artifactId>
-      <version>0.23.0</version>
+      <version>${opentelemetry.exporter.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>com.google.cloud.opentelemetry</groupId>
+      <artifactId>exporter-metrics</artifactId>
+      <version>${opentelemetry.exporter.version}</version>
     </dependency>
     <dependency>
       <groupId>org.jetbrains.kotlin</groupId>

src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java

@@ -25,9 +25,11 @@
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.TextFormat;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
+import com.google.cloud.spanner.pgadapter.utils.Metrics;
 import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
 import com.google.common.collect.ImmutableList;
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.trace.Tracer;
 import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
@@ -59,6 +61,7 @@ public class ProxyServer extends AbstractApiService {
   private static final Logger logger = Logger.getLogger(ProxyServer.class.getName());
   private final OptionsMetadata options;
   private final OpenTelemetry openTelemetry;
+  private final Metrics metrics;
   private final Properties properties;
   private final List<ConnectionHandler> handlers = Collections.synchronizedList(new LinkedList<>());
 
@@ -117,6 +120,10 @@ public ProxyServer(
       OptionsMetadata optionsMetadata, OpenTelemetry openTelemetry, Properties properties) {
     this.options = optionsMetadata;
     this.openTelemetry = openTelemetry;
+    this.metrics =
+        optionsMetadata.isEnableOpenTelemetryMetrics()
+            ? new Metrics(openTelemetry)
+            : new Metrics(OpenTelemetry.noop());
     this.localPort = optionsMetadata.getProxyPort();
     this.properties = properties;
     this.debugMode = optionsMetadata.isDebugMode();
@@ -372,6 +379,16 @@ public OpenTelemetry getOpenTelemetry() {
     return this.openTelemetry;
   }
 
+  public Tracer getTracer(String name, String version) {
+    return getOptions().isEnableOpenTelemetry()
+        ? getOpenTelemetry().getTracer(name, version)
+        : OpenTelemetry.noop().getTracer(name, version);
+  }
+
+  public Metrics getMetrics() {
+    return this.metrics;
+  }
+
   /** @return the JDBC connection properties that are used by this server */
   public Properties getProperties() {
     return (Properties) this.properties.clone();

src/main/java/com/google/cloud/spanner/pgadapter/Server.java

@@ -15,6 +15,8 @@
 package com.google.cloud.spanner.pgadapter;
 
 import com.google.auth.Credentials;
+import com.google.cloud.opentelemetry.metric.GoogleCloudMetricExporter;
+import com.google.cloud.opentelemetry.metric.MetricConfiguration;
 import com.google.cloud.opentelemetry.trace.TraceConfiguration;
 import com.google.cloud.opentelemetry.trace.TraceExporter;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
@@ -23,6 +25,9 @@
 import com.google.devtools.cloudtrace.v2.TruncatableString;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk;
+import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder;
+import io.opentelemetry.sdk.metrics.export.MetricExporter;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
 import io.opentelemetry.sdk.trace.samplers.Sampler;
@@ -56,7 +61,8 @@ public static void main(String[] args) {
 
   /** Creates an {@link OpenTelemetry} object from the given options. */
   static OpenTelemetry setupOpenTelemetry(OptionsMetadata optionsMetadata) {
-    if (!optionsMetadata.isEnableOpenTelemetry()) {
+    if (!optionsMetadata.isEnableOpenTelemetry()
+        && !optionsMetadata.isEnableOpenTelemetryMetrics()) {
       return OpenTelemetry.noop();
     }
 
@@ -73,45 +79,62 @@ static OpenTelemetry setupOpenTelemetry(OptionsMetadata optionsMetadata) {
       System.setProperty("otel.service.name", "pgadapter");
     }
 
-    TraceConfiguration.Builder builder =
-        TraceConfiguration.builder().setDeadline(Duration.ofSeconds(60L));
-    String projectId = optionsMetadata.getTelemetryProjectId();
-    if (projectId != null) {
-      builder.setProjectId(projectId);
-    }
     try {
+      String projectId = optionsMetadata.getTelemetryProjectId();
       Credentials credentials = optionsMetadata.getTelemetryCredentials();
-      if (credentials != null) {
-        builder.setCredentials(credentials);
+      AutoConfiguredOpenTelemetrySdkBuilder openTelemetryBuilder =
+          AutoConfiguredOpenTelemetrySdk.builder();
+      if (optionsMetadata.isEnableOpenTelemetry()) {
+        TraceConfiguration.Builder builder =
+            TraceConfiguration.builder().setDeadline(Duration.ofSeconds(60L));
+        if (projectId != null) {
+          builder.setProjectId(projectId);
+        }
+        if (credentials != null) {
+          builder.setCredentials(credentials);
+        }
+        builder.setFixedAttributes(
+            ImmutableMap.of(
+                "service.name",
+                AttributeValue.newBuilder()
+                    .setStringValue(
+                        TruncatableString.newBuilder()
+                            .setValue(
+                                Objects.requireNonNull(
+                                    getOpenTelemetrySetting("otel.service.name")))
+                            .build())
+                    .build()));
+        TraceConfiguration configuration = builder.build();
+        SpanExporter traceExporter = TraceExporter.createWithConfiguration(configuration);
+        Sampler sampler;
+        if (optionsMetadata.getOpenTelemetryTraceRatio() == null) {
+          sampler = Sampler.parentBased(Sampler.traceIdRatioBased(0.05d));
+        } else {
+          sampler =
+              Sampler.parentBased(
+                  Sampler.traceIdRatioBased(optionsMetadata.getOpenTelemetryTraceRatio()));
+        }
+        openTelemetryBuilder.addTracerProviderCustomizer(
+            (sdkTracerProviderBuilder, configProperties) ->
+                sdkTracerProviderBuilder
+                    .setSampler(sampler)
+                    .addSpanProcessor(BatchSpanProcessor.builder(traceExporter).build()));
       }
-      builder.setFixedAttributes(
-          ImmutableMap.of(
-              "service.name",
-              AttributeValue.newBuilder()
-                  .setStringValue(
-                      TruncatableString.newBuilder()
-                          .setValue(
-                              Objects.requireNonNull(getOpenTelemetrySetting("otel.service.name")))
-                          .build())
-                  .build()));
-      TraceConfiguration configuration = builder.build();
-      SpanExporter traceExporter = TraceExporter.createWithConfiguration(configuration);
-      Sampler sampler;
-      if (optionsMetadata.getOpenTelemetryTraceRatio() == null) {
-        sampler = Sampler.parentBased(Sampler.traceIdRatioBased(0.05d));
-      } else {
-        sampler =
-            Sampler.parentBased(
-                Sampler.traceIdRatioBased(optionsMetadata.getOpenTelemetryTraceRatio()));
+      if (optionsMetadata.isEnableOpenTelemetryMetrics()) {
+        MetricExporter cloudMonitoringExporter =
+            GoogleCloudMetricExporter.createWithConfiguration(
+                MetricConfiguration.builder()
+                    // Configure the cloud project id.
+                    .setProjectId(projectId)
+                    // Set the credentials to use when writing to the Cloud Monitoring API
+                    .setCredentials(credentials)
+                    .build());
+        openTelemetryBuilder.addMeterProviderCustomizer(
+            (sdkMeterProviderBuilder, configProperties) ->
+                sdkMeterProviderBuilder.registerMetricReader(
+                    PeriodicMetricReader.builder(cloudMonitoringExporter).build()));
       }
-      return AutoConfiguredOpenTelemetrySdk.builder()
-          .addTracerProviderCustomizer(
-              (sdkTracerProviderBuilder, configProperties) ->
-                  sdkTracerProviderBuilder
-                      .setSampler(sampler)
-                      .addSpanProcessor(BatchSpanProcessor.builder(traceExporter).build()))
-          .build()
-          .getOpenTelemetrySdk();
+      return openTelemetryBuilder.build().getOpenTelemetrySdk();
     } catch (IOException exception) {
       throw new RuntimeException(exception);
     }

src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java

@@ -550,6 +550,7 @@ public enum DdlTransactionMode {
   private static final String OPTION_BINARY_FORMAT = "b";
   private static final String OPTION_AUTHENTICATE = "a";
   private static final String OPTION_ENABLE_OPEN_TELEMETRY = "enable_otel";
+  private static final String OPTION_ENABLE_OPEN_TELEMETRY_METRICS = "enable_otel_metrics";
   private static final String OPTION_OPEN_TELEMETRY_TRACE_RATIO = "otel_trace_ratio";
   private static final String OPTION_SSL = "ssl";
   private static final String OPTION_DISABLE_AUTO_DETECT_CLIENT = "disable_auto_detect_client";
@@ -593,6 +594,7 @@ public enum DdlTransactionMode {
   private final boolean binaryFormat;
   private final boolean authenticate;
   private final boolean enableOpenTelemetry;
+  private final boolean enableOpenTelemetryMetrics;
   private final Double openTelemetryTraceRatio;
   private final SslMode sslMode;
   private final boolean disableAutoDetectClient;
@@ -679,6 +681,7 @@ private OptionsMetadata(Builder builder) {
     this.binaryFormat = commandLine.hasOption(OPTION_BINARY_FORMAT);
     this.authenticate = commandLine.hasOption(OPTION_AUTHENTICATE);
     this.enableOpenTelemetry = commandLine.hasOption(OPTION_ENABLE_OPEN_TELEMETRY);
+    this.enableOpenTelemetryMetrics = commandLine.hasOption(OPTION_ENABLE_OPEN_TELEMETRY_METRICS);
     this.openTelemetryTraceRatio =
         parseOpenTelemetryTraceRatio(commandLine.getOptionValue(OPTION_OPEN_TELEMETRY_TRACE_RATIO));
     this.sslMode = parseSslMode(commandLine.getOptionValue(OPTION_SSL));
@@ -757,6 +760,7 @@ public OptionsMetadata(
     this.binaryFormat = forceBinary;
     this.authenticate = authenticate;
     this.enableOpenTelemetry = false;
+    this.enableOpenTelemetryMetrics = false;
     this.openTelemetryTraceRatio = null;
     this.sslMode = SslMode.Disable;
     this.disableAutoDetectClient = false;
@@ -1124,6 +1128,8 @@ private CommandLine buildOptions(String[] args) {
         false,
         "Whether you wish the proxy to perform an authentication step.");
     options.addOption(null, OPTION_ENABLE_OPEN_TELEMETRY, false, "Enable OpenTelemetry tracing.");
+    options.addOption(
+        null, OPTION_ENABLE_OPEN_TELEMETRY_METRICS, false, "Enable OpenTelemetry metrics.");
     options.addOption(
         null,
         OPTION_OPEN_TELEMETRY_TRACE_RATIO,
@@ -1464,6 +1470,10 @@ public boolean isEnableOpenTelemetry() {
     return this.enableOpenTelemetry;
   }
 
+  public boolean isEnableOpenTelemetryMetrics() {
+    return this.enableOpenTelemetryMetrics;
+  }
+
   public Double getOpenTelemetryTraceRatio() {
     return this.openTelemetryTraceRatio;
   }

src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java

@@ -64,6 +64,7 @@
 import com.google.cloud.spanner.pgadapter.utils.CopyDataReceiver;
 import com.google.cloud.spanner.pgadapter.utils.Logging;
 import com.google.cloud.spanner.pgadapter.utils.Logging.Action;
+import com.google.cloud.spanner.pgadapter.utils.Metrics;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
@@ -80,6 +81,7 @@
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
+import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.SpanBuilder;
 import io.opentelemetry.api.trace.StatusCode;
@@ -123,6 +125,10 @@ public class BackendConnection {
 
   private final Tracer tracer;
 
+  private final Metrics metrics;
+
+  private final Attributes metricAttributes;
+
   private final Deque<Context> statementContext = new ConcurrentLinkedDeque<>();
 
   private final String connectionId;
@@ -472,6 +478,7 @@ private StatementResult executeOnSpannerWithLogging(Statement statement) {
         Stopwatch stopwatch = Stopwatch.createStarted();
         StatementResult result = spannerConnection.execute(statement);
         Duration executionDuration = stopwatch.elapsed();
+        metrics.recordClientLibLatency(executionDuration.toMillis(), metricAttributes);
         logger.log(
             Level.FINER,
             Logging.format(
@@ -629,9 +636,7 @@ void doExecute() {
 
         // Make sure both the front-end CopyDataReceiver and the backend MutationWriter processes
         // have finished before we proceed.
-        //noinspection UnstableApiUsage
         Futures.successfulAsList(copyDataReceiverFuture, statementResultFuture).get();
-        //noinspection UnstableApiUsage
         Futures.allAsList(copyDataReceiverFuture, statementResultFuture).get();
       } catch (ExecutionException executionException) {
         result.setException(executionException.getCause());
@@ -838,6 +843,8 @@ void doExecute() {
   /** Creates a PG backend connection that uses the given Spanner {@link Connection} and options. */
   BackendConnection(
       Tracer tracer,
+      Metrics metrics,
+      Attributes metricAttributes,
       String connectionId,
       Runnable closeAllPortals,
       DatabaseId databaseId,
@@ -846,6 +853,8 @@ void doExecute() {
       OptionsMetadata optionsMetadata,
       Supplier<ImmutableList<LocalStatement>> localStatements) {
     this.tracer = tracer;
+    this.metrics = metrics;
+    this.metricAttributes = metricAttributes;
     this.connectionId = connectionId;
     this.closeAllPortals = closeAllPortals;
     this.sessionState = new SessionState(optionsMetadata);
@@ -914,6 +923,20 @@ public ConnectionState getConnectionState() {
     return this.connectionState;
   }
 
+  @VisibleForTesting
+  public Tracer getTracer() {
+    return this.tracer;
+  }
+
+  @VisibleForTesting
+  public Metrics getMetrics() {
+    return this.metrics;
+  }
+
+  Attributes getMetricAttributes() {
+    return metricAttributes;
+  }
+
   private Span createSpan(String name, Statement statement) {
     return createSpan(name, statement, null);
   }
@@ -1474,7 +1497,10 @@ int executeStatementsInBatch(int fromIndex) {
       }
       Span runBatchSpan = createSpan("execute_batch_on_spanner", null);
       try (Scope ignoreRunBatchSpan = runBatchSpan.makeCurrent()) {
+        Stopwatch stopwatch = Stopwatch.createStarted();
         long[] counts = spannerConnection.runBatch();
+        Duration executionDuration = stopwatch.elapsed();
+        metrics.recordClientLibLatency(executionDuration.toMillis(), metricAttributes);
         if (batchType == StatementType.DDL) {
           counts = extractDdlUpdateCounts(statementResults, counts);
         }