장기 실행 작업 모니터링

이 페이지에서는 AI 애플리케이션에서 장기 실행 작업(LRO)의 수명 주기를 관리하는 방법을 설명합니다.

메서드 호출을 완료하는 데 시간이 오래 걸릴 수 있는 장기 실행 작업 객체가 반환됩니다. 예를 들어 AI Applications API는 API 또는 클라이언트 라이브러리를 통해 documents.import를 호출하면 장기 실행 작업을 만듭니다. 작업 트랙은 처리 작업의 상태를 추적합니다.

AI Applications 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: 엔진으로 만든 AI 애플리케이션 데이터 스토어의 ID입니다. Google Cloud 콘솔 URL에서 데이터 스토어 ID는 engines/ 뒤, /data 앞에 표시됩니다.

Python

자세한 내용은 AI 애플리케이션 Python API 참조 문서를 참고하세요.

AI 애플리케이션에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 인증 설정을 참조하세요. 

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

자세한 내용은 AI 애플리케이션 Python API 참조 문서를 참고하세요.

AI 애플리케이션에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 인증 설정을 참조하세요. 

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

자세한 내용은 AI 애플리케이션 Python API 참조 문서를 참고하세요.

AI 애플리케이션에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 인증 설정을 참조하세요. 

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 메서드의 응답에 CANCELLED 상태 코드를 나타내는 "code": 1 오류가 포함됩니다.

        예를 들면 다음과 같습니다.

        {
  "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

자세한 내용은 AI 애플리케이션 Python API 참조 문서를 참고하세요.

AI 애플리케이션에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 인증 설정을 참조하세요. 

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