From e91f75b3672dc7d914df7f50b1300f73a0eaff13 Mon Sep 17 00:00:00 2001 From: Vadim Vlasov Date: Mon, 13 Feb 2023 22:47:48 +0800 Subject: [PATCH] add gRPC-client compression (#136) * add upstream grpc-client compression * fix review issues --- docs/03-server-config.adoc | 53 ++++++++++++++++--- .../kotlin/io/emeraldpay/dshackle/Config.kt | 16 +++++- .../dshackle/config/CompressionConfig.kt | 2 +- .../dshackle/startup/ConfiguredUpstreams.kt | 8 ++- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 7 ++- .../startup/ConfiguredUpstreamsSpec.groovy | 6 +++ 6 files changed, 80 insertions(+), 12 deletions(-) diff --git a/docs/03-server-config.adoc b/docs/03-server-config.adoc index f60e1583..8270cf2b 100644 --- a/docs/03-server-config.adoc +++ b/docs/03-server-config.adoc @@ -26,6 +26,8 @@ compression: grpc: server: enabled: false + client: + enabled: false cluster: include: "upstreams.yaml" @@ -35,7 +37,8 @@ It configures following: - server is listening with gRCP API on `0.0.0.0:2449` - TLS is enabled -- compression is disabled for gRPC server (enabled by default) +- compression is disabled for gRPC-server (enabled by default) +- compression is disabled for upstream gRPC-client (enabled by default) - server certificate is located at `server.crt` with the key for it at `server.p8.key` - the server requires a client authentication by TLS client certificate signed by `ca.crt` certificate - no JSON RPC is configured @@ -72,12 +75,23 @@ a| `upstreams` |=== === Compression -Compression is enabled by default on the gRPC server -(the server accepts compressed requests and can send compressed responses). -Responses will be compressed if a client supports compression (sends relevant headers), -otherwise, communication will be uncompressed. -But if for some reason you need to disable compression forcibly, -add the lines below to the config: +Compression is enabled by default on the gRPC-server. +Responses will be compressed if a client supports compression +(sends relevant headers), otherwise, communication will be uncompressed. +But if for some reason you need to disable responses compression +forcibly (the server will still be able to accept compressed requests), +add following lines to the config: +[source,yaml] +---- +compression: + grpc: + server: + enabled: false +---- +Compression is enabled by default for gRPC-requests to upstreams. +If you want to disable it, +(gRPC-client will still be able to accept compressed responses), +add following lines to the config: [source,yaml] ---- compression: @@ -85,6 +99,31 @@ compression: client: enabled: false ---- +Thus, all possible combinations of compression configuration for +interacting dshackle grpc-client and grpc-server look like this: +|=== +| Client | Server | Requests | Responses + +| enabled +| enabled +| compressed +| compressed + +| enabled +| disabled +| compressed +| plain + +| disabled +| enabled +| plain +| compressed + +| disabled +| disabled +| plain +| plain +|=== === Enabling JSON RPC proxy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt index ffc4913a..ea21e069 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt @@ -16,7 +16,16 @@ */ package io.emeraldpay.dshackle -import io.emeraldpay.dshackle.config.* +import io.emeraldpay.dshackle.config.CacheConfig +import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.CompressionConfig +import io.emeraldpay.dshackle.config.HealthConfig +import io.emeraldpay.dshackle.config.MainConfig +import io.emeraldpay.dshackle.config.MainConfigReader +import io.emeraldpay.dshackle.config.MonitoringConfig +import io.emeraldpay.dshackle.config.SignatureConfig +import io.emeraldpay.dshackle.config.TokensConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig import org.bouncycastle.jce.provider.BouncyCastleProvider import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired @@ -110,6 +119,11 @@ open class Config( return mainConfig.upstreams } + @Bean + open fun compressionConfig(@Autowired mainConfig: MainConfig): CompressionConfig { + return mainConfig.compression + } + @Bean open fun cacheConfig(@Autowired mainConfig: MainConfig): CacheConfig { return mainConfig.cache ?: CacheConfig() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/CompressionConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/CompressionConfig.kt index efbcb6f4..93baf709 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/CompressionConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/CompressionConfig.kt @@ -1,6 +1,6 @@ package io.emeraldpay.dshackle.config -class CompressionConfig( +open class CompressionConfig( var grpc: GRPC = GRPC() ) { /** diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 0945972e..b90a4322 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.CompressionConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.JsonRpcReader import io.emeraldpay.dshackle.upstream.* @@ -62,6 +63,7 @@ import kotlin.math.abs open class ConfiguredUpstreams( private val fileResolver: FileResolver, private val config: UpstreamsConfig, + private val compressionConfig: CompressionConfig, private val callTargets: CallTargetsHolder, private val eventPublisher: ApplicationEventPublisher, @Qualifier("grpcChannelExecutor") @@ -86,7 +88,7 @@ open class ConfiguredUpstreams( log.debug("Start upstream ${up.id}") if (up.connection is UpstreamsConfig.GrpcConnection) { val options = up.options ?: UpstreamsConfig.Options() - buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options) + buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options, compressionConfig.grpc.clientEnabled) } else { val chain = Global.chainById(up.chain) if (chain == Chain.UNSPECIFIED) { @@ -307,7 +309,8 @@ open class ConfiguredUpstreams( private fun buildGrpcUpstream( nodeId: Int?, config: UpstreamsConfig.Upstream, - options: UpstreamsConfig.Options + options: UpstreamsConfig.Options, + compression: Boolean ) { if (!this::grpcUpstreamsScheduler.isInitialized) { grpcUpstreamsScheduler = Schedulers.fromExecutorService( @@ -324,6 +327,7 @@ open class ConfiguredUpstreams( endpoint.host!!, endpoint.port, endpoint.auth, + compression, fileResolver, endpoint.upstreamRating, config.labels, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 4465e933..64f0de5b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -34,6 +34,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics +import io.grpc.Codec import io.grpc.netty.NettyChannelBuilder import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Metrics @@ -63,6 +64,7 @@ class GrpcUpstreams( private val host: String, private val port: Int, private val auth: AuthConfig.ClientTlsAuth? = null, + private val compression: Boolean, private val fileResolver: FileResolver, private val nodeRating: Int, private val labels: UpstreamsConfig.Labels, @@ -94,7 +96,10 @@ class GrpcUpstreams( chanelBuilder.usePlaintext() } - val client = ReactorBlockchainGrpc.newReactorStub(chanelBuilder.build()) + var client = ReactorBlockchainGrpc.newReactorStub(chanelBuilder.build()) + if (compression) { + client = client.withCompression(Codec.Gzip().messageEncoding) + } this.client = client val statusSubscription = AtomicReference() diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index 9304818e..aad6f7f4 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.startup import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.config.ChainsConfig +import io.emeraldpay.dshackle.config.CompressionConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.quorum.NonEmptyQuorum import io.emeraldpay.dshackle.upstream.CallTargetsHolder @@ -20,6 +21,7 @@ class ConfiguredUpstreamsSpec extends Specification { def configurer = new ConfiguredUpstreams( Stub(FileResolver), Stub(UpstreamsConfig), + Stub(CompressionConfig), callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), @@ -47,6 +49,7 @@ class ConfiguredUpstreamsSpec extends Specification { def configurer = new ConfiguredUpstreams( Stub(FileResolver), Stub(UpstreamsConfig), + Stub(CompressionConfig), callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), @@ -73,6 +76,7 @@ class ConfiguredUpstreamsSpec extends Specification { def configurer = new ConfiguredUpstreams( Stub(FileResolver), Stub(UpstreamsConfig), + Stub(CompressionConfig), callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), @@ -94,6 +98,7 @@ class ConfiguredUpstreamsSpec extends Specification { def configurer = new ConfiguredUpstreams( Stub(FileResolver), Stub(UpstreamsConfig), + Stub(CompressionConfig), callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1), @@ -120,6 +125,7 @@ class ConfiguredUpstreamsSpec extends Specification { def configurer = new ConfiguredUpstreams( Stub(FileResolver), Stub(UpstreamsConfig), + Stub(CompressionConfig), callTargetsHolder, Mock(ApplicationEventPublisher), Executors.newFixedThreadPool(1),