蛇ノ目の記

技術のことも。そうでないことも。

Lambda Labs 小ネタ - インスタンスの空き状況を通知するスクリプトを書いた

AI Python

前回に引き続いてLambda Labsネタです。

nao-y.hatenablog.com

短所として

GPUインスタンスのキャバ不足が常態化している

と書いたわけですが、常にキャパ不足だからこそ空きができたときは秒でどこかの誰かに使われてしまうという面もあります。

そこで適当なインスタンス(ここでは社の常に稼働している環境)で動かしていて、Lambda Labs APIでインスタンスの空き状況をチェック・WenhookでSlackに通知するスクリプトを書いてみた、というだけのネタです。定期実行はsystemd.timerです。

こんな感じで通知してくれます。

Capacity available instances:
1x H100 (80 GB PCIe), $2.49/hr
us-west-3(Utah, USA)
1x A10 (24 GB PCIe), $0.75/hr
us-east-1(Virginia, USA)
us-west-1(California, USA)
1x A100 (40 GB SXM4), $1.29/hr
us-west-2(Arizona, USA)

APIのドキュメントはこちら。

https://cloud.lambdalabs.com/api/v1/docs

Pythonまわり

main.py

import json

import requests

LAMBDA_LABS_API_BASE = "https://cloud.lambdalabs.com/api/v1"
LAMBDA_LABS_API_KEY = (
    "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
)
WEBHOOK_URL = "https://hooks.slack.com/services/xxxxxxxx/yyyyyyyyyy/zzzzzzzzzz"


def notify_to_slack(message: str) -> None:
    message = f"Capacity available instances:\n{message}"
    payload = {
        "text": message,
        "username": "Instance Capacity Information",
        "icon_emoji": "information_source",
    }
    requests.post(
        WEBHOOK_URL,
        data=json.dumps(payload),
        headers={"Content-Type": "application/json"},
    )


def fetch_available_instances() -> str:
    message = ""
    endpoint = f"{LAMBDA_LABS_API_BASE}/instance-types"
    headers = {
        "Authorization": f"Bearer {LAMBDA_LABS_API_KEY}",
        "Content-Type": "application/json",
    }
    result = requests.get(endpoint, headers=headers).json()
    for instance_name, instance_type in result["data"].items():
        instance_description = instance_type["instance_type"]["description"]
        instance_price = instance_type["instance_type"]["price_cents_per_hour"] / 100
        available_regions = instance_type["regions_with_capacity_available"]
        if available_regions:
            message += f"{instance_description}, ${instance_price:.2f}/hr\n"
            for region in  available_regions:
                name = region["name"]
                description = region["description"]
                message += f"{name}({description})\n"
    return message


if __name__ == "__main__":
    message = fetch_available_instances()
    if message:
        notify_to_slack(message)

requirements.txt

requests==2.31.0

Dockerまわり

Dockerfile

FROM python:3.12.1-slim
COPY requirements.txt /workspace
COPY main.py /workspace

docker-compose.yml

version: "3"
services:
  lambdalabs-instance-watcher:
    build:
      dockerfile: Dockerfile
    container_name: lambdalabs-instance-watcher
    working_dir: /workspace
    command: >
        /bin/sh -c 'python main.py'

systemdまわり

home/ec2-user/Script/instance-watcher.sh

cd /home/ec2-user/Script/lambdalabs-instance-watcher-main
docker compose build
docker compose up -d

/etc/systemd/system/instance-watcher.service

[Unit]
Description=Notify available Lambda Labs instance.

[Service]
User=ec2-user
Type=oneshot
ExecStart=/bin/sh /home/ec2-user/Script/instance-watcher.sh

[Install]
WantedBy=multi-user.target

/etc/systemd/system/instance-watcher.timer

[Timer]
OnCalendar=*-*-* *:00/30:00
Unit=instance-watcher.service

30分に一度実行する設定。

感想

通知スクリプトを動かしてから実感したことですが、土日はVRAM多めのインスタンスが割と空きやすいですね。 逆に平日は1x A10 (24 GB PCIe)のみということがほとんどです。だからこそVRAM多めのやつが空いたときにすぐに気づけて便利。 みなさんも通知スクリプトでより快適なLambda Labs生活を。