NCP 비용 알람 받기

NCP에서 비용 알람을 받기 위해서 Cloud Functions 서비스의 Action과 Trigger를 활용합니다.

가장 먼저 Trigger를 생성해줄거에요. 평일 오전 10시마다 알람을 발송해주기를 원하므로 Cron 타입으로 트리거를 등록해줄거에요.

다음으로 액션을 생성해줄건데 트리거 종류엔 위에서 만든 cron으로 선택하여 액션 생성 화면으로 이동해줍니다.

런타임은 python 3.11 버전으로 선택하고 타입은 일반 액션으로 선택해요, 저는 NCP 내 서비스별로 사용한 내역과 전체 비용이 얼마나왔는지 조회하고 싶어서 아래처럼 코드를 작성했어요.

import hashlib
import hmac
import base64
import requests
import time
import json
import datetime


def tstamp():
    ts = int(time.time() * 1000)
    ts = str(ts)

    return ts


def lookup_period():
    now = datetime.datetime.now()
    period = now.strftime("%Y%m")
    period = str(period)

    return period


def create_api_signature(args, url):
    timestamp = tstamp()
    per = lookup_period()

    access_key = args.get('NCLOUD_ACCESS_KEY')
    secret_key = args.get('NCLOUD_SECRET_KEY')
    secret_key = bytes(secret_key, 'UTF-8')

    method = "GET"

    uri = url + "?startMonth=" + \
        per + "&endMonth=" + per + "&responseFormatType=json"

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

    http_header = {
        'x-ncp-apigw-signature-v2': signingKey,
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key,
        "Content-Type": "application/json",
    }

    response = requests.get(
        args.get('BILLING_API_URL') + uri, headers=http_header)
    return json.loads(response.text)


"""
NCP 클라우드 총 사용량을 조회합니다.
https://api.ncloud-docs.com/docs/platform-costandusage-getdemandcostlist
"""


def getTotalCost(args):
    data = create_api_signature(args, args.get('GET_DEMAND_COST_LIST_URL'))

    used_total_amount = 0
    if len(data['getDemandCostListResponse']['demandCostList']) > 0:
        used_total_amount = data['getDemandCostListResponse']['demandCostList'][0]['useAmount']

    today = datetime.datetime.today()
    result = "[ NCP 일일 운용 비용 알람 ]\n" \
        + today.strftime("- 기준일 : %Y년 %m월 %d일") + "\n" \
        + "- 총 비용(원) : " + str(f"{used_total_amount:,}") + "원 (월 누적)\n" \
        + "\n[ 서비스별 사용 내역 ]\n"

    return result


"""
계약 사용량 리스트를 조회하는 API를 호출하여 현재 사용중인 상품들의 코드와 코드명을 가져옵니다.
https://api.ncloud-docs.com/docs/platform-costandusage-getcontractusagelist
"""


def getContractUsageList(args):
    data = create_api_signature(args, args.get('GET_CONTRACT_USAGE_LIST_URL'))

    product_set = set()
    for contract in data['getContractUsageListResponse']['contractList']:
        for product in contract['contractProductList']:
            product_item = product['productItemKind']
            product_set.add((product_item['code'], product_item['codeName']))

    return product_set


"""
상품 청구 비용 리스트를 조회하는 API를 호출하여 상세한 사용 내역을 가져옵니다.
https://api.ncloud-docs.com/docs/platform-costandusage-getproductdemandcostlist
"""


def getTotalCostInDetail(args, prouct_set):
    timestamp = tstamp()
    per = lookup_period()

    access_key = args.get('NCLOUD_ACCESS_KEY')
    secret_key = args.get('NCLOUD_SECRET_KEY')
    secret_key = bytes(secret_key, 'UTF-8')

    method = "GET"

    data_list = []
    for product in prouct_set:
        code = product[0]
        code_name = product[1]
        uri = args.get('GET_PRODUCT_DEMAND_COST_LIST_URL') + "?startMonth=" + \
            per + "&endMonth=" + per + "&responseFormatType=json" + \
            "&productDemandTypeCode=" + code

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

        http_header = {
            'x-ncp-apigw-signature-v2': signingKey,
            'x-ncp-apigw-timestamp': timestamp,
            'x-ncp-iam-access-key': access_key,
            "Content-Type": "application/json",
        }

        response = requests.get(
            args.get('BILLING_API_URL') + uri, headers=http_header)
        data = json.loads(response.text)
        if data['getProductDemandCostListResponse']['totalRows'] > 0:
            data_list.append(
                (code_name, data['getProductDemandCostListResponse']['productDemandCostList']))

    svc_total_cost = 0
    message = ""
    for code_name, data in data_list:
        for cost in data:
            svc_total_cost += cost['useAmount']
        message += f"- {code_name} / {svc_total_cost:,}원\n"
        svc_total_cost = 0

    return message


def sendSlackMessage(message, args):
    slack_webhook = args.get('SLACK_WEBHOOK')

    try:
        payload = {
            "text": message,
        }
        headers = {'Content-type': 'application/json'}
        requests.post(
            slack_webhook, data=json.dumps(payload), headers=headers)

        return {"done": True}

    except Exeption as e:
      print(f'Error sending message: {e}')
      return {"done": False, "error_message": str(e)}


def main(args):
    message = getTotalCost(args)
    product_set = getContractUsageList(args)
    message += getTotalCostInDetail(args, product_set)
    return sendSlackMessage(message, args)

코드를 모두 작성하고 난 뒤에는 디폴트 파라미터에 크레덴셜 값을 넣어주어야 하는데 JSON 형식으로 넣어주면 되고 NCLOUD_ACCESS_KEY, NCLOUD_SECRET_KEY, SLACK_WEBHOOK 환경변수만 암호화를 설정했어요. 모든 값을 붙여넣기 한 이후에 암호화를 설정하면 설정한 파라미터 키에 한하여 암호화가 진행되며 사전에 NCP KMS 서비스를 활성화하여 키를 발급받아야 할 수 있어요.

{
  "BILLING_API_URL": "https://billingapi.apigw.ntruss.com",
  "GET_PRODUCT_DEMAND_COST_LIST_URL": "/billing/v1/cost/getProductDemandCostList",
  "GET_CONTRACT_USAGE_LIST_URL": "/billing/v1/cost/getContractUsageList",
  "NCLOUD_ACCESS_KEY": "",
  "NCLOUD_SECRET_KEY": "",
  "GET_DEMAND_COST_LIST_URL": "/billing/v1/cost/getDemandCostList",
  "SLACK_WEBHOOK": ""
}

다음으로 VPC 연결 정보를 설정하는데 VPC 연결 정보에 들어갈 서브넷은 항상 Private Subnet이어야 하므로 이점 유의해주세요! 생성이 완료되었다면 실행을 눌러 슬랙에 정상적으로 알람이 발송되는지 확인해주세요.

정상적으로 수행되는 모습입니다.