Skip to content
This repository was archived by the owner on May 8, 2026. It is now read-only.

Commit 89703d5

Browse files
igorbernstein2igorbernstein2
committed
chore: introduce PeerInfo & MetadataExtractor
Centralize sideband metadata collection using a new interceptor. Which gets injected into the GrpcCallContext channel. This provides the following benefits: - it works even if the end user sets their own channel provider - centralizes fetching of sideband metadata - removes the need for fetching directpath signals from grpc internals Change-Id: I42917074d65ccd7b8680f4a2a10c904b7646e4b6
1 parent 504bb51 commit 89703d5

13 files changed

Lines changed: 322 additions & 248 deletions

File tree

‎google-cloud-bigtable/pom.xml‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,6 @@
147147
<dependency>
148148
<groupId>io.grpc</groupId>
149149
<artifactId>grpc-alts</artifactId>
150-
<scope>runtime</scope>
151150
</dependency>
152151
<dependency>
153152
<groupId>org.checkerframework</groupId>

‎google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -922,7 +922,8 @@ private Builder() {
922922
.setReverseScans(true)
923923
.setLastScannedRowResponses(true)
924924
.setDirectAccessRequested(DIRECT_PATH_ENABLED)
925-
.setTrafficDirectorEnabled(DIRECT_PATH_ENABLED);
925+
.setTrafficDirectorEnabled(DIRECT_PATH_ENABLED)
926+
.setPeerInfo(true);
926927
}
927928

