push キューの Cloud Tasks への移行（Python）

このページでは、push キューコードをタスクキューからクラウドタスクに移行する方法について説明します。現在のところ、クラウドタスクが App Engine の push キューを使用する場合のおすすめの方法となっています。

Cloud Tasks は、Task Queues RPC API でアクセスするのと同じサービスにアクセスします。つまり、既存の push キューや push タスクを再作成する必要はありません。ただし、Cloud Tasks API を使用するには、push キューまたは push タスクを作成、操作するコードを移行する必要があります。

Cloud Tasks REST と RPC API、Cloud Tasks クライアント ライブラリ、Google Cloud CLI、Google Cloud コンソールを使用して、push キューと push タスクを作成して操作できます。このページでは、gcloud CLI と Cloud Tasks クライアント ライブラリを使用した場合の例を示します。

クラウドタスクでは、すべてのキューが push キューとして動作します。このガイドの残りの部分とクラウドタスクのドキュメントでは、「キュー」という用語が「push キュー」という用語と同義で使用されています。同様に、「タスク」という用語も「push タスク」という用語と同義です。

Cloud Tasks で利用できない機能

現在、Cloud Tasks では次の機能は使用できません。

• Datastore トランザクションの中でタスクをキューに追加する
• ワーカー サービスの代わりに遅延タスク ライブラリを使用する
• マルチテナント アプリケーションでタスクを扱う
• ローカル開発サーバーでのシミュレーション
• 非同期でのタスクの追加

料金と割り当て

push キューをクラウドタスクに移行すると、アプリの料金と割り当てに影響する場合があります。

料金

クラウドタスクには独自の料金が設定されています。タスクキューと同様に、App Engine アプリにタスクを使用してリクエストを送信するとアプリに費用がかかることがあります。

割り当て

クラウドタスクの割り当ては、タスクキューの割り当てとは異なります。タスクキューと同様に、Cloud Tasks から App Engine アプリにリクエストを送信すると、App Engine リクエストの割り当てに影響する場合があります。

移行前

以降のセクションでは、push キューを Cloud Tasks に移行する前の設定手順について説明します。

pull キューの移行

開始する前に、pull キューを移行してから、このガイドの手順に沿って push キューを移行してください。push キューの移行後に pull キューを移行することは、queue.yaml ファイルの必要な使用がクラウドタスクで予期しない動作を引き起こす可能性があるため、おすすめしません。

キュー構成の保護

クラウドタスクへの移行プロセスを開始すると、queue.yaml ファイルを変更する際に予期しない動作が発生する場合があるためおすすめしません。キュー構成を queue.yaml ファイルによる変更から保護するには、次の手順を行います。

1. 今後のデプロイで queue.yaml ファイルを省略するように gcloud CLI を構成します。

   queue.yaml ファイルを .gcloudignore ファイルに追加します。.gcloudignore ファイルがすでに存在しているか確認するには、ターミナルでデバイスの最上位レベルのディレクトリから次のコマンドを実行します。ファイルが存在する場合、このコマンドでファイル名が出力されます。

   ls -a | grep .gcloudignore

   .gcloudignore ファイルの詳細については、.gcloudignore リファレンスをご覧ください。

2. queue.yaml ファイルの権限を制限します。

   キュー構成の保護に関するガイドに記載されているおすすめの方法に従ってください。

3. クラウドタスクと queue.yaml ファイルについて学習する（省略可）。

   Cloud Tasks API を使用してキュー構成を管理する場合、queue.yaml ファイルをデプロイすると、Cloud Tasks によって設定された構成がオーバーライドされ、予期しない動作が発生する可能性があります。詳細については、キュー管理と queue.yaml の使用をご覧ください。

Cloud Tasks API の有効化

Cloud Tasks API を有効にするには、API ライブラリの [Cloud Tasks API] で [有効にする] をクリックします。[有効にする] ボタンの代わりに [管理] ボタンが表示されている場合は、以前にプロジェクトの Cloud Tasks API を有効にしているため、再び有効にする必要はありません。

Cloud Tasks API に対するアプリの認証

Cloud Tasks API に対してアプリを認証する必要があります。このセクションでは、2 つの異なるユースケースの認証について説明します。

アプリをローカルで開発またはテストするには、サービス アカウントを使用することをおすすめします。サービス アカウントを設定してアプリに接続する手順については、サービス アカウントの認証情報を手動で取得して提供するをご覧ください。

アプリを App Engine にデプロイする場合、新たに認証を行う必要はありません。アプリケーションのデフォルト認証情報（ADC）では、App Engine アプリの認証の詳細が推測されます。

