logo NodeSeekbeta
search for post search for people use google search

阿里 抢占式ECS实例 CDT，分享费用数据&TG通知脚本&保活脚本增强版

MJJYYDS
MJJYYDS楼主
edited 228days ago in 测评
#0

感谢大佬教程 https://www.nodeseek.com/post-572284-1

魔都移动+CDT，直连晚高峰油管：

image

image

在国内站aliyun.com和国际站alibabacloud.com分别开了最低配2C512M1G硬盘的ecs.e-c4m1.large
---注意香港区域D比BC贵一点

国内站开EIP还是共享带宽的时候会弹出来让充值100元，国际站绑了美区PP随用随付
---美区PP记得在付款-自动付款-ali***-改成支付卡换汇

---国内站开共享带宽的时候弹错后联系客服提交手持后小姐姐打回电告知解除后开通，国内站共享带宽可以拉2G，国际站共享带宽只能拉到1G

实测：账单详情里，节省停机后，扣费项目只有【系统盘，1G盘费用极低一个月仅仅6-7毛】,实例费，EIP，操作系统都不扣费

所以可以实现两只CDT小鸡交替12小时开机12小时节省停机，实现两台合计年付56元，流量400G单向/月

定时节省停机可以通过OOS系统运维管理-常用运维任务-定时开关机实现，然后保活脚本cron定时在开机期间运行即可

为什么这么折腾，因为无聊缺🐔玩，因为那100退不了，因为想用用好久没用的美区PP xhj001

以下：把两台的账单详情导入AI得出的对比表，有需要的参考下，没需要的看个乐 xhj009

image

image

image

AI还是挺好玩的，简单折腾了个TG通知脚本 xhj001

image

TG通知脚本，自行填写相关字段，推荐设置23:59运行。
（为符合大部分人的需求，已修改为单实例专用）

保存：/root/cdt_daily_report.sh
加权：chmod +x /root/cdt_daily_report.sh
cron：

每天 23:59 自动执行 CDT 每日流量报告脚本(脚本已包含写入日志任务)

59 23 * * * /root/cdt_daily_report.sh
重载：systemctl restart cron

#!/bin/bash

# ============================================
# 使用说明
# ============================================
# 1. Telegram Bot 配置
#    - TG_BOT_TOKEN：你的 Telegram Bot Token
#    - TG_CHAT_ID：你的 Telegram Chat ID
#
# 2. 阿里云 ECS / CDT 配置
#    - ACCESS_KEY_ID / ACCESS_KEY_SECRET：AK/SK
#    - REGION_ID：实例所在地域 ID（如 cn-hongkong）
#    - ECS_INSTANCE_ID：实例的 ECS ID
#    - ECS_NAME：实例名称（自定义）
# ============================================

# ================================
# 1. 配置信息（请填写）
# ================================
TG_BOT_TOKEN="REDACTED_TG_BOT_TOKEN"
TG_CHAT_ID="REDACTED_CHAT_ID"

ACCESS_KEY_ID='REDACTED_AK'
ACCESS_KEY_SECRET='REDACTED_SK'
REGION_ID='cn-hongkong'
ECS_INSTANCE_ID='REDACTED_ECS_ID'
ECS_NAME='AliCloud-CDT'

# 流量阈值设置（GB），CDT 免费额度通常为 200GB
LIMIT=200

# ================================
# 发送 Telegram 消息函数
# ================================
send_tg() {
    curl -s -o /dev/null -X POST "https://api.telegram.org/bot${TG_BOT_TOKEN}/sendMessage" \
        -H "Content-Type: application/json" \
        -d "{\"chat_id\": \"${TG_CHAT_ID}\", \"text\": \"$1\", \"parse_mode\": \"HTML\"}"
}

