ELK) Logstash 기본 개념

1. Logstash 개념 

ELK의 L을 의미하는 Logstash는 무료 오픈 소스 서버 데이터 처리 실시간 파이프라인

다양한 소스에서 데이터를 수집하여 변환한 후 자주 사용하는 저장소 (ELK 스택에서는 elasticsearch)로 전달한다. 데이터의 형식이나 복잡성과 관계 없이 데이터를 동적으로 수집, 전환, 전송한다. 즉, 서로 다른 소스의 데이터일지라도 동적으로 통합하여 원하는 대상으로 데이터를 정규화할 수 있다. 

2. Logstash 작업

• 입력 : 다양한 형태, 크기, 소스의 데이터를 수집한다. ㅛ소 
• 필터 : 소스에서 데이터로 변환하여 저장소로 이동하는 과정에서 각 이벤트를 구문 분석하고 명명된 필드(column)을 식별하여 구조를 구축하며, 이를 공통 형식으로 변환 통합한다. grok를 통해서 비구조화된 데이터에서 구조를 도출하고 IP 주소에서 위치 좌표를 해독하고, 데이터의 익명화와 민간한 정보 필드 완전 지외 등의 기능을 제공한다.
• 출력 : 원하는 곳으로 데이터를 라우팅할 수 있어서 여러 저장소(ex. elasticsearch)로 데이터를 다운스트림하는 유연성을 확보한다. 

3. Logstash 특징

• 확장성 : 200개 이상의 플러그인 프레임워크가 포함되어 있어 위 작업들을 원하는대로 생성 및 구성할 수 있다. 
• 안정성 및 보안성 : persistent queue를 사용하여 현재 처리 중인 이벤트에 최소 1회의 전송을 보장하며, 실패한 이벤트는 검증 및 재전송을 위해 dead letter queue로 분류되다. 수집량을 조절할 수 있어서 수집중인 데이터량이 급증하는 경우에도 정상적인 동작이 가능하며, 수집 파이프라인에 대한 완전한 보안이 설정되어 있다. 
• 모니터링 : 파이프라인 탐색기가 제공된다. 

4. Logstash 파이프라인

• 파이프라인 구성 요소 : inputs, outputs, filters(optional) 

출처 : https://soyoung-new-challenge.tistory.com/99

• logstash.yml 초기 설정

  path.data : /var/lib/logstash
  path.logs : /var/log/logstash​

• pipeline.yml 초기 설정

  - pipeline.id: main
    path.config: "/etc/logstash/conf.d/*.conf"​

   conf.d/custom.conf - conf파일명은 원하는대로 해서 직접 지정해주거나 *.conf로 작성 
  • input example

    # input
    input {
      beats {
        port => 5044
        host => "0.0.0.0"
        client_inactivity_timeout => 86400
      }
    }​

    파일 비트로부터 데이터 받을 포트 지정 (기본이 5044), 호스트 상관없이 모든 데이터 받을 경우 0.0.0.0으로 작성 
  • fileters exmple
    엘라스틱에 데이터 전달하기 전에 원하는 데이터 타입, 형태 등으로 데이터 전처리하는 과정
    

    filter {
      if "IMP" in [log][file][path] {
        mutate {
          gsub => ["message", ", ", "| "]
        }
        grok {
          match => { "message" => ["%{NUMBER:[imp][date]},%{NUMBER:[imp][h]},%{NUMBER:[imp][cu_id]},%{NUMBER:[imp][au_id]},%{NUMBER:[imp][pu_id]},%{WORD:[imp][c_key]},%{WORD:[imp][p_key]},%{GREEDYDATA:[imp][no_info]},%{NUMBER:[imp][place_id]},%{WORD:[imp][nation]},%{WORD:[imp][device]},%{NUMBER:[imp][no_info2]},%{NUMBER:[imp][user_key]},%{WORD:[imp][p_set_id]},%{GREEDYDATA:[imp][url]},\"%{TIMESTAMP_ISO8601:[imp][cre_tt]}\",%{GREEDYDATA:[imp][remote_addr]},%{NUMBER:[click][ar_id]}"]}
          remove_field => ["message"]
          }
        grok {
          match => { "message" => ["%{NUMBER:[imp][date]},%{NUMBER:[imp][h]},%{NUMBER:[imp][cu_id]},%{NUMBER:[imp][au_id]},%{NUMBER:[imp][pu_id]},%{WORD:[imp][c_key]},%{WORD:[imp][p_key]},%{GREEDYDATA:[imp][no_info]},%{NUMBER:[imp][place_id]},%{WORD:[imp][nation]},%{WORD:[imp][device]},%{NUMBER:[imp][no_info2]},%{NUMBER:[imp][user_key]},%{WORD:[imp][p_set_id]},%{GREEDYDATA:[imp][url]},\"%{TIMESTAMP_ISO8601:[imp][cre_tt]}\""]
          remove_field => ["message"]
          }
        }
        date {
          match => [ "[imp][cre_tt]", "YYYY-MM-dd H:m:s" ]
          target => "@timestamp"
          timezone => "Asia/Seoul"
        }
        mutate {
          gsub => [ '[imp][url]', '"', '']
          convert => ["[imp][au_id]","integer"]
          convert => ["[imp][cu_id]","integer"]
          convert => ["[imp][date]","integer"]
          convert => ["[imp][h]","integer"]
          convert => ["[imp][place_id]","integer"]
          convert => ["[imp][pu_id]","integer"]
        }
      }
    }​

    [log][file][path] : 로그의 위치 
    mutate { gsub => [필드명, 원래값, 변경할값] } : 최상단에서 grok에 보낼 메시지를 전처리할 작업이 있을 때 사용하는 것으로 위 예시에서는 구분자를 ,에서 |로 바꾸는 작업 
    grok { match => {"message" => [grok 정규표현식] }, removed_field => [지우고 싶은 필드명]  } : message안에 담긴 데이터를 grok 패턴으로 파싱하며, 여러 grok 패턴을 사용할 수 있다. 
    data { match => {변경할 필드명, 날짜 format }, target => "@timestamp", timezone=>"Asia/Seoul" } : 엘라스틱에 데이터 저장할 때 데이터 보내는 시간이 아닌 실제 로그 파일에 찍힌 시간으로 저장하기 위해 원하는 필드로 재설정하는 구문 
    mutate { convert => [ 변경할 필드명, 데이터 타입 ] } 
  • output example 
    전처리한 데이터를 전송할 곳 지정 
    

    # 기본 틀
    output {
        elasticsearch {
            index => "%{[some_field][sub_field]}-%{+YYYY.MM.dd}"
        }
    }
    
    # 실제로 구성한 내용
    output {
      if "IMP_PRODUCT" in [log][file][path] {
        elasticsearch {
          hosts => ["ip 주소:9200"]
          manage_template => false
          index => "2020-imp-%{[@metadata][beat]}-%{[host][name]}"
        }
      }
      else if "CLICK" in [log][file][path] {
        elasticsearch {
          hosts => ["ip 주소:9200"]
          manage_template => false
          index => "2020-click-%{[@metadata][beat]}-%{[host][name]}"
        }
      }
    }

5. Logstash 실행과 로그 

• 실행
  설정 파일 변경 사용 적용을 위해 재시작 필요 
• $ systemctl restart logstash​
• 로그 보기

  $ tail -f /var/log/logstash/logstash-plain.log​

---

References

• https://soyoung-new-challenge.tistory.com/99
• https://www.elastic.co/kr/logstash/