Logstash에서 REST API 연동하기

Logstash에서 REST API를 연동하여 데이터를 수집하는 방법을 정리한다.

 

Logstash에서 REST API를 연동하기 위해서는 http_poller 플러그인을 사용한다.

라고 되어있지만... 막상 사용해 보면 생각보다 제약조건이 많기 때문에 제대로 된 REST API 데이터 수집이 불가하다. 이는 input 플러그인에서 특정 정보를 동적으로 변경하는 것이 어렵기 때문이다.(불가한 건 아니지만 배보다 배꼽이 더 크다;;)

가령, 일자별로 미세먼지 정보를 제공하는 API가 있다고 가정하자. http_poller에서 해당 API를 사용하기 위해서는 파라미터로 일자 정보를 설정해줘야 한다. 고정된 날짜 정보를 설정해주는 건 가능하지만 오늘 날짜를 가져와서 파라미터로 설정하는 건 현재 버전의 logstash에서는 불가능하다.(필자가 못 찾은 걸 수도 있으니, 가능하다면 댓글을 부탁드린다.)

input에서 OS 환경변수 값을 가져올 수 있기 때문에 이를 활용하는 방법을 엘라스틱 커뮤니티에서 안내하기도 하는데, 주기적으로 호출이 일어나는 상황에서는 이마저도 쉽지 않다. 

고민 끝에 필자가 적용한 방법은,
더미 http_poller를 사용하여 주기적 호출을 수행하고 실제 REST API의 호출은 필터에서 http 플러그인을 사용했다.

input {
  # polling을 위한 더미 input
  http_poller {

    urls => {
      dummy => {
        url => "http://localhost"
      }
    }

    request_timeout => 60
    schedule => { cron => "0 10 * * *"}
    codec => "json"
  }
}


filter {
  # 어제날짜를 변수에 설정
  ruby {
    code => "event.set('yesterday', (Time.now - 86400).strftime('%Y%m%d'))"
  }

  # REST API 호출
  http {
    url => "https://api.test.com/getdata?date=%{yesterday}"
    body_format => "json"
  }

  # 배열 데이터 분리
  split {
    field => "[body][total]"
  }

  mutate {
    add_field => {
      "cpid" => "%{[body][cpid]}"
      "totalcount" => "%{[body][total][totalcount]}"
      "programid" => "%{[body][total][programid]}"
      "channelid" => "%{[body][total][channelid]}"
      "date" => "%{[body][total][date]}"
    }
    remove_field => [ "headers", "@version", "body", "yesterday" ]
  }

  # date 필드 추가
  date {
    match => ["date", "yyyyMMdd"]
    timezone => "Asia/Seoul"
    locale => "ko"
    target => "date"
  }

  # mutate 필터 안에서 처리할 경우, 필터 내부의 순서를 보장받지 못함
  mutate {
    convert => {
      "totalcount" => "integer"
    }
  }
}

무의미한 호출이 한번 일어나기 때문에, 이는 비용의 낭비로 볼 수 있다. 하지만 현재로서는 이 방법이 가장 명확해 보인다.

ruby 플러그인을 사용해서 변수(yesterday)에 값(어제 날짜)을 설정했고, 이를 http 플러그인에서 동적 파라미터로 활용했다. 다양한 파라미터를 이렇게 설정하여 호출할 수 있다. 

그밖에, 반환된 JSON 데이터를 분리하기 위해 split 플러그인을 사용했고, kibana에서 사용하기 위해 totalcount의 타입을 숫자로 변경했다. 이때, mutate 플러그인을 두 개로 나눠서 사용한 이유는 필터 내부의 순서를 보장할 수 없기 때문에 타입 전환이 정상적으로 일어나지 않기 때문이다.

ps. REST API를 DB에 저장해서 jdbc 플러그인으로 가져올 수도 있겠지만, 본 글에서는 HTTP 호출을 통한 직접 수집을 목표로 했다. 독자분들께서 더 좋은 방법을 알고 계시다면, 댓글로 소통을 꼭 부탁드린다.