ELK Stack 테스트 전략 완벽 가이드

Elasticsearch, Logstash, Kibana로 구성된 ELK Stack의 통합 테스트 전략을 다룹니다. 실무에서 활용할 수 있는 단위 테스트부터 통합 테스트까지 체계적인 테스트 방법을 제공합니다.

---

---

들어가며

이 글에서는 ELK Stack 테스트 전략 완벽 가이드에 대해 상세히 알아보겠습니다. 총 12가지 주요 개념을 다루며, 각각의 개념에 대한 설명과 실제 코드 예제를 함께 제공합니다.

목차

1. Elasticsearch_단위_테스트_기본
2. Testcontainers로_실제_Elasticsearch_테스트
3. Logstash_파이프라인_테스트
4. Kibana_API_테스트
5. ELK_통합_테스트_시나리오
6. 인덱스_매핑_검증_테스트
7. 성능_테스트_벤치마크
8. Logstash_에러_핸들링_테스트
9. Kibana_대시보드_자동_배포_테스트
10. 알람_규칙_테스트
11. 인덱스_라이프사이클_관리_테스트
12. 멀티_노드_클러스터_테스트

---

1. Elasticsearch 단위 테스트 기본

개요

Elasticsearch 클라이언트를 Mock하여 독립적인 단위 테스트를 수행합니다. 실제 ES 인스턴스 없이도 로직을 검증할 수 있습니다.

코드 예제

from unittest.mock import Mock
from elasticsearch import Elasticsearch

def test_index_document():
    es_mock = Mock(spec=Elasticsearch)
    es_mock.index.return_value = {'result': 'created'}

    result = es_mock.index(index='test', body={'field': 'value'})

    assert result['result'] == 'created'
    es_mock.index.assert_called_once()

설명

Mock 객체를 사용해 Elasticsearch 클라이언트의 동작을 시뮬레이션하고, 메서드 호출 여부와 반환값을 검증합니다.

---

2. Testcontainers로 실제 Elasticsearch 테스트

개요

Testcontainers를 활용하여 실제 Elasticsearch 컨테이너를 띄워 통합 테스트를 수행합니다. 실제 환경과 동일한 조건에서 테스트할 수 있습니다.

코드 예제

from testcontainers.elasticsearch import ElasticSearchContainer
from elasticsearch import Elasticsearch

def test_real_elasticsearch():
    with ElasticSearchContainer("elasticsearch:8.11.0") as es:
        client = Elasticsearch(es.get_url())

        client.index(index='test', document={'name': 'John'})
        result = client.search(index='test', query={'match_all': {}})

        assert result['hits']['total']['value'] == 1

설명

Docker 컨테이너로 실제 Elasticsearch를 실행하여 인덱싱, 검색 등의 전체 워크플로우를 테스트합니다.

---

3. Logstash 파이프라인 테스트

개요

Logstash 파이프라인의 필터 로직을 검증합니다. 입력 데이터가 올바르게 파싱되고 변환되는지 확인합니다.

코드 예제

import subprocess
import json

def test_logstash_pipeline():
    input_data = '{"message": "ERROR: Database connection failed"}'

    result = subprocess.run(
        ['logstash', '-e', 'input{stdin{}} filter{grok{match=>{"message"=>"%{LOGLEVEL:level}"}}} output{stdout{codec=>json}}'],
        input=input_data, capture_output=True, text=True
    )

    output = json.loads(result.stdout)
    assert output['level'] == 'ERROR'

설명

Logstash를 명령줄로 실행하여 파이프라인 필터가 로그 메시지에서 로그 레벨을 정확히 추출하는지 검증합니다.

---

4. Kibana API 테스트

개요

Kibana의 REST API를 통해 대시보드와 시각화 설정이 올바른지 테스트합니다. 자동화된 설정 검증이 가능합니다.

코드 예제

import requests

def test_kibana_dashboard():
    kibana_url = 'http://localhost:5601'
    headers = {'kbn-xsrf': 'true'}

    response = requests.get(
        f'{kibana_url}/api/saved_objects/dashboard/my-dashboard',
        headers=headers
    )

    assert response.status_code == 200
    assert 'my-dashboard' in response.json()['id']

설명

Kibana API를 호출하여 대시보드가 존재하는지, 설정이 올바른지 확인합니다.

---

5. ELK 통합 테스트 시나리오

개요

전체 ELK Stack을 통합하여 엔드투엔드 시나리오를 테스트합니다. 로그 수집부터 검색까지 전체 플로우를 검증합니다.

코드 예제

import time
from elasticsearch import Elasticsearch

def test_elk_integration():
    es = Elasticsearch('http://localhost:9200')

    # Logstash를 통해 로그 전송 (가정)
    log_entry = {'timestamp': '2025-01-15', 'level': 'ERROR'}
    es.index(index='logs-2025.01.15', document=log_entry)

    time.sleep(2)  # 인덱싱 대기
    result = es.search(index='logs-*', query={'match': {'level': 'ERROR'}})
    assert result['hits']['total']['value'] > 0

설명

Elasticsearch에 직접 로그를 삽입하고, 검색 쿼리로 데이터가 올바르게 저장되고 조회되는지 확인합니다.

---

6. 인덱스 매핑 검증 테스트

개요

Elasticsearch 인덱스의 매핑 설정이 의도한 대로 구성되었는지 검증합니다. 데이터 타입과 분석기 설정을 확인합니다.

