关于Oracle Cloud降配的问题

mlzl1426

各位道友，简单聊下Oracle云免费配额降配这件事。


Oracle在6月中旬官宣将A1免费实例从4C24G缩减至2C12G，我这台日本4C24G节点是今年7月才新建的。最开始先部署了一台1C1G纯免费实例，主要用来预留公网IP，闲置也先保留；后续将账户升级为PAYG按量计费模式，扩容搭建4C24G主力节点，当前节点承载Hermes Agent、Ollama、1Panel、Docker容器集群等全套服务。


我其实有点后知后觉，直到8月5日才留意到配额下调公告，第一时间登录控制台核验：两台实例均未产生计费账单。此前我已配置两层官方账单告警，预测费用告警、实际扣费告警阈值均设置1SGD(1新加坡元)，每条告警分别绑定3个独立邮箱接收。


考虑到本人邮件查看频次较低，常常3-4天才查阅一次，邮件告警存在漏看风险。于是直接在节点Hermes Agent对接飞书消息渠道，在 Oracle控制台导出API密钥、指纹、租户OCID、用户OCID完成鉴权配置，自定义告警策略：
每日0点、12点两轮定时轮询账单状态，无异常静默不推送；
每月1日、15日推送月度账单汇总报表；
一旦识别到扣费异常，立刻连续推送3条紧急告警消息。


下方为Agent落地配置代码：
#!/usr/bin/env python3
"""
Oracle Cloud 费用监控示例 — 脱敏公开版
========================================
背景: Oracle Cloud Always Free的Ampere A1配额在2026-06静默减半(4OCPU/24GB→2OCPU/12GB)。
      本文演示如何用官方Usage/Cost API做「定时费用监控+异常飞书告警」,
      并强调「不可绕过的规则级调度」=systemd timer(权威)+cron(兜底)双保险。


用法(三个子命令, 一次性执行, 由调度器触发):
    monitor_cost.py check     # 例行检查: 有异常连发3条告警, 无异常静默
    monitor_cost.py report    # 月度平安汇总(每月1/15号调用)
    monitor_cost.py status    # 仅显示当前费用(调试用, 不通知)


前置:
    pip install oci
    在 OCI 控制台生成API密钥对, 配置后可读取Usage/Cost。
"""


import sys
import json
import time
import urllib.request
from datetime import datetime, timedelta, timezone


import oci
from oci.usage_api.models import RequestSummarizedUsagesDetails


# ================== 配置区(请替换为真实值) ==================
OCI_CONFIG = {
    # 在OCI控制台「个人资料 → API密钥」生成
    "user": "ocid1.user.oc1..<User OCID>",
    "key_file": "/path/to/your/oci_api_key.pem",       # 私钥文件路径(权限600)
    "fingerprint": "aa:bb:cc:dd:ee:ff:00:<密钥指纹>",
    "tenancy": "ocid1.tenancy.oc1..<Tenancy OCID>",
    "region": "ap..",                             # 你租户的home region
}
TENANCY = OCI_CONFIG["tenancy"]


# --- 通知webhook(通用: 飞书/钉钉/Telegram 机器人自定义机器人webhook皆可) ---
NOTIFY_WEBHOOK = "https://your-msg-platform-webhook.example/bot/hook/<token>"


COST_THRESHOLD = 0.01      # 累计费用超过此值(SGD)视为异常
STATE_FILE = "/path/to/monitor_state.json"


# ================== 通知 ==================
def send_notify(message: str) -> bool:
    """通过自定义 webhook 发消息(通用, 不依赖 Hermes/飞书环境)
    用 urllib 而非 requests, 减少依赖; 关键: 加超时防挂起。"""
    payload = {"msg_type": "text", "content": {"text": message}}
    try:
        req = urllib.request.Request(
            NOTIFY_WEBHOOK,
            data=json.dumps(payload).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=15) as resp:
            return resp.status == 200
    except Exception as e:
        print(f"[通知失败] {e}", flush=True)
        return False


def send_alert_triple(total, currency, services):
    """异常: 连发3条警告(带递增编号)"""
    detail = "、".join(f"{k}={v:g}" for k, v in services.items() if v > 0) or "无明细"
    msg = (
        f"🚨 Oracle 计费异常警告\n"
        f"本月当前费用: {total:.4f} {currency}\n"
        f"服务明细: {detail}\n"
        f"请登录 Oracle 控制台核查(可能已对超配部分计费)。"
    )
    for i in range(3):
        ok = send_notify(f"[严重警告 {i+1}/3]\n{msg}")
        if not ok:
            print(f"  第{i+1}条发送失败", flush=True)
        time.sleep(2)


