Serverless 實戰：通過 Serverless 架構實作監控告警

在實際生產中，我們經常需要做一些監控腳本來監控網站服務或者 API 服務是否可用，傳統的方法是使用網站監控平臺（例如 DNSPod 監控、360 網站服務監控，以及阿里云監控等），它們的原理是通過用戶自己設定要監控的服務地址和監測的時間閾值，由監控平臺定期發起請求對網站或服務的可用性進行判斷，

這些方法很大眾化，通用性很強，但也不是所有場景都適合，例如，如果我們的需求是監控網站狀態碼，不同區域的延時，并且通過監控得到的資料，設定一個閾值，一旦超過閾值就通過郵件等進行統治告警，目前大部分的監控平臺是很難滿足這些需求的，這時就需要定制開發一個監控工具，

Serverless 服務的一個重要應用場景就是運維、監控與告警，所以本文將會通過現有的 Serverless 平臺，部署一個網站狀態監控腳本，對目標網站的可用性進行監控告警，

Web 服務監控告警

針對 Web 服務，我們先設計一個簡單的監控告警功能的流程：

在這個流程中，我們僅對網站的狀態碼進行監控，即回傳的狀態為 200，則判定網站可正常使用，否則進行告警：

# -*- coding: utf8 -*-
import ssl
import json
import smtplib
import urllib.request
from email.mime.text import MIMEText
from email.header import Header
 
ssl._create_default_https_context = ssl._create_unverified_context
 
 
def sendEmail(content, to_user):
    sender = '[email protected]'
    receivers = [to_user]
 
    mail_msg = content
    message = MIMEText(mail_msg, 'html', 'utf-8')
    message['From'] = Header(" 網站監控 ", 'utf-8')
    message['To'] = Header(" 站長 ", 'utf-8')
 
    subject = " 網站監控告警 "
    message['Subject'] = Header(subject, 'utf-8')
 
    try:
        smtpObj = smtplib.SMTP_SSL("smtp.exmail.qq.com", 465)
        smtpObj.login('發送郵件的郵箱地址', '密碼')
        smtpObj.sendmail(sender, receivers, message.as_string())
    except smtplib.SMTPException as e:
        print(e)
 
 
def getStatusCode(url):
    return urllib.request.urlopen(url).getcode()
 
 
def main_handler(event, context):
    url = "http://www.anycodes.cn"
    if getStatusCode(url) == 200:
        print(" 您的網站 %s 可以訪問！" % (url))
    else:
        sendEmail(" 您的網站 %s 不可以訪問！" % (url), " 接受人郵箱地址 ")
    return None

通過 ServerlessFramework 可以部署，在部署的時候可以增加時間觸發器：

MyWebMonitor:
  component: "@serverless/tencent-scf"
  inputs:
    name: MyWebMonitor
    codeUri: ./code
    handler: index.main_handler
    runtime: Python3.6
    region: ap-guangzhou
    description: 網站監控
    memorySize: 64
    timeout: 20
    events:
      - timer:
          name: timer
          parameters:
            cronExpression: '*/5 * * * *'
            enable: true

在這里，timer 表示時間觸發器，cronExpression是運算式：

創建定時觸發器時，用戶能夠使用標準的 Cron 運算式的形式自定義何時觸發，定時觸發器現已推出秒級觸發功能，為了兼容老的定時觸發器，因此 Cron 運算式有兩種寫法，

Cron 運算式語法一（推薦）

Cron 運算式有七個必需欄位，按空格分隔，

其中，每個欄位都有相應的取值范圍：

Cron 運算式語法二（不推薦）

Cron 運算式有五個必需欄位，按空格分隔，

其中，每個欄位都有相應的取值范圍：

通配符

注意事項

在 Cron 運算式中的“日”和“星期”欄位同時指定值時，兩者為“或”關系，即兩者的條件分別均生效，

示例

