ref(kafka): Move code into separate crate [INGEST-1627] #1563

olksdr · 2022-11-03T06:51:52Z

Create relay-kafka crate with its own configuration and producer functionality.

Having dedicated crate will help to keep all the related code together and it also simplifies the dependency tree.

Create `relay-kafka` crate with its own configuration and producer functionality. Having dedicated crate will help to keep all the related code together and it also simplifies the dependency tree.

jjbayer · 2022-11-03T14:13:08Z

relay-server/src/actors/store.rs

+            attachments: Producer::create(
+                &config
+                    .kafka_config(KafkaTopic::Attachments)
+                    .map_err(|_| ServerErrorKind::KafkaError)?,


I'm not sure how make_producers create_with_context worked, but are we now erasing information from the original error that used to be preserved? Or was it always erased and mapped to a ServerErrorKind::KafkaError?

The create_with_context call did not go anywhere, it was moved into the kafka crate and still used but in Producer::create function - which captures and reports the errors.

But we do erase the error here and replace it with ServerErrorKind::KafkaError which means that something wrong with kafka configuration and the producer could not be initialized.

Do you think we wanna keep the error around here as well? I can add required changes

This is a nit by the way, but yes, I would give KafkaError a #[cause] ProducerError to propagate the information. Would that work?

jjbayer · 2022-11-04T09:38:01Z

relay-server/src/actors/store.rs

+                    .map_err(|_| ServerErrorKind::KafkaError)?,
+                &mut reused_producers,
+            )
+            .map_err(|_| ServerErrorKind::KafkaError)?,


Another nit: These expressions all follow the same pattern, should we replace them with a helper function that just takes a KafkaTopic as argument?

jan-auer

Please also list the relay-kafka crate in the doc comment of the main relay crate.

jan-auer · 2022-11-04T13:04:24Z

relay-kafka/Cargo.toml

