Skip to content

fix(spanner): scope server-timing metrics per call and guard interceptor lifecycle callbacks #14053

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
olavloite merged 1 commit into from
Aug 12, 2026
Merged

fix(spanner): scope server-timing metrics per call and guard interceptor lifecycle callbacks #14053

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 89 additions & 46 deletions ...google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/HeaderInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,23 @@
import static com.google.cloud.spanner.spi.v1.SpannerRpcViews.SPANNER_GFE_LATENCY;

import com.google.api.gax.tracing.ApiTracer;
import com.google.cloud.spanner.*;
import com.google.cloud.spanner.BuiltInMetricsConstant;
import com.google.cloud.spanner.CompositeTracer;
import com.google.cloud.spanner.SpannerExceptionFactory;
import com.google.cloud.spanner.SpannerRpcMetrics;
import com.google.cloud.spanner.XGoogSpannerRequestId;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.spanner.admin.database.v1.DatabaseName;
import io.grpc.*;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.alts.AltsContextUtil;
import io.opencensus.stats.MeasureMap;
import io.opencensus.stats.Stats;
Expand All @@ -52,6 +62,7 @@
import java.util.logging.Logger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.annotation.Nullable;

/**
* Intercepts all gRPC calls to extract server-timing header. Captures GFE Latency and GFE Header
Expand Down Expand Up @@ -89,8 +100,6 @@ class HeaderInterceptor implements ClientInterceptor {
private static final Logger LOGGER = Logger.getLogger(HeaderInterceptor.class.getName());
private static final Level LEVEL = Level.INFO;
private final SpannerRpcMetrics spannerRpcMetrics;
private Float gfeLatency;
private Float afeLatency;

HeaderInterceptor(SpannerRpcMetrics spannerRpcMetrics) {
this.spannerRpcMetrics = spannerRpcMetrics;
Expand Down Expand Up @@ -118,48 +127,65 @@ public void start(Listener<RespT> responseListener, Metadata headers) {

super.start(
new SimpleForwardingClientCallListener<RespT>(responseListener) {
private Float gfeLatency;
private Float afeLatency;

@Override
public void onHeaders(Metadata metadata) {
recordFirstResponseLatency(requestId, startedAtNanos, firstResponseRecorded);
String serverTiming = metadata.get(SERVER_TIMING_HEADER_KEY);
try {
// Get gfe and afe Latency value
Map<String, Float> serverTimingMetrics = parseServerTimingHeader(serverTiming);
gfeLatency = serverTimingMetrics.get(GFE_TIMING_HEADER);
afeLatency = serverTimingMetrics.get(AFE_TIMING_HEADER);
} catch (NumberFormatException e) {
LOGGER.log(LEVEL, "Invalid server-timing object in header: {}", serverTiming);
recordFirstResponseLatency(requestId, startedAtNanos, firstResponseRecorded);
String serverTiming = metadata.get(SERVER_TIMING_HEADER_KEY);
try {
// Get gfe and afe Latency value
Map<String, Float> serverTimingMetrics =
parseServerTimingHeader(serverTiming);
gfeLatency = serverTimingMetrics.get(GFE_TIMING_HEADER);
afeLatency = serverTimingMetrics.get(AFE_TIMING_HEADER);
} catch (NumberFormatException e) {
LOGGER.log(
LEVEL, "Invalid server-timing object in header: {0}", serverTiming);
}
} catch (Throwable throwable) {
LOGGER.log(
Level.WARNING, "Error processing headers in HeaderInterceptor", throwable);
} finally {
super.onHeaders(metadata);
}

super.onHeaders(metadata);
}

@Override
public void onClose(Status status, Metadata trailers) {
// Record Built-in Metrics
boolean isDirectPathUsed = AltsContextUtil.check(getAttributes());
boolean isAfeEnabled = GapicSpannerRpc.isEnableAFEServerTiming();
recordSpan(span, requestId);
recordCustomMetrics(tagContext, attributes, isDirectPathUsed);
Map<String, String> builtInMetricsAttributes = new HashMap<>();
try {
builtInMetricsAttributes =
new HashMap<>(getBuiltInMetricAttributes(key, databaseName));
} catch (ExecutionException e) {
LOGGER.log(
LEVEL, "Unable to get built-in metric attributes {}", e.getMessage());
// Record Built-in Metrics
boolean isDirectPathUsed = AltsContextUtil.check(getAttributes());
boolean isAfeEnabled = GapicSpannerRpc.isEnableAFEServerTiming();
recordSpan(span, requestId, gfeLatency, afeLatency);
recordCustomMetrics(tagContext, attributes, isDirectPathUsed, gfeLatency);
Map<String, String> builtInMetricsAttributes = new HashMap<>();
try {
builtInMetricsAttributes =
new HashMap<>(getBuiltInMetricAttributes(key, databaseName));
} catch (ExecutionException e) {
LOGGER.log(
LEVEL, "Unable to get built-in metric attributes {0}", e.getMessage());
}
if (status.isOk()) {
recordFirstResponseLatency(requestId, startedAtNanos, firstResponseRecorded);
}
recordBuiltInMetrics(
compositeTracer,
builtInMetricsAttributes,
requestId,
isDirectPathUsed,
isAfeEnabled,
gfeLatency,
afeLatency);
} catch (Throwable throwable) {
LOGGER.log(Level.WARNING, "Error recording metrics in onClose", throwable);
} finally {
RequestIdTargetTracker.remove(requestId);
super.onClose(status, trailers);
}
if (status.isOk()) {
recordFirstResponseLatency(requestId, startedAtNanos, firstResponseRecorded);
}
recordBuiltInMetrics(
compositeTracer,
builtInMetricsAttributes,
requestId,
isDirectPathUsed,
isAfeEnabled);
RequestIdTargetTracker.remove(requestId);
super.onClose(status, trailers);
}
},
headers);
Expand All @@ -172,11 +198,14 @@ public void onClose(Status status, Metadata trailers) {
}

private void recordCustomMetrics(
TagContext tagContext, Attributes attributes, Boolean isDirectPathUsed) {
TagContext tagContext,
Attributes attributes,
Boolean isDirectPathUsed,
@Nullable Float gfeLatency) {
// Record OpenCensus and Custom OpenTelemetry Metrics
MeasureMap measureMap = STATS_RECORDER.newMeasureMap();

if (!isDirectPathUsed) {
if (!Boolean.TRUE.equals(isDirectPathUsed)) {
if (gfeLatency != null) {
long gfeVal = gfeLatency.longValue();
measureMap.put(SPANNER_GFE_LATENCY, gfeVal);
Expand All @@ -191,31 +220,45 @@ private void recordCustomMetrics(
measureMap.record(tagContext);
}

private void recordSpan(Span span, String requestId) {
private void recordSpan(
@Nullable Span span,
@Nullable String requestId,
@Nullable Float gfeLatency,
@Nullable Float afeLatency) {
if (span != null) {
if (gfeLatency != null) {
span.setAttribute("gfe_latency", gfeLatency.toString());
}
if (afeLatency != null) {
span.setAttribute("afe_latency", afeLatency.toString());
}
span.setAttribute(XGoogSpannerRequestId.REQUEST_ID_HEADER_NAME, requestId);
if (requestId != null) {
span.setAttribute(XGoogSpannerRequestId.REQUEST_ID_HEADER_NAME, requestId);
}
}
}

private void recordBuiltInMetrics(
CompositeTracer compositeTracer,
@Nullable CompositeTracer compositeTracer,
Map<String, String> builtInMetricsAttributes,
String requestId,
@Nullable String requestId,
Boolean isDirectPathUsed,
Boolean isAfeEnabled) {
Boolean isAfeEnabled,
@Nullable Float gfeLatency,
@Nullable Float afeLatency) {
if (compositeTracer != null) {
builtInMetricsAttributes.put(BuiltInMetricsConstant.REQUEST_ID_KEY.getKey(), requestId);
if (requestId != null) {
builtInMetricsAttributes.put(BuiltInMetricsConstant.REQUEST_ID_KEY.getKey(), requestId);
}
builtInMetricsAttributes.put(
BuiltInMetricsConstant.DIRECT_PATH_USED_KEY.getKey(), Boolean.toString(isDirectPathUsed));
BuiltInMetricsConstant.DIRECT_PATH_USED_KEY.getKey(),
Boolean.toString(Boolean.TRUE.equals(isDirectPathUsed)));
compositeTracer.addAttributes(builtInMetricsAttributes);
compositeTracer.recordServerTimingHeaderMetrics(
gfeLatency, afeLatency, isDirectPathUsed, isAfeEnabled);
gfeLatency,
afeLatency,
Boolean.TRUE.equals(isDirectPathUsed),
Boolean.TRUE.equals(isAfeEnabled));
}
}

Expand Down
9 changes: 6 additions & 3 deletions ...-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/SpannerErrorInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,10 +118,13 @@ public void onClose(Status status, Metadata trailers) {
if (trailers.containsKey(RETRY_INFO_KEY)) {
status = status.augmentDescription(trailers.get(RETRY_INFO_KEY).toString());
}
} catch (IllegalArgumentException e) {
} catch (Throwable throwable) {
// Messages could be invalid if, say, some invalid UTF8 is echoed back in some
// error text.
logger.log(Level.WARNING, "Invalid protocol message in metadata", e);
// error text, or if an unexpected exception occurs during metadata inspection.
logger.log(
Level.WARNING,
"Error processing error details in SpannerErrorInterceptor",
throwable);
} finally {
super.onClose(status, trailers);
}
Expand Down
Loading
Loading