[ELK] Elasticsearch Watcher 설정

회사에서 ELK를 통한 알림 시스템을 구축하려고 하는데 Alerting 및 Watcher 기능을 살펴보려고 한다.

 

Alerting : kibana 에서 발생시킴

Watcher : Elasticsearch 에서 발생시킴

 

Alerting 

Kibana 8.10.2 기준으로 

Kibana에서 Alerting 설정은 Rules을 만들어서 알림을 발생시키는데 GUI 기반으로 편하게 만들 수 있는 장점이 있다. 

다양한 임계값 설정과 알림 예시들이 많아서 쉽게 사용가능하고 QueryDSL도 사용가능해서 검색옵션을 풍부하게 할 수 있다. 하지만 여러개의 인덱스를 통해 조건을 거는건 불가능하며 간단한 알림 설정을 할 때 유용하다.

 

Watcher

watcher는 Alerting 보다 더 자세하게 알림을 만들 수 있다. 

Watcher | Elasticsearch Guide [8.13] | Elastic

 

watcher는 input, condition, transform, action, trigger로 이루어져 있으며 각각의 역할이 다르다.

 

trigger

말 그대로 트리거이며 시간 설정을 하면서 몇초 간격으로 watcher를 실행시킬지 정하는 부분

input

watcher에 데이터를 수집하는 부분이다. 가장 많이 쓰는 것이 search input이며 elasticsearch index를 검색해서 그 결과값을 input으로 쓰겠다는 얘기다.

http input을 통해 웹훅을 elasticsearch index에 저장하여 사용가능하고 여러개의 input을 받고 싶으면 chain input을 사용하는 등 여러 방법이 있다.

여기서 유의할 점은 input마다 condition(조건)을 걸 수 있는데 "데이터 수집 후 조건 실행" 이라는 점이다.

 

예를 들어 A index를 현재 시간 기준 10초전까지의 로그를 input으로 하려는데 조건을 hits.total(count)가 100 이상일때만 input으로 쓰겠다는 조건을 쓴 input이 있다. 이런 input이 chain input으로 여러개 있다고 치면 사용되지도 않을 로그들을 먼저 수집해버리기 때문에 리소스 낭비가 존재한다. 

따라서 input condition을 할 때 낭비되는 리소스가 발생하지 않게 세부적으로 하는 것이 중요하다.

 

search input이 가장 많이 쓰이며 문법도 search query랑 동일하기 때문에 문법을 잘 알아야한다.

대표적으로 중요한 3가지를 살펴보면 match, match_all, match_phrase 가 있다.

> match_all

{
  "query":{
    "match_all":{ }
  }
}

 

> match

아래처럼 2단어 이상을 match로 했을 때 문제점은 "Crystal", "Street" 각각 포함되어있는 로그도 검색이 된다는 것이다.

# OR 연산
{
  "query": {
    "match": {
      "address": "Crystal Street"
    }
  }
}

 

and를 걸어버리면 어느정도 해결은 되지만 "Crystal test Street" 라고 해도 검색이 된다.

# AND 연산
{
  "query": {
    "match": {
      "address": {
        "query": "Crystal Street",
        "operator": "and"
      }
    }
  }
}

> match_phrase

정확한 "Crystal Street"을 검색하기 위해선 match_phrase를 사용하면 된다.

{
  "query": {
    "match_phrase": {
      "message": "Crystal Street"
    }
  }
}

transforms

Input으로 가져온 로그 데이터 중에서 특정 필드를 추출하거나, 특정 시간 간격으로 데이터를 집계하여 Watcher의 조건을 더 정확하게 만든다. script를 적용할 수 있기 때문에 좀 더 정교한 작업이 가능하다.

최상단에 위치할 수 있고 action안에 위치할 수 있다.

{
  "trigger" : { ...}
  "input" : { ... },
  "condition" : { ... },
  "transform" : { 
    "search" : {
      "body" : { "query" : { "match_all" : {} } }
    }
  },
  "actions" : {
    "my_webhook": {
      "transform" : { 
      	"script" : "return ctx.payload.hits"
      },
      "webhook" : {
      	"host" : "host.domain",
      	"port" : 8089,
      	"path" : "/notify/{{ctx.watch_id}}"
      }
    }
  ]
}

그래서 특정 search query로 로그를 검색한다음에 script로 걸러서 결과를 받고 싶으면 chain payload transform을 사용하면 유용할 것이다.

"transform" : {
  "chain" : [ 
    {
      "search" : {  
        "indices" : [ "logstash-*" ],
        "body" : {
          "size" : 0,
          "query" : {
            "match" : { "priority" : "error" }
          }
        }
      }
    },
    {
      "script" : "return [ error_count : ctx.payload.hits.total ]"  
    }
  ]
}

 

 

1. gte (Greater-than or equal to) - 이상 (같거나 큼)
2. gt (Greater-than) – 초과 (큼)
3. lte (Less-than or equal to) - 이하 (같거나 작음)
4. lt (Less-than) - 미만 (작음)

 

condition

input에 대해서 조건을 거는 부분이다. source 부분은 항상 return 값이 있어야하며 보통 painless 언어로 작성을 많이 한다.

source 부분은 보통 들여쓰기없이 한줄로 적는데 여러줄로 적고 싶으면 """에 감싸서 적으면 된다.

내 경험상 watcher에서 설정할 때 """로 source를 감싸면 빨간줄이 뜨긴하는데 save는 되는거보니 큰 에러는 아닌거 같고 WARNING 정도인거 같기도 하고...???

"condition" : {
  "script" : {
    "source" : "return params.result",
    "lang" : "painless",
    "params" : {
      "result" : true
    }
  }
}

action

action은 간단하게 결과를 어떤 방식으로 알릴꺼냐를 설정하는 구간이며 webhook, email 다양하게 사용 가능하다.

**  webhook을 설정할 때 내/외부 host와 통신이 안되면 kibana의 connector를 보는게 아닌 Elasticsearch을 봐야한다.

나도 connection timeout이 지속적으로 떠서 kibana connector를 이용해 연결해봤는데 OK가 뜨는걸 보고 Watcher action도 Elasticsearch에서 수행하는 것으로 판단했고 실제로 elastic node log를 봤을 때 해당 오류가 떴다.watcher를 실행하는 Elastic node는 master node인데 node를 여러개 운용함과 동시에 master rules를 설정했다면 내부적으로 밸런싱으로 인해 watcher를 실행하는 node가 바뀔 수 있으니 주의!!!!

"actions" : {
  "my_webhook" : { 
    "transform" : { ... }, 
    "throttle_period" : "5m", 
    "webhook" : {
      "method" : "POST", 
      "host" : "mylisteningserver", 
      "port" : 9200, 
      "path": ":/{{ctx.watch_id}}", 
      "body" : "{{ctx.watch_id}}:{{ctx.payload.hits.total}}" 
    }
  }
}

chatgpt는 host,port,path를 url 하나로 합쳐서 써라라고 하는데 에러가 나던데 뭐가 맞는지는 모르겠다 

"foreach" : "ctx.payload.hits.hits"

foreach를 추가하면 input 로그 개별적으로 수행가능