+
+[features]
+default = []
+processing = [


This feature effectively gives us producer implementations, should we name it as such?

Yep, sounds good.

jan-auer · 2022-11-04T13:06:07Z

relay-kafka/src/lib.rs

+
+/// Kafka configuration errors.
+#[derive(Fail, Debug)]
+pub enum ConfigError {


Is there a reason this is not in config? This causes some strange imports now where most types get imported from Config except for the related error.

yeah, I will reorganize the modules and make it simpler and uniform.

jan-auer · 2022-11-04T13:07:03Z

relay-kafka/src/lib.rs

@@ -0,0 +1,54 @@
+pub mod config;


It would be great if we could use this as a chance to improve documentation. Particularly, the root doc comment of the crate is valuable because it can explain the purpose of the crate and point to the various entry-points.

👍 I will create some initial docs here.

relay-kafka/src/lib.rs

jan-auer · 2022-11-04T13:11:03Z

relay-kafka/src/producer.rs

@@ -0,0 +1,170 @@
+use std::{collections::BTreeMap, sync::Arc};


Please un-merge those imports. Same with the crate imports below.

jan-auer · 2022-11-04T13:14:11Z

relay-kafka/src/producer.rs

+    /// the producer could not be created.
+    pub fn create<'a>(
+        config: &'a KafkaConfig,
+        reused_producers: &mut ReusedProducersMap<'a>,


This was fine while this was still in a private module, but this mixes several concerns for a public signature:

It creates a new producer instance

It memoizes them and determines the data structure for memoization

I realize that this increases the effort for this refactor, but it would be good to clean this up and separate those concerns. Ideally the memo-map is either generic, or becomes a higher-level utility that uses create internally.

In the end, I think the abstraction here is entirely wrong. The real producer is the Arc'ed threaded producer inside. The Producer type here then gets one or more of them and wraps them together with a topic name.

I will revisit this and make some changes into this refactor.

jan-auer · 2022-11-04T13:17:46Z

relay-kafka/src/producer.rs

+type ReusedProducersMap<'a> = BTreeMap<Option<&'a str>, Arc<ThreadedProducer>>;
+
+/// This object containes the Kafka producer variants for single and sharded configurations.
+pub enum Producer {


nit: The internals of the enum are now part of the public signature. Do we want this explicitly?

If yes, then let's also introduce a type for the Single variant. This allows us to move specific functionality to a method on that type rather than having large functions covering all variants here.

If not, let's make this a struct with a hidden inner field that is then a private enum. I'd still propose the refactor from the previous point, though.

jan-auer · 2022-11-04T13:19:17Z

relay-kafka/src/lib.rs

+#[cfg(feature = "processing")]
+mod producer;
+#[cfg(feature = "processing")]
+mod utils;


Since this module just contains producer utilities, why don't we merge them into the producer module? That would also be quite nice if we rename the feature to producer.

jan-auer · 2022-11-04T13:21:58Z

relay-kafka/src/utils.rs

+/// [`CaptureErrorContext`] context.
+pub type ThreadedProducer = rdkafka::producer::ThreadedProducer<CaptureErrorContext>;
+
+enum KafkaCounters {


In order to document these metrics, we need to list them in the documentation CI job here:

relay/.github/workflows/deploy.yml

Line 134 in 453a5d0

args: -p document-metrics -- -o relay_metrics.json relay-server/src/statsd.rs relay-metrics/src/statsd.rs

nit: I would propose to move these into a statsd module so the path stays stable.

Sounds good 👍

jan-auer · 2022-11-04T13:27:59Z

relay-kafka/src/lib.rs

@@ -0,0 +1,54 @@
+pub mod config;


Please add this to the preamble:

#![warn(missing_docs)] #![warn(missing_debug_implementations)] #![doc( html_logo_url = "https://raw.githubusercontent.com/getsentry/relay/master/artwork/relay-icon.png", html_favicon_url = "https://raw.githubusercontent.com/getsentry/relay/master/artwork/relay-icon.png" )]

Particularly, note that this will require Debug to be implemented on most types. Let's ensure this is the case.

jjbayer · 2022-11-14T09:46:46Z

.github/workflows/deploy.yml

@@ -131,7 +131,7 @@ jobs:
        uses: actions-rs/cargo@v1
        with:
          command: run
-          args: -p document-metrics -- -o relay_metrics.json relay-server/src/statsd.rs relay-metrics/src/statsd.rs
+          args: -p document-metrics -- -o relay_metrics.json relay-server/src/statsd.rs relay-metrics/src/statsd.rs relay-kafka/src/statsd.rs


This is easy to miss when adding a crate, would a glob pattern work here? That is, relay-*/src/statsd.rs

relay-kafka/src/producer/mod.rs

jjbayer · 2022-11-14T10:01:40Z

relay-server/src/actors/outcome.rs

+                .map_err(|_| ServerErrorKind::KafkaError)?;
+            client_builder = client_builder
+                .add_kafka_topic_config(kafka_config)
+                .map_err(|_| ServerErrorKind::KafkaError)?;


This nit still stands, but it's not a blocker:

I would give KafkaError a #[cause] ProducerError to propagate the information.

https://github.com/getsentry/relay/pull/1563/files/3050fca26247f6d2e61d77c2b74842a93a721465#r1013800430

@jjbayer so, the reason that the error neither here nor in any other case when the ServerErrorKind is used is that, the enum contains all the "empty" variants and also implement Copy and Clone and also Hash where most of the error do not do that, see:

relay/relay-server/src/service.rs

Lines 41 to 75 in 1e2bbf9

/// Indicates the type of failure of the server.

#[derive(Debug, Fail, Copy, Clone, PartialEq, Eq, Hash)]

pub enum ServerErrorKind {

/// Binding failed.

#[fail(display = "bind to interface failed")]

BindFailed,

/// Listening on the HTTP socket failed.

#[fail(display = "listening failed")]

ListenFailed,

/// A TLS error ocurred.

#[fail(display = "could not initialize the TLS server")]

TlsInitFailed,

/// TLS support was not compiled in.

#[fail(display = "compile with the `ssl` feature to enable SSL support")]

TlsNotSupported,

/// GeoIp construction failed.

#[fail(display = "could not load the Geoip Db")]

GeoIpError,

/// Configuration failed.

#[fail(display = "configuration error")]

ConfigError,

/// Initializing the Kafka producer failed.

#[fail(display = "could not initialize kafka producer")]

KafkaError,

/// Initializing the Redis cluster client failed.

#[fail(display = "could not initialize redis cluster client")]

RedisError,

}

In case we want to propagate the errors - we would need either to implement Hash trait on kafka ClientError and remove Clone and Copy, or if we add only additional formatted message inside the variant - we would have to at least remove Copy and Clone and return reference in

relay/relay-server/src/service.rs

Line 96 in 1e2bbf9

*self.inner.get_context()

What do you think would be better?

I see. Ignore my comment then, given that this error only happens on initialization I think we can live without the extra information being bubbled up.

jan-auer

Excellent documentation, thank you for taking the time to write this!

This is good to merge from my end, see minor and optional comments below.

jan-auer · 2022-11-15T12:28:07Z

.github/workflows/ci.yml

@@ -103,7 +103,7 @@ jobs:
          args: --workspace --all-targets --no-deps -- -D warnings

  test:
-    timeout-minutes: 15
+    timeout-minutes: 20


Is that timeout change still required?

I think it can be removed now. Added before rust cache fix.

jan-auer · 2022-11-15T12:30:10Z

relay-kafka/Cargo.toml

+description = "Kafka related functionality for Relay"
+homepage = "https://getsentry.github.io/relay/"
+repository = "https://github.com/getsentry/relay"
+version = "22.9.0"


The version is outdated.

Suggested change

version = "22.9.0"

version = "22.10.0"

Please also note that there's a self-hosted release today, so ideally we merge before that.

jan-auer · 2022-11-15T12:31:41Z

relay-kafka/src/config.rs

+impl KafkaTopic {
+    /// Returns iterator over the variants of [`KafkaTopic`].
+    /// It will have to be adjusted if the new variants are added.
+    pub fn iter() -> std::slice::Iter<'static, KafkaTopic> {


Suggested change

pub fn iter() -> std::slice::Iter<'static, KafkaTopic> {

pub fn iter() -> std::slice::Iter<'static, Self> {

nit: This iterates references of KafkaTopic. It would be slightly more ergonomic to iterate owned references, but it's likely not worth the added complexity. There would be std::array::IntoIter, but that will internally own the array.

jan-auer · 2022-11-15T12:36:44Z

relay-kafka/src/config.rs

+/// Configuration for topics.
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(default)]
+pub struct TopicAssignments {


To separate concerns, it would be beneficial if the relay-kafka crate doesn't know about the topic assignments. For instance, we could leave the TopicAssignment type here, but move TopicAssignments and KafkaTopic back into relay-config or wherever it's best defined. That way, we don't have to rebuild it every time we add a new topic.

I'm happy to merge the PR as-is though, and follow up with this later.

Agreed! I will followup on this

jan-auer · 2022-11-15T12:37:54Z

relay-kafka/src/config.rs

+    },
+}
+
+/// Sharded Kafka config


Suggested change

/// Sharded Kafka config

/// Sharded Kafka config.

jan-auer · 2022-11-15T12:41:16Z

relay-kafka/src/lib.rs

+//!
+//! # Usage
+//!
+//! ```compile_fail


It's unfortunate that we cannot even compile this because of missing features. ~~Maybe we move the documentation of the producer part into the producer module so it can compile?~~ (that won't work since the module is not public, just its contents)

Regardless, we should at least insert the correct imports here.

ref(kafka): Move functionality into separate crate.

ab5b660

Create `relay-kafka` crate with its own configuration and producer functionality. Having dedicated crate will help to keep all the related code together and it also simplifies the dependency tree.

olksdr self-assigned this Nov 3, 2022

olksdr requested a review from a team November 3, 2022 06:51

olksdr added 2 commits November 3, 2022 07:53

Add CHANGELOG record

ef55250

Merge branch 'master' into feat/relay-kafka-crate

3050fca

jjbayer reviewed Nov 3, 2022

View reviewed changes

olksdr added 2 commits November 3, 2022 17:57

Merge branch 'master' into feat/relay-kafka-crate

f2422ca

set ci timouts for tests on 20 min

845a5ef

jjbayer approved these changes Nov 4, 2022

View reviewed changes

jan-auer reviewed Nov 4, 2022

View reviewed changes

olksdr added 6 commits November 7, 2022 15:59

Merge branch 'master' into feat/relay-kafka-crate

881c9b7

ref(kafka): Address the review comments: docs, debug, mods, features

0d97340

ref(kafka): Create statsd module

ead25b2

Merge branch 'master' into feat/relay-kafka-crate

228bd89

ref(kafka): Simplify kafka client/producer API

c549186

Merge branch 'master' into feat/relay-kafka-crate

3e932ac

olksdr requested review from jjbayer and jan-auer November 14, 2022 08:44

olksdr added 2 commits November 14, 2022 09:49

fix lint

ca5aeac

fix lint for default features

1016670

jjbayer approved these changes Nov 14, 2022

View reviewed changes

olksdr added 2 commits November 14, 2022 12:01

ref(kafka): make sure all existing test cases pass

53d67b4

doc compile fail

45fa6e5

jan-auer approved these changes Nov 15, 2022

View reviewed changes

ref(kafka): Address final review comments

19548cd

olksdr merged commit d4b7e54 into master Nov 15, 2022

olksdr deleted the feat/relay-kafka-crate branch November 15, 2022 14:11

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ref(kafka): Move code into separate crate [INGEST-1627] #1563

ref(kafka): Move code into separate crate [INGEST-1627] #1563

olksdr commented Nov 3, 2022

jjbayer Nov 3, 2022

olksdr Nov 4, 2022

jjbayer Nov 4, 2022

jjbayer Nov 4, 2022

jan-auer left a comment

jan-auer Nov 4, 2022

olksdr Nov 7, 2022

jan-auer Nov 4, 2022

olksdr Nov 7, 2022

jan-auer Nov 4, 2022

olksdr Nov 7, 2022

jan-auer Nov 4, 2022

jan-auer Nov 4, 2022

olksdr Nov 7, 2022

jan-auer Nov 4, 2022

jan-auer Nov 4, 2022

jan-auer Nov 4, 2022

olksdr Nov 7, 2022

jan-auer Nov 4, 2022

jjbayer Nov 14, 2022

jjbayer Nov 14, 2022

olksdr Nov 14, 2022

jjbayer Nov 14, 2022

jan-auer left a comment

jan-auer Nov 15, 2022

olksdr Nov 15, 2022

jan-auer Nov 15, 2022

jan-auer Nov 15, 2022

jan-auer Nov 15, 2022

olksdr Nov 15, 2022

jan-auer Nov 15, 2022

jan-auer Nov 15, 2022

		@@ -0,0 +1,170 @@
		use std::{collections::BTreeMap, sync::Arc};

	/// Indicates the type of failure of the server.
	#[derive(Debug, Fail, Copy, Clone, PartialEq, Eq, Hash)]
	pub enum ServerErrorKind {
	/// Binding failed.
	#[fail(display = "bind to interface failed")]
	BindFailed,

	/// Listening on the HTTP socket failed.
	#[fail(display = "listening failed")]
	ListenFailed,

	/// A TLS error ocurred.
	#[fail(display = "could not initialize the TLS server")]
	TlsInitFailed,

	/// TLS support was not compiled in.
	#[fail(display = "compile with the `ssl` feature to enable SSL support")]
	TlsNotSupported,

	/// GeoIp construction failed.
	#[fail(display = "could not load the Geoip Db")]
	GeoIpError,

	/// Configuration failed.
	#[fail(display = "configuration error")]
	ConfigError,

	/// Initializing the Kafka producer failed.
	#[fail(display = "could not initialize kafka producer")]
	KafkaError,

	/// Initializing the Redis cluster client failed.
	#[fail(display = "could not initialize redis cluster client")]
	RedisError,
	}

	pub fn iter() -> std::slice::Iter<'static, KafkaTopic> {
	pub fn iter() -> std::slice::Iter<'static, Self> {

ref(kafka): Move code into separate crate [INGEST-1627] #1563

ref(kafka): Move code into separate crate [INGEST-1627] #1563

Conversation

olksdr commented Nov 3, 2022

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

jan-auer 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

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

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

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

jan-auer 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

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