# ================================
# Python 查询 CDT 流量
# ================================
query_cdt() {
python3 <<EOF
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
import json

client = AcsClient("$ACCESS_KEY_ID", "$ACCESS_KEY_SECRET", "$REGION_ID")

request = CommonRequest()
request.set_domain('cdt.aliyuncs.com')
request.set_version('2021-08-13')
request.set_action_name('ListCdtInternetTraffic')
request.set_method('POST')

try:
    response = client.do_action_with_exception(request)
    data = json.loads(response.decode('utf-8'))
    total_bytes = sum(d.get('Traffic', 0) for d in data.get('TrafficDetails', []))
    print(f"{total_bytes / (1024**3):.2f}")
except Exception as e:
    print("0.00")
EOF
}

# ================================
# Python 获取 ECS 公网 IP
# ================================
get_public_ip() {
python3 <<EOF
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
import json

client = AcsClient("$ACCESS_KEY_ID", "$ACCESS_KEY_SECRET", "$REGION_ID")

request = CommonRequest()
request.set_domain('ecs.aliyuncs.com')
request.set_version('2014-05-26')
request.set_action_name('DescribeInstances')
request.set_method('POST')
request.add_query_param('InstanceIds', f'["$ECS_INSTANCE_ID"]')

try:
    response = client.do_action_with_exception(request)
    data = json.loads(response.decode('utf-8'))
    instance = data['Instances']['Instance'][0]
    eip = instance.get('EipAddress', {}).get('IpAddress')
    pub = instance.get('PublicIpAddress', {}).get('IpAddress', [])
    if eip:
        print(eip)
    elif pub:
        print(pub[0])
    else:
        print("无公网IP")
except:
    print("获取失败")
EOF
}

# ================================
# 执行查询与计算
# ================================
TRAFFIC=$(query_cdt)
PUBLIC_IP=$(get_public_ip)

# 计算剩余与百分比
REMAIN=$(awk "BEGIN {r=$LIMIT-$TRAFFIC; if(r<0) r=0; printf \"%.2f\", r}")
USAGE_PERCENT=$(awk "BEGIN {printf \"%.2f\", ($TRAFFIC/$LIMIT)*100}")

# 颜色等级判定
COLOR=$(awk 'BEGIN {p='$USAGE_PERCENT'; if(p<60) print "🟢 正常"; else if(p<90) print "🟡 注意"; else print "🔴 危险";}')

# ================================
# 生成消息内容
# ================================
MESSAGE="<b>📊 CDT 每日流量报告</b>

<b>🖥 ${ECS_NAME}</b>
┃ 📦 实例：${ECS_INSTANCE_ID}
┃ 🌐 公网 IP：${PUBLIC_IP}
┃ 📍 地域：${REGION_ID}
┃ 📈 已用：<b>${TRAFFIC} GB</b>
┃ 💾 剩余：<b>${REMAIN} GB</b>
┃ 🔥 使用率：<b>${USAGE_PERCENT}%</b>  ${COLOR}

⏰ <i>更新时间：$(date '+%Y-%m-%d %H:%M')</i>"

# 发送通知
send_tg "$MESSAGE"

# ================================
# 写入日志
# ================================
LOG_DIR="/var/log/cdt_daily_report"
mkdir -p $LOG_DIR

{
    echo "-------------------------------------------"
    echo "更新时间：$(date '+%Y-%m-%d %H:%M')"
    echo "${ECS_NAME}"
    echo "┃ 实例：${ECS_INSTANCE_ID}"
    echo "┃ 公网 IP：${PUBLIC_IP}"
    echo "┃ 地域：${REGION_ID}"
    echo "┃ 已用：${TRAFFIC} GB"
    echo "┃ 剩余：${REMAIN} GB"
    echo "┃ 使用率：${USAGE_PERCENT}%  $(echo ${COLOR} | sed 's/.* //')"
    echo "-------------------------------------------"
} >> "${LOG_DIR}/cdt_report_$(date '+%Y%m').log"

保活脚本增强版（设置节省停机模式）
（以下为单实例保活，双实例保活请复制双份，命名为不同名称.py,交替时间段运行）

