Skip to content

Commit 07629e0

Browse files
kannanjgithubkannanjgithub
committed
Refactoring to allow common code between the ext_proc client and server interceptors:
1. Moved the Otel metric instruments from the client interceptor to the filter class. Made the static initialization conditional on either the client flag or the server flag being true. 2. Moved `KnownLengthInputStream` used during observability mode to send message to the data plane to its own class in io.grpc.xds.internal package. 3. Moved method `outboundStreamFromByteString` used to construct the ProcessingRequest payload to ExternalProcessorFilter as a static utility method.
1 parent 71a10dc commit 07629e0

4 files changed

Lines changed: 185 additions & 150 deletions

File tree

‎xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java‎

Lines changed: 10 additions & 142 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,11 @@
1717
package io.grpc.xds;
1818

1919
import static com.google.common.base.Preconditions.checkNotNull;
20+
import static io.grpc.xds.ExternalProcessorFilter.clientHalfCloseDuration;
21+
import static io.grpc.xds.ExternalProcessorFilter.clientHeadersDuration;
22+
import static io.grpc.xds.ExternalProcessorFilter.outboundStreamToByteString;
23+
import static io.grpc.xds.ExternalProcessorFilter.serverHeadersDuration;
24+
import static io.grpc.xds.ExternalProcessorFilter.serverTrailersDuration;
2025