*/5 * * * * * * 表示每 5 秒觸發一次
0 0 2 1 * * * 表示在每月的 1 日的凌晨 2 點觸發
0 15 10 * * MON-FRI * 表示在周一到周五每天上午 10：15 觸發
0 0 10,14,16 * * * * 表示在每天上午 10 點，下午 2 點，4 點觸發
0 */30 9-17 * * * * 表示在每天上午 9 點到下午 5 點內每半小時觸發
0 0 12 * * WED * 表示在每個星期三中午 12 點觸發

因此，我們上面的代碼可以認為是每 5 秒觸發一次，當然，也可以根據網站監控密度，自定義設定觸發的間隔時間，當我們網站服務不可用時，就可以收到告警：

這種網站監控方法比較簡單，準確度可能會有問題，對于網站或服務的監控不能簡單的看回傳值，還要看鏈接耗時、下載耗時以及不同區域、不同運營商訪問網站或者服務的延時資訊等，

所以，我們需要對這個代碼進行額外的更新與優化：

1. 通過在線網速測驗的網站，抓包獲取不同地區不同運營商的請求特征；
2. 撰寫爬蟲程式，進行在線網速測驗模塊的撰寫；
3. 集成到剛剛的專案中；

下面以站長工具網站中國內網站測速工具 為例，通過網頁查閱相關資訊，

對網站測速工具進行封裝，例如：

通過對網頁進行分析，獲取請求特征，包括 Url，Form data，以及 Headers 等相關資訊，其中該網站在使用不同監測點對網站進行請求時，是通過 Form data 中的 guid 的引數實作的，例如部分監測點的 guid：

廣東佛山	電信	f403cdf2-27f8-4ccd-8f22-6f5a28a01309
江蘇宿遷	多線	74cb6a5c-b044-49d0-abee-bf42beb6ae05
江蘇常州	移動	5074fb13-4ab9-4f0a-87d9-f8ae51cb81c5
浙江嘉興	聯通	ddfeba9f-a432-4b9a-b0a9-ef76e9499558

此時，我們可以撰寫基本的爬蟲代碼，來對 Response 進行初步決議，以62a55a0e-387e-4d87-bf69-5e0c9dd6b983 江蘇宿遷 [電信]為例，撰寫代碼：

import urllib.request
import urllib.parse
 
url = "* 某測速網站地址 *"
form_data = https://www.cnblogs.com/serverlesscloud/p/{'guid': '62a55a0e-387e-4d87-bf69-5e0c9dd6b983',
    'host': 'anycodes.cn',
    'ishost': '1',
    'encode': 'ECvBP9vjbuXRi0CVhnXAbufDNPDryYzO',
    'checktype': '1',
}
headers = {
    'Host': 'tool.chinaz.com',
    'Origin': '* 某測速網站地址 *',
    'Referer': '* 某測速網站地址 *',
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.108 Safari/537.36',
    'X-Requested-With': 'XMLHttpRequest'
}
 
print(urllib.request.urlopen(
    urllib.request.Request(
        url=url,
        data=https://www.cnblogs.com/serverlesscloud/p/urllib.parse.urlencode(form_data).encode('utf-8'),
        headers=headers
    )
).read().decode("utf-8"))
 

獲得結果：

({
	state: 1,
	msg: '',
	result: {
		ip: '119.28.190.46',
		httpstate: 200,
		alltime: '212',
		dnstime: '18',
		conntime: '116',
		downtime: '78',
		filesize: '-',
		downspeed: '4.72',
		ipaddress: '新加坡新加坡',
		headers: 'HTTP/1.1 200 OK br>Server: ...',
		pagehtml: ''
	}
})
 

在這個結果中，我們可以提取部分資料，例如江蘇宿遷 [電信] 訪問目標網站的基礎資料：

總耗時：alltime:'212'
鏈接耗時：conntime:'116'
下載耗時：downtime:'78'
 

此時，我們可以改造代碼對更多的節點，進行測驗：

