feat(cogs): Create inventory-tracker crate for storage oplog - #588
feat(cogs): Create inventory-tracker crate for storage oplog#588matt-codecov wants to merge 1 commit into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #588 +/- ##
==========================================
+ Coverage 87.99% 88.21% +0.21%
==========================================
Files 96 101 +5
Lines 15956 16488 +532
==========================================
+ Hits 14041 14545 +504
- Misses 1915 1943 +28
☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
jan-auer
left a comment
jan-auer
left a comment
There was a problem hiding this comment.
Initial review. Some of the below comments can be left for follow-up to get a first version in, as long as they don't change the overall concept and can be fixed in isolation. Particularly, shutdown is a larger topic.
| /// Enqueues one record. | ||
| /// | ||
| /// `key` controls which partition receives the message. | ||
| fn send(&self, key: &[u8], payload: Vec<u8>) -> Result<(), Self::Error>; |
There was a problem hiding this comment.
At least in arroyo / librdkafka, my understanding is that because of message batching sending a message can also fail after they were initially recorded. With the fire-and-forget style send API, these errors will not be returned anymore.
For us, this is likely fine - we want non-blocking best effort submission. Though we should pick one of the below options:
- Acknowledge this in the doc comment and call out that not all errors can be captured
- Make the signature infallible and move the responsibility for handling errors to the producer implementation.
There was a problem hiding this comment.
The on_delivery_failure argument to the Kafka producer should give callers a hook into failed messages. Failures here are when, like, the local queue is full or something.
There was a problem hiding this comment.
This is the high-level interface, so callers should not have a concept about Kafka and should not need to hook the delivery callback. This is something the kafka producer impl can do internally.
To clarify my original message: we don't have to change anything, but we can write this into the doc comment so callers know that Ok() is not a guarantee.
01463b9 to
71877b3
Compare
jan-auer
left a comment
jan-auer
left a comment
There was a problem hiding this comment.
Looks good. See the open comments before merge, please.
Either in this PR or in a follow up, I'd suggest to add metrics, especially for errors.
71877b3 to
d180831
Compare
d180831 to
6373531
Compare
jan-auer
changed the title
jan-auer
left a comment
jan-auer
left a comment
There was a problem hiding this comment.
Thanks! Approving for merge, provided the comments below are solved
| /// Enqueues one record. | ||
| /// | ||
| /// `key` controls which partition receives the message. | ||
| fn send(&self, key: &[u8], payload: Vec<u8>) -> Result<(), Self::Error>; |
There was a problem hiding this comment.
This is the high-level interface, so callers should not have a concept about Kafka and should not need to hook the delivery callback. This is something the kafka producer impl can do internally.
To clarify my original message: we don't have to change anything, but we can write this into the doc comment so callers know that Ok() is not a guarantee.
|
|
||
| fn delivery(&self, result: &DeliveryResult<'_>, _opaque: Self::DeliveryOpaque) { | ||
| if let Err((error, _)) = result { | ||
| tracing::warn!(%error, "failed to deliver inventory record"); |
There was a problem hiding this comment.
There's dedicated syntax to pass errors so they get captured properly.
| tracing::warn!(%error, "failed to deliver inventory record"); | |
| tracing::warn!(!!error, "failed to deliver inventory record"); |
There was a problem hiding this comment.
note to self, look into objectstore's tracing wrapper macro and basically inline what the !! does
| /// | ||
| /// Sending is asynchronous, so records handed over just before a process exits are | ||
| /// still sitting in a local queue. Call this during shutdown to deliver them. | ||
| fn flush(&self, timeout: Duration) -> Result<(), Self::Error>; |
There was a problem hiding this comment.
Note that backends use join terminology for this, but flush can also work here.
Let's make this async on the interface, so we can join it in the same way that we join backends. Internally, try to delegate to a spawn_blocking so that we do not block the executor when flushing the producer.
| /// let producer = shared_producer(NoopProducer); | ||
| /// let tracker = InventoryTracker::new(producer.clone(), "my_gcs_bucket", 1.0); | ||
| /// ``` | ||
| pub fn shared_producer<P>(producer: P) -> SharedProducer |
There was a problem hiding this comment.
I think you can make this a provided method on the Producer trait so that you simply have to call .shared() for this.
| organization_id: Option<u64>, | ||
| project_id: Option<u64>, |
There was a problem hiding this comment.
Not blocking for this PR, but for later as we integrate and revise the API of this layer: Here we may want to take in scopes and usecase and then map this internally. This way, the backends and whoever is using inventory tracker don't need to perform this mapping.
2 participants
Ref FS-210
Related to getsentry/sentry-kafka-schemas#497
Creates the
inventory-trackercrate in our project that we will use to emit Kafka messages for a change stream that we will turn into a COGS pipeline.This crate isn't Objectstore-specific at all. If/when another service wants to use it, we can move it to a separate repository and set up a release process for it. I just couldn't be bothered if nobody's asking for it.
PR(s) that use this new crate will be along soon.