CloudWatchClient로 CPUUtilization 메트릭 Data 가져오기 (feat. CPU 상태 확인) for Kafka Throttling

implementation("software.amazon.awssdk:cloudwatch:2.25.30")

 

import software.amazon.awssdk.services.cloudwatch.CloudWatchClient
import software.amazon.awssdk.services.cloudwatch.model.*

@Component
class CloudWatchMetricsService {

    private val cloudWatchClient: CloudWatchClient = CloudWatchClient.create()

    fun getCpuUtilization(instanceId: String): Double {
        val request = GetMetricStatisticsRequest.builder()
            .namespace("AWS/EC2")
            .metricName("CPUUtilization")
            .dimensions(
                Dimension.builder().name("InstanceId").value(instanceId).build()
            )
            .startTime(Instant.now().minus(Duration.ofMinutes(5)))
            .endTime(Instant.now())
            .period(60)
            .statistics(Statistic.AVERAGE)
            .build()

        val response = cloudWatchClient.getMetricStatistics(request)
        val dataPoints = response.datapoints()

        return dataPoints.maxByOrNull { it.timestamp() }?.average() ?: 0.0
    }
}

 

@Component
class KafkaThrottler(
    private val metricsService: CloudWatchMetricsService,
    private val kafkaListenerEndpointRegistry: KafkaListenerEndpointRegistry
) {
    private val listenerId = "id"

    @Scheduled(fixedRate = 10000)
    fun monitorCpuAndThrottleKafka() {
        val cpu = metricsService.getCpuUtilization("   ")
        val container = kafkaListenerEndpointRegistry.getListenerContainer(listenerId)

        if (cpu > 75) {
            container.pause()
            println("CPU high ($cpu%) → Kafka paused")
        } else {
            container.resume()
            println("CPU normal ($cpu%) → Kafka resumed")
        }
    }
}

 

instance ID

IAM 권한 - cloudwatch - GetMetricsStatistics

Kafka Listener ID