Logstash, 데이터 파이프라인의 마술사

 
오늘은 오픈소스 데이터 수집, 처리, 전송 도구 Logstash에 대해 알아보자.
 
Logstash는 다양한 소스에서 데이터를 수집하고, 필터링, 가공하여 Elasticsearch 등의 저장소에 전송한다.
 
Logstash는 크게 입력(input), 필터(filter), 출력(output) 3가지 파이프라인으로 구성되어있는데 각각은 다채롭게 활용 가능하여 강력한 확장성을 가진다.

 

input, filter, output 설정들에 대해 예제를 통해 알아보자.

 

기본적으로 conf.d 폴더 내에 설정 파일들을 둬서 설정한다.

 

 

input { beats { port => 5044 host => "0.0.0.0" }}

input 설정을 이런 식으로 해준다면 모든 IP에서 오는 데이터들을 5044 포트로 들어오게 한다.

 

 

filter {
  grok {
    match => {
      "message" => [
        "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{DATA:logger} - %{GREEDYDATA:log_message}",
        "%{GREEDYDATA:log_message}",
        "%{DATA:service_name}/%{DATA:module_name}:%{DATA:instance_id} - %{GREEDYDATA:log_message}"
      ]
    }
  }

  if [level] == "INFO" {
    mutate {
      add_field => {
        "thread" => ""
      }
      rename => {
        "level" => "log_level"
      }
    }
    date {
      match => [
        "timestamp",
        "ISO8601"
      ]
      target => "@timestamp"
      remove_field => [
        "timestamp"
      ]
    }
    if [service_name] == "your-service" {
      mutate {
        add_field => {
          "[@metadata][index]" => "your-service-log"
        }
      }
    } else {
      mutate {
        add_field => {
          "[@metadata][index]" => "app-log"
        }
      }
    }
  } else if [level] == "WARN" or [level] == "ERROR" or [level] == "FATAL" {
    mutate {
      add_field => {
        "thread" => ""
      }
      rename => {
        "level" => "log_level"
      }
    }
    if [service_name] == "your-service-2" {
      mutate {
        add_field => {
          "[@metadata][index]" => "your-service-2-log"
        }
      }
    } else {
      mutate {
        add_field => {
          "[@metadata][index]" => "warn-log"
        }
      }
    }
  } else {
    drop {}
  }
}

 

filter 설정 예시.

 

 

로그스태시에서 사용되는 몇 가지 필터에 대해 알아보자.

 

• Grok

Grok 필터는 로그 메시지에서 필드를 추출하는 필터다.

 

로그 메시지에서 특정 패턴에 해당하는 문자열을 찾아 추출할 수 있다.

 

이 필터는 match 옵션으로 패턴을 지정하고, add_field 옵션으로 새로운 필드를 추가하거나 rename 옵션으로 기존 필드의 이름을 변경할 수 있다.

 

• Date

로그 메시지에서 날짜와 시간 정보를 추출하여 @timestamp 필드에 저장하는 필터다.

 

이 필터는 로그 메시지에서 날짜와 시간 정보를 추출할 수 있는 match 옵션을 제공한다.

 

• Mutate

Mutate 필터는 필드를 조작하는 데 사용된다.

 

필드를 추가하거나 삭제하거나, 필드의 이름을 변경하거나, 필드의 데이터 형식을 변경하는 등의 작업을 수행할 수 있다.

 

이 필터 또한 Grok 필터와 마찬가지로 add_field, remove_field, rename 등의 옵션을 제공한다.

 

• Drop

Drop 필터는 이벤트를 삭제하는 데 사용된다.

 

예를 들어, if 문에서 level이 특정 값을 가지지 않는 이벤트를 삭제할 수 있다.

 

이 필터는 drop 옵션만 가진다.

 

 

output { 
	elasticsearch { 
		hosts => ["http://192.168.56.101:9200", "http://192.168.56.102:9200", "http://192.168.56.103:9200"] 
		index =>  "%{[@metadata][beat]}-%{[@metadata][version]}-%{[@metadata][index]}"
        }
}

마지막으로 output 설정이다.

 

엘라스틱서치로 전송하는 output 예시로 주로 엘라스틱서치에 보내긴 하지만, 다른 데이터 파이프라인으로 전송해도 무방하다.

 

이렇게 다양한 입출력과 다채로운 필터를 가진 로그스태시를 여러분의 데이터 파이프라인에 추가해보길 바란다.

 

 

Thanks for watching, Have a nice day.