Google APP Engineの活用方法–定期ジョブ管理

皆さん

こんにちは、GMOアドマーケティングのS.Rです。

GCP(Google Cloud Platform)を使ってる時に定期ジョブ管理をどうやって実現していますか。今回はGCPでよく使われてる定期ジョブ管理方法を紹介します。※この記事を理解するには、Python、Http、Linuxのshellコマンドの基本知識が必要です。本記事中の図説は、筆者が自らの環境で作成したものを含みます。

すぐに思いつく定期ジョブ管理方法

定期ジョブ管理というとすぐ思いつくツールがCronです。図１は誰でも思いつく定期ジョブの解決方法です。一つGCE（Google Compute Engine）を立ち上げて、 CronサーバーとしてCron Scriptを登録し、24時間動かし続けます。 指定された時間にGCPのSDKと連携して、定期ジョブを動かします。

 

この方法は下記3つの大きな欠点があります：

1. Logが管理しづらい：GoogleSSHでCompute Engineへ連携しなければLogを確認できません。
2. 無駄な料金がかかります：定期ジョブを実行してない時間にも料金がかかります。
3. スケーリングできない：定期ジョブの計算量が変わってもCronサーバーの計算力を調整できません。

この3つ欠点に対して今回は改良できる管理方法を紹介します

Google APP Engineで定期ジョブを管理する

Google App Engineは公式のサードに説明は下記です。

Google App Engine は、Google のインフラの上でアプリケーションを作り、実行できるようにする PaaS （ Platform as a Service : サービスとしてのプラットフォーム）です。App Engine アプリケーションは、作るのもメンテナンスするのも簡単で、トラフィックやデータストレージのニーズの増減にあわせてスケーリングするのも簡単です。App Engine を使えば、自分でメンテナンスしなければならないサーバーはなくなります。単純にアプリケーションをアップロードすれば、それだけで実行できます。

GAEは定期ジョブを管理する機能があります。今回はGAEで定期ジョブを管理する方法を紹介します。今回の例として、GAEの定期ジョブで一日毎にDataprocのインスタンスを立ち上げ、タスクを実行した後、インスタンスを削除するタスクを管理します

環境設定

1. GAEを利用するにあたり、Google SDKの設定をします。設定方法はGoogle Cloudの公式サイトを参考にしてください。
2. Pythonのか仮想的な環境ツールvirtualenvをインストールします。コマンドは下記の通りです
   
   pip install virtualenv

   1	pip install virtualenv

3. GAE SDKにGCPのユーザーにログインしてKeyファイルを作りましょう。
   
   gcloud auth application-default login

   1	gcloud auth application-default login

   
   

   

   作成されたKeyファイルは~/.bashrcに追加します。

   export GOOGLE_APPLICATION_CREDENTIALS="Keyファイルのパス"

   1	export GOOGLE_APPLICATION_CREDENTIALS="Keyファイルのパス"

   設定を再読込します。

   source ~/.bashrc

   1	source ~/.bashrc

4. Pythonの仮想環境を作成します。
   virtualenv env source env/bin/activate

   1 2	virtualenv env source env/bin/activate

ソースコードを作成

例のソースコードを下記のsetpで作成します。

1. 図４ようなソースコードツリーを作ります。

   

2. appengine_config.pyの中身は下記のコードを入ります。

   from google.appengine.ext import vendor # Add any libraries installed in the "lib" folder. vendor.add('lib')

   1 2 3

   from google.appengine.ext import vendor # Add any libraries installed in the "lib" folder. vendor.add('lib')

3. requirements.txtの中身は下記のコードを記述します
   Flask==0.10 Markdown==2.5.2 google-api-python-client

   1 2 3

   Flask==0.10 Markdown==2.5.2 google-api-python-client

4. 定期ジョブが依存しているLibrary をインストールします。
   pip install -t lib -r requirements.txt

   1	pip install -t lib -r requirements.txt

5. cron.yamlの中身は下書き方は簡単な説明は下記です。
   YAML

   cron: - description:"[定期ジョブの説明]" url: 「書いたGAE applicationのURL」 target: beta schedule: 「実行する時間」

   1 2 3 4 5

   cron: - description:"[定期ジョブの説明]"   url: 「書いたGAE applicationのURL」   target: beta   schedule: 「実行する時間」

   詳しくcron.yamlの書き方を知りたい方はGoogle公式サイトに参考してください。今回の例のCron.yamlの中身は下記です。

   cron: - description: "daily summary job" url: /tasks/create_dataproc target: beta schedule: every 24 hours

   1 2 3 4 5

   cron: - description: "daily summary job"   url: /tasks/create_dataproc   target: beta   schedule: every 24 hours

