Google Cloud Tasksから呼び出すApp Engine TaskとGAEバージョンの関係

Google Cloud TasksからApp Engine HTTPタスクでバックグラウンド処理をするアプリケーションを作っている。

トラフィック分割で様子を見ながらリリースしたい場合、App Engine HTTPはどのバージョンのタスクを実行するのかがわからなかったので実験してみた。

前提条件

• GAE/Python3.7 (2nd Generation Runtime)
• google-cloud-tasks 1.5.0
  • ※執筆時点で最新版は 2.0.0
  • gcloud-tasks-emulator を使いたいという理由でバージョンを固定している。

実験環境ではApp Engine のバージョンは v1とv2があり、v2にトラフィック100%割り当てとした。どちらにも同じコードがデプロイされている。

通常の呼び出しをした場合

まずは、普通に実行した場合。

project = os.environ['GOOGLE_CLOUD_PROJECT']
service = os.environ['GAE_SERVICE']
version = os.environ['GAE_VERSION']
queue_region = "us-central1"
queue_name = 'test-dispatch'

@app.route('/enqueue1', methods=['POST'])
def enqueue1():
    print("Enqueue1 in PROJECT: {}, SERVICE: {}, VERSION: {}".format(project, service, version))
    request_task = {
        'app_engine_http_request': {
            'http_method': 'POST',
            'relative_uri': '/task'
        }
    }
    tasks_client = tasks.CloudTasksClient()
    tasks_queue = tasks_client.queue_path(project, queue_region, queue_name)
    tasks_client.create_task(tasks_queue, request_task)
    return 'ok'


@app.route('/task', methods=['POST'])
def task():
    print("Tasks Running in PROJECT: {}, SERVICE: {}, VERSION: {}".format(project, service, version))
    return 'ok'

結果

# デフォルトルート
$ curl -X POST -d '' https://[project].uc.r.appspot.com/enqueue1
# Cloud Logging の stdout
# Tasks Running in PROJECT: [project], SERVICE: default, VERSION: v2

# v2直指定
$ curl -X POST -d '' https://v2-dot-[project].uc.r.appspot.com/enqueue1
# Cloud Logging の stdout
# Tasks Running in PROJECT: [project], SERVICE: default, VERSION: v2

# v1直指定
$ curl -X POST -d '' https://v1-dot-[project].uc.r.appspot.com/enqueue1
# Cloud Logging の stdout
# Tasks Running in PROJECT: [project], SERVICE: default, VERSION: v2

v1を指定してもタスクはv2に向かった。これは、App Engineプロジェクト全体のデフォルトのルーティングによってリクエスト先が決まるからとのこと。マルチサービス環境下でdispatch.yamlがある場合はその内容が使用される。

app_engine_routingを指定する

app_engine_http_requestのオプションとしてapp_engine_routingがあり、これを指定することで任意のバージョンやサービスに向けることができる。

app_engine_routingを書くのはapp_engine_http_requestの下。同一階層に書いて動かなくて時間を浪費した。

project = os.environ['GOOGLE_CLOUD_PROJECT']
service = os.environ['GAE_SERVICE']
version = os.environ['GAE_VERSION']
queue_region = "us-central1"
queue_name = 'test-dispatch'

@app.route('/enqueue2', methods=['POST'])
def enqueue2():
    print("Enqueue2 in PROJECT: {}, SERVICE: {}, VERSION: {}".format(project, service, version))
    request_task = {
        'app_engine_http_request': {
            'http_method': 'POST',
            'relative_uri': '/task',
            'app_engine_routing': {
                'service': service,
                'version': version,
                'instance': ''
            }
        }
    }
    tasks_client = tasks.CloudTasksClient()
    tasks_queue = tasks_client.queue_path(project, queue_region, queue_name)
    tasks_client.create_task(tasks_queue, request_task)
    return 'ok'

結果

# デフォルトルート
$ curl -X POST -d '' https://[project].uc.r.appspot.com/enqueue2
# Cloud Logging の stdout
# Tasks Running in PROJECT: [project], SERVICE: default, VERSION: v2

# v2直指定
$ curl -X POST -d '' https://v2-dot-[project].uc.r.appspot.com/enqueue2
# Cloud Logging の stdout
# Tasks Running in PROJECT: [project], SERVICE: default, VERSION: v2

# v1直指定
$ curl -X POST -d '' https://v1-dot-[project].uc.r.appspot.com/enqueue2
# Cloud Logging の stdout
# Tasks Running in PROJECT: [project], SERVICE: default, VERSION: v1

app_engine_routingを指定してもデフォルトにルーティングされる

app_engine_routingを指定してもデフォルトサービスにルーティングされる場合がある。これはCloud Tasksキューにはタスクに指定されているapp_engine_routingを強制上書きするapp_engine_routing_overrideという設定があり、Webのコンソールから作成するとデフォルトで入ってしまうらしい。

