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

xds: log raw response messages in sync context (#7441)

This commit is contained in:
Chengyuan Zhang 2020-09-22 17:28:45 -07:00 committed by GitHub
parent b434df25cd
commit bc8c758a3c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 164 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
xds/src/main/java/io/grpc/xds/LoadReportClient.java
View File

@ -185,10 +185,8 @@ final class LoadReportClient {
abstract void sendError(Exception error);
final void handleResponse(final LoadStatsResponseData response) {
syncContext.execute(new Runnable() {
@Override
public void run() {
// Must run in syncContext.
final void handleResponse(LoadStatsResponseData response) {
if (closed) {
return;
}
@ -208,26 +206,15 @@ final class LoadReportClient {
loadReportIntervalNano = interval;
scheduleNextLoadReport();
}
});
}
final void handleRpcError(final Throwable t) {
syncContext.execute(new Runnable() {
@Override
public void run() {
// Must run in syncContext.
final void handleRpcError(Throwable t) {
handleStreamClosed(Status.fromThrowable(t));
}
});
}
final void handleRpcComplete() {
syncContext.execute(new Runnable() {
@Override
public void run() {
handleStreamClosed(
Status.UNAVAILABLE.withDescription("Closed by server"));
}
});
// Must run in syncContext.
final void handleRpcCompleted() {
handleStreamClosed(Status.UNAVAILABLE.withDescription("Closed by server"));
}
private void sendLoadReport() {
@ -324,19 +311,34 @@ final class LoadReportClient {
new StreamObserver<io.envoyproxy.envoy.service.load_stats.v2.LoadStatsResponse>() {
@Override
public void onNext(
io.envoyproxy.envoy.service.load_stats.v2.LoadStatsResponse response) {
logger.log(XdsLogLevel.DEBUG, "Received LRS response:\n{0}", response);
final io.envoyproxy.envoy.service.load_stats.v2.LoadStatsResponse response) {
syncContext.execute(new Runnable() {
@Override
public void run() {
logger.log(XdsLogLevel.DEBUG, "Received LoadStatsResponse:\n{0}", response);
handleResponse(LoadStatsResponseData.fromEnvoyProtoV2(response));
}
});
}
@Override
public void onError(Throwable t) {
public void onError(final Throwable t) {
syncContext.execute(new Runnable() {
@Override
public void run() {
handleRpcError(t);
}
});
}
@Override
public void onCompleted() {
handleRpcComplete();
syncContext.execute(new Runnable() {
@Override
public void run() {
handleRpcCompleted();
}
});
}
};
io.envoyproxy.envoy.service.load_stats.v2.LoadReportingServiceGrpc.LoadReportingServiceStub
@ -369,19 +371,34 @@ final class LoadReportClient {
StreamObserver<LoadStatsResponse> lrsResponseReaderV3 =
new StreamObserver<LoadStatsResponse>() {
@Override
public void onNext(LoadStatsResponse response) {
public void onNext(final LoadStatsResponse response) {
syncContext.execute(new Runnable() {
@Override
public void run() {
logger.log(XdsLogLevel.DEBUG, "Received LRS response:\n{0}", response);
handleResponse(LoadStatsResponseData.fromEnvoyProtoV3(response));
}
});
}
@Override
public void onError(Throwable t) {
public void onError(final Throwable t) {
syncContext.execute(new Runnable() {
@Override
public void run() {
handleRpcError(t);
}
});
}
@Override
public void onCompleted() {
handleRpcComplete();
syncContext.execute(new Runnable() {
@Override
public void run() {
handleRpcCompleted();
}
});
}
};
LoadReportingServiceStub stubV3 =

98
xds/src/main/java/io/grpc/xds/XdsClientImpl.java
View File

@ -1455,11 +1455,8 @@ final class XdsClientImpl extends XdsClient {
abstract void sendError(Exception error);
final void onDiscoveryResponse(final DiscoveryResponseData response) {
syncContext.execute(
new Runnable() {
@Override
public void run() {
// Must run in syncContext.
final void handleResponse(DiscoveryResponseData response) {
if (closed) {
return;
}
@ -1497,26 +1494,15 @@ final class XdsClientImpl extends XdsClient {
throw new AssertionError("Missing case in enum switch: " + resourceType);
}
}
});
}
final void onError(final Throwable t) {
syncContext.execute(new Runnable() {
@Override
public void run() {
// Must run in syncContext.
final void handleRpcError(Throwable t) {
handleStreamClosed(Status.fromThrowable(t));
}
});
}
final void onCompleted() {
syncContext.execute(new Runnable() {
@Override
public void run() {
handleStreamClosed(
Status.UNAVAILABLE.withDescription("Closed by server"));
}
});
// Must run in syncContext.
final void handleRpcCompleted() {
handleStreamClosed(Status.UNAVAILABLE.withDescription("Closed by server"));
}
private void handleStreamClosed(Status error) {
@ -1733,27 +1719,40 @@ final class XdsClientImpl extends XdsClient {
StreamObserver<io.envoyproxy.envoy.api.v2.DiscoveryResponse> responseReaderV2 =
new StreamObserver<io.envoyproxy.envoy.api.v2.DiscoveryResponse>() {
@Override
public void onNext(io.envoyproxy.envoy.api.v2.DiscoveryResponse response) {
DiscoveryResponseData responseData =
DiscoveryResponseData.fromEnvoyProtoV2(response);
public void onNext(final io.envoyproxy.envoy.api.v2.DiscoveryResponse response) {
syncContext.execute(new Runnable() {
@Override
public void run() {
if (logger.isLoggable(XdsLogLevel.DEBUG)) {
logger.log(
XdsLogLevel.DEBUG,
"Received {0} response:\n{1}",
responseData.getResourceType(),
logger.log(XdsLogLevel.DEBUG, "Received {0} response:\n{1}",
ResourceType.fromTypeUrl(response.getTypeUrl()),
respPrinter.print(response));
}
onDiscoveryResponse(responseData);
DiscoveryResponseData responseData =
DiscoveryResponseData.fromEnvoyProtoV2(response);
handleResponse(responseData);
}
});
}
@Override
public void onError(Throwable t) {
AdsStreamV2.this.onError(t);
public void onError(final Throwable t) {
syncContext.execute(new Runnable() {
@Override
public void run() {
handleRpcError(t);
}
});
}
@Override
public void onCompleted() {
AdsStreamV2.this.onCompleted();
syncContext.execute(new Runnable() {
@Override
public void run() {
handleRpcCompleted();
}
});
}
};
requestWriterV2 = stubV2.withWaitForReady().streamAggregatedResources(responseReaderV2);
@ -1787,27 +1786,38 @@ final class XdsClientImpl extends XdsClient {
void start() {
StreamObserver<DiscoveryResponse> responseReader = new StreamObserver<DiscoveryResponse>() {
@Override
public void onNext(DiscoveryResponse response) {
DiscoveryResponseData responseData =
DiscoveryResponseData.fromEnvoyProto(response);
public void onNext(final DiscoveryResponse response) {
syncContext.execute(new Runnable() {
@Override
public void run() {
if (logger.isLoggable(XdsLogLevel.DEBUG)) {
logger.log(
XdsLogLevel.DEBUG,
"Received {0} response:\n{1}",
responseData.getResourceType(),
respPrinter.print(response));
logger.log(XdsLogLevel.DEBUG, "Received {0} response:\n{1}",
ResourceType.fromTypeUrl(response.getTypeUrl()), respPrinter.print(response));
}
onDiscoveryResponse(responseData);
DiscoveryResponseData responseData = DiscoveryResponseData.fromEnvoyProto(response);
handleResponse(responseData);
}
});
}
@Override
public void onError(Throwable t) {
AdsStream.this.onError(t);
public void onError(final Throwable t) {
syncContext.execute(new Runnable() {
@Override
public void run() {
handleRpcError(t);
}
});
}
@Override
public void onCompleted() {
AdsStream.this.onCompleted();
syncContext.execute(new Runnable() {
@Override
public void run() {
handleRpcCompleted();
}
});
}
};
requestWriter = stub.withWaitForReady().streamAggregatedResources(responseReader);