928929
private Builder(EnhancedBigtableStubSettings settings) {

‎google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/MetadataExtractorInterceptor.java‎

Lines changed: 192 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,192 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.google.cloud.bigtable.data.v2.stub;
17+
18+
import com.google.api.core.InternalApi;
19+
import com.google.api.gax.grpc.GrpcCallContext;
20+
import com.google.bigtable.v2.PeerInfo;
21+
import com.google.bigtable.v2.ResponseParams;
22+
import com.google.common.base.Strings;
23+
import com.google.protobuf.InvalidProtocolBufferException;
24+
import io.grpc.Attributes;
25+
import io.grpc.CallOptions;
26+
import io.grpc.Channel;
27+
import io.grpc.ClientCall;
28+
import io.grpc.ClientInterceptor;
29+
import io.grpc.ClientInterceptors;
30+
import io.grpc.ForwardingClientCall;
31+
import io.grpc.ForwardingClientCallListener;
32+
import io.grpc.Metadata;
33+
import io.grpc.MethodDescriptor;
34+
import io.grpc.Status;
35+
import io.grpc.alts.AltsContextUtil;
36+
37+
import javax.annotation.Nullable;
38+
import java.util.Base64;
39+
import java.util.regex.Matcher;
40+
import java.util.regex.Pattern;
41+
42+
@InternalApi
43+
public class MetadataExtractorInterceptor implements ClientInterceptor {
44+
private final SidebandData sidebandData = new SidebandData();
45+
46+
public GrpcCallContext injectInto(GrpcCallContext ctx) {
47+
return ctx
48+
.withChannel(ClientInterceptors.intercept(ctx.getChannel(), this))
49+
.withCallOptions(ctx.getCallOptions().withOption(SidebandData.KEY, sidebandData));
50+
}
51+
52+
@Override
53+
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions, Channel channel) {
54+
return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(channel.newCall(methodDescriptor, callOptions)) {
55+
@Override
56+
public void start(Listener<RespT> responseListener, Metadata headers) {
57+
sidebandData.reset();
58+
59+
super.start(new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(responseListener) {
60+
@Override
61+
public void onHeaders(Metadata headers) {
62+
sidebandData.onResponseHeaders(headers, getAttributes());
63+
super.onHeaders(headers);
64+
}
65+
66+
@Override
67+
public void onClose(Status status, Metadata trailers) {
68+
sidebandData.onClose(status, trailers);
69+
super.onClose(status, trailers);
70+
}
71+
}, headers);
72+
}
73+
};
74+
}
75+
76+
public SidebandData getSidebandData() {
77+
return sidebandData;
78+
}
79+
80+
public static class SidebandData {
81+
private static final CallOptions.Key<SidebandData> KEY = CallOptions.Key.create("bigtable-sideband");
82+
83+
private static final Metadata.Key<String> SERVER_TIMING_HEADER_KEY =
84+
Metadata.Key.of("server-timing", Metadata.ASCII_STRING_MARSHALLER);
85+
private static final Pattern SERVER_TIMING_HEADER_PATTERN = Pattern.compile(".*dur=(?<dur>\\d+)");
86+
private static final Metadata.Key<byte[]> LOCATION_METADATA_KEY =
87+
Metadata.Key.of("x-goog-ext-425905942-bin", Metadata.BINARY_BYTE_MARSHALLER);
88+
private static final Metadata.Key<String> PEER_INFO_KEY =
89+
Metadata.Key.of("bigtable-peer-info", Metadata.ASCII_STRING_MARSHALLER);
90+
91+
@Nullable
92+
private volatile ResponseParams responseParams;
93+
@Nullable
94+
private volatile PeerInfo peerInfo;
95+
@Nullable
96+
private volatile Long gfeTiming;
97+
98+
@Nullable
99+
public ResponseParams getResponseParams() {
100+
return responseParams;
101+
}
102+
103+
@Nullable
104+
public PeerInfo getPeerInfo() {
105+
return peerInfo;
106+
}
107+
108+
@Nullable
109+
public Long getGfeTiming() {
110+
return gfeTiming;
111+
}
112+
113+
114+
private void reset() {
115+
responseParams = null;
116+
peerInfo = null;
117+
gfeTiming = null;
118+
}
119+
void onResponseHeaders(Metadata md, Attributes attributes) {
120+
responseParams = extractResponseParams(md);
121+
gfeTiming = extractGfeLatency(md);
122+
peerInfo = extractPeerInfo(md, gfeTiming, attributes);
123+
}
124+
void onClose(Status status, Metadata trailers) {
125+
if (responseParams == null) {
126+
responseParams = extractResponseParams(trailers);
127+
}
128+
}
129+
130+
@Nullable
131+
private static Long extractGfeLatency(Metadata metadata) {
132+
String serverTiming = metadata.get(SERVER_TIMING_HEADER_KEY);
133+
if (serverTiming == null) {
134+
return null;
135+
}
136+
Matcher matcher = SERVER_TIMING_HEADER_PATTERN.matcher(serverTiming);
137+
// this should always be true
138+
if (matcher.find()) {
139+
return Long.parseLong(matcher.group("dur"));
140+
}
141+
return null;
142+
}
143+
144+
@Nullable
145+
private static PeerInfo extractPeerInfo(Metadata metadata, Long gfeTiming, Attributes attributes) {
146+
String encodedStr = metadata.get(PEER_INFO_KEY);
147+
if (Strings.isNullOrEmpty(encodedStr)) {
148+
return null;
149+
}
150+
151+
try {
152+
byte[] decoded = Base64.getUrlDecoder().decode(encodedStr);
153+
PeerInfo peerInfo = PeerInfo.parseFrom(decoded);
154+
PeerInfo.TransportType effectiveTransport = peerInfo.getTransportType();
155+
156+
if (effectiveTransport == PeerInfo.TransportType.TRANSPORT_TYPE_UNKNOWN) {
157+
boolean isAlts = AltsContextUtil.check(attributes);
158+
if (isAlts) {
159+
effectiveTransport = PeerInfo.TransportType.TRANSPORT_TYPE_DIRECT_ACCESS;
160+
} else if (gfeTiming != null) {
161+
effectiveTransport = PeerInfo.TransportType.TRANSPORT_TYPE_CLOUD_PATH;
162+
}
163+
}
164+
if (effectiveTransport != PeerInfo.TransportType.TRANSPORT_TYPE_UNKNOWN) {
165+
peerInfo = peerInfo.toBuilder()
166+
.setTransportType(effectiveTransport)
167+
.build();
168+
}
169+
return peerInfo;
170+
} catch (Exception e) {
171+
throw new IllegalArgumentException(
172+
"Failed to parse "
173+
+ PEER_INFO_KEY.name()
174+
+ " from the response header value: "
175+
+ encodedStr);
176+
}
177+
}
178+
179+
180+
@Nullable
181+
private static ResponseParams extractResponseParams(Metadata metadata) {
182+
byte[] responseParams = metadata.get(LOCATION_METADATA_KEY);
183+
if (responseParams != null) {
184+
try {
185+
return ResponseParams.parseFrom(responseParams);
186+
} catch (InvalidProtocolBufferException e) {
187+
}
188+
}
189+
return null;
190+
}
191+
}
192+
}