# ================== OCI 费用查询 ==================
def query_cost() -> dict:
    """查询当月累计费用, 返回 {total, currency, services}"""
    now_utc = datetime.now(timezone.utc)
    # 注意: MONTHLY granularity 要求时间戳精确到日(时分秒必须为0), 否则报 InvalidParameter
    start = now_utc.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    end = (now_utc + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)


    client = oci.usage_api.UsageapiClient(OCI_CONFIG)
    req = RequestSummarizedUsagesDetails(
        tenant_id=TENANCY,
        time_usage_started=start.isoformat(),
        time_usage_ended=end.isoformat(),
        granularity="MONTHLY",
        is_aggregate_by_time=False,
        group_by=["service"],
        query_type="COST",
    )
    resp = client.request_summarized_usages(req)


    total, currency, services = 0.0, "SGD", {}
    if resp.data.items:
        for item in resp.data.items:
            cost = float(item.computed_amount or 0)
            service = getattr(item, "service", "unknown")
            if getattr(item, "currency", None):
                currency = item.currency
            services[service] = round(cost, 4)
            total += cost
    return {"total": round(total, 4), "currency": currency, "services": services}


# ================== 状态持久化 (避免重复轰炸) ==================
def load_state() -> dict:
    try:
        with open(STATE_FILE) as f:
            return json.load(f)
    except Exception:
        return {"last_alert_total": 0.0, "last_report_day": ""}


def save_state(state: dict):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f)


# ================== 三个动作 ==================
def check_and_alert():
    """例行检查: 异常连发3条, 无异常静默"""
    stamp = (datetime.now(timezone.utc) + timedelta(hours=8)).strftime("%Y-%m-%d %H:%M")
    print(f"[{stamp}] 例行费用检查...", flush=True)
    state = load_state()
    result = query_cost()
    total, currency = result["total"], result["currency"]
    print(f"  当月累计: {total:.4f} {currency}", flush=True)


    if total > COST_THRESHOLD:
        # 仅当金额相对上次告警上涨(>0.005)或首次告警才连发3条, 避免重复轰炸
        last = state.get("last_alert_total", 0.0)
        if total > last + 0.005 or last == 0.0:
            print("  ⚠️ 异常! 连发3条告警", flush=True)
            send_alert_triple(total, currency, result["services"])
            state["last_alert_total"] = total
        else:
            print("  已告警过, 仅发1条提醒", flush=True)
            send_notify(f"⚠️ Oracle费用仍超阈值: {total:.4f} {currency}")
    else:
        print("  正常, 无异常(静默)", flush=True)
    save_state(state)
    return 0


def monthly_report():
    """月度平安汇总(每月1/15号调用)"""
    stamp = (datetime.now(timezone.utc) + timedelta(hours=8)).strftime("%Y-%m-%d")
    print(f"[{stamp}] 月度平安汇总...", flush=True)
    result = query_cost()
    total, currency = result["total"], result["currency"]
    state = load_state()
    state["last_report_day"] = stamp
    if total <= COST_THRESHOLD:
        print("  无异常, 发送平安汇报", flush=True)
        send_notify(
            f"📊 Oracle 费用月度汇报({stamp})\n"
            f"本月累计: {total:.4f} {currency}\n✅ 一切正常, 无异常。"
        )
    else:
        print("  有异常! 连发3条告警", flush=True)
        send_alert_triple(total, currency, result["services"])
        state["last_alert_total"] = total
    save_state(state)
    return 0


def status():
    """显示当前费用(调试), 不通知"""
    result = query_cost()
    print(f"当月累计: {result['total']:.4f} {result['currency']}")
    print("服务明细:", json.dumps(result["services"], ensure_ascii=False))
    return 0


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "check"
    {"check": check_and_alert, "report": monthly_report, "status": status} \
        .get(cmd, lambda: (print("用法: monitor_cost.py [check|report|status]"), 2)[1])()




systemd timer（权威调度）
# /etc/systemd/system/oracle-cost-check.timer  —— 每日12:00 & 00:00 北京
[Timer]
OnCalendar=*-*-* 12:00:00 Asia/Shanghai
OnCalendar=*-*-* 00:00:00 Asia/Shanghai
Persistent=true            # 关机漏跑, 开机自动补
Unit=oracle-cost-check.service




# /etc/systemd/system/oracle-cost-report.timer  —— 每月1/15号 12:00 北京
[Timer]
OnCalendar=*-*-01 12:00:00 Asia/Shanghai
OnCalendar=*-*-15 12:00:00 Asia/Shanghai
Persistent=true
Unit=oracle-cost-report.service




cron 兜底（双保险）
# 每日北京12:00(=UTC04:00) & 00:00(=UTC16:00) 例行
0 4,16 * * * /path/venv/bin/python /path/monitor_cost.py check
# 每月1/15号 北京12:00(=UTC04:00) 月度汇报
0 4 1,15 * *  /path/venv/bin/python /path/monitor_cost.py report




至此搭建三层费用防护机制：
① Oracle 官方预测费用邮件告警
② Oracle 官方实际扣费邮件告警
③ Hermes Agent 飞书实时轮询告警
所以我暂时不打算手动降配，如果oracle真开始收费了，那我也会第一时间收到通知，然后手动降配。