前幾天簡書崩潰的事情大家還記得么？

現(xiàn)在數(shù)據(jù)已經(jīng)恢復(fù)了，大家不必?fù)?dān)心，這件事雖然處理慢了點(diǎn)，但終歸是沒有造成什么嚴(yán)重后果。

這件事讓我熬了個(gè)夜，跑了無數(shù)次單元測試，回復(fù)了近百條用戶反饋，更新兩個(gè)公告，同時(shí)還要對自己的服務(wù)進(jìn)行停運(yùn)和降級處理。

我發(fā)現(xiàn)這個(gè)故障，是在守聯(lián)的官方群里，而證實(shí)的方法不只是自己打開簡書 App 看看，而是需要連上服務(wù)器跑 JRT 的單元測試，因?yàn)楣收峡赡苤淮嬖谟谑謾C(jī)端，也可能軟件并沒有報(bào)錯，但數(shù)據(jù)有問題。

但除了用戶反饋，是不是還有其它方式，可以更快發(fā)現(xiàn)故障呢？

用戶反饋這一渠道，我們是被動方，而我們可以通過每隔一段時(shí)間主動嘗試的方式，進(jìn)行主動監(jiān)控。

國內(nèi)外有很多相關(guān)的服務(wù)，但大多數(shù)是收費(fèi)的，而且可定制性不高。我找了一圈，GitHub 上并沒有符合我心意、足夠輕量級的服務(wù)監(jiān)控框架。

軟件開發(fā)行業(yè)有句老話：不要重復(fù)發(fā)明輪子，意思是有現(xiàn)成的東西就要加以利用，不要自己實(shí)現(xiàn)一套一樣的。

但是既然在這個(gè)領(lǐng)域還沒有輪子，那咱們就自己造！

需求分析與設(shè)計(jì)

簡單梳理了一下，這個(gè)系統(tǒng)要實(shí)現(xiàn)以下功能：

• 管理監(jiān)控任務(wù)，按照一定規(guī)則定時(shí)執(zhí)行
• 處理各種成功和失敗的情況
• 執(zhí)行日志需要存檔
• 發(fā)生異常時(shí)，將信息推送到手機(jī)上
• 有一個(gè)網(wǎng)頁服務(wù)，可以讓用戶查看各監(jiān)控任務(wù)的狀態(tài)

第五個(gè)的實(shí)現(xiàn)方式與今天的主題關(guān)系不大，暫且不提。

監(jiān)控任務(wù)是多樣的，有些任務(wù)要訪問網(wǎng)址，根據(jù)錯誤碼進(jìn)行服務(wù)狀態(tài)的判斷，有些任務(wù)要尋找特定網(wǎng)頁元素，還有些要判斷一個(gè)值是否在區(qū)間內(nèi)，我們不可能逐個(gè)定義好這些任務(wù)類型，否則程序的后期維護(hù)將會很繁瑣。

我們使用函數(shù)來封裝每個(gè)監(jiān)控任務(wù)。由于所有監(jiān)控函數(shù)中都沒有需要動態(tài)變化的規(guī)則，所以我們不需要向函數(shù)傳入?yún)?shù)。

函數(shù)的內(nèi)部邏輯可能是多種多樣的，但可以分為成功和失敗兩種類型，成功只有一種可能性，而失敗有很多種，所以我們引入了錯誤碼機(jī)制。為了便于分析，我們引入一個(gè)可選的消息字段。

我們定義的函數(shù)返回值如下：

• success(bool): 是否成功
• status_code(int): 狀態(tài)碼
  • -2：執(zhí)行過程中的未捕獲異常
  • -1：通用異常
  • 0：執(zhí)行成功
  • 大于 0 的值：HTTP 錯誤碼
• message(str, optional): 附加信息

提到定時(shí)執(zhí)行，我們第一個(gè)想到的就是 Python 開發(fā)中非常常用的 apscheduler 庫，它可以幫助我們處理復(fù)雜的調(diào)度邏輯，同時(shí)保持簡潔與高度可定制化。

這里需要提到它的“事件”機(jī)制，簡單來說，它允許我們設(shè)定一個(gè)函數(shù)，在特定事件發(fā)生時(shí)自動調(diào)用這個(gè)函數(shù)，并將事件的詳細(xì)信息作為參數(shù)傳入。

