从零构建 Serverless 定时巡检与告警系统：华为云 FunctionGraph + 定时触发器 + SMN + LTS 全链路实践

作者：华为云开发者社区
适用场景：定时任务、自动化巡检、监控告警、运维编排
阅读时长：约 25 分钟
前置要求：华为云账号、基础 Linux/Python 知识

一、为什么需要 Serverless 定时巡检？

在传统运维中，定时巡检（如磁盘空间检查、服务存活探测、日志异常扫描）通常依赖 Cron + 脚本 + 邮件通知的组合。这套方案虽然经典，却有几个绕不开的痛点：

• 服务器成本：巡检脚本需要常驻一台 ECS，哪怕一天只跑 3 分钟，这台机器也得 24 小时计费。
• 运维负担：机器宕机、Cron 配置漂移、脚本依赖环境缺失，都是隐性成本。
• 扩展性差：巡检项从 1 个涨到 20 个，单机脚本很快变成"意大利面条"。

Serverless 定时巡检 的核心思路是：用事件驱动代替常驻进程，用按需计费代替包年包月。华为云 FunctionGraph 提供原生定时触发器，配合 SMN（消息通知服务）和 LTS（云日志服务），可以构建一套零运维、低成本的自动化巡检与告警系统。

本文将带你从零搭建一个多维度定时巡检平台，涵盖：

1. 系统资源巡检（磁盘、内存、CPU）
2. 服务连通性探测（HTTP 健康检查）
3. 日志异常关键词扫描
4. 告警消息分发（短信 + 邮件 + Webhook）

二、架构总览

┌─────────────────────────────────────────────────────┐
│                  FunctionGraph 函数                  │
│  ┌───────────────┐  ┌───────────────┐               │
│  │  巡检主函数    │  │  告警分发函数  │               │
│  │  (inspector)  │  │ (dispatcher)  │               │
│  └───────┬───────┘  └───────┬───────┘               │
│          │ 定时触发器        │ SMN 主题               │
└──────────┼──────────────────┼───────────────────────┘
           │                  │
           ▼                  ▼
    ┌─────────────┐   ┌──────────────┐
    │ LTS 日志    │   │ 短信 / 邮件   │
    │ (存储+查询)  │   │ / Webhook    │
    └─────────────┘   └──────────────┘

核心组件说明：

与已发布文章的差异：

三、环境准备

3.1 账号与权限

确保你的华为云账号具备以下权限（若为子账号，需主账号授权）：

• FunctionGraph Administrator（函数工作流管理员）
• SMN Administrator（消息通知管理员）
• LTS Administrator（云日志服务管理员）
• Tenant Guest（只读权限，用于巡检其他资源）

3.2 本地开发环境

# 确认 Python 版本（建议 3.9+）
python3 --version

# 安装 Huawei Cloud CLI（可选，用于命令行管理）
pip install huaweicloudsdkcore huaweicloudsdkfunctiongraph huaweicloudsdksmn

# 配置 CLI 凭据
hcloud configure set --ak YOUR_AK --sk YOUR_SK --region cn-north-4

提示：AK/SK 请在华为云控制台「我的凭证 → 访问密钥」中创建。切勿将 AK/SK 硬编码在函数代码中，应使用 FunctionGraph 的「密钥管理」功能注入环境变量。

四、第一步：创建 SMN 主题与订阅

告警消息需要通过 SMN 分发。我们先创建主题（Topic），再添加订阅（Subscription）。

4.1 创建主题

登录华为云控制台，进入 消息通知服务 SMN：

1. 点击「创建主题」
2. 主题名称：inspector-alerts
3. 显示名称：巡检告警主题
4. 点击「确定」

记录下生成的 主题 URN，格式类似：

smn:cn-north-4:123456789012:topic:inspector-alerts

4.2 添加订阅

在主题详情页，点击「添加订阅」：

注意：邮件和短信订阅需要先在 SMN 的「订阅管理」中完成确认（点击确认链接）。HTTPS 订阅需要目标地址返回 200 OK 才能确认。

4.3 通过 CLI 创建（可选）

# 创建主题
huaweicloud smn create-topic \
  --name "inspector-alerts" \
  --display-name "巡检告警主题"

# 添加邮件订阅
huaweicloud smn subscribe \
  --topic-urn "smn:cn-north-4:123456789012:topic:inspector-alerts" \
  --protocol email \
  --endpoint "your-email-domain"