gcloud CLI のダウンロード

以前にインストールしていない場合は、gcloud CLI をダウンロードしてインストールし、Cloud Tasks API で gcloud CLI を使用します。gcloud CLI がすでにインストールされている場合は、ターミナルで次のコマンドを実行します。

gcloud components update

Cloud クライアント ライブラリのインポート

App Engine アプリで Cloud Tasks クライアント ライブラリを使用するには、次の手順を行います。

1. app.yaml ファイルを更新します。 Python のバージョンに応じた手順に沿って操作します。

   Python 2

   Python 2 アプリの場合は、grpcio ライブラリと setuptools ライブラリの最新バージョンを追加します。

   app.yaml ファイルの例を次に示します。

   runtime: python27
   threadsafe: yes
   api_version: 1
   
   libraries:
   - name: grpcio
     version: latest
   - name: setuptools
     version: latest
   

   Python 3

   Python 3 アプリの場合は、app.yaml ファイルの runtime 要素に、サポートされている Python 3 バージョンを指定します。例:

   runtime: python310 # or another support version
   

   Python 3 ランタイムでは、ライブラリが自動的にインストールされるため、以前の Python 2 ランタイムの組み込みライブラリを指定する必要はありません。移行時に Python 3 アプリが他の以前のバンドル サービスを使用している場合は、引き続き必要な組み込みライブラリを指定できます。それ以外の場合は、app.yaml ファイル内の不要な行を削除してください。

2. requirements.txt ファイルを更新します。Python のバージョンに応じた手順に沿って操作します。

   Python 2

   Cloud Tasks 用の Cloud クライアント ライブラリを requirements.txt ファイルの依存関係のリストに追加します。

   google-cloud-tasks
   

   次に pip install -t lib -r requirements.txt を実行して、アプリで使用可能なライブラリのリストを更新します。

   Python 3

   Cloud Tasks 用の Cloud クライアント ライブラリを requirements.txt ファイルの依存関係のリストに追加します。

   google-cloud-tasks
   

   Python 3 ランタイムにアプリをデプロイするときに、これらの依存関係が自動的にインストールされるため、lib フォルダが存在する場合は削除します。

3. Python 2 アプリで組み込みライブラリまたはコピーされたライブラリを使用している場合、そのパスを app.yaml ファイルと同じフォルダにある appengine_config.py ファイル内に指定する必要があります。

   import pkg_resources
   from google.appengine.ext import vendor
   
   # Set PATH to your libraries folder.
   PATH = 'lib'
   # Add libraries installed in the PATH folder.
   vendor.add(PATH)
   # Add libraries to pkg_resources working set to find the distribution.
   pkg_resources.working_set.add_entry(PATH)
   

   appengine_config.py ファイルでは、現在の作業ディレクトリが lib フォルダのあるディレクトリであることを想定しています。単体テストなどの一部のケースでは、現在の作業ディレクトリはこれとは異なる場合があります。エラーが発生しないようにするために、lib フォルダのフルパスを次のようにして明示的に指定できます。

   import os
   path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'lib')

4. Task Queues API から push キューを使用するファイルに Cloud Tasks クライアント ライブラリをインポートします。

   from google.cloud import tasks

   push キューの作成や、push キューとのやり取りを行うすべてのコードの Cloud Tasks への完全な移行が正常に完了したら、Task Queues API（たとえば、from google.appengine.api import taskqueue）をインポートするステートメントを削除します。

キューの作成と管理

このセクションでは、Cloud Tasks API を使用してキューを作成および管理する方法について説明します。

Cloud Tasks では、キューの作成や管理に queue.yaml ファイルを使用しません。その代わりに、Cloud Tasks API が使用されます。queue.yaml ファイルと Cloud Tasks API の両方の使用はおすすめしませんが、アプリによってはタスクキューから Cloud Tasks への移行が避けられない場合があります。ベスト プラクティスについては、キュー管理と queue.yaml の使用をご覧ください。

キューの作成

アプリでプログラム的にキューが作成される場合、またはコマンドラインから追加のキューを作成する場合は、このセクションをお読みください。

クラウドタスクのキューの名前は projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID という形式に従います。キュー名の LOCATION_ID 部分が、Google Cloud のリージョンに対応しています。キュー名の QUEUE_ID の部分は、Task Queues キューの name フィールドと同じになります。キュー名は、プロジェクト、リージョン、およびユーザーが指定した QUEUE_ID に基づいて、キューが作成される際に生成されます。