為了實(shí)現(xiàn)各種執(zhí)行結(jié)果的處理，我們使用它提供的以下兩個(gè)事件：

• EVENT_JOB_EXECUTED：任務(wù)執(zhí)行成功事件
• EVENT_JOB_ERROR：任務(wù)執(zhí)行出錯事件

執(zhí)行日志存檔這個(gè)需求，我們第一個(gè)想到的就是將日志存入 txt 文件，但隨著數(shù)據(jù)的不斷增加，日志文件將越來越大，讀取速度也會受到影響。在大量結(jié)構(gòu)化數(shù)據(jù)的處理中，txt 文件并不是值得推薦的解決方案。

提到大量結(jié)構(gòu)化數(shù)據(jù)，對信息技術(shù)稍有了解的小伙伴可能已經(jīng)想到了：我們可以用數(shù)據(jù)庫存儲它們。

為了保證程序的簡潔，我們使用 SQLite 數(shù)據(jù)庫。這個(gè)名字大家可能有些陌生，但它是世界上裝機(jī)量最大的數(shù)據(jù)庫，微信的本機(jī)聊天記錄就存放在這個(gè)數(shù)據(jù)庫里。

然后呢？我們需要手寫 SQL 語句來實(shí)現(xiàn)存儲？這樣未免有些太麻煩了，而且會帶來一些不必要的安全問題。

好在我們有 ORM，全稱 Object Relational Mapping，對象關(guān)系映射。

簡單來說，它可以幫助我們把數(shù)據(jù)庫變成 Python 中的對象，我們可以通過修改對象屬性的方式，對數(shù)據(jù)庫進(jìn)行操作，而 SQL 的拼接與執(zhí)行則隱藏在內(nèi)部，不需要我們過多干預(yù)。

我們使用輕量級 ORM 框架 Peewee。

最后是消息推送。大家想到的可能是推送到微信上，但微信官方并沒有對個(gè)人賬號提供這一接口，而企業(yè)微信注冊流程較為復(fù)雜，Server 醬這類服務(wù)需要付出額外的成本，雖然不高，但體驗(yàn)終究不算良好。

我之前寫過一篇文章：用飛書實(shí)現(xiàn)腳本運(yùn)行狀態(tài)推送，所以，飛書，今天我又來了。

實(shí)現(xiàn)思路和技術(shù)選型完成，可以開始寫代碼了。

監(jiān)控函數(shù)注冊

打開編輯器，新建一個(gè)文件夾，git init 初始化存儲庫，然后新建 .gitignore 文件，讓 Git 忽略部分文件，不將其添加到版本庫中。

除了加入各類緩存的目錄外，我們還需要將 *.db 和 *.yaml 這兩項(xiàng)加入排除列表。數(shù)據(jù)庫文件很好理解，無用數(shù)據(jù)提交上去會污染版本庫，YAML 則是一種配置文件格式，我們要在其中填入自己的飛書機(jī)器人鑒權(quán)信息，如果提交到了版本庫上會存在安全風(fēng)險(xiǎn)。

軟件開發(fā)安全守則指出，嚴(yán)禁將鑒權(quán)信息寫入代碼，嚴(yán)格做好數(shù)據(jù)與代碼的分離，我們在開發(fā)過程中，應(yīng)時(shí)刻謹(jǐn)記這一原則。

我們使用設(shè)計(jì)模式中的策略模式實(shí)現(xiàn)監(jiān)控函數(shù)的注冊：

策略模式，指對象有某個(gè)行為，但是在不同的場景中，該行為有不同的實(shí)現(xiàn)算法。

這里，我們使用 Python 的裝飾器對其稍加修改，其實(shí)現(xiàn)思路來源于《流暢的 Python》一書，如果大家有基礎(chǔ)的 Python 程序開發(fā)經(jīng)驗(yàn)，可以找來細(xì)細(xì)閱讀。

注冊監(jiān)控函數(shù)的裝飾器代碼如下：

