こんにちは、香田です。
今回はCloud Functionsのローカル開発環境でFunction Frameworkを使用する方法について紹介していきます。
本記事ではPython3の環境を前提に紹介しています。
Function Frameworkについて
Functions Frameworkを使用すると、Cloud Functionsで作成する関数をローカル環境で実行しテストすることが可能となります。
Functions Frameworkは、Node.js、Go、Pythonなど各言語毎に提供されており、今回はPythonを使用して紹介していきます。
Function Frameworkのインストール
はじめにFunction Frameworkをインストールしていきます。
作業ディレクトリを作成します。
mkdir sandbox
cd sandboxvenv仮想環境を初期化します。
python -m venv venv
source venv/bin/activateFunction Frameworkをインストールします。
pip install functions-framework関数の作成
次にテスト実行する関数を作成していきます。
今回使用する関数は、Cloud SchedulerとPub/SubをトリガーにCloud Functionsが起動し、Compute Engine インスタンスが定期的に起動、停止する関数を想定し作成しています。
関数内で利用する為、Google API クライアント ライブラリをインストールします。
pip install google-api-python-client下記のコードをコピーし、main.pyという名前で保存します。
import json
import os
import base64
import googleapiclient.discovery
compute = googleapiclient.discovery.build('compute', 'v1')
project = os.getenv('GCP_PROJECT')
def get_data(event):
if 'data' in event:
data = json.loads(base64.b64decode(event['data']).decode('utf-8'))
else:
raise RuntimeError('No data in event')
return data
def get_zone(data):
if 'zone' in data:
zone = data['zone']
else:
raise RuntimeError('No zone in data')
return zone
def get_label(data):
if 'label' in data:
label = data['label']
else:
raise RuntimeError('No label in data')
return label
def stop_instance(event, context):
print("EVENT", event)
data = get_data(event)
zone = get_zone(data)
label = get_label(data)
key = label.split('=')[0]
val = label.split('=')[1]
instances = compute.instances().list(project=project, zone=zone).execute()
if 'items' in instances:
for instance in instances['items']:
if key in instance['labels']:
if val == instance['labels'][key]:
name = instance['name']
compute.instances().stop(
project=project,
zone=zone,
instance=name).execute()
print('Stop instance name {name}'.format(name=name))
def start_instance(event, context):
print("EVENT", event)
data = get_data(event)
zone = get_zone(data)
label = get_label(data)
key = label.split('=')[0]
val = label.split('=')[1]
instances = compute.instances().list(project=project, zone=zone).execute()
if 'items' in instances:
for instance in instances['items']:
if key in instance['labels']:
if val == instance['labels'][key]:
name = instance['name']
compute.instances().start(
project=project,
zone=zone,
instance=name).execute()
print('Start instance name {name}'.format(name=name))Functions Frameworkを使用してテスト実行する関数は、上記コード内で定義されているstop_instance、start_instanceという関数になります。
Compute Engine インスタンスのlabelでkeyにschedule-instance、valueにtrueと設定されていたら、処理がそれぞれ実行されるような内容になっています。
Functions Frameworkを実行する
関数が用意できたので、Functions Frameworkを実行していきます。
関数毎にテストしたい場合、Function Frameworkの起動ポートをそれぞれ変更し実行することでテスト可能です。
stop_instance関数のテスト用にport 8080でFunctions Frameworkを実行します。
export GCP_PROJECT=$(gcloud config get-value project)
functions-framework --target stop_instance --signature-type event --debug --port 8080start_instance関数のテスト用にport 8081でFunctions Frameworkを実行します。
export GCP_PROJECT=$(gcloud config get-value project)
functions-framework --target start_instance --signature-type event --debug --port 8081関数にリクエストを送信する
HTTPサーバがぞれぞれのポートで起動しているので、関数にリクエストを送信していきます。
Pub/Subのトリガーを想定しデータをbase64でエンコードし、それぞれリクエストを送信します。
下記ではゾーンがasia-northeast1-aのインスタンス対して、対象のラベルが設定されている場合、関数内の処理が実行される内容です。
stop_instance関数へリクエスト送信
export DATA=$(echo '{"zone":"asia-northeast1-a", "label":"schedule-instance=true"}' | base64)
curl -d '{"data": {"data": "'$DATA'"}}' -X POST -H "Content-Type: application/json" http://localhost:8080start_instance関数へリクエスト送信
export DATA=$(echo '{"zone":"asia-northeast1-a", "label":"schedule-instance=true"}' | base64)
curl -d '{"data": {"data": "'$DATA'"}}' -X POST -H "Content-Type: application/json" http://localhost:8081それぞれレスポンス結果としてOKと表示されていれば成功です。
さいごに
Cloud Functionsのローカル開発環境でFunction Frameworkを使用する方法いかがでしたでしょうか。
Function Frameworkを使用する上で本記事が参考になれば幸いです。
また、Function Frameworkについて詳しく知りたい場合、下記のCloud Functionsのドキュメントページも参考にしてみてください。
最後までご覧いただきありがとうございます。