# 添加短信订阅
huaweicloud smn subscribe \
  --topic-urn "smn:cn-north-4:123456789012:topic:inspector-alerts" \
  --protocol sms \
  --endpoint "138****1234"

五、第二步：创建巡检主函数

5.1 函数基本信息

进入 FunctionGraph 控制台：

1. 点击「创建函数」
2. 函数名称：inspector-main
3. 运行时：Python 3.9
4. 函数类型：事件函数（由定时触发器调用）
5. 内存：256 MB（巡检任务资源消耗低，256MB 足够）
6. 超时：90 秒

5.2 函数代码

将以下代码粘贴到函数编辑器中：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
华为云 FunctionGraph 定时巡检函数
功能：系统资源巡检 + 服务连通性探测 + 日志异常扫描 + 告警分发
"""

import json
import os
import time
import logging
import urllib.request
import urllib.error
from datetime import datetime
from typing import Dict, List, Any, Optional

# 华为云 SDK（函数内置，无需额外安装）
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkcore.http.http_config import HttpConfig
from huaweicloudsdkcore.region.region import Region
from huaweicloudsdkfunctiongraph.v1.region_functiongraph_region import FunctionGraphRegion
from huaweicloudsdkfunctiongraph.v1 import *
from huaweicloudsdkcore.auth import GlobalCredentials

# 配置日志（自动输出到 LTS）
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# ========== 从环境变量读取配置 ==========
SMN_TOPIC_URN = os.environ.get("SMN_TOPIC_URN", "")
ALERT_WEBHOOK_URL = os.environ.get("ALERT_WEBHOOK_URL", "")
LOG_GROUP = os.environ.get("LTS_LOG_GROUP", "inspector")
LOG_STREAM = os.environ.get("LTS_LOG_STREAM", "main")

# 巡检目标配置（可从环境变量或函数配置中读取）
HEALTH_CHECK_TARGETS = json.loads(os.environ.get("HEALTH_CHECK_TARGETS", "[]"))
DISK_CHECK_MOUNTS = json.loads(os.environ.get("DISK_CHECK_MOUNTS", '["/", "/data"]'))
LOG_KEYWORDS = json.loads(os.environ.get("LOG_KEYWORDS", '["ERROR", "CRITICAL", "FATAL"]'))


class Inspector:
    """巡检器核心类"""

    def __init__(self):
        self.results: List[Dict[str, Any]] = []
        self.alerts: List[Dict[str, Any]] = []
        self.start_time = datetime.now()

    def run_all(self) -> Dict[str, Any]:
        """执行全部巡检项"""
        logger.info("===== 巡检任务开始 =====")

        # 1. 系统资源巡检
        self._check_system_resources()

        # 2. 服务连通性探测
        self._check_service_health()

        # 3. 日志异常扫描（示例：模拟扫描，实际可对接 LTS 查询接口）
        self._check_log_anomalies()

        # 汇总结果
        end_time = datetime.now()
        duration_ms = int((end_time - self.start_time).total_seconds() * 1000)

        summary = {
            "timestamp": self.start_time.isoformat(),
            "duration_ms": duration_ms,
            "total_checks": len(self.results),
            "passed": sum(1 for r in self.results if r["status"] == "PASS"),
            "failed": sum(1 for r in self.results if r["status"] == "FAIL"),
            "alerts": len(self.alerts),
            "details": self.results,
            "alerts_list": self.alerts
        }

        logger.info(f"巡检完成：通过 {summary['passed']} / 失败 {summary['failed']} / 告警 {summary['alerts']}")

        # 4. 发送告警
        if self.alerts:
            self._dispatch_alerts(summary)

        return summary

    def _check_system_resources(self):
        """系统资源巡检（磁盘、内存、CPU）"""
        # 磁盘使用率检查
        for mount in DISK_CHECK_MOUNTS:
            try:
                stat = os.statvfs(mount)
                total = stat.f_blocks * stat.f_frsize
                free = stat.f_bavail * stat.f_frsize
                used = total - free
                usage_pct = (used / total) * 100

                status = "PASS" if usage_pct < 85 else "FAIL"
                self.results.append({
                    "check": f"disk_usage_{mount}",
                    "status": status,
                    "detail": f"已用 {usage_pct:.1f}% (总量 {total/1024/1024/1024:.1f}GB)",
                    "threshold": "85%",
                    "severity": "WARNING" if usage_pct >= 85 else "INFO"
                })

                if usage_pct >= 85:
                    self.alerts.append({
                        "title": f"[磁盘告警] {mount} 使用率过高",
                        "content": f"{mount} 磁盘使用率 {usage_pct:.1f}%，超过阈值 85%",
                        "level": "WARNING",
                        "check": f"disk_{mount}"
                    })
            except Exception as e:
                self.results.append({
                    "check": f"disk_usage_{mount}",
                    "status": "FAIL",
                    "detail": f"检查失败: {str(e)}",
                    "threshold": "N/A",
                    "severity": "ERROR"
                })

        # 内存检查（Linux 系统）
        try:
            with open("/proc/meminfo", "r") as f:
                meminfo = {}
                for line in f:
                    parts = line.split(":")
                    if len(parts) == 2:
                        meminfo[parts[0]] = parts[1].strip()

            total_kb = int(meminfo.get("MemTotal", "0").split()[0])
            available_kb = int(meminfo.get("MemAvailable", "0").split()[0])
            used_pct = ((total_kb - available_kb) / total_kb) * 100 if total_kb > 0 else 0

            status = "PASS" if used_pct < 90 else "FAIL"
            self.results.append({
                "check": "memory_usage",
                "status": status,
                "detail": f"已用 {used_pct:.1f}% (总量 {total_kb/1024/1024:.1f}GB)",
                "threshold": "90%",
                "severity": "WARNING" if used_pct >= 90 else "INFO"
            })
        except Exception as e:
            self.results.append({
                "check": "memory_usage",
                "status": "FAIL",
                "detail": f"检查失败: {str(e)}",
                "threshold": "N/A",
                "severity": "ERROR"
            })

    def _check_service_health(self):
        """服务连通性探测"""
        if not HEALTH_CHECK_TARGETS:
            # 默认检查几个常用服务
            default_targets = [
                {"name": "华为云官网", "url": "华为云官网", "expected_status": 200},
                {"name": "FunctionGraph 控制台", "url": "FunctionGraph 控制台", "expected_status": 200},
            ]
            targets = default_targets
        else:
            targets = HEALTH_CHECK_TARGETS

        for target in targets:
            name = target.get("name", "unknown")
            url = target.get("url", "")
            expected = target.get("expected_status", 200)
            timeout = target.get("timeout", 10)

            try:
                req = urllib.request.Request(url, headers={"User-Agent": "Inspector/1.0"})
                with urllib.request.urlopen(req, timeout=timeout) as resp:
                    actual = resp.getcode()

                status = "PASS" if actual == expected else "FAIL"
                self.results.append({
                    "check": f"health_{name}",
                    "status": status,
                    "detail": f"HTTP {actual} (期望 {expected})",
                    "threshold": f"HTTP {expected}",
                    "severity": "CRITICAL" if actual != expected else "INFO"
                })

                if actual != expected:
                    self.alerts.append({
                        "title": f"[服务异常] {name} 返回 {actual}",
                        "content": f"服务 {name} ({url}) 返回 HTTP {actual}，期望 {expected}",
                        "level": "CRITICAL",
                        "check": f"health_{name}"
                    })
            except urllib.error.URLError as e:
                self.results.append({
                    "check": f"health_{name}",
                    "status": "FAIL",
                    "detail": f"连接失败: {e.reason}",
                    "threshold": f"HTTP {expected}",
                    "severity": "CRITICAL"
                })
                self.alerts.append({
                    "title": f"[服务不可达] {name}",
                    "content": f"服务 {name} ({url}) 无法连接: {e.reason}",
                    "level": "CRITICAL",
                    "check": f"health_{name}"
                })
            except Exception as e:
                self.results.append({
                    "check": f"health_{name}",
                    "status": "FAIL",
                    "detail": f"检查异常: {str(e)}",
                    "threshold": f"HTTP {expected}",
                    "severity": "ERROR"
                })

    def _check_log_anomalies(self):
        """日志异常关键词扫描"""
        # 实际场景中，这里可以调用 LTS 的查询接口
        # 此处为示例逻辑：模拟从日志中检测关键词
        simulated_log_sample = (
            "2024-01-15 10:23:45 INFO  Service started\n"
            "2024-01-15 10:24:01 ERROR Database connection failed\n"
            "2024-01-15 10:24:02 WARN  Retrying connection...\n"
            "2024-01-15 10:25:30 CRITICAL Disk I/O timeout on /dev/vdb\n"
        )

        for keyword in LOG_KEYWORDS:
            count = simulated_log_sample.count(keyword)
            if count > 0:
                severity = "CRITICAL" if keyword in ("CRITICAL", "FATAL") else "WARNING"
                self.results.append({
                    "check": f"log_keyword_{keyword}",
                    "status": "FAIL",
                    "detail": f"发现 {count} 条 '{keyword}' 日志",
                    "threshold": "0",
                    "severity": severity
                })
                self.alerts.append({
                    "title": f"[日志异常] 发现关键词 '{keyword}'",
                    "content": f"在日志中检测到 {count} 条 '{keyword}' 级别记录，请及时处理",
                    "level": severity,
                    "check": f"log_{keyword}"
                })

    def _dispatch_alerts(self, summary: Dict[str, Any]):
        """分发告警到 SMN 和 Webhook"""
        # 1. 发送到 SMN 主题
        if SMN_TOPIC_URN:
            self._send_smn_alert(summary)

        # 2. 发送到 Webhook
        if ALERT_WEBHOOK_URL:
            self._send_webhook_alert(summary)

    def _send_smn_alert(self, summary: Dict[str, Any]):
        """通过 SMN 发送告警"""
        try:
            from huaweicloudsdkcore.auth.credentials import BasicCredentials
            from huaweicloudsdkcore.region.region import Region
            from huaweicloudsdksmn.v2.region_smn_region import SmnRegion
            from huaweicloudsdksmn.v2 import *

            credentials = BasicCredentials(
                ak=os.environ.get("HUAWEICLOUD_AK", ""),
                sk=os.environ.get("HUAWEICLOUD_SK", "")
            )

            client = SmnClient.new_builder() \
                .with_credentials(credentials) \
                .with_region(SmnRegion.value_of("cn-north-4")) \
                .build()

            # 构建告警消息
            alert_body = self._build_alert_message(summary)

            request = PublishRequest(
                topic_urn=SMN_TOPIC_URN,
                subject=alert_body["subject"],
                message=alert_body["message"],
                message_structure="json"
            )

            response = client.publish(request)
            logger.info(f"SMN 告警发送成功，消息ID: {response.message_id}")

        except Exception as e:
            logger.error(f"SMN 告警发送失败: {str(e)}")

    def _send_webhook_alert(self, summary: Dict[str, Any]):
        """通过 Webhook 发送告警"""
        try:
            alert_body = self._build_alert_message(summary)
            payload = json.dumps({
                "msg_type": "alert",
                "timestamp": summary["timestamp"],
                "subject": alert_body["subject"],
                "content": alert_body["message"],
                "alerts": summary["alerts_list"],
                "summary": {
                    "total": summary["total_checks"],
                    "passed": summary["passed"],
                    "failed": summary["failed"]
                }
            }).encode("utf-8")

            req = urllib.request.Request(
                ALERT_WEBHOOK_URL,
                data=payload,
                headers={
                    "Content-Type": "application/json",
                    "User-Agent": "Inspector/1.0"
                },
                method="POST"
            )
            with urllib.request.urlopen(req, timeout=10) as resp:
                logger.info(f"Webhook 告警发送成功，HTTP {resp.getcode()}")
        except Exception as e:
            logger.error(f"Webhook 告警发送失败: {str(e)}")

    def _build_alert_message(self, summary: Dict[str, Any]) -> Dict[str, str]:
        """构建告警消息内容"""
        failed_items = [r for r in summary["details"] if r["status"] == "FAIL"]
        alert_lines = []
        for item in failed_items:
            alert_lines.append(f"• [{item['severity']}] {item['check']}: {item['detail']}")

        message = f"""【巡检告警】