def register_task(task_name: str = None, run_cron: str = None) -> Callable:
    """將一個(gè)函數(shù)注冊為監(jiān)控任務(wù)

    Args:
        func (Callable): 監(jiān)控任務(wù)函數(shù)
        task_name (str): 監(jiān)控任務(wù)名稱, Defaults to None.
        run_cron (str): 監(jiān)控任務(wù)運(yùn)行的 cron 表達(dá)式, Defaults to None.
    """
    if not run_cron:
        raise ValueError("run_cron 不能為空")

    def outer(func: Callable):
        @wraps(func)
        def inner(task_name: str = None, run_cron: str = None):
            if task_name:
                registered_funcs[task_name] = [func, run_cron]
            else:
                registered_funcs[func.__name__] = [func, run_cron]
            return func
        return inner(task_name, run_cron)
    return outer

這一裝飾器函數(shù)將會接收兩個(gè)參數(shù)：監(jiān)控任務(wù)的名稱和任務(wù)運(yùn)行的 cron 表達(dá)式，并將它們存入字典 registered_funcs 中。

cron 是 Linux 下的一種計(jì)劃任務(wù)語法，可以用簡單的字符串表示定時(shí)執(zhí)行規(guī)則。

我們可以這樣使用這個(gè)裝飾器函數(shù)。

@register_task(name="測試監(jiān)控任務(wù)", cron="0 1-59 * * * *")
def test_task():
    pass

注冊這一函數(shù)后，registered_funcs 的值如下：

{'測試監(jiān)控任務(wù)': [<function test_task at 0x000001C04B677700>, '0 1-59 * * * *']}

這里的 cron 表達(dá)式代表每分鐘的第一秒執(zhí)行一次。

這里給出一個(gè)簡書主站的監(jiān)控任務(wù)實(shí)現(xiàn)：

@register_task("簡書主站監(jiān)控", "0 1-59 * * * *")  # 每分鐘執(zhí)行一次
def JianshuMainPageMonitor():
    try:
        response = httpx_get("http://www.itdecent.cn/")
    except Exception as e:
        success = False
        status_code = -1
        message = str(e)
    else:
        message = ""
        status_code = response.status_code
        if status_code != 200:
            success = False
        else:
            success = True
    return (success, status_code, message)

可以看到，我們的函數(shù)返回上文定義的三個(gè)字段。

日志存儲

我們需要用定義 Python 類的方式定義數(shù)據(jù)庫結(jié)構(gòu)。

運(yùn)行日志：

class RunLog(Model):
    id = IntegerField(primary_key=True)
    time = DateTimeField()
    level = IntegerField()
    message = CharField()

    class Meta:
        database = SqliteDatabase("log.db")

監(jiān)控日志：

class MonitorLog(Model):
    id = IntegerField(primary_key=True)
    time = DateTimeField()
    monitor_name = CharField()
    successed = BooleanField()
    status_code = IntegerField()
    message = CharField()

    class Meta:
        database = SqliteDatabase("log.db")

這里的各種字段，都是 Peewee 提供的對象，它們會在數(shù)據(jù)庫表創(chuàng)建時(shí)對應(yīng)到不同的數(shù)據(jù)類型。

我們將兩類日志存放在同一個(gè)數(shù)據(jù)庫文件 log.db 中，便于統(tǒng)一管理。程序中對數(shù)據(jù)庫的查詢不多，性能需求不敏感，故不需要新建索引。

最后，我們編寫一個(gè)函數(shù)，初始化數(shù)據(jù)庫并創(chuàng)建對應(yīng)的表：

def init_db() -> None:
    """初始化數(shù)據(jù)庫"""
    RunLog.create_table()
    MonitorLog.create_table()

數(shù)據(jù)庫有了，但為了后期維護(hù)考慮，我們需要封裝一下，將高級接口暴露給程序的其它部分使用。

先定義數(shù)字與日志等級對應(yīng)如下：

• 0：CRITICAL
• 1：ERROR
• 2：WARNING
• 3：INFO
• 4：DEBUG

封裝部分代碼省略，注意將錯誤捕獲并寫入運(yùn)行日志。

配置文件管理

我們使用 YAML 格式的配置文件，這是一種比 Json 更簡介易讀的格式，PyYAML 庫實(shí)現(xiàn)了對它的讀寫操作。

目前我們要配置的字段不多，先定義默認(rèn)配置文件，Python 命名規(guī)范中，常量使用全大寫字母，下劃線分隔：