江蘇宿遷 [電信]	總耗時:223	鏈接耗時:121	下載耗時:81
廣東佛山 [電信]	總耗時:44	鏈接耗時:27	下載耗時:17
廣東惠州 [電信]	總耗時:56	鏈接耗時:34	下載耗時:22
廣東深圳 [電信]	總耗時:149	鏈接耗時:36	下載耗時:25
浙江湖州 [電信]	總耗時:3190	鏈接耗時:3115	下載耗時:75
遼寧大連 [電信]	總耗時:468	鏈接耗時:255	下載耗時:170
江蘇泰州 [電信]	總耗時:180	鏈接耗時:104	下載耗時:69
安徽合肥 [電信]	總耗時:196	鏈接耗時:110	下載耗時:73
...
 

并對專案中的 index.py 進行代碼修改：

# -*- coding: utf8 -*-
import ssl
import json
import re
import socket
import smtplib
import urllib.request
from email.mime.text import MIMEText
from email.header import Header
 
socket.setdefaulttimeout(2.5)
ssl._create_default_https_context = ssl._create_unverified_context
 
def getWebTime():
 
    final_list = []
    final_status = True
 
    total_list = '''62a55a0e-387e-4d87-bf69-5e0c9dd6b983 江蘇宿遷 [電信]
    f403cdf2-27f8-4ccd-8f22-6f5a28a01309 廣東佛山 [電信]
    5bea1430-f7c2-4146-88f4-17a7dc73a953 河南新鄉 [多線]
    1f430ff0-eae9-413a-af2a-1c2a8986cff0 河南新鄉 [多線]
    ea551b59-2609-4ab4-89bc-14b2080f501a 河南新鄉 [多線]
    2805fa9f-05ea-46bc-8ac0-1769b782bf52 黑龍江哈爾濱 [聯通]
    722e28ca-dd02-4ccd-a134-f9d4218505a5 廣東深圳 [移動]
8e7a403c-d998-4efa-b3d1-b67c0dfabc41 廣東深圳 [移動]'''
 
    url = "* 某測速網站地址 *"
    for eve in total_list.split('\n'):
        id_data, node_name = eve.strip().split(" ")
        form_data = https://www.cnblogs.com/serverlesscloud/p/{'guid': id_data,
            'host': 'anycodes.cn',
            'ishost': '1',
            'encode': 'ECvBP9vjbuXRi0CVhnXAbufDNPDryYzO',
            'checktype': '1',
        }
        headers = {
            'Host': '* 某測速網站地址 *',
            'Origin': '* 某測速網站地址 *',
            'Referer': '* 某測速網站地址 *',
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.108 Safari/537.36',
            'X-Requested-With': 'XMLHttpRequest'
        }
        try:
            result_data = urllib.request.urlopen(
                urllib.request.Request(
                    url=url,
                    data=https://www.cnblogs.com/serverlesscloud/p/urllib.parse.urlencode(form_data).encode('utf-8'),
                    headers=headers
                )
            ).read().decode("utf-8")
            try:
                alltime = re.findall("alltime:'(.*?)'", result_data)[0]
                conntime = re.findall("conntime:'(.*?)'", result_data)[0]
                downtime = re.findall("downtime:'(.*?)'", result_data)[0]
                final_string = "%s\t 總耗時:%s\t 鏈接耗時:%s\t 下載耗時:%s" % (node_name, alltime, conntime, downtime)
            except:
                final_string = "%s 鏈接例外！" % (node_name)
                final_status = False
        except:
            final_string = "%s 鏈接超時！" % (node_name)
            final_status = False
        final_list.append(final_string)
        print(final_string)
    return (final_status,final_list)
def sendEmail(content, to_user):
    sender = '[email protected]'
    receivers = [to_user]
    mail_msg = content
    message = MIMEText(mail_msg, 'html', 'utf-8')
    message['From'] = Header(" 網站監控 ", 'utf-8')
    message['To'] = Header(" 站長 ", 'utf-8')
    subject = " 網站監控告警 "
    message['Subject'] = Header(subject, 'utf-8')
    try:
        smtpObj = smtplib.SMTP_SSL("smtp.exmail.qq.com", 465)
        smtpObj.login('[email protected]', '密碼')
        smtpObj.sendmail(sender, receivers, message.as_string())
    except smtplib.SMTPException:
        pass
 
