ELK 스택과 FileBeat 사용

#  ElasticSearch + LogStash + Kibana + filebeat 을 이용한 데이터분석

---

• 동작 방식

# ELK stack을 한 서버에 구축하며, 각 클라이언트 별 filebeat를 이용하여 로그를 수집한다.

• filebeat로 부터 로그를 수집한 후 logstash로 보내진다.
• logstash에서 데이터를 정제한 후 elasticsearch로 전달한다.
• 최종적으로 elasticsearch에 데이터를 kibana로 visualizing 한다.

 

---

LogStash

• 로그파일을 필요 용도에 맞게 정제를 한다.
• 항상 logstash 를 실행 시   tail -f /var/log/logstash/logstash-plain.log  켜서 에러가없이 포트가 정상적으로 열리는지 확인한다. 문제가 없을 시 port가 열리고나면 로그가 적재되지않는다.
• 제일 많이 오류가 발생하는 부분이 오타일 경우가 많다.
• 현재 여러 conf 파일들을 사용할 때, 하나의 conf파일이 먼저 실행되면 다음 conf 파일들은 인식을 못하고 기본 source 정보만 저장된다. 명확한 이유는 찾지못했고, logstash가 a b c 순으로 파일을 처리하는 부분에 있어 여러 conf 파일을 드나들 때 문제가 있다는 내용이 있다. pipeline id를 추가하는 방법이 해결방안으로 나와있는데, 현재 안됨.
• 그래서 하나의 conf파일에 IF문등 조건식을 사용하여 파일을 분배해 filtering 하고 있다.
• 또 conf 파일별로 포트를 같게 주면 오류가 발생하여 만약 여러개의 conf 파일을 사용하려면 각각 포트번호를 다르게 주어야한다.

elastico 에 rpm 다운 기준 /etc/logstash/conf.d/ 경로를 사용.

*.conf    양식으로 파일을 생성.  

--기본구조

 

input {

  beats {

     port => 5044              # filebeat과 통신을 위해 사용할 포트

  }

}

 

filter {

 

}

 

output {

   elasticsearch {

     hosts => ["http://localhost:9200"]                  # elasticsearch 위치

     index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"           # index명 임의로 지정가능

   }

}

 

filter

- grok : 패턴 매칭을 사용해서 추출

- date : 필드로 부터 timestamp 파싱함

- mutate : 이벤트에서 필드를 rename, remove, replace, modify

- csv : CSV 데이터 형태로 파싱함

- geoip : IP 주소로 지리를 판단

- kv : 이벤트 데이터에서 key-value 쌍을 파싱함

- ruby : 파이프라인에서의 모호한 Ruby code 실행될때 등등

 

• grok을 이용해 로그내용을 key , value 형태로 저장할 수 있다.
• 정규표현식을 이용하기 때문에 정규표현식을 이해하면 좋다.
• 보통 message 필드를 match하여 해당 내용을 조각낸다.
• 이때 정규표현식 또는 자체적으로 제공하는 pattern을 이용하여 데이터를 추출해내어 key,value 형태로 저장할 수 있다.
• https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns  자체적으로 제공하는 pattern 식
• 제공된 패턴 사용시 %{패턴명} , 자체적으로 패턴을 만들어 사용할때 (?<패턴명>추출할 양식=정규표현식)
• if문 사용이 가능하다. 여러가지 방식으로 이용가능.
• https://grokdebug.herokuapp.com/ 해당 표현식이 정삭적으로 작동하는지 디버깅해주는 사이트\

 

• & 자주쓰는 정규표현식

\s  :  스페이스 (= %{SPACE})

\w : 특수문자를 제외한 문자와 숫자를 추출

\d : 숫자를 추출

특수문자를 문자처럼 사용시 앞에 \를 붙여주고 사용할 것

...등등

필요없는 필드는 remove_field 를 이용하여 지울수있다.

• grok 부분에 사용할 수도 있고, mutate를 이용할 수 도 있다.

