Logstash

: logstash는 실시간 파이프라인 기능을 지원하는 데이터 엔진 오픈소스이다.

1. Logstash 동작방식

: Logstash의 파이프라인은 input -> filters -> output 세단계로 데이터를 처리한다

input : logstash가 읽을 데이터를 넣는 부분
- file : 파일 시스템의 파일을 읽어온다 (해당 파일 권한 확인 666 이상)
filter : 데이터를 가공하는 부분 (데이터 파싱, 문자열 가공 등)
- grok : 데이터 파싱을 지원하는 plugin
- mutate : field(=grouping하여 변수에 저장한 것)를 재가공하는 plugin
- date : datetime을 가공(pasing, formating 지원)하는 plugin
- ruby : ruby 언어를 통해 직접 filtering을 구현할 수 있는 plugin
output: 데이터 입력 부분
- file : 파일 시스템의 파일에 쓴다. (해당 파일 권한 확인 666 이상)
- jdbc : 공식지원하지 않는 plugin, 오픈소스를 통해 다운로드하여 사용가능.
  
  theangryangel/logstash-output-jdbc

2. Logstash config file setting

: logstash는 Pipeline config와 Setting files 2가지로 나뉜다.

Pipleline config

: /etc/logstash/conf.d에 .conf 확장자 파일을 생성해 넣어주어야 한다.

: 아래 예시와 같이, input, filter, output에 원하는 plugin을 설정해 준다.

input { stdin { } }

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch { hosts => ["localhost:9200"] }
  stdout { codec => rubydebug }
}

Setting Files

: 설치시 /etc/logstash에 아래의 파일들이 default값으로 생성되어 있다.
- logstash.yml
  
  : pipeline, config, queue, log 등 logstash의 모든 config를 설정할 수 있다.
- pipelines.yml
  
  : pipline과 관련된 설정을 한다. logstash로도 설정할 수 있지만 piplines.yml의 설정을 우선적으로 읽어온다.
- jvm.options
  
  : logstash는 JAVA로 만들어진 프로그램으로, JVM 설정을 할 수 있다.
- log4j2.properties
  
  : JAVA의 로그 관련된 설정이다.

3. Pipeline Config Plugin Detail

: pipeline config에서 사용할수 있는 plugin 중 일부 핵심 내용 소개.

Input
- file
  
  : 개행문자가 올때까지를 하나의 로그로 판단하며, 하나의 로그당 하나의 이벤트가 동작한다. (별도 설정을 통해 변경은 가능)
  
  : file plugin은 설정한 path에 있는 파일들에 상태값을 부여하고, 상태값에 따라 다르게 처리한다.
  
  : file plugin은 "closed" 나 "ignored" 상태인 파일의 수정날짜가 갱신되었다면, 해당 파일듸 상태를 "watched"로 변경한다.
  
  : "watched" 상태인 파일 중, 갱신된 파일 양이 서버의 여유공간보다 적은 파일들은 "active" 상태로 변경한다.
  
  : "active" 상태인 파일을 open하고 read하고 Pipeline Config의 Filter, Output을 모두 처리한 후, EOF까지 처리하면(chunk 설정을 하면 변경 가능), sincedb(어디까지 읽었는지를 저장한 offset)를 저장하고 완료된다.
```
 input {
     file {
         path => "/var/log/dovecot"
         start_position => "end"
         sincedb_write_interval => 10
         sincedb_clean_after => "7 day"
         discover_interval => 15
         stat_interval  => 1
         mode => "tail"
         file_chunk_count => 5
         file_chunk_size => 32768
     }
 }
```
  - path
    
    : file plugin이 감시할 경로
    
    : 해당 경로의 파일은 권한 666이상이어야한다.
  - start_position : 읽기 시작할 위치, sincedb가 없는 경우에만 적용된다.
  - sincedb_write_interval : sincedb를 저장할 주기
  - sincedb_clean_after : sincedb를 삭제할 기간
  - discover_interval : path에 새로운 파일을 감시하는 주기
  - stat_intercval : path의 수정날짜 갱신을 감시하는 주기
  - mode : Tail mode와 Read mode 2가지가 존재한다.
    
    (1) Tail mode
    
    : path의 파일이 수정날짜가 끝임없이 갱신되는 경우 사용하는 mode.
    
    : file이 rotate(읽던 파일을 다른이름으로 변경하고, 기존이름으로 새로운 파일을 생성하는 것)되는 것을 자동으로 감지한다.
    
    (2) Read mode
    
    : path의 파일이 수정날짜가 갱신되지 않는 경우 사용하는 mode.
  - file_chunk_count
    
    : default value = 4611686018427387903
  - file_chunk_size
    
    : 기본설정 상, file plugin은 한번 활성화 상태가 되면, EOF까지 읽도록 되어 있다. (그래서 default값이 매우큼)
    
    : 하지만, 해당 값을 수정하면, 해당 file_chunk_count * file_chunk_size 만큼만 읽고 sincedb를 저장한 후, 다시 읽는 방식으로 변경된다.