def getStatusCode(url):
    return urllib.request.urlopen(url).getcode()
 
def main_handler(event, context):
    url = "http://www.anycodes.cn"
    final_status,final_list = getWebTime()
    if not final_status:
        sendEmail(" 您的網站 %s 的狀態：<br>%s" % (url, "<br>".join(final_list)), "[email protected]")
 

由于本文是以學習為主，所以我們將節點串列進行縮減，只保留幾個，通過部署，可得到結果：

告警的靈敏度和監控的頻率，在實際生產程序中可以根據自己的需求進行調整，

云服務監控告警

前文，我們對網站狀態以及健康等資訊進行了監控與告警，在實際的生產運維中，還需要對服務進行監控，例如在使用 Hadoop、Spark 的時候對節點的健康進行監控，在使用 K8S 的時候對 API 網關、ETCD 等多維度的指標進行監控，在使用 Kafka 的時候，對資料積壓量，以及 Topic、Consumer 等進行監控…

而這些服務的監控，往往不能通過簡單的 URL 以及某些狀態來進行判斷，傳統運維的做法是在額外的機器上設定一個定時任務，對相關的服務進行旁路監控，而在本文中，我們則通過 Serverless 技術，對云產品進行相關的監控與告警，

在使用云上的 Kafka 時，我們通常要看資料積壓量，因為如果 Consumer 集群掛掉了，或者消費能力突然降低導致資料積壓，很可能會對服務產生不可預估的影響，這個時候對 Kafka 的資料積壓量進行監控告警，就顯得額外重要，

本文以監控騰訊云的 Ckafka 為例進行實踐，并通過多個云產品進行組合（包括云監控、Ckafka、云 API 以及云短信等）來實作短信告警、郵件告警以及企業微信告警功能，

首先，可以設計簡單的流程圖：

在開始專案之前，我們要準備一些基礎的模塊：

• Kafka 資料積壓量獲取模塊：

def GetSignature(param):
    # 公共引數
    param["SecretId"] = ""
    param["Timestamp"] = int(time.time())
    param["Nonce"] = random.randint(1, sys.maxsize)
    param["Region"] = "ap-guangzhou"
    # param["SignatureMethod"] = "HmacSHA256"
    # 生成待簽名字串
    sign_str = "GETckafka.api.qcloud.com/v2/index.php?"
    sign_str += "&".join("%s=%s" % (k, param[k]) for k in sorted(param))
    # 生成簽名
    secret_key = ""
    if sys.version_info[0] > 2:
        sign_str = bytes(sign_str, "utf-8")
        secret_key = bytes(secret_key, "utf-8")
    hashed = hmac.new(secret_key, sign_str, hashlib.sha1)
    signature = binascii.b2a_base64(hashed.digest())[:-1]
    if sys.version_info[0] > 2:
        signature = signature.decode()
    # 簽名串編碼
    signature = urllib.parse.quote(signature)
    return signature
 
def GetGroupOffsets(max_lag, phoneList):
    param = {}
    param["Action"] = "GetGroupOffsets"
    param["instanceId"] = ""
    param["group"] = ""
    signature = GetSignature(param)
    # 生成請求地址
    param["Signature"] = signature
    url = "https://ckafka.api.qcloud.com/v2/index.php?Action=GetGroupOffsets&"
    url += "&".join("%s=%s" % (k, param[k]) for k in sorted(param))
    req_attr = urllib.request.urlopen(url)
    res_data = https://www.cnblogs.com/serverlesscloud/p/req_attr.read().decode("utf-8")
    json_data = json.loads(res_data)
    for eve_topic in json_data['data']['topicList']:
        temp_lag = 0
        result_list = []
        for eve_partition in eve_topic["partitions"]:
            lag = eve_partition["lag"]
            temp_lag = temp_lag + lag
        if temp_lag > max_lag:
            result_list.append(
                {
                    "topic": eve_topic["topic"],
                    "lag": lag
                }
            )
        print(result_list)
        if len(result_list)>0:
            KafkaLagRobot(result_list)
            KafkaLagSMS(result_list,phoneList)
 

