UUUMエンジニアブログ

UUUMのエンジニアによる技術ブログです

Cloud Scheduler を使って、Cloud Functions を定期実行させたい！

▼ 自己紹介

こんにちは！UUUMシステムユニットの高橋です！


私はUUUM内でデータ出入力関連の作業を担当しておりますが、そこで頻繁に利用するサービスとして Google Cloud Scheduler というものがあります。

これは様々なジョブを定期実行してくれるGoogleのサービスで、私の場合で言えば、毎日一回作成するレポートを自動的に作成してくれています。

今回は、Google Cloud Scheduler の使用例の一つとして、Cloud Functionsの定期実行について書こうと思います！

▼ 構成の概要


利用するサービスの繋がりを図にしました↓

今回使用する言語がpythonなのでそのアイコンを使いましたが、他の言語も使えます！


▼ ローカルでコードを書く

まずは定期実行するコードを書いていきます
今回は実行結果の確認を簡単にするために、Slackにメッセージを送るコードを書こうと思います！


とりあえず、Slackにメッセージを送る関数 send_text() を作り↓

from slack_sdk import WebClient
import requests


def send_text(input_token,input_channel_id,input_text):
    '''---
    内容：
        指定したチャンネルにテキストを送信する
    引数:
        input_token: slack bot のトークン
        input_channel_id: 投稿先チャンネルのID(ex C02ABCDEFGH Slackのチャンネルリンク内に記載)
        input_text: 投稿するテキスト
    '''
    urlpo='https://slack.com/api/chat.postMessage'
    headers = {'Authorization' : f'Bearer {input_token}'}
    data = {
        'token': input_token,
        'channel': input_channel_id,
        'text':input_text
    }

    requests.post(urlpo, headers = headers, data=data)
    return


これをmain() で呼び出します↓

import datetime,os

def main():
    #タイムゾーン設定
    JST=datetime.timezone(datetime.timedelta(hours=+9), 'JST')
    now = str('{0:%Y-%m-%d_%H:%M:%S}'.format(datetime.datetime.now(JST)))

    # slackへメッセージを送る

    # slack_token, slack_channel_idを環境変数から取得する
    SLACK_TOKEN=os.getenv('SLACK_TOKEN')
    SLACK_CHANNEL_ID=os.getenv('SLACK_CHANNEL_ID')
    # messageを作る
    str_message=f'''
    【実行時間】{now}
    Cloud Pub/Sub トピックのメッセージからトリガーされました\n
    ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
    '''
    # Slackへ送信
    send_text(SLACK_TOKEN,SLACK_CHANNEL_ID,str_message)

main()


全体としてはこんな感じで完成↓

import datetime,os,requests
from slack_sdk import WebClient


def main():
    #タイムゾーン設定
    JST=datetime.timezone(datetime.timedelta(hours=+9), 'JST')
    now = str('{0:%Y-%m-%d_%H:%M:%S}'.format(datetime.datetime.now(JST)))

    # slackへメッセージを送る

    # slack_token, slack_channel_idを環境変数から取得する
    SLACK_TOKEN=os.getenv('SLACK_TOKEN')
    SLACK_CHANNEL_ID=os.getenv('SLACK_CHANNEL_ID')
    # messageを作る
    str_message=f'''
    【実行時間】{now}
    Cloud Pub/Sub トピックのメッセージからトリガーされました\n
    ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
    '''
    # Slackへ送信
    send_text(SLACK_TOKEN,SLACK_CHANNEL_ID,str_message)

#------------------------------------------------------------------------

def send_text(input_token,input_channel_id,input_text):
    '''---
    内容：
        指定したチャンネルにテキストを送信する
    引数:
        input_token: slack bot のトークン
        input_channel_id: 投稿先チャンネルのID(ex C02ABCDEFGH Slackのチャンネルリンク内に記載)
        input_text: 投稿するテキスト
    '''
    urlpo='https://slack.com/api/chat.postMessage'
    headers = {'Authorization' : f'Bearer {input_token}'}
    data = {
        'token': input_token,
        'channel': input_channel_id,
        'text':input_text
    }

    requests.post(urlpo, headers = headers, data=data)
    return

#------------------------------------------------------------------------

main()


実行するとSlack にメッセージを送信できていたのでOK↓(ここはまだ手動で実行しただけ)


これでコードの作成は完了です！


▼ Pub/Subにトピックを作る

今回はCloud Functionsに登録したコードをPub/Subをトリガーに実行したいと思います！
その為に、Pub/Subにトピックを作っておきます


Pub/Subのページから「トピックの作成」を押下し、トピックIDを決めて作成できます↓

今回は「test_pubsub」というIDで作成しました
このIDは次のステップで使用します！


これでトピックの作成は完了です！


▼ Cloud Functionsにコードを移す

次に、Cloud Functionsにコードを移していきます！

Cloud Functionsのページから「関数の作成」を押下し、
関数名を決めて、先ほど作成したトピックを選択します↓


作ったコードの中で使用する環境変数はこの欄で設定できます↓


コードを転記します↓


requirements.txtの中身は↓のようになっています

# Function dependencies, for example:
# package>=version
requests==2.28.1
slack-sdk==3.18.4


これでCloud Functionsへの移行は完了です！


▼ Pub/Subから実行できるか試してみる

ここまでで、Pub/SubをトリガーにCloud Functionに登録したコードを実行できるようになったはずなので、テストしてみます！


まずはPub/Subのページから作成したトピックを押下し、メッセージタブを選択します。
次にトピックをプルダウンで指定して、「メッセージをパブリッシュ」を押下します↓


メッセージを入力し、「公開」を押下します↓


期待通りSlack に届きました↓


これでPub/Sub → Cloud Functionsの起動テストは完了です！


▼ Google Cloud Scheduler で定期実行してみる

最後に Cloud Scheduler の設定をしていきます！


Cloud Schedulerのページから「ジョブの作成」を押下し↓


各欄に値を入力して登録できます↓


「2分毎に実行」で登録しましたが、強制実行で確認してみると...↓

Slackにメッセージが届いたので、大丈夫そうです！↓


これでCloud Scheduler の設定は完了です！


▼ 定期実行されてるか確認

さて、これで定期実行されるはずですが、
実際どうなのか10分ほど見守ってみると...↓

問題なく動いているみたいです！


これで全て完了です！

▼ 最後に

お疲れ様でした！
簡単な例ではありましたが、定期実行を登録することができました
実行されるコードの部分を書き換えれば、

  • 毎日の株価を取得してくる　とか
  • Youtubeの急上昇動画のURLを通知する　とか
  • データベースの点検をさせる　とか

自分が普段行っている作業を代わってもらうことができそうですね
皆様の何かの参考になれば幸いです！