Split up BuildEventServiceTransport into smaller classes.
Divide the class in more logical smaller classes in order
to have better understanding of the lifecycle of their creation,
improve the documentation and slightly improve the encapsulation.
PiperOrigin-RevId: 234041535
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
index d6031c5..67084ee 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
@@ -14,69 +14,26 @@
package com.google.devtools.build.lib.buildeventservice;
-import static com.google.common.base.Preconditions.checkState;
-import static com.google.common.util.concurrent.Uninterruptibles.getUninterruptibly;
-import static com.google.devtools.build.v1.BuildStatus.Result.COMMAND_FAILED;
-import static com.google.devtools.build.v1.BuildStatus.Result.COMMAND_SUCCEEDED;
-import static com.google.devtools.build.v1.BuildStatus.Result.UNKNOWN_STATUS;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.MoreObjects;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
-import com.google.common.collect.Iterables;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.devtools.build.lib.buildeventservice.client.BuildEventServiceClient;
-import com.google.devtools.build.lib.buildeventservice.client.BuildEventServiceClient.StreamContext;
import com.google.devtools.build.lib.buildeventstream.ArtifactGroupNamer;
-import com.google.devtools.build.lib.buildeventstream.BuildCompletingEvent;
import com.google.devtools.build.lib.buildeventstream.BuildEvent;
-import com.google.devtools.build.lib.buildeventstream.BuildEvent.LocalFile;
import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
-import com.google.devtools.build.lib.buildeventstream.BuildEventContext;
import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
-import com.google.devtools.build.lib.buildeventstream.PathConverter;
import com.google.devtools.build.lib.clock.Clock;
import com.google.devtools.build.lib.util.ExitCode;
import com.google.devtools.build.lib.util.JavaSleeper;
-import com.google.devtools.build.lib.util.LoggingUtil;
import com.google.devtools.build.lib.util.Sleeper;
-import com.google.devtools.build.lib.vfs.Path;
-import com.google.devtools.build.v1.BuildStatus.Result;
-import com.google.devtools.build.v1.PublishBuildToolEventStreamRequest;
-import com.google.devtools.build.v1.PublishLifecycleEventRequest;
-import com.google.protobuf.Any;
-import com.google.protobuf.Timestamp;
-import com.google.protobuf.util.Timestamps;
-import io.grpc.Status;
-import io.grpc.Status.Code;
-import io.grpc.StatusException;
import java.time.Duration;
-import java.util.Collection;
-import java.util.Map;
-import java.util.TreeMap;
-import java.util.concurrent.BlockingDeque;
-import java.util.concurrent.ConcurrentLinkedDeque;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedBlockingDeque;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Consumer;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-import javax.annotation.concurrent.GuardedBy;
-import javax.annotation.concurrent.Immutable;
/** A {@link BuildEventTransport} that streams {@link BuildEvent}s to BuildEventService. */
public class BuildEventServiceTransport implements BuildEventTransport {
-
private final BuildEventServiceUploader besUploader;
/** A builder for {@link BuildEventServiceTransport}. */
@@ -171,7 +128,7 @@
@Override
public BuildEventArtifactUploader getUploader() {
- return besUploader.localFileUploader;
+ return besUploader.getLocalFileUploader();
}
@Override
@@ -184,6 +141,11 @@
besUploader.enqueueEvent(event);
}
+ @VisibleForTesting
+ public BuildEventServiceUploader getBesUploader() {
+ return besUploader;
+ }
+
/** BuildEventLogger can be used to log build event (stats). */
@FunctionalInterface
public interface BuildEventLogger {
@@ -198,791 +160,4 @@
public interface ExitFunction {
void accept(String message, Throwable cause, ExitCode code);
}
-
- /**
- * This method is only used in tests. Once TODO(b/113035235) is fixed the close future will also
- * carry error messages.
- */
- @VisibleForTesting
- public void throwUploaderError() throws Throwable {
- synchronized (besUploader.lock) {
- checkState(besUploader.closeFuture != null && besUploader.closeFuture.isDone());
- try {
- besUploader.closeFuture.get();
- } catch (ExecutionException e) {
- throw e.getCause();
- }
- }
- }
-
- /** Implements the BES upload which includes uploading the lifecycle and build events. */
- private static class BuildEventServiceUploader implements Runnable {
-
- private static final Logger logger =
- Logger.getLogger(BuildEventServiceUploader.class.getName());
-
- /** Configuration knobs related to RPC retries. Values chosen by good judgement. */
- private static final int MAX_NUM_RETRIES = 4;
-
- private static final int DELAY_MILLIS = 1000;
-
- private final BuildEventServiceClient besClient;
- private final BuildEventArtifactUploader localFileUploader;
- private final BuildEventServiceProtoUtil besProtoUtil;
- private final BuildEventProtocolOptions protocolOptions;
- private final boolean publishLifecycleEvents;
- private final Duration closeTimeout;
- private final ExitFunction exitFunc;
- private final Sleeper sleeper;
- private final Clock clock;
- private final BuildEventLogger buildEventLogger;
- private final ArtifactGroupNamer namer;
-
- /**
- * The event queue contains two types of events: - Build events, sorted by sequence number, that
- * should be sent to the server - Command events that are used by {@link #publishBuildEvents()}
- * to change state.
- */
- private final BlockingDeque<EventLoopCommand> eventQueue = new LinkedBlockingDeque<>();
-
- /**
- * Computes sequence numbers for build events. As per the BES protocol, sequence numbers must be
- * consecutive monotonically increasing natural numbers.
- */
- private final AtomicLong nextSeqNum = new AtomicLong(1);
-
- private final Object lock = new Object();
-
- @GuardedBy("lock")
- private Result buildStatus = UNKNOWN_STATUS;
-
- /**
- * Initialized only after the first call to {@link #close()} or if the upload fails before that.
- * The {@code null} state is used throughout the code to make multiple calls to {@link #close()}
- * idempotent.
- */
- @GuardedBy("lock")
- private SettableFuture<Void> closeFuture;
-
- /**
- * The thread that calls the lifecycle RPCs and does the build event upload. It's started lazily
- * on the first call to {@link #enqueueEvent(BuildEvent)} or {@link #close()} (which ever comes
- * first).
- */
- @GuardedBy("lock")
- private Thread uploadThread;
-
- @GuardedBy("lock")
- private boolean interruptCausedByTimeout;
-
- private StreamContext streamContext;
-
- public BuildEventServiceUploader(
- BuildEventServiceClient besClient,
- BuildEventArtifactUploader localFileUploader,
- BuildEventServiceProtoUtil besProtoUtil,
- BuildEventProtocolOptions protocolOptions,
- boolean publishLifecycleEvents,
- Duration closeTimeout,
- ExitFunction exitFunc,
- Sleeper sleeper,
- Clock clock,
- BuildEventLogger buildEventLogger,
- ArtifactGroupNamer namer) {
- this.besClient = Preconditions.checkNotNull(besClient);
- this.localFileUploader = Preconditions.checkNotNull(localFileUploader);
- this.besProtoUtil = Preconditions.checkNotNull(besProtoUtil);
- this.protocolOptions = Preconditions.checkNotNull(protocolOptions);
- this.publishLifecycleEvents = publishLifecycleEvents;
- this.closeTimeout = Preconditions.checkNotNull(closeTimeout);
- this.exitFunc = Preconditions.checkNotNull(exitFunc);
- this.sleeper = Preconditions.checkNotNull(sleeper);
- this.clock = Preconditions.checkNotNull(clock);
- this.buildEventLogger = Preconditions.checkNotNull(buildEventLogger);
- this.namer = Preconditions.checkNotNull(namer);
- }
-
- /** Enqueues an event for uploading to a BES backend. */
- public void enqueueEvent(BuildEvent event) {
- // This needs to happen outside a synchronized block as it may trigger
- // stdout/stderr and lead to a deadlock. See b/109725432
- ListenableFuture<PathConverter> localFileUploadFuture =
- uploadReferencedLocalFiles(event.referencedLocalFiles());
-
- synchronized (lock) {
- if (closeFuture != null) {
- // Close has been called and thus we silently ignore any further events and cancel
- // any pending file uploads
- closeFuture.addListener(
- () -> {
- if (!localFileUploadFuture.isDone()) {
- localFileUploadFuture.cancel(true);
- }
- },
- MoreExecutors.directExecutor());
- return;
- }
- // BuildCompletingEvent marks the end of the build in the BEP event stream.
- if (event instanceof BuildCompletingEvent) {
- this.buildStatus = extractBuildStatus((BuildCompletingEvent) event);
- }
- ensureUploadThreadStarted();
- eventQueue.addLast(
- new SendRegularBuildEventCommand(
- event, namer, localFileUploadFuture, nextSeqNum.getAndIncrement(), currentTime()));
- }
- }
-
- /**
- * Gracefully stops the BES upload. All events enqueued before the call to close will be
- * uploaded and events enqueued after the call will be discarded.
- *
- * <p>The returned future completes when the upload completes. It's guaranteed to never fail.
- */
- public ListenableFuture<Void> close() {
- synchronized (lock) {
- if (closeFuture != null) {
- return closeFuture;
- }
- ensureUploadThreadStarted();
-
- closeFuture = SettableFuture.create();
-
- // Enqueue the last event which will terminate the upload.
- eventQueue.addLast(
- new SendLastBuildEventCommand(nextSeqNum.getAndIncrement(), currentTime()));
-
- if (!closeTimeout.isZero()) {
- startCloseTimer(closeFuture, closeTimeout);
- }
- return closeFuture;
- }
- }
-
- /** Stops the upload immediately. Enqueued events that have not been sent yet will be lost. */
- void closeOnTimeout() {
- synchronized (lock) {
- if (uploadThread != null) {
- if (uploadThread.isInterrupted()) {
- return;
- }
-
- interruptCausedByTimeout = true;
- uploadThread.interrupt();
- }
- }
- }
-
- @Override
- public void run() {
- try {
- if (publishLifecycleEvents) {
- publishLifecycleEvent(besProtoUtil.buildEnqueued(currentTime()));
- publishLifecycleEvent(besProtoUtil.invocationStarted(currentTime()));
- }
-
- try {
- publishBuildEvents();
- } finally {
- if (publishLifecycleEvents) {
- Result buildStatus;
- synchronized (lock) {
- buildStatus = this.buildStatus;
- }
- publishLifecycleEvent(besProtoUtil.invocationFinished(currentTime(), buildStatus));
- publishLifecycleEvent(besProtoUtil.buildFinished(currentTime(), buildStatus));
- }
- }
- exitFunc.accept(
- "The Build Event Protocol upload finished successfully",
- /*cause=*/ null,
- ExitCode.SUCCESS);
- synchronized (lock) {
- // Invariant: closeFuture is not null.
- // publishBuildEvents() only terminates successfully after SendLastBuildEventCommand
- // has been sent successfully and that event is only added to the eventQueue during a
- // call to close() which initializes the closeFuture.
- closeFuture.set(null);
- }
- } catch (InterruptedException e) {
- try {
- logInfo(e, "Aborting the BES upload due to having received an interrupt");
- synchronized (lock) {
- Preconditions.checkState(
- interruptCausedByTimeout, "Unexpected interrupt on BES uploader thread");
- exitFunc.accept(
- "The Build Event Protocol upload timed out",
- e,
- ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR);
- }
- } finally {
- // TODO(buchgr): Due to b/113035235 exitFunc needs to be called before the close future
- // completes.
- failCloseFuture(e);
- }
- } catch (StatusException e) {
- try {
- String message =
- "The Build Event Protocol upload failed: " + besClient.userReadableError(e);
- logInfo(e, message);
- ExitCode code =
- shouldRetryStatus(e.getStatus())
- ? ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR
- : ExitCode.PERSISTENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR;
- exitFunc.accept(message, e, code);
- } finally {
- failCloseFuture(e);
- }
- } catch (LocalFileUploadException e) {
- Throwables.throwIfUnchecked(e.getCause());
- try {
- String message =
- "The Build Event Protocol local file upload failed: " + e.getCause().getMessage();
- logInfo(e.getCause(), message);
- exitFunc.accept(
- message, e.getCause(), ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR);
- } finally {
- failCloseFuture(e.getCause());
- }
- } catch (Throwable e) {
- failCloseFuture(e);
- logError(e, "BES upload failed due to a RuntimeException / Error. This is a bug.");
- throw e;
- } finally {
- localFileUploader.shutdown();
- }
- }
-
- private void publishBuildEvents()
- throws StatusException, LocalFileUploadException, InterruptedException {
- eventQueue.addFirst(new OpenStreamCommand());
-
- // Every build event sent to the server needs to be acknowledged by it. This queue stores
- // the build events that have been sent and still have to be acknowledged by the server.
- // The build events are stored in the order they were sent.
- ConcurrentLinkedDeque<SendBuildEventCommand> ackQueue = new ConcurrentLinkedDeque<>();
- boolean lastEventSent = false;
- int acksReceived = 0;
- int retryAttempt = 0;
-
- try {
- // OPEN_STREAM is the first event and opens a bidi streaming RPC for sending build events
- // and receiving ACKs.
- // SEND_BUILD_EVENT sends a build event to the server. Sending of the Nth build event does
- // does not wait for the ACK of the N-1th build event to have been received.
- // SEND_LAST_EVENT sends the last build event and half closes the RPC.
- // ACK_RECEIVED is executed for every ACK from the server and checks that the ACKs are in
- // the correct order.
- // STREAM_COMPLETE checks that all build events have been sent and all ACKs have been
- // received. If not it invokes a retry logic that may decide to re-send every build event
- // for which an ACK has not been received. If so, it adds an OPEN_STREAM event.
- while (true) {
- EventLoopCommand event = eventQueue.takeFirst();
- switch (event.type()) {
- case OPEN_STREAM:
- {
- // Invariant: the eventQueue only contains events of type SEND_BUILD_EVENT
- // or SEND_LAST_EVENT
- logInfo("Starting publishBuildEvents: eventQueue=%d", eventQueue.size());
- streamContext =
- besClient.openStream(
- (ack) ->
- eventQueue.addLast(new AckReceivedCommand(ack.getSequenceNumber())));
- addStreamStatusListener(
- streamContext.getStatus(),
- (status) -> eventQueue.addLast(new StreamCompleteCommand(status)));
- }
- break;
-
- case SEND_BUILD_EVENT:
- {
- // Invariant: the eventQueue may contain events of any type
- SendRegularBuildEventCommand buildEvent = (SendRegularBuildEventCommand) event;
- ackQueue.addLast(buildEvent);
- PathConverter pathConverter = waitForLocalFileUploads(buildEvent);
- streamContext.sendOverStream(buildEvent.serialize(pathConverter));
- }
- break;
-
- case SEND_LAST_EVENT:
- {
- // Invariant: the eventQueue may contain events of any type
- SendLastBuildEventCommand lastEvent = (SendLastBuildEventCommand) event;
- ackQueue.addLast(lastEvent);
- lastEventSent = true;
- streamContext.sendOverStream(lastEvent.serialize());
- streamContext.halfCloseStream();
- }
- break;
-
- case ACK_RECEIVED:
- {
- // Invariant: the eventQueue may contain events of any type
- AckReceivedCommand ackEvent = (AckReceivedCommand) event;
- if (!ackQueue.isEmpty()) {
- SendBuildEventCommand expected = ackQueue.removeFirst();
- long actualSeqNum = ackEvent.getSequenceNumber();
- if (expected.getSequenceNumber() == actualSeqNum) {
- acksReceived++;
- } else {
- ackQueue.addFirst(expected);
- String message =
- String.format(
- "Expected ACK with seqNum=%d but received ACK with seqNum=%d",
- expected.getSequenceNumber(), actualSeqNum);
- logInfo(message);
- streamContext.abortStream(Status.FAILED_PRECONDITION.withDescription(message));
- }
- } else {
- String message =
- String.format(
- "Received ACK (seqNum=%d) when no ACK was expected",
- ackEvent.getSequenceNumber());
- logInfo(message);
- streamContext.abortStream(Status.FAILED_PRECONDITION.withDescription(message));
- }
- }
- break;
-
- case STREAM_COMPLETE:
- {
- // Invariant: the eventQueue only contains events of type SEND_BUILD_EVENT
- // or SEND_LAST_EVENT
- streamContext = null;
- StreamCompleteCommand completeEvent = (StreamCompleteCommand) event;
- Status streamStatus = completeEvent.status();
- if (streamStatus.isOk()) {
- if (lastEventSent && ackQueue.isEmpty()) {
- logInfo("publishBuildEvents was successful");
- // Upload successful. Break out from the while(true) loop.
- return;
- } else {
- throw (lastEventSent
- ? ackQueueNotEmptyStatus(ackQueue.size())
- : lastEventNotSentStatus())
- .asException();
- }
- }
-
- if (!shouldRetryStatus(streamStatus)) {
- logInfo("Not retrying publishBuildEvents: status='%s'", streamStatus);
- throw streamStatus.asException();
- }
- if (retryAttempt == MAX_NUM_RETRIES) {
- logInfo(
- "Not retrying publishBuildEvents, no more attempts left: status='%s'",
- streamStatus);
- throw streamStatus.asException();
- }
-
- // Retry logic
- // Adds events from the ackQueue to the front of the eventQueue, so that the
- // events in the eventQueue are sorted by sequence number (ascending).
- SendBuildEventCommand unacked;
- while ((unacked = ackQueue.pollLast()) != null) {
- eventQueue.addFirst(unacked);
- }
-
- long sleepMillis = retrySleepMillis(retryAttempt);
- logInfo("Retrying stream: status='%s', sleepMillis=%d", streamStatus, sleepMillis);
- sleeper.sleepMillis(sleepMillis);
-
- // If we made progress, meaning the server ACKed events that we sent, then reset
- // the retry counter to 0.
- if (acksReceived > 0) {
- retryAttempt = 0;
- } else {
- retryAttempt++;
- }
- acksReceived = 0;
- eventQueue.addFirst(new OpenStreamCommand());
- }
- break;
- }
- }
- } catch (InterruptedException | LocalFileUploadException e) {
- int limit = 30;
- logInfo(
- String.format(
- "Publish interrupt. Showing up to %d items from queues: ack_queue_size: %d, "
- + "ack_queue: %s, event_queue_size: %d, event_queue: %s",
- limit,
- ackQueue.size(),
- Iterables.limit(ackQueue, limit),
- eventQueue.size(),
- Iterables.limit(eventQueue, limit)));
- if (streamContext != null) {
- streamContext.abortStream(Status.CANCELLED);
- }
- throw e;
- } finally {
- // Cancel all local file uploads that may still be running
- // of events that haven't been uploaded.
- EventLoopCommand event;
- while ((event = ackQueue.pollFirst()) != null) {
- if (event instanceof SendRegularBuildEventCommand) {
- cancelLocalFileUpload((SendRegularBuildEventCommand) event);
- }
- }
- while ((event = eventQueue.pollFirst()) != null) {
- if (event instanceof SendRegularBuildEventCommand) {
- cancelLocalFileUpload((SendRegularBuildEventCommand) event);
- }
- }
- }
- }
-
- private void cancelLocalFileUpload(SendRegularBuildEventCommand event) {
- ListenableFuture<PathConverter> localFileUploaderFuture = event.localFileUploadProgress();
- if (!localFileUploaderFuture.isDone()) {
- localFileUploaderFuture.cancel(true);
- }
- }
-
- /** Sends a {@link PublishLifecycleEventRequest} to the BES backend. */
- private void publishLifecycleEvent(PublishLifecycleEventRequest request)
- throws StatusException, InterruptedException {
- int retryAttempt = 0;
- StatusException cause = null;
- while (retryAttempt <= MAX_NUM_RETRIES) {
- try {
- besClient.publish(request);
- return;
- } catch (StatusException e) {
- if (!shouldRetryStatus(e.getStatus())) {
- logInfo("Not retrying publishLifecycleEvent: status='%s'", e.getStatus().toString());
- throw e;
- }
-
- cause = e;
-
- long sleepMillis = retrySleepMillis(retryAttempt);
- logInfo(
- "Retrying publishLifecycleEvent: status='%s', sleepMillis=%d",
- e.getStatus().toString(), sleepMillis);
- sleeper.sleepMillis(sleepMillis);
- retryAttempt++;
- }
- }
-
- // All retry attempts failed
- throw cause;
- }
-
- private ListenableFuture<PathConverter> uploadReferencedLocalFiles(
- Collection<LocalFile> localFiles) {
- Map<Path, LocalFile> localFileMap = new TreeMap<>();
- for (LocalFile localFile : localFiles) {
- // It is possible for targets to have duplicate artifacts (same path but different owners)
- // in their output groups. Since they didn't trigger an artifact conflict they are the
- // same file, so just skip either one
- localFileMap.putIfAbsent(localFile.path, localFile);
- }
- return localFileUploader.upload(localFileMap);
- }
-
- private void ensureUploadThreadStarted() {
- synchronized (lock) {
- if (uploadThread == null) {
- uploadThread = new Thread(this, "bes-uploader");
- uploadThread.start();
- }
- }
- }
-
- private void startCloseTimer(ListenableFuture<Void> closeFuture, Duration closeTimeout) {
- Thread closeTimer =
- new Thread(
- () -> {
- // Call closeOnTimeout() if the future does not complete within closeTimeout
- try {
- getUninterruptibly(closeFuture, closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
- } catch (TimeoutException e) {
- closeOnTimeout();
- } catch (ExecutionException e) {
- if (e.getCause() instanceof TimeoutException) {
- // This is likely due to an internal timeout doing the local file uploading.
- closeOnTimeout();
- } else {
- // This code only cares about calling closeOnTimeout() if the closeFuture does
- // not complete within closeTimeout.
- String failureMsg = "BES close failure";
- logError(e, failureMsg);
- LoggingUtil.logToRemote(Level.SEVERE, failureMsg, e);
- }
- }
- },
- "bes-uploader-close-timer");
- closeTimer.start();
- }
-
- private void failCloseFuture(Throwable cause) {
- synchronized (lock) {
- if (closeFuture == null) {
- closeFuture = SettableFuture.create();
- }
- closeFuture.setException(cause);
- }
- }
-
- private PathConverter waitForLocalFileUploads(SendRegularBuildEventCommand orderedBuildEvent)
- throws LocalFileUploadException, InterruptedException {
- try {
- // Wait for the local file upload to have been completed.
- return orderedBuildEvent.localFileUploadProgress().get();
- } catch (ExecutionException e) {
- logger.log(
- Level.WARNING,
- String.format(
- "Failed to upload local files referenced by build event: %s", e.getMessage()),
- e);
- Throwables.throwIfUnchecked(e.getCause());
- throw new LocalFileUploadException(e.getCause());
- }
- }
-
- private Timestamp currentTime() {
- return Timestamps.fromMillis(clock.currentTimeMillis());
- }
-
- private static Result extractBuildStatus(BuildCompletingEvent event) {
- if (event.getExitCode() != null && event.getExitCode().getNumericExitCode() == 0) {
- return COMMAND_SUCCEEDED;
- } else {
- return COMMAND_FAILED;
- }
- }
-
- private static Status lastEventNotSentStatus() {
- return Status.FAILED_PRECONDITION.withDescription(
- "Server closed stream with status OK but not all events have been sent");
- }
-
- private static Status ackQueueNotEmptyStatus(int ackQueueSize) {
- return Status.FAILED_PRECONDITION.withDescription(
- String.format(
- "Server closed stream with status OK but not all ACKs have been"
- + " received (ackQueue=%d)",
- ackQueueSize));
- }
-
- private static void addStreamStatusListener(
- ListenableFuture<Status> stream, Consumer<Status> onDone) {
- Futures.addCallback(
- stream,
- new FutureCallback<Status>() {
- @Override
- public void onSuccess(Status result) {
- onDone.accept(result);
- }
-
- @Override
- public void onFailure(Throwable t) {}
- },
- MoreExecutors.directExecutor());
- }
-
- private static boolean shouldRetryStatus(Status status) {
- return !status.getCode().equals(Code.INVALID_ARGUMENT)
- && !status.getCode().equals(Code.FAILED_PRECONDITION);
- }
-
- private static long retrySleepMillis(int attempt) {
- // This somewhat matches the backoff used for gRPC connection backoffs.
- return (long) (DELAY_MILLIS * Math.pow(1.6, attempt));
- }
-
- private static void logInfo(String message, Object... args) {
- logger.log(Level.INFO, String.format(message, args));
- }
-
- private static void logInfo(Throwable cause, String message, Object... args) {
- logger.log(Level.INFO, String.format(message, args), cause);
- }
-
- private static void logError(Throwable cause, String message, Object... args) {
- logger.log(Level.SEVERE, String.format(message, args), cause);
- }
-
- /** A command that may be added to the {@code eventQueue}. */
- private interface EventLoopCommand {
-
- /**
- * The event types are used to switch between states in the event loop in {@link
- * #publishBuildEvents()}
- */
- enum Type {
- /** Tells the event loop to open a new BES stream */
- OPEN_STREAM,
- /** Tells the event loop to send the build event */
- SEND_BUILD_EVENT,
- /** Tells the event loop that an ACK was received */
- ACK_RECEIVED,
- /** Tells the event loop that this is the last event of the stream */
- SEND_LAST_EVENT,
- /** Tells the event loop that the streaming RPC completed */
- STREAM_COMPLETE
- }
-
- Type type();
- }
-
- @Immutable
- private static final class OpenStreamCommand implements EventLoopCommand {
-
- @Override
- public Type type() {
- return Type.OPEN_STREAM;
- }
- }
-
- @Immutable
- private static final class StreamCompleteCommand implements EventLoopCommand {
-
- private final Status status;
-
- public StreamCompleteCommand(Status status) {
- this.status = status;
- }
-
- public Status status() {
- return status;
- }
-
- @Override
- public Type type() {
- return Type.STREAM_COMPLETE;
- }
- }
-
- @Immutable
- private static final class AckReceivedCommand implements EventLoopCommand {
-
- private final long sequenceNumber;
-
- public AckReceivedCommand(long sequenceNumber) {
- this.sequenceNumber = sequenceNumber;
- }
-
- public long getSequenceNumber() {
- return sequenceNumber;
- }
-
- @Override
- public Type type() {
- return Type.ACK_RECEIVED;
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(this).add("seq_num", getSequenceNumber()).toString();
- }
- }
-
- private abstract static class SendBuildEventCommand implements EventLoopCommand {
-
- abstract long getSequenceNumber();
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(this).add("seq_num", getSequenceNumber()).toString();
- }
- }
-
- private final class SendRegularBuildEventCommand extends SendBuildEventCommand {
-
- private final BuildEvent event;
- private final ArtifactGroupNamer namer;
- private final ListenableFuture<PathConverter> localFileUpload;
- private final long sequenceNumber;
- private final Timestamp creationTime;
-
- private SendRegularBuildEventCommand(
- BuildEvent event,
- ArtifactGroupNamer namer,
- ListenableFuture<PathConverter> localFileUpload,
- long sequenceNumber,
- Timestamp creationTime) {
- this.event = event;
- this.namer = namer;
- this.localFileUpload = localFileUpload;
- this.sequenceNumber = sequenceNumber;
- this.creationTime = creationTime;
- }
-
- public ListenableFuture<PathConverter> localFileUploadProgress() {
- return localFileUpload;
- }
-
- public PublishBuildToolEventStreamRequest serialize(PathConverter pathConverter) {
- BuildEventContext ctx =
- new BuildEventContext() {
- @Override
- public PathConverter pathConverter() {
- return pathConverter;
- }
-
- @Override
- public ArtifactGroupNamer artifactGroupNamer() {
- return namer;
- }
-
- @Override
- public BuildEventProtocolOptions getOptions() {
- return protocolOptions;
- }
- };
- BuildEventStreamProtos.BuildEvent serializedBepEvent = event.asStreamProto(ctx);
- buildEventLogger.log(serializedBepEvent);
- return besProtoUtil.bazelEvent(sequenceNumber, creationTime, Any.pack(serializedBepEvent));
- }
-
- @Override
- public long getSequenceNumber() {
- return sequenceNumber;
- }
-
- @Override
- public Type type() {
- return Type.SEND_BUILD_EVENT;
- }
-
- @Override
- public String toString() {
- return super.toString() + " - [" + event + "]";
- }
- }
-
- @Immutable
- private final class SendLastBuildEventCommand extends SendBuildEventCommand {
-
- private final long sequenceNumber;
- private final Timestamp creationTime;
-
- SendLastBuildEventCommand(long sequenceNumber, Timestamp creationTime) {
- this.sequenceNumber = sequenceNumber;
- this.creationTime = creationTime;
- }
-
- public PublishBuildToolEventStreamRequest serialize() {
- return besProtoUtil.streamFinished(sequenceNumber, creationTime);
- }
-
- @Override
- public long getSequenceNumber() {
- return sequenceNumber;
- }
-
- @Override
- public Type type() {
- return Type.SEND_LAST_EVENT;
- }
- }
-
- private static class LocalFileUploadException extends Exception {
- LocalFileUploadException(Throwable cause) {
- super(cause);
- }
- }
- }
}
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploader.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploader.java
new file mode 100644
index 0000000..90b4181
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploader.java
@@ -0,0 +1,744 @@
+// Copyright 2019 The Bazel Authors. All rights reserved.
+//
+// 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 com.google.devtools.build.lib.buildeventservice;
+
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.util.concurrent.Uninterruptibles.getUninterruptibly;
+import static com.google.devtools.build.v1.BuildStatus.Result.COMMAND_FAILED;
+import static com.google.devtools.build.v1.BuildStatus.Result.COMMAND_SUCCEEDED;
+import static com.google.devtools.build.v1.BuildStatus.Result.UNKNOWN_STATUS;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceTransport.BuildEventLogger;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceTransport.ExitFunction;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceUploaderCommands.AckReceivedCommand;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceUploaderCommands.EventLoopCommand;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceUploaderCommands.OpenStreamCommand;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceUploaderCommands.SendBuildEventCommand;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceUploaderCommands.SendLastBuildEventCommand;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceUploaderCommands.SendRegularBuildEventCommand;
+import com.google.devtools.build.lib.buildeventservice.BuildEventServiceUploaderCommands.StreamCompleteCommand;
+import com.google.devtools.build.lib.buildeventservice.client.BuildEventServiceClient;
+import com.google.devtools.build.lib.buildeventservice.client.BuildEventServiceClient.StreamContext;
+import com.google.devtools.build.lib.buildeventstream.ArtifactGroupNamer;
+import com.google.devtools.build.lib.buildeventstream.BuildCompletingEvent;
+import com.google.devtools.build.lib.buildeventstream.BuildEvent;
+import com.google.devtools.build.lib.buildeventstream.BuildEvent.LocalFile;
+import com.google.devtools.build.lib.buildeventstream.BuildEventArtifactUploader;
+import com.google.devtools.build.lib.buildeventstream.BuildEventContext;
+import com.google.devtools.build.lib.buildeventstream.BuildEventProtocolOptions;
+import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
+import com.google.devtools.build.lib.buildeventstream.PathConverter;
+import com.google.devtools.build.lib.clock.Clock;
+import com.google.devtools.build.lib.util.ExitCode;
+import com.google.devtools.build.lib.util.LoggingUtil;
+import com.google.devtools.build.lib.util.Sleeper;
+import com.google.devtools.build.lib.vfs.Path;
+import com.google.devtools.build.v1.BuildStatus.Result;
+import com.google.devtools.build.v1.PublishBuildToolEventStreamRequest;
+import com.google.devtools.build.v1.PublishLifecycleEventRequest;
+import com.google.protobuf.Any;
+import com.google.protobuf.Timestamp;
+import com.google.protobuf.util.Timestamps;
+import io.grpc.Status;
+import io.grpc.Status.Code;
+import io.grpc.StatusException;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.concurrent.GuardedBy;
+
+/**
+ * Uploader of Build Events to the Build Event Service (BES).
+ *
+ * <p>The purpose is of this class is to manage the interaction between the BES client and the BES
+ * server. It implements the event loop pattern based on the commands defined by {@link
+ * BuildEventServiceUploaderCommands}.
+ */
+// TODO(lpino): This class should be package-private but there are unit tests that are in the
+// different packages and rely on this.
+@VisibleForTesting
+public final class BuildEventServiceUploader implements Runnable {
+ private static final Logger logger = Logger.getLogger(BuildEventServiceUploader.class.getName());
+
+ /** Configuration knobs related to RPC retries. Values chosen by good judgement. */
+ private static final int MAX_NUM_RETRIES = 4;
+
+ private static final int DELAY_MILLIS = 1000;
+
+ private final BuildEventServiceClient besClient;
+ private final BuildEventArtifactUploader localFileUploader;
+ private final BuildEventServiceProtoUtil besProtoUtil;
+ private final BuildEventProtocolOptions buildEventProtocolOptions;
+ private final boolean publishLifecycleEvents;
+ private final Duration closeTimeout;
+ private final ExitFunction exitFunc;
+ private final Sleeper sleeper;
+ private final Clock clock;
+ private final BuildEventLogger buildEventLogger;
+ private final ArtifactGroupNamer namer;
+
+ /**
+ * The event queue contains two types of events: - Build events, sorted by sequence number, that
+ * should be sent to the server - Command events that are used by {@link #publishBuildEvents()} to
+ * change state.
+ */
+ private final BlockingDeque<EventLoopCommand> eventQueue = new LinkedBlockingDeque<>();
+
+ /**
+ * Computes sequence numbers for build events. As per the BES protocol, sequence numbers must be
+ * consecutive monotonically increasing natural numbers.
+ */
+ private final AtomicLong nextSeqNum = new AtomicLong(1);
+
+ private final Object lock = new Object();
+
+ @GuardedBy("lock")
+ private Result buildStatus = UNKNOWN_STATUS;
+
+ /**
+ * Initialized only after the first call to {@link #close()} or if the upload fails before that.
+ * The {@code null} state is used throughout the code to make multiple calls to {@link #close()}
+ * idempotent.
+ */
+ @GuardedBy("lock")
+ private SettableFuture<Void> closeFuture;
+
+ /**
+ * The thread that calls the lifecycle RPCs and does the build event upload. It's started lazily
+ * on the first call to {@link #enqueueEvent(BuildEvent)} or {@link #close()} (which ever comes
+ * first).
+ */
+ @GuardedBy("lock")
+ private Thread uploadThread;
+
+ @GuardedBy("lock")
+ private boolean interruptCausedByTimeout;
+
+ private StreamContext streamContext;
+
+ BuildEventServiceUploader(
+ BuildEventServiceClient besClient,
+ BuildEventArtifactUploader localFileUploader,
+ BuildEventServiceProtoUtil besProtoUtil,
+ BuildEventProtocolOptions buildEventProtocolOptions,
+ boolean publishLifecycleEvents,
+ Duration closeTimeout,
+ ExitFunction exitFunc,
+ Sleeper sleeper,
+ Clock clock,
+ BuildEventLogger buildEventLogger,
+ ArtifactGroupNamer namer) {
+ this.besClient = Preconditions.checkNotNull(besClient);
+ this.localFileUploader = Preconditions.checkNotNull(localFileUploader);
+ this.besProtoUtil = Preconditions.checkNotNull(besProtoUtil);
+ this.buildEventProtocolOptions = buildEventProtocolOptions;
+ this.publishLifecycleEvents = publishLifecycleEvents;
+ this.closeTimeout = Preconditions.checkNotNull(closeTimeout);
+ this.exitFunc = Preconditions.checkNotNull(exitFunc);
+ this.sleeper = Preconditions.checkNotNull(sleeper);
+ this.clock = Preconditions.checkNotNull(clock);
+ this.buildEventLogger = Preconditions.checkNotNull(buildEventLogger);
+ this.namer = namer;
+ }
+
+ BuildEventArtifactUploader getLocalFileUploader() {
+ return localFileUploader;
+ }
+
+ /** Enqueues an event for uploading to a BES backend. */
+ void enqueueEvent(BuildEvent event) {
+ // This needs to happen outside a synchronized block as it may trigger
+ // stdout/stderr and lead to a deadlock. See b/109725432
+ ListenableFuture<PathConverter> localFileUploadFuture =
+ uploadReferencedLocalFiles(event.referencedLocalFiles());
+
+ synchronized (lock) {
+ if (closeFuture != null) {
+ // Close has been called and thus we silently ignore any further events and cancel
+ // any pending file uploads
+ closeFuture.addListener(
+ () -> {
+ if (!localFileUploadFuture.isDone()) {
+ localFileUploadFuture.cancel(true);
+ }
+ },
+ MoreExecutors.directExecutor());
+ return;
+ }
+ // BuildCompletingEvent marks the end of the build in the BEP event stream.
+ if (event instanceof BuildCompletingEvent) {
+ this.buildStatus = extractBuildStatus((BuildCompletingEvent) event);
+ }
+ ensureUploadThreadStarted();
+ eventQueue.addLast(
+ new SendRegularBuildEventCommand(
+ event,
+ localFileUploadFuture,
+ nextSeqNum.getAndIncrement(),
+ Timestamps.fromMillis(clock.currentTimeMillis())));
+ }
+ }
+
+ /**
+ * Gracefully stops the BES upload. All events enqueued before the call to close will be uploaded
+ * and events enqueued after the call will be discarded.
+ *
+ * <p>The returned future completes when the upload completes. It's guaranteed to never fail.
+ */
+ public ListenableFuture<Void> close() {
+ synchronized (lock) {
+ if (closeFuture != null) {
+ return closeFuture;
+ }
+ ensureUploadThreadStarted();
+
+ closeFuture = SettableFuture.create();
+
+ // Enqueue the last event which will terminate the upload.
+ eventQueue.addLast(
+ new SendLastBuildEventCommand(nextSeqNum.getAndIncrement(), currentTime()));
+
+ if (!closeTimeout.isZero()) {
+ startCloseTimer(closeFuture, closeTimeout);
+ }
+ return closeFuture;
+ }
+ }
+
+ /** Stops the upload immediately. Enqueued events that have not been sent yet will be lost. */
+ private void closeOnTimeout() {
+ synchronized (lock) {
+ if (uploadThread != null) {
+ if (uploadThread.isInterrupted()) {
+ return;
+ }
+
+ interruptCausedByTimeout = true;
+ uploadThread.interrupt();
+ }
+ }
+ }
+
+ @Override
+ public void run() {
+ try {
+ if (publishLifecycleEvents) {
+ publishLifecycleEvent(besProtoUtil.buildEnqueued(currentTime()));
+ publishLifecycleEvent(besProtoUtil.invocationStarted(currentTime()));
+ }
+
+ try {
+ publishBuildEvents();
+ } finally {
+ if (publishLifecycleEvents) {
+ Result buildStatus;
+ synchronized (lock) {
+ buildStatus = this.buildStatus;
+ }
+ publishLifecycleEvent(besProtoUtil.invocationFinished(currentTime(), buildStatus));
+ publishLifecycleEvent(besProtoUtil.buildFinished(currentTime(), buildStatus));
+ }
+ }
+ exitFunc.accept(
+ "The Build Event Protocol upload finished successfully",
+ /*cause=*/ null,
+ ExitCode.SUCCESS);
+ synchronized (lock) {
+ // Invariant: closeFuture is not null.
+ // publishBuildEvents() only terminates successfully after SendLastBuildEventCommand
+ // has been sent successfully and that event is only added to the eventQueue during a
+ // call to close() which initializes the closeFuture.
+ closeFuture.set(null);
+ }
+ } catch (InterruptedException e) {
+ try {
+ logger.info("Aborting the BES upload due to having received an interrupt");
+ synchronized (lock) {
+ Preconditions.checkState(
+ interruptCausedByTimeout, "Unexpected interrupt on BES uploader thread");
+ exitFunc.accept(
+ "The Build Event Protocol upload timed out",
+ e,
+ ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR);
+ }
+ } finally {
+ // TODO(buchgr): Due to b/113035235 exitFunc needs to be called before the close future
+ // completes.
+ failCloseFuture(e);
+ }
+ } catch (StatusException e) {
+ try {
+ String message =
+ "The Build Event Protocol upload failed: " + besClient.userReadableError(e);
+ logger.info(message);
+ ExitCode code =
+ shouldRetryStatus(e.getStatus())
+ ? ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR
+ : ExitCode.PERSISTENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR;
+ exitFunc.accept(message, e, code);
+ } finally {
+ failCloseFuture(e);
+ }
+ } catch (LocalFileUploadException e) {
+ Throwables.throwIfUnchecked(e.getCause());
+ try {
+ String message =
+ "The Build Event Protocol local file upload failed: " + e.getCause().getMessage();
+ logger.info(message);
+ exitFunc.accept(message, e.getCause(), ExitCode.TRANSIENT_BUILD_EVENT_SERVICE_UPLOAD_ERROR);
+ } finally {
+ failCloseFuture(e.getCause());
+ }
+ } catch (Throwable e) {
+ failCloseFuture(e);
+ logger.severe("BES upload failed due to a RuntimeException / Error. This is a bug.");
+ throw e;
+ } finally {
+ localFileUploader.shutdown();
+ }
+ }
+
+ private BuildEventStreamProtos.BuildEvent createSerializedRegularBuildEvent(
+ PathConverter pathConverter,
+ SendRegularBuildEventCommand buildEvent) {
+ BuildEventContext ctx =
+ new BuildEventContext() {
+ @Override
+ public PathConverter pathConverter() {
+ return pathConverter;
+ }
+
+ @Override
+ public ArtifactGroupNamer artifactGroupNamer() {
+ return namer;
+ }
+
+ @Override
+ public BuildEventProtocolOptions getOptions() {
+ return buildEventProtocolOptions;
+ }
+ };
+ BuildEventStreamProtos.BuildEvent serializedBepEvent =
+ buildEvent.getEvent().asStreamProto(ctx);
+ buildEventLogger.log(serializedBepEvent);
+ return serializedBepEvent;
+ }
+
+ private void publishBuildEvents()
+ throws StatusException, LocalFileUploadException, InterruptedException {
+ eventQueue.addFirst(new OpenStreamCommand());
+
+ // Every build event sent to the server needs to be acknowledged by it. This queue stores
+ // the build events that have been sent and still have to be acknowledged by the server.
+ // The build events are stored in the order they were sent.
+ ConcurrentLinkedDeque<SendBuildEventCommand> ackQueue = new ConcurrentLinkedDeque<>();
+ boolean lastEventSent = false;
+ int acksReceived = 0;
+ int retryAttempt = 0;
+
+ try {
+ // {@link BuildEventServiceUploaderCommands#OPEN_STREAM} is the first event and opens a
+ // bidi streaming RPC for sending build events and receiving ACKs.
+ // {@link BuildEventServiceUploaderCommands#SEND_REGULAR_BUILD_EVENT} sends a build event to
+ // the server. Sending of the Nth build event does
+ // does not wait for the ACK of the N-1th build event to have been received.
+ // {@link BuildEventServiceUploaderCommands#SEND_LAST_BUILD_EVENT} sends the last build event
+ // and half closes the RPC.
+ // {@link BuildEventServiceUploaderCommands#ACK_RECEIVED} is executed for every ACK from
+ // the server and checks that the ACKs are in the correct order.
+ // {@link BuildEventServiceUploaderCommands#STREAM_COMPLETE} checks that all build events
+ // have been sent and all ACKs have been received. If not it invokes a retry logic that may
+ // decide to re-send every build event for which an ACK has not been received. If so, it
+ // adds an OPEN_STREAM event.
+ while (true) {
+ EventLoopCommand event = eventQueue.takeFirst();
+ switch (event.type()) {
+ case OPEN_STREAM:
+ {
+ // Invariant: the eventQueue only contains events of type SEND_REGULAR_BUILD_EVENT
+ // or SEND_LAST_BUILD_EVENT
+ logger.info(
+ String.format("Starting publishBuildEvents: eventQueue=%d", eventQueue.size()));
+ streamContext =
+ besClient.openStream(
+ (ack) -> eventQueue.addLast(new AckReceivedCommand(ack.getSequenceNumber())));
+ addStreamStatusListener(
+ streamContext.getStatus(),
+ (status) -> eventQueue.addLast(new StreamCompleteCommand(status)));
+ }
+ break;
+
+ case SEND_REGULAR_BUILD_EVENT:
+ {
+ // Invariant: the eventQueue may contain events of any type
+ SendRegularBuildEventCommand buildEvent = (SendRegularBuildEventCommand) event;
+ ackQueue.addLast(buildEvent);
+
+ PathConverter pathConverter = waitForLocalFileUploads(buildEvent);
+
+ BuildEventStreamProtos.BuildEvent serializedRegularBuildEvent =
+ createSerializedRegularBuildEvent(pathConverter, buildEvent);
+
+ PublishBuildToolEventStreamRequest request =
+ besProtoUtil.bazelEvent(
+ buildEvent.getSequenceNumber(),
+ buildEvent.getCreationTime(),
+ Any.pack(serializedRegularBuildEvent));
+
+ streamContext.sendOverStream(request);
+ }
+ break;
+
+ case SEND_LAST_BUILD_EVENT:
+ {
+ // Invariant: the eventQueue may contain events of any type
+ SendBuildEventCommand lastEvent = (SendLastBuildEventCommand) event;
+ ackQueue.addLast(lastEvent);
+ lastEventSent = true;
+ PublishBuildToolEventStreamRequest request =
+ besProtoUtil.streamFinished(
+ lastEvent.getSequenceNumber(), lastEvent.getCreationTime());
+ streamContext.sendOverStream(request);
+ streamContext.halfCloseStream();
+ }
+ break;
+
+ case ACK_RECEIVED:
+ {
+ // Invariant: the eventQueue may contain events of any type
+ AckReceivedCommand ackEvent = (AckReceivedCommand) event;
+ if (!ackQueue.isEmpty()) {
+ SendBuildEventCommand expected = ackQueue.removeFirst();
+ long actualSeqNum = ackEvent.getSequenceNumber();
+ if (expected.getSequenceNumber() == actualSeqNum) {
+ acksReceived++;
+ } else {
+ ackQueue.addFirst(expected);
+ String message =
+ String.format(
+ "Expected ACK with seqNum=%d but received ACK with seqNum=%d",
+ expected.getSequenceNumber(), actualSeqNum);
+ logger.info(message);
+ streamContext.abortStream(Status.FAILED_PRECONDITION.withDescription(message));
+ }
+ } else {
+ String message =
+ String.format(
+ "Received ACK (seqNum=%d) when no ACK was expected",
+ ackEvent.getSequenceNumber());
+ logger.info(message);
+ streamContext.abortStream(Status.FAILED_PRECONDITION.withDescription(message));
+ }
+ }
+ break;
+
+ case STREAM_COMPLETE:
+ {
+ // Invariant: the eventQueue only contains events of type SEND_REGULAR_BUILD_EVENT
+ // or SEND_LAST_BUILD_EVENT
+ streamContext = null;
+ StreamCompleteCommand completeEvent = (StreamCompleteCommand) event;
+ Status streamStatus = completeEvent.status();
+ if (streamStatus.isOk()) {
+ if (lastEventSent && ackQueue.isEmpty()) {
+ logger.info("publishBuildEvents was successful");
+ // Upload successful. Break out from the while(true) loop.
+ return;
+ } else {
+ throw (lastEventSent
+ ? ackQueueNotEmptyStatus(ackQueue.size())
+ : lastEventNotSentStatus())
+ .asException();
+ }
+ }
+
+ if (!shouldRetryStatus(streamStatus)) {
+ logger.info(
+ String.format("Not retrying publishBuildEvents: status='%s'", streamStatus));
+ throw streamStatus.asException();
+ }
+ if (retryAttempt == MAX_NUM_RETRIES) {
+ logger.info(
+ String.format(
+ "Not retrying publishBuildEvents, no more attempts left: status='%s'",
+ streamStatus));
+ throw streamStatus.asException();
+ }
+
+ // Retry logic
+ // Adds events from the ackQueue to the front of the eventQueue, so that the
+ // events in the eventQueue are sorted by sequence number (ascending).
+ SendBuildEventCommand unacked;
+ while ((unacked = ackQueue.pollLast()) != null) {
+ eventQueue.addFirst(unacked);
+ }
+
+ long sleepMillis = retrySleepMillis(retryAttempt);
+ logger.info(
+ String.format(
+ "Retrying stream: status='%s', sleepMillis=%d", streamStatus, sleepMillis));
+ sleeper.sleepMillis(sleepMillis);
+
+ // If we made progress, meaning the server ACKed events that we sent, then reset
+ // the retry counter to 0.
+ if (acksReceived > 0) {
+ retryAttempt = 0;
+ } else {
+ retryAttempt++;
+ }
+ acksReceived = 0;
+ eventQueue.addFirst(new OpenStreamCommand());
+ }
+ break;
+ }
+ }
+ } catch (InterruptedException | LocalFileUploadException e) {
+ int limit = 30;
+ logger.info(
+ String.format(
+ "Publish interrupt. Showing up to %d items from queues: ack_queue_size: %d, "
+ + "ack_queue: %s, event_queue_size: %d, event_queue: %s",
+ limit,
+ ackQueue.size(),
+ Iterables.limit(ackQueue, limit),
+ eventQueue.size(),
+ Iterables.limit(eventQueue, limit)));
+ if (streamContext != null) {
+ streamContext.abortStream(Status.CANCELLED);
+ }
+ throw e;
+ } finally {
+ // Cancel all local file uploads that may still be running
+ // of events that haven't been uploaded.
+ EventLoopCommand event;
+ while ((event = ackQueue.pollFirst()) != null) {
+ if (event instanceof SendRegularBuildEventCommand) {
+ cancelLocalFileUpload((SendRegularBuildEventCommand) event);
+ }
+ }
+ while ((event = eventQueue.pollFirst()) != null) {
+ if (event instanceof SendRegularBuildEventCommand) {
+ cancelLocalFileUpload((SendRegularBuildEventCommand) event);
+ }
+ }
+ }
+ }
+
+ private void cancelLocalFileUpload(SendRegularBuildEventCommand event) {
+ ListenableFuture<PathConverter> localFileUploaderFuture = event.localFileUploadProgress();
+ if (!localFileUploaderFuture.isDone()) {
+ localFileUploaderFuture.cancel(true);
+ }
+ }
+
+ /** Sends a {@link PublishLifecycleEventRequest} to the BES backend. */
+ private void publishLifecycleEvent(PublishLifecycleEventRequest request)
+ throws StatusException, InterruptedException {
+ int retryAttempt = 0;
+ StatusException cause = null;
+ while (retryAttempt <= MAX_NUM_RETRIES) {
+ try {
+ besClient.publish(request);
+ return;
+ } catch (StatusException e) {
+ if (!shouldRetryStatus(e.getStatus())) {
+ logger.info(
+ String.format(
+ "Not retrying publishLifecycleEvent: status='%s'", e.getStatus().toString()));
+ throw e;
+ }
+
+ cause = e;
+
+ long sleepMillis = retrySleepMillis(retryAttempt);
+ logger.info(
+ String.format(
+ "Retrying publishLifecycleEvent: status='%s', sleepMillis=%d",
+ e.getStatus().toString(), sleepMillis));
+ sleeper.sleepMillis(sleepMillis);
+ retryAttempt++;
+ }
+ }
+
+ // All retry attempts failed
+ throw cause;
+ }
+
+ private ListenableFuture<PathConverter> uploadReferencedLocalFiles(
+ Collection<LocalFile> localFiles) {
+ Map<Path, LocalFile> localFileMap = new TreeMap<>();
+ for (LocalFile localFile : localFiles) {
+ // It is possible for targets to have duplicate artifacts (same path but different owners)
+ // in their output groups. Since they didn't trigger an artifact conflict they are the
+ // same file, so just skip either one
+ localFileMap.putIfAbsent(localFile.path, localFile);
+ }
+ return localFileUploader.upload(localFileMap);
+ }
+
+ private void ensureUploadThreadStarted() {
+ synchronized (lock) {
+ if (uploadThread == null) {
+ uploadThread = new Thread(this, "bes-uploader");
+ uploadThread.start();
+ }
+ }
+ }
+
+ private void startCloseTimer(ListenableFuture<Void> closeFuture, Duration closeTimeout) {
+ Thread closeTimer =
+ new Thread(
+ () -> {
+ // Call closeOnTimeout() if the future does not complete within closeTimeout
+ try {
+ getUninterruptibly(closeFuture, closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ closeOnTimeout();
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof TimeoutException) {
+ // This is likely due to an internal timeout doing the local file uploading.
+ closeOnTimeout();
+ } else {
+ // This code only cares about calling closeOnTimeout() if the closeFuture does
+ // not complete within closeTimeout.
+ String failureMsg = "BES close failure";
+ logger.severe(failureMsg);
+ LoggingUtil.logToRemote(Level.SEVERE, failureMsg, e);
+ }
+ }
+ },
+ "bes-uploader-close-timer");
+ closeTimer.start();
+ }
+
+ private void failCloseFuture(Throwable cause) {
+ synchronized (lock) {
+ if (closeFuture == null) {
+ closeFuture = SettableFuture.create();
+ }
+ closeFuture.setException(cause);
+ }
+ }
+
+ private PathConverter waitForLocalFileUploads(SendRegularBuildEventCommand orderedBuildEvent)
+ throws LocalFileUploadException, InterruptedException {
+ try {
+ // Wait for the local file upload to have been completed.
+ return orderedBuildEvent.localFileUploadProgress().get();
+ } catch (ExecutionException e) {
+ logger.log(
+ Level.WARNING,
+ String.format(
+ "Failed to upload local files referenced by build event: %s", e.getMessage()),
+ e);
+ Throwables.throwIfUnchecked(e.getCause());
+ throw new LocalFileUploadException(e.getCause());
+ }
+ }
+
+ private Timestamp currentTime() {
+ return Timestamps.fromMillis(clock.currentTimeMillis());
+ }
+
+ private static Result extractBuildStatus(BuildCompletingEvent event) {
+ if (event.getExitCode() != null && event.getExitCode().getNumericExitCode() == 0) {
+ return COMMAND_SUCCEEDED;
+ } else {
+ return COMMAND_FAILED;
+ }
+ }
+
+ private static Status lastEventNotSentStatus() {
+ return Status.FAILED_PRECONDITION.withDescription(
+ "Server closed stream with status OK but not all events have been sent");
+ }
+
+ private static Status ackQueueNotEmptyStatus(int ackQueueSize) {
+ return Status.FAILED_PRECONDITION.withDescription(
+ String.format(
+ "Server closed stream with status OK but not all ACKs have been"
+ + " received (ackQueue=%d)",
+ ackQueueSize));
+ }
+
+ private static void addStreamStatusListener(
+ ListenableFuture<Status> stream, Consumer<Status> onDone) {
+ Futures.addCallback(
+ stream,
+ new FutureCallback<Status>() {
+ @Override
+ public void onSuccess(Status result) {
+ onDone.accept(result);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {}
+ },
+ MoreExecutors.directExecutor());
+ }
+
+ private static boolean shouldRetryStatus(Status status) {
+ return !status.getCode().equals(Code.INVALID_ARGUMENT)
+ && !status.getCode().equals(Code.FAILED_PRECONDITION);
+ }
+
+ private static long retrySleepMillis(int attempt) {
+ // This somewhat matches the backoff used for gRPC connection backoffs.
+ return (long) (DELAY_MILLIS * Math.pow(1.6, attempt));
+ }
+
+ /**
+ * This method is only used in tests. Once TODO(b/113035235) is fixed the close future will also
+ * carry error messages.
+ */
+ @VisibleForTesting // productionVisibility = Visibility.PRIVATE
+ public void throwUploaderError() throws Throwable {
+ synchronized (lock) {
+ checkState(closeFuture != null && closeFuture.isDone());
+ try {
+ closeFuture.get();
+ } catch (ExecutionException e) {
+ throw e.getCause();
+ }
+ }
+ }
+
+ /** Thrown when encountered problems while uploading build event artifacts. */
+ private class LocalFileUploadException extends Exception {
+ LocalFileUploadException(Throwable cause) {
+ super(cause);
+ }
+ }
+}
+
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploaderCommands.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploaderCommands.java
new file mode 100644
index 0000000..4934284
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceUploaderCommands.java
@@ -0,0 +1,181 @@
+// Copyright 2019 The Bazel Authors. All rights reserved.
+//
+// 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 com.google.devtools.build.lib.buildeventservice;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.devtools.build.lib.buildeventstream.BuildEvent;
+import com.google.devtools.build.lib.buildeventstream.PathConverter;
+import com.google.protobuf.Timestamp;
+import io.grpc.Status;
+import javax.annotation.concurrent.Immutable;
+
+/** Commands that can be used in the event loop implemented by {@link BuildEventServiceUploader}. */
+public class BuildEventServiceUploaderCommands {
+ /** A command that may be added to the {@code BuildEventServiceUploader#eventQueue}. */
+ interface EventLoopCommand {
+ enum Type {
+ /** Tells the event loop to open a new BES stream */
+ OPEN_STREAM,
+ /** Tells the event loop to send the build event */
+ SEND_REGULAR_BUILD_EVENT,
+ /** Tells the event loop that an ACK was received */
+ ACK_RECEIVED,
+ /** Tells the event loop that this is the last event of the stream */
+ SEND_LAST_BUILD_EVENT,
+ /** Tells the event loop that the streaming RPC completed */
+ STREAM_COMPLETE
+ }
+
+ Type type();
+ }
+
+ /** Implementation of {@link EventLoopCommand.Type#OPEN_STREAM}. */
+ @Immutable
+ static final class OpenStreamCommand implements EventLoopCommand {
+ @Override
+ public Type type() {
+ return Type.OPEN_STREAM;
+ }
+ }
+
+ /** Implementation of {@link EventLoopCommand.Type#STREAM_COMPLETE}. */
+ @Immutable
+ static final class StreamCompleteCommand implements EventLoopCommand {
+ private final Status status;
+
+ StreamCompleteCommand(Status status) {
+ this.status = status;
+ }
+
+ public Status status() {
+ return status;
+ }
+
+ @Override
+ public Type type() {
+ return Type.STREAM_COMPLETE;
+ }
+ }
+
+ /** Implementation of {@link EventLoopCommand.Type#ACK_RECEIVED}. */
+ @Immutable
+ static final class AckReceivedCommand implements EventLoopCommand {
+ private final long sequenceNumber;
+
+ AckReceivedCommand(long sequenceNumber) {
+ this.sequenceNumber = sequenceNumber;
+ }
+
+ public long getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ @Override
+ public Type type() {
+ return Type.ACK_RECEIVED;
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this).add("seq_num", getSequenceNumber()).toString();
+ }
+ }
+
+ /** Common interface for {@link EventLoopCommand}s related to builds. */
+ abstract static class SendBuildEventCommand implements EventLoopCommand {
+ abstract long getSequenceNumber();
+
+ abstract Timestamp getCreationTime();
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this).add("seq_num", getSequenceNumber()).toString();
+ }
+ }
+
+ /** Implementation of {@link EventLoopCommand.Type#SEND_REGULAR_BUILD_EVENT}. */
+ static final class SendRegularBuildEventCommand extends SendBuildEventCommand {
+ private final BuildEvent event;
+ private final ListenableFuture<PathConverter> localFileUpload;
+ private final long sequenceNumber;
+ private final Timestamp creationTime;
+
+ SendRegularBuildEventCommand(
+ BuildEvent event,
+ ListenableFuture<PathConverter> localFileUpload,
+ long sequenceNumber,
+ Timestamp creationTime) {
+ this.event = event;
+ this.localFileUpload = localFileUpload;
+ this.sequenceNumber = sequenceNumber;
+ this.creationTime = creationTime;
+ }
+
+ BuildEvent getEvent() {
+ return event;
+ }
+
+ ListenableFuture<PathConverter> localFileUploadProgress() {
+ return localFileUpload;
+ }
+
+ public long getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ @Override
+ Timestamp getCreationTime() {
+ return creationTime;
+ }
+
+ @Override
+ public Type type() {
+ return Type.SEND_REGULAR_BUILD_EVENT;
+ }
+
+ @Override
+ public String toString() {
+ return super.toString() + " - [" + event + "]";
+ }
+ }
+
+ /** Implementation of {@link EventLoopCommand.Type#SEND_LAST_BUILD_EVENT}. */
+ @Immutable
+ static final class SendLastBuildEventCommand extends SendBuildEventCommand {
+ private final long sequenceNumber;
+ private final Timestamp creationTime;
+
+ SendLastBuildEventCommand(long sequenceNumber, Timestamp creationTime) {
+ this.sequenceNumber = sequenceNumber;
+ this.creationTime = creationTime;
+ }
+
+ @Override
+ Timestamp getCreationTime() {
+ return creationTime;
+ }
+
+ @Override
+ public Type type() {
+ return Type.SEND_LAST_BUILD_EVENT;
+ }
+
+ @Override
+ public long getSequenceNumber() {
+ return sequenceNumber;
+ }
+ }
+
+}
