Skip to content

Commit 32fc451

Browse files
werktcopybara-github
werkt
authored and
copybara-github
committed
Write/QueryWriteStatus logging refinement/addition
Improve the logging of WriteRequests to include offset and finish_write information. Offsets are logged for the initial and non-sequential per successive write request. Each finish_write true request is logged with the effective size of the resource at the completion of the write request, including the current offset and payload. Clarified comments for WriteDetails, and corrected some comment inconsistencies. Add logging for QueryWriteStatus calls which occur on progressive writes to determine an offset to begin a write call on a retry. Closes bazelbuild#12928. PiperOrigin-RevId: 355545331
1 parent 8fc7787 commit 32fc451

File tree

6 files changed

+212
-8
lines changed

6 files changed

+212
-8
lines changed

‎src/main/java/com/google/devtools/build/lib/remote/logging/BUILD‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ java_library(
1919
"//src/main/java/com/google/devtools/build/lib/util/io",
2020
"//src/main/protobuf:remote_execution_log_java_proto",
2121
"//third_party:flogger",
22+
"//third_party:guava",
2223
"//third_party:jsr305",
2324
"//third_party/grpc:grpc-jar",
2425
"//third_party/protobuf:protobuf_java",

‎src/main/java/com/google/devtools/build/lib/remote/logging/LoggingInterceptor.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,8 @@ public LoggingInterceptor(AsynchronousFileOutputStream rpcLogFile, Clock clock)
7474
return new ReadHandler(); // <ReadRequest, ReadResponse>
7575
} else if (method == ByteStreamGrpc.getWriteMethod()) {
7676
return new WriteHandler(); // <WriteRequest, WriteResponse>
77+
} else if (method == ByteStreamGrpc.getQueryWriteStatusMethod()) {
78+
return new QueryWriteStatusHandler(); // <QueryWriteStatusRequest, QueryWriteStatusResponse>
7779
} else if (method == CapabilitiesGrpc.getGetCapabilitiesMethod()) {
7880
return new GetCapabilitiesHandler(); // <GetCapabilitiesRequest, ServerCapabilities>
7981
}

‎src/main/java/com/google/devtools/build/lib/remote/logging/QueryWriteStatusHandler.java‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
// Copyright 2021 The Bazel Authors. All rights reserved.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package com.google.devtools.build.lib.remote.logging;
16+
17+
import com.google.bytestream.ByteStreamProto.QueryWriteStatusRequest;
18+
import com.google.bytestream.ByteStreamProto.QueryWriteStatusResponse;
19+
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.QueryWriteStatusDetails;
20+
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.RpcCallDetails;
21+
22+
/** LoggingHandler for {@link google.bytestream.QueryWriteStatus} gRPC call. */
23+
public class QueryWriteStatusHandler
24+
implements LoggingHandler<QueryWriteStatusRequest, QueryWriteStatusResponse> {
25+
private final QueryWriteStatusDetails.Builder builder = QueryWriteStatusDetails.newBuilder();
26+
27+
@Override
28+
public void handleReq(QueryWriteStatusRequest message) {
29+
builder.setRequest(message);
30+
}
31+
32+
@Override
33+
public void handleResp(QueryWriteStatusResponse message) {
34+
builder.setResponse(message);
35+
}
36+
37+
@Override
38+
public RpcCallDetails getDetails() {
39+
return RpcCallDetails.newBuilder().setQueryWriteStatus(builder).build();
40+
}
41+
}

‎src/main/java/com/google/devtools/build/lib/remote/logging/WriteHandler.java‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,24 +16,40 @@
1616

1717
import com.google.bytestream.ByteStreamProto.WriteRequest;
1818
import com.google.bytestream.ByteStreamProto.WriteResponse;
19+
import com.google.common.collect.Iterables;
1920
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.RpcCallDetails;
2021
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.WriteDetails;
22+
import java.util.ArrayList;
2123
import java.util.LinkedHashSet;
24+
import java.util.List;
2225
import java.util.Set;
2326