一般に、キューのロケーション（つまりリージョン）はアプリのリージョンと同じである必要があります。このルールに対する 2 つの例外は、europe-west リージョンを使用するアプリと us-central リージョンを使用するアプリの場合です。クラウドタスクのこれらのリージョンは、europe-west1 および us-central1 とそれぞれ呼ばれます。

キューを作成する際にオプションのキュー構成を指定できるだけでなく、作成の後にキューを更新することもできます。

既存のキューを再作成する必要はありません。その代わりに、このガイドの関連部分を読んで、既存のキューを操作するコードを移行してください。

キュー名の再利用

同じプロジェクトとロケーション（つまりリージョン）に同じキュー ID を持つキューを作成する場合、キューを削除してから 7 日間待つ必要があります。

次の例では、クラウドタスクを使用して 2 つのキューを作成しています。最初のキューのキュー ID は queue-blue です。すべてのタスクが 5/s のレートでバージョン v2 のサービス task-module に送信されるように構成されています。2 番目のキューのキュー ID は queue-red です。1/s のレートでタスクがディスパッチされます。これらの両方が、ロケーション us-central1 の my-project-id というプロジェクト ID のプロジェクトに作成されています。これは、タスクキューでキューの作成に相当する Cloud Tasks です。

gcloud tasks queues create queue-blue \
--max-dispatches-per-second=5 \
--routing-override=service:task-module,version:v2

gcloud tasks queues create queue-red \
--max-dispatches-per-second=1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue_blue_name = 'queue-blue'
# queue_red_name = 'queue-red'

parent = f"projects/{project}/locations/{location}"

queue_blue = {
    "name": client.queue_path(project, location, queue_blue_name),
    "rate_limits": {"max_dispatches_per_second": 5},
    "app_engine_routing_override": {"version": "v2", "service": "task-module"},
}

queue_red = {
    "name": client.queue_path(project, location, queue_red_name),
    "rate_limits": {"max_dispatches_per_second": 1},
}

queues = [queue_blue, queue_red]
for queue in queues:
    response = client.create_queue(parent=parent, queue=queue)
    print(response)

詳しくは、Cloud Tasks リファレンスの Cloud Tasks キューの作成をご覧ください。

キューの処理速度の設定

次の表に、Task Queues と Cloud Tasks で異なるフィールドを示します。

キューの作成やキューの更新の際に、キューの処理速度を設定できます。以下の例では、Cloud Tasks を使用して、すでに作成された queue-blue という名前のキューの処理速度を設定しています。queue-blue が queue.yaml を使用して作成または構成された場合、以下の例では、20 の max_dispatches_per_second の値に基づいて max_burst_size がリセットされます。この Cloud Tasks は、Task Queues でキュー処理率を設定する場合と同等です。

gcloud tasks queues update queue-blue \
--max-dispatches-per-second=20 \
--max-concurrent-dispatches=10

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue-blue'

# Get queue object
queue_path = client.queue_path(project, location, queue)
queue = client.get_queue(name=queue_path)

# Update queue object
queue.rate_limits.max_dispatches_per_second = 20
queue.rate_limits.max_concurrent_dispatches = 10

response = client.update_queue(queue=queue)
print(response)

詳細については、レート上限の定義をご覧ください。

キューの無効化と再開

クラウドタスクでは、タスクキューの「無効」という用語と同じ意味で、「一時停止」という用語が使用されます。キューを一時停止すると、キューが再開されるまで、キュー内のタスクの実行が停止します。ただし、一時停止中のキューに対してもタスクを追加し続けることができます。Cloud Tasks では、Task Queues の場合と同じ意味で、「再開」という用語が使用されます。

次の例では、 queue1 というキュー ID のキューを一時停止しています。これは、Task Queues でキューを無効にするに相当する Cloud Tasks です。

gcloud tasks queues pause 

queue1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.pause_queue(name=queue_path)

詳細については、Cloud Tasks リファレンスのキューの一時停止をご覧ください。

キューの削除

キューを削除した場合、同じ名前のキューを作成するまで 7 日間待つ必要があります。7 日間待てない場合は、キューからすべてのタスクを削除し、キューを再構成することを検討してください。

次の例では、キュー ID が queue1 のキューを削除しています。これは、Task Queues でキューを削除するに相当する Cloud Tasks です。

gcloud tasks queues delete 

queue1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.delete_queue(name=queue_path)

Cloud Tasks リファレンスのキューの削除をご覧ください。

タスクの作成と管理

このセクションでは、Cloud Tasks API を使用してタスクを作成および管理する方法について説明します。

