logstash를 활용한 원격 로그 수집 - 시작하기

각 컴포넌트 설치

elasticsearch & logstash & filebeat 설치

설치는 간단하다. 자세한 설명은 하지 않겠다. elastic 공식 사이트로 접속하여 스테이블한 tar파일을 받아서 로컬의 적절한 경로에 압축 해제 하면 된다.

각 컴포넌트 구동 및 설정

elasticsearch 구동

결과적으로 elasticsearch에 데이터를 저장하므로 먼저 elasticsearch를 구동한다. 
가장 심플하게 설치디렉토리로 가서 아래의 커맨드로 구동한다.

./elasticsearch

ingestion with filebeat

filebeat는 파일의 변경을 수집하여 logstash로 전달하는 역할을 담당한다. 
가장 filebeat를 압축해제한 디렉토리를 보면 filebeat.yml 설정파일이 존재하는데 다음과 같은 내용을 추가해보자.

첫번째 크롤한 패스를 적절한 패스를 아래와 변경하자.

- input_type: log

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /daum/logs/sdb-admin-homo-new/*

다음은 멀티 라인 옵션을 변경하자. 멀티라인옵션이 없을 경우 new line 기준으로 아이템을 구분한다. 
아래의 예는 로그에서 [yyyy-mm-dd 형태로 시작할 경우 item을 구분하겠다는 의미이다.

  multiline.pattern: '^\[[0-9]{4}-[0-9]{2}-[0-9]{2}'
  multiline.negate: true
  multiline.match: after 

그리고 전송할 logstash의 주소를 입력한다.

#------------------------ Logstash output ---------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5043"]

마지막으로 아래와 같은 커맨드로 실행한다.

./filebeat -e -c filebeat.yml -d "publish"

아직 입력을 처리할 logstash가 설정되지 않아서 아래와 같은 로그가 발생할 것이다.

2017/08/25 05:11:42.786091 single.go:140: ERR Connecting error publishing events (retrying): dial tcp 127.0.0.1:5043: getsockopt: connection refused
2017/08/25 05:12:04.341011 metrics.go:34: INFO No non-zero metrics in the last 30s
2017/08/25 05:12:34.340861 metrics.go:34: INFO No non-zero metrics in the last 30s

processing with logstash

아래와 같이 가장 간단한 형태의 설정파일을 만들어보자. 예제에서는 이름을 first-pipeline.conf라고 하겠다. 
5043 포트로 filebeat의 송신을 listen하고 그 결과를 엘라스틱 서치에 전달한다.

input {
        beats {
                host => "localhost"
                port => "5043"
                tags => ["sdb-homo-admin-new"]
        }
}

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

마지막으로 아래와 같은 커맨드로 logstash를 기동한다.

./logstash -f first-pipeline.conf --config.reload.automatic

logstash가 정상적으로 기동되면 filebeat에는 다음과 같은 로그가 보일 것이다.

2017/08/25 05:29:44.044839 sync.go:70: DBG  Events sent: 1148
2017/08/25 05:29:44.044969 client.go:214: DBG  Publish: {
  "@timestamp": "2017-08-25T05:02:34.358Z",
  "beat": {
    "hostname": "sonjeong-ug-ui-MacBook-Pro.local",
    "name": "sonjeong-ug-ui-MacBook-Pro.local",
    "version": "5.5.2"
  },
  "input_type": "log",
  "message": "[2017-08-25 09:54:17,950] [localhost-startStop-2] INFO  o.s.w.c.s.XmlWebApplicationContext - Closing WebApplicationContext for namespace 'admin-servlet-servlet': startup date [Thu Aug 24 18:36:38 KST 2017]; parent: Root WebApplicationContext",
  "offset": 501734,
  "source": "/daum/logs/sdb-admin-homo-new/undefined.log.log",
  "type": "log"
}
2017/08/25 05:29:44.044995 output.go:109: DBG  output worker: publish 1 events
2017/08/25 05:29:44.053219 sync.go:70: DBG  Events sent: 1
2017/08/25 05:29:44.053257 sync.go:70: DBG  Events sent: 1

search in elasticsearch

elasticsearch에 생성되는 인덱스명은 logstash-$DATE 형식이다. 아래와 같은 쿼리로 조회를 해보자.

curl "http://localhost:9200/logstash-2017.08.25/_search?pretty&q=response=200"