Synchronize shutdown in stanza adapter #34638
Conversation
There was a problem hiding this comment.
These appear to be nice improvements. Ranging over channels is a lot cleaner. Having a channel / wg for each function is much more precise.
One thing I'm wondering about is the use of context.Background(). I wonder if this indicates there are still issues to iron out. What is your take on this @michalpristas?
Also would really appreciate a review by @swiatekm if possible.
|
with context.Background used i'm fine with how it is signalling |
|
This PR was marked stale due to lack of activity. It will be closed in 14 days. |
|
Unless anyone else wants to review this, I think we just need to resolve the CI failures |
|
@djaglowski I'm going to take this up and get it over the finish line, as @michalpristas is currently on extended leave. I reviewed the change and I do think it fixes the problem it claims to fix. Some parts are still a bit awkward (I think we should refactor the emitter to just use a channel for signalling finalization instead of messing around with contexts), but those can be addressed in a follow up imo. Since you already reviewed this change, how about I just rebase and fix any remaining CI issues? |
Co-authored-by: Daniel Jaglowski <jaglows3@gmail.com>
Co-authored-by: Daniel Jaglowski <jaglows3@gmail.com>
swiatekm
force-pushed
the
fix/shutdown
branch
from
1094d8f
to
edb0639
Compare
After the shutdown changes in previous commits, a receiver can only shut down after its emitter was shut down first.
swiatekm
force-pushed
the
fix/shutdown
branch
from
edb0639
to
489b6d7
Compare
|
Thank you @swiatekm! |
| @@ -151,7 +151,10 @@ func TestEmitterToConsumer(t *testing.T) { | |||
|
|
|||
| err = logsReceiver.Start(context.Background(), componenttest.NewNopHost()) | |||
| require.NoError(t, err) | |||
| defer func() { require.NoError(t, logsReceiver.Shutdown(context.Background())) }() | |||
| defer func() { | |||
| require.NoError(t, logsReceiver.emitter.Stop()) | |||
There was a problem hiding this comment.
Note: This is now necessary, because the receiver is not allowed to stop until the emitter does. In an actual pipeline, the emitter itself is owned by the pipeline, and the pipeline is responsible for closing it. Here, we need to do it manually.
|
@djaglowski could you re-review? It's Michal's PR, so I don't have the power to re-request it. 😅 |
|
Thanks again @michalpristas and @swiatekm. It'll be great to have this fix in place. I'm happy to review a followup PR if there's any cleanup in mind. |
| } | ||
| case <-ctx.Done(): | ||
| // flush currently batched entries | ||
| for oldBatch := e.makeNewBatch(); len(oldBatch) > 0; oldBatch = e.makeNewBatch() { |
There was a problem hiding this comment.
Any specific scenario that warrants using a for instead of doing it just once? I would assume that at the point when LogEmitter.Stop is called the rest of the Stanza pipeline is stopped, so no new entries should be landing in the emitter's batch.
Not that the for hurts, it will work "more than" fine. 🙂 I just wonder if I'm missing something.
There was a problem hiding this comment.
I think it's not necessary in practice, because other components shouldn't be calling Process after Stop was already called. And this is how the pipeline acts. @djaglowski am I wrong here?
Clean up emitter shutdown in stanza. Instead of using contexts for signaling shutdown, use an idiomatic close channel. We still have a bit of context usage, but that's required by the operator API, and is only used for cancelling a given entry, not for component shutdown. Follow up to #34638.
**Description:** This PR takes emitter-converter-receiver pipeline and synchronize shutdown in a sense that lower level of the pipeline needs to be fully finished before next one is shut down. **Link to tracking Issue:** open-telemetry#31074 **Testing:** UT run ``` go test -run ^TestShutdownFlush$ . -count 10000 -failfast ``` **Documentation:** in code comments --------- Co-authored-by: Daniel Jaglowski <jaglows3@gmail.com> Co-authored-by: Mikołaj Świątek <mail@mikolajswiatek.com>
Clean up emitter shutdown in stanza. Instead of using contexts for signaling shutdown, use an idiomatic close channel. We still have a bit of context usage, but that's required by the operator API, and is only used for cancelling a given entry, not for component shutdown. Follow up to open-telemetry#34638.
Description:
This PR takes emitter-converter-receiver pipeline and synchronize shutdown in a sense that lower level of the pipeline needs to be fully finished before next one is shut down.
Link to tracking Issue: #31074
Testing: UT
run
Documentation: in code comments