長時間実行オペレーションをモニタリングする

このページでは、Vertex AI Search で長時間実行オペレーション（LRO）のライフサイクルを管理する方法について説明します。

メソッドの呼び出しが完了するまでに時間がかかる場合、長時間実行オペレーション オブジェクトが返されます。たとえば、Discovery Engine API は、API またはクライアント ライブラリを介して documents.import を呼び出すときに、長時間実行オペレーションを作成します。オペレーションは、処理ジョブのステータスを追跡します。

Discovery Engine API が提供する長時間実行オペレーション メソッドを使用して、オペレーションのステータスを確認できます。オペレーションを一覧表示したり、ポーリングしたりすることもできます。

オペレーションのレコードは、オペレーションの終了後約 30 日間保持されます。それ以降は、オペレーションを表示または一覧表示できません。

長時間実行オペレーションを一覧表示する

以下では、 Google Cloudリソースのオペレーションを一覧表示する方法を示します。

REST

Google Cloud リソースの長時間実行オペレーションを一覧表示するには、次の手順を行います。

  • operations.list メソッドを呼び出します。

    curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    "https://discoveryengine.googleapis.com/v1/projects/PROJECT_ID/locations/global/collections/default_collection/dataStores/DATA_STORE_ID/operations"

    DATA_STORE_ID: エンジンで作成された Vertex AI Search データストアの ID。 Google Cloud コンソールの URL では、データストア ID は engines/ の後、/data の前に表示されます。

Python

詳細については、Vertex AI Search Python API のリファレンス ドキュメントをご覧ください。

Vertex AI Search に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、ローカル開発環境の認証の設定をご覧ください。

from typing import Optional

from google.cloud import discoveryengine
from google.longrunning import operations_pb2

# TODO(developer): Uncomment these variables before running the sample.
# project_id = "YOUR_PROJECT_ID"
# location = "YOUR_PROCESSOR_LOCATION"  # Options: "global"
# search_engine_id = "YOUR_SEARCH_ENGINE_ID"

# Create filter in https://google.aip.dev/160 syntax
# operations_filter = "YOUR_FILTER"


def list_operations_sample(
    project_id: str,
    location: str,
    search_engine_id: str,
    operations_filter: Optional[str] = None,
) -> operations_pb2.ListOperationsResponse:
    # Create a client
    client = discoveryengine.DocumentServiceClient()

    # The full resource name of the search engine branch.
    name = f"projects/{project_id}/locations/{location}/collections/default_collection/dataStores/{search_engine_id}"

    # Make ListOperations request
    request = operations_pb2.ListOperationsRequest(
        name=name,
        filter=operations_filter,
    )

    # Make ListOperations request
    response = client.list_operations(request=request)

    # Print the Operation Information
    for operation in response.operations:
        print(operation)

    return response

長時間実行オペレーションの詳細の取得

以下では、オペレーションの詳細を取得する方法を示します。

REST

長時間実行オペレーションのステータスを取得して詳細を表示する手順は次のとおりです。

  1. オペレーションの名前は、次のいずれかの方法で確認できます。

    • 長時間実行オペレーションを返すメソッドを呼び出したら、レスポンスを確認します。

      たとえば、documents.import を呼び出すと、レスポンスの先頭は次のようになります。

      {
  "operations": [
    {
      "name": "projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.discoveryengine.v1.ImportDocumentsMetadata",
      }
    }
  ]
}

      レスポンスの name 値は、オペレーションのステータスのクエリに使用できるオペレーション名を提供します。オペレーション名をコピーするときに、引用符は含めないでください。

    • 長時間実行オペレーションを一覧表示して、オペレーション名を取得します。

  2. オペレーションを作成したリソースで operations.get メソッドを呼び出します。

    curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    "https://discoveryengine.googleapis.com/v1/OPERATION_NAME"

    OPERATION_NAME: 情報を取得するオペレーションの名前を指定します。オペレーション名は、長時間実行オペレーションを一覧表示して確認できます。

    GET コマンドのレスポンスの最初の行は次のようになります。

    {
      "name": "projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.discoveryengine.v1.ImportDocumentsMetadata"
      }
    }

Python

詳細については、Vertex AI Search Python API のリファレンス ドキュメントをご覧ください。

Vertex AI Search に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、ローカル開発環境の認証の設定をご覧ください。

from google.cloud import discoveryengine
from google.longrunning import operations_pb2

# TODO(developer): Uncomment these variables before running the sample.
# Example: `projects/{project}/locations/{location}/collections/{default_collection}/dataStores/{search_engine_id}/branches/{0}/operations/{operation_id}`
# operation_name = "YOUR_OPERATION_NAME"


