fix(splunk_hec source): Flush messages #5069

jszwedko · 2020-11-16T19:06:27Z

#4875 introduced buffering
during Pipeline.start_send. Events were sent to the inner sink only on
when start_send was called again or poll_complete is called.
A side-effect of this was that when using Sink.send_all and
Stream.forward, if an error was returned on a poll() of the stream
being sent, the previously written event would never be sent to the
inner sink as poll_complete() was never called.

futures01 Sink.send_all:
https://docs.rs/futures/0.1.30/src/futures/sink/send_all.rs.html#59-88
futures01 Stream.forward:
https://docs.rs/futures/0.1.30/src/futures/stream/forward.rs.html#78-110

We can see in both cases it will return early if there is an error. If
the stream is not ready or is finished, then it will call
poll_complete / close on the sink.

We use an errorable futures01 stream for the splunk_hec source:
https://github.com/timberio/vector/blob/3bd8076be4097e4e9cf903a4f32e34a8b145a339/src/sources/splunk_hec.rs#L175-L189
which meant that, if there was an error processing a single event, the
previous event was never flushed downstream.

This adds a Drop implementation for Pipeline that flushes via
close(). I'm not 100% on this idea over requiring usages of
Pipelines that use errorable streams to explicitly call .close(),
but pushing up to get feedback.

I also implement close to ensure that the inner sink also has
close() called on it.

Signed-off-by: Jesse Szwedko jesse@szwedko.me

Fixes #5058 #4875 introduced buffering during `Pipeline.start_send`. Events were sent to the inner sink only on when `start_send` was called again or `poll_complete` is called. A side-effect of this was that when using `Sink.send_all` and `Stream.forward`, if an error was returned on a `poll()` of the stream being sent, the previously written event would never be sent to the inner sink as `poll_complete()` was never called. futures01 `Sink.send_all`: https://docs.rs/futures/0.1.30/src/futures/sink/send_all.rs.html#59-88 futures01 `Stream.forward`: https://docs.rs/futures/0.1.30/src/futures/stream/forward.rs.html#78-110 We can see in both cases it will return early if there is an error. If the stream is not ready or is finished, then it will call `poll_complete` / `close` on the sink. We use an errorable futures01 stream for the splunk_hec source: https://github.com/timberio/vector/blob/3bd8076be4097e4e9cf903a4f32e34a8b145a339/src/sources/splunk_hec.rs#L175-L189 which meant that, if there was an error processing a single event, the previous event was never flushed downstream. This adds a `Drop` implementation for `Pipeline` that flushes via `close()`. I'm not 100% on this idea over requiring usages of `Pipeline`s that use errorable streams to explicitly call `.close()`, but pushing up to get feedback. I also implement `close` to ensure that the inner sink also has `close()` called on it. Signed-off-by: Jesse Szwedko <jesse@szwedko.me>

jszwedko · 2020-11-16T19:06:50Z

src/pipeline.rs

