#20276

Committed 11 May 2026 08:24AM UTC coverage: 88.821% (-0.02%) from 88.838%

Build # #20276

Build Type

push

github

Committed by

web-flow

Commit Message

Allow injecting bootstrap info into xDS Filter API for config parsing (#12724)

Extend the xDS Filter API to support injecting bootstrap information
into filters during configuration parsing. This allows filters to access
context information (e.g., allowed gRPC services) from the resource loading
layer during configuration validation and parsing.

- Update `Filter.Provider.parseFilterConfig` and
`parseFilterConfigOverride`
  to accept a `FilterContext` parameter.
- Introduce `BootstrapInfoGrpcServiceContextProvider` to encapsulate
  bootstrap info for context resolution.
- Update `XdsListenerResource` and `XdsRouteConfigureResource` to
  construct and pass `FilterContext` during configuration parsing.
- Update sub-filters (`FaultFilter`, `RbacFilter`,
`GcpAuthenticationFilter`,
  `RouterFilter`) to match the updated `FilterContext` signature.

Known Gaps & Limitations:
1. **MetricHolder**: Propagation of `MetricHolder` is not supported with
   this approach currently and is planned for support in a later phase.
2. **NameResolverRegistry**: Propagation is deferred for consistency.
While it could be passed from `XdsNameResolver` on the client side, there is
no equivalent mechanism on the server side. To ensure consistent
behavior, `DefaultRegistry` is used when validating schemes and creating channels.

Coverage Stats

36254 of 40817 relevant lines covered (88.82%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.77

/../xds/src/main/java/io/grpc/xds/FaultFilter.java

/*
 * Copyright 2021 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.xds;

import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;
import com.google.protobuf.util.Durations;
import io.envoyproxy.envoy.extensions.filters.http.fault.v3.HTTPFault;
import io.envoyproxy.envoy.type.v3.FractionalPercent;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.Context;
import io.grpc.Deadline;
import io.grpc.ForwardingClientCall;
import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.Status.Code;
import io.grpc.internal.DelayedClientCall;
import io.grpc.internal.GrpcUtil;
import io.grpc.xds.FaultConfig.FaultAbort;
import io.grpc.xds.FaultConfig.FaultDelay;
import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
import java.util.Locale;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;

/** HttpFault filter implementation. */
final class FaultFilter implements Filter {

  private static final FaultFilter INSTANCE =
      new FaultFilter(ThreadSafeRandomImpl.instance, new AtomicLong());

  @VisibleForTesting
  static final Metadata.Key<String> HEADER_DELAY_KEY =
      Metadata.Key.of("x-envoy-fault-delay-request", Metadata.ASCII_STRING_MARSHALLER);
  @VisibleForTesting
  static final Metadata.Key<String> HEADER_DELAY_PERCENTAGE_KEY =
      Metadata.Key.of("x-envoy-fault-delay-request-percentage", Metadata.ASCII_STRING_MARSHALLER);
  @VisibleForTesting
  static final Metadata.Key<String> HEADER_ABORT_HTTP_STATUS_KEY =
      Metadata.Key.of("x-envoy-fault-abort-request", Metadata.ASCII_STRING_MARSHALLER);
  @VisibleForTesting
  static final Metadata.Key<String> HEADER_ABORT_GRPC_STATUS_KEY =
      Metadata.Key.of("x-envoy-fault-abort-grpc-request", Metadata.ASCII_STRING_MARSHALLER);
  @VisibleForTesting
  static final Metadata.Key<String> HEADER_ABORT_PERCENTAGE_KEY =
      Metadata.Key.of("x-envoy-fault-abort-request-percentage", Metadata.ASCII_STRING_MARSHALLER);
  static final String TYPE_URL =
      "type.googleapis.com/envoy.extensions.filters.http.fault.v3.HTTPFault";

  private final ThreadSafeRandom random;
  private final AtomicLong activeFaultCounter;

  @VisibleForTesting
  FaultFilter(ThreadSafeRandom random, AtomicLong activeFaultCounter) {
    this.random = random;
    this.activeFaultCounter = activeFaultCounter;
  }

  static final class Provider implements Filter.Provider {
    @Override
    public String[] typeUrls() {
      return new String[]{TYPE_URL};
    }

    @Override
    public boolean isClientFilter() {
      return true;
    }

    @Override
    public FaultFilter newInstance(String name) {
      return INSTANCE;
    }

    @Override
    public ConfigOrError<FaultConfig> parseFilterConfig(
        Message rawProtoMessage, FilterConfigParseContext context) {
      HTTPFault httpFaultProto;
      if (!(rawProtoMessage instanceof Any)) {
        return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());
      }
      Any anyMessage = (Any) rawProtoMessage;
      try {
        httpFaultProto = anyMessage.unpack(HTTPFault.class);
      } catch (InvalidProtocolBufferException e) {
        return ConfigOrError.fromError("Invalid proto: " + e);
      }
      return parseHttpFault(httpFaultProto);
    }

    @Override
    public ConfigOrError<FaultConfig> parseFilterConfigOverride(
        Message rawProtoMessage, FilterConfigParseContext context) {
      return parseFilterConfig(rawProtoMessage, context);
    }

    private static ConfigOrError<FaultConfig> parseHttpFault(HTTPFault httpFault) {
      FaultDelay faultDelay = null;
      FaultAbort faultAbort = null;
      if (httpFault.hasDelay()) {
        faultDelay = parseFaultDelay(httpFault.getDelay());
      }
      if (httpFault.hasAbort()) {
        ConfigOrError<FaultAbort> faultAbortOrError = parseFaultAbort(httpFault.getAbort());
        if (faultAbortOrError.errorDetail != null) {
          return ConfigOrError.fromError(
              "HttpFault contains invalid FaultAbort: " + faultAbortOrError.errorDetail);
        }
        faultAbort = faultAbortOrError.config;
      }
      Integer maxActiveFaults = null;
      if (httpFault.hasMaxActiveFaults()) {
        maxActiveFaults = httpFault.getMaxActiveFaults().getValue();
        if (maxActiveFaults < 0) {
          maxActiveFaults = Integer.MAX_VALUE;
        }
      }
      return ConfigOrError.fromConfig(FaultConfig.create(faultDelay, faultAbort, maxActiveFaults));
    }

    private static FaultDelay parseFaultDelay(
        io.envoyproxy.envoy.extensions.filters.common.fault.v3.FaultDelay faultDelay) {
      FaultConfig.FractionalPercent percent = parsePercent(faultDelay.getPercentage());
      if (faultDelay.hasHeaderDelay()) {
        return FaultDelay.forHeader(percent);
      }
      return FaultDelay.forFixedDelay(Durations.toNanos(faultDelay.getFixedDelay()), percent);
    }

    @VisibleForTesting
    static ConfigOrError<FaultAbort> parseFaultAbort(
        io.envoyproxy.envoy.extensions.filters.http.fault.v3.FaultAbort faultAbort) {
      FaultConfig.FractionalPercent percent = parsePercent(faultAbort.getPercentage());
      switch (faultAbort.getErrorTypeCase()) {
        case HEADER_ABORT:
          return ConfigOrError.fromConfig(FaultAbort.forHeader(percent));
        case HTTP_STATUS:
          return ConfigOrError.fromConfig(FaultAbort.forStatus(
              GrpcUtil.httpStatusToGrpcStatus(faultAbort.getHttpStatus()), percent));
        case GRPC_STATUS:
          return ConfigOrError.fromConfig(FaultAbort.forStatus(
              Status.fromCodeValue(faultAbort.getGrpcStatus()), percent));
        case ERRORTYPE_NOT_SET:
        default:
          return ConfigOrError.fromError(
              "Unknown error type case: " + faultAbort.getErrorTypeCase());
      }
    }

    private static FaultConfig.FractionalPercent parsePercent(FractionalPercent proto) {
      switch (proto.getDenominator()) {
        case HUNDRED:
          return FaultConfig.FractionalPercent.perHundred(proto.getNumerator());
        case TEN_THOUSAND:
          return FaultConfig.FractionalPercent.perTenThousand(proto.getNumerator());
        case MILLION:
          return FaultConfig.FractionalPercent.perMillion(proto.getNumerator());
        case UNRECOGNIZED:
        default:
          throw new IllegalArgumentException("Unknown denominator type: " + proto.getDenominator());
      }
    }
  }

  @Nullable
  @Override
  public ClientInterceptor buildClientInterceptor(
      FilterConfig config, @Nullable FilterConfig overrideConfig,
      final ScheduledExecutorService scheduler) {
    checkNotNull(config, "config");
    if (overrideConfig != null) {
      config = overrideConfig;
    }
    FaultConfig faultConfig = (FaultConfig) config;

    final class FaultInjectionInterceptor implements ClientInterceptor {
      @Override
      public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
          final MethodDescriptor<ReqT, RespT> method, final CallOptions callOptions,
          final Channel next) {
        boolean checkFault = false;
        if (faultConfig.maxActiveFaults() == null
            || activeFaultCounter.get() < faultConfig.maxActiveFaults()) {
          checkFault = faultConfig.faultDelay() != null || faultConfig.faultAbort() != null;
        }
        if (!checkFault) {
          return next.newCall(method, callOptions);
        }
        final class DeadlineInsightForwardingCall extends ForwardingClientCall<ReqT, RespT> {
          private ClientCall<ReqT, RespT> delegate;

          @Override
          protected ClientCall<ReqT, RespT> delegate() {
            return delegate;
          }

          @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();
            }

            Long delayNanos;
            Status abortStatus = null;
            if (faultConfig.faultDelay() != null) {
              delayNanos = determineFaultDelayNanos(faultConfig.faultDelay(), headers);
            } else {
              delayNanos = null;
            }
            if (faultConfig.faultAbort() != null) {
              abortStatus = getAbortStatusWithDescription(
                  determineFaultAbortStatus(faultConfig.faultAbort(), headers));
            }

            Supplier<? extends ClientCall<ReqT, RespT>> callSupplier;
            if (abortStatus != null) {
              callSupplier = Suppliers.ofInstance(
                  new FailingClientCall<ReqT, RespT>(abortStatus, callExecutor));
            } else {
              callSupplier = new Supplier<ClientCall<ReqT, RespT>>() {
                @Override
                public ClientCall<ReqT, RespT> get() {
                  return next.newCall(method, callOptions);
                }
              };
            }
            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();
                      }
                      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);
          }
        }

        return new DeadlineInsightForwardingCall();
      }
    }

    return new FaultInjectionInterceptor();
  }

  private static Status getAbortStatusWithDescription(Status abortStatus) {
    Status finalAbortStatus = null;
    if (abortStatus != null) {
      String abortDesc = "RPC terminated due to fault injection";
      if (abortStatus.getDescription() != null) {
        abortDesc = abortDesc + ": " + abortStatus.getDescription();
      }
      finalAbortStatus = abortStatus.withDescription(abortDesc);
    }
    return finalAbortStatus;
  }

  @Nullable
  private Long determineFaultDelayNanos(FaultDelay faultDelay, Metadata headers) {
    Long delayNanos;
    FaultConfig.FractionalPercent fractionalPercent = faultDelay.percent();
    if (faultDelay.headerDelay()) {
      try {
        int delayMillis = Integer.parseInt(headers.get(HEADER_DELAY_KEY));
        delayNanos = TimeUnit.MILLISECONDS.toNanos(delayMillis);
        String delayPercentageStr = headers.get(HEADER_DELAY_PERCENTAGE_KEY);
        if (delayPercentageStr != null) {
          int delayPercentage = Integer.parseInt(delayPercentageStr);
          if (delayPercentage >= 0 && delayPercentage < fractionalPercent.numerator()) {
            fractionalPercent = FaultConfig.FractionalPercent.create(
                delayPercentage, fractionalPercent.denominatorType());
          }
        }
      } catch (NumberFormatException e) {
        return null; // treated as header_delay not applicable
      }
    } else {
      delayNanos = faultDelay.delayNanos();
    }
    if (random.nextInt(1_000_000) >= getRatePerMillion(fractionalPercent)) {
      return null;
    }
    return delayNanos;
  }

  @Nullable
  private Status determineFaultAbortStatus(FaultAbort faultAbort, Metadata headers) {
    Status abortStatus = null;
    FaultConfig.FractionalPercent fractionalPercent = faultAbort.percent();
    if (faultAbort.headerAbort()) {
      try {
        String grpcCodeStr = headers.get(HEADER_ABORT_GRPC_STATUS_KEY);
        if (grpcCodeStr != null) {
          int grpcCode = Integer.parseInt(grpcCodeStr);
          abortStatus = Status.fromCodeValue(grpcCode);
        }
        String httpCodeStr = headers.get(HEADER_ABORT_HTTP_STATUS_KEY);
        if (httpCodeStr != null) {
          int httpCode = Integer.parseInt(httpCodeStr);
          abortStatus = GrpcUtil.httpStatusToGrpcStatus(httpCode);
        }
        String abortPercentageStr = headers.get(HEADER_ABORT_PERCENTAGE_KEY);
        if (abortPercentageStr != null) {
          int abortPercentage =
              Integer.parseInt(headers.get(HEADER_ABORT_PERCENTAGE_KEY));
          if (abortPercentage >= 0 && abortPercentage < fractionalPercent.numerator()) {
            fractionalPercent = FaultConfig.FractionalPercent.create(
                abortPercentage, fractionalPercent.denominatorType());
          }
        }
      } catch (NumberFormatException e) {
        return null; // treated as header_abort not applicable
      }
    } else {
      abortStatus = faultAbort.status();
    }
    if (random.nextInt(1_000_000) >= getRatePerMillion(fractionalPercent)) {
      return null;
    }
    return abortStatus;
  }

  private static int getRatePerMillion(FaultConfig.FractionalPercent percent) {
    int numerator = percent.numerator();
    FaultConfig.FractionalPercent.DenominatorType type = percent.denominatorType();
    switch (type) {
      case TEN_THOUSAND:
        numerator *= 100;
        break;
      case HUNDRED:
        numerator *= 10_000;
        break;
      case MILLION:
      default:
        break;
    }
    if (numerator > 1_000_000 || numerator < 0) {
      numerator = 1_000_000;
    }
    return numerator;
  }

  /** A {@link DelayedClientCall} with a fixed delay. */
  private final class DelayInjectedCall<ReqT, RespT> extends DelayedClientCall<ReqT, RespT> {
    final Object lock = new Object();
    ScheduledFuture<?> delayTask;
    boolean cancelled;

    DelayInjectedCall(
        long delayNanos, Executor callExecutor, ScheduledExecutorService scheduler,
        @Nullable Deadline deadline,
        final Supplier<? extends ClientCall<ReqT, RespT>> callSupplier) {
      super(callExecutor, scheduler, deadline);
      activeFaultCounter.incrementAndGet();
      ScheduledFuture<?> task = scheduler.schedule(
          new Runnable() {
            @Override
            public void run() {
              synchronized (lock) {
                if (!cancelled) {
                  activeFaultCounter.decrementAndGet();
                }
              }
              Runnable toRun = setCall(callSupplier.get());
              if (toRun != null) {
                toRun.run();
              }
            }
          },
          delayNanos,
          NANOSECONDS);
      synchronized (lock) {
        if (!cancelled) {
          delayTask = task;
          return;
        }
      }
      task.cancel(false);
    }

    @Override
    protected void callCancelled() {
      ScheduledFuture<?> savedDelayTask;
      synchronized (lock) {
        cancelled = true;
        activeFaultCounter.decrementAndGet();
        savedDelayTask = delayTask;
      }
      if (savedDelayTask != null) {
        savedDelayTask.cancel(false);
      }
    }
  }

  /** An implementation of {@link ClientCall} that fails when started. */
  private final class FailingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
    final Status error;
    final Executor callExecutor;
    final Context context;

    FailingClientCall(Status error, Executor callExecutor) {
      this.error = error;
      this.callExecutor = callExecutor;
      this.context = Context.current();
    }

    @Override
    public void start(final ClientCall.Listener<RespT> listener, Metadata headers) {
      activeFaultCounter.incrementAndGet();
      callExecutor.execute(
          new Runnable() {
            @Override
            public void run() {
              Context previous = context.attach();
              try {
                listener.onClose(error, new Metadata());
                activeFaultCounter.decrementAndGet();
              } finally {
                context.detach(previous);
              }
            }
          });
    }

    @Override
    public void request(int numMessages) {}

    @Override
    public void cancel(String message, Throwable cause) {}

    @Override
    public void halfClose() {}

    @Override
    public void sendMessage(ReqT message) {}
  }
}

