Kafka + HPA 테스트

쿠버네티스를 공부해보며 도입해볼만한 것들을 추리고 있는데, HPA를 활용한 어플리케이션 병렬처리가 Kafka의 Consumer Group을 다루기에 궁합이 좋다고 느껴 테스트를 해보았습니다. 

 

++ Kafka 선택 이유

레디스와 카프카를 고민해보긴했는데 온프레미스 환경에서 서비스가 돌아가기 때문에 In-memory로는 Scale-out, Scale-up에 한계가 있어 카프카를 선택하게 되었습니다. 

---

 

테스트

 

실행환경

local : minikube

내부망 : Server 1 (Kafka)

 

동작

1. Elasticsearch 에서 문서를 가져와 Kafka (topic 1)에 적재

2. topic 1에 적재된 문서들을 가져와 전처리 (간단한 Key, value 추가) 후 다시 Kafka (topic 2)에 적재

 

Kafka client

public class KafkaUtil {

    private static final String bootstrapServer = "192.168.125.64:9092";
    Properties properties = new Properties();

    private Properties setProperties(SwitchParameters switchParameters) {
        switch (switchParameters) {
            case producer:
                properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
                // 전송/수신 하는 데이터에 맞추어 Serializer 설정 -> 문자열일 시 StringSerializer
                properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
                properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
                properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
                properties.setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, "30000");
                return properties;
            case consumer:
                properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
                // 전송/수신 하는 데이터에 맞추어 Serializer 설정 -> 문자열일 시 StringSerializer
                properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "testtopic");
                properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

                return properties;
        }
        return null;
    }

    public KafkaProducer getProducer() {
        properties = setProperties(SwitchParameters.producer);
        System.out.println("get Properties : " + properties);
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        return producer;
    }

    public KafkaConsumer getConsumer() {
        properties = setProperties(SwitchParameters.consumer);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        return consumer;
    }
}

Maven 환경

- Client를 이용해 pub, sub 기능 구현

 

Kafka-hpa.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
	name: kafka-hpa
spec:
	selector:
		matchLabels:
			run: kafka-hpa
	replicas: 1
	template:
		metadata:
			labels:
				run: kafka-hpa
		spec:
			containers:
				- name: kafka-hpa
					image: baektothefuture98/kafkatest:0.0.5
					ports: 
						- container: 9092
					resources:
						limits:
							memory: 128Mi
							cpu: 500m
						requests:
							memory: 64Mi
							cpu: 300m
___
apiVersion: v1
kind: Service
metadata:
	name: kafka-hpa
	labels:
		run: kafka-hpa
spec:
	ports:
		- port: 0
	selector:
		run: kafka-hpa

해당 Yaml 파일 작성 후 아래 명령어로 등록

1. kubectl apply -f <kafka-hpa.yaml 파일 위치>

2. kubectl autoscale deployment/kafka-hpa --min=1 --max=4 --cpu-percent=10
-> 최소 1개의 pod - 최대 4개의 pod까지 / cpu 한계 10 퍼센트일 시 적용 (테스트 차 낮게 적용)

Kafka 모니터링 도구 - 임시

https://github.com/provectus/kafka-ui

GitHub - provectus/kafka-ui: Open-Source Web UI for Apache Kafka Management

 

결과

kubectl get hpa를 통해 CPU 한계 시 replicas(pod)가 늘어나는지 확인

-> Consumer 어플리케이션에 HTTP API가 없기에 K6 테스트가 아닌 한계를 낮춰 파드 증가 확인

Replicas가 1 -> 4로 CPU 증가량에 따라 설정된 MaxPods만큼 증가하는 것을 확인

minikube dashboard  확인

 

총 100만건 인입
평균 전송 속도 (100건 기준) : 0.071 sec
CPU 사용률 (LOAD)  : 최대 90%