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