1	/*
2	* Copyright 2021 The gRPC Authors
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	* http://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
17	package io.grpc.xds;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20	import static java.util.concurrent.TimeUnit.NANOSECONDS;
21
22	import com.google.common.annotations.VisibleForTesting;
23	import com.google.common.base.Supplier;
24	import com.google.common.base.Suppliers;
25	import com.google.common.util.concurrent.MoreExecutors;
26	import com.google.protobuf.Any;
27	import com.google.protobuf.InvalidProtocolBufferException;
28	import com.google.protobuf.Message;
29	import com.google.protobuf.util.Durations;
30	import io.envoyproxy.envoy.extensions.filters.http.fault.v3.HTTPFault;
31	import io.envoyproxy.envoy.type.v3.FractionalPercent;
32	import io.grpc.CallOptions;
33	import io.grpc.Channel;
34	import io.grpc.ClientCall;
35	import io.grpc.ClientInterceptor;
36	import io.grpc.Context;
37	import io.grpc.Deadline;
38	import io.grpc.ForwardingClientCall;
39	import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
40	import io.grpc.Metadata;
41	import io.grpc.MethodDescriptor;
42	import io.grpc.Status;
43	import io.grpc.Status.Code;
44	import io.grpc.internal.DelayedClientCall;
45	import io.grpc.internal.GrpcUtil;
46	import io.grpc.xds.FaultConfig.FaultAbort;
47	import io.grpc.xds.FaultConfig.FaultDelay;
48	import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
49	import java.util.Locale;
50	import java.util.concurrent.Executor;
51	import java.util.concurrent.ScheduledExecutorService;
52	import java.util.concurrent.ScheduledFuture;
53	import java.util.concurrent.TimeUnit;
54	import java.util.concurrent.atomic.AtomicLong;
55	import javax.annotation.Nullable;
56
57	/** HttpFault filter implementation. */
58	final class FaultFilter implements Filter {
59
60	private static final FaultFilter INSTANCE =	1✔
61	new FaultFilter(ThreadSafeRandomImpl.instance, new AtomicLong());
62
63	@VisibleForTesting
64	static final Metadata.Key<String> HEADER_DELAY_KEY =	1✔
65	Metadata.Key.of("x-envoy-fault-delay-request", Metadata.ASCII_STRING_MARSHALLER);	1✔
66	@VisibleForTesting
67	static final Metadata.Key<String> HEADER_DELAY_PERCENTAGE_KEY =	1✔
68	Metadata.Key.of("x-envoy-fault-delay-request-percentage", Metadata.ASCII_STRING_MARSHALLER);	1✔
69	@VisibleForTesting
70	static final Metadata.Key<String> HEADER_ABORT_HTTP_STATUS_KEY =	1✔
71	Metadata.Key.of("x-envoy-fault-abort-request", Metadata.ASCII_STRING_MARSHALLER);	1✔
72	@VisibleForTesting
73	static final Metadata.Key<String> HEADER_ABORT_GRPC_STATUS_KEY =	1✔
74	Metadata.Key.of("x-envoy-fault-abort-grpc-request", Metadata.ASCII_STRING_MARSHALLER);	1✔
75	@VisibleForTesting
76	static final Metadata.Key<String> HEADER_ABORT_PERCENTAGE_KEY =	1✔
77	Metadata.Key.of("x-envoy-fault-abort-request-percentage", Metadata.ASCII_STRING_MARSHALLER);	1✔
78	static final String TYPE_URL =
79	"type.googleapis.com/envoy.extensions.filters.http.fault.v3.HTTPFault";
80
81	private final ThreadSafeRandom random;
82	private final AtomicLong activeFaultCounter;
83
84	@VisibleForTesting
85	FaultFilter(ThreadSafeRandom random, AtomicLong activeFaultCounter) {	1✔
86	this.random = random;	1✔
87	this.activeFaultCounter = activeFaultCounter;	1✔
88	}	1✔
89
90	static final class Provider implements Filter.Provider {	1✔
91	@Override
92	public String[] typeUrls() {
93	return new String[]{TYPE_URL};	1✔
94	}
95
96	@Override
97	public boolean isClientFilter() {
98	return true;	1✔
99	}
100
101	@Override
102	public FaultFilter newInstance(String name) {
103	return INSTANCE;	×
104	}
105
106	@Override
107	public ConfigOrError<FaultConfig> parseFilterConfig(
108	Message rawProtoMessage, FilterConfigParseContext context) {
109	HTTPFault httpFaultProto;
110	if (!(rawProtoMessage instanceof Any)) {	1✔
111	return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());	×
112	}
113	Any anyMessage = (Any) rawProtoMessage;	1✔
114	try {
115	httpFaultProto = anyMessage.unpack(HTTPFault.class);	1✔
116	} catch (InvalidProtocolBufferException e) {	×
117	return ConfigOrError.fromError("Invalid proto: " + e);	×
118	}	1✔
119	return parseHttpFault(httpFaultProto);	1✔
120	}
121
122	@Override
123	public ConfigOrError<FaultConfig> parseFilterConfigOverride(
124	Message rawProtoMessage, FilterConfigParseContext context) {
125	return parseFilterConfig(rawProtoMessage, context);	1✔
126	}
127
128	private static ConfigOrError<FaultConfig> parseHttpFault(HTTPFault httpFault) {
129	FaultDelay faultDelay = null;	1✔
130	FaultAbort faultAbort = null;	1✔
131	if (httpFault.hasDelay()) {	1✔
132	faultDelay = parseFaultDelay(httpFault.getDelay());	1✔
133	}
134	if (httpFault.hasAbort()) {	1✔
135	ConfigOrError<FaultAbort> faultAbortOrError = parseFaultAbort(httpFault.getAbort());	1✔
136	if (faultAbortOrError.errorDetail != null) {	1✔
137	return ConfigOrError.fromError(	×
138	"HttpFault contains invalid FaultAbort: " + faultAbortOrError.errorDetail);
139	}
140	faultAbort = faultAbortOrError.config;	1✔
141	}
142	Integer maxActiveFaults = null;	1✔
143	if (httpFault.hasMaxActiveFaults()) {	1✔
144	maxActiveFaults = httpFault.getMaxActiveFaults().getValue();	1✔
145	if (maxActiveFaults < 0) {	1✔
146	maxActiveFaults = Integer.MAX_VALUE;	×
147	}
148	}
149	return ConfigOrError.fromConfig(FaultConfig.create(faultDelay, faultAbort, maxActiveFaults));	1✔
150	}
151
152	private static FaultDelay parseFaultDelay(
153	io.envoyproxy.envoy.extensions.filters.common.fault.v3.FaultDelay faultDelay) {
154	FaultConfig.FractionalPercent percent = parsePercent(faultDelay.getPercentage());	1✔
155	if (faultDelay.hasHeaderDelay()) {	1✔
156	return FaultDelay.forHeader(percent);	×
157	}
158	return FaultDelay.forFixedDelay(Durations.toNanos(faultDelay.getFixedDelay()), percent);	1✔
159	}
160
161	@VisibleForTesting
162	static ConfigOrError<FaultAbort> parseFaultAbort(
163	io.envoyproxy.envoy.extensions.filters.http.fault.v3.FaultAbort faultAbort) {
164	FaultConfig.FractionalPercent percent = parsePercent(faultAbort.getPercentage());	1✔
165	switch (faultAbort.getErrorTypeCase()) {	1✔
166	case HEADER_ABORT:
167	return ConfigOrError.fromConfig(FaultAbort.forHeader(percent));	1✔
168	case HTTP_STATUS:
169	return ConfigOrError.fromConfig(FaultAbort.forStatus(	1✔
170	GrpcUtil.httpStatusToGrpcStatus(faultAbort.getHttpStatus()), percent));	1✔
171	case GRPC_STATUS:
172	return ConfigOrError.fromConfig(FaultAbort.forStatus(	1✔
173	Status.fromCodeValue(faultAbort.getGrpcStatus()), percent));	1✔
174	case ERRORTYPE_NOT_SET:
175	default:
176	return ConfigOrError.fromError(	×
177	"Unknown error type case: " + faultAbort.getErrorTypeCase());	×
178	}
179	}
180
181	private static FaultConfig.FractionalPercent parsePercent(FractionalPercent proto) {
182	switch (proto.getDenominator()) {	1✔
183	case HUNDRED:
184	return FaultConfig.FractionalPercent.perHundred(proto.getNumerator());	1✔
185	case TEN_THOUSAND:
186	return FaultConfig.FractionalPercent.perTenThousand(proto.getNumerator());	1✔
187	case MILLION:
188	return FaultConfig.FractionalPercent.perMillion(proto.getNumerator());	1✔
189	case UNRECOGNIZED:
190	default:
191	throw new IllegalArgumentException("Unknown denominator type: " + proto.getDenominator());	×
192	}
193	}
194	}
195
196	@Nullable
197	@Override
198	public ClientInterceptor buildClientInterceptor(
199	FilterConfig config, @Nullable FilterConfig overrideConfig,
200	final ScheduledExecutorService scheduler) {
201	checkNotNull(config, "config");	1✔
202	if (overrideConfig != null) {	1✔
203	config = overrideConfig;	1✔
204	}
205	FaultConfig faultConfig = (FaultConfig) config;	1✔
206
207	final class FaultInjectionInterceptor implements ClientInterceptor {	1✔
208	@Override
209	public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
210	final MethodDescriptor<ReqT, RespT> method, final CallOptions callOptions,
211	final Channel next) {
212	boolean checkFault = false;	1✔
213	if (faultConfig.maxActiveFaults() == null	1✔
214	\|\| activeFaultCounter.get() < faultConfig.maxActiveFaults()) {	1✔
215	checkFault = faultConfig.faultDelay() != null \|\| faultConfig.faultAbort() != null;	1✔
216	}
217	if (!checkFault) {	1✔
218	return next.newCall(method, callOptions);	1✔
219	}
220	final class DeadlineInsightForwardingCall extends ForwardingClientCall<ReqT, RespT> {	1✔
221	private ClientCall<ReqT, RespT> delegate;
222
223	@Override
224	protected ClientCall<ReqT, RespT> delegate() {
225	return delegate;	1✔
226	}
227
228	@Override
229	public void start(Listener<RespT> listener, Metadata headers) {
230	Executor callExecutor = callOptions.getExecutor();	1✔
231	if (callExecutor == null) { // This should never happen in practice because	1✔
232	// ManagedChannelImpl.ConfigSelectingClientCall always provides CallOptions with
233	// a callExecutor.
234	// TODO(https://github.com/grpc/grpc-java/issues/7868)
235	callExecutor = MoreExecutors.directExecutor();	1✔
236	}
237
238	Long delayNanos;
239	Status abortStatus = null;	1✔
240	if (faultConfig.faultDelay() != null) {	1✔
241	delayNanos = determineFaultDelayNanos(faultConfig.faultDelay(), headers);	1✔
242	} else {
243	delayNanos = null;	1✔
244	}
245	if (faultConfig.faultAbort() != null) {	1✔
246	abortStatus = getAbortStatusWithDescription(	1✔
247	determineFaultAbortStatus(faultConfig.faultAbort(), headers));	1✔
248	}
249
250	Supplier<? extends ClientCall<ReqT, RespT>> callSupplier;
251	if (abortStatus != null) {	1✔
252	callSupplier = Suppliers.ofInstance(	1✔
253	new FailingClientCall<ReqT, RespT>(abortStatus, callExecutor));
254	} else {
255	callSupplier = new Supplier<ClientCall<ReqT, RespT>>() {	1✔
256	@Override
257	public ClientCall<ReqT, RespT> get() {
258	return next.newCall(method, callOptions);	1✔
259	}
260	};
261	}
262	if (delayNanos == null) {	1✔
263	delegate = callSupplier.get();	1✔
264	delegate().start(listener, headers);	1✔
265	return;	1✔
266	}
267
268	delegate = new DelayInjectedCall<>(	1✔
269	delayNanos, callExecutor, scheduler, callOptions.getDeadline(), callSupplier);	1✔
270
271	Listener<RespT> finalListener =	1✔
272	new SimpleForwardingClientCallListener<RespT>(listener) {	1✔
273	@Override
274	public void onClose(Status status, Metadata trailers) {
275	if (status.getCode().equals(Code.DEADLINE_EXCEEDED)) {	1✔
276	// TODO(zdapeng:) check effective deadline locally, and
277	// do the following only if the local deadline is exceeded.
278	// (If the server sends DEADLINE_EXCEEDED for its own deadline, then the
279	// injected delay does not contribute to the error, because the request is
280	// only sent out after the delay. There could be a race between local and
281	// remote, but it is rather rare.)
282	String description = String.format(	1✔
283	Locale.US,
284	"Deadline exceeded after up to %d ns of fault-injected delay",
285	delayNanos);
286	if (status.getDescription() != null) {	1✔
287	description = description + ": " + status.getDescription();	1✔
288	}
289	status = Status.DEADLINE_EXCEEDED	1✔
290	.withDescription(description).withCause(status.getCause());	1✔
291	// Replace trailers to prevent mixing sources of status and trailers.
292	trailers = new Metadata();	1✔
293	}
294	delegate().onClose(status, trailers);	1✔
295	}	1✔
296	};
297	delegate().start(finalListener, headers);	1✔
298	}	1✔
299	}
300
301	return new DeadlineInsightForwardingCall();	1✔
302	}
303	}
304
305	return new FaultInjectionInterceptor();	1✔
306	}
307
308	private static Status getAbortStatusWithDescription(Status abortStatus) {
309	Status finalAbortStatus = null;	1✔
310	if (abortStatus != null) {	1✔
311	String abortDesc = "RPC terminated due to fault injection";	1✔
312	if (abortStatus.getDescription() != null) {	1✔
313	abortDesc = abortDesc + ": " + abortStatus.getDescription();	1✔
314	}
315	finalAbortStatus = abortStatus.withDescription(abortDesc);	1✔
316	}
317	return finalAbortStatus;	1✔
318	}
319
320	@Nullable
321	private Long determineFaultDelayNanos(FaultDelay faultDelay, Metadata headers) {
322	Long delayNanos;
323	FaultConfig.FractionalPercent fractionalPercent = faultDelay.percent();	1✔
324	if (faultDelay.headerDelay()) {	1✔
325	try {
326	int delayMillis = Integer.parseInt(headers.get(HEADER_DELAY_KEY));	1✔
327	delayNanos = TimeUnit.MILLISECONDS.toNanos(delayMillis);	1✔
328	String delayPercentageStr = headers.get(HEADER_DELAY_PERCENTAGE_KEY);	1✔
329	if (delayPercentageStr != null) {	1✔
330	int delayPercentage = Integer.parseInt(delayPercentageStr);	1✔
331	if (delayPercentage >= 0 && delayPercentage < fractionalPercent.numerator()) {	1✔
332	fractionalPercent = FaultConfig.FractionalPercent.create(	1✔
333	delayPercentage, fractionalPercent.denominatorType());	1✔
334	}
335	}
336	} catch (NumberFormatException e) {	1✔
337	return null; // treated as header_delay not applicable	1✔
338	}	1✔
339	} else {
340	delayNanos = faultDelay.delayNanos();	1✔
341	}
342	if (random.nextInt(1_000_000) >= getRatePerMillion(fractionalPercent)) {	1✔
343	return null;	1✔
344	}
345	return delayNanos;	1✔
346	}
347
348	@Nullable
349	private Status determineFaultAbortStatus(FaultAbort faultAbort, Metadata headers) {
350	Status abortStatus = null;	1✔
351	FaultConfig.FractionalPercent fractionalPercent = faultAbort.percent();	1✔
352	if (faultAbort.headerAbort()) {	1✔
353	try {
354	String grpcCodeStr = headers.get(HEADER_ABORT_GRPC_STATUS_KEY);	1✔
355	if (grpcCodeStr != null) {	1✔
356	int grpcCode = Integer.parseInt(grpcCodeStr);	1✔
357	abortStatus = Status.fromCodeValue(grpcCode);	1✔
358	}
359	String httpCodeStr = headers.get(HEADER_ABORT_HTTP_STATUS_KEY);	1✔
360	if (httpCodeStr != null) {	1✔
361	int httpCode = Integer.parseInt(httpCodeStr);	1✔
362	abortStatus = GrpcUtil.httpStatusToGrpcStatus(httpCode);	1✔
363	}
364	String abortPercentageStr = headers.get(HEADER_ABORT_PERCENTAGE_KEY);	1✔
365	if (abortPercentageStr != null) {	1✔
366	int abortPercentage =	1✔
367	Integer.parseInt(headers.get(HEADER_ABORT_PERCENTAGE_KEY));	1✔
368	if (abortPercentage >= 0 && abortPercentage < fractionalPercent.numerator()) {	1✔
369	fractionalPercent = FaultConfig.FractionalPercent.create(	1✔
370	abortPercentage, fractionalPercent.denominatorType());	1✔
371	}
372	}
373	} catch (NumberFormatException e) {	×
374	return null; // treated as header_abort not applicable	×
375	}	1✔
376	} else {
377	abortStatus = faultAbort.status();	1✔
378	}
379	if (random.nextInt(1_000_000) >= getRatePerMillion(fractionalPercent)) {	1✔
380	return null;	1✔
381	}
382	return abortStatus;	1✔
383	}
384
385	private static int getRatePerMillion(FaultConfig.FractionalPercent percent) {
386	int numerator = percent.numerator();	1✔
387	FaultConfig.FractionalPercent.DenominatorType type = percent.denominatorType();	1✔
388	switch (type) {	1✔
389	case TEN_THOUSAND:
390	numerator *= 100;	×
391	break;	×
392	case HUNDRED:
393	numerator *= 10_000;	1✔
394	break;	1✔
395	case MILLION:
396	default:
397	break;
398	}
399	if (numerator > 1_000_000 \|\| numerator < 0) {	1✔
400	numerator = 1_000_000;	×
401	}
402	return numerator;	1✔
403	}
404
405	/** A {@link DelayedClientCall} with a fixed delay. */
406	private final class DelayInjectedCall<ReqT, RespT> extends DelayedClientCall<ReqT, RespT> {
407	final Object lock = new Object();	1✔
408	ScheduledFuture<?> delayTask;
409	boolean cancelled;
410
411	DelayInjectedCall(
412	long delayNanos, Executor callExecutor, ScheduledExecutorService scheduler,
413	@Nullable Deadline deadline,
414	final Supplier<? extends ClientCall<ReqT, RespT>> callSupplier) {	1✔
415	super(callExecutor, scheduler, deadline);	1✔
416	activeFaultCounter.incrementAndGet();	1✔
417	ScheduledFuture<?> task = scheduler.schedule(	1✔
418	new Runnable() {	1✔
419	@Override
420	public void run() {
421	synchronized (lock) {	1✔
422	if (!cancelled) {	1✔
423	activeFaultCounter.decrementAndGet();	1✔
424	}
425	}	1✔
426	Runnable toRun = setCall(callSupplier.get());	1✔
427	if (toRun != null) {	1✔
428	toRun.run();	1✔
429	}
430	}	1✔
431	},
432	delayNanos,
433	NANOSECONDS);
434	synchronized (lock) {	1✔
435	if (!cancelled) {	1✔
436	delayTask = task;	1✔
437	return;	1✔
438	}
439	}	×
440	task.cancel(false);	×
441	}	×
442
443	@Override
444	protected void callCancelled() {
445	ScheduledFuture<?> savedDelayTask;
446	synchronized (lock) {	1✔
447	cancelled = true;	1✔
448	activeFaultCounter.decrementAndGet();	1✔
449	savedDelayTask = delayTask;	1✔
450	}	1✔
451	if (savedDelayTask != null) {	1✔
452	savedDelayTask.cancel(false);	1✔
453	}
454	}	1✔
455	}
456
457	/** An implementation of {@link ClientCall} that fails when started. */
458	private final class FailingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
459	final Status error;
460	final Executor callExecutor;
461	final Context context;
462
463	FailingClientCall(Status error, Executor callExecutor) {	1✔
464	this.error = error;	1✔
465	this.callExecutor = callExecutor;	1✔
466	this.context = Context.current();	1✔
467	}	1✔
468
469	@Override
470	public void start(final ClientCall.Listener<RespT> listener, Metadata headers) {
471	activeFaultCounter.incrementAndGet();	1✔
472	callExecutor.execute(	1✔
473	new Runnable() {	1✔
474	@Override
475	public void run() {
476	Context previous = context.attach();	1✔
477	try {
478	listener.onClose(error, new Metadata());	1✔
479	activeFaultCounter.decrementAndGet();	1✔
480	} finally {
481	context.detach(previous);	1✔
482	}
483	}	1✔
484	});
485	}	1✔
486
487	@Override
488	public void request(int numMessages) {}	×
489
490	@Override
491	public void cancel(String message, Throwable cause) {}	×
492
493	@Override
494	public void halfClose() {}	×
495
496	@Override
497	public void sendMessage(ReqT message) {}	×
498	}
499	}

grpc / grpc-java / #20276

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous