chore(architecture): consolidate sink I/O driver logic into reusable component #9215

tobz · 2021-09-17T16:20:31Z

This PR moves the logic previously baked into the run_io method for the S3 sink into a reusable component called Driver. Simply put, you give it a Stream of items which can be used as the request for a Service, and it handles building calls for each item, as well as providing finalization and acking as the responses come through.

As it stands, we've got a lot of changes going on here as part of trying to make the boilerplate specific to sinks be as simple/flexible as possible. To name a few:

a new trait, Encoder, that allows defining a type that can "encode" a single event (EncodingConfiguration + Encoding enums)
a new generalized StandardEncodings enum that implements Encoder for the most common encodings
Compression tweaks slightly to depend more on defined compression levels from flate2 itself
a new trait, RequestBuilder, to try and provide a starting point for generic "how do we take events and make a request?" interface
new stream combinators, from SinkBuilderExt, to wrap streams of events with common building blocks, such as batching, request building, Stream -> Service driving, for building a complete sink

…component Signed-off-by: Toby Lawrence <[email protected]>

Signed-off-by: Toby Lawrence <[email protected]>

netlify · 2021-09-17T16:20:37Z

✔️ Deploy Preview for vector-project canceled.

🔨 Explore the source changes: 8957e8f

🔍 Inspect the deploy log: https://app.netlify.com/sites/vector-project/deploys/615b528c1dec8e0008419f77

blt · 2021-09-17T16:51:53Z

lib/vector-core/src/stream/driver.rs

