BACK
Featured image of post 【AWS】利用 SNS + Lambda,在 S3 儲存貯體檔案更新時透過 Telegram 通知我!

【AWS】利用 SNS + Lambda,在 S3 儲存貯體檔案更新時透過 Telegram 通知我!

我的靜態專案放在 Amazon Web Services(AWS) 的 S3 儲存貯體上,最近使用 AWS 的 SNS 服務搭配 AWS Lambda,實作當 S3 儲存貯體內的檔案發生異動時,可以在我的 Telegram 上跳出通知,相當有趣,以下就來分享我的實作過程!

參考網站
參考網站
參考網站
參考網站
參考網站

我的靜態專案放在 Amazon Web Services(AWS) 的 S3 儲存貯體上,最近使用 AWS 的 SNS 服務搭配 AWS Lambda,實作當 S3 儲存貯體內的檔案發生異動時,可以在我的 Telegram 上跳出通知,相當有趣,以下就來分享我的實作過程!

因為我的專案 Git 倉放在 Github 上,暫不考慮 AWS 的 CodeCommit,以目前的架構來規劃玩玩!


架構概括

以下就是將要建立的流程,我將建立一個 SNS 主題,該主題透過接收 S3 的事件並將其發布到 Lambda 函數,再透過 Lambda 發布通知到 Telegram。


創建一個 Telegram bot

首先我們需要先在 Telegram 中創建一個機器人,透過之前的文章 - 「Telegram 機器人說明,並為你的 Drone CI/CD 配置 Telegram 機器人」,向 @BotFather 申請一隻吧!

取得 Telegram bot Token:

先將 Telegram bot 的 Token 存起來,後面會用到。


取得我的用戶 ID

有了 Telegram bot 後,也需要取得我的用戶 ID,才知道機器人要將訊息傳給誰。

先跟機器人講幾句話,以便 getUpdates API 取資料:

透過 telegram 的 getUpdates API 取得我的用戶 ID:

請使用瀏覽器打開以下網址:

1
https://api.telegram.org/bot<機器人的Token>/getUpdates

注意: <機器人的Token> 前面的前綴 bot 不可省略,否則會找不到。

用戶 ID 在這裡: [result][message][from][id]

將用戶 ID 也存起來,後面會用到。


創建一個 SNS 主題

登入 AWS,在控制台找到 “SNS” 服務。

進入「主題」,點選「建立主題」。

注意:建立主題時,地區一定要與你的 S3 儲存貯體相同,否則 S3 儲存貯體是看不到這個主題的。

選擇「標準」,並給它取個名字:「TelegramSNSNotifierTopic」,然後往下點擊「建立主題」。


創建 Lambda 函數

AWS 的控制台,找到 “Lambda” 服務。

點選「建立函式」,選擇「從頭開始撰寫」,給函數命名:「TelegramSNSNotifierLambda」,runtime 選擇「Python 3.9」,然後往下點擊「建立函式」。

設定 Lambda 的環境變數

進入剛剛建立的函數,在「組態」>「環境變數」內,點選「編輯」。

新增 TOKENUSER_ID 兩個環境變數,值為:

  • TOKEN: 剛剛創建的 Telegram bot token
  • USER_ID: 剛剛取得的我的 Telegram 用戶 ID

儲存後,來編輯一下 Lambda 函數的程式碼。

編輯 Lambda 函數

回到「程式碼」頁籤,打開 lambda_function.py 檔案,會看到預設的程式碼。

將以下程式碼覆蓋掉當前 lambda_function.py 內的程式碼:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import json
import os
import logging
import urllib3
import boto3

s3_client = boto3.client("s3")

# S3 Bucket Name
S3_BUCKET = "test-bucket-01"

# 專案中,用來紀錄當前版本號的檔案
object_key = "version.yaml"

http = urllib3.PoolManager()

# 初始化 logger,並設定為 INFO
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 取得環境變數,並組成 Telegram Bot API URL
TOKEN = os.environ['TOKEN']
USER_ID = os.environ['USER_ID']
TELEGRAM_URL = "https://api.telegram.org/bot{}/sendMessage".format(TOKEN)

# Main Lambda handler
def lambda_handler(event, context):
  # 紀錄事件觸發,以便 debugging
  logger.info("event=")
  logger.info(json.dumps(event))
  
  try:
    file_content = s3_client.get_object(Bucket=S3_BUCKET, Key=object_key)
    version_info = file_content["Body"].read().decode("utf-8")
    message = f"當前環境版本,{version_info}"

    # 組成用於呼叫 Telegram Bot API 的 Payload
    payload = {
      "text": message.encode("utf8"),
      "chat_id": USER_ID
    }

    # 送出請求到 Telegram Bot API
    http.request('POST', TELEGRAM_URL, payload)

  except Exception as e:
    raise e

