Logstash, Elasticsearch, Kibana를 사용하여 로그 분석하기

ELK 스택(Elasticsearch, Logstash, Kibana)은 로그 데이터를 수집, 저장, 분석 및 시각화하는 데 강력한 도구입니다.
이 글에서는 ELK 스택을 사용하여 로그 파일을 분석하고, 이를 기반으로 유용한 인사이트를 얻는 방법을 단계별로 설명하겠습니다.

로그 분석은 시스템의 상태를 모니터링하고, 문제를 진단하며,
성능을 최적화하는 데 중요합니다. ELK 스택을 사용하면 실시간으로 로그 데이터를 분석할 수 있어,
다양한 로그 데이터를 효과적으로 관리할 수 있습니다.

ELK 스택 설치 및 설정

Elasticsearch, logstash, kibana 설치 및 설정

Elasticsearch는 로그 데이터를 저장하고 검색할 수 있는 강력한 검색 엔진입니다.
다음 명령어를 사용하여 Elasticsearch를 설치할 수 있습니다.

Ubuntu 예시:

sudo apt update

# elasticsearch 설치 및 실행
sudo apt install elasticsearch
sudo systemctl start elasticsearch
sudo systemctl enable elasticsearch

# logstash 설치
sudo apt install logstash

# kibana 설치 및 실행
sudo apt install kibana
sudo systemctl start kibana
sudo systemctl enable kibana

고급 로그 분석

특정 문자열 빈도수 분석

• 예시 입니다.

특정 문자열(AAA1, AAA2, Module3, Module4)의 빈도수를 분석하기 위해 Logstash.conf 설정을 다음과 같이 수정합니다.

input {
  file {
    path => "../data/*.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}

filter {
  #grok {
    #match => { "message" => "%{COMMONAPACHELOG}" }
  #}
  if [message] =~ /AAA1/ {
    mutate { add_tag => ["AAA1"] }
  }
  if [message] =~ /AAA2/ {
    mutate { add_tag => ["AAA2"] }
  }
  if [message] =~ /Module3/ {
    mutate { add_tag => ["Module3"] }
  }
  if [message] =~ /Module4/ {
    mutate { add_tag => ["Module4"] }
  }
  # 로그 메시지에서 타임스탬프를 추출하여 @timestamp 필드에 할당합니다.
  date {
    match => [ "message", "YYYY-MM-dd HH:mm:ss" ]  # 로그 형식에 맞게 조정
    target => "@timestamp"
  }
  metrics {
    meter => [ "AAA1", "AAA2", "Module3", "Module4" ]
    add_tag => "metric"
    flush_interval => 60
  }
}



output {
  if "metric" in [tags] {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "metrics-%{+YYYY.MM.dd}"
    }
  } else {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "logs-%{+YYYY.MM.dd}"
    }
  }
}

로그 데이터 저장 및 인덱싱

Elasticsearch에 로그 데이터를 저장하기 위해 Logstash 설정 파일을 실행합니다.

/path/to/logstash/bin/logstash -f /path/to/your/config/logstash.conf

로그 데이터 시각화

Kibana에서 인덱스 패턴 생성

