Merge branch 'main' into devel

Cargo.lock

@@ -777,6 +777,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"
@@ -4280,10 +4289,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
@@ -36,6 +43,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
     cargo install \
     --features tokio-uring \
     --locked \
+    --features rdkafka-src \
     --no-default-features \
     --path ./web3_proxy \
     --profile faster_release \

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"
 
 # a timeseries database is optional. it is used for making pretty graphs
 influxdb_host = "http://127.0.0.1:18086"

migration/src/lib.rs

@@ -40,10 +40,10 @@ impl MigratorTrait for Migrator {
             Box::new(m20221213_134158_move_login_into_database::Migration),
             Box::new(m20230117_191358_admin_table::Migration),
             Box::new(m20230119_204135_better_free_tier::Migration),
+            Box::new(m20230125_204810_stats_v2::Migration),
             Box::new(m20230130_124740_read_only_login_logic::Migration),
             Box::new(m20230130_165144_prepare_admin_imitation_pre_login::Migration),
             Box::new(m20230215_152254_admin_trail::Migration),
-            Box::new(m20230125_204810_stats_v2::Migration),
             Box::new(m20230307_002623_migrate_rpc_accounting_to_rpc_accounting_v2::Migration),
         ]
     }

web3_proxy/Cargo.toml

@@ -10,6 +10,7 @@ default-run = "web3_proxy_cli"
 default = ["deadlock_detection"]
 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"]
 
 [dependencies]
 deferred-rate-limiter = { path = "../deferred-rate-limiter" }

web3_proxy/src/app/mod.rs

@@ -504,9 +504,14 @@ impl Web3ProxyApp {
 
         let mut kafka_producer: Option<rdkafka::producer::FutureProducer> = None;
         if let Some(kafka_brokers) = top_config.app.kafka_urls.clone() {
+            info!("Connecting to kafka");
+
+            let security_protocol = &top_config.app.kafka_protocol;
+
             match rdkafka::ClientConfig::new()
                 .set("bootstrap.servers", kafka_brokers)
                 .set("message.timeout.ms", "5000")
+                .set("security.protocol", security_protocol)
                 .create()
             {
                 Ok(k) => kafka_producer = Some(k),

web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs

@@ -52,16 +52,19 @@ impl SearchKafkaSubCommand {
 
         let wanted_kafka_key = wanted_kafka_key.as_ref().map(|x| &x[..]);
 
-        let brokers = top_config
+        let kafka_brokers = top_config
             .app
             .kafka_urls
             .context("top_config.app.kafka_urls is required")?;
 
         let mut consumer = ClientConfig::new();
 
+        let security_protocol = &top_config.app.kafka_protocol;
+
         consumer
-            .set("bootstrap.servers", &brokers)
+            .set("bootstrap.servers", &kafka_brokers)
             .set("enable.partition.eof", "false")
+            .set("security.protocol", security_protocol)
             .set("session.timeout.ms", "6000")
             .set("enable.auto.commit", "false");
 

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>,
 
@@ -214,6 +217,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