タスクの作成

次の表に、Task Queues と Cloud Tasks で異なるフィールドを示します。

次の例では、/update_counter ハンドラで指定された worker という名前の App Engine サービスにルーティングするタスクを作成します。これは、タスクキューにおいてタスクを作成することと同等の Cloud Tasks です。

gcloud tasks create-app-engine-task --queue=default \
--method=POST --relative-uri=/update_counter --routing=service:worker \
--body-content=10

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'
amount = 10

parent = client.queue_path(project, location, queue)

task = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter",
        "app_engine_routing": {"service": "worker"},
        "body": str(amount).encode(),
    }
}

response = client.create_task(parent=parent, task=task)
eta = response.schedule_time.strftime("%m/%d/%Y, %H:%M:%S")
print(f"Task {response.name} enqueued, ETA {eta}.")

詳細については、Cloud Tasks リファレンスの App Engine タスクの作成をご覧ください。

対象サービスとルーティングの指定

App Engine タスクに対して、App Engine のターゲット サービス、バージョン、インスタンスの指定はオプションです。デフォルトでは、App Engine のタスクは、タスクを実行するときにデフォルトのサービス、バージョン、インスタンスにルーティングされます。

タスクを作成する際に、タスクの app_engine_routing プロパティを設定して、タスクに別のApp Engineサービス、バージョン、またはインスタンスを指定します。

特定のキューで実行されているすべてのタスクを同じ App Engine のサービス、バージョン、インスタンスにルーティングするには、キューの app_engine_routing_override プロパティを設定します。

詳しくは、Cloud Tasks リファレンスのルーティングの構成をご覧ください。

データのハンドラへの引き渡し

タスクキューの場合と同様に、クラウドタスクを使用してハンドラにデータを渡す方法は 2 つあります。相対 URI でデータをクエリ パラメータとして渡す方法と、HTTP メソッドの POST または PUT を使用してリクエスト本文でデータを渡す方法があります。

クラウドタスクでは、タスクキューで「ペイロード」という用語を使用するのと同じ意味で、「本文」という用語を使用します。クラウドタスクでは、本文のコンテンツ タイプのデフォルトはプレーン テキストではなくオクテット ストリームです。本文コンテンツ タイプを設定するには、ヘッダーで指定します。

次の例では、2 つの異なる方法で、ハンドラ /update_counter にキーを渡しています。これは、タスクキューでハンドラにデータを渡すことに相当するクラウドタスクです。

gcloud tasks create-app-engine-task --queue=default --method=GET  \
--relative-uri=

/update_counter

?key=blue --routing=service:worker

gcloud tasks create-app-engine-task --queue=default --method=POST \
--relative-uri=

/update_counter

 --routing=service:worker \
--body-content="{'key': 'blue'}"

import json

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'

parent = client.queue_path(project, location, queue)

task1 = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter?key=blue",
        "app_engine_routing": {"service": "worker"},
    }
}

task2 = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter",
        "app_engine_routing": {"service": "worker"},
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"key": "blue"}).encode(),
    }
}

response = client.create_task(parent=parent, task=task1)
print(response)
response = client.create_task(parent=parent, task=task2)
print(response)

タスクに名前を付ける

タスク名の指定は省略可です。タスク名を指定しない場合、Cloud Tasks によってタスク ID が生成され、タスクを作成する際に指定したキューに基づいてプロジェクトとロケーション（つまりリージョン）を推定することによりタスク名が作成されます。

タスク名の形式は projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID/tasks/TASK_ID です。タスク名の TASK_ID の部分が Task Queues タスクの name フィールドと同等になります。

タスク名の再利用

タスク名を再利用する際には、待機する必要があります。待機する時間は、タスクをディスパッチするキューがクラウドタスクまたはタスクキューのどちらで作成されたかによって異なります。

タスクキューを使用して作成されたキュー（デフォルト キューを含む）のタスクの場合、元のタスクが削除または実行された後、約 9 日間待つ必要があります。クラウドタスクを使用して作成されたキューのタスクの場合は、元のタスクが削除または実行されてから約 1 時間待ちます。

次の例では、TASK_ID がfirst-try に設定されたタスクを作成し、デフォルトのキューに追加します。クラウドタスクにおいてこれは、タスクキューの名前付けタスクに相当します。

gcloud tasks create-app-engine-task first-try --queue=default \
--method=GET --relative-uri=

/url/path

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'
# task_name = 'first-try'

parent = client.queue_path(project, location, queue)