$ pipenv run gcloud tasks queues describe test --project [project]
appEngineRoutingOverride:
  host: [project].uc.r.appspot.com                 # <- !!?!!???
name: projects/[project]/locations/us-central1/queues/test2
rateLimits:
  maxBurstSize: 100
  maxConcurrentDispatches: 1000
  maxDispatchesPerSecond: 500.0
retryConfig:
  maxAttempts: 100
  maxBackoff: 3600s
  maxDoublings: 16
  minBackoff: 0.100s
state: RUNNING

gcloud tasks queues update [queue] --clear-routing-override という操作もあったが、これも効果がなかった。

gcloudコマンドで作成すると、この値は設定されないので、キューの作成・変更操作はコマンドラインでやった方が良さそう。

$ pipenv run gcloud tasks queues create test-dispatch --project [project]
WARNING: You are managing queues with gcloud, do not use queue.yaml or queue.xml in the future. More details at: https://cloud.google.com/tasks/docs/queue-yaml.
Created queue [test-dispatch].
$ pipenv run gcloud tasks queues describe test-dispatch --project [project]
name: projects/[project]/locations/us-central1/queues/test-dispatch
rateLimits:
  maxBurstSize: 100
  maxConcurrentDispatches: 1000
  maxDispatchesPerSecond: 500.0
retryConfig:
  maxAttempts: 100
  maxBackoff: 3600s
  maxDoublings: 16
  minBackoff: 0.100s
state: RUNNING

なお、Cloud Tasksキューは一度削除するとしばらく同一の名前では作ることができない。設定を間違えた場合は別の名前で作り直すしかない。

$ pipenv run gcloud tasks queues delete test --project [project]
WARNING: You are managing queues with gcloud, do not use queue.yaml or queue.xml in the future. More details at: https://cloud.google.com/tasks/docs/queue-yaml.
Are you sure you want to delete: 
[projects/[project]/locations/us-central1/queues/test] 
(Y/n)?  

Deleted queue [test].
$ pipenv run gcloud tasks queues create test --project [project]
WARNING: You are managing queues with gcloud, do not use queue.yaml or queue.xml in the future. More details at: https://cloud.google.com/tasks/docs/queue-yaml.
ERROR: (gcloud.tasks.queues.create) FAILED_PRECONDITION: The queue cannot be created because a queue with this name existed too recently.

実験コード

main.py

import flask
import os

from google.cloud import tasks_v2 as tasks

app = flask.Flask(__name__)
project = os.environ['GOOGLE_CLOUD_PROJECT']
service = os.environ['GAE_SERVICE']
version = os.environ['GAE_VERSION']
queue_region = "us-central1"
queue_name = 'test-dispatch'


@app.route('/enqueue1', methods=['POST'])
def enqueue1():
    print("Enqueue1 in PROJECT: {}, SERVICE: {}, VERSION: {}".format(project, service, version))
    request_task = {
        'app_engine_http_request': {
            'http_method': 'POST',
            'relative_uri': '/task'
        }
    }
    tasks_client = tasks.CloudTasksClient()
    tasks_queue = tasks_client.queue_path(project, queue_region, queue_name)
    tasks_client.create_task(tasks_queue, request_task)
    return 'ok'


@app.route('/enqueue2', methods=['POST'])
def enqueue2():
    print("Enqueue2 in PROJECT: {}, SERVICE: {}, VERSION: {}".format(project, service, version))
    request_task = {
        'app_engine_http_request': {
            'http_method': 'POST',
            'relative_uri': '/task',
            'app_engine_routing': {
                'service': service,
                'version': version,
                'instance': ''
            }
        }
    }
    tasks_client = tasks.CloudTasksClient()
    tasks_queue = tasks_client.queue_path(project, queue_region, queue_name)
    tasks_client.create_task(tasks_queue, request_task)
    return 'ok'


@app.route('/task', methods=['POST'])
def task():
    print("Tasks Running in PROJECT: {}, SERVICE: {}, VERSION: {}".format(project, service, version))
    return 'ok'


if __name__ == '__main__':
    app.run()

Pipfile

[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[packages]
flask = "*"
google = "*"
google-cloud-tasks = "==1.5.0"  # gloud-tasks-emulator not supports 2.0.0
gunicorn = "*"

[dev-packages]

[requires]
python_version = "3.7"

[scripts]
start = "flask run --debugger --reload"
deploy = "gcloud app deploy --project [project]"

app.yaml

runtime: python37
entrypoint: gunicorn -b :$PORT main:app

.gcloudignore

.gcloudignore
.git
.gitignore
__pycache__/
/setup.cfg
Pipfile
Pipfile.lock

デプロイ時の操作

$ pipenv install
$ pipenv lock -r > requirements.txt
$ pipevn run deploy