保存：/root/alicdt.py
目录：mkdir -p /var/log/alicdt
加权：chmod +x /root/alicdt.py
cron:

alicdt 保活脚本（交替12小时运行，延迟2分钟运行给OOS自动运维开机的时间）

1. 每天 00:02 - 00:59 运行（每分钟执行）

2-59 0 * * * /usr/bin/python3 /root/alicdt.py >> /var/log/alicdt/alicdt_date +\%Y\%m\%d.log 2>&1

2. 每天 01:00 - 11:59 运行（每分钟执行）

  • 1-11 * * * /usr/bin/python3 /root/alicdt.py >> /var/log/alicdt/alicdt_date +\%Y\%m\%d.log 2>&1
    重载：systemctl restart cron
# -*- coding: utf-8 -*-
"""
MJJYYDS-阿里云 CDT 流量保活脚本（增强注释版）

功能说明：
    - 每分钟运行一次
    - 查询当月 CDT 流量
    - 流量低于阈值 → 自动启动 ECS
    - 流量超过阈值 → 自动停止 ECS（StopCharging 节省停机）
    - 日志输出 ECS 当前状态与停机模式
"""

from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyunsdkecs.request.v20140526 import StartInstancesRequest, StopInstancesRequest, DescribeInstancesRequest
import json
import sys
import logging

# ================== 1. 配置日志 ==================
# 输出格式：时间 - 级别 - 内容
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    stream=sys.stdout
)
logger = logging.getLogger(__name__)

# ================== 2. 配置阿里云凭证和 ECS 实例信息 ==================
# 注意：生产环境建议使用环境变量或 RAM 角色，不要写死 AK/SK
ACCESS_KEY_ID = 'LT***'                # ← 填写你的 AccessKey ID
ACCESS_KEY_SECRET = 'yY***'      # ← 填写你的 AccessKey Secret
REGION_ID = 'cn-hongkong'                                 # ← 填写 ECS 所在地域 ID（如 cn-hongkong）
ECS_INSTANCE_ID = 'i-j***'                # ← 填写你的 ECS 实例 ID

# 流量阈值（GB）
# 当月流量 >= 阈值 → 停机
# 当月流量 < 阈值 → 启动
TRAFFIC_THRESHOLD_GB = 188

# ================== 3. 初始化阿里云客户端 ==================
try:
    client = AcsClient(ACCESS_KEY_ID, ACCESS_KEY_SECRET, REGION_ID)
    logger.info("AcsClient initialized successfully.")
except Exception as e:
    logger.error(f"Failed to initialize AcsClient: {e}")
    sys.exit(1)

# ================== 4. 查询当前 CDT 总流量 ==================
def get_total_traffic_gb(client):
    """
    调用 CDT API：ListCdtInternetTraffic
    返回当月总流量（GB）
    """
    request = CommonRequest()
    request.set_domain('cdt.aliyuncs.com')
    request.set_version('2021-08-13')
    request.set_action_name('ListCdtInternetTraffic')
    request.set_method('POST')

    try:
        response = client.do_action_with_exception(request)
        response_json = json.loads(response.decode('utf-8'))

        # TrafficDetails 是一个数组，每个元素包含 Traffic 字段（字节数）
        total_bytes = sum(d.get('Traffic', 0) for d in response_json.get('TrafficDetails', []))
        total_gb = total_bytes / (1024 ** 3)

        logger.info(f"当前总互联网流量: {total_gb:.2f} GB")
        return total_gb

    except Exception as e:
        logger.error(f"获取 CDT 流量失败: {e}")
        sys.exit(1)