-- mutate { remove_field => [ "a", "message","beat","host","input","meta","log","@version","offset","prospector" ] }              # 필요없는 필드는 지운다.

 

ex) 이 로그를 필터로 정제.

2019/01/10 07:01:07 [error] 23986#23986: *29 open() "/var/www/html/404.html" failed (2: No such file or directory), client: 1.1.1.1, server: ji.com, request: "GET /test.php HTTP/1.1", host: "2.2.2.2"

 

• filter 부분

filter {
  grok {
    match => { "message" => ["(?<log_time>%{DATE}\s%{TIME})\s\[%{DATA:title}\]\s%{DATA:identified}\s(?<status>%{DATA}\s%{DATA})\s\"%{DATA:path}\"\s%{DATA:event},\sclient\:\s%{IP:cilent},\sserver\:\s%{DATA:server},\srequest\:\s\"%{DATA:request}\",\shost\:\s\"%{IP:host}\"%{GREEDYDATA:hint}"] }
  }
}

---

* 결과 :

"server": "ji.com",
"request": "GET /test.php HTTP/1.1",
"title": "error",
"log_time": "19/01/10 07:01:07",
"cilent": "1.1.1.1",
"path": "/var/www/html/404.html",
"identified": "23986#23986:",
"hint": "",
"host": "2.2.2.2",
"event": "failed (2: No such file or directory)",
"status": "*29 open()"

---

& geoip 

• IP를 기반으로 지역정보를 추출한다.
• filter 에 작성한다.

grok 부분에서 IP를 패턴으로 추출하였다면 이 키값을 geoip 에 매칭시키면 된다.

ex)    client 값에 IP 정보가 있다.

geoip {

   source => "client"

   target => "geoip"

}

이렇게하면 client값에 대한 지역정보가 담긴 값들이 key:value 형태로 저장되어진다.

 

• geoip.city_name

• geoip.continent_code

• geoip.country_code2

• geoip.country_code3

• geoip.country_name

• geoip.dma_code

• geoip.ip

• geoip.latitude

• geoip.location.lat

• geoip.location.lon

• geoip.longitude

• geoip.postal_code

• geoip.region_code

• geoip.region_name

• geoip.timezone

해당 데이터들이 저장됨.

 

---

정제된 데이터는 elasticsearch로 전송된다.

• curl -XGET 'http://localhost:9200/_cat/indices?v&pretty'              # 현재 elasticsearch에 저장된 인덱스를 보여줌.

• curl -X DELETE 'http://localhost:9200/[data명 or _all]'                # 특정 인덱스만 지우거나 모든 데이터를 지운다.

• elasticsearch head 를 사용해 웹형식으로 관리할 수도 있다.              # 9100포트를 이용한다.
• https://github.com/mobz/elasticsearch-head                                # github 에서 제공하는 source를 이용
• 기본적으로 elasticsearch 와 연동된 kibana를 구동할때 데이터가 elasticsearch에 저장되는데, 혹시 모든데이터를 지웠을 경우 kibana를 재시작해준다. 

---

Kibana 를 이용해 데이터를 시각화한다.

• elasticsearch에 쌓여진 인덱스들은 kibana에서 확인이 가능하다.
• 인덱스를 분석하고 visualizing 하기 위해서는 index pattern을 생성해준다.

• 생성된 index pattern은 Discover 탭에서 확인가능하다.
• 앞서 logstash에서 넘겨준 데이터가 잘 정제되어 들어왔는지 확인한다.       
• # 참고 : tags 필드에  failure 단어가 포함된 메시지값이 있다면 필터쪽에서 해당 작업이 실패했다는 것.

• 특정 데이터를 보기 좋게 시각화작업을 할 수 있다.
• 먼저 Visualize 탭에서 만들고 Dashboard에 보기좋게 꾸며준다.

 

정규표현식이 많이 필요하군요..

---

THE END.