[Elasticsearch] Logstash를 사용한 Syslog

Syslog

Syslog는 네트워크 장치에서 생성된 로그 데이터를 일정한 포맷으로 한군데 모아주는 대중적인 장치이다. 이번에는 Logstash를 사용하여 서버와 장치에서 Syslog 메세지를 수집하는 방법에 대해 알아보자.
Syslog는 프로그램 에러, 알림, 경고 상태 메세지 등과 같은 로그 정보를 생성하고 수집하는 표준화된 방식으로 Linux 등 대부분의 Unix 계열의 운영 시스템에서 Syslog Daemon을 사용하여 정보를 수집하고 보관한다. 일반적으로 Syslog는 로컬에 저장되지만 관리자가 한 곳에서 모든 로그에 Access하길 바란다면 모든 log를 중앙 서버로 streaming할 수도 있다. 기본적으로 PORT 514와 UDP 연결이 SYslog 전달에 사용되지만, 네트워크가 불안정하거나 혼잡할 때 모든 로그가 정상적으로 송수신 된다는 보장이 없기 때문에 가능하면 UDP를 사용하지 않는 것이 좋다. 보안과 안정성을 위해서 PORT 6514가 TCP연결, TLS암호화와 함께 자주 사용된다.
이를 Elasticsearch와 연결해서 사용하면 원하는 데이터를 빠르고 효율적으로 검색할 수 있다.
syslog의 형태를 확인하고 싶다면 sudo head -10 /var/log/syslog 로 간단히 확인해 볼 수 있다.

Syslog Daemon

Syslog Daemon을 사용하여 Logstash를 실행하는 중앙 서버로 TCP 연결을 해서 로그를 전송하는 방법이다.
Syslog Daemon은 수집한 모든 로그 이벤트를 TCP연결을 통해 다른 장치로 전송할 수 있고, Logstash는 TCP 포트를 열어 수신되는 Syslog Data를 선별하는 기능이 있다.
현재는 VirtualBox 환경에서 진행하고 있기 때문에 이를 기준으로 연습을 해 보려고 한다.
Syslog를 가상 머신의 10514 TCP Port로 전달하면 Logstash가 Port 10514의 모든 메세지를 수신 할 것이다.
Syslog Daemon의 config file을 수정해보자. sudo vim /etc/rsyslog.d/50-default.conf
파일의 상단 아무 부분에 다음의 내용을 추가하자.

• *.* @@127.0.0.1:10514
  • *.* 는 모든 메세지르르 전송하라는 뜻이고
  • @@는 TCP 연결을 통해 데이터를 전송하겠다는 뜻이다.
• 이제 Syslog Daemon을 재시작해줘야한다.
  • sudo systemctl restart rsyslog.service
• logstash config file을 구성한다.

input {
  tcp {
    port => 10514
    type => syslog
  }
}

filter {
  if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
      target => "syslog_timestamp"
    }
  }
}

output{
  elasticsearch{
    hosts => ["localhost:9200"]
    index => "syslog-received-on-tcp"
  }
  stdout {
    codec => "rubydebug"
   }
}

• 정상적으로 로그가 수집된다면 정상적을 동작하고 있는 것이다.

Logstash 사용

Logstash를 사용하여 각 서버 또는 장치의 로그 파일을 모니터링하고 메세지를 Elasticsearch에 자동으로 index화한다.

input {
  file {
     path => ["/var/log/syslog"]
     type => "syslog"
     start_position => "beginning"
   }
}
filter {
       if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
      target => "syslog_timestamp"
    }
  }
}
output{
  elasticsearch{
    hosts => ["localhost:9200"]
    index => "syslog-monitor"
  }
}

• 여기서 중요한 것은 /var/log/syslog 파일을 모니터하는 것이다. 위의 구성으로 Logstash를 실행시켜보자.
• curl -XGET localhost:9200/syslog-monitor/_search?pretty -d '{"size":1}' 커맨드로 확인해보니 정상적으로 데이터가 index화 되었다.

{
  "took" : 5,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 2075,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "syslog-monitor",
        "_type" : "_doc",
        "_id" : ,
        "_score" : 1.0,
        "_source" : {
          "syslog_hostname" : ,
          "received_at" : ,
          "@timestamp" : ,
          "syslog_timestamp" : ,
          "received_from" : ,
          "path" : "/var/log/syslog",
          "message" : 
        }
      }
    ]
  }
}

위와 같이 결과가 나오면 성공이다.

이렇듯 모든 로그를 한 곳에 수집하는 것은 여러 장점이 있다. 예를 들어 모든 항목을 더욱 쉽게 접근하고 검색할 수 있다. 또한 자주 발생하는 서버 장애에 대한 로그를 잃지 않아 이에 대응이 가능하다. Elasticsearch를 사용하여 시스템이 손상되더라도 원본 로그는 그대로 유지되고 이를 통해 문제가 발생하기 전에 어떤 일이 일어났는지 확인할 수 있다.