‎google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BigtableGrpcStreamTracer.java‎

Lines changed: 2 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -26,16 +26,9 @@
2626
* asking gRPC to start an RPC and gRPC actually serializing that RPC.
2727
*/
2828
class BigtableGrpcStreamTracer extends ClientStreamTracer {
29-
private static final String GRPC_LB_LOCALITY_KEY = "grpc.lb.locality";
30-
private static final String GRPC_LB_BACKEND_SERVICE_KEY = "grpc.lb.backend_service";
31-
32-
private final StreamInfo info;
3329
private final BigtableTracer tracer;
34-
private volatile String backendService = null;
35-
private volatile String locality = null;
3630

37-
public BigtableGrpcStreamTracer(StreamInfo info, BigtableTracer tracer) {
38-
this.info = info;
31+
public BigtableGrpcStreamTracer(BigtableTracer tracer) {
3932
this.tracer = tracer;
4033
}
4134

@@ -44,26 +37,6 @@ public void outboundMessageSent(int seqNo, long optionalWireSize, long optionalU
4437
tracer.grpcMessageSent();
4538
}
4639

47-
@Override
48-
public void addOptionalLabel(String key, String value) {
49-
switch (key) {
50-
case GRPC_LB_LOCALITY_KEY:
51-
this.locality = value;
52-
break;
53-
case GRPC_LB_BACKEND_SERVICE_KEY:
54-
this.backendService = value;
55-
break;
56-
}
57-
58-
super.addOptionalLabel(key, value);
59-
}
60-
61-
@Override
62-
public void streamClosed(Status status) {
63-
tracer.setTransportAttrs(TransportAttrs.create(locality, backendService));
64-
super.streamClosed(status);
65-
}
66-
6740
static class Factory extends ClientStreamTracer.Factory {
6841

6942
private final BigtableTracer tracer;
@@ -75,7 +48,7 @@ static class Factory extends ClientStreamTracer.Factory {
7548
@Override
7649
public ClientStreamTracer newClientStreamTracer(
7750
ClientStreamTracer.StreamInfo info, Metadata headers) {
78-
return new BigtableGrpcStreamTracer(info, tracer);
51+
return new BigtableGrpcStreamTracer(tracer);
7952
}
8053
}
8154
}

‎google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BigtableTracer.java‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
import com.google.api.gax.rpc.ApiCallContext;
2121
import com.google.api.gax.tracing.ApiTracer;
2222
import com.google.api.gax.tracing.BaseApiTracer;
23+
import com.google.cloud.bigtable.data.v2.stub.MetadataExtractorInterceptor;
24+
2325
import java.time.Duration;
2426
import javax.annotation.Nullable;
2527

@@ -74,7 +76,9 @@ public int getAttempt() {
7476
* Record the latency between Google's network receives the RPC and reads back the first byte of
7577
* the response from server-timing header. If server-timing header is missing, increment the
7678
* missing header count.
79+
* @deprecated Use {@link #setSidebandData(MetadataExtractorInterceptor.SidebandData)}
7780
*/
81+
@Deprecated
7882
public void recordGfeMetadata(@Nullable Long latency, @Nullable Throwable throwable) {
7983
// noop
8084
}
@@ -84,15 +88,23 @@ public void batchRequestThrottled(long throttledTimeMs) {
8488
// noop
8589
}
8690

91+
public void setSidebandData(MetadataExtractorInterceptor.SidebandData sidebandData) {
92+
// noop
93+
}
8794
/**
8895
* Set the Bigtable zone and cluster so metrics can be tagged with location information. This will
8996
* be called in BuiltinMetricsTracer.
97+
* @deprecated Use {@link #setSidebandData(MetadataExtractorInterceptor.SidebandData)}
9098
*/
99+
@Deprecated
91100
public void setLocations(String zone, String cluster) {
92-
// noop
93101
}
94102

95-
/** Set the underlying transport used to process the attempt */
103+
/**
104+
* Set the underlying transport used to process the attempt
105+
* @deprecated Use {@link #setSidebandData(MetadataExtractorInterceptor.SidebandData)}
106+
*/
107+
@Deprecated
96108
public void setTransportAttrs(BuiltinMetricsTracer.TransportAttrs attrs) {}
97109

98110
@Deprecated

0 commit comments

Comments
 (0)