DEFALUT_CONFIG = {
    "message_service": {
        "app_id": "",
        "app_secret": "",
        "email": ""
    }
}

讀寫配置文件的函數(shù)如下：

def CreateDefaultConfig() -> None:
    with open("config.yaml", "w", encoding="utf-8") as f:
        dump(DEFALUT_CONFIG, f, indent=4, allow_unicode=True)


def GetConfig() -> Dict:
    try:
        with open("config.yaml", "r", encoding="utf-8") as f:
            return load(f, SafeLoader)
    except FileNotFoundError:  # 配置文件不存在
        CreateDefaultConfig()
        return GetConfig()

這里我們對配置文件不存在的處理方式是使用默認(rèn)值創(chuàng)建，然后返回其內(nèi)容，實(shí)際情況下可以改為文件不存在時(shí)報(bào)錯，將第一次運(yùn)行時(shí)無配置文件的處理邏輯交給主程序?qū)崿F(xiàn)。

消息推送

申請飛書開發(fā)者賬號、新建應(yīng)用、授權(quán)與獲取鑒權(quán)的操作方式已經(jīng)在用飛書實(shí)現(xiàn)腳本運(yùn)行狀態(tài)推送這篇文章中寫過，這里不再贅述。

我們從原先的項(xiàng)目中將請求飛書 API 相關(guān)的代碼復(fù)制過來，并稍作修改：

def SendFeishuMessage(app_id: str, app_secret: str, email: str, message: str) -> None:
    headers = {"Content-Type": "application/json; charset=utf-8"}
    data_to_get_token = {"app_id": app_id, "app_secret": app_secret}
    headers["Authorization"] = "Bearer " + httpx_post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", headers=headers, json=data_to_get_token).json()["tenant_access_token"]

    data_to_send_message = {"email": email, "msg_type": "text", "content": {"text": message}}
    response = httpx_post("https://open.feishu.cn/open-apis/message/v4/send/", headers=headers, json=data_to_send_message)
    if response.json()["code"] != 0:
        AddRunLog(1, f"向{email}發(fā)送飛書消息失敗，錯誤碼：{response.json()['code']}")
    else:
        AddRunLog(3, f"向{email}發(fā)送飛書消息成功，消息代碼：{response.json()['data']['message_id']}")

然后使用類似的方式，通過更改 data_to_send_message 的方式自定義消息的格式，實(shí)現(xiàn)以下三種消息類型的封裝：

• 服務(wù)告警消息：紅色，可傳入任務(wù)名稱、錯誤碼、附加信息
• 服務(wù)恢復(fù)消息：綠色，可傳入任務(wù)名稱
• 系統(tǒng)消息：橘色，可傳入消息內(nèi)容

這三個(gè)函數(shù)都有三個(gè)相同的參數(shù)：

• app_id：鑒權(quán)信息，相當(dāng)于賬號
• app_secret：鑒權(quán)信息，相當(dāng)于密碼
• email：目標(biāo)用戶的電子郵件地址

我們對配置文件管理模塊稍加改動，生成空白的配置文件，然后將信息填入其中：

message_service:
    app_id: 'cli_a157**********0d'
    app_secret: 'TSll************************ituf'
    email: 'ye************11@qq.com'

事件處理

上文已經(jīng)提到，我們需要用到 apscheduler 的兩個(gè)事件，分別是任務(wù)執(zhí)行成功和任務(wù)執(zhí)行失敗。

但注意，任務(wù)執(zhí)行失敗只有可能是出現(xiàn)了未捕獲的異常，但執(zhí)行成功有多種情況，有可能因?yàn)殄e誤被捕獲，或者數(shù)據(jù)超出范圍造成失敗，但在這些情況下，該任務(wù)在調(diào)度器看來依然是“執(zhí)行成功”。

當(dāng)然，也可以通過更改代碼邏輯的方式，使任務(wù)在滿足一定條件時(shí)主動拋出異常，兩者孰優(yōu)孰劣可以自行定奪。

我們定義的事件處理函數(shù)需要接收一個(gè)參數(shù)：event，這是 apscheduler 中對事件信息的封裝。

首先，我們定義一個(gè)任務(wù)運(yùn)行成功事件的處理函數(shù)：

