Logstash 및 Beats 활용 (ELK Stack 구성)
📌 ELK Stack은 Elasticsearch, Logstash, Kibana를 포함하는 로그 및 데이터 분석 솔루션입니다.
- Logstash: 다양한 데이터 소스로부터 데이터를 수집 및 변환하여 Elasticsearch에 전달
- Beats: 경량 데이터 수집기(Filebeat, Metricbeat, Heartbeat)
- Kafka 연동: 대용량 데이터 스트리밍을 위한 Apache Kafka와 Elasticsearch 연결
Logstash를 이용한 데이터 수집 및 변환
Logstash는 다양한 소스로부터 데이터를 수집하여 필터링, 변환 후 Elasticsearch로 전송하는 도구입니다.
✅ (1) Logstash 설치
wget https://artifacts.elastic.co/downloads/logstash/logstash-8.x.x-linux-x86_64.tar.gz
tar -xzf logstash-8.x.x-linux-x86_64.tar.gz
cd logstash-8.x.x-linux-x86_64
✅ Logstash 다운로드 및 압축 해제 후 실행 준비.
✅ (2) Logstash 기본 구성 (logstash.conf)
Logstash의 설정 파일은 Input → Filter → Output 구조를 가집니다.
input {
file {
path => "/var/log/syslog"
start_position => "beginning"
}
}
filter {
grok {
match => { "message" => "%{SYSLOGTIMESTAMP:timestamp} %{HOSTNAME:host} %{GREEDYDATA:log}" }
}
date {
match => [ "timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
target => "@timestamp"
}
}
output {
elasticsearch {
hosts => ["http://localhost:9200"]
index => "syslog-%{+YYYY.MM.dd}"
}
}
✅ 파일 로그(/var/log/syslog)를 읽고 변환하여 Elasticsearch에 저장.
✅ grok을 사용하여 timestamp, host, log 필드로 분리.
✅ 데이터를 날짜별 인덱스로 저장 (syslog-YYYY.MM.dd).
✅ (3) Logstash 실행
bin/logstash -f logstash.conf
✅ Logstash 실행 후 데이터를 Elasticsearch로 전송.
✅ (4) Elasticsearch에서 데이터 확인
GET syslog-*/_search
{
"query": {
"match_all": {}
}
}
✅ Elasticsearch에서 Logstash를 통해 수집된 로그 데이터 조회.
Beats 활용 (Filebeat, Metricbeat, Heartbeat)
Beats는 경량 데이터 수집기로, 개별 서버에서 실행되며 Logstash 또는 Elasticsearch로 데이터를 전송할 수 있습니다.
✅ (1) Filebeat - 로그 파일 수집
Filebeat는 파일 로그를 모니터링하고 전송하는 경량 데이터 수집기입니다.
📌 Filebeat 설정 (filebeat.yml)
filebeat.inputs:
- type: log
paths:
- /var/log/syslog
output.elasticsearch:
hosts: ["http://localhost:9200"]
index: "filebeat-%{+YYYY.MM.dd}"
✅ 파일 로그(/var/log/syslog)를 모니터링하고 Elasticsearch에 저장.
✅ (2) Metricbeat - 시스템 메트릭 수집
Metricbeat는 CPU, 메모리, 네트워크 등의 시스템 메트릭을 수집합니다.
📌 Metricbeat 설정 (metricbeat.yml)
metricbeat.modules:
- module: system
metricsets:
- cpu
- memory
- network
enabled: true
period: 10s
output.elasticsearch:
hosts: ["http://localhost:9200"]
index: "metricbeat-%{+YYYY.MM.dd}"
✅ CPU, 메모리, 네트워크 메트릭을 10초마다 수집하여 Elasticsearch에 저장.
✅ (3) Heartbeat - 서비스 상태 모니터링
Heartbeat는 웹 서비스 및 TCP, ICMP 등의 연결 상태를 모니터링합니다.
📌 Heartbeat 설정 (heartbeat.yml)
heartbeat.monitors:
- type: http
schedule: '@every 10s'
urls:
- "http://localhost:5601"
- "http://localhost:9200"
output.elasticsearch:
hosts: ["http://localhost:9200"]
index: "heartbeat-%{+YYYY.MM.dd}"
✅ Elasticsearch와 Kibana 상태를 10초마다 체크하여 Elasticsearch에 저장.
Kafka와 Elasticsearch 연동
대량의 로그 및 이벤트 데이터를 실시간으로 처리할 때 Kafka를 중간 버퍼로 사용하여 Elasticsearch로 전송할 수 있습니다.
✅ (1) Kafka 설치 및 실행
wget https://downloads.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz
tar -xzf kafka_2.13-3.0.0.tgz
cd kafka_2.13-3.0.0
# Zookeeper 실행
bin/zookeeper-server-start.sh config/zookeeper.properties
# Kafka 실행
bin/kafka-server-start.sh config/server.properties
✅ Zookeeper와 Kafka를 실행하여 메시지 큐 준비.
✅ (2) Kafka에 로그 데이터 전송
bin/kafka-topics.sh --create --topic logs --bootstrap-server localhost:9092
bin/kafka-console-producer.sh --topic logs --bootstrap-server localhost:9092
✅ Kafka 토픽(logs)을 생성하고 로그 메시지를 입력하여 테스트.
✅ (3) Logstash를 사용하여 Kafka → Elasticsearch 데이터 전송
input {
kafka {
bootstrap_servers => "localhost:9092"
topics => ["logs"]
group_id => "logstash-group"
}
}
filter {
json {
source => "message"
}
}
output {
elasticsearch {
hosts => ["http://localhost:9200"]
index => "kafka-logs-%{+YYYY.MM.dd}"
}
}
✅ Kafka에서 데이터를 읽어 Elasticsearch로 저장.
✅ (4) 데이터 확인
GET kafka-logs-*/_search
{
"query": {
"match_all": {}
}
}
✅ Kafka를 통해 전달된 로그 데이터가 Elasticsearch에 정상적으로 저장되었는지 확인.
Logstash 및 Beats 활용 정리
| 기능 |
설명 |
| Logstash |
다양한 데이터 소스를 수집하여 Elasticsearch로 전달 |
| Filebeat |
로그 파일을 모니터링하고 전송 |
| Metricbeat |
시스템 메트릭(CPU, 메모리 등) 수집 |
| Heartbeat |
서비스 및 네트워크 상태 모니터링 |
| Kafka 연동 |
대용량 로그를 Kafka → Elasticsearch로 스트리밍 |
