Elastic Agent로 수집된 로그 데이터를 Elasticsearch에서 유형별로 집계하고, 최근 30일 간의 일별 카운트를 집계하는 과정을 단계별로 설명합니다. 아래에 제시된 명령어와 옵션은 Kibana의 Dev Tools에서 사용할 수 있습니다.
1단계: 데이터 구조 확인
Elasticsearch에 저장된 로그 데이터의 필드 구조를 확인해야 합니다.
- Elastic Agent가 수집한 로그에는 일반적으로
log.type또는event.dataset과 같은 필드가 사용됩니다. @timestamp필드는 로그의 시간 정보를 나타냅니다.
샘플 쿼리
GET /your-index-pattern-*/_mapping위 명령어를 실행하면 인덱스의 매핑 정보가 반환됩니다. 여기에서 사용할 필드 이름(log.type 또는 event.dataset)을 확인합니다.
2단계: 쿼리 작성
최근 30일 간의 데이터를 가져와 로그 유형별로 그룹화하고, 일별 카운트를 집계하기 위해 아래와 같은 쿼리를 작성합니다.
쿼리 예제
GET /your-index-pattern-*/_search
{
"size": 0,
"query": {
"bool": {
"filter": [
{
"range": {
"@timestamp": {
"gte": "now-30d/d",
"lte": "now/d"
}
}
}
]
}
},
"aggs": {
"log_types": {
"terms": {
"field": "log.type.keyword", // 로그 유형별로 그룹화
"size": 10 // 상위 10개 유형만 표시
},
"aggs": {
"daily_counts": {
"date_histogram": {
"field": "@timestamp", // 일별로 그룹화
"calendar_interval": "day",
"format": "yyyy-MM-dd"
}
}
}
}
}
}size: 0: 검색 결과를 출력하지 않고 집계 데이터만 반환.range: 최근 30일 동안의 데이터만 필터링.terms:log.type필드를 기준으로 로그 유형별로 그룹화.date_histogram:@timestamp필드를 기준으로 일별 카운트 계산.
3단계: Kibana Dev Tools에서 실행
- Kibana에 로그인합니다.
- 왼쪽 사이드바에서 Dev Tools 메뉴를 선택합니다.
- 작성한 쿼리를 붙여넣고 실행 버튼을 클릭합니다.
4단계: 결과 확인 및 해석
쿼리가 실행되면 JSON 형태로 결과가 반환됩니다. 주요 구조는 아래와 같습니다.
{
"aggregations": {
"log_types": {
"buckets": [
{
"key": "system", // 로그 유형
"doc_count": 5000, // 총 로그 수
"daily_counts": {
"buckets": [
{
"key_as_string": "2025-01-01",
"doc_count": 200 // 해당 날짜의 로그 수
},
{
"key_as_string": "2025-01-02",
"doc_count": 180
}
// ...
]
}
}
// 추가 로그 유형
]
}
}
}log_types.buckets.key: 로그 유형.log_types.buckets.daily_counts.buckets: 해당 로그 유형의 일별 카운트.
5단계: 결과 데이터 활용
1. Kibana 대시보드 구성
위 결과를 기반으로 비주얼라이제이션을 생성합니다.
- Kibana에서 Visualize 메뉴로 이동합니다.
- 새 차트를 생성하고,
log.type필드를 기준으로 데이터를 분류합니다. - X축에는
@timestamp를 설정하여 일별 데이터를 표시합니다.
2. 경고(Alert) 설정
특정 로그 유형의 카운트가 기준치를 초과하거나 예상치 못한 패턴이 발견되었을 때 경고를 생성합니다.
- Kibana에서 Alerts and Actions 메뉴로 이동합니다.
- 새 경고를 생성하고, 위 쿼리를 기준으로 조건을 설정합니다.
6단계: 자동화된 스크립트 예제
Elasticsearch API를 호출하여 데이터를 자동으로 가져오는 Python 스크립트를 작성할 수도 있습니다.
Python 예제
import requests
import json
# Elasticsearch 정보
es_url = "http://your-elasticsearch-host:9200"
index_pattern = "your-index-pattern-*"
query = {
"size": 0,
"query": {
"bool": {
"filter": [
{
"range": {
"@timestamp": {
"gte": "now-30d/d",
"lte": "now/d"
}
}
}
]
}
},
"aggs": {
"log_types": {
"terms": {
"field": "log.type.keyword",
"size": 10
},
"aggs": {
"daily_counts": {
"date_histogram": {
"field": "@timestamp",
"calendar_interval": "day",
"format": "yyyy-MM-dd"
}
}
}
}
}
}
# Elasticsearch 쿼리 실행
response = requests.get(f"{es_url}/{index_pattern}/_search", headers={"Content-Type": "application/json"}, data=json.dumps(query))
# 결과 출력
if response.status_code == 200:
result = response.json()
print(json.dumps(result, indent=2))
else:
print(f"Error: {response.status_code} - {response.text}")고려 사항
- 인덱스 관리: 오래된 인덱스를 자동으로 삭제하려면 ILM(인덱스 라이프사이클 관리)를 설정하세요.
- 시각화 도구: 데이터를 Long-term 모니터링할 때 Grafana 또는 Kibana를 적극 활용합니다.
- 알림 시스템: Slack, 이메일, 또는 PagerDuty와 같은 외부 알림 시스템에 통합합니다.
위 과정을 통해 Elastic Agent로 수집된 로그 데이터를 효과적으로 관리할 수 있습니다.
728x90
댓글