基于华为云 FunctionGraph 与 API 网关构建 Serverless 轻量级服务巡检与智能通知实战

举报
yd_297008388 发表于 2026/10/29 09:39:25 2026/10/29
【摘要】 在微服务与分布式架构下,业务接口的高可用拨测与证书监控至关重要。本文深入讲解如何利用华为云 Serverless 核心产品 FunctionGraph 配合定时触发器与消息通知服务(SMN),零服务器成本打造一套全天候自动化智能巡检与告警系统。

在现代化云原生应用架构中,服务可用性拨测、核心 API 接口响应延时监控以及 SSL/TLS 证书到期预警是保障系统稳定运行的基础防线。传统方案通常依赖常驻虚拟机(ECS)搭建 Prometheus 探针或运行定制化 Python 巡检脚本,但对于中小团队或轻量化业务而言,这种方式存在机器闲置开销大、探针单点故障以及系统补丁维护琐碎等痛点。华为云函数工作流(FunctionGraph)提供了按需调用、毫秒级弹性、零运维的 Serverless 算力,是实现轻量级敏捷巡检的绝佳选择。

一、Serverless 智能巡检整体架构设计

本方案依托华为云 Serverless 生态产品构建闭环检测链路:

  • 定时调度层(Timer Trigger):在 FunctionGraph 内部配置 Cron 定时触发器(如每 5 分钟执行一次),驱动函数实例启动。
  • 探针计算核心(FunctionGraph):运行轻量 Python 运行环境,并发向目标业务端点发起 HTTP/HTTPS 探活请求,采集状态码、响应耗时(TTFB)、DNS 解析时间并解析证书有效期。
  • 告警通知下发(SMN 服务):当探针检测到接口异常(如非 2xx 响应、超时严重)或 SSL 证书将在 15 天内过期时,触发华为云消息通知服务(SMN),通过邮件、短信或企业微信/钉钉 Webhook 实时告警。
  • 对外管理能力(API 网关 APIG):可选绑定专享型/共享型 API 网关,供运维人员在移动端或控制台通过鉴权 API 手动触发即时全量巡检。

二、FunctionGraph 巡检函数核心实现

以下为在 FunctionGraph 中运行的生产级探针核心代码示例,基于 Python 3.9 编写:

# -*- coding:utf-8 -*-
import json
import time
import socket
import ssl
from datetime import datetime
import urllib.request
import urllib.error

# 巡检目标列表
TARGET_TARGETS = [
    {"name": "官网主站", "url": "https://developer.huaweicloud.com", "max_latency_ms": 1200},
    {"name": "API 网关健康检测端点", "url": "https://api.example.com/healthz", "max_latency_ms": 800}
]

def check_ssl_expiry(hostname, port=443):
    """检测 SSL 证书剩余有效天数"""
    context = ssl.create_default_context()
    with socket.create_connection((hostname, port), timeout=5) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            cert = ssock.getpeercert()
            expire_date_str = cert['notAfter']
            expire_date = datetime.strptime(expire_date_str, '%b %d %H:%M:%S %Y %Z')
            remaining_days = (expire_date - datetime.utcnow()).days
            return remaining_days

def handler(event, context):
    logger = context.getLogger()
    report = []
    has_alarm = False
    
    for item in TARGET_TARGETS:
        target_name = item["name"]
        url = item["url"]
        threshold = item["max_latency_ms"]
        
        start_time = time.time()
        status_code = 0
        error_msg = ""
        
        try:
            req = urllib.request.Request(url, headers={'User-Agent': 'HuaweiCloud-Probe/1.0'})
            with urllib.request.urlopen(req, timeout=5) as response:
                status_code = response.getcode()
        except urllib.error.HTTPError as e:
            status_code = e.code
            error_msg = str(e)
        except Exception as e:
            status_code = 599
            error_msg = str(e)
            
        elapsed_ms = int((time.time() - start_time) * 1000)
        
        # 证书检测(针对 https)
        cert_days = -1
        if url.startswith("https://"):
            try:
                host = url.split("://")[1].split("/")[0].split(":")[0]
                cert_days = check_ssl_expiry(host)
            except Exception as e:
                logger.warn("SSL Check Failed: %s" % str(e))
                
        # 判定告警
        is_unhealthy = (status_code < 200 or status_code >= 400) or (elapsed_ms > threshold) or (cert_days != -1 and cert_days < 15)
        if is_unhealthy:
            has_alarm = True
            
        report.append({
            "target": target_name,
            "url": url,
            "status_code": status_code,
            "latency_ms": elapsed_ms,
            "cert_remaining_days": cert_days,
            "healthy": not is_unhealthy,
            "error": error_msg
        })
        
    logger.info("Probe Result: %s" % json.dumps(report, ensure_ascii=False))
    
    # 若存在告警,可在此处调用 SMN SDK 发送告警通知
    return {
        "statusCode": 200,
        "isBase64Encoded": False,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"alarm": has_alarm, "details": report}, ensure_ascii=False)
    }

三、关键配置与安全落地要点

  1. 运行规格推荐:由于巡检主要为网络 I/O 密集型任务,FunctionGraph 内存配置选用最低档位(128MB 或 256MB)即可平稳运行,单次执行耗时通常在 1 秒以内,每月几十万次调用都在免费额度范围内。
  2. 最小权限委托(Agency):若需将结果写入对象存储 OBS 或调用云监控(Cloud Eye)打点,应在华为云 IAM 中创建最小权限的 FunctionGraph 服务委托,切勿在代码中明文硬编码 AK/SK。
  3. 防抖与告警抑制:在生产中偶发网络抖动可能造成误报。建议在函数内引入连续 2 次探针失败判定机制(利用 FunctionGraph 临时缓存或写入云缓存 DCS Redis),提高告警准确率。

四、总结

通过 FunctionGraph + Timer 触发器构建服务巡检方案,我们彻底抛弃了冗杂的基础设施运维,实现了“用完即走、按毫秒计费、零维护负担”的现代化云原生运维模式,非常适合广大开发者在中小型业务中快速落地。

【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0)

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。