[Logstash] 설치, 필터 예시, 동작방식

김태원·2023년 2월 6일
JDBCLogstashgroklogstash examplelogstash fieldlogstash filterlogstash pipelinelogstash pluginlogstash 사용법logstash 설치logstash-output-jdbcmutateruby
1
post-thumbnail

Logstash

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

1. Logstash 동작방식

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

  1. input : logstash가 읽을 데이터를 넣는 부분

    • file : 파일 시스템의 파일을 읽어온다 (해당 파일 권한 확인 666 이상)

  2. filter : 데이터를 가공하는 부분 (데이터 파싱, 문자열 가공 등)

    • grok : 데이터 파싱을 지원하는 plugin

    • mutate : field(=grouping하여 변수에 저장한 것)를 재가공하는 plugin

    • date : datetime을 가공(pasing, formating 지원)하는 plugin

    • ruby : ruby 언어를 통해 직접 filtering을 구현할 수 있는 plugin

  3. output: 데이터 입력 부분

    • file : 파일 시스템의 파일에 쓴다. (해당 파일 권한 확인 666 이상)

    • jdbc : 공식지원하지 않는 plugin, 오픈소스를 통해 다운로드하여 사용가능.

      theangryangel/logstash-output-jdbc

2. Logstash config file setting

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

  1. 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 }
}

  1. 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 중 일부 핵심 내용 소개.

  1. 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를 저장한 후, 다시 읽는 방식으로 변경된다.

  2. 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" 필드의 값이 다르게 설정한다.

  3. 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

  1. /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

  2. yum 을 통해 설치한다.

    %> yum install logstash

5. Install Lostash-Output-JDBC

  1. online download

    %> /usr/share/logstash/bin/logstash-plugin install logstash-output-jdbc
  1. 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"을 통해 설치파일을 실행한다.

profile
김태원
존경받는 개발자를 꿈꾸는 사람. 3년차 웹프로그래머입니다.

0개의 댓글