2126
import com.google.common.annotations.VisibleForTesting;
2227
import com.google.common.base.Joiner;
@@ -48,25 +53,24 @@
4853
import io.grpc.Context;
4954
import io.grpc.Deadline;
5055
import io.grpc.DoubleHistogramMetricInstrument;
51-
import io.grpc.Drainable;
5256
import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
5357
import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
54-
import io.grpc.KnownLength;
5558
import io.grpc.ManagedChannel;
5659
import io.grpc.Metadata;
5760
import io.grpc.MethodDescriptor;
58-
import io.grpc.MetricInstrumentRegistry;
5961
import io.grpc.MetricRecorder;
6062
import io.grpc.Status;
6163
import io.grpc.StatusRuntimeException;
6264
import io.grpc.internal.DelayedClientCall;
63-
import io.grpc.internal.GrpcUtil;
6465
import io.grpc.internal.SerializingExecutor;
6566
import io.grpc.stub.ClientCallStreamObserver;
6667
import io.grpc.stub.ClientResponseObserver;
68+
import io.grpc.xds.ExternalProcessorFilter.DataPlaneCallState;
69+
import io.grpc.xds.ExternalProcessorFilter.ExtProcStreamState;
6770
import io.grpc.xds.ExternalProcessorFilter.ExternalProcessorFilterConfig;
6871
import io.grpc.xds.ExternalProcessorFilter.HeaderForwardingRulesConfig;
6972
import io.grpc.xds.Filter.FilterContext;
73+
import io.grpc.xds.internal.KnownLengthInputStream;
7074
import io.grpc.xds.internal.grpcservice.CachedChannelManager;
7175
import io.grpc.xds.internal.grpcservice.HeaderValue;
7276
import io.grpc.xds.internal.grpcservice.HeaderValueValidationUtils;
@@ -98,76 +102,6 @@
98102
*/
99103
final class ExternalProcessorClientInterceptor implements ClientInterceptor {
100104

101-
102-
103-
@VisibleForTesting
104-
static final DoubleHistogramMetricInstrument clientHeadersDuration;
105-
@VisibleForTesting
106-
static final DoubleHistogramMetricInstrument clientHalfCloseDuration;
107-
@VisibleForTesting
108-
static final DoubleHistogramMetricInstrument serverHeadersDuration;
109-
@VisibleForTesting
110-
static final DoubleHistogramMetricInstrument serverTrailersDuration;
111-
112-
// Copied from io.grpc.opentelemetry.internal.OpenTelemetryConstants.LATENCY_BUCKETS
113-
private static final List<Double> LATENCY_BUCKETS = ImmutableList.of(
114-
0d, 0.00001d, 0.00005d, 0.0001d, 0.0003d, 0.0006d, 0.0008d, 0.001d, 0.002d,
115-
0.003d, 0.004d, 0.005d, 0.006d, 0.008d, 0.01d, 0.013d, 0.016d, 0.02d,
116-
0.025d, 0.03d, 0.04d, 0.05d, 0.065d, 0.08d, 0.1d, 0.13d, 0.16d,
117-
0.2d, 0.25d, 0.3d, 0.4d, 0.5d, 0.65d, 0.8d, 1d, 2d,
118-
5d, 10d, 20d, 50d, 100d);
119-
120-
static {
121-
if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)) {
122-
MetricInstrumentRegistry registry = MetricInstrumentRegistry.getDefaultRegistry();
123-
124-
clientHeadersDuration = registry.registerDoubleHistogram(
125-
"grpc.client_ext_proc.client_headers_duration",
126-
"Time between when the ext_proc filter sees the client's headers and when "
127-
+ "it allows those headers to continue on to the next filter",
128-
"s",
129-
LATENCY_BUCKETS,
130-
ImmutableList.of("grpc.target"),
131-
ImmutableList.of("grpc.lb.backend_service"),
132-
true);
133-
134-
clientHalfCloseDuration = registry.registerDoubleHistogram(
135-
"grpc.client_ext_proc.client_half_close_duration",
136-
"Time between when the ext_proc filter sees the client's half-close and when "
137-
+ "it allows that half-close to continue on to the next filter",
138-
"s",
139-
LATENCY_BUCKETS,
140-
ImmutableList.of("grpc.target"),
141-
ImmutableList.of("grpc.lb.backend_service"),
142-
true);
143-
144-
serverHeadersDuration = registry.registerDoubleHistogram(
145-
"grpc.client_ext_proc.server_headers_duration",
146-
"Time between when the ext_proc filter sees the server's headers and when "
147-
+ "it allows those headers to continue on to the next filter",
148-
"s",
149-
LATENCY_BUCKETS,
150-
ImmutableList.of("grpc.target"),
151-
ImmutableList.of("grpc.lb.backend_service"),
152-
true);
153-
154-
serverTrailersDuration = registry.registerDoubleHistogram(
155-
"grpc.client_ext_proc.server_trailers_duration",
156-
"Time between when the ext_proc filter sees the server's trailers and when "
157-
+ "it allows those trailers to continue on to the next filter",
158-
"s",
159-
LATENCY_BUCKETS,
160-
ImmutableList.of("grpc.target"),
161-
ImmutableList.of("grpc.lb.backend_service"),
162-
true);
163-
} else {
164-
clientHeadersDuration = null;
165-
clientHalfCloseDuration = null;
166-
serverHeadersDuration = null;
167-
serverTrailersDuration = null;
168-
}
169-
}
170-
171105
private final ExternalProcessorFilterConfig filterConfig;
172106
private final ScheduledExecutorService scheduler;
173107
private final MetricRecorder metricsRecorder;
@@ -182,6 +116,7 @@ final class ExternalProcessorClientInterceptor implements ClientInterceptor {
182116
checkNotNull(cachedChannelManager, "cachedChannelManager");
183117
this.scheduler = checkNotNull(scheduler, "scheduler");
184118
this.metricsRecorder = checkNotNull(context.metricsRecorder(), "metricsRecorder");
119+
ExternalProcessorFilter.initMetricInstruments();
185120
this.extProcChannel = cachedChannelManager.getChannel(filterConfig.getGrpcServiceConfig());
186121
}
187122

