From f62e16af459803dc36956353e91712dac55c5e54 Mon Sep 17 00:00:00 2001 From: fatemeh imanipour Date: Tue, 11 Aug 2026 20:31:22 +0330 Subject: [PATCH 1/2] Optimize wallet and api --- .../src/main/resources/application.yml | 1 + .../walletproxy/config/WebClientConfig.kt | 39 ++++++++++++++++--- .../opex/api/app/config/RateLimitConfig.kt | 13 +++++-- .../src/main/resources/application.yml | 9 +++-- 4 files changed, 48 insertions(+), 14 deletions(-) diff --git a/accountant/accountant-app/src/main/resources/application.yml b/accountant/accountant-app/src/main/resources/application.yml index 1f08d8860..f191ca34c 100644 --- a/accountant/accountant-app/src/main/resources/application.yml +++ b/accountant/accountant-app/src/main/resources/application.yml @@ -47,6 +47,7 @@ spring: instance-id: ${spring.application.name}:${server.port} healthCheckInterval: 20s prefer-ip-address: true + query-passing: true config: import: vault://secret/${spring.application.name} management: diff --git a/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/config/WebClientConfig.kt b/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/config/WebClientConfig.kt index 2ba1c3c02..2f9bbe64c 100644 --- a/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/config/WebClientConfig.kt +++ b/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/config/WebClientConfig.kt @@ -1,25 +1,52 @@ package co.nilin.opex.accountant.ports.walletproxy.config +import io.netty.channel.ChannelOption import org.springframework.cloud.client.ServiceInstance import org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer import org.springframework.cloud.client.loadbalancer.reactive.ReactorLoadBalancerExchangeFilterFunction import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration +import org.springframework.http.client.reactive.ReactorClientHttpConnector import org.springframework.web.reactive.function.client.WebClient import org.zalando.logbook.Logbook import org.zalando.logbook.netty.LogbookClientHandler import reactor.netty.http.client.HttpClient +import reactor.netty.resources.ConnectionProvider +import java.time.Duration @Configuration class WebClientConfig { @Bean - fun webClient(loadBalancerFactory: ReactiveLoadBalancer.Factory, logbook: Logbook): WebClient { - val client = HttpClient.create().doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) } + fun webClient( + loadBalancerFactory: ReactiveLoadBalancer.Factory, + logbook: Logbook + ): WebClient { + + val connectionProvider = ConnectionProvider.builder("accountant-wallet") + .maxIdleTime(Duration.ofSeconds(20)) + .maxLifeTime(Duration.ofMinutes(5)) + .pendingAcquireTimeout(Duration.ofSeconds(5)) + .evictInBackground(Duration.ofSeconds(30)) + .lifo() + .build() + + val client = HttpClient.create(connectionProvider) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) + .responseTimeout(Duration.ofSeconds(10)) + .keepAlive(true) + .doOnConnected { + it.addHandlerLast(LogbookClientHandler(logbook)) + } + return WebClient.builder() - //.clientConnector(ReactorClientHttpConnector(client)) - .filter(ReactorLoadBalancerExchangeFilterFunction(loadBalancerFactory, emptyList())) + .clientConnector(ReactorClientHttpConnector(client)) + .filter( + ReactorLoadBalancerExchangeFilterFunction( + loadBalancerFactory, + emptyList() + ) + ) .build() } - -} +} \ No newline at end of file diff --git a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/RateLimitConfig.kt b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/RateLimitConfig.kt index fd9167451..054d639d8 100644 --- a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/RateLimitConfig.kt +++ b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/RateLimitConfig.kt @@ -53,12 +53,17 @@ class RateLimitConfig( return ReactiveSecurityContextHolder.getContext() .mapNotNull { it.authentication } .filter { it.isAuthenticated } - .flatMap { auth -> - applyRateLimit(auth.name, exchange, chain, groupId) + .map { auth -> + Mono.defer { + applyRateLimit(auth.name, exchange, chain, groupId) + } } - .switchIfEmpty( - chain.filter(exchange) + .defaultIfEmpty( + Mono.defer { + chain.filter(exchange) + } ) + .flatMap { it } } private fun applyRateLimit( diff --git a/wallet/wallet-app/src/main/resources/application.yml b/wallet/wallet-app/src/main/resources/application.yml index 488d21278..ec67e2e77 100644 --- a/wallet/wallet-app/src/main/resources/application.yml +++ b/wallet/wallet-app/src/main/resources/application.yml @@ -8,6 +8,8 @@ management: endpoint: health: show-details: when_authorized + probes: + enabled: true metrics: enabled: true prometheus: @@ -66,9 +68,10 @@ spring: host: ${CONSUL_HOST:localhost} port: 8500 discovery: - #healthCheckPath: ${management.context-path}/health instance-id: ${spring.application.name}:${server.port} - healthCheckInterval: 20s + health-check-path: /actuator/health/liveness + health-check-interval: 10s + health-check-timeout: 5s prefer-ip-address: true config: import: vault://secret/${spring.application.name} @@ -174,6 +177,4 @@ logging: co.nilin: INFO reactor.netty.http.client: INFO org.zalando.logbook: TRACE - org.hibernate.SQL: DEBUG - logging.level.org.hibernate.type.descriptor.sql.BasicBinder: TRACE swagger.authUrl: ${SWAGGER_AUTH_URL:https://api.opex.dev/auth}/realms/opex/protocol/openid-connect/token From d7d70113138323e0ca0dd3ba834efa2fa05ced0f Mon Sep 17 00:00:00 2001 From: fatemeh imanipour Date: Wed, 12 Aug 2026 15:11:00 +0330 Subject: [PATCH 2/2] Optimize connection parameters in wallet and api --- .../opex/api/app/config/WebClientConfig.kt | 71 ++++++++++++++----- .../src/main/resources/application.yml | 34 +++++++-- .../ports/proxy/config/ProxyDispatchers.kt | 28 +++++++- .../proxy/impl/MarketUserDataProxyImpl.kt | 7 +- .../proxy/impl/MatchingGatewayProxyImpl.kt | 8 ++- device-management/pom.xml | 2 +- pom.xml | 2 +- 7 files changed, 123 insertions(+), 29 deletions(-) diff --git a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt index a89032080..c2d67d527 100644 --- a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt +++ b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt @@ -26,22 +26,58 @@ class WebClientConfig( private val logbook: Logbook, @Value("\${app.auth.url}") private val url: String, + @Value("\${app.http.client.wiretap.enabled:false}") + private val wiretapEnabled: Boolean, + @Value("\${app.http.client.general.max-connections:300}") + private val generalMaxConnections: Int, + @Value("\${app.http.client.general.pending-acquire-max-count:1000}") + private val generalPendingAcquireMaxCount: Int, + @Value("\${app.http.client.general.max-idle-seconds:30}") + private val generalMaxIdleSeconds: Long, + @Value("\${app.http.client.general.max-life-seconds:120}") + private val generalMaxLifeSeconds: Long, + @Value("\${app.http.client.general.pending-acquire-timeout-seconds:30}") + private val generalPendingAcquireTimeoutSeconds: Long, + @Value("\${app.http.client.general.connect-timeout-millis:5000}") + private val generalConnectTimeoutMillis: Int, + @Value("\${app.http.client.general.response-timeout-seconds:30}") + private val generalResponseTimeoutSeconds: Long, + @Value("\${app.http.client.keycloak.max-connections:150}") + private val keycloakMaxConnections: Int, + @Value("\${app.http.client.keycloak.pending-acquire-max-count:500}") + private val keycloakPendingAcquireMaxCount: Int, + @Value("\${app.http.client.keycloak.max-idle-seconds:30}") + private val keycloakMaxIdleSeconds: Long, + @Value("\${app.http.client.keycloak.max-life-seconds:120}") + private val keycloakMaxLifeSeconds: Long, + @Value("\${app.http.client.keycloak.pending-acquire-timeout-seconds:60}") + private val keycloakPendingAcquireTimeoutSeconds: Long, + @Value("\${app.http.client.keycloak.connect-timeout-millis:10000}") + private val keycloakConnectTimeoutMillis: Int, + @Value("\${app.http.client.keycloak.response-timeout-seconds:10}") + private val keycloakResponseTimeoutSeconds: Long, ) { private val provider = ConnectionProvider.builder("apiPool") - .maxConnections(150) - .pendingAcquireMaxCount(100) - .maxIdleTime(Duration.ofSeconds(30)) - .maxLifeTime(Duration.ofMinutes(2)) - .pendingAcquireTimeout(Duration.ofSeconds(10)) + .maxConnections(generalMaxConnections) + .pendingAcquireMaxCount(generalPendingAcquireMaxCount) + .maxIdleTime(Duration.ofSeconds(generalMaxIdleSeconds)) + .maxLifeTime(Duration.ofSeconds(generalMaxLifeSeconds)) + .pendingAcquireTimeout(Duration.ofSeconds(generalPendingAcquireTimeoutSeconds)) .evictInBackground(Duration.ofMinutes(1)) .build() - private val client = HttpClient.create(provider) - .wiretap("reactor.netty.http.client.HttpClient", LogLevel.DEBUG, AdvancedByteBufFormat.SIMPLE) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000) - .responseTimeout(Duration.ofSeconds(30)) - .keepAlive(true) - .doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) } + private val client = HttpClient.create(provider).let { + val configured = if (wiretapEnabled) { + it.wiretap("reactor.netty.http.client.HttpClient", LogLevel.DEBUG, AdvancedByteBufFormat.SIMPLE) + } else { + it + } + configured + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, generalConnectTimeoutMillis) + .responseTimeout(Duration.ofSeconds(generalResponseTimeoutSeconds)) + .keepAlive(true) + .doOnConnected { conn -> conn.addHandlerLast(LogbookClientHandler(logbook)) } + } @Bean("generalWebClient") @@ -63,16 +99,17 @@ class WebClientConfig( @Bean("keycloakWebClient") fun keycloakWebClient(logbook: Logbook): WebClient { val provider = ConnectionProvider.builder("keycloakPool") - .maxConnections(100) - .maxIdleTime(Duration.ofSeconds(30)) - .maxLifeTime(Duration.ofMinutes(2)) - .pendingAcquireTimeout(Duration.ofSeconds(60)) + .maxConnections(keycloakMaxConnections) + .pendingAcquireMaxCount(keycloakPendingAcquireMaxCount) + .maxIdleTime(Duration.ofSeconds(keycloakMaxIdleSeconds)) + .maxLifeTime(Duration.ofSeconds(keycloakMaxLifeSeconds)) + .pendingAcquireTimeout(Duration.ofSeconds(keycloakPendingAcquireTimeoutSeconds)) .evictInBackground(Duration.ofMinutes(1)) .build() val client = HttpClient.create(provider) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) - .responseTimeout(Duration.ofSeconds(10)) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, keycloakConnectTimeoutMillis) + .responseTimeout(Duration.ofSeconds(keycloakResponseTimeoutSeconds)) .keepAlive(true) .doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) } diff --git a/api/api-app/src/main/resources/application.yml b/api/api-app/src/main/resources/application.yml index ec0a0b34f..aa3606c00 100644 --- a/api/api-app/src/main/resources/application.yml +++ b/api/api-app/src/main/resources/application.yml @@ -107,10 +107,10 @@ logging: level: co.nilin: INFO org.zalando.logbook: TRACE - reactor.netty.pool: DEBUG - reactor.netty.http.client: DEBUG - org.springframework.web.reactive.function.client: DEBUG - co.nilin.opex.api.ports.proxy.impl: DEBUG + reactor.netty.pool: WARN + reactor.netty.http.client: WARN + org.springframework.web.reactive.function.client: WARN + co.nilin.opex.api.ports.proxy.impl: INFO app: base: @@ -152,6 +152,31 @@ app: api: crypto: key: ${api_crypto_key:0e1fd29572ec8c85970d76e3433e96ee} + http: + client: + wiretap: + enabled: ${HTTP_CLIENT_WIRETAP_ENABLED:false} + general: + max-connections: ${HTTP_CLIENT_GENERAL_MAX_CONNECTIONS:300} + pending-acquire-max-count: ${HTTP_CLIENT_GENERAL_PENDING_ACQUIRE_MAX_COUNT:1000} + max-idle-seconds: ${HTTP_CLIENT_GENERAL_MAX_IDLE_SECONDS:30} + max-life-seconds: ${HTTP_CLIENT_GENERAL_MAX_LIFE_SECONDS:120} + pending-acquire-timeout-seconds: ${HTTP_CLIENT_GENERAL_PENDING_ACQUIRE_TIMEOUT_SECONDS:30} + connect-timeout-millis: ${HTTP_CLIENT_GENERAL_CONNECT_TIMEOUT_MILLIS:5000} + response-timeout-seconds: ${HTTP_CLIENT_GENERAL_RESPONSE_TIMEOUT_SECONDS:30} + keycloak: + max-connections: ${HTTP_CLIENT_KEYCLOAK_MAX_CONNECTIONS:150} + pending-acquire-max-count: ${HTTP_CLIENT_KEYCLOAK_PENDING_ACQUIRE_MAX_COUNT:500} + max-idle-seconds: ${HTTP_CLIENT_KEYCLOAK_MAX_IDLE_SECONDS:30} + max-life-seconds: ${HTTP_CLIENT_KEYCLOAK_MAX_LIFE_SECONDS:120} + pending-acquire-timeout-seconds: ${HTTP_CLIENT_KEYCLOAK_PENDING_ACQUIRE_TIMEOUT_SECONDS:60} + connect-timeout-millis: ${HTTP_CLIENT_KEYCLOAK_CONNECT_TIMEOUT_MILLIS:10000} + response-timeout-seconds: ${HTTP_CLIENT_KEYCLOAK_RESPONSE_TIMEOUT_SECONDS:10} + proxy: + market: + max-concurrent-requests: ${API_PROXY_MARKET_MAX_CONCURRENT_REQUESTS:64} + matching: + max-concurrent-requests: ${API_PROXY_MATCHING_MAX_CONCURRENT_REQUESTS:64} cors: enabled: true allowed-origins: ${ALLOWED_ORIGINS:"http://localhost:8110"} @@ -170,4 +195,3 @@ springdoc: display-request-duration: true operations-sorter: method tags-sorter: alpha - diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt index 1dbbda2aa..48e8aeecd 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt @@ -4,8 +4,30 @@ import kotlinx.coroutines.reactor.asCoroutineDispatcher import reactor.core.scheduler.Schedulers object ProxyDispatchers { + private fun envInt(name: String, default: Int): Int { + val value = System.getenv(name)?.toIntOrNull() ?: return default + return if (value > 0) value else default + } - val general = Schedulers.newBoundedElastic(8, 16, "general").asCoroutineDispatcher() - val market = Schedulers.newBoundedElastic(8, 16, "market").asCoroutineDispatcher() - val wallet = Schedulers.newBoundedElastic(10, 20, "wallet").asCoroutineDispatcher() + private val cpu = Runtime.getRuntime().availableProcessors().coerceAtLeast(4) + private val defaultThreads = cpu * 4 + private val defaultQueue = 10_000 + + val general = Schedulers.newBoundedElastic( + envInt("API_PROXY_GENERAL_THREADS", defaultThreads), + envInt("API_PROXY_GENERAL_QUEUE", defaultQueue), + "general" + ).asCoroutineDispatcher() + + val market = Schedulers.newBoundedElastic( + envInt("API_PROXY_MARKET_THREADS", defaultThreads), + envInt("API_PROXY_MARKET_QUEUE", defaultQueue), + "market" + ).asCoroutineDispatcher() + + val wallet = Schedulers.newBoundedElastic( + envInt("API_PROXY_WALLET_THREADS", defaultThreads), + envInt("API_PROXY_WALLET_QUEUE", defaultQueue), + "wallet" + ).asCoroutineDispatcher() } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt index b40cfdc03..f67f0c113 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt @@ -34,6 +34,9 @@ class MarketUserDataProxyImpl(@Qualifier("generalWebClient") private val webClie @Value("\${app.market.url}") private lateinit var baseUrl: String + + @Value("\${app.proxy.market.max-concurrent-requests:64}") + private var marketMaxConcurrentRequests: Int = 64 private suspend fun retryOnce(backoffMs: Long = 200, block: suspend () -> T): T = try { block() @@ -41,7 +44,9 @@ class MarketUserDataProxyImpl(@Qualifier("generalWebClient") private val webClie delay(backoffMs); block() } - private val mgLimiter = Semaphore(permits = 16, acquiredPermits = 0) + private val mgLimiter by lazy { + Semaphore(permits = marketMaxConcurrentRequests, acquiredPermits = 0) + } override suspend fun queryOrder( token: String, diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt index a0718dba0..58ef1533d 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt @@ -39,7 +39,13 @@ class MatchingGatewayProxyImpl(@Qualifier("generalWebClient") private val client @Value("\${app.matching-gateway.url}") private lateinit var baseUrl: String - private val mgLimiter = Semaphore(permits = 16, acquiredPermits = 0) // fair-like behavior + + @Value("\${app.proxy.matching.max-concurrent-requests:64}") + private var matchingMaxConcurrentRequests: Int = 64 + + private val mgLimiter by lazy { + Semaphore(permits = matchingMaxConcurrentRequests, acquiredPermits = 0) + } override suspend fun createNewOrder( uuid: String?, pair: String, diff --git a/device-management/pom.xml b/device-management/pom.xml index fc80e43f7..0923d7cdd 100644 --- a/device-management/pom.xml +++ b/device-management/pom.xml @@ -22,7 +22,7 @@ 2.1.0 3.4.2 2024.0.0 - 1.2.25 + 1.2.26 diff --git a/pom.xml b/pom.xml index 853d3c8d5..e49130f86 100644 --- a/pom.xml +++ b/pom.xml @@ -16,7 +16,7 @@ 1.9.0 2.7.6 2021.0.5 - 1.2.25 + 1.2.26 1.0.8 true 1.0.1-beta.38