코드 예제

from elasticsearch import Elasticsearch

def test_index_mapping():
    es = Elasticsearch('http://localhost:9200')

    mapping = es.indices.get_mapping(index='logs-*')
    properties = mapping['logs-2025.01.15']['mappings']['properties']

    assert properties['timestamp']['type'] == 'date'
    assert properties['level']['type'] == 'keyword'
    assert 'message' in properties

설명

인덱스 매핑 정보를 조회하여 각 필드의 데이터 타입이 올바르게 설정되었는지 확인합니다.

---

7. 성능 테스트 벤치마크

개요

Elasticsearch의 검색 및 인덱싱 성능을 측정합니다. 대용량 데이터 처리 시 응답 시간을 모니터링합니다.

코드 예제

import time
from elasticsearch import Elasticsearch

def test_bulk_indexing_performance():
    es = Elasticsearch('http://localhost:9200')
    docs = [{'_index': 'perf', '_source': {'i': i}} for i in range(10000)]

    start = time.time()
    es.bulk(operations=docs)
    duration = time.time() - start

    assert duration < 5.0  # 5초 이내 완료

설명

대량의 문서를 bulk API로 인덱싱하고, 수행 시간이 임계값 내에 있는지 검증합니다.

---

8. Logstash 에러 핸들링 테스트

개요

Logstash가 잘못된 형식의 로그를 처리할 때 에러를 올바르게 핸들링하는지 테스트합니다.

코드 예제

import subprocess
import json

def test_logstash_error_handling():
    invalid_input = 'this is not json'

    config = '''
    input{stdin{codec=>json}}
    output{stdout{codec=>json}}
    '''

    result = subprocess.run(['logstash', '-e', config],
                          input=invalid_input, capture_output=True, text=True)

    assert '_jsonparsefailure' in result.stdout

설명

JSON 파싱이 실패하는 입력을 전달하여 Logstash가 에러 태그를 추가하는지 확인합니다.

---

9. Kibana 대시보드 자동 배포 테스트

개요

Kibana 대시보드를 코드로 자동 생성하고 배포하는 프로세스를 테스트합니다. Infrastructure as Code 접근법입니다.

코드 예제

import requests
import json

def test_create_kibana_visualization():
    kibana_url = 'http://localhost:5601'
    viz_config = {'title': 'Error Logs', 'type': 'line'}

    response = requests.post(
        f'{kibana_url}/api/saved_objects/visualization',
        headers={'kbn-xsrf': 'true', 'Content-Type': 'application/json'},
        data=json.dumps({'attributes': viz_config})
    )

    assert response.status_code == 200

설명

Kibana API를 통해 시각화 객체를 프로그래밍 방식으로 생성하고, 성공 여부를 검증합니다.

---

10. 알람 규칙 테스트

개요

Elasticsearch의 Watcher 또는 알람 규칙이 올바른 조건에서 트리거되는지 테스트합니다.

코드 예제

from elasticsearch import Elasticsearch

def test_alerting_rule():
    es = Elasticsearch('http://localhost:9200')

    # 에러 로그 생성
    es.index(index='logs', document={'level': 'ERROR', 'count': 100})
    es.indices.refresh(index='logs')

    # 알람 조건 검색
    result = es.search(index='logs', query={'range': {'count': {'gte': 50}}})

    assert result['hits']['total']['value'] > 0  # 알람 트리거 조건 충족

설명

임계값을 초과하는 에러 로그를 생성하고, 알람이 트리거되어야 하는 조건을 만족하는지 검증합니다.

---

11. 인덱스 라이프사이클 관리 테스트

개요

ILM(Index Lifecycle Management) 정책이 오래된 인덱스를 자동으로 삭제하는지 테스트합니다.

코드 예제

from elasticsearch import Elasticsearch
import time

def test_ilm_policy():
    es = Elasticsearch('http://localhost:9200')

    policy = {'phases': {'delete': {'min_age': '1d', 'actions': {'delete': {}}}}}
    es.ilm.put_lifecycle(name='test_policy', policy=policy)

    es.indices.create(index='old-logs', settings={'index.lifecycle.name': 'test_policy'})

    # 정책 적용 확인
    settings = es.indices.get_settings(index='old-logs')
    assert settings['old-logs']['settings']['index']['lifecycle']['name'] == 'test_policy'

설명

ILM 정책을 생성하고 인덱스에 적용한 후, 설정이 올바르게 반영되었는지 확인합니다.

---

12. 멀티 노드 클러스터 테스트

개요

Elasticsearch 클러스터의 샤드 할당과 복제본이 올바르게 분산되는지 테스트합니다.

코드 예제

from elasticsearch import Elasticsearch

def test_cluster_health():
    es = Elasticsearch('http://localhost:9200')

    health = es.cluster.health()

    assert health['status'] in ['green', 'yellow']
    assert health['number_of_nodes'] >= 2
    assert health['active_shards'] > 0
    assert health['relocating_shards'] == 0

설명

클러스터 헬스 API를 호출하여 노드 수, 샤드 상태, 전체 클러스터 상태가 정상인지 확인합니다.

---

마치며

이번 글에서는 ELK Stack 테스트 전략 완벽 가이드에 대해 알아보았습니다. 총 12가지 개념을 다루었으며, 각각의 사용법과 예제를 살펴보았습니다.

관련 태그

#ELK #Elasticsearch #Testing #Integration #Monitoring