6. 定期ジョブの本体を作成します。定期ジョブは3つの部分があります。
   1. DataprocのInstanceを立ち上げます。DataprocのPython ApiはGoogle公式サイトを参考にしてください。
      @staticmethod def create_cluster(project, zone, region, cluster_name): dataproc = build('dataproc', 'v1') print('Creating cluster...') zone_uri = \ 'https://www.googleapis.com/compute/v1/projects/{}/zones/{}'.format( project, zone) cluster_data = { 'projectId': project, 'clusterName': cluster_name, 'config': { 'gceClusterConfig': { 'zoneUri': zone_uri }, 'masterConfig': { 'numInstances': 1, 'machineTypeUri': 'n1-standard-1' }, 'workerConfig': { 'numInstances': 2, 'machineTypeUri': 'n1-standard-1' } } } result = dataproc.projects().regions().clusters().create( projectId=project, region=region, body=cluster_data).execute() print('Waiting for cluster creation...') while True: time.sleep(5) instance_list = dataproc.projects().regions().clusters().list( projectId=project, region=region).execute() cluster_list = instance_list['clusters'] cluster = [c for c in cluster_list if c['clusterName'] == cluster_name][0] if cluster['status']['state'] == 'ERROR': raise Exception(result['status']['details']) if cluster['status']['state'] == 'RUNNING': print("Cluster created.") break return result

      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

      @staticmethod def create_cluster(project, zone, region, cluster_name):     dataproc = build('dataproc', 'v1')     print('Creating cluster...')     zone_uri = \         'https://www.googleapis.com/compute/v1/projects/{}/zones/{}'.format(             project, zone)     cluster_data = {         'projectId': project,         'clusterName': cluster_name,         'config': {             'gceClusterConfig': {                 'zoneUri': zone_uri             },             'masterConfig': {                 'numInstances': 1,                 'machineTypeUri': 'n1-standard-1'             },             'workerConfig': {                 'numInstances': 2,                 'machineTypeUri': 'n1-standard-1'             }         }     }     result = dataproc.projects().regions().clusters().create(         projectId=project,         region=region,         body=cluster_data).execute()     print('Waiting for cluster creation...')     while True:         time.sleep(5)         instance_list = dataproc.projects().regions().clusters().list(             projectId=project,             region=region).execute()         cluster_list = instance_list['clusters']         cluster = [c                    for c in cluster_list                    if c['clusterName'] == cluster_name][0]         if cluster['status']['state'] == 'ERROR':             raise Exception(result['status']['details'])         if cluster['status']['state'] == 'RUNNING':             print("Cluster created.")             break     return result

   2. DataprocのInstanceを削除します。
      @staticmethod def delete_cluster(dataproc, project, region, cluster): print('Tearing down cluster') return dataproc.projects().regions().clusters().delete(projectId=project, region=region, lusterName = cluster).execute()

      1 2 3 4 5 6

      @staticmethod def delete_cluster(dataproc, project, region, cluster):     print('Tearing down cluster')     return dataproc.projects().regions().clusters().delete(projectId=project,                                                            region=region,                                                            lusterName = cluster).execute()

   3. 定期ジョブのGet requestのHandler関数：GAEのCronサーバーからきたrequestであれば、requestのheaderに項目の’X-AppEngine-Cron’はTrueとなってます。
      def get(self): self.response.headers['Content-Type'] = 'text/plain' if self.request.headers.get('X-AppEngine-Cron') is None: self.abort(403) self.response.write("hello") self.create_cluster('test', 'asia-northeast1-a', 'asia-northeast1', 'test') self.delete_cluster('test', 'asia-northeast1-a', 'asia-northeast1', 'test')

      1 2 3 4 5 6 7

      def get(self):     self.response.headers['Content-Type'] = 'text/plain'     if self.request.headers.get('X-AppEngine-Cron') is None:         self.abort(403)     self.response.write("hello")     self.create_cluster('test', 'asia-northeast1-a', 'asia-northeast1', 'test')     self.delete_cluster('test', 'asia-northeast1-a', 'asia-northeast1', 'test')

      最後は作成したWSGIApplication Classを返却します。

      app = webapp2.WSGIApplication([ ('/tasks/create_dataproc', createDataproc), ], debug=True)

      1 2 3

      app = webapp2.WSGIApplication([     ('/tasks/create_dataproc', createDataproc), ], debug=True)

7. 定期ジョブの本体を作成します。
   def get(self): self.response.headers['Content-Type'] = 'text/plain' self.response.write("hello")

   1 2 3

   def get(self):     self.response.headers['Content-Type'] = 'text/plain'     self.response.write("hello")

8. Firewall rulesを設定します。GAEでCron サーバーのIP Address以外を拒否します

   1. 

9. 環境設定するScript .bashrcを実行します

Local環境で例を試す

本番のサーバーへDepolyする前にまずLocal環境で作成したコードを試します。まず下記のコマンドよってLocal環境でコードを起動します。

ブラウザでlocalhost:8000へ訪問し、 Local環境の管理画面を見えます。「Cron Jobs」を選択して「Run now」buttonを押してコードを起動できます。

GAEで例を試す

Local環境で定期ジョブのテストが完成したら 、下記のコマンドよってGAEでコードをDeployします。

GAEの管理画面から、Cron Jobsを選択して「Run now」buttonを押しコードを起動します。

GAEで定期ジョブを削除する

GAEで定期ジョブを停止したい時は下記ようなCron.yamlに一行目以外のコードを削除してもう一度GAEへDeployします。

Deployするコマンドは下記です。

３まとめ

今回はGAEで定期ジョブの一例を紹介しました。いかがだったでしょうか。GCPでプロダクトを開発する時に定期ジョブの開発が多いと思います。機会があれば皆さんぜひ試してみてください。