+            match self.close() {
+                Ok(Async::Ready(())) => break,
+                Ok(Async::NotReady) => {}
+                Err(err) => panic!(err),


Thoughts welcome on what to do with errors here.

binarylogic · 2020-11-16T19:19:26Z

Is it possible to test for this? Or is that what the broken test is doing?

Hoverbear · 2020-11-16T19:32:02Z

src/pipeline.rs

@@ -60,6 +60,23 @@ impl Sink for Pipeline {
        debug_assert!(self.enqueued.is_empty());
        self.inner.poll_complete()
    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.poll_complete()?;


Does this change alone fix it by chance?

I had the same thought before; it did not.

lukesteensen · 2020-11-16T19:33:36Z

src/pipeline.rs

+        loop {
+            match self.close() {
+                Ok(Async::Ready(())) => break,
+                Ok(Async::NotReady) => {}


This could turn into a pretty tight loop if there's something inner is waiting on before closing. That plus the error handling question below makes me think we'd be better off requiring the caller to use close than doing this through Drop.

👍 I'm happy to just call close() in this one particular case, but do you have thoughts on ensuring that we call close() everywhere forward or send_all is used?

(my concern is that we'll forget about this again)

We will need to move away from the futures 0.1 send_all/forward methods before too long, so we could add our own method to Pipeline directly that makes sure to flush before returning an error.

Hoverbear · 2020-11-16T20:06:41Z

@binarylogic In this case, this existing test is saving us. :)

Signed-off-by: Jesse Szwedko <jesse@szwedko.me>

jszwedko · 2020-11-16T22:59:51Z

@lukesteensen @Hoverbear I pushed another commit switching this to explicit flushing in the splunk_hec source.

I'll do a survey tomorrow for other places we are using Stream.forward and Sink.send_all to see if this issue exists there as well, but this at least fixes this specific case.

Given the switch to calling flush() directly, I'd say the existing test covers this @binarylogic

jszwedko · 2020-11-17T15:29:43Z

Noting that I'll do the survey separately from this PR; I'd like to merge this to unblock master.

* fix(architecture): Ensure that Pipelines are flushed Fixes #5058 #4875 introduced buffering during `Pipeline.start_send`. Events were sent to the inner sink only on when `start_send` was called again or `poll_complete` is called. A side-effect of this was that when using `Sink.send_all` and `Stream.forward`, if an error was returned on a `poll()` of the stream being sent, the previously written event would never be sent to the inner sink as `poll_complete()` was never called. futures01 `Sink.send_all`: https://docs.rs/futures/0.1.30/src/futures/sink/send_all.rs.html#59-88 futures01 `Stream.forward`: https://docs.rs/futures/0.1.30/src/futures/stream/forward.rs.html#78-110 We can see in both cases it will return early if there is an error. If the stream is not ready or is finished, then it will call `poll_complete` / `close` on the sink. We use an errorable futures01 stream for the splunk_hec source: https://github.com/timberio/vector/blob/3bd8076be4097e4e9cf903a4f32e34a8b145a339/src/sources/splunk_hec.rs#L175-L189 which meant that, if there was an error processing a single event, the previous event was never flushed downstream. This change ensures that it flushes after the request is handled to send any additional buffered events. I also implement `close` to ensure that the inner sink also has `close()` called on it. Signed-off-by: Jesse Szwedko <jesse@szwedko.me> Signed-off-by: casserni <nicholascassera@gmail.com>

* fix(architecture): Ensure that Pipelines are flushed Fixes vectordotdev#5058 vectordotdev#4875 introduced buffering during `Pipeline.start_send`. Events were sent to the inner sink only on when `start_send` was called again or `poll_complete` is called. A side-effect of this was that when using `Sink.send_all` and `Stream.forward`, if an error was returned on a `poll()` of the stream being sent, the previously written event would never be sent to the inner sink as `poll_complete()` was never called. futures01 `Sink.send_all`: https://docs.rs/futures/0.1.30/src/futures/sink/send_all.rs.html#59-88 futures01 `Stream.forward`: https://docs.rs/futures/0.1.30/src/futures/stream/forward.rs.html#78-110 We can see in both cases it will return early if there is an error. If the stream is not ready or is finished, then it will call `poll_complete` / `close` on the sink. We use an errorable futures01 stream for the splunk_hec source: https://github.com/timberio/vector/blob/3bd8076be4097e4e9cf903a4f32e34a8b145a339/src/sources/splunk_hec.rs#L175-L189 which meant that, if there was an error processing a single event, the previous event was never flushed downstream. This change ensures that it flushes after the request is handled to send any additional buffered events. I also implement `close` to ensure that the inner sink also has `close()` called on it. Signed-off-by: Jesse Szwedko <jesse@szwedko.me> Signed-off-by: Brian Menges <brian.menges@anaplan.com>

jszwedko requested review from Hoverbear and lukesteensen November 16, 2020 19:06

jszwedko commented Nov 16, 2020

View reviewed changes

Hoverbear reviewed Nov 16, 2020

View reviewed changes

lukesteensen reviewed Nov 16, 2020

View reviewed changes

Flush explicitely

6ffb14e

Signed-off-by: Jesse Szwedko <jesse@szwedko.me>

jszwedko requested a review from ktff as a code owner November 16, 2020 22:57

Merge remote-tracking branch 'origin/master' into flush-sinks

634af80

ktff approved these changes Nov 17, 2020

View reviewed changes

jszwedko changed the title ~~fix(architecture): Ensure that Pipelines are flushed~~ fix(splunk_hec source): Flush messages Nov 17, 2020

lukesteensen approved these changes Nov 17, 2020

View reviewed changes

jszwedko merged commit 6bce5ac into master Nov 17, 2020

jszwedko deleted the flush-sinks branch November 17, 2020 17:20

This was referenced Nov 17, 2020

chore(ci): Fix hanging splunk_hec source partial test #5066

Closed

Survey stream frowarding for missing flushes #5081

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix(splunk_hec source): Flush messages #5069

fix(splunk_hec source): Flush messages #5069

jszwedko commented Nov 16, 2020

jszwedko Nov 16, 2020

binarylogic commented Nov 16, 2020 •

edited

Hoverbear Nov 16, 2020

jszwedko Nov 16, 2020

lukesteensen Nov 16, 2020

jszwedko Nov 16, 2020

jszwedko Nov 16, 2020

lukesteensen Nov 16, 2020

Hoverbear commented Nov 16, 2020

jszwedko commented Nov 16, 2020 •

edited

jszwedko commented Nov 17, 2020

fix(splunk_hec source): Flush messages #5069

fix(splunk_hec source): Flush messages #5069

Conversation

jszwedko commented Nov 16, 2020

jszwedko Nov 16, 2020

Choose a reason for hiding this comment

binarylogic commented Nov 16, 2020 • edited

Hoverbear Nov 16, 2020

Choose a reason for hiding this comment

jszwedko Nov 16, 2020

Choose a reason for hiding this comment

lukesteensen Nov 16, 2020

Choose a reason for hiding this comment

jszwedko Nov 16, 2020

Choose a reason for hiding this comment

jszwedko Nov 16, 2020

Choose a reason for hiding this comment

lukesteensen Nov 16, 2020

Choose a reason for hiding this comment

Hoverbear commented Nov 16, 2020

jszwedko commented Nov 16, 2020 • edited

jszwedko commented Nov 17, 2020

binarylogic commented Nov 16, 2020 •

edited

jszwedko commented Nov 16, 2020 •

edited