Synology Chat Webhook 을 통한 메시지 전송(Feat. Jira, Python)

서론

Jira 를 통해서 이슈가 관리되어지고 있는 상황에서 회사 메신저가 Synology Chat 인 경우에는

일반적인 방법으로는 이슈알림을 받기가 어렵다.

Slack 의 경우 Jira에서 공식적으로 지원을 하고 있기에 이러한 부분에서 유용한 도움을 받을 수 있지만

만약에, Slack을 사용하지않고 다른 Chat 어플리케이션을 쓰는 경우에는 Jira 의 자동화 그 중에서 웹요청을 통해 이슈 알림을 받을 수 있다.

 

Synology Chat

일단 Synology Chat 의 경우 다른 외부에서 들어오는 요청을 통해서 채팅방 혹은 개인에게 메시지를 보내는 것이 가능하다.

물론 이 과정속에서 들어오는 Webhook 을 설정할 필요가 있다.

 

해당하는 부분은 설정 > 통합 > 들어오는 webhooks 을 통해서 가능하고 Bot 도 가능하다.

 

이러한 부분은 다른 곳에 자세히 설명이 되어있으니 참고하면 될 것 같다.

그리고 처음에는 shell script 로 한번 작성을 했었는데 더욱 더 로직이 많아지면서 Python 으로 작성하는 것이 유리하다고 판단해 Python으로 작성했다.

그러면 Python 을 통해서 어떻게 보내는 지 알아보자.

 

코드

환경은 Lambda 기반으로 Jira -> API Gateway -> lambda -> Synlogy Chat 흐름이다.

Lambda 의 경우에는 Layer 를 통해서 외부라이브러리를 올릴수 있지만 그렇게 하지는 않았다.

이유는 이 프로젝트에서 필요한 것이 requests 하나인데 그것 때문에 올리기에는 약간 애매했다.

그러다보니 문제가 있었는데

urllib3 기반으로 메시지데이터 전송하는 것과 requests 라이브러리로 데이터를 전송하는 것에 인코딩차이가 존재한다는 것이였고

requests 경우에는 사용자가 따로 별도의 로직을 작성할 필요가 없었다면, urllib3 는 별도의 로직 작성이 필요했다.

 

그래서 추가된 것이 to_key_val_list(), encode_params() 이다.

현재 보내는 데이터가 Iterable 한 데이터이기 때문에 그 Items 도 encoding 을 해줘야한다.

그래서 해당 부분까지 작성하는 코드를 추가적으로 작성했다.

참고로 해당 코드는 requests 의 라이브러리 코드를 참고했다.

 

코드의 흐름은 이벤트 호출 -> Synology Chat  에서 User list 조회 및 저장 -> 담당자 Email 과 Synology Email 비교 및 담당자 ID 저장 -> 워크플로우에 맞게 메시지 작성 -> Synology Chat 에 User id 기반으로 메시지 전송(인코딩 포함)

import json
import urllib3
import urllib

def to_key_val_list(value):
   
    if value is None:
        return None

    if isinstance(value, (str, bytes, bool, int)):
        raise ValueError("cannot encode objects that are not 2-tuples")

    value = value.items()

    return list(value)

def encode_params(data):
    
    basestring = (str, bytes)

    if isinstance(data, (str, bytes)):
        return data
    elif hasattr(data, "read"):
        return data
    elif hasattr(data, "__iter__"):
        result = []
        for k, vs in to_key_val_list(data):
            if isinstance(vs, basestring) or not hasattr(vs, "__iter__"):
                vs = [vs]
            for v in vs:
                if v is not None:
                    result.append(
                        (
                            k.encode("utf-8") if isinstance(k, str) else k,
                            v.encode("utf-8") if isinstance(v, str) else v,
                        )
                    )
        return urllib.parse.urlencode(result, doseq=True)
    else:
        return data


def get_user_list(domain, token):
    url = f'{domain}/webapi/entry.cgi?api=SYNO.Chat.External&method=user_list&version=2&token={token}'
    res = http.request('GET', url)
    res_data = json.loads(res.data.decode('utf-8'))
    users = []
    for user in res_data['data']['users']:
        users.append({'id': user['user_id'], 'email': user['user_props']['email']})
    return users

def returns_user_ids(email, users):
    return [user['id'] for user in list(filter(lambda u: u['email'] in email, users))]


def send_message(domain, token, ids, text):
    url = f'{domain}/webapi/entry.cgi?api=SYNO.Chat.External&method=chatbot&version=2&token={token}'
    json_data = json.dumps({
        'text': text, 'user_ids': ids
    })

    data = encode_params({'payload':json_data})
    
    res = http.request(
        "POST",
        url,
        body=data,
        )

def lambda_handler(event, context):
    domain = "" Synology Chat Domain
    token = "" Synology Chat API Token
    result = event
    assign = result['assignee']
    users = get_user_list(domain, token)
    assign_ids = returns_user_ids([assign], users)
    assignee_text = ''
    
    
    
    if result['type'] == 'workflow':
        assignee_text = f'*({result["key"]}) 이슈가 {result["status"]} 로 이동되었습니다* \n ({result["key"]}) - {result["summary"]} \n {result["url"]}'
        if assign != result['reporter']:
            reporter_ids = returns_user_ids([result['reporter']], users)
            send_message(domain, token, reporter_ids, assignee_text)
    elif result['type'] == 'comment':
        if assign != result['author']:
            assignee_text = f'*({result["key"]}) 이슈에 {result["displayName"]}님이 댓글을 작성하였습니다.* \n {result["comment"]} \n {result["url"]}'
    elif result['type'] == 'assign':
        assignee_text = f'*({result["key"]}) 이슈가 할당되었습니다.* \n {result["summary"]} \n {result["url"]}'
    elif result['type'] == 'create':
        assignee_text = f'*({result["key"]}) 이슈가 할당되었습니다.* \n {result["summary"]} \n {result["url"]}'

    if assignee_text:
        send_message(domain, token, assign_ids, assignee_text)

    return {'status':'good'}

 

후기

만약에 본인이 Synology Chat 을 통해서 이와 같이 개발한다고 말하면 나는 만류하고싶다.

이유는 크게 두가지인데 첫번째는 Synology Chat 의 경우 불안정하고 두번째는 API 가 불친절하다.

 

첫번째의 경우에는 Synology Chat 이 요청을 받고 메시지를 전송하는데 1분이 넘는 시간이 소요된다는 것이다.

그러다보니 실시간으로 받는 듯한 느낌이 덜하고 옆에서 이슈등록되었다고 말하는 것이 더 빠르다 그냥 메모용으로 전락할 가능성이 크다.

 

두번째의 경우에는 Python API 요청을 하면서 느꼈는데 메시지가 너무 불친절하게 나오고 스펙이 없으니

에러의 포인트를 잡는 것부터 난관이였다.

그러다보니 처음에는 무조건 되는 Shell script 기반부터 시작해서 requests 라이브러리 등 여러가지를 비교한 다음에 문제를 파악하고

거기에 맞는 스펙을 유저가 알아서 맞춰서 보냈다.

 

그러다보니 만약에 Synology Chat으로 한다면 나는 그들의 환경이 Synology Chat 가 밀접한 관련이 없다면 그냥 Slack을 쓰라고 권유를 하고싶다.

 

참고

Jira 웹 요청 설정은 아래와 같았다.

{
    "type" : "create",
    "assignee": "{{assignee.emailAddress}}",
    "summary": "{{issue.summary}}",
    "key" : "{{issue.key}}",
    "project_name": "{{issue.project.name}}",
    "url": "{{issue.url}}"
}