task = {
    "name": client.task_path(project, location, queue, task_name),
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.GET,
        "relative_uri": "/url/path",
    },
}
response = client.create_task(parent=parent, task=task)
print(response)

失敗したタスクの再試行

タスクの再試行の構成は、キューの作成中に行うか、キューの更新によって設定できます。Task Queues のフィールドと、それぞれに対応する Cloud Tasks のフィールドを次の表に示します。

タスク固有の再試行パラメータ

タスクキューで構成されたタスク固有の再試行パラメータはクラウドタスクでも機能しますが、編集や新規タスクの設定はできません。タスク固有の再試行パラメータを持つタスクの再試行パラメータを変更するには、目的の再試行パラメータを持つクラウド タスクキューを使用してタスクを再作成します。

次の例に、さまざまな再試行のシナリオを示します。

• fooqueue では、タスクは最初の試行から 2 日間を期限とし、7 回を上限として再試行されます。両方の上限に達すると、タスクは完全に失敗し、再試行されなくなります。
• barqueue では、App Engine は、再試行の回数に応じて間隔を長くしながらタスクを再試行し、最大バックオフに達すると、最大間隔で無限に再試行します（リクエストの間隔は 10 秒、20 秒、30 秒、...、190 秒、200 秒、200 秒、... となります）。
• bazqueue では、再試行間隔は 10 秒から開始し、3 回 2 倍ずつ間隔を長くして、その後は直線的に間隔が広げられます。最終的には、最大間隔で無限に再試行します（リクエストの間隔は 10 秒、20 秒、40 秒、80 秒、160 秒、240 秒、300 秒、300 秒、... となります）。

クラウドタスクにおいてこれは、タスクキューの再試行タスクに相当します。

gcloud tasks queues create fooqueue \
--max-attempts=7 \
--max-retry-duration=172800s  #2*60*60*24 seconds in 2 days

gcloud tasks queues create barqueue \
--min-backoff=10s \
--max-backoff=200s \
--max-doublings=0

gcloud tasks queues create bazqueue \
--min-backoff=10s \
--max-backoff=300s \
--max-doublings=3

from google.protobuf import duration_pb2

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# fooqueue = 'fooqueue'
# barqueue = 'barqueue'
# bazqueue = 'bazqueue'

parent = f"projects/{project}/locations/{location}"

max_retry = duration_pb2.Duration()
max_retry.seconds = 2 * 60 * 60 * 24

foo = {
    "name": client.queue_path(project, location, fooqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"max_attempts": 7, "max_retry_duration": max_retry},
}

min = duration_pb2.Duration()
min.seconds = 10

max = duration_pb2.Duration()
max.seconds = 200

bar = {
    "name": client.queue_path(project, location, barqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"min_backoff": min, "max_backoff": max, "max_doublings": 0},
}

max.seconds = 300
baz = {
    "name": client.queue_path(project, location, bazqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"min_backoff": min, "max_backoff": max, "max_doublings": 3},
}

queues = [foo, bar, baz]
for queue in queues:
    response = client.create_queue(parent=parent, queue=queue)
    print(response)

詳細については、Cloud Tasks リファレンスの再試行パラメータの設定をご覧ください。

タスクのキューからの削除

タスクを削除した場合、次に同じ名前のタスクを作成するまで、タスクが queue.yaml ファイルを使用して作成されたキューにある場合は 9 日間、クラウドタスクを使用して作成されたキューにある場合は 1 時間、それぞれ待つ必要があります。

次の例では、キュー ID が queue1 のキューからタスク ID が foo のタスクを削除しています。これは、タスクキューにおいてタスクを削除するのと同等のクラウドタスクです。

gcloud tasks delete foo --queue=queue1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

task_path = client.task_path(project, location, queue, "foo")
response = client.delete_task(name=task_path)

詳細については、Cloud Tasks リファレンスのタスクをキューから削除するをご覧ください。

タスクの完全な削除

次の例では、キュー ID が queue1 のキューからすべてのタスクを完全に削除しています。クラウドタスクにおいてこれは、タスクキューのタスクを完全に削除するに相当します。

gcloud tasks queues purge 

queue1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.purge_queue(name=queue_path)

詳細については、Cloud Tasks リファレンスのキューからすべてのタスクを完全に削除するをご覧ください。

次のステップ

• Cloud Tasks のドキュメント
• Cloud Tasks のクライアント ライブラリ
• Cloud Tasks の REST リファレンスの概要
• Cloud Tasks の RPC リファレンスの概要
• チュートリアルについては、Codelab: App Engine タスクキューの push キューから Cloud Tasks への移行をご覧ください。