@@ -407,31 +342,6 @@ private static class DataPlaneDelayedCall<ReqT, RespT> extends DelayedClientCall
407342
*/
408343
private static class DataPlaneClientCall
409344
extends SimpleForwardingClientCall<InputStream, InputStream> {
410-
enum ExtProcStreamState {
411-
ACTIVE,
412-
DRAINING,
413-
COMPLETED,
414-
FAILED;
415-
416-
boolean isCompleted() {
417-
return this == COMPLETED || this == FAILED;
418-
}
419-
420-
boolean isFailed() {
421-
return this == FAILED;
422-
}
423-
424-
boolean isDraining() {
425-
return this == DRAINING;
426-
}
427-
}
428-
429-
enum DataPlaneCallState {
430-
IDLE,
431-
ACTIVE,
432-
CLOSED
433-
}
434-
435345
private enum EventType {
436346
REQUEST_HEADERS,
437347
REQUEST_BODY,
@@ -1395,7 +1305,7 @@ public void onMessage(InputStream message) {
13951305
@Override
13961306
public void onClose(Status status, Metadata trailers) {
13971307
dataPlaneClientCall.setServerTrailersStartNanos(System.nanoTime());
1398-
DataPlaneClientCall.ExtProcStreamState extProcStreamState =
1308+
ExtProcStreamState extProcStreamState =
13991309
dataPlaneClientCall.getExtProcStreamState().get();
14001310
if (extProcStreamState.isFailed()
14011311
&& !dataPlaneClientCall.getConfig().getObservabilityMode()
@@ -1593,46 +1503,4 @@ private void sendResponseBodyToExtProc(
15931503
.build());
15941504
}
15951505
}
1596-
1597-
1598-
1599-
private static ByteString outboundStreamToByteString(InputStream message) throws IOException {
1600-
if (message instanceof Drainable) {
1601-
int size = message.available();
1602-
ByteString.Output output =
1603-
size > 0 ? ByteString.newOutput(size) : ByteString.newOutput();
1604-
((Drainable) message).drainTo(output);
1605-
return output.toByteString();
1606-
}
1607-
return ByteString.readFrom(message);
1608-
}
1609-
1610-
private static final class KnownLengthInputStream extends InputStream
1611-
implements KnownLength {
1612-
private final InputStream delegate;
1613-
1614-
KnownLengthInputStream(ByteString byteString) {
1615-
this.delegate = byteString.newInput();
1616-
}
1617-
1618-
@Override
1619-
public int read() throws IOException {
1620-
return delegate.read();
1621-
}
1622-
1623-
@Override
1624-
public int read(byte[] b, int off, int len) throws IOException {
1625-
return delegate.read(b, off, len);
1626-
}
1627-
1628-
@Override
1629-
public int available() throws IOException {
1630-
return delegate.available();
1631-
}
1632-
1633-
@Override
1634-
public void close() throws IOException {
1635-
delegate.close();
1636-
}
1637-
}
16381506
}

‎xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.google.common.annotations.VisibleForTesting;
2222
import com.google.common.collect.ImmutableList;
2323
import com.google.protobuf.Any;
24+
import com.google.protobuf.ByteString;
2425
import com.google.protobuf.Duration;
2526
import com.google.protobuf.InvalidProtocolBufferException;
2627
import com.google.protobuf.Message;
@@ -32,6 +33,9 @@
3233
import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.HeaderForwardingRules;
3334
import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ProcessingMode;
3435
import io.grpc.ClientInterceptor;
36+
import io.grpc.DoubleHistogramMetricInstrument;
37+
import io.grpc.Drainable;
38+
import io.grpc.MetricInstrumentRegistry;
3539
import io.grpc.internal.GrpcUtil;
3640
import io.grpc.xds.Filter.FilterConfigParseContext;
3741
import io.grpc.xds.Filter.FilterContext;
@@ -43,6 +47,8 @@
4347
import io.grpc.xds.internal.headermutations.HeaderMutationRulesConfig;
4448
import io.grpc.xds.internal.headermutations.HeaderMutationRulesParseException;
4549
import io.grpc.xds.internal.headermutations.HeaderMutationRulesParser;
50+
import java.io.IOException;
51+
import java.io.InputStream;
4652
import java.util.List;
4753
import java.util.Locale;
4854
import java.util.Optional;
@@ -57,6 +63,103 @@ public class ExternalProcessorFilter implements Filter {
5763
static final String TYPE_URL =
5864
"type.googleapis.com/envoy.extensions.filters.http.ext_proc.v3.ExternalProcessor";
5965

66+
@VisibleForTesting
67+
static DoubleHistogramMetricInstrument clientHeadersDuration;
68+
@VisibleForTesting
69+
static DoubleHistogramMetricInstrument clientHalfCloseDuration;
70+
@VisibleForTesting
71+
static DoubleHistogramMetricInstrument serverHeadersDuration;
72+
@VisibleForTesting
73+
static DoubleHistogramMetricInstrument serverTrailersDuration;
74+
75+
// Copied from io.grpc.opentelemetry.internal.OpenTelemetryConstants.LATENCY_BUCKETS
76+
private static final List<Double> LATENCY_BUCKETS = ImmutableList.of(
77+
0d, 0.00001d, 0.00005d, 0.0001d, 0.0003d, 0.0006d, 0.0008d, 0.001d, 0.002d,
78+
0.003d, 0.004d, 0.005d, 0.006d, 0.008d, 0.01d, 0.013d, 0.016d, 0.02d,
79+
0.025d, 0.03d, 0.04d, 0.05d, 0.065d, 0.08d, 0.1d, 0.13d, 0.16d,
80+
0.2d, 0.25d, 0.3d, 0.4d, 0.5d, 0.65d, 0.8d, 1d, 2d,
81+
5d, 10d, 20d, 50d, 100d);
82+
83+
static {
84+
if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)
85+
|| GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", false)) {
86+
initMetricInstruments();
87+
}
88+
}
89+
90+
static synchronized void initMetricInstruments() {
91+
if (clientHeadersDuration == null
92+
&& (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)
93+
|| GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_SERVER", false))) {
94+
MetricInstrumentRegistry registry = MetricInstrumentRegistry.getDefaultRegistry();
95+
96+
clientHeadersDuration = registry.registerDoubleHistogram(
97+
"grpc.client_ext_proc.client_headers_duration",
98+
"Time between when the ext_proc filter sees the client's headers and when "
99+
+ "it allows those headers to continue on to the next filter",
100+
"s",
101+
LATENCY_BUCKETS,
102+
ImmutableList.of("grpc.target"),
103+
ImmutableList.of("grpc.lb.backend_service"),
104+
true);
105+
106+
clientHalfCloseDuration = registry.registerDoubleHistogram(
107+
"grpc.client_ext_proc.client_half_close_duration",
108+
"Time between when the ext_proc filter sees the client's half-close and when "
109+
+ "it allows that half-close to continue on to the next filter",
110+
"s",
111+
LATENCY_BUCKETS,
112+
ImmutableList.of("grpc.target"),
113+
ImmutableList.of("grpc.lb.backend_service"),
114+
true);
115+
116+
serverHeadersDuration = registry.registerDoubleHistogram(
117+
"grpc.client_ext_proc.server_headers_duration",
118+
"Time between when the ext_proc filter sees the server's headers and when "
119+
+ "it allows those headers to continue on to the next filter",
120+
"s",
121+
LATENCY_BUCKETS,
122+
ImmutableList.of("grpc.target"),
123+
ImmutableList.of("grpc.lb.backend_service"),
124+
true);
125+
126+
serverTrailersDuration = registry.registerDoubleHistogram(
127+
"grpc.client_ext_proc.server_trailers_duration",
128+
"Time between when the ext_proc filter sees the server's trailers and when "
129+
+ "it allows those trailers to continue on to the next filter",
130+
"s",
131+
LATENCY_BUCKETS,
132+
ImmutableList.of("grpc.target"),
133+
ImmutableList.of("grpc.lb.backend_service"),
134+
true);
135+
}
136+
}
137+
138+
enum ExtProcStreamState {
139+
ACTIVE,
140+
DRAINING,
141+
COMPLETED,
142+
FAILED;
143+
144+
boolean isCompleted() {
145+
return this == COMPLETED || this == FAILED;
146+
}
147+
148+
boolean isFailed() {
149+
return this == FAILED;
150+
}
151+
152+
boolean isDraining() {
153+
return this == DRAINING;
154+
}
155+
}
156+
157+
enum DataPlaneCallState {
158+
IDLE,
159+
ACTIVE,
160+
CLOSED
161+
}
162+
60163
private final CachedChannelManager cachedChannelManager;
61164
private final FilterContext context;
62165

@@ -444,4 +547,15 @@ boolean isAllowed(String headerName) {
444547
return true;
445548
}
446549
}
550+
551+
static ByteString outboundStreamToByteString(InputStream message) throws IOException {
552+
if (message instanceof Drainable) {
553+
int size = message.available();
554+
ByteString.Output output =
555+
size > 0 ? ByteString.newOutput(size) : ByteString.newOutput();
556+
((Drainable) message).drainTo(output);
557+
return output.toByteString();
558+
}
559+
return ByteString.readFrom(message);
560+
}
447561
}

0 commit comments

Comments
 (0)