def JobExecutedSuccessfully(event) -> None:

查閱文檔，找到 event 對象的可用屬性，其中有我們需要的內(nèi)容：

• event.job_id：監(jiān)控任務(wù)名
• event.retval：任務(wù)函數(shù)返回值

使用 Python 的元組解包特性賦值相關(guān)變量，代碼如下：

success, status_code, message = event.retval

之后，我們需要判斷任務(wù)是否成功，如果不成功，就在終端輸出相應(yīng)提示，并推送告警信息，最后將相關(guān)數(shù)據(jù)存入數(shù)據(jù)庫。

但是，我們細(xì)細(xì)一想，會發(fā)現(xiàn)一個(gè)問題：我們的程序會每分鐘監(jiān)測一次簡書主頁的狀態(tài)，如果簡書崩潰 60 分鐘，我們就會收到 60 條相同的告警信息，這顯然是沒有必要的，而且如此頻繁地調(diào)用飛書的 API 可能導(dǎo)致訪問超限，繼而影響其它告警信息的發(fā)送。

所以我們需要一種方式，對該任務(wù)的歷史狀態(tài)進(jìn)行檢測，判斷其告警信息是否已經(jīng)發(fā)送過。

由于我們已經(jīng)設(shè)計(jì)過，每次任務(wù)執(zhí)行時(shí)，都會將數(shù)據(jù)入庫，所以我們只需以任務(wù)名稱作為篩選條件，獲取其上一次運(yùn)行的結(jié)果即可，如果上一次運(yùn)行未能成功，則告警信息已經(jīng)發(fā)送過。

在日志服務(wù)模塊中添加一個(gè)函數(shù)，實(shí)現(xiàn)這一功能：

def IsFailedUntilNow(monitor_name: str) -> bool:
    try:
        last_log = MonitorLog.select().where(MonitorLog.monitor_name == monitor_name).order_by(MonitorLog.id.desc()).get()
        return not last_log.successed
    except Exception as e:
        AddRunLog(level=1, message=f"查詢監(jiān)控日志失?。簕e}")
        return True  # 無法判斷時(shí)假設(shè)任務(wù)一直處于失敗狀態(tài)

這時(shí)，使用 ORM 框架的好處就體現(xiàn)出來了，我們可以很方便地指定篩選條件和排序規(guī)則。

注意，在事件處理函數(shù)中，這一查詢應(yīng)先于本次任務(wù)結(jié)果的入庫執(zhí)行，否則本次任務(wù)的結(jié)果將會作為函數(shù)的判斷依據(jù)，導(dǎo)致這一邏輯失效。

回到事件處理函數(shù)，我們使用這一函數(shù)判斷告警信息是否發(fā)送過，然后根據(jù)不同狀態(tài)執(zhí)行對應(yīng)邏輯：

• 任務(wù)失敗，告警信息沒有發(fā)送過：發(fā)送告警信息
• 任務(wù)失敗，告警信息已經(jīng)發(fā)送過：什么都不做
• 任務(wù)成功，該任務(wù)上一次運(yùn)行失?。喊l(fā)送服務(wù)恢復(fù)信息
• 任務(wù)成功，該任務(wù)上一次運(yùn)行成功：什么都不做

為了便于調(diào)試，不要忘記在每個(gè)分支中加上合適等級的運(yùn)行日志記錄。

用相同的方式封裝好任務(wù)失敗事件的處理函數(shù)，這一函數(shù)的邏輯較為簡單，只需要判斷并執(zhí)行上述邏輯的前兩條即可。

程序主邏輯

對程序開發(fā)有所了解的小伙伴可能已經(jīng)發(fā)現(xiàn)了：我們使用的是自底向上的開發(fā)方式。

其實(shí)，我們也可以先編寫好空函數(shù)，使用高層函數(shù)實(shí)現(xiàn)主要邏輯，然后逐個(gè)實(shí)現(xiàn)函數(shù)內(nèi)容。

主程序啟動時(shí)，先導(dǎo)入第三方庫和我們自己編寫的模塊：

from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_EXECUTED
from apscheduler.schedulers.background import BackgroundScheduler

