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

Commit 54fee08

Browse files
igorbernstein2igorbernstein2
authored
chore: introduce PeerInfo & MetadataExtractor (#2788)
* 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 * format Change-Id: Id9dfd28ca4a3f9e98474b33c98895e05b830b410 * oops Change-Id: I87f110743cd6261e07b59bdcf2bc005af0916d35 * format Change-Id: Ide93ae0406012e8e5779d65cb9770dbda5d0562e * remove replaced location & gfe methods Change-Id: I2aff3f13f2f07b400d6f2099b75c9f1462077e44 * add todo Change-Id: If07939e62e04e9c7a27862bf87ca5b8731711b75 * fix null handling of sideband data formating and remove stale code Change-Id: Icf1b8ed5d020c9bf5386173b817a89f1679369b4 * todo Change-Id: I99e5dd3b4b2397d32fbd41ac7c4e697f6788cd4f * remove stale dep Change-Id: I3c98eef573aea3364f2788135407acbba991d7c7 * Eagerly set sideband data instead of deferring until onClose Also defensively add null checks for it and a todo to remove them Change-Id: Ie8237bfcc8c5a0886735ca5b93c0f03f5373e24b
1 parent 0542794 commit 54fee08

14 files changed

Lines changed: 328 additions & 397 deletions

‎google-cloud-bigtable/pom.xml‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -136,18 +136,13 @@
136136
<groupId>com.google.protobuf</groupId>
137137
<artifactId>protobuf-java-util</artifactId>
138138
</dependency>
139-
<dependency>
140-
<groupId>com.google.code.gson</groupId>
141-
<artifactId>gson</artifactId>
142-
</dependency>
143139
<dependency>
144140
<groupId>io.opencensus</groupId>
145141
<artifactId>opencensus-api</artifactId>
146142
</dependency>
147143
<dependency>
148144
<groupId>io.grpc</groupId>
149145
<artifactId>grpc-alts</artifactId>
150-
<scope>runtime</scope>
151146
</dependency>
152147
<dependency>
153148
<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: 198 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,198 @@
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+
import java.util.Base64;
37+
import java.util.regex.Matcher;
38+
import java.util.regex.Pattern;
39+
import javax.annotation.Nullable;
40+
41+
@InternalApi
42+
public class MetadataExtractorInterceptor implements ClientInterceptor {
43+
private final SidebandData sidebandData = new SidebandData();
44+
45+
public GrpcCallContext injectInto(GrpcCallContext ctx) {
46+
// TODO: migrate to using .withTransportChannel
47+
// This will require a change on gax's side to expose the underlying ManagedChannel in
48+
// GrpcTransportChannel (its currently package private).
49+
return ctx.withChannel(ClientInterceptors.intercept(ctx.getChannel(), this))
50+
.withCallOptions(ctx.getCallOptions().withOption(SidebandData.KEY, sidebandData));
51+
}
52+
53+
@Override
54+
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
55+
MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions, Channel channel) {
56+
return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
57+
channel.newCall(methodDescriptor, callOptions)) {
58+
@Override
59+
public void start(Listener<RespT> responseListener, Metadata headers) {
60+
sidebandData.reset();
61+
62+
super.start(
63+
new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(
64+
responseListener) {
65+
@Override
66+
public void onHeaders(Metadata headers) {
67+
sidebandData.onResponseHeaders(headers, getAttributes());
68+
super.onHeaders(headers);
69+
}
70+
71+
@Override
72+
public void onClose(Status status, Metadata trailers) {
73+
sidebandData.onClose(status, trailers);
74+
super.onClose(status, trailers);
75+
}
76+
},
77+
headers);
78+
}
79+
};
80+
}
81+
82+
public SidebandData getSidebandData() {
83+
return sidebandData;
84+
}
85+
86+
public static class SidebandData {
87+
private static final CallOptions.Key<SidebandData> KEY =
88+
CallOptions.Key.create("bigtable-sideband");
89+
90+
private static final Metadata.Key<String> SERVER_TIMING_HEADER_KEY =
91+
Metadata.Key.of("server-timing", Metadata.ASCII_STRING_MARSHALLER);
92+
private static final Pattern SERVER_TIMING_HEADER_PATTERN =
93+
Pattern.compile(".*dur=(?<dur>\\d+)");
94+
private static final Metadata.Key<byte[]> LOCATION_METADATA_KEY =
95+
Metadata.Key.of("x-goog-ext-425905942-bin", Metadata.BINARY_BYTE_MARSHALLER);
96+
private static final Metadata.Key<String> PEER_INFO_KEY =
97+
Metadata.Key.of("bigtable-peer-info", Metadata.ASCII_STRING_MARSHALLER);
98+
99+
@Nullable private volatile ResponseParams responseParams;
100+
@Nullable private volatile PeerInfo peerInfo;
101+
@Nullable private volatile Long gfeTiming;
102+
103+
@Nullable
104+
public ResponseParams getResponseParams() {
105+
return responseParams;
106+
}
107+
108+
@Nullable
109+
public PeerInfo getPeerInfo() {
110+
return peerInfo;
111+
}
112+
113+
@Nullable
114+
public Long getGfeTiming() {
115+
return gfeTiming;
116+
}
117+
118+
private void reset() {
119+
responseParams = null;
120+
peerInfo = null;
121+
gfeTiming = null;
122+
}
123+
124+
void onResponseHeaders(Metadata md, Attributes attributes) {
125+
responseParams = extractResponseParams(md);
126+
gfeTiming = extractGfeLatency(md);
127+
peerInfo = extractPeerInfo(md, gfeTiming, attributes);
128+
}
129+
130+
void onClose(Status status, Metadata trailers) {
131+
if (responseParams == null) {
132+
responseParams = extractResponseParams(trailers);
133+
}
134+
}
135+
136+
@Nullable
137+
private static Long extractGfeLatency(Metadata metadata) {
138+
String serverTiming = metadata.get(SERVER_TIMING_HEADER_KEY);
139+
if (serverTiming == null) {
140+
return null;
141+
}
142+
Matcher matcher = SERVER_TIMING_HEADER_PATTERN.matcher(serverTiming);
143+
// this should always be true
144+
if (matcher.find()) {
145+
return Long.parseLong(matcher.group("dur"));
146+
}
147+
return null;
148+
}
149+
150+
@Nullable
151+
private static PeerInfo extractPeerInfo(
152+
Metadata metadata, Long gfeTiming, Attributes attributes) {
153+
String encodedStr = metadata.get(PEER_INFO_KEY);
154+
if (Strings.isNullOrEmpty(encodedStr)) {
155+
return null;
156+
}
157+
158+
try {
159+
byte[] decoded = Base64.getUrlDecoder().decode(encodedStr);
160+
PeerInfo peerInfo = PeerInfo.parseFrom(decoded);
161+
PeerInfo.TransportType effectiveTransport = peerInfo.getTransportType();
162+
163+
// TODO: remove this once transport_type is being sent by the server
164+
// This is a temporary workaround to detect directpath until its available from the server
165+
if (effectiveTransport == PeerInfo.TransportType.TRANSPORT_TYPE_UNKNOWN) {
166+
boolean isAlts = AltsContextUtil.check(attributes);
167+
if (isAlts) {
168+
effectiveTransport = PeerInfo.TransportType.TRANSPORT_TYPE_DIRECT_ACCESS;
169+
} else if (gfeTiming != null) {
170+
effectiveTransport = PeerInfo.TransportType.TRANSPORT_TYPE_CLOUD_PATH;
171+
}
172+
}
173+
if (effectiveTransport != PeerInfo.TransportType.TRANSPORT_TYPE_UNKNOWN) {
174+
peerInfo = peerInfo.toBuilder().setTransportType(effectiveTransport).build();
175+
}
176+
return peerInfo;
177+
} catch (Exception e) {
178+
throw new IllegalArgumentException(
179+
"Failed to parse "
180+
+ PEER_INFO_KEY.name()
181+
+ " from the response header value: "
182+
+ encodedStr);
183+
}
184+
}
185+
186+
@Nullable
187+
private static ResponseParams extractResponseParams(Metadata metadata) {
188+
byte[] responseParams = metadata.get(LOCATION_METADATA_KEY);
189+
if (responseParams != null) {
190+
try {
191+
return ResponseParams.parseFrom(responseParams);
192+
} catch (InvalidProtocolBufferException e) {
193+
}
194+
}
195+
return null;
196+
}
197+
}
198+
}

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

