Evolution of the Crescendo Lab LINE messaging system

漸強實驗室的產品 MAAC (https://www.cresclab.com/)是個 LINE CRM 行銷工具，簡單來說，就是我們 host 客戶的 LINE bot ，讓他們的可以透過平台的操作，使 LINE bot 變成 CRM 的行銷工具，而其中有以下兩個主要的功能會需要訊息發送的功能：

1. 分眾推播：在平台上客製化廣告訊息，然後圈選受眾以及訂定時間來進行排程廣告推播
2. 自動回覆：一些 chatbot 的關鍵字即時回覆以及互動功能

而本篇主要內容是想要分享漸強在分眾推播的訊息系統架構上的一些經驗，

首先談談要實現這兩個功能的基礎，就是 LINE 的訊息推播 api (https://developers.line.biz/zh-hant/docs/messaging-api/sending-messages/#methods-of-sending-message)

分眾推播需要兩種 api：

• Push message (推送一則訊息給一個 LINE user)：用在分眾訊息發送給每人的訊息長相都不同的時候
• Multicast messages (推送一則訊息給可選的一群人)：用在分眾推播訊息都長一樣的時候，目前上限為 500 人/次

MAAC 在上線後短短的一年多的時間，我們擁有的 LINE user 人數從 300 萬爬升到 4000萬(2020/04)，而每月訊息推播的數量也從 200萬爬升到接近 6000萬次(2020/04)，在人數與訊息推播次數急遽的上升的情況之下，漸強先後為了效能以及成本，在分眾推播的架構做了兩次的調整。

原始架構：

簡單的把訊息丟到 queue 裡面，然後靠著 worker 去 queue 裡面拿取訊息做推播

最初，漸強的訊息發送系統，在 queue 的部分主要由搭建在 kubernetes 內的 3 node rabbitmq cluster 提供，而 worker 則由 python celery 實作，也是運行在同一個 kubernetes 叢集內——其實就是一個非常簡單的分散式任務隊列應用，所以工作內容也很簡單，透過排程的 worker 把訊息一個一個準備好後，丟到 rabbitmq 上某個放 message 的 queue，再由 listen 那個 queue 的 worker 把訊息拿下來發送。

然而，在客戶推播的量體越來越大的時候，我們遇到了第一個問題：量太多發太久了！ 你可以想像如果受眾是 100 萬個 LINE user ，當推送的訊息都一樣的時候 ，利用 multicast messages api 一次推播 500 人次 (當時上限是 150人/次)，總共只需要 2000 個 http requests 就可以搞定，但是如果推播訊息需要附上一些個人化資訊讓每一個訊息都變得不一樣，這時候問題就變得不有趣了，因為一次訊息推播需要打 100 萬個 http requests，我們當時慘痛的代價是一個客戶 60 萬則訊息，加上準備訊息的時間，一共發了 40 分鐘 😰。

雖然一開始靠著 kubernetes 的彈性，scale 更多 worker 來加快發送速度，但提升的速度仍然非常有限，加上 rabbitmq 雖作為一個 message broker 的服務，但他本身並不適合堆放太大量的訊息在 queue 上 (會導致效能降低)，亦或是害怕訊息堆放速度太快超過 queue 最大長度，而導致漏發訊息等等問題，於是促使了第一次的改進。

第一次改進：

不只利用異步的 queue + multiple workers，更是利用異步 I/O 的方法，提高單次 worker 推播的訊息數

簡單來說就是把一堆 single push 包成一包當成 multicast 來發送，因為根據遇到的問題，我們發現要打巨量的 push message api，雖然已經利用 queue 以及 worker 的方式可以提升併發的推播數量，但是因為 worker 能夠 scale 的數量有限，以及堆放到 queue 上的長度也有限，相比於 multicast 一次能夠發送 150 則以上的訊息，不只速度是 150 倍的提升，堆到 queue 上的長度也會是 150 倍的變短，因此，如果可以一次把多個 single push request 包成一大包，然後也可以在一個 http request 的 I/O block 時間內完成，就可以在只改少量 code 的情況下，大大的增加訊息推播的效能。

說到這裡應該也很多人知道了，就是一些 Concurrency 的運用，然而漸強的後端系統主要是由 python 打造的，受困於 GIL(Global Interpreter Lock) 的關係，讓 python 在併發上比較難發揮，於是我們使用 python 支援 asyncio 的套件 aiohttp，利用他的 asynchronous http client，在 python coroutine 跟 event loop 運作之下，達成讓多個 single push requests ，在接近一次的 I/O block 時間，達到發送多則訊息的目標。

在這邊所遇到的主要問題是，由於後端的 python 服務都是使用同步的框架實作，所以在 code 裡面我們寫了一些 wrapper ，讓我們可以輕鬆的在同步的 code 中跑異步的 event loop，以下稍微 demo 一下最簡易的版本，實際上當然還有一些 headers 或是不同 http method 以及錯誤處理，不過這跟實際要處理的情況有關就不多贅述：

最終我們決定一次包裹 100 個 single push 到 wrapper 內，結論在 push message api 類型的推播上是接近 100 倍的速度有感提升，而因為大量的使用 asyncio 讓我們嚐到甜頭，往後許多批次大量要 call LINE api 的功能，我們也都用相同的方法實作，獲得還不錯的效率，誠心推薦 👍

這時發送速度已經不是瓶頸了之後，反而是要注意發送速度不要頂到 LINE 的 rate limit，甚至有時候會因為一次發送太大量的 LINE 訊息，導致訊息上帶有的一些追蹤網址，最後會跳轉到客戶網站因為流量太大，而導致客戶網站掛點的情況… 😵，反而開始有了想控制發送速度的需求，這也是非常有趣的問題。

第二次改進：

一堆 worker待機等待發送訊息實在太貴了！ 改尋求 serverless 解法，丟 pub/sub 用 cloud function 發送！

過了一陣子後，我們發現大部分的推播流量都在中午或下班時間，導致很長時間的 vm 資源使用率都很低，這些動態的 vm 資源大概一個月要花個 300~500 鎂左右，雖然我們可以知道排程的推播時間，並且透過 kubernetes 的 hpa 或一些 controller 的實作來控制 worker 數量，但是還是有個問題，就是平台提供”馬上發送” 的功能，亦或是可以在還沒發送之前，任意的更改發送時間，這時不只是中午或下班時間，還要考慮客戶更改發送時間或馬上發送的問題，要用這樣的邏輯來控制 worker 數量，實作上還是有些複雜。

再者雖然 scale container 的速度是夠快的，但某種程度上這並不會真的減少支出成本， scale node 才會！那才是真正付錢的資源，一般來說 scale node 的速度就比較跟不上瞬間大量的發送訊息需求，為了降低成本，以及解決隨機大量推播的狀況，於是我們開始尋求 serverless 的解法，研究了一下決定使用 GCP 的 Pub/Sub 搭配 trigger cloud function 的方式來解決。

所以就是以 Pub/Sub 替代 rabbitmq message broker 的角色，cloud function 替代原本的 celery worker 負責 call LINE push message api，而在這個解法當中需要注意三件事：

1. At-Least-Once delivery：
   這是在 pub/sub 技術文擋中提到的，代表是有情況可能會發生重複訊息被推送的可能，如果同一包訊息被 cloud function 發送兩次，代表有一群人收到兩封一樣的訊息，且客戶就要多付給 LINE 那一包重複發送的訊息費，為了預防這件事情，我們把丟到 pub/sub 的每一包訊息都給了一個 package_id，在 cloud function 發送訊息時，透過 firestore 來記錄這個 package_id 有沒有被發送過，來預防重複的訊息被 pub/sub push 到 cloud function，然後被 cloud function 發送出去的情況
2. LINE message API rate limit：
   在 cloud function 輕鬆的 scale 大量的 instance 來打 LINE API 的同時，也遇到上面提過的 LINE API 的 rate limit 限制，對此我們測試了許多不同量級訊息來測試發送，最後限制了 cloud function instance scale 的數量，來防止我們頂到這個 rate limit。
   但我們也因此踩到一個小雷，就是 pub/sub trigger cloud function 的機制是透過 push，因為 cloud function scale 數量的關係，pub/sub 的 push 可能收到 cloud function 回覆 429 的狀態碼而不是 200，進而導致 pub/sub 會過一段時間後才重試這些 429 訊息，後來才理解到這樣的模式不該控制 cloud function scale instance 的數量，但是幾經考慮後，衡量在 cloud function code 裡面處理 rate limit 的複雜度，跟我們其實可以接受這個 pub/sub retry 的延遲之下，就不作特別的調整，而這件事情 debug 釐清原因的過程其實蠻冗長的，要特別感謝 iKala 在這件情上提供的 GCP support 協助，幫助我們釐清這個問題。
3. Too many record need save
   大量的發送訊息後的結果，需要被記錄的問題，上面都沒有提到的是，其實我們不只要發送，發送完的結果，誰成功誰失敗，或失敗的原因等等，我們都希望可以鉅細彌遺的記錄下來，在之前是靠自己 host Redis 跟 Postgres 來負責紀錄，但是既然要 serverless ，就要 serverless 到底，己經嘗試後發現上面提到的 firestore 是撐得住的，於是我們就把每個訊息發送的結果，跟一些 metadata 都存到 firestore 上去，當然也是有遇到了一些問題，在這邊也 reference 一些文擋給大家參考，包括利用 Distributed counters 來記錄發送數量，以及利用 Transactions and batched writes 來批次寫入發送結果等等。

最後，提供我們 2020/04 用 Pub/Sub + Cloud function 發送六千萬則訊息後的帳單：

後記心得：

大家會發現我們用了一堆 GCP 的服務，其實也沒有什麼高大尚的原因，就是我們有拿到 GCP 的免費額度，現實世界的架構設計跟白板上的還是有點不一樣的，成本的支出常常是很大的考量因素，即使這樣的架構其實也不算是特別完美，但是以我們目前這個發送訊息的量級來說，出錯的機率低、花費的金額低，就連需要人力去維護的時間成本也低，這的確是個蠻不錯的解決方案！

至於這誇張便宜的金額，其實是包含了一些 GCP 每個專案都有的免費基本額度，而且這只是兩個主要的訊息系統架構，其他還有儲存系統的花費，更別說廣告推播訊息的圖片跟影片所造成的 CDN 網路流量，更是貴到爆炸！

往後有空我會持續分享在漸強搭建系統架構上的經驗與心得，也歡迎大家給予回饋與建議，在次感謝大家的耐心 🤘

最後，如果對我們有興趣的話，目前都有在徵才喔～
https://www.yourator.co/companies/CrescendoLab
誠摯歡迎各路大神來加入我們！