flasn4nt
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

xds: Allow FaultFilter's interceptor to be reused 

This is the only usage of PickSubchannelArgs when creating a filter's
ClientInterceptor, and a follow-up commit will remove the argument and
actually reuse the interceptors. Other filter's interceptors can
already be reused.

There doesn't seem to be any significant loss of legibility by making
FaultFilter a more ordinary interceptor, but the change does cause the
ForwardingClientCall to be present when faultDelay is configured,
independent of whether the fault delay ends up being triggered.

Reusing interceptors will move more state management out of the RPC path
which will be more relevant with RLQS.
This commit is contained in:
Eric Anderson 2025-01-28 16:38:32 -08:00
parent 9e8629914f
commit b3db8c2489
1 changed files with 82 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

156
xds/src/main/java/io/grpc/xds/FaultFilter.java
View File

@ -190,94 +190,102 @@ final class FaultFilter implements Filter, ClientInterceptorBuilder {
config = overrideConfig; config = overrideConfig;
} }
FaultConfig faultConfig = (FaultConfig) config; FaultConfig faultConfig = (FaultConfig) config;
Long delayNanos = null;
Status abortStatus = null;
if (faultConfig.maxActiveFaults() == null
|| activeFaultCounter.get() < faultConfig.maxActiveFaults()) {
Metadata headers = args.getHeaders();
if (faultConfig.faultDelay() != null) {
delayNanos = determineFaultDelayNanos(faultConfig.faultDelay(), headers);
}
if (faultConfig.faultAbort() != null) {
abortStatus = determineFaultAbortStatus(faultConfig.faultAbort(), headers);
}
}
if (delayNanos == null && abortStatus == null) {
return null;
}
final Long finalDelayNanos = delayNanos;
final Status finalAbortStatus = getAbortStatusWithDescription(abortStatus);
final class FaultInjectionInterceptor implements ClientInterceptor { final class FaultInjectionInterceptor implements ClientInterceptor {
@Override @Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall( public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
final MethodDescriptor<ReqT, RespT> method, final CallOptions callOptions, final MethodDescriptor<ReqT, RespT> method, final CallOptions callOptions,
final Channel next) { final Channel next) {
Executor callExecutor = callOptions.getExecutor(); boolean checkFault = false;
if (callExecutor == null) { // This should never happen in practice because if (faultConfig.maxActiveFaults() == null
// ManagedChannelImpl.ConfigSelectingClientCall always provides CallOptions with || activeFaultCounter.get() < faultConfig.maxActiveFaults()) {
// a callExecutor. checkFault = faultConfig.faultDelay() != null || faultConfig.faultAbort() != null;
// TODO(https://github.com/grpc/grpc-java/issues/7868)
callExecutor = MoreExecutors.directExecutor();
} }
if (finalDelayNanos != null) { if (!checkFault) {
Supplier<? extends ClientCall<ReqT, RespT>> callSupplier; return next.newCall(method, callOptions);
if (finalAbortStatus != null) { }
callSupplier = Suppliers.ofInstance( final class DeadlineInsightForwardingCall extends ForwardingClientCall<ReqT, RespT> {
new FailingClientCall<ReqT, RespT>(finalAbortStatus, callExecutor)); private ClientCall<ReqT, RespT> delegate;
} else {
callSupplier = new Supplier<ClientCall<ReqT, RespT>>() {
@Override
public ClientCall<ReqT, RespT> get() {
return next.newCall(method, callOptions);
}
};
}
final DelayInjectedCall<ReqT, RespT> delayInjectedCall = new DelayInjectedCall<>(
finalDelayNanos, callExecutor, scheduler, callOptions.getDeadline(), callSupplier);
final class DeadlineInsightForwardingCall extends ForwardingClientCall<ReqT, RespT> { @Override
@Override protected ClientCall<ReqT, RespT> delegate() {
protected ClientCall<ReqT, RespT> delegate() { return delegate;
return delayInjectedCall; }
@Override
public void start(Listener<RespT> listener, Metadata headers) {
Executor callExecutor = callOptions.getExecutor();
if (callExecutor == null) { // This should never happen in practice because
// ManagedChannelImpl.ConfigSelectingClientCall always provides CallOptions with
// a callExecutor.
// TODO(https://github.com/grpc/grpc-java/issues/7868)
callExecutor = MoreExecutors.directExecutor();
} }
@Override Long delayNanos;
public void start(Listener<RespT> listener, Metadata headers) { Status abortStatus = null;
Listener<RespT> finalListener = if (faultConfig.faultDelay() != null) {
new SimpleForwardingClientCallListener<RespT>(listener) { delayNanos = determineFaultDelayNanos(faultConfig.faultDelay(), headers);
@Override } else {
public void onClose(Status status, Metadata trailers) { delayNanos = null;
if (status.getCode().equals(Code.DEADLINE_EXCEEDED)) { }
// TODO(zdapeng:) check effective deadline locally, and if (faultConfig.faultAbort() != null) {
// do the following only if the local deadline is exceeded. abortStatus = getAbortStatusWithDescription(
// (If the server sends DEADLINE_EXCEEDED for its own deadline, then the determineFaultAbortStatus(faultConfig.faultAbort(), headers));
// injected delay does not contribute to the error, because the request is }
// only sent out after the delay. There could be a race between local and
// remote, but it is rather rare.) Supplier<? extends ClientCall<ReqT, RespT>> callSupplier;
String description = String.format( if (abortStatus != null) {
Locale.US, callSupplier = Suppliers.ofInstance(
"Deadline exceeded after up to %d ns of fault-injected delay", new FailingClientCall<ReqT, RespT>(abortStatus, callExecutor));
finalDelayNanos); } else {
if (status.getDescription() != null) { callSupplier = new Supplier<ClientCall<ReqT, RespT>>() {
description = description + ": " + status.getDescription(); @Override
} public ClientCall<ReqT, RespT> get() {
status = Status.DEADLINE_EXCEEDED return next.newCall(method, callOptions);
.withDescription(description).withCause(status.getCause()); }
// Replace trailers to prevent mixing sources of status and trailers. };
trailers = new Metadata(); }
if (delayNanos == null) {
delegate = callSupplier.get();
delegate().start(listener, headers);
return;
}
delegate = new DelayInjectedCall<>(
delayNanos, callExecutor, scheduler, callOptions.getDeadline(), callSupplier);
Listener<RespT> finalListener =
new SimpleForwardingClientCallListener<RespT>(listener) {
@Override
public void onClose(Status status, Metadata trailers) {
if (status.getCode().equals(Code.DEADLINE_EXCEEDED)) {
// TODO(zdapeng:) check effective deadline locally, and
// do the following only if the local deadline is exceeded.
// (If the server sends DEADLINE_EXCEEDED for its own deadline, then the
// injected delay does not contribute to the error, because the request is
// only sent out after the delay. There could be a race between local and
// remote, but it is rather rare.)
String description = String.format(
Locale.US,
"Deadline exceeded after up to %d ns of fault-injected delay",
delayNanos);
if (status.getDescription() != null) {
description = description + ": " + status.getDescription();
} }
delegate().onClose(status, trailers); status = Status.DEADLINE_EXCEEDED
.withDescription(description).withCause(status.getCause());
// Replace trailers to prevent mixing sources of status and trailers.
trailers = new Metadata();
} }
}; delegate().onClose(status, trailers);
delegate().start(finalListener, headers); }
} };
delegate().start(finalListener, headers);
} }
return new DeadlineInsightForwardingCall();
} else {
return new FailingClientCall<>(finalAbortStatus, callExecutor);
} }
return new DeadlineInsightForwardingCall();
} }
} }