# ================== 5. 查询 ECS 状态 + 停机模式 ==================
def get_ecs_status(client, instance_id):
    """
    调用 DescribeInstances 查询 ECS 状态与停机模式
    返回：
        status: Running / Stopped / Stopping / Starting
        stopped_mode: StopCharging / KeepCharging / Not-applicable
    """
    try:
        request = DescribeInstancesRequest.DescribeInstancesRequest()
        request.set_InstanceIds([instance_id])
        response = client.do_action_with_exception(request)
        response_json = json.loads(response.decode('utf-8'))

        instances = response_json.get("Instances", {}).get("Instance", [])
        if not instances:
            logger.error("未找到该 ECS 实例信息。")
            return None, None

        instance = instances[0]
        status = instance.get("Status")
        stopped_mode = instance.get("StoppedMode")

        logger.info(f"ECS实例 {instance_id} 当前状态: {status}")

        # 仅在实例已停止时打印停机模式
        if status == "Stopped" and stopped_mode:
            if stopped_mode == "StopCharging":
                logger.info(f"ECS实例当前停机模式: {stopped_mode}（节省停机）")
            else:
                logger.info(f"ECS实例当前停机模式: {stopped_mode}")

        return status, stopped_mode

    except Exception as e:
        logger.error(f"获取 ECS 实例状态失败: {e}")
        return None, None

# ================== 6. 启动 ECS ==================
def ecs_start(client, instance_id):
    """
    启动 ECS 实例
    """
    status, _ = get_ecs_status(client, instance_id)

    # 已经运行 → 不重复启动
    if status == "Running":
        logger.info(f"ECS实例 {instance_id} 已经是运行状态，无需启动。")
        return

    try:
        request = StartInstancesRequest.StartInstancesRequest()
        request.set_InstanceIds([instance_id])
        request.set_accept_format('json')

        response = client.do_action_with_exception(request)
        logger.info(f"ECS启动响应: {response.decode('utf-8')}")

    except Exception as e:
        logger.error(f"启动 ECS 实例失败: {e}")

# ================== 7. 停止 ECS（节省停机） ==================
def ecs_stop(client, instance_id):
    """
    停止 ECS 实例，停机模式强制设置为 StopCharging（节省停机）
    """
    status, stopped_mode = get_ecs_status(client, instance_id)

    # 已经停止 → 不重复停机
    if status == "Stopped":
        logger.info(f"ECS实例 {instance_id} 已经是停止状态，无需再次停止。")
        return

    try:
        request = StopInstancesRequest.StopInstancesRequest()
        request.set_InstanceIds([instance_id])
        request.set_ForceStop(False)

        # ⭐ 强制设置节省停机模式 ⭐
        request.set_StoppedMode("StopCharging")

        request.set_accept_format('json')

        response = client.do_action_with_exception(request)
        logger.info(f"ECS停止响应: {response.decode('utf-8')}")
        logger.info("停机模式: StopCharging（节省停机）")

    except Exception as e:
        logger.error(f"停止 ECS 实例失败: {e}")

# ================== 8. 主流程 ==================
def main():
    """
    主逻辑：
        - 查询流量
        - 根据阈值决定启动或停止 ECS
    """
    total_gb = get_total_traffic_gb(client)

    if total_gb < TRAFFIC_THRESHOLD_GB:
        logger.info(f"流量 {total_gb:.2f} GB < 阈值 {TRAFFIC_THRESHOLD_GB} GB，尝试启动 ECS")
        ecs_start(client, ECS_INSTANCE_ID)
    else:
        logger.info(f"流量 {total_gb:.2f} GB ≥ 阈值 {TRAFFIC_THRESHOLD_GB} GB，尝试停止 ECS")
        ecs_stop(client, ECS_INSTANCE_ID)

    logger.info("脚本执行完毕。")

# ================== 程序入口 ==================
if __name__ == "__main__":
    main()

以上两个脚本有需求的话可以提交给AI，合二为一更方便！

还有一个免OOS的三合一高度集成的自用脚本：自动轮换+故障转移+自动DDNS+TG通知，像我一样国内站和国际站都开的应该不多，脚本就不发了 xhj010