def get_operation_sample(operation_name: str) -> operations_pb2.Operation:
    # Create a client
    client = discoveryengine.DocumentServiceClient()

    # Make GetOperation request
    request = operations_pb2.GetOperationRequest(name=operation_name)
    operation = client.get_operation(request=request)

    # Print the Operation Information
    print(operation)

    return operation

長時間実行オペレーションをポーリングする

以下に、オペレーションのステータスをポーリングする方法を示します。

REST

長時間実行オペレーションが終了するまでポーリングする手順は次のとおりです。

  1. 次のコマンドを実行します。このコマンドは、各リクエスト間で 10 秒のバックオフを使用して、operations.get メソッドを繰り返し呼び出します。

    while true; \
    do curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    "https://discoveryengine.googleapis.com/v1/OPERATION_NAME"; \
    sleep 10; \
    done

    OPERATION_NAME: ポーリングするオペレーションの名前を指定します。オペレーション名は、長時間実行オペレーションを一覧表示して確認できます。例: projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789

  2. ステータスが "done": true になったら、ポーリング ジョブ（Control+Z）を停止します。

Python

詳細については、Vertex AI Search Python API のリファレンス ドキュメントをご覧ください。

Vertex AI Search に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、ローカル開発環境の認証の設定をご覧ください。

from time import sleep

from google.cloud import discoveryengine
from google.longrunning import operations_pb2

# TODO(developer): Uncomment these variables before running the sample.
# Example: `projects/{project}/locations/{location}/collections/{default_collection}/dataStores/{search_engine_id}/branches/{0}/operations/{operation_id}`
# operation_name = "YOUR_OPERATION_NAME"


def poll_operation_sample(
    operation_name: str, limit: int = 10
) -> operations_pb2.Operation:
    # Create a client
    client = discoveryengine.DocumentServiceClient()

    # Make GetOperation request
    request = operations_pb2.GetOperationRequest(name=operation_name)

    for _ in range(limit):
        operation = client.get_operation(request=request)
        # Print the Operation Information
        print(operation)

        # Stop polling when Operation is no longer running
        if operation.done:
            break

        # Wait 10 seconds before polling again
        sleep(10)

    return operation

長時間実行オペレーションをキャンセルする

オペレーションをキャンセルする方法は次のとおりです。

REST

長時間実行オペレーションをキャンセルする手順は次のとおりです。

  • operations.cancel メソッドを呼び出します。

    curl -X post \
   -H "Authorization: Bearer $(gcloud auth print-access-token)" \
   "https://discoveryengine.googleapis.com/v1/OPERATION_NAME":cancel

    OPERATION_NAME: キャンセルするオペレーションの名前を指定します。オペレーション名は、長時間実行オペレーションを一覧表示して確認できます。例: projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789

    API 呼び出しが行われると、サーバーはオペレーションのキャンセルを試みます。表示される結果と実行できる操作は次のとおりです。

    • "code": 400 と "status": "FAILED_PRECONDITION" のエラーは、リクエストをキャンセルできなかったことを示します。

    • 正常にキャンセルされると、空の JSON オブジェクトが返されます。解約を確認するには:

      • operations.get メソッドを使用します。

      • オペレーションが正常にキャンセルされると、operations.get メソッドからのレスポンスに "code": 1 エラーが含まれます。これは CANCELLED ステータス コードを表します。

        次に例を示します。

        {
  "name": "projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.discoveryengine.v1alpha.ImportDocumentsMetadata",
    "createTime": "2025-04-28T21:29:21.199190Z",
    "updateTime": "2025-04-28T21:31:29.076865Z"
  },
  "done": true,
  "error": {
    "code": 1,
    "message": "Operation projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789 is cancelled."
  }
}

Python

詳細については、Vertex AI Search Python API のリファレンス ドキュメントをご覧ください。

Vertex AI Search に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、ローカル開発環境の認証の設定をご覧ください。

from google.cloud import discoveryengine
from google.longrunning import operations_pb2

# TODO(developer): Uncomment these variables before running the sample.
# Example: `projects/{project}/locations/{location}/collections/{default_collection}/dataStores/{search_engine_id}/branches/{0}/operations/{operation_id}`
# operation_name = "YOUR_OPERATION_NAME"


def cancel_operation_sample(operation_name: str) -> None:
    # Create a client
    client = discoveryengine.DocumentServiceClient()

    # Make CancelOperation request
    request = operations_pb2.CancelOperationRequest(name=operation_name)
    client.cancel_operation(request=request)

    return