diff --git a/.gitignore b/.gitignore index 259113f73..440730d57 100644 --- a/.gitignore +++ b/.gitignore @@ -18,7 +18,7 @@ target/ *.iws *.iml *.ipr -./http-client.env.json +http-client.env.json ### NetBeans ### /nbproject/private/ @@ -33,3 +33,6 @@ build/ ### VS Code ### .vscode/ + +grafana/data/* +prometheus/data/* \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index cc6f2e042..3d9587cd0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,7 +6,7 @@ RUN mvn dependency:go-offline COPY src src RUN mvn package -FROM openjdk:17-jdk-slim +FROM eclipse-temurin:17-alpine-3.22 COPY --from=build /app/target/*.jar /high-load-course.jar diff --git a/grafana/provisioning/dashboards/ServicesStatistic.json b/grafana/provisioning/dashboards/ServicesStatistic.json index 684b97269..9de50eb73 100644 --- a/grafana/provisioning/dashboards/ServicesStatistic.json +++ b/grafana/provisioning/dashboards/ServicesStatistic.json @@ -1545,16 +1545,13 @@ "viz": false }, "insertNulls": false, - "lineInterpolation": "smooth", - "lineStyle": { - "fill": "solid" - }, + "lineInterpolation": "linear", "lineWidth": 1, "pointSize": 5, "scaleDistribution": { "type": "linear" }, - "showPoints": "never", + "showPoints": "auto", "spanNulls": false, "stacking": { "group": "A", @@ -1570,90 +1567,69 @@ "steps": [ { "color": "green", - "value": null + "value": 0 }, { "color": "red", "value": 80 } ] - }, - "unit": "ms" - }, - "overrides": [ - { - "matcher": { - "id": "byName", - "options": "success" - }, - "properties": [ - { - "id": "color", - "value": { - "fixedColor": "green", - "mode": "fixed" - } - } - ] - }, - { - "matcher": { - "id": "byName", - "options": "fail" - }, - "properties": [ - { - "id": "color", - "value": { - "fixedColor": "red", - "mode": "fixed" - } - } - ] } - ] + }, + "overrides": [] }, "gridPos": { "h": 8, "w": 12, - "x": 12, + "x": 0, "y": 4 }, - "id": 14, + "id": 101, "options": { "legend": { - "calcs": [ - "max" - ], - "displayMode": "table", - "placement": "right", - "showLegend": true, - "sortBy": "Max", - "sortDesc": true + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true }, "tooltip": { + "hideZeros": false, "mode": "single", "sort": "none" } }, - "pluginVersion": "11.4.0", + "pluginVersion": "12.1.1", "targets": [ + { + "disableTextWrap": false, + "editorMode": "builder", + "expr": "rate(http_requests_served_total{endpoint=~\"pay_order\"}[$__rate_interval])", + "fullMetaSearch": false, + "includeNullMetadata": true, + "legendFormat": "__auto", + "range": true, + "refId": "A", + "useBackend": false + }, { "datasource": { "type": "prometheus", "uid": "PBFA97CFB590B2093" }, - "editorMode": "code", - "exemplar": true, - "expr": "rate(http_external_duration_sum{service=~\"$service\"}[1m]) / rate(http_external_duration_count{service=~\"$service\"}[1m])", + "disableTextWrap": false, + "editorMode": "builder", + "expr": "rate(payment_service_sanded_total{service=~\"$service\"}[$__rate_interval])", + "fullMetaSearch": false, "hide": false, - "interval": "", - "legendFormat": "{{method}} - {{result}}", + "includeNullMetadata": true, + "instant": false, + "legendFormat": "__auto", "range": true, - "refId": "C" + "refId": "B", + "useBackend": false } ], - "title": "(Clients -> Online shop) http requests duration. Average", + "title": "sending rate & processing speed", "type": "timeseries" }, { diff --git a/http-client.env.json b/http-client.env.json new file mode 100644 index 000000000..3eac6dbd5 --- /dev/null +++ b/http-client.env.json @@ -0,0 +1,68 @@ +{ + "Lab4 :: acc-18": { + "accounts": "acc-18", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "200", + "testCount": "6000", + "processingTimeMillis": "3000" + }, + "Lab4 :: acc-23": { + "accounts": "acc-23", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "16", + "testCount": "1600", + "processingTimeMillis": "30000" + }, + "Lab5 :: acc-23 :: v1": { + "accounts": "acc-23", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "15", + "testCount": "3000", + "processingTimeMillis": "2500" + }, + "Lab5 :: acc-23 :: v2": { + "accounts": "acc-23", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "11", + "testCount": "2200", + "processingTimeMillis": "13000", + "profile": "s_0.7_60" + }, + "Lab5 :: acc-23 :: v3": { + "accounts": "acc-23", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "3", + "testCount": "1150", + "processingTimeMillis": "26000", + "runits": "90" + }, + "Lab6 :: acc-8 :: v3": { + "accounts": "acc-8", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "7", + "testCount": "800", + "processingTimeMillis": "3500" + }, + "Lab8 :: acc-9": { + "accounts": "acc-9", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "100", + "testCount": "5000", + "processingTimeMillis": "20000" + }, + "Lab9 :: acc-12": { + "accounts": "acc-12", + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "ratePerSecond": "1000", + "testCount": "200000", + "processingTimeMillis": "50000" + } +} diff --git a/pom.xml b/pom.xml index 0e7c0ad2d..204fcd924 100644 --- a/pom.xml +++ b/pom.xml @@ -126,6 +126,10 @@ io.micrometer micrometer-registry-prometheus + + io.micrometer + micrometer-registry-prometheus-simpleclient + org.jetbrains.kotlin kotlin-test @@ -175,7 +179,7 @@ spring - 1.8 + 17 diff --git a/src/main/kotlin/ru/quipy/apigateway/APIController.kt b/src/main/kotlin/ru/quipy/apigateway/APIController.kt index 6f23fa18d..9c2b06a97 100644 --- a/src/main/kotlin/ru/quipy/apigateway/APIController.kt +++ b/src/main/kotlin/ru/quipy/apigateway/APIController.kt @@ -1,18 +1,34 @@ package ru.quipy.apigateway +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.MeterRegistry import org.slf4j.Logger import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired -import org.springframework.web.bind.annotation.* +import org.springframework.http.HttpHeaders +import org.springframework.http.HttpStatus +import org.springframework.http.ResponseEntity +import org.springframework.web.bind.annotation.ExceptionHandler +import org.springframework.web.bind.annotation.PathVariable +import org.springframework.web.bind.annotation.PostMapping +import org.springframework.web.bind.annotation.RequestBody +import org.springframework.web.bind.annotation.RequestParam +import org.springframework.web.bind.annotation.RestController +import ru.quipy.common.utils.TooManyRequestsException import ru.quipy.orders.repository.OrderRepository import ru.quipy.payments.logic.OrderPayer -import java.util.* +import java.util.UUID @RestController -class APIController { +class APIController(@Autowired val meterRegistry: MeterRegistry) { val logger: Logger = LoggerFactory.getLogger(APIController::class.java) + private val orderCounter: Counter = Counter.builder("http_requests_served") + .description("Total number of served http requests for order payment") + .tag("endpoint", "pay_order") + .register(meterRegistry) + @Autowired private lateinit var orderRepository: OrderRepository @@ -21,6 +37,7 @@ class APIController { @PostMapping("/users") fun createUser(@RequestBody req: CreateUserRequest): User { + orderCounter.increment() return User(UUID.randomUUID(), req.name) } @@ -30,6 +47,7 @@ class APIController { @PostMapping("/orders") fun createOrder(@RequestParam userId: UUID, @RequestParam price: Int): Order { + orderCounter.increment() val order = Order( UUID.randomUUID(), userId, @@ -57,6 +75,7 @@ class APIController { @PostMapping("/orders/{orderId}/payment") fun payOrder(@PathVariable orderId: UUID, @RequestParam deadline: Long): PaymentSubmissionDto { val paymentId = UUID.randomUUID() + orderCounter.increment() val order = orderRepository.findById(orderId)?.let { orderRepository.save(it.copy(status = OrderStatus.PAYMENT_IN_PROGRESS)) it @@ -67,6 +86,12 @@ class APIController { return PaymentSubmissionDto(createdAt, paymentId) } + @ExceptionHandler(TooManyRequestsException::class) + fun tooManyRequestsExceptionHandler(exception: TooManyRequestsException) = + ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS) + .headers(HttpHeaders().apply { add(HttpHeaders.RETRY_AFTER, exception.delay.toString()) }) + .body(exception.message) + class PaymentSubmissionDto( val timestamp: Long, val transactionId: UUID diff --git a/src/main/kotlin/ru/quipy/common/utils/InstantRateLimitSemaphore.kt b/src/main/kotlin/ru/quipy/common/utils/InstantRateLimitSemaphore.kt new file mode 100644 index 000000000..7c8853ca8 --- /dev/null +++ b/src/main/kotlin/ru/quipy/common/utils/InstantRateLimitSemaphore.kt @@ -0,0 +1,46 @@ +package ru.quipy.common.utils + +import kotlinx.coroutines.sync.Semaphore +import java.time.Duration +import java.time.Instant +import java.util.concurrent.TimeUnit +import kotlin.math.floor + +/** + * Обёртка над семафором, которая учитывает сколько семафоров можно взять в единицу времени и если до указанного + * момента осталось меньше, чем limitDuration единиц времени, то ограничивает возможность взятия семафора + */ +class InstantRateLimitSemaphore( + limitDuration: Duration, + private val rateDuration: Duration, + private val unitsPerRate: Int +): RateLimitSemaphore(Semaphore(durationCapacity(limitDuration, rateDuration, unitsPerRate)) ) { + + constructor(limitDuration: Duration, timeUnit: TimeUnit, unitsPerRate: Int) : + this(limitDuration, Duration.ofNanos(timeUnit.toNanos(1)), unitsPerRate) + + /** + * Функция вычисляет допустимый limit взятий семафора на заданный период + */ + override fun limit(volume: Instant): Int = + Duration.between(Instant.now(), volume) + ?.takeIf { it.toNanos() > 0 } + ?.let { durationCapacity(it, rateDuration, unitsPerRate) } + ?: 0 + + companion object { + /** + * Вычисление количества ресурса на Duration при известном объёме на единицу TimeUnit + * @param duration Длительность + * @param rateDuration Длительность периода + * @param unitsPerRate Количество ресурса на период + * @return количество элементов, доступных на duration + */ + private fun durationCapacity( + duration: Duration, + rateDuration: Duration, + unitsPerRate: Int) = + floor(unitsPerRate.toDouble() * duration.toNanos() / rateDuration.toNanos()).toInt() + } + +} diff --git a/src/main/kotlin/ru/quipy/common/utils/RateLimitSemaphore.kt b/src/main/kotlin/ru/quipy/common/utils/RateLimitSemaphore.kt new file mode 100644 index 000000000..b96daaef4 --- /dev/null +++ b/src/main/kotlin/ru/quipy/common/utils/RateLimitSemaphore.kt @@ -0,0 +1,44 @@ +package ru.quipy.common.utils + +import kotlinx.coroutines.sync.Semaphore +import java.util.concurrent.atomic.AtomicInteger + +abstract class RateLimitSemaphore protected constructor (private val semaphore: Semaphore) { + + val acquires = AtomicInteger() + + fun acquire(value: T): Boolean { + return limitAcquire(limit(value)) + } + + fun release() { + try { + semaphore.release() + acquires.decrementAndGet() + } catch (ex : IllegalStateException) { + throw ex + } + } + + /** + * Функция вычисления limit для заданного объёма + */ + protected abstract fun limit(volume: T): Int + + /** + * Метод берёт семафор, но ограничивая его capacity в пределах limit, который меньше, чем даёт сам семафор + */ + private fun limitAcquire(limit: Int): Boolean { + // Берём семафор + if (limit > 0 && semaphore.tryAcquire()) { + // Если успешно получили, то оцениваем количество захватов + if (acquires.incrementAndGet() <= limit) { + // Если не превысили лимит, то возвращаем, что всё ОК + return true + } + release() + } + return false + } + +} \ No newline at end of file diff --git a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt index 6ff3092ab..348fb2f3a 100644 --- a/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt +++ b/src/main/kotlin/ru/quipy/common/utils/SlidingWindowRateLimiter.kt @@ -10,8 +10,7 @@ import java.time.Duration import java.util.concurrent.Executors import java.util.concurrent.PriorityBlockingQueue import java.util.concurrent.atomic.AtomicLong -import java.util.concurrent.locks.ReentrantLock -import kotlin.concurrent.withLock +import java.util.concurrent.locks.LockSupport class SlidingWindowRateLimiter( private val rate: Long, @@ -39,6 +38,17 @@ class SlidingWindowRateLimiter( } } + /** + * Пытаемся взять блокировку до заданного момента времени + */ + fun blockingUntil(instant: Long): Boolean { + while(System.currentTimeMillis() < instant) { + if (tick()) return true + LockSupport.parkNanos(Duration.ofMillis(1).toNanos()) + } + return false + } + data class Measure( val value: Long, val timestamp: Long @@ -67,4 +77,4 @@ class SlidingWindowRateLimiter( companion object { private val logger: Logger = LoggerFactory.getLogger(SlidingWindowRateLimiter::class.java) } -} \ No newline at end of file +} diff --git a/src/main/kotlin/ru/quipy/common/utils/TooManyRequestsException.kt b/src/main/kotlin/ru/quipy/common/utils/TooManyRequestsException.kt new file mode 100644 index 000000000..8da775dae --- /dev/null +++ b/src/main/kotlin/ru/quipy/common/utils/TooManyRequestsException.kt @@ -0,0 +1,3 @@ +package ru.quipy.common.utils + +class TooManyRequestsException(val delay: Int = 1) : Exception("Too many requests") \ No newline at end of file diff --git a/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt b/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt index 9bcb80d07..a2958c354 100644 --- a/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt +++ b/src/main/kotlin/ru/quipy/config/EventSourcingLibConfiguration.kt @@ -66,12 +66,23 @@ class EventSourcingLibConfiguration { } } - @Bean // hack Jetty to tweak the number of possible https2 streams + @Bean // hack Jetty to tweak the number of possible https2 streams and optimize for high load fun jettyServerCustomizer(): JettyServletWebServerFactory { val jettyServletWebServerFactory = JettyServletWebServerFactory() - val c = JettyServerCustomizer { - (it.connectors[0].getConnectionFactory("h2c") as HTTP2CServerConnectionFactory).maxConcurrentStreams = 10_000_000 + val c = JettyServerCustomizer { server -> + // Настройка HTTP/2 + (server.connectors[0].getConnectionFactory("h2c") as HTTP2CServerConnectionFactory).maxConcurrentStreams = 10_000_000 + + // Оптимизация пула потоков для обработки входящих запросов + // Acceptors и selectors настраиваются через application.properties + server.connectors.forEach { connector -> + val executor = connector.executor + if (executor is org.eclipse.jetty.util.thread.QueuedThreadPool) { + executor.minThreads = 64 + executor.maxThreads = 200 + } + } } jettyServletWebServerFactory.serverCustomizers.add(c) diff --git a/src/main/kotlin/ru/quipy/payments/config/PaymentAccountsConfig.kt b/src/main/kotlin/ru/quipy/payments/config/PaymentAccountsConfig.kt index eceb90cff..edaa7a8bf 100644 --- a/src/main/kotlin/ru/quipy/payments/config/PaymentAccountsConfig.kt +++ b/src/main/kotlin/ru/quipy/payments/config/PaymentAccountsConfig.kt @@ -8,12 +8,15 @@ import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate -import ru.quipy.payments.logic.* +import ru.quipy.payments.logic.PaymentAccountProperties +import ru.quipy.payments.logic.PaymentAggregateState +import ru.quipy.payments.logic.PaymentExternalSystemAdapter +import ru.quipy.payments.logic.PaymentExternalSystemAdapterImpl import java.net.URI import java.net.http.HttpClient import java.net.http.HttpRequest import java.net.http.HttpResponse -import java.util.* +import java.util.UUID @Configuration @@ -36,7 +39,7 @@ class PaymentAccountsConfig { lateinit var allowedAccounts: List @Bean - fun accountAdapters(paymentService: EventSourcingService): List { + fun paymentAccountProperties(): List { val request = HttpRequest.newBuilder() .uri(URI("http://${paymentProviderHostPort}/external/accounts?serviceName=$serviceName&token=$token")) .GET() @@ -51,6 +54,14 @@ class PaymentAccountsConfig { ) .filter { it.accountName in allowedAccounts } .map { it.copy(enabled = true) } + } + + @Bean + fun accountAdapters( + paymentAccountProperties: List, + paymentService: EventSourcingService + ): List { + return paymentAccountProperties .onEach(::println) .map { PaymentExternalSystemAdapterImpl( diff --git a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt index a5909b85b..e610c1bd3 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/OrderPayer.kt @@ -5,19 +5,29 @@ import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import ru.quipy.common.utils.CallerBlockingRejectedExecutionHandler +import ru.quipy.common.utils.InstantRateLimitSemaphore import ru.quipy.common.utils.NamedThreadFactory +import ru.quipy.common.utils.TooManyRequestsException import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate -import java.util.* +import java.time.Duration +import java.time.Instant +import java.util.UUID import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit +import kotlin.math.roundToInt +import kotlin.math.roundToLong @Service -class OrderPayer { +class OrderPayer(paymentAccountProperties: List) { companion object { val logger: Logger = LoggerFactory.getLogger(OrderPayer::class.java) + const val MIN_PARALLEL_PROCESS = 64 + const val MAX_PARALLEL_PROCESS = 512 + const val DELAY_COEFFICIENT = 0.0 + const val MIN_DELAY_ADD_MILLIS = 10L } @Autowired @@ -26,30 +36,79 @@ class OrderPayer { @Autowired private lateinit var paymentService: PaymentService + private val parallelThreads = paymentAccountProperties + .sumOf { it.rateLimitPerSec.coerceAtMost(MAX_PARALLEL_PROCESS) } + private val poolSize = (parallelThreads + 2) + .coerceAtLeast(MIN_PARALLEL_PROCESS) + .coerceAtMost(MAX_PARALLEL_PROCESS) + + // Увеличена очередь до 50000 для обработки большего количества запросов private val paymentExecutor = ThreadPoolExecutor( - 16, - 16, + poolSize, + poolSize, 0L, TimeUnit.MILLISECONDS, - LinkedBlockingQueue(8_000), + LinkedBlockingQueue(50_000), NamedThreadFactory("payment-submission-executor"), CallerBlockingRejectedExecutionHandler() ) + private val callsPerMinute = paymentAccountProperties.sumOf { + (TimeUnit.MINUTES.toNanos(1).toDouble() * + it.rateLimitPerSec.coerceAtMost(MAX_PARALLEL_PROCESS) / + it.averageProcessingTime.toNanos()).roundToInt()} + + private val minAverageProcessingTime = paymentAccountProperties.minOf { it.averageProcessingTime } + private val maxAverageProcessingTime = paymentAccountProperties.maxOf { it.averageProcessingTime } + + + /** + * Это предполагаемое время, которое может понадобиться внешнему сервису на выполнение нашего запроса + * Т.е.: если у нас до deathTime остаётся меньше callDelay, то мы его не ставим в очередь, а получаем + * отказ от семафора + */ + val callDelay: Duration = Duration + .ofMillis((minAverageProcessingTime.toMillis() * DELAY_COEFFICIENT) + .roundToLong() + .coerceAtLeast(MIN_DELAY_ADD_MILLIS)) + + val instantRateLimitSemaphore = + Triple( + // Делаем объем по задачам на несколько секунды вперед. + // То есть если есть свободные места в очереди на эти 3 величины обработки запроса и задаче не протухнет + // до того момента, когда сможет выполниться, то мы её ставим в очередь, а если нет, + // то возвращаем TooManyRequests + maxAverageProcessingTime.multipliedBy(3), + TimeUnit.MINUTES, + callsPerMinute + ).let { + logger.info("Create OrderPayer::InstantRateLimitSemaphore(duration=${it.first}, timeUnit=${it.second}, rate=${it.third})") + InstantRateLimitSemaphore(it.first, it.second, it.third) + } + fun processPayment(orderId: UUID, amount: Int, paymentId: UUID, deadline: Long): Long { val createdAt = System.currentTimeMillis() - paymentExecutor.submit { - val createdEvent = paymentESService.create { - it.create( - paymentId, - orderId, - amount - ) + val deadLineTime = Instant.ofEpochMilli(deadline) + if (instantRateLimitSemaphore.acquire(deadLineTime.minus(callDelay))) { + paymentExecutor.submit { + try { + val createdEvent = paymentESService.create { + it.create( + paymentId, + orderId, + amount + ) + } + logger.trace("Payment {} for order {} created.", createdEvent.paymentId, orderId) + } finally { + instantRateLimitSemaphore.release() + } + paymentService.submitPaymentRequest(paymentId, amount, createdAt, deadline) } - logger.trace("Payment ${createdEvent.paymentId} for order $orderId created.") - - paymentService.submitPaymentRequest(paymentId, amount, createdAt, deadline) + } else { + logger.error("Payment: $paymentId retried. Too many requests") + throw TooManyRequestsException() } return createdAt } -} \ No newline at end of file +} diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt index 5cb12106a..c4fdcfb29 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentExternalServiceImpl.kt @@ -2,15 +2,25 @@ package ru.quipy.payments.logic import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.module.kotlin.registerKotlinModule -import okhttp3.OkHttpClient -import okhttp3.Request -import okhttp3.RequestBody +import org.slf4j.Logger import org.slf4j.LoggerFactory +import ru.quipy.common.utils.NamedThreadFactory +import ru.quipy.common.utils.SlidingWindowRateLimiter import ru.quipy.core.EventSourcingService import ru.quipy.payments.api.PaymentAggregate -import java.net.SocketTimeoutException +import java.net.URI +import java.net.http.HttpClient +import java.net.http.HttpClient.Version +import java.net.http.HttpRequest +import java.net.http.HttpResponse import java.time.Duration -import java.util.* +import java.util.UUID +import java.util.concurrent.CompletableFuture +import java.util.concurrent.Executors +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger // Advice: always treat time as a Duration @@ -22,73 +32,151 @@ class PaymentExternalSystemAdapterImpl( ) : PaymentExternalSystemAdapter { companion object { - val logger = LoggerFactory.getLogger(PaymentExternalSystemAdapter::class.java) - - val emptyBody = RequestBody.create(null, ByteArray(0)) + val logger: Logger = LoggerFactory.getLogger(PaymentExternalSystemAdapter::class.java) val mapper = ObjectMapper().registerKotlinModule() } private val serviceName = properties.serviceName private val accountName = properties.accountName - private val requestAverageProcessingTime = properties.averageProcessingTime + private val averageProcessTime = properties.averageProcessingTime private val rateLimitPerSec = properties.rateLimitPerSec - private val parallelRequests = properties.parallelRequests - private val client = OkHttpClient.Builder().build() + private val httpClientExecutor = ThreadPoolExecutor( + 64, + 64, + 0, + TimeUnit.SECONDS, + LinkedBlockingQueue(100000), + NamedThreadFactory("payment-http-client") + ) + + private val dbExecutor = ThreadPoolExecutor( + 1000, + 1000, + 0, + TimeUnit.SECONDS, + LinkedBlockingQueue(50000), + NamedThreadFactory("payment-db-callback") + ) + + private val scheduler = Executors.newScheduledThreadPool(8, NamedThreadFactory("payment-hedge-scheduler")) + + private val client = HttpClient.newBuilder() + .version(Version.HTTP_2) + .executor(httpClientExecutor) + .connectTimeout(Duration.ofSeconds(3)) + .build() + + private val rateLimiter = SlidingWindowRateLimiter( + rateLimitPerSec.toLong(), + Duration.ofSeconds(1) + ) override fun performPaymentAsync(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { - logger.warn("[$accountName] Submitting payment request for payment $paymentId") - val transactionId = UUID.randomUUID() - // Вне зависимости от исхода оплаты важно отметить что она была отправлена. - // Это требуется сделать ВО ВСЕХ СЛУЧАЯХ, поскольку эта информация используется сервисом тестирования. + rateLimiter.tickBlocking() + paymentESService.update(paymentId) { it.logSubmission(success = true, transactionId, now(), Duration.ofMillis(now() - paymentStartedAt)) } - logger.info("[$accountName] Submit: $paymentId , txId: $transactionId") - - try { - val request = Request.Builder().run { - url("http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount") - post(emptyBody) - }.build() - - client.newCall(request).execute().use { response -> - val body = try { - mapper.readValue(response.body?.string(), ExternalSysResponse::class.java) - } catch (e: Exception) { - logger.error("[$accountName] [ERROR] Payment processed for txId: $transactionId, payment: $paymentId, result code: ${response.code}, reason: ${response.body?.string()}") - ExternalSysResponse(transactionId.toString(), paymentId.toString(),false, e.message) + performHedgedPayment(paymentId, amount, transactionId, deadline) + .exceptionally { exception -> + logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", exception) + paymentESService.update(paymentId) { + it.logProcessing(false, now(), transactionId, reason = exception.message ?: "Unknown error") } + false + } + } - logger.warn("[$accountName] Payment processed for txId: $transactionId, payment: $paymentId, succeeded: ${body.result}, message: ${body.message}") - - // Здесь мы обновляем состояние оплаты в зависимости от результата в базе данных оплат. - // Это требуется сделать ВО ВСЕХ ИСХОДАХ (успешная оплата / неуспешная / ошибочная ситуация) + /** + * Отправляет первый запрос и если за какое-то вермя не получен ответ, + * * то отправляет параллельный запрос с тем же transactionId в качестве + * ключа идемпотентности. Короче побеждает тот кто ответил первым + */ + private fun performHedgedPayment( + paymentId: UUID, + amount: Int, + transactionId: UUID, + deadline: Long, + ): CompletableFuture { + val result = CompletableFuture() + // сколько запросов у нас щас. когда будет 0, то значит что все запросы отправили уже и они вернулись с ответами(дай бог) + val pending = AtomicInteger(0) + + fun onSuccess(success: Boolean, reason: String?) { + // result.complete() атомарно вернёт true только первому вызову + if (result.complete(success)) { paymentESService.update(paymentId) { - it.logProcessing(body.result, now(), transactionId, reason = body.message) + it.logProcessing(success, now(), transactionId, reason = reason) } } - } catch (e: Exception) { - when (e) { - is SocketTimeoutException -> { - logger.error("[$accountName] Payment timeout for txId: $transactionId, payment: $paymentId", e) - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = "Request timeout.") - } - } + } - else -> { - logger.error("[$accountName] Payment failed for txId: $transactionId, payment: $paymentId", e) + fun onError(ex: Throwable) { + if (pending.decrementAndGet() == 0) { + result.completeExceptionally(ex) + } + } - paymentESService.update(paymentId) { - it.logProcessing(false, now(), transactionId, reason = e.message) - } - } + fun send() { + // Таймаут HTTP привязываем к дедлайну — не ждём 30с если платёж уже не успеет + val timeoutMs = remainingMillis(deadline).coerceAtLeast(500L) + sendSingleRequest(paymentId, amount, transactionId, timeoutMs) + .thenAcceptAsync({ (success, reason) -> onSuccess(success, reason) }, dbExecutor) + .exceptionally { ex -> onError(ex.cause ?: ex); null } + } + + pending.incrementAndGet() + send() + + fun scheduleHedge(delayMs: Long) { + if (remainingMillis(deadline) > delayMs) { + pending.incrementAndGet() + scheduler.schedule({ + if (!result.isDone) send() + else pending.decrementAndGet() + }, delayMs, TimeUnit.MILLISECONDS) } } + + val avg = averageProcessTime.toMillis() + // 5 попыток равномерно по шкале дедлайна + scheduleHedge((avg * 0.10).toLong().coerceAtLeast(150L)) // ~1300ms остаток + scheduleHedge((avg * 0.22).toLong().coerceAtLeast(280L)) // ~1170ms остаток + scheduleHedge((avg * 0.38).toLong().coerceAtLeast(450L)) // ~1000ms остаток + scheduleHedge((avg * 0.55).toLong().coerceAtLeast(650L)) // ~800ms остаток + + return result + } + + private fun sendSingleRequest( + paymentId: UUID, + amount: Int, + transactionId: UUID, + timeoutMs: Long = 30_000L, + ): CompletableFuture> { + val url = "http://$paymentProviderHostPort/external/process?serviceName=$serviceName&token=$token&accountName=$accountName&transactionId=$transactionId&paymentId=$paymentId&amount=$amount" + val request = HttpRequest.newBuilder() + .uri(URI.create(url)) + .version(Version.HTTP_2) + .POST(HttpRequest.BodyPublishers.noBody()) + .header("x-idempotency-key", transactionId.toString()) + .timeout(Duration.ofMillis(timeoutMs)) + .build() + + return client.sendAsync(request, HttpResponse.BodyHandlers.ofString()) + .thenApplyAsync({ response -> + val body = try { + mapper.readValue(response.body(), ExternalSysResponse::class.java) + } catch (e: Exception) { + logger.error("[$accountName] [ERROR] Payment response parse error for txId: $transactionId, payment: $paymentId, code: ${response.statusCode()}, body: ${response.body()}") + ExternalSysResponse(transactionId.toString(), paymentId.toString(), false, e.message) + } + Pair(body.result, body.message) + }, dbExecutor) } override fun price() = properties.price @@ -97,6 +185,11 @@ class PaymentExternalSystemAdapterImpl( override fun name() = properties.accountName + private fun remainingMillis(epocTime: Long) = + System.currentTimeMillis().takeIf { it < epocTime } + ?.let { epocTime - it } + ?: 0 + } -public fun now() = System.currentTimeMillis() \ No newline at end of file +fun now() = System.currentTimeMillis() \ No newline at end of file diff --git a/src/main/kotlin/ru/quipy/payments/logic/PaymentServiceImpl.kt b/src/main/kotlin/ru/quipy/payments/logic/PaymentServiceImpl.kt index 1c24e5a72..4b1f2d9e0 100644 --- a/src/main/kotlin/ru/quipy/payments/logic/PaymentServiceImpl.kt +++ b/src/main/kotlin/ru/quipy/payments/logic/PaymentServiceImpl.kt @@ -1,5 +1,7 @@ package ru.quipy.payments.logic +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.MeterRegistry import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @@ -15,8 +17,14 @@ import kotlin.concurrent.withLock @Service class PaymentSystemImpl( - private val paymentAccounts: List + private val paymentAccounts: List, + @Autowired val meterRegistry: MeterRegistry, ) : PaymentService { + private val ansCounter: Counter = Counter.builder("payment_service_sanded") + .description("Total number of sanded requests") + .tag("service", "payment_requests") + .register(meterRegistry) + companion object { val logger = LoggerFactory.getLogger(PaymentSystemImpl::class.java) } @@ -24,6 +32,7 @@ class PaymentSystemImpl( override fun submitPaymentRequest(paymentId: UUID, amount: Int, paymentStartedAt: Long, deadline: Long) { for (account in paymentAccounts) { account.performPaymentAsync(paymentId, amount, paymentStartedAt, deadline) + ansCounter.increment() } } } \ No newline at end of file diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 33d51a58b..2314a0ef3 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -1,8 +1,18 @@ server.address=0.0.0.0 -server.port=8081 +server.port=${SERVER_PORT:8081} server.http2.enabled=true spring.main.allow-bean-definition-overriding=true +# Jetty configuration for high load - максимальные ресурсы +server.jetty.threads.min=128 +server.jetty.threads.max=500 +server.jetty.acceptors=16 +server.jetty.selectors=32 +server.jetty.max-queue-capacity=2000000 + +# Spring MVC async configuration +spring.mvc.async.request-timeout=200000 + # MongoDB properties spring.data.mongodb.host=localhost spring.data.mongodb.port=27017 @@ -14,11 +24,16 @@ event.sourcing.scan-package=ru.quipy event.sourcing.snapshots-enabled=false event.sourcing.sagas-enabled=false -# Postgres event store properties +# Postgres event store properties - максимальные ресурсы для высокой нагрузки spring.datasource.hikari.jdbc-url=jdbc:postgresql://${POSTGRES_ADDRESS:localhost}:${POSTGRES_PORT:65432}/postgres spring.datasource.hikari.username=tiny_es spring.datasource.hikari.password=tiny_es -spring.datasource.hikari.leak-detection-threshold=2000 +spring.datasource.hikari.maximum-pool-size=150 +spring.datasource.hikari.minimum-idle=50 +spring.datasource.hikari.connection-timeout=20000 +spring.datasource.hikari.idle-timeout=300000 +spring.datasource.hikari.max-lifetime=1200000 +spring.datasource.hikari.leak-detection-threshold=5000 management.metrics.web.server.request.autotime.percentiles=0.95 management.metrics.export.prometheus.enabled=true @@ -26,5 +41,5 @@ management.endpoints.web.exposure.include=info,health,prometheus,metrics payment.service-name=${PAYMENT_SERVICE_NAME} payment.token=${PAYMENT_TOKEN} -payment.accounts=${PAYMENT_ACCOUNTS:acc-12,acc-20} +payment.accounts=acc-9 payment.hostPort=${PAYMENT_HOST:localhost}:${PAYMENT_PORT:1234} \ No newline at end of file diff --git a/test-local-run.http b/test-local-run.http index dc8dbeedf..f79c8cac1 100644 --- a/test-local-run.http +++ b/test-local-run.http @@ -5,11 +5,11 @@ Content-Type: application/json { "serviceName": "{{serviceName}}", "token": "{{token}}", - "ratePerSecond": 1, - "testCount": 100, - "processingTimeMillis": 80000 + "ratePerSecond": {{ratePerSecond}}, + "testCount": {{testCount}}, + "processingTimeMillis": {{processingTimeMillis}} } ### Stop running test to save time and resources # @timeout 120 -POST http://localhost:4321/test/stop/"{{serviceName}}" \ No newline at end of file +POST http://localhost:1234/test/stop/{{serviceName}} \ No newline at end of file diff --git a/test-on-prem-run.http b/test-on-prem-run.http index dd5765987..6947aff80 100644 --- a/test-on-prem-run.http +++ b/test-on-prem-run.http @@ -4,16 +4,15 @@ POST http://77.234.215.138:34321/run Content-Type: application/json { - "serviceName": "{{serviceName}}", - "token": "{{token}}", - "branch": "main", - "accounts": "acc-12,acc-20", - "ratePerSecond": 2, - "testCount": 10, - "processingTimeMillis": 80000, - "onPremises": true + "serviceName": "m3403-8", + "token": "bTgVPIlM5Jbr0T03=8", + "branch": "lab9_M", + "accounts": "acc-12", + "ratePerSecond": "1000", + "testCount": "200000", + "processingTimeMillis": "50000" } ### Stop running test to save credits # @timeout 120 -POST http://77.234.215.138:31234/test/stop/"{{serviceName}}" \ No newline at end of file +POST http://77.234.215.138:31234/test/stop/{{serviceName}}