#18957

Committed 19 Dec 2023 05:39PM UTC coverage: 88.134% (-0.003%) from 88.137%

Build # #18957

Build Type

push

github

Committed by

web-flow

Commit Message

buildscripts: Use the Kokoro shared install lib from the new repo (#10757) (#10763)

Source: https://github.com/grpc/grpc/blob/4f7ead2c7/tools/internal_ci/linux/grpc_xds_k8s_install_test_driver.sh
New repo: https://github.com/grpc/psm-interop
New path: `.kokoro/psm_interop_kokoro_lib.sh`

Run Details

30832 of 34983 relevant lines covered (88.13%)

0.88 hits per line

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

96.53

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

/*
 * Copyright 2019 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.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Stopwatch;
import com.google.common.base.Supplier;
import com.google.protobuf.util.Durations;
import io.envoyproxy.envoy.service.load_stats.v3.LoadReportingServiceGrpc;
import io.envoyproxy.envoy.service.load_stats.v3.LoadStatsRequest;
import io.envoyproxy.envoy.service.load_stats.v3.LoadStatsResponse;
import io.grpc.Channel;
import io.grpc.Context;
import io.grpc.InternalLogId;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.internal.BackoffPolicy;
import io.grpc.stub.StreamObserver;
import io.grpc.xds.EnvoyProtoData.Node;
import io.grpc.xds.Stats.ClusterStats;
import io.grpc.xds.Stats.DroppedRequests;
import io.grpc.xds.Stats.UpstreamLocalityStats;
import io.grpc.xds.XdsLogger.XdsLogLevel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;

/**
 * Client of xDS load reporting service based on LRS protocol, which reports load stats of
 * gRPC client's perspective to a management server.
 */
final class LoadReportClient {
  private final InternalLogId logId;
  private final XdsLogger logger;
  private final Channel channel;
  private final Context context;
  private final Node node;
  private final SynchronizationContext syncContext;
  private final ScheduledExecutorService timerService;
  private final Stopwatch retryStopwatch;
  private final BackoffPolicy.Provider backoffPolicyProvider;
  @VisibleForTesting
  final LoadStatsManager2 loadStatsManager;

  private boolean started;
  @Nullable
  private BackoffPolicy lrsRpcRetryPolicy;
  @Nullable
  private ScheduledHandle lrsRpcRetryTimer;
  @Nullable
  @VisibleForTesting
  LrsStream lrsStream;

  LoadReportClient(
      LoadStatsManager2 loadStatsManager,
      Channel channel,
      Context context,
      Node node,
      SynchronizationContext syncContext,
      ScheduledExecutorService scheduledExecutorService,
      BackoffPolicy.Provider backoffPolicyProvider,
      Supplier<Stopwatch> stopwatchSupplier) {
    this.loadStatsManager = checkNotNull(loadStatsManager, "loadStatsManager");
    this.channel = checkNotNull(channel, "xdsChannel");
    this.context = checkNotNull(context, "context");
    this.syncContext = checkNotNull(syncContext, "syncContext");
    this.timerService = checkNotNull(scheduledExecutorService, "timeService");
    this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
    this.retryStopwatch = checkNotNull(stopwatchSupplier, "stopwatchSupplier").get();
    this.node = checkNotNull(node, "node").toBuilder()
        .addClientFeatures("envoy.lrs.supports_send_all_clusters").build();
    logId = InternalLogId.allocate("lrs-client", null);
    logger = XdsLogger.withLogId(logId);
    logger.log(XdsLogLevel.INFO, "Created");
  }

  /**
   * Establishes load reporting communication and negotiates with traffic director to report load
   * stats periodically. Calling this method on an already started {@link LoadReportClient} is
   * no-op.
   */
  void startLoadReporting() {
    syncContext.throwIfNotInThisSynchronizationContext();
    if (started) {
      return;
    }
    started = true;
    logger.log(XdsLogLevel.INFO, "Starting load reporting RPC");
    startLrsRpc();
  }

  /**
   * Terminates load reporting. Calling this method on an already stopped
   * {@link LoadReportClient} is no-op.
   */
  void stopLoadReporting() {
    syncContext.throwIfNotInThisSynchronizationContext();
    if (!started) {
      return;
    }
    started = false;
    logger.log(XdsLogLevel.INFO, "Stopping load reporting RPC");
    if (lrsRpcRetryTimer != null && lrsRpcRetryTimer.isPending()) {
      lrsRpcRetryTimer.cancel();
    }
    if (lrsStream != null) {
      lrsStream.close(Status.CANCELLED.withDescription("stop load reporting").asException());
    }
    // Do not shutdown channel as it is not owned by LrsClient.
  }

  @VisibleForTesting
  static class LoadReportingTask implements Runnable {
    private final LrsStream stream;

    LoadReportingTask(LrsStream stream) {
      this.stream = stream;
    }

    @Override
    public void run() {
      stream.sendLoadReport();
    }
  }

  @VisibleForTesting
  class LrsRpcRetryTask implements Runnable {

    @Override
    public void run() {
      startLrsRpc();
    }
  }

  private void startLrsRpc() {
    if (!started) {
      return;
    }
    checkState(lrsStream == null, "previous lbStream has not been cleared yet");
    lrsStream = new LrsStream();
    retryStopwatch.reset().start();
    Context prevContext = context.attach();
    try {
      lrsStream.start();
    } finally {
      context.detach(prevContext);
    }
  }

  private final class LrsStream {
    boolean initialResponseReceived;
    boolean closed;
    long intervalNano = -1;
    boolean reportAllClusters;
    List<String> clusterNames;  // clusters to report loads for, if not report all.
    ScheduledHandle loadReportTimer;
    StreamObserver<LoadStatsRequest> lrsRequestWriterV3;

    void start() {
      StreamObserver<LoadStatsResponse> lrsResponseReaderV3 =
          new StreamObserver<LoadStatsResponse>() {
            @Override
            public void onNext(final LoadStatsResponse response) {
              syncContext.execute(new Runnable() {
                @Override
                public void run() {
                  logger.log(XdsLogLevel.DEBUG, "Received LRS response:\n{0}", response);
                  handleRpcResponse(response.getClustersList(), response.getSendAllClusters(),
                      Durations.toNanos(response.getLoadReportingInterval()));
                }
              });
            }

            @Override
            public void onError(final Throwable t) {
              syncContext.execute(new Runnable() {
                @Override
                public void run() {
                  handleRpcError(t);
                }
              });
            }

            @Override
            public void onCompleted() {
              syncContext.execute(new Runnable() {
                @Override
                public void run() {
                  handleRpcCompleted();
                }
              });
            }
          };
      lrsRequestWriterV3 = LoadReportingServiceGrpc.newStub(channel).withWaitForReady()
          .streamLoadStats(lrsResponseReaderV3);
      logger.log(XdsLogLevel.DEBUG, "Sending initial LRS request");
      sendLoadStatsRequest(Collections.<ClusterStats>emptyList());
    }

    void sendLoadStatsRequest(List<ClusterStats> clusterStatsList) {
      LoadStatsRequest.Builder requestBuilder =
          LoadStatsRequest.newBuilder().setNode(node.toEnvoyProtoNode());
      for (ClusterStats stats : clusterStatsList) {
        requestBuilder.addClusterStats(buildClusterStats(stats));
      }
      LoadStatsRequest request = requestBuilder.build();
      lrsRequestWriterV3.onNext(request);
      logger.log(XdsLogLevel.DEBUG, "Sent LoadStatsRequest\n{0}", request);
    }

    void sendError(Exception error) {
      lrsRequestWriterV3.onError(error);
    }

    void handleRpcResponse(List<String> clusters, boolean sendAllClusters,
                           long loadReportIntervalNano) {
      if (closed) {
        return;
      }
      if (!initialResponseReceived) {
        logger.log(XdsLogLevel.DEBUG, "Initial LRS response received");
        initialResponseReceived = true;
      }
      reportAllClusters = sendAllClusters;
      if (reportAllClusters) {
        logger.log(XdsLogLevel.INFO, "Report loads for all clusters");
      } else {
        logger.log(XdsLogLevel.INFO, "Report loads for clusters: ", clusters);
        clusterNames = clusters;
      }
      intervalNano = loadReportIntervalNano;
      logger.log(XdsLogLevel.INFO, "Update load reporting interval to {0} ns", intervalNano);
      scheduleNextLoadReport();
    }

    void handleRpcError(Throwable t) {
      handleStreamClosed(Status.fromThrowable(t));
    }

    void handleRpcCompleted() {
      handleStreamClosed(Status.UNAVAILABLE.withDescription("Closed by server"));
    }

    private void sendLoadReport() {
      if (closed) {
        return;
      }
      List<ClusterStats> clusterStatsList;
      if (reportAllClusters) {
        clusterStatsList = loadStatsManager.getAllClusterStatsReports();
      } else {
        clusterStatsList = new ArrayList<>();
        for (String name : clusterNames) {
          clusterStatsList.addAll(loadStatsManager.getClusterStatsReports(name));
        }
      }
      sendLoadStatsRequest(clusterStatsList);
      scheduleNextLoadReport();
    }

    private void scheduleNextLoadReport() {
      // Cancel pending load report and reschedule with updated load reporting interval.
      if (loadReportTimer != null && loadReportTimer.isPending()) {
        loadReportTimer.cancel();
        loadReportTimer = null;
      }
      if (intervalNano > 0) {
        loadReportTimer = syncContext.schedule(
            new LoadReportingTask(this), intervalNano, TimeUnit.NANOSECONDS, timerService);
      }
    }

    private void handleStreamClosed(Status status) {
      checkArgument(!status.isOk(), "unexpected OK status");
      if (closed) {
        return;
      }
      logger.log(
          XdsLogLevel.ERROR,
          "LRS stream closed with status {0}: {1}. Cause: {2}",
          status.getCode(), status.getDescription(), status.getCause());
      closed = true;
      cleanUp();

      if (initialResponseReceived || lrsRpcRetryPolicy == null) {
        // Reset the backoff sequence if balancer has sent the initial response, or backoff sequence
        // has never been initialized.
        lrsRpcRetryPolicy = backoffPolicyProvider.get();
      }
      // The back-off policy determines the interval between consecutive RPC upstarts, thus the
      // actual delay may be smaller than the value from the back-off policy, or even negative,
      // depending how much time was spent in the previous RPC.
      long delayNanos =
          lrsRpcRetryPolicy.nextBackoffNanos() - retryStopwatch.elapsed(TimeUnit.NANOSECONDS);
      logger.log(XdsLogLevel.INFO, "Retry LRS stream in {0} ns", delayNanos);
      if (delayNanos <= 0) {
        startLrsRpc();
      } else {
        lrsRpcRetryTimer = syncContext.schedule(
            new LrsRpcRetryTask(), delayNanos, TimeUnit.NANOSECONDS, timerService);
      }
    }

    private void close(Exception error) {
      if (closed) {
        return;
      }
      closed = true;
      cleanUp();
      sendError(error);
    }

    private void cleanUp() {
      if (loadReportTimer != null && loadReportTimer.isPending()) {
        loadReportTimer.cancel();
        loadReportTimer = null;
      }
      if (lrsStream == this) {
        lrsStream = null;
      }
    }

    private io.envoyproxy.envoy.config.endpoint.v3.ClusterStats buildClusterStats(
        ClusterStats stats) {
      io.envoyproxy.envoy.config.endpoint.v3.ClusterStats.Builder builder =
          io.envoyproxy.envoy.config.endpoint.v3.ClusterStats.newBuilder()
              .setClusterName(stats.clusterName());
      if (stats.clusterServiceName() != null) {
        builder.setClusterServiceName(stats.clusterServiceName());
      }
      for (UpstreamLocalityStats upstreamLocalityStats : stats.upstreamLocalityStatsList()) {
        builder.addUpstreamLocalityStats(
            io.envoyproxy.envoy.config.endpoint.v3.UpstreamLocalityStats.newBuilder()
                .setLocality(
                    io.envoyproxy.envoy.config.core.v3.Locality.newBuilder()
                        .setRegion(upstreamLocalityStats.locality().region())
                        .setZone(upstreamLocalityStats.locality().zone())
                        .setSubZone(upstreamLocalityStats.locality().subZone()))
            .setTotalSuccessfulRequests(upstreamLocalityStats.totalSuccessfulRequests())
            .setTotalErrorRequests(upstreamLocalityStats.totalErrorRequests())
            .setTotalRequestsInProgress(upstreamLocalityStats.totalRequestsInProgress())
            .setTotalIssuedRequests(upstreamLocalityStats.totalIssuedRequests()));
      }
      for (DroppedRequests droppedRequests : stats.droppedRequestsList()) {
        builder.addDroppedRequests(
            io.envoyproxy.envoy.config.endpoint.v3.ClusterStats.DroppedRequests.newBuilder()
                .setCategory(droppedRequests.category())
                .setDroppedCount(droppedRequests.droppedCount()));
      }
      return builder
          .setTotalDroppedRequests(stats.totalDroppedRequests())
          .setLoadReportInterval(Durations.fromNanos(stats.loadReportIntervalNano()))
          .build();
    }
  }
}

1	/*
2	* Copyright 2019 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.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static com.google.common.base.Preconditions.checkState;
22
23	import com.google.common.annotations.VisibleForTesting;
24	import com.google.common.base.Stopwatch;
25	import com.google.common.base.Supplier;
26	import com.google.protobuf.util.Durations;
27	import io.envoyproxy.envoy.service.load_stats.v3.LoadReportingServiceGrpc;
28	import io.envoyproxy.envoy.service.load_stats.v3.LoadStatsRequest;
29	import io.envoyproxy.envoy.service.load_stats.v3.LoadStatsResponse;
30	import io.grpc.Channel;
31	import io.grpc.Context;
32	import io.grpc.InternalLogId;
33	import io.grpc.Status;
34	import io.grpc.SynchronizationContext;
35	import io.grpc.SynchronizationContext.ScheduledHandle;
36	import io.grpc.internal.BackoffPolicy;
37	import io.grpc.stub.StreamObserver;
38	import io.grpc.xds.EnvoyProtoData.Node;
39	import io.grpc.xds.Stats.ClusterStats;
40	import io.grpc.xds.Stats.DroppedRequests;
41	import io.grpc.xds.Stats.UpstreamLocalityStats;
42	import io.grpc.xds.XdsLogger.XdsLogLevel;
43	import java.util.ArrayList;
44	import java.util.Collections;
45	import java.util.List;
46	import java.util.concurrent.ScheduledExecutorService;
47	import java.util.concurrent.TimeUnit;
48	import javax.annotation.Nullable;
49
50	/**
51	* Client of xDS load reporting service based on LRS protocol, which reports load stats of
52	* gRPC client's perspective to a management server.
53	*/
54	final class LoadReportClient {
55	private final InternalLogId logId;
56	private final XdsLogger logger;
57	private final Channel channel;
58	private final Context context;
59	private final Node node;
60	private final SynchronizationContext syncContext;
61	private final ScheduledExecutorService timerService;
62	private final Stopwatch retryStopwatch;
63	private final BackoffPolicy.Provider backoffPolicyProvider;
64	@VisibleForTesting
65	final LoadStatsManager2 loadStatsManager;
66
67	private boolean started;
68	@Nullable
69	private BackoffPolicy lrsRpcRetryPolicy;
70	@Nullable
71	private ScheduledHandle lrsRpcRetryTimer;
72	@Nullable
73	@VisibleForTesting
74	LrsStream lrsStream;
75
76	LoadReportClient(
77	LoadStatsManager2 loadStatsManager,
78	Channel channel,
79	Context context,
80	Node node,
81	SynchronizationContext syncContext,
82	ScheduledExecutorService scheduledExecutorService,
83	BackoffPolicy.Provider backoffPolicyProvider,
84	Supplier<Stopwatch> stopwatchSupplier) {	1✔
85	this.loadStatsManager = checkNotNull(loadStatsManager, "loadStatsManager");	1✔
86	this.channel = checkNotNull(channel, "xdsChannel");	1✔
87	this.context = checkNotNull(context, "context");	1✔
88	this.syncContext = checkNotNull(syncContext, "syncContext");	1✔
89	this.timerService = checkNotNull(scheduledExecutorService, "timeService");	1✔
90	this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");	1✔
91	this.retryStopwatch = checkNotNull(stopwatchSupplier, "stopwatchSupplier").get();	1✔
92	this.node = checkNotNull(node, "node").toBuilder()	1✔
93	.addClientFeatures("envoy.lrs.supports_send_all_clusters").build();	1✔
94	logId = InternalLogId.allocate("lrs-client", null);	1✔
95	logger = XdsLogger.withLogId(logId);	1✔
96	logger.log(XdsLogLevel.INFO, "Created");	1✔
97	}	1✔
98
99	/**
100	* Establishes load reporting communication and negotiates with traffic director to report load
101	* stats periodically. Calling this method on an already started {@link LoadReportClient} is
102	* no-op.
103	*/
104	void startLoadReporting() {
105	syncContext.throwIfNotInThisSynchronizationContext();	1✔
106	if (started) {	1✔
107	return;	×
108	}
109	started = true;	1✔
110	logger.log(XdsLogLevel.INFO, "Starting load reporting RPC");	1✔
111	startLrsRpc();	1✔
112	}	1✔
113
114	/**
115	* Terminates load reporting. Calling this method on an already stopped
116	* {@link LoadReportClient} is no-op.
117	*/
118	void stopLoadReporting() {
119	syncContext.throwIfNotInThisSynchronizationContext();	1✔
120	if (!started) {	1✔
121	return;	1✔
122	}
123	started = false;	1✔
124	logger.log(XdsLogLevel.INFO, "Stopping load reporting RPC");	1✔
125	if (lrsRpcRetryTimer != null && lrsRpcRetryTimer.isPending()) {	1✔
126	lrsRpcRetryTimer.cancel();	1✔
127	}
128	if (lrsStream != null) {	1✔
129	lrsStream.close(Status.CANCELLED.withDescription("stop load reporting").asException());	1✔
130	}
131	// Do not shutdown channel as it is not owned by LrsClient.
132	}	1✔
133
134	@VisibleForTesting
135	static class LoadReportingTask implements Runnable {
136	private final LrsStream stream;
137
138	LoadReportingTask(LrsStream stream) {	1✔
139	this.stream = stream;	1✔
140	}	1✔
141
142	@Override
143	public void run() {
144	stream.sendLoadReport();	1✔
145	}	1✔
146	}
147
148	@VisibleForTesting
149	class LrsRpcRetryTask implements Runnable {	1✔
150
151	@Override
152	public void run() {
153	startLrsRpc();	1✔
154	}	1✔
155	}
156
157	private void startLrsRpc() {
158	if (!started) {	1✔
159	return;	×
160	}
161	checkState(lrsStream == null, "previous lbStream has not been cleared yet");	1✔
162	lrsStream = new LrsStream();	1✔
163	retryStopwatch.reset().start();	1✔
164	Context prevContext = context.attach();	1✔
165	try {
166	lrsStream.start();	1✔
167	} finally {
168	context.detach(prevContext);	1✔
169	}
170	}	1✔
171
172	private final class LrsStream {	1✔
173	boolean initialResponseReceived;
174	boolean closed;
175	long intervalNano = -1;	1✔
176	boolean reportAllClusters;
177	List<String> clusterNames; // clusters to report loads for, if not report all.
178	ScheduledHandle loadReportTimer;
179	StreamObserver<LoadStatsRequest> lrsRequestWriterV3;
180
181	void start() {
182	StreamObserver<LoadStatsResponse> lrsResponseReaderV3 =	1✔
183	new StreamObserver<LoadStatsResponse>() {	1✔
184	@Override
185	public void onNext(final LoadStatsResponse response) {
186	syncContext.execute(new Runnable() {	1✔
187	@Override
188	public void run() {
189	logger.log(XdsLogLevel.DEBUG, "Received LRS response:\n{0}", response);	1✔
190	handleRpcResponse(response.getClustersList(), response.getSendAllClusters(),	1✔
191	Durations.toNanos(response.getLoadReportingInterval()));	1✔
192	}	1✔
193	});
194	}	1✔
195
196	@Override
197	public void onError(final Throwable t) {
198	syncContext.execute(new Runnable() {	1✔
199	@Override
200	public void run() {
201	handleRpcError(t);	1✔
202	}	1✔
203	});
204	}	1✔
205
206	@Override
207	public void onCompleted() {
208	syncContext.execute(new Runnable() {	1✔
209	@Override
210	public void run() {
211	handleRpcCompleted();	1✔
212	}	1✔
213	});
214	}	1✔
215	};
216	lrsRequestWriterV3 = LoadReportingServiceGrpc.newStub(channel).withWaitForReady()	1✔
217	.streamLoadStats(lrsResponseReaderV3);	1✔
218	logger.log(XdsLogLevel.DEBUG, "Sending initial LRS request");	1✔
219	sendLoadStatsRequest(Collections.<ClusterStats>emptyList());	1✔
220	}	1✔
221
222	void sendLoadStatsRequest(List<ClusterStats> clusterStatsList) {
223	LoadStatsRequest.Builder requestBuilder =
224	LoadStatsRequest.newBuilder().setNode(node.toEnvoyProtoNode());	1✔
225	for (ClusterStats stats : clusterStatsList) {	1✔
226	requestBuilder.addClusterStats(buildClusterStats(stats));	1✔
227	}	1✔
228	LoadStatsRequest request = requestBuilder.build();	1✔
229	lrsRequestWriterV3.onNext(request);	1✔
230	logger.log(XdsLogLevel.DEBUG, "Sent LoadStatsRequest\n{0}", request);	1✔
231	}	1✔
232
233	void sendError(Exception error) {
234	lrsRequestWriterV3.onError(error);	1✔
235	}	1✔
236
237	void handleRpcResponse(List<String> clusters, boolean sendAllClusters,
238	long loadReportIntervalNano) {
239	if (closed) {	1✔
240	return;	×
241	}
242	if (!initialResponseReceived) {	1✔
243	logger.log(XdsLogLevel.DEBUG, "Initial LRS response received");	1✔
244	initialResponseReceived = true;	1✔
245	}
246	reportAllClusters = sendAllClusters;	1✔
247	if (reportAllClusters) {	1✔
248	logger.log(XdsLogLevel.INFO, "Report loads for all clusters");	1✔
249	} else {
250	logger.log(XdsLogLevel.INFO, "Report loads for clusters: ", clusters);	1✔
251	clusterNames = clusters;	1✔
252	}
253	intervalNano = loadReportIntervalNano;	1✔
254	logger.log(XdsLogLevel.INFO, "Update load reporting interval to {0} ns", intervalNano);	1✔
255	scheduleNextLoadReport();	1✔
256	}	1✔
257
258	void handleRpcError(Throwable t) {
259	handleStreamClosed(Status.fromThrowable(t));	1✔
260	}	1✔
261
262	void handleRpcCompleted() {
263	handleStreamClosed(Status.UNAVAILABLE.withDescription("Closed by server"));	1✔
264	}	1✔
265
266	private void sendLoadReport() {
267	if (closed) {	1✔
268	return;	×
269	}
270	List<ClusterStats> clusterStatsList;
271	if (reportAllClusters) {	1✔
272	clusterStatsList = loadStatsManager.getAllClusterStatsReports();	1✔
273	} else {
274	clusterStatsList = new ArrayList<>();	1✔
275	for (String name : clusterNames) {	1✔
276	clusterStatsList.addAll(loadStatsManager.getClusterStatsReports(name));	1✔
277	}	1✔
278	}
279	sendLoadStatsRequest(clusterStatsList);	1✔
280	scheduleNextLoadReport();	1✔
281	}	1✔
282
283	private void scheduleNextLoadReport() {
284	// Cancel pending load report and reschedule with updated load reporting interval.
285	if (loadReportTimer != null && loadReportTimer.isPending()) {	1✔
286	loadReportTimer.cancel();	1✔
287	loadReportTimer = null;	1✔
288	}
289	if (intervalNano > 0) {	1✔
290	loadReportTimer = syncContext.schedule(	1✔
291	new LoadReportingTask(this), intervalNano, TimeUnit.NANOSECONDS, timerService);	1✔
292	}
293	}	1✔
294
295	private void handleStreamClosed(Status status) {
296	checkArgument(!status.isOk(), "unexpected OK status");	1✔
297	if (closed) {	1✔
298	return;	1✔
299	}
300	logger.log(	1✔
301	XdsLogLevel.ERROR,
302	"LRS stream closed with status {0}: {1}. Cause: {2}",
303	status.getCode(), status.getDescription(), status.getCause());	1✔
304	closed = true;	1✔
305	cleanUp();	1✔
306
307	if (initialResponseReceived \|\| lrsRpcRetryPolicy == null) {	1✔
308	// Reset the backoff sequence if balancer has sent the initial response, or backoff sequence
309	// has never been initialized.
310	lrsRpcRetryPolicy = backoffPolicyProvider.get();	1✔
311	}
312	// The back-off policy determines the interval between consecutive RPC upstarts, thus the
313	// actual delay may be smaller than the value from the back-off policy, or even negative,
314	// depending how much time was spent in the previous RPC.
315	long delayNanos =	1✔
316	lrsRpcRetryPolicy.nextBackoffNanos() - retryStopwatch.elapsed(TimeUnit.NANOSECONDS);	1✔
317	logger.log(XdsLogLevel.INFO, "Retry LRS stream in {0} ns", delayNanos);	1✔
318	if (delayNanos <= 0) {	1✔
319	startLrsRpc();	×
320	} else {
321	lrsRpcRetryTimer = syncContext.schedule(	1✔
322	new LrsRpcRetryTask(), delayNanos, TimeUnit.NANOSECONDS, timerService);	1✔
323	}
324	}	1✔
325
326	private void close(Exception error) {
327	if (closed) {	1✔
328	return;	×
329	}
330	closed = true;	1✔
331	cleanUp();	1✔
332	sendError(error);	1✔
333	}	1✔
334
335	private void cleanUp() {
336	if (loadReportTimer != null && loadReportTimer.isPending()) {	1✔
337	loadReportTimer.cancel();	1✔
338	loadReportTimer = null;	1✔
339	}
340	if (lrsStream == this) {	1✔
341	lrsStream = null;	1✔
342	}
343	}	1✔
344
345	private io.envoyproxy.envoy.config.endpoint.v3.ClusterStats buildClusterStats(
346	ClusterStats stats) {
347	io.envoyproxy.envoy.config.endpoint.v3.ClusterStats.Builder builder =
348	io.envoyproxy.envoy.config.endpoint.v3.ClusterStats.newBuilder()	1✔
349	.setClusterName(stats.clusterName());	1✔
350	if (stats.clusterServiceName() != null) {	1✔
351	builder.setClusterServiceName(stats.clusterServiceName());	1✔
352	}
353	for (UpstreamLocalityStats upstreamLocalityStats : stats.upstreamLocalityStatsList()) {	1✔
354	builder.addUpstreamLocalityStats(	1✔
355	io.envoyproxy.envoy.config.endpoint.v3.UpstreamLocalityStats.newBuilder()	1✔
356	.setLocality(	1✔
357	io.envoyproxy.envoy.config.core.v3.Locality.newBuilder()	1✔
358	.setRegion(upstreamLocalityStats.locality().region())	1✔
359	.setZone(upstreamLocalityStats.locality().zone())	1✔
360	.setSubZone(upstreamLocalityStats.locality().subZone()))	1✔
361	.setTotalSuccessfulRequests(upstreamLocalityStats.totalSuccessfulRequests())	1✔
362	.setTotalErrorRequests(upstreamLocalityStats.totalErrorRequests())	1✔
363	.setTotalRequestsInProgress(upstreamLocalityStats.totalRequestsInProgress())	1✔
364	.setTotalIssuedRequests(upstreamLocalityStats.totalIssuedRequests()));	1✔
365	}	1✔
366	for (DroppedRequests droppedRequests : stats.droppedRequestsList()) {	1✔
367	builder.addDroppedRequests(	1✔
368	io.envoyproxy.envoy.config.endpoint.v3.ClusterStats.DroppedRequests.newBuilder()	1✔
369	.setCategory(droppedRequests.category())	1✔
370	.setDroppedCount(droppedRequests.droppedCount()));	1✔
371	}	1✔
372	return builder	1✔
373	.setTotalDroppedRequests(stats.totalDroppedRequests())	1✔
374	.setLoadReportInterval(Durations.fromNanos(stats.loadReportIntervalNano()))	1✔
375	.build();	1✔
376	}
377	}
378	}

grpc / grpc-java / #18957

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