2427
/** LoggingHandler for {@link google.bytestream.Write} gRPC call. */
2528
public class WriteHandler implements LoggingHandler<WriteRequest, WriteResponse> {
2629
private final WriteDetails.Builder builder = WriteDetails.newBuilder();
2730
private final Set<String> resources = new LinkedHashSet<>();
31+
private final List<Long> offsets = new ArrayList<>();
32+
private final List<Long> finishWrites = new ArrayList<>();
33+
private long bytesSentInSequence = 0;
2834
private long numWrites = 0;
2935
private long bytesSent = 0;
3036

3137
@Override
3238
public void handleReq(WriteRequest message) {
3339
resources.add(message.getResourceName());
40+
long writeOffset = message.getWriteOffset();
41+
if (numWrites == 0 || Iterables.getLast(offsets) + bytesSentInSequence != writeOffset) {
42+
offsets.add(writeOffset);
43+
bytesSentInSequence = 0;
44+
}
45+
int size = message.getData().size();
46+
if (message.getFinishWrite()) {
47+
finishWrites.add(writeOffset + size);
48+
}
3449

3550
numWrites++;
36-
bytesSent += message.getData().size();
51+
bytesSent += size;
52+
bytesSentInSequence += size;
3753
}
3854

3955
@Override
@@ -44,6 +60,8 @@ public void handleResp(WriteResponse message) {
4460
@Override
4561
public RpcCallDetails getDetails() {
4662
builder.addAllResourceNames(resources);
63+
builder.addAllOffsets(offsets);
64+
builder.addAllFinishWrites(finishWrites);
4765
builder.setNumWrites(numWrites);
4866
builder.setBytesSent(bytesSent);
4967
return RpcCallDetails.newBuilder().setWrite(builder).build();

‎src/main/protobuf/remote_execution_log.proto‎

Lines changed: 29 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,11 @@ syntax = "proto3";
1616

1717
package remote_logging;
1818

19-
import "build/bazel/remote/execution/v2/remote_execution.proto";
20-
import "google/protobuf/timestamp.proto";
2119
import "google/bytestream/bytestream.proto";
2220
import "google/longrunning/operations.proto";
21+
import "google/protobuf/timestamp.proto";
2322
import "google/rpc/status.proto";
23+
import "build/bazel/remote/execution/v2/remote_execution.proto";
2424

2525
option java_package = "com.google.devtools.build.lib.remote.logging";
2626

@@ -83,7 +83,7 @@ message GetActionResultDetails {
8383
// Details for a call to
8484
// build.bazel.remote.execution.v2.ActionCache.UpdateActionResult.
8585
message UpdateActionResultDetails {
86-
// The build.bazel.remote.execution.v2.GetActionResultRequest sent by
86+
// The build.bazel.remote.execution.v2.UpdateActionResultRequest sent by
8787
// the call.
8888
build.bazel.remote.execution.v2.UpdateActionResultRequest request = 1;
8989

@@ -128,10 +128,24 @@ message ReadDetails {
128128
message WriteDetails {
129129
// The names of resources requested to be written to in this call in the order
130130
// they were first requested in. If the ByteStream protocol is followed
131-
// according to specification, this should only contain have a single element,
132-
// which is the resource name specified in the first message of the stream.
131+
// according to specification, this should contain at most two elements:
132+
// The resource name specified in the first message of the stream, and an
133+
// empty string specified in each successive request if num_writes > 1.
133134
repeated string resource_names = 1;
134135

136+
// The offsets sent for the initial request and any non-sequential offsets
137+
// specified over the course of the call. If the ByteStream protocol is
138+
// followed according to specification, this should contain a single element
139+
// which is the starting point for the write call.
140+
repeated int64 offsets = 5;
141+
142+
// The effective final size for each request sent with finish_write true
143+
// specified over the course of the call. If the ByteStream protocol is
144+
// followed according to specification, this should contain a single element
145+
// which is the total size of the written resource, including the initial
146+
// offset.
147+
repeated int64 finish_writes = 6;
148+
135149
// The number of writes performed in this call.
136150
int64 num_writes = 2;
137151

@@ -142,6 +156,15 @@ message WriteDetails {
142156
google.bytestream.WriteResponse response = 4;
143157
}
144158

159+
// Details for a call to google.bytestream.QueryWriteStatus.
160+
message QueryWriteStatusDetails {
161+
// The google.bytestream.QueryWriteStatusRequest sent by the call.
162+
google.bytestream.QueryWriteStatusRequest request = 1;
163+
164+
// The received google.bytestream.QueryWriteStatusResponse.
165+
google.bytestream.QueryWriteStatusResponse response = 2;
166+
}
167+
145168
// Contains details for specific types of calls.
146169
message RpcCallDetails {
147170
reserved 1 to 4, 11;
@@ -152,6 +175,7 @@ message RpcCallDetails {
152175
FindMissingBlobsDetails find_missing_blobs = 10;
153176
ReadDetails read = 5;
154177
WriteDetails write = 6;
178+
QueryWriteStatusDetails query_write_status = 14;
155179
GetCapabilitiesDetails get_capabilities = 12;
156180
UpdateActionResultDetails update_action_result = 13;
157181
}

‎src/test/java/com/google/devtools/build/lib/remote/logging/LoggingInterceptorTest.java‎

Lines changed: 120 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,8 @@
4949
import com.google.bytestream.ByteStreamGrpc.ByteStreamBlockingStub;
5050
import com.google.bytestream.ByteStreamGrpc.ByteStreamImplBase;
5151
import com.google.bytestream.ByteStreamGrpc.ByteStreamStub;
52+
import com.google.bytestream.ByteStreamProto.QueryWriteStatusRequest;
53+
import com.google.bytestream.ByteStreamProto.QueryWriteStatusResponse;
5254
import com.google.bytestream.ByteStreamProto.ReadRequest;
5355
import com.google.bytestream.ByteStreamProto.ReadResponse;
5456
import com.google.bytestream.ByteStreamProto.WriteRequest;
@@ -58,6 +60,7 @@
5860
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.GetActionResultDetails;
5961
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.GetCapabilitiesDetails;
6062
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.LogEntry;
63+
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.QueryWriteStatusDetails;
6164
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.ReadDetails;
6265
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.RpcCallDetails;
6366
import com.google.devtools.build.lib.remote.logging.RemoteExecutionLog.UpdateActionResultDetails;
@@ -854,6 +857,10 @@ public void onCompleted() {
854857
WriteDetails.newBuilder()
855858
.addResourceNames("test1")
856859
.addResourceNames("test2")
860+
.addOffsets(0)
861+
.addOffsets(0)
862+
.addOffsets(0)
863+
// finish write is empty
857864
.setResponse(response)
858865
.setBytesSent(9)
859866
.setNumWrites(3)))
@@ -865,6 +872,81 @@ public void onCompleted() {
865872
verify(logStream).write(expectedEntry);
866873
}
867874

875+
@Test
876+
public void testWriteCallOffsetAndFinishWriteCompounding() {
877+
WriteRequest request1 =
878+
WriteRequest.newBuilder()
879+
.setResourceName("test1")
880+
.setData(ByteString.copyFromUtf8("abc"))
881+
.setWriteOffset(10)
882+
.build();
883+
WriteRequest request2 =
884+
WriteRequest.newBuilder()
885+
.setData(ByteString.copyFromUtf8("def"))
886+
.setWriteOffset(request1.getWriteOffset() + request1.getData().size())
887+
.build();
888+
WriteResponse response = WriteResponse.newBuilder().setCommittedSize(6).build();
889+
serviceRegistry.addService(
890+
new ByteStreamImplBase() {
891+
@Override
892+
public StreamObserver<WriteRequest> write(StreamObserver<WriteResponse> streamObserver) {
893+
return new StreamObserver<WriteRequest>() {
894+
@Override
895+
public void onNext(WriteRequest writeRequest) {}
896+
897+
@Override
898+
public void onError(Throwable throwable) {}
899+
900+
@Override
901+
public void onCompleted() {
902+
streamObserver.onNext(response);
903+
streamObserver.onCompleted();
904+
}
905+
};
906+
}
907+
});
908+
ByteStreamStub stub = ByteStreamGrpc.newStub(loggedChannel);
909+
@SuppressWarnings("unchecked")
910+
StreamObserver<WriteResponse> responseObserver = Mockito.mock(StreamObserver.class);
911+
912+
clock.advanceMillis(10000);
913+
// Request three writes, the first identical with the third, but offset correctly and
914+
// finish_writing
915+
StreamObserver<WriteRequest> requester = stub.write(responseObserver);
916+
requester.onNext(request1);
917+
clock.advanceMillis(100);
918+
requester.onNext(request2);
919+
clock.advanceMillis(200);
920+
requester.onNext(
921+
request1.toBuilder()
922+
.setWriteOffset(request2.getWriteOffset() + request2.getData().size())
923+
.setFinishWrite(true)
924+
.build());
925+
clock.advanceMillis(100);
926+
requester.onCompleted();
927+
928+
LogEntry expectedEntry =
929+
LogEntry.newBuilder()
930+
.setMethodName(ByteStreamGrpc.getWriteMethod().getFullMethodName())
931+
.setDetails(
932+
RpcCallDetails.newBuilder()
933+
.setWrite(
934+
WriteDetails.newBuilder()
935+
.addResourceNames("test1")
936+
.addResourceNames("")
937+
.addOffsets(request1.getWriteOffset())
938+
.addFinishWrites(
939+
10 + request1.getData().size() * 2 + request2.getData().size())
940+
.setResponse(response)
941+
.setBytesSent(9)
942+
.setNumWrites(3)))
943+
.setStatus(com.google.rpc.Status.getDefaultInstance())
944+
.setStartTime(Timestamp.newBuilder().setSeconds(10))
945+
.setEndTime(Timestamp.newBuilder().setSeconds(10).setNanos(400000000))
946+
.build();
947+
verify(logStream).write(expectedEntry);
948+
}
949+
868950
@Test
869951
public void testWriteCallFail() {
870952
WriteRequest request =
@@ -881,7 +963,6 @@ public StreamObserver<WriteRequest> write(StreamObserver<WriteResponse> streamOb
881963
return Mockito.mock(StreamObserver.class);
882964
}
883965
});
884-
885966
ByteStreamStub stub = ByteStreamGrpc.newStub(loggedChannel);
886967
@SuppressWarnings("unchecked")
887968
StreamObserver<WriteResponse> responseObserver = Mockito.mock(StreamObserver.class);
@@ -894,7 +975,6 @@ public StreamObserver<WriteRequest> write(StreamObserver<WriteResponse> streamOb
894975
requester.onError(error.asRuntimeException());
895976

896977
Status expectedCancel = Status.CANCELLED.withCause(error.asRuntimeException());
897-
898978
LogEntry expectedEntry =
899979
LogEntry.newBuilder()
900980
.setMethodName(ByteStreamGrpc.getWriteMethod().getFullMethodName())
@@ -907,12 +987,50 @@ public StreamObserver<WriteRequest> write(StreamObserver<WriteResponse> streamOb
907987
.setWrite(
908988
WriteDetails.newBuilder()
909989
.addResourceNames("test")
990+
.addOffsets(0)
910991
.setNumWrites(1)
911992
.setBytesSent(3)))
912993
.setStartTime(Timestamp.newBuilder().setSeconds(10000000))
913994
.setEndTime(Timestamp.newBuilder().setSeconds(20000000))
914995
.build();
996+
verify(logStream).write(expectedEntry);
997+
}
915998

999+
@Test
1000+
public void testQueryWriteStatusCallOk() {
1001+
QueryWriteStatusRequest request =
1002+
QueryWriteStatusRequest.newBuilder().setResourceName("test").build();
1003+
QueryWriteStatusResponse response =
1004+
QueryWriteStatusResponse.newBuilder().setCommittedSize(10).build();
1005+
serviceRegistry.addService(
1006+
new ByteStreamImplBase() {
1007+
@Override
1008+
public void queryWriteStatus(
1009+
QueryWriteStatusRequest request,
1010+
StreamObserver<QueryWriteStatusResponse> responseObserver) {
1011+
clock.advanceMillis(22222);
1012+
responseObserver.onNext(response);
1013+
responseObserver.onCompleted();
1014+
}
1015+
});
1016+
ByteStreamBlockingStub stub = ByteStreamGrpc.newBlockingStub(loggedChannel);
1017+
1018+
clock.advanceMillis(11111);
1019+
stub.queryWriteStatus(request);
1020+
1021+
LogEntry expectedEntry =
1022+
LogEntry.newBuilder()
1023+
.setMethodName(ByteStreamGrpc.getQueryWriteStatusMethod().getFullMethodName())
1024+
.setDetails(
1025+
RpcCallDetails.newBuilder()
1026+
.setQueryWriteStatus(
1027+
QueryWriteStatusDetails.newBuilder()
1028+
.setRequest(request)
1029+
.setResponse(response)))
1030+
.setStatus(com.google.rpc.Status.getDefaultInstance())
1031+
.setStartTime(Timestamp.newBuilder().setSeconds(11).setNanos(111000000))
1032+
.setEndTime(Timestamp.newBuilder().setSeconds(33).setNanos(333000000))
1033+
.build();
9161034
verify(logStream).write(expectedEntry);
9171035
}
9181036
}

0 commit comments

Comments
 (0)