#19702

/*
 * 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() {
      return INSTANCE;
    }

    @Override
    public ConfigOrError<FaultConfig> parseFilterConfig(Message rawProtoMessage) {
      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) {
      return parseFilterConfig(rawProtoMessage);
    }

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

grpc / grpc-java / #19702

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