feat(flagd): introduce fatalStatusCodes option #1624
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,5 +3,6 @@ | |
| /** Payload type emitted by {@link QueueSource}. */ | ||
| public enum QueuePayloadType { | ||
| DATA, | ||
| ERROR, | ||
| FATAL | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,6 +19,7 @@ | |
| import io.grpc.Status; | ||
| import io.grpc.StatusRuntimeException; | ||
| import io.grpc.stub.StreamObserver; | ||
| import java.util.List; | ||
| import java.util.concurrent.BlockingQueue; | ||
| import java.util.concurrent.LinkedBlockingQueue; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
@@ -38,6 +39,7 @@ public class SyncStreamQueueSource implements QueueSource { | |
|
|
||
| private final AtomicBoolean shutdown = new AtomicBoolean(false); | ||
| private final AtomicBoolean shouldThrottle = new AtomicBoolean(false); | ||
| private final AtomicBoolean successfulSync = new AtomicBoolean(false); | ||
|
Member
There was a problem hiding this comment. My biggest question with this whole concept (not your implementation) is whether or not we should care about whether this is the initial sync or not. I'm actually leaning towards "not"... here is my reasoning (anyone feel free to disagree):
WDYT?
Member
Author
There was a problem hiding this comment. If it's easy to get the same behaviour through event handlers, I think that might be better, because it allows for more customization. I get both sides, that one might not want to completely shut down if a valid flag config was previously received, but also that one might not want to work with stale data given that a non-transient error was received |
||
| private final int streamDeadline; | ||
| private final int deadline; | ||
| private final int maxBackoffMs; | ||
|
|
@@ -47,6 +49,7 @@ public class SyncStreamQueueSource implements QueueSource { | |
| private final boolean reinitializeOnError; | ||
| private final FlagdOptions options; | ||
| private final BlockingQueue<QueuePayload> outgoingQueue = new LinkedBlockingQueue<>(QUEUE_SIZE); | ||
| private final List<String> fatalStatusCodes; | ||
|
Contributor
There was a problem hiding this comment. Since we do lots of
Member
Author
There was a problem hiding this comment. That's hard for me to estimate, what do the others think? The currently defined default is an empty list |
||
| private volatile GrpcComponents grpcComponents; | ||
|
|
||
| /** | ||
|
|
@@ -77,6 +80,7 @@ public SyncStreamQueueSource(final FlagdOptions options) { | |
| providerId = options.getProviderId(); | ||
| maxBackoffMs = options.getRetryBackoffMaxMs(); | ||
| syncMetadataDisabled = options.isSyncMetadataDisabled(); | ||
| fatalStatusCodes = options.getFatalStatusCodes(); | ||
| reinitializeOnError = options.isReinitializeOnError(); | ||
| this.options = options; | ||
| initializeChannelComponents(); | ||
|
|
@@ -94,6 +98,7 @@ protected SyncStreamQueueSource( | |
| providerId = options.getProviderId(); | ||
| maxBackoffMs = options.getRetryBackoffMaxMs(); | ||
| syncMetadataDisabled = options.isSyncMetadataDisabled(); | ||
| fatalStatusCodes = options.getFatalStatusCodes(); | ||
| reinitializeOnError = options.isReinitializeOnError(); | ||
| this.options = options; | ||
| this.grpcComponents = new GrpcComponents(connectorMock, stubMock, blockingStubMock); | ||
|
|
@@ -185,23 +190,42 @@ private void observeSyncStream() { | |
| } | ||
|
|
||
| log.debug("Initializing sync stream request"); | ||
| SyncStreamObserver observer = new SyncStreamObserver(outgoingQueue, shouldThrottle, fatalStatusCodes); | ||
| try { | ||
| observer.metadata = getMetadata(); | ||
| } catch (StatusRuntimeException metaEx) { | ||
| if (fatalStatusCodes.contains(metaEx.getStatus().getCode().name()) && !successfulSync.get()) { | ||
| log.debug("Fatal status code for metadata request: {}, not retrying", | ||
| metaEx.getStatus().getCode()); | ||
| enqueueFatal(String.format( | ||
| "Fatal: Failed to connect for metadata request, not retrying for error %s", | ||
| metaEx.getStatus().getCode())); | ||
| return; | ||
| } else { | ||
| // retry for other status codes | ||
| String message = metaEx.getMessage(); | ||
| log.debug("Metadata request error: {}, will restart", message, metaEx); | ||
| enqueueError(String.format("Error in getMetadata request: %s", message)); | ||
| } | ||
| shouldThrottle.set(true); | ||
| continue; | ||
| } | ||
|
|
||
| try { | ||
| syncFlags(observer); | ||
| successfulSync.set(true); | ||
| } catch (StatusRuntimeException ex) { | ||
| if (fatalStatusCodes.contains(ex.getStatus().getCode().toString()) && !successfulSync.get()) { | ||
| log.debug("Fatal status code during sync stream: {}, not retrying", ex.getStatus().getCode()); | ||
| enqueueFatal(String.format( | ||
| "Fatal: Failed to connect for metadata request, not retrying for error %s", | ||
| ex.getStatus().getCode())); | ||
| return; | ||
| } else { | ||
| // retry for other status codes | ||
| log.error("Unexpected sync stream exception, will restart.", ex); | ||
| enqueueError(String.format("Error in syncStream: %s", ex.getMessage())); | ||
| } | ||
| shouldThrottle.set(true); | ||
| } | ||
| } catch (InterruptedException ie) { | ||
|
|
@@ -276,16 +300,29 @@ private static void enqueueError(BlockingQueue<QueuePayload> queue, String messa | |
| } | ||
| } | ||
|
|
||
| private void enqueueFatal(String message) { | ||
| enqueueFatal(outgoingQueue, message); | ||
| } | ||
|
|
||
| private static void enqueueFatal(BlockingQueue<QueuePayload> queue, String message) { | ||
| if (!queue.offer(new QueuePayload(QueuePayloadType.FATAL, message, null))) { | ||
| log.error("Failed to convey FATAL status, queue is full"); | ||
| } | ||
| } | ||
|
|
||
| private static class SyncStreamObserver implements StreamObserver<SyncFlagsResponse> { | ||
| private final BlockingQueue<QueuePayload> outgoingQueue; | ||
| private final AtomicBoolean shouldThrottle; | ||
| private final Awaitable done = new Awaitable(); | ||
| private final List<String> fatalStatusCodes; | ||
|
|
||
| private Struct metadata; | ||
|
|
||
| public SyncStreamObserver(BlockingQueue<QueuePayload> outgoingQueue, AtomicBoolean shouldThrottle, | ||
| List<String> fatalStatusCodes) { | ||
| this.outgoingQueue = outgoingQueue; | ||
| this.shouldThrottle = shouldThrottle; | ||
| this.fatalStatusCodes = fatalStatusCodes; | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -303,9 +340,14 @@ public void onNext(SyncFlagsResponse syncFlagsResponse) { | |
| @Override | ||
| public void onError(Throwable throwable) { | ||
| try { | ||
| Status status = Status.fromThrowable(throwable); | ||
| String message = throwable != null ? throwable.getMessage() : "unknown"; | ||
| log.debug("Stream error: {}, will restart", message, throwable); | ||
| if (fatalStatusCodes.contains(status.getCode().name())) { | ||
| enqueueFatal(outgoingQueue, String.format("Error from stream: %s", message)); | ||
| } else { | ||
| enqueueError(outgoingQueue, String.format("Error from stream: %s", message)); | ||
| } | ||
|
|
||
| // Set throttling flag to ensure backoff before retry | ||
| this.shouldThrottle.set(true); | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should print an info/warn that the env vars are invalid
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Just for this method? Or the other ones too? I'd either leave it or add it in all cases to be consistent
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Then we should add it everywhere, but in a different PR
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Alright, sounds good. Should we create a new issue for this or is that overkill?