Allow diagnostics endpoints command to receive more than one message #13285
Conversation
This file contains 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
Signed-off-by: Alex Leong <[email protected]>
cli/cmd/endpoints.go
Outdated
Comment on lines
167
to
186
| event, err := rsp.Recv() | ||
| if err != nil { | ||
| if grpcError, ok := status.FromError(err); ok { | ||
| err = errors.New(grpcError.Message()) | ||
| // Endpoint state may be sent in multiple messages so it's not | ||
| // sufficient to read only the first message. Instead, we | ||
| // continuously read from the stream. This goroutine will never | ||
| // terminate if there are no errors, but this is okay for a | ||
| // short lived CLI command. | ||
| for { | ||
| event, err := rsp.Recv() | ||
| if errors.Is(err, io.EOF) { | ||
| return | ||
| } else if err != nil { | ||
| if grpcError, ok := status.FromError(err); ok { | ||
| err = errors.New(grpcError.Message()) | ||
| } | ||
| errs <- err | ||
| return | ||
| } | ||
| errs <- err | ||
| return | ||
| events <- event | ||
| } | ||
| events <- event | ||
| } | ||
| }(authority) | ||
| } | ||
| // Block till all goroutines above are done | ||
| wg.Wait() | ||
| // Wait an amount of time for some endpoint responses to be received. | ||
| time.Sleep(5 * time.Second) |
There was a problem hiding this comment.
My gut feeling is that this work group is fine here. I think depending on time.Sleep has one problem that you might encounter - if you hit an error while consuming from the stream, you will fail to report it right away but would still need to wait 5 seconds. Here is an alternative approach.
- Keep the workgroup
- Pass the worklgroup to the streaming goroutine along with a context with a timeout of 5 seconds.
- Keep reading in this routine until you hit an error or the context elapses (need to do a select over the stream chan and the context expiration chan)
- Whatever happens (error or context elapsed), before exiting the goroutine call Done on the workgroup.
- Keep the rest of the logic the same, wait for the workgroup, then read the errors and events.
There was a problem hiding this comment.
Unfortunately, I don't think that will work because the stream is not a chan; rsp.Recv() is a synchronous blocking call.
Instead, I've replaced the Sleep with a Timer so that we can select over the timeout, the messages, and the errors. This lets us exit early if there is an error instead of needing to wait the entire timeout.
Signed-off-by: Alex Leong <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
The
linkerd diagnostics endpointscommand initiates aGetlookup to the destination controller to get the set of endpoints for a destination. This is a streaming response API and the command takes only the first response message and displays it. However, the full current state of endpoints may be split across multiple messages, resulting in an incomplete list of endpoints displayed.We instead read continuously from the response stream for a short amount of time (5 seconds) before displaying the full set of endpoints received.