diff --git a/Cargo.lock b/Cargo.lock index e62cea88..91ba8b69 100644 --- a/Cargo.lock +++ b/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]] diff --git a/Dockerfile b/Dockerfile index c657919e..e1bbb8e2 100644 --- a/Dockerfile +++ b/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 \ diff --git a/config/example.toml b/config/example.toml index 9058b8dd..d393b405 100644 --- a/config/example.toml +++ b/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" diff --git a/migration/src/lib.rs b/migration/src/lib.rs index 12606a2f..cc031348 100644 --- a/migration/src/lib.rs +++ b/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), ] } diff --git a/web3_proxy/Cargo.toml b/web3_proxy/Cargo.toml index b756d5dd..97083593 100644 --- a/web3_proxy/Cargo.toml +++ b/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" } diff --git a/web3_proxy/src/app/mod.rs b/web3_proxy/src/app/mod.rs index b27ccb40..eec2883f 100644 --- a/web3_proxy/src/app/mod.rs +++ b/web3_proxy/src/app/mod.rs @@ -504,9 +504,14 @@ impl Web3ProxyApp { let mut kafka_producer: Option = 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), diff --git a/web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs b/web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs index 35d11e9d..89553f71 100644 --- a/web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs +++ b/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"); diff --git a/web3_proxy/src/config.rs b/web3_proxy/src/config.rs index 1606e03d..68bed7a0 100644 --- a/web3_proxy/src/config.rs +++ b/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, + #[serde(default = "default_kafka_protocol")] + pub kafka_protocol: String, + /// domain in sign-in-with-ethereum messages pub login_domain: Option, @@ -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