from config_service import GetConfig
from event_handlers import JobExecutedFailure, JobExecutedSuccessfully
from log_service import AddRunLog
from message_service import SendFeishuSystemMessage
from monitors import init_monitors
from register import get_registered_funcs_info
from utils import CronToKwargs

為了保證導(dǎo)入語句符合規(guī)范，這里日志模塊不是最先導(dǎo)入的，但如果其它模塊的初始化過程中可能發(fā)生異常，建議變更導(dǎo)入順序，將日志模塊的導(dǎo)入提前，避免發(fā)生錯誤時(shí)無法記錄日志。

之后，我們需要校驗(yàn)配置文件是否完整填寫：

if not all((GetConfig()["message_service"]["app_id"],
           GetConfig()["message_service"]["app_secret"],
           GetConfig()["message_service"]["email"])):
    AddRunLog(1, "消息服務(wù)配置不完整")
    print("消息服務(wù)配置不完整，請檢查配置文件")
    AddRunLog(0, "程序退出，原因：消息服務(wù)配置不完整")
    exit(1)

其實(shí)配置文件的讀取其實(shí)還有待優(yōu)化，以現(xiàn)有的邏輯，每調(diào)用一次這個(gè)函數(shù)，就會對磁盤進(jìn)行一次訪問，會拖慢程序的速度，但考慮到在程序中配置文件的讀取不算頻繁，而且服務(wù)器上使用的是固態(tài)硬盤，隨機(jī)讀取性能尚可，現(xiàn)階段不需要優(yōu)化。

使用一行代碼來初始化調(diào)度器：

scheduler = BackgroundScheduler()

AddRunLog(4, "初始化調(diào)度器成功")

然后用我們封裝好的接口獲取已注冊的函數(shù)，遍歷數(shù)據(jù)，進(jìn)行任務(wù)注冊：

funcs_to_schedule = get_registered_funcs_info()

AddRunLog(3, f"已注冊的監(jiān)控任務(wù)數(shù)量：{len(funcs_to_schedule)}")

AddRunLog(4, "開始添加監(jiān)控任務(wù)")
for task_name, (func, run_cron) in funcs_to_schedule.items():
    scheduler.add_job(func, "cron", **CronToKwargs(run_cron), id=task_name)
    AddRunLog(4, f"添加監(jiān)控任務(wù)：{task_name} 成功，運(yùn)行 cron 表達(dá)式：{run_cron}")

AddRunLog(4, "監(jiān)控任務(wù)添加完成")

這里引入了一個(gè)函數(shù) CronToKwargs，這一函數(shù)來自于我們編寫的 utils 模塊，作用為將任務(wù)注冊裝飾器中的 cron 表達(dá)式進(jìn)行拆分，返回一個(gè)字典，我們在此處使用字典解包將其作為參數(shù)傳入 scheduler.add_job() 方法。

然后，注冊我們的事件處理函數(shù)：

AddRunLog(4, "開始注冊事件回調(diào)")
scheduler.add_listener(JobExecutedSuccessfully, EVENT_JOB_EXECUTED)
scheduler.add_listener(JobExecutedFailure, EVENT_JOB_ERROR)
AddRunLog(4, "事件回調(diào)注冊完成")

最后，啟動調(diào)度器，并發(fā)送系統(tǒng)消息：

scheduler.start()
AddRunLog(3, "調(diào)度器啟動成功")
SendFeishuSystemMessage(GetConfig()["message_service"]["app_id"],
                        GetConfig()["message_service"]["app_secret"],
                        GetConfig()["message_service"]["email"],
                        "調(diào)度器啟動成功")
print("調(diào)度器啟動成功")

最后，使用 while True 開始一個(gè)無限循環(huán)，使用 input() 函數(shù)阻塞獲取用戶輸入，并提供命令行管理能力：

• 輸出當(dāng)前注冊任務(wù)信息
• 停止調(diào)度器
• 強(qiáng)制停止調(diào)度器

Debug

運(yùn)行程序，通過填入錯誤監(jiān)控地址的方式使監(jiān)控任務(wù)失敗，發(fā)現(xiàn)雖然監(jiān)控日志中有相應(yīng)的記錄，但告警信息并沒有發(fā)送。