+    /// return an error for a legitimate reason in the future.
+    pub async fn run(self) -> Result<(), ()> {
+        let in_flight = FuturesUnordered::new();
+        let mut pending_acks = HashMap::new();


Small note that we could probably parameterize this with twox hash or some such if we wanted.

I had the exact same thought. There's actually a nohash-hasher crate since our keys are already u64. I guess if we've both had this thought, maybe I should just go ahead and do it.

TIL about nohash-hasher.

This is almost a glorified VecDeque. I wonder if it could be emulated with a wrapper around VecDequeue<Option<(u64, u64)>> where inserting inserts "blanks" if it's not actually sequentially next.

That's a good point regarding VecDeque. I'll have to think about your idea... off the top of my head, I can't visualize it, so might just be one of those things I need to write down first to grok.

In 05bc5f3, I reworked this as a BinaryHeap because I didn't like the idea of having to insert blanks that we would then had to do a linear scan to find when the call finishes and we want to mark the pending acknowledgement as completed.

Admittedly, though, the code using BinaryHeap is longer than the HashMap approach.

src/sinks/aws_s3/sink.rs

blt · 2021-09-17T16:53:09Z

I really like where this is headed.

bruceg

A big 👍🏻 to the reusable component in vector-core.

src/sinks/aws_s3/sink.rs

bruceg · 2021-09-17T19:01:00Z

lib/vector-core/src/stream/driver.rs

+    /// return an error for a legitimate reason in the future.
+    pub async fn run(self) -> Result<(), ()> {
+        let in_flight = FuturesUnordered::new();
+        let mut pending_acks = HashMap::new();


This is almost a glorified VecDeque. I wonder if it could be emulated with a wrapper around VecDequeue<Option<(u64, u64)>> where inserting inserts "blanks" if it's not actually sequentially next.

Signed-off-by: Toby Lawrence <[email protected]>

Co-authored-by: Bruce Guenter <[email protected]>

Signed-off-by: Toby Lawrence <[email protected]>

src/sinks/util/concurrent_map.rs

src/sinks/util/builder.rs

…eamify-new-sink-io-task Signed-off-by: Toby Lawrence <[email protected]>

src/sinks/s3_common/sink.rs

lukesteensen

Still working through this, but a couple quick notes.

lib/vector-core/src/stream/driver.rs

Signed-off-by: Toby Lawrence <[email protected]>

lukesteensen

Looks great to me! Just one note about a log message.

lukesteensen · 2021-10-04T22:26:48Z

lib/vector-core/src/stream/driver.rs

+                    let result: Result<(u64, usize), JoinError> = result;
+                    match result {
+                        Ok((seq_no, ack_size)) => {
+                            trace!(message = "Sending request.", seq_no, ack_size);


Is this message accurate? It seems like we've already sent the request and gotten a response at this point.

Oh! Good catch.

blt · 2021-10-04T23:59:08Z

src/sinks/aws_s3/tests.rs

@@ -220,7 +220,7 @@ mod integration_tests {

        let (_lines, events, receiver) = make_events_batch(1, 1);
        sink.run(events).await.unwrap();
-        assert_eq!(receiver.await, BatchStatus::Errored);
+        assert_eq!(receiver.await, BatchStatus::Failed);


It's not clear to me why this switch would happen, but I'm not opposed. Mostly I want to flag that there's some logic behind this change, which I am missing. I have to look up the difference between 'failed' and 'errored' every time; they are "at least one event in the batch had a permanent failure" and "at least one event in the batch had a transient error in delivery" respectively. These definitions lead to tricky questions about priority if a batch has an event that suffered a permanent failure and also a transient error, so I would tend to imagine the line between them is blurry in practice in vector today.

blt

I've noticed you've left the Datadog log sink's run_io in place, here. I'd be happy to see that converted in this PR or in a follow-up issue, whichever your preference. I had a question about a follow-up issue for a movement of code into core as well.

That said, this is excellent and I look forward to it being merged up.

tobz · 2021-10-05T00:09:31Z

@blt I'll be working on an immediate follow-up PR where I handle some of the code movement into vector_core as well as converting the Datadog Logs sink. I just wanted to get this merged in since now other folks have PRs rebased on my branch, and it's all starting to get a bit too long in the tooth. 😅

blt · 2021-10-05T00:41:07Z

@blt I'll be working on an immediate follow-up PR where I handle some of the code movement into vector_core as well as converting the Datadog Logs sink. I just wanted to get this merged in since now other folks have PRs rebased on my branch, and it's all starting to get a bit too long in the tooth. sweat_smile

I'm for it. Merge this thing.

…component (#9215) * chore(architecture): consolidate sink I/O driver logic into reusable component Signed-off-by: Toby Lawrence <[email protected]>

tobz added 3 commits September 17, 2021 10:30

chore(architecture): consolidate sink I/O driver logic into reusable …

69c398a

…component Signed-off-by: Toby Lawrence <[email protected]>

Merge branch 'master' into tobz/streamify-new-sink-io-task

eec948f

cleaning up clippy nits

c968754

Signed-off-by: Toby Lawrence <[email protected]>

blt reviewed Sep 17, 2021

View reviewed changes

src/sinks/aws_s3/sink.rs Outdated Show resolved Hide resolved

bruceg reviewed Sep 17, 2021

View reviewed changes

blt mentioned this pull request Sep 20, 2021

Rewrite sinks into the "new-style" #9261

Open

40 tasks

tobz added 2 commits September 21, 2021 10:36

temporary commit to avoid losing too much work before more futzing

ae2de65

Signed-off-by: Toby Lawrence <[email protected]>

ahhh it compiles

ef8dd9e

Signed-off-by: Toby Lawrence <[email protected]>

tobz added the ci-condition: integration tests enable Run integration tests on this PR label Sep 24, 2021

tobz force-pushed the tobz/streamify-new-sink-io-task branch from a8a2430 to d24f943 Compare September 24, 2021 15:17

tobz and others added 4 commits September 24, 2021 11:25

Merge branch 'master' into tobz/streamify-new-sink-io-task

7e239f6

Signed-off-by: Toby Lawrence <[email protected]>

Update src/sinks/aws_s3/sink.rs

1bfd7d4

Co-authored-by: Bruce Guenter <[email protected]>

PR feedback

55f8062

Signed-off-by: Toby Lawrence <[email protected]>

formatting and spelling errors and what not

45d95cd

Signed-off-by: Toby Lawrence <[email protected]>

tobz force-pushed the tobz/streamify-new-sink-io-task branch from d24f943 to 45d95cd Compare September 24, 2021 15:25

blt reviewed Sep 24, 2021

View reviewed changes

src/sinks/util/concurrent_map.rs Show resolved Hide resolved

blt reviewed Sep 24, 2021

View reviewed changes

src/sinks/util/builder.rs Show resolved Hide resolved

Merge branch 'master' of github.com:vectordotdev/vector into tobz/str…

021a974

…eamify-new-sink-io-task Signed-off-by: Toby Lawrence <[email protected]>

tobz force-pushed the tobz/streamify-new-sink-io-task branch from 1f30482 to 021a974 Compare September 24, 2021 18:50

fuchsnj reviewed Sep 24, 2021

View reviewed changes

src/sinks/s3_common/sink.rs Outdated Show resolved Hide resolved

fuchsnj mentioned this pull request Sep 28, 2021

chore(kafka sink): rewrite kafka sink with new style #9361

Merged

bruceg self-requested a review September 28, 2021 20:29

lukesteensen reviewed Sep 28, 2021

View reviewed changes

lib/vector-core/src/stream/driver.rs Outdated Show resolved Hide resolved

lib/vector-core/src/stream/driver.rs Show resolved Hide resolved

vladimir-dd mentioned this pull request Oct 1, 2021

feat(datadog_archives sink): Add gcp_cloud_storage support #9403

Merged

tobz added 3 commits October 4, 2021 12:10

address PR feedback

05bc5f3

Signed-off-by: Toby Lawrence <[email protected]>

fix more nits and lints

101d461

Signed-off-by: Toby Lawrence <[email protected]>

fix nit

f4c2196

Signed-off-by: Toby Lawrence <[email protected]>

try it this way

8957e8f

Signed-off-by: Toby Lawrence <[email protected]>

binarylogic requested review from blt, lukesteensen and fuchsnj October 4, 2021 21:39

lukesteensen approved these changes Oct 4, 2021

View reviewed changes

blt reviewed Oct 4, 2021

View reviewed changes

blt approved these changes Oct 5, 2021

View reviewed changes

tobz merged commit 6b5141b into master Oct 5, 2021

tobz deleted the tobz/streamify-new-sink-io-task branch October 5, 2021 00:43

fuchsnj mentioned this pull request Oct 27, 2021

fix(architecture): fix bug in Driver that causes hang #9804

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

chore(architecture): consolidate sink I/O driver logic into reusable component #9215

chore(architecture): consolidate sink I/O driver logic into reusable component #9215

tobz commented Sep 17, 2021 •

edited

Loading

netlify bot commented Sep 17, 2021 •

edited

Loading

blt Sep 17, 2021

tobz Sep 17, 2021

blt Sep 17, 2021

bruceg Sep 17, 2021

tobz Sep 24, 2021

tobz Oct 4, 2021 •

edited

Loading

blt commented Sep 17, 2021

bruceg left a comment

bruceg Sep 17, 2021

lukesteensen left a comment

lukesteensen left a comment

lukesteensen Oct 4, 2021

tobz Oct 5, 2021

blt Oct 4, 2021

blt left a comment

tobz commented Oct 5, 2021

blt commented Oct 5, 2021

chore(architecture): consolidate sink I/O driver logic into reusable component #9215

chore(architecture): consolidate sink I/O driver logic into reusable component #9215

Conversation

tobz commented Sep 17, 2021 • edited Loading

netlify bot commented Sep 17, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tobz Oct 4, 2021 • edited Loading

Choose a reason for hiding this comment

blt commented Sep 17, 2021

bruceg left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukesteensen left a comment

Choose a reason for hiding this comment

lukesteensen left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

blt left a comment

Choose a reason for hiding this comment

tobz commented Oct 5, 2021

blt commented Oct 5, 2021

tobz commented Sep 17, 2021 •

edited

Loading

netlify bot commented Sep 17, 2021 •

edited

Loading

tobz Oct 4, 2021 •

edited

Loading