more packages for kafka+ssl

Cargo.lock

@@ -841,6 +841,15 @@ dependencies = [
  "os_str_bytes",
 ]
 
+[[package]]
+name = "cmake"
+version = "0.1.50"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a31c789563b815f77f4250caee12365734369f942439b7defd71e18a48197130"
+dependencies = [
+ "cc",
+]
+
 [[package]]
 name = "codespan-reporting"
 version = "0.11.1"
@@ -3266,6 +3275,18 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "openssl-sys"
+version = "0.9.85"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0d3d193fb1488ad46ffe3aaabc912cc931d02ee8518fe2959aea8ef52718b0c0"
+dependencies = [
+ "cc",
+ "libc",
+ "pkg-config",
+ "vcpkg",
+]
+
 [[package]]
 name = "ordered-float"
 version = "3.4.0"
@@ -3956,10 +3977,13 @@ version = "4.3.0+1.9.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d222a401698c7f2010e3967353eae566d9934dcda49c29910da922414ab4e3f4"
 dependencies = [
+ "cmake",
  "libc",
  "libz-sys",
  "num_enum",
+ "openssl-sys",
  "pkg-config",
+ "zstd-sys",
 ]
 
 [[package]]

Dockerfile

@@ -18,7 +18,14 @@ RUN curl -L https://foundry.paradigm.xyz | bash && foundryup
 
 # install web3-proxy system dependencies. most things are rust-only, but not everything
 RUN apt-get update && \
-    apt-get install --yes librdkafka-dev && \
+    apt-get install --yes \
+    cmake \
+    liblz4-dev \
+    libpthread-stubs0-dev \
+    libssl-dev \
+    libzstd-dev \
+    make \
+    && \
     rm -rf /var/lib/apt/lists/*
 
 # copy the application
@@ -35,6 +42,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
     --mount=type=cache,target=/app/target \
     cargo install \
     --locked \
+    --features rdkafka-src \
     --no-default-features \
     --profile faster_release \
     --root /opt/bin \

config/example.toml

@@ -12,6 +12,7 @@ db_url = "mysql://root:dev_web3_proxy@127.0.0.1:13306/dev_web3_proxy"
 db_replica_url = "mysql://root:dev_web3_proxy@127.0.0.1:13306/dev_web3_proxy"
 
 kafka_urls = "127.0.0.1:19092"
+kafka_protocol = "plaintext"
 
 # thundering herd protection
 # only mark a block as the head block if the sum of their soft limits is greater than or equal to min_sum_soft_limit

web3_proxy/Cargo.toml

@@ -10,6 +10,7 @@ default-run = "web3_proxy_cli"
 default = ["deadlock_detection", "tokio-console"]
 deadlock_detection = ["parking_lot/deadlock_detection"]
 tokio-console = ["dep:tokio-console", "dep:console-subscriber"]
+rdkafka-src = ["rdkafka/cmake-build", "rdkafka/libz", "rdkafka/ssl", "rdkafka/zstd-pkg-config"]
 
 # TODO: turn tokio-console on with a feature. console-subscriber = { version = "0.1.7" }
 

web3_proxy/src/app/mod.rs

@@ -468,13 +468,7 @@ impl Web3ProxyApp {
 
         let mut kafka_producer: Option<rdkafka::producer::FutureProducer> = None;
         if let Some(kafka_brokers) = top_config.app.kafka_urls.clone() {
-            let security_protocol = if kafka_brokers.starts_with("SSL:") {
-                "ssl"
-            } else if kafka_brokers.starts_with("TCP:") {
-                "plaintext"
-            } else {
-                return Err(anyhow::anyhow!("unexpected kafka protocol"));
-            };
+            let security_protocol = &top_config.app.kafka_protocol;
 
             match rdkafka::ClientConfig::new()
                 .set("bootstrap.servers", kafka_brokers)

web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs

@@ -59,13 +59,7 @@ impl SearchKafkaSubCommand {
 
         let mut consumer = ClientConfig::new();
 
-        let security_protocol = if kafka_brokers.starts_with("SSL:") {
-            "ssl"
-        } else if kafka_brokers.starts_with("TCP:") {
-            "plaintext"
-        } else {
-            return Err(anyhow::anyhow!("unexpected kafka protocol"));
-        };
+        let security_protocol = &top_config.app.kafka_protocol;
 
         consumer
             .set("bootstrap.servers", &kafka_brokers)

web3_proxy/src/config.rs

@@ -108,6 +108,9 @@ pub struct AppConfig {
     /// Used by /debug/:rpc_key urls for logging requests and responses. No other endpoints log request/response data.
     pub kafka_urls: Option<String>,
 
+    #[serde(default = "default_kafka_protocol")]
+    pub kafka_protocol: String,
+
     /// domain in sign-in-with-ethereum messages
     pub login_domain: Option<String>,
 
@@ -202,6 +205,10 @@ fn default_login_rate_limit_per_period() -> u64 {
     10
 }
 
+fn default_kafka_protocol() -> String {
+    "ssl".to_string()
+}
+
 fn default_response_cache_max_bytes() -> u64 {
     // TODO: default to some percentage of the system?
     // 100 megabytes