查看運(yùn)行日志，發(fā)現(xiàn)程序認(rèn)為告警信息已經(jīng)發(fā)送過，但事實(shí)并非如此。在相應(yīng)判斷邏輯位置打斷點(diǎn)，并在此處停止調(diào)度器避免重復(fù)運(yùn)行帶來的干擾。

單步運(yùn)行函數(shù)，可以發(fā)現(xiàn)，由于監(jiān)控?cái)?shù)據(jù)庫為空，程序在數(shù)據(jù)庫查詢中拋出了異常，異常被 except 語句捕獲，執(zhí)行了備用邏輯，導(dǎo)致告警信息發(fā)送被跳過。

在 IsFailedUntilNow 函數(shù)的 try 語句塊中加入如下代碼：

if MonitorLog.select().where(MonitorLog.monitor_name == monitor_name).count() == 0:
            return False  # 數(shù)據(jù)庫中沒有對應(yīng)的記錄，第一次運(yùn)行即出現(xiàn)錯誤

再次運(yùn)行，問題解決。

至此，程序編寫完成。來看看我們寫了多少行代碼：

代碼行數(shù)統(tǒng)計(jì)

我們只用了不到四百行代碼，就實(shí)現(xiàn)了一個(gè)服務(wù)監(jiān)控系統(tǒng)。

實(shí)戰(zhàn)

湊巧的是，就在我編寫好這個(gè)系統(tǒng)的幾小時(shí)后，簡書又一次出現(xiàn)故障。在本次故障的過程中，我收到的消息推送如下：

消息推送記錄

簡單梳理一下時(shí)間線：

• 00:31 調(diào)度器啟動成功：這一故障從 00:07 第一次被用戶發(fā)現(xiàn)，我啟動時(shí)距離故障發(fā)生已經(jīng)過去了一段時(shí)間

• 00:32 簡書主站監(jiān)控預(yù)警：調(diào)度器啟動后，在下一分鐘對簡書主站進(jìn)行了自動檢測，發(fā)現(xiàn)服務(wù)異常后推送了告警信息

• 00:41 調(diào)度器啟動成功：當(dāng)時(shí)的程序并沒有對簡書 API 進(jìn)行監(jiān)測的任務(wù)函數(shù)，我臨時(shí)編寫了一個(gè)，上傳到服務(wù)器上覆蓋之后，手動重啟了系統(tǒng)

• 00:42 簡書 Json API 監(jiān)控預(yù)警：Json API 的監(jiān)控任務(wù)也是一分鐘執(zhí)行一次

• 01:06 服務(wù)恢復(fù)：簡書服務(wù)恢復(fù)正常，系統(tǒng)第一時(shí)間檢測到并推送了信息

這里解釋一下流程中的一些問題：

為什么程序完成之后沒有立刻上線啟動運(yùn)行？

我本來打算完善幾個(gè)功能點(diǎn)，所以沒有直接上線。

為什么沒有調(diào)度器停止的信息？

因?yàn)槲覜]有使用常規(guī)的退出方式（命令行輸入退出指令），所以沒有發(fā)送退出信息。后續(xù)打算改成注冊退出事件的方式，保證程序無論因?yàn)楹畏N原因退出，均可以推送系統(tǒng)信息。

為什么系統(tǒng)重新上線之后簡書主站的監(jiān)控預(yù)警沒有再發(fā)送一遍？

因?yàn)閿?shù)據(jù)庫中還有上次運(yùn)行的失敗記錄，重復(fù)發(fā)送預(yù)警信息在這種場景下也沒有必要。

后續(xù)優(yōu)化

• 優(yōu)化獲取配置文件的邏輯，在保證配置能夠及時(shí)更新的前提下，降低磁盤讀取頻率，提高運(yùn)行速度
• 支持在不停止程序運(yùn)行的情況下動態(tài)增刪監(jiān)控任務(wù)
• 優(yōu)化程序退出邏輯
• 制作一個(gè)網(wǎng)頁實(shí)時(shí)顯示相關(guān)信息供查閱

后記

程序已經(jīng)以 MIT 協(xié)議開源：JianshuAvailabilityMonitor

希望大家學(xué)編程不是只學(xué)會語法和特性，而能夠?qū)⒕幊套鳛橐环N工具，真正應(yīng)用到特定場景中。