Notify Azure Cost from AWS Lambd with python

notify_azure_cost.py

import json
import os
from datetime import datetime

import requests

SLACK_WEBHOOK_URL = os.environ['SLACK_WEBHOOK_URL']
ENV = os.environ['ENV']
SUBSCRIPTION_ID = os.environ["SUBSCRIPTION_ID"]
TENANT = os.environ["TENANT"]
APP_ID = os.environ["APP_ID"]
PASSWORD = os.environ["PASSWORD"]


def lambda_handler(event, context) -> None:
    access_token = generate_access_token()
    billing = get_billing(access_token)

    (title, detail) = generate_message(billing)
    print(title, detail)

    post_slack(title, detail, "Azure Cost", ":microsoft:")


def generate_access_token() -> str:
    url = f"https://login.microsoftonline.com/{TENANT}/oauth2/token"
    payload = {
        "grant_type": "client_credentials",
        "resource": "https://management.core.windows.net/",
        "client_id": APP_ID,
        "client_secret": PASSWORD
    }
    print(url, payload)
    response = requests.post(url, data=payload)
    print(response.status_code)
    print(response.json())
    return response.json()['access_token']


def get_billing(token) -> dict:
    """
    Call Azure API to get billing information
    https://docs.microsoft.com/ja-jp/rest/api/cost-management/query/usage
    """

    url = f"https://management.azure.com/subscriptions/{SUBSCRIPTION_ID}/providers/Microsoft.CostManagement/query?api-version=2019-11-01"
    headers = {
        "Authorization": f"Bearer {token}",
        "content-type": "application/json"
    }
    payload = {
        "dataset": {
            "aggregation": {
                "totalCost": {
                    "function": "Sum",
                    "name": "PreTaxCost"
                }
            },
            "granularity": "None",
            "grouping": [
                {
                    "name": "ResourceType",
                    "type": "Dimension"
                }
            ]
        },
        "timeframe": "BillingMonthToDate",
        "type": "Usage"
    }
    response = requests.post(url, data=json.dumps(payload), headers=headers)
    print(response.status_code)
    print(response.json())
    return response.json()


def generate_message(billing: dict) -> (str, str):
    dt_now = datetime.now()
    start = f"{dt_now.month:02d}/01"
    end = f"{dt_now.month:02d}/{dt_now.day:02d}"

    details = []
    total = 0
    currency = ""
    for item in billing["properties"]["rows"]:
        service_bill = item[0]
        service_name = item[1]
        currency = item[2]
        details.append(f'　・{service_name}: {service_bill:.2f} {currency}')

        total += float(service_bill)

    title = f'【{ENV}】 {start}-{end} {round(total, 2):.2f} {currency}'
    return title, '\n'.join(details)


def post_slack(title: str, detail: str, username: str, icon_emoji: str) -> None:
    payload = {
        "username": username,
        "icon_emoji": icon_emoji,
        'attachments': [
            {
                'color': '#36a64f',
                'pretext': title,
                'text': detail
            }
        ]
    }

    try:
        response = requests.post(SLACK_WEBHOOK_URL, data=json.dumps(payload))
    except requests.exceptions.RequestException as e:
        print(e)
    else:
        print(response.status_code)


if __name__ == '__main__':
    lambda_handler("", "")

notify_azure_cost_function.tf

module "lambda_function" {
  source = "terraform-aws-modules/lambda/aws"

  function_name = "notify-azure-cost-function"
  description   = "notify azure cost every day"
  handler       = "notify_azure_cost_function.lambda_handler"
  runtime       = "python3.9"
  timeout       = 300

  source_path = [
    "../src/notify_azure_cost_function.py",
    {
      path             = "../src",
      pip_requirements = true,
      pip_tmp_dir      = "${path.cwd}/tmp",
    }
  ]

  environment_variables = {
    SLACK_WEBHOOK_URL = var.slack_webhook_url
    ENV               = var.env
    SUBSCRIPTION_ID   = var.subscription_id
    TENANT            = var.tenant
    APP_ID            = var.app_id
    PASSWORD          = var.password
  }

  tags = {
    Name = "notify-azure-cost-function"
  }
}


resource "aws_lambda_function_event_invoke_config" "notify_azure_cost_function_invoke_config" {
  function_name          = module.lambda_function.lambda_function_name
  maximum_retry_attempts = 2
}

resource "aws_cloudwatch_event_rule" "notify_azure_cost_function" {
  name                = "notify-azure-cost-function-schedule"
  schedule_expression = "cron(55 23 ? * * *)"
  description         = "Notify azure cost at 23:55(UTC) every day"
}


resource "aws_cloudwatch_event_target" "notify_azure_cost_function" {
  rule  = aws_cloudwatch_event_rule.notify_azure_cost_function.name
  arn   = module.lambda_function.lambda_function_arn
  input = ""
}

resource "aws_lambda_permission" "notify_azure_cost_function_invoke" {
  action        = "lambda:InvokeFunction"
  function_name = module.lambda_function.lambda_function_name
  principal     = "events.amazonaws.com"
  source_arn    = aws_cloudwatch_event_rule.notify_azure_cost_function.arn
}

You can generate Azure App like below;

$ az ad sp create-for-rbac --name slack-cost --role "Billing Reader"
Creating 'Billing Reader' role assignment under scope '/subscriptions/06dd82e8-93c3-4f27-b599-adc661e5075e'
The output includes credentials that you must protect. Be sure that you do not include these credentials in your code or check the credentials into your source control. For more information, see https://aka.ms/azadsp-cli
'name' property in the output is deprecated and will be removed in the future. Use 'appId' instead.

{
  "appId": <YOUR_ APP_ID >,
  "displayName": "slack-cost",
  "name": <YOUR_APP_ID >,
  "password": <YOUR_PASSWORD>,
  "tenant": <YOUR_TENANT>
}

$ az account subscription list -o tsv --query '[*][@.displayName,@.subscriptionId]'
Command group 'account subscription' is experimental and under development. Reference and support levels: https://aka.ms/CLI_refstatus
Azure subscription 1	<YOUR_ SUBSCRIPTION_ID >