Browsing Tag

API

NCLOUD

[NCLOUD] SourceDeploy로 배포하고 성공 및 실패 알림을 받아보자

안녕하세요 MANVSCLOUD 김수현입니다.

네이버 클라우드에서 Developer Tools 카테고리에 있는 서비스들을 이용하여 CI/CD 파이프라인을 효과적으로 구축할 수 있습니다. 특히 Source Deploy 서비스는 자동화 배포 과정에 큰 도움이 됩니다. 오늘은 Source Deploy를 사용할 때 배포가 성공적으로 완료되었는지 혹은 실패했는지에 대한 알림을 받는 방법에 대해 알아보겠습니다.


Developer Tools, 배포 후 성공/실패 알람이 가능할까?

이 포스팅을 작성하게 된 계기는 네이버 클라우드 플랫폼 유저 커뮤니티 (NCUC)에서 제기된 특정 요구사항 때문입니다. 사용자들이 배포 후 성공 또는 실패에 대한 알림 기능의 필요성을 토론하였습니다.


아직 SourceDeploy 서비스는 직접적으로 이러한 알림 기능을 제공하지 않고 있지만 네이버 클라우드는 API를 통해 이를 가능하게 하는 방법이 있기에 이 글을 통해 배포 결과에 대한 알림을 받는 방법에 대해 설명해보도록 하겠습니다.


SourceDeploy API + Cloud Functions + Slack

방법은 다양하게 생각할 수 있겠지만 저는 Cloud Functions을 이용하여 Slack으로 알림을 받을 수 있도록 구축해보았습니다.

프로젝트는 여러 개일 수 있으니 SourceDeploy 프로젝트 목록을 조회하는 API로 모든 프로젝트의 ID를 가져오고 모든 프로젝트ID에 대한 배포 이력 목록을 조회하여 최근 5분 이내에 실행된 배포에 대한 결과 값을 가져오는 코드입니다.

  • Cloud Functions Action 소스코드 (런타임 : Python 3.11)
import requests
import time
import hashlib
import hmac
import base64
import json
from datetime import datetime, timedelta

def generate_signature(secret_key, method, uri, timestamp, access_key):
    message = method + " " + uri + "\n" + timestamp + "\n" + access_key
    message = bytes(message, 'UTF-8')
    return base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())

def get_projects(api_server, uri, http_header):
    response = requests.get(api_server + uri, headers=http_header)
    return response.json()

def get_project_history(api_server, uri, http_header):
    response = requests.get(api_server + uri, headers=http_header)
    return response.json()

def send_to_slack(webhook_url, message):
    payload = {'text': message}
    requests.post(webhook_url, json=payload)

def format_message(project, history):
    timestamp = datetime.fromtimestamp(history['startTime'] / 1000)
    formatted_time = timestamp.strftime("%Y-%m-%d %H:%M (UTC%z)")
    return (f"프로젝트 이름 : {project['name']}\n"
            f"시나리오 이름 : {history['scenario']['name']}\n"
            f"스테이지 : {history['stage']['name']}\n"
            f"최종 실행시간 : {formatted_time}\n"
            f"상태 : {history['status']}")

def main(args):
    try:
        timestamp = str(int(time.time() * 1000))

        access_key = args["access_key"]
        secret_key = bytes(args["secret_key"], 'UTF-8')
        slack_webhook_url = args["slack_webhook_url"]

        method = "GET"
        api_server = "https://vpcsourcedeploy.apigw.ntruss.com"
        project_uri = "/api/v1/project"

        signingKey = generate_signature(secret_key, method, project_uri, timestamp, access_key)

        http_header = {
            'x-ncp-apigw-signature-v2': signingKey,
            'x-ncp-apigw-timestamp': timestamp,
            'x-ncp-iam-access-key': access_key
        }

        projects = get_projects(api_server, project_uri, http_header)
        messages = []

        for project in projects['result']['projectList']:
            project_id = project['id']
            history_uri = f"/api/v1/project/{project_id}/history"

            signingKey = generate_signature(secret_key, method, history_uri, timestamp, access_key)
            http_header['x-ncp-apigw-signature-v2'] = signingKey

            history_data = get_project_history(api_server, history_uri, http_header)
            current_time = datetime.now()

            for history in history_data['result']['historyList']:
                history_time = datetime.fromtimestamp(history['startTime'] / 1000)
                if current_time - history_time < timedelta(minutes=5):
                    slack_message = format_message(project, history)
                    send_to_slack(slack_webhook_url, slack_message)
                    messages.append(slack_message)

        return {"status": "success", "messages": messages}

    except Exception as e:
        return {"status": "error", "error": str(e)}
  • 디폴트 파라미터
{"access_key":"[ACCESS_KEY]","secret_key":"[SECRET_KEY]","slack_webhook_url":"[WEBHOOK_URL]"}
  • ACCESS_KEY : SourceDeploy의 배포 이력 목록 조회 권한이 있는 Sub Account의 Access Key
  • SECRET_KEY : SourceDeploy의 배포 이력 목록 조회 권한이 있는 Sub Account의 Secret Key
  • WEBHOOK_URL : Slack Webhook URL

format_message 함수에서 Slack에 전달할 폼을 변경할 수 있으며 최근 5분 내에 실행된 배포가 아니라 원하는 시간대로 변경하고 싶다면 minutes=5 부분을 코드에서 변경할 수 있습니다.

Cloud Functions 생성 후 배포를 일부러 실패할 수 있도록 구성한 뒤 배포를 진행하였습니다.

배포가 끝나고 Cloud Functions을 실행해보면 위와 같이 어떤 프로젝트에서 어떤 시나리오의 이름으로 언제 실패했고 성공했는지 알림을 받을 수 있습니다.

우리는 배포 때마다 Cloud Functions을 수동으로 실행해줄 수 없습니다.
따라서 Cloud Functions 실행을 자동화하기 위해 Trigger를 추가해주어야 합니다.

여러가지 Trigger 종류를 제공하고 있기에 방법은 다양할 것입니다.
cron을 이용하여 주기적으로 배포 여부를 체크할 수도 있을 것이고 저는 Cloud Insight를 좋아하니 Cloud Insight API 중 SendData로 배포 이력을 수집하고 배포 이력이 발생했을 때 모니터링 알림과 함께 Slack으로 내용을 보도록 해볼 것같습니다.


Personal Comments

네이버 클라우드에서는 다양한 서비스들을 제공하고 있지만 가끔 원하는 기능이 없을 때도 있습니다. 하지만 이 경우 네이버 클라우드에서 제공하는 API를 이용하여 필요한 기능을 만들어서 사용할 수도 있습니다.

긴 글 읽어주셔서 감사합니다.