#20338

Committed 26 Jun 2026 11:37AM UTC coverage: 89.043% (+0.2%) from 88.876%

Build # #20338

Build Type

push

github

Committed by

web-flow

Commit Message

Ext_proc filter and client interceptor (#12792)

Implements ext_proc filter from [grfc A93](https://github.com/grpc/proposal/pull/484)  (internal [design doc](http://go/ext-proc-design-java)).

Coverage Stats

37578 of 42202 relevant lines covered (89.04%)

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

grpc / grpc-java / #20338

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