时间：{summary['timestamp']}
状态：{summary['failed']} 项失败 / {summary['passed']} 项通过

异常项：
{chr(10).join(alert_lines)}

请及时登录华为云控制台查看详情。"""

        subject = f"[巡检告警] 发现 {summary['failed']} 项异常"

        return {"subject": subject, "message": message}


def handler(event: Dict, context):
    """
    FunctionGraph 入口函数
    
    参数：
        event: 触发事件（定时触发器会传入调度信息）
        context: 函数上下文（包含函数名、请求ID等）
    
    返回：
        Dict: 巡检结果摘要
    """
    logger.info(f"函数被调用，请求ID: {context.request_id}")
    logger.info(f"事件来源: {event.get('source', 'unknown')}")

    inspector = Inspector()
    result = inspector.run_all()

    return {
        "code": 0,
        "message": "success",
        "data": result
    }

5.3 环境变量配置

在函数控制台的「配置 → 环境变量」中，添加以下变量：

安全建议：AK/SK 不要直接写在环境变量中。在 FunctionGraph 控制台「配置 → 密钥管理」中创建密钥，然后在环境变量中引用 ${secret_name}。

六、第三步：配置定时触发器

6.1 创建定时触发器

在函数详情页，点击「触发器 → 创建触发器」：

6.2 Cron 表达式速查

6.3 多触发器配置

可以为同一个函数配置多个定时触发器，实现分级巡检：

触发器 A（每 5 分钟）：健康检查（快速探测）
触发器 B（每 1 小时）：资源巡检（磁盘、内存）
触发器 C（每天 0 点）：全量巡检 + 日志扫描

七、第四步：配置 LTS 日志接入

FunctionGraph 的日志默认输出到 LTS，但我们需要做以下配置以便检索和分析：

7.1 创建日志组和日志流

进入 云日志服务 LTS 控制台：

1. 点击「日志组 → 创建日志组」

   • 名称：inspector
   • 保留天数：30（巡检日志建议保留 30 天）

2. 在日志组中创建日志流：

   • 名称：main

7.2 配置日志索引

在日志组 inspector 的「索引配置」中，添加以下字段索引：

7.3 日志查询示例

在 LTS 控制台，使用以下查询语句分析巡检结果：

# 查询最近 1 小时所有失败的巡检
status: FAIL

# 查询 CRITICAL 级别告警
severity: CRITICAL

# 查询磁盘相关巡检
check: disk_*

# 统计各巡检项通过率
* | select check, count(*) as total, 
       count_if(status='PASS') as passed,
       count_if(status='FAIL') as failed,
       round(count_if(status='PASS')*100.0/count(*), 2) as pass_rate
    group by check order by pass_rate desc

八、第五步：测试与验证

8.1 手动触发测试

在函数控制台，点击「测试」按钮，输入测试事件：

{
  "source": "timer",
  "trigger_name": "inspector-schedule",
  "event_time": "2024-01-15T10:00:00Z"
}

点击「执行」，观察：

1. 函数日志：在「日志」标签页查看执行输出
2. 执行结果：在「结果」标签页查看返回数据
3. 告警接收：检查邮箱/手机是否收到告警

8.2 预期输出

{
  "code": 0,
  "message": "success",
  "data": {
    "timestamp": "2024-01-15T10:00:00.123456",
    "duration_ms": 1234,
    "total_checks": 5,
    "passed": 3,
    "failed": 2,
    "alerts": 2,
    "details": [
      {
        "check": "disk_usage_/",
        "status": "PASS",
        "detail": "已用 45.2% (总量 40.0GB)",
        "threshold": "85%",
        "severity": "INFO"
      },
      {
        "check": "health_我的服务",
        "status": "FAIL",
        "detail": "HTTP 502 (期望 200)",
        "threshold": "HTTP 200",
        "severity": "CRITICAL"
      }
    ],
    "alerts_list": [
      {
        "title": "[服务异常] 我的服务 返回 502",
        "content": "服务 我的服务 返回 HTTP 502，期望 200",
        "level": "CRITICAL",
        "check": "health_我的服务"
      }
    ]
  }
}

8.3 验证 SMN 通知

1. 登录 SMN 控制台，查看「主题订阅 → 消息发布记录」
2. 确认消息状态为「成功」
3. 检查邮件/短信是否收到告警内容

九、进阶：多账号巡检与跨区域监控

9.1 跨账号巡检

如果你有多个华为云账号，可以通过委托授权实现统一巡检：

1. 在被巡检账号中创建委托：

   • 委托名称：inspector-delegation
   • 委托对象：主账号的账号 ID
   • 权限：Tenant Guest

2. 在主账号函数中，使用 STS 临时凭证访问被巡检账号：

from huaweicloudsdkcore.auth.credentials import StsCredentials

# 获取临时安全凭证
sts_response = assume_role(
    role_arn="iam::123456789012:role/inspector-delegation",
    session_name="inspector-session"
)

# 使用临时凭证创建客户端
credentials = StsCredentials(
    ak=sts_response.credentials.access_key,
    sk=sts_response.credentials.secret_key,
    security_token=sts_response.credentials.security_token
)

9.2 跨区域监控

函数部署在 cn-north-4，但可以巡检其他区域的资源：

# 在 HEALTH_CHECK_TARGETS 中配置跨区域服务
HEALTH_CHECK_TARGETS = [
    {"name": "北京四 ECS", "url": "/health"},
    {"name": "上海一 RDS", "url": "/health"},
    {"name": "广州 OBS", "url": "/health"}
]

十、成本分析

10.1 计费项

10.2 与 ECS 方案对比

十一、最佳实践与注意事项

11.1 函数设计原则

1. 幂等性：定时触发器可能因网络重试导致重复执行，函数逻辑必须幂等。
2. 超时设置：巡检任务建议设置 30-90 秒超时，避免长时间阻塞。
3. 错误处理：网络异常、API 限流等场景应有重试和降级逻辑。
4. 日志规范：使用结构化日志（JSON），便于 LTS 检索。

11.2 告警分级

建议将告警分为三个等级：

11.3 安全加固

1. 最小权限原则：函数使用的 AK/SK 仅授予巡检所需的只读权限。
2. 密钥管理：使用 FunctionGraph 密钥管理功能，不要硬编码凭据。
3. 网络隔离：如需访问 VPC 内资源，配置函数启用 VPC 访问。
4. Webhook 校验：对 Webhook 地址添加签名验证，防止伪造告警。

11.4 常见问题排查

十二、完整部署脚本（Terraform）

如果你偏好 IaC 方式，以下 Terraform 脚本可一键部署整套巡检系统：

# main.tf

terraform {
  required_providers {
    huaweicloud = {
      source  = "huaweicloud/huaweicloud"
      version = ">= 4.0"
    }
  }
}

provider "huaweicloud" {
  region = "cn-north-4"
}

# 1. SMN 主题
resource "huaweicloud_smn_topic" "inspector_alerts" {
  name         = "inspector-alerts"
  display_name = "巡检告警主题"
}

# 2. SMN 邮件订阅
resource "huaweicloud_smn_subscription" "email" {
  topic_urn = huaweicloud_smn_topic.inspector_alerts.id
  protocol  = "email"
  endpoint  = "your-email-domain"
}

# 3. FunctionGraph 函数
resource "huaweicloud_fgc_function" "inspector" {
  name        = "inspector-main"
  handler     = "index.handler"
  memory_size = 256
  timeout     = 90
  runtime     = "Python3.9"
  code_type   = "inline"

  func_code = filebase64("${path.module}/inspector.py")

  environment {
    variables = {
      SMN_TOPIC_URN      = huaweicloud_smn_topic.inspector_alerts.id
      HEALTH_CHECK_TARGETS = jsonencode([
        {
          name  = "华为云官网"
          url   = "华为云官网"
          expected_status = 200
        }
      ])
      DISK_CHECK_MOUNTS = jsonencode(["/", "/data"])
      LOG_KEYWORDS      = jsonencode(["ERROR", "CRITICAL", "FATAL"])
    }
  }
}

# 4. 定时触发器
resource "huaweicloud_fgc_func_trigger" "schedule" {
  function_urn = huaweicloud_fgc_function.inspector.urn
  type         = "TIMER"

  timer_config {
    schedule_type = "Cron"
    cron_expression = "0 */2 * * *"
    enabled       = true
  }
}

# 5. LTS 日志组
resource "huaweicloud_lts_group" "inspector" {
  group_name  = "inspector"
  ttl_in_days = 30
}

# 6. 日志流
resource "huaweicloud_lts_stream" "inspector_main" {
  group_id    = huaweicloud_lts_group.inspector.id
  stream_name = "main"
}

# 输出
output "smn_topic_urn" {
  value = huaweicloud_smn_topic.inspector_alerts.id
}

output "function_urn" {
  value = huaweicloud_fgc_function.inspector.urn
}

部署命令：

terraform init
terraform plan -out=tfplan
terraform apply tfplan

十三、总结

本文从零构建了一套基于华为云 FunctionGraph 的 Serverless 定时巡检与告警系统，核心要点回顾：

1. 零服务器成本：用 FunctionGraph 定时触发器替代 ECS + Cron，月成本从 ¥70+ 降至 ¥2-5。
2. 全链路告警：通过 SMN 实现短信、邮件、Webhook 多渠道分发，确保告警必达。
3. 可观测性：LTS 提供结构化日志存储与检索，支持巡检历史追溯和趋势分析。
4. 可扩展性：函数化设计让新增巡检项只需修改代码，无需改动架构。

下一步可以探索的方向：

• 接入 云监控服务（CES），实现资源指标自动采集
• 集成 工作流服务（Workflow），编排复杂巡检流程
• 对接 函数工作流 DAG，实现巡检结果的自动修复（如自动清理磁盘）
• 结合 API 网关，开放巡检结果查询接口

本文首发于华为云开发者社区，转载请注明出处。