如需取得 SNS 服務傳來的 message,可以使用以下版本:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import json
import os
import logging
import urllib3

http = urllib3.PoolManager()

# 初始化 logger,並設定為 INFO
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 取得環境變數,並組成 Telegram Bot API URL
TOKEN = os.environ['TOKEN']
USER_ID = os.environ['USER_ID']
TELEGRAM_URL = "https://api.telegram.org/bot{}/sendMessage".format(TOKEN)

# JSON Helper function to prettify the message if it's in JSON
def process_message(input):
  try:
    raw_json = json.loads(input)
    output = json.dumps(raw_json, indent=4)
  except:
    output = input
  return output

# Main Lambda handler
def lambda_handler(event, context):
  # 紀錄事件觸發,以便 debugging
  logger.info("event=")
  logger.info(json.dumps(event))

  try:
    # 從回傳的 message 內取得 SNS message 欄位
    message = process_message(event['Records'][0]['Sns']['Message'])

    # 組成用於呼叫 Telegram Bot API 的 Payload
    payload = {
      "text": message.encode("utf8"),
      "chat_id": USER_ID
    }

    # 送出請求到 Telegram Bot API
    http.request('POST', TELEGRAM_URL, payload)

  except Exception as e:
    raise e

程式碼僅依我個人需求撰寫,大家可以自行依自己的需求做撰寫。

接著儲存後進行發布,點選「Deploy」。

成功更新函數!


添加 SNS 觸發器

點選「新增觸發」,來源選擇「SNS」,SNS 主題選擇剛剛建立的「TelegramSNSNotifierTopic」,接著點擊「新增」。

SNS 與 Lambda 已完成配置!


到 S3 儲存貯體設定「事件通知」

前往 AWS S3 服務,並進入你欲設定通知的儲存貯體詳細資訊。

切換到「屬性」頁籤,下滑找到「事件通知」,並點選「建立事件通知」。

依照我的需求,我的設定如下:

  • 事件名稱,我取名「開發環境版本更新」
  • 事件類型,因為我的專案在版本更新時,會自動將 Git 當前版本的 Commit ID(SHA) 更新到根目錄下的 version.yaml 檔案中,所以我選擇:
    • PUT「s3:ObjectCreated:Put」
    • POST「s3:ObjectCreated:Post」
    • 複製「s3:ObjectCreated:Copy」
  • 目的地選擇「SNS 主題」,指定 SNS 主題選擇「從 SNS 主題中選擇」,最後選擇前面建立的 SNS 主題 - 「TelegramSNSNotifierTopic

接著點選「儲存變更」,就完成設定囉!


測試看看

這邊來手動上傳 version.yaml 到 S3 儲存貯體,試試看是否有成功。

上傳成功後,就會發現 Telegram bot 通知我了!


小結

首先,S3 是一個非常強大的對象儲存服務,可以存儲和檢索大量數據,它的存儲和訪問速度非常快。在使用 S3 時,其簡單易用的界面可以輕鬆地創建存儲桶(bucket)、設置存取權限、上傳和下載文件等。此外,S3 還支持版本控制、交叉區域複製等高級功能。

其次,SNS 是一個強大的消息發布和訂閱服務,可以幫助我將消息發送到多個端點,如電子郵件、短信、移動設備等。SNS 的設置非常簡單,可以輕鬆地創建主題(topic)、訂閱端點(subscription)、設置過濾器(policy)等。同時,SNS 還提供了許多高級功能,如死信隊列(dead-letter queue)、消息篩選等,這些功能可以更好地控制和管理我的消息流。

最後,Lambda 是一個靈活且高效的無伺服器計算服務,可以讓我在不擔心伺服器設置和管理的情況下運行代碼。使用 Lambda 非常方便,只需創建一個函數(function),然後設置觸發器(trigger),就可以開始運行我的代碼。同時,Lambda 還支持許多高級功能,如環境變量(environment variables)、VPC 訪問(virtual private cloud access)、調用其他服務等,這些功能可以讓我的代碼更加強大和靈活。

本篇只是分享我試玩一下 AWS SNS 與 Lambda,相當有趣,期待後續有更靈活、更好玩的運用!


comments powered by Disqus