Filter
- grok
  
  : grok plugin은 데이터 파싱, 재가공을 해준다.
  
  : 아래 사이트를 통해 grok pattern을 테스트 할 수 있다. 매우 유용하다
  
  Test grok patterns
```
grok {
    match => { "message" => "(?<time>\w{3} \d{2} \d{2}:\d{2}:\d{2}) (?:imap|pop3)-login: (?<log_name>Info: Login): user=\<(?<login_email>.+\@(?<login_domain>.+))\>.+, rip=(?<ip>(?<flag>[\d]{1,3}).[\d]{1,3}.[\d]{1,3}.[\d]{1,3}), lip=.+,.+session=\<(?<session>.+)\>"}
}
```
  : 정규표현식을 사용 할 수 있고, naming grouping한 값은 feild(변수 개념으로 보면된다)에 저장된다.
  
  : 저장된 feild는 "필드명"으로 호출 시, 저장된 값이 불러와진다.
- date
  
  : filed에 저장된 date를 파싱하고, date format을 변경할 수 있다.
```
date {
    match => ["time", "MMM dd HH:mm:ss"]
    target => "log_time"
}
```
  - match : "time" field의 값을 "MMM dd HH:mm:ss" format으로 읽어온다.
  - target
    
    : match에서 읽어온 값을 "log_time" field에 저장한다.
    
    : 선언하지 않을 경우, 기본값은 @timestamp에 저장된다.
    
    @timestamp
    
    : reserved field로서, 따로 생성하여 값을 저장하지 않더라도, logstash가 항상 생성하고 값을 저장하는 필드이다.
    
    : @timestamp에는 현재시간 값이 저장되어 있다.
- ruby
  
  : ruby언어 코드를 실행한다.
  
  : 실행 방법은 inline 방법과 file 방법으로 나뉜다.
  
  : inline - config 파일에 코드를 직접 입력하여, 해당 코드를 실행 하는 방식
  
  : file - ruby code 파일의 path를 config에 입력하여, 해당 파일을 실행하는 방식
```
ruby {
    code => "
      begin
        # random integer
        event.set('rand_int', rand(1000000))

      rescue Exception => e
        event['ruby_exception'] = 'EEK: ' + e.message
      end
    "
}
```
  : rand() 함수를 통해 무작위 자연수를 생성하고, "rand_int" field에 저장한다.
- mutate
  
  : field의 이름을 변경, 특정부분 변환, 수정을 할 수 있다.
```
if [ip] == "10.10.10.250" or [ip] == "172.29.12.214"{
    mutate{
         add_field => {
            "type" => "webmail"
        }
    }
}
else {
    mutate{
         add_field => {
            "type" => "outlook"
        }
    }
}
```
  : "ip" 필드의 값에 따라, "type" 필드의 값이 다르게 설정한다.
Output
- file
  
  : 디스크의 특정 파일에 데이터를 쓴다.
  
  : file filter 호출전에, if문이 없다면 각 log line마다 모두 실행되니 주의한다..
  - path
    
    : 파일의 경로이다.
    
    : 해당 경로의 파일은 권한 666이상이어야한다.
  - codec : 출력할 데이터를 명시한다.
```
if [daemon] == " postfix/pipe" {
    file {
        path => "/var/log/maillogS1"
        codec => line { format => "%{message}"}
    }
}
```
- jdbc
  
  : Logstash에서 공식지원하는 plugin이 아니다. Logstash는 input plugin으로만 jdbc를 지원하지만, 아래 오픈소스를 통해, 별도 설치하여 사용가능하다.
  
  theangryangel/logstash-output-jdbc
  
  : 설치 방법은 밑에 별도 기재.
```
 output {
     jdbc {
         driver_class => "com.mysql.jdbc.Driver"
         connection_string => "jdbc:mysql://10.10.10.152:5101/ECOUNT_MAIL?user=ecountdev&password=acct@0000"
         statement => ["INSERT INTO test_tw (id) VALUES(?)", 2]
     }
 }
```
  - driver_class : jdbc driver 이름을 명시한다.
  - connection_string : db connection 정보를 명시한다.
  - statement : 실행할 쿼리를 명시한다.

4. Install Logstash

/etc/yum.repos.d 에 .repo 파일을 셋팅한다.

[logstash-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

yum 을 통해 설치한다.
```
%> yum install logstash
```

5. Install Lostash-Output-JDBC

online download

%> /usr/share/logstash/bin/logstash-plugin install logstash-output-jdbc

offline download
- 이미 설치되어 있는 서버에서 오프라인 설치파일 만들기
```
%> /usr/share/logstash/bin/logstash-plugin prepare-offline-pack --output ./logstash-offline-plugins-8.6.0.zip --overwrite logstash-output-jdbc
```
  : "logstash-plugin"을 통해 설치 파일을 생성한다.
- 인터넷이 안되는 미설치 서버에 설치하기
```
%> /usr/share/logstash/bin/logstash-plugin install file:///logstash-offline-plugins-8.6.0.zip
```
  : "logstash-plugin"을 통해 설치파일을 실행한다.

신입개발자 뮤말라의 노트

[Logstash] 설치, 동작방식, 필터예시(grok, mutate, ruby, jdbc, file)