Lines changed: 2 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -15,27 +15,18 @@
1515
*/
1616
package com.google.cloud.bigtable.data.v2.stub.metrics;
1717

18-
import com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsTracer.TransportAttrs;
1918
import io.grpc.ClientStreamTracer;
2019
import io.grpc.Metadata;
21-
import io.grpc.Status;
2220

2321
/**
2422
* Records the time a request is enqueued in a grpc channel queue. This a bridge between gRPC stream
2523
* tracing and Bigtable tracing. Its primary purpose is to measure the transition time between
2624
* asking gRPC to start an RPC and gRPC actually serializing that RPC.
2725
*/
2826
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;
3327
private final BigtableTracer tracer;
34-
private volatile String backendService = null;
35-
private volatile String locality = null;
3628

37-
public BigtableGrpcStreamTracer(StreamInfo info, BigtableTracer tracer) {
38-
this.info = info;
29+
public BigtableGrpcStreamTracer(BigtableTracer tracer) {
3930
this.tracer = tracer;
4031
}
4132

@@ -44,26 +35,6 @@ public void outboundMessageSent(int seqNo, long optionalWireSize, long optionalU
4435
tracer.grpcMessageSent();
4536
}
4637

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-
6738
static class Factory extends ClientStreamTracer.Factory {
6839

6940
private final BigtableTracer tracer;
@@ -75,7 +46,7 @@ static class Factory extends ClientStreamTracer.Factory {
7546
@Override
7647
public ClientStreamTracer newClientStreamTracer(
7748
ClientStreamTracer.StreamInfo info, Metadata headers) {
78-
return new BigtableGrpcStreamTracer(info, tracer);
49+
return new BigtableGrpcStreamTracer(tracer);
7950
}
8051
}
8152
}

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

Lines changed: 2 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
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;
2324
import java.time.Duration;
2425
import javax.annotation.Nullable;
2526

@@ -70,36 +71,12 @@ public int getAttempt() {
7071
return attempt;
7172
}
7273

73-
/**
74-
* Record the latency between Google's network receives the RPC and reads back the first byte of
75-
* the response from server-timing header. If server-timing header is missing, increment the
76-
* missing header count.
77-
*/
78-
public void recordGfeMetadata(@Nullable Long latency, @Nullable Throwable throwable) {
79-
// noop
80-
}
81-
8274
/** Adds an annotation of the total throttled time of a batch. */
8375
public void batchRequestThrottled(long throttledTimeMs) {
8476
// noop
8577
}
8678

87-
/**
88-
* Set the Bigtable zone and cluster so metrics can be tagged with location information. This will
89-
* be called in BuiltinMetricsTracer.
90-
*/
91-
public void setLocations(String zone, String cluster) {
92-
// noop
93-
}
94-
95-
/** Set the underlying transport used to process the attempt */
96-
public void setTransportAttrs(BuiltinMetricsTracer.TransportAttrs attrs) {}
97-
98-
@Deprecated
99-
/**
100-
* @deprecated {@link #grpcMessageSent()} is called instead.
101-
*/
102-
public void grpcChannelQueuedLatencies(long queuedTimeMs) {
79+
public void setSidebandData(MetadataExtractorInterceptor.SidebandData sidebandData) {
10380
// noop
10481
}
10582

0 commit comments

Comments
 (0)