Splunk로 효율적인 실시간 누적 통계 검색하기

스플렁크를 보안관제, 모니터링에 이용하는 사람들이 자주 쓰는 검색 / 경고 / 보고서 중 하나는
바로 '00분간 00건의 로그가 들어왔는지 확인하기'다.

예를 들어 30분간 방화벽 로그가 미수집 되면 로그 수집 상태 경고를 띄운다던가,
10분간 특정 방화벽에 1만건의 로그가 들어오면 DDoS 공격 경고를 띄운다던가.

DDoS 경고 쿼리를 짠다고 가정할 시,
아무 생각 없이 짜서 넣을 수 있는 쿼리는  다음과 같다.

index=보안로그 sourcetype=방화벽 earliest=-10m latest=now
|stats count by host
|where count > 10000

이 쿼리를 10분마다 검색한다면, 딱히 더 최적화를 할 여지는 없다.

하지만 이런 식의 관제, 모니터링용 시나리오들은 실시간이거나 비슷한 수준의 실시간성을 요하기 때문에,
대부분의 사용처는 1~3분 간격으로 이를 반복 검색하게 마련이다.

예를들어 1분 간격으로 해당 쿼리를 반복 검색하는 경고를 만들었다면,
1분간 쌓인 새로운 데이터를 검색하기 위해 이미 예전에 검색했던 9분간의 데이터를 매번 중복 검색해야하지 않는가?

방화벽이 몇 개 없다면 또 모르겠지만,
수 백개의 방화벽에서 수 만개의 로그를 한 시간마다 수집하는 회사에서는 매번 새로 검색하는 9분간의 데이터도 시간이 몇 십초는 걸리게 마련이다.

검색은 어차피 1분마다 하니 새로운 데이터는 1분 어치씩만 업데이트 하는 것이 효율적이다.
하지만 이전 9분간의 검색 결과를 따로 저장해야 할 곳이 필요하니,  룩업 기능을 이용해 이를 구현한다.

| inputlookup 디도스탐지.csv
| append [
    index=보안로그 sourcetype=방화벽 earliest=-1m
    |stats count by host
    |eval time=now()
         ]
| where time > relative_time(now(),"-10m")
| outputlookup 디도스탐지.csv
| stats sum(count) as count by host
| where count > 10000

이 방식은 지난 10분 간의 기록이 담긴 디도스탐지.csv 를 불러와
append 명령어를 이용해 지난 1분 간의 신규 데이터를 검색한다.
stats 명령어를 통해 나온 결과표에는 시간 정보가 없기 때문에 eval 구문을 통해  따로 time 정보를 기록한다.

이로서 11분간의 데이터가 나오면  where 명령어를 이용해 최신 10분간의 데이터만 보여주도록 하고,
이를 outputlookup 명령어를 이용해 디도스탐지.csv에 다시 덮어쓴다.

이후 stats 명령어를 다시 사용해 지난 10분간 모인 count 값을 sum으로 모두 합해주면,
우리가 원하는 결과가 나오게 된다.

이 방식은 쿼리 실행 주기가 짧을 수록, 전체 대상 기간이 길 수록 효율이 뛰어나다.
1분간격으로 30분간의 데이터를 검색할 경우 정직한(?) 방식보다 효율이 10배 이상 차이가 날 정도다.