• 接入企業微信機器人模塊：

def KafkaLagRobot(content):
    url = ""
    data = https://www.cnblogs.com/serverlesscloud/p/{
        "msgtype": "markdown",
        "markdown": {
            "content": content,
        }
    }
    data = json.dumps(data).encode("utf-8")
    req_attr = urllib.request.Request(url, data)
    resp_attr = urllib.request.urlopen(req_attr)
    return_msg = resp_attr.read().decode("utf-8")
 

• 接入騰訊云短信服務模塊：

def KafkaLagSMS(infor, phone_list):
    url = ""
    strMobile = phone_list
    strAppKey = ""
    strRand = str(random.randint(1, sys.maxsize))
    strTime = int(time.time())
    strSign = "appkey=%s&random=%s&time=%s&mobile=%s" % (strAppKey, strRand, strTime, ",".join(strMobile))
    sig = hashlib.sha256()
    sig.update(strSign.encode("utf-8"))
 
    phone_dict = []
    for eve_phone in phone_list:
        phone_dict.append(
            {
                "mobile": eve_phone,
                "nationcode": "86"
            }
        )
    data = https://www.cnblogs.com/serverlesscloud/p/{
        "ext": "",
        "extend": "",
        "params": [
            infor,
        ],
        "sig": sig.hexdigest(),
        "sign": " 你的 sign",
        "tel": phone_dict,
        "time": strTime,
        "tpl_id": 你的模板 id
    }
    data = json.dumps(data).encode("utf-8")
    req_attr = urllib.request.Request(url=url, data=data)
    resp_attr = urllib.request.urlopen(req_attr)
    return_msg = resp_attr.read().decode("utf-8")
 

• 發送郵件告警模塊：

def sendEmail(content, to_user):
    sender = '[email protected]'
    message = MIMEText(content, 'html', 'utf-8')
    message['From'] = Header(" 監控 ", 'utf-8')
    message['To'] = Header(" 站長 ", 'utf-8')
    message['Subject'] = Header(" 告警 ", 'utf-8')
    try:
        smtpObj = smtplib.SMTP_SSL("smtp.exmail.qq.com", 465)
        smtpObj.login('[email protected]', '密碼')
        smtpObj.sendmail(sender, [to_user], message.as_string())
    except smtplib.SMTPException as e:
        logging.debug(e)
 

完成模塊撰寫，和上面的方法一樣，進行專案部署，部署成功之后進行測驗，測驗可看到功能可用：

• 短信告警樣式：

• 企業微信告警樣式：

總結

通過本文的實踐，希望讀者可以了解到 Serverless 相關產品在運維行業中的基本應用，尤其是監控告警的基本使用方法和初步靈感，設計一個網站監控程式實際上是一個很初級的入門場景，希望大家可以將更多的監控告警功與 Serverless 技術進行結合，例如監控自己的 MySQL 壓力情況、監控已有服務器的資料指標等，通過對這些指標的監控告警，不僅僅可以讓管理者及時發現服務的潛在風險，也可以通過一些自動化流程實作專案的自動化運維，

通過本場景實踐，我們也可以對專案進行額外的優化或者應用在不同的領域以及場景中，例如，我們可以通過增加短信告警、微信告警、企業微信告警等多個維度，來確保相關人員可以及時收到告警資訊；我們也可以通過監控某個小說網站、視頻網站等，看到我們關注的小說或者視頻的更新情況，便于追更等，

傳送門：

• GitHub: github.com/serverless
• 官網：serverless.com

歡迎訪問：Serverless 中文網，您可以在 最佳實踐 里體驗更多關于 Serverless 應用的開發！

推薦閱讀：《Serverless 架構：從原理、設計到專案實戰》

標籤：其他

上一篇：請問怎么解決啊

下一篇：bili云開發講解