This method should only be used when multiplexed sessions are enabled to create a retry for + * a previously aborted transaction. This method can be used instead of {@link + * #resetForRetryAsync()} to create a retry. Using this method or {@link #resetForRetryAsync()} + * will have the same effect. You must pass in the {@link AbortedException} from the previous + * attempt to preserve the transaction's priority. + * + *
For regular sessions, this behaves the same as {@link #beginAsync()}.
+ */
+ TransactionContextFuture beginAsync(AbortedException exception);
+
/**
* Rolls back the currently active transaction. In most cases there should be no need to call this
* explicitly since {@link #close()} would automatically roll back any active transaction.
@@ -188,7 +203,10 @@ interface AsyncTransactionFunction This class extends the {@link OpenTelemetryMetricsRecorder} which implements the *
+ * measurements related to the lifecyle of an RPC.
+ */
+class BuiltInMetricsRecorder extends OpenTelemetryMetricsRecorder {
+
+ private final DoubleHistogram gfeLatencyRecorder;
+ private final DoubleHistogram afeLatencyRecorder;
+ private final LongCounter gfeHeaderMissingCountRecorder;
+ private final LongCounter afeHeaderMissingCountRecorder;
+
+ /**
+ * Creates the following instruments for the following metrics:
+ *
+ * This class extends the {@link MetricsTracer} which computes generic metrics that can be
+ * observed in the lifecycle of an RPC operation.
+ */
+class BuiltInMetricsTracer extends MetricsTracer implements ApiTracer {
+
+ private final BuiltInMetricsRecorder builtInOpenTelemetryMetricsRecorder;
+ // These are RPC specific attributes and pertain to a specific API Trace
+ private final Map Failed attempt extracts the error from the throwable and adds it to the status attribute
+ * key.
+ */
+ @Override
+ public void attemptFailedDuration(Throwable error, java.time.Duration delay) {
+ try (IScope s = this.traceWrapper.withSpan(this.currentSpan)) {
+ super.attemptFailedDuration(error, delay);
+ attributes.put(STATUS_ATTRIBUTE, extractStatus(error));
+ builtInOpenTelemetryMetricsRecorder.recordServerTimingHeaderMetrics(
+ gfeLatency, afeLatency, attributes, isDirectPathUsed, isAfeEnabled);
+ }
+ }
+
+ /**
+ * Adds an annotation that the attempt failed and that no further attempts will be made because
+ * retry limits have been reached. This extracts the error from the throwable and adds it to the
+ * status attribute key.
+ *
+ * @param error the last error received before retries were exhausted.
+ */
+ @Override
+ public void attemptFailedRetriesExhausted(Throwable error) {
+ try (IScope s = this.traceWrapper.withSpan(this.currentSpan)) {
+ super.attemptFailedRetriesExhausted(error);
+ attributes.put(STATUS_ATTRIBUTE, extractStatus(error));
+ builtInOpenTelemetryMetricsRecorder.recordServerTimingHeaderMetrics(
+ gfeLatency, afeLatency, attributes, isDirectPathUsed, isAfeEnabled);
+ }
+ }
+
+ /**
+ * Adds an annotation that the attempt failed and that no further attempts will be made because
+ * the last error was not retryable. This extracts the error from the throwable and adds it to the
+ * status attribute key.
+ *
+ * @param error the error that caused the final attempt to fail.
+ */
+ @Override
+ public void attemptPermanentFailure(Throwable error) {
+ try (IScope s = this.traceWrapper.withSpan(this.currentSpan)) {
+ super.attemptPermanentFailure(error);
+ attributes.put(STATUS_ATTRIBUTE, extractStatus(error));
+ builtInOpenTelemetryMetricsRecorder.recordServerTimingHeaderMetrics(
+ gfeLatency, afeLatency, attributes, isDirectPathUsed, isAfeEnabled);
+ }
+ }
+
+ public void recordServerTimingHeaderMetrics(
+ Float gfeLatency, Float afeLatency, boolean isDirectPathUsed, boolean isAfeEnabled) {
+ this.gfeLatency = gfeLatency;
+ this.isDirectPathUsed = isDirectPathUsed;
+ this.afeLatency = afeLatency;
+ this.isAfeEnabled = isAfeEnabled;
+ }
+
+ @Override
+ public void addAttributes(Map This class extends the {@link MetricsTracerFactory} which wraps the {@link
+ * BuiltInMetricsRecorder} and pass it to {@link BuiltInMetricsTracer}. It will be * used to record
+ * metrics in {@link BuiltInMetricsTracer}.
+ *
+ * This class is expected to be initialized once during client initialization.
+ */
+class BuiltInMetricsTracerFactory extends MetricsTracerFactory {
+
+ protected BuiltInMetricsRecorder builtInMetricsRecorder;
+ private final Map A {@link StatementFactory} can be used to create statements with unnamed parameters. This is
+ * primarily intended for framework developers who want to integrate the Spanner client with
+ * frameworks that use unnamed parameters. Developers who just want to use the Spanner client in
+ * their application, should use named parameters.
+ *
+ * Examples using {@link StatementFactory}
+ *
+ *
+ *
+ *
+ * @param openTelemetry OpenTelemetry instance
+ * @param serviceName Service Name
+ */
+ BuiltInMetricsRecorder(OpenTelemetry openTelemetry, String serviceName) {
+ super(openTelemetry, serviceName);
+ Meter meter =
+ openTelemetry
+ .meterBuilder(BuiltInMetricsConstant.SPANNER_METER_NAME)
+ .setInstrumentationVersion(GaxProperties.getLibraryVersion(getClass()))
+ .build();
+ this.gfeLatencyRecorder =
+ meter
+ .histogramBuilder(serviceName + '/' + BuiltInMetricsConstant.GFE_LATENCIES_NAME)
+ .setDescription(
+ "Latency between Google's network receiving an RPC and reading back the first byte"
+ + " of the response")
+ .setUnit("ms")
+ .setExplicitBucketBoundariesAdvice(BuiltInMetricsConstant.BUCKET_BOUNDARIES)
+ .build();
+ this.afeLatencyRecorder =
+ meter
+ .histogramBuilder(serviceName + '/' + BuiltInMetricsConstant.AFE_LATENCIES_NAME)
+ .setDescription(
+ "Latency between Spanner API Frontend receiving an RPC and starting to write back"
+ + " the response.")
+ .setExplicitBucketBoundariesAdvice(BuiltInMetricsConstant.BUCKET_BOUNDARIES)
+ .setUnit("ms")
+ .build();
+ this.gfeHeaderMissingCountRecorder =
+ meter
+ .counterBuilder(serviceName + '/' + BuiltInMetricsConstant.GFE_CONNECTIVITY_ERROR_NAME)
+ .setDescription("Number of requests that failed to reach the Google network.")
+ .setUnit("1")
+ .build();
+ this.afeHeaderMissingCountRecorder =
+ meter
+ .counterBuilder(serviceName + '/' + BuiltInMetricsConstant.AFE_CONNECTIVITY_ERROR_NAME)
+ .setDescription("Number of requests that failed to reach the Spanner API Frontend.")
+ .setUnit("1")
+ .build();
+ }
+
+ /**
+ * Record the latency between Google's network receiving an RPC and reading back the first byte of
+ * the response. Data is stored in a Histogram.
+ *
+ * @param gfeLatency Attempt Latency in ms
+ * @param attributes Map of the attributes to store
+ */
+ void recordServerTimingHeaderMetrics(
+ Float gfeLatency,
+ Float afeLatency,
+ Map{@code
+ * Statement statement = databaseClient
+ * .getStatementFactory()
+ * .withUnnamedParameters("SELECT NAME FROM TABLE WHERE ID = ?", 10);
+ * }
+ */
+ default StatementFactory getStatementFactory() {
+ throw new UnsupportedOperationException("method should be overwritten");
+ }
}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java
index f571354dacb..bae8067e33d 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java
@@ -20,12 +20,20 @@
import com.google.cloud.Timestamp;
import com.google.cloud.spanner.Options.TransactionOption;
import com.google.cloud.spanner.Options.UpdateOption;
-import com.google.cloud.spanner.SessionPool.PooledSessionFuture;
import com.google.cloud.spanner.SpannerImpl.ClosedException;
+import com.google.cloud.spanner.Statement.StatementFactory;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
+import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.spanner.v1.BatchWriteResponse;
+import io.opentelemetry.api.common.Attributes;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nullable;
class DatabaseClientImpl implements DatabaseClient {
@@ -33,100 +41,76 @@ class DatabaseClientImpl implements DatabaseClient {
private static final String READ_ONLY_TRANSACTION = "CloudSpanner.ReadOnlyTransaction";
private static final String PARTITION_DML_TRANSACTION = "CloudSpanner.PartitionDMLTransaction";
private final TraceWrapper tracer;
+ private final Attributes databaseAttributes;
@VisibleForTesting final String clientId;
- @VisibleForTesting final SessionPool pool;
@VisibleForTesting final MultiplexedSessionDatabaseClient multiplexedSessionDatabaseClient;
- @VisibleForTesting final boolean useMultiplexedSessionPartitionedOps;
- @VisibleForTesting final boolean useMultiplexedSessionForRW;
-
- final boolean useMultiplexedSessionBlindWrite;
-
- @VisibleForTesting
- DatabaseClientImpl(SessionPool pool, TraceWrapper tracer) {
- this(
- "",
- pool,
- /* useMultiplexedSessionBlindWrite = */ false,
- /* multiplexedSessionDatabaseClient = */ null,
- /* useMultiplexedSessionPartitionedOps= */ false,
- tracer,
- /* useMultiplexedSessionForRW = */ false);
- }
-
- @VisibleForTesting
- DatabaseClientImpl(String clientId, SessionPool pool, TraceWrapper tracer) {
- this(
- clientId,
- pool,
- /* useMultiplexedSessionBlindWrite = */ false,
- /* multiplexedSessionDatabaseClient = */ null,
- /* useMultiplexedSessionPartitionedOps= */ false,
- tracer,
- /* useMultiplexedSessionForRW = */ false);
- }
+ @VisibleForTesting final int dbId;
+ private final AtomicInteger nthRequest;
+ private final Map