Kibana에 접속하여(http://localhost:5601), 인덱스 패턴을 생성합니다.

stack management tap -> Data view에서 설정

logs-*를 인덱스 패턴으로 설정하고, @timestamp 필드를 시간 필드로 지정합니다.

Kibana에서 시각화 대시보드 생성

시각화 도구를 사용하여 로그 데이터를 시각화할 수 있습니다. 예를 들어,
Vertical Bar 차트를 생성하여 시간대별 로그 트래픽을 시각화할 수 있습니다.

Visualize Library tap 에서 시각화 저장 -> dashboard 에서 데이터 확인 가능

실행 순서

Logstash, Elasticsearch, Kibana를 실행하는 순서는 중요합니다.

일반적으로 올바른 순서는 다음과 같습니다:

Elasticsearch 실행: Elasticsearch는 데이터 저장소로서 모든 로그 데이터가 저장될 곳입니다.
따라서 먼저 Elasticsearch를 실행하여 데이터베이스가 준비되도록 합니다.

Logstash 실행: Logstash는 로그 데이터를 수집하고 변환하여 Elasticsearch에 저장합니다.
Elasticsearch가 준비된 후 Logstash를 실행합니다.

Kibana 실행: Kibana는 Elasticsearch에서 데이터를 검색하고 시각화합니다.
따라서 Elasticsearch가 실행되고 데이터가 수집된 후에 Kibana를 실행합니다.

이 순서를 따르는 이유는 각 구성 요소가 의존성을 가지고 있기 때문입니다.
Elasticsearch가 먼저 실행되어 있어야 Logstash가 데이터를 저장할 수 있고,
Kibana가 Elasticsearch에서 데이터를 검색할 수 있습니다.

시행 착오

1. 권한 문제 해결

엘라스틱 서치를 처음 동작시키면

Password for the elastic user (reset with `bin/elasticsearch-reset-password -u elastic`):
  qh=qUe20Ar-Iel7arWQL

ℹ️  HTTP CA certificate SHA-256 fingerprint:
  80caed3c763209a2c67879a31999c36724b817a4747b255797696acd65e5f805

ℹ️  Configure Kibana to use this cluster:
• Run Kibana and click the configuration link in the terminal when Kibana starts.
• Copy the following enrollment token and paste it into Kibana in your browser (valid for the next 30 minutes):
  $

ℹ️  Configure other nodes to join this cluster:
• On this node:
  ⁃ Create an enrollment token with `bin/elasticsearch-create-enrollment-token -s node`.
  ⁃ Uncomment the transport.host setting at the end of config/elasticsearch.yml.
  ⁃ Restart Elasticsearch.
• On other nodes:
  ⁃ Start Elasticsearch with `bin/elasticsearch --enrollment-token <token>`, using the enrollment token that you generated.
‹‹

Go to \u001b[96m\u001b[4mhttp://localhost:5601/?code=029830\u001b[24m\u001b[39m to get started.

이러한 로그가 나오는데 권한 문제를 해결하기에 꼭 필요하므로 메모장같은데 저장해 둘것

 

2. ssl 문제 및 localhost 해결

엘라스틱 서치, logstash, kibana가 ssl을 쓴다고 가정할때 전부 https로 통일해줘야한다.
저는 localhost로 진행했기 때문에 http로 통일하였습니다.
해당 정보는 각각의 .yml 파일에서 변경 가능

 

3. 특정 시점의 데이터를 파악하기 위한 필터 적용 방법

filter {
  # Grok 필터를 사용하여 로그 메시지를 파싱합니다.
  grok {
    match => { "message" => "%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day} %{HOUR:hour}:%{MINUTE:minute}:%{SECOND:second} %{LOGLEVEL:loglevel} \[%{DATA:thread}\] %{GREEDYDATA:msg}" }
    add_field => { "customtime" => "%{year}-%{month}-%{day} %{hour}:%{minute}:%{second}" }
  }

  # 커스텀 시간 필드를 @timestamp 필드에 매핑합니다.
  date {
    match => [ "customtime", "yy-MM-dd HH:mm:ss" ]
    target => "@timestamp"
    timezone => "Asia/Seoul"
  }
  # 로그 메시지에서 특정 문자열을 추출하고, 해당 문자열이 있으면 태그를 추가합니다.
  if [message] =~ /Sample1/ {
    mutate {
      add_tag => ["Sample1"]
    }
  }
  if [message] =~ /Sample2/ {
    mutate {
      add_tag => ["Sample2"]
    }
  }

  # metrics 필터를 사용하여 태그별 빈도수를 계산합니다.
  metrics {
    meter => [ "Sample1", "Sample2" ]
    add_tag => "metric"
    flush_interval => 60
  }
}

결론

이 글에서는 ELK 스택을 사용하여 로그 데이터를 수집, 저장, 분석 및 시각화하는 방법을 살펴보았습니다.
Logstash, Elasticsearch, Kibana를 통해 로그 데이터를 효과적으로 관리하고,
실시간으로 분석하여 유용한 인사이트를 얻을 수 있습니다.

추가적으로, 보안 및 성능 최적화를 위해 SSL 설정, 로드 밸런싱, 캐싱 등을 고려할 수 있습니다.
더 자세한 내용은 ELK 스택의 공식 문서와 기타 리소스를 참고하세요.

참고 자료

• Elasticsearch 공식 문서
• Logstash 공식 문서
• Kibana 공식 문서