======================================================================
阿里云 CDT 抢占式 ECS 自动轮换保活脚本（故障转移，自动DDNS，TG通知）
======================================================================
核心逻辑说明：

1. 📅 流量保护机制与双账号轮换 (12H 班次):
   - [流量监控]：实时监测流量，流量 < 188GB 正常运行；流量 ≥ 188GB 触发保护。
   - [分时当班]：AM(00-12) 账号A当班，PM(12-24) 账号B当班。非当班实例强制执行“节省停机”。
   - [动态调度]：首选账号超限自动切换备选；若双账号流量均超限，则进入“全局熔断”全线关机，确保处于 Stopped (节省停机)状态。

2. 🌐 DDNS 自动化同步 (Cloudflare):
   - [原子化同步]：DDNS 更新操作严格前置于 Telegram 通知，确保通知中的解析状态真实有效。
   - [状态回显]：通知信息中实时回传 DDNS 同步结果，并在系统日志中保留完整的 API 交互记录。   
   - [多场景触发]：在准点换班、故障转移、抢占恢复等 IP 变动场景下，自动同步最新 IP 至 Cloudflare。

3. 🔔 智能告警与状态核实系统:
   - [异步核实机制]：所有“停机”类通知（熔断、限额停机）均在确认实例进入 Stopped (节省停机) 状态后发出。
   - [全生命周期通知]：
     - 🟢 抢占恢复：检测到抢占式实例被回收并自动拉起成功后，同步解析并发送抢占恢复通知。
     - 🔴 超限停机：当班账号流量达标且无法换班时，下发指令并确认进入“节省停机”后发送超限停机通知。
     - 🚨 全局熔断：双账号均确认“节省停机”后，发送包含真实状态参数的全局熔断报告。
     - ⚠️ 换班异常：若准点换班超时（10分钟），实例状态仍未达到预期的“运行”或“节省停机”，实时发送换班异常报告。
     - ⚠️ 临时换班：非换班时段因流量超限触发转移，在确认新旧实例状态就绪后发送临时换班报告。
     - ✅ 换班完成：00:00/12:00 准点触发，严格校验“节省停机”参数生效且 DDNS 同步后推送换班完成报告。
     - [每日汇总]：23:58 定时推送双账号流量清单及额度预警，实现全天候资产监控。

======================================================================
1234
  • Flanker
    Flanker
    #2

    要能做成自动化联动就好了 人醒来了开机 人睡着了关机

  • xbbazt
    xbbazt
    #1

    好活！

  • MJJYYDS
    MJJYYDS楼主
    #3

    @Flanker #2
    两个鸡12小时交替，然后用CLASH系客户端自动选择功能，我基本上用在手机上，这两鸡延迟最低，总能被优选到一个 xhj008

  • xmanx
    xmanx
    #4

    你是会玩鸡的

  • Tokin
    Tokin
    #5

    分享一下我vibe的shell自动切换dns解析脚本：https://www.nodeseek.com/post-579497-1

    我现在就是，白天自动解析到ecs-cdt，晚上ecs关机前自动解析到另一台小鸡

  • AEgean
    AEgean
    #6

    负载不均衡可是要破产的

  • mozart229
    mozart229
    #7

    没看明白，24H是47.8，那交替的意义是什么？

  • Tokin
    Tokin
    #8
  • FCB
    FCB
    #9

    人一多阿里就要改套路了

  • Demonking
    Demonking
    #10

    搞个阿里国际 开个免费鸡，开启cdt 每个月200G起飞！

 1234

你好啊，陌生人!

我的朋友，看起来你是新来的，如果想参与到讨论中，点击下面的按钮！
登录 注册

快捷功能区

所有版块

📈用户数目📈

目前论坛共有72105位seeker

🎉欢迎新用户🎉

vastness
vastness
ggkb00
ggkb00
TideLi
TideLi
奔驰的兔子
奔驰的兔子
simon334
simon334
岂是沧海
岂是沧海
worxift
worxift
niccfun
niccfun