[ELK] #2 Logstash.yml

ddalkigum·2022년 2월 11일

ELK Logstash

Elastic

목록 보기

2/2

로깅 시스템 구축부터 데이터 시각화까지

어떤식으로 로그를 전달할지

각 서버에서 로그 저장
Filebeat를 이용하여 Logstash로 전달
Logstash에서 필터링하여 Elasticsearch에 저장
Kibana를 통해 시각화

Log stash 설정에서 생각보다 시간을 많이 잡아먹음

Filebeat

일단 로컬에서 실행했기에 맥북에 Filebeat을 설치

filebeat.yml은 지난번에 올렸던 글에서 사용했던 파일 그대로 사용했고, 달라진건 EC2서버를 종료해뒀다가 재기동해서 IP가 변경됨

Logstash IP만 변경해주고 실행,
path는 로그 쌓아두는 곳으로 설정

Brew로 설치

brew install filebeat

직접 설치가 필요하다면 이전글을 보면 됩니다
https://velog.io/@ddalkigum/ELK-Filebeat-1-%EC%84%A4%EC%B9%98

Brew는 정말 사용할때마다 느끼지만 정말 사기다

filebeat.yml

filebeat.inputs:
	- type: log
	enabled: true
	encoding: utf-8
    paths:
    - $LOG_PATH

Filebeat가 동작 하는 방식

일단 filebeat의 로그를 보면 Go로 개발이 되어있고,
아직 여러 설정을 건드려보진 않았지만

Elastic 공식홈페이지에 나와있는 걸 보면 로그를 보내는
interval을 logstash에서 설정을 해줄 수 있는 것으로 보인다

자세한 옵션설정은 천천히 살펴보기로 하자

Filebeat는 inputs와 harvesters로 구성되어있다

harvesters
각 파일을 한줄씩 읽고 내용을 보내는 역할
파일당 하나의 havesters가 담당한다

scan_frequency라는 옵션이 있고 기본값이 10초로 설정되어 있다
10초에 한번씩 파일을 읽고 수집된 로그를 보내는 방식인 셈이다

완전히 실시간은 아니다 엘라스틱에서도 웬만하면 설정하지 않는 것을 추천한다

실시간으로 작동하도록 하려면 close_inactive 옵션을 설정하면 되는 것 같다

로그 파일을 오픈하여 지켜보는 시간을 설정해주는 것이고, 5분으로 설정했다면
5분내로 로그가 쌓이지 않을 경우 오픈했던 로그파일을 닫아주는 것이다

이건 굳이? 라는 생각이 드는게 계속 지켜본다면 그 만큼의 메모리가 할당되게 되는 것이고
실시간 처리가 중요하다면 감수하겠지만, 계속 할당하는 건 비효율적이라고 생각된다

inputs
input type을 지정해줄수 있고, 내가 사용한 설정은 log로 되어있다

harvesters가 각각의 파일을 지켜보도록 설정한다 했는데
여기서 고루틴이 사용된다
고루틴은 멀티쓰레드와 비슷하고, OS에서 쓰레드를 가지고오는 것이 아니기 때문에 멀티쓰레드 처리보다 비용이 적게 사용되는 것으로 알고 있다

Go언어는 사용해본적이 없어서 정확히는 모르겠네요

마지막 위치를 기억하는 방법
harvester가 마지막으로 읽은 위치에 offset을 저장하고, 이후 추가된 로그가 있는지 확인한다

Filebeat가 실행되는 동안 log파일에 입력된 상태정보를 메모리에 유지한다고 나와있는데
Filebeat를 종료하고 로그 중간을 삭제하면 어떻게 되는거지?

궁금하네

레퍼런스: https://www.elastic.co/guide/en/beats/filebeat/current/how-filebeat-works.html
filebeat옵션: https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-input-log.html#filebeat-input-log-scan-frequency

Logstash

사실 오늘 글을 쓰는 이유는 Filebeat보다는 이녀석 때문이었다

Filebeat는 애초에 로그를 전달하는 역할로만 사용하려 했기에 깊게 파고들지 않았다
설정에서 너무 애먹었고, 지금되돌아보면 정말 별거 아닌 것 같다

뭐가 잘못된지 전혀 몰랐다

filter {
  grok {
    match => {"message" => "(?<request_time>%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{TIME}) (?<log_level>%{WORD}): %{WORD:http_method} %{URIPATHPARAM:request} %{NUMBER:http_status} %{NUMBER:content_length} - %{NUMBER:response_time}"}
  }
}

거의 한시간에서 두시간동안 삽질한 것같다
왜 로그가 전달이 안되지? grok문법이 잘못됬나? kibana페이지에서 테스트도 했는데
하면서 Logstash가 잘못된게 분명하다고 생각했다

filter {
  grok {
    match => {"message" => ["(?<request_time>%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{TIME}) (?<log_level>%{WORD}): %{WORD:http_method} %{URIPATHPARAM:request} %{NUMBER:http_status} %{NUMBER:content_length} - %{NUMBER:response_time}"]}
  }
}

분명 나와 같은 실수를 하는 사람이 있을거다
grok문법에는 틀린게 없었고, []를 사용하지 않아서 로그가 제대로 필터링 되지 않았다...

grok 문법

처음 사용해본 친구인데 일단 문법관련 페이지 부터

Logstash github
https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns

위에 들어가보면 정규식을 어떤걸 사용해야하는지 나와있다
이걸보고 그대로 따라하면 된다

이것도 모르고 다른 블로그를 보면서 필터링 어떻게 사용해야 하지 Logstash에 바로적용했는데 안되면 어떡하지 생각했는데

Management > Dev tools

다행히도 Kibana페이지에서 Grok Debugger 기능을 제공해준다

여기서 테스트후 적용해주면 깔끔하게 나온다

데이터를 정제해주고 Elasticsearch로 보내주면 된다

위 두가지에서 가장 애먹었고 나머지는 나쁘지 않았던 것 같다

작동 방식

이전글 그대로 설치했다면 logstash.yml파일은 수정할 필요가 없다

logstash 디렉토리에 들어가면 위의 파일들을 볼 수 있는데
여기서 conf.d에 pipe line설정만 해주면 된다

yml파일

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    match => {"message" => ["(?<request_time>%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{TIME}) (?<log_level>%{WORD}): %{WORD:http_method} %{URIPATHPARAM:request} %{NUMBER:http_status} %{NUMBER:content_length} - %{NUMBER:response_time}"]}
  }
}

output {
  elasticsearch {
    hosts => "52.79.234.81:9200"
    index => "filebeat-"
  }
}

보는 것처럼 간단하게 사용했고
다양한 옵션을 사용하지 않은 것을 변명하자면...
백엔드쪽 공부하는데 내가 구축해보고 싶은건 로그에서 시각화까지의 파이프라인이지
어떤 옵션을 사용해야 될지가 아니어서... 이건 간단하게 넘어가고
현재 하고있는 프로젝트에 시간을 좀더 쓰고 싶었다

Express 로그 세팅

Winston logger

const logFormat = printf(({ level, timestamp, message }) => {
      return `${timestamp} ${level}: ${message}`;
    });

    const debugFormat = printf(({ level, message }) => {
      return `${level}: ${message}`;
    });

    this.logger = winston.createLogger({
      format: combine(
        timestamp({
          format: 'YYYY-MM-DD HH:mm:ss',
        }),
        logFormat,
      ),

Morgan logger

stream: StreamOptions = {
    write: (message) => this.logger.http(message),
  };

  public init = () => {
    return morgan(':method :url :status :res[content-length] - :response-time ms', { stream: this.stream })
  };